ntex_rt/

signals.rs

use std::sync::{Arc, OnceLock, atomic::AtomicBool, atomic::Ordering, mpsc};
use std::{cell::RefCell, future::poll_fn, panic, task::Poll};

use atomic_waker::AtomicWaker;
use ntex_error::Backtrace;
use parking_lot::{Mutex, MutexGuard};

use crate::System;

thread_local! {
    static STOP: RefCell<Option<oneshot::Sender<()>>> = const { RefCell::new(None) };
    static HANDLERS: RefCell<Vec<oneshot::Sender<Arc<[Signal]>>>> = RefCell::default();
}

static CUR_SYS: Mutex<Option<System>> = Mutex::new(None);
// Mirrors `CUR_SYS`, the panic hook must not lock it.
static ENABLED: AtomicBool = AtomicBool::new(false);
static SIGS: Mutex<Vec<Signal>> = Mutex::new(Vec::new());
static HND_WAKER: AtomicWaker = AtomicWaker::new();
// The panic hook runs on the panicking thread, some runtimes (compio) abort
// the process if a task is woken there. Panics are woken from a dedicated thread.
static PANIC_WAKER: OnceLock<mpsc::SyncSender<()>> = OnceLock::new();

/// Process and application signals delivered to the runtime.
#[derive(Clone, Debug)]
pub enum Signal {
    /// SIGHUP
    Hup,
    /// SIGINT
    Int,
    /// SIGTERM
    Term,
    /// SIGQUIT
    Quit,
    /// Application panic
    Panic(PanicSource),
}

/// Source of a panic signal.
#[derive(Clone, Debug)]
pub enum PanicSource {
    /// A `SIGSEGV` or `SIGABRT` signal was received.
    ///
    /// Delivery is best effort; a fatal fault can terminate the process before
    /// the runtime or server handles this notification.
    Sig(&'static str),
    /// An application panic and its captured backtrace.
    App(Arc<str>, Backtrace),
}

/// Registers interest in the next batch of signals.
///
/// The returned one-shot receiver handles one notification. Call this function
/// again after each notification to continue receiving signals.
pub fn signal() -> oneshot::AsyncReceiver<Arc<[Signal]>> {
    let (tx, rx) = oneshot::async_channel();
    System::current().handle().spawn_detached(async move {
        HANDLERS.with(|handlers| {
            handlers.borrow_mut().push(tx);
        });
    });

    rx
}

/// Returns whether signal handling is enabled.
pub fn is_enabled() -> bool {
    CUR_SYS.lock().is_some()
}

type Registration = MutexGuard<'static, Option<System>>;

/// Registers the system as the signal handler.
///
/// Returns the lock guard, so handlers are installed before other systems
/// can register or unregister.
fn register_system(sys: &System) -> Option<Registration> {
    let mut cur = CUR_SYS.lock();
    if cur.is_some() {
        None
    } else {
        *cur = Some(sys.clone());
        ENABLED.store(true, Ordering::Release);

        let (tx, rx) = oneshot::async_channel();
        sys.handle().spawn_detached(signals(rx));
        STOP.with(|stop| {
            *stop.borrow_mut() = Some(tx);
        });
        Some(cur)
    }
}

/// Unregisters the system if it handles signals.
///
/// Returns the lock guard, so handlers are removed before other systems
/// can register.
fn unregister_system(sys: &System) -> Option<Registration> {
    let mut cur = CUR_SYS.lock();
    if cur.as_ref().is_some_and(|cur| cur.id() == sys.id()) {
        cur.take();
        ENABLED.store(false, Ordering::Release);
        sys.handle().spawn_detached(async move {
            STOP.with(|stop| {
                if let Some(tx) = stop.borrow_mut().take() {
                    let _ = tx.send(());
                }
            });
        });
        Some(cur)
    } else {
        None
    }
}

/// Queue signal and wake the system.
///
/// Must not be called from a signal handler.
fn handle_signal(sig: Signal) {
    SIGS.lock().push(sig);
    HND_WAKER.wake();
}

#[cfg(target_family = "unix")]
/// Signal delivery thread handle and `SIGUSR2` handler id.
static SIG_HANDLERS: Mutex<(
    Option<signal_hook::iterator::Handle>,
    Option<signal_hook::SigId>,
)> = Mutex::new((None, None));

#[cfg(target_family = "unix")]
/// Register signal handler.
///
/// Returns `false` if signals are handled by another system.
pub(crate) fn start(sys: &System) -> bool {
    static ONCE: std::sync::Once = std::sync::Once::new();

    if let Some(_registration) = register_system(sys) {
        use nix::sys::signal;
        use signal_hook::consts::signal::{SIGHUP, SIGINT, SIGQUIT, SIGTERM, SIGUSR2};
        use signal_hook::{iterator::Signals, low_level::register};

        ONCE.call_once(|| {
            if let Err(e) = start_fatal_signal_thread() {
                log::error!("Cannot start fatal signal handling thread: {e:?}");
            }

            // Use u128 for alignment.
            let buf = Vec::leak(vec![0u128; 4096]);
            let stack = libc::stack_t {
                ss_sp: buf.as_ptr() as *mut libc::c_void,
                ss_flags: 0,
                ss_size: std::mem::size_of_val(buf),
            };
            let mut old = libc::stack_t {
                ss_sp: std::ptr::null_mut(),
                ss_flags: 0,
                ss_size: 0,
            };
            let result = unsafe { libc::sigaltstack(&raw const stack, &raw mut old) };
            if result != 0 {
                log::error!("Cannot set signal stack");
            }

            let sig_action = signal::SigAction::new(
                signal::SigHandler::Handler(sig_segv),
                signal::SaFlags::SA_ONSTACK,
                signal::SigSet::empty(),
            );
            unsafe {
                match signal::sigaction(signal::SIGSEGV, &sig_action) {
                    Ok(prev) => _ = PREV_SIGSEGV.set(prev),
                    Err(_) => log::error!("Cannot install signal handler for SIGSEGV"),
                }
                match signal::sigaction(signal::SIGABRT, &sig_action) {
                    Ok(prev) => _ = PREV_SIGABRT.set(prev),
                    Err(_) => log::error!("Cannot install signal handler for SIGABRT"),
                }
            }
        });

        // signal handlers only write to a self-pipe, signals are
        // dispatched from a regular thread
        let handle = match Signals::new([SIGHUP, SIGINT, SIGTERM, SIGQUIT]) {
            Ok(mut signals) => {
                let handle = signals.handle();
                let result = std::thread::Builder::new()
                    .name("ntex-rt:signals".to_string())
                    .spawn(move || {
                        for sig in signals.forever() {
                            handle_signal(match sig {
                                SIGHUP => Signal::Hup,
                                SIGINT => Signal::Int,
                                SIGTERM => Signal::Term,
                                SIGQUIT => Signal::Quit,
                                _ => continue,
                            });
                        }
                    });
                match result {
                    Ok(_) => Some(handle),
                    Err(e) => {
                        log::error!("Cannot start signal handling thread: {e:?}");
                        None
                    }
                }
            }
            Err(e) => {
                log::error!("Cannot install signal handlers: {e:?}");
                None
            }
        };

        let usr2 = unsafe { register(SIGUSR2, || crate::watchdog::sig_usr2()) };
        if usr2.is_err() {
            log::error!("Cannot install signal handler for SIGUSR2");
        }
        *SIG_HANDLERS.lock() = (handle, usr2.ok());
        true
    } else {
        false
    }
}

#[cfg(target_family = "unix")]
/// Unregister signal handler.
pub(crate) fn stop(sys: &System) {
    if let Some(_registration) = unregister_system(sys) {
        let (handle, usr2) = std::mem::take(&mut *SIG_HANDLERS.lock());
        if let Some(handle) = handle {
            // the thread exits and unregisters handlers
            handle.close();
        }
        if let Some(usr2) = usr2 {
            signal_hook::low_level::unregister(usr2);
        }
    }
}

#[cfg(target_family = "windows")]
static CTRLC_ENABLED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

#[cfg(target_family = "windows")]
/// Register signal handler.
///
/// Signals are handled by oneshots, you have to re-register
/// after each signal. Returns `false` if signals are handled by another system.
pub(crate) fn start(sys: &System) -> bool {
    use std::sync::atomic::Ordering;
    static ONCE: std::sync::Once = std::sync::Once::new();

    if let Some(_registration) = register_system(sys) {
        // the handler cannot be removed, it ignores signals while disabled
        ONCE.call_once(|| {
            let result = ctrlc::set_handler(|| {
                if CTRLC_ENABLED.load(Ordering::Acquire) {
                    handle_signal(Signal::Int);
                }
            });
            if let Err(e) = result {
                log::error!("Cannot install Ctrl-C handler: {e:?}");
            }
        });
        CTRLC_ENABLED.store(true, Ordering::Release);
        true
    } else {
        false
    }
}

#[cfg(target_family = "windows")]
/// Unregister signal handler.
pub(crate) fn stop(sys: &System) {
    if let Some(_registration) = unregister_system(sys) {
        CTRLC_ENABLED.store(false, std::sync::atomic::Ordering::Release);
        log::info!("Signals handling is disabled");
    }
}

async fn signals(rx: oneshot::AsyncReceiver<()>) {
    let mut rx = std::pin::pin!(rx);

    poll_fn(|cx| {
        if rx.as_mut().poll(cx).is_ready() {
            Poll::Ready(())
        } else {
            HND_WAKER.register(cx.waker());

            let sigs = std::mem::take(&mut *SIGS.lock());
            if !sigs.is_empty() {
                let sigs: Arc<[Signal]> = Arc::from(sigs);

                HANDLERS.with(|handlers| {
                    for tx in handlers.borrow_mut().drain(..) {
                        let _ = tx.send(sigs.clone());
                    }
                });
            }

            Poll::Pending
        }
    })
    .await;
}

#[cfg(target_family = "unix")]
static PREV_SIGSEGV: std::sync::OnceLock<nix::sys::signal::SigAction> = std::sync::OnceLock::new();
#[cfg(target_family = "unix")]
static PREV_SIGABRT: std::sync::OnceLock<nix::sys::signal::SigAction> = std::sync::OnceLock::new();

// Both ends stay open for the process lifetime: a handler must never write to a
// recycled descriptor or raise SIGPIPE if the delivery thread exits.
#[cfg(target_family = "unix")]
static FATAL_PIPE: OnceLock<(
    libc::pid_t,
    std::os::unix::net::UnixStream,
    std::os::unix::net::UnixStream,
)> = OnceLock::new();

#[cfg(target_family = "unix")]
fn start_fatal_signal_thread() -> std::io::Result<()> {
    use std::{io::Read, os::unix::net::UnixStream};

    let (read, write) = UnixStream::pair()?;
    write.set_nonblocking(true)?;
    let (_, read, _) = FATAL_PIPE.get_or_init(|| (unsafe { libc::getpid() }, read, write));
    std::thread::Builder::new()
        .name("ntex-rt:crashes".to_string())
        .spawn(move || {
            let mut read = read;
            let mut byte = [0];
            while read.read_exact(&mut byte).is_ok() {
                let name = match byte[0] {
                    b'A' => "SIGABRT",
                    b'S' => "SIGSEGV",
                    _ => continue,
                };
                if ENABLED.load(Ordering::Acquire) {
                    handle_signal(Signal::Panic(PanicSource::Sig(name)));
                }
            }
        })?;
    Ok(())
}

#[cfg(target_family = "unix")]
extern "C" fn sig_segv(v: i32) {
    use std::os::fd::AsRawFd;

    use nix::errno::Errno;
    use nix::sys::signal::{self, SaFlags, SigAction, SigHandler, SigSet};

    let errno = Errno::last_raw();
    let (sig, prev, byte) = if v == libc::SIGABRT {
        (signal::SIGABRT, &PREV_SIGABRT, b"A")
    } else {
        (signal::SIGSEGV, &PREV_SIGSEGV, b"S")
    };
    if let Some((pid, _, write)) = FATAL_PIPE.get() {
        unsafe {
            // A forked child must not shut down its parent's server.
            if libc::getpid() == *pid {
                let _ = libc::write(write.as_raw_fd(), byte.as_ptr().cast(), 1);
            }
        }
    }

    // restore the previous handler, it handles the signal raised again by
    // the faulting instruction or by `abort()`, otherwise the process never exits
    let prev = prev
        .get()
        .copied()
        .unwrap_or_else(|| SigAction::new(SigHandler::SigDfl, SaFlags::empty(), SigSet::empty()));
    unsafe {
        let _ = signal::sigaction(sig, &prev);
    }
    Errno::set_raw(errno);
}

/// Installs the panic hook, once per process.
///
/// The previous hook is called first. Panics are delivered as
/// `Signal::Panic` only while signal handling is enabled.
pub(crate) fn enable_panic_handling() {
    static ONCE: std::sync::Once = std::sync::Once::new();

    ONCE.call_once(|| {
        let (tx, rx) = mpsc::sync_channel::<()>(1);
        let result = std::thread::Builder::new()
            .name("ntex-rt:panics".to_string())
            .spawn(move || {
                for () in rx {
                    HND_WAKER.wake();
                }
            });
        match result {
            Ok(_) => _ = PANIC_WAKER.set(tx),
            Err(e) => log::error!("Cannot start panic handling thread: {e:?}"),
        }

        let prev = panic::take_hook();
        panic::set_hook(Box::new(move |panic_info| {
            prev(panic_info);

            if !ENABLED.load(Ordering::Acquire) {
                return;
            }

            let info: Arc<str> = if let Some(s) = panic_info.payload().downcast_ref::<&str>() {
                Arc::from(s.to_string())
            } else if let Some(s) = panic_info.payload().downcast_ref::<String>() {
                Arc::from(s.clone())
            } else {
                Arc::from("panic")
            };
            let bt = if let Some(loc) = panic_info.location() {
                let s = Box::new(loc.file().to_string());
                let filename = Box::leak(s);
                Backtrace::with_filename(filename)
            } else {
                Backtrace::new(panic::Location::caller())
            };

            SIGS.lock().push(Signal::Panic(PanicSource::App(info, bt)));
            if let Some(tx) = PANIC_WAKER.get() {
                // a pending wake covers this panic too
                let _ = tx.try_send(());
            }
        }));
    });
}

#[cfg(all(test, any(target_family = "windows", target_family = "unix")))]
mod tests {
    use crate::testing::TestRunner;

    use super::*;

    // signal handling is global
    static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    async fn recv(rx: oneshot::AsyncReceiver<Arc<[Signal]>>) -> Option<Arc<[Signal]>> {
        let mut rx = std::pin::pin!(rx);
        let mut timeout =
            std::pin::pin!(futures_timer::Delay::new(std::time::Duration::from_secs(5)));
        poll_fn(|cx| {
            if let Poll::Ready(res) = rx.as_mut().poll(cx) {
                Poll::Ready(res.ok())
            } else if timeout.as_mut().poll(cx).is_ready() {
                Poll::Ready(None)
            } else {
                Poll::Pending
            }
        })
        .await
    }

    /// Raises a signal with `raise` until a batch with a signal matching
    /// `expected` is delivered, returns the number of attempts.
    ///
    /// The panic hook is process-wide, panics of tests running in parallel are
    /// delivered too. A batch without the expected signal consumes the
    /// handler, and the expected signal may be dropped before the handler is
    /// registered again, so it is raised again.
    async fn deliver(raise: impl Fn(), expected: impl Fn(&Signal) -> bool) -> usize {
        for attempt in 1..=10 {
            let rx = signal();
            // let the registration task run
            futures_timer::Delay::new(std::time::Duration::from_millis(50)).await;
            raise();
            if recv(rx)
                .await
                .is_some_and(|sigs| sigs.iter().any(&expected))
            {
                return attempt;
            }
        }
        panic!("signal is not delivered");
    }

    fn is_app_panic(sig: &Signal, expected: &str) -> bool {
        matches!(sig, Signal::Panic(PanicSource::App(msg, _)) if &**msg == expected)
    }

    #[test]
    fn panic_hook_chains_and_follows_signals() {
        use std::sync::atomic::AtomicUsize;

        static CALLS: AtomicUsize = AtomicUsize::new(0);

        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let prev = panic::take_hook();
        panic::set_hook(Box::new(move |info| {
            // panics of other tests reach the hook too
            let msg = info.payload().downcast_ref::<&str>();
            if matches!(msg, Some(&("no signals" | "boom"))) {
                CALLS.fetch_add(1, Ordering::Relaxed);
            }
            prev(info);
        }));
        enable_panic_handling();
        // installed once
        enable_panic_handling();

        // previous hook is called, nothing is queued without signal handling
        SIGS.lock().clear();
        let _ = panic::catch_unwind(|| panic!("no signals"));
        assert_eq!(CALLS.load(Ordering::Relaxed), 1);
        assert!(
            !SIGS
                .lock()
                .iter()
                .any(|sig| is_app_panic(sig, "no signals"))
        );

        let attempts = System::new("test", TestRunner).block_on(async {
            let sys = System::current();
            sys.enable_signals();

            deliver(
                || {
                    let _ = panic::catch_unwind(|| panic!("boom"));
                },
                |sig| is_app_panic(sig, "boom"),
            )
            .await
        });
        // the hook is installed once and chains the previous one
        assert_eq!(CALLS.load(Ordering::Relaxed), 1 + attempts);
    }

    #[test]
    fn panic_on_other_thread_delivered() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        enable_panic_handling();

        System::new("test", TestRunner).block_on(async {
            System::current().enable_signals();

            // compio aborts the process if the task is woken on a panicking thread
            deliver(
                || {
                    let res = std::thread::spawn(|| panic!("thread boom")).join();
                    assert!(res.is_err());
                },
                |sig| is_app_panic(sig, "thread boom"),
            )
            .await;
        });
    }

    #[test]
    fn signals_released_when_future_panics() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let res = panic::catch_unwind(|| {
            System::new("block_on", TestRunner).block_on(async {
                System::current().enable_signals();
                assert!(is_enabled());
                panic!("block_on fails");
            });
        });
        assert!(res.is_err());
        assert!(!is_enabled());

        let res = panic::catch_unwind(|| {
            System::build().signals(true).build(TestRunner).run(|| {
                assert!(is_enabled());
                panic!("run fails");
            })
        });
        assert!(res.is_err());
        assert!(!is_enabled());
    }

    #[test]
    fn signals_released_when_system_stops() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let first = System::new("first", TestRunner).block_on(async {
            let sys = System::current();
            sys.enable_signals();
            assert!(sys.signals());
            sys
        });
        assert!(!is_enabled());
        assert!(!first.signals());

        System::new("second", TestRunner).block_on(async {
            let sys = System::current();
            sys.enable_signals();
            assert!(sys.signals());
            assert!(is_enabled());

            // signals are handled by one system at a time
            let other = std::thread::spawn(|| {
                System::new("other", TestRunner).block_on(async {
                    let sys = System::current();
                    sys.enable_signals();
                    sys.signals()
                })
            })
            .join()
            .unwrap();
            assert!(!other);
            assert!(is_enabled());
        });
        assert!(!is_enabled());
    }

    #[test]
    fn builder_signals() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        // stopped by another thread
        let stopper = std::thread::spawn(|| {
            loop {
                let sys = CUR_SYS.lock().clone().filter(|sys| sys.name() == "signals");
                if let Some(sys) = sys {
                    sys.stop();
                    break;
                }
                std::thread::sleep(std::time::Duration::from_millis(1));
            }
        });
        System::build()
            .name("signals")
            .enable_signals()
            .build(TestRunner)
            .run_until_stop()
            .unwrap();
        stopper.join().unwrap();
        assert!(!is_enabled());

        // signals are released if `f` fails
        let err = System::build()
            .signals(true)
            .build(TestRunner)
            .run(|| {
                assert!(System::current().signals());
                assert!(is_enabled());
                Err(std::io::Error::other("failed"))
            })
            .unwrap_err();
        assert_eq!(err.to_string(), "failed");
        assert!(!is_enabled());
    }

    #[test]
    fn builder_panic_handling() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        System::build()
            .signals(true)
            .panic_handling(true)
            .build(TestRunner)
            .run(|| {
                crate::spawn_detached(async {
                    deliver(
                        || {
                            // a `String` payload, like a formatted message
                            let res =
                                std::thread::spawn(|| panic::panic_any("run".to_string())).join();
                            assert!(res.is_err());
                        },
                        |sig| is_app_panic(sig, "run"),
                    )
                    .await;
                    System::current().stop();
                });
                Ok(())
            })
            .unwrap();
        assert!(!is_enabled());

        System::build()
            .signals(true)
            .panic_handling(true)
            .build(TestRunner)
            .block_on(async {
                assert!(is_enabled());
                deliver(
                    || {
                        let res = std::thread::spawn(|| panic!("block_on")).join();
                        assert!(res.is_err());
                    },
                    |sig| is_app_panic(sig, "block_on"),
                )
                .await;
            });
        assert!(!is_enabled());
    }

    #[test]
    fn signals_registered_by_one_system() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        let barrier = Arc::new(std::sync::Barrier::new(4));
        let handles: Vec<_> = (0..4)
            .map(|i| {
                let barrier = barrier.clone();
                std::thread::spawn(move || {
                    System::new(&format!("sys{i}"), TestRunner).block_on(async move {
                        let sys = System::current();
                        barrier.wait();
                        sys.enable_signals();
                        let enabled = sys.signals();
                        // keep signals registered until all systems tried
                        barrier.wait();
                        enabled
                    })
                })
            })
            .collect();
        let enabled = handles
            .into_iter()
            .map(|h| h.join().unwrap())
            .filter(|enabled| *enabled)
            .count();
        assert_eq!(enabled, 1);
        assert!(!is_enabled());
    }

    #[test]
    fn signals_queued_from_many_threads() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        SIGS.lock().clear();

        let handles: Vec<_> = (0..4)
            .map(|_| {
                std::thread::spawn(|| {
                    for _ in 0..25 {
                        handle_signal(Signal::Hup);
                    }
                })
            })
            .collect();
        let hups = || {
            let sigs = std::mem::take(&mut *SIGS.lock());
            sigs.iter().filter(|sig| matches!(sig, Signal::Hup)).count()
        };
        let mut received = 0;
        while received < 100 && !handles.iter().all(std::thread::JoinHandle::is_finished) {
            received += hups();
        }
        for h in handles {
            h.join().unwrap();
        }
        received += hups();
        assert_eq!(received, 100);
    }

    #[cfg(target_family = "unix")]
    #[test]
    fn os_signal_delivered() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        System::new("test", TestRunner).block_on(async {
            let sys = System::current();
            sys.enable_signals();
            assert!(sys.signals());

            deliver(
                || unsafe {
                    libc::kill(libc::getpid(), libc::SIGHUP);
                },
                |sig| matches!(sig, Signal::Hup),
            )
            .await;
        });
        assert!(!is_enabled());
    }

    #[cfg(target_family = "windows")]
    #[test]
    fn reenable_signals() {
        let _lock = LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        System::new("test", TestRunner).block_on(async {
            let sys = System::current();
            sys.enable_signals();
            sys.disable_signals();
            sys.enable_signals();
            assert!(sys.signals());
            sys.disable_signals();
            assert!(!sys.signals());
        });
    }

    #[cfg(target_os = "linux")]
    fn signal_child_mode() -> Option<String> {
        let mode = std::env::var("NTEX_FATAL_SIGNAL_CHILD").ok()?;
        for (resource, limit) in [(libc::RLIMIT_CORE, 0), (libc::RLIMIT_CPU, 10)] {
            let limits = libc::rlimit {
                rlim_cur: limit,
                rlim_max: limit,
            };
            assert_eq!(unsafe { libc::setrlimit(resource, &raw const limits) }, 0);
        }
        Some(mode)
    }

    #[cfg(target_os = "linux")]
    fn signal_child(test: &str, mode: &str) -> std::process::ExitStatus {
        use std::process::{Command, Stdio};
        use std::time::{Duration, Instant};

        let mut child = Command::new(std::env::current_exe().unwrap())
            .args(["--exact", &format!("signals::tests::{test}")])
            .env("NTEX_FATAL_SIGNAL_CHILD", mode)
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .unwrap();
        let start = Instant::now();
        loop {
            if let Some(status) = child.try_wait().unwrap() {
                return status;
            }
            if start.elapsed() > Duration::from_secs(10) {
                child.kill().unwrap();
                child.wait().unwrap();
                panic!("{test} ({mode}) did not terminate");
            }
            std::thread::sleep(Duration::from_millis(10));
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn fatal_signals_terminate_process() {
        use std::os::unix::process::ExitStatusExt;

        if let Some(mode) = signal_child_mode() {
            System::new("test", TestRunner).block_on(async {
                System::current().enable_signals();
                let other_thread = mode.starts_with("thread-");
                let crash = move || {
                    if mode.ends_with("abort") {
                        std::process::abort();
                    }
                    unsafe {
                        let page = libc::mmap(
                            std::ptr::null_mut(),
                            4096,
                            libc::PROT_NONE,
                            libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
                            -1,
                            0,
                        );
                        assert_ne!(page, libc::MAP_FAILED);
                        page.cast::<u8>().write_volatile(1);
                    }
                };
                if other_thread {
                    let _stderr = std::io::stderr().lock();
                    let _sigs = SIGS.lock();
                    std::thread::spawn(crash).join().unwrap();
                } else {
                    crash();
                }
            });
            return;
        }

        for mode in ["abort", "segv", "thread-abort", "thread-segv"] {
            let status = signal_child("fatal_signals_terminate_process", mode);
            let expected = if mode.ends_with("abort") {
                libc::SIGABRT
            } else {
                libc::SIGSEGV
            };
            assert_eq!(status.signal(), Some(expected), "{mode}: {status}");
        }
    }

    #[cfg(target_os = "linux")]
    async fn wait_for_signal_handler() {
        poll_fn(|cx| {
            if HANDLERS.with(|handlers| handlers.borrow().is_empty()) {
                cx.waker().wake_by_ref();
                Poll::Pending
            } else {
                Poll::Ready(())
            }
        })
        .await;
    }

    #[cfg(target_os = "linux")]
    fn raise_fatal_signal(sig: i32) {
        std::thread::spawn(move || {
            nix::errno::Errno::EBUSY.set();
            assert_eq!(unsafe { libc::raise(sig) }, 0);
            assert_eq!(nix::errno::Errno::last(), nix::errno::Errno::EBUSY);
        })
        .join()
        .unwrap();
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn fatal_signal_notifications() {
        use std::os::fd::AsRawFd;

        if let Some(mode) = signal_child_mode() {
            System::new("test", TestRunner).block_on(async move {
                System::current().enable_signals();
                if mode == "full" {
                    // Block the receiver and fill the pipe: the signal handler
                    // must still return without waiting or changing errno.
                    let _sigs = SIGS.lock();
                    let (_, _, write) = FATAL_PIPE.get().unwrap();
                    let fd = write.as_raw_fd();
                    assert_eq!(unsafe { libc::write(fd, b"A".as_ptr().cast(), 1) }, 1);
                    let bytes = [0u8; 1];
                    while unsafe { libc::write(fd, bytes.as_ptr().cast(), bytes.len()) } >= 0 {}
                    assert_eq!(nix::errno::Errno::last(), nix::errno::Errno::EAGAIN);
                    raise_fatal_signal(libc::SIGABRT);
                    return;
                }

                if mode == "fork" {
                    let rx = signal();
                    wait_for_signal_handler().await;
                    let pid = unsafe { libc::fork() };
                    assert!(pid >= 0);
                    if pid == 0 {
                        unsafe {
                            libc::alarm(2);
                            libc::raise(libc::SIGABRT);
                            libc::_exit(0);
                        }
                    }
                    let mut status = 0;
                    assert_eq!(unsafe { libc::waitpid(pid, &raw mut status, 0) }, pid);
                    assert_eq!(status, 0);
                    raise_fatal_signal(libc::SIGSEGV);
                    let sigs = recv(rx).await.unwrap();
                    assert!(
                        sigs.iter()
                            .all(|sig| matches!(sig, Signal::Panic(PanicSource::Sig("SIGSEGV"))))
                    );
                    return;
                }

                for (sig, name) in [(libc::SIGABRT, "SIGABRT"), (libc::SIGSEGV, "SIGSEGV")] {
                    let rx = signal();
                    wait_for_signal_handler().await;
                    {
                        let _stderr = std::io::stderr().lock();
                        let _sigs = SIGS.lock();
                        raise_fatal_signal(sig);
                    }
                    let sigs = recv(rx).await.expect("fatal signal was not delivered");
                    assert!(sigs.iter().any(
                        |sig| matches!(sig, Signal::Panic(PanicSource::Sig(s)) if *s == name)
                    ));
                }
            });
            return;
        }

        for mode in ["notify", "full", "fork"] {
            let status = signal_child("fatal_signal_notifications", mode);
            assert!(status.success(), "{mode}: {status}");
        }
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn fatal_signal_stops_server() {
        if let Some(mode) = signal_child_mode() {
            ntex::rt::System::new("test", ntex::rt::DefaultRuntime).block_on(async move {
                let srv = ntex::server::build()
                    .workers(1)
                    .graceful_shutdown()
                    .bind(
                        "test",
                        "127.0.0.1:0",
                        ntex::service::cfg::SharedCfg::default(),
                        async |()| async |_| Ok::<_, ()>(()),
                    )
                    .unwrap()
                    .run();
                while srv.status() != ntex::server::ServerStatus::Ready {
                    srv.status_changed().await;
                }
                futures_timer::Delay::new(std::time::Duration::from_millis(50)).await;
                raise_fatal_signal(if mode == "abort" {
                    libc::SIGABRT
                } else {
                    libc::SIGSEGV
                });
                let stopped = srv.clone();
                srv.await.unwrap();
                assert_eq!(stopped.status(), ntex::server::ServerStatus::Stopped);
            });
            return;
        }

        for mode in ["abort", "segv"] {
            let status = signal_child("fatal_signal_stops_server", mode);
            assert!(status.success(), "{mode}: {status}");
        }
    }
}