From 220faee3128d723406fc17ffe46fb9d95334a233 Mon Sep 17 00:00:00 2001 From: barlowa124 Date: Tue, 6 Oct 2026 18:11:23 -0400 Subject: [PATCH 1/3] 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 Date: Tue, 6 Oct 2026 18:12:00 -0400 Subject: [PATCH 2/3] 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. From a2243fbc6e9fb2cd32481624259f1b74bbce5fe4 Mon Sep 17 00:00:00 2001 From: Davis Vann Bennett Date: Fri, 9 Oct 2026 19:26:22 +0200 Subject: [PATCH 3/3] test(codec_pipeline): pin pool size in the fork regression test Move the #4478 regression test next to the other FusedCodecPipeline pool tests and pin `codec_pipeline.max_workers` to 4 through the shared `_FUSED_POOL_CONFIG`. With the default, a 1-CPU host resolves `max_workers` to 1, never creates the pool, and passes the test without exercising the bug. The test now asserts that the parent created the pool before forking, and it runs on macOS, where the fork start method is available. Assisted-by: ClaudeCode:claude-opus-5-5 Co-Authored-By: Claude Opus 5.5 --- tests/test_array.py | 35 ---------------------------- tests/test_fused_pipeline.py | 45 ++++++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 35 deletions(-) diff --git a/tests/test_array.py b/tests/test_array.py index 3e4835e686..ef54320e86 100644 --- a/tests/test_array.py +++ b/tests/test_array.py @@ -2152,41 +2152,6 @@ 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 diff --git a/tests/test_fused_pipeline.py b/tests/test_fused_pipeline.py index 06e79d3de7..ea44eda324 100644 --- a/tests/test_fused_pipeline.py +++ b/tests/test_fused_pipeline.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio +import multiprocessing as mp +import sys from dataclasses import dataclass, field, replace from typing import TYPE_CHECKING, Any from unittest.mock import patch @@ -37,6 +39,7 @@ if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable, Iterable + from pathlib import Path from zarr.abc.store import ByteRequest from zarr.core.buffer import Buffer, NDBuffer @@ -543,6 +546,48 @@ def read_row_block(i: int, handle: zarr.Array[Any] = fresh) -> np.ndarray: np.testing.assert_array_equal(fut.result(), data[i * 4 : (i + 1) * 4, :]) +def _read_region(arr: zarr.Array[Any], selection: Any) -> np.ndarray: + return np.asarray(arr[selection]) + + +@pytest.mark.skipif(sys.platform == "win32", reason="fork is not available on Windows") +@pytest.mark.filterwarnings( + r"ignore:This process \(pid=\d+\) is multi-threaded, use of fork\(\):DeprecationWarning" +) +def test_forked_child_does_not_reuse_parent_pool(tmp_path: Path) -> 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. + Executor threads do not survive `fork()`, but the inherited executor keeps the + idle-semaphore permits its parent's workers released. Each submission in the + child consumes a permit instead of starting a thread, so a read of no more + chunks than there are permits queues work that nothing runs. + + `max_workers` is pinned rather than left to `os.cpu_count()`: on a 1-CPU host + the default resolves to 1, the parent never creates the pool, and the test + would pass without exercising the bug. + """ + import zarr.core.codec_pipeline as cp_mod + + data = np.arange(100).reshape(10, 10) + with zarr_config.set(_FUSED_POOL_CONFIG): + arr = zarr.create_array( + store=str(tmp_path / "a.zarr"), shape=data.shape, chunks=(5, 5), dtype=data.dtype + ) + arr[:] = data + # Each multi-chunk read releases one idle permit per completed task, so after + # two 4-chunk reads the pool holds more permits than the child read submits. + arr[:] + arr[:] + assert cp_mod._pool is not None, "the parent read did not create the pool" + with mp.get_context("fork").Pool(1) as pool: + # `[:5]` spans 2 chunks. Without the after-fork reset, both submissions + # are swallowed by the dead pool and `get` times out. + result = pool.starmap_async(_read_region, [(arr, np.s_[:5])]) + (read_back,) = result.get(timeout=30) + np.testing.assert_array_equal(read_back, data[:5]) + + def test_shared_transform_decode_alternating_specs() -> None: """A single ChunkTransform must decode chunks of DIFFERENT specs correctly when calls alternate, exercising eviction/refill of its single-entry