mailbox: Phase 4a — direct identity-bound send to Pid<A> (RFC 014 §4.2)
Make a Pid<A> messageable: install + send_to, the direct addressing mode that dies with the actor and never redirects (the counterpart to the re-resolving Name). - install::<A: Addressable>(tx: Sender<A::Msg>) -> Pid<A>: opt-in, lazy, nameless publish (RFC 014 §5). Files the actor's inbox into the by_index mailbox table under TypeId::of::<A::Msg>() and re-types self's identity as Pid<A>. Infallible: no name to collide on, self is always live in run(). - register and install now share publish_channel (the insert-or-extend-mailbox step factored out); register additionally binds the name. Byte-identical storage, so name- and pid-addressing populate the same table. - send_to::<A>(Pid<A>, A::Msg): resolve by raw pid, deliver with NO redirect. The stored mailbox must be this exact incarnation (generation included) and live, else SendError::Dead — even when the slot now holds a different live actor, which is left untouched. Resolution clones the Sender under the Leaf lock, releases, then sends (Leaf -> Channel), as name send / pg / finalize. - send_to_pid is the shared raw-pid core; send_dyn (§4.6) lands on it next. - SendError gains Dead(M), the pid-path counterpart to name-path Unresolved; into_inner / Display / Debug updated. Name send never yields Dead, pid send never yields Unresolved — disjoint by construction, documented on the enum. tests/registry.rs: install_then_send_to_pid_delivers; and the load-bearing send_to_does_not_redirect_after_takeover (slot reused by a new incarnation, the stale Pid<A> send returns Dead and the new occupant gets only its own message). Full suite + order-checker green, warning-free across all targets.
This commit is contained in:
+1
-1
@@ -56,7 +56,7 @@ pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId};
|
|||||||
pub use mutex::{LockTimeout, Mutex, MutexGuard};
|
pub use mutex::{LockTimeout, Mutex, MutexGuard};
|
||||||
pub use pid::{Addressable, Erased, Name, Pid, RawPid};
|
pub use pid::{Addressable, Erased, Name, Pid, RawPid};
|
||||||
pub use pg::{join, leave, members, pick, Incarnation, Member, NodeId};
|
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 runtime::{init, Config, Runtime};
|
||||||
pub use scheduler::{
|
pub use scheduler::{
|
||||||
block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_under, wait_readable,
|
block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_under, wait_readable,
|
||||||
|
|||||||
+100
-8
@@ -52,7 +52,7 @@
|
|||||||
//! **Leaf -> Channel**, as `pg`/`finalize`.
|
//! **Leaf -> Channel**, as `pg`/`finalize`.
|
||||||
|
|
||||||
use crate::channel::Sender;
|
use crate::channel::Sender;
|
||||||
use crate::pid::{Name, Pid};
|
use crate::pid::{Addressable, Name, Pid};
|
||||||
use crate::scheduler::{self_pid, with_runtime};
|
use crate::scheduler::{self_pid, with_runtime};
|
||||||
use std::any::{type_name, Any, TypeId};
|
use std::any::{type_name, Any, TypeId};
|
||||||
use std::collections::HashMap;
|
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
|
/// `Debug`/`Display` are hand-written so neither demands `M: Debug` — the
|
||||||
/// payload is returned, not printed.
|
/// payload is returned, not printed.
|
||||||
pub enum SendError<M> {
|
pub enum SendError<M> {
|
||||||
/// 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),
|
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<A>` 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.
|
/// The actor is live but exposes no channel for this message type.
|
||||||
NoChannel(M),
|
NoChannel(M),
|
||||||
/// The actor's channel for this type is closed (its receiver is gone).
|
/// The actor's channel for this type is closed (its receiver is gone).
|
||||||
@@ -98,13 +104,17 @@ impl<M> SendError<M> {
|
|||||||
/// Recover the undelivered message.
|
/// Recover the undelivered message.
|
||||||
pub fn into_inner(self) -> M {
|
pub fn into_inner(self) -> M {
|
||||||
match self {
|
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 {
|
fn variant(&self) -> &'static str {
|
||||||
match self {
|
match self {
|
||||||
SendError::Unresolved(_) => "Unresolved",
|
SendError::Unresolved(_) => "Unresolved",
|
||||||
|
SendError::Dead(_) => "Dead",
|
||||||
SendError::NoChannel(_) => "NoChannel",
|
SendError::NoChannel(_) => "NoChannel",
|
||||||
SendError::Closed(_) => "Closed",
|
SendError::Closed(_) => "Closed",
|
||||||
}
|
}
|
||||||
@@ -121,6 +131,7 @@ impl<M> std::fmt::Display for SendError<M> {
|
|||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
match self {
|
match self {
|
||||||
SendError::Unresolved(_) => write!(f, "no live actor registered under that name"),
|
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::NoChannel(_) => write!(f, "actor has no channel for this message type"),
|
||||||
SendError::Closed(_) => write!(f, "the actor's channel for this type is closed"),
|
SendError::Closed(_) => write!(f, "the actor's channel for this type is closed"),
|
||||||
}
|
}
|
||||||
@@ -220,8 +231,19 @@ pub fn register<M: Send + 'static>(name: Name<M>, tx: Sender<M>) -> Result<(), R
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Insert or extend my mailbox. A leftover mailbox at this slot index
|
// Publish (or extend) my mailbox with this channel, then bind the name.
|
||||||
// from a dead prior incarnation (pid mismatch) is replaced wholesale.
|
publish_channel::<M>(&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<M: Send + 'static>(reg: &mut Registry, me: Pid, tx: Sender<M>) {
|
||||||
let mb = reg.by_index.entry(me.index()).or_insert_with(|| Mailbox::new(me));
|
let mb = reg.by_index.entry(me.index()).or_insert_with(|| Mailbox::new(me));
|
||||||
if mb.pid != me {
|
if mb.pid != me {
|
||||||
*mb = Mailbox::new(me);
|
*mb = Mailbox::new(me);
|
||||||
@@ -230,9 +252,28 @@ pub fn register<M: Send + 'static>(name: Name<M>, tx: Sender<M>) -> Result<(), R
|
|||||||
TypeId::of::<M>(),
|
TypeId::of::<M>(),
|
||||||
Channel { sender: Box::new(tx), msg_type: type_name::<M>() },
|
Channel { sender: Box::new(tx), msg_type: type_name::<M>() },
|
||||||
);
|
);
|
||||||
reg.by_name.insert(key, me.index());
|
}
|
||||||
Ok(())
|
|
||||||
})
|
/// Publish the current actor's `Sender<A::Msg>` into its mailbox **without**
|
||||||
|
/// binding a name, and hand back the typed [`Pid<A>`] 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<A>`] (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<A: Addressable>(tx: Sender<A::Msg>) -> Pid<A> {
|
||||||
|
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::<A::Msg>(&mut reg, me, tx);
|
||||||
|
});
|
||||||
|
// `me` is this actor; re-type the identity as `Pid<A>` (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
|
/// The single actor currently registered under `name`, or `None` if unbound or
|
||||||
@@ -306,3 +347,54 @@ pub fn send<M: Send + 'static>(name: Name<M>, msg: M) -> Result<(), SendError<M>
|
|||||||
tx.send(msg).map_err(|crate::channel::SendError(m)| SendError::Closed(m))
|
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<M: Send + 'static>(
|
||||||
|
inner: &crate::runtime::RuntimeInner,
|
||||||
|
pid: Pid,
|
||||||
|
msg: M,
|
||||||
|
) -> Result<(), SendError<M>> {
|
||||||
|
// 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::<M>) {
|
||||||
|
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<A>`-reachable inbox. Panics if called outside
|
||||||
|
/// `Runtime::run()`.
|
||||||
|
pub fn send_to<A: Addressable>(pid: Pid<A>, msg: A::Msg) -> Result<(), SendError<A::Msg>> {
|
||||||
|
with_runtime(|inner| send_to_pid::<A::Msg>(inner, pid.erase(), msg))
|
||||||
|
}
|
||||||
|
|||||||
+68
-1
@@ -5,7 +5,10 @@
|
|||||||
//! the root closure resolves and sends by name. A `ready` handshake closes the
|
//! the root closure resolves and sends by name. A `ready` handshake closes the
|
||||||
//! register-then-send race without busy-waiting on `whereis`.
|
//! 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<u64> = Name::new("svc");
|
const SVC: Name<u64> = Name::new("svc");
|
||||||
|
|
||||||
@@ -142,3 +145,67 @@ fn unregister_frees_the_name_only() {
|
|||||||
h.join().unwrap();
|
h.join().unwrap();
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- RFC 014 §4.2: direct, identity-bound addressing via `Pid<A>` ----------
|
||||||
|
|
||||||
|
// A stand-in single-message actor. `install::<Worker>` publishes its inbox and
|
||||||
|
// returns a `Pid<Worker>` 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::<Pid<Worker>>();
|
||||||
|
let h = spawn(move || {
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
let me = install::<Worker>(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<Worker>
|
||||||
|
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<A>` 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::<Pid<Worker>>();
|
||||||
|
let (rt1, rr1) = channel::<()>();
|
||||||
|
let a = spawn(move || {
|
||||||
|
let (tx, _rx) = channel::<u64>();
|
||||||
|
let me = install::<Worker>(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::<Pid<Worker>>();
|
||||||
|
let (rt2, rr2) = channel::<()>();
|
||||||
|
let b = spawn(move || {
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
let me = install::<Worker>(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();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user