From a9e1081058ee65e917e317a10f99c6e5c9b23aca Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 13:49:11 +0700 Subject: [PATCH 1/8] Give a worker a way to stop hoarding self-waking tasks A job displaced from a worker's LIFO slot went to a private inbox behind it. Neither is stealable, and the path that was supposed to rescue them cannot run: 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 sits ahead of that in the loop. A task that keeps waking itself keeps the slot occupied, so the heartbeat never ticks, so nothing this worker holds ever becomes reachable. Measured on sixteen workers with eight self-waking read-only client tasks: four runs in eight collapsed to a single active worker with the other fifteen parked in `__ulock_wait`, at exactly the one-thread rate. That looks exactly like a lost wakeup and is not one, which is why the `SeqCst` Dekker pair did not touch it. Every wake was delivered. The work became invisible. `share_displaced` sends the displaced job to the injector instead, so at most one task is ever private to a worker. It is off by default and it is a policy rather than a fix: every displacement then costs an injector push and a wake, and on work that contends on row locks there is no parallelism to win and the churn is pure cost. A 50% update workload fell from 943,606 to 651,554 while cores busy went from 1.13 to 13.16. `nudge` also woke exactly one sleeper however many jobs had just been made stealable, so a worker sharing eight jobs published seven that nobody was told about. It now wakes one per job. --- src/fanout/mod.rs | 96 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 90 insertions(+), 6 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 9dfbc15..ce01806 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -104,6 +104,33 @@ pub struct Tuning { /// contention-free work; too large and one worker holds a backlog its /// neighbours cannot see until the next heartbeat. pub injector_batch: usize, + /// 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`. + 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 @@ -164,6 +191,7 @@ impl Tuning { #[must_use] pub fn locality() -> Self { Self { + share_displaced: false, rounds_before_park: ROUNDS_BEFORE_PARK, backoff_spins: BACKOFF_SPINS, promote_every: PROMOTE_EVERY, @@ -212,6 +240,13 @@ impl Tuning { } } + /// Set [`share_displaced`](Self::share_displaced). + #[must_use] + pub fn with_share_displaced(mut self, share: bool) -> Self { + self.share_displaced = share; + self + } + /// Set [`rounds_before_park`](Self::rounds_before_park). #[must_use] pub fn with_rounds_before_park(mut self, rounds: u32) -> Self { @@ -660,11 +695,38 @@ impl Pool { self.push(None, job); return; } - // The slot first, and whatever it held goes to the inbox behind it. - // One swap on the common path, against a `SegQueue` push and pop. + // 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. let displaced = self.local[worker].lifo.lock().replace(job); if let Some(displaced) = displaced { - self.local[worker].inbox.push(displaced); + if self.tuning.share_displaced { + self.push(Some(worker), displaced); + } else { + self.local[worker].inbox.push(displaced); + } } // Publish, then read the bitmap, with a barrier between. // @@ -716,15 +778,34 @@ impl Pool { /// Used when work becomes *stealable* rather than newly submitted: a batch /// pulled off the injector, or a private queue shared on the heartbeat. fn nudge(&self, except: Option) { + self.nudge_many(except, 1); + } + + /// Wake up to `wanted` sleepers, other than `except`. + /// + /// **One wake per job made available, not one per sharing event.** A + /// worker that shares eight jobs and wakes one sleeper has published work + /// seven other parked workers cannot see: the woken one takes what it + /// needs, and nothing else generates a wake, because a task that self-wakes + /// stays local by design and never touches the injector again. + /// + /// That is not a lost wakeup, which is why the `SeqCst` Dekker pair did not + /// fix it. Every wake that was sent arrived. There were simply too few of + /// them, and the pool has no timeout that would notice. + fn nudge_many(&self, except: Option, wanted: usize) { let spare = match except { Some(id) => 1usize << id, None => 0, }; - let sleeping = self.sleeping.load(Ordering::SeqCst) & !spare; - if sleeping != 0 { + let mut sleeping = self.sleeping.load(Ordering::SeqCst) & !spare; + for _ in 0..wanted { + if sleeping == 0 { + return; + } #[allow(clippy::cast_possible_truncation)] let candidate = sleeping.trailing_zeros() as usize; self.host.unpark(candidate); + sleeping &= !(1usize << candidate); } } @@ -1080,7 +1161,10 @@ impl Pool { } } if shared > 0 { - self.nudge(Some(id)); + // 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. + self.nudge_many(Some(id), shared); } } From f47bffffd0a31f227989e47387bf7c3ded53f872 Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 14:10:26 +0700 Subject: [PATCH 2/8] Let a caller say how long the LIFO slot may hold a worker The slot sits ahead of everything in the worker loop, so a task that keeps waking itself is served from it indefinitely and the worker never reaches the `mine.pop_back` that counts down to the heartbeat. Anything it is holding for somebody else therefore never becomes stealable, however long the queue behind it grows. The cap existed but was a `const` nobody could reach. Lower means the worker looks at its own queue sooner, and that costs nothing when there is nothing there: `share` returns immediately unless the private queue holds at least two jobs. Which is what makes it a cheaper answer to a pile-up than routing displaced work to the injector, where a wake is paid whether or not anybody needed one. Measured on YCSB C, sixteen workers, eight self-waking client tasks. At the old 32 the run collapsed to the one-thread rate in one attempt out of six, with the slowest of six at 2,718,656. At 4 the slowest of six was 12,912,445 and the median rose from 12,466,259 to 13,398,239. It is not free: the same change costs a 50% update workload 3.7% and a read-modify-write workload 4.1%, which is why it is a tuning and not a new constant. --- src/fanout/mod.rs | 60 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 57 insertions(+), 3 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index ce01806..666af7f 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -104,6 +104,21 @@ pub struct Tuning { /// contention-free work; too large and one worker holds a backlog its /// neighbours cannot see until the next heartbeat. pub injector_batch: usize, + /// How many self-wakes may be served from the LIFO slot back to back + /// before the loop insists on looking elsewhere. + /// + /// The slot sits ahead of everything in the worker loop, so a task that + /// keeps waking itself is served from it indefinitely and the worker never + /// reaches the `mine.pop_back` that counts down to the heartbeat. Anything + /// 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. + 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 @@ -191,6 +206,7 @@ impl Tuning { #[must_use] pub fn locality() -> Self { Self { + lifo_run_limit: LIFO_RUN_LIMIT, share_displaced: false, rounds_before_park: ROUNDS_BEFORE_PARK, backoff_spins: BACKOFF_SPINS, @@ -240,6 +256,13 @@ impl Tuning { } } + /// Set [`lifo_run_limit`](Self::lifo_run_limit). + #[must_use] + pub fn with_lifo_run_limit(mut self, limit: u32) -> Self { + self.lifo_run_limit = limit; + self + } + /// Set [`share_displaced`](Self::share_displaced). #[must_use] pub fn with_share_displaced(mut self, share: bool) -> Self { @@ -287,6 +310,10 @@ impl Tuning { } } +/// Self-wakes served from the LIFO slot back to back before the worker looks +/// 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 @@ -722,7 +749,34 @@ impl Pool { // one somebody else's to take. let displaced = self.local[worker].lifo.lock().replace(job); if let Some(displaced) = displaced { - if self.tuning.share_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() { self.push(Some(worker), displaced); } else { self.local[worker].inbox.push(displaced); @@ -880,7 +934,7 @@ impl Pool { // tasks ping-ponging through the slot starve everything this worker // holds, including work it has already promised to share. let mut lifo_run = 0u32; - const LIFO_RUN_LIMIT: u32 = 32; + let lifo_run_limit = self.tuning.lifo_run_limit; loop { // 0. The task that just woke itself, if there is one. Ahead of @@ -888,7 +942,7 @@ impl Pool { // slot is one uncontended swap to check. // Written without a `let` chain: this crate supports Rust 1.60 // and those are 2024. - let from_slot = if lifo_run < LIFO_RUN_LIMIT { + let from_slot = if lifo_run < lifo_run_limit { self.local[w.id].lifo.lock().take() } else { None From 434effec18d53682a3f9f8a86363353cf37449d0 Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 14:28:26 +0700 Subject: [PATCH 3/8] Publish displaced work without announcing it `share_displaced` sent the job through `push`, which also unparks a sleeper, and that wake was most of what the policy cost. On work that contends on row locks the woken worker steals the job, runs it, blocks on the same lock and parks again, so the pool pays a park and an unpark per displacement for work that was serialised anyway. The wake is not needed for the job to be found. A worker re-checks the injector after announcing sleep and before parking, so it cannot park while this job is queued, and the worker that displaced it is by definition awake and reaches the injector itself. Anybody already asleep is asleep because there was no work, and one job does not change that. Measured against the old engine, eight client threads, six paired rounds: the policy went from 14,734 behind on a 50% update workload to 4,428 behind, and from 55,526 behind on read-modify-write to 26,835. At sixteen threads on a read-only workload it took the same policy from a median of 11,671,193 to 15,049,665. --- src/fanout/mod.rs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 666af7f..5def368 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -777,7 +777,21 @@ impl Pool { // 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() { - self.push(Some(worker), displaced); + // **Published, not announced.** `push` would also unpark a + // sleeper, and that 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 wake is not needed for the job to be found. A worker + // re-checks the injector after announcing sleep and before + // parking, so it cannot park while this job is queued; the + // worker that displaced it is by definition awake and will + // reach the injector itself. Anybody already asleep is asleep + // because there was no work, and this one job does not change + // that. + self.injector.push(displaced); } else { self.local[worker].inbox.push(displaced); } From 528ddc261ffe6f9a7b73d28fa987cde033a8d162 Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 14:33:10 +0700 Subject: [PATCH 4/8] Release 0.6.1, the tunings the flavor sweep needed --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index a2b8959..8bc671c 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.0" +version = "0.6.1" authors = ["Serge Barral "] edition = "2021" rust-version = "1.60" From 54a03f7329ecc7fc31e7ec956ab749fd02fd3eee Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 20:07:39 +0700 Subject: [PATCH 5/8] Claim a sleeper when waking one, do not merely name it wake read sleeping.trailing_zeros() and unparked that worker without clearing the bit. The bit is not cleared until the woken worker actually runs, so every submit in a burst read the same candidate and unparked one worker over and over while the rest stayed parked. Every wake was delivered. They all went to the same worker, which is why this never looked like a lost wakeup and why the SeqCst Dekker pair around it was beside the point. Measured through nagoya on a 16 core machine, 16 workers and 32 tasks, counting cores kept busy, which is task CPU time over wall time, so 16.0 is the whole machine and higher is better. Worst of eight runs: tuning shape before after locality self-waking 3.9 10.3 locality single-poll 4.0 11.4 spread self-waking 2.9 12.6 spread single-poll 2.9 13.9 shared_slot self-waking 4.0 10.4 shared_slot single-poll 4.0 13.8 A fresh pool always reached about 12.7 and every pool that had parked once fell to about 4.0, on repeat, for every tuning. That degradation is gone. Tasks polled exactly once collapsed identically to self-waking ones. That rules out the LIFO slot, since a task polled once never touches it, and names the unpark path instead. nudge_many already masked each candidate it took, but only inside one call, which is why the bug was invisible there. Clearing the bit is that same rule held across calls. Clearing early only claims the sleeper: the worker clears it itself on resume and sets it again if it parks again. Two deterministic tests, checked by mutation, and no wall clock in either. --- src/fanout/mod.rs | 153 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 144 insertions(+), 9 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 5def368..41b1789 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -825,11 +825,7 @@ impl Pool { /// and another is parked is the bug this exists to avoid: the job waits out /// unrelated work while a free worker sleeps beside it. fn wake(&self, hint: Option) { - // Ordered against publishing the job above; see `sleeping`. - let sleeping = self.sleeping.load(Ordering::SeqCst); - if sleeping != 0 { - #[allow(clippy::cast_possible_truncation)] - let candidate = sleeping.trailing_zeros() as usize; + if let Some(candidate) = self.claim_sleeper(0) { self.host.unpark(candidate); return; } @@ -865,15 +861,52 @@ impl Pool { Some(id) => 1usize << id, None => 0, }; - let mut sleeping = self.sleeping.load(Ordering::SeqCst) & !spare; for _ in 0..wanted { - if sleeping == 0 { + let Some(candidate) = self.claim_sleeper(spare) else { return; + }; + self.host.unpark(candidate); + } + } + + /// Take one sleeper, rather than merely naming one. + /// + /// The bit says "announced sleep and not yet woken". Reading it and + /// unparking left the bit set until the woken worker actually ran, so a + /// burst of submits all read the same `trailing_zeros` candidate and + /// unparked one worker over and over while the rest stayed parked. The + /// wakes were all delivered; they all went to the same worker. + /// + /// Measured on a 16 core machine with 16 workers and 32 tasks, counting + /// cores kept busy (task CPU time over wall time, so 16.0 is the whole + /// machine and higher is better). A fresh pool held 12.7 of 16 cores busy. + /// Every pool that had parked once held 4.0, on repeat, for every tuning. + /// + /// Self-waking made no difference to it, which is what ruled out the LIFO + /// slot: a task polled exactly once cannot touch that machinery and + /// collapsed the same way. After this change the worst run of eight held + /// 10.3 of 16 cores busy rather than 2.9. + /// + /// `nudge_many` already masked each candidate it took, but only inside one + /// call. Clearing the bit here is that same rule, held across calls. + /// + /// Clearing early only claims the sleeper: the worker clears this itself + /// when it resumes and sets it again if it parks again, so a claim that + /// races a worker waking on its own costs nothing. The `SeqCst` read + /// stays ordered against publishing the job, and the read-modify-write is + /// no weaker. + fn claim_sleeper(&self, blocked: usize) -> Option { + loop { + let sleeping = self.sleeping.load(Ordering::SeqCst) & !blocked; + if sleeping == 0 { + return None; } #[allow(clippy::cast_possible_truncation)] let candidate = sleeping.trailing_zeros() as usize; - self.host.unpark(candidate); - sleeping &= !(1usize << candidate); + let bit = 1usize << candidate; + if self.sleeping.fetch_and(!bit, Ordering::SeqCst) & bit != 0 { + return Some(candidate); + } } } @@ -1296,3 +1329,105 @@ impl core::fmt::Debug for Pool { .finish_non_exhaustive() } } + +#[cfg(all(test, feature = "host"))] +mod wake_distribution { + use super::{Host, Pool, Tuning}; + use alloc::sync::Arc; + use alloc::vec::Vec; + use core::sync::atomic::Ordering; + use std::collections::HashSet; + use std::sync::Mutex; + + /// Records who was told to wake, and parks nobody. + /// + /// The pathology is about *which* worker each wake names, so the host only + /// has to remember the names. + struct RecordingHost { + unparked: Mutex>, + } + + impl Host for RecordingHost { + fn park(&self, _worker: usize) {} + fn unpark(&self, worker: usize) { + self.unparked.lock().expect("no state a panic could corrupt").push(worker); + } + fn now_ns(&self) -> u64 { + 0 + } + } + + /// A burst of submits must reach distinct sleepers. + /// + /// `wake` used to read `sleeping.trailing_zeros()` and unpark it without + /// clearing the bit, so every submit in a burst named the same worker: the + /// bit stays set until the woken worker actually runs. Every wake was + /// delivered and they all went to one worker, which is why it did not look + /// like a lost wakeup. + /// + /// Measured through `nagoya` on a 16 core machine, 16 workers and 32 + /// tasks, counting cores kept busy (task CPU time over wall time, so 16.0 + /// is the whole machine): a fresh pool held 12.7 busy and a pool that had + /// parked once held 4.0, on repeat, for every tuning. Tasks polled exactly + /// once collapsed identically to self-waking ones, which is what ruled out + /// the LIFO slot and named this instead. + #[test] + fn a_burst_of_wakes_reaches_distinct_sleepers() { + const WORKERS: usize = 8; + + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning(WORKERS, 64, host.clone(), Tuning::default()); + + // Everyone is asleep, which is the state a pool sits in between bursts + // of work and the state this used to collapse from. + pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + + for _ in 0..WORKERS { + pool.wake(None); + } + + let unparked = host.unparked.lock().expect("the log").clone(); + let distinct: HashSet = unparked.iter().copied().collect(); + assert_eq!( + distinct.len(), + WORKERS, + "{WORKERS} submits against {WORKERS} sleepers woke {} distinct workers, {unparked:?}: \ + a wake that names a sleeper without claiming it names the same one every time", + distinct.len() + ); + assert_eq!( + pool.sleeping.load(Ordering::SeqCst), + 0, + "a claimed sleeper must leave the bitmap, or the next submit picks it again" + ); + } + + /// `nudge_many` must not hand the same worker to two callers. + /// + /// It already masked each candidate inside one call, which is why the bug + /// above was invisible here. Two calls had no such memory. + #[test] + fn two_nudges_do_not_pick_the_same_sleeper() { + const WORKERS: usize = 8; + + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning(WORKERS, 64, host.clone(), Tuning::default()); + pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + + pool.nudge_many(None, 3); + pool.nudge_many(None, 3); + + let unparked = host.unparked.lock().expect("the log").clone(); + let distinct: HashSet = unparked.iter().copied().collect(); + assert_eq!( + distinct.len(), + 6, + "two nudges of three woke {} distinct workers, {unparked:?}", + distinct.len() + ); + } +} From 5db19762e8f8208806e2d89de7bb3ab7149f09bd Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 20:16:55 +0700 Subject: [PATCH 6/8] Wake the worker that holds a private job, not any sleeper submit_local puts a job in one worker's LIFO slot or inbox. Neither is stealable. It then called wake(Some(worker)), but wake treats its argument as a fallback and prefers trailing_zeros() whenever anybody is asleep. So a job handed to worker 7 while worker 0 was also asleep unparked worker 0. Worker 0 woke, found nothing it was allowed to take, and parked again. Worker 7 slept on work no other worker could reach. A wake was sent every time, which is why this survived a hunt for a lost wakeup and the SeqCst fence added directly above it. The fence was needed and fixed a different miss; this one is not an ordering problem at all. The wake went to the wrong worker. This is the shape that shows up as one busy core with tasks outstanding, because the stranded worker is stranded for as long as nothing else happens to wake it. wake and wake_specific are now distinct and the doc comments say which is which: the difference is whether the work being announced is stealable. The regression test hands a job to worker 5 while every worker is asleep and asserts worker 5 is the one woken. Under the old call it reports "a job private to worker 5 woke [0]". Worker 5 rather than worker 1 on purpose, since a two-worker test where the holder is also the lowest sleeper cannot see this. --- src/fanout/mod.rs | 65 +++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 63 insertions(+), 2 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 41b1789..5b5e6f5 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -808,8 +808,34 @@ 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); - if self.sleeping.load(Ordering::SeqCst) & (1usize << worker) != 0 { - self.wake(Some(worker)); + // **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. + // + // `wake` cannot be used here. It treats its argument as a fallback and + // takes `trailing_zeros()` whenever any worker is asleep, so a private + // job handed to worker 7 while worker 0 was also asleep unparked + // worker 0, which found nothing it could take and parked again, while + // worker 7 slept on work nobody could reach. That is a lost wakeup + // 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); + } + + /// Wake one named worker, because the work is private to it. + /// + /// Distinct from [`Pool::wake`], which is free to prefer any sleeper + /// because the job it is announcing went to the injector where anyone can + /// reach it. The two must not be confused: the difference is whether the + /// work is stealable. + /// + /// The read of the bit is folded into the claim. It is still a `SeqCst` + /// read-modify-write, so the Dekker pair the fence above exists for is + /// unchanged. + fn wake_specific(&self, worker: usize) { + let bit = 1usize << worker; + if self.sleeping.fetch_and(!bit, Ordering::SeqCst) & bit != 0 { + self.host.unpark(worker); } } @@ -1404,6 +1430,41 @@ mod wake_distribution { ); } + /// Private work must wake the worker that holds it, not any sleeper. + /// + /// `submit_local` puts the job in one worker's slot or inbox, and neither + /// is stealable. It then called `wake(Some(worker))`, but `wake` treats its + /// argument as a fallback and prefers `trailing_zeros()` whenever anybody + /// is asleep. So a job handed to worker 5 while worker 0 was also asleep + /// unparked worker 0, which found nothing it was allowed to take and + /// parked again, while worker 5 slept on the job. + /// + /// A wake was sent every time, which is why this survived a hunt for a + /// lost wakeup and an added `SeqCst` fence. It went to the wrong worker. + #[test] + fn private_work_wakes_the_worker_that_holds_it() { + const WORKERS: usize = 8; + // Deliberately not worker 0: the bug is invisible when the worker that + // holds the job is also the lowest-numbered sleeper, which is exactly + // how a two-worker test would miss it. + const HOLDER: usize = 5; + + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning(WORKERS, 64, host.clone(), Tuning::locality()); + pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + + pool.submit_local(HOLDER, super::Job::from_boxed(|| {})); + + let unparked = host.unparked.lock().expect("the log").clone(); + assert_eq!( + unparked, + alloc::vec![HOLDER], + "a job private to worker {HOLDER} woke {unparked:?}; only {HOLDER} can run it" + ); + } + /// `nudge_many` must not hand the same worker to two callers. /// /// It already masked each candidate inside one call, which is why the bug From 9495ae3e04f0b8ecd42f57cf2752cdb4928a79a4 Mon Sep 17 00:00:00 2001 From: meh Date: Thu, 10 Sep 2026 21:04:59 +0700 Subject: [PATCH 7/8] Nudge on what is left in the injector, not on what was taken The condition was `taken > 1`, and injector_batch defaults to 1, so the take loop can never take more than one and the nudge never fired for any flavor that ships. It was dead code. A worker that woke, found sixteen jobs queued and took one told nobody. It then looped and took another, and another, draining the injector before its neighbours had finished waking. Under local_wakes that concentration is permanent, because a task's waker returns it to the worker that last polled it, so whichever worker grabbed a task at startup owns it for the run. YCSB C, 16 client threads, 16 workers, 1.5M operations, 16 repetitions: min p50 max cpu_x ops/core tokio 13,026,746 14,792,857 20,115,733 12.8 1,157,605 before 8,697,719 10,216,732 21,301,082 7.5 1,361,285 after 12,021,390 15,484,396 16,737,289 12.9 1,202,396 cpu_x 7.5 to 12.9 against tokio's 12.8 is the result. Per-core rates were always within 3% of tokio, so the entire gap was cores left idle, and the p50 above tokio's sits inside the overlap and is parity rather than a win. The regression test asserts distinct workers woken, not timing. Against the old condition it reports "7 jobs were left queued and 0 workers were woken". --- src/fanout/mod.rs | 69 +++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 67 insertions(+), 2 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 5b5e6f5..8ee60be 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -1234,8 +1234,26 @@ impl Pool { None => break, } } - if taken > 1 { - self.nudge(None); + // **What is left, not what was taken.** This was `taken > 1`, which is + // dead code for every flavor that ships: `injector_batch` defaults to + // 1, so the loop above can never take more than one and the condition + // is never true. A worker that woke, found sixteen jobs queued and took + // one told nobody, then looped and took another, and another, draining + // the injector before its neighbours had finished waking. + // + // That concentration is permanent under `local_wakes`, which is the + // point: a task's waker sends it back to the worker that last polled + // it, so whichever worker grabbed a task at startup owns it for the + // whole run. Measured on YCSB C with sixteen client tasks on sixteen + // workers: 7.3 cores kept busy against tokio's 13.0, at per-core rates + // within 3% of each other, so the entire gap was cores left idle. + // + // One wake per job still queued, which is the rule `nudge_many` + // documents. Bounded by the worker count because there is no point + // naming more sleepers than exist. + let remaining = self.injector.len(); + if remaining > 0 { + self.nudge_many(None, remaining.min(self.workers())); } taken } @@ -1430,6 +1448,53 @@ mod wake_distribution { ); } + /// Taking from a queue that still has work must wake somebody. + /// + /// The condition here was `taken > 1`, which is dead code for every flavor + /// that ships: `injector_batch` defaults to 1, so the take loop can never + /// take more than one and the nudge never fired. A worker that woke, found + /// sixteen jobs queued and took one told nobody, then looped and took + /// another, draining the injector before its neighbours finished waking. + /// + /// Under `local_wakes` that concentration is permanent, because a task's + /// waker returns it to the worker that last polled it. YCSB C with sixteen + /// client tasks on sixteen workers kept 7.3 cores busy against tokio's + /// 13.0, at per-core rates within 3%, so the whole gap was idle cores. + /// After this, 12.9. + /// + /// The assertion is on distinct workers woken rather than on timing. + #[test] + fn taking_one_job_wakes_someone_for_the_rest() { + const WORKERS: usize = 8; + const QUEUED: usize = 8; + + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + // injector_batch 1 is the default and the case that was broken. + let pool = Pool::with_tuning(WORKERS, 64, host.clone(), Tuning::locality()); + assert_eq!(pool.tuning.injector_batch, 1, "the default this guards"); + + for _ in 0..QUEUED { + pool.injector.push(super::Job::from_boxed(|| {})); + } + pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + + let mut mine = super::VecDeque::new(); + let taken = pool.take_from_injector(&mut mine); + + assert_eq!(taken, 1, "one, because that is the batch size"); + let unparked = host.unparked.lock().expect("the log").clone(); + let distinct: HashSet = unparked.iter().copied().collect(); + assert_eq!( + distinct.len(), + QUEUED - 1, + "{} jobs were left queued and {} workers were woken, {unparked:?}", + QUEUED - 1, + distinct.len() + ); + } + /// Private work must wake the worker that holds it, not any sleeper. /// /// `submit_local` puts the job in one worker's slot or inbox, and neither From b55035957c51af9b9421f731d117827b8a41ceb2 Mon Sep 17 00:00:00 2001 From: meh Date: Fri, 11 Sep 2026 05:32:10 +0700 Subject: [PATCH 8/8] Fold PR 5's review findings, minus one that was already answered Applied from `ps-st3-pr5-source-fixes.patch` in ~/code/patch, whose author states that nothing was compiled or run. It is compiled and run now: 44 tests pass, and WorkTable's 999 pass against this tree, which is what its `[patch.crates-io]` points at. Taken as written: A zero `lifo_run_limit` is rejected at construction. It would never consume the slot, so a job put there is stranded, and the failure is silent. The wake-selection path uses `match` rather than `let`-`else`, which was over the declared 1.60 floor. `wake_specific` reads the bitmap before it writes it. An already-awake owner no longer takes a shared cache line exclusively for a bit that is not set. The publication fence and the SeqCst load keep the Dekker pair intact. The worker services the injector at the LIFO fairness boundary, so a one-worker pool, or one whose peers are all busy, makes progress on displaced work instead of waiting behind self-wakes. Not taken: the wake on displacement. The patch adds `nudge(Some(worker))` after publishing a displaced job, against a comment that had removed exactly that call. The comment is right, and the review's objection is right too, and they do not actually conflict. The objection is starvation: an owner servicing self-wakes forever leaves the job behind while peers sleep. The fairness-boundary service above already fixes that, and it ends in `take_from_injector`, which wakes one sleeper per job still queued. The sleepers are told. They are told once per fairness boundary rather than once per displacement, which is the same news for a fraction of the round trips. A wake here would be a third mechanism for one job, and the one measured to cost the most: on lock-bound work the woken worker steals the job, blocks on the same row lock and parks again, paying a park/unpark per displacement for work that was serialised anyway. Its test asserted the wake happened. It asserts the absence now, and says why, so a future patch adding the wake back fails and has to argue with the comment rather than landing quietly. --- src/fanout/mod.rs | 167 +++++++++++++++++++++++++++++++++++++++------- 1 file changed, 143 insertions(+), 24 deletions(-) diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 8ee60be..4a11fff 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -256,7 +256,8 @@ impl Tuning { } } - /// Set [`lifo_run_limit`](Self::lifo_run_limit). + /// Set [`lifo_run_limit`](Self::lifo_run_limit). Must be nonzero; + /// pool construction rejects zero because it would never consume the slot. #[must_use] pub fn with_lifo_run_limit(mut self, limit: u32) -> Self { self.lifo_run_limit = limit; @@ -583,6 +584,10 @@ impl Pool { tuning: Tuning, ) -> Arc { assert!(workers > 0, "a pool needs at least one worker"); + assert!( + tuning.lifo_run_limit > 0, + "`lifo_run_limit` must be nonzero" + ); assert!( tuning.promote_every > 0, "`promote_every` is a countdown to zero: 0 never reaches it" @@ -777,20 +782,26 @@ impl Pool { // 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` would also unpark a - // sleeper, and that 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. + // **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 wake is not needed for the job to be found. A worker - // re-checks the injector after announcing sleep and before - // parking, so it cannot park while this job is queued; the - // worker that displaced it is by definition awake and will - // reach the injector itself. Anybody already asleep is asleep - // because there was no work, and this one job does not change - // that. + // 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. self.injector.push(displaced); } else { self.local[worker].inbox.push(displaced); @@ -829,11 +840,14 @@ impl Pool { /// reach it. The two must not be confused: the difference is whether the /// work is stealable. /// - /// The read of the bit is folded into the claim. It is still a `SeqCst` - /// read-modify-write, so the Dekker pair the fence above exists for is - /// unchanged. + /// Avoid modifying the shared bitmap for an already-awake owner. The + /// preceding publication fence and this SeqCst load retain the sleep + /// handshake; only an observed sleeper requires a modifying claim. fn wake_specific(&self, worker: usize) { let bit = 1usize << worker; + if self.sleeping.load(Ordering::SeqCst) & bit == 0 { + return; + } if self.sleeping.fetch_and(!bit, Ordering::SeqCst) & bit != 0 { self.host.unpark(worker); } @@ -888,8 +902,9 @@ impl Pool { None => 0, }; for _ in 0..wanted { - let Some(candidate) = self.claim_sleeper(spare) else { - return; + let candidate = match self.claim_sleeper(spare) { + Some(candidate) => candidate, + None => return, }; self.host.unpark(candidate); } @@ -1027,6 +1042,12 @@ impl Pool { self.local[w.id].completed.fetch_add(1, Ordering::Relaxed); continue; } + // 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 { + self.take_from_injector(&mut mine); + } lifo_run = 0; // 1. This worker's own queue. No synchronisation whatsoever. @@ -1391,10 +1412,104 @@ mod wake_distribution { unparked: Mutex>, } + /// Displacement publishes to the injector without waking anybody. + /// + /// The wake is deliberately absent. It costs a park/unpark round trip per + /// displacement on lock-bound work, where the woken worker steals the job, + /// blocks on the same lock and parks again. The sleepers are reached from + /// `take_from_injector` instead, which the owner reaches at the LIFO + /// fairness boundary and which wakes one per job still queued. + /// + /// This asserts the absence, because the absence is the decision. A patch + /// that adds the wake back should fail here and argue with the comment + /// rather than land quietly. + #[test] + fn displacement_publishes_without_paying_for_a_wake() { + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning( + 2, + 64, + host.clone(), + Tuning::locality().with_share_displaced(true), + ); + pool.sleeping.store(1 << 1, Ordering::SeqCst); + pool.submit_local(0, super::Job::from_boxed(|| {})); + 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!(pool.injector.len(), 1, "the third displacement is public"); + assert!( + host.unparked.lock().unwrap().is_empty(), + "publishing a displaced job must not cost a wake" + ); + } + + #[test] + #[should_panic(expected = "lifo_run_limit")] + fn zero_lifo_budget_is_rejected_before_work_can_be_stranded() { + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let _ = Pool::with_tuning(1, 64, host, Tuning::locality().with_lifo_run_limit(0)); + } + + #[test] + fn one_worker_services_displaced_work_before_local_retry_exhaustion() { + use alloc::sync::Weak; + use core::sync::atomic::{AtomicBool, AtomicUsize}; + fn retry(pool: Weak, done: Arc, polls: Arc) -> super::Job { + super::Job::from_boxed(move || { + if done.load(Ordering::Acquire) { + return; + } + // A finite sentinel makes a broken scheduler fail an assertion, + // rather than making the test itself wait forever. + if polls.fetch_add(1, Ordering::Relaxed) < 512 { + let owner = pool.upgrade().unwrap(); + owner.submit_local(0, retry(pool, done, polls)); + } + }) + } + let host = Arc::new(RecordingHost { + unparked: Mutex::new(Vec::new()), + }); + let pool = Pool::with_tuning( + 1, + 64, + host, + Tuning::locality() + .with_share_displaced(true) + .with_lifo_run_limit(4), + ); + let done = Arc::new(AtomicBool::new(false)); + let polls = Arc::new(AtomicUsize::new(0)); + pool.submit_local(0, retry(Arc::downgrade(&pool), done.clone(), polls.clone())); + let (owner, completed) = (Arc::downgrade(&pool), done.clone()); + pool.submit_local( + 0, + super::Job::from_boxed(move || { + completed.store(true, Ordering::Release); + owner.upgrade().unwrap().shut_down(); + }), + ); + pool.submit_local(0, retry(Arc::downgrade(&pool), done.clone(), polls.clone())); + assert!(pool.run(pool.runner(0))); + assert!(done.load(Ordering::Acquire)); + assert!( + polls.load(Ordering::Relaxed) < 512, + "public work starved behind private retries" + ); + } + impl Host for RecordingHost { fn park(&self, _worker: usize) {} fn unpark(&self, worker: usize) { - self.unparked.lock().expect("no state a panic could corrupt").push(worker); + self.unparked + .lock() + .expect("no state a panic could corrupt") + .push(worker); } fn now_ns(&self) -> u64 { 0 @@ -1426,7 +1541,8 @@ mod wake_distribution { // Everyone is asleep, which is the state a pool sits in between bursts // of work and the state this used to collapse from. - pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + pool.sleeping + .store((1usize << WORKERS) - 1, Ordering::SeqCst); for _ in 0..WORKERS { pool.wake(None); @@ -1478,7 +1594,8 @@ mod wake_distribution { for _ in 0..QUEUED { pool.injector.push(super::Job::from_boxed(|| {})); } - pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + pool.sleeping + .store((1usize << WORKERS) - 1, Ordering::SeqCst); let mut mine = super::VecDeque::new(); let taken = pool.take_from_injector(&mut mine); @@ -1518,7 +1635,8 @@ mod wake_distribution { unparked: Mutex::new(Vec::new()), }); let pool = Pool::with_tuning(WORKERS, 64, host.clone(), Tuning::locality()); - pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + pool.sleeping + .store((1usize << WORKERS) - 1, Ordering::SeqCst); pool.submit_local(HOLDER, super::Job::from_boxed(|| {})); @@ -1542,7 +1660,8 @@ mod wake_distribution { unparked: Mutex::new(Vec::new()), }); let pool = Pool::with_tuning(WORKERS, 64, host.clone(), Tuning::default()); - pool.sleeping.store((1usize << WORKERS) - 1, Ordering::SeqCst); + pool.sleeping + .store((1usize << WORKERS) - 1, Ordering::SeqCst); pool.nudge_many(None, 3); pool.nudge_many(None, 3);