diff --git a/src/lib.rs b/src/lib.rs index 6231b25..f723bf7 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::{Addressable, Erased, Name, Pid, RawPid}; pub use pg::{join, leave, members, pick, Incarnation, Member, NodeId}; -pub use registry::{register, send, unregister, whereis, RegisterError, SendError}; +pub use registry::{install, register, send, send_to, 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/registry.rs b/src/registry.rs index 6c9a133..f26d716 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -52,7 +52,7 @@ //! **Leaf -> Channel**, as `pg`/`finalize`. use crate::channel::Sender; -use crate::pid::{Name, Pid}; +use crate::pid::{Addressable, Name, Pid}; use crate::scheduler::{self_pid, with_runtime}; use std::any::{type_name, Any, TypeId}; use std::collections::HashMap; @@ -86,8 +86,14 @@ impl std::error::Error for RegisterError {} /// `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. + /// No live actor is currently registered under this name. Name-addressed + /// [`send`] only; the pid-addressed counterpart is [`SendError::Dead`]. Unresolved(M), + /// The pid-addressed actor is no longer the live incarnation this pid names + /// — it has died, even if its slot now holds a *different* actor (a direct + /// `Pid` send never redirects; contrast name-addressed [`send`]). Pid + /// paths ([`send_to`] / [`send_dyn`]) only. + Dead(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). @@ -98,13 +104,17 @@ impl SendError { /// Recover the undelivered message. pub fn into_inner(self) -> M { match self { - SendError::Unresolved(m) | SendError::NoChannel(m) | SendError::Closed(m) => m, + SendError::Unresolved(m) + | SendError::Dead(m) + | SendError::NoChannel(m) + | SendError::Closed(m) => m, } } fn variant(&self) -> &'static str { match self { SendError::Unresolved(_) => "Unresolved", + SendError::Dead(_) => "Dead", SendError::NoChannel(_) => "NoChannel", SendError::Closed(_) => "Closed", } @@ -121,6 +131,7 @@ 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::Dead(_) => write!(f, "the addressed actor is no longer the live incarnation"), 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"), } @@ -220,21 +231,51 @@ pub fn register(name: Name, tx: Sender) -> Result<(), R } } } - // 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); - } - mb.channels.insert( - TypeId::of::(), - Channel { sender: Box::new(tx), msg_type: type_name::() }, - ); + // Publish (or extend) my mailbox with this channel, then bind the name. + publish_channel::(&mut reg, me, tx); reg.by_name.insert(key, me.index()); Ok(()) }) } +/// Insert or extend the current actor's mailbox with one typed channel, filed +/// under its message [`TypeId`]. Shared by [`register`] (which then binds a +/// name) and [`install`] (which does not). A leftover mailbox at this slot +/// index from a dead prior incarnation (pid mismatch) is replaced wholesale. +/// Caller holds the registry lock and has established that `me` is live. +fn publish_channel(reg: &mut Registry, me: Pid, tx: Sender) { + let mb = reg.by_index.entry(me.index()).or_insert_with(|| Mailbox::new(me)); + if mb.pid != me { + *mb = Mailbox::new(me); + } + mb.channels.insert( + TypeId::of::(), + Channel { sender: Box::new(tx), msg_type: type_name::() }, + ); +} + +/// Publish the current actor's `Sender` into its mailbox **without** +/// binding a name, and hand back the typed [`Pid`] that addresses this +/// actor directly. This is the opt-in, lazy install of RFC 014 §5: an actor +/// that wants to be reachable by a direct, identity-bound [`Pid`] (rather +/// than only via a re-resolving [`Name`]) calls this once with its inbox +/// sender, then hands the returned pid out. +/// +/// Unlike [`register`] there is no name to collide on, and `self` is always a +/// live actor inside `run()`, so this is infallible. Panics if called outside +/// `Runtime::run()`. +pub fn install(tx: Sender) -> Pid { + let me = self_pid(); + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + debug_assert!(live(inner, me), "self_pid() is a live actor inside run()"); + publish_channel::(&mut reg, me, tx); + }); + // `me` is this actor; re-type the identity as `Pid` (the channel for + // `A::Msg` was just published, so the typed address is now messageable). + Pid::from_raw(me.raw()) +} + /// 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 { @@ -306,3 +347,54 @@ pub fn send(name: Name, msg: M) -> Result<(), SendError tx.send(msg).map_err(|crate::channel::SendError(m)| SendError::Closed(m)) }) } + +/// Resolve a *raw* pid to its mailbox and deliver `msg` on the channel for `M`, +/// with **no redirect**. The stored mailbox must be this exact incarnation +/// (generation included) and still live; otherwise the actor this pid named is +/// gone and the result is [`SendError::Dead`] — even when the slot now holds a +/// different, live actor (which we leave untouched). Shared by [`send_to`] +/// (typed, `M = A::Msg`, channel guaranteed on an installed actor) and +/// [`send_dyn`] (explicit `M`, where `NoChannel` is a real outcome). +fn send_to_pid( + inner: &crate::runtime::RuntimeInner, + pid: Pid, + msg: M, +) -> Result<(), SendError> { + // Resolve + clone the sender under the Leaf lock, then drop the lock before + // sending (a send can unpark a receiver) — Leaf -> Channel, as name `send`. + let tx = { + let mut reg = inner.registry.lock(); + match reg.by_index.get(&pid.index()).map(|m| m.pid) { + // Exact incarnation, still alive: its `M` channel, or NoChannel. + Some(stored) if stored == pid && live(inner, pid) => { + match reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::) { + Some(tx) => tx, + None => return Err(SendError::NoChannel(msg)), + } + } + // Our incarnation's mailbox, but the actor has died: prune + Dead. + Some(stored) if stored == pid => { + reg.prune(pid.index()); + return Err(SendError::Dead(msg)); + } + // A different incarnation (or nothing) occupies the slot: the actor + // this pid named is gone. Do not disturb any newer occupant. + _ => return Err(SendError::Dead(msg)), + } + }; + tx.send(msg).map_err(|crate::channel::SendError(m)| SendError::Closed(m)) +} + +/// Deliver `msg` to the exact actor named by `pid` — RFC 014 §4.2's direct, +/// identity-bound addressing mode. Unlike name-addressed [`send`] there is **no +/// redirect**: if that incarnation has died the message comes back as +/// [`SendError::Dead`], even if its slot now holds a different actor. +/// +/// The message type is the actor's `A::Msg`, so on a live actor that has +/// installed its inbox (via [`install`] or [`register`]) the channel is always +/// present; [`SendError::NoChannel`] therefore means the actor is live but +/// never published a `Pid`-reachable inbox. Panics if called outside +/// `Runtime::run()`. +pub fn send_to(pid: Pid, msg: A::Msg) -> Result<(), SendError> { + with_runtime(|inner| send_to_pid::(inner, pid.erase(), msg)) +} diff --git a/tests/registry.rs b/tests/registry.rs index 532d864..41e7308 100644 --- a/tests/registry.rs +++ b/tests/registry.rs @@ -5,7 +5,10 @@ //! 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, register, run, send, spawn, unregister, whereis, Name, RegisterError, SendError}; +use smarm::{ + channel, install, register, run, send, send_to, spawn, unregister, whereis, Addressable, Name, + Pid, RegisterError, SendError, +}; const SVC: Name = Name::new("svc"); @@ -142,3 +145,67 @@ fn unregister_frees_the_name_only() { h.join().unwrap(); }); } + +// --- RFC 014 §4.2: direct, identity-bound addressing via `Pid` ---------- + +// A stand-in single-message actor. `install::` publishes its inbox and +// returns a `Pid` that delivers `u64` to exactly that incarnation. +struct Worker; +impl Addressable for Worker { + type Msg = u64; +} + +#[test] +fn install_then_send_to_pid_delivers() { + run(|| { + let (addr_tx, addr_rx) = channel::>(); + let h = spawn(move || { + let (tx, rx) = channel::(); + let me = install::(tx); // nameless publish, typed pid back + addr_tx.send(me).unwrap(); + assert_eq!(rx.recv().unwrap(), 42); + }); + let addr = addr_rx.recv().unwrap(); // worker installed, handed its Pid + assert_eq!(addr.erase(), h.pid()); // same identity, just re-typed + send_to(addr, 42u64).unwrap(); + h.join().unwrap(); + }); +} + +#[test] +fn send_to_does_not_redirect_after_takeover() { + // The load-bearing §4.2 property: a `Pid` is identity-bound. When the + // actor dies and a new incarnation reuses the slot, sending to the *old* + // address fails `Dead` — it must never silently reach the new occupant + // (that redirect is the re-resolving `Name`'s job, not a pid's). + run(|| { + let (addr_a_tx, addr_a_rx) = channel::>(); + let (rt1, rr1) = channel::<()>(); + let a = spawn(move || { + let (tx, _rx) = channel::(); + let me = install::(tx); + addr_a_tx.send(me).unwrap(); + rt1.send(()).unwrap(); // then return -> die + }); + let a_addr = addr_a_rx.recv().unwrap(); + rr1.recv().unwrap(); + a.join().unwrap(); // a dead; a_addr names a dead incarnation + + let (addr_b_tx, addr_b_rx) = channel::>(); + let (rt2, rr2) = channel::<()>(); + let b = spawn(move || { + let (tx, rx) = channel::(); + let me = install::(tx); // reuses a's slot index, new generation + addr_b_tx.send(me).unwrap(); + rt2.send(()).unwrap(); + // Only b's own message ever arrives; the stale send never redirects. + assert_eq!(rx.recv().unwrap(), 7); + }); + let b_addr = addr_b_rx.recv().unwrap(); + rr2.recv().unwrap(); + assert_ne!(b_addr.erase(), a_addr.erase()); // distinct incarnation + assert!(matches!(send_to(a_addr, 99u64), Err(SendError::Dead(_)))); // no redirect + send_to(b_addr, 7u64).unwrap(); // b's real message + b.join().unwrap(); + }); +}