fix: reset the codec-pipeline thread pool after fork

changes/4483.bugfix.md

@@ -0,0 +1 @@
+FusedCodecPipeline reads no longer hang in a process forked after a read populated the module-level codec thread pool. `_pool`, `_pool_size`, and `_pool_lock` are now reset in the child via an `os.register_at_fork` hook. Previously the child inherited an executor whose threads did not exist, so small reads were queued on a pool nothing would run.

src/zarr/core/codec_pipeline.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import os
 import threading
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
@@ -100,6 +101,24 @@ def _get_pool(max_workers: int) -> ThreadPoolExecutor:
     return _pool
 
 
+def _reset_pool_after_fork() -> None:
+    """Discard the pool a forked child inherits from its parent.
+
+    Executor threads do not survive `fork()`, so the inherited pool accepts
+    submissions that no thread ever runs. The lock is recreated too: it may
+    have been held by a thread that no longer exists in the child.
+    """
+    global _pool, _pool_size, _pool_lock
+    _pool = None  # pragma: no cover
+    _pool_size = 0  # pragma: no cover
+    _pool_lock = threading.Lock()  # pragma: no cover
+
+
+# this is only available on certain operating systems
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=_reset_pool_after_fork)
+
+
 def _unzip2[T, U](iterable: Iterable[tuple[T, U]]) -> tuple[list[T], list[U]]:
     out0: list[T] = []
     out1: list[U] = []

tests/test_array.py

@@ -2152,6 +2152,41 @@ def test_multiprocessing(
     assert all(np.array_equal(r, data) for r in results)
 
 
+@pytest.mark.skipif(
+    sys.platform in ("win32", "darwin"),
+    reason="fork not supported on Windows or OSX",
+)
+@pytest.mark.filterwarnings(
+    r"ignore:This process \(pid=\d+\) is multi-threaded, use of fork\(\):DeprecationWarning"
+)
+@pytest.mark.parametrize("store", ["local"], indirect=True)
+def test_multiprocessing_fork_codec_pipeline_pool(store: Store) -> None:
+    """A forked child must not reuse the parent's codec-pipeline thread pool.
+
+    Regression test for https://github.com/zarr-developers/zarr-python/issues/4478:
+    submitting to the inherited pool hung forever because its threads did not
+    survive the fork. The child must read FEWER chunks than the pool's inherited
+    idle permits: once submissions exceed the stale permits, a real thread is
+    spawned and the queued work drains normally. Single-chunk batches bypass the
+    pool entirely.
+    """
+    data = np.arange(100).reshape(10, 10)
+    with zarr.config.set({"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}):
+        arr = zarr.create_array(store=store, shape=data.shape, chunks=(5, 5), dtype=data.dtype)
+        arr[:] = data
+        # Two multi-chunk reads: spawn 4 pool threads and leave all of their
+        # idle-semaphore permits released before the fork.
+        arr[:]
+        arr[:]
+        ctx = mp.get_context("fork")
+        with ctx.Pool() as pool:
+            # [:5] spans 2 chunks — below the ~4 inherited idle permits, so on
+            # unfixed code both submissions are swallowed by the dead pool.
+            result = pool.starmap_async(_index_array, [(arr, np.s_[:5])])
+            (read_back,) = result.get(timeout=30)
+        assert np.array_equal(read_back, data[:5])
+
+
 def test_create_array_method_signature() -> None:
     """
     Test that the signature of the ``AsyncGroup.create_array`` function has nearly the same signature