nordisk/znippy-zoomies

Gatling — practical guide

Class: guide · Function: gatling.

The how-to. The chosen design is in gatling-design.md, the rationale in gatling-reasoning.md, and the plain-words intro in gatling-for-idiots.md. All of these live in znippy-zoomies (use znippy_zoomies::gatling; / gatling_forkjoin;).

Rule zero: no rayon. Every gatling engine below is std::thread::scope (sync) or tokio (async). The reader only reads, the collector only stitches in order, all heavy work happens in the workers.

Rule zero holds for the whole workspace, not just the root engines. No crate here declares a rayon dependency, and none stands up a private thread pool. The two root scanners that used to reach for it — src/json.rs (NDJSON line-scan) and src/vtd.rs (sub-chunk parse) — now map across gatling_forkjoin::gatling_for_each (on the json_scan bench, no slower than the rayon version they replaced), and the codec crates ljar / lzip-parallel were ported off their OnceLock<rayon::ThreadPool> singletons onto the same shared engine (lgz / lbzip2 folded their private scoped-thread clones onto it too). The guard test tests/rayon_free_law.rs now walks every workspace member's src/ + Cargo.toml (whole_workspace_is_rayon_free) and bans private pools (rayon::ThreadPool and the pool builders), not just root src/. (rayon still appears once transitively, as a dev-dependency of criterion for the benches; it never reaches a shipped build — cargo tree -e no-dev -i rayon prints nothing.) The history of how the law got holed and then fixed is in gatling-reasoning.md.

Picking a variant in 10 seconds

Is the job "decode N independent units and WRITE them" (archive extract, concat stream),
sizes known up front, ordering droppable? ── yes ─→ parwrite (each worker writes its own
                │                                    output; no single writer thread)
                │ no
                ▼
Is the work I/O-bound (network/PUT/GET/fsync)? ── yes ─→ gatling::io::run / run_ordered
                │ no (CPU-bound)
                ▼
Do I have the whole work set already (a Vec / file list / grid)?
   │ no — bytes stream in from a Read           │ yes
   ▼                                            ▼
 gatling::run        (reassemble bytes)     Is it a 2D row/pixel grid?
 gatling::run_typed  (finished records)        │ yes ─→ gatling_forkjoin::gatling_scanlines
                                               │ no
                                               ▼
                              Are unit costs heavily skewed?
                                 │ yes ─→ is the work an owned Vec of tags
                                 │        I want to map-then-group-by-key?
                                 │          │ yes ─→ gatling_map_balanced (item-ref in, Vec<R> out)
                                 │          │ no  ─→ gatling_for_each_balanced (0..n, LPT + batch)
                                 │ no
                                 ▼
                        Do I need the results back?
                           │ yes ─→ gatling_for_each
                           │ no  ─→ gatling_run

Fork-join (work known upfront) — gatling_forkjoin

The everyday family. You hold a Vec/range/grid; map across it on all cores.

gatling_for_each — fixed 0..n, results in index order

pub fn gatling_for_each<T, F>(n: usize, n_workers: usize, f: F) -> Vec<T>
where T: Send, F: Fn(usize) -> T + Sync;

n_workers == 0 ⇒ one worker per core. Each worker claims the next index via the shared atomic and runs without waiting on any sibling; results come back in 0..n order regardless of which worker finished when.

use znippy_zoomies::gatling_forkjoin::gatling_for_each;
let doubled = gatling_for_each(1000, 0, |i| i as u32 * 2);
assert_eq!(doubled[500], 1000);

gatling_run — fixed 0..n, side-effects only

pub fn gatling_run<F>(n: usize, n_workers: usize, f: F)
where F: Fn(usize) + Sync;

No collected output — use when each worker writes into per-i external state you own (an array of atomics, a pre-sized slice indexed by i, etc.).

gatling_for_each_balanced — skewed costs (LPT + batched)

pub fn gatling_for_each_balanced<T, F, W>(
    n: usize,
    n_workers: usize,
    batch: usize,           // indices claimed per atomic fetch_add; 0 ⇒ 1
    weight: W,              // weight(i): bigger = scheduled earlier (LPT)
    f: F,
) -> Vec<T>
where T: Send, F: Fn(usize) -> T + Sync, W: Fn(usize) -> u64;

Same self-dispatch + index-ordered output as gatling_for_each, plus: units are scheduled heaviest-first by weight(i), and workers steal batch at a time. Reach for this when a few units dwarf the rest (one huge file among many tiny ones).

use znippy_zoomies::gatling_forkjoin::gatling_for_each_balanced;
// One expensive unit (index 0) among many cheap ones; still index-ordered out.
let out = gatling_for_each_balanced(
    1000, 0, 16,
    |i| if i == 0 { 1_000_000 } else { 1 },   // weight = cost estimate
    |i| i * 2,
);
assert_eq!(out[500], 1000);

Real-world: parse every .rs of a repo, scheduling by file byte size so a 5 000-line generated file is claimed first and idle workers mop up the small modules:

let scans = gatling_for_each_balanced(
    files.len(), 0, 8,
    |i| files[i].size_bytes,                  // heaviest-first
    |i| parse_one(&files[i]),                 // syn::parse_file, the CPU cost
);

gatling_map_balanced — owned work list in, owned results out (LPT + batched)

pub fn gatling_map_balanced<T, R, S, F>(
    items: &[T],
    n_workers: usize,
    batch: usize,           // items claimed per atomic fetch_add; 0 ⇒ 1
    size_of: S,             // size_of(&item): bigger = scheduled earlier (LPT)
    f: F,                   // f(index, &item) -> R
) -> Vec<R>
where T: Sync, R: Send, S: Fn(&T) -> u64, F: Fn(usize, &T) -> R + Sync;

gatling_for_each_balanced's sibling for when the work set isn't a bare 0..n but a Vec of heterogeneous tags you want to map then group. Same heaviest-first LPT schedule + batched self-dispatch + original-order output, but the closure gets the item by reference (f(i, &item)) and the weight reads the item itself (size_of(&item)) — so the LPT key and the work both come off one borrow, no re-indexing an external slice inside the hot loop. The slice is borrowed for the scope (T: Sync); only the tag is read up front — a file-parsing f reads the file inside the worker, so item bytes are never copied into the pool (zero-copy distribution). Results land in their own disjoint output slots (no per-worker Vec

• sort), in original items order, ready to fold.

use znippy_zoomies::gatling_forkjoin::gatling_map_balanced;
// Flatten every (member, file) into one tag list, parse all on all cores scheduled
// by byte size, then group the owned per-file scans back per member.
let parsed = gatling_map_balanced(
    &union, 0, 8,
    |tag| tag.size_bytes,                         // heaviest-first
    |_, tag| (tag.member_idx, parse_one(tag)),    // owned (key, result)
);
for (member_idx, scan) in parsed {               // original order ⇒ deterministic
    out[member_idx].merge(scan);
}

Reach for it over gatling_for_each_balanced when you hold the work as owned items (not a count) and want owned per-item results to group-by-key afterwards.

gatling_scanlines — a 2D buffer, one disjoint row per worker

pub fn gatling_scanlines<P, F>(
    buf: &mut [P], rows: usize, stride: usize,
    n_workers: usize, min_rows_for_threads: usize, f: F,
) where P: Send, F: Fn(usize, &mut [P]) + Sync;

buf is rows * stride; row y owns buf[y*stride .. (y+1)*stride]. Workers never touch the same row, so no lock and no false sharing. Below min_rows_for_threads it stays single-threaded (no pool spin-up for a tooltip-sized frame).

gatling_map_owned — owned items consumed, results in input order

pub fn gatling_map_owned<T, R, F>(items: Vec<T>, f: F) -> Vec<R>
where T: Send, R: Send, F: Fn(T) -> R + Sync;

The owned-item sibling of gatling_for_each. Where gatling_for_each calls Fn(usize) -> T (the worker fetches its own input), this consumes a Vec<T>: each T is moved by value into exactly one worker. Reach for it when the per-item payload is owned and possibly !Sync (a handle that can't sit behind a shared Fn(usize)) — it's built on the caller thread, then handed off to a single worker.

• Ownership. Each input element is moved out exactly once (a per-index atomic cursor claims each slot once; disjoint UnsafeCell<MaybeUninit<_>> input/output slots, so the move-out and the result write never alias — no lock).
• Ordering. Results come back in input order, identical to a serial items.into_iter().map(f).collect().
• Sizing. Worker count is available_parallelism() capped at items.len(); empty input returns Vec::new(), and ≤ 1 worker just runs the serial map.

use znippy_zoomies::gatling_forkjoin::gatling_map_owned;
let items: Vec<u64> = (0..1000).collect();
let out = gatling_map_owned(items, |v| v * 2);   // items consumed
assert_eq!(out[500], 1000);

Streaming (work arrives a chunk at a time) — gatling

You have a Read stream to carve up. Plug in a codec + sink.

Byte mode — run + Codec / Sink

pub fn run<C: Codec, S: Sink>(
    reader: impl Read + Send, codec: C, sink: &mut S,
    n_workers: usize, cfg: Config,
) -> Result<()>;

pub trait Codec: Sync {
    type Seg: Send + Copy + 'static;
    fn split(&self, data: &[u8], n_workers: usize, is_last: bool) -> Option<Split<Self::Seg>>;
    fn decode(&self, data: &[u8], seg: &Self::Seg) -> Vec<u8>;
}
pub trait Sink: Send {
    fn safe_end(&self, assembled: &[u8], is_last: bool) -> usize; // largest safe prefix
    fn process(&mut self, bytes: &[u8]) -> Result<()>;            // one boundary-aligned run
}

Workers decode segments → bytes; the collector concatenates and cuts at safe_end (e.g. end of an XML element) before calling process. Use for bz2/gz → XML where the output must be reassembled across segment boundaries.

Typed mode — run_typed + TypedCodec / TypedSink

pub fn run_typed<C: TypedCodec, S: TypedSink<C::Output>>(
    reader: impl Read + Send, codec: C, sink: &mut S,
    n_workers: usize, cfg: Config,
) -> Result<()>;

pub trait TypedCodec: Sync {
    type Seg: Send + Copy + 'static;
    type Output: Send + 'static;
    fn split(&self, data: &[u8], n_workers: usize, is_last: bool) -> Option<Split<Self::Seg>>;
    fn transform(&self, data: &[u8], seg: &Self::Seg) -> Self::Output;
    fn finish_worker(&self) -> Option<Self::Output> { None } // flush per-worker tail state
}
pub trait TypedSink<T>: Send {
    fn process(&mut self, out: T, is_last: bool) -> Result<()>; // each output, in stream order
    fn finish(&mut self) -> Result<()> { Ok(()) }               // called once after all segments
}

split must cut logically complete segments (no record spanning a boundary); the collector then forwards each transform output individually, in order — no reassembly. Use for parse-style transforms (nornir deep_scan: one stage decompresses crates, the next syn-parses each file, both via run_typed).

Config

pub struct Config {
    pub chunk_size: usize,        // compressed bytes read per slot
    pub carry_headroom: usize,    // head room for prepended carry (slot = headroom + chunk_size)
    pub ring_slots: usize,        // slots in flight = steady-state pipeline width
    pub initial_carry: Vec<u8>,   // bytes prepended to the first chunk (e.g. a header already read)
    pub slot_fill: SlotFill,      // Incremental (default) | Big
}

• SlotFill::Incremental (default) only faults the bytes actually read — best for small / single-chunk / partial-tail inputs.
• SlotFill::Big faults the full slot once and reuses it — best for planet-scale inputs where nearly every chunk is full.

Async I/O fan-out — gatling::io

// fastest-first:
pub async fn run<...>(jobs, max_in_flight, on_result) -> Result<...>;
// submission order:
pub async fn run_ordered<...>(jobs, max_in_flight, map) -> Result<Vec<...>>;

Drive an iterator of async jobs with ≤ max_in_flight running at once (admission control + backpressure, like buffer_unordered with a cap). Payloads are bytes::Bytes — moved into the job and out as a result, never deep-copied. The first error short-circuits. Use for S3 PUT/GET fan-out, batched fsync, network calls — latency-bound work where CPU threads would just block on sockets.

use znippy_zoomies::gatling::io;
use bytes::Bytes;
let jobs = (0..100u64).map(|seq| async move { upload(seq, Bytes::from(vec![0u8; 1024])).await });
let descriptors = io::run_ordered(jobs, 16, |d| d).await?; // ≤16 PUTs in flight, in order

Ordered streaming sink — gatling::ordered (gatling_ordered)

The item-oriented sibling of the byte streaming engine, in src/gatling_ordered.rs (the gatling::ordered module). Where gatling::run owns the reader and splits one byte stream, this does no reading and no splitting: the caller stays the producer, yielding discrete labelled items (Label, Input) (e.g. znippy's io_uring batched small-file reader handing over (path, bytes)). N workers run a caller map, and outputs stream to a caller OrderedSink in producer order through a bounded reorder buffer.

pub trait OrderedSink<O> {
    fn emit(&mut self, seq: u64, output: O) -> Result<()>; // called once per item, in order
}
// Blanket impl for FnMut(u64, O) -> Result<()>, so a closure is a sink.

pub fn run_ordered_sink<L, I, O, P, M, S>(
    producer: P,          // FnMut() -> Option<(L, I)>: pulled lazily, None = end of stream
    n_workers: usize,     // map workers; 0 ⇒ 1
    cap: usize,           // in-flight / reorder-buffer bound; 0 ⇒ max(2 * n_workers, 1)
    map: M,               // Fn(L, I) -> O + Sync, run on the workers
    sink: &mut S,         // OrderedSink<O>, called in strict producer order
) -> Result<()>
where
    L: Send, I: Send, O: Send,
    P: FnMut() -> Option<(L, I)> + Send,
    M: Fn(L, I) -> O + Sync,
    S: OrderedSink<O>;

• Ordering guarantee. sink.emit is called exactly once per item in strict producer sequence (seq = 0, 1, 2, …) as each in-order output becomes ready — never the whole output set at once, so the sink can pwrite to an archive and record offsets/metadata incrementally. Determinism: same producer sequence + same (pure) map ⇒ identical ordered emit sequence for any n_workers.
• Ownership model. The producer is pulled by value ((L, I) moved in); the map consumes (L, I) and produces an owned O handed to the sink by value. A single counting-permit pool (primed with cap tokens) bounds work: the dispatch thread must hold a permit to pull the producer and the collector returns one after each in-order emit, so at every instant items_pulled − items_emitted ≤ cap. Memory is bounded by cap, never by stream length; a slow item near the front back-pressures the producer.
• Threading. No barrier / self-dispatch: workers pull the next item the instant they finish the last (LPT). The collector runs on the calling thread (so the sink need not be Send); the dispatch loop and the n_workers map workers run on std::thread::scope threads joined before return. A sink returning Err, or a worker panicking, aborts the run and propagates out.

use znippy_zoomies::gatling::ordered::run_ordered_sink;
let mut next = 0usize;
let producer = move || if next < 100 { let i = next; next += 1; Some((i, vec![i as u8; 4])) }
                       else { None };
let mut out: Vec<(u64, usize)> = Vec::new();
run_ordered_sink(
    producer, /* workers */ 8, /* cap */ 16,
    |label: usize, bytes: Vec<u8>| (label, bytes.len()),
    &mut |seq, (label, _len)| { out.push((seq, label)); Ok(()) },
)?;

Parallel writers — remove the single writer (parwrite)

gatling::run and the fork-join family funnel decoded output through one collector + one writer thread. When the output units are independent (archive entries) or pre-offset regions of one stream (concatenated members), that lone writer is the last un-parallelised stage — the serial drain that left the DEFLATE codecs writer-bound (lgz ~3, ljar ~3.5, lzip ~4 of 12 cores). parwrite (module gatling::parwrite, landed on master) removes it: the same no-barrier atomic-cursor pool, but each worker writes its own output — no collector, no writer thread. Reach for it when output sizes are known up front and ordering can be dropped. Shipped on the DEFLATE codecs this lifts cores_busy to ljar/lzip ~11.5/12 and lgz ~7.8/12 (offset-pwrite on -o files), with lbz2 held at ~10.4.

extract_entries_unordered — independent files, each worker writes its own

pub fn extract_entries_unordered<D>(
    input: &File, jobs: &[EntryJob], out_dir: &Path, n_workers: usize, decode: D,
) -> WriteStats
where D: Fn(&[u8], usize, &mut Vec<u8>) -> Result<(), String> + Sync;

pub struct EntryJob { pub data_offset: u64, pub comp_len: usize, pub out_size: usize,
                      pub is_store: bool, pub name: String }

Each worker claims the next EntryJob, preads its comp_len compressed bytes at data_offset from the shared input (thread-safe positional read — no shared file offset, so N workers read disjoint regions at once), decodes into a reused output buffer (is_store entries are copied verbatim), and writes out_dir.join(name). Entries are independent so file writes never conflict; no ordering. Returns WriteStats { ok, failed } — a unit that fails to read/decode/write is logged to stderr and skipped, its siblings keep running. This is the ZIP/JAR extract shape.

write_segments_positional — decode into one pre-sized stream at offsets

pub fn write_segments_positional<D>(
    out: &File, n_segments: usize, offsets: &[u64], n_workers: usize, decode: D,
) -> WriteStats
where D: Fn(usize, &mut Vec<u8>) -> Result<(), String> + Sync;

The caller has already laid out the stream: offsets[i] is where segment i's decoded bytes belong (a prefix-sum of the segments' known output sizes) and out is pre-sized (set_len). Each worker claims the next segment, decode(i, &mut buf) fills its reused buffer, and write_all_at (pwrite) writes it at offsets[i]. Positional writes to disjoint regions preserve exact stream byte-order with no ordered collector. decode should verify the decoded length matches the assumed size (return Err on mismatch); the caller can detect failed > 0 and fall back to a safe path.

write_slices_positional — ready slices, zero-copy parallel pwrite

pub fn write_slices_positional(
    out: &File, slices: &[&[u8]], offsets: &[u64], n_workers: usize,
) -> WriteStats;

The two-phase sibling: use it when the segments are already materialised (e.g. members decoded by a prior gatling_for_each pass) and their sizes — hence offsets — are known. out must be pre-sized. Each worker pwrites slices[i] at offsets[i]; the slices are never copied through the pool.

Limits (honest). pwrite needs a seekable file — stdout cannot be offset-written, so a piped lgz input.gz | … falls back to the ordered writer. A single-stream format with a serial boundary-scan tail (lgz) only partly lifts (measured 7.8/12 on -o files); independent-entry formats (ljar/lzip) reach ~11.5/12. See gatling-reasoning.md for the full writer story and numbers.

Common mistakes (each one re-serializes the pool)

• Heavy work on the reader — finding every segment boundary up front, reading the whole input before starting. Hand the workers a rough range and let each find its own start.
• Heavy work on the collector — building/sorting the final result there instead of just stitching in order. Workers build and pack their own output.
• A serial for "just for now" over the work set — that's the single-core path. Use gatling_for_each[_balanced].
• Plain for_each on skewed costs — the monster unit becomes a long-pole. Use gatling_for_each_balanced with a weight.
• Reaching for rayon — don't: its global pool is exactly what bit osm-katana. Use gatling_forkjoin::gatling_for_each for a fixed 0..n of units (the shape json.rs and vtd.rs use for their sub-chunk scans), or one of the streaming engines above. The crate carries no rayon dependency; keep it that way.