gatling/src/gatling_map.rs
//! **The PARKED pool** — workers that live on the cores and pick up work, for the same
//! shaped job run over and over.
//!
//! # Why this exists
//! [`gatling_forkjoin`](crate::gatling_forkjoin) is the ONE-SHOT fan-out: it calls
//! `std::thread::scope` and spawns fresh OS threads **on every call**, then joins them.
//! That is right for a job you run once — a file to decode, a planet to convert.
//!
//! It is wrong for a job you run **sixty times a second**. Measured on oden
//! (Threadripper PRO 3975WX, 32 threads), `facett-graph3d`'s per-frame point
//! projection:
//!
//! | lane | 100 000 items |
//! |---|---|
//! | serial, one core | **0.093 ms** |
//! | `gatling_for_each` (grain 1) | 4.202 ms |
//! | static contiguous split, threads spawned per call | 0.639 ms |
//! | **32 threads spawned + joined doing NOTHING** | **0.559 ms** |
//!
//! Two separate costs, and the fix for one does not touch the other:
//!
//! * **grain** — `gatling_for_each` claims ONE index per contended `fetch_add`, so a
//! 100k map pays 100 000 atomics on a single cache line across every core. Fixing the
//! grain alone is worth 6.6× (4.202 → 0.639 ms).
//! * **spawn** — and then 0.559 ms of that 0.639 ms is *thread creation*. At 60 fps that
//! is ~1900 OS threads per second created and destroyed to do 5.6 ms of actual work.
//! No split can fix it, because the floor is paid before the split is even reached.
//!
//! # What this is
//! Workers are spawned **once** and park on a condvar. A call publishes the job and wakes
//! them; they take a **contiguous range each** (static split, so **zero atomics inside the
//! loop**) and park again. The threads outlive the call, so a render loop pays the spawn
//! cost once for the process instead of once per frame.
//!
//! # The split, and why it is trivial here
//! A `&[T]` of fixed-stride items has **no border problem**: every index is already a
//! valid boundary. The 32-offset-candidates-then-scan-to-a-real-border dance is for
//! *variable-length records in a byte stream* (XML/JSON/PBF), where a cut can land
//! mid-record. Fixed-stride data is pre-split by construction, so the split is
//! `n.div_ceil(workers)` and nothing more.
//!
//! # Not a replacement
//! This does NOT deprecate `gatling_forkjoin`. Use that when the job runs once, when
//! items cost wildly different amounts (its per-item claim IS the load balancer), or when
//! you want no long-lived threads. Use this when the SAME shaped job repeats and items
//! cost about the same.
//!
//! # wasm
//! Same rule as the rest of gatling: [`workers`](Pool::workers) collapses to 1 where
//! `available_parallelism` is unsupported, and the map then runs inline on the caller —
//! no pool is created, no thread is spawned.
use std::sync::{Arc, Condvar, Mutex};
/// A raw pointer to the caller's `&dyn Fn(worker, workers)`, parked in the job slot.
///
/// # Safety
/// The referent lives on the **caller's stack** inside [`Pool::for_each_range`], which
/// publishes this pointer, wakes the workers, and then blocks until every worker has
/// reported done. So the closure is alive for every dereference, and no worker can touch
/// it after the call returns. This is the whole safety argument; it is why the wait loop
/// at the end of `for_each_range` is not optional.
#[derive(Clone, Copy)]
struct JobPtr(*const (dyn Fn(usize, usize) + Send + Sync));
// SAFETY: see JobPtr — the pointee is `Send + Sync` and outlives every dereference.
unsafe impl Send for JobPtr {}
unsafe impl Sync for JobPtr {}
/// Completion bookkeeping for one dispatch: how many workers finished, and whether any
/// of them panicked (so the caller can re-raise instead of returning a half-written
/// buffer as if it were a result).
#[derive(Default)]
struct Done {
finished: usize,
panicked: bool,
}
#[derive(Default)]
struct Shared {
job: Option<JobPtr>,
/// Bumped once per dispatch; a worker runs each generation exactly once.
generation: u64,
/// Set at shutdown so parked workers wake and exit.
stop: bool,
}
/// **A pool of parked workers**, spawned once and reused for every call.
pub struct Pool {
workers: usize,
shared: Arc<(Mutex<Shared>, Condvar)>,
done: Arc<(Mutex<Done>, Condvar)>,
handles: Vec<std::thread::JoinHandle<()>>,
}
impl Pool {
/// Spawn `workers` parked threads (`0` => one per core). A pool of one spawns
/// nothing and runs every call inline on the caller.
#[must_use]
pub fn new(workers: usize) -> Self {
let workers = if workers == 0 {
crate::gatling_forkjoin::default_workers()
} else {
workers.max(1)
};
let shared: Arc<(Mutex<Shared>, Condvar)> = Arc::new((Mutex::new(Shared::default()), Condvar::new()));
let done: Arc<(Mutex<Done>, Condvar)> = Arc::new((Mutex::new(Done::default()), Condvar::new()));
let mut handles = Vec::new();
// One worker means the caller does the work itself — spawning a thread to hand
// the job straight back is the exact waste this module exists to remove. This is
// also the wasm arm: `available_parallelism` is unsupported there, so
// `default_workers()` is 1 and no thread is ever created.
if workers > 1 {
for w in 0..workers {
let shared = Arc::clone(&shared);
let done = Arc::clone(&done);
handles.push(std::thread::spawn(move || {
let mut seen: u64 = 0;
loop {
let job = {
let (lock, cv) = &*shared;
let mut g = lock.lock().unwrap();
while !g.stop && g.generation == seen {
g = cv.wait(g).unwrap();
}
if g.stop {
return;
}
seen = g.generation;
g.job
};
let mut panicked = false;
if let Some(j) = job {
// THE STATIC SPLIT: the worker derives its own contiguous
// range from `(w, workers)`. No shared counter, no atomic in
// the loop, no stealing.
//
// CAUGHT, and this is not defensive dressing: without it a
// panicking job never reaches the done-count below, the
// caller waits on the condvar forever, and the failure
// presents as a HANG rather than an error. That is exactly
// what happened the first time this module's own guards were
// driven red — the test suite hung instead of failing.
// `thread::scope` (which `gatling_forkjoin` uses) propagates
// panics on join for free; a parked pool has to do it itself.
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
// SAFETY: see `JobPtr` — the caller blocks until `done`
// reaches `workers`, so the closure outlives this call.
unsafe { (*j.0)(w, workers) }
}));
panicked = r.is_err();
}
let (lock, cv) = &*done;
let mut d = lock.lock().unwrap();
d.finished += 1;
d.panicked |= panicked;
drop(d);
cv.notify_all();
}
}));
}
}
Self { workers, shared, done, handles }
}
/// The number of workers actually parked (1 => everything runs inline).
#[must_use]
pub fn workers(&self) -> usize {
self.workers
}
/// **Run `f(worker_index, worker_count)` once on every parked worker**, then return.
///
/// Each worker computes its own contiguous slice of the problem from those two
/// numbers, so the caller does the splitting and this function does the waking. The
/// call does not return until every worker has finished.
pub fn for_each_worker<F>(&self, f: F)
where
F: Fn(usize, usize) + Send + Sync,
{
if self.workers <= 1 {
f(0, 1);
return;
}
// Erase the closure's lifetime through a RAW pointer. This is the step the
// safety note on `JobPtr` is about: the pointee is this stack frame's `f`, and
// the wait at the bottom of this function is what keeps it alive for every
// worker that dereferences it.
let dynref: &(dyn Fn(usize, usize) + Send + Sync + '_) = &f;
// SAFETY: a lifetime erasure on a fat pointer, nothing more — the vtable and
// data pointer are unchanged. Soundness comes from the wait below, not from here.
let ptr = JobPtr(unsafe {
std::mem::transmute::<
*const (dyn Fn(usize, usize) + Send + Sync + '_),
*const (dyn Fn(usize, usize) + Send + Sync),
>(dynref as *const _)
});
*self.done.0.lock().unwrap() = Done::default();
{
let (lock, cv) = &*self.shared;
let mut g = lock.lock().unwrap();
g.job = Some(ptr);
g.generation += 1;
cv.notify_all();
}
// NOT optional: this wait is the lifetime guarantee behind `JobPtr`.
let (lock, cv) = &*self.done;
let mut d = lock.lock().unwrap();
while d.finished < self.workers {
d = cv.wait(d).unwrap();
}
if d.panicked {
drop(d);
panic!("gatling_map: a pooled worker panicked — see the worker's own message above");
}
}
/// **Map a fixed-stride slice into an output slice, in parallel, in index order.**
///
/// The split is `n.div_ceil(workers)` contiguous items per worker — the only split a
/// fixed-stride array needs, because every index is already a valid boundary.
pub fn map_slice<A, B, F>(&self, src: &[A], dst: &mut [B], f: F)
where
A: Sync,
B: Send,
F: Fn(&A) -> B + Send + Sync,
{
assert_eq!(src.len(), dst.len(), "map_slice needs matching lengths");
let n = src.len();
if n == 0 {
return;
}
if self.workers <= 1 || n < 2 {
for (d, s) in dst.iter_mut().zip(src) {
*d = f(s);
}
return;
}
let chunk = n.div_ceil(self.workers);
// Disjoint mutable chunks, one per worker: `chunks_mut` proves non-aliasing to
// the compiler, so no unsafe is needed for the output at all.
let dst_cells: Vec<std::sync::Mutex<&mut [B]>> = dst.chunks_mut(chunk).map(std::sync::Mutex::new).collect();
let fr = &f;
self.for_each_worker(|w, _| {
if let Some(cell) = dst_cells.get(w) {
let mut out = cell.lock().unwrap();
let start = w * chunk;
let sl = &src[start..start + out.len()];
for (d, s) in out.iter_mut().zip(sl) {
*d = fr(s);
}
}
});
}
/// **Fill `dst[i]` with `f(i)`, in parallel, in index order.**
///
/// The index-aware sibling of [`map_slice`](Self::map_slice), for the common case
/// where the work needs the position and not just one input element — gathering from
/// two arrays at once, say, or reading a neighbour. Same static contiguous split, so
/// each worker owns a `[start, end)` and writes only its own slots.
pub fn map_indexed<B, F>(&self, dst: &mut [B], f: F)
where
B: Send,
F: Fn(usize) -> B + Send + Sync,
{
let n = dst.len();
if n == 0 {
return;
}
if self.workers <= 1 {
for (i, d) in dst.iter_mut().enumerate() {
*d = f(i);
}
return;
}
let chunk = n.div_ceil(self.workers);
// `chunks_mut` proves the output slices disjoint to the compiler, so the write
// side needs no unsafe at all.
let cells: Vec<std::sync::Mutex<&mut [B]>> = dst.chunks_mut(chunk).map(std::sync::Mutex::new).collect();
let fr = &f;
self.for_each_worker(|w, _| {
if let Some(cell) = cells.get(w) {
let mut out = cell.lock().unwrap();
let start = w * chunk;
for (k, d) in out.iter_mut().enumerate() {
*d = fr(start + k);
}
}
});
}
}
impl Drop for Pool {
fn drop(&mut self) {
{
let (lock, cv) = &*self.shared;
let mut g = lock.lock().unwrap();
g.stop = true;
cv.notify_all();
}
for h in self.handles.drain(..) {
let _ = h.join();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// **The pool must compute exactly what one core computes** — same values, same
/// index order — across worker counts and sizes that do NOT divide evenly, which is
/// where a contiguous split gets its off-by-one wrong (last chunk short, or a worker
/// with no work at all when `workers > n`).
#[test]
fn a_pooled_map_equals_the_serial_map() {
for &n in &[0usize, 1, 2, 7, 31, 32, 33, 1000, 4097] {
for &w in &[1usize, 2, 3, 8, 32, 64] {
let src: Vec<u64> = (0..n as u64).map(|i| i * 7 + 1).collect();
let want: Vec<u64> = src.iter().map(|v| v * v).collect();
let pool = Pool::new(w);
let mut got = vec![0u64; n];
pool.map_slice(&src, &mut got, |v| v * v);
assert_eq!(got, want, "n={n} workers={w} (pool reported {})", pool.workers());
}
}
}
/// Every index is visited **exactly once** — the property a contiguous split can
/// break silently by overlapping or skipping a chunk. A value check alone would not
/// catch a double-write of the same correct value.
#[test]
fn every_index_is_covered_exactly_once() {
let n = 5_000usize;
for &w in &[2usize, 7, 32] {
let pool = Pool::new(w);
let src: Vec<usize> = (0..n).collect();
let mut got = vec![0usize; n];
pool.map_slice(&src, &mut got, |i| *i + 1);
for (i, g) in got.iter().enumerate() {
assert_eq!(*g, i + 1, "index {i} not written exactly once (workers={w})");
}
}
}
/// **A panicking job must FAIL the caller, not hang it.** `gatling_forkjoin` gets
/// this free from `thread::scope`, which propagates on join; a parked pool has to
/// count the worker as finished even when its job unwound, or the caller waits on the
/// condvar forever. Found the hard way: driving the split guards above red made the
/// whole test binary HANG instead of fail, because an out-of-range slice panicked a
/// worker that then never reported done.
#[test]
fn a_panicking_job_raises_in_the_caller_instead_of_hanging() {
let pool = Pool::new(0);
let src: Vec<u64> = (0..1_000).collect();
let mut got = vec![0u64; src.len()];
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
pool.map_slice(&src, &mut got, |v| {
assert!(*v < 500, "deliberate job panic");
*v
});
}));
assert!(r.is_err(), "a panicking job must raise in the caller");
// And the pool is still USABLE afterwards — a failed dispatch must not poison the
// parked workers, or one bad frame would kill every frame after it.
let mut ok = vec![0u64; src.len()];
pool.map_slice(&src, &mut ok, |v| v + 1);
assert_eq!(ok[999], 1000, "the pool still serves work after a panicking dispatch");
}
/// `map_indexed` must agree with the serial `(0..n).map(f)`, at sizes and worker
/// counts that do not divide evenly — the index arithmetic is `w * chunk + k`, which
/// is exactly where a contiguous split gets its off-by-one.
#[test]
fn an_indexed_map_equals_the_serial_indexed_map() {
for &n in &[0usize, 1, 5, 31, 32, 33, 999, 4097] {
for &w in &[1usize, 3, 8, 32] {
let want: Vec<usize> = (0..n).map(|i| i * 3 + 1).collect();
let pool = Pool::new(w);
let mut got = vec![0usize; n];
pool.map_indexed(&mut got, |i| i * 3 + 1);
assert_eq!(got, want, "n={n} workers={w}");
}
}
}
/// The pool is REUSED: many dispatches through one pool, each correct. This is the
/// whole point — if generation/park bookkeeping were wrong, the second call would
/// hang or return the first call's results.
#[test]
fn the_same_pool_serves_many_dispatches() {
let pool = Pool::new(0);
let src: Vec<u64> = (0..10_000).collect();
let mut got = vec![0u64; src.len()];
for round in 1..=50u64 {
pool.map_slice(&src, &mut got, |v| v * round);
assert_eq!(got[9_999], 9_999 * round, "round {round} returned a stale generation");
assert_eq!(got[0], 0);
}
}
}