feat(monitor): unidirectional one-shot process monitors

`monitor(pid) -> Receiver<Down>` 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<Sender<Down>>`; 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.
This commit is contained in:
smarm-dev
2026-06-07 21:51:56 +00:00
parent 14dc7a79cf
commit 461de2c451
5 changed files with 209 additions and 4 deletions
+2
View File
@@ -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};
+92
View File
@@ -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<Down> {
let (tx, rx) = channel::<Down>();
// 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
}
+20 -4
View File
@@ -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<Pid>,
pub(crate) outcome: Option<Outcome>,
pub(crate) supervisor_channel: Option<Sender<Signal>>,
/// 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<Sender<Down>>,
pub(crate) outstanding_handles: u32,
pub(crate) pending_io_result: Option<crate::io::IoResult>,
/// 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<RuntimeInner>, 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<dyn std::any::Any + Send>),
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<RuntimeInner>, 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<RuntimeInner>, 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);
+1
View File
@@ -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);