mailbox: typed-path producers, by-name gen_servers, root-exit teardown (RFC 014)
Implements the new addressing surface the examples in examples/ specify: - spawn_addr::<A>(FnOnce(Receiver<A::Msg>)) -> Pid<A>: the typed-path producer. Makes the inbox, spawns the body with its receiver, and publishes the sender from the PARENT side (registry::install_for) before returning the pid, so an immediate send_to always resolves (no race on the body installing itself). Detached, like ServerBuilder::start. - lookup_as / pick_as / members_as: re-type an erased pid as Pid<A> over the heterogeneous registry/pg stores, sharing one helper (pid::assert_type). Unchecked but sound: routing is by message TypeId, so a wrong A degrades to SendError::NoChannel on the next send, never a misdelivery. - dispatch::<A>(group, msg) -> Result<Pid<A>, SendError<A::Msg>>: pick_as + send_to, with the message handed back on failure. New SendError::NoMember variant for the empty/all-dead group case. - gen_server naming: ServerName<G> backed internally by the registry's existing typed-channel store keyed by TypeId::of::<Envelope<G>>() — Envelope stays private, no separate directory. Type-state NamedServerBuilder<G> keeps the current infallible ServerBuilder::start() untouched; its start() is fallible (parent-side register, NameTaken). Free call/cast/whereis_server resolve per use. ServerRef::shutdown (+ free shutdown) is the sys-style synchronous stop. - root-exit teardown: the run's initial actor is recorded as root; when it finalizes it flags root_exited, and the scheduler's idle verdict then stops the parked-forever remainder (a one-shot RootDrain sweep). Deferred to the idle point on purpose: the run queue drains first, so actors with queued work finish rather than unwinding on the stop. This lets a named-server daemon (or any pinned actor) wind the run down instead of hanging on live_actors > 0. request_stop is refactored to a request_stop_inner core so the sweep can drive it from inside the runtime without re-borrowing the thread-local. Lib + tests + examples build warning-free; full suite green.
This commit is contained in:
@@ -509,6 +509,19 @@ pub(crate) struct RuntimeInner {
|
||||
/// Incremented in `spawn` before the enqueue; decremented at the very end
|
||||
/// of `finalize_actor`, after every wakeup finalize produces.
|
||||
pub(crate) live_actors: AtomicU32,
|
||||
/// Packed `(index << 32 | generation)` of the run's root (initial) actor,
|
||||
/// or `u64::MAX` (the ROOT_PID sentinel) before one is set. When this actor
|
||||
/// finalizes it flags `root_exited`; the scheduler's idle verdict then
|
||||
/// stops the remaining (parked-forever) actors. Set once per `run()`, right
|
||||
/// after the initial spawn.
|
||||
pub(crate) root_bits: AtomicU64,
|
||||
/// Set when the root actor finalizes; read by the scheduler's idle verdict
|
||||
/// to trigger the one-shot teardown sweep. Reset per `run()`.
|
||||
pub(crate) root_exited: AtomicBool,
|
||||
/// Guards the teardown sweep to fire at most once per run (a parked-forever
|
||||
/// remainder that survives the sweep falls through to the normal idle wait
|
||||
/// rather than busy-spinning). Reset per `run()`.
|
||||
pub(crate) root_swept: AtomicBool,
|
||||
/// Timer heap. Independent lock: never nested with any other.
|
||||
pub(crate) timers: Mutex<Timers>,
|
||||
/// IO subsystem. `None` between runs. Lock order: io before everything.
|
||||
@@ -569,6 +582,9 @@ impl RuntimeInner {
|
||||
slots,
|
||||
free: RawMutex::new(free),
|
||||
live_actors: AtomicU32::new(0),
|
||||
root_bits: AtomicU64::new(u64::MAX),
|
||||
root_exited: AtomicBool::new(false),
|
||||
root_swept: AtomicBool::new(false),
|
||||
timers: Mutex::new(Timers::new()),
|
||||
io: Mutex::new(None),
|
||||
next_monitor_id: AtomicU64::new(0),
|
||||
@@ -597,6 +613,25 @@ impl RuntimeInner {
|
||||
self.slots.get(pid.index() as usize)
|
||||
}
|
||||
|
||||
/// Record `pid` as this run's root actor. Called once per `run()`, right
|
||||
/// after the initial spawn and before any scheduler thread starts, so no
|
||||
/// finalize can observe the count before the root is set.
|
||||
#[inline]
|
||||
pub(crate) fn set_root(&self, pid: Pid) {
|
||||
self.root_bits.store(Self::pack(pid), Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Is `pid` (index + generation) this run's root actor?
|
||||
#[inline]
|
||||
pub(crate) fn is_root(&self, pid: Pid) -> bool {
|
||||
self.root_bits.load(Ordering::Relaxed) == Self::pack(pid)
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn pack(pid: Pid) -> u64 {
|
||||
((pid.index() as u64) << 32) | pid.generation() as u64
|
||||
}
|
||||
|
||||
/// Push to the run queue. Callers must have just transitioned the pid
|
||||
/// into `Queued` (spawn's publish, the unpark protocol, or the
|
||||
/// scheduler's yield/notified-park return paths).
|
||||
@@ -818,6 +853,11 @@ impl Runtime {
|
||||
// requires a running runtime in the thread-local).
|
||||
RUNTIME.with(|r| *r.borrow_mut() = Some(self.inner.clone()));
|
||||
let initial_handle = crate::scheduler::spawn(f);
|
||||
// The initial actor is the run's root: when it exits, remaining actors
|
||||
// are stopped so the run winds down (see finalize_actor / schedule_loop).
|
||||
self.inner.root_exited.store(false, Ordering::Relaxed);
|
||||
self.inner.root_swept.store(false, Ordering::Relaxed);
|
||||
self.inner.set_root(initial_handle.pid());
|
||||
|
||||
// Launch N-1 extra scheduler threads. The calling thread is thread 0.
|
||||
let mut os_threads = Vec::new();
|
||||
@@ -1117,6 +1157,16 @@ fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
|
||||
// Reclaim if no outstanding handles (re-verified inside).
|
||||
reclaim_slot(inner, pid);
|
||||
|
||||
// Root-exit teardown is DEFERRED to the scheduler's idle verdict, not done
|
||||
// here: stopping eagerly would cut off actors that still have queued work
|
||||
// (they'd unwind on the stop before draining their mailbox). Flagging it
|
||||
// instead lets the run queue drain naturally first; only the parked-forever
|
||||
// remainder (e.g. a server pinned alive by a registered name) is then
|
||||
// stopped, once nothing runnable is left. See `schedule_loop`.
|
||||
if inner.is_root(pid) {
|
||||
inner.root_exited.store(true, Ordering::Release);
|
||||
}
|
||||
|
||||
// The decrement is LAST: every wakeup this finalize produced (joiners,
|
||||
// monitor/trap sends, stop cascades) is enqueued before `live_actors`
|
||||
// can be observed at its decremented value. See the termination note in
|
||||
@@ -1125,6 +1175,19 @@ fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
|
||||
debug_assert!(prev >= 1, "live_actors underflow — double finalize");
|
||||
}
|
||||
|
||||
/// Cooperatively stop every live actor — the root-exit teardown sweep, run from
|
||||
/// `schedule_loop` once the run queue is empty after the root has exited. Each
|
||||
/// [`request_stop_inner`](crate::scheduler::request_stop_inner) re-verifies the
|
||||
/// target under its cold lock, so the racy per-slot generation read is safe: a
|
||||
/// vacant, dead, or reused slot no-ops. The swept actors unpark, unwind at their
|
||||
/// next observation point, and finalize, dropping `live_actors` to zero.
|
||||
fn stop_live_actors(inner: &Arc<RuntimeInner>) {
|
||||
for idx in 0..inner.slots.len() as u32 {
|
||||
let pid = Pid::new(idx, inner.slots[idx as usize].generation());
|
||||
crate::scheduler::request_stop_inner(inner, pid);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// schedule_loop — runs on each scheduler OS thread
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -1220,6 +1283,9 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
Got(Pid),
|
||||
Idle { io_outstanding: u32, wake_fd: Option<std::os::fd::RawFd> },
|
||||
AllDone,
|
||||
/// Root has exited and nothing is runnable: stop the parked-forever
|
||||
/// remainder, then re-pop. Fires at most once per run.
|
||||
RootDrain,
|
||||
}
|
||||
|
||||
// 2a. RFC 005: drain this thread's wake slot before touching the
|
||||
@@ -1268,6 +1334,15 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
let live = inner.live_actors.load(Ordering::Acquire);
|
||||
if live == 0 && io_out == 0 {
|
||||
Pop::AllDone
|
||||
} else if inner.root_exited.load(Ordering::Acquire)
|
||||
&& !inner.root_swept.swap(true, Ordering::AcqRel)
|
||||
{
|
||||
// Root gone and nothing runnable — the live remainder
|
||||
// are parked-forever daemons (Queued actors with pending
|
||||
// work drained before the queue emptied). Stop them so
|
||||
// the run can end. One-shot: a survivor falls through to
|
||||
// the idle wait below on the next pass.
|
||||
Pop::RootDrain
|
||||
} else {
|
||||
Pop::Idle { io_outstanding: io_out, wake_fd: io_fd }
|
||||
}
|
||||
@@ -1294,6 +1369,13 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||||
}
|
||||
return;
|
||||
}
|
||||
Pop::RootDrain => {
|
||||
// Root has exited and nothing is runnable: stop the
|
||||
// parked-forever remainder, then loop back to re-pop the
|
||||
// now-runnable (stopping) actors.
|
||||
stop_live_actors(inner);
|
||||
continue;
|
||||
}
|
||||
Pop::Idle { io_outstanding, wake_fd } => {
|
||||
// Something is still in flight. Sleep on the appropriate
|
||||
// source to avoid hammering the queue mutex; retry on wake.
|
||||
|
||||
Reference in New Issue
Block a user