# Parallel data-processing flows in znippy-zoomies
<!-- nornir:class: reasoning -->
The parallel decoders (`lgz`, `lzip-parallel`, `lbzip2`, `ljar`) and the
streaming parsers (OSM XML → PBF, NDJSON) all live in `znippy-zoomies`, and they
all share one idea: a **no-barrier slot-pool stream** — fill a slot, find the cut
points, fan the segments out to N persistent workers with no per-chunk join,
reassemble in stream order. That idea is the **gatling** engine
([gatling.md](gatling.md)); some flows call the engine directly, others
hand-roll the same ring-slot pipeline. The interesting part is the **split**:
sometimes the boundaries are markers in the bitstream (a cheap forward scan);
sometimes there are none, and the split *itself* has to run in parallel —
speculative resync. Source: `znippy-zoomies/{lgz,lzip-parallel,lbzip2}/`,
`znippy-zoomies/src/gatling.rs`, `znippy-zoomies/src/json.rs`, and
`znippy-zoomies/osm-katana/src/{xml_to_pbf,xml_reader}.rs`. The shared inflate
core under all the gzip/zip/jar decoders is its own chapter:
[inflate.md](inflate.md).
## The engine they ride: gatling
`gatling` (`znippy-zoomies/src/gatling.rs`) is the substrate. A `Read` source is
cut into segments held in a bounded, zero-copy **slot ring**; **N worker
threads** transform each segment with **no cross-worker barrier**; an **ordered
collector** reassembles by `(chunk_id, seg_id)` and flushes strictly in stream
order. The full design — the two modes (`run` + `Codec` byte mode vs `run_typed`
+ `TypedCodec` typed mode), the `Config`/`SlotFill` knobs, the carry-headroom
mechanism — is in [gatling.md](gatling.md). The two facts that matter for every
flow below:
- **Zero-copy.** Inter-thread messages are a slot pointer + `(chunk_id, seg_id)`
(`WorkItem`, `gatling.rs:163`), never a `Vec<u8>` copy. The slot is held alive
in the collector's `InFlight` record (`gatling.rs:182`) until every segment of
the chunk is done.
- **Backpressure on pool exhaustion.** The reader allocates up to `ring_slots`
and then **blocks** on `slot_return_rx.recv()` (`gatling.rs:324`) until the
collector recycles one. A fast reader cannot outrun slow workers.
The decoder *binaries* (`lgz`, `lbzip2`) re-implement this exact ring inline —
same `RING_SLOTS` / `CARRY_HEADROOM` / `SLOT_SIZE` idiom (`lgz/src/bin/lgz.rs:42`)
— so they can fold gzip-header parsing and flush scanning straight into the
pipeline. The library decode entry points (`decompress_gz`, `decompress_zip`,
`lbzip2::parallel::decompress_parallel`) use scoped-thread / rayon `par_map`
over the discovered segments. The gatling *engine module itself* is consumed
today by osm-katana's XML→PBF and PBF paths (typed mode). Same shape everywhere;
the engine is the canonical expression of it.
## Where to cut: easy split vs hard split
Every parallel decoder's hard problem is the same: a compressed stream is one
long entropy-coded run, and you cannot start decoding in the middle unless you
know where an independently-decodable unit begins. There are two regimes.
**EASY — the boundaries are hints in the bitstream.** A single forward scan
finds them and the split is cheap:
- **gzip with flush markers** (pigz `-i`, BGZF): a sync-flush emits an empty
stored block — the byte pattern `00 00 FF FF` (LEN=0x0000, NLEN=0xFFFF).
`deflate_scan::find_next_flush` (`lgz/src/deflate_scan.rs:54`, SIMD AVX512 →
AVX2 → scalar) scans for it; `find_next_valid_flush` (`:187`) confirms each
candidate with a `probe_decode` to reject a coincidental pattern.
- **bzip2** — boundaries are *always* present. Every block starts with the
48-bit magic `0x314159265359` (the digits of π, `lbzip2/src/lib.rs:28`).
`block_scan::find_next_block` (`lbzip2/src/block_scan.rs:37`) searches for it
at the **bit** level (bzip2 blocks are not byte-aligned): a 64-bit big-endian
sliding window, 16 candidate bit-shifts checked per 2-byte step. So bzip2
never needs speculation — the magic is the boundary. `quick_verify_block`
(`:157`) reads ~73 header bits to drop false positives.
- **zip** — the container already lists every entry. `central_dir` parses the
End-Of-Central-Directory record at the tail (`EOCD_SIG = 0x06054b50`,
`lzip-parallel/src/central_dir.rs:12`) and `walk_cd_entries` (`:107`) yields
one `EntryLocation` per file with its absolute `local_header_offset`. **No
boundary search at all**, by design — and so **no speculative resync** in
`lzip-parallel`, deliberately.
In the easy regime each segment is independently decodable, so the split fans
straight out to the workers with no window stitching.
**HARD — no markers (a plain gzip stream).** This is where the split becomes a
parallel computation in its own right. `lgz`'s `speculative.rs` is a pugz-style
resync:
1. **Per-core forward search.** `split_boundaries_parallel`
(`lgz/src/speculative.rs:159`) launches N−1 scoped-thread tasks, each starting
from a nominal offset and forward-searching for a DEFLATE stored-block header
(`find_next_block` → `find_next_stored_block_end`, `:107`/`:75`) up to
`FORWARD_SEARCH_LIMIT = 64 * 1024` bytes (`:39`). **The split itself now runs
on N cores.**
2. **Probe-decode confirm.** A candidate offset is only accepted if
`probe_decode` (`:131`) actually decodes at least `PROBE_THRESHOLD = 4096`
bytes (`:32`) from it (up to `PROBE_LIMIT = 32768`, `:35`) via
`linflate::inflate_segment`. Garbage candidates are rejected and the scan
continues; cuts closer than the threshold apart are deduplicated.
3. **Two-pass window fixup.** Segments after the first have back-references that
point into the previous segment's output — across the cut. `speculative_decode`
(`:256`) handles this in two passes:
- **Pass 1 (parallel):** decode every segment with a **zeroed**
`LZ77_WINDOW = 32 * 1024` (`:43`) window. Back-refs over the cut resolve to
garbage, but the output *lengths* are correct.
- **Pass 2 (cheap, sequential):** for each segment, re-decode only the first
≤ 32 KiB with the **correct** window — the last 32 KiB of the previous
segment's real output (`decode_with_window_limited`, `:327`). Only ~32 KiB ×
N bytes are touched sequentially; the bulk stayed parallel.
### lgz's 3-tier fallback
`lgz`'s binary (`lgz/src/bin/lgz.rs`) tries strategies cheapest-first, falling
through only when the one above returns `None`:
1. **`decode_concatenated_members`** (`speculative.rs:407`) — multiple gzip
members concatenated (`1f 8b 08` headers). Each member is fully independent,
so this is embarrassingly parallel and needs no window stitching. Returns
`None` if only one member is found.
2. **`speculative_decode`** (`speculative.rs:256`) — the per-core resync above.
Works on *any* gzip stream; the universal-but-slower path.
3. **`flate2::read::GzDecoder`** — single-threaded, always correct. The
guaranteed floor when no parallel split is possible.
The library `decompress_gz` (`lgz/src/parallel.rs:96`) takes the flush-marker
fast path first (`chunk::decode_chunk`, only above `PARALLEL_THRESHOLD = 4 MB`,
`lib.rs:42`) and falls back to single-stream `flate2` when no flush boundary
exists.
## Parsing flows: XML through gatling, JSON through rayon
![parsing flows: OSM XML → PBF runs through gatling::run_typed — a Reader fills a 200 MiB slot with 32 MiB carry headroom over 6 slots, xml_split_typed finds the last complete element (find_safe_slot_end) and cuts N segments at top-level <node>/<way> starts (a .bz2 source produces bit-granular segment ranges instead), no-barrier workers decode+VTD-parse+PBF-encode each segment into a PbfSegment with left/right stubs, and the PbfTypedSink forwards blobs in strict stream order while stitching the element that straddles each cut from [prev_stub, left_stub] via flush_stub; separately, NDJSON record scanning uses rayon (NOT gatling): RFC 8259 makes every raw newline a record boundary so find_safe_split cuts on \n, scan_records_parallel scans each slice, and find_key probes fields zero-copy](assets/zoomies_parsing.svg)
### OSM XML → PBF — `gatling::run_typed` (typed mode)
The XML→PBF path (`osm-katana/src/xml_to_pbf.rs`) is the canonical gatling
*engine* consumer and the engine's original extraction site (it was lifted out of
`xml_to_pbf_bz2`). It uses **typed mode** because each worker produces a
fully-processed `PbfSegment`, not raw bytes needing reassembly:
- **split** — `xml_split_typed` (`osm-katana/src/xml_reader.rs:98`) returns
`Split<(usize, usize)>` (byte ranges). It calls `find_safe_slot_end`
(`znippy-zoomies/src/xml.rs:217`) to locate the last complete element (via
`memmem::rfind` on closing tags like `</way>` and `/>` for self-closing
elements), then divides the consumed region into N candidate segments, each
snapped forward to a top-level element start (`find_top_level_start`, `:191`).
The straddling tail past the last complete element becomes gatling's carry,
prepended to the next slot.
- **transform** — `encode_decoded_to_pbf` (`xml_to_pbf.rs:166`) VTD-parses the
middle and PBF-encodes it, returning a `PbfSegment` carrying the encoded blobs
plus a `left_stub` / `right_stub` (the partial elements at the segment edges).
- **sink** — `PbfTypedSink` (`xml_to_pbf.rs:497`) forwards each segment's blobs
in strict stream order, and `flush_stub` (`:521`) re-parses the element that
straddles a cut by concatenating `[prev_stub, left_stub]`. `finish` flushes the
trailing stub.
Source codecs differ only in the split:
- **Raw XML** — `RawPbfTypedCodec` (`xml_to_pbf.rs:395`), byte-range segments.
- **bzip2 XML** — `Bz2PbfTypedCodec` (`xml_to_pbf.rs:465`) maps
`lbzip2::chunk::split_chunk` block boundaries to **bit-granular** `(start_bit,
end_bit)` segments, so decode parallelizes across bzip2 block boundaries; the
decoded XML then flows through the same boundary-trimming.
- **gzip XML** — `gz_split` (`xml_to_pbf.rs:593`) uses lgz flush boundaries.
The same engine also drives the **PBF** path: `pbf_split` (`xml_to_pbf.rs:696`)
cuts at OSMData blob frames.
### NDJSON — rayon, not gatling
NDJSON record scanning (`znippy-zoomies/src/json.rs`) is **not** gatling-driven,
and that is the right call. RFC 8259 §7 makes an unescaped newline illegal inside
a JSON string, so **every raw `\n` is a record boundary** — there is no
escape-state machine needed to split, and no straddling unit to stitch.
`find_safe_split` (`json.rs:55`) cuts on the next `\n`; `scan_records_parallel`
(`json.rs:122`) fans the `\n`-bounded slices out with rayon `into_par_iter`;
`find_key` (`json.rs:266`) probes a record's fields with a zero-copy,
string-aware depth scan, never fully parsing.
The contrast is the point of this chapter: **gatling earns its keep when a
logical unit can straddle a chunk edge** (an XML element split across a 200 MiB
slot boundary → carry + stub stitching). NDJSON records are independent, uniform,
and self-delimiting — no carry, no stream-order reassembly, no stub — so plain
rayon over `\n`-cut slices is simpler and just as fast.
## Multi-core scoreboard: znippy-zoomies vs the field
The point of the whole chapter, in one table: **znippy-zoomies parallelizes every
format in every situation** — including the ones where the input gives no help.
The mainstream tools only light up all cores when the *producer* left them
markers (pigz `-i`, BGZF) or when the *container* lists the offsets (zip).
| situation | znippy-zoomies | pigz | igzip / libdeflate / zlib | unzip | lbzip2(C) / pbzip2 |
|---|---|---|---|---|---|
| **zip / jar** — many entries | `lzip-parallel` / `ljar` ✅ | — | — | ✗ single-thread | — |
| **gzip** — flush-marked (pigz `-i`, BGZF) | `lgz` ✅ | ✅ | ✗ single-stream | — | — |
| **gzip** — arbitrary, no markers | `lgz` ✅ *(speculative resync)* | ✗ falls back to 1 core | ✗ single-stream | — | — |
| **gzip** — concatenated members | `lgz` ✅ | ✗ serial | ✗ serial | — | — |
| **bzip2** — any stream | `lbzip2` ✅ | — | — | — | lbzip2(C) ✅ · pbzip2 ◐ own-output only |
✅ multi-core · ◐ only parallelizes its *own* output · ✗ single-thread · — format not handled
**Reading the matrix honestly** (so the claim survives scrutiny):
- The **unique** wins are the two markerless gzip rows — *arbitrary, no markers*
and *concatenated members*. No mainstream gzip tool parallelizes a hostile,
hint-free stream; `lgz` does it with the per-core speculative resync from
[§Where to cut](#where-to-cut-easy-split-vs-hard-split). This is the
differentiator.
- On **zip/jar** the win is over stock `unzip` (single-thread); the capability
itself (parallel across central-directory entries) is not exotic, but it is
ours by default.
- On **bzip2**, the C `lbzip2` is a genuine peer — it also finds block magic and
decodes every block in parallel, so it ticks the same box. `pbzip2` only
parallelizes files *it* compressed (multi-stream). We match the best-in-class
here rather than beating it.
- **igzip (ISA-L) is faster per core**, but on a single stream — it does not
multi-core a decode at all. That is the GB/s-vs-scaling distinction recorded in
the bencher (`xtask/examples/nornir-bench.rs`, the `Lgz` arm): igzip owns raw
single-stream throughput; znippy-zoomies owns *scaling the unscalable input*.
So the column to point at is `znippy-zoomies`: **every row ticked, the only one
with no gaps.**
## Benchmarks: the scoreboard, measured
The scoreboard above is a capability claim; these are the numbers behind it. They
are **not hand-written** — the tables are expanded from the nornir warehouse at
render time (the same `nornir:gen:benches` markers that feed the README; see
[benching.md](benching.md) for why there is exactly one bench source of truth).
Every figure is `MB/s`, single-shot on a real corpus (guava / scala jars,
recompressed per codec), `threadripper-3975wx · 32 cores`. Rows ending `_st` are
the **single-threaded** baseline (one worker) — the gap between a workload and
its `_st` twin *is* the multi-core scaling.
**Parallel JAR decode — `ljar` vs `unzip -p`**
<!-- nornir:gen:start:benches name=ljar -->
**v0.9.6 · oden · 32 cores · 2026-07-22**
| workload | ljar_mbs | speedup_x | unzip_mbs |
|---|---|---|---|
| ljar_guava | **1063** | 18.54 | 57.36 |
| ljar_scala | **1357** | 22.93 | 59.19 |
<!-- nornir:gen:end:benches -->
**Parallel ZIP decode — `lzip-parallel` vs `unzip -p`**
<!-- nornir:gen:start:benches name=lzip -->
**v0.9.6 · oden · 32 cores · 2026-07-22**
| workload | lzip_mbs | speedup_x | unzip_mbs |
|---|---|---|---|
| lzip_guava | **1266** | 22.95 | 55.13 |
| lzip_scala | **1747** | 31.19 | 56 |
<!-- nornir:gen:end:benches -->
**Parallel bzip2 decode — `lbzip2` vs C `lbzip2` (1995-lineage bzip2)**
<!-- nornir:gen:start:benches name=lbzip2 -->
**v0.9.6 · oden · 32 cores · 2026-07-22**
| workload | lbzip2_c_mbs | lbzip2_mbs | speedup_x |
|---|---|---|---|
| lbzip2_guava | **74.70** | 60.78 | 0.81 |
| lbzip2_scala | **134.73** | 86.07 | 0.64 |
<!-- nornir:gen:end:benches -->
**Parallel gzip decode — `lgz` vs `pigz -dc`**
<!-- nornir:gen:start:benches name=lgz -->
**v0.9.6 · oden · 32 cores · 2026-07-22**
| workload | lgz_mbs | pigz_mbs | speedup_x |
|---|---|---|---|
| lgz_guava | **334.32** | 274.19 | 1.22 |
| lgz_scala | **314.08** | 293.68 | 1.07 |
<!-- nornir:gen:end:benches -->
Reading them against the scoreboard: `ljar`/`lzip` show the ~20–30× win over
single-thread `unzip`; `lgz` beats `pigz` even single-threaded (`_st`) and scales
from there; `lbzip2` edges the C `lbzip2` it matches on capability. The one number
that *isn't* here is igzip — it has no parallel-decode row to report, which is
exactly the point made at the `Lgz` arm of the bencher.
## Map
| flow | crate | split signal | speculative? | parallelism |
|---|---|---|---|---|
| gzip | `lgz` | `00 00 FF FF` flush marker, else resync | **yes** (no markers) | ring-slot / scoped-thread |
| zip | `lzip-parallel` | central directory (every offset listed) | no (by design) | rayon pool |
| bzip2 | `lbzip2` | 48-bit block magic `0x314159265359` at any bit | no (magic suffices) | ring-slot / scoped-thread |
| OSM XML → PBF | `osm-katana` | last complete element / top-level start | n/a | **gatling `run_typed`** |
| NDJSON | `znippy-zoomies` | raw `\n` (RFC 8259) | n/a | rayon |