pipeline: throughput starves at low array count (parallelism limited by #arrays, not pool size)

[nclack]

Symptom

damacy throughput (and the decode stage's output rate) collapses when a scenario
has only a few distinct arrays, even though there are many chunks/files available
and the IO pool is large. The pipeline appears unable to extract enough
independent work from a small number of arrays to keep IO + GPU decode saturated.

Evidence

From a cold-cache survey on the fast /mnt/main0 mount (L40), comparing two
real scenarios with the same chunk geometry (32 MiB uint16 decode chunks) but
very different array counts:

arrays	throughput	decode stage GB/s_out	decode load
512	1669 samp/s	51.3 GB/s	99%
2	38 samp/s	3.2 GB/s	100%

Same GPU, same chunk type — the decode stage shows ~100% wall occupancy in both,
but its output rate is ~16x lower with 2 arrays. "Decode-bound" at 3.2 GB/s
means the decode stage is occupied yet starved/serialized, not saturated.

The cleanest single-variable evidence: on the same 2-array scenario, with the
same patches (a CPU tensorstore benchmark was made to reproduce damacy's
xorshift sampling bit-for-bit), a 32-thread CPU reader gets 203 samp/s vs
damacy's 38 — 5.3x faster. tensorstore happily issues 32 concurrent reads
into 2 arrays; damacy does not keep that many independent operations in flight.
So this is a damacy pipeline-parallelism limit, not a property of the data or
chunking. (The 2-array dataset has ~576 distinct chunk files, so per-file
parallelism is available — it just isn't being used.)

Hypothesis (needs confirming)

The read-ahead / decode parallelism keys off the number of distinct arrays (or
shards) present in the lookahead window. With few arrays, in-flight depth can't
grow regardless of n_io_threads / metadata_io_concurrency, so IO and GPU
decode idle behind a serialized per-array dependency. Worth checking against the
counters (metadata_backend_read_max_active, in-flight read depth, distinct
shards per wave) whether depth tracks array count rather than the configured
pool size.

Suggested repro / investigation

The synthetic scenario machinery (bench/scenario.py + gen_dataset.py) can
isolate this cleanly: hold chunk_shape/shard_shape/dtype/codec and the patch
fixed, and sweep n_zarrs (e.g. 1, 2, 4, 16, 64, 512). If throughput and decode
GB/s_out scale with n_zarrs and plateau, that confirms the pipeline derives its
parallelism from array count rather than from available chunks/files. From there:
can the planner/dispatcher interleave more in-flight reads/decodes within a single
array (across its chunks/shards) so that a 1-2 array dataset still saturates the
IO pool and the GPU decode stream?

Why it matters

Several training datasets have very few arrays (e.g. timelapse sets with 2-11
large arrays). On those, damacy currently loses to a CPU reader despite GPU
decode, purely from this starvation — so it's worth fixing independently of the
data-format and backend decisions.

[nclack]

Root-cause read (from the source, not yet measured)

The parallelism unit is the post-coalesce read_op, and its concurrency is
effectively bounded by the number of distinct shard files in a wave — which
correlates directly with array count. The mechanism is in coalesce_chunks
(src/planner/coalesce.c), and it's two compounding effects:

1. Fusion collapses op count (coalesce.c:47-82): contiguous reads within one
   shard_path fuse into a single read_op, capped at max_read_op_bytes = 4 MiB
   (damacy_config.c:185). One read_op = one io job = one worker = one pread.
   With few arrays, a batch's samples cluster into few shards, within-shard
   contiguity fuses aggressively, and you get a handful of large serial reads
   instead of many concurrent ones.

2. Interleave width = distinct shard count (coalesce.c:86-105): the
   round-robin spreads emitted reads across distinct shard_path runs because
   "simultaneous reads into one file serialize on NFS." That's optimal at 512
   arrays (512 files → 512-wide spread → all 64 io workers on distinct files). At
   2 arrays it backfires: ~2 shard runs → 64 workers pile onto 2 files → NFS
   serializes per-file → effective concurrency ~2.

This matches the tensorstore contrast exactly: tensorstore issues 32 concurrent
reads into 2 files (accepts per-file serialization, wins on aggregate); damacy's
coalescer is designed to avoid concurrent same-file reads, so it leaves the pool
idle when there aren't enough files. Decode reads ~100% occupied but starved —
each wave carries few/large serial reads.

This is a distinct axis from the #146–#151 work (that was cold-open depth sag
within a well-fed pipeline; this is the pipeline failing to generate enough
independent work).

Approach: confirm the mechanism, then make coalescing pool-aware

Don't jump to a fix — the fork between "few distinct read_ops per wave" vs "enough
ops but they serialize on few files" changes the fix, and only counters
distinguish them.

Step 1 — sweep. bench/scenario.py + gen_dataset.py, hold
chunk/shard/dtype/codec/patch fixed, sweep n_zarrs ∈ {1,2,4,16,64,512}.
Deterministic single-variable repro, cheap.

Step 2 — instrument planner output + wave dispatch. Per wave, log: post-coalesce
n_read_ops, distinct shard_path count, fused-bytes histogram, chunks-per-wave,
and which budget cut the wave (input_cap / max_chunks_per_wave /
dev_decompressed_cap). Cross with metadata_backend_read_max_active and in-flight
read depth. Decision question: does post-coalesce n_read_ops and in-flight depth
track n_zarrs, or the pool size? Plot both against the sweep; if they track
n_zarrs and plateau, the hypothesis is confirmed. (Probe branch, in the
probe/wave-trace lineage.)

Step 3 — fix, contingent on Step 2. The tension is real: per-file
sequential-large is fastest per file, but with few files you must parallelize
within a file to use the pool. Candidates, preferred first:

• Pool-aware fusion split (primary): make fusion adaptive to
  n_io_threads / n_distinct_shards. When distinct shards in a wave is well below
  the worker count, split each shard's chunk set into K ~ pool/distinct_shards
  concurrent sub-reads instead of fusing to 4 MiB. Trade a little per-file NFS
  serialization for filling the pool — what tensorstore does to win. Localized
  change in the coalescer's fusion decision.
• Cap fused-op size by available width: lower effective max_read_op_bytes
  when few shards are present so op count stays >= pool size.
• Decouple interleave from concurrency: if Step 2 shows ops exist but serialize
  on-file, the round-robin only reorders — the fix moves to the io-submission layer
  (allow N concurrent reads per file when total files < pool).

The fix must be a response to the distinct-shard-vs-pool ratio, not a fixed
constant — the 512-array case is already optimal and must keep its current behavior.

First action

Build the n_zarrs sweep + per-wave instrumentation on a probe branch, run it
cold-cache on an L40, and confirm whether n_read_ops/depth track array count.
That single plot decides the fix.

[nclack]

Findings: measured, not reproduced as framed

Built the per-wave instrumentation + a n_zarrs sweep (probe branch
probe/array-starve), then ran the damacy-vs-tensorstore comparison on real
datasets chosen to isolate the variable. Conclusion up front: the "parallelism
limited by #arrays" framing does not hold on well-formed data — damacy beats
tensorstore even at 1 array. The original loss was decode amplification, not
array count.

What the synthetic sweep showed

A synthetic n_zarrs sweep (hold chunk/shard/dtype/patch fixed, vary array count)
did not reproduce the collapse. With small shards, 2 arrays already expose
enough shard files that shards/wave ≈ 100 and waves fill to the chunk cap. With
large shards (2/array, so shard count = 2→64 across the sweep) throughput was flat
~11→9.6 GB/s and fell slightly with more arrays — a small-working-set page-cache
artifact (few arrays = cacheable). Decode stayed fed at ~28 GB/s throughout. So
array/shard count alone doesn't starve the pipeline; the synthetic data lacked the
real datasets' decode amplification.

Identifying the original case

The tensorstore comparison that motivated this (#153) lost on
allencell-dynamic-nuc-morph_mip — the 2-array row: damacy 0.14 GB/s
(38 samp/s) vs tensorstore 0.43 GB/s (203 samp/s, 32 threads). That dataset is
decode-amplification-bound (~20x): 32 MiB unsharded chunks
[16,1,1,1024,1024] over tiny arrays, read with a 256² patch. The 512-array
comparison row was allencell-static-hipsc-fov_mip (116x amp). Both are
data-format problems (rechunk offline), not pipeline bugs.

Comparative benchmark on low-decode-amp few-shard datasets

Picked datasets with few arrays/shards but reasonable decode-amp (where #154,
if real, is the only possible cause), plus dynacell (raw sharded, ~9.8k shards) as
a control. Cold-cache, fast /mnt/main0, tuned preset (n_io_threads=64), matched
xorshift patches, tensorstore swept 1→32 threads:

dataset	arrays	shards/wave	damacy io GB/s	damacy GPU	tensorstore CPU (best)	winner
chad_actin	1	6	1.09	1.97 GB/s (517/s)	1.44 (343, 32t)	damacy 1.4x
alfi	58	37	2.37	5.69 GB/s (1491/s)	1.58 (751, 32t)	damacy 3.6x
dynacell (control)	201	236	4.32	4.49 GB/s (236/s)	3.46 (165, 32t)	damacy 1.3x

damacy wins on all three — including the 1-array case. All three are io-bound;
decode is well-fed everywhere (28/22/89 GB/s out, 23/53/24% of wall).

The real, residual effect (the kernel worth keeping)

io throughput scales cleanly with shards/wave:

shards/wave    6  ->  io 1.09 GB/s   (chad, 1 array)
shards/wave   37  ->  io 2.37 GB/s   (alfi)
shards/wave  236  ->  io 4.32 GB/s   (dynacell)

With few shards the 64-thread io pool can't spread across enough files —
per-file NFS read concurrency saturates before the pool does (chad issues 137
read ops into just 6 files). This matches the chammi reformat result (one-shard-
per-array regressed ~25% on the fast mount). But it's a ~4x spread on the io
stage, and GPU decode is ~free, so damacy stays io-bound and ahead of
tensorstore regardless of shard count. It leaves io throughput on the table; it
does not make damacy uncompetitive.

Recommendation: re-scope #154

• Not "parallelism limited by array count" (false — 1-array chad wins).
• The catastrophic tensorstore losses were decode amplification → fix in data
  (rechunk), tracked separately from the pipeline.
• The genuine pipeline item is narrower: io-pool under-utilization on
  few-shard datasets — raise per-file read concurrency, or split reads within a
  shard when shard count < pool. Upside ~2–4x on the io stage of few-shard sets;
  not a correctness or competitiveness fix. Suggest re-titling to that scope, or
  closing in favor of the data-format work + a smaller io-utilization issue.

Instrumentation (per-wave shards/wave/reads/wave/stop-reason counters +
DAMACY_TRACE_WAVES) is on probe/array-starve if worth landing.

[nclack]

Closing — not a pipeline-parallelism bug as framed

Resolved by the measurements above. Disposition:

• The catastrophic tensorstore losses were decode amplification, not array
  count. That's data-format work (rechunk), tracked with the corpus/reformatting
  effort — not a pipeline fix.
• damacy beats tensorstore on every well-formed dataset measured, including a
  1-array one. Array count is not the limiter.
• The real residual is a mild read-width effect: bulk-read throughput scales
  with how many distinct files are in flight (6→1.09, 37→2.37, 236→4.32 GB/s). The
  cross-shard interleave (planner: interleave reads across shards #146, bulk reads) already handles the "many files, don't
  clump" half; the leftover is "too few files to spread across," which reordering
  can't fix — it would need multiple readers splitting work within one file. The
  chammi reformat sweep confirms this is monotonic (more shards/array = faster,
  unsharded fastest, one-shard-per-array ~half rate). It only costs read-stage
  speed; GPU decode is ~free, so damacy stays ahead regardless.

So the framing ("parallelism limited by #arrays") doesn't hold, and both real
effects are tracked elsewhere: decode-amp = data format; read-width residual = a
minor io-stage optimization (split a shard's reads when files < pool). Reopen or
spin that out if the few-shard io headroom is worth chasing.

Diagnostics that produced this: PR #159.