Skip to content

fix: keep FusedCodecPipeline compute off the event-loop thread - #247

Open
d-v-b wants to merge 9 commits into
mainfrom
claude/fusedcodecpipeline-zstd-perf-0bb07d
Open

fix: keep FusedCodecPipeline compute off the event-loop thread#247
d-v-b wants to merge 9 commits into
mainfrom
claude/fusedcodecpipeline-zstd-perf-0bb07d

Conversation

@d-v-b

@d-v-b d-v-b commented Jul 28, 2026

Copy link
Copy Markdown
Owner

🤖 AI text below 🤖

fix: keep FusedCodecPipeline compute off the event-loop thread

Problem

Users reported that the opt-in FusedCodecPipeline is slower than
BatchedCodecPipeline for zstd-compressed data. Investigation showed the
fused pipeline is actually ~2x faster in every single-threaded scenario; the
regression only appears under concurrent access from multiple user threads
(the dask/xarray pattern: many threads, each reading roughly one chunk per
call).

Root cause: FusedCodecPipeline.read/write ran their synchronous fast path
(read_sync/write_sync) inline on the coroutine servicing the request —
i.e. on the global zarr_io event-loop thread. Single-chunk batches
decoded inline on the loop; multi-chunk batches blocked the loop in
pool.map until the whole batch finished. Since every sync-API call from
every user thread is serviced by that one loop, concurrent operations
serialized behind each other's codec compute. The blocked-loop window is the
codec compute time, so zstd-compressed data (real CPU work per chunk) showed
the regression prominently while uncompressed data barely moved.

Evidence (8192x8192 f32, 64 x 4 MB chunks, zstd level 3, LocalStore; each
user thread reads one chunk per call):

user threads batched fused (before)
1 669 ms 336 ms
2 361 ms 334 ms
4 204 ms 407 ms
8 121 ms 439 ms

Batched scales with threads; fused was flat-to-degrading — the signature of
everything funneling through one thread. Tracing confirmed decode executed on
the thread named zarr_io for single-chunk reads.

Fix

Run the synchronous batch on a worker thread via asyncio.to_threadone
hop per batch, not per chunk
, so the fused design's win over per-chunk
async scheduling (the reason this pipeline exists) is preserved, while the
event loop stays free to service concurrent callers.

Benchmarks (after)

Same workload as above, interleaved in-process A/B (medians of 7 rounds;
"inline" = old behavior, "to_thread" = this PR):

workload inline (before) to_thread (after)
full-array read, 1 call 102 ms 102 ms
64 single-chunk reads, 8 threads 469-497 ms 105-107 ms

Thread-scaling after the fix (same benchmark as the table above):

user threads batched fused (after)
1 649 ms 625 ms
2 366 ms 329 ms
4 206 ms 183 ms
8 118 ms 109 ms

Fused now scales with reader threads and is at least as fast as batched at
every point. Single-threaded workloads are unchanged: the isolated cost of the
offload hop is ~75 µs per batch (measured with a noop through
sync(asyncio.to_thread(...))), which is noise next to per-chunk decode.
(An earlier A/B that appeared to show a ~2x single-thread regression for
sequential single-chunk reads turned out to be a thermal/ordering artifact:
reversing the variant order moved the inflated median to the other variant;
per-call interleaved p50s are 4.78 ms vs 4.89 ms.)

Single-threaded sweep (unchanged, fused/batched time ratios after this PR,
LocalStore): zstd write 0.62, zstd read 0.61, raw write 0.69, raw read 0.85.

Changes

  • FusedCodecPipeline.read/write: offload read_sync/write_sync to a
    worker thread via asyncio.to_thread, with a comment explaining why inline
    execution is forbidden.
  • New regression test test_sync_api_compute_off_event_loop: deterministic
    (no timing) — asserts codec compute never runs on a thread with a running
    event loop, for single- and multi-chunk reads and writes through the sync
    API.
  • New benchmark test_read_array_concurrent in tests/benchmarks/test_e2e.py
    covering the dask-style many-threads/one-chunk-per-call access pattern for
    both pipelines, zstd and uncompressed.
  • Changelog entry changes/247.bugfix.md.

🤖 Generated with Claude Code

d-v-b added 5 commits July 14, 2026 13:25
* fix: byte-order handling for structured dtypes in the bytes codec

The bytes codec neither byte-swapped structured-dtype fields to its
configured endian on encode (numpy reports byteorder '|' for void
dtypes, so the top-level byteorder comparison never detected a
mismatch) nor honored its endian when decoding, silently corrupting
any structured data whose field byte order differed from the stored
one (e.g. virtual references to external big-endian data).

Encode now detects byte-order mismatches by comparing full dtypes via
newbyteorder, and decode reinterprets raw bytes in the stored byte
order before converting to the data type's declared byte order, so the
stored layout (codec state) and the in-memory layout (array data type)
are independent.

Closes zarr-developers#4141

Assisted-by: ClaudeCode:claude-fable-5

* test: fold structured byte-order cases into existing bytes codec tests

Extend test_endian's parametrization with structured dtypes and
test_bytes_codec_sync_roundtrip with endian/dtype parametrization plus
stored-layout and decoded-dtype assertions, instead of adding parallel
test functions for the same properties.

Assisted-by: ClaudeCode:claude-fable-5

* refactor: rename stored_dtype to view_dtype in BytesCodec decode

The variable is the dtype used to view the raw chunk bytes (byte order
from the codec's endian configuration), not a property of the stored
data or of the returned buffer, which always carries the array's
declared dtype.

Assisted-by: ClaudeCode:claude-fable-5

* docs: note that the decode-side byte-order conversion copies the chunk

Assisted-by: ClaudeCode:claude-fable-5
…op thread

FusedCodecPipeline.read/write ran their synchronous fast path inline on
the coroutine servicing the request — i.e. on the global zarr_io event
loop thread. Single-chunk batches decoded inline on the loop and
multi-chunk batches blocked the loop in pool.map, so every sync-API call
from every user thread serialized behind each other's codec compute.
The blocked window scales with codec cost, which is why users reported
the fused pipeline as "slower for zstd-compressed data" under
multi-threaded (dask-style, one chunk per call) access: at 8 reader
threads on 4 MiB zstd chunks it was 3.4x slower than
BatchedCodecPipeline, and throughput did not scale with threads at all
(336 -> 439 ms from 1 to 8 threads, versus 669 -> 121 ms for batched).

Offload the synchronous batch to a worker thread with asyncio.to_thread:
one hop per batch, not per chunk, preserving the fused pipeline's win
over per-chunk async scheduling while keeping the loop free. After the
fix the same workload scales 625 -> 109 ms from 1 to 8 threads, beating
batched at every thread count; single-threaded performance is unchanged
(the hop costs ~75 us per batch).

The regression test asserts deterministically (no timing) that codec
compute never runs on a thread with a running event loop, covering
single- and multi-chunk reads and writes through the sync API. A new
benchmark covers the many-threads/one-chunk-per-call access pattern.

Assisted-by: ClaudeCode:claude-fable-5
d-v-b added 4 commits July 28, 2026 16:41
towncrier's issue_format links to zarr-developers/zarr-python issues, so
247 (the fork PR number) would render a link to an unrelated old issue.

Assisted-by: ClaudeCode:claude-fable-5
…ps triggering

The test asserts a negative (compute never ran on the loop thread). If a
refactor made the sync fast path stop triggering, the traced ChunkTransform
methods would never be called (the async fallback uses AsyncChunkTransform)
and the test would pass while guarding nothing. Assert the traced hooks
actually ran.

Assisted-by: ClaudeCode:claude-fable-5
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant