From 220faee3128d723406fc17ffe46fb9d95334a233 Mon Sep 17 00:00:00 2001
From: barlowa124 <barlowa124@users.noreply.github.com>
Date: Tue, 6 Oct 2026 18:11:23 -0400
Subject: [PATCH 1/2] fix: reset the codec-pipeline thread pool after fork

A forked child inherits codec_pipeline._pool pointing at threads that
do not exist in the child, plus the parent's idle-semaphore permits.
Submissions up to the permit count are consumed without spawning a
thread, so small FusedCodecPipeline reads hang forever.

Add an after_in_child hook that drops the inherited pool, size, and
lock. The lock is recreated because it may have been held by a thread
that no longer exists in the child.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 src/zarr/core/codec_pipeline.py | 19 ++++++++++++++++++
 tests/test_array.py             | 35 +++++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+)

diff --git a/src/zarr/core/codec_pipeline.py b/src/zarr/core/codec_pipeline.py
index 9ed16bc381..dd893ea8d8 100644
--- a/src/zarr/core/codec_pipeline.py
+++ b/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] = []
diff --git a/tests/test_array.py b/tests/test_array.py
index 611524d474..91b80d7f1b 100644
--- a/tests/test_array.py
+++ b/tests/test_array.py
@@ -2097,6 +2097,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

From 14e58499424f8ee22acb7842c2b5d0779f59d232 Mon Sep 17 00:00:00 2001
From: barlowa124 <barlowa124@users.noreply.github.com>
Date: Tue, 6 Oct 2026 18:12:00 -0400
Subject: [PATCH 2/2] docs: add changelog entry for #4483

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 changes/4483.bugfix.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changes/4483.bugfix.md

diff --git a/changes/4483.bugfix.md b/changes/4483.bugfix.md
new file mode 100644
index 0000000000..7f90482d1d
--- /dev/null
+++ b/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.