perf: use sync methods for chunk encoding / decoding

.gitignore

@@ -92,5 +92,9 @@ tests/.hypothesis
 zarr/version.py
 zarr.egg-info/
 
+# Local agent / planning notes (not versioned)
+.claude/
+CLAUDE.md
+docs/superpowers/
 # zarr-metadata package lockfile (a library, not an app)
 packages/zarr-metadata/uv.lock

changes/3885.feature.1.md

@@ -0,0 +1 @@
+Added `SyncByteGetter` and `SyncByteSetter` runtime-checkable protocols and a `get_ranges_sync` method on the `Store` ABC. These let custom byte getters/setters opt into the synchronous codec pipeline's fast path for in-memory IO, which the sharding codec uses for its inner chunks.

changes/3885.feature.md

@@ -0,0 +1 @@
+Added `FusedCodecPipeline`, an opt-in codec pipeline that runs codec compute synchronously and in bulk (avoiding the per-chunk async scheduling overhead of the default `BatchedCodecPipeline`), giving large speedups for sharded arrays (up to ~24x writes / ~14x reads on many-chunks-per-shard layouts, more with compression) and no regressions on compute-bound workloads. The default pipeline is unchanged; enable the new one with `zarr.config.set({"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"})`.

docs/user-guide/experimental.md

@@ -2,6 +2,96 @@
 
 This section contains documentation for experimental Zarr Python features. The features described here are exciting and potentially useful, but also volatile -- we might change them at any time. Take this into account if you consider depending on these features.
 
+## `FusedCodecPipeline`
+
+A *codec pipeline* is the machinery that turns chunks of array data into stored bytes and back, by running the configured codecs (filters, serializer, compressors) and performing the storage IO.
+The default pipeline, `BatchedCodecPipeline`, schedules both the IO and codec work asynchronously -- roughly one coroutine per chunk operation.
+
+`FusedCodecPipeline` is an experimental alternative that runs codec compute and synchronous IO *synchronously*, avoiding that per-chunk async scheduling overhead and nasty [`asyncio.to_thread` overhead](https://github.com/python/cpython/issues/136084).
+On real workloads the scheduling cost can dominate the actual codec work, so removing it is a significant speedup -- especially for **sharded arrays**, where a single shard read or write involves many inner chunks.
+
+> **Note:** The win is *not* a faster compressor or a different on-disk format -- the bytes written are
+> identical. It is purely the removal of async scheduling overhead, plus a few vectorized fast paths
+> for dense, uncompressed shards i.e., removing compute where it is not needed.
+
+### When it helps
+
+There are two main benefits in this new pipeline:
+
+1. When storage IO is fast enough that the *scheduling* overhead, not the IO itself, is the bottleneck. That means **low-latency stores** that are themselves synchronous -- in particular [`zarr.storage.MemoryStore`][] and [`zarr.storage.LocalStore`][].
+
+2. Whenever codec work that is truly synchronous will not need the overhead of `async` scheduling i.e., inner-chunk codec work in sharding using something like `zstd`. We also now make use of `asyncio.as_completed` so that IO from asynchronous sources can begin decompression immediately.
+
+### Opting in
+
+`FusedCodecPipeline` is opt-in: the default pipeline is unchanged, so existing code behaves exactly as
+before. Select it through the [runtime configuration](config.md), by setting `codec_pipeline.path`:
+
+```python exec="true" session="experimental-fused" source="above" result="ansi"
+import zarr
+
+zarr.config.set(
+    {"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}
+)
+```
+
+You can set this globally as above (affecting every array created or opened afterwards), or scope it to
+a block of code using `zarr.config.set` as a context manager:
+
+```python exec="true" session="experimental-fused" source="above"
+import numpy as np
+import zarr
+from zarr.storage import MemoryStore
+
+with zarr.config.set(
+    {"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}
+):
+    # A sharded array on an in-memory store -- the low-latency case the
+    # synchronous pipeline targets.
+    arr = zarr.create_array(
+        store=MemoryStore(),
+        shape=(1000, 1000),
+        chunks=(100, 100),
+        shards=(1000, 1000),
+        dtype="float32",
+    )
+    arr[:] = np.random.random((1000, 1000)).astype("float32")
+    result = arr[:]
+
+print(result.shape)
+```
+
+To return to the default pipeline, set `codec_pipeline.path` back to the batched implementation:
+
+```python exec="true" session="experimental-fused" source="above"
+import zarr
+
+zarr.config.set(
+    {"codec_pipeline.path": "zarr.core.codec_pipeline.BatchedCodecPipeline"}
+)
+```
+
+### Threading
+
+By default the synchronous pipeline runs sequentially (`codec_pipeline.max_workers` is `1`). For
+CPU-heavy decode/encode on large chunks you can opt into a thread pool by setting a larger worker
+count -- or `None` to use `os.cpu_count()`:
+
+```python exec="true" session="experimental-fused" source="above"
+import zarr
+
+# Use a fixed-size thread pool for codec compute.
+zarr.config.set({"codec_pipeline.max_workers": 8})
+
+# Or "auto", sized to the number of CPUs.
+zarr.config.set({"codec_pipeline.max_workers": None})
+```
+
+Threading is off by default because it is not universally a win: it slows small chunks (where the
+per-task overhead outweighs the work), and on many-core nodes a pool sized to `cpu_count` can
+oversubscribe workloads that already parallelize at a higher level (e.g. Dask). `max_workers` only
+affects `FusedCodecPipeline`; the default batched pipeline ignores it.
+
 ## `CacheStore`
 
 Zarr Python 3.1.4 adds [`zarr.experimental.cache_store.CacheStore`][], which provides a dual-store caching implementation

pyproject.toml

@@ -457,6 +457,18 @@ filterwarnings = [
     # collected without being entered. This is a known issue in s3fs/aiobotocore, and pytest
     # per-test filterwarnings markers can't catch it (https://github.com/pytest-dev/pytest/issues/14096).
     "ignore:Exception ignored ((on calling weakref callback)|(in[\\s\\S]*Session was never entered)):pytest.PytestUnraisableExceptionWarning",
+    # pytest-asyncio implicitly creates an event loop in _get_event_loop_no_warn during
+    # fixture setup/teardown and never closes it (allocation site verified with
+    # PYTHONTRACEMALLOC: pytest_asyncio/plugin.py). When the garbage collector reclaims
+    # that loop (and its self-pipe socketpair: AF_UNIX family=1 on POSIX, emulated with
+    # AF_INET family=2 on Windows) mid-test, the unraisable hook fails whichever unrelated
+    # test happens to be running — the long-standing "random cross-file failure" in the
+    # pipeline suites. The message contains only the __del__ repr, so these patterns cannot
+    # scope to pytest-asyncio specifically: a loop/socketpair leak in zarr's own sync
+    # machinery would also be silenced. Accepted tradeoff — revisit if zarr.core.sync grows
+    # loop-lifecycle changes.
+    "ignore:Exception ignored in[\\s\\S]*<function BaseEventLoop.__del__:pytest.PytestUnraisableExceptionWarning",
+    "ignore:Exception ignored in[\\s\\S]*<socket.socket fd=\\d+, family=(1|2), type=1:pytest.PytestUnraisableExceptionWarning",
 ]
 markers = [
     "asyncio: mark test as asyncio test",

src/zarr/abc/codec.py

@@ -93,7 +93,15 @@ class BaseCodec[CI: CodecInput, CO: CodecOutput](Metadata):
     ArrayArrayCodec, ArrayBytesCodec or BytesBytesCodec for subclassing.
     """
 
-    is_fixed_size: bool
+    # Whether this codec's encoded output is a fixed size given a fixed input
+    # size. Defaults to False (the conservative answer): a codec that does not
+    # explicitly opt in is treated as variable-size, which only disables
+    # size-dependent fast paths (e.g. the sharding bulk-decode), never
+    # correctness. Codecs with genuinely fixed-size output (BytesCodec,
+    # TransposeCodec, ...) override this with True. The default also keeps
+    # third-party / variable-length codecs (VLenUTF8, numcodecs wrappers) that
+    # never set the attribute from raising AttributeError where it is read.
+    is_fixed_size: bool = False
 
     @abstractmethod
     def compute_encoded_size(self, input_byte_length: int, chunk_spec: ArraySpec) -> int:

src/zarr/abc/store.py

@@ -22,6 +22,8 @@
     "SupportsGetSync",
     "SupportsSetSync",
     "SupportsSyncStore",
+    "SyncByteGetter",
+    "SyncByteSetter",
     "set_or_delete",
 ]
 
@@ -469,6 +471,73 @@ async def get_ranges(
         ):
             yield group
 
+    def get_ranges_sync(
+        self,
+        key: str,
+        byte_ranges: Sequence[ByteRequest | None],
+        *,
+        prototype: BufferPrototype,
+        max_gap_bytes: int = 1 << 20,  # 1 MiB
+        max_coalesced_bytes: int = 16 << 20,  # 16 MiB
+    ) -> Sequence[tuple[int, Buffer | None]]:
+        """Synchronous, coalescing counterpart of `get_ranges`.
+
+        Plans merged fetches with the same `coalesce_ranges` policy as the async
+        path, then issues one synchronous `get_sync` per merged group (or per
+        uncoalescable request) and slices results back into per-input buffers.
+        Used by the sync codec pipeline's partial-shard reads so they get the
+        same byte-range coalescing as the async path, without an event loop.
+
+        Returns a list of `(input_index, Buffer | None)`. Raises
+        `BaseExceptionGroup` containing a `FileNotFoundError` if the key is
+        absent (matching `get_ranges`), so callers can handle a deleted shard
+        uniformly across the sync and async paths.
+
+        Requires the store to implement `get_sync` (`SupportsGetSync`).
+        """
+        from zarr.core._coalesce import coalesce_ranges
+
+        if not isinstance(self, SupportsGetSync):
+            raise TypeError(f"{type(self).__name__} does not support synchronous reads")
+
+        groups, uncoalescable = coalesce_ranges(
+            byte_ranges, max_gap_bytes=max_gap_bytes, max_coalesced_bytes=max_coalesced_bytes
+        )
+        results: list[tuple[int, Buffer | None]] = []
+        errors: list[BaseException] = []
+
+        def _get(req: ByteRequest | None) -> Buffer | None:
+            return self.get_sync(key, prototype=prototype, byte_range=req)
+
+        for idx, req in uncoalescable:
+            buf = _get(req)
+            if buf is None:
+                errors.append(FileNotFoundError(key))
+            else:
+                results.append((idx, buf))
+
+        for members in groups:
+            if len(members) == 1:
+                solo_idx, solo_req = members[0]
+                buf = _get(solo_req)
+                if buf is None:
+                    errors.append(FileNotFoundError(key))
+                else:
+                    results.append((solo_idx, buf))
+                continue
+            start = members[0][1].start
+            end = max(r.end for _, r in members)
+            big = _get(RangeByteRequest(start, end))
+            if big is None:
+                errors.append(FileNotFoundError(key))
+                continue
+            for member_idx, r in members:
+                results.append((member_idx, big[r.start - start : r.end - start]))
+
+        if errors:
+            raise BaseExceptionGroup("chunk read failed", errors)
+        return results
+
     async def getsize(self, key: str) -> int:
         """
         Return the size, in bytes, of a value in a Store.
@@ -564,6 +633,33 @@ async def delete(self) -> None: ...
     async def set_if_not_exists(self, default: Buffer) -> None: ...
 
 
+@runtime_checkable
+class SyncByteGetter(Protocol):
+    """A `ByteGetter` that can also fetch synchronously, without an event loop.
+
+    Non-StorePath byte getters (e.g. the sharding codec's in-memory
+    `_ShardingByteGetter`) implement this so a synchronous codec pipeline can
+    take its sync fast path on them instead of scheduling one coroutine per
+    chunk. Note that `StorePath` also *has* a `get_sync` method (so it matches
+    this protocol structurally) but it only works when its store supports
+    synchronous IO — callers gate `StorePath` on the store's `SupportsGetSync`
+    instead of on this protocol.
+    """
+
+    def get_sync(
+        self, prototype: BufferPrototype | None = None, byte_range: ByteRequest | None = None
+    ) -> Buffer | None: ...
+
+
+@runtime_checkable
+class SyncByteSetter(SyncByteGetter, Protocol):
+    """A `ByteSetter` that can also write synchronously. See `SyncByteGetter`."""
+
+    def set_sync(self, value: Buffer) -> None: ...
+
+    def delete_sync(self) -> None: ...
+
+
 @runtime_checkable
 class SupportsGetSync(Protocol):
     def get_sync(

src/zarr/codecs/_v2.py

@@ -23,22 +23,22 @@ class V2Codec(ArrayBytesCodec):
 
     is_fixed_size = False
 
-    async def _decode_single(
+    def _decode_sync(
         self,
         chunk_bytes: Buffer,
         chunk_spec: ArraySpec,
     ) -> NDBuffer:
         cdata = chunk_bytes.as_array_like()
         # decompress
         if self.compressor:
-            chunk = await asyncio.to_thread(self.compressor.decode, cdata)
+            chunk = self.compressor.decode(cdata)
         else:
             chunk = cdata
 
         # apply filters
         if self.filters:
             for f in reversed(self.filters):
-                chunk = await asyncio.to_thread(f.decode, chunk)
+                chunk = f.decode(chunk)
 
         # view as numpy array with correct dtype
         chunk = ensure_ndarray_like(chunk)
@@ -70,7 +70,7 @@ async def _decode_single(
 
         return get_ndbuffer_class().from_ndarray_like(chunk)
 
-    async def _encode_single(
+    def _encode_sync(
         self,
         chunk_array: NDBuffer,
         chunk_spec: ArraySpec,
@@ -83,18 +83,32 @@ async def _encode_single(
         # apply filters
         if self.filters:
             for f in self.filters:
-                chunk = await asyncio.to_thread(f.encode, chunk)
+                chunk = f.encode(chunk)
         # check object encoding
         if ensure_ndarray_like(chunk).dtype == object:
             raise RuntimeError("cannot write object array without object codec")
 
         # compress
         if self.compressor:
-            cdata = await asyncio.to_thread(self.compressor.encode, chunk)
+            cdata = self.compressor.encode(chunk)
         else:
             cdata = chunk
         cdata = ensure_bytes(cdata)
         return chunk_spec.prototype.buffer.from_bytes(cdata)
 
+    async def _decode_single(
+        self,
+        chunk_bytes: Buffer,
+        chunk_spec: ArraySpec,
+    ) -> NDBuffer:
+        return await asyncio.to_thread(self._decode_sync, chunk_bytes, chunk_spec)
+
+    async def _encode_single(
+        self,
+        chunk_array: NDBuffer,
+        chunk_spec: ArraySpec,
+    ) -> Buffer | None:
+        return await asyncio.to_thread(self._encode_sync, chunk_array, chunk_spec)
+
     def compute_encoded_size(self, _input_byte_length: int, _chunk_spec: ArraySpec) -> int:
         raise NotImplementedError