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

feat(safety): phase 5 — loom model checking + invariant audit/assertion sweep

Browse Source 
State machine extracted to src/slot_state.rs as a standalone unit (StateWord:
publish_queued / try_claim / yield_return / park_return / unpark / set_done /
reclaim, plus the Status view for cold paths). runtime.rs keeps the protocol
rationale and consumes the mechanism; every transition self-asserts its
precondition per the assert-the-invariants rule.

src/sync_shim.rs: std vs loom indirection (atomics + UnsafeCell with the
with/with_mut access API) for the two loom-modeled modules. Loom models run
the PRODUCTION code, not a replica:
- slot_state: no-lost-wakeup (park vs unpark), two-unparkers-one-enqueue,
  stale-unpark-never-hits-reused-slot (the ABA theorem), unpark-vs-claim.
- run_queue: mpmc exactly-once through a lap wraparound, push/pop race,
  striped two-producer drain.
RUSTFLAGS="--cfg loom" cargo test --lib --release — 7 models, all pass.
RawMutex deliberately not loom-modeled (futexes can't be; textbook mutex3
with stress + unwind coverage).

Assertion sweep (invariants now checked at the point of reliance):
- enqueue debug-asserts the word reads EXACTLY (gen, Queued) — the
  at-most-once-enqueued invariant the ring capacity proof leans on.
- RawMutex enforces the leaf rule mechanically: debug-build thread-local
  held-count, panics at the acquisition that violates it.
- live_actors underflow (double finalize) asserted.
- StateWord transition preconditions asserted (yield/park/done/reclaim/
  publish/claim).

Audit: with_runtime, RawMutex guards, run-queue ops, and trace::record all
gate preemption (and thereby the stop sentinel) for their span; trace was
already self-gating. Sole remaining exception is the channel std MutexGuard
— the documented first post-v0.5 fast-follow.

Review fixes to the branched-in work:
- slot_state::status_for mapped (matching gen, Vacant) to Stale instead of
  unreachable!: Pid::new is public, so a forged/never-issued pid (e.g.
  monitor(Pid::new(5,0)) on a fresh slab) could reach it — "no such actor"
  is the correct total answer; issued pids still can't get there.
- Tightened enqueue's assert from Status::Live to the exact (gen, Queued)
  word its own comment argues for.

Validated: 22 suites in debug (all asserts + leaf counter live) for
rq-mutex/rq-mpmc/rq-striped, release for rq-mutex, smarm-trace build +
stress, loom 7/7.
This commit is contained in:
Claude
2026-06-09 21:27:45 +00:00
parent 6d9f3698d4
commit 039703dbeb
9 changed files with 651 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib.rs
+2
View File
@@ -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;
src/raw_mutex.rs
+31
View File
@@ -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<u32> = 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<T> {
state: AtomicU32,
data: UnsafeCell<T>,
@@ -69,6 +98,7 @@ impl<T> RawMutex<T> {
// 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<T> 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));
}
src/run_queue.rs
+94 -5
View File
@@ -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());
});
}
}
src/runtime.rs
+55 -127
View File
@@ -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<dyn FnOnce() + Send>;
/// 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<u64, u64> {
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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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<RuntimeInner>, 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);
}
}
}
src/scheduler.rs
+25 -19
View File
@@ -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")
src/slot_state.rs
+381
View File
@@ -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));
});
}
}
src/sync_shim.rs
+37
View File
@@ -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<T>(std::cell::UnsafeCell<T>);
#[cfg(not(loom))]
impl<T> UnsafeCell<T> {
pub(crate) fn new(v: T) -> Self {
Self(std::cell::UnsafeCell::new(v))
}
#[inline]
pub(crate) fn with<R>(&self, f: impl FnOnce(*const T) -> R) -> R {
f(self.0.get())
}
#[inline]
pub(crate) fn with_mut<R>(&self, f: impl FnOnce(*mut T) -> R) -> R {
f(self.0.get())
}
}