diff --git a/src/lib.rs b/src/lib.rs index e866f04..58dff9a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -46,7 +46,7 @@ pub use scheduler::{ block_on_io, run, self_pid, sleep, spawn, spawn_under, wait_readable, wait_writable, yield_now, JoinError, JoinHandle, }; -pub use supervisor::Signal; +pub use supervisor::{ChildSpec, OneForOne, Restart, Signal}; // --------------------------------------------------------------------------- // check!() diff --git a/src/supervisor.rs b/src/supervisor.rs index 2ecc35d..c0d69dc 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -35,3 +35,158 @@ impl Signal { } } } + +// --------------------------------------------------------------------------- +// One-for-one supervisor +// +// A supervisor is itself an actor. It spawns each child under its own pid so +// that every child death funnels into one mailbox (the `supervisor_channel`), +// then loops: receive a Signal, decide per the child's `Restart` policy +// whether to restart, and enforce a restart-intensity cap so a child that +// crashes in a tight loop eventually gives up instead of spinning forever. +// +// What this deliberately does NOT do: forcibly terminate a running child. +// smarm actors share a heap and rely on Drop/RAII, so async teardown of a +// peer's stack from outside is unsound. That rules out `one_for_all` / +// `rest_for_one` (which must stop siblings) and true bidirectional links +// until there is a cooperative-cancellation primitive. When the intensity cap +// trips, this supervisor simply stops restarting and returns; it does not kill +// whatever children are still alive. +// --------------------------------------------------------------------------- + +use crate::channel::channel; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +/// When a terminated child should be restarted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Restart { + /// Always restart, on normal exit or panic. + Permanent, + /// Restart only on panic; a normal exit is treated as "done". + Transient, + /// Never restart. + Temporary, +} + +/// A child managed by a supervisor: a restart policy plus a *factory* that +/// produces a fresh instance of the child's work on each (re)start. +/// +/// The factory is `Fn` (not `FnOnce`) precisely so it can be called again to +/// restart; it is shared via `Arc` so the spec stays cheap to clone. +#[derive(Clone)] +pub struct ChildSpec { + start: Arc, + restart: Restart, +} + +impl ChildSpec { + pub fn new(restart: Restart, start: impl Fn() + Send + Sync + 'static) -> Self { + Self { start: Arc::new(start), restart } + } +} + +/// A one-for-one supervisor: each child is restarted independently of the +/// others. Build with `new()`, add children with `child()`, tune the cap with +/// `intensity()`, then drive it with `run()` from inside an actor (typically +/// `spawn(|| OneForOne::new()....run())`). +pub struct OneForOne { + children: Vec, + intensity: u32, + period: Duration, +} + +impl Default for OneForOne { + fn default() -> Self { + // Erlang's default supervisor intensity: 1 restart per 5 seconds. We + // start a little more permissive but in the same spirit. + Self { children: Vec::new(), intensity: 3, period: Duration::from_secs(5) } + } +} + +impl OneForOne { + pub fn new() -> Self { + Self::default() + } + + /// Allow at most `max` restarts within any `period`-long window before the + /// supervisor gives up and `run()` returns. + pub fn intensity(mut self, max: u32, period: Duration) -> Self { + self.intensity = max; + self.period = period; + self + } + + pub fn child(mut self, spec: ChildSpec) -> Self { + self.children.push(spec); + self + } + + /// Run the supervision loop on the current actor. Returns when every child + /// has reached a terminal, non-restartable state, or when the restart + /// intensity cap is tripped. + pub fn run(self) { + let me = crate::scheduler::self_pid(); + let (tx, rx) = channel::(); + crate::scheduler::register_supervisor_channel(me, tx); + + // pid -> index into `self.children`, for the children currently alive. + let mut by_pid: HashMap = HashMap::new(); + let mut active: usize = 0; + // Sliding window of recent restart instants, for the intensity cap. + let mut restarts: Vec = Vec::new(); + + let start_child = |idx: usize, by_pid: &mut HashMap| { + let start = self.children[idx].start.clone(); + let h = crate::scheduler::spawn_under(me, move || (start)()); + by_pid.insert(h.pid(), idx); + // We supervise via the signal funnel, not by joining; drop the + // handle so the child's slot is reclaimed promptly on death (the + // termination Signal is delivered before reclamation regardless). + drop(h); + }; + + for idx in 0..self.children.len() { + start_child(idx, &mut by_pid); + active += 1; + } + + while active > 0 { + let sig = match rx.recv() { + Ok(s) => s, + Err(_) => break, // mailbox closed: nothing left to supervise + }; + let idx = match by_pid.remove(&sig.pid()) { + Some(i) => i, + None => continue, // stray/duplicate signal + }; + + let abnormal = matches!(sig, Signal::Panic(..)); + let should_restart = match self.children[idx].restart { + Restart::Permanent => true, + Restart::Transient => abnormal, + Restart::Temporary => false, + }; + + if !should_restart { + active -= 1; + continue; + } + + // Intensity cap: prune the window, then give up if we are already + // at the limit. `run()` returns; surviving children are left as-is. + let now = Instant::now(); + restarts.retain(|t| now.duration_since(*t) <= self.period); + if restarts.len() as u32 >= self.intensity { + break; + } + restarts.push(now); + + // Restart in place: one child died, one replaces it, so `active` + // is unchanged. + start_child(idx, &mut by_pid); + } + } +} + diff --git a/tests/supervisor.rs b/tests/supervisor.rs new file mode 100644 index 0000000..8a7e6eb --- /dev/null +++ b/tests/supervisor.rs @@ -0,0 +1,130 @@ +//! One-for-one supervisor tests. +//! +//! A `OneForOne` runs as an ordinary actor: it spawns its children under +//! itself, collects their termination `Signal`s on a single mailbox, and +//! restarts them per policy until either every child is in a terminal, +//! non-restartable state (then `run()` returns) or the restart intensity cap +//! is tripped (then `run()` gives up and returns). +//! +//! Policies: +//! - `Permanent` — restart on any termination (normal or panic). +//! - `Transient` — restart only on panic; a normal exit is "done". +//! - `Temporary` — never restart. +//! +//! All cases below are deterministic under the single-thread runtime: the +//! supervisor parks in `recv()` while a child runs to completion, so signals +//! arrive one at a time in a fixed order. + +use smarm::supervisor::{ChildSpec, OneForOne, Restart}; +use smarm::{run, spawn}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +fn counting_child( + counter: &Arc, + behavior: impl Fn(usize) + Send + Sync + 'static, +) -> impl Fn() + Send + Sync + 'static { + let c = counter.clone(); + move || { + let n = c.fetch_add(1, Ordering::SeqCst) + 1; + behavior(n); + } +} + +#[test] +fn transient_child_is_restarted_on_panic_then_settles() { + let runs = Arc::new(AtomicUsize::new(0)); + let r = runs.clone(); + run(move || { + let child = counting_child(&r, |n| { + if n < 3 { + panic!("crash {n}"); + } + // 3rd run returns normally -> Transient does not restart. + }); + let sup = spawn(move || { + OneForOne::new() + .intensity(10, Duration::from_secs(60)) + .child(ChildSpec::new(Restart::Transient, child)) + .run(); + }); + sup.join().unwrap(); + }); + assert_eq!(runs.load(Ordering::SeqCst), 3, "two restarts then a clean exit"); +} + +#[test] +fn transient_child_normal_exit_is_not_restarted() { + let runs = Arc::new(AtomicUsize::new(0)); + let r = runs.clone(); + run(move || { + let child = counting_child(&r, |_| { /* return normally */ }); + let sup = spawn(move || { + OneForOne::new() + .child(ChildSpec::new(Restart::Transient, child)) + .run(); + }); + sup.join().unwrap(); + }); + assert_eq!(runs.load(Ordering::SeqCst), 1); +} + +#[test] +fn temporary_child_is_not_restarted_on_panic() { + let runs = Arc::new(AtomicUsize::new(0)); + let r = runs.clone(); + run(move || { + let child = counting_child(&r, |_| panic!("once")); + let sup = spawn(move || { + OneForOne::new() + .child(ChildSpec::new(Restart::Temporary, child)) + .run(); + }); + sup.join().unwrap(); + }); + assert_eq!(runs.load(Ordering::SeqCst), 1); +} + +#[test] +fn intensity_cap_stops_a_permanent_crash_loop() { + let runs = Arc::new(AtomicUsize::new(0)); + let r = runs.clone(); + run(move || { + let child = counting_child(&r, |n| panic!("always {n}")); + let sup = spawn(move || { + OneForOne::new() + .intensity(3, Duration::from_secs(60)) + .child(ChildSpec::new(Restart::Permanent, child)) + .run(); + }); + sup.join().unwrap(); + }); + // 1 initial run + 3 restarts, then the 4th failure trips the cap. + assert_eq!(runs.load(Ordering::SeqCst), 4); +} + +#[test] +fn one_for_one_restarts_only_the_failed_child() { + let a = Arc::new(AtomicUsize::new(0)); + let b = Arc::new(AtomicUsize::new(0)); + let (ra, rb) = (a.clone(), b.clone()); + run(move || { + let child_a = counting_child(&ra, |n| { + if n < 2 { + panic!("a crash {n}"); + } + }); + let child_b = counting_child(&rb, |_| { /* runs once, normal */ }); + let sup = spawn(move || { + OneForOne::new() + .intensity(10, Duration::from_secs(60)) + .child(ChildSpec::new(Restart::Transient, child_a)) + .child(ChildSpec::new(Restart::Temporary, child_b)) + .run(); + }); + sup.join().unwrap(); + }); + assert_eq!(a.load(Ordering::SeqCst), 2, "A restarts once then settles"); + assert_eq!(b.load(Ordering::SeqCst), 1, "B is untouched by A's restart"); +}