diff --git a/Cargo.toml b/Cargo.toml index 25da28e..acca772 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,9 @@ version = "0.4.0" edition = "2021" rust-version = "1.95" +[lints.rust] +unexpected_cfgs = { level = "warn", check-cfg = ["cfg(loom)"] } + [features] default = ["rq-mutex"] smarm-trace = [] @@ -16,6 +19,9 @@ rq-striped = [] [dependencies] libc = "0.2" +[target.'cfg(loom)'.dependencies] +loom = "0.7" + [dev-dependencies] libc = "0.2" tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "sync", "time"] } diff --git a/ROADMAP_v0.5.md b/ROADMAP_v0.5.md index 02e6f20..2b62af5 100644 --- a/ROADMAP_v0.5.md +++ b/ROADMAP_v0.5.md @@ -98,11 +98,26 @@ Make slot lookup lock-free and per-slot state independently mutable. contention curves and the actual variant decision come from the 20-core box. -## Phase 5 — Safety hardening & model checking -- [ ] `loom` (dev-dependency only, x86 Linux) model tests for the slot state - machine and each ring variant. -- [ ] `NoPreempt` / no-unwind-under-lock audit across all internal critical - sections; assert the invariant in debug builds where feasible. +## Phase 5 — Safety hardening & model checking ✅ DONE +- [x] State machine extracted to `src/slot_state.rs` (mechanism only; the + protocol rationale stays in `runtime.rs`) so loom checks the PRODUCTION + transitions. Models: lost-wakeup (park vs unpark), at-most-once + (two unparkers), ABA (stale unpark vs reclaim+reuse), claim coalescing. +- [x] Loom on the rings via `src/sync_shim.rs` (std normally, `loom::sync` + under `--cfg loom`): exactly-once through lap wraparound, push/pop + race, striped two-producer drain. `[target.'cfg(loom)'.dependencies]`; + run with `RUSTFLAGS="--cfg loom" cargo test --lib --release`. + RawMutex is deliberately NOT loom-modeled: futexes can't be, and it is + Drepper's textbook mutex3 with stress + unwind tests. +- [x] NoPreempt / no-unwind / TLS-guard audit: with_runtime + RawMutex guards + + run-queue ops + trace::record all gate preemption (and thereby the + stop sentinel) for their span. Sole remaining exception: channel's std + MutexGuard — the documented first fast-follow. +- [x] Invariant-assertion sweep (the fast-follow, pulled forward): every + StateWord transition self-asserts its precondition; enqueue asserts + exactly (gen, Queued); live_actors underflow asserted; RawMutex + enforces the leaf rule with a debug-build held-count; slab overflow and + ring overflow stay loud panics. House style from here on. --- diff --git a/src/lib.rs b/src/lib.rs index b31e4c3..4b0e424 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,6 +27,8 @@ pub mod link; pub mod gen_server; pub mod runtime; pub(crate) mod raw_mutex; +pub(crate) mod slot_state; +pub(crate) mod sync_shim; #[doc(hidden)] // pub only so benches/rq_micro.rs can drive the raw structures pub mod run_queue; pub mod trace; diff --git a/src/raw_mutex.rs b/src/raw_mutex.rs index c5b03e8..3152a56 100644 --- a/src/raw_mutex.rs +++ b/src/raw_mutex.rs @@ -45,6 +45,35 @@ const CONTENDED: u32 = 2; /// sender), so a short spin almost always avoids the syscall. const SPIN_LIMIT: u32 = 64; +// The leaf rule, mechanically enforced (debug builds): slot cold locks, the +// free list, and the stack pool — i.e. every RawMutex — are mutual leaves. +// Holding two at once is a deadlock waiting for the right interleaving, so +// fail at the acquisition that violates it, not in the eventual hang. +#[cfg(debug_assertions)] +thread_local! { + static RAW_MUTEXES_HELD: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +#[inline] +fn leaf_check_acquire() { + #[cfg(debug_assertions)] + RAW_MUTEXES_HELD.with(|c| { + debug_assert_eq!( + c.get(), + 0, + "leaf rule violated: acquiring a RawMutex while already holding one \ + (cold locks / free list / stack pool are mutual leaves)" + ); + c.set(c.get() + 1); + }); +} + +#[inline] +fn leaf_check_release() { + #[cfg(debug_assertions)] + RAW_MUTEXES_HELD.with(|c| c.set(c.get() - 1)); +} + pub(crate) struct RawMutex { state: AtomicU32, data: UnsafeCell, @@ -69,6 +98,7 @@ impl RawMutex { // Enter NoPreempt *before* acquiring, so a preemption can't fire // between acquisition and guard construction. let prev_preempt = crate::preempt::PREEMPTION_ENABLED.with(|c| c.replace(false)); + leaf_check_acquire(); if self .state .compare_exchange(UNLOCKED, LOCKED, Ordering::Acquire, Ordering::Relaxed) @@ -142,6 +172,7 @@ impl Drop for RawMutexGuard<'_, T> { #[inline] fn drop(&mut self) { self.m.unlock(); + leaf_check_release(); // Restore preemption only after the lock is released. crate::preempt::PREEMPTION_ENABLED.with(|c| c.set(self.prev_preempt)); } diff --git a/src/run_queue.rs b/src/run_queue.rs index df46cee..91775f7 100644 --- a/src/run_queue.rs +++ b/src/run_queue.rs @@ -47,9 +47,8 @@ //! - `len()` is approximate (stats only). use crate::pid::Pid; -use std::cell::UnsafeCell; +use crate::sync_shim::{AtomicUsize, Ordering, UnsafeCell}; use std::mem::MaybeUninit; -use std::sync::atomic::{AtomicUsize, Ordering}; // --------------------------------------------------------------------------- // Feature selection @@ -208,7 +207,7 @@ impl MpmcRing { Ok(_) => { // SAFETY: the claim gives us exclusive write access // to this cell until we publish below. - unsafe { (*cell.pid.get()).write(pid) }; + cell.pid.with_mut(|p| unsafe { (*p).write(pid) }); cell.seq.store(pos + 1, Ordering::Release); return true; } @@ -237,7 +236,7 @@ impl MpmcRing { // 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() }; + let pid = cell.pid.with(|p| unsafe { (*p).assume_init_read() }); // Release the cell for the next lap. cell.seq.store(pos + self.mask + 1, Ordering::Release); return Some(pid); @@ -344,7 +343,7 @@ impl StripedRing { // Tests — all variants, in every build (the feature only picks the alias) // --------------------------------------------------------------------------- -#[cfg(test)] +#[cfg(all(test, not(loom)))] mod tests { use super::*; use std::collections::HashSet; @@ -467,3 +466,93 @@ mod tests { assert_eq!(seen.len(), 64); } } + +// --------------------------------------------------------------------------- +// loom model tests — RUSTFLAGS="--cfg loom" cargo test --lib --release +// --------------------------------------------------------------------------- + +#[cfg(all(test, loom))] +mod loom_tests { + use super::*; + use loom::sync::Arc; + use loom::thread; + + fn pid(i: u32) -> Pid { + Pid::new(i, 0) + } + + /// Two producers, main-thread consumer: both elements arrive exactly + /// once, across every interleaving — including through a lap wraparound + /// (capacity 2 forces cell reuse). + #[test] + fn mpmc_two_producers_exactly_once() { + loom::model(|| { + let q = Arc::new(MpmcRing::with_capacity(2)); + let mut hs = Vec::new(); + for i in 0..2u32 { + let q = q.clone(); + hs.push(thread::spawn(move || q.push(pid(i)))); + } + let mut got = Vec::new(); + while got.len() < 2 { + match q.pop() { + Some(p) => got.push(p.index()), + None => thread::yield_now(), + } + } + for h in hs { + h.join().unwrap(); + } + got.sort_unstable(); + assert_eq!(got, vec![0, 1]); + assert!(q.pop().is_none()); + }); + } + + /// Producer races a consumer on a single element: the consumer either + /// gets it or sees a clean None — never a torn/duplicated element. + #[test] + fn mpmc_push_pop_race() { + loom::model(|| { + let q = Arc::new(MpmcRing::with_capacity(2)); + let q2 = q.clone(); + let prod = thread::spawn(move || q2.push(pid(7))); + let seen = q.pop(); + prod.join().unwrap(); + match seen { + Some(p) => { + assert_eq!(p.index(), 7); + assert!(q.pop().is_none()); + } + None => assert_eq!(q.pop().map(|p| p.index()), Some(7)), + } + }); + } + + /// Striped: two producers landing in (potentially) different stripes, + /// main-thread consumer drains both exactly once. + #[test] + fn striped_two_producers_exactly_once() { + loom::model(|| { + let q = Arc::new(StripedRing::new(2, 4)); + let mut hs = Vec::new(); + for i in 0..2u32 { + let q = q.clone(); + hs.push(thread::spawn(move || q.push(pid(i)))); + } + let mut got = Vec::new(); + while got.len() < 2 { + match q.pop() { + Some(p) => got.push(p.index()), + None => thread::yield_now(), + } + } + for h in hs { + h.join().unwrap(); + } + got.sort_unstable(); + assert_eq!(got, vec![0, 1]); + assert!(q.pop().is_none()); + }); + } +} diff --git a/src/runtime.rs b/src/runtime.rs index 420e0e9..42dc7b8 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -104,6 +104,7 @@ 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}; @@ -304,27 +305,6 @@ impl RuntimeStats { pub(crate) const ACTOR_STACK_SIZE: usize = 64 * 1024; -// State encodings for the low 8 bits of `Slot::word`. -pub(crate) const ST_VACANT: u64 = 0; -pub(crate) const ST_QUEUED: u64 = 1; -pub(crate) const ST_RUNNING: u64 = 2; -pub(crate) const ST_RUNNING_NOTIFIED: u64 = 3; -pub(crate) const ST_PARKED: u64 = 4; -pub(crate) const ST_DONE: u64 = 5; - -#[inline] -pub(crate) const fn pack(gen: u32, st: u64) -> u64 { - ((gen as u64) << 32) | st -} -#[inline] -pub(crate) const fn word_gen(w: u64) -> u32 { - (w >> 32) as u32 -} -#[inline] -pub(crate) const fn word_state(w: u64) -> u64 { - w & 0xFF -} - pub(crate) type Closure = Box; /// Lifecycle data, mutated only under the slot's cold [`RawMutex`]. Everything @@ -357,8 +337,9 @@ pub(crate) struct SlotCold { /// between unrelated actors. #[repr(align(128))] pub(crate) struct Slot { - /// `(generation << 32) | state`. See the module docs for the machine. - word: AtomicU64, + /// `(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`. @@ -377,7 +358,7 @@ pub(crate) struct Slot { impl Slot { fn vacant() -> Self { Self { - word: AtomicU64::new(pack(0, ST_VACANT)), + word: StateWord::new(), sp: AtomicUsize::new(0), stop_ptr: AtomicPtr::new(std::ptr::null_mut()), closure: AtomicPtr::new(std::ptr::null_mut()), @@ -394,16 +375,18 @@ impl Slot { } } - #[inline] - pub(crate) fn word(&self) -> u64 { - self.word.load(Ordering::Acquire) - } - - /// Current generation (of whatever occupies the slot — pair with a state - /// check or a CAS before acting on it). + /// 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 { - word_gen(self.word()) + self.word.generation() + } + + /// 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 @@ -411,24 +394,7 @@ impl Slot { /// CAS on the word.) #[inline] pub(crate) fn is_live_for(&self, pid: Pid) -> bool { - let w = self.word(); - word_gen(w) == pid.generation() - && !matches!(word_state(w), ST_VACANT | ST_DONE) - } - - #[inline] - fn store_word(&self, gen: u32, st: u64) { - self.word.store(pack(gen, st), Ordering::Release); - } - - #[inline] - fn cas_word(&self, gen: u32, from: u64, to: u64) -> Result { - self.word.compare_exchange( - pack(gen, from), - pack(gen, to), - Ordering::AcqRel, - Ordering::Acquire, - ) + self.status_for(pid) == Status::Live } fn store_closure(&self, c: Closure) { @@ -449,43 +415,6 @@ impl Slot { } } - /// The unpark protocol — the one way anything outside the scheduler makes - /// an actor runnable. Returns `true` iff the caller must enqueue `pid`. - /// - /// Loops on the packed word: - /// - generation mismatch → stale pid → no-op (atomic with the CAS: a - /// recycled slot can never be transitioned by an old pid). - /// - `Parked` → `Queued`: caller enqueues. - /// - `Running` → `RunningNotified`: the actor is inside a prep-to-park - /// window (registered somewhere, not yet parked). The scheduler's - /// park-return CAS will fail against this and re-queue immediately — - /// the lost-wakeup window is a state, not a flag. - /// - `Queued` / `RunningNotified`: someone already notified — coalesce. - /// - `Done` / `Vacant`: nothing to wake. - #[must_use] - pub(crate) fn unpark_action(&self, pid: Pid) -> bool { - let gen = pid.generation(); - loop { - let w = self.word(); - if word_gen(w) != gen { - return false; - } - match word_state(w) { - ST_PARKED => { - if self.cas_word(gen, ST_PARKED, ST_QUEUED).is_ok() { - return true; - } - } - ST_RUNNING => { - if self.cas_word(gen, ST_RUNNING, ST_RUNNING_NOTIFIED).is_ok() { - crate::te!(crate::trace::Event::UnparkDeferred(pid)); - return false; - } - } - _ => return false, // Queued | RunningNotified | Done | Vacant - } - } - } } // --------------------------------------------------------------------------- @@ -571,6 +500,18 @@ 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) { + // 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_eq!( + self.slot_at(pid).map(|s| s.word.load()), + Some(crate::slot_state::pack( + pid.generation(), + 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)); } @@ -579,9 +520,15 @@ impl RuntimeInner { /// The runtime-internal core of `scheduler::unpark`. pub(crate) fn unpark(&self, pid: Pid) { if let Some(slot) = self.slot_at(pid) { - if slot.unpark_action(pid) { - crate::te!(crate::trace::Event::UnparkDirect(pid)); - self.enqueue(pid); + match slot.word.unpark(pid.generation()) { + Unpark::Enqueue => { + crate::te!(crate::trace::Event::UnparkDirect(pid)); + self.enqueue(pid); + } + Unpark::Notified => { + crate::te!(crate::trace::Event::UnparkDeferred(pid)); + } + Unpark::Noop => {} } } } @@ -797,7 +744,6 @@ pub(crate) fn install_actor( 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); - debug_assert_eq!(word_state(slot.word()), ST_VACANT); let stop = Arc::new(AtomicBool::new(false)); slot.stop_ptr.store(Arc::as_ptr(&stop) as *mut _, Ordering::Release); @@ -816,7 +762,7 @@ pub(crate) fn install_actor( // Publish: only now can pops, unparks, or stops find the actor. The // Release store orders everything above before any Acquire reader. - slot.store_word(gen, ST_QUEUED); + slot.word.publish_queued(gen); inner.enqueue(pid); crate::te!(crate::trace::Event::Spawn { parent: supervisor, child: pid }); pid @@ -839,11 +785,7 @@ pub(crate) fn reclaim_slot(inner: &RuntimeInner, pid: Pid) { let dropped_outside; { let mut cold = slot.cold.lock(); - let w = slot.word(); - if word_gen(w) != pid.generation() - || word_state(w) != ST_DONE - || cold.outstanding_handles != 0 - { + 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"); @@ -859,7 +801,7 @@ pub(crate) fn reclaim_slot(inner: &RuntimeInner, pid: Pid) { 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.store_word(pid.generation().wrapping_add(1), ST_VACANT); + slot.word.reclaim(pid.generation()); } drop(dropped_outside); inner.free.lock().push(pid.index()); @@ -886,7 +828,6 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { let slot = inner.slot_at(pid).expect("finalize_actor: pid out of range"); let (waiters, monitors, links, actor) = { let mut cold = slot.cold.lock(); - debug_assert_eq!(slot.generation(), pid.generation(), "finalize: slot vanished"); let actor = cold.actor.take().expect("finalize_actor: actor vanished"); cold.outcome = Some(joiner_outcome); slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release); @@ -894,7 +835,8 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { // 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. - slot.store_word(pid.generation(), ST_DONE); + // (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), @@ -948,10 +890,7 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { let trap = match inner.slot_at(peer) { Some(ps) => { let mut cold = ps.cold.lock(); - let w = ps.word(); - if word_gen(w) == peer.generation() - && !matches!(word_state(w), ST_DONE | ST_VACANT) - { + 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())) @@ -985,7 +924,8 @@ fn finalize_actor(inner: &Arc, pid: Pid, outcome: Outcome) { // 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. - inner.live_actors.fetch_sub(1, Ordering::Release); + let prev = inner.live_actors.fetch_sub(1, Ordering::Release); + debug_assert!(prev >= 1, "live_actors underflow — double finalize"); } // --------------------------------------------------------------------------- @@ -1165,11 +1105,7 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { Some(s) => s, None => continue, // can't happen for real pids; defensive }; - if let Err(actual) = slot.cas_word(pid.generation(), ST_QUEUED, ST_RUNNING) { - debug_assert_ne!( - word_gen(actual), pid.generation(), - "queued pid in unexpected state {}", word_state(actual) - ); + if !slot.word.try_claim(pid.generation()) { continue; // stale pid: retry immediately (never the idle path) } crate::te!(crate::trace::Event::Dequeue(pid)); @@ -1214,26 +1150,18 @@ fn schedule_loop(inner: &Arc, slot_idx: usize) { // Running OR RunningNotified → Queued; a notification // arriving mid-run coalesces into the re-queue. crate::te!(crate::trace::Event::Yield(pid)); - let prev = slot.word.swap(pack(gen, ST_QUEUED), Ordering::AcqRel); - debug_assert!(matches!( - word_state(prev), ST_RUNNING | ST_RUNNING_NOTIFIED - )); + slot.word.yield_return(gen); inner.enqueue(pid); } YieldIntent::Park => { - match slot.cas_word(gen, ST_RUNNING, ST_PARKED) { - Ok(_) => { - crate::te!(crate::trace::Event::Park(pid)); - } - Err(actual) => { - // An unpark landed in the prep-to-park window: - // we are RunningNotified. Re-queue instead of - // parking — the lost-wakeup window, closed. - debug_assert_eq!(word_state(actual), ST_RUNNING_NOTIFIED); - slot.store_word(gen, ST_QUEUED); - crate::te!(crate::trace::Event::UnparkFlagConsumed(pid)); - inner.enqueue(pid); - } + if slot.word.park_return(gen) { + 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. + crate::te!(crate::trace::Event::UnparkFlagConsumed(pid)); + inner.enqueue(pid); } } } diff --git a/src/scheduler.rs b/src/scheduler.rs index 9c59785..f0d3bcd 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -86,18 +86,19 @@ impl JoinHandle { let slot = inner.slot_at(self.pid) .expect("join: pid index out of range"); let mut cold = slot.cold.lock(); - let w = slot.word(); - // Our outstanding handle pins the slot: it cannot be - // reclaimed (generation cannot change) while we hold it. - assert_eq!( - crate::runtime::word_gen(w), self.pid.generation(), - "join: target slot has been reused" - ); - if crate::runtime::word_state(w) == crate::runtime::ST_DONE { - Some(cold.outcome.take().expect("Done slot must have outcome")) - } else { - cold.waiters.push(me); - None + match slot.status_for(self.pid) { + // Our outstanding handle pins the slot: it cannot be + // reclaimed (generation cannot change) while we hold it. + crate::slot_state::Status::Stale => { + panic!("join: target slot has been reused") + } + crate::slot_state::Status::Done => { + Some(cold.outcome.take().expect("Done slot must have outcome")) + } + crate::slot_state::Status::Live => { + cold.waiters.push(me); + None + } } }); @@ -127,13 +128,14 @@ impl JoinHandle { let should_reclaim = match inner.slot_at(self.pid) { Some(slot) => { let mut cold = slot.cold.lock(); - if slot.generation() == self.pid.generation() { - cold.outstanding_handles = cold.outstanding_handles.saturating_sub(1); - cold.outstanding_handles == 0 - && crate::runtime::word_state(slot.word()) - == crate::runtime::ST_DONE - } else { - false + match slot.status_for(self.pid) { + crate::slot_state::Status::Stale => false, + status => { + cold.outstanding_handles = + cold.outstanding_handles.saturating_sub(1); + cold.outstanding_handles == 0 + && status == crate::slot_state::Status::Done + } } } None => false, @@ -324,6 +326,10 @@ where let result = with_runtime(|inner| { let slot = inner.slot_at(me).expect("block_on_io: own slot vanished"); let mut cold = slot.cold.lock(); + debug_assert_eq!( + slot.generation(), me.generation(), + "block_on_io: own slot reused mid-park" + ); cold.pending_io_result .take() .expect("block_on_io: resumed without a result") diff --git a/src/slot_state.rs b/src/slot_state.rs new file mode 100644 index 0000000..80f4ca2 --- /dev/null +++ b/src/slot_state.rs @@ -0,0 +1,381 @@ +//! The per-slot scheduling state machine, as a standalone unit. +//! +//! One atomic word packs `(generation << 32) | state`; every transition is a +//! CAS on the packed word, so the generation check is atomic with the +//! transition — no ABA, no acting on a recycled slot. The diagram and the +//! full protocol rationale live in `runtime.rs`; this module is the +//! mechanism, factored out so that: +//! +//! - loom can model-check the production transitions directly (see the +//! `loom_tests` module; built with `RUSTFLAGS="--cfg loom"`), and +//! - every method asserts the precondition it relies on (`debug_assert!` — +//! these are hot paths), per the assert-the-invariants house rule. +//! +//! Atomics come from `sync_shim` (std normally, `loom::sync` under +//! `cfg(loom)`). + +use crate::sync_shim::{AtomicU64, Ordering}; + +pub(crate) const ST_VACANT: u64 = 0; +pub(crate) const ST_QUEUED: u64 = 1; +pub(crate) const ST_RUNNING: u64 = 2; +pub(crate) const ST_RUNNING_NOTIFIED: u64 = 3; +pub(crate) const ST_PARKED: u64 = 4; +pub(crate) const ST_DONE: u64 = 5; + +#[inline] +pub(crate) const fn pack(gen: u32, st: u64) -> u64 { + ((gen as u64) << 32) | st +} +#[inline] +pub(crate) const fn word_gen(w: u64) -> u32 { + (w >> 32) as u32 +} +#[inline] +pub(crate) const fn word_state(w: u64) -> u64 { + w & 0xFF +} + +/// What an unpark amounted to. The caller owns the side effects (enqueue, +/// trace events) — this module is pure state. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub(crate) enum Unpark { + /// Parked → Queued: the caller must enqueue the pid. + Enqueue, + /// Running → RunningNotified: the scheduler's park-return will re-queue. + Notified, + /// Stale generation, already queued/notified, done, or vacant. + Noop, +} + +/// A pid's-eye view of the slot, for cold paths that hold the slot lock. +#[derive(Debug, PartialEq, Eq, Clone, Copy)] +pub(crate) enum Status { + /// The generation no longer matches: the slot was reclaimed (and possibly + /// reused) — the pid is stale. + Stale, + /// The actor terminated; its outcome is (or was) in the slot. + Done, + /// Alive in some scheduling state (Queued / Running / Notified / Parked). + Live, +} + +pub(crate) struct StateWord(AtomicU64); + +impl StateWord { + pub(crate) fn new() -> Self { + Self(AtomicU64::new(pack(0, ST_VACANT))) + } + + #[inline] + pub(crate) fn load(&self) -> u64 { + self.0.load(Ordering::Acquire) + } + + #[inline] + pub(crate) fn generation(&self) -> u32 { + word_gen(self.load()) + } + + #[inline] + pub(crate) fn status_for(&self, gen: u32) -> Status { + let w = self.load(); + if word_gen(w) != gen { + return Status::Stale; + } + match word_state(w) { + ST_DONE => Status::Done, + // A matching generation on a Vacant slot is unreachable for any + // ISSUED pid — reclaim bumps the generation in the very store + // that vacates, and install publishes Queued before the pid + // escapes. But `Pid::new` is public, so a forged / never-issued + // pid (e.g. `Pid::new(5, 0)` against a fresh slab) can land + // here; for those, "no such actor" is the correct total answer. + ST_VACANT => Status::Stale, + _ => Status::Live, + } + } + + /// Spawn-side publish: Vacant → Queued. The caller owns the vacant slot + /// exclusively (it popped the index from the free list), so this is a + /// plain Release store; it is the moment the actor becomes visible to + /// pops, unparks, and stops. + pub(crate) fn publish_queued(&self, gen: u32) { + debug_assert_eq!( + self.load(), + pack(gen, ST_VACANT), + "publish over a non-vacant slot" + ); + self.0.store(pack(gen, ST_QUEUED), Ordering::Release); + } + + /// Scheduler pop-side claim: Queued → Running. `false` means the popped + /// pid is stale — by the at-most-once-enqueued invariant, a generation + /// mismatch is the only possible failure (asserted). + #[must_use] + pub(crate) fn try_claim(&self, gen: u32) -> bool { + match self.0.compare_exchange( + pack(gen, ST_QUEUED), + pack(gen, ST_RUNNING), + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => true, + Err(actual) => { + debug_assert_ne!( + word_gen(actual), + gen, + "queued pid found in unexpected state {} — double enqueue?", + word_state(actual) + ); + false + } + } + } + + /// Yield return path: Running | RunningNotified → Queued. A notification + /// that arrived mid-run coalesces into the re-queue. Caller must enqueue. + pub(crate) fn yield_return(&self, gen: u32) { + let prev = self.0.swap(pack(gen, ST_QUEUED), Ordering::AcqRel); + debug_assert!( + matches!(word_state(prev), ST_RUNNING | ST_RUNNING_NOTIFIED) + && word_gen(prev) == gen, + "yield return from invalid word {prev:#x}" + ); + } + + /// Park return path. `true` = actually parked. `false` = an unpark landed + /// in the prep-to-park window (RunningNotified); the word is already back + /// to Queued and the caller must enqueue — the lost-wakeup window, closed. + #[must_use] + pub(crate) fn park_return(&self, gen: u32) -> bool { + match self.0.compare_exchange( + pack(gen, ST_RUNNING), + pack(gen, ST_PARKED), + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => true, + Err(actual) => { + debug_assert_eq!( + actual, + pack(gen, ST_RUNNING_NOTIFIED), + "park return from invalid word {actual:#x}" + ); + self.0.store(pack(gen, ST_QUEUED), Ordering::Release); + false + } + } + } + + /// The unpark protocol — the one way anything outside the scheduler makes + /// an actor runnable. See [`Unpark`] for the caller's obligations. + #[must_use] + pub(crate) fn unpark(&self, gen: u32) -> Unpark { + loop { + let w = self.load(); + if word_gen(w) != gen { + return Unpark::Noop; + } + match word_state(w) { + ST_PARKED => { + if self + .0 + .compare_exchange( + w, + pack(gen, ST_QUEUED), + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + return Unpark::Enqueue; + } + } + ST_RUNNING => { + if self + .0 + .compare_exchange( + w, + pack(gen, ST_RUNNING_NOTIFIED), + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok() + { + return Unpark::Notified; + } + } + _ => return Unpark::Noop, // Queued | Notified | Done | Vacant + } + } + } + + /// Finalize: Running | RunningNotified → Done. Called by the scheduler + /// that just ran the actor to completion (so those are the only legal + /// prior states), under the slot's cold lock so join's check-or-register + /// is linearized against it. + pub(crate) fn set_done(&self, gen: u32) { + let prev = self.0.swap(pack(gen, ST_DONE), Ordering::AcqRel); + debug_assert!( + matches!(word_state(prev), ST_RUNNING | ST_RUNNING_NOTIFIED) + && word_gen(prev) == gen, + "finalize from invalid word {prev:#x}" + ); + } + + /// Reclaim: Done → Vacant(gen + 1). The generation bump IS the reclaim: + /// every stale pid is dead from this store onwards. Caller holds the cold + /// lock and has verified eligibility (asserted). + pub(crate) fn reclaim(&self, gen: u32) { + debug_assert_eq!( + self.load(), + pack(gen, ST_DONE), + "reclaim of a non-Done slot" + ); + self.0 + .store(pack(gen.wrapping_add(1), ST_VACANT), Ordering::Release); + } +} + +// --------------------------------------------------------------------------- +// loom model tests — RUSTFLAGS="--cfg loom" cargo test --lib --release +// --------------------------------------------------------------------------- + +#[cfg(all(test, loom))] +mod loom_tests { + use super::*; + use loom::sync::atomic::{AtomicBool, AtomicUsize}; + use loom::sync::Arc; + use loom::thread; + use std::sync::atomic::Ordering as O; + + /// THE lost-wakeup theorem. A waiter registers a condition check then + /// parks (as every parking site does); a waker sets the condition then + /// unparks. In every interleaving the waiter must end up runnable — + /// parked-forever-with-condition-set must be unreachable. + #[test] + fn no_lost_wakeup_park_vs_unpark() { + loom::model(|| { + let word = Arc::new(StateWord::new()); + word.publish_queued(0); + assert!(word.try_claim(0)); // scheduler claimed: actor Running + + let ready = Arc::new(AtomicBool::new(false)); + let enqueues = Arc::new(AtomicUsize::new(0)); + + // Waker: make the condition true, then unpark. + let w = word.clone(); + let r = ready.clone(); + let e = enqueues.clone(); + let waker = thread::spawn(move || { + r.store(true, O::SeqCst); + if w.unpark(0) == Unpark::Enqueue { + e.fetch_add(1, O::SeqCst); + } + }); + + // Waiter (as the scheduler executes it): re-check the condition, + // park only if still false; a Notified park-return re-queues. + let parked = if ready.load(O::SeqCst) { + false // condition already visible: doesn't park at all + } else if word.park_return(0) { + true + } else { + enqueues.fetch_add(1, O::SeqCst); // notified → re-queued + false + }; + + waker.join().unwrap(); + + let w = word.load(); + if parked { + // Parked is only a FINAL state if the waker's unpark moved it + // back to Queued (+ one enqueue). Parked-and-stays-parked + // would be the lost wakeup. + assert_eq!(word_state(w), ST_QUEUED, "lost wakeup: parked forever"); + assert_eq!(enqueues.load(O::SeqCst), 1); + } else { + // Never more than one enqueue (at-most-once-enqueued). + assert!(enqueues.load(O::SeqCst) <= 1); + } + }); + } + + /// Two concurrent unparkers, one parked actor: exactly one wins the + /// enqueue (at-most-once), regardless of interleaving. + #[test] + fn two_unparkers_one_enqueue() { + loom::model(|| { + let word = Arc::new(StateWord::new()); + word.publish_queued(0); + assert!(word.try_claim(0)); + assert!(word.park_return(0)); // actor parked + + let enqueues = Arc::new(AtomicUsize::new(0)); + let mut hs = Vec::new(); + for _ in 0..2 { + let w = word.clone(); + let e = enqueues.clone(); + hs.push(thread::spawn(move || { + if w.unpark(0) == Unpark::Enqueue { + e.fetch_add(1, O::SeqCst); + } + })); + } + for h in hs { + h.join().unwrap(); + } + assert_eq!(enqueues.load(O::SeqCst), 1); + assert_eq!(word_state(word.load()), ST_QUEUED); + }); + } + + /// The ABA theorem: a stale-generation unpark racing reclaim + reuse can + /// never touch the slot's new occupant. + #[test] + fn stale_unpark_never_hits_reused_slot() { + loom::model(|| { + let word = Arc::new(StateWord::new()); + // Gen-0 actor runs to completion. + word.publish_queued(0); + assert!(word.try_claim(0)); + + let w = word.clone(); + let stale = thread::spawn(move || w.unpark(0)); + + // Scheduler: finalize, reclaim, and a new spawn reuses the slot. + word.set_done(0); + word.reclaim(0); + word.publish_queued(1); + + // The stale unpark may have squeezed in only while gen 0 was + // still Running (→ Notified) — in which case set_done's swap + // absorbed it — or it observed Done/Vacant/gen-1 and no-op'd. + // Either way it must never claim an enqueue. + assert_ne!(stale.join().unwrap(), Unpark::Enqueue); + // And the new occupant is exactly where its spawn put it. + assert_eq!(word.load(), pack(1, ST_QUEUED)); + }); + } + + /// Unpark racing the claim itself: whatever the interleaving, the actor + /// is Running or RunningNotified afterwards and nobody enqueued (it was + /// never parked). + #[test] + fn unpark_vs_claim_coalesces() { + loom::model(|| { + let word = Arc::new(StateWord::new()); + word.publish_queued(0); + + let w = word.clone(); + let unparker = thread::spawn(move || w.unpark(0)); + + assert!(word.try_claim(0)); // the entry is ours; claim must win + let r = unparker.join().unwrap(); + assert_ne!(r, Unpark::Enqueue); + let st = word_state(word.load()); + assert!(matches!(st, ST_RUNNING | ST_RUNNING_NOTIFIED)); + }); + } +} diff --git a/src/sync_shim.rs b/src/sync_shim.rs new file mode 100644 index 0000000..08d766b --- /dev/null +++ b/src/sync_shim.rs @@ -0,0 +1,37 @@ +//! std vs loom indirection for the modules that loom model-checks +//! (`slot_state`, `run_queue`). Everything else uses std paths directly — +//! the full runtime (context switches, futexes, real TLS) is not loom-able +//! and is never executed under `cfg(loom)`. +//! +//! Build the loom models with: `RUSTFLAGS="--cfg loom" cargo test --lib --release` + +#[cfg(loom)] +pub(crate) use loom::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; + +#[cfg(not(loom))] +pub(crate) use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; + +/// `UnsafeCell` with loom's `with`/`with_mut` access API; pass-through cost +/// is zero in normal builds (`#[inline]`, newtype over std's cell). +#[cfg(loom)] +pub(crate) use loom::cell::UnsafeCell; + +#[cfg(not(loom))] +pub(crate) struct UnsafeCell(std::cell::UnsafeCell); + +#[cfg(not(loom))] +impl UnsafeCell { + pub(crate) fn new(v: T) -> Self { + Self(std::cell::UnsafeCell::new(v)) + } + + #[inline] + pub(crate) fn with(&self, f: impl FnOnce(*const T) -> R) -> R { + f(self.0.get()) + } + + #[inline] + pub(crate) fn with_mut(&self, f: impl FnOnce(*mut T) -> R) -> R { + f(self.0.get()) + } +}