char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somrearrange libs into lib and extensions into extsf4a0f5b

main
branchesmainall refs →
13.2 KiB400 linesraw
1//! Process-wide signal listeners (`addSignalListener`).
2//!
3//! Signals are process-wide but avril runs workers as threads with their own
4//! isolates, so delivery fans out: one process-lifetime feed per signal
5//! pushes each delivery into every live isolate that currently binds it.
6//! Listeners therefore fire in the entry isolate, the REPL, and workers
7//! alike, and need no pledge: observing a delivery grants no authority.
8//! But only the entry isolate (and the REPL's main isolate) suppress the
9//! default death: worker listeners observe every delivery but never count
10//! toward suppression, so with no entry listener a lethal delivery still
11//! kills the process and worker delivery is best-effort.
12//!
13//! Listener feeds build on tokio's signal handling, where every
14//! registration shares one chaining handler instead of replacing the
15//! last. Serve claims SIGINT for shutdown and the REPL claims it for eval
16//! interrupt through the same registry, so those owners keep running
17//! alongside listeners regardless of install order: a listened SIGINT
18//! still shuts serve down and still interrupts REPL evaluation. The
19//! supervisor (supervision.rs) checks the listener count and skips its
20//! own shutdown-and-die path while listeners exist.
21//!
22//! tokio's registry never restores the previous disposition once installed,
23//! so feeds are never dropped: when the last listener goes away the feed
24//! emulates the recorded prior fate (re-raising under the default
25//! disposition, or nothing when it was ignored or owned) instead of
26//! swallowing the signal forever.
27
28use std::collections::{HashMap, VecDeque};
29use std::convert::Infallible;
30use std::io;
31use std::sync::{Arc, Mutex, OnceLock, Weak};
32
33use avril_core::Op;
34use tokio::runtime::Handle;
35use tokio::sync::Notify;
36
37#[cfg(any(target_os = "linux", target_os = "macos"))]
38use super::registry::supervision::Tree;
39use avril_host::io::response;
40#[cfg(any(target_os = "linux", target_os = "macos"))]
41use avril_subprocess::signals::{Prior, original_prior};
42
43struct Globals {
44    /// Entry-isolate bindings per signal. Worker bindings fan out
45    /// deliveries but never suppress: the supervisor consults these
46    /// before carrying out a default death.
47    counts: Mutex<HashMap<&'static str, usize>>,
48    /// Feeds already running; feeds are process-lifetime, never removed.
49    #[cfg(any(target_os = "linux", target_os = "macos"))]
50    feeds: Mutex<HashMap<&'static str, ()>>,
51    /// Every live per-isolate state, for fanning deliveries out.
52    registries: Mutex<Vec<Weak<State>>>,
53}
54
55static GLOBALS: OnceLock<Globals> = OnceLock::new();
56
57fn globals() -> &'static Globals {
58    GLOBALS.get_or_init(|| Globals {
59        counts: Mutex::new(HashMap::new()),
60        #[cfg(any(target_os = "linux", target_os = "macos"))]
61        feeds: Mutex::new(HashMap::new()),
62        registries: Mutex::new(Vec::new()),
63    })
64}
65
66/// Per-isolate listener state, owned by the process extension's registry.
67pub(crate) struct State {
68    /// Whether this isolate is a worker. Worker bindings receive fan-out
69    /// deliveries but never count toward death suppression.
70    worker: bool,
71    #[cfg(any(target_os = "linux", target_os = "macos"))]
72    runtime: Handle,
73    bindings: Mutex<HashMap<&'static str, usize>>,
74    pending: Mutex<VecDeque<&'static str>>,
75    notify: Notify,
76}
77
78impl State {
79    pub(crate) fn create(runtime: Handle, worker: bool) -> Arc<Self> {
80        let _ = &runtime;
81        let state = Arc::new(Self {
82            worker,
83            #[cfg(any(target_os = "linux", target_os = "macos"))]
84            runtime,
85            bindings: Mutex::new(HashMap::new()),
86            pending: Mutex::new(VecDeque::new()),
87            notify: Notify::new(),
88        });
89        globals()
90            .registries
91            .lock()
92            .unwrap()
93            .push(Arc::downgrade(&state));
94        state
95    }
96
97    pub(crate) fn bind(&self, name: &str) -> io::Result<()> {
98        bind_impl(self, name)
99    }
100
101    /// The next delivered signal bound by this isolate, or `None` once it
102    /// binds nothing: the JS pump loop ends, releasing the event loop.
103    pub(crate) async fn poll(&self) -> Option<String> {
104        loop {
105            let notified = self.notify.notified();
106            tokio::pin!(notified);
107            notified.as_mut().enable();
108            if let Some(name) = self.pending.lock().unwrap().pop_front() {
109                return Some(name.to_owned());
110            }
111            if self.bindings.lock().unwrap().values().sum::<usize>() == 0 {
112                return None;
113            }
114            notified.await;
115        }
116    }
117
118    pub(crate) fn unbind(&self, name: &str) {
119        let Some(signal) = lookup(name) else {
120            return;
121        };
122        let mut bindings = self.bindings.lock().unwrap();
123        let Some(count) = bindings.get_mut(signal) else {
124            return;
125        };
126        *count = count.saturating_sub(1);
127        if *count == 0 {
128            bindings.remove(signal);
129        }
130        drop(bindings);
131        if !self.worker {
132            let mut counts = globals().counts.lock().unwrap();
133            if let Some(count) = counts.get_mut(signal) {
134                *count = count.saturating_sub(1);
135                if *count == 0 {
136                    counts.remove(signal);
137                }
138            }
139            drop(counts);
140        }
141        // Wake a polling pump so it observes the removed binding.
142        self.notify.notify_waiters();
143    }
144
145    /// Drops every binding this isolate still holds: entry close and
146    /// isolate teardown restore the supervisor's default fate. Worker
147    /// bindings never counted, so they have nothing to release.
148    pub(crate) fn shutdown(&self) {
149        let bindings = std::mem::take(&mut *self.bindings.lock().unwrap());
150        if !self.worker && !bindings.is_empty() {
151            let mut counts = globals().counts.lock().unwrap();
152            for (signal, held) in &bindings {
153                if let Some(count) = counts.get_mut(signal) {
154                    *count = count.saturating_sub(*held);
155                    if *count == 0 {
156                        counts.remove(signal);
157                    }
158                }
159            }
160        }
161        globals()
162            .registries
163            .lock()
164            .unwrap()
165            .retain(|state| state.strong_count() != 0);
166        self.pending.lock().unwrap().clear();
167        self.notify.notify_waiters();
168    }
169
170    #[cfg(any(target_os = "linux", target_os = "macos"))]
171    fn bound(&self, signal: &'static str) -> bool {
172        self.bindings.lock().unwrap().contains_key(signal)
173    }
174
175    #[cfg(any(target_os = "linux", target_os = "macos"))]
176    fn push(&self, signal: &'static str) {
177        self.pending.lock().unwrap().push_back(signal);
178        self.notify.notify_waiters();
179    }
180}
181
182/// How many entry isolates currently bind this signal. Worker bindings
183/// are excluded: only the entry (and REPL main) isolate suppress death.
184pub(crate) fn listener_count(name: &str) -> usize {
185    globals()
186        .counts
187        .lock()
188        .unwrap()
189        .get(name)
190        .copied()
191        .unwrap_or(0)
192}
193
194pub(crate) fn ops(state: &Arc<State>) -> Vec<Op> {
195    let mut ops = Vec::new();
196    ops.push(Op::new("op_signal_bind", {
197        let state = state.clone();
198        move |name: String| Ok::<_, Infallible>(response(state.bind(&name)))
199    }));
200    ops.push(Op::asynchronous("op_signal_poll", {
201        let state = state.clone();
202        move |(): ()| {
203            let state = state.clone();
204            async move {
205                // A response tuple like every other op: result() unwraps it.
206                Ok::<_, Infallible>(response(Ok::<_, io::Error>(state.poll().await)))
207            }
208        }
209    }));
210    ops.push(Op::new("op_signal_unbind", {
211        let state = state.clone();
212        move |name: String| {
213            state.unbind(&name);
214            Ok::<_, Infallible>(())
215        }
216    }));
217    ops
218}
219
220#[cfg(any(target_os = "linux", target_os = "macos"))]
221struct Signal {
222    name: &'static str,
223    number: libc::c_int,
224    kind: fn() -> tokio::signal::unix::SignalKind,
225    /// Whether the default disposition terminates the process. SIGWINCH's is
226    /// to ignore the delivery.
227    kills: bool,
228}
229
230#[cfg(any(target_os = "linux", target_os = "macos"))]
231const LISTENABLE: [Signal; 6] = [
232    Signal {
233        name: "SIGINT",
234        number: libc::SIGINT,
235        kind: tokio::signal::unix::SignalKind::interrupt,
236        kills: true,
237    },
238    Signal {
239        name: "SIGTERM",
240        number: libc::SIGTERM,
241        kind: tokio::signal::unix::SignalKind::terminate,
242        kills: true,
243    },
244    Signal {
245        name: "SIGHUP",
246        number: libc::SIGHUP,
247        kind: tokio::signal::unix::SignalKind::hangup,
248        kills: true,
249    },
250    Signal {
251        name: "SIGUSR1",
252        number: libc::SIGUSR1,
253        kind: tokio::signal::unix::SignalKind::user_defined1,
254        kills: true,
255    },
256    Signal {
257        name: "SIGUSR2",
258        number: libc::SIGUSR2,
259        kind: tokio::signal::unix::SignalKind::user_defined2,
260        kills: true,
261    },
262    Signal {
263        name: "SIGWINCH",
264        number: libc::SIGWINCH,
265        kind: tokio::signal::unix::SignalKind::window_change,
266        kills: false,
267    },
268];
269
270#[cfg(any(target_os = "linux", target_os = "macos"))]
271fn lookup(name: &str) -> Option<&'static str> {
272    LISTENABLE
273        .iter()
274        .find(|signal| signal.name == name)
275        .map(|signal| signal.name)
276}
277
278#[cfg(not(any(target_os = "linux", target_os = "macos")))]
279fn lookup(_name: &str) -> Option<&'static str> {
280    None
281}
282
283#[cfg(any(target_os = "linux", target_os = "macos"))]
284fn bind_impl(state: &State, name: &str) -> io::Result<()> {
285    use super::sys::known_signal;
286
287    let Some(signal) = LISTENABLE.iter().find(|signal| signal.name == name) else {
288        if matches!(name, "SIGKILL" | "SIGSTOP") {
289            return Err(io::Error::new(
290                io::ErrorKind::InvalidInput,
291                format!("{name} cannot be intercepted"),
292            ));
293        }
294        if known_signal(name) {
295            return Err(io::Error::new(
296                io::ErrorKind::Unsupported,
297                format!("{name} signal listeners are not supported"),
298            ));
299        }
300        return Err(io::Error::new(
301            io::ErrorKind::InvalidInput,
302            format!("Unknown signal {name}"),
303        ));
304    };
305    // Install the process-wide feed before marking anything: a failed
306    // install must not pin the signal. The single insert doubles as the
307    // already-running check, so concurrent binds cannot start two feeds.
308    let fresh = globals()
309        .feeds
310        .lock()
311        .unwrap()
312        .insert(signal.name, ())
313        .is_none();
314    if fresh {
315        // An inherited SIG_IGN stays ignored for supervision, but an
316        // explicit listener takes over the signal: that is the point of
317        // listening.
318        let _enter = state.runtime.enter();
319        match tokio::signal::unix::signal((signal.kind)()) {
320            Ok(stream) => {
321                state
322                    .runtime
323                    .spawn(feed(signal.name, signal.number, stream));
324            }
325            Err(error) => {
326                globals().feeds.lock().unwrap().remove(signal.name);
327                return Err(error);
328            }
329        }
330    }
331    *state
332        .bindings
333        .lock()
334        .unwrap()
335        .entry(signal.name)
336        .or_insert(0) += 1;
337    if !state.worker {
338        *globals()
339            .counts
340            .lock()
341            .unwrap()
342            .entry(signal.name)
343            .or_insert(0) += 1;
344    }
345    Ok(())
346}
347
348#[cfg(not(any(target_os = "linux", target_os = "macos")))]
349fn bind_impl(_state: &State, _name: &str) -> io::Result<()> {
350    Err(io::Error::new(
351        io::ErrorKind::Unsupported,
352        "Signal listeners are not supported on this platform",
353    ))
354}
355
356#[cfg(any(target_os = "linux", target_os = "macos"))]
357async fn feed(name: &'static str, number: libc::c_int, mut stream: tokio::signal::unix::Signal) {
358    let kills = LISTENABLE
359        .iter()
360        .find(|signal| signal.name == name)
361        .is_some_and(|signal| signal.kills);
362    loop {
363        if stream.recv().await.is_none() {
364            return;
365        }
366        // Fan out first: worker listeners observe every delivery even
367        // when nobody suppresses, and delivery racing a real death is
368        // best-effort.
369        let targets = {
370            let mut registries = globals().registries.lock().unwrap();
371            registries.retain(|state| state.strong_count() != 0);
372            registries
373                .iter()
374                .filter_map(Weak::upgrade)
375                .filter(|state| state.bound(name))
376                .collect::<Vec<_>>()
377        };
378        for target in targets {
379            target.push(name);
380        }
381        if listener_count(name) == 0
382            && kills
383            && original_prior(number) == Prior::Default
384            && !Tree::global().holds(number)
385        {
386            // Nobody suppresses: emulate the startup fate, since the
387            // installed handler would otherwise swallow the delivery. The
388            // supervisor owns its signals' deaths itself.
389            die(number);
390        }
391    }
392}
393
394#[cfg(any(target_os = "linux", target_os = "macos"))]
395fn die(number: libc::c_int) {
396    unsafe {
397        libc::signal(number, libc::SIG_DFL);
398        libc::raise(number);
399    }
400}