ps-st3 0.6.2: correct parking, short-idle locality and no_std OS waits

.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

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

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 <serge.barral@asynchronics.com>"]
 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"

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.

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<T, W: GenericWorker<u64> + '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<T, W: GenericWorker<u64> + '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<T, W: GenericWorker<u64> + '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;
                                     }

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 {

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<CachePadded<Slot>>,
+    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)()
+    }
+}