# Gatling — practical guide
> **Class:** `guide` · **Function:** `gatling`.
The how-to. The chosen design is in [gatling-design.md](gatling-design.md), the
rationale in [gatling-reasoning.md](gatling-reasoning.md), and the plain-words intro
in [gatling-for-idiots.md](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`](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
```rust
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.
```rust
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
```rust
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)
```rust
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).
```rust
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:
```rust
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)
```rust
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.
```rust
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
```rust
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
```rust
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.
```rust
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`
```rust
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`
```rust
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`
```rust
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`
```rust
// 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.
```rust
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.
```rust
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.
```rust
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
```rust
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`, `pread`s 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
```rust
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`
```rust
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 `pwrite`s `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`](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.