diff --git a/README.md b/README.md index 77b5c69..773b4c7 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ convenience wrapper around `runtime::init(Config::exact(1)).run(f)`. | `timer` | Min-heap of `(deadline, reason)`; `Sleep` and `WaitTimeout` reasons | | `io` | `block_on_io` for blocking work; `wait_readable`/`wait_writable` + `read`/`write` via epoll | | `supervisor` | `Signal::Exit`/`Panic`/`Stopped` funnelled to a parent; `OneForOne`/`OneForAll`/`RestForOne` strategies + restart-intensity cap | -| `monitor` | `monitor(pid)` → one-shot `Receiver`; unidirectional death notice | +| `monitor` | `monitor(pid)` → `Monitor { id, target, rx }`; one-shot `Down` via `rx`; `demonitor(&m)` tears one registration down; unidirectional death notice | | `link` | bidirectional `link`/`unlink`; abnormal death propagates (cooperative stop, or an `ExitSignal` message under `trap_exit`) | | `gen_server` | `call` (sync request-reply) / `cast` (async) over one inbox; `ServerRef` + `init`/`terminate` hooks; server-down via channel closure | diff --git a/src/lib.rs b/src/lib.rs index 839ca40..3faa229 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,7 +42,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; pub use channel::{channel, Receiver, RecvError, Sender}; pub use gen_server::{CallError, CastError, GenServer, ServerRef}; pub use link::{link, trap_exit, unlink, ExitSignal}; -pub use monitor::{monitor, Down, DownReason}; +pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; 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 index 44d6495..c5aee8a 100644 --- a/src/monitor.rs +++ b/src/monitor.rs @@ -1,8 +1,9 @@ //! 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: +//! `target` terminates, and hands back a [`Monitor`] — the [`Receiver`] to read +//! it from, plus the identity (`id`, `target`) needed to take the registration +//! back down with [`demonitor`]. 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 @@ -24,14 +25,27 @@ //! Monitoring a pid that is already gone (reclaimed, or never alive) yields //! [`DownReason::NoProc`] immediately, mirroring Erlang's `noproc`. //! +//! ## Demonitoring +//! +//! Each `monitor()` registration is tagged with a process-unique [`MonitorId`]. +//! [`demonitor`] removes the registration named by a [`Monitor`] from its +//! target's slot, returning `Some(id)` if a live registration was found or +//! `None` if it had already fired (or the target is gone). Dropping the +//! [`Monitor`] afterwards discards any `Down` that the target had *already* +//! queued — the equivalent of Erlang's `demonitor(Ref, [flush])`. +//! //! ## 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. +//! `demonitor` is protected by the generation half of the pid: if the target +//! has died and its slot index been recycled, `slot_mut(target)` fails the +//! generation check and `demonitor` is a clean no-op — it can never strip a +//! *different* actor's monitor that happens to share the slot index. -use crate::channel::{channel, Receiver}; +use crate::channel::{channel, Receiver, Sender}; use crate::pid::Pid; use crate::runtime::State; use crate::scheduler::with_runtime; @@ -63,26 +77,56 @@ pub struct Down { pub reason: DownReason, } -/// Monitor `target`. Returns a channel that will receive exactly one [`Down`]. +/// A process-unique identifier for one `monitor()` registration. +/// +/// Opaque and `Copy`. Allocated from a monotonic counter in shared state, so +/// it is never reused for the lifetime of the runtime — distinct `monitor()` +/// calls on the same target get distinct ids, which is what lets [`demonitor`] +/// tear down exactly one of several monitors on a target. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct MonitorId(pub(crate) u64); + +/// A live monitor: the receiving end of the one-shot [`Down`] channel, plus the +/// identity needed to [`demonitor`] it. +/// +/// Read the notification from [`Monitor::rx`]. Not `Clone` (the receiver is a +/// single consumer). Dropping it closes the receiving end; if a `Down` was +/// already queued it is discarded with the channel. +pub struct Monitor { + /// This registration's process-unique id. + pub id: MonitorId, + /// The pid being monitored. + pub target: Pid, + /// The one-shot channel the `Down` arrives on. + pub rx: Receiver, +} + +/// Monitor `target`. Returns a [`Monitor`] whose `rx` receives 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 { +/// immediately so the caller's `rx.recv()` returns without parking. +pub fn monitor(target: Pid) -> Monitor { 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, + // Register under the shared lock. We allocate the id and (if the target is + // live) clone the sender into its monitor list, keeping the original `tx` + // for the NoProc fallback. `tx.clone()` only touches the channel's own + // mutex, never the shared runtime mutex, so it is safe under the lock — but + // we must not *send* here, as `Sender::send` can call back in to unpark a + // parked receiver and the shared mutex is not reentrant. + let (id, registered) = with_runtime(|inner| { + inner.with_shared(|s| { + let id = s.alloc_monitor_id(); + let registered = match s.slot_mut(target) { + Some(slot) if !matches!(slot.state, State::Done) => { + slot.monitors.push((id, tx.clone())); + true + } + _ => false, + }; + (id, registered) }) }); @@ -90,5 +134,30 @@ pub fn monitor(target: Pid) -> Receiver { let _ = tx.send(Down { pid: target, reason: DownReason::NoProc }); } - rx + Monitor { id, target, rx } +} + +/// Cancel the monitor `m`. Returns `Some(id)` if a live registration was found +/// on the target's slot and removed, or `None` if there was nothing to remove +/// — the target already fired its `Down` (the registration is drained on +/// finalize), was never alive (`NoProc`), or has been reclaimed. +/// +/// This stops any *future* `Down`. To also discard a `Down` the target may have +/// *already* queued (the finalize-races-demonitor case), drop `m` afterwards; +/// dropping the [`Monitor`] closes its receiver and the queued notice goes with +/// it — the analogue of Erlang's `demonitor(Ref, [flush])`. +pub fn demonitor(m: &Monitor) -> Option { + // Remove the registration under the lock, but move the `Sender` *out* and + // let it drop only after the lock is released: dropping the last sender + // runs `Sender::drop`, which may unpark a parked receiver → `with_shared`, + // and the shared mutex is not reentrant. + let removed: Option<(MonitorId, Sender)> = with_runtime(|inner| { + inner.with_shared(|s| { + let slot = s.slot_mut(m.target)?; + let pos = slot.monitors.iter().position(|(mid, _)| *mid == m.id)?; + Some(slot.monitors.remove(pos)) + }) + }); + // `removed`'s sender drops here, outside the lock. + removed.map(|(id, _sender)| id) } diff --git a/src/runtime.rs b/src/runtime.rs index b33946d..ee484bf 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -37,7 +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::monitor::{Down, DownReason, MonitorId}; use crate::pid::Pid; use crate::preempt::PREEMPTION_ENABLED; use crate::supervisor::Signal; @@ -224,10 +224,11 @@ 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>, + /// Watchers registered via `monitor()`, each tagged with its + /// `MonitorId` so `demonitor` can remove exactly one. 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<(MonitorId, Sender)>, /// Bidirectional links (roadmap #3). Each entry is a peer whose abnormal /// death propagates to this actor (and vice versa — the relationship is /// recorded on both slots). Walked + cleared in `finalize_actor`. Reset in @@ -272,6 +273,10 @@ pub(crate) struct SharedState { /// `pending_closures[idx]` is `Some` iff the actor at slot `idx` has not /// yet been resumed for the first time. Grows on demand; never shrinks. pub(crate) pending_closures: Vec>, + /// Monotonic source of `MonitorId`s, bumped under the shared lock by + /// `alloc_monitor_id`. Never reused, so `demonitor` can name one of several + /// monitors on the same target unambiguously. + pub(crate) next_monitor_id: u64, } impl SharedState { @@ -284,9 +289,17 @@ impl SharedState { timers: Timers::new(), io: None, pending_closures: Vec::new(), // indexed by slot index + next_monitor_id: 0, } } + /// Allocate the next process-unique `MonitorId`. Caller holds the shared + /// lock (this is only reached from `monitor()` under `with_shared`). + pub(crate) fn alloc_monitor_id(&mut self) -> MonitorId { + self.next_monitor_id += 1; + MonitorId(self.next_monitor_id) + } + pub(crate) fn allocate_slot(&mut self) -> (u32, u32) { if let Some(idx) = self.free_list.pop() { let gen = self.slots[idx as usize].generation; @@ -602,7 +615,7 @@ 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 { + for (_, m) in monitors { let _ = m.send(Down { pid, reason: down_reason }); } diff --git a/task.md b/task.md index 77365be..2caa43b 100644 --- a/task.md +++ b/task.md @@ -162,6 +162,37 @@ Shipped. What landed vs the plan: - `demonitor`: needs a per-monitor id to remove a specific sender. Decide the monitor API NOW before more code depends on it — likely return a `Monitor { id, rx }` instead of a bare `Receiver`. + ✅ DONE (this commit). What landed vs the plan: + - `monitor()` now returns `Monitor { id, target, rx }` (added `target` over + the sketched `{id, rx}` so `demonitor` jumps straight to the slot instead of + scanning every slot for the id). `MonitorId(u64)` is opaque, from a + monotonic `next_monitor_id` counter on `SharedState`, bumped under the + shared lock in `monitor()` — no atomics, deterministic, never reused. + - `Slot.monitors: Vec<(MonitorId, Sender)>`. The three slot-reset sites + were untouched — they `.clear()`/`Vec::new()`, which is element-type- + agnostic, so no new reset obligation. `finalize_actor` just destructures + `(_, m)` and sends as before. + - `demonitor(&Monitor) -> Option`: `Some(id)` when a live + registration was found+removed, `None` when it had already fired (drained on + finalize), was `NoProc`, or the slot was reclaimed. Chose `Option` + over a bare bool — names which registration went. Generation half of the pid + makes a stale demonitor a clean no-op: a recycled slot index fails + `slot_mut`'s generation check, so it can never strip a different actor's + monitor. + - ⚠️ Reentrancy: the removed `Sender` is `remove`d out of the Vec under the + lock but **dropped after the lock is released** — `Sender::drop` can unpark a + parked receiver → `with_shared`, and the shared mutex is non-reentrant. Same + discipline as `finalize_actor`. + - "Flush" (discard a `Down` the target already queued) falls out of dropping + the `Monitor`: `demonitor(&m); drop(m)`. That's the cleanup the still-to-come + gen_server **call timeout** wants — monitor the server, wait reply-or-Down- + or-deadline, then demonitor+drop so a timed-out call leaks no registration + and no stale `Down`. + - Perf: touches `Slot` + `finalize_actor`, but `chained_spawn`/`yield_many` + register no monitors, so the Vec stays empty (take-empty is identical cost, + finalize loop runs zero times). before/after `general` probe medians within + noise. Tests (`tests/monitor.rs`): demonitor-stops-delivery, one-of-many + (siblings untouched), after-fire-is-None. - Named registry: `register(name,pid)`/`whereis`/`send_by_name`; a `HashMap` in `SharedState`. - gen_server-style call/cast: request-reply correlation as a thin layer over diff --git a/tests/cancel.rs b/tests/cancel.rs index f166e61..1d97a2f 100644 --- a/tests/cancel.rs +++ b/tests/cancel.rs @@ -42,7 +42,7 @@ fn looping_actor_on_check_is_stopped() { // Flag the stop before the child is ever resumed; it will observe the // flag once its check!() loop reaches the amortised preempt check. request_stop(pid); - let dn = down.recv().expect("monitor channel closed before Down"); + let dn = down.rx.recv().expect("monitor channel closed before Down"); assert_eq!(dn.pid, pid); if matches!(dn.reason, DownReason::Stopped) { s.store(true, Ordering::SeqCst); @@ -72,7 +72,7 @@ fn parked_on_recv_actor_is_stopped() { // Let the child run and park in recv() before we request the stop. yield_now(); request_stop(pid); - let dn = down.recv().expect("monitor channel closed before Down"); + let dn = down.rx.recv().expect("monitor channel closed before Down"); assert_eq!(dn.pid, pid); if matches!(dn.reason, DownReason::Stopped) { s.store(true, Ordering::SeqCst); @@ -103,7 +103,7 @@ fn no_check_no_alloc_loop_is_not_stopped() { let pid = h.pid(); let down = monitor(pid); request_stop(pid); // no observation point => ignored - let dn = down.recv().expect("monitor channel closed before Down"); + let dn = down.rx.recv().expect("monitor channel closed before Down"); if matches!(dn.reason, DownReason::Exit) { s.store(true, Ordering::SeqCst); } diff --git a/tests/link.rs b/tests/link.rs index b1730fa..f82a22e 100644 --- a/tests/link.rs +++ b/tests/link.rs @@ -57,7 +57,7 @@ fn linked_pair_one_panics_other_is_stopped() { panic!("boom"); }); - let dn = down_b.recv().expect("monitor channel closed before Down"); + let dn = down_b.rx.recv().expect("monitor channel closed before Down"); assert_eq!(dn.pid, b, "Down reported the wrong pid"); if matches!(dn.reason, DownReason::Stopped) { s.store(true, Ordering::SeqCst); @@ -152,7 +152,7 @@ fn link_to_dead_pid_stops_a_nontrapping_caller() { }); let b = hb.pid(); let down_b = monitor(b); - let dn = down_b.recv().expect("monitor channel closed before Down"); + let dn = down_b.rx.recv().expect("monitor channel closed before Down"); if matches!(dn.reason, DownReason::Stopped) { s.store(true, Ordering::SeqCst); } diff --git a/tests/monitor.rs b/tests/monitor.rs index 29db352..428b69a 100644 --- a/tests/monitor.rs +++ b/tests/monitor.rs @@ -1,10 +1,11 @@ //! Monitor tests: `monitor(pid)` delivers exactly one `Down` describing how -//! the target terminated. +//! the target terminated, and `demonitor(&m)` tears a registration back down. //! //! 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. +//! the fan-out case where several monitors watch the same target, and that +//! `demonitor` removes exactly the one registration it names. //! //! Scheduling note: under the default single-thread runtime the parent runs //! until it parks, so every `monitor()` call below registers while the target @@ -13,7 +14,7 @@ //! alive at registration time always produces a real `Down`, never a missed //! one. -use smarm::{monitor, run, spawn, DownReason}; +use smarm::{demonitor, monitor, run, spawn, DownReason}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; @@ -25,7 +26,7 @@ fn monitor_sees_normal_exit() { let h = spawn(|| {}); let pid = h.pid(); let down = monitor(pid); - let d = down.recv().expect("monitor channel closed before Down"); + let d = down.rx.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); @@ -43,7 +44,7 @@ fn monitor_sees_panic() { let h = spawn(|| panic!("boom")); let pid = h.pid(); let down = monitor(pid); - let d = down.recv().expect("monitor channel closed before Down"); + let d = down.rx.recv().expect("monitor channel closed before Down"); assert_eq!(d.pid, pid); if matches!(d.reason, DownReason::Panic) { o.store(true, Ordering::SeqCst); @@ -66,7 +67,7 @@ fn monitor_already_dead_target_is_noproc() { // 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"); + let d = down.rx.recv().expect("NoProc Down should be delivered immediately"); assert_eq!(d.pid, pid); if matches!(d.reason, DownReason::NoProc) { o.store(true, Ordering::SeqCst); @@ -85,10 +86,57 @@ fn multiple_monitors_all_notified() { let monitors = [monitor(pid), monitor(pid), monitor(pid)]; let _ = h.join(); for m in monitors { - if matches!(m.recv().unwrap().reason, DownReason::Exit) { + if matches!(m.rx.recv().unwrap().reason, DownReason::Exit) { c.fetch_add(1, Ordering::SeqCst); } } }); assert_eq!(count.load(Ordering::SeqCst), 3, "every monitor should see the Down"); } + +#[test] +fn demonitor_stops_delivery() { + // Demonitoring a live registration removes the slot's only sender for that + // channel; the channel closes, so a later recv() observes Err rather than + // a Down. demonitor reports the id it tore down. + run(|| { + let h = spawn(|| {}); + let pid = h.pid(); + let m = monitor(pid); + assert_eq!(demonitor(&m), Some(m.id), "live registration should be removed"); + assert!(m.rx.recv().is_err(), "no Down should arrive after demonitor"); + let _ = h.join(); + }); +} + +#[test] +fn demonitor_one_of_many() { + // Three monitors on one target; demonitor the middle. The other two still + // fire; the demonitored channel is closed with no Down. Distinct ids mean + // tearing one down never disturbs its siblings. + run(|| { + let h = spawn(|| {}); + let pid = h.pid(); + let ms = [monitor(pid), monitor(pid), monitor(pid)]; + assert_eq!(demonitor(&ms[1]), Some(ms[1].id)); + let _ = h.join(); + assert!(matches!(ms[0].rx.recv().unwrap().reason, DownReason::Exit)); + assert!(matches!(ms[2].rx.recv().unwrap().reason, DownReason::Exit)); + assert!(ms[1].rx.recv().is_err(), "demonitored channel should be closed"); + }); +} + +#[test] +fn demonitor_after_fire_is_none() { + // Once the Down has fired, the registration is already drained from the + // slot, so demonitor finds nothing to remove and reports None. + run(|| { + let h = spawn(|| {}); + let pid = h.pid(); + let m = monitor(pid); + let d = m.rx.recv().expect("Down before close"); + assert!(matches!(d.reason, DownReason::Exit)); + assert_eq!(demonitor(&m), None, "already-fired monitor has nothing to remove"); + let _ = h.join(); + }); +}