diff --git a/src/lib.rs b/src/lib.rs index 956b259..c23c313 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -46,7 +46,7 @@ pub use scheduler::{ block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_under, wait_readable, wait_writable, yield_now, JoinError, JoinHandle, }; -pub use supervisor::{ChildSpec, OneForOne, Restart, Signal}; +pub use supervisor::{ChildSpec, OneForOne, Restart, Signal, Strategy}; // --------------------------------------------------------------------------- // check!() diff --git a/src/supervisor.rs b/src/supervisor.rs index e0ad0d1..99f5507 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -61,7 +61,7 @@ impl Signal { // --------------------------------------------------------------------------- use crate::channel::channel; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -93,12 +93,33 @@ impl ChildSpec { } } -/// 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())`). +/// How a supervisor reacts when one child terminates and a restart is due. +/// +/// The *triggering* child's [`Restart`] policy still decides whether a restart +/// happens at all; the strategy only decides *which other children* are cycled +/// along with it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Strategy { + /// Restart only the child that terminated. Siblings are untouched. + OneForOne, + /// Restart every child: the survivors are cooperatively stopped (in + /// reverse start order) and the whole set is restarted in start order. + OneForAll, + /// Restart the terminated child and every child started *after* it; those + /// started before it are untouched. + RestForOne, +} + +/// A supervisor over a fixed set of children. +/// +/// Despite the name (kept for backwards compatibility), the restart strategy +/// is selectable via [`OneForOne::strategy`]; the default is +/// [`Strategy::OneForOne`]. 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, + strategy: Strategy, intensity: u32, period: Duration, } @@ -107,7 +128,12 @@ 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) } + Self { + children: Vec::new(), + strategy: Strategy::OneForOne, + intensity: 3, + period: Duration::from_secs(5), + } } } @@ -116,6 +142,12 @@ impl OneForOne { Self::default() } + /// Select the restart strategy (default [`Strategy::OneForOne`]). + pub fn strategy(mut self, strategy: Strategy) -> Self { + self.strategy = strategy; + self + } + /// 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 { @@ -158,17 +190,31 @@ impl OneForOne { active += 1; } + // A signal that arrives while we are awaiting stop-confirmations (for a + // child we are *not* currently stopping) is stashed here and processed + // by the main loop before it blocks on `recv` again. + let mut pending: VecDeque = VecDeque::new(); + let next_signal = |pending: &mut VecDeque| -> Option { + if let Some(s) = pending.pop_front() { + Some(s) + } else { + rx.recv().ok() + } + }; + while active > 0 { - let sig = match rx.recv() { - Ok(s) => s, - Err(_) => break, // mailbox closed: nothing left to supervise + let sig = match next_signal(&mut pending) { + Some(s) => s, + None => 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(..)); + // The triggering child's own policy decides whether *anything* + // restarts. A cooperative stop counts as abnormal alongside a panic. + let abnormal = matches!(sig, Signal::Panic(..) | Signal::Stopped(..)); let should_restart = match self.children[idx].restart { Restart::Permanent => true, Restart::Transient => abnormal, @@ -181,7 +227,10 @@ impl OneForOne { } // Intensity cap: prune the window, then give up if we are already - // at the limit. `run()` returns; surviving children are left as-is. + // at the limit. One triggering failure counts as one restart event, + // even when the strategy cycles several children. On giving up we + // fall out of the loop and the ordered shutdown below stops any + // survivors. let now = Instant::now(); restarts.retain(|t| now.duration_since(*t) <= self.period); if restarts.len() as u32 >= self.intensity { @@ -189,9 +238,78 @@ impl OneForOne { } restarts.push(now); - // Restart in place: one child died, one replaces it, so `active` - // is unchanged. - start_child(idx, &mut by_pid); + // Which *live* siblings get cycled along with the failed child. + // (The failed child is already gone — removed from `by_pid` above.) + let mut to_stop: Vec<(Pid, usize)> = match self.strategy { + Strategy::OneForOne => Vec::new(), + Strategy::OneForAll => by_pid.iter().map(|(p, i)| (*p, *i)).collect(), + Strategy::RestForOne => by_pid + .iter() + .filter(|(_, i)| **i > idx) + .map(|(p, i)| (*p, *i)) + .collect(), + }; + // Stop survivors in reverse start order (highest child index first). + to_stop.sort_unstable_by(|a, b| b.1.cmp(&a.1)); + + // The set we will restart: the failed child plus every sibling we + // are about to stop, restarted in start (ascending index) order. + let mut restart_set: Vec = Vec::with_capacity(to_stop.len() + 1); + restart_set.push(idx); + + // Request stops, then await each survivor's termination signal + // before restarting. `request_stop` on an already-dead pid is a + // no-op; in that case its (already-sent) Exit signal serves as the + // confirmation. Any signal for a pid we are *not* awaiting is + // stashed for the main loop. + let mut awaiting: Vec = Vec::with_capacity(to_stop.len()); + for (pid, cidx) in &to_stop { + by_pid.remove(pid); + restart_set.push(*cidx); + crate::scheduler::request_stop(*pid); + awaiting.push(*pid); + } + while !awaiting.is_empty() { + let s = match next_signal(&mut pending) { + Some(s) => s, + None => break, // mailbox closed mid-await; stop waiting + }; + if let Some(pos) = awaiting.iter().position(|p| *p == s.pid()) { + awaiting.swap_remove(pos); + } else { + pending.push_back(s); + } + } + + // Restart the whole set in start order. Net effect on `active`: + // one child died (idx), `to_stop.len()` were stopped, and + // `restart_set.len() == 1 + to_stop.len()` are started — so + // `active` is unchanged and needs no adjustment here. + restart_set.sort_unstable(); + for cidx in restart_set { + start_child(cidx, &mut by_pid); + } + } + + // Ordered shutdown: stop any survivors in reverse start order and await + // their termination. On the normal `active == 0` exit `by_pid` is empty + // and this is a no-op; on a cap-trip or mailbox-closed break it tears + // the remaining children down deterministically instead of leaking them. + let mut survivors: Vec<(Pid, usize)> = by_pid.iter().map(|(p, i)| (*p, *i)).collect(); + survivors.sort_unstable_by(|a, b| b.1.cmp(&a.1)); + let mut awaiting: Vec = Vec::with_capacity(survivors.len()); + for (pid, _) in &survivors { + crate::scheduler::request_stop(*pid); + awaiting.push(*pid); + } + while !awaiting.is_empty() { + let s = match next_signal(&mut pending) { + Some(s) => s, + None => break, + }; + if let Some(pos) = awaiting.iter().position(|p| *p == s.pid()) { + awaiting.swap_remove(pos); + } } } } diff --git a/tests/supervisor.rs b/tests/supervisor.rs index 8a7e6eb..4344235 100644 --- a/tests/supervisor.rs +++ b/tests/supervisor.rs @@ -15,10 +15,10 @@ //! 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 smarm::supervisor::{ChildSpec, OneForOne, Restart, Strategy}; +use smarm::{run, sleep, spawn}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; fn counting_child( @@ -128,3 +128,144 @@ fn one_for_one_restarts_only_the_failed_child() { 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"); } + +// --------------------------------------------------------------------------- +// one_for_all / rest_for_one (roadmap #2) +// +// These exercise the group-restart strategies. The discriminator across all +// three strategies is how a *sibling* of the failed child is treated: +// - one_for_one : sibling untouched. +// - rest_for_one: only siblings started *after* the failed child are cycled. +// - one_for_all : every sibling is cycled, regardless of its own policy. +// --------------------------------------------------------------------------- + +#[test] +fn one_for_all_restarts_a_normally_exited_sibling() { + // A panics once then settles; B always exits normally. Under one_for_all, + // A's failure cycles the whole group, so B is *restarted even though it had + // exited normally* and its own Transient policy would never self-restart. + // That second B run is what distinguishes one_for_all from one_for_one + // (where B would run exactly once). + 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, |_| { /* always normal */ }); + let sup = spawn(move || { + OneForOne::new() + .strategy(Strategy::OneForAll) + .intensity(10, Duration::from_secs(60)) + .child(ChildSpec::new(Restart::Transient, child_a)) + .child(ChildSpec::new(Restart::Transient, child_b)) + .run(); + }); + sup.join().unwrap(); + }); + assert_eq!(a.load(Ordering::SeqCst), 2, "A: crash then clean run"); + assert_eq!(b.load(Ordering::SeqCst), 2, "B cycled with the group despite a clean exit"); +} + +#[test] +fn rest_for_one_cycles_only_the_suffix() { + // Three children A(0) B(1) C(2). B panics once. Under rest_for_one only the + // children started after B (i.e. C) are cycled with it; A — started before + // B — is left running untouched. + // + // A must still be *alive* when B's failure is processed, otherwise "not + // restarted" is indistinguishable from "already gone". So A sleeps on its + // first run (parking across B's failure) and returns immediately on any + // later run. Result: + // one_for_one : A=1 B=2 C=1 + // rest_for_one: A=1 B=2 C=2 <- this test + // one_for_all : A=2 B=2 C=2 + let a = Arc::new(AtomicUsize::new(0)); + let b = Arc::new(AtomicUsize::new(0)); + let c = Arc::new(AtomicUsize::new(0)); + let (ra, rb, rc) = (a.clone(), b.clone(), c.clone()); + run(move || { + let child_a = counting_child(&ra, |n| { + if n == 1 { + // Stay alive (parked) across B's failure, then exit on its own. + sleep(Duration::from_millis(20)); + } + }); + let child_b = counting_child(&rb, |n| { + if n < 2 { + panic!("b crash {n}"); + } + }); + let child_c = counting_child(&rc, |_| { /* always normal */ }); + let sup = spawn(move || { + OneForOne::new() + .strategy(Strategy::RestForOne) + .intensity(10, Duration::from_secs(60)) + .child(ChildSpec::new(Restart::Transient, child_a)) + .child(ChildSpec::new(Restart::Transient, child_b)) + .child(ChildSpec::new(Restart::Transient, child_c)) + .run(); + }); + sup.join().unwrap(); + }); + assert_eq!(a.load(Ordering::SeqCst), 1, "A (prefix) is left untouched"); + assert_eq!(b.load(Ordering::SeqCst), 2, "B (the failure) is restarted"); + assert_eq!(c.load(Ordering::SeqCst), 2, "C (suffix) is cycled with B"); +} + +#[test] +fn group_restart_and_shutdown_stop_in_reverse_start_order() { + // A(0) and B(1) park (long sleep) and record their teardown order on Drop; + // C(2) panics on every run. Under one_for_all with intensity(1): + // - C's first panic cycles the group: live siblings B,A are stopped in + // reverse start order -> Drop order [B, A]. + // - after the restart C panics again, tripping the cap; the supervisor + // shuts down, stopping the survivors B,A in reverse order again. + // We assert the first teardown wave is [B, A] (reverse of start order A,B). + let order = Arc::new(Mutex::new(Vec::<&'static str>::new())); + + struct Rec(&'static str, Arc>>); + impl Drop for Rec { + fn drop(&mut self) { + self.1.lock().unwrap().push(self.0); + } + } + + let oa = order.clone(); + run(move || { + let o_a = oa.clone(); + let o_b = oa.clone(); + let sup = spawn(move || { + let o_a2 = o_a.clone(); + let o_b2 = o_b.clone(); + OneForOne::new() + .strategy(Strategy::OneForAll) + .intensity(1, Duration::from_secs(60)) + .child(ChildSpec::new(Restart::Permanent, move || { + let _g = Rec("A", o_a2.clone()); + sleep(Duration::from_secs(30)); // interrupted by the stop + })) + .child(ChildSpec::new(Restart::Permanent, move || { + let _g = Rec("B", o_b2.clone()); + sleep(Duration::from_secs(30)); + })) + .child(ChildSpec::new(Restart::Permanent, || panic!("c always"))) + .run(); + }); + sup.join().unwrap(); + }); + + let recorded = order.lock().unwrap().clone(); + assert!( + recorded.len() >= 2, + "expected at least one teardown wave, got {recorded:?}" + ); + assert_eq!( + &recorded[..2], + &["B", "A"], + "siblings must be torn down in reverse start order (got {recorded:?})" + ); +}