Markk116/smarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(scheduler): user-facing rewrite of the actor/spawn/run entry point

Browse Source 
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.
This commit is contained in:
Markk116
2026-07-24 08:40:26 +02:00
parent d9addeba5e
commit 8625ae4c35
1 changed files with 352 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/scheduler.rs
+352 -178
View File
@@ -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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> 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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> 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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> R) -> Option<R> {
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<R>(f: impl FnOnce(&Arc<RuntimeInner>) -> 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<dyn std::any::Any + Send>,
}
/// 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<A>(supervisor: Pid<A>, 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<A>(supervisor: Pid<A>, f: impl FnOnce() + Send + 'static) ->
JoinHandle { pid, consumed: false }
}
/// Spawn a typed, single-message actor and hand back its identity-bound
/// [`Pid<A>`] (RFC 014's typed-path producer). The runtime makes the actor's
/// inbox, hands the body its [`Receiver<A::Msg>`], installs the sender, and
/// returns the parent a `Pid<A>`.
/// Spawn an actor that other actors can message directly by its [`Pid<A>`],
/// 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<A::Msg>`](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<A: crate::pid::Addressable>(
@@ -237,6 +337,11 @@ pub fn spawn_addr<A: crate::pid::Addressable>(
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<A>(pid: Pid<A>) {
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<A>` that has since died yields `SendError::Dead`, a `Name<M>` 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<A: crate::pid::Addressable>(
after: std::time::Duration,
dest: Pid<A>,
@@ -475,9 +607,11 @@ pub fn send_after<A: crate::pid::Addressable>(
})
}
/// 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<M: Send + 'static>(
after: std::time::Duration,
dest: Name<M>,
@@ -497,13 +631,12 @@ pub fn send_after_named<M: Send + 'static>(
})
}
/// 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<A: crate::pid::Addressable>(
after: std::time::Duration,
dest: Pid<A>,
@@ -521,8 +654,8 @@ pub fn send_after_wall<A: crate::pid::Addressable>(
})
}
/// 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<M: Send + 'static>(
after: std::time::Duration,
dest: Name<M>,
@@ -542,17 +675,12 @@ pub fn send_after_named_wall<M: Send + 'static>(
})
}
/// 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<T: Send + 'static>(
after: std::time::Duration,
tx: Sender<T>,
@@ -571,9 +699,9 @@ pub(crate) fn send_after_to<T: Send + 'static>(
})
}
/// 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, T>(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<bool> {
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<bool> {
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<usize> {
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<usize> {
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<usize> {
}
// ---------------------------------------------------------------------------
// register_supervisor_channel
// register_supervisor_channel: internal wiring used by supervisor.rs
// ---------------------------------------------------------------------------
pub fn register_supervisor_channel(pid: Pid, sender: Sender<Signal>) {
@@ -874,11 +1038,21 @@ pub fn register_supervisor_channel(pid: Pid, sender: Sender<Signal>) {
}
// ---------------------------------------------------------------------------
// 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: FnOnce() + Send + 'static>(f: F) {
crate::runtime::init(crate::runtime::Config::exact(1)).run(f);
}