feat(link): bidirectional links + trap_exit (roadmap #3)
Add Erlang-style process links so an abnormal death fate-shares across a link set, with trap_exit to convert that into a message instead. - Slot.links: Vec<Pid>, bidirectional; reset in all three lifecycle sites. - Actor.trap: Option<Sender<ExitSignal>>, fresh per spawn (a restarted child starts un-trapped; no fourth reset site). - link/unlink free functions on self_pid(); trap_exit() -> Receiver<ExitSignal> (a dedicated inbox, distinct from the monitor Down channel). - finalize_actor clears reverse links under the lock (always; keeps the cascade acyclic), then propagates abnormal deaths post-lock: trapping peer gets an ExitSignal message and survives, non-trapping peer is request_stop'd. Normal exit never propagates. Dead-pid link delivers an immediate NoProc (message if trapping, else request_stop(self) -- never a silent no-op). - ExitSignal reuses DownReason and carries no panic payload (joiner-only). Tests in tests/link.rs cover the propagate/trap/normal/dead-pid/unlink cases.
This commit is contained in:
+45
-2
@@ -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<Sender<Down>>,
|
||||
/// 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<Pid>,
|
||||
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).
|
||||
@@ -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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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);
|
||||
|
||||
Reference in New Issue
Block a user