diff --git a/src/actor.rs b/src/actor.rs index 145f93c..c1e61a2 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -134,4 +134,11 @@ pub struct Actor { /// (constructed fresh at spawn), so unlike a `Slot` field it needs no reset /// in the slot-recycling paths. pub stop: std::sync::Arc, + /// Trap-exit inbox (roadmap #3). `None` until the actor calls + /// `link::trap_exit()`; `Some(tx)` means an abnormal death of a linked + /// peer is delivered here as an `ExitSignal` *message* instead of + /// cooperatively stopping this actor. Lives on the `Actor` (fresh per + /// spawn) for the same reason as `stop`: no slot-recycling reset needed, + /// and a restarted child correctly starts out *not* trapping. + pub trap: Option>, } diff --git a/src/lib.rs b/src/lib.rs index c23c313..65067d4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -23,6 +23,7 @@ pub mod timer; pub mod io; pub mod mutex; pub mod monitor; +pub mod link; pub mod runtime; pub mod trace; @@ -38,6 +39,7 @@ static ALLOCATOR: preempt::PreemptingAllocator = preempt::PreemptingAllocator; // --------------------------------------------------------------------------- pub use channel::{channel, Receiver, RecvError, Sender}; +pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{monitor, Down, DownReason}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; pub use pid::Pid; diff --git a/src/link.rs b/src/link.rs new file mode 100644 index 0000000..65f6925 --- /dev/null +++ b/src/link.rs @@ -0,0 +1,164 @@ +//! Process links + `trap_exit`. +//! +//! A *link* is bidirectional and persistent, in contrast to a [`monitor`] +//! (unidirectional, one-shot): once two actors are linked, the *abnormal* +//! death of either propagates to the other. "Abnormal" means a panic or a +//! cooperative [`request_stop`] — a normal return never propagates. +//! +//! What propagation *does* depends on whether the surviving peer traps exits: +//! +//! - **not trapping** (the default): the peer is cooperatively +//! [`request_stop`]'d, so a crash fate-shares across the whole link set — +//! Erlang's "let it crash". Because the peer's own links are walked when +//! *it* finalizes, the stop cascades transitively. +//! - **trapping** (called [`trap_exit`]): the peer instead receives an +//! [`ExitSignal`] *message* on its trap inbox and keeps running, turning +//! peer failures into ordinary inbox events (Erlang's +//! `process_flag(trap_exit, true)`). +//! +//! Linking an already-dead pid is not a silent no-op: it behaves exactly like +//! an immediate abnormal death of the target with reason [`DownReason::NoProc`] +//! — a trapping caller gets a message, a non-trapping caller is stopped. +//! +//! ## Payload +//! +//! [`ExitSignal`] reuses [`DownReason`] and, like a monitor, does **not** +//! carry the panic payload: a panic's payload has a single owner and is +//! delivered to whoever `join()`s the actor. A trap inbox only learns *that* +//! (and *how*) a peer died. +//! +//! ## Inbox +//! +//! The trap inbox is a dedicated channel, distinct from the monitor [`Down`] +//! channel: monitors are documented as one-shot/unidirectional, whereas a trap +//! inbox receives one [`ExitSignal`] per linked peer death over the actor's +//! lifetime. [`trap_exit`] enables trapping and hands back the inbox in one +//! call; calling it again installs a fresh inbox (the previous one closes). +//! +//! ## Races +//! +//! [`link`] registration and `finalize_actor` (in `runtime`) both serialize on +//! the shared mutex, so a target that is live when the link is registered is +//! guaranteed to propagate its eventual death; there is no window in which the +//! death slips between the liveness check and the registration. As elsewhere, +//! we never `send` or `request_stop` while holding the shared lock — those +//! re-enter the runtime — so the act is always performed after the lock is +//! released. +//! +//! [`monitor`]: crate::monitor::monitor +//! [`Down`]: crate::monitor::Down +//! [`request_stop`]: crate::scheduler::request_stop + +use crate::channel::{channel, Receiver, Sender}; +use crate::monitor::DownReason; +use crate::pid::Pid; +use crate::runtime::State; +use crate::scheduler::{request_stop, self_pid, with_runtime}; + +/// A linked peer's death, delivered to a trapping actor's inbox. +/// +/// `Copy` because it carries no payload — see the module docs for why the +/// panic payload is *not* included. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ExitSignal { + /// The linked pid that died (or the dead pid that was linked). + pub from: Pid, + /// How it went down. Only ever `Panic`, `Stopped`, or `NoProc` — a normal + /// `Exit` does not propagate, so it never appears here. + pub reason: DownReason, +} + +/// Trap exits on the current actor and return its exit inbox. +/// +/// After this call, the abnormal death of a linked peer arrives here as an +/// [`ExitSignal`] message instead of cooperatively stopping this actor. +/// Calling it again installs a fresh inbox; the previously returned receiver +/// then closes. +pub fn trap_exit() -> Receiver { + let (tx, rx) = channel::(); + let me = self_pid(); + with_runtime(|inner| { + inner.with_shared(|s| { + if let Some(actor) = s.slot_mut(me).and_then(|slot| slot.actor.as_mut()) { + actor.trap = Some(tx); + } + }) + }); + rx +} + +/// Bidirectionally link the current actor to `target`. +/// +/// If `target` is live, the link is recorded on both actors and either's +/// abnormal death will propagate to the other. If `target` is already gone, +/// this delivers an immediate [`DownReason::NoProc`] exit signal to the caller +/// (a message if trapping, otherwise a cooperative stop). Linking yourself, or +/// re-linking an existing peer, is a no-op. +pub fn link(target: Pid) { + let me = self_pid(); + if target == me { + return; + } + + // Under the lock: if the target is live, record the link both ways and + // return `None`. If it is gone, return the caller's trap sender (if any) + // so we can deliver the NoProc signal after releasing the lock. + let dead_action: Option>> = with_runtime(|inner| { + inner.with_shared(|s| { + let target_live = matches!( + s.slot(target), + Some(slot) if slot.actor.is_some() && !matches!(slot.state, State::Done) + ); + if target_live { + if let Some(slot) = s.slot_mut(me) { + if !slot.links.contains(&target) { + slot.links.push(target); + } + } + if let Some(slot) = s.slot_mut(target) { + if !slot.links.contains(&me) { + slot.links.push(me); + } + } + None + } else { + // Grab our own trap sender so the NoProc delivery (below) + // doesn't need a second lock acquisition. + Some( + s.slot(me) + .and_then(|slot| slot.actor.as_ref()) + .and_then(|a| a.trap.clone()), + ) + } + }) + }); + + match dead_action { + None => {} // linked successfully + Some(Some(tx)) => { + let _ = tx.send(ExitSignal { from: target, reason: DownReason::NoProc }); + } + Some(None) => request_stop(me), + } +} + +/// Remove the link between the current actor and `target`, in both directions. +/// +/// After this, neither actor's death propagates to the other. A no-op if the +/// two were not linked. +pub fn unlink(target: Pid) { + let me = self_pid(); + if target == me { + return; + } + with_runtime(|inner| { + inner.with_shared(|s| { + if let Some(slot) = s.slot_mut(me) { + slot.links.retain(|p| *p != target); + } + if let Some(slot) = s.slot_mut(target) { + slot.links.retain(|p| *p != me); + } + }) + }); +} diff --git a/src/runtime.rs b/src/runtime.rs index 2042ced..b33946d 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -228,6 +228,11 @@ pub(crate) struct Slot { /// actor terminates (drained in `finalize_actor`). Distinct from /// `supervisor_channel`, which is the parent's single funnel. pub(crate) monitors: Vec>, + /// 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 + /// all three slot-lifecycle sites, like `monitors`. + pub(crate) links: Vec, pub(crate) outstanding_handles: u32, pub(crate) pending_io_result: Option, /// Set by `unpark()` when the actor is still running (not yet Parked). @@ -246,6 +251,7 @@ impl Slot { outcome: None, supervisor_channel: None, monitors: Vec::new(), + links: Vec::new(), outstanding_handles: 0, pending_io_result: None, pending_unpark: false, @@ -523,6 +529,7 @@ pub(crate) fn reclaim_slot(s: &mut SharedState, pid: Pid) { slot.waiters.clear(); slot.supervisor_channel = None; slot.monitors.clear(); + slot.links.clear(); slot.state = State::Done; slot.outstanding_handles = 0; slot.pending_unpark = false; @@ -548,7 +555,7 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { Outcome::Stopped => (Outcome::Stopped, Signal::Stopped(pid), DownReason::Stopped), }; - let (waiters, supervisor_pid, monitors, recycled_stack) = inner.with_shared(|s| { + let (waiters, supervisor_pid, monitors, recycled_stack, links) = 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 @@ -557,7 +564,18 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { slot.outcome = Some(joiner_outcome); slot.state = State::Done; let monitors = std::mem::take(&mut slot.monitors); - (std::mem::take(&mut slot.waiters), sup, monitors, stack) + let waiters = std::mem::take(&mut slot.waiters); + let links = std::mem::take(&mut slot.links); + // Drop this (dying) pid from each linked peer's list now, under the + // same lock. Done regardless of how we died, so a peer that later + // finalizes won't try to propagate back to this about-to-be-reclaimed + // slot — which also makes the abnormal-death cascade acyclic. + for &peer in &links { + if let Some(ps) = s.slot_mut(peer) { + ps.links.retain(|p| *p != pid); + } + } + (waiters, sup, monitors, stack, links) }); // Return the stack to the pool outside the shared lock. The pool grows @@ -588,6 +606,31 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { let _ = m.send(Down { pid, reason: down_reason }); } + // Propagate to linked peers. A normal exit never propagates; an abnormal + // death (panic or cooperative stop) reaches each linked peer — a trapping + // peer gets an `ExitSignal` message and survives, a non-trapping peer is + // cooperatively stopped (which cascades along *its* links in turn). The + // reverse links were cleared above, so this never ping-pongs back. Decide + // trap-vs-stop under the lock, then act after releasing it: both + // `Sender::send` and `request_stop` re-enter the shared mutex. + if matches!(down_reason, DownReason::Panic | DownReason::Stopped) { + for peer in links { + let trap = inner.with_shared(|s| match s.slot(peer) { + Some(slot) if !matches!(slot.state, State::Done) => { + Some(slot.actor.as_ref().and_then(|a| a.trap.clone())) + } + _ => None, // peer already gone; nothing to do + }); + match trap { + Some(Some(tx)) => { + let _ = tx.send(crate::link::ExitSignal { from: pid, reason: down_reason }); + } + Some(None) => crate::scheduler::request_stop(peer), + None => {} + } + } + } + // Unpark joiners. for joiner in waiters { crate::scheduler::unpark(joiner); diff --git a/src/scheduler.rs b/src/scheduler.rs index b9d3639..9011fae 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -157,6 +157,7 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa sp, supervisor, stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + trap: None, }); slot.state = crate::runtime::State::Runnable; slot.outstanding_handles = 1; @@ -164,6 +165,7 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHa slot.waiters.clear(); slot.supervisor_channel = None; slot.monitors.clear(); + slot.links.clear(); slot.pending_unpark = false; slot.pending_io_result = None; s.run_queue.push_back(pid); diff --git a/tests/link.rs b/tests/link.rs new file mode 100644 index 0000000..b1730fa --- /dev/null +++ b/tests/link.rs @@ -0,0 +1,223 @@ +//! Link + trap_exit tests (roadmap item #3). +//! +//! A *link* is bidirectional and persistent (contrast a monitor, which is +//! unidirectional and one-shot). When a linked actor dies *abnormally* — a +//! panic or a cooperative `request_stop` — the death propagates to the peer: +//! +//! - a peer that has NOT trapped exits is cooperatively `request_stop`'d, so +//! a crash fate-shares across the link set (Erlang's "let it crash"); +//! - a peer that HAS called `trap_exit()` instead receives an [`ExitSignal`] +//! *message* on its trap inbox and survives. +//! +//! A *normal* exit never propagates — not even to a trapping peer. Linking an +//! already-dead pid delivers an immediate exit signal (`NoProc`): a message to +//! a trapping caller, a `request_stop` to a non-trapping one. +//! +//! These cases are deterministic under the single-thread runtime: an actor +//! runs until it parks/yields, so the relative order of link setup, the +//! triggering death, and the propagated signal is fixed. + +use smarm::{ + channel, link, monitor, run, self_pid, spawn, trap_exit, unlink, yield_now, DownReason, +}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; + +/// Sets its flag when dropped — proves a propagated stop unwound the peer's +/// stack (running Drop) rather than leaking it. +struct DropFlag(Arc); +impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } +} + +/// Abnormal death of one linked actor cooperatively stops its non-trapping +/// peer, and the peer's Drop guards run during the propagated unwind. +#[test] +fn linked_pair_one_panics_other_is_stopped() { + let dropped = Arc::new(AtomicBool::new(false)); + let saw_stopped = Arc::new(AtomicBool::new(false)); + let (d, s) = (dropped.clone(), saw_stopped.clone()); + run(move || { + // B parks on recv (kept open by a retained sender) so it is alive and + // suspended when the propagated stop arrives. + let hb = spawn(move || { + let _g = DropFlag(d); + let (tx, rx) = channel::(); + let _keep = tx; + let _ = rx.recv(); // parks until the link-propagated stop unwinds us + }); + let b = hb.pid(); + let down_b = monitor(b); + + // A links B, then panics. The panic propagates along the link to B. + let ha = spawn(move || { + link(b); + panic!("boom"); + }); + + let dn = down_b.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); + } + drop(ha); + let _ = hb.join(); + }); + assert!( + saw_stopped.load(Ordering::SeqCst), + "peer should be Stopped by the propagated abnormal exit" + ); + assert!( + dropped.load(Ordering::SeqCst), + "peer's Drop guard must run during the propagated cancellation unwind" + ); +} + +/// A trapping peer receives an abnormal death as an `ExitSignal` message and +/// keeps running instead of being torn down. +#[test] +fn trap_exit_turns_abnormal_death_into_a_message() { + let ok = Arc::new(AtomicBool::new(false)); + let o = ok.clone(); + run(move || { + let h = spawn(move || { + let inbox = trap_exit(); + // Spawn the peer from here so we control ordering: it is enqueued + // behind us and runs once we park on `inbox.recv()`. + let ha = spawn(|| panic!("peer down")); + let a = ha.pid(); + link(a); + // Parks here; A then runs, panics, and the trap message wakes us. + let sig = inbox.recv().expect("trap inbox closed without a signal"); + if sig.from == a && matches!(sig.reason, DownReason::Panic) { + o.store(true, Ordering::SeqCst); + } + drop(ha); + }); + let _ = h.join(); + }); + assert!( + ok.load(Ordering::SeqCst), + "trapping peer should receive ExitSignal{{from, Panic}} and survive" + ); +} + +/// A normal exit never propagates — a trapping linked peer sees no message and +/// stays alive. +#[test] +fn normal_exit_does_not_propagate() { + let empty = Arc::new(AtomicBool::new(false)); + let e = empty.clone(); + run(move || { + let h = spawn(move || { + let inbox = trap_exit(); + let ha = spawn(|| { /* exits normally, immediately */ }); + let a = ha.pid(); + link(a); + yield_now(); // let A run to completion and finalize + // A exited normally: nothing should have landed on the inbox. + if let Ok(None) = inbox.try_recv() { + e.store(true, Ordering::SeqCst); + } + drop(ha); + }); + let _ = h.join(); + }); + assert!( + empty.load(Ordering::SeqCst), + "normal exit must not deliver any ExitSignal to a trapping peer" + ); +} + +/// Linking an already-dead pid from a non-trapping actor stops the caller +/// (no silent no-op). +#[test] +fn link_to_dead_pid_stops_a_nontrapping_caller() { + let dropped = Arc::new(AtomicBool::new(false)); + let saw_stopped = Arc::new(AtomicBool::new(false)); + let (d, s) = (dropped.clone(), saw_stopped.clone()); + run(move || { + let ha = spawn(|| {}); + let a = ha.pid(); + ha.join().unwrap(); // A finishes and is reclaimed → `a` is now stale + + let hb = spawn(move || { + let _g = DropFlag(d); + link(a); // dead target, not trapping → request_stop(self) + loop { + smarm::check!(); // observation point to realize the stop + } + }); + let b = hb.pid(); + let down_b = monitor(b); + let dn = down_b.recv().expect("monitor channel closed before Down"); + if matches!(dn.reason, DownReason::Stopped) { + s.store(true, Ordering::SeqCst); + } + let _ = hb.join(); + }); + assert!( + saw_stopped.load(Ordering::SeqCst), + "linking a dead pid (non-trapping) should stop the caller" + ); + assert!(dropped.load(Ordering::SeqCst), "Drop guard must run"); +} + +/// Linking an already-dead pid from a trapping actor delivers an immediate +/// `NoProc` exit message instead of killing it. +#[test] +fn link_to_dead_pid_messages_a_trapping_caller() { + let ok = Arc::new(AtomicBool::new(false)); + let o = ok.clone(); + run(move || { + let ha = spawn(|| {}); + let a = ha.pid(); + ha.join().unwrap(); // `a` is now stale + + let hb = spawn(move || { + let inbox = trap_exit(); + link(a); // dead target, trapping → immediate NoProc message + let sig = inbox.recv().expect("trap inbox closed without a signal"); + if sig.from == a && matches!(sig.reason, DownReason::NoProc) { + o.store(true, Ordering::SeqCst); + } + }); + let _ = hb.join(); + }); + assert!( + ok.load(Ordering::SeqCst), + "linking a dead pid (trapping) should deliver a NoProc ExitSignal" + ); +} + +/// `unlink` removes the relationship in both directions: an abnormal death no +/// longer reaches the formerly-linked peer. +#[test] +fn unlink_prevents_propagation() { + let survived = Arc::new(AtomicBool::new(false)); + let sv = survived.clone(); + run(move || { + let h = spawn(move || { + let b = self_pid(); + let inbox = trap_exit(); + let ha = spawn(move || { + link(b); + unlink(b); + panic!("boom"); // abnormal, but the link is gone + }); + yield_now(); // let A link, unlink, and panic + // Unlinked before death → no ExitSignal should have arrived. + if let Ok(None) = inbox.try_recv() { + sv.store(true, Ordering::SeqCst); + } + drop(ha); + }); + let _ = h.join(); + }); + assert!( + survived.load(Ordering::SeqCst), + "unlinked peer must not receive a propagated exit" + ); +}