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 e648df7..acb4500 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,39 @@ +# 0.6.2 (2026-09-12) + +### Added + +- `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` + 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 becomes stealable at its fairness + quota, including a lone self-waking task. Existing presets keep their + 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 + 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. +- 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..218da20 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" @@ -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 bd96ee2..4330779 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 } ``` @@ -247,3 +247,29 @@ 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 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. + +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 +when tuning this budget. 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/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 4a11fff..e188528 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}; @@ -113,65 +117,34 @@ 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. - /// - /// `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. + /// Keep displaced jobs in a FIFO inbox that idle peers can steal. + /// + /// 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, + /// 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 @@ -191,23 +164,17 @@ 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 { 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, @@ -216,19 +183,28 @@ impl Tuning { } } - /// Send every wake to the injector, where any worker can take it. + /// Keep a warm LIFO slot for three polls, then expose its task to peers. /// - /// 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. + /// 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. /// - /// 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. + /// 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) + .with_rounds_before_park(0) + } + + /// Send every wake to the injector, where any worker can take it. /// - /// 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 { @@ -271,6 +247,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 { @@ -315,46 +298,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; @@ -539,6 +490,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 +577,8 @@ impl Pool { tuning, sleeping: AtomicUsize::new(0), searching: AtomicUsize::new(0), + #[cfg(test)] + idle_backoffs: AtomicUsize::new(0), }) } @@ -727,81 +682,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. - if 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. + // 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() + { + // 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); @@ -819,9 +715,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 @@ -831,6 +726,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. @@ -1045,7 +943,11 @@ 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.stealable_inbox { + 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); } lifo_run = 0; @@ -1063,6 +965,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 @@ -1078,7 +991,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); @@ -1158,6 +1071,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 +1103,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 @@ -1185,26 +1117,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; @@ -1279,6 +1195,36 @@ impl Pool { taken } + /// 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); + if has_other_ready_work || self.local[id].inbox.len() > 1 { + 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 { + 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 @@ -1327,6 +1273,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. @@ -1366,6 +1315,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); @@ -1412,6 +1371,135 @@ 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" + ); + } + + #[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); + } + + #[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, 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 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/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)) }); } 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)); +}