From 461de2c4517970e5a502fc21311fc7c5e51f04f1 Mon Sep 17 00:00:00 2001 From: smarm-dev Date: Fri, 29 May 2026 07:04:38 +0000 Subject: [PATCH] feat(monitor): unidirectional one-shot process monitors `monitor(pid) -> Receiver` lets any actor watch any other and receive a single Down{pid, reason} when it terminates. Generalizes the existing single supervisor_channel (which is just a hard-wired monitor the parent installs at spawn). - src/monitor.rs: Down / DownReason{Exit,Panic,NoProc} + monitor(). Payload-free: a panic payload has one owner and goes to the joiner via JoinError, so monitors learn only that it panicked. Monitoring an already-gone pid yields NoProc immediately. - runtime: Slot grows `monitors: Vec>`; finalize_actor drains and notifies them outside the shared lock (send can unpark a receiver, which re-takes the non-reentrant mutex). Cleared in reclaim_slot and on slot reuse at spawn. - Registration and finalize both serialize on the shared mutex, so a target alive at registration always delivers a real Down (no race between liveness check and registration). tests/monitor.rs: Exit, Panic, NoProc-on-dead-target, and fan-out to multiple monitors. Full suite green. --- src/lib.rs | 2 ++ src/monitor.rs | 92 +++++++++++++++++++++++++++++++++++++++++++++++ src/runtime.rs | 24 ++++++++++--- src/scheduler.rs | 1 + tests/monitor.rs | 94 ++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 209 insertions(+), 4 deletions(-) create mode 100644 src/monitor.rs create mode 100644 tests/monitor.rs diff --git a/src/lib.rs b/src/lib.rs index cdcf05c..e866f04 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod supervisor; pub mod timer; pub mod io; pub mod mutex; +pub mod monitor; pub mod runtime; pub mod trace; @@ -37,6 +38,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; // --------------------------------------------------------------------------- pub use channel::{channel, Receiver, RecvError, Sender}; +pub use monitor::{monitor, Down, DownReason}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; pub use pid::Pid; pub use runtime::{init, Config, Runtime}; diff --git a/src/monitor.rs b/src/monitor.rs new file mode 100644 index 0000000..3821633 --- /dev/null +++ b/src/monitor.rs @@ -0,0 +1,92 @@ +//! Process monitors. +//! +//! `monitor(target)` asks the runtime to deliver a single [`Down`] when +//! `target` terminates, and hands back the [`Receiver`] to read it from. A +//! monitor is: +//! +//! - **unidirectional** — the watcher learns of the target's death, but the +//! target learns nothing of the watcher, and the watcher is unaffected by +//! the death beyond the notification (contrast a *link*, which propagates +//! failure); +//! - **one-shot** — exactly one `Down` is ever sent for a given monitor. +//! The returned channel closes afterwards, so a second `recv()` yields +//! `Err(RecvError)`. +//! +//! This generalizes the older single-`supervisor_channel` mechanism: a +//! supervisor is just a hard-wired monitor that the parent installs at spawn +//! time. Here any actor may monitor any pid, any number of times. +//! +//! ## Reasons +//! +//! [`DownReason`] is deliberately payload-free. A panicking actor's payload +//! has a single owner and is delivered to whoever `join()`s the actor (as +//! `JoinError`); a monitor only learns *that* it panicked, not the value. +//! Monitoring a pid that is already gone (reclaimed, or never alive) yields +//! [`DownReason::NoProc`] immediately, mirroring Erlang's `noproc`. +//! +//! ## Races +//! +//! Registration (below) and `finalize_actor` (in `runtime`) both run under the +//! shared-state mutex, so a target that is still alive when its monitor is +//! registered is guaranteed to deliver a real `Down`; there is no window in +//! which the death slips between the liveness check and the registration. + +use crate::channel::{channel, Receiver}; +use crate::pid::Pid; +use crate::runtime::State; +use crate::scheduler::with_runtime; + +/// Why a monitored actor went down. +/// +/// `Copy` because it carries no payload — see the module docs for why the +/// panic payload is *not* included here. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DownReason { + /// The target returned normally. + Exit, + /// The target panicked. The payload is delivered to the actor's joiner, + /// not to monitors. + Panic, + /// The target was already gone (finished and reclaimed, or never alive) + /// at the moment `monitor()` was called. + NoProc, +} + +/// A monitored actor's termination notice. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Down { + /// The pid that was being monitored. + pub pid: Pid, + /// How it went down. + pub reason: DownReason, +} + +/// Monitor `target`. Returns a channel that will receive exactly one [`Down`]. +/// +/// If `target` is still live, the `Down` arrives when it terminates. If +/// `target` is already gone, a [`DownReason::NoProc`] `Down` is queued +/// immediately so the caller's `recv()` returns without parking. +pub fn monitor(target: Pid) -> Receiver { + let (tx, rx) = channel::(); + + // Register under the shared lock. We clone the sender into the target's + // monitor list and keep the original for the NoProc fallback; we must not + // send while holding the lock, since `Sender::send` can call back into the + // runtime (to unpark a parked receiver) and the shared mutex is not + // reentrant. + let registered = with_runtime(|inner| { + inner.with_shared(|s| match s.slot_mut(target) { + Some(slot) if !matches!(slot.state, State::Done) => { + slot.monitors.push(tx.clone()); + true + } + _ => false, + }) + }); + + if !registered { + let _ = tx.send(Down { pid: target, reason: DownReason::NoProc }); + } + + rx +} diff --git a/src/runtime.rs b/src/runtime.rs index 19e8e0d..20ab2a3 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -37,6 +37,7 @@ use crate::actor::{ use crate::channel::Sender; use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor}; use crate::io::IoThread; +use crate::monitor::{Down, DownReason}; use crate::pid::Pid; use crate::preempt::PREEMPTION_ENABLED; use crate::supervisor::Signal; @@ -223,6 +224,10 @@ pub(crate) struct Slot { pub(crate) waiters: Vec, pub(crate) outcome: Option, pub(crate) supervisor_channel: Option>, + /// Watchers registered via `monitor()`. Each receives one `Down` when this + /// actor terminates (drained in `finalize_actor`). Distinct from + /// `supervisor_channel`, which is the parent's single funnel. + pub(crate) monitors: Vec>, pub(crate) outstanding_handles: u32, pub(crate) pending_io_result: Option, /// Set by `unpark()` when the actor is still running (not yet Parked). @@ -240,6 +245,7 @@ impl Slot { waiters: Vec::new(), outcome: None, supervisor_channel: None, + monitors: Vec::new(), outstanding_handles: 0, pending_io_result: None, pending_unpark: false, @@ -516,6 +522,7 @@ pub(crate) fn reclaim_slot(s: &mut SharedState, pid: Pid) { slot.outcome = None; slot.waiters.clear(); slot.supervisor_channel = None; + slot.monitors.clear(); slot.state = State::Done; slot.outstanding_handles = 0; slot.pending_unpark = false; @@ -528,15 +535,16 @@ pub(crate) fn reclaim_slot(s: &mut SharedState, pid: Pid) { // --------------------------------------------------------------------------- fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { - let (joiner_outcome, sup_signal) = match outcome { - Outcome::Exit => (Outcome::Exit, Signal::Exit(pid)), + let (joiner_outcome, sup_signal, down_reason) = match outcome { + Outcome::Exit => (Outcome::Exit, Signal::Exit(pid), DownReason::Exit), Outcome::Panic(payload) => ( Outcome::Panic(payload), Signal::Panic(pid, Box::new(()) as Box), + DownReason::Panic, ), }; - let (waiters, supervisor_pid, recycled_stack) = inner.with_shared(|s| { + let (waiters, supervisor_pid, monitors, recycled_stack) = inner.with_shared(|s| { let slot = s.slot_mut(pid).expect("finalize_actor: slot vanished"); let sup = slot.actor.as_ref().map(|a| a.supervisor); // Extract the stack before clearing the actor so we can recycle it @@ -544,7 +552,8 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { let stack = slot.actor.take().map(|a| a.stack); slot.outcome = Some(joiner_outcome); slot.state = State::Done; - (std::mem::take(&mut slot.waiters), sup, stack) + let monitors = std::mem::take(&mut slot.monitors); + (std::mem::take(&mut slot.waiters), sup, monitors, stack) }); // Return the stack to the pool outside the shared lock. The pool grows @@ -568,6 +577,13 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { } } + // Notify monitors. Sent outside the shared lock for the same reason as the + // supervisor signal: `send` may unpark a parked receiver, which re-takes + // the shared mutex. + for m in monitors { + let _ = m.send(Down { pid, reason: down_reason }); + } + // Unpark joiners. for joiner in waiters { crate::scheduler::unpark(joiner); diff --git a/src/scheduler.rs b/src/scheduler.rs index eae3dda..1df6a5e 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -153,6 +153,7 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa slot.outcome = None; slot.waiters.clear(); slot.supervisor_channel = None; + slot.monitors.clear(); slot.pending_unpark = false; slot.pending_io_result = None; s.run_queue.push_back(pid); diff --git a/tests/monitor.rs b/tests/monitor.rs new file mode 100644 index 0000000..29db352 --- /dev/null +++ b/tests/monitor.rs @@ -0,0 +1,94 @@ +//! Monitor tests: `monitor(pid)` delivers exactly one `Down` describing how +//! the target terminated. +//! +//! A monitor is unidirectional and one-shot: it never propagates failure to +//! the watcher (that's a link), it just drops a `Down` into the returned +//! channel. These tests pin the three reasons — `Exit`, `Panic`, `NoProc` — +//! and the fan-out case where several monitors watch the same target. +//! +//! Scheduling note: under the default single-thread runtime the parent runs +//! until it parks, so every `monitor()` call below registers while the target +//! is still `Runnable` (it hasn't been resumed yet). Registration and +//! `finalize_actor` both serialize on the shared mutex, so a target that is +//! alive at registration time always produces a real `Down`, never a missed +//! one. + +use smarm::{monitor, run, spawn, DownReason}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::Arc; + +#[test] +fn monitor_sees_normal_exit() { + let ok = Arc::new(AtomicBool::new(false)); + let o = ok.clone(); + run(move || { + let h = spawn(|| {}); + let pid = h.pid(); + let down = monitor(pid); + let d = down.recv().expect("monitor channel closed before Down"); + assert_eq!(d.pid, pid, "Down reported the wrong pid"); + if matches!(d.reason, DownReason::Exit) { + o.store(true, Ordering::SeqCst); + } + let _ = h.join(); + }); + assert!(ok.load(Ordering::SeqCst), "expected DownReason::Exit"); +} + +#[test] +fn monitor_sees_panic() { + let ok = Arc::new(AtomicBool::new(false)); + let o = ok.clone(); + run(move || { + let h = spawn(|| panic!("boom")); + let pid = h.pid(); + let down = monitor(pid); + let d = down.recv().expect("monitor channel closed before Down"); + assert_eq!(d.pid, pid); + if matches!(d.reason, DownReason::Panic) { + o.store(true, Ordering::SeqCst); + } + // Drain the panic outcome so the joiner path is exercised too; the + // payload goes here, not to the monitor. + let _ = h.join(); + }); + assert!(ok.load(Ordering::SeqCst), "expected DownReason::Panic"); +} + +#[test] +fn monitor_already_dead_target_is_noproc() { + let ok = Arc::new(AtomicBool::new(false)); + let o = ok.clone(); + run(move || { + let h = spawn(|| {}); + let pid = h.pid(); + // Consume the only handle on a finished child: the slot is reclaimed + // and its generation bumped, so `pid` is now stale. + h.join().unwrap(); + let down = monitor(pid); + let d = down.recv().expect("NoProc Down should be delivered immediately"); + assert_eq!(d.pid, pid); + if matches!(d.reason, DownReason::NoProc) { + o.store(true, Ordering::SeqCst); + } + }); + assert!(ok.load(Ordering::SeqCst), "expected DownReason::NoProc"); +} + +#[test] +fn multiple_monitors_all_notified() { + let count = Arc::new(AtomicUsize::new(0)); + let c = count.clone(); + run(move || { + let h = spawn(|| {}); + let pid = h.pid(); + let monitors = [monitor(pid), monitor(pid), monitor(pid)]; + let _ = h.join(); + for m in monitors { + if matches!(m.recv().unwrap().reason, DownReason::Exit) { + c.fetch_add(1, Ordering::SeqCst); + } + } + }); + assert_eq!(count.load(Ordering::SeqCst), 3, "every monitor should see the Down"); +}