diff --git a/src/gen_server.rs b/src/gen_server.rs index d8e09de..a68faf7 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -50,6 +50,7 @@ //! see `handle_down` — because monitors are inherently created at runtime.) use crate::channel::{channel, select, Receiver, Selectable, Sender}; +use crate::monitor::{Down, Monitor}; use crate::pid::Pid; use crate::scheduler::{spawn, spawn_under}; @@ -71,8 +72,11 @@ pub trait GenServer: Send + 'static { /// sources enum them up into one `Info`. Use `()` if unused. type Info: Send + 'static; - /// Runs once inside the server actor before any message is handled. - fn init(&mut self) {} + /// Runs once inside the server actor before any message is handled. The + /// [`ServerCtx`] is the loop's one runtime hook: clone its [`Watcher`] + /// into the state here to be able to [`watch`](Watcher::watch) monitors + /// from any later handler. + fn init(&mut self, _ctx: &ServerCtx) {} /// Handle a synchronous call and produce the reply sent back to the caller. fn handle_call(&mut self, request: Self::Call) -> Self::Reply; @@ -84,6 +88,10 @@ pub trait GenServer: Send + 'static { /// drop it. fn handle_info(&mut self, _info: Self::Info) {} + /// Handle a [`Down`] from a monitor handed to the loop via + /// [`Watcher::watch`]. Default: drop it. + fn handle_down(&mut self, _down: Down) {} + /// Runs as the server actor exits, on any exit path (see module docs). fn terminate(&mut self) {} } @@ -190,6 +198,43 @@ impl ServerRef { } } +/// The server loop's runtime hook, passed to [`GenServer::init`]. Currently +/// carries only the [`Watcher`]; opaque so fields can grow without breaking. +pub struct ServerCtx { + watcher: Watcher, +} + +impl ServerCtx { + /// A clonable handle to the loop's monitor intake. Store it in the state + /// during `init` to watch monitors from later handlers. + pub fn watcher(&self) -> Watcher { + self.watcher.clone() + } + + /// Shorthand for `ctx.watcher().watch(m)` when watching during `init`. + pub fn watch(&self, m: Monitor) { + self.watcher.watch(m) + } +} + +/// Hands [`Monitor`]s to a server loop; their [`Down`]s are dispatched to +/// [`GenServer::handle_down`]. Down arms outrank every other arm (a death +/// notice cannot be starved), and a delivered `Down` retires its arm — +/// monitors are one-shot. +#[derive(Clone)] +pub struct Watcher { + tx: Sender, +} + +impl Watcher { + /// Transfer `m` to the server loop. If the server is already gone the + /// monitor is silently dropped (its queued `Down`, if any, with it) — + /// there is no loop left to care. + pub fn watch(&self, m: Monitor) { + let _ = self.tx.send(m); + } +} + /// Configure-then-start construction for servers that need more than a bare /// [`start`]: info channels, a supervisor. Consumed by [`start`](Self::start). /// @@ -278,10 +323,19 @@ fn server_loop( } let mut guard = Terminate(state); - guard.0.init(); + + // The control arm: Watchers feed Monitors to the loop through it. The + // ctx (and with it the loop's own sender) drops right after init — a + // state that didn't clone the Watcher closes the arm, the first select + // observes the closure, and the loop falls back to the plain-inbox park. + let (watch_tx, watch_rx) = channel::(); + guard.0.init(&ServerCtx { watcher: Watcher { tx: watch_tx } }); + + let mut monitors: Vec = Vec::new(); + let mut watch_open = true; loop { - if infos.is_empty() { + if monitors.is_empty() && !watch_open && infos.is_empty() { // Nothing to select over: park on the inbox alone, exactly the // pre-v0.8 loop. match rx.recv() { @@ -290,27 +344,51 @@ fn server_loop( Err(_) => break, } } else { - // Arm priority: infos (declaration order), then the inbox. The - // arm slice is rebuilt per iteration because `infos` shrinks as - // arms close. + // Arm priority: downs, then the control arm, then infos (each in + // declaration order), then the inbox. The arm slice is rebuilt + // per iteration because every set but the inbox shrinks or grows. + let nd = monitors.len(); + let nw = watch_open as usize; let i = { - let mut arms: Vec<&dyn Selectable> = Vec::with_capacity(infos.len() + 1); + let mut arms: Vec<&dyn Selectable> = + Vec::with_capacity(nd + nw + infos.len() + 1); + for m in &monitors { + arms.push(&m.rx); + } + if watch_open { + arms.push(&watch_rx); + } for r in &infos { arms.push(r); } arms.push(&rx); select(&arms) }; - if i < infos.len() { - match infos[i].try_recv() { - Ok(Some(info)) => guard.0.handle_info(info), + if i < nd { + // A Down retires its arm either way: delivered (one-shot) or + // closed without delivering (defensive; shouldn't happen). + let m = monitors.remove(i); + if let Ok(Some(down)) = m.rx.try_recv() { + guard.0.handle_down(down); + } + } else if i < nd + nw { + match watch_rx.try_recv() { + Ok(Some(m)) => monitors.push(m), // Single-receiver: nothing can drain the arm between // select's ready and our try_recv. + Ok(None) => debug_assert!(false, "ready control arm was empty"), + // Every Watcher gone: stop selecting on the arm. + Err(_) => watch_open = false, + } + } else if i < nd + nw + infos.len() { + let j = i - nd - nw; + match infos[j].try_recv() { + Ok(Some(info)) => guard.0.handle_info(info), Ok(None) => debug_assert!(false, "ready info arm was empty"), // Senders all gone: drop the arm, keep serving. `remove` // (not swap_remove) — order is priority. Err(_) => { - infos.remove(i); + infos.remove(j); } } } else { diff --git a/src/lib.rs b/src/lib.rs index b78407c..56874f2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -46,7 +46,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; // --------------------------------------------------------------------------- pub use channel::{channel, select, select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender}; -pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerRef}; +pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerCtx, ServerRef, Watcher}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; diff --git a/tests/gen_server.rs b/tests/gen_server.rs index 35f7bba..cc042ab 100644 --- a/tests/gen_server.rs +++ b/tests/gen_server.rs @@ -71,7 +71,7 @@ impl GenServer for Lifecycle { type Cast = (); type Info = (); - fn init(&mut self) { + fn init(&mut self, _ctx: &smarm::gen_server::ServerCtx) { self.log.lock().unwrap().push("init"); } @@ -339,3 +339,101 @@ fn closed_info_arm_is_dropped_silently() { }); assert_eq!(*got.lock().unwrap(), vec!["cast"]); } + +// --------------------------------------------------------------------------- +// handle_down: monitors handed to the loop via Watcher (v0.8) +// --------------------------------------------------------------------------- + +use smarm::gen_server::Watcher; +use smarm::{monitor, spawn, DownReason, Pid}; + +/// The motivating pattern: a server that spawns workers from a handler, +/// watches them, and logs their deaths. +struct Pool { + watcher: Option, + log: Vec, +} + +enum PoolCast { + SpawnDoomedWorker, + Watch(Pid), +} + +impl GenServer for Pool { + type Call = (); + type Reply = Vec; + type Cast = PoolCast; + type Info = (); + + fn init(&mut self, ctx: &smarm::gen_server::ServerCtx) { + self.watcher = Some(ctx.watcher()); + } + + fn handle_call(&mut self, _: ()) -> Vec { + self.log.clone() + } + + fn handle_cast(&mut self, cast: PoolCast) { + let watcher = self.watcher.as_ref().expect("init ran first"); + match cast { + PoolCast::SpawnDoomedWorker => { + let h = spawn(|| panic!("worker died")); + watcher.watch(monitor(h.pid())); + } + PoolCast::Watch(pid) => watcher.watch(monitor(pid)), + } + } + + fn handle_down(&mut self, down: smarm::Down) { + self.log.push(down.reason); + } +} + +// A worker spawned and watched from inside a handler delivers its Down to +// handle_down. Down arms outrank the inbox, so the death is in the log by +// the time the follow-up call is answered. +#[test] +fn worker_pool_down_reaches_handle_down() { + let got = Arc::new(Mutex::new(Vec::new())); + let got2 = got.clone(); + run(move || { + let server = start(Pool { watcher: None, log: Vec::new() }); + server.cast(PoolCast::SpawnDoomedWorker).unwrap(); + let _ = server.call(()).unwrap(); // sync point: cast handled, worker live + *got2.lock().unwrap() = server.call(()).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), vec![DownReason::Panic]); +} + +// Watching an already-dead pid yields an immediate NoProc Down, and the down +// arm outranks the inbox: the call cast *after* the watch still observes it. +#[test] +fn watch_dead_pid_is_noproc_down() { + let got = Arc::new(Mutex::new(Vec::new())); + let got2 = got.clone(); + run(move || { + let h = spawn(|| {}); + let dead = h.pid(); + h.join().unwrap(); + let server = start(Pool { watcher: None, log: Vec::new() }); + server.cast(PoolCast::Watch(dead)).unwrap(); + *got2.lock().unwrap() = server.call(()).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), vec![DownReason::NoProc]); +} + +// A state that never clones the Watcher closes the control arm; the loop +// falls back to the plain-inbox park and keeps serving. (Every pre-v0.8 test +// in this file also exercises this path.) +#[test] +fn unused_ctx_closes_control_arm_silently() { + let got = Arc::new(Mutex::new(0i64)); + let got2 = got.clone(); + run(move || { + let server = start(Counter { n: 0 }); + server.cast(Op::Add(2)).unwrap(); + server.cast(Op::Add(40)).unwrap(); + *got2.lock().unwrap() = server.call(Req::Get).unwrap(); + }); + assert_eq!(*got.lock().unwrap(), 42); +}