The corgi backend, on more than one worker

[frankmcsherry]

The corgi backend's arrange used Pipeline and asserted peers() == 1, because a multi-worker run would have MIS-PLACED keys — silently wrong, not slow. This replaces that assert with the exchange it was waiting for, and the backend now renders on any number of workers, in any number of processes.

The distributor

arrange is the only exchange point in the Backend trait, so substituting a pact for Pipeline is the whole of multi-worker support: every other operator (linear, join, reduce, as_collection) is key-local once arrangements are placed correctly, and both sides of a join agree on placement because they route by the same function of the key.

Timely's stock DrainContainerDistributor is unusable here. A CorgiContainer has no items to drain — it is four columns — and taking it apart row-wise would undo the representation before the data left the worker. CorgiDistributor partitions the way a columnar engine does:

1. hash the key column once, columnar, with corgi::hash — the same structural hash present_key prepends as a key's identifier lane, so the distributor and the arrangement agree on what a key's identifier is;
2. counting-sort row indices by destination (mask when peers is a power of two, modulus otherwise);
3. gather each destination's rows into fresh contiguous columns — which is exactly what the wire format wants to write, one memcpy per leaf.

Stable, so rows keep their order within a destination. A container whose rows all share a destination moves whole, with no gather at all. The index buffers are fields, not locals, so a batch costs no allocation beyond the hash column itself.

The wire format

corgi/bytes.rs is ContainerBytes for CorgiContainer: four length words, then the key and value columns through corgi::bytes (landed as frankmcsherry/WIP#13) and the time and diff columns through columnar's Stash — the encoder T: Columnar was already chosen for. Every section is a whole number of words, so a receiver can install the bytes rather than relocate them.

The contrast is the point: the row backend's Vec<((Row, Row), T, R)> reaches the wire through bincode, walking every Value of every row and emitting varints. This walks four columns and emits memcpys.

The container decoder checks the four columns agree with the time column's length. corgi::bytes guarantees a structurally sound Value but not a small one — the payload-free constructors declare rows without spending bytes, so a Unit names a trillion rows in sixteen bytes and nothing inside corgi can call that wrong. Here it is wrong, and this is the layer that knows.

A latent bug the exchange made reachable

A container with no rows has no shape either: every path that rebuilds one from rows infers its columns by scanning them, so from_updates on an empty vector returns the shape-erased default. Hand that to the next op and compile_projection lowers $1[0] against a Unit, because a Field is legal in the abstract; eval_graph then panics with "Field: expected a product".

Reaching it takes two steps, which is why it survived: an op that empties a batch while KEEPING its shape (the columnar filter path — gather of an empty index list is still a typed zero-row column), and then an op whose term the lowering DECLINES, taking the row-wise path through from_updates. One worker rarely empties a batch it was given; the key-hash exchange makes it routine.

Fixed by skipping ops once a container is empty, at the top of the loop rather than before it — a chain empties itself, a filter dropping every row and handing the erased container to the very next op in the same slice. That is the whole answer rather than a shortcut: every LinearOp maps zero rows to zero rows, so the only thing the ops could contribute is the output shape, and no consumer reads an empty container's shape anyway.

Testing

The gate. tests/corgi_backend.rs runs each of its programs at 1, 2, 3 and 4 workers — 3 deliberately, so the modulus path runs rather than the mask and no input divides evenly — and again over ProcessBinary channels, which serializes every exchanged container. That last one is the multi-process path without the processes, so the wire format is exercised in-tree rather than only in a cluster. evaluate_with_workers / evaluate_with_config are the entry points that makes possible.

The AoC suite. The 33 parts added in #847 have known answers, so they were run through the corgi backend at 1, 3 and 4 workers. This is what found the bug above: day03 part2 answered correctly at one worker and panicked at four. With the fix, all 33 give identical results at every worker count. Four still disagree with the oracle — day01 p1/p2 and day05 p1/p2 — but those are the pre-existing ones that suite's README already records, and they do not vary with worker count. tests/programs/empty_batch.ddp reproduces the failure in the gate, built from the failing day03 export rather than guessed at: two earlier attempts passed with the fix reverted, because the filters I reached for compiled columnar and kept their shape.

Real processes. scc at n=20k with 200 rounds of incremental insertions and retractions gives the identical net count (25298) for vec at 1/2/4 workers, corgi at 1/2/4 workers, and corgi across two OS processes over TCP.

Unit tests cover the partition (it is a partition; a key goes to one destination; routing does not depend on the batch, which is what makes a join correct; strided keys still spread, which routing by the raw key would not) and the container codec (round trips per shape family, truncation, and that a key column costs its payload).

Performance

M4, 10 cores. Load time, median of 3, rounds = 0.

program	backend	-w1	-w2	-w4
scc n=100k e=200k	corgi	2.01 s	1.12 s	716 ms
scc n=100k e=200k	vec	2.70 s	1.49 s	876 ms
reach n=200k e=400k	corgi	279 ms	153 ms	99 ms
reach n=200k e=400k	vec	357 ms	195 ms	109 ms

2.8× at four workers on both programs, and corgi stays ahead of vec at every worker count. Both backends regress past four workers, which is this machine's P-core/E-core boundary rather than anything about the exchange.

Serialization costs +6.3%. Same program, four workers either way: 715 ms as one process with typed channels, 760 ms as two OS processes with the wire format plus TCP. vec pays +9.0% for bincode over the same step. corgi paying full serialization beats vec paying none.

The exchange itself is 1.2% — everything under Exchange::push in a samply profile of scc at -w4 is 35 of 2983 samples.

Why 2.8× and not 4×, since that is the obvious next question. Total CPU rises 1.54× from one worker to four (1931 → 2983 samples), and 4 / 1.54 = 2.6. Of the 1052 added samples: step_or_park self time is 0 at one worker and 362 at four — timely's scheduler, a third of the loss — then ProxyReduceTactic::retire +178, MergeBatcher::seal +122, SmallVec::try_grow +91, which are per-batch fixed costs multiplying by worker count as each worker seals and retires and merges its own quarter. The exchange is +35. The lever for better scaling is batch granularity, not the shuffle.

Two ideas measured and set aside

Ship each destination pre-sorted by key hash. The receiver's chunker sorts anyway, and has to — it is merging W senders' blocks, not one. Sorting at the sender adds a sort there and leaves the receiver's merge in place. The profile puts the receiver's whole ingest sort at 3.3%; there is no 20% sitting there.

Carry the hash lane across the exchange so the receiver's present_key need not recompute it. Only bites compound keys — present_key uses a primitive-integer key as it stands and hashes nothing, and scc, reach and stable are all scalar-keyed after lowering. It would need a flag on the container whose invariant is held by the dataflow's topology rather than by the type, which is fragile for a slice of an already-1.2% budget.

Known cost

length_in_bytes and into_bytes are separate calls on &self, and the time and diff columns must be built to be measured, so they are built twice — two linear passes over times, on the multi-process path only. The fix is the one container.rs already names: hold times columnar in the container instead of as Vec<T>. Keys and values do not have this problem, since sizing a Value walks its shape without touching a payload byte.

🤖 Generated with Claude Code