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" diff --git a/src/fanout/mod.rs b/src/fanout/mod.rs index 9dfbc15..4a11fff 100644 --- a/src/fanout/mod.rs +++ b/src/fanout/mod.rs @@ -104,6 +104,48 @@ 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 + /// 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 +206,8 @@ 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, promote_every: PROMOTE_EVERY, @@ -212,6 +256,21 @@ impl Tuning { } } + /// 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; + self + } + + /// 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 { @@ -252,6 +311,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 @@ -521,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" @@ -660,11 +727,85 @@ 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); + // **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. + self.injector.push(displaced); + } else { + self.local[worker].inbox.push(displaced); + } } // Publish, then read the bitmap, with a barrier between. // @@ -678,8 +819,37 @@ 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. + /// + /// 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); } } @@ -695,11 +865,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; } @@ -716,15 +882,72 @@ 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 { + for _ in 0..wanted { + let candidate = match self.claim_sleeper(spare) { + Some(candidate) => candidate, + None => 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); + let bit = 1usize << candidate; + if self.sleeping.fetch_and(!bit, Ordering::SeqCst) & bit != 0 { + return Some(candidate); + } } } @@ -799,7 +1022,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 @@ -807,7 +1030,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 @@ -819,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. @@ -1026,8 +1255,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 } @@ -1080,7 +1327,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); } } @@ -1144,3 +1394,285 @@ 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>, + } + + /// 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); + } + 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" + ); + } + + /// 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 + /// 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 + /// 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() + ); + } +}