Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pg: Phase 2 — liveness & the monitor death hook (RFC 012)

Browse Source 
Wire eviction to actor death via the monitor subsystem, using drain-on-
contact (no scheduler hot-path hook).

- Each membership now carries its own Monitor alongside the group entry;
  join() installs monitor(pid) (registration races finalize_actor under the
  cold lock, as every monitor does) and a redundant join tears its extra
  monitor back down.
- reap_group: every group op drains the touched group's monitors with a
  non-blocking try_recv (a delivered Down or closed channel = dead) and, on
  the first death, sweeps that pid out of *every* group via remove_where —
  so a death vanishes from all its groups on next contact.
- Lock discipline: monitor()/demonitor() (cold = Leaf) run outside the group
  lock; try_recv (Channel) runs under it, permitted by the Leaf->Channel
  order; evicted/rejected monitors are dropped after the lock releases so no
  receiver-drop wakeup runs under it. Validated by the debug-build lock-order
  checker passing under the integration tests.
- remove_where now returns the evicted monitors (for deferred drop) and
  preserves intra-group insertion order.
- Tests: unit (synthetic monitors) for reap live/dead/closed + sweep;
  integration (real actors) for death-vanishes-from-every-group, pick on an
  all-dead group, peer survival, leave isolation, dead-at-join eviction.

Outstanding: a miri pass (cargo +nightly miri test --lib pg::) is untried —
first build exceeds the sandbox window and the futex RawMutex may need a miri
shim; the mechanical Leaf->Channel checker is the primary soundness guard.
This commit is contained in:
smarm-agent
2026-06-15 17:24:48 +00:00
parent b78311bc88
commit 6464c3e92b
2 changed files with 419 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pg.rs
+288 -105
View File
@@ -4,26 +4,32 @@
//! registry is a *bimap*: at most one pid per name. A group is the opposite — //! registry is a *bimap*: at most one pid per name. A group is the opposite —
//! many pids per name, the same pid in many groups — so it cannot be a //! many pids per name, the same pid in many groups — so it cannot be a
//! generalization of the bimap; it is its own keyspace sharing only the //! generalization of the bimap; it is its own keyspace sharing only the
//! liveness helper and the lock *class*. //! liveness signal source (monitors) and the lock *class*.
//! //!
//! ## Cleanup is eager (unlike the registry) //! ## Cleanup is eager and monitor-driven (unlike the registry)
//! //!
//! The registry prunes a stale binding lazily, on contact, because it only //! The registry prunes a stale binding lazily, on contact, because it only
//! ever resolves one binding at a time. A group is *iterated* — `members` //! ever resolves one binding at a time. A group is *iterated* — `members` fans
//! fans out to every member — so it must not carry dead members across a //! out to every member — so it must not carry dead members across a broadcast.
//! broadcast. Removal is therefore eager and monitor-driven: each `join` //! Each [`join`] installs a `monitor(pid)` and keeps the resulting [`Monitor`]
//! installs a `monitor(pid)` and the one-shot `Down` evicts the member //! *alongside the group entry*; the actor's one-shot `Down` lands in that
//! (wired in Phase 2). A generation-checked liveness read stays on the read //! monitor's channel when it dies. Every group operation [`reap`s][reap] the
//! path as a belt-and-braces backstop (Phase 3), matching the registry's //! group it touches first — draining each membership's monitor with a
//! prune-on-contact guard. //! non-blocking `try_recv` — and on the first sign of death sweeps that pid out
//! of *every* group via the predicate primitive. So the set is self-pruning on
//! contact rather than filtered on read; Phase 3 adds a generation-checked
//! liveness read as a belt-and-braces backstop on the read path.
//!
//! [reap]: ProcessGroups::reap_group
//! //!
//! ## Eviction is one dumb primitive //! ## Eviction is one dumb primitive
//! //!
//! [`ProcessGroups::remove_where`] removes every member matching a predicate //! [`ProcessGroups::remove_where`] removes every member matching a predicate
//! from every group. The primitive never knows *why* a member leaves — that //! from every group. The primitive never knows *why* a member leaves — that is
//! is the caller's concern. Its first caller is the monitor death hook (this //! the caller's concern. Its first caller is the monitor death hook (this RFC,
//! RFC); the later `evict_incarnation(node, inc)` sweep (RFC 010) reuses the //! via `reap_group`); the later `evict_incarnation(node, inc)` sweep (RFC 010)
//! same predicate path, which is the whole reason to shape it as a predicate. //! reuses the same predicate path, which is the whole reason to shape it as a
//! predicate.
//! //!
//! ## Identity is cluster-shaped from the first commit //! ## Identity is cluster-shaped from the first commit
//! //!
@@ -38,11 +44,23 @@
//! //!
//! ## Locking //! ## Locking
//! //!
//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry` //! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry`. The
//! exactly: never held with any other lock, and the eviction path keeps it //! group lock is never held with another Leaf (it never touches the registry
//! off the send path (it does not send under the lock). Liveness checks under //! or a slot's cold lock). The two places that *do* need another lock are kept
//! it read only the atomic slot word, so the leaf rule holds trivially. //! off the group-lock path:
//!
//! - `monitor()` / `demonitor()` acquire the target's cold lock (Leaf) and
//! so run *before* / *after* the group lock, never under it.
//! - draining a monitor with `try_recv` takes the channel's Channel-class
//! lock — permitted *under* a Leaf by the lock order (`raw_mutex.rs`), and
//! a channel critical section only does the lock-free unpark protocol, so
//! no Leaf is ever nested under it.
//!
//! Evicted and rejected [`Monitor`]s are dropped only *after* the group lock is
//! released, so a receiver-drop never runs a wakeup under the lock (same
//! discipline as `demonitor`).
use crate::monitor::{demonitor, monitor, Monitor};
use crate::pid::Pid; use crate::pid::Pid;
use crate::scheduler::with_runtime; use crate::scheduler::with_runtime;
use std::collections::HashMap; use std::collections::HashMap;
@@ -118,12 +136,21 @@ pub struct Member {
pub pid: Pid, pub pid: Pid,
} }
/// One membership: a [`Member`] and the [`Monitor`] that watches its liveness.
/// The monitor lives *alongside* the group entry (RFC 012) so a group is
/// self-contained: draining the membership tells us whether the member is
/// still alive, and dropping the membership drops its monitor.
struct Membership {
member: Member,
monitor: Monitor,
}
/// The store: `name → multiset<Member>`. Within a single group a `Member` /// The store: `name → multiset<Member>`. Within a single group a `Member`
/// appears at most once (`join` is idempotent); the *multiset* framing is for /// appears at most once (`join` is idempotent); the *multiset* framing is for
/// cluster-readiness — the same pid is freely a member of many groups, and the /// cluster-readiness — the same pid is freely a member of many groups, and the
/// width admits multiples in general. Held under one Leaf-class `RawMutex`. /// width admits multiples in general. Held under one Leaf-class `RawMutex`.
pub(crate) struct ProcessGroups { pub(crate) struct ProcessGroups {
groups: HashMap<String, Vec<Member>>, groups: HashMap<String, Vec<Membership>>,
} }
impl ProcessGroups { impl ProcessGroups {
@@ -131,50 +158,93 @@ impl ProcessGroups {
Self { groups: HashMap::new() } Self { groups: HashMap::new() }
} }
/// Add `member` to `group`. Idempotent within a group: returns `true` if /// Insert `ms` into `group`. Idempotent on the *member*: if the member is
/// it was newly inserted, `false` if it was already present. /// already present the new membership is handed back (`Some`) so the caller
pub(crate) fn join(&mut self, group: &str, member: Member) -> bool { /// can tear its now-redundant monitor down outside the lock; `None` means
/// it was inserted.
fn join(&mut self, group: &str, ms: Membership) -> Option<Membership> {
let v = self.groups.entry(group.to_owned()).or_default(); let v = self.groups.entry(group.to_owned()).or_default();
if v.contains(&member) { if v.iter().any(|e| e.member == ms.member) {
return false; return Some(ms);
} }
v.push(member); v.push(ms);
true None
} }
/// Remove `member` from `group`. Returns whether anything was removed. /// Remove `member`'s membership from `group`, returning it (so the caller
/// An emptied group is pruned so the keyspace does not leak empty vecs. /// can `demonitor` it outside the lock). An emptied group is pruned.
pub(crate) fn leave(&mut self, group: &str, member: Member) -> bool { fn leave(&mut self, group: &str, member: Member) -> Option<Membership> {
let Some(v) = self.groups.get_mut(group) else { let v = self.groups.get_mut(group)?;
return false; let pos = v.iter().position(|e| e.member == member)?;
}; let removed = v.remove(pos);
let before = v.len();
v.retain(|m| *m != member);
let removed = v.len() != before;
if v.is_empty() { if v.is_empty() {
self.groups.remove(group); self.groups.remove(group);
} }
removed Some(removed)
} }
/// The one dumb eviction primitive: drop every member matching `pred` from /// The one dumb eviction primitive: drop every member matching `pred` from
/// every group, pruning emptied groups. The primitive does not know *why* /// every group, pruning emptied groups, and return the evicted memberships'
/// a member leaves. First caller: the monitor death hook (Phase 2). Later: /// monitors for the caller to drop outside the lock. The primitive does not
/// `evict_incarnation` (RFC 010) over the same path. /// know *why* a member leaves. Callers: the death hook (`reap_group`) and,
// /// later, `evict_incarnation` (RFC 010), over the same path. Insertion
// No non-test caller until Phase 2 wires the death hook to it. /// order within a group is preserved (`members` / `pick` are order-stable).
#[allow(dead_code)] fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) -> Vec<Monitor> {
pub(crate) fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) { let mut evicted = Vec::new();
self.groups.retain(|_, v| { self.groups.retain(|_, v| {
v.retain(|m| !pred(m)); let mut i = 0;
while i < v.len() {
if pred(&v[i].member) {
evicted.push(v.remove(i).monitor);
} else {
i += 1;
}
}
!v.is_empty() !v.is_empty()
}); });
evicted
} }
/// Raw enumeration of a group's members — no liveness filtering (the /// Drain-on-contact death hook. Drains every membership monitor in `group`
/// public `members` layers that on in Phase 3). Absent group → empty. /// with a non-blocking `try_recv`: a delivered `Down` (any reason) or a
pub(crate) fn members_of(&self, group: &str) -> Vec<Member> { /// closed channel means that member is dead. On the first death detected,
self.groups.get(group).cloned().unwrap_or_default() /// sweep *all* of the dead pids out of *every* group via [`remove_where`]
/// — a death is removed from each group it joined, not just the one being
/// touched. Returns the evicted monitors to drop outside the lock.
fn reap_group(&mut self, group: &str) -> Vec<Monitor> {
let dead: Vec<Pid> = {
let Some(v) = self.groups.get(group) else {
return Vec::new();
};
v.iter()
.filter_map(|e| match e.monitor.rx.try_recv() {
// A Down arrived, or the channel closed and drained: dead.
Ok(Some(_)) | Err(_) => Some(e.member.pid),
// Empty but open — the sender still lives in the slot: alive.
Ok(None) => None,
})
.collect()
};
if dead.is_empty() {
return Vec::new();
}
self.remove_where(|m| dead.contains(&m.pid))
}
/// Raw enumeration of a group's members — no liveness filtering (Phase 3
/// layers a generation-checked backstop on the public `members`). Absent
/// group → empty.
fn members_of(&self, group: &str) -> Vec<Member> {
self.groups
.get(group)
.map(|v| v.iter().map(|e| e.member).collect())
.unwrap_or_default()
}
/// The first member of `group`, if any — backing the stateless first-live
/// `pick` without cloning the whole group.
fn first_member(&self, group: &str) -> Option<Member> {
self.groups.get(group).and_then(|v| v.first().map(|e| e.member))
} }
} }
@@ -184,62 +254,107 @@ fn member_for(inner: &crate::runtime::RuntimeInner, pid: Pid) -> Member {
} }
/// Add `pid` to `group`. The same pid may join many groups; within one group a /// Add `pid` to `group`. The same pid may join many groups; within one group a
/// pid is a member at most once (idempotent). Returns `true` if this call /// pid is a member at most once (idempotent). Returns `true` if this call newly
/// newly added the membership. /// added the membership, `false` if it was already a member.
///
/// Installs a `monitor(pid)` whose one-shot `Down` drives eviction: the
/// registration races `finalize_actor` under the slot's cold lock exactly as
/// every other monitor does, so no death slips between the join and the
/// registration. A redundant (idempotent) join tears its extra monitor back
/// down.
/// ///
/// Panics if called outside `Runtime::run()`. /// Panics if called outside `Runtime::run()`.
//
// Phase 2 will install `monitor(pid)` on first join (registered under the
// shared mutex, racing `finalize_actor` exactly as every other monitor does)
// and route its one-shot `Down` to `remove_where(|m| m.pid == pid)`.
pub fn join(group: impl Into<String>, pid: Pid) -> bool { pub fn join(group: impl Into<String>, pid: Pid) -> bool {
let group = group.into(); let group = group.into();
with_runtime(|inner| { // Install the monitor BEFORE taking the group lock: monitor() acquires the
let member = member_for(inner, pid); // target's cold lock (Leaf), and two Leaf locks are never held at once.
inner.process_groups.lock().join(&group, member) let mon = monitor(pid);
})
let (rejected, reaped) = with_runtime(|inner| {
let ms = Membership { member: member_for(inner, pid), monitor: mon };
let mut pg = inner.process_groups.lock();
let reaped = pg.reap_group(&group);
let rejected = pg.join(&group, ms);
(rejected, reaped)
});
// Outside the group lock: drop the reaped (dead) monitors, and if this join
// was redundant, demonitor + drop the extra monitor we just installed.
drop(reaped);
match rejected {
Some(dup) => {
demonitor(&dup.monitor);
false
}
None => true,
}
} }
/// Drop `pid`'s membership of `group`. Returns whether a membership was /// Drop `pid`'s membership of `group`. Returns whether a membership was
/// removed. /// removed. The membership's monitor is demonitored and dropped.
/// ///
/// Panics if called outside `Runtime::run()`. /// Panics if called outside `Runtime::run()`.
pub fn leave(group: &str, pid: Pid) -> bool { pub fn leave(group: &str, pid: Pid) -> bool {
with_runtime(|inner| { let (removed, reaped) = with_runtime(|inner| {
let member = member_for(inner, pid); let member = member_for(inner, pid);
inner.process_groups.lock().leave(group, member) let mut pg = inner.process_groups.lock();
}) let reaped = pg.reap_group(group);
let removed = pg.leave(group, member);
(removed, reaped)
});
drop(reaped);
match removed {
Some(ms) => {
demonitor(&ms.monitor);
true
}
None => false,
}
} }
/// Fan-out read: every member of `group`. /// Fan-out read: every live member of `group`.
/// ///
/// The set is self-pruning once the death hook lands (Phase 2), so liveness is /// The touched group is reaped first, so dead members are evicted before the
/// maintained by eviction rather than filtered on read; Phase 3 adds a /// read rather than filtered out of it. Phase 3 adds a generation-checked
/// generation-checked liveness read as the belt-and-braces backstop. /// liveness read as a belt-and-braces backstop.
/// ///
/// Panics if called outside `Runtime::run()`. /// Panics if called outside `Runtime::run()`.
pub fn members(group: &str) -> Vec<Pid> { pub fn members(group: &str) -> Vec<Pid> {
with_runtime(|inner| { let (pids, reaped) = with_runtime(|inner| {
inner.process_groups.lock().members_of(group).into_iter().map(|m| m.pid).collect() let mut pg = inner.process_groups.lock();
}) let reaped = pg.reap_group(group);
let pids: Vec<Pid> = pg.members_of(group).into_iter().map(|m| m.pid).collect();
(pids, reaped)
});
drop(reaped);
pids
} }
/// Pool/discovery read: one member of `group`, or `None` if empty. Stateless /// Pool/discovery read: one live member of `group`, or `None` if empty.
/// first-live selection (the first-*live* refinement lands in Phase 3; smarter /// Stateless first-live selection (the touched group is reaped first, so the
/// routing is an explicitly later, clustered concern). /// first surviving member is live; smarter routing is an explicitly later,
/// clustered concern per RFC 010).
/// ///
/// Panics if called outside `Runtime::run()`. /// Panics if called outside `Runtime::run()`.
pub fn pick(group: &str) -> Option<Pid> { pub fn pick(group: &str) -> Option<Pid> {
with_runtime(|inner| { let (picked, reaped) = with_runtime(|inner| {
inner.process_groups.lock().members_of(group).into_iter().next().map(|m| m.pid) let mut pg = inner.process_groups.lock();
}) let reaped = pg.reap_group(group);
let picked = pg.first_member(group).map(|m| m.pid);
(picked, reaped)
});
drop(reaped);
picked
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::channel::{channel, Sender};
use crate::monitor::{Down, DownReason, MonitorId};
fn m(index: u32, generation: u32) -> Member { fn member(index: u32, generation: u32) -> Member {
Member { Member {
node: DEFAULT_NODE_ID, node: DEFAULT_NODE_ID,
incarnation: DEFAULT_INCARNATION, incarnation: DEFAULT_INCARNATION,
@@ -247,69 +362,137 @@ mod tests {
} }
} }
/// A synthetic membership with a real (but slot-less) monitor channel. The
/// returned `Sender` stands in for the slot's `Down` sender: hold it to
/// keep the member "alive" (`try_recv` → `Ok(None)`), `send` a `Down` to
/// simulate death, or `drop` it to simulate a drained/closed channel.
fn synth(index: u32, generation: u32) -> (Membership, Sender<Down>) {
let pid = Pid::new(index, generation);
let (tx, rx) = channel::<Down>();
let ms = Membership {
member: member(index, generation),
monitor: Monitor { id: MonitorId(0), target: pid, rx },
};
(ms, tx)
}
#[test] #[test]
fn join_is_idempotent_within_a_group() { fn join_is_idempotent_within_a_group() {
let mut pg = ProcessGroups::new(); let mut pg = ProcessGroups::new();
assert!(pg.join("workers", m(1, 0)), "first join is new"); let (a, _ta) = synth(1, 0);
assert!(!pg.join("workers", m(1, 0)), "second identical join is a no-op"); let (b, _tb) = synth(1, 0);
assert_eq!(pg.members_of("workers"), vec![m(1, 0)]); assert!(pg.join("workers", a).is_none(), "first join inserts");
assert!(pg.join("workers", b).is_some(), "second identical join is handed back");
assert_eq!(pg.members_of("workers"), vec![member(1, 0)]);
} }
#[test] #[test]
fn same_pid_in_many_groups_is_independent() { fn same_pid_in_many_groups_is_independent() {
let mut pg = ProcessGroups::new(); let mut pg = ProcessGroups::new();
pg.join("a", m(1, 0)); let (a, _ta) = synth(1, 0);
pg.join("b", m(1, 0)); let (b, _tb) = synth(1, 0);
pg.join("b", m(2, 0)); let (c, _tc) = synth(2, 0);
assert_eq!(pg.members_of("a"), vec![m(1, 0)]); pg.join("a", a);
assert_eq!(pg.members_of("b"), vec![m(1, 0), m(2, 0)]); pg.join("b", b);
pg.join("b", c);
assert_eq!(pg.members_of("a"), vec![member(1, 0)]);
assert_eq!(pg.members_of("b"), vec![member(1, 0), member(2, 0)]);
} }
#[test] #[test]
fn distinct_generations_are_distinct_members() { fn distinct_generations_are_distinct_members() {
// ABA guard: same slot index, different generation = different actor. // ABA guard: same slot index, different generation = different actor.
let mut pg = ProcessGroups::new(); let mut pg = ProcessGroups::new();
assert!(pg.join("g", m(1, 0))); let (a, _ta) = synth(1, 0);
assert!(pg.join("g", m(1, 1)), "different generation is a distinct member"); let (b, _tb) = synth(1, 1);
assert_eq!(pg.members_of("g"), vec![m(1, 0), m(1, 1)]); assert!(pg.join("g", a).is_none());
assert!(pg.join("g", b).is_none(), "different generation is a distinct member");
assert_eq!(pg.members_of("g"), vec![member(1, 0), member(1, 1)]);
} }
#[test] #[test]
fn leave_removes_one_membership_and_prunes_empty_groups() { fn leave_removes_one_membership_and_prunes_empty_groups() {
let mut pg = ProcessGroups::new(); let mut pg = ProcessGroups::new();
pg.join("g", m(1, 0)); let (a, _ta) = synth(1, 0);
pg.join("g", m(2, 0)); let (b, _tb) = synth(2, 0);
assert!(pg.leave("g", m(1, 0))); pg.join("g", a);
assert_eq!(pg.members_of("g"), vec![m(2, 0)]); pg.join("g", b);
assert!(!pg.leave("g", m(1, 0)), "second leave finds nothing"); assert!(pg.leave("g", member(1, 0)).is_some());
assert!(pg.leave("g", m(2, 0))); assert_eq!(pg.members_of("g"), vec![member(2, 0)]);
assert!(pg.leave("g", member(1, 0)).is_none(), "second leave finds nothing");
assert!(pg.leave("g", member(2, 0)).is_some());
assert!(pg.members_of("g").is_empty(), "group is now empty"); assert!(pg.members_of("g").is_empty(), "group is now empty");
// Leaving a never-joined group is a clean no-op. assert!(pg.leave("never", member(9, 0)).is_none(), "leaving an unknown group is a no-op");
assert!(!pg.leave("never", m(9, 0)));
} }
#[test] #[test]
fn remove_where_sweeps_every_group() { fn remove_where_sweeps_every_group() {
let mut pg = ProcessGroups::new(); let mut pg = ProcessGroups::new();
pg.join("a", m(1, 0)); for (g, (m, _t)) in [("a", synth(1, 0)), ("a", synth(2, 0)), ("b", synth(1, 0)), ("c", synth(3, 0))] {
pg.join("a", m(2, 0)); pg.join(g, m);
pg.join("b", m(1, 0)); }
pg.join("c", m(3, 0));
// Death of pid index 1 (any generation) evicts it everywhere. // Death of pid index 1 (any generation) evicts it everywhere.
pg.remove_where(|mem| mem.pid.index() == 1); let evicted = pg.remove_where(|mem| mem.pid.index() == 1);
assert_eq!(pg.members_of("a"), vec![m(2, 0)]); assert_eq!(evicted.len(), 2, "pid 1 was in a and b");
assert_eq!(pg.members_of("a"), vec![member(2, 0)]);
assert!(pg.members_of("b").is_empty(), "b held only pid 1; pruned"); assert!(pg.members_of("b").is_empty(), "b held only pid 1; pruned");
assert_eq!(pg.members_of("c"), vec![m(3, 0)]); assert_eq!(pg.members_of("c"), vec![member(3, 0)]);
} }
#[test] #[test]
fn remove_where_can_match_an_incarnation_sweep() { fn remove_where_can_match_an_incarnation_sweep() {
// Shape check for the later evict_incarnation(node, inc) caller. // Shape check for the later evict_incarnation(node, inc) caller.
let mut pg = ProcessGroups::new(); let mut pg = ProcessGroups::new();
let dead = Member { incarnation: Incarnation::new(7), ..m(1, 0) }; let pid = Pid::new(1, 0);
let (tx, rx) = channel::<Down>();
let dead = Membership {
member: Member { node: DEFAULT_NODE_ID, incarnation: Incarnation::new(7), pid },
monitor: Monitor { id: MonitorId(0), target: pid, rx },
};
let _keep = tx;
let (live, _tl) = synth(2, 0);
pg.join("g", dead); pg.join("g", dead);
pg.join("g", m(2, 0)); pg.join("g", live);
pg.remove_where(|mem| mem.incarnation == Incarnation::new(7)); let evicted = pg.remove_where(|mem| mem.incarnation == Incarnation::new(7));
assert_eq!(pg.members_of("g"), vec![m(2, 0)]); assert_eq!(evicted.len(), 1);
assert_eq!(pg.members_of("g"), vec![member(2, 0)]);
}
#[test]
fn reap_keeps_live_members() {
let mut pg = ProcessGroups::new();
let (a, _ta) = synth(1, 0); // sender held: member stays alive
pg.join("a", a);
assert!(pg.reap_group("a").is_empty(), "no deaths");
assert_eq!(pg.members_of("a"), vec![member(1, 0)]);
}
#[test]
fn reap_evicts_a_dead_member_and_sweeps_all_its_groups() {
let mut pg = ProcessGroups::new();
let (a1, ta1) = synth(1, 0); // pid 1 in group a
let (a2, _ta2) = synth(2, 0); // pid 2 in group a (stays alive)
let (b1, _tb1) = synth(1, 0); // pid 1 in group b
pg.join("a", a1);
pg.join("a", a2);
pg.join("b", b1);
// pid 1 dies: its group-a monitor receives a Down. Its group-b monitor
// has not — reap must still sweep pid 1 out of b by the pid predicate.
ta1.send(Down { pid: Pid::new(1, 0), reason: DownReason::Exit }).unwrap();
let evicted = pg.reap_group("a");
assert_eq!(evicted.len(), 2, "pid 1's memberships in both a and b are evicted");
assert_eq!(pg.members_of("a"), vec![member(2, 0)]);
assert!(pg.members_of("b").is_empty(), "swept from b too; pruned");
}
#[test]
fn reap_treats_a_closed_channel_as_dead() {
let mut pg = ProcessGroups::new();
let (a, ta) = synth(1, 0);
pg.join("a", a);
drop(ta); // sender gone, queue empty → try_recv = Err(RecvError) = dead
let evicted = pg.reap_group("a");
assert_eq!(evicted.len(), 1);
assert!(pg.members_of("a").is_empty());
} }
} }
tests/pg.rs
+131
View File
@@ -0,0 +1,131 @@
//! Process-group tests that run under the scheduler: `join` installs a real
//! monitor on a live actor, and a real death drives eviction on next contact.
//! (Pure structural invariants live in the `pg` unit tests.)
use smarm::{channel, members, pick, run, spawn};
use smarm::{join, leave};
#[test]
fn join_then_members_lists_a_live_member() {
run(|| {
let (tx, rx) = channel::<()>();
let h = spawn(move || {
rx.recv().unwrap();
});
let pid = h.pid();
assert!(join("workers", pid), "first join is new");
assert!(!join("workers", pid), "second join is idempotent");
assert_eq!(members("workers"), vec![pid]);
assert_eq!(pick("workers"), Some(pid));
tx.send(()).unwrap();
h.join().unwrap();
});
}
#[test]
fn a_dead_actor_vanishes_from_every_group_it_joined() {
run(|| {
let (tx, rx) = channel::<()>();
let h = spawn(move || {
rx.recv().unwrap();
});
let pid = h.pid();
join("g1", pid);
join("g2", pid);
assert_eq!(members("g1"), vec![pid]);
assert_eq!(members("g2"), vec![pid]);
// Release and reap the actor. finalize_actor queues the Down to our
// monitors before unparking joiners, so by the time join() returns the
// Down is already waiting in the membership channel.
tx.send(()).unwrap();
h.join().unwrap();
// Drain-on-contact: touching g1 detects the death and sweeps the pid
// out of every group (g2 included), not just g1.
assert!(members("g1").is_empty(), "evicted from the touched group");
assert!(members("g2").is_empty(), "and swept from the untouched group");
assert_eq!(pick("g1"), None);
});
}
#[test]
fn pick_returns_none_when_the_only_member_is_dead() {
run(|| {
let (tx, rx) = channel::<()>();
let h = spawn(move || {
rx.recv().unwrap();
});
let pid = h.pid();
join("pool", pid);
tx.send(()).unwrap();
h.join().unwrap();
assert_eq!(pick("pool"), None);
assert!(members("pool").is_empty());
});
}
#[test]
fn live_members_survive_a_peers_death() {
run(|| {
let (tx_a, rx_a) = channel::<()>();
let (tx_b, rx_b) = channel::<()>();
let a = spawn(move || {
rx_a.recv().unwrap();
});
let b = spawn(move || {
rx_b.recv().unwrap();
});
join("svc", a.pid());
join("svc", b.pid());
// Kill a; b is still parked on its channel.
tx_a.send(()).unwrap();
a.join().unwrap();
assert_eq!(members("svc"), vec![b.pid()], "only the dead peer is reaped");
assert_eq!(pick("svc"), Some(b.pid()));
tx_b.send(()).unwrap();
b.join().unwrap();
});
}
#[test]
fn leave_drops_a_membership_without_affecting_others() {
run(|| {
let (tx_a, rx_a) = channel::<()>();
let (tx_b, rx_b) = channel::<()>();
let a = spawn(move || {
rx_a.recv().unwrap();
});
let b = spawn(move || {
rx_b.recv().unwrap();
});
join("g", a.pid());
join("g", b.pid());
assert!(leave("g", a.pid()), "a was a member");
assert!(!leave("g", a.pid()), "leaving twice finds nothing");
assert_eq!(members("g"), vec![b.pid()]);
tx_a.send(()).unwrap();
tx_b.send(()).unwrap();
a.join().unwrap();
b.join().unwrap();
});
}
#[test]
fn joining_an_already_dead_pid_is_evicted_on_next_contact() {
run(|| {
let h = spawn(|| {});
let pid = h.pid();
h.join().unwrap(); // actor is finalized before we join it to anything
// monitor() on a gone pid queues a NoProc Down immediately, so the
// membership is reaped the next time the group is touched.
join("late", pid);
assert!(members("late").is_empty(), "dead-at-join member is reaped on read");
assert_eq!(pick("late"), None);
});
}