diff --git a/src/gen_server.rs b/src/gen_server.rs index a68faf7..b079425 100644 --- a/src/gen_server.rs +++ b/src/gen_server.rs @@ -50,9 +50,11 @@ //! see `handle_down` — because monitors are inherently created at runtime.) use crate::channel::{channel, select, Receiver, Selectable, Sender}; -use crate::monitor::{Down, Monitor}; +use crate::monitor::{demonitor, monitor, Down, Monitor}; use crate::pid::Pid; -use crate::scheduler::{spawn, spawn_under}; +use crate::registry::{register_with, resolve_named_sender, RegisterError}; +use crate::scheduler::{request_stop, spawn, spawn_under}; +use std::marker::PhantomData; /// Behaviour for a gen_server: a state value plus call/cast handlers. /// @@ -196,6 +198,27 @@ impl ServerRef { .send(Envelope::Cast(request)) .map_err(|_| CastError::ServerDown) } + + /// Ask the server to terminate and block until it has — the `sys`-style + /// stop (cf. Erlang's `gen_server:stop/1`). Sends a cooperative stop to the + /// server actor and waits for its `Down`, so [`GenServer::terminate`] has + /// run by the time this returns (it fires from the loop's drop guard on the + /// stop unwind). Returns immediately if the server was already gone. + /// + /// This is the explicit teardown for a server pinned alive by a registered + /// [`ServerName`] (whose stored sender means dropping every external + /// [`ServerRef`] no longer closes the inbox). Best-effort like all + /// cooperative cancellation: a server wedged in a tight loop with no + /// observation point cannot be stopped. Panics if called outside + /// `Runtime::run()`. + pub fn shutdown(&self) { + let mon = monitor(self.pid); + request_stop(self.pid); + // The Down lands when the server finalizes; an already-dead target makes + // `monitor` deliver NoProc immediately, so this never blocks forever. + let _ = mon.rx.recv(); + demonitor(&mon); + } } /// The server loop's runtime hook, passed to [`GenServer::init`]. Currently @@ -274,6 +297,22 @@ impl ServerBuilder { /// lifetime is governed by its refs, not by joining, so the backing join /// handle is dropped. pub fn start(self) -> ServerRef { + self.spawn_server() + } + + /// Bind the server to a durable [`ServerName`] as it starts. Switches to the + /// fallible [`NamedServerBuilder::start`] (the name may already be held by a + /// live server). Consumes the builder, carrying its `with_info` / `under` + /// configuration through. + pub fn named(self, name: ServerName) -> NamedServerBuilder { + NamedServerBuilder { builder: self, name: name.as_str() } + } + + /// The shared spawn body behind [`start`](Self::start) and + /// [`NamedServerBuilder::start`]: make the inbox, spawn the loop, return the + /// ref. (The named path additionally publishes the inbox sender under the + /// name before returning.) + fn spawn_server(self) -> ServerRef { let (tx, rx) = channel::>(); let ServerBuilder { state, infos, supervisor } = self; let handle = match supervisor { @@ -284,6 +323,133 @@ impl ServerBuilder { } } +/// A durable name typed by the *server* (RFC 014): a gen_server is multi-message +/// (call / cast over one inbox), so it is addressed by `G` rather than by a +/// single message type. By-name [`call`] / [`cast`] check the request against +/// `G`'s `Call` / `Cast` / `Reply`. Declared as a constant and shared freely: +/// +/// ```ignore +/// const COUNTER: ServerName = ServerName::new("counter"); +/// ``` +/// +/// Under the hood the server's inbox is published into the registry as a +/// `Sender>` keyed by its message `TypeId` — the same channel store +/// every other name uses — so naming needs no separate directory. `Envelope` +/// stays private: only `ServerName` opens that door. +pub struct ServerName { + name: &'static str, + _marker: PhantomData G>, +} + +impl ServerName { + /// Bind a static string as a server name. `const`, so names live as + /// associated constants at call sites. + #[inline] + pub const fn new(name: &'static str) -> Self { + Self { name, _marker: PhantomData } + } + + /// The underlying registry key. + #[inline] + pub const fn as_str(self) -> &'static str { + self.name + } +} + +impl Copy for ServerName {} +impl Clone for ServerName { + fn clone(&self) -> Self { + *self + } +} + +/// A [`ServerBuilder`] that will bind a [`ServerName`] as it starts. Reached via +/// [`ServerBuilder::named`]; its [`start`](Self::start) is fallible because the +/// name may already be held by a live server. `with_info` / `under` stay +/// available so configuration can come before or after `named`. +pub struct NamedServerBuilder { + builder: ServerBuilder, + name: &'static str, +} + +impl NamedServerBuilder { + /// Add an out-of-band info channel (see [`ServerBuilder::with_info`]). + pub fn with_info(mut self, rx: Receiver) -> Self { + self.builder = self.builder.with_info(rx); + self + } + + /// Spawn under an explicit supervisor (see [`ServerBuilder::under`]). + pub fn under(mut self, supervisor: Pid) -> Self { + self.builder = self.builder.under(supervisor); + self + } + + /// Spawn the server and bind its name in one step. Fallible: returns + /// [`RegisterError::NameTaken`] if the name is already held by a different + /// live server. + /// + /// The inbox sender is published under the name **from the parent side, + /// before this returns**, so a by-name `call` / `cast` resolves the instant + /// `start()` returns — no race with the server body. On a name clash the + /// just-spawned server is wound down (its only ref is dropped, closing the + /// inbox), so a failed bind leaks no actor. + pub fn start(self) -> Result, RegisterError> { + let NamedServerBuilder { builder, name } = self; + let server = builder.spawn_server(); + match register_with::>(server.pid, name, server.tx.clone()) { + Ok(()) => Ok(server), + Err(e) => { + drop(server); // inbox closes → loop exits gracefully + Err(e) + } + } + } +} + +/// Resolve a [`ServerName`] to a [`ServerRef`] when you want a handle to hold or +/// pass on rather than resolve per call. Rebuilds the ref from the registry's +/// stored inbox sender; `None` if no live server holds the name. +/// +/// Panics if called outside `Runtime::run()`. +pub fn whereis_server(name: ServerName) -> Option> { + resolve_named_sender::>(name.as_str()).map(|(pid, tx)| ServerRef { tx, pid }) +} + +/// Synchronous request-reply to the server currently registered under `name`, +/// resolving through the registry on every call (so a server restarted under +/// the same name is reached transparently). [`CallError::ServerDown`] if no live +/// server holds the name, or if it dies before replying. +/// +/// Panics if called outside `Runtime::run()`. +pub fn call(name: ServerName, request: G::Call) -> Result { + match whereis_server(name) { + Some(server) => server.call(request), + None => Err(CallError::ServerDown), + } +} + +/// Fire-and-forget to the server registered under `name`, resolving per cast. +/// [`CastError::ServerDown`] if no live server holds the name. +/// +/// Panics if called outside `Runtime::run()`. +pub fn cast(name: ServerName, request: G::Cast) -> Result<(), CastError> { + match whereis_server(name) { + Some(server) => server.cast(request), + None => Err(CastError::ServerDown), + } +} + +/// Terminate the server registered under `name` and block until it is down (see +/// [`ServerRef::shutdown`]). A no-op if no live server holds the name. +/// +/// Panics if called outside `Runtime::run()`. +pub fn shutdown(name: ServerName) { + if let Some(server) = whereis_server(name) { + server.shutdown(); + } +} + /// Spawn `state` as a server under the current actor (via [`spawn`]). Returns a /// [`ServerRef`]. Shorthand for `ServerBuilder::new(state).start()`. pub fn start(state: G) -> ServerRef { diff --git a/src/lib.rs b/src/lib.rs index 49a6ad6..809c70e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -50,18 +50,22 @@ pub use channel::{ channel, select, select_timeout, try_select, try_select_timeout, Receiver, RecvError, RecvTimeoutError, Selectable, Sender, }; -pub use gen_server::{CallError, CallTimeoutError, CastError, GenServer, ServerBuilder, ServerCtx, ServerRef, Watcher}; +pub use gen_server::{ + call, cast, shutdown, whereis_server, CallError, CallTimeoutError, CastError, GenServer, + NamedServerBuilder, ServerBuilder, ServerCtx, ServerName, ServerRef, Watcher, +}; 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::{Addressable, Erased, Name, Pid, RawPid}; -pub use pg::{join, leave, members, pick, Incarnation, Member, NodeId}; +pub use pg::{dispatch, join, leave, members, members_as, pick, pick_as, Incarnation, Member, NodeId}; pub use registry::{ - install, register, send, send_dyn, send_to, unregister, whereis, RegisterError, SendError, + install, lookup_as, register, send, send_dyn, 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, + block_on_io, request_stop, run, self_pid, sleep, spawn, spawn_addr, spawn_under, wait_readable, wait_readable_timeout, wait_writable, wait_writable_timeout, yield_now, FdArm, JoinError, JoinHandle, }; diff --git a/src/pg.rs b/src/pg.rs index 12d3e48..8e8cc40 100644 --- a/src/pg.rs +++ b/src/pg.rs @@ -97,7 +97,8 @@ //! discipline as `demonitor`). use crate::monitor::{demonitor, monitor, Monitor}; -use crate::pid::Pid; +use crate::pid::{assert_type, Addressable, Pid}; +use crate::registry::{send_to, SendError}; use crate::scheduler::with_runtime; use std::collections::HashMap; @@ -408,6 +409,43 @@ pub fn pick(group: &str) -> Option { picked } +/// Typed `pick`: one live member of `group` as a [`Pid`] (RFC 014 §4.4). +/// For a homogeneous pool every member is an `A`, so the picked member comes +/// back typed and dispatch is an ordinary compile-checked [`send_to`] rather +/// than the [`send_dyn`](crate::send_dyn) escape hatch. Re-types via the +/// unchecked [`assert_type`] primitive — a wrong `A` degrades to +/// [`SendError::NoChannel`] on the next send, never a misdelivery. +/// +/// Panics if called outside `Runtime::run()`. +pub fn pick_as(group: &str) -> Option> { + pick(group).map(assert_type::) +} + +/// Typed `members`: every live member of `group` as a [`Pid`], same +/// unchecked re-type as [`pick_as`]. Fan-out stays compile-checked end to end. +/// +/// Panics if called outside `Runtime::run()`. +pub fn members_as(group: &str) -> Vec> { + members(group).into_iter().map(assert_type::).collect() +} + +/// Pick a live member of `group` and send it `msg` in one step, returning the +/// member it reached on success. The pick-a-live-member-and-send combinator +/// over [`pick_as`] + [`send_to`]. +/// +/// Errors hand `msg` back undelivered: [`SendError::NoMember`] if the pool is +/// empty (or all-dead), otherwise whatever the underlying [`send_to`] returns +/// (e.g. the picked member died in the window between pick and send → +/// [`SendError::Dead`]). +/// +/// Panics if called outside `Runtime::run()`. +pub fn dispatch(group: &str, msg: A::Msg) -> Result, SendError> { + match pick_as::(group) { + Some(pid) => send_to::(pid, msg).map(|()| pid), + None => Err(SendError::NoMember(msg)), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/pid.rs b/src/pid.rs index c8acfd5..42bd8bc 100644 --- a/src/pid.rs +++ b/src/pid.rs @@ -119,6 +119,22 @@ impl Pid { } } +/// Re-type an erased pid as `Pid` *unchecked* — the one shared primitive +/// behind `lookup_as` / `pick_as` / `members_as` (RFC 014 §4.4). The registry +/// and pg stores are heterogeneous in `A` (they hold actors of every type at +/// once), so resolving them yields a bare [`Pid`]; recovering the typed address +/// is necessarily an assertion the store cannot make for us. +/// +/// **Not unsound.** Delivery routes on the message's [`TypeId`](std::any::TypeId) +/// (every send path keys the channel store by it), so a wrong `A` here does not +/// mis-deliver: the next [`send_to`](crate::send_to) finds no channel for +/// `A::Msg` on that actor and returns [`SendError::NoChannel`](crate::SendError::NoChannel). +/// A mistyped pid degrades to a clean send error, never a silent misroute. +#[inline] +pub(crate) fn assert_type(pid: Pid) -> Pid { + Pid::from_raw(pid.raw()) +} + impl Copy for Pid {} impl Clone for Pid { fn clone(&self) -> Self { diff --git a/src/registry.rs b/src/registry.rs index 1ecaf47..0025eb6 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -96,8 +96,13 @@ pub enum SendError { 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). + /// The actor's channel for this message type is closed (its receiver is gone). Closed(M), + /// No live member to deliver to — a [`dispatch`](crate::dispatch) over an + /// empty (or all-dead) process group. Group-addressed dispatch only; the + /// name-addressed counterpart is [`SendError::Unresolved`]. The message is + /// handed back undelivered. + NoMember(M), } impl SendError { @@ -107,7 +112,8 @@ impl SendError { SendError::Unresolved(m) | SendError::Dead(m) | SendError::NoChannel(m) - | SendError::Closed(m) => m, + | SendError::Closed(m) + | SendError::NoMember(m) => m, } } @@ -117,6 +123,7 @@ impl SendError { SendError::Dead(_) => "Dead", SendError::NoChannel(_) => "NoChannel", SendError::Closed(_) => "Closed", + SendError::NoMember(_) => "NoMember", } } } @@ -134,6 +141,7 @@ impl std::fmt::Display for SendError { 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"), + SendError::NoMember(_) => write!(f, "no live member in the process group"), } } } @@ -212,8 +220,20 @@ fn live(inner: &crate::runtime::RuntimeInner, pid: Pid) -> bool { /// 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(); + register_with(self_pid(), name.as_str(), tx) +} + +/// Bind `name` to `pid`'s mailbox and publish `tx` under `M`'s [`TypeId`], for +/// an explicit (already-live) actor rather than `self`. The shared core of +/// [`register`] (which passes `self_pid()`) and the parent-side server-name +/// bind in `gen_server` (which names a freshly spawned server before its body +/// has run, so the name resolves the instant `start()` returns). Same collision +/// rules and lock discipline as `register`. +pub(crate) fn register_with( + me: Pid, + key: &'static str, + tx: Sender, +) -> Result<(), RegisterError> { with_runtime(|inner| { let mut reg = inner.registry.lock(); if !live(inner, me) { @@ -231,7 +251,7 @@ pub fn register(name: Name, tx: Sender) -> Result<(), R } } } - // Publish (or extend) my mailbox with this channel, then bind the name. + // Publish (or extend) the mailbox with this channel, then bind the name. publish_channel::(&mut reg, me, tx); reg.by_name.insert(key, me.index()); Ok(()) @@ -276,6 +296,23 @@ pub fn install(tx: Sender) -> Pid { Pid::from_raw(me.raw()) } +/// Publish `tx` into `pid`'s mailbox under `M`'s [`TypeId`], for an explicit +/// (freshly minted, already-live) actor rather than `self`. The parent-side +/// half of [`spawn_addr`](crate::spawn_addr): the spawner makes the inbox and +/// publishes the sender here *before* handing back the `Pid`, so an immediate +/// `send_to` on the returned pid always resolves — the address is live the +/// instant the caller holds it, with no dependence on the body having run yet. +/// +/// Caller guarantees `pid` is the just-installed actor (Queued, this exact +/// incarnation); `publish_channel` replaces any stale leftover at the slot. +pub(crate) fn install_for(pid: Pid, tx: Sender) { + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + debug_assert!(live(inner, pid), "install_for: pid must be a freshly spawned, live actor"); + publish_channel::(&mut reg, pid, tx); + }); +} + /// 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 { @@ -296,6 +333,45 @@ pub fn whereis(name: &str) -> Option { }) } +/// Resolve `name` to a *typed* [`Pid`] — the identity-bound counterpart of +/// [`whereis`] (RFC 014 §4.4). Recovers the compile-checked +/// [`send_to`](crate::send_to) path from a durable name: looks the name up, +/// then re-types the erased pid as `Pid` via the unchecked +/// [`assert_type`](crate::pid::assert_type) primitive. A wrong `A` is not +/// unsound — it degrades to [`SendError::NoChannel`] on the next send (routing +/// is by message `TypeId`), never a misdelivery. `None` if unbound or dead. +/// +/// Panics if called outside `Runtime::run()`. +pub fn lookup_as(name: &str) -> Option> { + whereis(name).map(crate::pid::assert_type::) +} + +/// Resolve `name` to its actor's pid and a cloned `Sender`, under the Leaf +/// lock (clone-under-lock, then release). The crate-internal building block for +/// `gen_server`'s by-name addressing: a named server publishes its inbox as a +/// `Sender>` (via [`register_with`]), and `whereis_server` / `call` +/// / `cast` recover that exact typed sender here to rebuild a `ServerRef`. +/// `None` if unbound, dead (pruned on the way out), or holding no `M` channel. +pub(crate) fn resolve_named_sender(name: &str) -> Option<(Pid, Sender)> { + with_runtime(|inner| { + let mut reg = inner.registry.lock(); + let idx = *reg.by_name.get(name)?; + let pid = match reg.by_index.get(&idx).map(|m| m.pid) { + Some(pid) if live(inner, pid) => pid, + Some(_) => { + reg.prune(idx); + return None; + } + None => { + reg.by_name.remove(name); + return None; + } + }; + let tx = reg.by_index.get(&idx).and_then(Mailbox::clone_sender::)?; + Some((pid, tx)) + }) +} + /// 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`. diff --git a/src/runtime.rs b/src/runtime.rs index 1d50353..acdbf41 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -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, /// 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, 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, 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) { + 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, slot_idx: usize) { Got(Pid), Idle { io_outstanding: u32, wake_fd: Option }, 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, 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, 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. diff --git a/src/scheduler.rs b/src/scheduler.rs index bf0ec30..c1a9131 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -195,6 +195,33 @@ pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHandle { pid, consumed: false } } +/// Spawn a typed, single-message actor and hand back its identity-bound +/// [`Pid`] (RFC 014's typed-path producer). The runtime makes the actor's +/// inbox, hands the body its [`Receiver`], installs the sender, and +/// returns the parent a `Pid`. +/// +/// The inbox is published from the parent side **before** the pid is returned +/// (see [`registry::install_for`](crate::registry::install_for)), so the address +/// is live the instant the caller holds it: an immediate +/// [`send_to`](crate::send_to) always resolves, never racing the body's first +/// instruction. The actor is detached — its lifetime is governed by its own +/// logic (an explicit stop message, or returning), like +/// [`ServerBuilder::start`](crate::ServerBuilder::start) — so the backing join +/// handle is dropped. Spawns under the current actor (via [`spawn`]). +/// +/// Panics if called outside `Runtime::run()`. +pub fn spawn_addr( + body: impl FnOnce(crate::channel::Receiver) + Send + 'static, +) -> Pid { + let (tx, rx) = crate::channel::channel::(); + let handle = spawn(move || body(rx)); + let pid = handle.pid(); + // Publish the sender for `pid` before returning the typed address. `handle` + // drops at end of scope (detached). + crate::registry::install_for::(pid, tx); + crate::pid::assert_type::(pid) +} + use crate::context::init_actor_stack; pub fn self_pid() -> Pid { @@ -290,21 +317,28 @@ pub(crate) fn retire_wait() { /// if `pid` is already gone. pub fn request_stop(pid: Pid) { let pid = pid.erase(); - let _ = try_with_runtime(|inner| { - if let Some(slot) = inner.slot_at(pid) { - { - let cold = slot.cold.lock(); - // Verify under the cold lock: generation can't change while - // we hold it (reclaim takes the same lock). - if slot.generation() == pid.generation() { - if let Some(actor) = cold.actor.as_ref() { - actor.stop.store(true, std::sync::atomic::Ordering::Relaxed); - } + let _ = try_with_runtime(|inner| request_stop_inner(inner, pid)); +} + +/// The core of [`request_stop`], taking the runtime directly so it can be +/// driven from inside the runtime (e.g. the root-exit sweep in +/// `finalize_actor`) without re-borrowing the thread-local. Sets the stop flag +/// under the target's cold lock (generation re-verified there; a mismatch or a +/// slot with no live actor is a no-op) and wakes it. +pub(crate) fn request_stop_inner(inner: &RuntimeInner, pid: Pid) { + if let Some(slot) = inner.slot_at(pid) { + { + let cold = slot.cold.lock(); + // Verify under the cold lock: generation can't change while + // we hold it (reclaim takes the same lock). + if slot.generation() == pid.generation() { + if let Some(actor) = cold.actor.as_ref() { + actor.stop.store(true, std::sync::atomic::Ordering::Relaxed); } } - inner.unpark(pid); } - }); + inner.unpark(pid); + } } // ---------------------------------------------------------------------------