diff --git a/src/park.rs b/src/park.rs index 2cc5727..9142474 100644 --- a/src/park.rs +++ b/src/park.rs @@ -22,18 +22,20 @@ //! 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). +//! The publish/re-check pair is a store-buffer (Dekker) shape. Two sound +//! resolutions coexist here, chosen per call-site cost profile: the +//! **fence handshake** on the hot producer path +//! ([`Coordinator::wake_one_if_idle`]: publish work; `fence(SeqCst)`; +//! one *Relaxed* mask load — pairing with the consumer's `fetch_or(bit)`; +//! `fence(SeqCst)`; re-check inside [`Coordinator::park`]), so the +//! pure-compute hot path (everyone busy, mask 0) never takes the shared +//! mask line exclusive — the RFC's "one relaxed load" fast path, made +//! sound; and the **same-location-RMW read** (`fetch_or(0)`, in +//! [`Coordinator::wake_one`] / [`Coordinator::idle_mask`]) for the rare +//! paths (chain rule — at most one per wake) where reading the latest +//! mask by modification-order coherence is worth an RMW. The loom models +//! drive the fence pattern end to end (a fence-less plain-load draft +//! would — and did — fail model 1/2 with a lost wake, as it must). //! - **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)` @@ -53,7 +55,7 @@ //! Standalone until the runtime swap (RFC 018 commit 2): nothing outside //! tests constructs a [`Coordinator`] yet. -use crate::sync_shim::{AtomicU64, Ordering}; +use crate::sync_shim::{fence, AtomicU64, Ordering}; use std::time::Instant; /// Sentinel for "no timekeeper" in the holder word. @@ -286,9 +288,17 @@ pub(crate) struct Coordinator { /// 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. + /// Armed deadline as nanos since `origin`, or `NO_DEADLINE`. Written + /// only by the timekeeper arm/disarm protocol. tk_armed: AtomicU64, + /// Earliest KNOWN timer deadline (nanos since `origin`), or + /// `NO_DEADLINE` — independent of whether any scheduler is parked, + /// which is what the timekeeper's `tk_armed` cannot give: under + /// saturation nobody parks and nobody arms, yet due timers must still + /// fire (ratified design point (a)). Maintained under the caller's + /// timers mutex (`note_deadline` on insert, `refresh_deadline` after a + /// pop/peek); read lock-free by the busy-path due-check. + next_deadline: AtomicU64, /// Time origin for the nanos encoding. origin: Instant, } @@ -304,6 +314,7 @@ impl Coordinator { idle: AtomicU64::new(0), tk_holder: AtomicU64::new(NO_TIMEKEEPER), tk_armed: AtomicU64::new(NO_DEADLINE), + next_deadline: AtomicU64::new(NO_DEADLINE), origin: Instant::now(), } } @@ -342,6 +353,11 @@ impl Coordinator { // 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"); + // Fence half of the producer handshake (see `wake_one_if_idle`): + // orders our bit-publish before the re-check's loads, so it pairs + // with the producer's publish→fence→mask-load — at least one side + // must see the other's store, whichever queue backend is in play. + fence(Ordering::SeqCst); // (2) the mandatory post-publish re-check. if recheck() { self.idle.fetch_and(!bit, Ordering::AcqRel); @@ -394,6 +410,23 @@ impl Coordinator { } } + /// The producer-side wake tail (`enqueue`'s fast path, RFC 018 "enqueue + /// wakes"). The caller has just published work (queue push); we fence, + /// then read the mask with ONE Relaxed load — 0 means every scheduler + /// is awake and the pure-compute hot path pays no RMW on the shared + /// mask line. Soundness is the fence handshake (module docs): our + /// fence orders the caller's push before the mask load; the consumer's + /// fence (in `park`) orders its bit-publish before its re-check — at + /// least one side must observe the other's store, so a consumer we + /// miss here is a consumer whose re-check finds the caller's work. + pub(crate) fn wake_one_if_idle(&self) -> bool { + fence(Ordering::SeqCst); + if self.idle.load(Ordering::Relaxed) == 0 { + return false; + } + self.wake_one() + } + /// 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 @@ -470,11 +503,53 @@ impl Coordinator { } /// 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. + /// none). One Relaxed load. pub(crate) fn armed_deadline_nanos(&self) -> u64 { self.tk_armed.load(Ordering::Relaxed) } + + // ----- earliest-deadline snapshot (busy-path due-check) ----- + + /// Record a newly inserted timer deadline. MUST be called under the + /// timers mutex (same serialization rule as `try_arm_timer`), which is + /// why plain compare+store suffices for the min-maintenance. Also runs + /// the timekeeper re-arm check (`timer_inserted`) — one call site for + /// both consequences of an insert. + pub(crate) fn note_deadline(&self, deadline: Instant) { + let n = self.deadline_nanos(deadline); + if n < self.next_deadline.load(Ordering::Relaxed) { + self.next_deadline.store(n, Ordering::Release); + } + self.timer_inserted(deadline); + } + + /// Re-anchor the snapshot to the heap minimum (`None` = heap empty) + /// after a `pop_due` / `clear`. MUST be called under the timers mutex. + pub(crate) fn refresh_deadline(&self, next: Option) { + let n = next.map_or(NO_DEADLINE, |d| self.deadline_nanos(d)); + self.next_deadline.store(n, Ordering::Release); + } + + /// Busy-path due-check: is the earliest known deadline at or past + /// `now`? One Relaxed load when no deadline is armed — the clock is + /// read only when a timer actually exists (matching the old drain + /// phase's is_empty guard), so the pure-compute hot path pays a load + /// and a branch. + pub(crate) fn deadline_due(&self) -> bool { + let n = self.next_deadline.load(Ordering::Relaxed); + n != NO_DEADLINE && self.deadline_nanos(Instant::now()) >= n + } + + /// The earliest-deadline snapshot as an `Instant`, for the idle path's + /// park timeout (`None` = no timer pending). + pub(crate) fn next_deadline_instant(&self) -> Option { + let n = self.next_deadline.load(Ordering::Acquire); + if n == NO_DEADLINE { + None + } else { + self.origin.checked_add(std::time::Duration::from_nanos(n)) + } + } } // --------------------------------------------------------------------------- @@ -662,6 +737,63 @@ mod tests { fn more_than_64_schedulers_asserts() { let _ = Coordinator::new(65); } + + #[test] + fn wake_one_if_idle_noop_on_empty_and_wakes_on_parked() { + let c = Arc::new(Coordinator::new(1)); + assert!(!c.wake_one_if_idle(), "empty mask must be a no-op"); + let c2 = c.clone(); + let t = std::thread::spawn(move || { + assert_eq!(c2.park(0, None, || false), ParkResult::Woken); + }); + let t0 = Instant::now(); + while c.idle_mask() == 0 { + assert!(t0.elapsed() < Duration::from_secs(5), "never parked"); + std::thread::yield_now(); + } + assert!(c.wake_one_if_idle()); + match t.join() { + Ok(()) => {} + Err(p) => std::panic::resume_unwind(p), + } + } + + #[test] + fn deadline_snapshot_min_maintenance_and_due_check() { + let c = Coordinator::new(1); + assert!(!c.deadline_due(), "no deadline: never due"); + assert_eq!(c.next_deadline_instant(), None); + let far = Instant::now() + Duration::from_secs(60); + let near = Instant::now() + Duration::from_millis(1); + c.note_deadline(far); + assert!(!c.deadline_due()); + c.note_deadline(near); // min wins + assert!(c.next_deadline_instant().is_some_and(|d| d <= near)); + c.note_deadline(far); // later insert must NOT raise the snapshot + assert!(c.next_deadline_instant().is_some_and(|d| d <= near)); + std::thread::sleep(Duration::from_millis(2)); + assert!(c.deadline_due(), "past deadline not reported due"); + c.refresh_deadline(Some(far)); + assert!(!c.deadline_due(), "refresh did not re-anchor"); + c.refresh_deadline(None); + assert_eq!(c.next_deadline_instant(), None); + // A deadline at/before origin encodes as 0: always due. + c.note_deadline(Instant::now() - Duration::from_secs(1)); + assert!(c.deadline_due()); + } + + #[test] + fn note_deadline_earlier_wakes_timekeeper_via_snapshot_path() { + // note_deadline must carry the timer_inserted re-arm wake too. + 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)); + c.note_deadline(near); + let r = c.park(0, Some(far), || false); + assert_eq!(r, ParkResult::Woken, "re-arm wake lost through note_deadline"); + c.disarm_timer(0); + } } // --------------------------------------------------------------------------- @@ -691,8 +823,10 @@ mod loom_tests { let c = c.clone(); let item = item.clone(); thread::spawn(move || { + // The enqueue shape: publish, then the fenced fast-path + // wake tail (this is what the runtime's enqueue calls). item.store(1, O::SeqCst); - c.wake_one(); + c.wake_one_if_idle(); }) }; @@ -753,9 +887,10 @@ mod loom_tests { })); } - // Producer (main): two items, ONE wake. + // Producer (main): two items, ONE wake, via the enqueue-shaped + // fenced fast path. items.store(2, O::SeqCst); - c.wake_one(); + c.wake_one_if_idle(); for h in hs { h.join().unwrap(); diff --git a/src/sync_shim.rs b/src/sync_shim.rs index f8c1cd5..901e9cf 100644 --- a/src/sync_shim.rs +++ b/src/sync_shim.rs @@ -6,10 +6,10 @@ //! Build the loom models with: `RUSTFLAGS="--cfg loom" cargo test --lib --release` #[cfg(loom)] -pub(crate) use loom::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +pub(crate) use loom::sync::atomic::{fence, AtomicU64, AtomicUsize, Ordering}; #[cfg(not(loom))] -pub(crate) use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +pub(crate) use std::sync::atomic::{fence, 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