diff --git a/src/lib.rs b/src/lib.rs index 2ffb49c..fcd5ae1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,6 +24,7 @@ pub mod io; pub mod mutex; pub mod monitor; pub mod registry; +pub mod pg; pub mod link; pub mod gen_server; pub mod runtime; @@ -54,6 +55,7 @@ pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; pub use pid::Pid; +pub use pg::{join, leave, members, pick, Incarnation, Member, NodeId}; pub use registry::{name_of, register, unregister, whereis, RegisterError}; pub use runtime::{init, Config, Runtime}; pub use scheduler::{ diff --git a/src/pg.rs b/src/pg.rs new file mode 100644 index 0000000..85ee2f8 --- /dev/null +++ b/src/pg.rs @@ -0,0 +1,315 @@ +//! Process groups — a `name → multiset` map (RFC 012). +//! +//! Sits parallel to [`registry`](crate::registry), not on top of it. The +//! 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 +//! generalization of the bimap; it is its own keyspace sharing only the +//! liveness helper and the lock *class*. +//! +//! ## Cleanup is eager (unlike the registry) +//! +//! The registry prunes a stale binding lazily, on contact, because it only +//! ever resolves one binding at a time. A group is *iterated* — `members` +//! fans out to every member — so it must not carry dead members across a +//! broadcast. Removal is therefore eager and monitor-driven: each `join` +//! installs a `monitor(pid)` and the one-shot `Down` evicts the member +//! (wired in Phase 2). A generation-checked liveness read stays on the read +//! path as a belt-and-braces backstop (Phase 3), matching the registry's +//! prune-on-contact guard. +//! +//! ## Eviction is one dumb primitive +//! +//! [`ProcessGroups::remove_where`] removes every member matching a predicate +//! from every group. The primitive never knows *why* a member leaves — that +//! is the caller's concern. Its first caller is the monitor death hook (this +//! RFC); the later `evict_incarnation(node, inc)` sweep (RFC 010) reuses the +//! same predicate path, which is the whole reason to shape it as a predicate. +//! +//! ## Identity is cluster-shaped from the first commit +//! +//! A [`Member`] is not a bare [`Pid`]: it is `(NodeId, Incarnation, Pid)`, a +//! field-for-field image of a modern BEAM pid (`NEW_PID_EXT`) so the eventual +//! ETF codec (RFC 011) is a near-identity mapping. In this RFC `NodeId` and +//! `Incarnation` are runtime-init constants — one node, one fixed incarnation +//! — threaded through storage and eviction anyway so the public surface never +//! has to change to acquire them once clustering (RFC 010) supplies real +//! values. No cluster types leak out of the public functions: callers pass and +//! receive [`Pid`]; the node/incarnation are filled from runtime identity. +//! +//! ## Locking +//! +//! One `RawMutex` (Leaf class) on `RuntimeInner`, mirroring `registry` +//! exactly: never held with any other lock, and the eviction path keeps it +//! off the send path (it does not send under the lock). Liveness checks under +//! it read only the atomic slot word, so the leaf rule holds trivially. + +use crate::pid::Pid; +use crate::scheduler::with_runtime; +use std::collections::HashMap; + +/// A cluster node handle. A `u32` integer handle, *not* an interned atom — the +/// single deliberate divergence from the BEAM wire shape (RFC 011 names it). +#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)] +pub struct NodeId(u32); + +impl NodeId { + #[inline] + pub const fn new(v: u32) -> Self { + Self(v) + } + #[inline] + pub const fn get(self) -> u32 { + self.0 + } +} + +impl From for NodeId { + #[inline] + fn from(v: u32) -> Self { + Self(v) + } +} + +/// A node's incarnation epoch — the BEAM `Creation` field adopted verbatim: it +/// separates a crashed node from its restart. Fixed for the life of a run +/// until clustering supplies a real one. +#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)] +pub struct Incarnation(u32); + +impl Incarnation { + #[inline] + pub const fn new(v: u32) -> Self { + Self(v) + } + #[inline] + pub const fn get(self) -> u32 { + self.0 + } +} + +impl From for Incarnation { + #[inline] + fn from(v: u32) -> Self { + Self(v) + } +} + +/// The fixed single-node identity used until clustering (RFC 010) supplies +/// real values. Carried like `wake_slot` so the public API never has to change +/// to acquire it. +pub const DEFAULT_NODE_ID: NodeId = NodeId(0); +/// The fixed incarnation for the single-node default. Non-zero so it never +/// collides with a BEAM "any creation" wildcard at interop time. +pub const DEFAULT_INCARNATION: Incarnation = Incarnation(1); + +/// A group member's full identity: `(node, incarnation, pid)`. +/// +/// Deliberately a field-for-field image of a modern BEAM pid (`NEW_PID_EXT`). +/// In memory it is a plain struct — no wire packing; the packed representation +/// belongs to the `RemoteRef` boundary (RFC 010/011), not here. +#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)] +pub struct Member { + /// Which node the pid lives on. `DEFAULT_NODE_ID` while single-node. + pub node: NodeId, + /// The node's incarnation epoch at the time of joining. + pub incarnation: Incarnation, + /// Pure local slot identity — unchanged; cluster identity is layered + /// *around* it here rather than overloading `Pid::generation`. + pub pid: Pid, +} + +/// The store: `name → multiset`. Within a single group a `Member` +/// 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 +/// width admits multiples in general. Held under one Leaf-class `RawMutex`. +pub(crate) struct ProcessGroups { + groups: HashMap>, +} + +impl ProcessGroups { + pub(crate) fn new() -> Self { + Self { groups: HashMap::new() } + } + + /// Add `member` to `group`. Idempotent within a group: returns `true` if + /// it was newly inserted, `false` if it was already present. + pub(crate) fn join(&mut self, group: &str, member: Member) -> bool { + let v = self.groups.entry(group.to_owned()).or_default(); + if v.contains(&member) { + return false; + } + v.push(member); + true + } + + /// Remove `member` from `group`. Returns whether anything was removed. + /// An emptied group is pruned so the keyspace does not leak empty vecs. + pub(crate) fn leave(&mut self, group: &str, member: Member) -> bool { + let Some(v) = self.groups.get_mut(group) else { + return false; + }; + let before = v.len(); + v.retain(|m| *m != member); + let removed = v.len() != before; + if v.is_empty() { + self.groups.remove(group); + } + removed + } + + /// The one dumb eviction primitive: drop every member matching `pred` from + /// every group, pruning emptied groups. The primitive does not know *why* + /// a member leaves. First caller: the monitor death hook (Phase 2). Later: + /// `evict_incarnation` (RFC 010) over the same path. + // + // No non-test caller until Phase 2 wires the death hook to it. + #[allow(dead_code)] + pub(crate) fn remove_where(&mut self, mut pred: impl FnMut(&Member) -> bool) { + self.groups.retain(|_, v| { + v.retain(|m| !pred(m)); + !v.is_empty() + }); + } + + /// Raw enumeration of a group's members — no liveness filtering (the + /// public `members` layers that on in Phase 3). Absent group → empty. + pub(crate) fn members_of(&self, group: &str) -> Vec { + self.groups.get(group).cloned().unwrap_or_default() + } +} + +/// Build the full member identity for `pid` from runtime identity. +fn member_for(inner: &crate::runtime::RuntimeInner, pid: Pid) -> Member { + Member { node: inner.node_id, incarnation: inner.incarnation, pid } +} + +/// 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 +/// newly added the membership. +/// +/// 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, pid: Pid) -> bool { + let group = group.into(); + with_runtime(|inner| { + let member = member_for(inner, pid); + inner.process_groups.lock().join(&group, member) + }) +} + +/// Drop `pid`'s membership of `group`. Returns whether a membership was +/// removed. +/// +/// Panics if called outside `Runtime::run()`. +pub fn leave(group: &str, pid: Pid) -> bool { + with_runtime(|inner| { + let member = member_for(inner, pid); + inner.process_groups.lock().leave(group, member) + }) +} + +/// Fan-out read: every member of `group`. +/// +/// The set is self-pruning once the death hook lands (Phase 2), so liveness is +/// maintained by eviction rather than filtered on read; Phase 3 adds a +/// generation-checked liveness read as the belt-and-braces backstop. +/// +/// Panics if called outside `Runtime::run()`. +pub fn members(group: &str) -> Vec { + with_runtime(|inner| { + inner.process_groups.lock().members_of(group).into_iter().map(|m| m.pid).collect() + }) +} + +/// Pool/discovery read: one member of `group`, or `None` if empty. Stateless +/// first-live selection (the first-*live* refinement lands in Phase 3; smarter +/// routing is an explicitly later, clustered concern). +/// +/// Panics if called outside `Runtime::run()`. +pub fn pick(group: &str) -> Option { + with_runtime(|inner| { + inner.process_groups.lock().members_of(group).into_iter().next().map(|m| m.pid) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn m(index: u32, generation: u32) -> Member { + Member { + node: DEFAULT_NODE_ID, + incarnation: DEFAULT_INCARNATION, + pid: Pid::new(index, generation), + } + } + + #[test] + fn join_is_idempotent_within_a_group() { + let mut pg = ProcessGroups::new(); + assert!(pg.join("workers", m(1, 0)), "first join is new"); + assert!(!pg.join("workers", m(1, 0)), "second identical join is a no-op"); + assert_eq!(pg.members_of("workers"), vec![m(1, 0)]); + } + + #[test] + fn same_pid_in_many_groups_is_independent() { + let mut pg = ProcessGroups::new(); + pg.join("a", m(1, 0)); + pg.join("b", m(1, 0)); + pg.join("b", m(2, 0)); + assert_eq!(pg.members_of("a"), vec![m(1, 0)]); + assert_eq!(pg.members_of("b"), vec![m(1, 0), m(2, 0)]); + } + + #[test] + fn distinct_generations_are_distinct_members() { + // ABA guard: same slot index, different generation = different actor. + let mut pg = ProcessGroups::new(); + assert!(pg.join("g", m(1, 0))); + assert!(pg.join("g", m(1, 1)), "different generation is a distinct member"); + assert_eq!(pg.members_of("g"), vec![m(1, 0), m(1, 1)]); + } + + #[test] + fn leave_removes_one_membership_and_prunes_empty_groups() { + let mut pg = ProcessGroups::new(); + pg.join("g", m(1, 0)); + pg.join("g", m(2, 0)); + assert!(pg.leave("g", m(1, 0))); + assert_eq!(pg.members_of("g"), vec![m(2, 0)]); + assert!(!pg.leave("g", m(1, 0)), "second leave finds nothing"); + assert!(pg.leave("g", m(2, 0))); + 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", m(9, 0))); + } + + #[test] + fn remove_where_sweeps_every_group() { + let mut pg = ProcessGroups::new(); + pg.join("a", m(1, 0)); + pg.join("a", m(2, 0)); + pg.join("b", m(1, 0)); + pg.join("c", m(3, 0)); + // Death of pid index 1 (any generation) evicts it everywhere. + pg.remove_where(|mem| mem.pid.index() == 1); + assert_eq!(pg.members_of("a"), vec![m(2, 0)]); + assert!(pg.members_of("b").is_empty(), "b held only pid 1; pruned"); + assert_eq!(pg.members_of("c"), vec![m(3, 0)]); + } + + #[test] + fn remove_where_can_match_an_incarnation_sweep() { + // Shape check for the later evict_incarnation(node, inc) caller. + let mut pg = ProcessGroups::new(); + let dead = Member { incarnation: Incarnation::new(7), ..m(1, 0) }; + pg.join("g", dead); + pg.join("g", m(2, 0)); + pg.remove_where(|mem| mem.incarnation == Incarnation::new(7)); + assert_eq!(pg.members_of("g"), vec![m(2, 0)]); + } +} diff --git a/src/runtime.rs b/src/runtime.rs index dc1d9a2..1d50353 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -155,6 +155,8 @@ pub struct Config { stack_pool_cap: usize, max_actors: usize, wake_slot: bool, + node_id: crate::pg::NodeId, + incarnation: crate::pg::Incarnation, } impl Config { @@ -168,6 +170,8 @@ impl Config { stack_pool_cap: n * 4, max_actors: DEFAULT_MAX_ACTORS, wake_slot: false, + node_id: crate::pg::DEFAULT_NODE_ID, + incarnation: crate::pg::DEFAULT_INCARNATION, } } @@ -185,6 +189,8 @@ impl Config { stack_pool_cap: max * 4, max_actors: DEFAULT_MAX_ACTORS, wake_slot: false, + node_id: crate::pg::DEFAULT_NODE_ID, + incarnation: crate::pg::DEFAULT_INCARNATION, } } @@ -241,6 +247,23 @@ impl Config { self } + /// This runtime's node identity (RFC 012). Defaults to a fixed single-node + /// value; clustering (RFC 010) will supply a real one. Threaded through pg + /// storage/eviction so the process-group public API never changes to + /// acquire it. + pub fn node_id(mut self, id: impl Into) -> Self { + self.node_id = id.into(); + self + } + + /// This runtime's incarnation epoch (RFC 012) — the BEAM `Creation` analogue + /// that separates a crashed node from its restart. Defaults to a fixed + /// single-node value; constant for the life of a run. + pub fn incarnation(mut self, inc: impl Into) -> Self { + self.incarnation = inc.into(); + self + } + /// The number of scheduler threads this config resolves to. pub fn resolved_thread_count(&self) -> usize { if let Some(e) = self.exact { @@ -265,6 +288,8 @@ impl Default for Config { stack_pool_cap: avail * 4, max_actors: DEFAULT_MAX_ACTORS, wake_slot: false, + node_id: crate::pg::DEFAULT_NODE_ID, + incarnation: crate::pg::DEFAULT_INCARNATION, } } } @@ -507,6 +532,17 @@ pub(crate) struct RuntimeInner { /// with any other lock; liveness checks under it read only the atomic /// slot word. pub(crate) registry: RawMutex, + /// Runtime identity (RFC 012). Read-only after init — one node, one fixed + /// incarnation until clustering (RFC 010) supplies real values. Carried + /// like `wake_slot`; pg fills these into every `Member` so the public + /// surface stays Pid-shaped. + pub(crate) node_id: crate::pg::NodeId, + pub(crate) incarnation: crate::pg::Incarnation, + /// Process groups: `name -> multiset` (RFC 012). RawMutex Leaf, + /// exactly like `registry`: never held with any other lock; liveness + /// checks under it read only the atomic slot word, and the eviction path + /// keeps it off the send path. + pub(crate) process_groups: RawMutex, /// Recycled stacks waiting to be reused by the next spawn. pub(crate) stack_pool: RawMutex>, /// Maximum number of stacks to retain in the pool. @@ -521,6 +557,8 @@ impl RuntimeInner { stack_pool_cap: usize, max_actors: usize, wake_slot: bool, + node_id: crate::pg::NodeId, + incarnation: crate::pg::Incarnation, ) -> Arc { let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect(); let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect(); @@ -542,6 +580,9 @@ impl RuntimeInner { timeslice_cycles, wake_slot, registry: RawMutex::new(crate::registry::Registry::new()), + node_id, + incarnation, + process_groups: RawMutex::new(crate::pg::ProcessGroups::new()), stack_pool: RawMutex::new(Vec::new()), stack_pool_cap, }) @@ -712,6 +753,8 @@ pub fn init(config: Config) -> Runtime { config.stack_pool_cap, config.max_actors, config.wake_slot, + config.node_id, + config.incarnation, ), thread_count: n, }