From 7b026cfe56b91514b4516181e17530b9c367ec51 Mon Sep 17 00:00:00 2001 From: smarm Date: Fri, 24 Jul 2026 08:39:58 +0200 Subject: [PATCH] =?UTF-8?q?feat(park):=20scheduler=20coordination=20layer?= =?UTF-8?q?=20=E2=80=94=20parkers,=20idle=20mask,=20wake=20protocol=20(RFC?= =?UTF-8?q?=20018)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Schedulers get an IO-agnostic sleep/wake primitive of their own: one futex Parker per scheduler thread (permit semantics, std::thread::park shaped — closes the check-then-park race), an AtomicU64 idle mask with a set-bit → re-check → wait park protocol, wake_one (highest-bit LIFO, CAS-clear before unpark: exactly one wakeup per call by construction), wake_all for the terminal path, and the timekeeper role — at most one parked scheduler holds the timer deadline, with an atomic armed-deadline snapshot for the busy-path due-check and an insert-side re-arm wake. Deadlines travel as nanosecond timespecs end to end; the wake pipe's as_millis truncation is unrepresentable here. The Dekker publish/re-check shape is resolved by the same-location-RMW handshake (AcqRel), not SeqCst loads; loom verifies exactly this in four models (no-lost-wake, chain propagation, timekeeper handoff, termination), run with LOOM_MAX_PREEMPTIONS=3 — unbounded exploration is impractical for the looped models. Loom/non-Linux builds park on a Mutex+Condvar via sync_shim. Standalone until the runtime swap (next commit): nothing outside tests constructs a Coordinator yet, hence the temporary dead_code allow in lib.rs. --- src/lib.rs | 4 + src/park.rs | 851 +++++++++++++++++++++++++++++++++++++++++++++++ src/sync_shim.rs | 9 + 3 files changed, 864 insertions(+) create mode 100644 src/park.rs diff --git a/src/lib.rs b/src/lib.rs index 2920bc2..a290d37 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,6 +32,10 @@ pub mod introspect; #[cfg(feature = "observer")] pub mod observer; pub mod runtime; +// TEMPORARY dead_code allow: park is standalone until the RFC 018 runtime +// swap (next commit) wires it into schedule_loop/enqueue; the allow dies there. +#[allow(dead_code)] +pub(crate) mod park; pub(crate) mod raw_mutex; pub(crate) mod slot_state; pub(crate) mod sync_shim; diff --git a/src/park.rs b/src/park.rs new file mode 100644 index 0000000..2cc5727 --- /dev/null +++ b/src/park.rs @@ -0,0 +1,851 @@ +//! Scheduler park/wake coordination layer (RFC 018). +//! +//! Schedulers never touch an fd to sleep: they park on a per-thread +//! [`Parker`] and are woken through an idle-mask protocol the runtime owns +//! outright. IO backends (epoll today, io_uring later) are *producers* +//! behind a two-call contract — make actors runnable, then wake — which is +//! what makes backend selection tractable (RFC 018 §step-back). +//! +//! Three pieces, all in [`Coordinator`]: +//! +//! - **Parker** (one per scheduler): permit semantics, `std::thread::park` +//! shaped — an unpark delivered before the park sets a permit; the next +//! park consumes it and returns immediately. This single property closes +//! the check-then-park race. Linux: `futex(2)` `FUTEX_WAIT`/`FUTEX_WAKE` +//! (private) with a nanosecond-precision relative `timespec` — the +//! `as_millis` truncation defect of the retired wake pipe is +//! unrepresentable here. Loom / non-Linux: `Mutex` + `Condvar` +//! (the loom models run against this build). +//! - **Idle mask**: an `AtomicU64` bitmask of parked scheduler ids +//! (construction asserts ≤ 64 schedulers). Park protocol: set own bit, +//! run the caller's mandatory post-publish re-check, then wait. A +//! producer that published work before observing our bit has left us +//! work the re-check finds; one that observes the bit wakes us. +//! +//! The publish/re-check pair is a store-buffer (Dekker) shape, and it is +//! resolved by the **same-location-RMW handshake**, not by `SeqCst` +//! loads: every mask read on the producer side ([`Coordinator::wake_one`], +//! [`Coordinator::idle_mask`]) is an atomic RMW (`fetch_or(0)`). An RMW +//! always reads the *latest* value in the mask's modification order, so +//! a producer can never miss a bit that a consumer's `fetch_or(bit)` +//! has already written; conversely, a consumer whose `fetch_or(bit)` +//! lands *after* the producer's release-RMW reads-from it, acquires the +//! producer's prior work publication, and its re-check finds the work. +//! Both directions hold under plain acquire/release — the loom models +//! verify exactly this (loom models `SeqCst` as `AcqRel`, so a +//! plain-load design would — and did — fail model 1/2 with a lost wake). +//! - **Timekeeper**: at most one parked scheduler holds the timer +//! deadline (RFC 018 §timers) so a timer expiry wakes one scheduler, +//! not a herd. The role is an atomic `(holder id, armed deadline)` +//! pair; a timer insertion with an earlier deadline wakes the holder to +//! re-peek. Arm / insert-check MUST be serialized by the caller (the +//! timers mutex in the runtime) — the atomics exist so the *busy-path +//! due-check* (one Relaxed load, [`Coordinator::armed_deadline_nanos`]) +//! and the wake stay lock-free. All races are biased over-wake: a +//! spurious permit costs one failed pop; a missed wake would cost a +//! stranded actor, and is unrepresentable under the serialization rule. +//! +//! Every wake here is *at most one* futex round-trip and wakes *exactly +//! one* scheduler by construction (`wake_one` CASes a bit clear before +//! unparking its owner) — there is no shared level-triggered anything +//! left to herd on. +//! +//! Standalone until the runtime swap (RFC 018 commit 2): nothing outside +//! tests constructs a [`Coordinator`] yet. + +use crate::sync_shim::{AtomicU64, Ordering}; +use std::time::Instant; + +/// Sentinel for "no timekeeper" in the holder word. +const NO_TIMEKEEPER: u64 = u64::MAX; +/// Sentinel for "no armed deadline" in the deadline word. Also what the +/// busy-path due-check compares against: `now_nanos < armed` is one branch. +pub(crate) const NO_DEADLINE: u64 = u64::MAX; + +/// Outcome of a [`Coordinator::park`] call. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum ParkResult { + /// A permit was consumed (wake delivered before or during the park). + Woken, + /// The deadline passed with no wake. Only possible when a deadline was + /// supplied (and never under loom, which has no time — see `park`). + TimedOut, + /// The post-publish re-check found work; the thread never blocked. + /// A permit may still be pending (a racing `wake_one` picked us after + /// the bit was set) — it will surface as one spurious `Woken` on a + /// later park. Benign: over-wake by design. + WorkFound, +} + +// --------------------------------------------------------------------------- +// Parker — permit-semantics thread parking +// --------------------------------------------------------------------------- + +/// Linux, non-loom: futex on a state word. +/// +/// States: EMPTY (no permit, nobody waiting), PARKED (a thread is, or is +/// about to be, in `futex_wait`), NOTIFIED (permit pending). The classic +/// std-parker protocol: `unpark` swaps to NOTIFIED and futex-wakes iff it +/// displaced PARKED; `park` CASes EMPTY→PARKED, waits, and consumes +/// NOTIFIED on every exit path. +#[cfg(all(target_os = "linux", not(loom)))] +mod parker { + use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Instant; + + const EMPTY: u32 = 0; + const PARKED: u32 = 1; + const NOTIFIED: u32 = 2; + + pub(super) struct Parker { + state: AtomicU32, + } + + impl Parker { + pub(super) fn new() -> Self { + Self { state: AtomicU32::new(EMPTY) } + } + + /// Returns `true` = woken (permit consumed), `false` = timed out. + pub(super) fn park(&self, deadline: Option) -> bool { + // Fast path: consume a pending permit without blocking. + if self + .state + .compare_exchange(EMPTY, PARKED, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + // Only NOTIFIED can be here (one thread parks at a time). + self.state.store(EMPTY, Ordering::Release); + return true; + } + loop { + let timeout = match deadline { + None => None, + Some(d) => { + let now = Instant::now(); + if d <= now { + // Deadline passed: cancel the park. The swap + // races a concurrent unpark — if it delivered + // NOTIFIED first, report Woken (never lose a + // permit). + return self.state.swap(EMPTY, Ordering::AcqRel) == NOTIFIED; + } + Some(d - now) + } + }; + futex_wait(&self.state, PARKED, timeout); + if self + .state + .compare_exchange(NOTIFIED, EMPTY, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + return true; + } + // Spurious wake or timeout with state still PARKED: loop — + // the deadline check at the top decides. + } + } + + pub(super) fn unpark(&self) { + if self.state.swap(NOTIFIED, Ordering::AcqRel) == PARKED { + futex_wake(&self.state, 1); + } + } + } + + /// `FUTEX_WAIT` with a *relative* nanosecond timeout (`CLOCK_MONOTONIC` + /// per futex(2) for relative waits). No millisecond conversion anywhere: + /// the timespec carries the full sub-ms remainder (RFC 018 kills the + /// `as_millis` truncation structurally). + fn futex_wait(word: &AtomicU32, expected: u32, timeout: Option) { + let ts; + let ts_ptr: *const libc::timespec = match timeout { + Some(d) => { + ts = libc::timespec { + tv_sec: d.as_secs() as libc::time_t, + tv_nsec: d.subsec_nanos() as libc::c_long, + }; + &ts + } + None => std::ptr::null(), + }; + // Errors (EAGAIN: word changed; ETIMEDOUT; EINTR) all mean "return + // and let the caller's state machine decide" — deliberately ignored. + unsafe { + libc::syscall( + libc::SYS_futex, + word.as_ptr(), + libc::FUTEX_WAIT | libc::FUTEX_PRIVATE_FLAG, + expected, + ts_ptr, + ); + } + } + + fn futex_wake(word: &AtomicU32, n: u32) { + unsafe { + libc::syscall( + libc::SYS_futex, + word.as_ptr(), + libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG, + n, + ); + } + } +} + +/// Loom / non-Linux: `Mutex` permit + `Condvar` — loom's own model +/// of a parker, and the portable fallback. Under loom the deadline is +/// ignored (loom has no time); the models exercise wake paths only. +#[cfg(any(loom, not(target_os = "linux")))] +mod parker { + use crate::sync_shim::{Condvar, Mutex}; + use std::time::Instant; + + pub(super) struct Parker { + permit: Mutex, + cv: Condvar, + } + + impl Parker { + pub(super) fn new() -> Self { + Self { permit: Mutex::new(false), cv: Condvar::new() } + } + + /// Returns `true` = woken (permit consumed), `false` = timed out. + pub(super) fn park(&self, deadline: Option) -> bool { + let mut permit = match self.permit.lock() { + Ok(g) => g, + Err(_) => panic!("smarm: parker permit lock poisoned (core corrupt)"), + }; + loop { + if *permit { + *permit = false; + return true; + } + #[cfg(loom)] + { + // Loom has no clock: block until a wake. Models must + // deliver one (a park nobody wakes is a real deadlock + // and loom reports it as such). + let _ = deadline; + permit = match self.cv.wait(permit) { + Ok(g) => g, + Err(_) => panic!("smarm: parker cv poisoned (core corrupt)"), + }; + } + #[cfg(not(loom))] + { + match deadline { + None => { + permit = match self.cv.wait(permit) { + Ok(g) => g, + Err(_) => panic!("smarm: parker cv poisoned (core corrupt)"), + }; + } + Some(d) => { + let now = Instant::now(); + if d <= now { + return false; + } + permit = match self.cv.wait_timeout(permit, d - now) { + Ok((g, _)) => g, + Err(_) => panic!("smarm: parker cv poisoned (core corrupt)"), + }; + } + } + } + } + } + + pub(super) fn unpark(&self) { + let mut permit = match self.permit.lock() { + Ok(g) => g, + Err(_) => panic!("smarm: parker permit lock poisoned (core corrupt)"), + }; + *permit = true; + self.cv.notify_one(); + } + } +} + +use parker::Parker; + +// --------------------------------------------------------------------------- +// Coordinator — idle mask + wake protocol + timekeeper +// --------------------------------------------------------------------------- + +pub(crate) struct Coordinator { + parkers: Box<[Parker]>, + /// Bit `i` set = scheduler `i` is parked or committed to parking (set + /// before the re-check; cleared by `wake_one`'s CAS or by the parker + /// itself on return). AcqRel same-location-RMW handshake — see module + /// docs (no SeqCst needed: every producer-side read is an RMW). + idle: AtomicU64, + /// Timekeeper holder id, or `NO_TIMEKEEPER`. Written under the + /// caller's timer serialization (arm/disarm/insert-check); read + /// lock-free by the insert wake path. + tk_holder: AtomicU64, + /// Armed deadline as nanos since `origin`, or `NO_DEADLINE`. The + /// busy-path due-check reads this Relaxed. + tk_armed: AtomicU64, + /// Time origin for the nanos encoding. + origin: Instant, +} + +impl Coordinator { + pub(crate) fn new(schedulers: usize) -> Self { + assert!( + (1..=64).contains(&schedulers), + "smarm: scheduler count must be 1..=64 (idle mask is one u64); got {schedulers}" + ); + Self { + parkers: (0..schedulers).map(|_| Parker::new()).collect(), + idle: AtomicU64::new(0), + tk_holder: AtomicU64::new(NO_TIMEKEEPER), + tk_armed: AtomicU64::new(NO_DEADLINE), + origin: Instant::now(), + } + } + + /// Encode a deadline for the armed snapshot / busy-path compare. + /// Saturating: a deadline at-or-before `origin` encodes as 0 (always + /// due), one beyond ~584 years as `NO_DEADLINE - 1`. + pub(crate) fn deadline_nanos(&self, deadline: Instant) -> u64 { + let nanos = deadline + .checked_duration_since(self.origin) + .map(|d| d.as_nanos()) + .unwrap_or(0); + if nanos >= NO_DEADLINE as u128 { + NO_DEADLINE - 1 + } else { + nanos as u64 + } + } + + /// Park scheduler `id` until a wake, the deadline, or a positive + /// re-check. Protocol: (1) publish own idle bit (SeqCst), (2) run + /// `recheck` — it MUST re-read the work source with an ordering that + /// pairs with the producer's publish (SeqCst load, or take the mutex + /// the producer publishes under); a `true` aborts the park, (3) block. + pub(crate) fn park( + &self, + id: usize, + deadline: Option, + recheck: impl FnOnce() -> bool, + ) -> ParkResult { + debug_assert!(id < self.parkers.len(), "park: scheduler id out of range"); + let bit = 1u64 << id; + // (1) publish. AcqRel RMW: the acquire half is the handshake — if + // this lands after a producer's mask RMW in modification order, we + // read-from it and the producer's earlier work publication is + // visible to the re-check below (see module docs). + let prev = self.idle.fetch_or(bit, Ordering::AcqRel); + debug_assert_eq!(prev & bit, 0, "park: idle bit already set for this id"); + // (2) the mandatory post-publish re-check. + if recheck() { + self.idle.fetch_and(!bit, Ordering::AcqRel); + return ParkResult::WorkFound; + } + // (3) block. The permit closes the window between the re-check and + // the futex wait: a wake_one that picked us in that window has + // already CASed our bit clear and set the permit. + let woken = self.parkers[id].park(deadline); + // Clear own bit — a no-op when a waker already CASed it clear. + self.idle.fetch_and(!bit, Ordering::AcqRel); + if woken { + ParkResult::Woken + } else { + ParkResult::TimedOut + } + } + + /// Wake exactly one parked scheduler, if any: pick the highest set idle + /// bit (LIFO-ish — warmest cache), CAS it clear, deliver a permit. + /// Empty mask = no-op (everyone is awake and will find work by + /// popping). Returns whether a scheduler was woken. + pub(crate) fn wake_one(&self) -> bool { + // RMW read, not a load: reads the latest mask by modification-order + // coherence, closing the Dekker race with a parking consumer (see + // module docs). The release side of the RMW is what a later-parking + // consumer's fetch_or acquires to make its re-check sound. + let mut mask = self.idle.fetch_or(0, Ordering::AcqRel); + loop { + if mask == 0 { + return false; + } + let id = 63 - mask.leading_zeros() as usize; // highest set bit + let bit = 1u64 << id; + // The CAS is the exactly-one guarantee: whoever clears the bit + // owns the wake; a racing wake_one retries on the observed value + // (coherence: a failed CAS can never read older than `mask`). + match self.idle.compare_exchange( + mask, + mask & !bit, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => { + self.parkers[id].unpark(); + return true; + } + Err(m) => mask = m, + } + } + } + + /// Terminal wake (replaces the AllDone wake-pipe byte): clear the mask + /// and deliver a permit to *every* parker, parked or not. A permit set + /// on a busy scheduler costs one spurious park return — nothing at the + /// terminal boundary. Idempotent. + pub(crate) fn wake_all(&self) { + self.idle.store(0, Ordering::Release); + for p in self.parkers.iter() { + p.unpark(); + } + } + + /// Latest idle mask (RMW read — same handshake as `wake_one`; used by + /// the chain rule: "queue non-empty and mask non-zero ⇒ wake_one + /// again", where missing a just-parked sibling would strand its item). + pub(crate) fn idle_mask(&self) -> u64 { + self.idle.fetch_or(0, Ordering::AcqRel) + } + + // ----- timekeeper ----- + + /// Try to take the timekeeper role for scheduler `id` with `deadline`. + /// MUST be called under the caller's timer serialization (the timers + /// mutex), with `deadline` the heap minimum peeked under that same + /// hold — this is what makes the insert-check race-free. Returns + /// whether the role was taken (false = someone else holds it; park + /// with no deadline). + pub(crate) fn try_arm_timer(&self, id: usize, deadline: Instant) -> bool { + debug_assert!(id < self.parkers.len(), "try_arm_timer: id out of range"); + if self + .tk_holder + .compare_exchange(NO_TIMEKEEPER, id as u64, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { + return false; + } + // Holder-then-deadline order: an insert-check that sees the holder + // with the deadline still NO_DEADLINE compares `new < MAX` = true + // and over-wakes — the benign direction. (Under the mandated timer + // serialization this interleaving cannot occur anyway.) + self.tk_armed.store(self.deadline_nanos(deadline), Ordering::SeqCst); + true + } + + /// Release the timekeeper role (the holder, on wake, before it + /// re-peeks/fires). Callable without the timer serialization: a + /// racing insert may wake a no-longer-holder — over-wake, benign. + pub(crate) fn disarm_timer(&self, id: usize) { + debug_assert_eq!( + self.tk_holder.load(Ordering::SeqCst), + id as u64, + "disarm_timer by a non-holder" + ); + // Deadline first: a concurrent insert-check then sees NO_DEADLINE + // and skips the (now pointless) wake instead of waking a stale + // holder id. Either order is correct; this one wastes less. + self.tk_armed.store(NO_DEADLINE, Ordering::SeqCst); + self.tk_holder.store(NO_TIMEKEEPER, Ordering::SeqCst); + } + + /// Insert-side re-arm check: if `deadline` is earlier than the armed + /// snapshot, wake the timekeeper to re-peek. MUST be called under the + /// same timer serialization as `try_arm_timer` (see there). + pub(crate) fn timer_inserted(&self, deadline: Instant) { + if self.deadline_nanos(deadline) < self.tk_armed.load(Ordering::SeqCst) { + let holder = self.tk_holder.load(Ordering::SeqCst); + if holder != NO_TIMEKEEPER { + // Direct unpark, not wake_one: the wake targets the + // timekeeper specifically (it must re-peek the heap). Its + // idle bit stays set until it returns from park — a + // concurrent wake_one may pick it too; over-wake, benign. + self.parkers[holder as usize].unpark(); + } + } + } + + /// The armed-deadline snapshot (nanos since origin; `NO_DEADLINE` = + /// none). One Relaxed load — the busy-path due-check (RFC 018 design + /// point (a)) reads this every scheduler-loop iteration. + pub(crate) fn armed_deadline_nanos(&self) -> u64 { + self.tk_armed.load(Ordering::Relaxed) + } +} + +// --------------------------------------------------------------------------- +// Unit tests (std build) +// --------------------------------------------------------------------------- + +#[cfg(all(test, not(loom)))] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering as O}; + use std::sync::Arc; + use std::time::{Duration, Instant}; + + #[test] + fn permit_before_park_returns_immediately() { + let c = Coordinator::new(1); + // Deliver the wake first (nobody parked: wake_one no-ops on the + // mask, so use the timekeeper-direct path? No — permit semantics + // are the parker's own; exercise via wake_all which permits all). + c.wake_all(); + let t0 = Instant::now(); + let r = c.park(0, None, || false); + assert_eq!(r, ParkResult::Woken); + assert!(t0.elapsed() < Duration::from_millis(100), "park blocked despite permit"); + } + + #[test] + fn submillisecond_deadline_is_honored() { + // Regression for the retired as_millis truncation: a 500µs deadline + // must neither busy-return instantly forever nor round to 0/∞. + let c = Coordinator::new(1); + let t0 = Instant::now(); + let r = c.park(0, Some(t0 + Duration::from_micros(500)), || false); + let dt = t0.elapsed(); + assert_eq!(r, ParkResult::TimedOut); + assert!(dt >= Duration::from_micros(400), "woke too early: {dt:?}"); + assert!(dt < Duration::from_millis(50), "overslept: {dt:?}"); + } + + #[test] + fn recheck_true_aborts_park_and_clears_bit() { + let c = Coordinator::new(2); + let r = c.park(1, None, || true); + assert_eq!(r, ParkResult::WorkFound); + assert_eq!(c.idle_mask(), 0, "bit not cleared after WorkFound"); + } + + #[test] + fn recheck_observes_own_bit_published() { + let c = Coordinator::new(2); + let seen = std::cell::Cell::new(0u64); + let r = c.park(1, None, || { + seen.set(c.idle_mask()); + true + }); + assert_eq!(r, ParkResult::WorkFound); + assert_eq!(seen.get() & 0b10, 0b10, "bit not published before re-check"); + } + + #[test] + fn wake_one_wakes_exactly_one_of_n() { + const N: usize = 4; + let c = Arc::new(Coordinator::new(N)); + let woken = Arc::new(AtomicUsize::new(0)); + let mut ts = Vec::new(); + for id in 0..N { + let c = c.clone(); + let woken = woken.clone(); + ts.push(std::thread::spawn(move || { + let r = c.park(id, None, || false); + assert_eq!(r, ParkResult::Woken); + woken.fetch_add(1, O::SeqCst); + })); + } + // Wait until all four are published idle. + let t0 = Instant::now(); + while c.idle_mask().count_ones() != N as u32 { + assert!(t0.elapsed() < Duration::from_secs(5), "threads never parked"); + std::thread::yield_now(); + } + assert!(c.wake_one()); + // Exactly one wakes; give the others a beat to (incorrectly) wake. + let t0 = Instant::now(); + while woken.load(O::SeqCst) == 0 { + assert!(t0.elapsed() < Duration::from_secs(5), "wake_one woke nobody"); + std::thread::yield_now(); + } + std::thread::sleep(Duration::from_millis(100)); + assert_eq!(woken.load(O::SeqCst), 1, "wake_one woke more than one"); + assert_eq!(c.idle_mask().count_ones(), (N - 1) as u32); + c.wake_all(); + for t in ts { + match t.join() { + Ok(()) => {} + Err(p) => std::panic::resume_unwind(p), + } + } + assert_eq!(woken.load(O::SeqCst), N); + assert_eq!(c.idle_mask(), 0); + } + + #[test] + fn wake_one_prefers_highest_bit() { + let c = Arc::new(Coordinator::new(3)); + let woken_id = Arc::new(AtomicUsize::new(usize::MAX)); + let mut ts = Vec::new(); + for id in 0..3 { + let c = c.clone(); + let woken_id = woken_id.clone(); + ts.push(std::thread::spawn(move || { + if c.park(id, None, || false) == ParkResult::Woken { + let _ = woken_id.compare_exchange(usize::MAX, id, O::SeqCst, O::SeqCst); + } + })); + } + let t0 = Instant::now(); + while c.idle_mask() != 0b111 { + assert!(t0.elapsed() < Duration::from_secs(5)); + std::thread::yield_now(); + } + assert!(c.wake_one()); + let t0 = Instant::now(); + while woken_id.load(O::SeqCst) == usize::MAX { + assert!(t0.elapsed() < Duration::from_secs(5)); + std::thread::yield_now(); + } + assert_eq!(woken_id.load(O::SeqCst), 2, "LIFO-ish: highest bit first"); + c.wake_all(); + for t in ts { + let _ = t.join(); + } + } + + #[test] + fn wake_one_on_empty_mask_is_noop() { + let c = Coordinator::new(2); + assert!(!c.wake_one()); + assert_eq!(c.idle_mask(), 0); + } + + #[test] + fn timekeeper_arm_is_exclusive_and_snapshot_readable() { + let c = Coordinator::new(2); + let d2 = Instant::now() + Duration::from_secs(10); + let d1 = Instant::now() + Duration::from_secs(1); + assert_eq!(c.armed_deadline_nanos(), NO_DEADLINE); + assert!(c.try_arm_timer(0, d2)); + assert!(!c.try_arm_timer(1, d1), "second arm must fail while held"); + assert_eq!(c.armed_deadline_nanos(), c.deadline_nanos(d2)); + c.disarm_timer(0); + assert_eq!(c.armed_deadline_nanos(), NO_DEADLINE); + assert!(c.try_arm_timer(1, d1), "role must be re-takeable after disarm"); + c.disarm_timer(1); + } + + #[test] + fn earlier_insert_wakes_timekeeper() { + let c = Coordinator::new(2); + let far = Instant::now() + Duration::from_secs(60); + let near = Instant::now() + Duration::from_millis(1); + assert!(c.try_arm_timer(0, far)); + // Holder not yet parked: the wake must land as a permit. + c.timer_inserted(near); + let t0 = Instant::now(); + let r = c.park(0, Some(far), || false); + assert_eq!(r, ParkResult::Woken, "re-arm wake lost"); + assert!(t0.elapsed() < Duration::from_secs(5), "slept toward the stale deadline"); + c.disarm_timer(0); + } + + #[test] + fn later_insert_does_not_wake_timekeeper() { + let c = Coordinator::new(2); + let near = Instant::now() + Duration::from_millis(20); + let far = Instant::now() + Duration::from_secs(60); + assert!(c.try_arm_timer(0, near)); + c.timer_inserted(far); // later than armed: no wake + let r = c.park(0, Some(near), || false); + assert_eq!(r, ParkResult::TimedOut, "spurious wake for a later insert"); + c.disarm_timer(0); + } + + #[test] + #[should_panic(expected = "1..=64")] + fn more_than_64_schedulers_asserts() { + let _ = Coordinator::new(65); + } +} + +// --------------------------------------------------------------------------- +// loom models — RUSTFLAGS="--cfg loom" cargo test --lib --release park +// --------------------------------------------------------------------------- + +#[cfg(all(test, loom))] +mod loom_tests { + use super::*; + use loom::sync::atomic::{AtomicU64 as LAtomicU64, Ordering as O}; + use loom::sync::{Arc, Mutex as LMutex}; + use loom::thread; + use std::time::{Duration, Instant}; + + /// RFC 018 loom model 1 — no lost wake. + /// producer{publish item; wake_one} ∥ consumer{set bit; re-check; park}: + /// the consumer always observes the item or a permit; it can never + /// sleep past a published item (loom's deadlock detector is the + /// assertion — a consumer parked forever fails the model). + #[test] + fn no_lost_wake() { + loom::model(|| { + let c = Arc::new(Coordinator::new(1)); + let item = Arc::new(LAtomicU64::new(0)); + + let prod = { + let c = c.clone(); + let item = item.clone(); + thread::spawn(move || { + item.store(1, O::SeqCst); + c.wake_one(); + }) + }; + + // Consumer: loop until the item is popped. Guarded load+CAS, + // not a blind swap — a swap writes 0 even when empty, and + // coherence allows that write to land after the producer's + // store in modification order, destroying the item (a model + // bug loom caught in an earlier draft of this test). + loop { + if item.load(O::SeqCst) == 1 + && item.compare_exchange(1, 0, O::SeqCst, O::SeqCst).is_ok() + { + break; + } + let _ = c.park(0, None, || item.load(O::SeqCst) == 1); + } + prod.join().unwrap(); + }); + } + + /// RFC 018 loom model 2 — chain propagation. + /// Two items, two sleepers, ONE producer wake: the chain rule (a woken + /// consumer that sees surplus work and a non-empty mask wakes again) + /// must get both items consumed with no further producer action. + #[test] + fn chain_propagation() { + loom::model(|| { + let c = Arc::new(Coordinator::new(2)); + let items = Arc::new(LAtomicU64::new(0)); + let consumed = Arc::new(LAtomicU64::new(0)); + + let mut hs = Vec::new(); + for id in 0..2usize { + let c = c.clone(); + let items = items.clone(); + let consumed = consumed.clone(); + hs.push(thread::spawn(move || loop { + if consumed.load(O::SeqCst) == 2 { + c.wake_all(); // release a sibling still parked + break; + } + let cur = items.load(O::SeqCst); + if cur > 0 + && items + .compare_exchange(cur, cur - 1, O::SeqCst, O::SeqCst) + .is_ok() + { + consumed.fetch_add(1, O::SeqCst); + // THE CHAIN RULE: surplus + idle sibling ⇒ wake. + if items.load(O::SeqCst) > 0 && c.idle_mask() != 0 { + c.wake_one(); + } + continue; + } + let _ = c.park(id, None, || { + items.load(O::SeqCst) > 0 || consumed.load(O::SeqCst) == 2 + }); + })); + } + + // Producer (main): two items, ONE wake. + items.store(2, O::SeqCst); + c.wake_one(); + + for h in hs { + h.join().unwrap(); + } + assert_eq!(consumed.load(O::SeqCst), 2); + assert_eq!(items.load(O::SeqCst), 0); + }); + } + + /// RFC 018 loom model 3 — timekeeper handoff. + /// An earlier-deadline insert racing the parking timekeeper: the + /// earlier deadline is always honored — either the timekeeper armed it + /// directly (insert landed first under the timers lock) or the insert + /// wakes the timekeeper to re-peek. A timekeeper sleeping toward the + /// stale later deadline would deadlock the model (loom has no time). + #[test] + fn timekeeper_handoff() { + loom::model(|| { + let origin = Instant::now(); + let d_far = origin + Duration::from_secs(60); + let d_near = origin + Duration::from_secs(1); + + let c = Arc::new(Coordinator::new(1)); + // The timers-mutex stand-in: heap min under a lock. + let heap_min = Arc::new(LMutex::new(d_far)); + + let tk = { + let c = c.clone(); + let heap_min = heap_min.clone(); + thread::spawn(move || { + // Peek + arm under the lock (the serialization rule). + let armed = { + let g = heap_min.lock().unwrap(); + let min = *g; + assert!(c.try_arm_timer(0, min)); + min + }; + if armed == d_far { + // Insert hadn't landed: it MUST wake us. Parking + // toward d_far with no wake = model deadlock. + let r = c.park(0, Some(armed), || false); + assert_eq!(r, ParkResult::Woken, "re-arm wake lost"); + } + // Woken (or armed the near deadline directly): re-peek. + c.disarm_timer(0); + let g = heap_min.lock().unwrap(); + assert_eq!(*g, d_near, "earlier deadline not visible on re-peek"); + }) + }; + + // Inserter (main): publish the earlier deadline under the lock, + // then the insert-check. + { + let mut g = heap_min.lock().unwrap(); + *g = d_near; + c.timer_inserted(d_near); + } + + tk.join().unwrap(); + }); + } + + /// RFC 018 loom model 4 — termination. + /// The AllDone verdict: producer flips done and wake_all()s; consumers + /// must never park forever past done (park's re-check + wake_all's + /// permits close every interleaving; a stuck consumer = loom deadlock). + #[test] + fn termination_no_park_past_done() { + loom::model(|| { + let c = Arc::new(Coordinator::new(2)); + let done = Arc::new(LAtomicU64::new(0)); + + let mut hs = Vec::new(); + for id in 0..2usize { + let c = c.clone(); + let done = done.clone(); + hs.push(thread::spawn(move || loop { + if done.load(O::SeqCst) == 1 { + break; + } + let _ = c.park(id, None, || done.load(O::SeqCst) == 1); + })); + } + + done.store(1, O::SeqCst); + c.wake_all(); + + for h in hs { + h.join().unwrap(); + } + }); + } +} diff --git a/src/sync_shim.rs b/src/sync_shim.rs index 08d766b..f8c1cd5 100644 --- a/src/sync_shim.rs +++ b/src/sync_shim.rs @@ -11,6 +11,15 @@ pub(crate) use loom::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; #[cfg(not(loom))] pub(crate) use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +// park.rs condvar-parker (loom + non-Linux builds only; the Linux non-loom +// build parks on a futex and never touches these — gating them identically +// keeps the default build free of unused imports). +#[cfg(loom)] +pub(crate) use loom::sync::{Condvar, Mutex}; + +#[cfg(all(not(loom), not(target_os = "linux")))] +pub(crate) use std::sync::{Condvar, Mutex}; + /// `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)]