From 8625ae4c354afec16593c3189afca8b237418685 Mon Sep 17 00:00:00 2001 From: Mark Kalsbeek Date: Fri, 24 Jul 2026 08:40:26 +0200 Subject: [PATCH] docs(scheduler): user-facing rewrite of the actor/spawn/run entry point Lead with what an actor is and how to start one with run()/spawn(), following gen_server.rs's example-first style. Add a compiling module doctest. Drop RFC references and em-dashes; keep internal mechanics (preemption gating, thread-local borrow rules) as plain contributor comments rather than public-facing doc prose. --- src/scheduler.rs | 530 +++++++++++++++++++++++++++++++---------------- 1 file changed, 352 insertions(+), 178 deletions(-) diff --git a/src/scheduler.rs b/src/scheduler.rs index 40b8eff..dea289f 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -1,11 +1,69 @@ -//! Scheduler public API — thin façade over the multi-scheduler runtime. +//! Start actors and control them: `run`, `spawn`, `sleep`, timers, and IO waits. //! -//! All heavy lifting lives in `runtime.rs`. This module exposes the same -//! surface that the rest of the codebase (channel, mutex, io, timer, actor) -//! calls into, plus the public API re-exported from `lib.rs`. +//! ## What is an actor? //! -//! The single-threaded `run()` entry point is kept as a convenience wrapper -//! around `runtime::init(Config::exact(1)).run(f)`. +//! An actor in smarm is a *green thread*: a lightweight, cooperatively +//! scheduled unit of execution with its own stack, running alongside +//! thousands of others on a handful of OS threads. You give it a closure; +//! smarm gives it a [`Pid`] and a [`JoinHandle`]. Actors talk to each other by +//! sending messages over [`channel`](mod@crate::channel)s, not by sharing memory, +//! so most of the concurrency bugs that come from shared mutable state simply +//! don't arise. +//! +//! Everything in smarm (channels, [`Mutex`](crate::Mutex), timers, IO waits, +//! `gen_server`) needs to run on top of a scheduler. [`run`] starts one and +//! blocks the calling OS thread until the actor you give it finishes; from +//! inside that actor (or any actor it spawns), call [`spawn`] to start more. +//! +//! ``` +//! use smarm::{run, spawn}; +//! +//! run(|| { +//! let handle = spawn(|| { +//! println!("hello from another actor"); +//! }); +//! handle.join().unwrap(); +//! }); +//! ``` +//! +//! ## Waiting for an actor: `JoinHandle` +//! +//! [`spawn`] returns a [`JoinHandle`], your only handle on the actor's +//! outcome. Call [`JoinHandle::join`] to block the calling actor until the +//! spawned one finishes: +//! +//! - if it returned normally, `join` returns `Ok(())`; +//! - if it panicked, `join` returns `Err(`[`JoinError`]`)` carrying the panic +//! payload, so a crash in one actor never silently vanishes and never +//! crashes the process; the caller decides what to do with it (log it, +//! propagate it, ignore it). +//! +//! If you don't need the result, drop the `JoinHandle` (or never bind it): +//! the actor keeps running independently. To watch an actor without +//! blocking, or to react to failures across a whole tree of actors, see +//! [`monitor`](mod@crate::monitor), [`link`](mod@crate::link), and +//! [`supervisor`](crate::supervisor) instead. +//! +//! ## Sleeping, timeouts, and IO +//! +//! [`sleep`] parks only the calling actor, not the OS thread underneath it, +//! so thousands of sleeping actors cost nothing but the timer entry. +//! [`send_after`] and [`send_after_named`] schedule a message to be delivered +//! later (Erlang-style `send_after`), and [`cancel_timer`] can call one off +//! before it fires. +//! +//! For blocking or file-descriptor-based IO, see [`block_on_io`], +//! [`wait_readable`], and [`wait_writable`]: they park the calling actor and +//! resume it when the work completes or the fd is ready, again without +//! blocking an OS thread. +//! +//! ## Stopping an actor from the outside +//! +//! [`request_stop`] asks an actor to cooperatively unwind: it's the +//! mechanism `gen_server` shutdown, timeouts, and supervisor restarts are +//! built on. It's best-effort: an actor that never yields, allocates, or +//! blocks (a tight loop with nothing else in it) has no opportunity to +//! notice the request. use crate::actor::current_pid; use crate::channel::Sender; @@ -20,21 +78,14 @@ use std::sync::Arc; // with_runtime / try_with_runtime // --------------------------------------------------------------------------- -/// Borrow the current runtime. Panics if called outside `Runtime::run()`. -/// -/// The whole span runs with preemption disabled. Two reasons, both load- -/// bearing: -/// -/// - The `RUNTIME` thread-local borrow is live across `f`. A preemption- -/// driven context switch inside `f` can resume the actor on a DIFFERENT OS -/// thread; the borrow guard would then increment this thread's RefCell but -/// decrement the other's — a count underflow that leaves that thread's -/// RefCell permanently "mutably borrowed". Holding a thread-local guard -/// across a potential switch point is the one unforgivable sin of green -/// threads; disabling preemption makes "no switch inside `f`" structural -/// instead of an accident of which bodies happen to allocate. -/// - `f` is runtime bookkeeping. Suspending an actor halfway through it (or -/// unwinding via the stop sentinel, which shares the gate) is never wanted. +// Borrow the current runtime. Panics if called outside `Runtime::run()`. +// +// Preemption is disabled for the whole span. `f` holds a thread-local borrow +// of `RUNTIME`; if a preemption-driven context switch moved the actor to a +// different OS thread in the middle of `f`, the borrow guard would be +// released on the wrong thread's copy of the thread-local, corrupting its +// borrow count. `f` is also always runtime bookkeeping that should run to +// completion without the actor being suspended or unwound partway through. pub(crate) fn with_runtime(f: impl FnOnce(&Arc) -> R) -> R { let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); let result = RUNTIME.with(|r| { @@ -49,9 +100,9 @@ pub(crate) fn with_runtime(f: impl FnOnce(&Arc) -> R) -> R { result } -/// Borrow the runtime if present; returns `None` otherwise. -/// Used on cleanup paths (channel Drop during teardown). -/// Same preemption gate as [`with_runtime`]; same reasons. +// Borrow the runtime if present, otherwise `None`. Used on cleanup paths +// (e.g. a channel's Drop impl during teardown) that may run after the +// runtime has already gone away. Same preemption gate as `with_runtime`. pub(crate) fn try_with_runtime(f: impl FnOnce(&Arc) -> R) -> Option { let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); let result = RUNTIME.with(|r| r.borrow().as_ref().map(f)); @@ -63,19 +114,49 @@ pub(crate) fn try_with_runtime(f: impl FnOnce(&Arc) -> R) -> Op // JoinHandle / JoinError // --------------------------------------------------------------------------- +/// The spawned actor panicked. Returned by [`JoinHandle::join`]; `payload` +/// is exactly what the panic carried (the value passed to `panic!`, or +/// whatever a library panicked with), the same payload you'd get from +/// [`std::thread::JoinHandle::join`]. Downcast it if you need to inspect it: +/// +/// ``` +/// use smarm::{run, spawn}; +/// +/// run(|| { +/// let h = spawn(|| panic!("boom")); +/// let err = h.join().unwrap_err(); +/// let msg = err.payload.downcast_ref::<&str>().copied().unwrap_or("?"); +/// assert_eq!(msg, "boom"); +/// }); +/// ``` #[derive(Debug)] pub struct JoinError { pub payload: Box, } +/// A handle to a spawned actor, returned by [`spawn`], [`spawn_under`], and +/// friends. Use [`join`](Self::join) to wait for the actor to finish and +/// collect its outcome, or [`pid`](Self::pid) to get its identity for use +/// with [`request_stop`], [`monitor`](crate::monitor::monitor), or +/// [`link`](crate::link::link). +/// +/// If you never call `join` (or drop the handle instead), the actor is not +/// affected: it keeps running and its resources are still reclaimed when it +/// finishes. `join` is how you find out *what happened*, not a requirement +/// for the actor to make progress or clean up. pub struct JoinHandle { pid: Pid, consumed: bool, } impl JoinHandle { + /// The identity of the actor this handle refers to. pub fn pid(&self) -> Pid { self.pid } + /// Block the calling actor until the spawned actor finishes, then + /// report how it finished: `Ok(())` if it returned normally or stopped + /// cooperatively via [`request_stop`], `Err(`[`JoinError`]`)` if it + /// panicked. pub fn join(mut self) -> Result<(), JoinError> { use crate::actor::Outcome; @@ -177,6 +258,19 @@ impl Drop for JoinHandle { // spawn / spawn_under / self_pid // --------------------------------------------------------------------------- +/// Start a new actor running `f`, and return a [`JoinHandle`] for it. +/// +/// The new actor runs concurrently with its caller and with every other +/// actor in the runtime; smarm schedules it cooperatively across the +/// available OS threads. `spawn` can be called from inside `run`'s closure, +/// or from inside any actor (spawning a child from a child works the same +/// way); it cannot be called before `run` has started or after it returns. +/// +/// The returned [`JoinHandle`] is how you learn how the actor finished. If +/// you don't need that, it's fine to drop it: the actor still runs to +/// completion either way. +/// +/// Panics if called outside `Runtime::run()`. pub fn spawn(f: impl FnOnce() + Send + 'static) -> JoinHandle { let parent = current_pid().unwrap_or_else(|| { // Outside an actor but inside run(): the initial spawn. with_runtime @@ -186,6 +280,11 @@ pub fn spawn(f: impl FnOnce() + Send + 'static) -> JoinHandle { spawn_under(parent, f) } +/// Like [`spawn`], but explicitly attaches the new actor to `supervisor` +/// instead of the calling actor. Ordinary code should reach for [`spawn`]; +/// this exists for supervision trees (see [`supervisor`](crate::supervisor)) +/// and other cases that need to place a child under a specific ancestor +/// rather than its true caller. pub fn spawn_under(supervisor: Pid, f: impl FnOnce() + Send + 'static) -> JoinHandle { let supervisor = supervisor.erase(); // Stack + closure boxing happen before ANY runtime lock is taken: no @@ -208,19 +307,20 @@ 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`. +/// Spawn an actor that other actors can message directly by its [`Pid`], +/// rather than only by holding on to a channel `Sender` you passed it +/// yourself. /// -/// 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 -/// [`GenServerBuilder::start`](crate::GenServerBuilder::start) — so the backing join -/// handle is dropped. Spawns under the current actor (via [`spawn`]). +/// `body` receives the [`Receiver`](crate::channel::Receiver) smarm +/// creates for it; `spawn_addr` publishes the matching `Sender` and hands +/// back the actor's typed address. That address is usable the instant you +/// hold it: an immediate [`send_to`](crate::send_to) on the returned pid +/// always finds the inbox, even if `body` hasn't started running yet. +/// +/// The spawned actor is detached (there is no [`JoinHandle`] to join): its +/// lifetime is up to its own logic, for example running until it receives a +/// stop message, or until the actor decides to return. This mirrors how +/// [`GenServerBuilder::start`](crate::GenServerBuilder::start) works. /// /// Panics if called outside `Runtime::run()`. pub fn spawn_addr( @@ -237,6 +337,11 @@ pub fn spawn_addr( use crate::context::init_actor_stack; +/// The identity of the actor currently running. Use it to hand your own +/// address to another actor (for a reply, a monitor, or a link). +/// +/// Panics if called outside an actor (for example, from the closure passed +/// to [`run`] itself, before any [`spawn`]). pub fn self_pid() -> Pid { match current_pid() { Some(pid) => pid, @@ -248,6 +353,12 @@ pub fn self_pid() -> Pid { // yield_now / park_current / unpark // --------------------------------------------------------------------------- +/// Voluntarily give up the CPU so another runnable actor gets a turn, then +/// resume as soon as the scheduler gets back around to you. Use this in a +/// long-running, allocation-free loop that you want to stay cooperative with +/// the rest of the runtime (see also the [`check!`](crate::check) macro, +/// which does the same thing conditionally, only when your timeslice has +/// actually run out). pub fn yield_now() { runtime::set_yield_intent(YieldIntent::Yield); unsafe { crate::context::switch_to_scheduler() }; @@ -255,48 +366,44 @@ pub fn yield_now() { crate::preempt::check_cancelled(); } +// Suspend the current actor until something wakes it (a message arrives, a +// timer fires, a lock is granted, and so on). This is the low-level parking +// primitive that every blocking smarm operation (channel recv, sleep, +// Mutex::lock, IO waits, JoinHandle::join) is built on; application code +// should reach for one of those rather than calling this directly. +// +// Checks for a pending cooperative-stop request both before parking (a stop +// requested while merely queued to run would otherwise have no future wake +// to catch it) and after resuming (so a stop that arrived while parked is +// noticed as soon as we wake, and unwinds from here exactly like any other +// blocking call would). pub fn park_current() { - // Entry-side observation point: a stop flagged while we were QUEUED is - // otherwise lost — the stop's wildcard unpark no-ops on a Queued actor - // (the pending run "is" the wake), so if our first action on resume is - // this park, no wake is ever coming and the wake-side check below is - // unreachable. Checking here closes that hole; a flag that lands after - // this check is covered by the existing protocol (the stop's unpark - // finds Running / the prep-to-park window, sets Notified, and the - // park-return re-queues us into the wake-side check). Unwinding from - // here is the same unwind path as the wake side: leftover wait - // registrations are stale-epoch / dead-generation and self-clean at - // their wakers' failed CAS. crate::preempt::check_cancelled(); runtime::set_yield_intent(YieldIntent::Park); unsafe { crate::context::switch_to_scheduler() }; - // Observation point on the wakeup side of every blocking primitive - // (recv/sleep/mutex/io/join). Past the prep-to-park window, so this never - // races a wakeup: a stop unparks us, we resume here, and unwind out of - // whatever blocking call parked us — running Drop along the way. crate::preempt::check_cancelled(); } +// Wake `pid` unconditionally, regardless of what it's currently waiting for. +// Reserved for terminal wakes (`request_stop`); anything waking an actor from +// a specific registered wait (a channel send, a mutex grant, a timer, an IO +// completion) must use `unpark_at` instead, so a stale wakeup can never be +// mistaken for the one the actor is actually waiting on. pub fn unpark(pid: Pid) { - // The whole protocol lives on the slot's packed word (gen + epoch + - // state in one CAS) — see slot_state.rs. No runtime lock unless we - // enqueue. WILDCARD form: reserved for terminal wakes (request_stop); - // every registration-based waker must use `unpark_at`. let _ = try_with_runtime(|inner| inner.unpark(pid)); } -/// Epoch-matched unpark: wake `pid` only if its current wait is still the -/// one this waker registered for. The form every registration-based waker -/// (channel senders, mutex grants, wait-timers, io completions, joiner -/// wakes) must use — see slot_state.rs for the consuming-wake rules. +// Wake `pid` only if its current wait is still the one this waker +// registered for (an "epoch-matched" unpark). Every registration-based +// waker (channel senders, mutex grants, wait-timers, IO completions, joiner +// wakes) must use this rather than the unconditional `unpark`. pub(crate) fn unpark_at(pid: Pid, epoch: u32) { let _ = try_with_runtime(|inner| inner.unpark_at(pid, epoch)); } -/// Open a new wait for the CURRENT actor: bump its park-epoch and return -/// it. Call once per wait, before registering `(pid, epoch)` with any -/// waker. Lock-free (one CAS on the own slot word), so it is legal under -/// any lock, including a Channel-class lock. +// Open a new wait for the current actor and return its wait identity +// ("epoch"). Call once per wait, before registering with any waker. Lock-free, +// so it's legal to call while already holding another internal lock. pub(crate) fn begin_wait() -> u32 { let me = match current_pid() { Some(pid) => pid, @@ -305,48 +412,47 @@ pub(crate) fn begin_wait() -> u32 { with_runtime(|inner| inner.begin_wait(me)) } -/// Close the current actor's wait WITHOUT parking on it. For the no-park -/// exit of multi-registration waits (`select` finding an arm ready at -/// registration time): bumps the epoch so in-flight wakes die at their CAS, -/// eats a notification that already landed, then observes a pending stop — -/// in that order. After this, leftover registrations are stale-epoch and -/// self-clean at their wakers' failed CAS; nothing can fault the actor's -/// next one-shot park. +// Close the current actor's wait without parking on it: the no-park exit +// used when a multi-arm `select` finds an arm already ready at registration +// time. Leftover registrations from the other arms are left to self-clean +// when their wakers try to use them. pub(crate) fn retire_wait() { let me = match current_pid() { Some(pid) => pid, None => panic!("retire_wait() called outside an actor"), }; with_runtime(|inner| inner.retire_wait(me)); - // A request_stop that fired before the clear had its notification - // eaten, but it set the stop flag first — observe it here and unwind. - // One that fires after re-notifies a Running word as usual. crate::preempt::check_cancelled(); } -/// Request cooperative cancellation of `pid`. +/// Ask `pid` to stop cooperatively. /// -/// Sets the actor's stop flag and wakes it so it observes the flag promptly: -/// a parked target is re-queued; a running target is marked notified, so its -/// next park returns immediately to an observation point. The actor realises -/// the stop as a controlled unwind at its next observation point -/// (`check!()`/allocation, or the wakeup side of a blocking park), -/// terminating with `Outcome::Stopped`. +/// This sets a flag on the target actor and wakes it so it notices promptly; +/// the actor itself decides when it's safe to actually unwind, at its next +/// natural checkpoint (a blocking call returning, a `check!()`, or an +/// allocation). Once it does, it terminates as if it panicked, except that +/// [`JoinHandle::join`] reports it as a normal, non-error exit: cooperative +/// stop is a controlled shutdown, not a failure. /// -/// This is best-effort and cooperative: an actor that never reaches an -/// observation point — a tight loop with no `check!()`, no allocation, and no -/// blocking op — cannot be stopped, exactly as it cannot be preempted. A no-op -/// if `pid` is already gone. +/// This is exactly the mechanism `gen_server` shutdown, supervisor restarts, +/// and structured teardown are built from: reach for [`GenServerRef::shutdown`](crate::GenServerRef::shutdown) +/// or a [`supervisor`](crate::supervisor) instead of calling this directly +/// where those apply. +/// +/// Because it's cooperative, an actor stuck in a tight loop with no +/// blocking call, no [`check!`](crate::check), and no allocation cannot be +/// stopped, for the same reason it cannot be preempted. Calling this on an +/// actor that has already finished is a harmless no-op. pub fn request_stop(pid: Pid) { let pid = pid.erase(); 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. +// The core of `request_stop`, taking the runtime directly so it can also be +// driven from inside the runtime itself (the root-exit sweep) without +// re-borrowing the thread-local. Sets the stop flag under the target's lock +// (a generation mismatch, or no live actor there, makes it a no-op) and +// wakes the target. pub(crate) fn request_stop_inner(inner: &RuntimeInner, pid: Pid) { if let Some(slot) = inner.slot_at(pid) { { @@ -367,6 +473,10 @@ pub(crate) fn request_stop_inner(inner: &RuntimeInner, pid: Pid) { // NoPreempt // --------------------------------------------------------------------------- +/// A guard that disables preemption for its lifetime, restoring the +/// previous setting on drop. Internal-use: application code has no need to +/// disable preemption directly. See [`check!`](crate::check) for the +/// user-facing side of preemption. pub struct NoPreempt(bool); impl NoPreempt { @@ -386,6 +496,21 @@ impl Drop for NoPreempt { // sleep / insert_wait_timer // --------------------------------------------------------------------------- +/// Suspend the calling actor for `duration`. Unlike +/// [`std::thread::sleep`], this parks only the actor, not the underlying OS +/// thread, so every other actor (including others sharing the same OS +/// thread) keeps running normally while this one waits. +/// +/// ``` +/// use smarm::{run, sleep}; +/// use std::time::Duration; +/// +/// run(|| { +/// sleep(Duration::from_millis(1)); +/// }); +/// ``` +/// +/// Panics if called outside an actor. pub fn sleep(duration: std::time::Duration) { let me = match current_pid() { Some(pid) => pid, @@ -403,11 +528,11 @@ pub fn sleep(duration: std::time::Duration) { park_current(); } -/// Like [`sleep`], but wall-anchored: under causal profiling (feature -/// `smarm-causal`) the deadline is honoured in wall time instead of chasing -/// injected virtual delay. Identical to [`sleep`] without the feature. For -/// measurement machinery whose durations define wall time (the causal -/// controller's windows, TSC calibration) — workload code wants [`sleep`]. +/// Like [`sleep`], but the deadline is always measured in real (wall-clock) +/// time. Ordinary code should use [`sleep`]; this variant exists for smarm's +/// own profiling and measurement tooling, which can otherwise stretch or +/// compress simulated time. Without that tooling active, `sleep_wall` and +/// `sleep` behave identically. pub fn sleep_wall(duration: std::time::Duration) { let me = match current_pid() { Some(pid) => pid, @@ -425,6 +550,11 @@ pub fn sleep_wall(duration: std::time::Duration) { park_current(); } +// Building block for bounded waits elsewhere in the crate (Mutex::lock_timeout, +// Receiver::recv_timeout, select_timeout): arm a timer that, on expiry, asks +// `target` whether this particular wait is still pending and should be woken +// with a timeout. Not part of the public API; application code wants +// `sleep`, `send_after`, or one of the `*_timeout` methods instead. pub fn insert_wait_timer( deadline: std::time::Instant, pid: Pid, @@ -444,20 +574,22 @@ pub fn insert_wait_timer( } // --------------------------------------------------------------------------- -// send_after / cancel_timer — message-delivery timers (Erlang send_after). -// -// Arm a timer that delivers `msg` to an address after `after`, returning a -// `TimerId`. The destination is resolved *on fire*, not at arm time: a -// `Pid` that has since died yields `SendError::Dead`, a `Name` resolves -// to whoever currently holds it (so a restarted server is reached). Either way -// a failed resolve / closed inbox is dropped, matching `erlang:send_after`. -// `cancel_timer` prevents an as-yet-unfired delivery; it returns whether the -// timer was still armed. +// send_after / cancel_timer: message-delivery timers, in the same spirit as +// Erlang's `erlang:send_after/3`. Each schedules `msg` for delivery after a +// delay and returns a TimerId; `cancel_timer` can call one off before it +// fires. The destination is resolved when the timer actually fires, not when +// it's armed, so a `Name`-addressed timer always reaches whoever currently +// holds that name, even if the original holder has since restarted. If the +// destination is gone by fire time, the message is silently dropped, exactly +// as a live send to a dead address would be. // --------------------------------------------------------------------------- -/// Deliver `msg` to the exact actor named by `dest` (identity-bound, no -/// redirect — see [`send_to`](crate::registry::send_to)) after `after`. -/// Returns a [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. +/// Deliver `msg` to the exact actor identified by `dest` after `after` has +/// elapsed. Unlike [`send_after_named`], this targets one specific actor: if +/// that actor is gone by the time the timer fires, the message is dropped +/// (it is never redirected to a different actor, even one that inherited the +/// same name). Returns a [`TimerId`](crate::timer::TimerId) you can pass to +/// [`cancel_timer`] to call it off early. pub fn send_after( after: std::time::Duration, dest: Pid, @@ -475,9 +607,11 @@ pub fn send_after( }) } -/// Deliver `msg` to whichever actor holds the name `dest` at fire time -/// (re-resolving [`send`](crate::registry::send) semantics) after `after`. -/// Returns a [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. +/// Deliver `msg` to whichever actor holds the name `dest` when the timer +/// fires, not necessarily whoever holds it now: if the named actor restarts +/// (for example under a supervisor) and re-registers before the deadline, +/// the message reaches the new instance. Returns a +/// [`TimerId`](crate::timer::TimerId) you can pass to [`cancel_timer`]. pub fn send_after_named( after: std::time::Duration, dest: Name, @@ -497,13 +631,12 @@ pub fn send_after_named( }) } -/// Wall-anchored [`send_after`] (RFC 007): under causal profiling the timer -/// opts out of the virtual-time shift and fires at its raw deadline instead -/// of dilating with injected delay — the user-facing opt-out whose substrate -/// [`sleep_wall`] landed. Use it for deadlines that reflect the outside -/// world (protocol timeouts, wall-clock schedules) rather than workload -/// pacing. Without the `smarm-causal` feature it is identical to -/// [`send_after`]. Cancellation via [`cancel_timer`] is unchanged. +/// Like [`send_after`], but the deadline is always measured in real +/// (wall-clock) time rather than being subject to smarm's own profiling and +/// measurement tooling. Use this for deadlines that need to reflect the +/// outside world (a protocol timeout, a wall-clock schedule) rather than +/// simulated workload pacing. Without that tooling active, it behaves +/// identically to [`send_after`]. pub fn send_after_wall( after: std::time::Duration, dest: Pid, @@ -521,8 +654,8 @@ pub fn send_after_wall( }) } -/// Wall-anchored [`send_after_named`] (RFC 007): re-resolving name delivery, -/// raw-deadline anchor — see [`send_after_wall`] for the semantics. +/// Wall-clock-anchored [`send_after_named`]: re-resolving name delivery, +/// with the same real-time deadline guarantee as [`send_after_wall`]. pub fn send_after_named_wall( after: std::time::Duration, dest: Name, @@ -542,17 +675,12 @@ pub fn send_after_named_wall( }) } -/// Deliver `msg` onto a channel the caller owns after `after`, rather than to a -/// registry address. The sibling of [`send_after`] used by the gen_server timer -/// layer (RFC 015 §5): arming a server timer must land the fire on the loop's -/// own `Sys` channel — giving it the loop's arm position — not in the inbox. -/// -/// Same substrate as [`send_after`]: a `Reason::Send` entry, the same `armed` -/// set, the same [`TimerId`](crate::timer::TimerId) for [`cancel_timer`]. Only -/// the fire thunk differs — `tx.send(msg)` instead of a registry resolve — so a -/// send onto a channel whose receiver is gone is dropped, exactly as a failed -/// address resolve is (Erlang `send_after` semantics). `pid` is informational -/// (who armed it); delivery lives entirely in the thunk. +// Deliver `msg` onto a channel the caller already holds, after a delay, +// rather than resolving a registry address at fire time. Used internally by +// gen_server's timer support, which needs the fire to land on the server +// loop's own dedicated channel instead of its public inbox. Otherwise +// identical to `send_after`: same cancellation via `cancel_timer`, and a +// send to a channel whose receiver is gone is silently dropped. pub(crate) fn send_after_to( after: std::time::Duration, tx: Sender, @@ -571,9 +699,9 @@ pub(crate) fn send_after_to( }) } -/// Cancel a timer armed by [`send_after`] / [`send_after_named`]. Returns -/// `true` if it was still pending (delivery now prevented), `false` if it had -/// already fired or been cancelled. +/// Call off a timer armed by [`send_after`] or [`send_after_named`] before +/// it fires. Returns `true` if the timer was still pending and delivery is +/// now prevented, `false` if it had already fired or was already cancelled. pub fn cancel_timer(id: crate::timer::TimerId) -> bool { with_runtime(|inner| { match inner.timers.lock() { @@ -587,6 +715,22 @@ pub fn cancel_timer(id: crate::timer::TimerId) -> bool { // block_on_io / wait_readable / wait_writable / read / write // --------------------------------------------------------------------------- +/// Run a blocking closure (blocking file IO, a synchronous library call, +/// anything that isn't itself actor-aware) on a dedicated worker thread, +/// while the calling actor parks and every other actor keeps making +/// progress. When `f` completes, the calling actor resumes with its result. +/// +/// Reach for this whenever you need to call into something that would +/// otherwise block the underlying OS thread outright: a blocking C library, +/// a synchronous filesystem call, DNS resolution via the system resolver, +/// and so on. For plain readiness-based network IO on a file descriptor, +/// [`wait_readable`] / [`wait_writable`] are cheaper since they don't need a +/// dedicated thread. +/// +/// If `f` panics, that panic is carried over and re-raised in the calling +/// actor, exactly as if the call had been made inline. +/// +/// Panics if called outside an actor. pub fn block_on_io(f: F) -> T where F: FnOnce() -> T + Send + 'static, @@ -639,10 +783,19 @@ where } } +/// Park the calling actor until `fd` becomes readable. Other actors keep +/// running while you wait; when the kernel reports the fd ready, the actor +/// resumes and you perform the actual `read(2)` yourself (see [`read`] for +/// a convenience wrapper that does both steps). +/// +/// Only one actor may wait on a given fd for a given direction at a time. +/// Panics if called outside an actor. pub fn wait_readable(fd: std::os::fd::RawFd) -> std::io::Result<()> { wait_fd(fd, true, false) } +/// Park the calling actor until `fd` becomes writable. See [`wait_readable`] +/// for the read-side counterpart; the same notes apply. pub fn wait_writable(fd: std::os::fd::RawFd) -> std::io::Result<()> { wait_fd(fd, false, true) } @@ -666,12 +819,12 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R })?; // If a terminal stop unwinds us out of the park below, the registration - // must not outlive us: a stale `waiters` entry fails every future - // `wait_*` on this fd with AlreadyExists, and the kernel-side ADD leaks - // until the fd happens to be reused. Clean up iff the entry is still - // ours — a `FdReady` racing the stop may have consumed it already (it - // removes + DELs under the io lock), after which the fd may even carry - // ANOTHER actor's fresh registration; in that case touch nothing. + // must not outlive us: a stale registration would fail every future + // `wait_*` on this fd, and the kernel-side registration would leak until + // the fd happens to be reused. Clean up only if the entry is still + // ours; a wakeup racing the stop may have already consumed it, possibly + // leaving a different actor's fresh registration in its place, which + // must not be disturbed. struct Dereg { fd: std::os::fd::RawFd, me: Pid, @@ -695,24 +848,28 @@ fn wait_fd(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::R } let guard = Dereg { fd, me, epoch }; park_current(); - // Normal wake: the FdReady path removed the entry and DEL'd the fd - // before the unpark, so the guard's check would be a guaranteed no-op — - // skip the io lock on the hot path. (Dereg owns nothing; forget leaks - // no resource.) + // Normal wake: the ready-fd path already removed the entry and + // deregistered from epoll before waking us, so the guard's check on drop + // is a guaranteed no-op. Skip it on this hot path (Dereg owns no + // resource itself, so forgetting it leaks nothing). std::mem::forget(guard); Ok(()) } // --------------------------------------------------------------------------- -// FdArm — fd readiness as a select arm (RFC 008) +// FdArm: fd readiness as a select arm // --------------------------------------------------------------------------- -/// An fd-readiness arm for [`crate::select`] / [`crate::select_timeout`]: -/// ready when the fd is readable (resp. writable), composable with channel -/// receivers on one wait epoch. Phase-1 rules apply: one waiter per fd at a -/// time, one direction per arm (duplex on a single fd needs `dup`; epoll -/// registrations key on the open file description, so dup'd fds register -/// independently). +/// A file-descriptor readiness condition usable as an arm of +/// [`select`](crate::select) / [`select_timeout`](crate::select_timeout), +/// so you can wait on "this fd is readable" alongside ordinary channel +/// receivers in the same call. Build one with [`FdArm::readable`] or +/// [`FdArm::writable`]. +/// +/// Only one actor may wait on a given fd for a given direction at a time. A +/// single fd open in both directions needs two `FdArm`s (or `dup` the fd) if +/// you want to wait on both; you cannot wait on read and write readiness +/// with one arm. pub struct FdArm { fd: std::os::fd::RawFd, readable: bool, @@ -720,10 +877,12 @@ pub struct FdArm { } impl FdArm { + /// An arm that becomes ready when `fd` is readable. pub fn readable(fd: std::os::fd::RawFd) -> Self { FdArm { fd, readable: true, writable: false } } + /// An arm that becomes ready when `fd` is writable. pub fn writable(fd: std::os::fd::RawFd) -> Self { FdArm { fd, readable: false, writable: true } } @@ -732,14 +891,12 @@ impl FdArm { impl crate::channel::sealed::Sealed for FdArm {} impl crate::channel::Selectable for FdArm { - /// Ready-now check is a zero-timeout `poll(2)`; if the requested events - /// are pending the wait is retired without registering (`Ok(false)`, - /// the channel-arm contract). Otherwise register with the io thread — - /// every failure surfaces as `Err` (EBADF including a closed-fd - /// POLLNVAL, EMFILE on the epoll set, AlreadyExists for a second - /// waiter on the fd): the fallible-out, nothing-left-behind rule, in - /// deviation from RFC 008's permanently-ready lean, which would spin a - /// consumer whose fd is healthy but unregistrable (EMFILE). + // Ready-now check is a zero-timeout poll(2); if the requested events are + // already pending, the wait is retired without registering. Otherwise + // register with the IO thread. Any registration failure (a closed fd, + // too many fds registered, a second waiter already on this fd) is + // surfaced as an error rather than silently treated as "always ready", + // so a caller never spins on an fd that genuinely can't be registered. fn sel_register(&self, pid: Pid, epoch: u32) -> std::io::Result { if poll_events(self.fd, self.readable, self.writable)? { return Ok(false); @@ -759,20 +916,19 @@ impl crate::channel::Selectable for FdArm { Ok(true) } - /// Classification is the same zero-timeout poll: a pure function of fd - /// state, independent of the registration the cleanup pass removed. An - /// error here (EBADF: fd closed mid-wait) reports READY — the - /// consumer's read/write surfaces the errno; a dead arm is an event, - /// not a hang. + // Same zero-timeout poll used for classification: a pure function of + // fd state. An error here (fd closed mid-wait) is reported as ready + // rather than pending forever: the caller's own read/write then + // surfaces the real error, so a dead fd is an observable event, not a + // silent hang. fn sel_ready(&self) -> bool { poll_events(self.fd, self.readable, self.writable).unwrap_or(true) } - /// `wait_fd`'s `Dereg` compare, verbatim: remove the waiters entry and - /// kernel-side registration iff the entry is still `(pid, epoch)`-ours. - /// A `FdReady` racing the wake may have consumed it (it removes + DELs - /// under the io lock), after which the fd may even carry ANOTHER - /// actor's fresh registration; in that case touch nothing. + // Mirrors wait_fd's cleanup guard: remove the registration only if it's + // still ours. A wakeup racing this cleanup may have already consumed + // it, possibly leaving a different actor's fresh registration on the + // same fd, which must not be touched. fn sel_unregister(&self, pid: Pid, epoch: u32) { with_runtime(|inner| { let mut io = match inner.io.lock() { @@ -793,9 +949,9 @@ impl crate::channel::Selectable for FdArm { } } -/// Zero-timeout `poll(2)`: are any of the requested events (or ERR/HUP, -/// which make the consumer's read/write fail loudly rather than park -/// forever) pending on `fd`? POLLNVAL maps to `Err(EBADF)`. +// Zero-timeout poll(2): are any of the requested events (or an error/hangup +// condition, so the caller's read/write fails loudly instead of parking +// forever) pending on `fd` right now? fn poll_events(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::io::Result { let mut events: libc::c_short = 0; if readable { @@ -824,8 +980,9 @@ fn poll_events(fd: std::os::fd::RawFd, readable: bool, writable: bool) -> std::i } } -/// Wait until `fd` is readable or `timeout` elapses: `Ok(true)` = ready, -/// `Ok(false)` = timed out. A one-arm [`crate::try_select_timeout`]. +/// Like [`wait_readable`], but gives up after `timeout` instead of waiting +/// indefinitely: `Ok(true)` means the fd became ready, `Ok(false)` means the +/// timeout elapsed first. pub fn wait_readable_timeout( fd: std::os::fd::RawFd, timeout: std::time::Duration, @@ -834,8 +991,9 @@ pub fn wait_readable_timeout( Ok(crate::channel::try_select_timeout(&[&arm], timeout)?.is_some()) } -/// Wait until `fd` is writable or `timeout` elapses: `Ok(true)` = ready, -/// `Ok(false)` = timed out. +/// Like [`wait_writable`], but gives up after `timeout` instead of waiting +/// indefinitely: `Ok(true)` means the fd became ready, `Ok(false)` means the +/// timeout elapsed first. pub fn wait_writable_timeout( fd: std::os::fd::RawFd, timeout: std::time::Duration, @@ -844,12 +1002,18 @@ pub fn wait_writable_timeout( Ok(crate::channel::try_select_timeout(&[&arm], timeout)?.is_some()) } +/// Convenience wrapper: park until `fd` is readable, then perform the +/// `read(2)` into `buf`. Equivalent to calling [`wait_readable`] yourself +/// followed by a raw read, provided as a shorthand for the common case. pub fn read(fd: std::os::fd::RawFd, buf: &mut [u8]) -> std::io::Result { wait_readable(fd)?; let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut _, buf.len()) }; if n < 0 { Err(std::io::Error::last_os_error()) } else { Ok(n as usize) } } +/// Convenience wrapper: park until `fd` is writable, then perform the +/// `write(2)` of `buf`. Equivalent to calling [`wait_writable`] yourself +/// followed by a raw write, provided as a shorthand for the common case. pub fn write(fd: std::os::fd::RawFd, buf: &[u8]) -> std::io::Result { wait_writable(fd)?; let n = unsafe { libc::write(fd, buf.as_ptr() as *const _, buf.len()) }; @@ -857,7 +1021,7 @@ pub fn write(fd: std::os::fd::RawFd, buf: &[u8]) -> std::io::Result { } // --------------------------------------------------------------------------- -// register_supervisor_channel +// register_supervisor_channel: internal wiring used by supervisor.rs // --------------------------------------------------------------------------- pub fn register_supervisor_channel(pid: Pid, sender: Sender) { @@ -874,11 +1038,21 @@ pub fn register_supervisor_channel(pid: Pid, sender: Sender) { } // --------------------------------------------------------------------------- -// Legacy run() — convenience wrapper +// run(): the single-threaded convenience entry point // --------------------------------------------------------------------------- -/// Single-threaded runtime entry point (backwards-compatible wrapper). -/// Equivalent to `runtime::init(Config::exact(1)).run(f)`. +/// Start the smarm runtime on a single OS thread, run `f` as the first +/// (root) actor, and block until every actor it transitively spawned has +/// finished. +/// +/// This is the simplest way to get started, and is all you need for most +/// programs: `f` typically calls [`spawn`] to create more actors and waits +/// on their [`JoinHandle`]s. If you want smarm to schedule actors across +/// multiple OS threads instead, use +/// [`crate::runtime::init`] with a +/// [`Config`](crate::runtime::Config) that requests more than one scheduler +/// thread, then call [`Runtime::run`](crate::runtime::Runtime::run) on it; +/// `run` here is exactly that, pinned to one thread. pub fn run(f: F) { crate::runtime::init(crate::runtime::Config::exact(1)).run(f); }