From b207c11c950d1ec1ebaa9753488e7e7d1dc45bd0 Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 00:30:57 +0700 Subject: [PATCH 1/8] Resume ready local work after the fairness probe --- CHANGELOG.md | 10 +++++++ Cargo.toml | 2 +- README.md | 4 +-- benches/benchmark.rs | 9 ++---- benches/generic_queue.rs | 1 - src/fanout/mod.rs | 59 +++++++++++++++++++++++++++++++++++++++- src/transfer.rs | 4 +-- 7 files changed, 76 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e648df7..cbcbbd9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,13 @@ +# 0.6.2 (2026-09-12) + +### Fixed + +- After reaching its LIFO fairness quota, a worker still probes other queues + first, then resumes ready local work without an idle backoff. Previously + self-waking tasks paid an unnecessary delay at every quota boundary. +- A deterministic regression test checks that a continuously ready local chain + performs zero idle backoffs; displaced-work fairness remains covered. + # 0.6.0 (2026-09-10) ### Added diff --git a/Cargo.toml b/Cargo.toml index 8bc671c..7b17ac4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ name = "ps-st3" # - Update CHANGELOG.md # - Update if necessary copyright notice in LICENSE-MIT # - Create a "vX.Y.Z" git tag -version = "0.6.1" +version = "0.6.2" authors = ["Serge Barral "] edition = "2021" rust-version = "1.60" diff --git a/README.md b/README.md index bd96ee2..0541c2b 100644 --- a/README.md +++ b/README.md @@ -39,14 +39,14 @@ Add this to your `Cargo.toml`: ```toml [dependencies] -ps-st3 = "0.5.0" +ps-st3 = "0.6.2" ``` For the queues alone, without the pool below and without its one dependency: ```toml [dependencies] -ps-st3 = { version = "0.5.0", default-features = false } +ps-st3 = { version = "0.6.2", default-features = false } ``` diff --git a/benches/benchmark.rs b/benches/benchmark.rs index 5a70fa4..06aa47a 100644 --- a/benches/benchmark.rs +++ b/benches/benchmark.rs @@ -4,8 +4,6 @@ use std::thread::spawn; use std::time::Instant; use criterion::{criterion_group, criterion_main, Criterion}; -use num_cpus; -use st3; mod generic_queue; mod tokio_queue; @@ -109,8 +107,7 @@ pub fn executor + 'static>(name: &str, c: &mut Criterio // Preallocate an array that will be populated with the IDs // of other workers. - let mut other_workers_id = Vec::with_capacity(thread_count - 1); - other_workers_id.resize(thread_count - 1, 0); + let mut other_workers_id = vec![0; thread_count - 1]; // Block until all threads are ready. let start_time = { @@ -144,7 +141,7 @@ pub fn executor + 'static>(name: &str, c: &mut Criterio // queue on a push when many pop/push operations // are made while a stealer is preempted. In // such rare cases, spinning is necessary. - while let Err(_) = worker.push(repeat_count - 1) {} + while worker.push(repeat_count - 1).is_err() {} } } else { // No more local tasks, try to steal. @@ -169,7 +166,7 @@ pub fn executor + 'static>(name: &str, c: &mut Criterio Ok(repeat_count) => { // Re-inject the task if needed. if repeat_count > 0 { - while let Err(_) = worker.push(repeat_count - 1) {} + while worker.push(repeat_count - 1).is_err() {} } continue 'new_task; } diff --git a/benches/generic_queue.rs b/benches/generic_queue.rs index 2a8e07d..91a07d5 100644 --- a/benches/generic_queue.rs +++ b/benches/generic_queue.rs @@ -1,7 +1,6 @@ //! Generic traits for queue benchmarking. use crate::tokio_queue; -use st3; /// Error returned on stealing failure. pub enum GenericStealError { diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 4a11fff..6a80121 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -539,6 +539,8 @@ pub struct Pool { /// on both sides. Too few searchers and work sits in a deque with nobody /// coming for it; too many and they cost more than they redistribute. searching: AtomicUsize, + #[cfg(test)] + idle_backoffs: AtomicUsize, } impl Pool { @@ -624,6 +626,8 @@ impl Pool { tuning, sleeping: AtomicUsize::new(0), searching: AtomicUsize::new(0), + #[cfg(test)] + idle_backoffs: AtomicUsize::new(0), }) } @@ -1045,7 +1049,8 @@ impl Pool { // Sharing must still make progress with one worker, or when all // peers are busy. Service the injector at the local fairness // boundary so displaced work cannot wait forever behind self-wakes. - if lifo_run == lifo_run_limit && self.tuning.share_displaced { + let slot_budget_exhausted = lifo_run == lifo_run_limit; + if slot_budget_exhausted && self.tuning.share_displaced { self.take_from_injector(&mut mine); } lifo_run = 0; @@ -1158,6 +1163,23 @@ impl Pool { continue; } + // The fairness boundary skips the local slot to give other work a + // chance. If those probes found nothing, the slot is still ready + // work: run it instead of paying an idle backoff. Release its lock + // before running, since this job may wake itself into the slot. + let ready_slot = if slot_budget_exhausted { + self.local[w.id].lifo.lock().take() + } else { + None + }; + if let Some(job) = ready_slot { + lifo_run = 1; + spins = 0; + job.run(); + self.local[w.id].completed.fetch_add(1, Ordering::Relaxed); + continue; + } + // Nothing found this round, and nothing left to run. Shutdown is // checked **here**, not only on the park path below: a worker with // a large `rounds_before_park` would otherwise spin out its whole @@ -1173,6 +1195,8 @@ impl Pool { // `Tuning::rounds_before_park` for why the wake, not the park, // costs. if spins < self.tuning.rounds_before_park { + #[cfg(test)] + self.idle_backoffs.fetch_add(1, Ordering::Relaxed); spins += 1; // Back off *without* touching anything shared. Looking again // means a pop off the injector and a probe of every other @@ -1412,6 +1436,39 @@ mod wake_distribution { unparked: Mutex>, } + #[test] + fn fairness_probe_does_not_back_off_with_ready_local_work() { + fn chain(pool: Arc, remaining: usize) -> super::Job { + super::Job::from_boxed(move || { + if remaining == 0 { + pool.shut_down(); + } else { + pool.submit_local(0, chain(pool.clone(), remaining - 1)); + } + }) + } + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning( + 1, + 64, + host, + Tuning::locality() + .with_lifo_run_limit(1) + .with_rounds_before_park(16) + .with_backoff_spins(1), + ); + pool.submit_local(0, chain(pool.clone(), 8)); + assert!(pool.run(pool.runner(0))); + assert_eq!(pool.completed(0), 9); + assert_eq!( + pool.idle_backoffs.load(Ordering::Relaxed), + 0, + "a fairness probe may check other queues, but ready work is not idle" + ); + } + /// Displacement publishes to the injector without waking anybody. /// /// The wake is deliberately absent. It costs a park/unpark round trip per diff --git a/src/transfer.rs b/src/transfer.rs index e386b94..552122f 100644 --- a/src/transfer.rs +++ b/src/transfer.rs @@ -170,8 +170,8 @@ mod tests { for dst_pos in 0..(dst_cap as UnsignedShort * 2) { for count in 0..=(src_cap.min(dst_cap) as UnsignedShort) { let src = buffer(src_cap); - for i in 0..src_cap { - src[i].with_mut(|s| unsafe { + for (i, slot) in src.iter().enumerate() { + slot.with_mut(|s| unsafe { s.write(MaybeUninit::new(0xC0DE_0000u32 + i as u32)) }); } From 754b57f0b18674c740efa2e6f38ae4cf1508599b Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 00:37:27 +0700 Subject: [PATCH 2/8] Describe current displacement and fairness behavior --- src/fanout/mod.rs | 41 ++++++++++++----------------------------- 1 file changed, 12 insertions(+), 29 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 6a80121..631c826 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -113,38 +113,21 @@ pub struct Tuning { /// this worker is holding for somebody else therefore never becomes /// stealable, however long the queue behind it grows. /// - /// Lower means the worker looks at its own queue sooner, which costs - /// nothing when there is nothing there: sharing returns immediately unless - /// the private queue holds at least two jobs. That is what makes this a - /// cheaper answer to a pile-up than routing displaced work to the injector, - /// which pays a wake whether or not anybody needed one. + /// Lower means the worker probes alternative work sources more often. + /// When those probes find nothing, it resumes the ready local slot without + /// an idle backoff. More frequent probes still incur queue and steal costs. pub lifo_run_limit: u32, /// Where a job displaced from the LIFO slot goes. /// - /// `false`, the default, sends it to a private inbox behind the slot: one - /// swap and a `SegQueue` push, reachable by nobody until this worker - /// drains it. `true` sends it to the injector, where any worker can take - /// it, at the cost of a wake. - /// - /// **This is the difference between two failure modes, and both are real.** - /// - /// Private: a worker that accumulates several self-waking tasks keeps all - /// of them. The inbox drains into the private deque, the deque is shared - /// only on a heartbeat counted in `mine.pop_back` calls, and the LIFO slot - /// ahead of that in the loop starves the heartbeat for as long as any task - /// keeps waking itself. Measured on sixteen workers with eight self-waking - /// read-only tasks: four runs in eight collapsed to a single active worker - /// with the other fifteen parked, at exactly the one-thread rate. Every - /// wake was delivered. The work became invisible, which is why this looks - /// like a lost wakeup and is not one. - /// - /// Shared: every displacement is an injector push and a wake. On work that - /// is *lock*-bound rather than CPU-bound there is no parallelism to win and - /// the churn is pure cost. Measured on a 50% update workload: throughput - /// fell from 943,606 to 651,554 while cores busy went from 1.13 to 13.16. - /// - /// So it is a policy, not a fix. Read-mostly work with independent tasks - /// wants `true`; work that contends on row locks wants `false`. + /// `false` sends it to the worker's private inbox. `true` keeps the + /// first displaced job in that inbox and sends further displaced jobs to + /// the shared injector while the inbox is occupied. The owner services + /// that injector at its LIFO fairness boundary; publication itself does + /// not wake another worker. The slot and first inbox job remain private. + /// + /// Sharing makes excess ready work available to peers. Keeping it local + /// avoids queue contention on workloads dominated by lock handoffs. Both + /// policies still probe other work after `lifo_run_limit` slot jobs. pub share_displaced: bool, /// Whether [`Pool::submit_local`] keeps work on the calling worker. /// From 3da107ec54b8543fa5687424836233c1c1459f3f Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 00:55:25 +0700 Subject: [PATCH 3/8] Expose no_std OS parking and close final-search sleep race --- .github/workflows/ci.yml | 7 ++ CHANGELOG.md | 14 ++++ Cargo.toml | 12 +++- README.md | 12 ++++ src/fanout/atomic_host.rs | 134 +++++++++++++++++++++++++++++++++++ src/fanout/host.rs | 145 ++++---------------------------------- src/fanout/mod.rs | 139 ++++++++++++++++++++++++++++++------ src/lib.rs | 4 +- tests/atomic_host.rs | 45 ++++++++++++ tests/fanout.rs | 59 ++++++++++++++++ 10 files changed, 415 insertions(+), 156 deletions(-) create mode 100644 src/fanout/atomic_host.rs create mode 100644 tests/atomic_host.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3c33185..b721137 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,6 +30,10 @@ jobs: - name: Run cargo check run: cargo check + - name: Check OS parking without Rust std + if: matrix.rust == 'stable' + run: cargo check --no-default-features --features atomic-host + test: name: Test suite runs-on: ubicloud-standard-2 @@ -49,6 +53,9 @@ jobs: - name: Run cargo test (fanout with a std host) run: cargo test --release --features host + - name: Test atomic parking without the std host + run: cargo test --release --no-default-features --features atomic-host --test atomic-host + loom: name: Test suite (Loom) runs-on: ubicloud-standard-2 diff --git a/CHANGELOG.md b/CHANGELOG.md index cbcbbd9..856b37e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,21 @@ # 0.6.2 (2026-09-12) +### Added + +- `atomic-host` exposes the blocking OS parker as `AtomicHost` without Rust + `std`; callers supply a monotonic clock and worker lifecycle. `StdHost` + delegates to the same permit protocol. +- `Tuning::with_stealable_inbox` exposes displaced local jobs to idle peers + through a FIFO inbox. The warm LIFO slot remains private; existing presets + keep their previous displacement policy. + + ### Fixed +- The final check after announcing sleep now includes peer queues, closing + the interval between a worker's last search and its sleep announcement. + Sharing uses a publication fence before checking for sleeping peers. + - After reaching its LIFO fairness quota, a worker still probes other queues first, then resumes ready local work without an idle backoff. Previously self-waking tasks paid an unnecessary delay at every quota boundary. diff --git a/Cargo.toml b/Cargo.toml index 7b17ac4..218da20 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,7 +61,12 @@ fanout = ["dep:spin", "dep:crossbeam-queue"] # which on Linux, Android and FreeBSD depends on `libc`, and `libc` 0.2.189 # requires **1.65**. macOS and Windows do not pull it. CI checks 1.60 without # this feature for that reason. -host = ["fanout", "dep:atomic-wait"] +host = ["atomic-host"] + +# OS blocking waits without Rust std. The caller supplies a monotonic clock +# and worker lifecycle. Like `host`, supported platforms come from atomic-wait +# and its libc dependency can require a newer compiler than the portable core. +atomic-host = ["fanout", "dep:atomic-wait"] [dev-dependencies] criterion = "0.3" @@ -91,6 +96,11 @@ name = "fanout" path = "tests/fanout.rs" required-features = ["host"] +[[test]] +name = "atomic-host" +path = "tests/atomic_host.rs" +required-features = ["atomic-host"] + [[example]] name = "floor" path = "examples/floor.rs" diff --git a/README.md b/README.md index 0541c2b..c2de489 100644 --- a/README.md +++ b/README.md @@ -247,3 +247,15 @@ terms, which are explicitly outlined within those assets. Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions. + +## Blocking without Rust std + +Enable `atomic-host` with `default-features = false` for `fanout::AtomicHost`. +Pass the worker count and a `fn() -> u64` monotonic nanosecond clock. The host +uses atomic-wait platform calls, retains wake permits, and has no periodic +parking timeout. Your embedding owns worker creation and shutdown. The `host` +feature adds `StdHost` with an `Instant` clock over this same implementation. + +`Tuning::with_stealable_inbox(true)` lets idle peers take displaced local jobs +while retaining a private warm LIFO slot. It is a Rust callsite option, useful +for independent tasks; measure its extra sharing cost on lock handoffs. diff --git a/src/fanout/atomic_host.rs b/src/fanout/atomic_host.rs new file mode 100644 index 0000000..3308b75 --- /dev/null +++ b/src/fanout/atomic_host.rs @@ -0,0 +1,134 @@ +//! Blocking atomic waits with core/alloc and the platform ABI, without Rust std. +//! +//! Available with `atomic-host` on the platforms supported by `atomic-wait`. +//! The caller provides workers and a monotonic clock; no threads are created. + +use crate::config::AtomicUnsignedLong; +use alloc::vec::Vec; +use core::sync::atomic::{AtomicU32, Ordering}; +use crossbeam_utils::CachePadded; + +use super::Host; + +/// Futex parking, one word per worker. +/// +/// # Where this came from +/// +/// The three-state protocol below is `forte`'s `Semaphore` (`src/latch.rs`, +/// MIT OR Apache-2.0, the same licence as this crate), ported because it is +/// better than what was here and there is no point pretending otherwise. What +/// makes it better is one observation: +/// +/// **Both sides swap the same word, and modification order on a single atomic +/// is total even under `Relaxed`.** So there is no store-load race to close and +/// no `SeqCst` to pay for it. The protocol this replaces used a pool-wide +/// bitmap of who was asleep, which is a different word from the queue, so it +/// *did* need `SeqCst`, and every park attempt and every submit hit that one +/// cache line with a read-modify-write. Measured: parking sooner made the pool +/// spend *more* CPU, up to 818 ms against 445, because the bookkeeping cost +/// more than the sleep saved. +#[derive(Debug)] +pub struct AtomicHost { + slots: Vec>, + clock: fn() -> u64, + /// How many parks woke with no signal to show for it. + spurious: AtomicUnsignedLong, +} + +/// One worker's park word. +/// +/// * `LOCKED` is running, and no signal outstanding. +/// * `ASLEEP` is blocked, or about to block, on this address. +/// * `SIGNAL` is a permit left by an unpark, which the next park consumes +/// instead of sleeping. That is the invariant [`Host`] requires. +#[derive(Debug, Default)] +struct Slot { + state: AtomicU32, +} + +const LOCKED: u32 = 0; +const SIGNAL: u32 = 1; +const ASLEEP: u32 = 2; + +impl AtomicHost { + #[must_use] + /// One park slot per worker and a host-supplied monotonic nanosecond clock. + /// + /// The clock callback must be thread-safe and monotonic. It is not called + /// while parking and is independent of any runtime timer installation. + pub fn new(workers: usize, clock: fn() -> u64) -> Self { + let mut slots = Vec::new(); + slots.resize_with(workers, || CachePadded::new(Slot::default())); + Self { + slots, + clock, + spurious: AtomicUnsignedLong::new(0), + } + } + + /// How many parks woke with no signal to show for it. + /// + /// The platform may return from a wait for its own reasons; the pool loops, + /// so one costs a pass and nothing else. + #[must_use] + #[allow(clippy::useless_conversion)] + pub fn spurious(&self) -> u64 { + u64::from(self.spurious.load(Ordering::Relaxed)) + } +} + +impl Host for AtomicHost { + /// Block until signalled, preserving a permit delivered before the wait. + fn park(&self, worker: usize) { + let state = &self.slots[worker].state; + // **`Acquire`, not `Relaxed`.** Total modification order on this word + // proves which transition happened first and *nothing* about what the + // unparker wrote before it. Consuming a permit has to acquire the + // publication that preceded it, or a worker can take the shutdown + // notification and then read `running` as `true`, park again, and never + // be told a second time. + if state.swap(ASLEEP, Ordering::Acquire) != LOCKED { + // A signal arrived first. Take it and do not sleep. + // + // Compare-exchange rather than a store: another unpark may have + // landed since the swap above, and its permit has to survive rather + // than be overwritten with `LOCKED`. A failure here means exactly + // that, and leaving `SIGNAL` in place is the right outcome. + let _ = state.compare_exchange(ASLEEP, LOCKED, Ordering::Acquire, Ordering::Relaxed); + return; + } + // **No timeout.** `Pool` publishes work and then unparks, and + // `shut_down` unparks every worker, so a wake is never lost; the + // timeout this used to have cost a wakeup per worker per 100 ms on a + // pool with nothing to do and guarded nothing. + while state.load(Ordering::Acquire) == ASLEEP { + atomic_wait::wait(state, ASLEEP); + if state.load(Ordering::Acquire) == ASLEEP { + self.spurious.fetch_add(1, Ordering::Relaxed); + } + } + // Consume the permit that ended the wait, and only that one. A plain + // store would clear a *newer* permit that arrived between the loop + // exiting and this line, which is a lost wakeup. + let _ = state.compare_exchange(SIGNAL, LOCKED, Ordering::Acquire, Ordering::Relaxed); + } + + fn unpark(&self, worker: usize) { + let state = &self.slots[worker].state; + // Leave a permit whatever the state was, because `Host` requires an + // unpark arriving before the park to make that park return at once. + // + // **The system call happens only when somebody is actually on the + // address.** Signalling a running worker is this one swap: no syscall, + // no lock, and no shared line except that worker's own, which is + // padded. That is what lets `Pool` unpark on every submit without + // asking first who is asleep. + if state.swap(SIGNAL, Ordering::Release) == ASLEEP { + atomic_wait::wake_one(state); + } + } + + fn now_ns(&self) -> u64 { + (self.clock)() + } +} diff --git a/src/fanout/host.rs b/src/fanout/host.rs index 1b352a2..e1ac0fe 100644 --- a/src/fanout/host.rs +++ b/src/fanout/host.rs @@ -1,163 +1,44 @@ -//! A `std`-backed [`Host`], behind the `host` feature. -//! -//! [`Host`]: super::Host -//! -//! The core of this crate knows nothing about an operating system: three -//! methods, `park`, `unpark` and `now_ns`. This is what those look like when -//! there is a `std` to implement them with, and it is the reference a target -//! without one replaces. -//! -//! # Minimum Rust version, and platforms -//! -//! **This feature needs more than the crate does.** The crate holds a 1.60 -//! floor; `atomic-wait` here depends on `libc` on Linux, Android and FreeBSD, -//! and `libc` 0.2.189 requires 1.65. macOS and Windows pull nothing extra. -//! -//! Parking is `futex` on Linux, `WaitOnAddress` on Windows, `__ulock_wait` on -//! macOS, and `_umtx_op` on FreeBSD. A platform outside that set has no -//! `StdHost`; it has a [`Host`] trait to implement, which is the point. - -use crossbeam_utils::CachePadded; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +//! Convenience clock adapter for the no_std atomic parker. +use super::{AtomicHost, Host}; use std::time::Instant; -use std::vec::Vec; - -use super::Host; -/// Futex parking, one word per worker. -/// -/// # Where this came from +/// Blocking atomic parking with a std monotonic clock. /// -/// The three-state protocol below is `forte`'s `Semaphore` (`src/latch.rs`, -/// MIT OR Apache-2.0, the same licence as this crate), ported because it is -/// better than what was here and there is no point pretending otherwise. What -/// makes it better is one observation: -/// -/// **Both sides swap the same word, and modification order on a single atomic -/// is total even under `Relaxed`.** So there is no store-load race to close and -/// no `SeqCst` to pay for it. The protocol this replaces used a pool-wide -/// bitmap of who was asleep, which is a different word from the queue, so it -/// *did* need `SeqCst`, and every park attempt and every submit hit that one -/// cache line with a read-modify-write. Measured: parking sooner made the pool -/// spend *more* CPU, up to 818 ms against 445, because the bookkeeping cost -/// more than the sleep saved. +/// For a host without Rust std, use [`AtomicHost`] with its own clock and +/// worker lifecycle. Both hosts use the same permit and wake implementation. #[derive(Debug)] pub struct StdHost { - slots: Vec>, + parking: AtomicHost, origin: Instant, - /// How many parks woke with no signal to show for it. - spurious: AtomicU64, } - -/// One worker's park word. -/// -/// * `LOCKED` is running, and no signal outstanding. -/// * `ASLEEP` is blocked, or about to block, on this address. -/// * `SIGNAL` is a permit left by an unpark, which the next park consumes -/// instead of sleeping. That is the invariant [`Host`] requires. -#[derive(Debug, Default)] -struct Slot { - state: AtomicU32, -} - -const LOCKED: u32 = 0; -const SIGNAL: u32 = 1; -const ASLEEP: u32 = 2; - impl StdHost { + /// One park slot per worker and a clock whose origin is now. #[must_use] - /// One park slot per worker, and a clock whose origin is now. pub fn new(workers: usize) -> Self { - let mut slots = Vec::new(); - slots.resize_with(workers, || CachePadded::new(Slot::default())); Self { - slots, + parking: AtomicHost::new(workers, || 0), origin: Instant::now(), - spurious: AtomicU64::new(0), } } - - /// How many parks woke with no signal to show for it. - /// - /// The platform may return from a wait for its own reasons; the pool loops, - /// so one costs a pass and nothing else. + /// How many waits woke without a signal. #[must_use] pub fn spurious(&self) -> u64 { - self.spurious.load(Ordering::Relaxed) + self.parking.spurious() } - - /// How many parks ended in a timeout rather than a wake. - /// - /// Always zero: parking has no timeout any more. It had one as a backstop - /// from a time when the pool did not wake its own sleepers and the timeout - /// *was* how stealable work got discovered. + /// Always zero: parking has no periodic timeout. #[must_use] #[deprecated(since = "0.6.0", note = "parking no longer times out; see `spurious`")] pub fn timeouts(&self) -> u64 { 0 } } - impl Host for StdHost { - /// **Bounded, but only as a backstop.** It used to be bounded at one - /// millisecond, and that was not a backstop: nothing woke a parked worker - /// when stealable work appeared, so the timeout *was* the notification. - /// A pool with nothing to do woke every worker 800 times a second and burnt - /// 10% of a core doing it. - /// - /// The pool now names its sleepers and wakes one when work is published, so - /// this bound covers only a wakeup genuinely lost below us. A hundred - /// milliseconds is short enough that such a bug costs latency rather than a - /// hang, and long enough that idling is free. fn park(&self, worker: usize) { - let state = &self.slots[worker].state; - // **`Acquire`, not `Relaxed`.** Total modification order on this word - // proves which transition happened first and *nothing* about what the - // unparker wrote before it. Consuming a permit has to acquire the - // publication that preceded it, or a worker can take the shutdown - // notification and then read `running` as `true`, park again, and never - // be told a second time. - if state.swap(ASLEEP, Ordering::Acquire) != LOCKED { - // A signal arrived first. Take it and do not sleep. - // - // Compare-exchange rather than a store: another unpark may have - // landed since the swap above, and its permit has to survive rather - // than be overwritten with `LOCKED`. A failure here means exactly - // that, and leaving `SIGNAL` in place is the right outcome. - let _ = state.compare_exchange(ASLEEP, LOCKED, Ordering::Acquire, Ordering::Relaxed); - return; - } - // **No timeout.** `Pool` publishes work and then unparks, and - // `shut_down` unparks every worker, so a wake is never lost; the - // timeout this used to have cost a wakeup per worker per 100 ms on a - // pool with nothing to do and guarded nothing. - while state.load(Ordering::Acquire) == ASLEEP { - atomic_wait::wait(state, ASLEEP); - if state.load(Ordering::Acquire) == ASLEEP { - self.spurious.fetch_add(1, Ordering::Relaxed); - } - } - // Consume the permit that ended the wait, and only that one. A plain - // store would clear a *newer* permit that arrived between the loop - // exiting and this line, which is a lost wakeup. - let _ = state.compare_exchange(SIGNAL, LOCKED, Ordering::Acquire, Ordering::Relaxed); + self.parking.park(worker); } - fn unpark(&self, worker: usize) { - let state = &self.slots[worker].state; - // Leave a permit whatever the state was, because `Host` requires an - // unpark arriving before the park to make that park return at once. - // - // **The system call happens only when somebody is actually on the - // address.** Signalling a running worker is this one swap: no syscall, - // no lock, and no shared line except that worker's own, which is - // padded. That is what lets `Pool` unpark on every submit without - // asking first who is asleep. - if state.swap(SIGNAL, Ordering::Release) == ASLEEP { - atomic_wait::wake_one(state); - } + self.parking.unpark(worker); } - fn now_ns(&self) -> u64 { u64::try_from(self.origin.elapsed().as_nanos()).unwrap_or(u64::MAX) } diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 631c826..f025bda 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -49,9 +49,13 @@ use crossbeam_utils::CachePadded; use crate::lifo::{Stealer, Worker as Queue}; use spin::Mutex; +#[cfg(feature = "atomic-host")] +mod atomic_host; #[cfg(feature = "host")] mod host; mod job; +#[cfg(feature = "atomic-host")] +pub use atomic_host::AtomicHost; #[cfg(feature = "host")] pub use host::StdHost; pub use job::{Act, Job}; @@ -129,6 +133,12 @@ pub struct Tuning { /// avoids queue contention on workloads dominated by lock handoffs. Both /// policies still probe other work after `lifo_run_limit` slot jobs. pub share_displaced: bool, + /// Keep displaced jobs in a FIFO inbox that idle peers can steal. + /// + /// Only the warm LIFO slot remains private. This takes precedence over + /// `share_displaced` and wakes a peer when displaced work is published. + /// It trades lock-handoff locality for prompt access to independent work. + pub stealable_inbox: bool, /// Whether [`Pool::submit_local`] keeps work on the calling worker. /// /// `false` makes it forward to [`Pool::submit_job`], which is the whole of @@ -191,6 +201,7 @@ impl Tuning { Self { lifo_run_limit: LIFO_RUN_LIMIT, share_displaced: false, + stealable_inbox: false, rounds_before_park: ROUNDS_BEFORE_PARK, backoff_spins: BACKOFF_SPINS, promote_every: PROMOTE_EVERY, @@ -254,6 +265,13 @@ impl Tuning { self } + /// Set [`stealable_inbox`](Self::stealable_inbox), without changing the LIFO quota. + #[must_use] + pub fn with_stealable_inbox(mut self, enabled: bool) -> Self { + self.stealable_inbox = enabled; + self + } + /// Set [`rounds_before_park`](Self::rounds_before_park). #[must_use] pub fn with_rounds_before_park(mut self, rounds: u32) -> Self { @@ -740,6 +758,7 @@ impl Pool { // win, which is entirely about the *first* task, and makes the second // one somebody else's to take. let displaced = self.local[worker].lifo.lock().replace(job); + let published_inbox = self.tuning.stealable_inbox && displaced.is_some(); if let Some(displaced) = displaced { // **The second displacement is the one that means something.** // @@ -768,7 +787,10 @@ impl Pool { // task costs 7% on lock-bound work for a tail that only appears on // read-only work with many independent tasks. Which of those a // table does is exactly what a flavor is for. - if self.tuning.share_displaced && !self.local[worker].inbox.is_empty() { + if !self.tuning.stealable_inbox + && self.tuning.share_displaced + && !self.local[worker].inbox.is_empty() + { // **Published, not announced.** // // `push` plus a wake is what made this expensive on lock-bound @@ -818,6 +840,9 @@ impl Pool { // even though a wake was sent, and it is the shape that shows up as // one busy core with tasks outstanding. self.wake_specific(worker); + if published_inbox { + self.nudge(Some(worker)); + } } /// Wake one named worker, because the work is private to it. @@ -1051,6 +1076,17 @@ impl Pool { continue; } + // With a stealable inbox, consume one FIFO job at a fairness + // boundary and leave the remaining backlog visible to peers. + if self.tuning.stealable_inbox { + if let Some(job) = self.local[w.id].inbox.pop() { + spins = 0; + job.run(); + self.local[w.id].completed.fetch_add(1, Ordering::Relaxed); + continue; + } + } + // 1.5. What this worker handed back to itself. // // Moved into `mine` rather than run from here. Running it @@ -1066,7 +1102,7 @@ impl Pool { // task that just yielded still runs next and keeps its cache // lines, and subject to the same promotion that makes local // work stealable on the heartbeat. - { + if !self.tuning.stealable_inbox { let mut moved = 0usize; while let Some(job) = self.local[w.id].inbox.pop() { mine.push_back(job); @@ -1192,26 +1228,10 @@ impl Pool { continue; } - // Nothing anywhere, so this worker sleeps. - // - // **Announce first, then look.** A submit publishes its job and - // then reads the bitmap; this sets its bit and then reads the - // injector. Two store-then-load pairs against different locations, - // so only `SeqCst` on all four makes at least one of them see the - // other. Announcing after the look, or announcing with a weaker - // ordering, lets a worker sleep on a job that is already queued. + // Announce before the final check, including work another worker + // published after our last steal attempt but before seeing our bit. let bit = 1usize << w.id; - self.sleeping.fetch_or(bit, Ordering::SeqCst); - - // The inbox is checked alongside the injector, and for the same - // reason: `submit_local` publishes and then reads this bitmap, so - // announcing before looking is what stops a worker sleeping on a - // job already handed to it. - if !self.injector.is_empty() - || !self.local[w.id].inbox.is_empty() - || self.local[w.id].lifo.lock().is_some() - { - self.sleeping.fetch_and(!bit, Ordering::SeqCst); + if !self.prepare_to_park(&queue, w.id, &mut rng) { continue; } // Checked before parking so a worker cannot sleep through the end; @@ -1286,6 +1306,22 @@ impl Pool { taken } + /// Publish sleep intent, then recheck every reachable source of work. + /// A publisher either observes our bit or its work appears in this check. + fn prepare_to_park(&self, queue: &Queue, id: usize, rng: &mut u64) -> bool { + let bit = 1usize << id; + self.sleeping.fetch_or(bit, Ordering::SeqCst); + if !self.injector.is_empty() + || !self.local[id].inbox.is_empty() + || self.local[id].lifo.lock().is_some() + || self.steal_once(queue, id, rng) > 0 + { + self.sleeping.fetch_and(!bit, Ordering::SeqCst); + return false; + } + true + } + /// Publish some of this worker's private queue where thieves can reach it. /// /// **Only when somebody is asleep.** With every worker busy there is nobody @@ -1334,6 +1370,9 @@ impl Pool { } } if shared > 0 { + // Deque publication is release-only. Pair it with the sleeper's + // SeqCst announcement before inspecting the sleep bitmap. + core::sync::atomic::fence(Ordering::SeqCst); // One wake per job published. Waking a single worker for a batch // leaves the rest of the batch reachable in principle and unseen // in practice. @@ -1373,6 +1412,16 @@ impl Pool { if victim == id { continue; } + if self.tuning.stealable_inbox { + if let Some(job) = self.local[victim].inbox.pop() { + // This worker owns `queue`; a full queue already contains + // work, but preserve the stolen job even in that case. + if let Err(job) = queue.push(job) { + self.push(Some(id), job); + } + return 1; + } + } let got = self.stealers[victim] .steal(queue, |n| (n.min(PULL_BATCH) + 1) / 2) .unwrap_or(0); @@ -1452,6 +1501,54 @@ mod wake_distribution { ); } + #[test] + fn work_shared_between_last_search_and_sleep_is_not_stranded() { + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning(2, 64, host.clone(), Tuning::locality()); + let owner = pool.local[0].queue.lock().take().unwrap(); + let peer = pool.local[1].queue.lock().take().unwrap(); + let mut rng = 123; + assert_eq!(pool.steal_once(&owner, 0, &mut rng), 0); + // The producer shares after our last search, while no sleeper is visible. + let mut mine = alloc::collections::VecDeque::new(); + mine.push_back(super::Job::from_boxed(|| {})); + mine.push_back(super::Job::from_boxed(|| {})); + pool.share(&mut mine, &peer, 1); + assert!(host.unparked.lock().unwrap().is_empty()); + assert!( + !pool.prepare_to_park(&owner, 0, &mut rng), + "shared work arrived before sleep intent; recheck it before blocking" + ); + assert!(owner.pop().is_some()); + assert_eq!(pool.sleeping.load(Ordering::SeqCst) & 1, 0); + } + + #[test] + fn stealable_inbox_exposes_displacement_but_keeps_the_warm_slot_private() { + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning( + 2, + 64, + host.clone(), + Tuning::locality().with_stealable_inbox(true), + ); + pool.sleeping.store(1 << 1, Ordering::SeqCst); + pool.submit_local(0, super::Job::from_boxed(|| {})); + assert!(host.unparked.lock().unwrap().is_empty()); + pool.submit_local(0, super::Job::from_boxed(|| {})); + assert_eq!(*host.unparked.lock().unwrap(), [1]); + let thief = pool.local[1].queue.lock().take().unwrap(); + assert_eq!(pool.steal_once(&thief, 1, &mut 123), 1); + assert!(thief.pop().is_some()); + assert!(pool.local[0].inbox.is_empty()); + assert!(pool.local[0].lifo.lock().is_some()); + assert_eq!(pool.steal_once(&thief, 1, &mut 456), 0); + } + /// Displacement publishes to the injector without waking anybody. /// /// The wake is deliberately absent. It costs a park/unpark round trip per diff --git a/src/lib.rs b/src/lib.rs index 4724fb1..c3d7a09 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -52,8 +52,8 @@ extern crate alloc; -// `host` brings an operating system with it, so it brings `std`. Everything -// else here, the queues and the pool alike, stays free of both. +// Only the convenience `host` clock adapter uses Rust std. `atomic-host` +// uses blocking platform calls without std; the queues and pool need neither. #[cfg(feature = "host")] extern crate std; diff --git a/tests/atomic_host.rs b/tests/atomic_host.rs new file mode 100644 index 0000000..835f513 --- /dev/null +++ b/tests/atomic_host.rs @@ -0,0 +1,45 @@ +use st3::fanout::{AtomicHost, Host}; +use std::{ + sync::{mpsc, Arc}, + thread, + time::Duration, +}; + +#[test] +fn signal_before_park_leaves_a_permit() { + let host = Arc::new(AtomicHost::new(1, || 42)); + assert_eq!(host.now_ns(), 42); + host.unpark(0); + let (send, receive) = mpsc::channel(); + let worker = thread::spawn(move || { + host.park(0); + send.send(()).unwrap(); + }); + receive + .recv_timeout(Duration::from_secs(5)) + .expect("permit was lost"); + worker.join().unwrap(); +} + +#[test] +fn repeated_signal_and_wait_handshakes_do_not_lose_wakes() { + let host = Arc::new(AtomicHost::new(1, || 0)); + let owner = host.clone(); + let (ready, receive_ready) = mpsc::channel(); + let (done, receive_done) = mpsc::channel(); + let worker = thread::spawn(move || { + for _ in 0..500 { + ready.send(()).unwrap(); + owner.park(0); + done.send(()).unwrap(); + } + }); + for _ in 0..500 { + receive_ready.recv_timeout(Duration::from_secs(5)).unwrap(); + host.unpark(0); + receive_done + .recv_timeout(Duration::from_secs(5)) + .expect("wake was lost"); + } + worker.join().unwrap(); +} diff --git a/tests/fanout.rs b/tests/fanout.rs index fb8889a..e5662a9 100644 --- a/tests/fanout.rs +++ b/tests/fanout.rs @@ -397,3 +397,62 @@ fn short_jobs_do_not_wait_behind_a_long_one() { ); let _ = finished; } + +/// A blocked owner keeps only its warm slot; peers can drain every displaced job. +#[test] +fn stealable_inbox_drains_behind_a_blocked_owner_exactly_once() { + use st3::fanout::{Job, Tuning}; + use std::sync::mpsc; + let pool = Pool::with_tuning( + 4, + 64, + Arc::new(StdHost::new(4)), + Tuning::locality() + .with_stealable_inbox(true) + .with_rounds_before_park(0), + ); + let (entered, ready) = mpsc::channel(); + let (release, resume) = mpsc::channel(); + pool.submit_local( + 0, + Job::from_boxed(move || { + entered.send(()).unwrap(); + resume.recv_timeout(Duration::from_secs(10)).unwrap(); + }), + ); + let threads: Vec<_> = (0..4) + .map(|id| { + let pool = pool.clone(); + std::thread::spawn(move || assert!(pool.run(pool.runner(id)))) + }) + .collect(); + ready.recv_timeout(Duration::from_secs(5)).unwrap(); + let counts: Arc> = Arc::new((0..2000).map(|_| AtomicUsize::new(0)).collect()); + let (finished, completed) = mpsc::channel(); + for id in 0..counts.len() { + let counts = counts.clone(); + let finished = finished.clone(); + pool.submit_local( + 0, + Job::from_boxed(move || { + counts[id].fetch_add(1, Ordering::Relaxed); + finished.send(()).unwrap(); + }), + ); + } + // All but the warm slot must run before the blocked owner is released. + for _ in 0..counts.len() - 1 { + completed + .recv_timeout(Duration::from_secs(5)) + .expect("displaced work was stranded"); + } + release.send(()).unwrap(); + completed.recv_timeout(Duration::from_secs(5)).unwrap(); + pool.shut_down(); + for worker in threads { + worker.join().unwrap(); + } + assert!(counts + .iter() + .all(|count| count.load(Ordering::Relaxed) == 1)); +} From 34ea68d702aa6f0a58bc06a2e4e6443c1f115cd2 Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 01:23:27 +0700 Subject: [PATCH 4/8] Expose capped warm tasks through almost_tokio and parking policies --- CHANGELOG.md | 9 ++- README.md | 8 ++- src/fanout/mod.rs | 149 +++++++++++++++++++++++----------------------- 3 files changed, 87 insertions(+), 79 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 856b37e..d8c5847 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,12 +2,17 @@ ### Added +- `Tuning::almost_tokio()` names the three-poll LIFO/FIFO-sharing policy; + `Tuning::parking()` adds immediate host parking after an unsuccessful search. + Both are opt-in and available without Rust std. + - `atomic-host` exposes the blocking OS parker as `AtomicHost` without Rust `std`; callers supply a monotonic clock and worker lifecycle. `StdHost` delegates to the same permit protocol. - `Tuning::with_stealable_inbox` exposes displaced local jobs to idle peers - through a FIFO inbox. The warm LIFO slot remains private; existing presets - keep their previous displacement policy. + through a FIFO inbox. The warm LIFO slot becomes stealable at its fairness + quota, including a lone self-waking task. Existing presets keep their + previous displacement policy. ### Fixed diff --git a/README.md b/README.md index c2de489..c85f0e5 100644 --- a/README.md +++ b/README.md @@ -257,5 +257,11 @@ parking timeout. Your embedding owns worker creation and shutdown. The `host` feature adds `StdHost` with an `Instant` clock over this same implementation. `Tuning::with_stealable_inbox(true)` lets idle peers take displaced local jobs -while retaining a private warm LIFO slot. It is a Rust callsite option, useful +while retaining a warm LIFO slot between fairness boundaries. At the LIFO quota, +the slot job also enters the stealable inbox, including a lone self-waking task. +It is a Rust callsite option, useful for independent tasks; measure its extra sharing cost on lock handoffs. + +The opt-in presets are `Tuning::almost_tokio()` (a three-poll LIFO quota and +shared FIFO inbox) and `Tuning::parking()` (the same policy with no idle spin +rounds). These are Nagoya scheduler policies, not a Tokio compatibility mode. diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index f025bda..555344d 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -135,7 +135,8 @@ pub struct Tuning { pub share_displaced: bool, /// Keep displaced jobs in a FIFO inbox that idle peers can steal. /// - /// Only the warm LIFO slot remains private. This takes precedence over + /// The warm LIFO slot stays private until its fairness quota expires, + /// then its ready job also enters the stealable inbox. This takes precedence over /// `share_displaced` and wakes a peer when displaced work is published. /// It trades lock-handoff locality for prompt access to independent work. pub stealable_inbox: bool, @@ -210,6 +211,27 @@ impl Tuning { } } + /// Keep a warm LIFO slot for three polls, then expose its task to peers. + /// + /// Displaced work enters a stealable FIFO inbox. This is a Nagoya policy + /// inspired by Tokios local scheduling, not a Tokio implementation or + /// compatibility mode. Measure its sharing cost on contended workloads. + #[must_use] + pub fn almost_tokio() -> Self { + Self::locality() + .with_stealable_inbox(true) + .with_lifo_run_limit(3) + } + + /// The almost_tokio policy with OS parking after an unsuccessful search. + /// + /// This removes the idle spin rounds; the host still determines how a + /// worker waits. With AtomicHost or StdHost, that wait blocks in the OS. + #[must_use] + pub fn parking() -> Self { + Self::almost_tokio().with_rounds_before_park(0) + } + /// Send every wake to the injector, where any worker can take it. /// /// For work whose wakes are **independent**: the woken task has no claim on @@ -732,85 +754,22 @@ impl Pool { self.push(None, job); return; } - // The slot first, and whatever it held goes wherever - // `Tuning::share_displaced` says. - // - // # At most one task may be private to a worker, under `share_displaced` - // - // The slot is not stealable, which is the point: the task that just - // woke itself runs next, on the worker whose cache lines it is still - // in. That is worth having for one task and is a trap for two. - // - // A displaced job used to go to a private inbox behind the slot. Both - // are unreachable by any other worker, so a worker that accumulated - // several distinct tasks kept them, and nothing could take them back: - // the inbox drains into the private deque, the deque is shared only on - // a heartbeat counted in `mine.pop_back` calls, and the LIFO slot ahead - // of that in the loop starves the heartbeat for as long as any task - // keeps waking itself. Every wake was delivered; the work simply became - // invisible, which is why the `SeqCst` Dekker pair below did not fix - // it and why it looked like a lost wakeup. - // - // Measured on sixteen workers with eight self-waking client tasks, - // YCSB C: four runs in eight collapsed to one active worker with the - // other fifteen parked in `__ulock_wait`, at exactly the one-thread - // rate. Sending the displaced job to the injector keeps the locality - // win, which is entirely about the *first* task, and makes the second - // one somebody else's to take. + // Keep the newest ready task warm. Displacement policy chooses whether + // older work stays private, spills excess to the injector, or enters + // an immediately announced stealable inbox. let displaced = self.local[worker].lifo.lock().replace(job); let published_inbox = self.tuning.stealable_inbox && displaced.is_some(); if let Some(displaced) = displaced { - // **The second displacement is the one that means something.** - // - // One task waking itself displaces nothing: it takes the slot, - // runs, and puts itself back into a slot the worker just emptied. - // Two tasks alternating displace each other, and the inbox behind - // the slot drains between them. Neither is an imbalance, and - // sending either to the injector would cost a wake for work that - // was going to be run here anyway. That is what made - // `share_displaced` lose 31% on a lock-bound update workload. - // - // A displacement arriving while the inbox is *already* occupied is - // different: it means a third task is queued behind two that this - // worker has not got to, which is the pile-up that cannot correct - // itself, because neither the slot nor the inbox is stealable and - // the slot starves the heartbeat that would share them. - // - // So under `share_displaced` the first one stays private and the - // rest go where anybody can reach them. Sharing from the *first* - // displacement was measured too and is worse on both sides: it - // took a lock-bound update workload from 943,606 to 651,554, and - // it did not buy anything the second-displacement rule does not. - // - // Off, this is the original behaviour: everything stays private. - // That is still the right default, because holding the line at one - // task costs 7% on lock-bound work for a tail that only appears on - // read-only work with many independent tasks. Which of those a - // table does is exactly what a flavor is for. + // share_displaced retains the first inbox job locally and sends + // further overflow to the injector. The stealable-inbox option + // instead keeps all displaced jobs in the shared FIFO inbox. if !self.tuning.stealable_inbox && self.tuning.share_displaced && !self.local[worker].inbox.is_empty() { - // **Published, not announced.** - // - // `push` plus a wake is what made this expensive on lock-bound - // work: the woken worker steals the job, runs it, blocks on the - // same row lock and parks again, so the pool pays a park/unpark - // round trip per displacement for work that was serialised - // anyway. - // - // The review's objection to that was starvation: an owner - // servicing self-wakes forever leaves this job behind while - // peers sleep. Real, and fixed below at the LIFO fairness - // boundary, where the owner services the injector itself. That - // path ends in `take_from_injector`, which already wakes one - // sleeper per job still queued. So the sleepers are told; they - // are told once per fairness boundary instead of once per - // displacement, which is the same news for a fraction of the - // round trips. - // - // A wake here would be the third mechanism for one job, and the - // one measured to cost the most. + // The owner services this injector at its fairness boundary, + // where excess work is announced in a batch. Avoid a peer + // wake for every lock handoff under this locality policy. self.injector.push(displaced); } else { self.local[worker].inbox.push(displaced); @@ -828,9 +787,8 @@ impl Pool { // caught in a profile with all sixteen workers parked in `__ulock_wait` // while the run had work outstanding, at a sixth of normal throughput. core::sync::atomic::fence(Ordering::SeqCst); - // **This worker, and no other.** The job above went into this worker's - // slot or inbox, and neither is stealable, so waking anybody else - // wakes somebody who is not allowed to touch it. + // Always notify the owner of its private warm slot. The optional + // stealable inbox additionally announces its displaced work below. // // `wake` cannot be used here. It treats its argument as a fallback and // takes `trailing_zeros()` whenever any worker is asleep, so a private @@ -1058,6 +1016,9 @@ impl Pool { // peers are busy. Service the injector at the local fairness // boundary so displaced work cannot wait forever behind self-wakes. let slot_budget_exhausted = lifo_run == lifo_run_limit; + if slot_budget_exhausted && self.tuning.stealable_inbox { + self.publish_capped_slot(w.id); + } if slot_budget_exhausted && self.tuning.share_displaced { self.take_from_injector(&mut mine); } @@ -1306,6 +1267,17 @@ impl Pool { taken } + /// A capped warm task becomes available to peers at the fairness boundary. + /// An idle peer may take it; otherwise its owner consumes the FIFO inbox. + fn publish_capped_slot(&self, id: usize) { + let job = self.local[id].lifo.lock().take(); + if let Some(job) = job { + self.local[id].inbox.push(job); + core::sync::atomic::fence(Ordering::SeqCst); + self.nudge(Some(id)); + } + } + /// Publish sleep intent, then recheck every reachable source of work. /// A publisher either observes our bit or its work appears in this check. fn prepare_to_park(&self, queue: &Queue, id: usize, rng: &mut u64) -> bool { @@ -1549,6 +1521,31 @@ mod wake_distribution { assert_eq!(pool.steal_once(&thief, 1, &mut 456), 0); } + #[test] + fn a_capped_warm_task_becomes_stealable_without_a_second_task() { + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning( + 2, + 64, + host.clone(), + Tuning::locality() + .with_stealable_inbox(true) + .with_lifo_run_limit(3), + ); + let thief = pool.local[1].queue.lock().take().unwrap(); + pool.sleeping.store(1 << 1, Ordering::SeqCst); + pool.submit_local(0, super::Job::from_boxed(|| {})); + assert_eq!(pool.steal_once(&thief, 1, &mut 123), 0); + pool.publish_capped_slot(0); + assert_eq!(*host.unparked.lock().unwrap(), [1]); + assert_eq!(pool.steal_once(&thief, 1, &mut 456), 1); + assert!(thief.pop().is_some()); + assert!(pool.local[0].lifo.lock().is_none()); + assert!(pool.local[0].inbox.is_empty()); + } + /// Displacement publishes to the injector without waking anybody. /// /// The wake is deliberately absent. It costs a park/unpark round trip per From 089918a5516dfebd2380bc0665241fb0bb654ede Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 01:42:21 +0700 Subject: [PATCH 5/8] Use a short idle budget for the locality baseline --- CHANGELOG.md | 6 +++ README.md | 4 ++ src/fanout/mod.rs | 108 +++++++++------------------------------------- 3 files changed, 30 insertions(+), 88 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d8c5847..14cf92e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,12 @@ previous displacement policy. +### Changed + +- Locality and its derived presets use four empty search rounds with 128 + spin hints per round before host parking. The former 64 by 1024 budget + burned several CPU cores between sparse bursts. + ### Fixed - The final check after announcing sleep now includes peer queues, closing diff --git a/README.md b/README.md index c85f0e5..8a76f73 100644 --- a/README.md +++ b/README.md @@ -265,3 +265,7 @@ for independent tasks; measure its extra sharing cost on lock handoffs. The opt-in presets are `Tuning::almost_tokio()` (a three-poll LIFO quota and shared FIFO inbox) and `Tuning::parking()` (the same policy with no idle spin rounds). These are Nagoya scheduler policies, not a Tokio compatibility mode. + +The default locality policy uses four empty search rounds and 128 spin hints +per round before host parking. Keep CPU use between bursts alongside throughput +when tuning this budget. diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 555344d..02db497 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -140,32 +140,11 @@ pub struct Tuning { /// `share_displaced` and wakes a peer when displaced work is published. /// It trades lock-handoff locality for prompt access to independent work. pub stealable_inbox: bool, - /// Whether [`Pool::submit_local`] keeps work on the calling worker. - /// - /// `false` makes it forward to [`Pool::submit_job`], which is the whole of - /// the switch. It is a parameter because the right answer depends on the - /// workload and the two answers are far apart. Measured at eight threads - /// against a tokio-driven build of the same storage engine: - /// - /// ```text - /// workload local injector - /// 50% update +6.4% -40.2% - /// read-modify-write +2.0% -38.4% - /// 95% read / 5% update -19.2% +74.4% - /// 95% read / 5% insert +137.1% +321.6% - /// ``` - /// - /// Contended work wants the handoff kept warm: its wakes are a chain, and - /// the successor wants the rows the releasing worker just touched. Sparse - /// work wants the opposite, because a rare wake queued behind a busy worker - /// waits while others sleep. - /// - /// `true` is the default because it is the better worst case: one workload - /// 19% behind, against two at about 40% behind the other way. There is no - /// static rule that gets both, and several were tried: gating on whether - /// the wake was a task waking *itself*, on whether any worker was idle, and - /// letting a thief take the slot on second sight. Each reproduced one - /// column or the other exactly. + /// Whether local wakes retain the calling worker's locality. + /// + /// False routes every wake through the shared injector. Local handoffs + /// generally favor contended tasks; independent tasks may benefit from + /// sharing. Compare throughput, tail latency and CPU on the same workload. pub local_wakes: bool, /// How many jobs a worker runs between two looks at whether anyone needs @@ -185,18 +164,11 @@ impl Default for Tuning { } impl Tuning { - /// Keep a woken task on the worker that woke it. - /// - /// For work whose wakes are a **chain**: a task releases something and the - /// task it releases wants the lines the first one just touched. Update-heavy - /// storage paths look like this, and so does anything with a lock handoff. - /// - /// Measured on YCSB at eight threads against a tokio-driven build of the - /// same storage engine: 50% update +6.4%, read-modify-write +2.0%, where - /// [`Tuning::spread`] puts both about 40% behind. + /// Keep local wake handoffs warm, with a short idle retry budget. /// - /// This is [`Tuning::default`], because being 19% behind on one shape beats - /// being 40% behind on two. + /// This is the default: four empty rounds, 128 spin hints per round, + /// then host parking. The release comparison favors this balance across + /// table operations and sparse bursts; it is not fastest in every cell. #[must_use] pub fn locality() -> Self { Self { @@ -214,7 +186,7 @@ impl Tuning { /// Keep a warm LIFO slot for three polls, then expose its task to peers. /// /// Displaced work enters a stealable FIFO inbox. This is a Nagoya policy - /// inspired by Tokios local scheduling, not a Tokio implementation or + /// inspired by local scheduling in Tokio, not a Tokio implementation or /// compatibility mode. Measure its sharing cost on contended workloads. #[must_use] pub fn almost_tokio() -> Self { @@ -234,17 +206,9 @@ impl Tuning { /// Send every wake to the injector, where any worker can take it. /// - /// For work whose wakes are **independent**: the woken task has no claim on - /// the waking worker's cache and would rather run now, somewhere else, than - /// wait behind it. Read-mostly and insert-mostly paths look like this. - /// - /// Measured on the same YCSB runs: 95% read / 5% update +74.4%, 95% read / - /// 5% insert +321.6%, where [`Tuning::locality`] is 19% behind on the first - /// and less than half as fast on the second. - /// - /// The cost is the other column: update-heavy work goes about 40% behind. - /// There is no setting that wins both, which is why this is a choice and - /// not a default. + /// Independent tasks may benefit from immediate sharing. Contended tasks + /// may instead pay for extra migration and wakeups. Uses the same short + /// idle budget as locality. #[must_use] pub fn spread() -> Self { Self { @@ -338,46 +302,14 @@ impl Tuning { /// at its own queue. const LIFO_RUN_LIMIT: u32 = 32; -/// Empty rounds a worker takes before it announces sleep. -/// -/// Parking is not expensive to the parker; it is expensive to whoever has to -/// wake it, because a wake is a mutex and a condvar notify. A worker that -/// parks the moment its queue runs dry makes every later submit pay one, and -/// at eight workers that was 0.95 parks a task: the producer became the -/// bottleneck, and the starvation that caused made the workers park again. -/// -/// A steady stream refills within a few hundred cycles, so looking again a few -/// times skips the protocol entirely. Swept at 16, 64, 256 and 1024 empty -/// rounds over 100,000 tasks and 8 workers: 445, 353, 372 and 365 ns a task. -/// Past a few dozen it stops mattering. -const ROUNDS_BEFORE_PARK: u32 = 64; +/// Empty search rounds before announcing sleep. The release baseline keeps +/// a small retry budget, then lets the host block instead of burning CPU +/// between sparse bursts. +const ROUNDS_BEFORE_PARK: u32 = 4; -/// How long to wait between two empty rounds. -/// -/// **This is worth more than anything else in this pool.** A worker that finds -/// nothing looks again, and looking means a pop off the injector and a probe of -/// every other worker's deque. Eight idle workers doing that in a tight loop -/// take the very lines the producer is trying to fill, so the pool starves -/// itself: the same 100,000 tasks that one worker finishes in 12.4 ms took -/// eight workers 49.2 ms. -/// -/// Swept over 100,000 tasks on 8 workers, with wake latency measured separately -/// on an idle pool over 2,000 samples: -/// -/// ```text -/// spins wall cpu latency median p99 -/// 64 49.2 ms 440 ms 3917 ns 10333 ns -/// 256 36.0 ms 325 ms 2500 ns 7958 ns -/// 1024 30.3 ms 277 ms 2375 ns 6000 ns -/// 4096 24.4 ms 237 ms 3375 ns 14209 ns -/// 16384 15.6 ms 213 ms 21000 ns 43375 ns -/// ``` -/// -/// 1024 is the last value that is better than the one before it on **every** -/// axis, so it is the default. Past it the trade is real: 16384 is three times -/// the throughput of 64 and five times the wake latency, which is the right -/// choice for a batch pool and the wrong one for anything waiting on a reply. -const BACKOFF_SPINS: u32 = 1024; +/// Spin hints between empty search rounds. Combined with four rounds this +/// preserves a brief retry opportunity without the old 65,536-spin budget. +const BACKOFF_SPINS: u32 = 128; /// Jobs run between two heartbeats. const PROMOTE_EVERY: u64 = 64; From 5dddcc8217cae27c86b9246797f57fce8160ce69 Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 01:52:42 +0700 Subject: [PATCH 6/8] Keep the parking policy as almost_tokio --- CHANGELOG.md | 7 ++++--- README.md | 8 +++++--- src/fanout/mod.rs | 20 ++++++++------------ 3 files changed, 17 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 14cf92e..acb4500 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,10 @@ ### Added -- `Tuning::almost_tokio()` names the three-poll LIFO/FIFO-sharing policy; - `Tuning::parking()` adds immediate host parking after an unsuccessful search. - Both are opt-in and available without Rust std. +- `Tuning::almost_tokio()` combines a three-poll LIFO quota, FIFO sharing and + immediate host parking after an unsuccessful search. It is opt-in and + available without Rust std. This is the policy named `parking` in early + benchmark reports; the earlier aggressive-spin preset was retired. - `atomic-host` exposes the blocking OS parker as `AtomicHost` without Rust `std`; callers supply a monotonic clock and worker lifecycle. `StdHost` diff --git a/README.md b/README.md index 8a76f73..38e7cdf 100644 --- a/README.md +++ b/README.md @@ -262,9 +262,11 @@ the slot job also enters the stealable inbox, including a lone self-waking task. It is a Rust callsite option, useful for independent tasks; measure its extra sharing cost on lock handoffs. -The opt-in presets are `Tuning::almost_tokio()` (a three-poll LIFO quota and -shared FIFO inbox) and `Tuning::parking()` (the same policy with no idle spin -rounds). These are Nagoya scheduler policies, not a Tokio compatibility mode. +The opt-in preset `Tuning::almost_tokio()` combines a three-poll LIFO quota, +a shared FIFO inbox and host parking without idle spin rounds. It is a Nagoya +scheduler policy, not a Tokio compatibility mode. Dated benchmark reports +called this policy `parking`; the earlier aggressive-spin `almost_tokio` +preset was retired before release. The default locality policy uses four empty search rounds and 128 spin hints per round before host parking. Keep CPU use between bursts alongside throughput diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 02db497..a780587 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -185,23 +185,19 @@ impl Tuning { /// Keep a warm LIFO slot for three polls, then expose its task to peers. /// - /// Displaced work enters a stealable FIFO inbox. This is a Nagoya policy - /// inspired by local scheduling in Tokio, not a Tokio implementation or - /// compatibility mode. Measure its sharing cost on contended workloads. + /// Displaced work enters a stealable FIFO inbox. An unsuccessful search + /// parks without idle spin rounds; the host determines how a worker waits. + /// With AtomicHost or StdHost, that wait blocks in the OS. + /// + /// This is a Nagoya policy inspired by local scheduling in Tokio, not a + /// Tokio implementation or compatibility mode. Measure its sharing cost + /// on contended workloads. #[must_use] pub fn almost_tokio() -> Self { Self::locality() .with_stealable_inbox(true) .with_lifo_run_limit(3) - } - - /// The almost_tokio policy with OS parking after an unsuccessful search. - /// - /// This removes the idle spin rounds; the host still determines how a - /// worker waits. With AtomicHost or StdHost, that wait blocks in the OS. - #[must_use] - pub fn parking() -> Self { - Self::almost_tokio().with_rounds_before_park(0) + .with_rounds_before_park(0) } /// Send every wake to the injector, where any worker can take it. From bd8c5c96f91b1e0a42e711cf45713388ba52378a Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 02:20:46 +0700 Subject: [PATCH 7/8] Wake peers for spare work at a LIFO fairness boundary --- src/fanout/mod.rs | 40 +++++++++++++++++++++++++++++++++------- 1 file changed, 33 insertions(+), 7 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index a780587..e188528 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -945,7 +945,7 @@ impl Pool { // boundary so displaced work cannot wait forever behind self-wakes. let slot_budget_exhausted = lifo_run == lifo_run_limit; if slot_budget_exhausted && self.tuning.stealable_inbox { - self.publish_capped_slot(w.id); + self.publish_capped_slot(w.id, !mine.is_empty() || !queue.is_empty()); } if slot_budget_exhausted && self.tuning.share_displaced { self.take_from_injector(&mut mine); @@ -1195,14 +1195,17 @@ impl Pool { taken } - /// A capped warm task becomes available to peers at the fairness boundary. - /// An idle peer may take it; otherwise its owner consumes the FIFO inbox. - fn publish_capped_slot(&self, id: usize) { + /// Expose a capped task without waking a peer merely to move the only job. + /// The active owner consumes a lone inbox entry on its next step. If it has + /// other work, announce the extra capacity a peer can use. + fn publish_capped_slot(&self, id: usize, has_other_ready_work: bool) { let job = self.local[id].lifo.lock().take(); if let Some(job) = job { self.local[id].inbox.push(job); core::sync::atomic::fence(Ordering::SeqCst); - self.nudge(Some(id)); + if has_other_ready_work || self.local[id].inbox.len() > 1 { + self.nudge(Some(id)); + } } } @@ -1466,14 +1469,37 @@ mod wake_distribution { pool.sleeping.store(1 << 1, Ordering::SeqCst); pool.submit_local(0, super::Job::from_boxed(|| {})); assert_eq!(pool.steal_once(&thief, 1, &mut 123), 0); - pool.publish_capped_slot(0); - assert_eq!(*host.unparked.lock().unwrap(), [1]); + pool.publish_capped_slot(0, false); + assert!(host.unparked.lock().unwrap().is_empty()); assert_eq!(pool.steal_once(&thief, 1, &mut 456), 1); assert!(thief.pop().is_some()); assert!(pool.local[0].lifo.lock().is_none()); assert!(pool.local[0].inbox.is_empty()); } + #[test] + fn a_capped_task_announces_spare_work_when_its_owner_has_other_jobs() { + for private_backlog in [false, true] { + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning(2, 64, host.clone(), Tuning::almost_tokio()); + pool.submit_local(0, super::Job::from_boxed(|| {})); + if !private_backlog { + // A second task displaces the first into the shared inbox. + pool.submit_local(0, super::Job::from_boxed(|| {})); + } + pool.sleeping.store(1 << 1, Ordering::SeqCst); + host.unparked.lock().unwrap().clear(); + pool.publish_capped_slot(0, private_backlog); + assert_eq!(*host.unparked.lock().unwrap(), [1]); + assert!(pool.local[0].lifo.lock().is_none()); + let thief = pool.local[1].queue.lock().take().unwrap(); + assert_eq!(pool.steal_once(&thief, 1, &mut 456), 1); + assert!(thief.pop().is_some()); + } + } + /// Displacement publishes to the injector without waking anybody. /// /// The wake is deliberately absent. It costs a park/unpark round trip per From 2374640e71c56b6fe67f64a52ad120c33cf7c2bf Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 12 Sep 2026 02:31:38 +0700 Subject: [PATCH 8/8] Document spare-work wakeups for almost_tokio --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index 38e7cdf..4330779 100644 --- a/README.md +++ b/README.md @@ -259,6 +259,8 @@ feature adds `StdHost` with an `Instant` clock over this same implementation. `Tuning::with_stealable_inbox(true)` lets idle peers take displaced local jobs while retaining a warm LIFO slot between fairness boundaries. At the LIFO quota, the slot job also enters the stealable inbox, including a lone self-waking task. +A lone job stays available to awake thieves without waking a sleeping peer; +the active owner can run it. Spare local or inbox work still triggers a wake. It is a Rust callsite option, useful for independent tasks; measure its extra sharing cost on lock handoffs.