fanout: thin jobs, one injector, a heartbeat, and a futex parker

CHANGELOG.md

@@ -1,3 +1,34 @@
+# 0.6.0 (2026-09-10)
+
+### Added
+
+- `Tuning`, the `fanout` pool's idle and wake policy, with three presets:
+  `Tuning::locality` (the default), `Tuning::spread` and `Tuning::throughput`.
+  Each carries the measurement that produced it in its doc comment, so a caller
+  chooses on evidence rather than on the name.
+- `Tuning::with_rounds_before_park`, `with_backoff_spins`, `with_injector_batch`,
+  `with_local_wakes` and `with_promote_every`, taking `self` by value so they
+  chain into `Pool::with_tuning`.
+- `Pool::submit_local`, which keeps a woken task on the worker that woke it.
+
+### Changed
+
+- `Pool` no longer wakes a sleeping worker when none is sleeping. The sleep path
+  is a Dekker pair, so publishing to a worker and then reading whether it sleeps
+  needs a full barrier on both sides; a `SeqCst` fence in `submit_local` closes
+  a lost wakeup that the spin-mutex release did not.
+- `INJECTOR_BATCH` is 1 rather than 8. Measured: a pure-read workload at sixteen
+  threads went from 8.9M to 15.2M operations per second.
+
+### Breaking
+
+- `Tuning` is `#[non_exhaustive]`, so it can gain a field without a major bump
+  next time. **This is why the bump is minor rather than patch**: a downstream
+  struct literal, including one using `..Default::default()`, no longer
+  compiles. Build one from `Tuning::locality()`, `spread()` or `throughput()`
+  and the `with_*` methods instead. Fields stay public to read, because callers
+  legitimately derive values from the defaults.
+
 # 0.5.0 (2026-09-06)
 
 ### 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.5.0"
+version = "0.6.0"
 authors = ["Serge Barral <serge.barral@asynchronics.com>"]
 edition = "2021"
 rust-version = "1.60"
@@ -27,24 +27,41 @@ name = "st3"
 
 [dependencies]
 crossbeam-utils = { version = "0.8.14", default-features = false }
-# Only the `fanout` pool needs a lock, for the per-worker intake and for the
+# The `fanout` injector: one lock-free queue every worker drains, in place of
+# the per-worker intakes nothing but their owner could reach. Same MSRV as this
+# crate, and its only dependency is `crossbeam-utils`, which is already here.
+crossbeam-queue = { version = "0.3.12", default-features = false, features = [
+    "alloc",
+], optional = true }
+# Only the `fanout` pool needs a lock, for the
 # one-time handover of each deque to the thread that runs it.
 spin = { version = "0.9", default-features = false, features = [
     "spin_mutex",
 ], optional = true }
+# The `host` parker. A condition variable costs a mutex on every wake, and this
+# pool's wakes are its hot path; `atomic-wait` is a futex on Linux, `WaitOnAddress`
+# on Windows and `__ulock_wait` on macOS, so an uncontended park or unpark is one
+# atomic and, when nobody has to sleep, no system call at all.
+atomic-wait = { version = "1.1", optional = true }
 
 [features]
 default = ["fanout"]
 
 # The work-stealing pool built on these queues. On by default: it is the reason
 # this fork exists, and `spin` with default features off pulls in nothing else.
 # Turn it off with `default-features = false` for the deques alone.
-fanout = ["dep:spin"]
+fanout = ["dep:spin", "dep:crossbeam-queue"]
 
 # The OS substrate for `fanout`: threads, parking, a clock. The pool itself is
 # `no_std` and knows nothing about any of it, so a target without an OS links
 # the pool alone and brings its own `Host`.
-host = ["fanout"]
+#
+# **This feature has a higher minimum than the crate.** `rust-version` above is
+# 1.60 and holds for the queues and for `fanout`. `host` pulls `atomic-wait`,
+# 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"]
 
 [dev-dependencies]
 criterion = "0.3"

docs/nagoya.md

@@ -0,0 +1,84 @@
+# Nagoya: what would sit on top of these queues, and what it is not
+
+A design note and a name, written 8 September 2026 so the argument survives the
+conversation that produced it. Nothing here is built.
+
+## Where this crate sits
+
+`ps-st3` is a fork of `st3`, and its README is precise about what the two queues
+target:
+
+> The FIFO queue is based on the Tokio queue, but provides a somewhat more
+> convenient and more flexible API. The LIFO queue is a novel design [...] a
+> faster, fixed-size alternative to the Chase-Lev deque.
+
+So the halves point at different rivals: `fifo` at Tokio's internal queue,
+`lifo` at the Chase-Lev deque that `crossbeam-deque` implements and `rayon-core`
+runs on.
+
+**Only one of those is substitutable.** Tokio's queue is private - nothing can
+depend on `tokio::runtime::scheduler::multi_thread::queue` - so nobody migrates
+to this crate *from* Tokio. `crossbeam-deque` is a public crate that rayon
+depends on, so it is the one thing here that could actually displace something.
+The customer is a library author building a scheduler, not an application
+choosing a runtime.
+
+`fanout` is in neither camp. No futures and no I/O driver, so it is not Tokio.
+No `join`, `scope` or `par_iter`, so it is not rayon. It is the layer both of
+them contain.
+
+## What would make it feel like a runtime
+
+`Host` is already the whole operating-system seam: park, unpark, a clock. That
+is the useful fact, because it means the first four of these need nothing more
+than what is already abstracted, and the `no_std` core survives all of them.
+
+| | adds | cost |
+|---|---|---|
+| 1 | **`JoinHandle`** - `spawn` returns a handle, the task writes its result to a slot and unparks the waiter | one allocation and one synchronisation point per task |
+| 2 | **`block_on`** - drives one future on the calling thread, waker is `Host::unpark` | small, and independent of the pool |
+| 3 | **futures and `Waker`** - a task becomes a `Future`, waking re-submits it to a worker | the change that makes this a runtime rather than a pool |
+| 4 | **timers** - a wheel over `Host::clock`, giving `sleep` and `timeout` | self-contained |
+| 5 | **an I/O driver** - epoll, kqueue, io_uring | the actual moat, needs an operating system, **ends `no_std`** |
+
+1 and 2 buy most of the felt difference for a fraction of the work. 3 is the
+fork in the road. **Stopping after 4 is a coherent product rather than a
+half-built Tokio: an async runtime that needs no operating system.** Tokio
+cannot be that. "Tokio but ours" is not a position; that is.
+
+## The tension, named
+
+`fanout` says of itself:
+
+> Tasks are independent by construction: no join, no continuation, no completion
+> signal.
+
+That is not a limitation, it is why it is cheap. Every layer above adds per-task
+state and spends some of it. So they belong beside `submit`, not instead of it,
+or the resemblance is bought with the advantage.
+
+## The name
+
+**Nagoya**, when it graduates. Not before: claiming a registry name for an
+unpublished idea is the thing this note objects to on Tokio's behalf below.
+
+A runtime with `JoinHandle`, `block_on`, futures and timers does not belong
+under a name that says "these are st3's queues". It would be its own crate
+depending on this one, and that crate is what is called `nagoya`. While it stays
+a module it stays `fanout`, and the name is premature.
+
+Rejected, with reasons:
+
+- **`tokyo`** - one letter from `tokio`. It reads as a typosquat or a joke at
+  the expense of the project whose queue design this crate builds on, and it
+  mislabels the thing: naming it after the product it deliberately is not,
+  exactly when the interesting claim is "no operating system required".
+- **`osaka`** - taken, and taken by an abandoned **async runtime** (0.3.0,
+  25,650 downloads, last touched 2019, "async for rust without the noise").
+  A new runtime that users must disambiguate from a dead runtime of the same
+  name is the one collision worth avoiding.
+- **`kyoto`** - taken by a dormant cache crate (0.0.1, 1,690 downloads, 2021).
+  Only a nominal collision, but unavailable.
+
+`nagoya` and `sapporo` were both free on 8 September 2026. `nagoya` because
+`sapporo` reads as the beer first.

examples/floor.rs

@@ -203,5 +203,5 @@ fn main() {
         n.as_secs_f64() * 1e3,
         s.as_secs_f64() / n.as_secs_f64()
     );
-    println!("\n  park timeouts during the run: {}", host.timeouts());
+    println!("\n  spurious wakes during the run: {}", host.spurious());
 }

src/fanout/host.rs

@@ -6,42 +6,95 @@
 //! 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 std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Condvar, Mutex};
-use std::time::{Duration, Instant};
+use crossbeam_utils::CachePadded;
+use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+use std::time::Instant;
 use std::vec::Vec;
 
 use super::Host;
 
-/// Condvar parking, one pair per worker.
+/// 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 StdHost {
-    slots: Vec<(Mutex<bool>, Condvar)>,
+    slots: Vec<CachePadded<Slot>>,
     origin: Instant,
-    /// How many times a park timed out rather than being woken. With the pool
-    /// waking its own sleepers this should stay near zero under load; what it
-    /// still counts on an idle pool is one tick per worker per 100 ms.
-    timeouts: AtomicU64,
+    /// 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 {
     #[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, || (Mutex::new(false), Condvar::new()));
+        slots.resize_with(workers, || CachePadded::new(Slot::default()));
         Self {
             slots,
             origin: Instant::now(),
-            timeouts: AtomicU64::new(0),
+            spurious: AtomicU64::new(0),
         }
     }
 
-    /// Parks that timed out instead of being woken.
+    /// 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]
+    pub fn spurious(&self) -> u64 {
+        self.spurious.load(Ordering::Relaxed)
+    }
+
+    /// 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.
     #[must_use]
+    #[deprecated(since = "0.6.0", note = "parking no longer times out; see `spurious`")]
     pub fn timeouts(&self) -> u64 {
-        self.timeouts.load(Ordering::Relaxed)
+        0
     }
 }
 
@@ -57,28 +110,52 @@ impl Host for StdHost {
     /// 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 (mutex, condvar) = &self.slots[worker];
-        let mut guard = mutex.lock().expect("the park mutex is not poisoned");
-        // A permit was left by an unpark that arrived before this park. Take it
-        // and do not sleep. Locking again here instead of writing through the
-        // guard deadlocks: the mutex is not reentrant, and it cost an hour.
-        if *guard {
-            *guard = false;
+        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;
         }
-        let (mut guard, timed_out) = condvar
-            .wait_timeout(guard, Duration::from_millis(100))
-            .expect("the park mutex is not poisoned");
-        if timed_out.timed_out() {
-            self.timeouts.fetch_add(1, Ordering::Relaxed);
+        // **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);
+            }
         }
-        *guard = false;
+        // 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 (mutex, condvar) = &self.slots[worker];
-        *mutex.lock().expect("the park mutex is not poisoned") = true;
-        condvar.notify_one();
+        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 {