Claude (sandbox)
a3be8f0977
Measure-only counters for the deficit hunt (~23ms short per 700ms window at 50% on the bottleneck site; superlinear vs 25%). Nothing here changes injection or absorption; the sweep decides the fix. Buckets, windowed per cell into new ExperimentResult fields (audit snapshot taken at end() — spin/attribution freeze there, forgiveness does not): - spin_absorbed / park_forgiven: where owed delay actually went. Spin during a 0% cell is the baseline-contamination signature — leftover debt from a prior window being paid in a later one (checks gate on the experiment word, so cooldowns pay nothing and debt carries over). - drop_park / drop_yield (+counts): the deschedule path flushes no sample tail — an in-target-site park or yield silently loses [last sample -> now]; on_resume re-arms before the actor runs again. New on_deschedule hook in all three intent arms (real park; explicit/ slice-expiry yield; requeued park counts as yield — it never blocked). Slice-expiry yields sample at the descheduling checkpoint, so a fat yield bucket points at explicit yield_now or requeued parks. - discard_overmax (+count, in would-be delta terms so columns compare against injected_cycles) / discard_unarmed: the attribute() clamps, previously silent. LedgerCounters + ledger_counters() expose cumulative totals (tests, run-level prints); render_ledger_audit() is the per-cell companion to render_summary, which stays byte-identical (pinned). ExperimentResult now derives Default so literals survive future audit-field growth. Tests: +7 (spin counted, forgiveness counted, in-site park drop, in-site yield drop, overmax discard, 0%-window leftover absorption — synthesized deterministically via inject_delay_cycles_for_test with no experiment active — and the audit render). 22/22 causal.
1799 lines
81 KiB
Rust
1799 lines
81 KiB
Rust
//! Multi-scheduler runtime: configuration, initialisation, and the shared
|
||
//! state that all scheduler OS threads operate against.
|
||
//!
|
||
//! # Architecture (post slot-table split, ROADMAP_v0.5 phase 2)
|
||
//!
|
||
//! ```text
|
||
//! init(Config) → Runtime (Arc<RuntimeInner>)
|
||
//!
|
||
//! RuntimeInner {
|
||
//! slots: Box<[Slot]> ← FIXED slab, max_actors entries, lock-free lookup
|
||
//! free: RawMutex<Vec<u32>> ← vacant slot indices
|
||
//! run_queue: RunQueue ← compile-time selected (src/run_queue.rs)
|
||
//! timers: Mutex<Timers>
|
||
//! io: Mutex<Option<IoThread>>
|
||
//! live_actors: AtomicU32 ← spawned-but-not-finalized count (termination)
|
||
//! stats: Vec<SchedulerStats> ← one per thread, lockless atomics (RFC 000)
|
||
//! }
|
||
//!
|
||
//! Slot {
|
||
//! word: AtomicU64 ← (gen << 32) | (epoch << 8) | state — THE state machine
|
||
//! sp: AtomicUsize ← saved stack pointer
|
||
//! stop_ptr:AtomicPtr<...> ← into the actor's Arc<AtomicBool>
|
||
//! closure: AtomicPtr<...> ← first-resume closure, swap-to-take
|
||
//! cold: RawMutex<SlotCold> ← lifecycle collections (waiters/monitors/links/…)
|
||
//! }
|
||
//! ```
|
||
//!
|
||
//! # The per-slot state machine
|
||
//!
|
||
//! Scheduling state lives in one atomic word per slot packing
|
||
//! `(generation, park-epoch, state)`, where state is one of:
|
||
//!
|
||
//! ```text
|
||
//! Vacant ─spawn→ Queued ─pop→ Running ─yield→ Queued
|
||
//! ↑ │ │
|
||
//! │ park │ unpark while running
|
||
//! │ ↓ ↓
|
||
//! unpark ←──── Parked RunningNotified ─park→ Queued
|
||
//! (re-queued immediately)
|
||
//! Running|RunningNotified ─actor returns→ Done ─reclaim→ Vacant(gen+1)
|
||
//! ```
|
||
//!
|
||
//! Every transition is a CAS on the packed word, so:
|
||
//!
|
||
//! - The generation check is **atomic with the transition** — a stale `Pid`
|
||
//! can never act on a recycled slot (no ABA, no spurious unparks).
|
||
//! - The park-epoch (middle 24 bits) is the actor's *wait identity*: opened
|
||
//! by `begin_wait` before any registration, consumed (bumped) by every
|
||
//! successful wake. Registration-based wakers carry `(pid, epoch)` and use
|
||
//! `unpark_at`, so a waker holding a registration from an already-woken
|
||
//! wait — a `select` loser arm, a satisfied wait's timer — fails the epoch
|
||
//! check and no-ops instead of faulting a later one-shot park. The only
|
||
//! wildcard wake is `request_stop`, which is terminal. Full rules in
|
||
//! slot_state.rs.
|
||
//! - `RunningNotified` replaces the old `pending_unpark` bool: an unpark that
|
||
//! races the prep-to-park window is a *state*, resolved by the scheduler's
|
||
//! park-return CAS, not a flag read under a lock. This also closes a latent
|
||
//! lost-wakeup in the old Blocking-IO completion path, which set the result
|
||
//! for a still-Running actor without flagging it.
|
||
//! - **A pid is in the run queue at most once**: the only pushes are paired
|
||
//! 1:1 with successful transitions *into* `Queued`, and only the scheduler
|
||
//! transitions `Queued → Running` (paired 1:1 with pops).
|
||
//!
|
||
//! Memory ordering: all word CASes are `AcqRel` (failure `Acquire`), plain
|
||
//! word stores are `Release`, loads are `Acquire`. The chain that matters:
|
||
//! the park path stores `sp` (Relaxed) *before* its Release transition; any
|
||
//! later Acquire transition/load of the word therefore observes that `sp`.
|
||
//! The run-queue mutex independently provides the same edges today; the
|
||
//! word's own ordering is what phase 3's lock-free queue will rely on.
|
||
//!
|
||
//! # Locks and ordering
|
||
//!
|
||
//! - The run queue is its own module (`run_queue.rs`), selected at compile
|
||
//! time (`rq-mutex` / `rq-mpmc` / `rq-striped`). Queue ops require
|
||
//! preemption disabled (debug-asserted there); when the mutex variant is in
|
||
//! play it is the innermost lock — nothing else is acquired under it.
|
||
//! - Per-slot `cold` locks ([`RawMutex`], non-poisoning, guard enters
|
||
//! `NoPreempt`) guard the lifecycle collections. **Leaf rule: never hold
|
||
//! two cold locks at once** — `finalize_actor`'s link cascade and `link()`
|
||
//! lock peers one at a time (correctness arguments at the call sites).
|
||
//! Holding a cold lock while pushing to the run queue is permitted.
|
||
//! - Lock order overall: `io` → (slot `cold` | `free` | `stack_pool`,
|
||
//! mutually leaf) → run queue (innermost). `timers` is independent (never
|
||
//! nested with any of the above on either side).
|
||
//!
|
||
//! # Termination (counter-based)
|
||
//!
|
||
//! The old all-clear scanned the slot table under the big lock. Now:
|
||
//! exit when `io_out == 0` (read *before* the queue lock, phase-1 ordering)
|
||
//! and, under the queue lock, the queue is empty and `live_actors == 0`.
|
||
//! `live_actors` is incremented in `spawn` before the enqueue and decremented
|
||
//! at the very END of `finalize_actor`, strictly after every wakeup that
|
||
//! finalize produces has been enqueued. The soundness crux: any enqueue
|
||
//! targets a live (not-yet-finalized) actor, so `live == 0` implies no wakeup
|
||
//! can still be in flight; combined with "spawner is itself live", observing
|
||
//! `(queue empty, live == 0)` under the queue lock means no work can ever
|
||
//! appear again.
|
||
//!
|
||
//! # Timer / IO drain (try-lock, one-winner)
|
||
//!
|
||
//! Unchanged from phase 1: one winner per round drains due timers and IO
|
||
//! completions from their own mutexes; wakeups go through the unpark
|
||
//! protocol like everyone else's.
|
||
|
||
use crate::actor::{
|
||
clear_current_pid, is_actor_done, reset_actor_done, set_current_actor_box,
|
||
set_current_pid, take_last_outcome, Actor, Outcome,
|
||
};
|
||
use crate::channel::Sender;
|
||
use crate::io::IoThread;
|
||
use crate::monitor::{Down, DownReason, MonitorId};
|
||
use crate::pid::Pid;
|
||
use crate::preempt::PREEMPTION_ENABLED;
|
||
use crate::raw_mutex::RawMutex;
|
||
use crate::slot_state::{StateWord, Status, Unpark};
|
||
use crate::supervisor::Signal;
|
||
use crate::timer::Timers;
|
||
use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor};
|
||
|
||
use std::sync::atomic::{
|
||
AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering,
|
||
};
|
||
use std::sync::{Arc, Mutex};
|
||
use std::thread;
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Config
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Default capacity of the actor slot table. Slots are ~256 bytes, so the
|
||
/// default costs ~4 MiB, allocated once at `init`. See [`Config::max_actors`].
|
||
pub const DEFAULT_MAX_ACTORS: usize = 16_384;
|
||
|
||
/// Runtime configuration.
|
||
///
|
||
/// ```
|
||
/// use smarm::runtime::Config;
|
||
///
|
||
/// // Use all available CPUs (default):
|
||
/// let c = Config::default();
|
||
///
|
||
/// // Exactly 4 scheduler threads:
|
||
/// let c = Config::exact(4);
|
||
///
|
||
/// // Between 2 and 8, clamped to available parallelism:
|
||
/// let c = Config::new(2, 8, None);
|
||
/// ```
|
||
#[derive(Clone, Debug)]
|
||
pub struct Config {
|
||
min: usize,
|
||
max: usize,
|
||
exact: Option<usize>,
|
||
alloc_interval: u32,
|
||
timeslice_cycles: u64,
|
||
stack_pool_cap: usize,
|
||
max_actors: usize,
|
||
wake_slot: bool,
|
||
node_id: crate::pg::NodeId,
|
||
incarnation: crate::pg::Incarnation,
|
||
}
|
||
|
||
impl Config {
|
||
/// Exact thread count; takes precedence over min/max.
|
||
pub fn exact(n: usize) -> Self {
|
||
assert!(n >= 1, "scheduler thread count must be ≥ 1");
|
||
Self {
|
||
min: n, max: n, exact: Some(n),
|
||
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
|
||
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
|
||
stack_pool_cap: n * 4,
|
||
max_actors: DEFAULT_MAX_ACTORS,
|
||
wake_slot: false,
|
||
node_id: crate::pg::DEFAULT_NODE_ID,
|
||
incarnation: crate::pg::DEFAULT_INCARNATION,
|
||
}
|
||
}
|
||
|
||
/// Bounded range. Thread count = clamp(available_parallelism, min, max).
|
||
pub fn new(min: usize, max: usize, exact: Option<usize>) -> Self {
|
||
assert!(min >= 1, "min must be ≥ 1");
|
||
assert!(max >= min, "max must be ≥ min");
|
||
if let Some(e) = exact {
|
||
assert!(e >= 1, "exact must be ≥ 1");
|
||
}
|
||
Self {
|
||
min, max, exact,
|
||
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
|
||
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
|
||
stack_pool_cap: max * 4,
|
||
max_actors: DEFAULT_MAX_ACTORS,
|
||
wake_slot: false,
|
||
node_id: crate::pg::DEFAULT_NODE_ID,
|
||
incarnation: crate::pg::DEFAULT_INCARNATION,
|
||
}
|
||
}
|
||
|
||
/// How many allocations (or `smarm::check!()` calls) between RDTSC checks.
|
||
/// Lower = more responsive preemption, higher = less overhead.
|
||
/// Default: 128.
|
||
pub fn alloc_interval(mut self, n: u32) -> Self {
|
||
assert!(n >= 1, "alloc_interval must be ≥ 1");
|
||
self.alloc_interval = n;
|
||
self
|
||
}
|
||
|
||
/// How many TSC cycles constitute one timeslice.
|
||
/// Default: 300_000 (≈ 100µs on a 3 GHz CPU).
|
||
pub fn timeslice_cycles(mut self, n: u64) -> Self {
|
||
assert!(n >= 1, "timeslice_cycles must be ≥ 1");
|
||
self.timeslice_cycles = n;
|
||
self
|
||
}
|
||
|
||
/// Maximum number of stacks kept in the pool for reuse across spawns.
|
||
/// A larger cap reduces `mmap`/`munmap` syscalls at the cost of idle memory.
|
||
/// Default: `thread_count * 4`.
|
||
pub fn stack_pool_cap(mut self, n: usize) -> Self {
|
||
self.stack_pool_cap = n;
|
||
self
|
||
}
|
||
|
||
/// Capacity of the actor slot table — the maximum number of
|
||
/// **simultaneously live** actors (total spawned over a run is unbounded;
|
||
/// slots are recycled). The table is a fixed slab allocated once at
|
||
/// `init`: slots never move, which is what makes lock-free slot lookup
|
||
/// sound. Exhausting it is a loud panic naming this knob.
|
||
/// Default: [`DEFAULT_MAX_ACTORS`] (16_384, ~4 MiB).
|
||
pub fn max_actors(mut self, n: usize) -> Self {
|
||
assert!(n >= 1, "max_actors must be ≥ 1");
|
||
assert!(
|
||
n < u32::MAX as usize,
|
||
"max_actors must fit a u32 slot index (ROOT_PID reserves u32::MAX)"
|
||
);
|
||
self.max_actors = n;
|
||
self
|
||
}
|
||
|
||
/// Enable the per-scheduler wake slot (RFC 005): a thread-local,
|
||
/// capacity-one wake cache checked before the shared run queue. A wake
|
||
/// performed from actor context parks the woken pid in the waking
|
||
/// thread's slot; it is resumed next on that core and inherits the
|
||
/// remainder of the waker's timeslice. Scheduler-context wakes
|
||
/// (timer/IO drain) and spawns always go to the shared queue.
|
||
/// Default: `false` (off until the slot shootout accepts it).
|
||
pub fn wake_slot(mut self, on: bool) -> Self {
|
||
self.wake_slot = on;
|
||
self
|
||
}
|
||
|
||
/// This runtime's node identity (RFC 012). Defaults to a fixed single-node
|
||
/// value; clustering (RFC 010) will supply a real one. Threaded through pg
|
||
/// storage/eviction so the process-group public API never changes to
|
||
/// acquire it.
|
||
pub fn node_id(mut self, id: impl Into<crate::pg::NodeId>) -> Self {
|
||
self.node_id = id.into();
|
||
self
|
||
}
|
||
|
||
/// This runtime's incarnation epoch (RFC 012) — the BEAM `Creation` analogue
|
||
/// that separates a crashed node from its restart. Defaults to a fixed
|
||
/// single-node value; constant for the life of a run.
|
||
pub fn incarnation(mut self, inc: impl Into<crate::pg::Incarnation>) -> Self {
|
||
self.incarnation = inc.into();
|
||
self
|
||
}
|
||
|
||
/// The number of scheduler threads this config resolves to.
|
||
pub fn resolved_thread_count(&self) -> usize {
|
||
if let Some(e) = self.exact {
|
||
return e;
|
||
}
|
||
let avail = thread::available_parallelism()
|
||
.map(|n| n.get())
|
||
.unwrap_or(1);
|
||
avail.clamp(self.min, self.max)
|
||
}
|
||
}
|
||
|
||
impl Default for Config {
|
||
fn default() -> Self {
|
||
let avail = thread::available_parallelism()
|
||
.map(|n| n.get())
|
||
.unwrap_or(1);
|
||
Self {
|
||
min: 1, max: avail, exact: None,
|
||
alloc_interval: crate::preempt::DEFAULT_ALLOC_INTERVAL,
|
||
timeslice_cycles: crate::preempt::DEFAULT_TIMESLICE_CYCLES,
|
||
stack_pool_cap: avail * 4,
|
||
max_actors: DEFAULT_MAX_ACTORS,
|
||
wake_slot: false,
|
||
node_id: crate::pg::DEFAULT_NODE_ID,
|
||
incarnation: crate::pg::DEFAULT_INCARNATION,
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Per-thread stats (RFC 000 Layer 1 primitives)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Lockless per-scheduler-thread counters. Written only by the owning thread;
|
||
/// readable from any thread (introspection actor, tests).
|
||
#[repr(align(64))]
|
||
pub struct SchedulerStats {
|
||
/// PID index of the actor currently on-CPU, or `u32::MAX` when idle.
|
||
pub current_pid_index: AtomicU32,
|
||
/// Snapshot of run queue length maintained on every push/pop.
|
||
pub run_queue_len: AtomicU64,
|
||
/// RFC 005: wakes resumed from this thread's wake slot.
|
||
pub slot_hits: AtomicU64,
|
||
/// RFC 005: slot occupants displaced to the shared queue by a newer wake.
|
||
pub slot_displacements: AtomicU64,
|
||
}
|
||
|
||
impl SchedulerStats {
|
||
fn new() -> Self {
|
||
Self {
|
||
current_pid_index: AtomicU32::new(u32::MAX),
|
||
run_queue_len: AtomicU64::new(0),
|
||
slot_hits: AtomicU64::new(0),
|
||
slot_displacements: AtomicU64::new(0),
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Runtime stats snapshot (for tests / introspection)
|
||
// ---------------------------------------------------------------------------
|
||
|
||
pub struct RuntimeStats {
|
||
pub(crate) inner: Arc<RuntimeInner>,
|
||
}
|
||
|
||
impl RuntimeStats {
|
||
/// Sum of run queue lengths across all scheduler threads.
|
||
pub fn total_run_queue_len(&self) -> u64 {
|
||
self.inner.stats.iter()
|
||
.map(|s| s.run_queue_len.load(Ordering::Relaxed))
|
||
.sum()
|
||
}
|
||
|
||
/// Number of scheduler threads.
|
||
pub fn scheduler_count(&self) -> usize {
|
||
self.inner.stats.len()
|
||
}
|
||
|
||
/// Actors currently parked on IO.
|
||
pub fn io_parked_count(&self) -> u32 {
|
||
self.inner.io_parked.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// Actors currently sleeping on a timer.
|
||
pub fn sleeping_count(&self) -> u32 {
|
||
self.inner.sleeping.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// RFC 005: total wakes resumed from a wake slot, summed across
|
||
/// scheduler threads. Counters are reset at the start of each `run()`,
|
||
/// so after a run this reads that run's total.
|
||
pub fn slot_hits(&self) -> u64 {
|
||
self.inner.stats.iter()
|
||
.map(|s| s.slot_hits.load(Ordering::Relaxed))
|
||
.sum()
|
||
}
|
||
|
||
/// RFC 005: total slot occupants displaced to the shared queue, summed
|
||
/// across scheduler threads. Reset at the start of each `run()`.
|
||
pub fn slot_displacements(&self) -> u64 {
|
||
self.inner.stats.iter()
|
||
.map(|s| s.slot_displacements.load(Ordering::Relaxed))
|
||
.sum()
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Slot — packed state word + hot atomics + cold lifecycle data
|
||
// ---------------------------------------------------------------------------
|
||
|
||
pub(crate) const ACTOR_STACK_SIZE: usize = 64 * 1024;
|
||
|
||
pub(crate) type Closure = Box<dyn FnOnce() + Send>;
|
||
|
||
/// Lifecycle data, mutated only under the slot's cold [`RawMutex`]. Everything
|
||
/// here is touched O(1) times per actor lifetime (spawn / join / monitor /
|
||
/// link / finalize), never on the yield/park/unpark hot path.
|
||
pub(crate) struct SlotCold {
|
||
pub(crate) actor: Option<Actor>,
|
||
/// Parked joiners as `(pid, park-epoch)`; finalize wakes each via the
|
||
/// epoch-matched unpark.
|
||
pub(crate) waiters: Vec<(Pid, u32)>,
|
||
pub(crate) outcome: Option<Outcome>,
|
||
pub(crate) supervisor_channel: Option<Sender<Signal>>,
|
||
/// Watchers registered via `monitor()`, each tagged with its
|
||
/// `MonitorId` so `demonitor` can remove exactly one. Each receives one
|
||
/// `Down` when this actor terminates (drained in `finalize_actor`).
|
||
/// Distinct from `supervisor_channel`, which is the parent's single funnel.
|
||
pub(crate) monitors: Vec<(MonitorId, Sender<Down>)>,
|
||
/// Bidirectional links (roadmap #3). Each entry is a peer whose abnormal
|
||
/// death propagates to this actor (and vice versa). Entries may be
|
||
/// momentarily or persistently stale (peer already dead) — every walk
|
||
/// re-verifies the peer's word, so stale entries are benign no-ops.
|
||
pub(crate) links: Vec<Pid>,
|
||
pub(crate) outstanding_handles: u32,
|
||
pub(crate) pending_io_result: Option<crate::io::IoResult>,
|
||
}
|
||
|
||
/// One actor slot. Hot scheduling state is atomic; cold lifecycle state is
|
||
/// behind `cold`. Slots live in a fixed slab and never move.
|
||
///
|
||
/// `align(128)` keeps two adjacent slots' hot words off each other's
|
||
/// cache-line pair (x86 prefetches lines in pairs), avoiding false sharing
|
||
/// between unrelated actors.
|
||
#[repr(align(128))]
|
||
pub(crate) struct Slot {
|
||
/// `(generation << 32) | state` — the state machine, factored into
|
||
/// `slot_state.rs` (loom-modeled there; every transition self-asserts).
|
||
word: StateWord,
|
||
/// Saved stack pointer. Written by the owning scheduler thread before the
|
||
/// Release transition out of Running; read after the Acquire transition
|
||
/// Queued→Running. Relaxed is sufficient — ordering rides on `word`.
|
||
sp: AtomicUsize,
|
||
/// Pointer into the actor's `Arc<AtomicBool>` stop flag. Set at spawn,
|
||
/// nulled at finalize. The box outlives every read: it is only ever read
|
||
/// on the resume path while the actor cannot be finalized (it is on-CPU).
|
||
stop_ptr: AtomicPtr<AtomicBool>,
|
||
/// First-resume closure, double-boxed so it fits an `AtomicPtr`
|
||
/// (`Box<Closure>` is a thin pointer). Swap-to-take; null when absent.
|
||
closure: AtomicPtr<Closure>,
|
||
/// RFC 016 Chunk 2 — per-actor timeslice overrun tally. Single-writer: only
|
||
/// the on-CPU actor's scheduler thread increments it (at the slice-expiry
|
||
/// site in `preempt.rs`, reached via the stashed slot pointer), so the
|
||
/// writes are plain Relaxed load+store with no atomic-RMW traffic; the
|
||
/// snapshot reads it Relaxed from any thread. Lives in the hot region rather
|
||
/// than `SlotCold` so the increment needs no lock; reset across reuse like
|
||
/// every other slot field (`vacant` / `reclaim_slot` / `install_actor`).
|
||
overruns: AtomicU64,
|
||
/// RFC 016 Chunk 2 — messages this actor has received (dequeued). Same
|
||
/// single-writer discipline as `overruns`: only the receiving actor, on its
|
||
/// own thread, increments it on the receive path (D4/D5), Relaxed load+store
|
||
/// with no RMW; snapshot reads Relaxed.
|
||
messages_received: AtomicU64,
|
||
/// RFC 016 Chunk 2 — cumulative on-CPU cycles this incarnation has consumed
|
||
/// (the cycle-accurate "budget used", an analogue of OTP reductions).
|
||
/// Written only by the scheduler thread that ran the actor, once per resume,
|
||
/// and only when the `budget-accounting` feature is on (it costs two RDTSC
|
||
/// per resume); stays 0 otherwise. Same single-writer Relaxed discipline.
|
||
budget_cycles: AtomicU64,
|
||
/// RFC 007 (`smarm-causal`) — id of the causal-profiling site this actor
|
||
/// is currently inside (0 = none). Lives in the slot, not a thread-local,
|
||
/// so it survives preemption and cross-scheduler migration. Written only
|
||
/// by the actor itself (guard enter/exit on its own thread), read by that
|
||
/// same thread in `maybe_preempt` — single-writer Relaxed, like `overruns`.
|
||
/// Exists regardless of the feature; stays 0 without it.
|
||
causal_site: AtomicU32,
|
||
/// RFC 007 (`smarm-causal`) — virtual-speedup delay cycles this actor has
|
||
/// absorbed (or been credited). Compared against the global ledger in
|
||
/// `causal::check`; fast-forwarded on resume-from-park so blocked time
|
||
/// absorbs delay for free (Coz's blocked-thread rule). Same single-writer
|
||
/// discipline.
|
||
causal_delay: AtomicU64,
|
||
/// RFC 007 (`smarm-causal`) — true iff this actor's last deschedule was a
|
||
/// real park (a successful `park_return`, not a slice yield and not the
|
||
/// instant-wake re-queue). Consumed by `causal::on_resume`: only a wake
|
||
/// from genuine blocking forgives outstanding virtual delay; a merely
|
||
/// preempted (runnable) actor stays in debt and must pay by spinning.
|
||
/// Starts true so a fresh spawn doesn't inherit the process's whole delay
|
||
/// history. Written by the owning scheduler thread at the deschedule /
|
||
/// resume boundary only — same single-writer discipline as `causal_delay`.
|
||
causal_parked: AtomicBool,
|
||
/// Cold lifecycle data. See [`SlotCold`].
|
||
pub(crate) cold: RawMutex<SlotCold>,
|
||
}
|
||
|
||
impl Slot {
|
||
fn vacant() -> Self {
|
||
Self {
|
||
word: StateWord::new(),
|
||
sp: AtomicUsize::new(0),
|
||
stop_ptr: AtomicPtr::new(std::ptr::null_mut()),
|
||
closure: AtomicPtr::new(std::ptr::null_mut()),
|
||
overruns: AtomicU64::new(0),
|
||
messages_received: AtomicU64::new(0),
|
||
budget_cycles: AtomicU64::new(0),
|
||
causal_site: AtomicU32::new(0),
|
||
causal_delay: AtomicU64::new(0),
|
||
causal_parked: AtomicBool::new(true),
|
||
cold: RawMutex::new(SlotCold {
|
||
actor: None,
|
||
waiters: Vec::new(),
|
||
outcome: None,
|
||
supervisor_channel: None,
|
||
monitors: Vec::new(),
|
||
links: Vec::new(),
|
||
outstanding_handles: 0,
|
||
pending_io_result: None,
|
||
}),
|
||
}
|
||
}
|
||
|
||
/// Current generation (of whatever occupies the slot — pair with a
|
||
/// status check or a CAS before acting on it).
|
||
#[inline]
|
||
pub(crate) fn generation(&self) -> u32 {
|
||
self.word.generation()
|
||
}
|
||
|
||
/// Raw packed state word, for introspection's lock-free classify
|
||
/// (`introspect.rs`). The coarse `status_for` only distinguishes
|
||
/// Live/Done/Stale; the snapshot needs the fine scheduling state.
|
||
#[inline]
|
||
pub(crate) fn state_word(&self) -> u64 {
|
||
self.word.load()
|
||
}
|
||
|
||
/// Tally one timeslice overrun (RFC 016 Chunk 2). Single-writer: only the
|
||
/// on-CPU actor's own thread calls this, at the slice-expiry site, so a
|
||
/// Relaxed load+store is sufficient and avoids the cache-line lock of an
|
||
/// atomic RMW.
|
||
#[inline]
|
||
pub(crate) fn record_overrun(&self) {
|
||
let v = self.overruns.load(Ordering::Relaxed);
|
||
self.overruns.store(v.wrapping_add(1), Ordering::Relaxed);
|
||
}
|
||
|
||
/// Read the overrun tally (Relaxed; the snapshot reads cross-thread).
|
||
#[inline]
|
||
pub(crate) fn overruns(&self) -> u64 {
|
||
self.overruns.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// Tally one received (dequeued) message. Same single-writer Relaxed
|
||
/// discipline as `record_overrun`; called by the receiving actor on its own
|
||
/// thread, so no RMW.
|
||
#[inline]
|
||
pub(crate) fn record_message(&self) {
|
||
let v = self.messages_received.load(Ordering::Relaxed);
|
||
self.messages_received.store(v.wrapping_add(1), Ordering::Relaxed);
|
||
}
|
||
|
||
/// Read the received-message tally (Relaxed; cross-thread snapshot read).
|
||
#[inline]
|
||
pub(crate) fn messages_received(&self) -> u64 {
|
||
self.messages_received.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// Accumulate on-CPU cycles consumed in one resume (RFC 016 Chunk 2,
|
||
/// `budget-accounting`). Single-writer (the scheduler thread that ran the
|
||
/// actor), Relaxed load+store. Approximate by design: the figure is
|
||
/// `now − slice-start`, and a wake-slot resume inherits the waker's slice,
|
||
/// so a handed-off actor is charged a little of the chain's time — noise
|
||
/// that averages out across runs, traded for one RDTSC instead of two.
|
||
#[cfg(feature = "budget-accounting")]
|
||
#[inline]
|
||
pub(crate) fn add_budget(&self, cycles: u64) {
|
||
let v = self.budget_cycles.load(Ordering::Relaxed);
|
||
self.budget_cycles.store(v.wrapping_add(cycles), Ordering::Relaxed);
|
||
}
|
||
|
||
/// Read the accumulated budget cycles (Relaxed). Always 0 unless the
|
||
/// `budget-accounting` feature is enabled.
|
||
#[inline]
|
||
pub(crate) fn budget_cycles(&self) -> u64 {
|
||
self.budget_cycles.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// Zero the per-actor introspection counters. Called at every point a slot
|
||
/// is recycled or freshly occupied (`reclaim_slot`, `install_actor`) so a
|
||
/// reused slot never carries a previous incarnation's counts — the standing
|
||
/// slot-lifecycle reset invariant (RFC 016 D7).
|
||
#[inline]
|
||
pub(crate) fn reset_counters(&self) {
|
||
self.overruns.store(0, Ordering::Relaxed);
|
||
self.messages_received.store(0, Ordering::Relaxed);
|
||
self.budget_cycles.store(0, Ordering::Relaxed);
|
||
self.causal_site.store(0, Ordering::Relaxed);
|
||
self.causal_delay.store(0, Ordering::Relaxed);
|
||
// True, not false: the first resume of a fresh actor is a "wake" —
|
||
// it must not owe the process's accumulated delay history.
|
||
self.causal_parked.store(true, Ordering::Relaxed);
|
||
}
|
||
|
||
/// RFC 007 — mark that this actor's deschedule was a genuine park.
|
||
/// Called from the scheduler's `YieldIntent::Park` branch on a successful
|
||
/// `park_return` only.
|
||
#[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
|
||
#[inline]
|
||
pub(crate) fn set_causal_parked(&self) {
|
||
self.causal_parked.store(true, Ordering::Relaxed);
|
||
}
|
||
|
||
/// RFC 007 — consume the parked marker at resume: returns whether the
|
||
/// last deschedule was a real park, and clears it so the next resume
|
||
/// defaults to "was runnable" unless the park branch says otherwise.
|
||
#[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
|
||
#[inline]
|
||
pub(crate) fn take_causal_parked(&self) -> bool {
|
||
self.causal_parked.swap(false, Ordering::Relaxed)
|
||
}
|
||
|
||
/// RFC 007 — current causal site id (0 = none). Single-writer: only the
|
||
/// on-CPU actor's thread writes, via the site-guard enter/exit.
|
||
#[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
|
||
#[inline]
|
||
pub(crate) fn causal_site(&self) -> u32 {
|
||
self.causal_site.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// RFC 007 — write the current causal site id (guard enter/exit).
|
||
#[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
|
||
#[inline]
|
||
pub(crate) fn set_causal_site(&self, id: u32) {
|
||
self.causal_site.store(id, Ordering::Relaxed);
|
||
}
|
||
|
||
/// RFC 007 — absorbed/credited virtual-delay cycles.
|
||
#[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
|
||
#[inline]
|
||
pub(crate) fn causal_delay(&self) -> u64 {
|
||
self.causal_delay.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// RFC 007 — set the absorbed-delay ledger (spin-absorb, credit, or the
|
||
/// resume-path fast-forward). Single-writer per the resume protocol: the
|
||
/// actor's own thread while on-CPU, the resuming scheduler thread at the
|
||
/// resume boundary — never both at once.
|
||
#[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
|
||
#[inline]
|
||
pub(crate) fn set_causal_delay(&self, v: u64) {
|
||
self.causal_delay.store(v, Ordering::Relaxed);
|
||
}
|
||
|
||
/// A pid's-eye snapshot of the slot. Cold paths re-read this under the
|
||
/// cold lock (generation can't change while it is held).
|
||
#[inline]
|
||
pub(crate) fn status_for(&self, pid: Pid) -> Status {
|
||
self.word.status_for(pid.generation())
|
||
}
|
||
|
||
/// Does the slot currently hold the actor `pid` names, in a non-terminal
|
||
/// state? (Snapshot — callers that mutate must re-verify under `cold` or
|
||
/// CAS on the word.)
|
||
#[inline]
|
||
pub(crate) fn is_live_for(&self, pid: Pid) -> bool {
|
||
self.status_for(pid) == Status::Live
|
||
}
|
||
|
||
fn store_closure(&self, c: Closure) {
|
||
let raw = Box::into_raw(Box::new(c));
|
||
let prev = self.closure.swap(raw, Ordering::Release);
|
||
debug_assert!(prev.is_null(), "slot already had a pending closure");
|
||
}
|
||
|
||
fn take_closure(&self) -> Option<Closure> {
|
||
let raw = self.closure.swap(std::ptr::null_mut(), Ordering::Acquire);
|
||
if raw.is_null() {
|
||
None
|
||
} else {
|
||
// SAFETY: non-null values in `closure` are exclusively
|
||
// `Box::into_raw(Box<Closure>)` from `store_closure`, and the
|
||
// swap above made us the unique owner.
|
||
Some(*unsafe { Box::from_raw(raw) })
|
||
}
|
||
}
|
||
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// RuntimeInner — the shared core behind an Arc
|
||
// ---------------------------------------------------------------------------
|
||
|
||
pub(crate) struct RuntimeInner {
|
||
/// The run queue, compile-time selected (see `run_queue.rs` for the
|
||
/// contract: ops require preemption disabled, push is infallible,
|
||
/// pop-None is a snapshot).
|
||
pub(crate) run_queue: crate::run_queue::RunQueue,
|
||
/// The fixed actor slot table. Allocated once; slots never move.
|
||
pub(crate) slots: Box<[Slot]>,
|
||
/// Vacant slot indices. RawMutex leaf; never held with a cold lock.
|
||
pub(crate) free: RawMutex<Vec<u32>>,
|
||
/// Spawned-but-not-finalized actor count; the termination criterion.
|
||
/// Incremented in `spawn` before the enqueue; decremented at the very end
|
||
/// of `finalize_actor`, after every wakeup finalize produces.
|
||
pub(crate) live_actors: AtomicU32,
|
||
/// Packed `(index << 32 | generation)` of the run's root (initial) actor,
|
||
/// or `u64::MAX` (the ROOT_PID sentinel) before one is set. When this actor
|
||
/// finalizes it flags `root_exited`; the scheduler's idle verdict then
|
||
/// stops the remaining (parked-forever) actors. Set once per `run()`, right
|
||
/// after the initial spawn.
|
||
pub(crate) root_bits: AtomicU64,
|
||
/// Set when the root actor finalizes; read by the scheduler's idle verdict
|
||
/// to trigger the one-shot teardown sweep. Reset per `run()`.
|
||
pub(crate) root_exited: AtomicBool,
|
||
/// Guards the teardown sweep to fire at most once per run (a parked-forever
|
||
/// remainder that survives the sweep falls through to the normal idle wait
|
||
/// rather than busy-spinning). Reset per `run()`.
|
||
pub(crate) root_swept: AtomicBool,
|
||
/// Timer heap. Independent lock: never nested with any other.
|
||
pub(crate) timers: Mutex<Timers>,
|
||
/// IO subsystem. `None` between runs. Lock order: io before everything.
|
||
pub(crate) io: Mutex<Option<IoThread>>,
|
||
/// Monotonic `MonitorId` source. Never reused.
|
||
pub(crate) next_monitor_id: AtomicU64,
|
||
/// Try-lock: exactly one scheduler thread drains timers/IO per iteration.
|
||
drain_lock: Mutex<()>,
|
||
/// Per-thread stats, indexed by scheduler thread slot (0..N).
|
||
pub(crate) stats: Vec<SchedulerStats>,
|
||
/// Global counters for RFC 000 primitives.
|
||
pub(crate) io_parked: AtomicU32,
|
||
pub(crate) sleeping: AtomicU32,
|
||
/// Preemption knobs, written into each scheduler thread's locals on startup.
|
||
pub(crate) alloc_interval: u32,
|
||
pub(crate) timeslice_cycles: u64,
|
||
/// RFC 005: whether actor-context wakes route through the per-scheduler
|
||
/// wake slot. Read-only after init; one predictable branch per wake.
|
||
pub(crate) wake_slot: bool,
|
||
/// The name <-> pid registry (bidirectional). RawMutex Leaf: never held
|
||
/// with any other lock; liveness checks under it read only the atomic
|
||
/// slot word.
|
||
pub(crate) registry: RawMutex<crate::registry::Registry>,
|
||
/// Runtime identity (RFC 012). Read-only after init — one node, one fixed
|
||
/// incarnation until clustering (RFC 010) supplies real values. Carried
|
||
/// like `wake_slot`; pg fills these into every `Member` so the public
|
||
/// surface stays Pid-shaped.
|
||
pub(crate) node_id: crate::pg::NodeId,
|
||
pub(crate) incarnation: crate::pg::Incarnation,
|
||
/// Process groups: `name -> multiset<Member>` (RFC 012). RawMutex Leaf,
|
||
/// exactly like `registry`: never held with any other lock; liveness
|
||
/// checks under it read only the atomic slot word, and the eviction path
|
||
/// keeps it off the send path.
|
||
pub(crate) process_groups: RawMutex<crate::pg::ProcessGroups>,
|
||
/// Recycled stacks waiting to be reused by the next spawn.
|
||
pub(crate) stack_pool: RawMutex<Vec<crate::stack::Stack>>,
|
||
/// Maximum number of stacks to retain in the pool.
|
||
pub(crate) stack_pool_cap: usize,
|
||
}
|
||
|
||
impl RuntimeInner {
|
||
// Private constructor taking the parsed Config fields one-for-one; a params
|
||
// struct would only move the same 8 values across the call boundary.
|
||
#[allow(clippy::too_many_arguments)]
|
||
fn new(
|
||
thread_count: usize,
|
||
alloc_interval: u32,
|
||
timeslice_cycles: u64,
|
||
stack_pool_cap: usize,
|
||
max_actors: usize,
|
||
wake_slot: bool,
|
||
node_id: crate::pg::NodeId,
|
||
incarnation: crate::pg::Incarnation,
|
||
) -> Arc<Self> {
|
||
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
|
||
let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
|
||
// Low indices on top of the stack so early spawns get low pids.
|
||
let free: Vec<u32> = (0..max_actors as u32).rev().collect();
|
||
Arc::new(Self {
|
||
run_queue: crate::run_queue::RunQueue::new(thread_count, max_actors),
|
||
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),
|
||
drain_lock: Mutex::new(()),
|
||
stats,
|
||
io_parked: AtomicU32::new(0),
|
||
sleeping: AtomicU32::new(0),
|
||
alloc_interval,
|
||
timeslice_cycles,
|
||
wake_slot,
|
||
registry: RawMutex::new(crate::registry::Registry::new()),
|
||
node_id,
|
||
incarnation,
|
||
process_groups: RawMutex::new(crate::pg::ProcessGroups::new()),
|
||
stack_pool: RawMutex::new(Vec::new()),
|
||
stack_pool_cap,
|
||
})
|
||
}
|
||
|
||
/// Slot lookup by index only — bounds-checked, NOT generation-checked.
|
||
/// `ROOT_PID` (index `u32::MAX`) is out of bounds by construction and
|
||
/// resolves to `None`. Callers verify the generation atomically: either
|
||
/// inside a CAS on the word, or by re-reading the word under the cold lock.
|
||
#[inline]
|
||
pub(crate) fn slot_at(&self, pid: Pid) -> Option<&Slot> {
|
||
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).
|
||
pub(crate) fn enqueue(&self, pid: Pid) {
|
||
// Every push pairs 1:1 with a transition INTO Queued, and nothing can
|
||
// move the word off Queued until this very entry is popped — so the
|
||
// word must read EXACTLY (gen, Queued) here. This is the at-most-once-
|
||
// enqueued invariant the bounded rings' capacity proof leans on.
|
||
debug_assert!(
|
||
self.slot_at(pid).map(|s| s.word.load()).is_some_and(|w| {
|
||
crate::slot_state::word_gen(w) == pid.generation()
|
||
&& crate::slot_state::word_state(w) == crate::slot_state::ST_QUEUED
|
||
}),
|
||
"enqueue of a pid not in (gen, Queued)"
|
||
);
|
||
self.run_queue.push(pid);
|
||
crate::te!(crate::trace::Event::Enqueue(pid));
|
||
}
|
||
|
||
/// Make `pid` runnable if it is parked; coalesce or defer otherwise.
|
||
/// The runtime-internal core of `scheduler::unpark`. WILDCARD wake:
|
||
/// consumes the epoch but does not check it — reserved for terminal
|
||
/// wakes (`request_stop`); see slot_state.rs.
|
||
pub(crate) fn unpark(&self, pid: Pid) {
|
||
self.unpark_inner(pid, None);
|
||
}
|
||
|
||
/// Epoch-matched wake: lands only if `pid`'s current wait is still the
|
||
/// one the waker registered for. The form every registration-based waker
|
||
/// (channel senders, mutex grants, wait-timers, …) must use.
|
||
pub(crate) fn unpark_at(&self, pid: Pid, epoch: u32) {
|
||
self.unpark_inner(pid, Some(epoch));
|
||
}
|
||
|
||
/// Open a new wait for `pid` (the calling actor itself): bump its
|
||
/// park-epoch and return it. See slot_state.rs for the rules.
|
||
#[must_use]
|
||
pub(crate) fn begin_wait(&self, pid: Pid) -> u32 {
|
||
let slot = match self.slot_at(pid) {
|
||
Some(slot) => slot,
|
||
None => panic!("begin_wait: own slot vanished: {:?}", pid),
|
||
};
|
||
slot.word.begin_wait(pid.generation())
|
||
}
|
||
|
||
/// Retire the calling actor's current wait without parking on it: bump
|
||
/// the epoch (invalidating every in-flight registration-based wake),
|
||
/// then eat a notification that already landed. The caller MUST
|
||
/// re-check its stop flag afterwards — see `StateWord::clear_notify`.
|
||
pub(crate) fn retire_wait(&self, pid: Pid) {
|
||
let slot = match self.slot_at(pid) {
|
||
Some(slot) => slot,
|
||
None => panic!("retire_wait: own slot vanished: {:?}", pid),
|
||
};
|
||
let _ = slot.word.begin_wait(pid.generation());
|
||
slot.word.clear_notify(pid.generation());
|
||
}
|
||
|
||
fn unpark_inner(&self, pid: Pid, want: Option<u32>) {
|
||
if let Some(slot) = self.slot_at(pid) {
|
||
match slot.word.unpark(pid.generation(), want) {
|
||
Unpark::Enqueue => {
|
||
crate::te!(crate::trace::Event::UnparkDirect(pid));
|
||
// RFC 005: a wake from ACTOR context is slot-eligible —
|
||
// the woken actor's message bytes are hot in this core's
|
||
// cache. Scheduler-context wakes (timer/IO drain, where
|
||
// current_pid is None) have no locality to exploit,
|
||
// arrive in bursts that would thrash the slot, and
|
||
// concentrate on the drain winner by construction:
|
||
// shared queue, always. Spawns never come through here
|
||
// (install_actor enqueues directly), so they bypass the
|
||
// slot by construction.
|
||
if self.wake_slot && crate::actor::current_pid().is_some() {
|
||
self.slot_push(pid);
|
||
} else {
|
||
self.enqueue(pid);
|
||
}
|
||
}
|
||
Unpark::Notified => {
|
||
crate::te!(crate::trace::Event::UnparkDeferred(pid));
|
||
}
|
||
Unpark::Noop => {}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// RFC 005: park `pid` in this thread's wake slot instead of the shared
|
||
/// queue. Replaces `enqueue` at the tail of the wake protocol's
|
||
/// `Parked → Queued` CAS, so the caller has just transitioned the pid
|
||
/// into `Queued` — same precondition, same invariant, different home.
|
||
/// Displacement: the NEW wake takes the slot (newest is hottest; the old
|
||
/// occupant was about to lose its locality window anyway) and the old
|
||
/// occupant is pushed to the shared queue.
|
||
fn slot_push(&self, pid: Pid) {
|
||
debug_assert!(
|
||
!crate::preempt::PREEMPTION_ENABLED.with(|c| c.get()),
|
||
"slot_push with preemption enabled — a switch mid-op could \
|
||
migrate the actor and split the slot access across threads"
|
||
);
|
||
debug_assert!(
|
||
self.slot_at(pid).map(|s| s.word.load()).is_some_and(|w| {
|
||
crate::slot_state::word_gen(w) == pid.generation()
|
||
&& crate::slot_state::word_state(w) == crate::slot_state::ST_QUEUED
|
||
}),
|
||
"slot_push of a pid not in (gen, Queued)"
|
||
);
|
||
let displaced = WAKE_SLOT.with(|s| s.replace(Some(pid)));
|
||
crate::te!(crate::trace::Event::SlotPush(pid));
|
||
if let Some(old) = displaced {
|
||
SCHED_SLOT.with(|s| {
|
||
self.stats[s.get()]
|
||
.slot_displacements
|
||
.fetch_add(1, Ordering::Relaxed)
|
||
});
|
||
self.enqueue(old);
|
||
}
|
||
}
|
||
|
||
/// Allocate the next process-unique `MonitorId`. Lock-free; monitors are a
|
||
/// cold path but there is no reason to serialize id minting under any lock.
|
||
pub(crate) fn alloc_monitor_id(&self) -> MonitorId {
|
||
MonitorId(self.next_monitor_id.fetch_add(1, Ordering::Relaxed) + 1)
|
||
}
|
||
|
||
/// Pop a vacant slot index, or die loudly. The fixed slab is a deliberate
|
||
/// v0.5 simplification (ROADMAP: "Deferred"); the panic names the fix.
|
||
pub(crate) fn allocate_slot(&self) -> u32 {
|
||
match self.free.lock().pop() {
|
||
Some(idx) => idx,
|
||
None => panic!(
|
||
"smarm: actor slot table exhausted — {} actors are live \
|
||
simultaneously, which is the configured maximum. \
|
||
Fix: raise the cap at runtime init, e.g. \
|
||
`smarm::init(Config::default().max_actors({}))`. \
|
||
(Slots are ~256 bytes each; the table is allocated up-front.)",
|
||
self.slots.len(),
|
||
self.slots.len() * 2
|
||
),
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Runtime — the public handle
|
||
// ---------------------------------------------------------------------------
|
||
|
||
pub struct Runtime {
|
||
inner: Arc<RuntimeInner>,
|
||
thread_count: usize,
|
||
}
|
||
|
||
/// Initialise the runtime with the given config. Returns a reusable handle.
|
||
pub fn init(config: Config) -> Runtime {
|
||
let n = config.resolved_thread_count();
|
||
Runtime {
|
||
inner: RuntimeInner::new(
|
||
n,
|
||
config.alloc_interval,
|
||
config.timeslice_cycles,
|
||
config.stack_pool_cap,
|
||
config.max_actors,
|
||
config.wake_slot,
|
||
config.node_id,
|
||
config.incarnation,
|
||
),
|
||
thread_count: n,
|
||
}
|
||
}
|
||
|
||
impl Runtime {
|
||
/// Run `f` as the initial actor, block until all actors finish.
|
||
/// Can be called multiple times sequentially on the same `Runtime`.
|
||
pub fn run(&self, f: impl FnOnce() + Send + 'static) {
|
||
// Install smarm's panic hook on first call. The default Rust hook is
|
||
// not reentrant — concurrent actor panics can trigger a double-panic
|
||
// abort when the backtrace printer takes an internal lock that is
|
||
// already held. smarm catches every actor panic via `catch_unwind` in
|
||
// the trampoline, so panics never need to reach the hook for runtime
|
||
// correctness; the hook fires only as a side-effect of unwinding before
|
||
// `catch_unwind` catches it.
|
||
//
|
||
// We install once and leave it installed: the previous hook is chained
|
||
// so that panics outside actor context (e.g. in the test harness
|
||
// itself) are still reported normally.
|
||
static HOOK_INSTALLED: std::sync::OnceLock<()> = std::sync::OnceLock::new();
|
||
HOOK_INSTALLED.get_or_init(|| {
|
||
let prev = std::panic::take_hook();
|
||
std::panic::set_hook(Box::new(move |info| {
|
||
// If we are currently executing inside an actor trampoline the
|
||
// panic will be caught by `catch_unwind` momentarily. Suppress
|
||
// the hook output to avoid interleaved noise and reentrancy.
|
||
// Outside actor context, delegate to the previous hook so that
|
||
// genuine runtime panics are still reported.
|
||
if crate::actor::current_pid().is_some() {
|
||
// Inside an actor — catch_unwind handles it; stay silent.
|
||
} else {
|
||
prev(info);
|
||
}
|
||
}));
|
||
});
|
||
|
||
// Open the trace store for this run (no-op without smarm-trace).
|
||
#[cfg(feature = "smarm-trace")]
|
||
crate::trace::open();
|
||
|
||
// Re-initialise shared state for this run.
|
||
assert_eq!(
|
||
self.inner.run_queue.len(), 0,
|
||
"run() called while previous run still active"
|
||
);
|
||
debug_assert_eq!(
|
||
self.inner.live_actors.load(Ordering::Acquire), 0,
|
||
"run() called while previous run still active"
|
||
);
|
||
let io_thread = match IoThread::start() {
|
||
Ok(io) => io,
|
||
Err(e) => panic!("failed to start IO thread: {e}"),
|
||
};
|
||
match self.inner.io.lock() {
|
||
Ok(mut io) => *io = Some(io_thread),
|
||
Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
|
||
}
|
||
|
||
// RFC 005: slot counters reset at the START of a run (not the end),
|
||
// so `stats()` read after `run()` returns reports that run's totals.
|
||
for stat in &self.inner.stats {
|
||
stat.slot_hits.store(0, Ordering::Relaxed);
|
||
stat.slot_displacements.store(0, Ordering::Relaxed);
|
||
}
|
||
|
||
// Spawn the initial actor through the public spawn path (which
|
||
// 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, named `smarm-sched-{slot}` so
|
||
// they are identifiable in `/proc/<pid>/task/*/comm`, stack dumps and
|
||
// debuggers. The calling thread is thread 0 and keeps its caller-given
|
||
// name (an embedder typically names it when spawning `run`).
|
||
let mut os_threads = Vec::new();
|
||
for slot in 1..self.thread_count {
|
||
let inner = self.inner.clone();
|
||
let t = thread::Builder::new()
|
||
.name(format!("smarm-sched-{slot}"))
|
||
.spawn(move || {
|
||
RUNTIME.with(|r| *r.borrow_mut() = Some(inner.clone()));
|
||
SCHED_SLOT.with(|s| s.set(slot));
|
||
schedule_loop(&inner, slot);
|
||
RUNTIME.with(|r| *r.borrow_mut() = None);
|
||
});
|
||
// `thread::spawn` (the previous form) also panics when the OS
|
||
// refuses a thread, so this keeps the failure semantics.
|
||
let t = match t {
|
||
Ok(t) => t,
|
||
Err(e) => panic!("failed to spawn smarm scheduler thread: {e}"),
|
||
};
|
||
os_threads.push(t);
|
||
}
|
||
|
||
// Thread 0 runs the loop on the calling thread.
|
||
SCHED_SLOT.with(|s| s.set(0));
|
||
schedule_loop(&self.inner, 0);
|
||
|
||
// Wait for all other scheduler threads.
|
||
for t in os_threads {
|
||
let _ = t.join();
|
||
}
|
||
|
||
// Drop initial handle (decrements outstanding_handles count).
|
||
drop(initial_handle);
|
||
|
||
// Tear down IO and clean up for the next run() call.
|
||
match self.inner.io.lock() {
|
||
Ok(mut io) => drop(io.take()), // joins IO threads
|
||
Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
|
||
}
|
||
match self.inner.timers.lock() {
|
||
Ok(mut timers) => timers.clear(),
|
||
Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
|
||
}
|
||
self.inner.next_monitor_id.store(0, Ordering::Relaxed);
|
||
// Every slot must have come back: any leak here is a runtime bug
|
||
// (a JoinHandle held across run() is decremented just above).
|
||
debug_assert_eq!(
|
||
self.inner.free.lock().len(),
|
||
self.inner.slots.len(),
|
||
"slot leak across run()"
|
||
);
|
||
// Reset per-thread stats.
|
||
for stat in &self.inner.stats {
|
||
stat.current_pid_index.store(u32::MAX, Ordering::Relaxed);
|
||
stat.run_queue_len.store(0, Ordering::Relaxed);
|
||
}
|
||
self.inner.io_parked.store(0, Ordering::Relaxed);
|
||
self.inner.sleeping.store(0, Ordering::Relaxed);
|
||
|
||
RUNTIME.with(|r| *r.borrow_mut() = None);
|
||
|
||
// Flush trace to disk (no-op without smarm-trace).
|
||
#[cfg(feature = "smarm-trace")]
|
||
crate::trace::flush();
|
||
}
|
||
|
||
/// Snapshot of runtime statistics for introspection / tests.
|
||
pub fn stats(&self) -> RuntimeStats {
|
||
RuntimeStats { inner: self.inner.clone() }
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Thread-locals
|
||
// ---------------------------------------------------------------------------
|
||
|
||
use std::cell::{Cell, RefCell};
|
||
|
||
thread_local! {
|
||
/// The RuntimeInner for the current run(). Set by run() on the calling
|
||
/// thread and by each spawned scheduler thread.
|
||
pub(crate) static RUNTIME: RefCell<Option<Arc<RuntimeInner>>> =
|
||
const { RefCell::new(None) };
|
||
|
||
/// This scheduler thread's index into RuntimeInner::stats.
|
||
static SCHED_SLOT: Cell<usize> = const { Cell::new(0) };
|
||
|
||
/// RFC 005: the per-scheduler wake slot — a capacity-one wake cache
|
||
/// checked before the shared run queue. Holds a pid in state
|
||
/// `(gen, Queued)` exactly as a shared-queue entry would; the
|
||
/// at-most-once-enqueued invariant reads "in (slot ⊕ shared queue) at
|
||
/// most once". All access is from the owning thread with preemption
|
||
/// disabled (the existing queue-op contract), so plain Cell ops suffice:
|
||
/// no atomics, nothing to steal, nothing to model. Empty whenever the
|
||
/// thread reaches the idle or termination path (pop order drains it
|
||
/// first), so it never holds a pid across the end of a run.
|
||
static WAKE_SLOT: Cell<Option<Pid>> = const { Cell::new(None) };
|
||
|
||
/// What the actor wants when it yields back to the scheduler.
|
||
static YIELD_INTENT: Cell<YieldIntent> = const { Cell::new(YieldIntent::Yield) };
|
||
}
|
||
|
||
#[derive(Copy, Clone)]
|
||
pub(crate) enum YieldIntent { Yield, Park }
|
||
|
||
pub(crate) fn set_yield_intent(i: YieldIntent) {
|
||
YIELD_INTENT.with(|c| c.set(i));
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Sentinel root PID
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Index `u32::MAX` is out of bounds for any slab (Config asserts
|
||
/// `max_actors < u32::MAX`), so every slot lookup on ROOT_PID resolves to
|
||
/// `None` — the root "actor" silently absorbs supervisor signals.
|
||
pub const ROOT_PID: Pid = Pid::new(u32::MAX, u32::MAX);
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Spawn-side slot installation
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Install a freshly spawned actor into the slot `idx` (which must have come
|
||
/// from `allocate_slot`) and publish it as Queued. Returns the new `Pid`.
|
||
/// Called by `scheduler::spawn_under`; lives here next to its inverse
|
||
/// (`reclaim_slot`) so the lifecycle is in one file.
|
||
pub(crate) fn install_actor(
|
||
inner: &RuntimeInner,
|
||
idx: u32,
|
||
sp: usize,
|
||
stack: crate::stack::Stack,
|
||
supervisor: Pid,
|
||
closure: Closure,
|
||
) -> Pid {
|
||
let slot = &inner.slots[idx as usize];
|
||
let gen = slot.generation(); // stable: we own the vacant slot via the free list
|
||
let pid = Pid::new(idx, gen);
|
||
|
||
let stop = Arc::new(AtomicBool::new(false));
|
||
slot.stop_ptr.store(Arc::as_ptr(&stop) as *mut _, Ordering::Release);
|
||
{
|
||
let mut cold = slot.cold.lock();
|
||
debug_assert!(cold.actor.is_none(), "install over live actor");
|
||
debug_assert!(cold.waiters.is_empty() && cold.monitors.is_empty() && cold.links.is_empty());
|
||
cold.actor = Some(Actor { pid, stack, supervisor, stop, trap: None });
|
||
cold.outstanding_handles = 1;
|
||
cold.outcome = None;
|
||
cold.pending_io_result = None;
|
||
}
|
||
slot.sp.store(sp, Ordering::Relaxed);
|
||
slot.store_closure(closure);
|
||
slot.reset_counters();
|
||
inner.live_actors.fetch_add(1, Ordering::Relaxed);
|
||
|
||
// Publish: only now can pops, unparks, or stops find the actor. The
|
||
// Release store orders everything above before any Acquire reader.
|
||
slot.word.publish_queued(gen);
|
||
inner.enqueue(pid);
|
||
crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid });
|
||
pid
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Slot reclamation
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/// Reclaim `pid`'s slot if (still) eligible: generation matches, state is
|
||
/// Done, and no handles are outstanding. Safe to call from racing sites
|
||
/// (finalize tail vs. JoinHandle drop): the first caller bumps the
|
||
/// generation under the cold lock, the loser sees the mismatch and no-ops.
|
||
///
|
||
/// Channel senders extracted from the slot are dropped *after* the cold lock
|
||
/// is released — a last-sender drop can unpark a receiver, which takes the
|
||
/// run-queue mutex; legal under a cold lock, but pointless to nest.
|
||
pub(crate) fn reclaim_slot(inner: &RuntimeInner, pid: Pid) {
|
||
let Some(slot) = inner.slot_at(pid) else { return };
|
||
let dropped_outside;
|
||
{
|
||
let mut cold = slot.cold.lock();
|
||
if slot.status_for(pid) != Status::Done || cold.outstanding_handles != 0 {
|
||
return; // already reclaimed, or not yet eligible
|
||
}
|
||
debug_assert!(cold.actor.is_none(), "reclaiming a slot that still owns an actor");
|
||
dropped_outside = (
|
||
cold.outcome.take(),
|
||
cold.supervisor_channel.take(),
|
||
cold.pending_io_result.take(),
|
||
slot.take_closure(), // an actor stopped before first resume
|
||
);
|
||
cold.waiters.clear();
|
||
cold.monitors.clear();
|
||
cold.links.clear();
|
||
slot.reset_counters();
|
||
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
||
// The generation bump IS the reclaim: every stale pid is dead from
|
||
// this store onwards (unpark protocol, pops, cold-path re-verifies).
|
||
slot.word.reclaim(pid.generation());
|
||
}
|
||
drop(dropped_outside);
|
||
inner.free.lock().push(pid.index());
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// finalize_actor
|
||
// ---------------------------------------------------------------------------
|
||
|
||
fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
|
||
let (joiner_outcome, sup_signal, down_reason) = match outcome {
|
||
Outcome::Exit => (Outcome::Exit, Signal::Exit(pid), DownReason::Exit),
|
||
Outcome::Panic(payload) => (
|
||
Outcome::Panic(payload),
|
||
Signal::Panic(pid, Box::new(()) as Box<dyn std::any::Any + Send>),
|
||
DownReason::Panic,
|
||
),
|
||
// Cooperative cancellation: kept distinct from a normal Exit so a
|
||
// supervisor's await logic (roadmap #2) can tell "I stopped it" apart
|
||
// from "it finished on its own".
|
||
Outcome::Stopped => (Outcome::Stopped, Signal::Stopped(pid), DownReason::Stopped),
|
||
};
|
||
|
||
let slot = match inner.slot_at(pid) {
|
||
Some(slot) => slot,
|
||
None => panic!("finalize_actor: pid out of range: {:?}", pid),
|
||
};
|
||
let (waiters, monitors, links, actor) = {
|
||
let mut cold = slot.cold.lock();
|
||
let actor = match cold.actor.take() {
|
||
Some(actor) => actor,
|
||
None => panic!("finalize_actor: actor vanished"),
|
||
};
|
||
cold.outcome = Some(joiner_outcome);
|
||
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
||
// Done is published under the cold lock, so join's
|
||
// check-Done-or-register-waiter (also under it) can never miss: it
|
||
// either sees Done and takes the outcome, or its waiter registration
|
||
// happens before our take() below and is woken further down.
|
||
// (set_done self-asserts the Running|Notified precondition + gen.)
|
||
slot.word.set_done(pid.generation());
|
||
(
|
||
std::mem::take(&mut cold.waiters),
|
||
std::mem::take(&mut cold.monitors),
|
||
std::mem::take(&mut cold.links),
|
||
actor,
|
||
)
|
||
};
|
||
|
||
// Recycle the stack outside the cold lock; drop the rest of the Actor
|
||
// (the trap sender can unpark its receiver — keep that outside too).
|
||
let supervisor_pid = actor.supervisor;
|
||
let Actor { stack, .. } = actor;
|
||
{
|
||
let mut pool = inner.stack_pool.lock();
|
||
if pool.len() < inner.stack_pool_cap {
|
||
pool.push(stack);
|
||
}
|
||
// else: drop here → munmap, same as before
|
||
}
|
||
|
||
// Deliver to supervisor. ROOT_PID resolves to no slot → silently absorbed.
|
||
let sender = inner.slot_at(supervisor_pid).and_then(|sup| {
|
||
let cold = sup.cold.lock();
|
||
if sup.generation() == supervisor_pid.generation() {
|
||
cold.supervisor_channel.clone()
|
||
} else {
|
||
None
|
||
}
|
||
});
|
||
if let Some(sender) = sender {
|
||
let _ = sender.send(sup_signal);
|
||
}
|
||
|
||
// Notify monitors. Sent outside any slot lock: `send` may unpark a parked
|
||
// receiver, which takes the run-queue mutex.
|
||
for (_, m) in monitors {
|
||
let _ = m.send(Down { pid, reason: down_reason });
|
||
}
|
||
|
||
// Walk linked peers ONE AT A TIME (cold locks are leaves). For every
|
||
// peer: remove the back-link to this (dying) actor; on abnormal death,
|
||
// also fetch its trap sender and deliver after unlocking.
|
||
//
|
||
// Acyclicity: the back-link removal happens under the peer's cold lock
|
||
// *before* any stop is delivered to it, so when the peer later dies its
|
||
// own cascade no longer contains us. Two peers finalizing concurrently
|
||
// each find the other already Done (set above, before any cascade) and
|
||
// skip — no ping-pong, no deadlock (never two cold locks held).
|
||
let abnormal = matches!(down_reason, DownReason::Panic | DownReason::Stopped);
|
||
for peer in links {
|
||
let trap = match inner.slot_at(peer) {
|
||
Some(ps) => {
|
||
let mut cold = ps.cold.lock();
|
||
if ps.status_for(peer) == Status::Live {
|
||
cold.links.retain(|p| *p != pid);
|
||
if abnormal {
|
||
Some(cold.actor.as_ref().and_then(|a| a.trap.clone()))
|
||
} else {
|
||
None // normal exit never propagates
|
||
}
|
||
} else {
|
||
None // peer already gone; nothing to do
|
||
}
|
||
}
|
||
None => None,
|
||
};
|
||
match trap {
|
||
Some(Some(tx)) => {
|
||
let _ = tx.send(crate::link::ExitSignal { from: pid, reason: down_reason });
|
||
}
|
||
Some(None) => crate::scheduler::request_stop(peer),
|
||
None => {}
|
||
}
|
||
}
|
||
|
||
// Unpark joiners (epoch-matched: each registered under the cold lock).
|
||
for (joiner, epoch) in waiters {
|
||
inner.unpark_at(joiner, epoch);
|
||
}
|
||
|
||
// 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
|
||
// the module docs.
|
||
let prev = inner.live_actors.fetch_sub(1, Ordering::Release);
|
||
debug_assert!(prev >= 1, "live_actors underflow — double finalize");
|
||
}
|
||
|
||
/// Cooperatively stop every live actor — the root-exit teardown sweep, run from
|
||
/// `schedule_loop` once the run queue is empty after the root has exited. Each
|
||
/// [`request_stop_inner`](crate::scheduler::request_stop_inner) re-verifies the
|
||
/// target under its cold lock, so the racy per-slot generation read is safe: a
|
||
/// vacant, dead, or reused slot no-ops. The swept actors unpark, unwind at their
|
||
/// next observation point, and finalize, dropping `live_actors` to zero.
|
||
fn stop_live_actors(inner: &Arc<RuntimeInner>) {
|
||
for idx in 0..inner.slots.len() as u32 {
|
||
let pid = Pid::new(idx, inner.slots[idx as usize].generation());
|
||
crate::scheduler::request_stop_inner(inner, pid);
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// schedule_loop — runs on each scheduler OS thread
|
||
// ---------------------------------------------------------------------------
|
||
|
||
fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
||
crate::preempt::configure_preempt(inner.alloc_interval, inner.timeslice_cycles);
|
||
let stats = &inner.stats[slot_idx];
|
||
|
||
loop {
|
||
// ----------------------------------------------------------------
|
||
// 1. Try to win the drain lock (timers + IO). One winner per round;
|
||
// losers skip immediately and proceed to step 2.
|
||
// ----------------------------------------------------------------
|
||
if let Ok(_drain_guard) = inner.drain_lock.try_lock() {
|
||
// Timers and IO live behind their own mutexes (phase 1), so the
|
||
// pure-yield / pure-compute hot path never contends a global lock
|
||
// just to discover there is nothing to drain. The clock is read
|
||
// only when the timer heap is non-empty.
|
||
let due = {
|
||
let mut t = match inner.timers.lock() {
|
||
Ok(t) => t,
|
||
Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
|
||
};
|
||
if t.is_empty() {
|
||
Vec::new()
|
||
} else {
|
||
t.pop_due(std::time::Instant::now())
|
||
}
|
||
};
|
||
let completions = match inner.io.lock() {
|
||
Ok(mut io) => io
|
||
.as_mut()
|
||
.map(|io| {
|
||
// Consume wake-pipe bytes ONLY here, under the drain
|
||
// lock and strictly before draining completions.
|
||
// Producers push their completion before writing the
|
||
// byte, so every byte consumed here has its completion
|
||
// visible to the drain below. Consuming bytes anywhere
|
||
// else — in particular after an idle poll, outside the
|
||
// lock — loses wakeups: a try_lock loser can eat the
|
||
// byte for a completion the winner never saw, leaving
|
||
// it stranded (and its EPOLLONESHOT fd disarmed) until
|
||
// an unrelated timer forces another drain pass.
|
||
crate::io::drain_wake_pipe(io.wake_fd());
|
||
io.drain_completions()
|
||
})
|
||
.unwrap_or_default(),
|
||
Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
|
||
};
|
||
for entry in due {
|
||
match entry.reason {
|
||
// A sleep expiry is just an unpark: the protocol handles
|
||
// every interleaving — Parked (re-queue), Running (the
|
||
// actor is between `timers.insert_sleep` and
|
||
// `park_current`; RunningNotified makes the upcoming park
|
||
// re-queue), or gone (no-op).
|
||
crate::timer::Reason::Sleep { epoch } => {
|
||
inner.unpark_at(entry.pid, epoch)
|
||
}
|
||
crate::timer::Reason::WaitTimeout { target, epoch } => {
|
||
// The callback may call unpark_at itself.
|
||
target.on_timeout(entry.pid, epoch);
|
||
}
|
||
// A `send_after` deadline: run the captured delivery thunk.
|
||
// It resolves the destination through the registry and
|
||
// sends now (a send can unpark a receiver) — same as any
|
||
// other in-loop unpark. The timers lock is already
|
||
// released; lock order Leaf -> Channel is preserved by the
|
||
// send itself. `pop_due` only returns still-armed Sends, so
|
||
// a cancelled one never reaches here.
|
||
crate::timer::Reason::Send { fire } => fire(),
|
||
}
|
||
}
|
||
|
||
for completion in completions {
|
||
match completion {
|
||
crate::io::Completion::Blocking { pid, epoch, result } => {
|
||
match inner.io.lock() {
|
||
Ok(mut io) => {
|
||
if let Some(io) = io.as_mut() {
|
||
io.outstanding = io.outstanding.saturating_sub(1);
|
||
}
|
||
}
|
||
Err(e) => {
|
||
panic!("smarm: io lock poisoned (core corrupt): {e}")
|
||
}
|
||
}
|
||
// Stash the result under the cold lock, then unpark.
|
||
// The protocol also covers the submit→park window
|
||
// (RunningNotified), which the old code missed for
|
||
// Blocking completions — a latent lost wakeup.
|
||
if let Some(slot) = inner.slot_at(pid) {
|
||
{
|
||
let mut cold = slot.cold.lock();
|
||
if slot.generation() == pid.generation() {
|
||
cold.pending_io_result = Some(result);
|
||
} else {
|
||
// Actor died (stopped) with the op in
|
||
// flight; discard the result.
|
||
}
|
||
}
|
||
inner.unpark_at(pid, epoch);
|
||
}
|
||
}
|
||
crate::io::Completion::FdReady { fd, events: _ } => {
|
||
// Resolve the parked pid under the io lock, then wake
|
||
// through the protocol. Lock order: io before all.
|
||
let parked = match inner.io.lock() {
|
||
Ok(mut io) => io.as_mut().and_then(|io| {
|
||
let entry = io.waiters.remove(&fd);
|
||
io.epoll_deregister(fd);
|
||
entry
|
||
}),
|
||
Err(e) => {
|
||
panic!("smarm: io lock poisoned (core corrupt): {e}")
|
||
}
|
||
};
|
||
if let Some((pid, epoch)) = parked {
|
||
inner.unpark_at(pid, epoch);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
} // drain_guard drops here
|
||
|
||
// ----------------------------------------------------------------
|
||
// 2. Pop a runnable pid. Pop order (RFC 005): wake slot first, then
|
||
// shared queue. The queue mutex covers ONLY the pop; the slot's
|
||
// own atomics carry everything needed to resume.
|
||
// ----------------------------------------------------------------
|
||
enum Pop {
|
||
Got(Pid),
|
||
Idle { io_outstanding: u32, wake_fd: Option<std::os::fd::RawFd> },
|
||
AllDone,
|
||
/// Root has exited and nothing is runnable: stop the parked-forever
|
||
/// remainder, then re-pop. Fires at most once per run.
|
||
RootDrain,
|
||
}
|
||
|
||
// 2a. RFC 005: drain this thread's wake slot before touching the
|
||
// shared queue. Two consequences fall out of slot-first order:
|
||
// the idle path below is only reachable with an empty slot, and so
|
||
// is AllDone — an occupied slot on ANOTHER thread holds a Queued
|
||
// (hence live) actor, so `live_actors > 0` and termination cannot
|
||
// fire; the counter-first argument is untouched.
|
||
let slot_pid = if inner.wake_slot {
|
||
WAKE_SLOT.with(|s| s.take())
|
||
} else {
|
||
None
|
||
};
|
||
let from_slot = slot_pid.is_some();
|
||
|
||
let pid = if let Some(pid) = slot_pid {
|
||
stats.slot_hits.fetch_add(1, Ordering::Relaxed);
|
||
crate::te!(crate::trace::Event::SlotPop(pid));
|
||
pid
|
||
} else {
|
||
// Read IO liveness BEFORE the queue lock (phase-1 ordering: a
|
||
// completion resurrects an actor only via the drain path, whose
|
||
// enqueue would be visible under the queue lock we take next).
|
||
let (io_out, io_fd) = {
|
||
let io = match inner.io.lock() {
|
||
Ok(io) => io,
|
||
Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
|
||
};
|
||
match io.as_ref() {
|
||
Some(io) => (io.outstanding + io.waiters.len() as u32, Some(io.wake_fd())),
|
||
None => (0, None),
|
||
}
|
||
};
|
||
|
||
stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed);
|
||
let pop = match inner.run_queue.pop() {
|
||
Some(pid) => Pop::Got(pid),
|
||
None => {
|
||
// Termination does not lean on pop-None being a fence (with
|
||
// the ring queues it is only a snapshot). The argument is
|
||
// counter-first: every queue entry's target stays `Queued` —
|
||
// hence un-finalized, hence counted live — until that very
|
||
// entry is popped. So `live == 0` (Acquire, pairing with
|
||
// finalize's Release decrement, which strictly follows all
|
||
// wakeup enqueues) by itself implies no entry is in, or can
|
||
// ever again enter, the queue: enqueues only target live
|
||
// actors, and a spawner is itself live. The pop-None above
|
||
// is then just the cheap fast-path filter; io_out was read
|
||
// before it per the phase-1 ordering. `live == 0` is also
|
||
// final — no spawn can resurrect the count — so every
|
||
// scheduler thread independently reaches this same verdict.
|
||
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 }
|
||
}
|
||
}
|
||
};
|
||
|
||
match pop {
|
||
Pop::Got(pid) => pid,
|
||
Pop::AllDone => {
|
||
// Remaining timer entries are orphaned (no live actor can be
|
||
// woken by them — e.g. a sleeper cancelled out of its sleep);
|
||
// they must not keep the runtime alive. Drop them on the way out.
|
||
match inner.timers.lock() {
|
||
Ok(mut timers) => timers.clear(),
|
||
Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
|
||
}
|
||
// Terminal wake: a sibling scheduler may be blocked in its
|
||
// idle wait on a snapshot that is now terminally stale — an
|
||
// orphaned long deadline (it would sleep it out in full) or
|
||
// a stale `io_outstanding > 0` from a stop-cancelled waiter
|
||
// (it would block in poll(-1) forever; cancellation produces
|
||
// no completion, so nothing else writes the wake pipe).
|
||
// One byte wakes every poller; each re-runs the verdict,
|
||
// reaches AllDone itself, and re-wakes — idempotent.
|
||
match inner.io.lock() {
|
||
Ok(io) => {
|
||
if let Some(io) = io.as_ref() {
|
||
io.wake();
|
||
}
|
||
}
|
||
Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
|
||
}
|
||
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.
|
||
let next_deadline = match inner.timers.lock() {
|
||
Ok(timers) => timers.peek_deadline(),
|
||
Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
|
||
};
|
||
match (next_deadline, wake_fd) {
|
||
(Some(deadline), fd_opt) => {
|
||
let now = std::time::Instant::now();
|
||
if deadline > now {
|
||
let timeout = deadline - now;
|
||
match fd_opt {
|
||
Some(fd) => {
|
||
// Wake only; the byte (if any) is
|
||
// consumed by the next drain-lock
|
||
// winner in phase 1. Level-triggered
|
||
// poll means an unconsumed byte makes
|
||
// this return immediately, so a loser
|
||
// spins briefly until the winner
|
||
// releases — never sleeps through it.
|
||
crate::io::poll_wake(fd, Some(timeout));
|
||
}
|
||
None => thread::sleep(timeout),
|
||
}
|
||
}
|
||
}
|
||
(None, Some(fd)) if io_outstanding > 0 => {
|
||
// See above: no byte consumption outside phase 1.
|
||
crate::io::poll_wake(fd, None);
|
||
}
|
||
_ => {
|
||
thread::sleep(std::time::Duration::from_micros(100));
|
||
}
|
||
}
|
||
continue;
|
||
}
|
||
}
|
||
};
|
||
|
||
// ----------------------------------------------------------------
|
||
// 3. Claim and resume the actor: CAS Queued → Running. A failure
|
||
// means the pid is stale (slot recycled — generation mismatch);
|
||
// by the at-most-once-enqueued invariant nothing else can have
|
||
// changed the state of a queued actor.
|
||
// ----------------------------------------------------------------
|
||
let slot = match inner.slot_at(pid) {
|
||
Some(s) => s,
|
||
None => continue, // can't happen for real pids; defensive
|
||
};
|
||
if !slot.word.try_claim(pid.generation()) {
|
||
continue; // stale pid: retry immediately (never the idle path)
|
||
}
|
||
crate::te!(crate::trace::Event::Dequeue(pid));
|
||
|
||
let sp = slot.sp.load(Ordering::Relaxed);
|
||
let stop_flag = slot.stop_ptr.load(Ordering::Relaxed);
|
||
// First resume: move the closure into the trampoline's thread-local.
|
||
if let Some(b) = slot.take_closure() {
|
||
set_current_actor_box(b);
|
||
}
|
||
|
||
// Update per-thread stats: record who's on-CPU.
|
||
stats.current_pid_index.store(pid.index(), Ordering::Relaxed);
|
||
|
||
set_actor_sp(sp);
|
||
set_current_pid(pid);
|
||
crate::preempt::set_current_stop(stop_flag);
|
||
crate::preempt::set_current_slot(slot as *const Slot);
|
||
reset_actor_done();
|
||
YIELD_INTENT.with(|c| c.set(YieldIntent::Yield));
|
||
// RFC 005 timeslice inheritance: a slot-popped actor does NOT get a
|
||
// fresh slice — it inherits the waker's remaining one (this thread's
|
||
// TIMESLICE_START/ALLOC_COUNT carry over from the waker's run, with
|
||
// only scheduler bookkeeping in between). A chain of slot handoffs
|
||
// is therefore collectively bounded by one slice, after which
|
||
// preemption fires and the preempt-yield re-enqueue goes to the
|
||
// SHARED queue (a yield is not a wake — never slot-eligible). The
|
||
// shared queue is thus consulted at least once per slice per
|
||
// scheduler: the one-slice starvation bound, zero new counters.
|
||
if !from_slot {
|
||
crate::preempt::reset_timeslice();
|
||
}
|
||
PREEMPTION_ENABLED.with(|c| c.set(true));
|
||
// RFC 007: delay accrued while this actor was off-CPU is absorbed for
|
||
// free (Coz's blocked-thread rule) — fast-forward its ledger and arm
|
||
// the per-thread sample clock before it runs.
|
||
#[cfg(feature = "smarm-causal")]
|
||
crate::causal::on_resume(slot);
|
||
|
||
crate::te!(crate::trace::Event::Resume(pid));
|
||
unsafe { switch_to_actor() };
|
||
|
||
PREEMPTION_ENABLED.with(|c| c.set(false));
|
||
// RFC 016 Chunk 2: charge the cycles this resume consumed to the actor
|
||
// (approximate; reuses the slice-start timestamp — one RDTSC). Read
|
||
// before the next resume re-arms TIMESLICE_START.
|
||
#[cfg(feature = "budget-accounting")]
|
||
slot.add_budget(crate::preempt::elapsed_slice_cycles());
|
||
stats.current_pid_index.store(u32::MAX, Ordering::Relaxed);
|
||
clear_current_pid();
|
||
crate::preempt::clear_current_stop();
|
||
crate::preempt::clear_current_slot();
|
||
|
||
let intent = YIELD_INTENT.with(|c| c.get());
|
||
slot.sp.store(get_actor_sp(), Ordering::Relaxed);
|
||
|
||
if is_actor_done() {
|
||
crate::te!(crate::trace::Event::Done(pid));
|
||
let outcome = take_last_outcome().unwrap_or(Outcome::Exit);
|
||
finalize_actor(inner, pid, outcome);
|
||
} else {
|
||
let gen = pid.generation();
|
||
match intent {
|
||
YieldIntent::Yield => {
|
||
// RFC 007 audit: measure the sample tail this yield drops
|
||
// (near-zero for slice-expiry yields, which sample at the
|
||
// same checkpoint; fat for explicit yield_now in-site).
|
||
#[cfg(feature = "smarm-causal")]
|
||
crate::causal::on_deschedule(slot, false);
|
||
// Running OR RunningNotified → Queued; a notification
|
||
// arriving mid-run coalesces into the re-queue.
|
||
crate::te!(crate::trace::Event::Yield(pid));
|
||
slot.word.yield_return(gen);
|
||
inner.enqueue(pid);
|
||
}
|
||
YieldIntent::Park => {
|
||
if slot.word.park_return(gen) {
|
||
// RFC 007 audit: an in-site park drops its sample
|
||
// tail (nothing flushes it; on_resume re-arms).
|
||
#[cfg(feature = "smarm-causal")]
|
||
crate::causal::on_deschedule(slot, true);
|
||
// RFC 007: a real park — the eventual wake forgives
|
||
// delay accrued while blocked (the instant-wake
|
||
// re-queue below does NOT: that actor never blocked).
|
||
#[cfg(feature = "smarm-causal")]
|
||
slot.set_causal_parked();
|
||
crate::te!(crate::trace::Event::Park(pid));
|
||
} else {
|
||
// An unpark landed in the prep-to-park window; the
|
||
// word is back to Queued — re-queue instead of
|
||
// parking. The lost-wakeup window, closed: this actor
|
||
// never blocked, so its dropped tail counts as a
|
||
// yield in the RFC 007 audit.
|
||
#[cfg(feature = "smarm-causal")]
|
||
crate::causal::on_deschedule(slot, false);
|
||
crate::te!(crate::trace::Event::UnparkFlagConsumed(pid));
|
||
inner.enqueue(pid);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|