feat(supervisor): one-for-one supervisor with restart policies + intensity cap
Builds on the existing supervisor_channel funnel (one mailbox per
supervisor; better than N monitor channels given there is no select).
- supervisor.rs: Restart{Permanent,Transient,Temporary}, ChildSpec (an
Fn factory so children can be re-instantiated), and OneForOne with a
builder (child/intensity) and run() supervision loop. Restart decision
keys off whether the Signal was a panic; a sliding-window intensity cap
stops crash loops and returns instead of spinning.
- Does NOT forcibly terminate children: shared-heap + Drop make async
teardown of a peer unsound, so one_for_all/rest_for_one and true links
wait on a cooperative-cancellation primitive. Cap-trip just stops
restarting; live children are left as-is.
tests/supervisor.rs: transient restart-then-settle, transient/temporary
no-restart, permanent crash-loop hitting the cap, and one-for-one
isolation across two children. Full suite green.
This commit is contained in:
smarm-dev
+1
-1
@@ -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!()
|
||||
|
||||
@@ -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<dyn Fn() + Send + Sync + 'static>,
|
||||
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<ChildSpec>,
|
||||
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::<Signal>();
|
||||
crate::scheduler::register_supervisor_channel(me, tx);
|
||||
|
||||
// pid -> index into `self.children`, for the children currently alive.
|
||||
let mut by_pid: HashMap<Pid, usize> = HashMap::new();
|
||||
let mut active: usize = 0;
|
||||
// Sliding window of recent restart instants, for the intensity cap.
|
||||
let mut restarts: Vec<Instant> = Vec::new();
|
||||
|
||||
let start_child = |idx: usize, by_pid: &mut HashMap<Pid, usize>| {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<AtomicUsize>,
|
||||
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");
|
||||
}
|
||||
Reference in New Issue
Block a user