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

feat(runtime): phase 3 — pluggable run queue (rq-mutex / rq-mpmc / rq-striped)

Browse Source 
src/run_queue.rs: the run queue extracted behind a compile-time-selected
type alias; mutually-exclusive cargo features with compile_error! guards
(zero or >1 selected). No runtime dispatch. All variants compile in every
build so their unit tests always run; the feature only picks the alias.

- rq-mutex (default): Mutex<VecDeque>, the control/baseline.
- rq-mpmc: hand-rolled Vyukov bounded MPMC ring, per-cell sequence numbers,
  padded enqueue/dequeue counters. Strict FIFO, lock-free.
- rq-striped: M Vyukov rings (M = thread_count rounded up to pow2),
  fetch-add ticket distribution, probe-from-home on push. Relaxed FIFO with
  stripe-bounded skew; Σcapacity ≈ 2×max_actors so the probe terminates.

Capacity soundness: occupancy ≤ max_actors by the at-most-once-enqueued
invariant + slab cap, rings sized ≥ that bound, so push is infallible; a
full ring panics as a double-enqueue invariant violation rather than spin.

Two contracts the extraction made explicit (documented + debug-asserted):
- Queue ops require preemption disabled: a producer suspended between
  claiming a cell and publishing its sequence stalls every consumer behind
  it — livelock, since the suspended actor's own resume entry is behind the
  hole. Structurally guaranteed since the phase-2 with_runtime NoPreempt fix.
- pop()==None is a snapshot, not a fence. Termination is counter-first:
  every queue entry's target stays Queued (hence live) until that entry is
  popped, so live==0 alone implies nothing actionable is or can be queued;
  argument rewritten at the schedule_loop site. SharedState and with_shared
  are deleted — nothing global is mutex-guarded on the run path under the
  ring variants.

Validated: 22 suites green per variant (release for all three; debug with
live asserts for all three), ring unit tests (FIFO, lap wraparound,
4p/4c exactly-once, skewed drain) in every build, both compile_error!
guards verified to fire.
This commit is contained in:
Claude
2026-06-09 20:34:32 +00:00
parent a78e17e1eb
commit 1b3b618aa7
6 changed files with 538 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/runtime.rs
+43 -62
View File
@@ -9,7 +9,7 @@
//! RuntimeInner {
//! slots: Box<[Slot]> ← FIXED slab, max_actors entries, lock-free lookup
//! free: RawMutex<Vec<u32>> ← vacant slot indices
//! shared: Mutex<SharedState> ← the run queue (and nothing else)
//! 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)
@@ -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<Pid>,
}
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<SharedState>,
/// 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<u32> = (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<R>(&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<RuntimeInner>, 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,