nuclei/signal/
unix.rs

1//! Unix signals as streams: [`signal`] listens to one [`SignalKind`].
2//!
3//! The first listener of a signal installs the process's handler for it
4//! (`SA_SIGINFO | SA_RESTART | SA_ONSTACK`), which stays for good and
5//! chains to the handler that was there before. The handler raises the
6//! signal's count, then writes to the process's wake descriptor: an
7//! eventfd on Linux, a pipe elsewhere, made once, non-blocking, and never
8//! closed, so a handler never writes to a descriptor that went or that
9//! another file reuses. Both are async-signal-safe; errno is kept.
10//!
11//! Each shard a listener is polled on runs a pump: a task of the runtime's
12//! own that reads the wake descriptor through the shard's driver and then
13//! dispatches. The pump drains the descriptor before it looks at the
14//! counts, and the handler raises the count before it writes, so a
15//! delivery whose write the pump read is in the counts it sees, and a later
16//! one writes again. A listener polled where no shard runs is served by a
17//! pump on a runtime worker.
18
19use core::cell::{Cell, UnsafeCell};
20use core::ffi::{c_int, c_void};
21use core::mem::MaybeUninit;
22use core::pin::Pin;
23use core::task::{Context, Poll};
24use std::io;
25use std::os::fd::{BorrowedFd, RawFd};
26use std::sync::OnceLock;
27use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU8, Ordering};
28
29use super::{Listener, SLOTS};
30use crate::io::Handle;
31
32/// A unix signal a [`Signal`] listens to.
33#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
34pub struct SignalKind(c_int);
35
36impl SignalKind {
37    /// The signal numbered `signum`.
38    pub const fn from_raw(signum: c_int) -> Self {
39        Self(signum)
40    }
41
42    /// The signal's number.
43    pub const fn as_raw(self) -> c_int {
44        self.0
45    }
46
47    /// `SIGALRM`.
48    pub const fn alarm() -> Self {
49        Self(libc::SIGALRM)
50    }
51
52    /// `SIGCHLD`.
53    pub const fn child() -> Self {
54        Self(libc::SIGCHLD)
55    }
56
57    /// `SIGHUP`.
58    pub const fn hangup() -> Self {
59        Self(libc::SIGHUP)
60    }
61
62    /// `SIGINT`: Ctrl-C at a terminal.
63    pub const fn interrupt() -> Self {
64        Self(libc::SIGINT)
65    }
66
67    /// `SIGPIPE`.
68    pub const fn pipe() -> Self {
69        Self(libc::SIGPIPE)
70    }
71
72    /// `SIGQUIT`.
73    pub const fn quit() -> Self {
74        Self(libc::SIGQUIT)
75    }
76
77    /// `SIGTERM`.
78    pub const fn terminate() -> Self {
79        Self(libc::SIGTERM)
80    }
81
82    /// `SIGUSR1`.
83    pub const fn user_defined1() -> Self {
84        Self(libc::SIGUSR1)
85    }
86
87    /// `SIGUSR2`.
88    pub const fn user_defined2() -> Self {
89        Self(libc::SIGUSR2)
90    }
91
92    /// `SIGWINCH`: the terminal's size changed.
93    pub const fn window_change() -> Self {
94        Self(libc::SIGWINCH)
95    }
96}
97
98impl From<c_int> for SignalKind {
99    fn from(signum: c_int) -> Self {
100        Self(signum)
101    }
102}
103
104/// Signals no handler can take, or whose handler cannot return.
105const FORBIDDEN: [c_int; 5] = [
106    libc::SIGILL,
107    libc::SIGFPE,
108    libc::SIGKILL,
109    libc::SIGSEGV,
110    libc::SIGSTOP,
111];
112
113/// The deliveries of one signal, as a stream: each item stands for one
114/// delivery or more since the item before.
115#[derive(Debug)]
116pub struct Signal {
117    listener: Listener,
118}
119
120/// Listens to `kind` from now on.
121///
122/// # Errors
123///
124/// When `kind` is no signal number, is one no handler can take (`SIGKILL`,
125/// `SIGSTOP`) or whose handler cannot return (`SIGILL`, `SIGFPE`,
126/// `SIGSEGV`), or when the handler or the wake descriptor cannot be made.
127pub fn signal(kind: SignalKind) -> io::Result<Signal> {
128    let signum = kind.0;
129    if signum <= 0 || signum as usize >= SLOTS {
130        return Err(io::Error::new(
131            io::ErrorKind::InvalidInput,
132            "not a signal number",
133        ));
134    }
135    if FORBIDDEN.contains(&signum) {
136        return Err(io::Error::new(
137            io::ErrorKind::InvalidInput,
138            "this signal cannot be listened to",
139        ));
140    }
141    install(signum)?;
142    Ok(Signal {
143        listener: Listener::new(signum as usize),
144    })
145}
146
147impl Signal {
148    /// Waits for the next delivery. Always `Some`: the stream never ends.
149    pub async fn recv(&mut self) -> Option<()> {
150        core::future::poll_fn(|cx| self.poll_recv(cx)).await
151    }
152
153    /// Ready with `Some` once the signal was delivered since the last item.
154    pub fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<()>> {
155        self.listener.poll_recv(cx, ensure_pump)
156    }
157}
158
159impl futures_core::Stream for Signal {
160    type Item = ();
161
162    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
163        self.get_mut().poll_recv(cx)
164    }
165}
166
167/// The wake descriptor: the end the handler writes and the end the pumps
168/// read (the same one for an eventfd), and how many bytes a write is.
169struct Wake {
170    write: RawFd,
171    read: RawFd,
172    len: usize,
173}
174
175/// The write end for the handler: -1 until the wake descriptor is made.
176static WAKE_WRITE: AtomicI32 = AtomicI32::new(-1);
177/// What a write is, in bytes, for the handler.
178static WAKE_LEN: core::sync::atomic::AtomicUsize = core::sync::atomic::AtomicUsize::new(1);
179
180/// The process's wake descriptor, made on first need. It is never closed:
181/// the handler may write to it at any time, and the pumps read it.
182fn wake() -> io::Result<&'static Wake> {
183    static WAKE: OnceLock<Result<Wake, i32>> = OnceLock::new();
184    let made = WAKE.get_or_init(|| {
185        let wake = make_wake().map_err(|err| err.raw_os_error().unwrap_or(libc::EIO))?;
186        WAKE_LEN.store(wake.len, Ordering::SeqCst);
187        WAKE_WRITE.store(wake.write, Ordering::SeqCst);
188        Ok(wake)
189    });
190    made.as_ref()
191        .map_err(|&errno| io::Error::from_raw_os_error(errno))
192}
193
194#[cfg(any(target_os = "linux", target_os = "android"))]
195fn make_wake() -> io::Result<Wake> {
196    // SAFETY: a plain syscall; the descriptor is the process's for good.
197    let fd = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK | libc::EFD_CLOEXEC) };
198    if fd < 0 {
199        return Err(io::Error::last_os_error());
200    }
201    Ok(Wake {
202        write: fd,
203        read: fd,
204        len: size_of::<u64>(),
205    })
206}
207
208#[cfg(not(any(target_os = "linux", target_os = "android")))]
209fn make_wake() -> io::Result<Wake> {
210    let mut fds = [-1; 2];
211    // SAFETY: the kernel writes the two descriptors into our array.
212    if unsafe { libc::pipe(fds.as_mut_ptr()) } != 0 {
213        return Err(io::Error::last_os_error());
214    }
215    for fd in fds {
216        // SAFETY: flag changes on the descriptors just made.
217        let set = unsafe {
218            libc::fcntl(fd, libc::F_SETFD, libc::FD_CLOEXEC) == 0
219                && libc::fcntl(fd, libc::F_SETFL, libc::O_NONBLOCK) == 0
220        };
221        if !set {
222            let err = io::Error::last_os_error();
223            // SAFETY: closes the two descriptors just made, which nothing
224            // else has.
225            unsafe {
226                libc::close(fds[0]);
227                libc::close(fds[1]);
228            }
229            return Err(err);
230        }
231    }
232    Ok(Wake {
233        write: fds[1],
234        read: fds[0],
235        len: 1,
236    })
237}
238
239const IDLE: u8 = 0;
240const INSTALLING: u8 = 1;
241const INSTALLED: u8 = 2;
242
243/// The process's handler for one signal.
244struct Handler {
245    state: AtomicU8,
246    /// The action that was in place before; written before the handler
247    /// is installed, and then only read.
248    previous: UnsafeCell<MaybeUninit<libc::sigaction>>,
249    /// `previous` is written.
250    chains: AtomicBool,
251}
252
253// SAFETY: `previous` is written once, before `chains` is raised and before
254// the handler that reads it is installed, and only read after.
255unsafe impl Sync for Handler {}
256
257static HANDLERS: [Handler; SLOTS] = [const {
258    Handler {
259        state: AtomicU8::new(IDLE),
260        previous: UnsafeCell::new(MaybeUninit::uninit()),
261        chains: AtomicBool::new(false),
262    }
263}; SLOTS];
264
265/// Installs the handler of `signum` (a number below `SLOTS`), once.
266fn install(signum: c_int) -> io::Result<()> {
267    let handler = &HANDLERS[signum as usize];
268    loop {
269        match handler
270            .state
271            .compare_exchange(IDLE, INSTALLING, Ordering::AcqRel, Ordering::Acquire)
272        {
273            Ok(_) => break,
274            Err(INSTALLED) => return Ok(()),
275            // Another thread installs it: its outcome is a moment away.
276            Err(_) => std::thread::yield_now(),
277        }
278    }
279    let installed = wake().and_then(|_| {
280        // SAFETY: plain data, all zeroes valid; the kernel fills it.
281        let mut previous: libc::sigaction = unsafe { core::mem::zeroed() };
282        // SAFETY: reads the action in place into our own struct.
283        if unsafe { libc::sigaction(signum, core::ptr::null(), &mut previous) } != 0 {
284            return Err(io::Error::last_os_error());
285        }
286        if !handler.chains.load(Ordering::Acquire) {
287            // SAFETY: the one writer (the installing thread), before the
288            // handler that reads it is in place.
289            unsafe { (*handler.previous.get()).write(previous) };
290            handler.chains.store(true, Ordering::Release);
291        }
292        // SAFETY: as above.
293        let mut action: libc::sigaction = unsafe { core::mem::zeroed() };
294        action.sa_sigaction = on_signal as *const () as libc::sighandler_t;
295        action.sa_flags = libc::SA_SIGINFO | libc::SA_RESTART | libc::SA_ONSTACK;
296        // SAFETY: an empty mask in our own struct; then the action is ours,
297        // its function lives for the process.
298        unsafe {
299            libc::sigemptyset(&mut action.sa_mask);
300            if libc::sigaction(signum, &action, core::ptr::null_mut()) != 0 {
301                return Err(io::Error::last_os_error());
302            }
303        }
304        Ok(())
305    });
306    let state = if installed.is_ok() { INSTALLED } else { IDLE };
307    handler.state.store(state, Ordering::Release);
308    installed
309}
310
311/// The process's handler: counts the delivery, wakes the pumps, and runs
312/// the handler that was there before. Async-signal-safe: atomics, one
313/// `write`, and errno as it found it.
314extern "C" fn on_signal(signum: c_int, info: *mut libc::siginfo_t, context: *mut c_void) {
315    // SAFETY: the thread's own errno.
316    let errno = unsafe { *errno_location() };
317    super::delivered(signum as usize);
318    let fd = WAKE_WRITE.load(Ordering::SeqCst);
319    if fd >= 0 {
320        let one = 1u64;
321        // SAFETY: writes the first bytes of our own u64 to a descriptor
322        // that is never closed; a full pipe (EAGAIN) has a wake waiting
323        // already.
324        unsafe {
325            libc::write(
326                fd,
327                (&one as *const u64).cast(),
328                WAKE_LEN.load(Ordering::SeqCst),
329            )
330        };
331    }
332    if let Some(handler) = HANDLERS.get(signum as usize)
333        && handler.chains.load(Ordering::Acquire)
334    {
335        // SAFETY: written before `chains` was raised, read-only since.
336        let previous = unsafe { (*handler.previous.get()).assume_init_ref() };
337        let action = previous.sa_sigaction;
338        if action != libc::SIG_DFL && action != libc::SIG_IGN {
339            if previous.sa_flags & libc::SA_SIGINFO != 0 {
340                // SAFETY: a handler installed with SA_SIGINFO takes these.
341                let chained: extern "C" fn(c_int, *mut libc::siginfo_t, *mut c_void) =
342                    unsafe { core::mem::transmute(action) };
343                chained(signum, info, context);
344            } else {
345                // SAFETY: a plain handler takes the signal number.
346                let chained: extern "C" fn(c_int) = unsafe { core::mem::transmute(action) };
347                chained(signum);
348            }
349        }
350    }
351    // SAFETY: as above.
352    unsafe { *errno_location() = errno };
353}
354
355#[cfg(target_os = "linux")]
356fn errno_location() -> *mut c_int {
357    // SAFETY: the thread's errno; no argument.
358    unsafe { libc::__errno_location() }
359}
360
361#[cfg(any(target_os = "android", target_os = "netbsd", target_os = "openbsd"))]
362fn errno_location() -> *mut c_int {
363    // SAFETY: as above.
364    unsafe { libc::__errno() }
365}
366
367#[cfg(any(
368    target_os = "macos",
369    target_os = "ios",
370    target_os = "freebsd",
371    target_os = "dragonfly"
372))]
373fn errno_location() -> *mut c_int {
374    // SAFETY: as above.
375    unsafe { libc::__error() }
376}
377
378#[cfg(target_os = "illumos")]
379fn errno_location() -> *mut c_int {
380    // SAFETY: as above.
381    unsafe { libc::___errno() }
382}
383
384thread_local! {
385    /// The shard this thread runs has a pump.
386    static PUMPING: Cell<bool> = const { Cell::new(false) };
387}
388
389/// A pump on a runtime worker serves listeners polled where no shard runs.
390static SERVICE_PUMP: AtomicBool = AtomicBool::new(false);
391
392/// Installs the handler of `signum` with no listener: for the process
393/// module's `SIGCHLD` reaper, which the dispatch calls.
394#[cfg(any(target_os = "linux", target_os = "android"))]
395pub(crate) fn listen_raw(signum: c_int) -> io::Result<()> {
396    install(signum)
397}
398
399/// Makes sure a pump serves listeners polled here: one on the shard this
400/// thread runs, or the one on a runtime worker.
401pub(crate) fn ensure_pump() {
402    let on_shard = crate::rt::context::current(|shard| shard.is_some());
403    let started = if on_shard {
404        PUMPING.try_with(|p| p.replace(true)).unwrap_or(true)
405    } else {
406        SERVICE_PUMP.swap(true, Ordering::SeqCst)
407    };
408    if started {
409        return;
410    }
411    let made = wake().and_then(|wake| {
412        // SAFETY: the wake descriptor is never closed, so the borrow is
413        // good for the process.
414        Handle::new(unsafe { BorrowedFd::borrow_raw(wake.read) })
415    });
416    match made {
417        Ok(handle) => crate::rt::runtime::spawn_service(pump(handle, on_shard)),
418        // No pump: the next poll tries again (a listener only misses its
419        // wakes meanwhile, never a delivery, which its count keeps).
420        Err(_) => pump_ended(on_shard),
421    }
422}
423
424fn pump_ended(on_shard: bool) {
425    if on_shard {
426        let _ = PUMPING.try_with(|p| p.set(false));
427    } else {
428        SERVICE_PUMP.store(false, Ordering::SeqCst);
429    }
430}
431
432/// Reads the wake descriptor, then dispatches, for as long as the shard
433/// runs it.
434async fn pump(wake: Handle<BorrowedFd<'static>>, on_shard: bool) {
435    let mut buf = [0u8; 64];
436    loop {
437        let read = core::future::poll_fn(|cx| wake.source().poll_read(cx, &mut buf, false)).await;
438        super::dispatch();
439        if read.is_err() {
440            break;
441        }
442    }
443    pump_ended(on_shard);
444}
nuclei/signal/unix.rs

nuclei/signal/
unix.rs
