diff --git a/Cargo.toml b/Cargo.toml index e9ae0fb..9e61d68 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,13 @@ edition = "2021" rust-version = "1.95" [features] +default = ["rq-mutex"] smarm-trace = [] +# Run-queue selection: exactly one, compile-time (see src/run_queue.rs). +# Non-default variants need --no-default-features (features are additive). +rq-mutex = [] +rq-mpmc = [] +rq-striped = [] [dependencies] libc = "0.2" diff --git a/ROADMAP_v0.5.md b/ROADMAP_v0.5.md index 47446d9..576d60a 100644 --- a/ROADMAP_v0.5.md +++ b/ROADMAP_v0.5.md @@ -63,19 +63,28 @@ Make slot lookup lock-free and per-slot state independently mutable. && queue empty && live == 0; decrement-last ordering documented as the correctness crux at the site. -## Phase 3 — Pluggable run queue + bench shootout -- [ ] Extract `RunQueue` behind a `type RunQueue = ...` alias selected at - compile time by mutually-exclusive features, with a `compile_error!` - guard if zero or >1 is set. No runtime dispatch. +## Phase 3 — Pluggable run queue ✅ DONE (shootout = phase 4 harness) +- [x] `RunQueue` alias in `src/run_queue.rs`, compile-time selected, + `compile_error!` guards for zero / >1 features. No runtime dispatch. + All variants compile in every build (unit tests always run); the + feature only picks the alias. Non-default variants: + `--no-default-features --features rq-…`. - `rq-mutex` — `Mutex`, the control/baseline. - `rq-mpmc` — single hand-rolled Vyukov bounded MPMC ring (per-cell seq numbers). Strict FIFO; "one hot cache line". - `rq-striped` — M Vyukov rings, fetch-add ticket distribution. Relaxed FIFO, reordering bounded by ~M. Predicted winner @20c. -- [ ] Bounded rings are sound because the slab cap + at-most-once-enqueued - invariant bound occupancy ≤ `max_actors`; size Σcapacity > `max_actors` - so push is infallible (probe next stripe when full; terminates). -- [ ] All hand-rolled, dependency-free (portability incl. possible embedded). +- [x] Bounded rings sound via slab cap + at-most-once-enqueued (occupancy ≤ + `max_actors`); mpmc capacity = next_pow2(max_actors), striped + Σcapacity ≈ 2×max_actors with probe-from-home-stripe. A full ring + panics as an invariant violation (double enqueue), never spins. +- [x] All hand-rolled, dependency-free. +- [x] Two contracts surfaced by the extraction, documented + debug-asserted + in `run_queue.rs`: queue ops require preemption disabled (a producer + suspended mid-publish stalls every consumer behind its cell — livelock); + and pop-None is a snapshot, not a fence — termination is counter-first + (`live == 0` alone implies the queue holds nothing actionable; argument + rewritten at the `schedule_loop` site). ## Phase 4 — Bench harness - [ ] Raw-structure microbench: N threads, push/pop throughput vs thread count, diff --git a/src/lib.rs b/src/lib.rs index d92d4ad..ebb2b6d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,6 +27,7 @@ pub mod link; pub mod gen_server; pub mod runtime; pub(crate) mod raw_mutex; +pub(crate) mod run_queue; pub mod trace; // --------------------------------------------------------------------------- diff --git a/src/raw_mutex.rs b/src/raw_mutex.rs index d6861ad..c5b03e8 100644 --- a/src/raw_mutex.rs +++ b/src/raw_mutex.rs @@ -27,9 +27,10 @@ //! //! Lock-order position: slot cold locks, the free list, and the stack pool //! are all `RawMutex`es and all *leaves among themselves* — never hold two at -//! once. Holding a `RawMutex` while taking the run-queue mutex (`with_shared`) -//! is permitted (e.g. unpark from inside a cold section); the reverse — -//! taking any `RawMutex` from inside `with_shared` — is forbidden. +//! once. Holding a `RawMutex` while pushing to the run queue is permitted +//! (e.g. unpark from inside a cold section); the reverse — taking any +//! `RawMutex` from inside a run-queue op — cannot arise (queue ops call +//! nothing). use std::cell::UnsafeCell; use std::ops::{Deref, DerefMut}; diff --git a/src/run_queue.rs b/src/run_queue.rs new file mode 100644 index 0000000..ef8f906 --- /dev/null +++ b/src/run_queue.rs @@ -0,0 +1,467 @@ +//! The run queue, selected at COMPILE TIME by mutually-exclusive cargo +//! features (no runtime dispatch — the scheduler's pop loop is the hottest +//! code in the runtime): +//! +//! - `rq-mutex` (default) — `Mutex`. The control/baseline: +//! strictly FIFO, trivially correct, one global lock. +//! - `rq-mpmc` — a single hand-rolled Vyukov bounded MPMC ring (per-cell +//! sequence numbers). Strict FIFO, lock-free, one hot +//! enqueue/dequeue cache-line pair. +//! - `rq-striped` — M Vyukov rings with fetch-add ticket distribution. +//! *Relaxed* FIFO: ordering across stripes is bounded-skewed +//! (≈ one ring's worth of reordering per stripe), in exchange +//! for spreading the hot line M ways. Predicted winner at +//! high core counts; phase 4's shootout decides. +//! +//! Select non-default variants with `--no-default-features --features rq-…` +//! (cargo features are additive, so the default must be switched off). +//! +//! All variants are compiled unconditionally (so every build runs every +//! variant's unit tests); the feature only picks which one the runtime uses +//! via the [`RunQueue`] alias. +//! +//! # Contract (shared by all variants) +//! +//! - **Occupancy is bounded by `max_actors`.** A pid is in the queue at most +//! once (pushes pair 1:1 with transitions into `Queued`; only the +//! scheduler transitions `Queued → Running` — see the state-machine docs +//! in `runtime.rs`), and at most `max_actors` actors exist. The bounded +//! rings are sized ≥ `max_actors`, so **`push` is infallible**; a full +//! ring is an invariant violation and panics loudly rather than spinning. +//! - **Preemption must be disabled around every push/pop** (debug-asserted). +//! For the mutex variant this is the usual no-switch/no-unwind-under-lock +//! rule. For the rings it is *load-bearing in a sharper way*: a producer +//! suspended between claiming a cell and publishing its sequence number +//! stalls every consumer behind that cell — on a busy runtime that is a +//! livelock, since the suspended actor's own resume entry sits behind the +//! hole. Callers get this for free: every queue op happens inside +//! `with_runtime`/`try_with_runtime` (NoPreempt for their span since +//! phase 2) or on a scheduler thread between resumes (preemption off). +//! - **`pop() == None` is a snapshot, not a fence.** A push that is mid- +//! publish (or in a stripe the probe already passed) may be missed; the +//! caller's idle path sleeps ≤ 100µs and retries, so the cost is a bounded +//! latency blip, never a lost entry. Termination does not lean on this: +//! the all-clear is `live_actors == 0` (+ io quiescent), and `live == 0` +//! already implies the queue holds nothing actionable — see the argument +//! in `schedule_loop`. +//! - `len()` is approximate (stats only). + +use crate::pid::Pid; +use std::cell::UnsafeCell; +use std::mem::MaybeUninit; +use std::sync::atomic::{AtomicUsize, Ordering}; + +// --------------------------------------------------------------------------- +// Feature selection +// --------------------------------------------------------------------------- + +#[cfg(not(any(feature = "rq-mutex", feature = "rq-mpmc", feature = "rq-striped")))] +compile_error!( + "smarm: no run queue selected. Enable exactly one of the features \ + `rq-mutex` (default), `rq-mpmc`, `rq-striped`." +); +#[cfg(all(feature = "rq-mutex", feature = "rq-mpmc"))] +compile_error!( + "smarm: features `rq-mutex` and `rq-mpmc` are mutually exclusive \ + (use --no-default-features to drop the default `rq-mutex`)." +); +#[cfg(all(feature = "rq-mutex", feature = "rq-striped"))] +compile_error!( + "smarm: features `rq-mutex` and `rq-striped` are mutually exclusive \ + (use --no-default-features to drop the default `rq-mutex`)." +); +#[cfg(all(feature = "rq-mpmc", feature = "rq-striped"))] +compile_error!("smarm: features `rq-mpmc` and `rq-striped` are mutually exclusive."); + +#[cfg(feature = "rq-mutex")] +pub(crate) type RunQueue = MutexQueue; +#[cfg(feature = "rq-mpmc")] +pub(crate) type RunQueue = MpmcRing; +#[cfg(feature = "rq-striped")] +pub(crate) type RunQueue = StripedRing; + +#[inline] +fn assert_no_preempt() { + debug_assert!( + !crate::preempt::PREEMPTION_ENABLED.with(|c| c.get()), + "run-queue op with preemption enabled — a switch mid-op stalls or \ + corrupts the queue; route through with_runtime or scheduler context" + ); +} + +// --------------------------------------------------------------------------- +// rq-mutex — the baseline +// --------------------------------------------------------------------------- + +#[allow(dead_code)] +pub(crate) struct MutexQueue { + q: std::sync::Mutex>, +} + +#[allow(dead_code)] +impl MutexQueue { + pub(crate) fn new(_threads: usize, max_actors: usize) -> Self { + Self { + // Pre-size: the queue can never outgrow the slab, and one + // allocation at init beats reallocating under the lock later. + q: std::sync::Mutex::new(std::collections::VecDeque::with_capacity(max_actors)), + } + } + + pub(crate) fn push(&self, pid: Pid) { + assert_no_preempt(); + self.q.lock().unwrap().push_back(pid); + } + + pub(crate) fn pop(&self) -> Option { + assert_no_preempt(); + self.q.lock().unwrap().pop_front() + } + + pub(crate) fn len(&self) -> u64 { + self.q.lock().unwrap().len() as u64 + } +} + +// --------------------------------------------------------------------------- +// rq-mpmc — Vyukov bounded MPMC ring +// --------------------------------------------------------------------------- +// +// Dmitry Vyukov's bounded MPMC queue: each cell carries a sequence number. +// A producer may write cell `i` when `seq == pos` (its turn); it publishes +// with `seq = pos + 1`. A consumer may read when `seq == pos + 1`; it +// releases the cell to the next lap with `seq = pos + capacity`. Producers +// and consumers each contend on one counter; cell handoff is a per-cell +// Acquire/Release pair, so unrelated push/pop pairs don't serialize. + +/// Pad to a cache-line pair so the producer and consumer counters (and the +/// cells) don't false-share. +#[repr(align(128))] +struct CachePadded(T); + +struct Cell { + seq: AtomicUsize, + pid: UnsafeCell>, +} + +#[allow(dead_code)] +pub(crate) struct MpmcRing { + buf: Box<[Cell]>, + mask: usize, + enqueue_pos: CachePadded, + dequeue_pos: CachePadded, +} + +// SAFETY: cells are handed off between threads via the per-cell seq +// (Release on publish, Acquire on claim); Pid is Copy + Send. +unsafe impl Send for MpmcRing {} +unsafe impl Sync for MpmcRing {} + +#[allow(dead_code)] +impl MpmcRing { + pub(crate) fn new(_threads: usize, max_actors: usize) -> Self { + Self::with_capacity(max_actors) + } + + fn with_capacity(min_cap: usize) -> Self { + // Occupancy ≤ max_actors (queue contract), so capacity = the next + // power of two ≥ max_actors can never overflow. (≥ 2 so mask works.) + let cap = min_cap.next_power_of_two().max(2); + let buf: Box<[Cell]> = (0..cap) + .map(|i| Cell { + seq: AtomicUsize::new(i), + pid: UnsafeCell::new(MaybeUninit::uninit()), + }) + .collect(); + Self { + buf, + mask: cap - 1, + enqueue_pos: CachePadded(AtomicUsize::new(0)), + dequeue_pos: CachePadded(AtomicUsize::new(0)), + } + } + + pub(crate) fn push(&self, pid: Pid) { + assert_no_preempt(); + assert!( + self.try_push(pid), + "smarm: run queue overflow — occupancy exceeded the slab bound, \ + which the at-most-once-enqueued invariant forbids. This is a \ + runtime bug (double enqueue), not a capacity tuning problem." + ); + } + + /// One full claim attempt; `false` only if the ring is full. + fn try_push(&self, pid: Pid) -> bool { + let mut pos = self.enqueue_pos.0.load(Ordering::Relaxed); + loop { + let cell = &self.buf[pos & self.mask]; + let seq = cell.seq.load(Ordering::Acquire); + let diff = seq as isize - pos as isize; + if diff == 0 { + // Our turn: claim the position. + match self.enqueue_pos.0.compare_exchange_weak( + pos, pos + 1, Ordering::Relaxed, Ordering::Relaxed, + ) { + Ok(_) => { + // SAFETY: the claim gives us exclusive write access + // to this cell until we publish below. + unsafe { (*cell.pid.get()).write(pid) }; + cell.seq.store(pos + 1, Ordering::Release); + return true; + } + Err(actual) => pos = actual, + } + } else if diff < 0 { + return false; // full — a whole lap behind + } else { + pos = self.enqueue_pos.0.load(Ordering::Relaxed); + } + } + } + + pub(crate) fn pop(&self) -> Option { + assert_no_preempt(); + let mut pos = self.dequeue_pos.0.load(Ordering::Relaxed); + loop { + let cell = &self.buf[pos & self.mask]; + let seq = cell.seq.load(Ordering::Acquire); + let diff = seq as isize - (pos + 1) as isize; + if diff == 0 { + match self.dequeue_pos.0.compare_exchange_weak( + pos, pos + 1, Ordering::Relaxed, Ordering::Relaxed, + ) { + Ok(_) => { + // SAFETY: the claim gives us exclusive read access; + // the producer's Release publish made `pid` visible + // to our Acquire load of `seq`. + let pid = unsafe { (*cell.pid.get()).assume_init_read() }; + // Release the cell for the next lap. + cell.seq.store(pos + self.mask + 1, Ordering::Release); + return Some(pid); + } + Err(actual) => pos = actual, + } + } else if diff < 0 { + // Empty (or the producer at this cell hasn't published yet — + // a snapshot miss the caller's idle-retry loop absorbs). + return None; + } else { + pos = self.dequeue_pos.0.load(Ordering::Relaxed); + } + } + } + + pub(crate) fn len(&self) -> u64 { + let e = self.enqueue_pos.0.load(Ordering::Relaxed); + let d = self.dequeue_pos.0.load(Ordering::Relaxed); + e.saturating_sub(d) as u64 + } +} + +// --------------------------------------------------------------------------- +// rq-striped — M Vyukov rings, ticket-distributed +// --------------------------------------------------------------------------- +// +// Producers fetch-add a ticket and start probing at stripe `ticket % M`; +// consumers do the same with their own ticket. Under symmetric load the +// tickets spread producers and consumers uniformly, so each stripe sees +// ~1/M of the traffic and the single hot cache-line pair becomes M cooler +// ones. FIFO is relaxed: two pushes that land in different stripes can be +// popped in either order, with skew bounded by stripe occupancy imbalance. +// +// Push probes forward from its home stripe until a `try_push` succeeds. +// Σ stripe capacity ≥ 2 × max_actors while occupancy ≤ max_actors, so at +// every instant at least half the total capacity is free and the probe +// terminates (in practice on the first stripe). + +#[allow(dead_code)] +pub(crate) struct StripedRing { + stripes: Box<[MpmcRing]>, + /// Stripe count minus one (count is a power of two). + stripe_mask: usize, + push_ticket: CachePadded, + pop_ticket: CachePadded, +} + +#[allow(dead_code)] +impl StripedRing { + pub(crate) fn new(threads: usize, max_actors: usize) -> Self { + // One stripe per scheduler thread, rounded up to a power of two — + // more stripes than threads buys nothing (at most `threads` ops are + // in flight) and costs pop-probe latency when mostly empty. + let n = threads.max(1).next_power_of_two(); + // Per-stripe capacity: 2 × max_actors / n in total, and never below + // a floor that keeps degenerate configs (tiny slab, many threads) + // trivially correct. + let per = ((2 * max_actors) / n).next_power_of_two().max(8); + let stripes: Box<[MpmcRing]> = (0..n).map(|_| MpmcRing::with_capacity(per)).collect(); + Self { + stripes, + stripe_mask: n - 1, + push_ticket: CachePadded(AtomicUsize::new(0)), + pop_ticket: CachePadded(AtomicUsize::new(0)), + } + } + + pub(crate) fn push(&self, pid: Pid) { + assert_no_preempt(); + let home = self.push_ticket.0.fetch_add(1, Ordering::Relaxed); + // Probe from the home stripe; capacity headroom (Σ ≥ 2×occupancy) + // guarantees a free stripe exists, so the outer loop terminates. + // The retry-from-home lap handles the racy case where every stripe + // momentarily refused us. + loop { + for i in 0..=self.stripe_mask { + let s = &self.stripes[(home + i) & self.stripe_mask]; + if s.try_push(pid) { + return; + } + } + std::hint::spin_loop(); + } + } + + pub(crate) fn pop(&self) -> Option { + assert_no_preempt(); + let home = self.pop_ticket.0.fetch_add(1, Ordering::Relaxed); + for i in 0..=self.stripe_mask { + if let Some(pid) = self.stripes[(home + i) & self.stripe_mask].pop() { + return Some(pid); + } + } + None // snapshot miss possible across stripes; idle-retry absorbs it + } + + pub(crate) fn len(&self) -> u64 { + self.stripes.iter().map(|s| s.len()).sum() + } +} + +// --------------------------------------------------------------------------- +// Tests — all variants, in every build (the feature only picks the alias) +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashSet; + use std::sync::Arc; + + fn pid(i: u32) -> Pid { + Pid::new(i, 0) + } + + fn fifo_smoke(q: &Q, push: impl Fn(&Q, Pid), pop: impl Fn(&Q) -> Option) { + for i in 0..100 { + push(q, pid(i)); + } + for i in 0..100 { + assert_eq!(pop(q), Some(pid(i))); + } + assert_eq!(pop(q), None); + } + + #[test] + fn mutex_fifo() { + let q = MutexQueue::new(1, 1024); + fifo_smoke(&q, |q, p| q.push(p), |q| q.pop()); + } + + #[test] + fn mpmc_fifo_single_thread() { + let q = MpmcRing::new(1, 1024); + fifo_smoke(&q, |q, p| q.push(p), |q| q.pop()); + } + + #[test] + fn mpmc_wraps_many_laps() { + let q = MpmcRing::with_capacity(8); + for lap in 0..1000u32 { + for i in 0..8 { + q.push(pid(lap * 8 + i)); + } + for i in 0..8 { + assert_eq!(q.pop(), Some(pid(lap * 8 + i))); + } + } + assert_eq!(q.pop(), None); + } + + /// N producers, M consumers, every element exactly once. Run on plain OS + /// threads (PREEMPTION_ENABLED defaults false, satisfying the contract). + fn exactly_once( + q: Q, + push: fn(&Q, Pid), + pop: fn(&Q) -> Option, + producers: u32, + consumers: u32, + per_producer: u32, + ) { + let q = Arc::new(q); + let total = (producers * per_producer) as usize; + let popped = Arc::new(std::sync::Mutex::new(Vec::with_capacity(total))); + let remaining = Arc::new(AtomicUsize::new(total)); + + let mut hs = Vec::new(); + for p in 0..producers { + let q = q.clone(); + hs.push(std::thread::spawn(move || { + for i in 0..per_producer { + push(&q, pid(p * per_producer + i)); + } + })); + } + for _ in 0..consumers { + let q = q.clone(); + let popped = popped.clone(); + let remaining = remaining.clone(); + hs.push(std::thread::spawn(move || { + let mut local = Vec::new(); + while remaining.load(Ordering::Relaxed) > 0 { + if let Some(pid) = pop(&q) { + remaining.fetch_sub(1, Ordering::Relaxed); + local.push(pid); + } else { + std::hint::spin_loop(); + } + } + popped.lock().unwrap().extend(local); + })); + } + for h in hs { + h.join().unwrap(); + } + + let popped = popped.lock().unwrap(); + assert_eq!(popped.len(), total, "count mismatch"); + let set: HashSet = popped.iter().map(|p| ((p.index() as u64) << 32) | p.generation() as u64).collect(); + assert_eq!(set.len(), total, "duplicate or lost element"); + assert_eq!(pop(&q), None); + } + + #[test] + fn mpmc_exactly_once_contended() { + exactly_once(MpmcRing::new(8, 4096), |q, p| q.push(p), |q| q.pop(), 4, 4, 1000); + } + + #[test] + fn striped_exactly_once_contended() { + exactly_once(StripedRing::new(8, 4096), |q, p| q.push(p), |q| q.pop(), 4, 4, 1000); + } + + #[test] + fn striped_drains_after_skewed_load() { + // Hammer pushes from one thread (all tickets walk the stripes in + // order) and verify a single consumer sees every element. + let q = StripedRing::new(4, 64); + let mut seen = HashSet::new(); + for i in 0..64 { + q.push(pid(i)); + } + while let Some(p) = q.pop() { + assert!(seen.insert(p.index())); + } + assert_eq!(seen.len(), 64); + } +} diff --git a/src/runtime.rs b/src/runtime.rs index fe26801..420e0e9 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -9,7 +9,7 @@ //! RuntimeInner { //! slots: Box<[Slot]> ← FIXED slab, max_actors entries, lock-free lookup //! free: RawMutex> ← vacant slot indices -//! shared: Mutex ← the run queue (and nothing else) +//! run_queue: RunQueue ← compile-time selected (src/run_queue.rs) //! timers: Mutex //! io: Mutex> //! live_actors: AtomicU32 ← spawned-but-not-finalized count (termination) @@ -62,15 +62,17 @@ //! //! # Locks and ordering //! -//! - `shared` now guards **only the run queue**. It is the innermost lock: -//! nothing else is ever acquired while holding it. +//! - 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) → `shared`/run-queue. `timers` is independent (never +//! mutually leaf) → run queue (innermost). `timers` is independent (never //! nested with any of the above on either side). //! //! # Termination (counter-based) @@ -106,7 +108,6 @@ use crate::supervisor::Signal; use crate::timer::Timers; use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor}; -use std::collections::VecDeque; use std::sync::atomic::{ AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering, }; @@ -487,28 +488,15 @@ impl Slot { } } -// --------------------------------------------------------------------------- -// Shared state (behind Mutex<>) — now: the run queue, nothing else -// --------------------------------------------------------------------------- - -pub(crate) struct SharedState { - pub(crate) run_queue: VecDeque, -} - -impl SharedState { - fn new() -> Self { - Self { run_queue: VecDeque::new() } - } -} - // --------------------------------------------------------------------------- // RuntimeInner — the shared core behind an Arc // --------------------------------------------------------------------------- pub(crate) struct RuntimeInner { - /// The run queue. Innermost lock: nothing else is acquired under it. - /// Phase 3 swaps this for the compile-time-selected `RunQueue`. - pub(crate) shared: Mutex, + /// 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. @@ -552,7 +540,7 @@ impl RuntimeInner { // Low indices on top of the stack so early spawns get low pids. let free: Vec = (0..max_actors as u32).rev().collect(); Arc::new(Self { - shared: Mutex::new(SharedState::new()), + run_queue: crate::run_queue::RunQueue::new(thread_count, max_actors), slots, free: RawMutex::new(free), live_actors: AtomicU32::new(0), @@ -570,17 +558,6 @@ impl RuntimeInner { }) } - pub(crate) fn with_shared(&self, f: impl FnOnce(&mut SharedState) -> R) -> R { - // Preemption must be off while we hold the queue mutex: a - // preemption-driven context switch under it would carry the lock off - // to a suspended actor and stall every scheduler thread. (The stop - // sentinel shares the gate, so nothing can unwind in here either.) - let prev = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); - let result = f(&mut self.shared.lock().unwrap()); - crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(prev)); - result - } - /// 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 @@ -594,7 +571,7 @@ impl RuntimeInner { /// into `Queued` (spawn's publish, the unpark protocol, or the /// scheduler's yield/notified-park return paths). pub(crate) fn enqueue(&self, pid: Pid) { - self.with_shared(|s| s.run_queue.push_back(pid)); + self.run_queue.push(pid); crate::te!(crate::trace::Event::Enqueue(pid)); } @@ -694,14 +671,14 @@ impl Runtime { crate::trace::open(); // Re-initialise shared state for this run. - { - let s = self.inner.shared.lock().unwrap(); - assert!(s.run_queue.is_empty(), "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" - ); - } + 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" + ); *self.inner.io.lock().unwrap() = Some(IoThread::start().expect("failed to start IO thread")); // Spawn the initial actor through the public spawn path (which @@ -1113,27 +1090,31 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { None => (0, None), }; - let pop = inner.with_shared(|s| { - let len = s.run_queue.len() as u64; - stats.run_queue_len.store(len, Ordering::Relaxed); - match s.run_queue.pop_front() { - Some(pid) => Pop::Got(pid), - None => { - // live_actors is read under the queue lock. If it is 0, - // every finalize fully completed — including its wakeup - // enqueues, which happen-before the decrement — so with - // the queue also empty, nothing can produce work again - // (any enqueue targets a live actor; a spawner is itself - // live). io_out was read before the lock per phase 1. - let live = inner.live_actors.load(Ordering::Acquire); - if live == 0 && io_out == 0 { - Pop::AllDone - } else { - Pop::Idle { io_outstanding: io_out, wake_fd: io_fd } - } + 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 { + Pop::Idle { io_outstanding: io_out, wake_fd: io_fd } } } - }); + }; let pid = match pop { Pop::Got(pid) => pid,