diff --git a/src/lib.rs b/src/lib.rs index 4b0e424..e90c8ab 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 registry; pub mod link; pub mod gen_server; pub mod runtime; @@ -50,6 +51,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 registry::{name_of, register, unregister, whereis, RegisterError}; pub use runtime::{init, Config, Runtime}; pub use scheduler::{ block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_under, wait_readable, diff --git a/src/registry.rs b/src/registry.rs new file mode 100644 index 0000000..3a63e93 --- /dev/null +++ b/src/registry.rs @@ -0,0 +1,169 @@ +//! Named pid registry. +//! +//! `register(name, pid)` binds a name to a live actor; `whereis(name)` looks +//! it up; `name_of(pid)` is the inverse. Erlang semantics: at most one name +//! per pid, at most one pid per name, registering over a live binding is an +//! error. The registry is a *bimap* (two `HashMap`s kept in exact inverse +//! under one lock) so both lookup directions are O(1) — `name_of` exists so +//! supervisors, tracing, and diagnostics can label a pid without scanning. +//! +//! ## Cleanup is lazy +//! +//! There is no hook in `finalize_actor` and no name field in `Slot` (which +//! would buy into the reset-in-three-places slot invariant). Instead every +//! operation that touches a binding checks the bound pid's liveness via the +//! generation-checked slot word — a `Pid` is `(index, generation)` and a +//! generation is never reused, so a stale binding is *detectable*, never +//! misdirected. Bindings to dead actors behave as absent and are pruned on +//! contact: `whereis`/`name_of` return `None`, `register` treats the name as +//! free. The cost is that a dead binding lingers until something looks at it; +//! the payoff is zero coupling to the actor lifecycle. +//! +//! ## Locking +//! +//! One `RawMutex` (Leaf class) in `RuntimeInner`. Liveness checks under it +//! read only the atomic slot state word — no second lock is ever taken, so +//! the leaf rule holds trivially. + +use crate::pid::Pid; +use crate::scheduler::with_runtime; +use std::collections::HashMap; + +/// Why a [`register`] call was rejected. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RegisterError { + /// The name is bound to a different, still-live actor. + NameTaken { holder: Pid }, + /// The pid already has a (live) registered name; one name per pid. + PidAlreadyRegistered { name: String }, + /// The pid does not refer to a live actor. + NoProc, +} + +impl std::fmt::Display for RegisterError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RegisterError::NameTaken { holder } => { + write!(f, "name is already registered to live actor {holder}") + } + RegisterError::PidAlreadyRegistered { name } => { + write!(f, "pid is already registered as {name:?}") + } + RegisterError::NoProc => write!(f, "pid does not refer to a live actor"), + } + } +} + +impl std::error::Error for RegisterError {} + +/// The bimap. Invariant (held under the registry lock): `by_name` and +/// `by_pid` are exact inverses of each other at all times. +pub(crate) struct Registry { + by_name: HashMap, + by_pid: HashMap, +} + +impl Registry { + pub(crate) fn new() -> Self { + Self { by_name: HashMap::new(), by_pid: HashMap::new() } + } + + fn remove_binding(&mut self, name: &str, pid: Pid) { + self.by_name.remove(name); + self.by_pid.remove(&pid); + debug_assert_eq!(self.by_name.len(), self.by_pid.len(), "bimap out of sync"); + } + + fn insert_binding(&mut self, name: String, pid: Pid) { + self.by_pid.insert(pid, name.clone()); + self.by_name.insert(name, pid); + debug_assert_eq!(self.by_name.len(), self.by_pid.len(), "bimap out of sync"); + } +} + +/// Is `pid` a live actor right now? Atomic slot-word read; no lock. +fn live(inner: &crate::runtime::RuntimeInner, pid: Pid) -> bool { + inner.slot_at(pid).is_some_and(|s| s.is_live_for(pid)) +} + +/// Bind `name` to the live actor `pid`. +/// +/// Fails with [`RegisterError::NoProc`] if `pid` is not live, +/// [`RegisterError::NameTaken`] if the name is bound to a *live* actor +/// (a binding to a dead actor is evicted and the name treated as free), and +/// [`RegisterError::PidAlreadyRegistered`] if `pid` already has a name. +/// +/// Panics if called outside `Runtime::run()`. +pub fn register(name: impl Into, pid: Pid) -> Result<(), RegisterError> { + let name = name.into(); + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + if !live(inner, pid) { + return Err(RegisterError::NoProc); + } + let prior = reg.by_name.get(&name).copied(); + if let Some(holder) = prior { + if holder == pid { + return Ok(()); // already exactly this binding: idempotent + } + if live(inner, holder) { + return Err(RegisterError::NameTaken { holder }); + } + // Stale binding: the holder died. Evict and treat the name as free. + reg.remove_binding(&name, holder); + } + if let Some(existing) = reg.by_pid.get(&pid) { + // `pid` is live (checked above) and pids are never reused, so this + // binding is necessarily current — one name per pid. + return Err(RegisterError::PidAlreadyRegistered { name: existing.clone() }); + } + reg.insert_binding(name, pid); + Ok(()) + }) +} + +/// Look up the pid bound to `name`. `None` if unbound, or if the bound actor +/// is no longer live (the stale binding is pruned on the way out). +pub fn whereis(name: &str) -> Option { + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + let pid = *reg.by_name.get(name)?; + if live(inner, pid) { + Some(pid) + } else { + reg.remove_binding(name, pid); + None + } + }) +} + +/// The inverse lookup: the name `pid` is registered under, if any. `None` if +/// unregistered or no longer live (pruning the stale binding). +pub fn name_of(pid: Pid) -> Option { + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + let name = reg.by_pid.get(&pid)?.clone(); + if live(inner, pid) { + Some(name) + } else { + reg.remove_binding(&name, pid); + None + } + }) +} + +/// Remove the binding for `name`, returning the pid it was bound to if that +/// actor is still live. A binding to a dead actor is pruned and reported as +/// `None`, consistent with [`whereis`]. +pub fn unregister(name: &str) -> Option { + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + let pid = *reg.by_name.get(name)?; + reg.remove_binding(name, pid); + if live(inner, pid) { + Some(pid) + } else { + None + } + }) +} diff --git a/src/runtime.rs b/src/runtime.rs index 42dc7b8..55a838a 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -450,6 +450,10 @@ pub(crate) struct RuntimeInner { /// Preemption knobs, written into each scheduler thread's locals on startup. pub(crate) alloc_interval: u32, pub(crate) timeslice_cycles: u64, + /// The name <-> pid registry (bidirectional). RawMutex Leaf: never held + /// with any other lock; liveness checks under it read only the atomic + /// slot word. + pub(crate) registry: 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. @@ -482,6 +486,7 @@ impl RuntimeInner { sleeping: AtomicU32::new(0), alloc_interval, timeslice_cycles, + registry: RawMutex::new(crate::registry::Registry::new()), stack_pool: RawMutex::new(Vec::new()), stack_pool_cap, }) diff --git a/tests/registry.rs b/tests/registry.rs new file mode 100644 index 0000000..cd13b7b --- /dev/null +++ b/tests/registry.rs @@ -0,0 +1,221 @@ +//! Named registry tests. Run under the scheduler: registration requires a +//! live runtime and live actors. + +use smarm::{channel, name_of, register, run, spawn, unregister, whereis, RegisterError}; + +#[test] +fn register_whereis_roundtrip() { + run(|| { + let (tx, rx) = channel::<()>(); + let h = spawn(move || { + rx.recv().unwrap(); + }); + register("worker", h.pid()).unwrap(); + assert_eq!(whereis("worker"), Some(h.pid())); + assert_eq!(name_of(h.pid()).as_deref(), Some("worker")); + tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn register_is_idempotent_for_same_binding() { + run(|| { + let (tx, rx) = channel::<()>(); + let h = spawn(move || { + rx.recv().unwrap(); + }); + register("svc", h.pid()).unwrap(); + // Same name, same pid: a no-op Ok, not NameTaken. + register("svc", h.pid()).unwrap(); + tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn duplicate_name_on_live_holder_is_rejected() { + run(|| { + let (tx, rx) = channel::<()>(); + let (tx2, rx2) = channel::<()>(); + let a = spawn(move || { + rx.recv().unwrap(); + }); + let b = spawn(move || { + rx2.recv().unwrap(); + }); + register("svc", a.pid()).unwrap(); + assert_eq!( + register("svc", b.pid()), + Err(RegisterError::NameTaken { holder: a.pid() }) + ); + tx.send(()).unwrap(); + tx2.send(()).unwrap(); + a.join().unwrap(); + b.join().unwrap(); + }); +} + +#[test] +fn one_name_per_pid() { + run(|| { + let (tx, rx) = channel::<()>(); + let h = spawn(move || { + rx.recv().unwrap(); + }); + register("first", h.pid()).unwrap(); + assert_eq!( + register("second", h.pid()), + Err(RegisterError::PidAlreadyRegistered { name: "first".into() }) + ); + tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn registering_a_dead_pid_is_noproc() { + run(|| { + let h = spawn(|| {}); + let pid = h.pid(); + h.join().unwrap(); + assert_eq!(register("ghost", pid), Err(RegisterError::NoProc)); + }); +} + +#[test] +fn binding_evaporates_on_death_and_name_is_reusable() { + run(|| { + let a = spawn(|| {}); + let pid_a = a.pid(); + register("svc", pid_a).unwrap(); + a.join().unwrap(); + + // Dead holder: both lookup directions report unbound. + assert_eq!(whereis("svc"), None); + assert_eq!(name_of(pid_a), None); + + // And the name is free for a successor. + let (tx, rx) = channel::<()>(); + let b = spawn(move || { + rx.recv().unwrap(); + }); + register("svc", b.pid()).unwrap(); + assert_eq!(whereis("svc"), Some(b.pid())); + tx.send(()).unwrap(); + b.join().unwrap(); + }); +} + +#[test] +fn register_over_a_dead_holder_succeeds_without_lookup_in_between() { + // The eviction path inside register() itself (not via whereis pruning). + run(|| { + let a = spawn(|| {}); + let pid_a = a.pid(); + register("svc", pid_a).unwrap(); + a.join().unwrap(); + + let (tx, rx) = channel::<()>(); + let b = spawn(move || { + rx.recv().unwrap(); + }); + register("svc", b.pid()).unwrap(); + assert_eq!(whereis("svc"), Some(b.pid())); + assert_eq!(name_of(pid_a), None); + tx.send(()).unwrap(); + b.join().unwrap(); + }); +} + +#[test] +fn unregister_frees_both_directions() { + run(|| { + let (tx, rx) = channel::<()>(); + let h = spawn(move || { + rx.recv().unwrap(); + }); + register("svc", h.pid()).unwrap(); + assert_eq!(unregister("svc"), Some(h.pid())); + assert_eq!(whereis("svc"), None); + assert_eq!(name_of(h.pid()), None); + assert_eq!(unregister("svc"), None); + + // The pid may take a new name afterwards. + register("svc2", h.pid()).unwrap(); + assert_eq!(name_of(h.pid()).as_deref(), Some("svc2")); + tx.send(()).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn whereis_from_another_actor_and_usable_with_runtime_apis() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + let stopped = Arc::new(AtomicBool::new(false)); + let stopped2 = stopped.clone(); + run(move || { + let (tx, rx) = channel::<()>(); + let svc = spawn(move || { + // Parks forever; only a request_stop ends it. + let _ = rx.recv(); + }); + register("stoppable", svc.pid()).unwrap(); + + let stopped3 = stopped2.clone(); + let client = spawn(move || { + let pid = whereis("stoppable").expect("name must resolve cross-actor"); + // The registry's pids plug into the rest of the runtime API. + smarm::request_stop(pid); + stopped3.store(true, Ordering::Relaxed); + }); + client.join().unwrap(); + svc.join().unwrap(); // a stopped actor joins Ok + drop(tx); + }); + assert!(stopped.load(Ordering::Relaxed)); +} + +#[test] +fn registry_under_concurrent_churn_multi_thread() { + // Many actors racing to claim the same names while holders die: the + // bimap invariant (debug-asserted internally) and Erlang semantics must + // hold under real parallelism. + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::Arc; + + let wins = Arc::new(AtomicU32::new(0)); + let wins2 = wins.clone(); + smarm::init(smarm::Config::exact(4)).run(move || { + let mut handles = Vec::new(); + for round in 0..8 { + let name = format!("contested-{}", round % 2); + for _ in 0..8 { + let name = name.clone(); + let wins = wins2.clone(); + handles.push(spawn(move || { + let me = smarm::self_pid(); + match register(&name, me) { + Ok(()) => { + wins.fetch_add(1, Ordering::Relaxed); + smarm::yield_now(); + // May have been pruned-by-contact never; we are + // alive, so our binding must still resolve to us. + assert_eq!(whereis(&name), Some(me)); + assert_eq!(unregister(&name), Some(me)); + } + Err(RegisterError::NameTaken { .. }) => {} + Err(e) => panic!("unexpected register error: {e}"), + } + })); + } + } + for h in handles { + h.join().unwrap(); + } + }); + // At least one registration per name must have succeeded. + assert!(wins.load(Ordering::Relaxed) >= 2); +}