diff --git a/src/lib.rs b/src/lib.rs index 80076cc..bc334e3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,7 +56,7 @@ pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; pub use pid::{Addr, Addressable, Name, Pid}; pub use pg::{join, leave, members, pick, Incarnation, Member, NodeId}; -pub use registry::{name_of, register, unregister, whereis, RegisterError}; +pub use registry::{register, send, unregister, whereis, RegisterError, SendError}; 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/pid.rs b/src/pid.rs index 1352b52..98bce27 100644 --- a/src/pid.rs +++ b/src/pid.rs @@ -177,7 +177,7 @@ mod typed_addr_tests { // A stand-in actor type with one message type, exercising `Addressable`. struct Counter; - enum CounterMsg { Inc, Get } + struct CounterMsg; // used only as a phantom key; no variants needed impl Addressable for Counter { type Msg = CounterMsg; } diff --git a/src/registry.rs b/src/registry.rs index 3a63e93..6c9a133 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -1,32 +1,60 @@ -//! Named pid registry. +//! Named mailbox registry — resolve a name (or pid) to a *messageable* actor. //! -//! `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. +//! ## What changed (RFC 014) //! -//! ## Cleanup is lazy +//! The old registry was a `name <-> pid` bimap: `whereis` handed back a `Pid` +//! you could not send to, because a pid is just `(index, generation)` with no +//! delivery endpoint. This rework makes resolution yield something messageable. //! -//! 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; +//! Two facts shape the structure: +//! +//! 1. **A name resolves to a single actor.** Many actors under one label is +//! what *process groups* (`pg`) are for; the registry is one-name-one-actor +//! (several names *may* point at the same actor). +//! 2. **Channels are typed**, so an actor has no single untyped mailbox. An +//! actor instead owns a *set* of typed channels — one [`Sender`] per message +//! type it accepts. So the registry maps name/pid to a [`Mailbox`]: a small +//! structure holding that actor's pid plus all of its typed channels, keyed +//! by message [`TypeId`]. +//! +//! Resolution is therefore: `name -> pid` (single actor) `-> Mailbox -> the +//! channel for message type M`. A `Name` and a `Name` on the *same* +//! actor select *different* channels purely by their type parameter, so +//! capability separation (RFC 014 §4.7) needs no extra machinery. +//! +//! ## Type erasure is contained +//! +//! Each stored channel is a `Box` that is concretely a +//! `Sender`, filed under `TypeId::of::()`. A resolve for `M` looks up +//! that exact `TypeId` and downcasts to `Sender` — keyed by the very type we +//! downcast to, so the downcast cannot fail on correct data; a failure is a +//! smarm bug, asserted in debug. The phantom `M` on [`Name`] re-imposes the +//! type at the call site, so callers never touch the erasure. +//! +//! ## Cleanup is lazy (prune-on-contact) +//! +//! As before, there is no `finalize` hook and no name field on the slot. Every +//! operation that touches a binding checks the target pid's liveness via the +//! generation-checked slot word; a binding to a dead actor behaves as absent +//! and is pruned on contact (its [`Mailbox`] and every name pointing at it are +//! dropped). The cost is a dead binding lingering 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. +//! One `RawMutex` (Leaf class) in `RuntimeInner`, exactly like the old +//! registry. The fold (name index *and* handles under the one lock) is what +//! keeps a name-addressed `send` on a single Leaf — `raw_mutex` panics on a +//! second Leaf acquired while one is held. The send path clones the `Sender` +//! **under** the Leaf lock (a `Sender::clone` takes a Channel lock, permitted +//! under a Leaf), then **releases** the Leaf and only *then* sends — a send can +//! unpark a receiver, and wakeup-bearing work runs outside the Leaf. Order is +//! **Leaf -> Channel**, as `pg`/`finalize`. -use crate::pid::Pid; -use crate::scheduler::with_runtime; +use crate::channel::Sender; +use crate::pid::{Name, Pid}; +use crate::scheduler::{self_pid, with_runtime}; +use std::any::{type_name, Any, TypeId}; use std::collections::HashMap; /// Why a [`register`] call was rejected. @@ -34,9 +62,8 @@ use std::collections::HashMap; 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. + /// The caller is not a live actor (cannot happen for `self`, kept for + /// symmetry / future explicit-pid registration). NoProc, } @@ -46,38 +73,117 @@ impl std::fmt::Display for RegisterError { 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"), + RegisterError::NoProc => write!(f, "caller is not 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. +/// Why a name-addressed [`send`] did not deliver. Carries the message back so +/// the caller never loses it (mirrors [`crate::channel::SendError`]). +/// +/// `Debug`/`Display` are hand-written so neither demands `M: Debug` — the +/// payload is returned, not printed. +pub enum SendError { + /// No live actor is currently registered under this name. + Unresolved(M), + /// The actor is live but exposes no channel for this message type. + NoChannel(M), + /// The actor's channel for this type is closed (its receiver is gone). + Closed(M), +} + +impl SendError { + /// Recover the undelivered message. + pub fn into_inner(self) -> M { + match self { + SendError::Unresolved(m) | SendError::NoChannel(m) | SendError::Closed(m) => m, + } + } + + fn variant(&self) -> &'static str { + match self { + SendError::Unresolved(_) => "Unresolved", + SendError::NoChannel(_) => "NoChannel", + SendError::Closed(_) => "Closed", + } + } +} + +impl std::fmt::Debug for SendError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "SendError::{}", self.variant()) + } +} + +impl std::fmt::Display for SendError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SendError::Unresolved(_) => write!(f, "no live actor registered under that name"), + SendError::NoChannel(_) => write!(f, "actor has no channel for this message type"), + SendError::Closed(_) => write!(f, "the actor's channel for this type is closed"), + } + } +} + +impl std::error::Error for SendError {} + +/// One typed channel of an actor, type-erased. Concretely a `Sender` filed +/// under `TypeId::of::()`; `msg_type` is `type_name::()`, kept for +/// observers (RFC 014 §4.5) and as the debug cross-check on the downcast. +struct Channel { + sender: Box, + msg_type: &'static str, +} + +/// An actor's messageable surface: its identity plus every typed channel it has +/// published, keyed by message [`TypeId`]. Stored once per live actor; reached +/// by pid (directly) or by any name pointing at that pid. +struct Mailbox { + pid: Pid, + channels: HashMap, +} + +impl Mailbox { + fn new(pid: Pid) -> Self { + Self { pid, channels: HashMap::new() } + } + + /// Clone the `Sender` for this actor, if it has one. Called **under the + /// registry Leaf lock**: `Sender::clone` takes a Channel lock, which is + /// legal under a Leaf (Leaf -> Channel). + fn clone_sender(&self) -> Option> { + let ch = self.channels.get(&TypeId::of::())?; + let tx = ch + .sender + .downcast_ref::>() + .expect("channel keyed by TypeId but downcast to its own type failed — smarm bug"); + debug_assert_eq!(ch.msg_type, type_name::(), "msg_type / TypeId disagree"); + Some(tx.clone()) + } +} + +/// The directory. Invariant (held under the registry lock): every value in +/// `by_name` is the index of a [`Mailbox`] present in `by_index`, and that +/// mailbox's `pid.index()` equals the key. Stale entries (dead actors) violate +/// nothing — they are simply pruned on contact. pub(crate) struct Registry { - by_name: HashMap, - by_pid: HashMap, + /// `pid.index() -> the actor's mailbox`. The handle store. + by_index: HashMap, + /// `name -> pid.index()`. Several names may map to one actor. + by_name: HashMap<&'static str, u32>, } impl Registry { pub(crate) fn new() -> Self { - Self { by_name: HashMap::new(), by_pid: HashMap::new() } + Self { by_index: HashMap::new(), by_name: 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"); + /// Drop a dead actor's mailbox and every name that pointed at it. + fn prune(&mut self, index: u32) { + self.by_index.remove(&index); + self.by_name.retain(|_, idx| *idx != index); } } @@ -86,84 +192,117 @@ 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`. +/// Publish the current actor's `Sender` under `name`, capturing the channel +/// so the name becomes messageable. Idempotent for the same `(name, type)`; +/// registering a *second* type under the same (or another) name on the same +/// actor just adds another channel to the actor's mailbox. /// -/// 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(); +/// Fails with [`RegisterError::NameTaken`] if the name is held by a *different* +/// live actor (a binding to a dead actor is pruned and the name treated as +/// free). Panics if called outside `Runtime::run()`. +pub fn register(name: Name, tx: Sender) -> Result<(), RegisterError> { + let me = self_pid(); + let key = name.as_str(); with_runtime(|inner| { let mut reg = inner.registry.lock(); - if !live(inner, pid) { + if !live(inner, me) { 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 let Some(&holder_idx) = reg.by_name.get(key) { + match reg.by_index.get(&holder_idx).map(|m| m.pid) { + Some(holder) if holder == me => {} // same actor: just add the channel below + Some(holder) if live(inner, holder) => { + return Err(RegisterError::NameTaken { holder }); + } + Some(_) => reg.prune(holder_idx), // dead holder: free the name + None => { + reg.by_name.remove(key); // dangling name: free it + } } - 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() }); + // Insert or extend my mailbox. A leftover mailbox at this slot index + // from a dead prior incarnation (pid mismatch) is replaced wholesale. + let mb = reg.by_index.entry(me.index()).or_insert_with(|| Mailbox::new(me)); + if mb.pid != me { + *mb = Mailbox::new(me); } - reg.insert_binding(name, pid); + mb.channels.insert( + TypeId::of::(), + Channel { sender: Box::new(tx), msg_type: type_name::() }, + ); + reg.by_name.insert(key, me.index()); 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). +/// The single actor currently registered under `name`, or `None` if unbound or +/// 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 + let idx = *reg.by_name.get(name)?; + match reg.by_index.get(&idx).map(|m| m.pid) { + Some(pid) if live(inner, pid) => Some(pid), + Some(_) => { + reg.prune(idx); + None + } + None => { + reg.by_name.remove(name); + 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`]. +/// Remove the binding for `name`, returning the actor it pointed at if still +/// live. Only the *name* is freed; the actor's mailbox (and any other names for +/// it) remain. A binding to a dead actor is reported as `None`. 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 + let idx = reg.by_name.remove(name)?; + match reg.by_index.get(&idx).map(|m| m.pid) { + Some(pid) if live(inner, pid) => Some(pid), + _ => None, } }) } + +/// Resolve `name` to its actor's `Sender` and deliver `msg`. The whole point +/// of the rework: a name you can *send* to. +/// +/// Errors (message returned in every case): [`SendError::Unresolved`] if no +/// live actor holds the name, [`SendError::NoChannel`] if that actor has no +/// channel for `M`, [`SendError::Closed`] if its `M` channel's receiver is +/// gone. Panics if called outside `Runtime::run()`. +pub fn send(name: Name, msg: M) -> Result<(), SendError> { + let key = name.as_str(); + with_runtime(|inner| { + // Resolve + clone the sender under the Leaf lock, then drop the lock + // before sending (a send can unpark a receiver). + let tx = { + let mut reg = inner.registry.lock(); + let idx = match reg.by_name.get(key) { + Some(&i) => i, + None => return Err(SendError::Unresolved(msg)), + }; + let pid = match reg.by_index.get(&idx).map(|m| m.pid) { + Some(pid) => pid, + None => { + reg.by_name.remove(key); + return Err(SendError::Unresolved(msg)); + } + }; + if !live(inner, pid) { + reg.prune(idx); + return Err(SendError::Unresolved(msg)); + } + match reg.by_index.get(&idx).and_then(Mailbox::clone_sender::) { + Some(tx) => tx, + None => return Err(SendError::NoChannel(msg)), + } + }; + tx.send(msg).map_err(|crate::channel::SendError(m)| SendError::Closed(m)) + }) +} diff --git a/tests/registry.rs b/tests/registry.rs index cd13b7b..532d864 100644 --- a/tests/registry.rs +++ b/tests/registry.rs @@ -1,221 +1,144 @@ -//! Named registry tests. Run under the scheduler: registration requires a -//! live runtime and live actors. +//! Mailbox-registry tests (RFC 014). Run under the scheduler: registration +//! captures a live actor's channel, resolution checks liveness. +//! +//! Workers register their *own* inbox (`register` claims the current actor); +//! the root closure resolves and sends by name. A `ready` handshake closes the +//! register-then-send race without busy-waiting on `whereis`. -use smarm::{channel, name_of, register, run, spawn, unregister, whereis, RegisterError}; +use smarm::{channel, register, run, send, spawn, unregister, whereis, Name, RegisterError, SendError}; + +const SVC: Name = Name::new("svc"); #[test] -fn register_whereis_roundtrip() { +fn register_then_send_by_name_delivers() { run(|| { - let (tx, rx) = channel::<()>(); + let (ready_tx, ready_rx) = channel::<()>(); + let (tx, rx) = channel::(); let h = spawn(move || { - rx.recv().unwrap(); + register(SVC, tx).unwrap(); + ready_tx.send(()).unwrap(); + assert_eq!(rx.recv().unwrap(), 42); }); - 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(); + ready_rx.recv().unwrap(); // worker has registered + assert_eq!(whereis("svc"), Some(h.pid())); + send(SVC, 42).unwrap(); h.join().unwrap(); }); } #[test] -fn register_is_idempotent_for_same_binding() { +fn one_actor_many_typed_channels_route_by_type() { + // Same name, two message types: capability separation falls out of the + // type parameter — `Name` and `Name<&str>` hit different channels of + // the one actor (RFC 014 §4.7). run(|| { - let (tx, rx) = channel::<()>(); + let (ready_tx, ready_rx) = channel::<()>(); + let (cmd_tx, cmd_rx) = channel::(); + let (adm_tx, adm_rx) = channel::<&'static str>(); let h = spawn(move || { - rx.recv().unwrap(); + register(Name::::new("port"), cmd_tx).unwrap(); + register(Name::<&'static str>::new("port"), adm_tx).unwrap(); + ready_tx.send(()).unwrap(); + assert_eq!(cmd_rx.recv().unwrap(), 7); + assert_eq!(adm_rx.recv().unwrap(), "halt"); }); - register("svc", h.pid()).unwrap(); - // Same name, same pid: a no-op Ok, not NameTaken. - register("svc", h.pid()).unwrap(); - tx.send(()).unwrap(); + ready_rx.recv().unwrap(); + send(Name::::new("port"), 7u64).unwrap(); + send(Name::<&'static str>::new("port"), "halt").unwrap(); h.join().unwrap(); }); } #[test] -fn duplicate_name_on_live_holder_is_rejected() { +fn name_held_by_live_actor_is_taken() { run(|| { - let (tx, rx) = channel::<()>(); - let (tx2, rx2) = channel::<()>(); + let (ready_tx, ready_rx) = channel::<()>(); + let (tx_a, rx_a) = channel::(); let a = spawn(move || { - rx.recv().unwrap(); + register(SVC, tx_a).unwrap(); + ready_tx.send(()).unwrap(); + assert_eq!(rx_a.recv().unwrap(), 0); // wait to be released }); - 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(); + ready_rx.recv().unwrap(); + // Root tries to claim a live actor's name for itself -> NameTaken. + let (tx_b, _rx_b) = channel::(); + assert_eq!(register(SVC, tx_b), Err(RegisterError::NameTaken { holder: a.pid() })); + send(SVC, 0).unwrap(); // release a (delivers to the holder, a) a.join().unwrap(); + }); +} + +#[test] +fn dead_holder_is_pruned_and_name_taken_over() { + // a registers, signals, then dies. Its slot index is typically reused by b; + // b's registration must prune the stale binding and take the name over. + run(|| { + let (rt1, rr1) = channel::<()>(); + let (tx1, _rx1) = channel::(); + let a = spawn(move || { + register(SVC, tx1).unwrap(); + rt1.send(()).unwrap(); // then return -> die, _rx1 dropped + }); + rr1.recv().unwrap(); + let a_pid = a.pid(); + a.join().unwrap(); // a is dead; "svc" now points at a stale pid + + let (rt2, rr2) = channel::<()>(); + let (tx2, rx2) = channel::(); + let b = spawn(move || { + register(SVC, tx2).unwrap(); // takes over the freed name + rt2.send(()).unwrap(); + assert_eq!(rx2.recv().unwrap(), 9); + }); + rr2.recv().unwrap(); + assert_ne!(b.pid(), a_pid); // distinct incarnation even if slot reused + assert_eq!(whereis("svc"), Some(b.pid())); + send(SVC, 9).unwrap(); b.join().unwrap(); }); } #[test] -fn one_name_per_pid() { +fn send_errors_unresolved_and_no_channel() { run(|| { - let (tx, rx) = channel::<()>(); + // No actor at all. + assert!(matches!(send(Name::::new("ghost"), 1u64), Err(SendError::Unresolved(_)))); + + let (ready_tx, ready_rx) = channel::<()>(); + let (tx, rx) = channel::(); let h = spawn(move || { - rx.recv().unwrap(); + register(Name::::new("svc2"), tx).unwrap(); + ready_tx.send(()).unwrap(); + assert_eq!(rx.recv().unwrap(), 0); }); - register("first", h.pid()).unwrap(); - assert_eq!( - register("second", h.pid()), - Err(RegisterError::PidAlreadyRegistered { name: "first".into() }) - ); - tx.send(()).unwrap(); + ready_rx.recv().unwrap(); + // Right actor, wrong message type: it has a u64 channel, not a String. + let e = send(Name::::new("svc2"), "x".to_string()); + assert!(matches!(e, Err(SendError::NoChannel(_)))); + assert_eq!(e.unwrap_err().into_inner(), "x"); // message handed back + send(Name::::new("svc2"), 0u64).unwrap(); h.join().unwrap(); }); } #[test] -fn registering_a_dead_pid_is_noproc() { +fn unregister_frees_the_name_only() { 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 (ready_tx, ready_rx) = channel::<()>(); + let (done_tx, done_rx) = channel::<()>(); + let (tx, _rx) = channel::(); // _rx moves into the actor, kept open let h = spawn(move || { - rx.recv().unwrap(); + register(SVC, tx).unwrap(); + let _keep_open = _rx; + ready_tx.send(()).unwrap(); + done_rx.recv().unwrap(); // released over a separate channel }); - register("svc", h.pid()).unwrap(); + ready_rx.recv().unwrap(); + assert_eq!(whereis("svc"), Some(h.pid())); 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(); + assert!(matches!(send(SVC, 1u64), Err(SendError::Unresolved(_)))); + done_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); -}