feat(supervisor): one_for_all and rest_for_one strategies + ordered shutdown
Adds a Strategy enum (OneForOne default, OneForAll, RestForOne) selected via OneForOne::strategy(). The triggering child's Restart policy still decides whether anything restarts; the strategy decides which siblings are cycled with it: - OneForAll : all live siblings - RestForOne: siblings started after the failed child Group restarts stop the affected survivors cooperatively (request_stop) in reverse start order, await each one's termination signal on the existing funnel (no new channel, no select), then restart the whole set in start order. Signals that arrive for a child we aren't currently stopping are stashed and replayed by the main loop. A Stopped signal now counts as abnormal for the restart decision. On giving up (intensity cap) or any exit with survivors, an ordered shutdown stops the remaining children in reverse start order instead of leaking them; on the normal all-settled exit it's a no-op. The struct keeps the OneForOne name for compatibility (existing tests unchanged); rename is a later refactor.
This commit is contained in:
smarm-dev
+1
-1
@@ -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!()
|
||||
|
||||
+132
-14
@@ -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<ChildSpec>,
|
||||
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<Signal> = VecDeque::new();
|
||||
let next_signal = |pending: &mut VecDeque<Signal>| -> Option<Signal> {
|
||||
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<usize> = 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<Pid> = 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<Pid> = 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user