From 80bdc8cc3beedfe3912beffed87bab221f3de584 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:00:15 +0200 Subject: [PATCH 01/13] Fix remote HTTP test stalls and tune HTTP HDF5 caching Squashed rebase of fix-remote-hdf5-http-cache onto dev. --- .github/workflows/runtests.yml | 2 +- dascore/constants.py | 6 +- dascore/io/core.py | 12 +- dascore/utils/hdf5.py | 96 +++++++++++-- dascore/utils/io.py | 38 +++-- dascore/utils/remote_io.py | 96 ++++++++++++- docs/changelog.qmd | 2 + tests/test_io/conftest.py | 94 +++++------- tests/test_io/test_io_core.py | 26 ++++ tests/test_io/test_remote_http.py | 26 ++-- tests/test_utils/test_gc_pause.py | 228 ++++++++++++++++++++++++++++++ tests/test_utils/test_io_utils.py | 181 +++++++++++++++++++++--- 12 files changed, 681 insertions(+), 126 deletions(-) create mode 100644 tests/test_utils/test_gc_pause.py diff --git a/.github/workflows/runtests.yml b/.github/workflows/runtests.yml index ab139e2ba..5d4b390ca 100644 --- a/.github/workflows/runtests.yml +++ b/.github/workflows/runtests.yml @@ -154,7 +154,7 @@ jobs: strategy: fail-fast: false matrix: - os: [ubuntu-latest, macos-latest] + os: [ubuntu-latest, macos-latest, windows-latest] # Keep remote-IO coverage visible without blocking unrelated changes. if: github.event_name == 'push' || !contains(github.event.pull_request.labels.*.name, 'no_ci') diff --git a/dascore/constants.py b/dascore/constants.py index de3f38e68..2491cd21b 100644 --- a/dascore/constants.py +++ b/dascore/constants.py @@ -66,10 +66,14 @@ def map(self, fn: Callable, iterable: Iterable, /) -> Iterable: # types used to represent paths path_types = str | Path | UPath +# Protocols served over HTTP. The one spelling of this set; HDF5 tuning and +# the remote-cache downloader both key off it. +http_protocols = ("http", "https") + # Remote protocols that should use DASCore's smaller HDF5 readahead blocks. # h5py performs many small metadata reads while opening files, and some S3-like # backends default to large readahead chunks that overfetch remote data. -remote_hdf5_tuned_protocols = ("s3", "s3a", "s3n") +remote_hdf5_tuned_protocols = ("s3", "s3a", "s3n", *http_protocols) # One second in numpy timedelta speak ONE_SECOND = np.timedelta64(1, "s") diff --git a/dascore/io/core.py b/dascore/io/core.py index b9eb9615e..229ee9887 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -17,6 +17,7 @@ Mapping, Sequence, ) +from contextlib import suppress from functools import cached_property, wraps from numbers import Integral from pathlib import Path @@ -860,13 +861,22 @@ def _wrapper(*args, _pre_cast=False, **kwargs): bound = sig.bind(*args, **kwargs) new_kw = bound.arguments resource = new_kw.pop(arg_name) + new_resource = None try: new_resource = get_handle_from_resource(resource, required_type) new_kw[arg_name] = new_resource # kwargs is included in bound arguments, need to re-attach new_kw.update(new_kw.pop("kwargs", {})) out = func(**new_kw) - except Exception as e: # get_format can't raise; must return false. + except BaseException as e: # get_format can't raise; must return false. + # A handle created here must close even on failure, including on + # KeyboardInterrupt; leaking a remote handle would leave garbage + # collection paused for as long as the traceback is retained. + if new_resource is not None and new_resource is not resource: + with suppress(Exception): + getattr(new_resource, "close", lambda: None)() + if not isinstance(e, Exception): + raise if fun_name == "get_format": out = False else: diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index ec86e7739..5bb821377 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -18,7 +18,7 @@ from dascore.compat import UPath from dascore.config import get_config -from dascore.constants import remote_hdf5_tuned_protocols +from dascore.constants import http_protocols, remote_hdf5_tuned_protocols from dascore.utils.misc import ( _maybe_make_parent_directory, _maybe_unpack, @@ -31,6 +31,8 @@ ensure_local_file, get_local_handle, is_no_range_http_error, + pause_gc, + resume_gc, ) ns_to_datetime = partial(pd.to_datetime, unit="ns") @@ -59,22 +61,38 @@ class _ManagedH5pyFile: therefore the point where DASCore tears down the entire HDF5 access stack. """ - def __init__(self, handle: H5pyFile, owned_fileobj=None): + def __init__(self, handle: H5pyFile, owned_fileobj=None, gc_paused=False): self._handle = handle self._owned_fileobj = owned_fileobj + # The pid that paused, so a handle inherited through a fork cannot + # resume a pause the child never took. + self._gc_paused_pid = os.getpid() if gc_paused else None self._closed = False def close(self): """Close the h5py file and, when present, the owned file object.""" if self._closed: return + self._closed = True try: self._handle.close() finally: - if self._owned_fileobj is not None: - with suppress(Exception): - self._owned_fileobj.close() - self._closed = True + # Nested so nothing raised by the teardown, including a + # BaseException, can skip the resume and strand the pause. + try: + if self._owned_fileobj is not None: + with suppress(Exception): + self._owned_fileobj.close() + finally: + # dict.pop is atomic, so racing closes resume exactly once. + pid = self.__dict__.pop("_gc_paused_pid", None) + if pid == os.getpid(): + resume_gc() + + def __del__(self): + """Backstop close so a leaked handle cannot pause collection forever.""" + with suppress(Exception): + self.close() def __enter__(self): return self @@ -101,6 +119,43 @@ def __getattr__(self, item): return getattr(self._handle, item) +def _is_loop_backed_fileobj(resource) -> bool: + """ + Return True when a file object's filesystem serves reads via a loop thread. + + fsspec async filesystems (http, s3, ...) bridge each read onto a shared + event-loop thread and mark themselves with ``async_impl``; duck-typing it + avoids importing fsspec here. Buffering wrappers hide the filesystem, so + unwrap a few levels: missing one would leave the deadlock window open. + """ + for _ in range(4): + if getattr(getattr(resource, "fs", None), "async_impl", False): + return True + wrapped = getattr(resource, "raw", None) or getattr(resource, "buffer", None) + if wrapped is None or wrapped is resource: + return False + resource = wrapped + return False + + +def _open_h5_paused(fileobj, constructor, mode) -> _ManagedH5pyFile: + """ + Open a loop-backed file object with h5py while automatic gc is paused. + + The returned wrapper owns the pause and releases it on close; if the open + fails, both the pause and the file object are released here. + """ + pause_gc() + try: + handle = constructor(fileobj, mode=mode, driver="fileobj") + return _ManagedH5pyFile(handle, fileobj, gc_paused=True) + except BaseException: + with suppress(Exception): + fileobj.close() + resume_gc() + raise + + def get_h5py_file(handle) -> H5pyFile: """ Return the underlying ``h5py.File`` for a DASCore h5 handle. @@ -148,6 +203,11 @@ def open_h5_resource( if isinstance(resource, H5pyFile): return _ManagedH5pyFile(resource) if isinstance(resource, io.IOBase): + # A user-supplied fsspec file object delegates reads to the same + # event-loop thread as the UPath branch below and needs the same + # GC pause; plain local/in-memory streams do not. + if _is_loop_backed_fileobj(resource): + return _open_h5_paused(resource, constructor, mode) handle = constructor(resource, mode=mode, driver="fileobj") return _ManagedH5pyFile(handle, resource) if isinstance(resource, UPath): @@ -160,19 +220,20 @@ def open_h5_resource( constructor=constructor, open_kwargs_getter=open_kwargs_getter, ) + # Note: only mode == "r" is a supported remote path here; H5Writer + # intercepts UPath targets with its temp-file write-back handle. file_mode = "rb" if mode == "r" else "r+b" open_kwargs = open_kwargs_getter(resource) + # h5py holds its global lock while blocking on fsspec's event-loop + # thread for remote fetches; an automatic garbage collection on that + # thread deallocating h5py objects then deadlocks on the same lock. + # Pause collection for the handle's lifetime (resumed in close()). handle = _FallbackFileObj( remote_opener=lambda: resource.open(file_mode, **open_kwargs), local_opener=lambda: ensure_local_file(resource).open(file_mode), error_predicate=is_no_range_http_error, ) - try: - h5_handle = constructor(handle, mode=mode, driver="fileobj") - return _ManagedH5pyFile(h5_handle, handle) - except Exception: - handle.close() - raise + return _open_h5_paused(handle, constructor, mode) try: if mode != "r": _maybe_make_parent_directory(resource) @@ -209,6 +270,17 @@ def _get_open_kwargs(resource: UPath) -> dict[str, object]: # h5py performs many small seeks while opening HDF5 metadata. # s3fs defaults to 50 MB readahead blocks, which can pull most of # a large remote file just to satisfy metadata probes. + # HTTP needs a block LRU cache instead: the metadata probe + # alternates between the file header and footer, and fsspec's + # default single-window cache refetches a full block (or the whole + # file on range-less servers) on every jump. + if protocol in http_protocols: + # Bound retained memory: 8 blocks of the configured size. + return { + "block_size": get_config().remote_hdf5_block_size, + "cache_type": "blockcache", + "cache_options": {"maxblocks": 8}, + } return { "block_size": get_config().remote_hdf5_block_size, "cache_type": "readahead", diff --git a/dascore/utils/io.py b/dascore/utils/io.py index f2f4e9431..b51f8a53e 100644 --- a/dascore/utils/io.py +++ b/dascore/utils/io.py @@ -254,28 +254,50 @@ def get_resource(self, required_type: RequiredType) -> RequiredType: self._cache[required_type] = out return self._cache[required_type] - def close_all(self): - """Close any open file handles.""" + def close_all(self, abort: bool = False): + """ + Close any open file handles. + + With ``abort=True``, handles that support it discard uncommitted + work (e.g. remote writers skip uploading a partial file). One + handle failing must not skip cleanup of the others (remote handles + resume garbage collection in close), so the first error is + re-raised only after every handle was attempted. BaseException is + caught for that reason too: a Ctrl-C mid-close would otherwise + strand the GC pause of every handle after it. + """ + first_exc = None with self._lock: for handle in self._cache.values(): - getattr(handle, "close", lambda: None)() + try: + if abort and hasattr(handle, "abort"): + handle.abort() + else: + getattr(handle, "close", lambda: None)() + except BaseException as exc: + first_exc = first_exc if first_exc is not None else exc + if first_exc is not None: + raise first_exc def clear_cache(self): """Close and forget any cached resources so they can be reopened fresh.""" with self._lock: - self.close_all() - self._cache.clear() + try: + self.close_all() + finally: + self._cache.clear() def __enter__(self): """Entering context manager.""" return self def __exit__(self, exc_type, exc_val, exc_tb): - """Simply ensure all file handles are closed.""" - self.close_all() + """Close all handles; on error, abort uncommitted writes instead.""" + self.close_all(abort=exc_type is not None) def __del__(self): - self.close_all() + with suppress(Exception): + self.close_all() def patch_to_xarray(patch: PatchType): diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 72bc6ba5b..ac705473e 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -2,10 +2,13 @@ from __future__ import annotations +import gc import json import os import shutil import tempfile +import threading +import time import warnings from contextlib import contextmanager from contextvars import ContextVar @@ -16,11 +19,11 @@ from dascore.compat import UPath from dascore.config import get_config +from dascore.constants import http_protocols from dascore.exceptions import RemoteCacheError from dascore.utils.misc import _reinit_after_fork from dascore.utils.paths import coerce_to_upath, is_local_path, is_pathlike -_HTTP_PROTOCOLS = {"http", "https"} _NO_RANGE_HTTP_PATTERNS = ( "doesn't appear to support range requests", "only reading this file from the beginning is supported", @@ -34,6 +37,77 @@ "remote_cache_scope", default="default" ) +_gc_pause_lock = threading.Lock() +_gc_pause_depth = 0 +_gc_was_enabled = False +_gc_collect_after = 0.0 + + +def pause_gc() -> None: + """ + Pause automatic garbage collection for one remote read session. + + h5py holds its process-global lock while blocking on fsspec's event-loop + thread for each remote fetch. An automatic collection on that thread which + has to deallocate a dead h5py object needs the same lock, so the two + threads deadlock. Disabling automatic collection for the handle's lifetime + closes that window; reference counting still frees non-cyclic garbage. + + Calls nest; every ``pause_gc`` needs one ``resume_gc``. ``_ManagedH5pyFile`` + pairs them with ``close``/``__del__``. + + The pause is process-global, so cyclic garbage from every thread + accumulates until the last remote handle closes. A handle which is never + closed keeps it paused; one which is dropped inside a reference cycle is + recovered by the collection below, on the next remote open. + """ + global _gc_pause_depth, _gc_was_enabled, _gc_collect_after + now = time.monotonic() + with _gc_pause_lock: + # Count first: an interrupt before ``disable`` leaves a pending resume + # (harmless); the reverse order could leave gc off with none pending. + _gc_pause_depth += 1 + if _gc_pause_depth == 1: + _gc_was_enabled = gc.isenabled() + gc.disable() + # Claim the valve here so concurrent openers cannot each run it. + collect = now >= _gc_collect_after + if collect: + _gc_collect_after = now + 10.0 + # Safety valve: finalize cyclic garbage, including a handle leaked by an + # earlier session, so a stranded pause cannot disable collection forever. + # Must run outside the lock, since finalizing such a handle calls + # resume_gc, which takes it. Rate limited; a full collect is O(live). + if collect: + gc.collect() + + +def resume_gc() -> None: + """Undo one ``pause_gc``; the last one restores the previous gc state.""" + global _gc_pause_depth + with _gc_pause_lock: + if _gc_pause_depth == 0: + return + # Enable first, for the mirror image of the reason pause counts first. + if _gc_pause_depth == 1 and _gc_was_enabled: + gc.enable() + _gc_pause_depth -= 1 + + +@_reinit_after_fork +def _reset_gc_pause_state(): + """ + Drop a pause inherited from a fork; no close in the child can undo it. + + Handles inherited by the child record the pid that paused for them, so + closing one there cannot resume a pause this child never took. + """ + global _gc_pause_depth, _gc_pause_lock + _gc_pause_lock = threading.Lock() + if _gc_pause_depth and _gc_was_enabled: + gc.enable() + _gc_pause_depth = 0 + @_reinit_after_fork def _reinit_remote_cache_locks(): @@ -150,7 +224,7 @@ def _download_remote_file(path, local_path: Path): """Download a remote path into its cache location.""" resource = coerce_to_upath(path) protocol = getattr(resource, "protocol", None) - open_kwargs = {"block_size": 0} if protocol in _HTTP_PROTOCOLS else {} + open_kwargs = {"block_size": 0} if protocol in http_protocols else {} local_path.parent.mkdir(parents=True, exist_ok=True) fd, temp_name = tempfile.mkstemp( dir=local_path.parent, @@ -252,9 +326,13 @@ def get_local_handle(resource, opener): def is_no_range_http_error(exc: Exception) -> bool: """Return True when an exception indicates no-range HTTP random access.""" message = str(exc).lower() - return isinstance(exc, ValueError) and all( - pattern in message for pattern in _NO_RANGE_HTTP_PATTERNS - ) + if not isinstance(exc, ValueError): + return False + # Servers with no reported size stream instead of ranging; seeking such + # a streaming file needs the same local-file fallback as no-range access. + if "cannot seek streaming http file" in message: + return True + return all(pattern in message for pattern in _NO_RANGE_HTTP_PATTERNS) class _FallbackFileObj: @@ -290,6 +368,14 @@ class _FallbackFileObj: This is not a general retry wrapper for arbitrary IO failures. It is meant for one known fallback condition where switching from remote access to a local cached file is safe and expected. + + Load-bearing invariant: references flow one way here - h5py objects may + reference this wrapper and the fsspec handle, but nothing that crosses to + fsspec's event-loop thread may ever reference an h5py object. Otherwise + that thread could perform the final decref of an h5py object, whose + C-level deallocation takes h5py's global lock, and deadlock against a + caller blocked on the loop while holding that lock (``pause_gc`` only + stops cyclic collection, not refcount-driven deallocation). """ def __init__(self, remote_opener, local_opener, error_predicate): diff --git a/docs/changelog.qmd b/docs/changelog.qmd index 7e723375f..2aae0a185 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -76,3 +76,5 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f - Removed `dascore.utils.patch.merge_patches`; use `dc.spool(...).chunk(...)`. - Removed deprecated compatibility parameters `gauge_multiple`, `lag`, and `notch`; use `step_multiple`, select on lag coordinates after correlation, and `invert`, respectively. - DASCore now requires Python 3.11 or later (was 3.10, which was never covered by CI and reaches end of life in October 2026). +- Writing to a remote path no longer commits a partial file when the write fails: `dc.write` (and any `IOResourceManager` context) now aborts uncommitted remote uploads on error instead of closing, which previously uploaded whatever had been written so far. +- Reading a remote HDF5 file over HTTP/S3 pauses Python's automatic garbage collection for the lifetime of the open handle. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the handle closes; cyclic garbage from all threads accumulates until then. diff --git a/tests/test_io/conftest.py b/tests/test_io/conftest.py index 57d188d0b..d9b4039d4 100644 --- a/tests/test_io/conftest.py +++ b/tests/test_io/conftest.py @@ -14,9 +14,10 @@ import threading import time from collections.abc import Callable +from contextlib import contextmanager from functools import partial from http import HTTPStatus -from http.server import HTTPServer, SimpleHTTPRequestHandler, ThreadingHTTPServer +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.error import URLError from urllib.request import urlopen @@ -243,7 +244,7 @@ def _ensure(fetch_name: str, relative_path: str | Path | None = None) -> Path: return _ensure -class _NoReverseDNSMixin: +class _ThreadingHTTPServer(ThreadingHTTPServer): """Bind without the reverse DNS lookup ``HTTPServer`` normally performs. ``HTTPServer.server_bind`` calls ``socket.getfqdn(host)`` purely to fill in @@ -261,38 +262,41 @@ def server_bind(self): self.server_port = port -class _HTTPServer(_NoReverseDNSMixin, HTTPServer): - """Single-threaded test server that skips the reverse DNS lookup.""" - - -class _ThreadingHTTPServer(_NoReverseDNSMixin, ThreadingHTTPServer): - """Threading test server that skips the reverse DNS lookup.""" - +@contextmanager +def _serve_das_tree(handler_cls, root, label): + """ + Serve one local tree over localhost HTTP for the life of a fixture. -@pytest.fixture(scope="session") -def http_das_path(http_test_data_root, ensure_http_fetch_file): - """Return a UPath pointing at a localhost HTTP view of DAS test data.""" - handler = partial( - _SilentSimpleHTTPRequestHandler, - directory=str(http_test_data_root), - ) + Threaded so one slow or abandoned connection cannot park the accept loop; + a parked single-threaded server left new connections in TCP SYN retry + backoff, which appeared as 15+ second stalls or hangs. + """ + handler = partial(handler_cls, directory=str(root)) server = _ThreadingHTTPServer(("127.0.0.1", 0), handler) server.daemon_threads = True thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() - probe_path = "example_dasdae_event_1.h5" try: host, port = server.server_address - probe_url = f"http://{host}:{port}/das/{probe_path}" - _wait_for_http_server(probe_url, "http_das_path readiness probe") + probe_url = f"http://{host}:{port}/das/example_dasdae_event_1.h5" + _wait_for_http_server(probe_url, f"{label} readiness probe") yield UPath(f"http://{host}:{port}/das") finally: - with skip_on_timeout(10, "http_das_path teardown"): + with skip_on_timeout(10, f"{label} teardown"): server.shutdown() server.server_close() thread.join(timeout=5) if thread.is_alive(): - raise TimeoutError("HTTP test server thread did not exit cleanly.") + raise TimeoutError(f"{label} server thread did not exit cleanly.") + + +@pytest.fixture(scope="session") +def http_das_path(http_test_data_root, ensure_http_fetch_file): + """Return a UPath pointing at a localhost HTTP view of DAS test data.""" + with _serve_das_tree( + _SilentSimpleHTTPRequestHandler, http_test_data_root, "http_das_path" + ) as path: + yield path @pytest.fixture(scope="session") @@ -321,51 +325,21 @@ def _ensure(fetch_name: str, relative_path: str | Path | None = None) -> Path: @pytest.fixture(scope="session") def http_regression_das_path(http_regression_data_root, ensure_http_regression_file): """Return an isolated HTTP tree containing only the regression fixtures.""" - handler = partial( + with _serve_das_tree( _RegressionHTTPRequestHandler, - directory=str(http_regression_data_root), - ) - server = _HTTPServer(("127.0.0.1", 0), handler) - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - probe_path = "example_dasdae_event_1.h5" - try: - host, port = server.server_address - probe_url = f"http://{host}:{port}/das/{probe_path}" - _wait_for_http_server(probe_url, "http_regression_das_path readiness probe") - yield UPath(f"http://{host}:{port}/das") - finally: - with skip_on_timeout(10, "http_regression_das_path teardown"): - server.shutdown() - server.server_close() - thread.join(timeout=5) - if thread.is_alive(): - raise TimeoutError( - "HTTP regression server thread did not exit cleanly." - ) + http_regression_data_root, + "http_regression_das_path", + ) as path: + yield path @pytest.fixture(scope="session") def http_range_das_path(http_test_data_root, ensure_http_fetch_file): """Return a UPath pointing at a localhost HTTP server with range support.""" - handler = partial(_RangeHTTPRequestHandler, directory=str(http_test_data_root)) - server = _HTTPServer(("127.0.0.1", 0), handler) - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - try: - host, port = server.server_address - probe_url = f"http://{host}:{port}/das/example_dasdae_event_1.h5" - _wait_for_http_server(probe_url, "http_range_das_path readiness probe") - yield UPath(f"http://{host}:{port}/das") - finally: - with skip_on_timeout(10, "http_range_das_path teardown"): - server.shutdown() - server.server_close() - thread.join(timeout=5) - if thread.is_alive(): - raise TimeoutError( - "Range-capable HTTP server thread did not exit cleanly." - ) + with _serve_das_tree( + _RangeHTTPRequestHandler, http_test_data_root, "http_range_das_path" + ) as path: + yield path @pytest.fixture(scope="session") diff --git a/tests/test_io/test_io_core.py b/tests/test_io/test_io_core.py index 375afbd91..1800051d6 100644 --- a/tests/test_io/test_io_core.py +++ b/tests/test_io/test_io_core.py @@ -1599,6 +1599,32 @@ def test_unsupported_type(self, dummy_text_file): out = dc.read(dummy_text_file, name, version) assert out == Path(dummy_text_file).read_text() + def test_handle_closed_when_method_raises(self, dummy_text_file, monkeypatch): + """A handle opened by the caster must close when the method raises.""" + + class _Recorder: + closed = False + + def close(self): + self.closed = True + + recorder = _Recorder() + monkeypatch.setattr( + "dascore.io.core.get_handle_from_resource", + lambda resource, required_type: recorder, + ) + + class _Exploder(FiberIO): + name = "_ExploderIO" + version = "1" + + def read(self, resource: BinaryReader, **kwargs): + raise ValueError("mid-read failure") + + with pytest.raises(ValueError, match="mid-read failure"): + _Exploder().read(dummy_text_file) + assert recorder.closed + class TestGetSupportedIOTable: """A test for creating the supported io table.""" diff --git a/tests/test_io/test_remote_http.py b/tests/test_io/test_remote_http.py index 3c3f82e0f..319baf7b6 100644 --- a/tests/test_io/test_remote_http.py +++ b/tests/test_io/test_remote_http.py @@ -2,7 +2,6 @@ from __future__ import annotations -import sys from urllib.request import Request, urlopen import pytest @@ -13,20 +12,14 @@ from dascore.exceptions import InvalidSpoolError, RemoteCacheError from dascore.utils.misc import suppress_warnings from dascore.utils.remote_io import clear_remote_file_cache, get_remote_cache_path -from tests.test_io._common_io_test_utils import skip_on_timeout -# The localhost HTTP + fsspec/aiohttp streaming path intermittently deadlocks on -# Windows (the async read stalls while h5py probes remote HDF5 metadata, which -# pytest-timeout then aborts). This is a known Windows flakiness in that fallback -# path, not a DASCore logic issue. Skip the localhost-HTTP tests on Windows to -# keep CI deterministic; Linux and macOS still exercise them fully. +# The intermittent stall formerly attributed to Windows was a platform-agnostic +# deadlock between h5py's global lock and garbage collection on fsspec's +# event-loop thread, fixed by pausing collection around remote h5py handles +# (see dascore.utils.remote_io.pause_gc), so these tests run on all platforms. pytestmark = [ pytest.mark.network, pytest.mark.timeout(30), - pytest.mark.skipif( - sys.platform == "win32", - reason="Flaky localhost-HTTP fsspec/aiohttp streaming on Windows.", - ), ] @@ -181,11 +174,12 @@ def test_http_range_hdf5_read_succeeds( path = http_range_das_path / "prodml_2.1.h5" fmt = dc.get_format(path) assert fmt == ("PRODML", "2.1") - # Note: this path is intermittently hanging in CI/local repro during - # the live ranged HDF5 read, so keep the skip narrowly scoped here. - # TODO: root-cause the ranged HTTP/fsspec/h5py stall and remove this. - with skip_on_timeout(15, "http_range_hdf5_read_succeeds dc.read"): - spool = dc.read(path) + # This read used to deadlock: h5py holds its global lock while + # delegating fetches to fsspec's event-loop thread, and a garbage + # collection on that thread deallocating h5py objects needed the same + # lock. Remote h5py handles now pause automatic collection, see + # dascore.utils.remote_io.pause_gc. + spool = dc.read(path) assert spool cached = list(get_remote_cache_path().rglob("prodml_2.1.h5")) assert not cached diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py new file mode 100644 index 000000000..d70d66589 --- /dev/null +++ b/tests/test_utils/test_gc_pause.py @@ -0,0 +1,228 @@ +"""Tests for the garbage-collection pause used by remote HDF5 reads.""" + +from __future__ import annotations + +import gc +import io +import os +import threading +from contextlib import suppress + +import pytest + +import dascore.utils.remote_io as remote_io +from dascore.utils.hdf5 import ( + _is_loop_backed_fileobj, + _ManagedH5pyFile, + _open_h5_paused, +) +from dascore.utils.remote_io import pause_gc, resume_gc + + +@pytest.fixture(autouse=True) +def _gc_state_is_restored(): + """Fail loudly if a test leaves collection paused.""" + assert remote_io._gc_pause_depth == 0 + yield + assert remote_io._gc_pause_depth == 0, "test left the gc pause held" + assert gc.isenabled(), "test left automatic collection disabled" + + +class _FakeFS: + """Stand-in for an fsspec filesystem.""" + + def __init__(self, async_impl): + self.async_impl = async_impl + + +class _FakeRemoteFile(io.RawIOBase): + """A file object whose filesystem serves reads on a loop thread.""" + + def __init__(self): + self.fs = _FakeFS(True) + + def readable(self): + return True + + +class TestDeadlockProperty: + """The pause must actually prevent the h5py/loop-thread deadlock.""" + + def test_pause_prevents_loop_thread_deadlock(self): + """A collection on the loop thread cannot wedge a lock-holding reader. + + Models the real cycle: a reader holds h5py's global lock and waits on + a loop thread, while that thread's collection finalizes an object + needing the same lock. Without the pause this deadlocks. + """ + phil = threading.RLock() + request = threading.Semaphore(0) + answer = threading.Semaphore(0) + stop = threading.Event() + + class _NeedsPhil: + """Its finalizer takes the lock, as a dead h5py object would.""" + + def __init__(self): + self.self_ref = self # a cycle: only gc can free it + + def __del__(self): + with phil: + pass + + def loop_thread(): + """Serve reads and make cyclic garbage while doing it.""" + while not stop.is_set(): + if not request.acquire(timeout=0.5): + continue + for _ in range(200): + _NeedsPhil() + answer.release() + + server = threading.Thread(target=loop_thread, daemon=True) + server.start() + pause_gc() + try: + for _ in range(20): + with phil: # h5py holds its lock across the fetch + request.release() + assert answer.acquire(timeout=20), "deadlocked" + finally: + stop.set() + resume_gc() + server.join(timeout=5) + gc.collect() + + +class TestPauseAccounting: + """The pause must never be left on, and never lifted early.""" + + def test_nests(self): + """Only the outermost resume re-enables collection.""" + pause_gc() + pause_gc() + try: + assert not gc.isenabled() + resume_gc() + assert not gc.isenabled() + finally: + resume_gc() + assert gc.isenabled() + + def test_unbalanced_resume_is_a_no_op(self): + """A stray resume cannot enable collection during a live session.""" + resume_gc() + pause_gc() + try: + assert not gc.isenabled() + finally: + resume_gc() + + def test_user_disabled_gc_is_not_re_enabled(self): + """A caller who disabled collection keeps it disabled.""" + gc.disable() + try: + pause_gc() + resume_gc() + assert not gc.isenabled() + finally: + gc.enable() + + def test_leaked_handle_resumes(self): + """Dropping a handle without closing it releases the pause.""" + _open_h5_paused(_FakeRemoteFile(), lambda fileobj, **kw: object(), "r") + gc.collect() + assert gc.isenabled() + + def test_stranded_cyclic_handle_is_recovered(self): + """A handle leaked inside a cycle is healed by the next remote open.""" + holder = {} + handle = _open_h5_paused(_FakeRemoteFile(), lambda fileobj, **kw: object(), "r") + holder["handle"], holder["self"] = handle, holder # unreachable cycle + del handle, holder + assert not gc.isenabled() + remote_io._gc_collect_after = 0.0 # the valve is rate limited + pause_gc() + resume_gc() + assert gc.isenabled() + + def test_open_failure_releases_pause_and_fileobj(self): + """A failed open leaves neither the pause nor the file object behind.""" + fileobj = _FakeRemoteFile() + + def _raise(*args, **kwargs): + raise ValueError("no") + + with pytest.raises(ValueError): + _open_h5_paused(fileobj, _raise, "r") + assert gc.isenabled() + assert fileobj.closed + + def test_teardown_error_still_resumes(self): + """A BaseException from the owned file object cannot strand the pause.""" + + class _BadClose: + def close(self): + raise KeyboardInterrupt("interrupted mid-close") + + class _Handle: + def close(self): + pass + + pause_gc() + managed = _ManagedH5pyFile(_Handle(), _BadClose(), gc_paused=True) + with suppress(KeyboardInterrupt): + managed.close() + assert gc.isenabled() + + def test_inherited_handle_does_not_resume_after_fork(self): + """A handle carried through a fork cannot resume the child's session.""" + pause_gc() + try: + + class _Handle: + def close(self): + pass + + managed = _ManagedH5pyFile(_Handle(), None, gc_paused=True) + managed._gc_paused_pid = os.getpid() + 1 # as if inherited + managed.close() + assert not gc.isenabled(), "an inherited close stole a live pause" + finally: + resume_gc() + + def test_fork_reset_clears_inherited_pause(self): + """The fork hook drops a pause no child close can rebalance.""" + pause_gc() + try: + remote_io._reset_gc_pause_state() + assert gc.isenabled() + assert remote_io._gc_pause_depth == 0 + except BaseException: # pragma: no cover - only on failure + resume_gc() + raise + + +class TestLoopBackedDetection: + """Missing a loop-backed object leaves the deadlock window open.""" + + def test_async_filesystem_detected(self): + """A file object over an async filesystem is loop backed.""" + assert _is_loop_backed_fileobj(_FakeRemoteFile()) + + def test_wrapped_async_filesystem_detected(self): + """Buffering wrappers hide the filesystem but not the loop thread.""" + wrapped = io.BufferedReader(_FakeRemoteFile()) + assert _is_loop_backed_fileobj(wrapped) + + def test_plain_stream_is_not_loop_backed(self): + """Local and in-memory streams must not pause collection.""" + assert not _is_loop_backed_fileobj(io.BytesIO(b"abc")) + + def test_sync_filesystem_is_not_loop_backed(self): + """A synchronous fsspec filesystem needs no pause.""" + + class _SyncFile: + fs = _FakeFS(False) + + assert not _is_loop_backed_fileobj(_SyncFile()) diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 0f2b6f995..45abb6fd0 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -2,6 +2,7 @@ from __future__ import annotations +import gc import threading from contextlib import closing from io import BufferedReader, BufferedWriter, BytesIO, StringIO, TextIOBase @@ -46,6 +47,32 @@ ) +class _DummyHandle: + """A file-like stand-in that only records being closed.""" + + closed = False + + def close(self): + self.closed = True + + +class _FakeRemoteFile(BytesIO): + """A file object whose fsspec filesystem serves reads on a loop thread.""" + + +def _make_loop_backed_file() -> _FakeRemoteFile: + """Return a file object that takes the loop-backed (GC-paused) branch.""" + from fsspec.asyn import AsyncFileSystem + + class _FakeAsyncFS(AsyncFileSystem): + def __init__(self): + pass + + out = _FakeRemoteFile() + out.fs = _FakeAsyncFS() + return out + + class _BadType: """A dummy type for testing.""" @@ -359,13 +386,6 @@ def test_h5_reader_closes_upath_handle_on_constructor_error( self, tmp_path, monkeypatch ): """Ensure constructor failures close UPath-opened file handles.""" - - class _DummyHandle: - closed = False - - def close(self): - self.closed = True - handle = _DummyHandle() path = UPath(tmp_path / "error.h5") path.write_bytes(b"not an hdf5") @@ -381,33 +401,93 @@ def close(self): H5Reader.get_handle(path) assert handle.closed - def test_h5_reader_uses_small_blocks_for_s3_upath(self, monkeypatch): - """S3-backed HDF5 readers should override s3fs's large default block.""" - - class _DummyHandle: - closed = False - - def close(self): - self.closed = True - + @pytest.mark.parametrize( + ("url", "options", "expected"), + [ + ( + "s3://example-bucket/example.h5", + {"anon": True}, + {"cache_type": "readahead"}, + ), + ( + "http://example.com/example.h5", + {}, + {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}}, + ), + ], + ) + def test_remote_h5_open_kwargs_are_tuned(self, monkeypatch, url, options, expected): + """Remote HDF5 opens must override backend defaults that overfetch. + + s3fs defaults to 50 MB readahead blocks. HTTP instead needs a block + LRU: the h5py metadata probe alternates between the file header and + footer, and a single-window cache refetches on every jump. + """ opened = {} - handle = _DummyHandle() - path = UPath("s3://example-bucket/example.h5", anon=True) + path = UPath(url, **options) def _open(_self, _mode, **kwargs): opened.update(kwargs) - return handle + return _DummyHandle() monkeypatch.setattr(type(path), "open", _open) monkeypatch.setattr( H5Reader, "constructor", - staticmethod(lambda *args, **kwargs: object()), + staticmethod(lambda *args, **kwargs: _DummyHandle()), ) with config_context(remote_hdf5_block_size=1234): - H5Reader.get_handle(path) + H5Reader.get_handle(path).close() assert opened["block_size"] == 1234 - assert opened["cache_type"] == "readahead" + for key, value in expected.items(): + assert opened[key] == value + + def test_remote_h5_handle_pauses_gc(self, monkeypatch): + """Automatic collection stays paused while a remote handle is open.""" + path = UPath("http://example.com/gc-pause.h5") + monkeypatch.setattr(type(path), "open", lambda *a, **k: _DummyHandle()) + monkeypatch.setattr( + H5Reader, + "constructor", + staticmethod(lambda *args, **kwargs: _DummyHandle()), + ) + assert gc.isenabled() + handle = H5Reader.get_handle(path) + try: + assert not gc.isenabled() + finally: + handle.close() + assert gc.isenabled() + # Closing twice must not unbalance the pause bookkeeping. + handle.close() + assert gc.isenabled() + + def test_loop_backed_fileobj_pauses_gc(self, monkeypatch): + """User-supplied fsspec async file objects need the GC pause too.""" + monkeypatch.setattr( + H5Reader, + "constructor", + staticmethod(lambda *args, **kwargs: BytesIO()), + ) + assert gc.isenabled() + handle = H5Reader.get_handle(_make_loop_backed_file()) + try: + assert not gc.isenabled() + finally: + handle.close() + assert gc.isenabled() + + def test_loop_backed_fileobj_constructor_error_resumes_gc(self, monkeypatch): + """A failed h5py construction must rebalance the GC pause.""" + + def _explode(*args, **kwargs): + raise ValueError("not an hdf5 fileobj") + + monkeypatch.setattr(H5Reader, "constructor", staticmethod(_explode)) + assert gc.isenabled() + with pytest.raises(ValueError, match="not an hdf5 fileobj"): + H5Reader.get_handle(_make_loop_backed_file()) + assert gc.isenabled() def test_h5_writer_to_remote_upath(self): """HDF5 writers should create remote UPath files via write-back.""" @@ -534,6 +614,57 @@ def test_get_none_resource_returns_source(self): with IOResourceManager(source) as man: assert man.get_resource(None) is source + def test_error_in_context_aborts_handles(self): + """An exception inside the context must abort, not commit, handles.""" + + class _Recorder: + aborted = False + closed = False + + def abort(self): + self.aborted = True + + def close(self): + self.closed = True + + recorder = _Recorder() + man = IOResourceManager("unused") + man._cache["key"] = recorder + with pytest.raises(ValueError, match="boom"): + with man: + raise ValueError("boom") + assert recorder.aborted + assert not recorder.closed + # A clean exit closes normally. + recorder2 = _Recorder() + man2 = IOResourceManager("unused") + man2._cache["key"] = recorder2 + with man2: + pass + assert recorder2.closed + assert not recorder2.aborted + + def test_close_all_survives_failing_handle(self): + """One handle raising must not skip cleanup of the others.""" + + class _Exploder: + def close(self): + raise OSError("close failed") + + class _Recorder: + closed = False + + def close(self): + self.closed = True + + recorder = _Recorder() + man = IOResourceManager("unused") + man._cache["bad"] = _Exploder() + man._cache["good"] = recorder + with pytest.raises(OSError, match="close failed"): + man.close_all() + assert recorder.closed + def test_non_pathlike_resource_passthrough(self): """Non-pathlike resources should bypass path coercion entirely.""" source = BytesIO(b"abc") @@ -922,6 +1053,12 @@ def test_no_range_error_predicate_matches_expected_message(self): assert not is_no_range_http_error(ValueError("different error")) assert not is_no_range_http_error(RuntimeError("range requests")) + def test_no_range_error_predicate_matches_streaming_seek(self): + """Seeking a streaming (size-less) HTTP file needs the same fallback.""" + exc = ValueError("Cannot seek streaming HTTP file") + assert is_no_range_http_error(exc) + assert not is_no_range_http_error(RuntimeError(str(exc))) + class TestFallbackFileObj: """Tests for switching failed remote handles to local cache files.""" From e941f932b27fd3097a776c09c200efd75ef0c68e Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:10:23 +0200 Subject: [PATCH 02/13] Simplify remote-IO changes and stop pausing gc for sync backends - Pause collection only for loop-backed (async fsspec) resources; local and memory UPaths took the remote branch and paused process-wide gc for every HDF5 read. - Fold the two fileobj-driver opens into one helper; keep a caller-supplied file object usable after a failed open, which get_format relies on. - Flatten the tuned-open-kwargs branch, the no-range predicate, and the type-caster error path. --- dascore/io/core.py | 10 ++-- dascore/utils/hdf5.py | 96 ++++++++++++++++++------------- dascore/utils/remote_io.py | 16 ++++-- tests/test_utils/test_gc_pause.py | 33 +++++------ tests/test_utils/test_io_utils.py | 42 +++++++++----- 5 files changed, 115 insertions(+), 82 deletions(-) diff --git a/dascore/io/core.py b/dascore/io/core.py index 229ee9887..58cf774aa 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -868,19 +868,17 @@ def _wrapper(*args, _pre_cast=False, **kwargs): # kwargs is included in bound arguments, need to re-attach new_kw.update(new_kw.pop("kwargs", {})) out = func(**new_kw) - except BaseException as e: # get_format can't raise; must return false. + except BaseException as e: # A handle created here must close even on failure, including on # KeyboardInterrupt; leaking a remote handle would leave garbage # collection paused for as long as the traceback is retained. if new_resource is not None and new_resource is not resource: with suppress(Exception): getattr(new_resource, "close", lambda: None)() - if not isinstance(e, Exception): + # get_format can't raise; it must return False instead. + if fun_name != "get_format" or not isinstance(e, Exception): raise - if fun_name == "get_format": - out = False - else: - raise e + out = False else: # if a new file handle was created we need to close it now. But it # shouldn't close any passed in, that should happen up the stack. diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index 5bb821377..5a8b7fe09 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -119,13 +119,14 @@ def __getattr__(self, item): return getattr(self._handle, item) -def _is_loop_backed_fileobj(resource) -> bool: +def _is_loop_backed(resource) -> bool: """ - Return True when a file object's filesystem serves reads via a loop thread. + Return True when reads on a resource are served by an event-loop thread. fsspec async filesystems (http, s3, ...) bridge each read onto a shared event-loop thread and mark themselves with ``async_impl``; duck-typing it - avoids importing fsspec here. Buffering wrappers hide the filesystem, so + avoids importing fsspec here. Local, memory, and other synchronous + backends need no GC pause. Buffering wrappers hide the filesystem, so unwrap a few levels: missing one would leave the deadlock window open. """ for _ in range(4): @@ -138,22 +139,33 @@ def _is_loop_backed_fileobj(resource) -> bool: return False -def _open_h5_paused(fileobj, constructor, mode) -> _ManagedH5pyFile: +def _open_h5_fileobj( + fileobj, constructor, mode, *, pause: bool, close_on_error: bool +) -> _ManagedH5pyFile: """ - Open a loop-backed file object with h5py while automatic gc is paused. + Open a file object with h5py through the fileobj driver. + + Loop-backed resources pause automatic collection first; the returned + wrapper owns that pause and releases it on close, and a failed open + releases it here. - The returned wrapper owns the pause and releases it on close; if the open - fails, both the pause and the file object are released here. + ``close_on_error`` is set only for file objects DASCore created. A + caller-supplied one must survive a failed open: ``get_format`` offers the + same object to every FiberIO in turn, and an HDF5 miss is the expected + outcome for most of them. """ - pause_gc() + if pause: + pause_gc() try: handle = constructor(fileobj, mode=mode, driver="fileobj") - return _ManagedH5pyFile(handle, fileobj, gc_paused=True) except BaseException: - with suppress(Exception): - fileobj.close() - resume_gc() + if close_on_error: + with suppress(Exception): + fileobj.close() + if pause: + resume_gc() raise + return _ManagedH5pyFile(handle, fileobj, gc_paused=pause) def get_h5py_file(handle) -> H5pyFile: @@ -206,10 +218,13 @@ def open_h5_resource( # A user-supplied fsspec file object delegates reads to the same # event-loop thread as the UPath branch below and needs the same # GC pause; plain local/in-memory streams do not. - if _is_loop_backed_fileobj(resource): - return _open_h5_paused(resource, constructor, mode) - handle = constructor(resource, mode=mode, driver="fileobj") - return _ManagedH5pyFile(handle, resource) + return _open_h5_fileobj( + resource, + constructor, + mode, + pause=_is_loop_backed(resource), + close_on_error=False, + ) if isinstance(resource, UPath): # Reuse an already-materialized local artifact when present so later # HDF5 reads do not re-enter the remote fallback path unnecessarily. @@ -224,16 +239,22 @@ def open_h5_resource( # intercepts UPath targets with its temp-file write-back handle. file_mode = "rb" if mode == "r" else "r+b" open_kwargs = open_kwargs_getter(resource) - # h5py holds its global lock while blocking on fsspec's event-loop - # thread for remote fetches; an automatic garbage collection on that - # thread deallocating h5py objects then deadlocks on the same lock. - # Pause collection for the handle's lifetime (resumed in close()). handle = _FallbackFileObj( remote_opener=lambda: resource.open(file_mode, **open_kwargs), local_opener=lambda: ensure_local_file(resource).open(file_mode), error_predicate=is_no_range_http_error, ) - return _open_h5_paused(handle, constructor, mode) + # h5py holds its global lock while blocking on fsspec's event-loop + # thread for remote fetches; an automatic garbage collection on that + # thread deallocating h5py objects then deadlocks on the same lock. + # Pause collection for the handle's lifetime (resumed in close()). + return _open_h5_fileobj( + handle, + constructor, + mode, + pause=_is_loop_backed(resource), + close_on_error=True, + ) try: if mode != "r": _maybe_make_parent_directory(resource) @@ -266,26 +287,19 @@ class H5Reader(_H5CasterBase): def _get_open_kwargs(resource: UPath) -> dict[str, object]: """Return backend-specific kwargs for remote HDF5 file objects.""" protocol = getattr(resource, "protocol", None) - if protocol in remote_hdf5_tuned_protocols: - # h5py performs many small seeks while opening HDF5 metadata. - # s3fs defaults to 50 MB readahead blocks, which can pull most of - # a large remote file just to satisfy metadata probes. - # HTTP needs a block LRU cache instead: the metadata probe - # alternates between the file header and footer, and fsspec's - # default single-window cache refetches a full block (or the whole - # file on range-less servers) on every jump. - if protocol in http_protocols: - # Bound retained memory: 8 blocks of the configured size. - return { - "block_size": get_config().remote_hdf5_block_size, - "cache_type": "blockcache", - "cache_options": {"maxblocks": 8}, - } - return { - "block_size": get_config().remote_hdf5_block_size, - "cache_type": "readahead", - } - return {} + if protocol not in remote_hdf5_tuned_protocols: + return {} + # h5py performs many small seeks while opening HDF5 metadata, and + # remote backends default to large readahead blocks (s3fs uses 50 MB) + # which can pull most of a file just to satisfy those probes. + out = {"block_size": get_config().remote_hdf5_block_size} + if protocol not in http_protocols: + return out | {"cache_type": "readahead"} + # HTTP needs a block LRU instead: the probe alternates between the + # file header and footer, and fsspec's default single-window cache + # refetches a full block (or the whole file on range-less servers) on + # every jump. Eight blocks keeps both ends resident and bounds memory. + return out | {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}} @classmethod def get_handle(cls, resource): diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index ac705473e..d7b56fcbd 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -28,6 +28,7 @@ "doesn't appear to support range requests", "only reading this file from the beginning is supported", ) +_NO_SIZE_HTTP_PATTERN = "cannot seek streaming http file" _REMOTE_RESOURCE_CACHE: dict[str, UPath] = {} # One lock per cached resource, so two threads never download the same # file at once while unrelated downloads still run together. Entries are @@ -41,6 +42,9 @@ _gc_pause_depth = 0 _gc_was_enabled = False _gc_collect_after = 0.0 +# Seconds between safety-valve collections; a full collect is O(live objects), +# measured at 7-160 ms here, so it must not run on every remote open. +_GC_COLLECT_INTERVAL = 10.0 def pause_gc() -> None: @@ -62,7 +66,6 @@ def pause_gc() -> None: recovered by the collection below, on the next remote open. """ global _gc_pause_depth, _gc_was_enabled, _gc_collect_after - now = time.monotonic() with _gc_pause_lock: # Count first: an interrupt before ``disable`` leaves a pending resume # (harmless); the reverse order could leave gc off with none pending. @@ -71,9 +74,10 @@ def pause_gc() -> None: _gc_was_enabled = gc.isenabled() gc.disable() # Claim the valve here so concurrent openers cannot each run it. + now = time.monotonic() collect = now >= _gc_collect_after if collect: - _gc_collect_after = now + 10.0 + _gc_collect_after = now + _GC_COLLECT_INTERVAL # Safety valve: finalize cyclic garbage, including a handle leaked by an # earlier session, so a stranded pause cannot disable collection forever. # Must run outside the lock, since finalizing such a handle calls @@ -325,12 +329,12 @@ def get_local_handle(resource, opener): def is_no_range_http_error(exc: Exception) -> bool: """Return True when an exception indicates no-range HTTP random access.""" - message = str(exc).lower() if not isinstance(exc, ValueError): return False - # Servers with no reported size stream instead of ranging; seeking such - # a streaming file needs the same local-file fallback as no-range access. - if "cannot seek streaming http file" in message: + message = str(exc).lower() + # A server reporting no size streams instead of ranging; seeking such a + # file needs the same local-file fallback as an outright no-range server. + if _NO_SIZE_HTTP_PATTERN in message: return True return all(pattern in message for pattern in _NO_RANGE_HTTP_PATTERNS) diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py index d70d66589..31c95ecf9 100644 --- a/tests/test_utils/test_gc_pause.py +++ b/tests/test_utils/test_gc_pause.py @@ -12,9 +12,9 @@ import dascore.utils.remote_io as remote_io from dascore.utils.hdf5 import ( - _is_loop_backed_fileobj, + _is_loop_backed, _ManagedH5pyFile, - _open_h5_paused, + _open_h5_fileobj, ) from dascore.utils.remote_io import pause_gc, resume_gc @@ -45,6 +45,11 @@ def readable(self): return True +def _open_paused(fileobj, constructor=lambda fileobj, **kwargs: object()): + """Open a fake loop-backed file object the way the UPath branch does.""" + return _open_h5_fileobj(fileobj, constructor, "r", pause=True, close_on_error=True) + + class TestDeadlockProperty: """The pause must actually prevent the h5py/loop-thread deadlock.""" @@ -130,14 +135,14 @@ def test_user_disabled_gc_is_not_re_enabled(self): def test_leaked_handle_resumes(self): """Dropping a handle without closing it releases the pause.""" - _open_h5_paused(_FakeRemoteFile(), lambda fileobj, **kw: object(), "r") + _open_paused(_FakeRemoteFile()) gc.collect() assert gc.isenabled() def test_stranded_cyclic_handle_is_recovered(self): """A handle leaked inside a cycle is healed by the next remote open.""" holder = {} - handle = _open_h5_paused(_FakeRemoteFile(), lambda fileobj, **kw: object(), "r") + handle = _open_paused(_FakeRemoteFile()) holder["handle"], holder["self"] = handle, holder # unreachable cycle del handle, holder assert not gc.isenabled() @@ -154,7 +159,7 @@ def _raise(*args, **kwargs): raise ValueError("no") with pytest.raises(ValueError): - _open_h5_paused(fileobj, _raise, "r") + _open_paused(fileobj, _raise) assert gc.isenabled() assert fileobj.closed @@ -194,13 +199,9 @@ def close(self): def test_fork_reset_clears_inherited_pause(self): """The fork hook drops a pause no child close can rebalance.""" pause_gc() - try: - remote_io._reset_gc_pause_state() - assert gc.isenabled() - assert remote_io._gc_pause_depth == 0 - except BaseException: # pragma: no cover - only on failure - resume_gc() - raise + remote_io._reset_gc_pause_state() + assert gc.isenabled() + assert remote_io._gc_pause_depth == 0 class TestLoopBackedDetection: @@ -208,16 +209,16 @@ class TestLoopBackedDetection: def test_async_filesystem_detected(self): """A file object over an async filesystem is loop backed.""" - assert _is_loop_backed_fileobj(_FakeRemoteFile()) + assert _is_loop_backed(_FakeRemoteFile()) def test_wrapped_async_filesystem_detected(self): """Buffering wrappers hide the filesystem but not the loop thread.""" wrapped = io.BufferedReader(_FakeRemoteFile()) - assert _is_loop_backed_fileobj(wrapped) + assert _is_loop_backed(wrapped) def test_plain_stream_is_not_loop_backed(self): """Local and in-memory streams must not pause collection.""" - assert not _is_loop_backed_fileobj(io.BytesIO(b"abc")) + assert not _is_loop_backed(io.BytesIO(b"abc")) def test_sync_filesystem_is_not_loop_backed(self): """A synchronous fsspec filesystem needs no pause.""" @@ -225,4 +226,4 @@ def test_sync_filesystem_is_not_loop_backed(self): class _SyncFile: fs = _FakeFS(False) - assert not _is_loop_backed_fileobj(_SyncFile()) + assert not _is_loop_backed(_SyncFile()) diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 45abb6fd0..9389f25dc 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -10,6 +10,7 @@ import h5py import pytest +from fsspec.asyn import AsyncFileSystem from upath import UPath import dascore as dc @@ -56,21 +57,17 @@ def close(self): self.closed = True -class _FakeRemoteFile(BytesIO): - """A file object whose fsspec filesystem serves reads on a loop thread.""" +class _FakeAsyncFS(AsyncFileSystem): + """A stand-in async fsspec filesystem which needs no event loop.""" + def __init__(self): + pass -def _make_loop_backed_file() -> _FakeRemoteFile: - """Return a file object that takes the loop-backed (GC-paused) branch.""" - from fsspec.asyn import AsyncFileSystem - class _FakeAsyncFS(AsyncFileSystem): - def __init__(self): - pass +class _FakeRemoteFile(BytesIO): + """A file object whose fsspec filesystem serves reads on a loop thread.""" - out = _FakeRemoteFile() - out.fs = _FakeAsyncFS() - return out + fs = _FakeAsyncFS() class _BadType: @@ -462,6 +459,25 @@ def test_remote_h5_handle_pauses_gc(self, monkeypatch): handle.close() assert gc.isenabled() + def test_failed_open_leaves_caller_fileobj_usable(self): + """A caller's file object must survive a failed HDF5 open. + + get_format hands the same object to every FiberIO in turn, so an + HDF5 miss cannot close it out from under the next one. + """ + buffer = BytesIO(b"not an hdf5 file") + with pytest.raises(OSError): + H5Reader.get_handle(buffer) + assert not buffer.closed + + def test_local_upath_does_not_pause_gc(self, generic_hdf5): + """A synchronous backend has no loop thread, so it must not pause gc.""" + handle = H5Reader.get_handle(UPath(generic_hdf5)) + try: + assert gc.isenabled() + finally: + handle.close() + def test_loop_backed_fileobj_pauses_gc(self, monkeypatch): """User-supplied fsspec async file objects need the GC pause too.""" monkeypatch.setattr( @@ -470,7 +486,7 @@ def test_loop_backed_fileobj_pauses_gc(self, monkeypatch): staticmethod(lambda *args, **kwargs: BytesIO()), ) assert gc.isenabled() - handle = H5Reader.get_handle(_make_loop_backed_file()) + handle = H5Reader.get_handle(_FakeRemoteFile()) try: assert not gc.isenabled() finally: @@ -486,7 +502,7 @@ def _explode(*args, **kwargs): monkeypatch.setattr(H5Reader, "constructor", staticmethod(_explode)) assert gc.isenabled() with pytest.raises(ValueError, match="not an hdf5 fileobj"): - H5Reader.get_handle(_make_loop_backed_file()) + H5Reader.get_handle(_FakeRemoteFile()) assert gc.isenabled() def test_h5_writer_to_remote_upath(self): From e459249904f044c69c2a3735a9fc3053d8675f66 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:15:15 +0200 Subject: [PATCH 03/13] Document the remote-read gc pause and reuse the shared CI OS matrix --- .github/workflows/runtests.yml | 4 +++- docs/tutorial/remote_patches.qmd | 8 ++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/.github/workflows/runtests.yml b/.github/workflows/runtests.yml index 5d4b390ca..8ad9dc169 100644 --- a/.github/workflows/runtests.yml +++ b/.github/workflows/runtests.yml @@ -154,7 +154,9 @@ jobs: strategy: fail-fast: false matrix: - os: [ubuntu-latest, macos-latest, windows-latest] + # The same OS list as test_code; the deadlock these tests cover was + # platform-agnostic, so every supported platform should exercise it. + os: ${{ fromJson(needs.setup.outputs.os-matrix) }} # Keep remote-IO coverage visible without blocking unrelated changes. if: github.event_name == 'push' || !contains(github.event.pull_request.labels.*.name, 'no_ci') diff --git a/docs/tutorial/remote_patches.qmd b/docs/tutorial/remote_patches.qmd index 62c01495d..f51e22618 100644 --- a/docs/tutorial/remote_patches.qmd +++ b/docs/tutorial/remote_patches.qmd @@ -136,3 +136,11 @@ The same distinction applies to spools: - `spool.get_contents()` relies on summary metadata - `spool[0]` or iteration loads patch data - if a remote backend cannot answer the metadata side without a download, use `allow_remote_cache_for_metadata=True` + +## Garbage Collection During Remote HDF5 Reads + +While a remote HDF5 handle is open, DASCore pauses Python's automatic garbage collection for the whole process, then restores it when the last such handle closes. + +This avoids a deadlock: h5py holds a process-global lock while waiting on the event loop thread [fsspec](https://filesystem-spec.readthedocs.io/) uses for each remote fetch, and an automatic collection running on that thread needs the same lock to deallocate a dead h5py object. + +Reference counting is unaffected, so most objects are still freed immediately; only cyclic garbage accumulates, and only until the handle closes. Reads of local files and of synchronous backends such as `memory://` are unaffected. From 50c4b325a6427d7e2e6c1ee883a00faee127e440 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:18:53 +0200 Subject: [PATCH 04/13] Address adversarial review: interrupt-safe safety collect, no masked errors - Run the rate-limited safety collection before the pause is taken, so an interrupt inside gc.collect cannot strand a pause nobody will resume. - Keep a failed abort from replacing the exception that triggered it. - Treat a UPath whose backend is not installed as not loop backed rather than raising while deciding whether to pause. --- dascore/utils/hdf5.py | 8 +++++++- dascore/utils/io.py | 9 ++++++++- dascore/utils/remote_io.py | 31 +++++++++++++++++++------------ tests/test_utils/test_gc_pause.py | 23 +++++++++++++++++++++++ tests/test_utils/test_io_utils.py | 14 ++++++++++++++ 5 files changed, 71 insertions(+), 14 deletions(-) diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index 5a8b7fe09..f67a7287e 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -130,7 +130,13 @@ def _is_loop_backed(resource) -> bool: unwrap a few levels: missing one would leave the deadlock window open. """ for _ in range(4): - if getattr(getattr(resource, "fs", None), "async_impl", False): + try: + fs = getattr(resource, "fs", None) + except Exception: + # A UPath whose backend is not installed raises here; the open + # which follows reports that properly, so just skip the pause. + return False + if getattr(fs, "async_impl", False): return True wrapped = getattr(resource, "raw", None) or getattr(resource, "buffer", None) if wrapped is None or wrapped is resource: diff --git a/dascore/utils/io.py b/dascore/utils/io.py index b51f8a53e..cd767b8b1 100644 --- a/dascore/utils/io.py +++ b/dascore/utils/io.py @@ -293,7 +293,14 @@ def __enter__(self): def __exit__(self, exc_type, exc_val, exc_tb): """Close all handles; on error, abort uncommitted writes instead.""" - self.close_all(abort=exc_type is not None) + if exc_type is None: + self.close_all() + return + try: + self.close_all(abort=True) + except Exception as cleanup_error: + # A cleanup failure must not replace the error which caused it. + exc_val.add_note(f"Aborting IO resources also failed: {cleanup_error!r}") def __del__(self): with suppress(Exception): diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index d7b56fcbd..0120ce475 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -47,6 +47,17 @@ _GC_COLLECT_INTERVAL = 10.0 +def _claim_safety_collect() -> bool: + """Return True when this caller wins the rate-limited safety collection.""" + global _gc_collect_after + with _gc_pause_lock: + now = time.monotonic() + if now < _gc_collect_after: + return False + _gc_collect_after = now + _GC_COLLECT_INTERVAL + return True + + def pause_gc() -> None: """ Pause automatic garbage collection for one remote read session. @@ -65,7 +76,14 @@ def pause_gc() -> None: closed keeps it paused; one which is dropped inside a reference cycle is recovered by the collection below, on the next remote open. """ - global _gc_pause_depth, _gc_was_enabled, _gc_collect_after + global _gc_pause_depth, _gc_was_enabled + # Safety valve: finalize cyclic garbage, including a handle leaked by an + # earlier session, so a stranded pause cannot disable collection forever. + # It runs before this pause is taken, so an interrupt during the collect + # cannot strand one, and outside the lock, since finalizing such a handle + # calls resume_gc, which takes it. + if _claim_safety_collect(): + gc.collect() with _gc_pause_lock: # Count first: an interrupt before ``disable`` leaves a pending resume # (harmless); the reverse order could leave gc off with none pending. @@ -73,17 +91,6 @@ def pause_gc() -> None: if _gc_pause_depth == 1: _gc_was_enabled = gc.isenabled() gc.disable() - # Claim the valve here so concurrent openers cannot each run it. - now = time.monotonic() - collect = now >= _gc_collect_after - if collect: - _gc_collect_after = now + _GC_COLLECT_INTERVAL - # Safety valve: finalize cyclic garbage, including a handle leaked by an - # earlier session, so a stranded pause cannot disable collection forever. - # Must run outside the lock, since finalizing such a handle calls - # resume_gc, which takes it. Rate limited; a full collect is O(live). - if collect: - gc.collect() def resume_gc() -> None: diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py index 31c95ecf9..6ad71f3b5 100644 --- a/tests/test_utils/test_gc_pause.py +++ b/tests/test_utils/test_gc_pause.py @@ -45,6 +45,11 @@ def readable(self): return True +def _raise_keyboard_interrupt(*args, **kwargs): + """Stand in for an interrupt landing inside an uninterruptible step.""" + raise KeyboardInterrupt("interrupted") + + def _open_paused(fileobj, constructor=lambda fileobj, **kwargs: object()): """Open a fake loop-backed file object the way the UPath branch does.""" return _open_h5_fileobj(fileobj, constructor, "r", pause=True, close_on_error=True) @@ -151,6 +156,14 @@ def test_stranded_cyclic_handle_is_recovered(self): resume_gc() assert gc.isenabled() + def test_interrupted_safety_collect_takes_no_pause(self, monkeypatch): + """An interrupt during the safety collect cannot strand a pause.""" + monkeypatch.setattr(remote_io.gc, "collect", _raise_keyboard_interrupt) + remote_io._gc_collect_after = 0.0 # the valve is rate limited + with pytest.raises(KeyboardInterrupt): + pause_gc() + assert gc.isenabled() + def test_open_failure_releases_pause_and_fileobj(self): """A failed open leaves neither the pause nor the file object behind.""" fileobj = _FakeRemoteFile() @@ -220,6 +233,16 @@ def test_plain_stream_is_not_loop_backed(self): """Local and in-memory streams must not pause collection.""" assert not _is_loop_backed(io.BytesIO(b"abc")) + def test_unavailable_backend_is_not_loop_backed(self): + """A path whose backend is not installed raises on ``fs``, not here.""" + + class _MissingBackend: + @property + def fs(self): + raise ImportError("please install some-fs") + + assert not _is_loop_backed(_MissingBackend()) + def test_sync_filesystem_is_not_loop_backed(self): """A synchronous fsspec filesystem needs no pause.""" diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 9389f25dc..7c962114c 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -660,6 +660,20 @@ def close(self): assert recorder2.closed assert not recorder2.aborted + def test_failed_abort_does_not_mask_original_error(self): + """A cleanup failure must not replace the error which caused it.""" + + class _BadAbort: + def abort(self): + raise OSError("abort failed") + + man = IOResourceManager("unused") + man._cache["key"] = _BadAbort() + with pytest.raises(ValueError, match="boom") as exc_info: + with man: + raise ValueError("boom") + assert any("abort failed" in note for note in exc_info.value.__notes__) + def test_close_all_survives_failing_handle(self): """One handle raising must not skip cleanup of the others.""" From 90c0d8401e6c8169c7ec5c87a680f0088682acb5 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:20:55 +0200 Subject: [PATCH 05/13] Address bot review comments - Keep the resume balanced when closing a fileobj raises a BaseException. - Cover https alongside http in the tuned-open-kwargs test. - Correct the handle-ownership docstring. --- dascore/utils/hdf5.py | 19 ++++++++++++------- tests/test_utils/test_io_utils.py | 5 +++++ 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index f67a7287e..82e699710 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -57,8 +57,9 @@ class _ManagedH5pyFile: For path-backed opens, this wrapper owns only the h5py handle. For ``h5py.File(..., driver="fileobj")`` paths, it also owns the Python - file-like object DASCore created on behalf of the caller. ``close()`` is - therefore the point where DASCore tears down the entire HDF5 access stack. + file-like object underneath, whether DASCore created it or the caller + supplied it. ``close()`` is therefore the point where DASCore tears down + the entire HDF5 access stack. """ def __init__(self, handle: H5pyFile, owned_fileobj=None, gc_paused=False): @@ -165,11 +166,15 @@ def _open_h5_fileobj( try: handle = constructor(fileobj, mode=mode, driver="fileobj") except BaseException: - if close_on_error: - with suppress(Exception): - fileobj.close() - if pause: - resume_gc() + # Nested so nothing raised while closing, including a BaseException, + # can skip the resume and strand the pause. + try: + if close_on_error: + with suppress(Exception): + fileobj.close() + finally: + if pause: + resume_gc() raise return _ManagedH5pyFile(handle, fileobj, gc_paused=pause) diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 7c962114c..8e54c5964 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -411,6 +411,11 @@ def test_h5_reader_closes_upath_handle_on_constructor_error( {}, {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}}, ), + ( + "https://example.com/example.h5", + {}, + {"cache_type": "blockcache", "cache_options": {"maxblocks": 8}}, + ), ], ) def test_remote_h5_open_kwargs_are_tuned(self, monkeypatch, url, options, expected): From 280146fb09b07f9fe99b9c625e9dcafd4bfbf8ab Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:21:41 +0200 Subject: [PATCH 06/13] Scope the changelog gc note to async backends and link the tutorial --- docs/changelog.qmd | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/changelog.qmd b/docs/changelog.qmd index 2aae0a185..eaf69b2ee 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -77,4 +77,4 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f - Removed deprecated compatibility parameters `gauge_multiple`, `lag`, and `notch`; use `step_multiple`, select on lag coordinates after correlation, and `invert`, respectively. - DASCore now requires Python 3.11 or later (was 3.10, which was never covered by CI and reaches end of life in October 2026). - Writing to a remote path no longer commits a partial file when the write fails: `dc.write` (and any `IOResourceManager` context) now aborts uncommitted remote uploads on error instead of closing, which previously uploaded whatever had been written so far. -- Reading a remote HDF5 file over HTTP/S3 pauses Python's automatic garbage collection for the lifetime of the open handle. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the handle closes; cyclic garbage from all threads accumulates until then. +- Reading an HDF5 file from an async fsspec backend (HTTP, S3, ...) pauses Python's automatic garbage collection for the lifetime of the open handle. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the handle closes; cyclic garbage from all threads accumulates until then. Local paths and synchronous backends such as `memory://` are unaffected. See [Working with Remote Patches](tutorial/remote_patches.qmd). From 801cff3c3984b76e84a3319f586a8bb0d32c14de Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 07:32:06 +0200 Subject: [PATCH 07/13] Address adversarial reviews: ownership, abort-on-failure, interrupt safety - __del__ only releases the gc pause; closing there would also close an h5py file or stream the caller still owns. - The type caster aborts a handle it created when the method raises, so a failed remote HDF5 write discards its temp file instead of uploading it. close_all and the caster now share one release_handle helper. - pause_gc keeps its depth consistent when interrupted, and the whole paused region of _open_h5_fileobj sits inside its rebalancing try. - _ManagedH5pyFile gets class-level defaults so a half-built wrapper is still closeable; the fork hook resets the safety-valve deadline. - Narrow the changelog abort note to remote HDF5 writes; drop the .buffer hop from the loop-backed check. --- dascore/io/core.py | 20 ++++++++---- dascore/utils/hdf5.py | 53 ++++++++++++++++++++----------- dascore/utils/io.py | 19 ++++++++--- dascore/utils/remote_io.py | 26 +++++++++------ docs/changelog.qmd | 2 +- tests/test_io/test_io_core.py | 31 ++++++++++++++++++ tests/test_utils/test_io_utils.py | 12 +++++++ 7 files changed, 123 insertions(+), 40 deletions(-) diff --git a/dascore/io/core.py b/dascore/io/core.py index 58cf774aa..e0edaf874 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -60,7 +60,11 @@ RemoteCacheError, UnknownFiberFormatError, ) -from dascore.utils.io import IOResourceManager, get_handle_from_resource +from dascore.utils.io import ( + IOResourceManager, + get_handle_from_resource, + release_handle, +) from dascore.utils.mapping import FrozenDict from dascore.utils.misc import ( _get_install_message, @@ -869,12 +873,14 @@ def _wrapper(*args, _pre_cast=False, **kwargs): new_kw.update(new_kw.pop("kwargs", {})) out = func(**new_kw) except BaseException as e: - # A handle created here must close even on failure, including on - # KeyboardInterrupt; leaking a remote handle would leave garbage - # collection paused for as long as the traceback is retained. + # A handle created here must be released even on failure, + # including on KeyboardInterrupt: leaking a remote handle leaves + # garbage collection paused for as long as the traceback is + # retained. Abort rather than close, so a failed remote write + # discards its temp file instead of uploading a partial one. if new_resource is not None and new_resource is not resource: with suppress(Exception): - getattr(new_resource, "close", lambda: None)() + release_handle(new_resource, abort=True) # get_format can't raise; it must return False instead. if fun_name != "get_format" or not isinstance(e, Exception): raise @@ -882,8 +888,8 @@ def _wrapper(*args, _pre_cast=False, **kwargs): else: # if a new file handle was created we need to close it now. But it # shouldn't close any passed in, that should happen up the stack. - if new_resource is not resource and hasattr(new_resource, "close"): - new_resource.close() + if new_resource is not resource: + release_handle(new_resource) return out # attach the function and required type for later use diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index 82e699710..dcc141e97 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -62,13 +62,18 @@ class _ManagedH5pyFile: the entire HDF5 access stack. """ + # Class defaults, so a half-built instance is still closeable rather than + # falling through __getattr__ to a handle that may not be set yet. + _closed = False + _gc_paused_pid = None + def __init__(self, handle: H5pyFile, owned_fileobj=None, gc_paused=False): self._handle = handle self._owned_fileobj = owned_fileobj - # The pid that paused, so a handle inherited through a fork cannot - # resume a pause the child never took. - self._gc_paused_pid = os.getpid() if gc_paused else None - self._closed = False + if gc_paused: + # The pid that paused, so a handle inherited through a fork + # cannot resume a pause the child never took. + self._gc_paused_pid = os.getpid() def close(self): """Close the h5py file and, when present, the owned file object.""" @@ -91,9 +96,17 @@ def close(self): resume_gc() def __del__(self): - """Backstop close so a leaked handle cannot pause collection forever.""" - with suppress(Exception): - self.close() + """ + Release a leaked handle's pause so it cannot stop collection forever. + + Only the pause: closing here would also close a caller-supplied h5py + file or stream that this wrapper never had permission to close. + Reference counting still tears the underlying handles down. + """ + pid = self.__dict__.pop("_gc_paused_pid", None) + if pid == os.getpid(): + with suppress(Exception): + resume_gc() def __enter__(self): return self @@ -127,10 +140,10 @@ def _is_loop_backed(resource) -> bool: fsspec async filesystems (http, s3, ...) bridge each read onto a shared event-loop thread and mark themselves with ``async_impl``; duck-typing it avoids importing fsspec here. Local, memory, and other synchronous - backends need no GC pause. Buffering wrappers hide the filesystem, so - unwrap a few levels: missing one would leave the deadlock window open. + backends need no GC pause. A buffered reader hides the filesystem behind + ``raw``, so unwrap it: missing it would leave the deadlock window open. """ - for _ in range(4): + while resource is not None: try: fs = getattr(resource, "fs", None) except Exception: @@ -139,10 +152,8 @@ def _is_loop_backed(resource) -> bool: return False if getattr(fs, "async_impl", False): return True - wrapped = getattr(resource, "raw", None) or getattr(resource, "buffer", None) - if wrapped is None or wrapped is resource: - return False - resource = wrapped + wrapped = getattr(resource, "raw", None) + resource = None if wrapped is resource else wrapped return False @@ -161,13 +172,18 @@ def _open_h5_fileobj( same object to every FiberIO in turn, and an HDF5 miss is the expected outcome for most of them. """ - if pause: - pause_gc() try: + if pause: + # Inside the try, and paired unconditionally below, because + # pause_gc always leaves the depth consistent with the pauses it + # took -- even when interrupted partway. + pause_gc() handle = constructor(fileobj, mode=mode, driver="fileobj") + return _ManagedH5pyFile(handle, fileobj, gc_paused=pause) except BaseException: - # Nested so nothing raised while closing, including a BaseException, - # can skip the resume and strand the pause. + # Everything the pause covers runs in here, so an interrupt at any + # point still rebalances. Nested so nothing raised while closing, + # including a BaseException, can skip the resume. try: if close_on_error: with suppress(Exception): @@ -176,7 +192,6 @@ def _open_h5_fileobj( if pause: resume_gc() raise - return _ManagedH5pyFile(handle, fileobj, gc_paused=pause) def get_h5py_file(handle) -> H5pyFile: diff --git a/dascore/utils/io.py b/dascore/utils/io.py index cd767b8b1..08725c877 100644 --- a/dascore/utils/io.py +++ b/dascore/utils/io.py @@ -209,6 +209,20 @@ def get_handle_from_resource(uri, required_type): return uri +def release_handle(handle, abort: bool = False): + """ + Release a file handle, closing it or discarding its uncommitted work. + + Only a few handles can ``abort``; a remote HDF5 writer does, because + closing it uploads whatever was written so far. Everything else is + closed, and a handle with no ``close`` needs no release at all. + """ + if abort and hasattr(handle, "abort"): + handle.abort() + else: + getattr(handle, "close", lambda: None)() + + class IOResourceManager: """ A class for managing opening/closing files. @@ -270,10 +284,7 @@ def close_all(self, abort: bool = False): with self._lock: for handle in self._cache.values(): try: - if abort and hasattr(handle, "abort"): - handle.abort() - else: - getattr(handle, "close", lambda: None)() + release_handle(handle, abort=abort) except BaseException as exc: first_exc = first_exc if first_exc is not None else exc if first_exc is not None: diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 0120ce475..5a1a90a5a 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -69,12 +69,15 @@ def pause_gc() -> None: closes that window; reference counting still frees non-cyclic garbage. Calls nest; every ``pause_gc`` needs one ``resume_gc``. ``_ManagedH5pyFile`` - pairs them with ``close``/``__del__``. + pairs them with ``close``/``__del__``. An interrupted ``pause_gc`` still + leaves the depth consistent with the pauses it took, so a caller which + resumes on any failure stays balanced. The pause is process-global, so cyclic garbage from every thread accumulates until the last remote handle closes. A handle which is never closed keeps it paused; one which is dropped inside a reference cycle is - recovered by the collection below, on the next remote open. + recovered by the collection below, on the next remote open -- if the + program makes one. Otherwise the pause outlives every remote read. """ global _gc_pause_depth, _gc_was_enabled # Safety valve: finalize cyclic garbage, including a handle leaked by an @@ -85,12 +88,15 @@ def pause_gc() -> None: if _claim_safety_collect(): gc.collect() with _gc_pause_lock: - # Count first: an interrupt before ``disable`` leaves a pending resume - # (harmless); the reverse order could leave gc off with none pending. - _gc_pause_depth += 1 - if _gc_pause_depth == 1: - _gc_was_enabled = gc.isenabled() - gc.disable() + # The count must move even if ``disable`` is interrupted, or the depth + # stops matching the live handles: it would never fall back to zero, + # so no later pause would ever disable collection again. + try: + if _gc_pause_depth == 0: + _gc_was_enabled = gc.isenabled() + gc.disable() + finally: + _gc_pause_depth += 1 def resume_gc() -> None: @@ -113,11 +119,13 @@ def _reset_gc_pause_state(): Handles inherited by the child record the pid that paused for them, so closing one there cannot resume a pause this child never took. """ - global _gc_pause_depth, _gc_pause_lock + global _gc_pause_depth, _gc_pause_lock, _gc_collect_after _gc_pause_lock = threading.Lock() if _gc_pause_depth and _gc_was_enabled: gc.enable() _gc_pause_depth = 0 + # The parent's rate limit does not describe this process. + _gc_collect_after = 0.0 @_reinit_after_fork diff --git a/docs/changelog.qmd b/docs/changelog.qmd index eaf69b2ee..f74785d37 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -76,5 +76,5 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f - Removed `dascore.utils.patch.merge_patches`; use `dc.spool(...).chunk(...)`. - Removed deprecated compatibility parameters `gauge_multiple`, `lag`, and `notch`; use `step_multiple`, select on lag coordinates after correlation, and `invert`, respectively. - DASCore now requires Python 3.11 or later (was 3.10, which was never covered by CI and reaches end of life in October 2026). -- Writing to a remote path no longer commits a partial file when the write fails: `dc.write` (and any `IOResourceManager` context) now aborts uncommitted remote uploads on error instead of closing, which previously uploaded whatever had been written so far. +- Writing HDF5 to a remote path no longer commits a partial file when the write fails. A remote HDF5 write buffers into a temp file and uploads it on close, so an error mid-write previously uploaded whatever had been written so far; the temp file is now discarded instead. Other remote writers stream directly and are unchanged. - Reading an HDF5 file from an async fsspec backend (HTTP, S3, ...) pauses Python's automatic garbage collection for the lifetime of the open handle. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the handle closes; cyclic garbage from all threads accumulates until then. Local paths and synchronous backends such as `memory://` are unaffected. See [Working with Remote Patches](tutorial/remote_patches.qmd). diff --git a/tests/test_io/test_io_core.py b/tests/test_io/test_io_core.py index 1800051d6..efe194df9 100644 --- a/tests/test_io/test_io_core.py +++ b/tests/test_io/test_io_core.py @@ -1625,6 +1625,37 @@ def read(self, resource: BinaryReader, **kwargs): _Exploder().read(dummy_text_file) assert recorder.closed + def test_handle_aborted_when_write_raises(self, dummy_text_file, monkeypatch): + """A failed write must discard its handle, not commit a partial file.""" + + class _Recorder: + aborted = False + closed = False + + def abort(self): + self.aborted = True + + def close(self): + self.closed = True + + recorder = _Recorder() + monkeypatch.setattr( + "dascore.io.core.get_handle_from_resource", + lambda resource, required_type: recorder, + ) + + class _BadWriter(FiberIO): + name = "_BadWriterIO" + version = "1" + + def write(self, patch, resource: BinaryWriter, **kwargs): + raise ValueError("mid-write failure") + + with pytest.raises(ValueError, match="mid-write failure"): + _BadWriter().write(None, dummy_text_file) + assert recorder.aborted + assert not recorder.closed + class TestGetSupportedIOTable: """A test for creating the supported io table.""" diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 8e54c5964..43ccfaa8b 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -464,6 +464,18 @@ def test_remote_h5_handle_pauses_gc(self, monkeypatch): handle.close() assert gc.isenabled() + def test_dropped_wrapper_leaves_caller_handle_open(self, generic_hdf5): + """Collecting a wrapper must not close the handle its caller owns.""" + file = h5py.File(generic_hdf5, "r") + try: + wrapper = H5Reader.get_handle(file) + assert wrapper is not file + del wrapper + gc.collect() + assert file # h5py files are falsey once closed + finally: + file.close() + def test_failed_open_leaves_caller_fileobj_usable(self): """A caller's file object must survive a failed HDF5 open. From e361e5ce2b568817d150a546788255a4595e0695 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 08:28:48 +0200 Subject: [PATCH 08/13] Address test review: repair the gc guard, scope it to every test - Move the pause guard to the root conftest and make it repair the state before failing, so one leak blames its own test instead of erroring out every test after it. This also covers the pause tests in test_io_utils. - Patch remote_io's gc reference rather than the stdlib module. - Restore gc conditionally in the user-disabled test. - Add a threaded nest/unnest test; drop a dead daemon_threads assignment. --- tests/conftest.py | 24 +++++++++++++++++++ tests/test_io/conftest.py | 1 - tests/test_utils/test_gc_pause.py | 39 ++++++++++++++++++++++--------- 3 files changed, 52 insertions(+), 12 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 396020883..7e78e83bf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,6 +2,7 @@ from __future__ import annotations +import gc import os import shutil import threading @@ -17,6 +18,7 @@ import dascore as dc import dascore.examples as ex +import dascore.utils.remote_io as remote_io from dascore.compat import random_state from dascore.config import get_config, set_config from dascore.constants import SpoolType @@ -156,6 +158,28 @@ def use_test_config(): yield +@pytest.fixture(autouse=True) +def gc_pause_is_not_leaked(): + """ + Blame the test which strands the remote-read gc pause, and repair it. + + Remote HDF5 handles disable automatic collection process-wide (see + `dascore.utils.remote_io.pause_gc`). Repairing here keeps one leak from + cascading into every later test, which would bury the real failure. + """ + was_enabled = gc.isenabled() + yield + depth = remote_io._gc_pause_depth + if not depth and gc.isenabled() == was_enabled: + return + remote_io._gc_pause_depth = 0 + if was_enabled: + gc.enable() + else: + gc.disable() + pytest.fail(f"test left the remote-read gc pause held (depth={depth})") + + @pytest.fixture(scope="session", autouse=True) def allow_legacy_dasdae_coord_unpickle(): """Test fixtures may rely on trusted historical DASDAE coord payloads.""" diff --git a/tests/test_io/conftest.py b/tests/test_io/conftest.py index d9b4039d4..eea228674 100644 --- a/tests/test_io/conftest.py +++ b/tests/test_io/conftest.py @@ -273,7 +273,6 @@ def _serve_das_tree(handler_cls, root, label): """ handler = partial(handler_cls, directory=str(root)) server = _ThreadingHTTPServer(("127.0.0.1", 0), handler) - server.daemon_threads = True thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py index 6ad71f3b5..3e97f2dcc 100644 --- a/tests/test_utils/test_gc_pause.py +++ b/tests/test_utils/test_gc_pause.py @@ -7,6 +7,7 @@ import os import threading from contextlib import suppress +from types import SimpleNamespace import pytest @@ -19,15 +20,6 @@ from dascore.utils.remote_io import pause_gc, resume_gc -@pytest.fixture(autouse=True) -def _gc_state_is_restored(): - """Fail loudly if a test leaves collection paused.""" - assert remote_io._gc_pause_depth == 0 - yield - assert remote_io._gc_pause_depth == 0, "test left the gc pause held" - assert gc.isenabled(), "test left automatic collection disabled" - - class _FakeFS: """Stand-in for an fsspec filesystem.""" @@ -64,6 +56,9 @@ def test_pause_prevents_loop_thread_deadlock(self): Models the real cycle: a reader holds h5py's global lock and waits on a loop thread, while that thread's collection finalizes an object needing the same lock. Without the pause this deadlocks. + + This calls pause_gc directly; that the HDF5 open paths reach it is + covered by test_remote_h5_handle_pauses_gc and its neighbours. """ phil = threading.RLock() request = threading.Semaphore(0) @@ -130,13 +125,15 @@ def test_unbalanced_resume_is_a_no_op(self): def test_user_disabled_gc_is_not_re_enabled(self): """A caller who disabled collection keeps it disabled.""" + was_enabled = gc.isenabled() gc.disable() try: pause_gc() resume_gc() assert not gc.isenabled() finally: - gc.enable() + if was_enabled: + gc.enable() def test_leaked_handle_resumes(self): """Dropping a handle without closing it releases the pause.""" @@ -158,7 +155,13 @@ def test_stranded_cyclic_handle_is_recovered(self): def test_interrupted_safety_collect_takes_no_pause(self, monkeypatch): """An interrupt during the safety collect cannot strand a pause.""" - monkeypatch.setattr(remote_io.gc, "collect", _raise_keyboard_interrupt) + fake_gc = SimpleNamespace( + collect=_raise_keyboard_interrupt, + isenabled=gc.isenabled, + disable=gc.disable, + enable=gc.enable, + ) + monkeypatch.setattr(remote_io, "gc", fake_gc) remote_io._gc_collect_after = 0.0 # the valve is rate limited with pytest.raises(KeyboardInterrupt): pause_gc() @@ -216,6 +219,20 @@ def test_fork_reset_clears_inherited_pause(self): assert gc.isenabled() assert remote_io._gc_pause_depth == 0 + @pytest.mark.concurrency + def test_concurrent_sessions_keep_the_count_exact(self, run_in_threads): + """Overlapping pauses must not lose or double-count each other.""" + + def _pause_and_resume(_index): + for _ in range(50): + pause_gc() + assert not gc.isenabled() + resume_gc() + + run_in_threads(_pause_and_resume) + assert remote_io._gc_pause_depth == 0 + assert gc.isenabled() + class TestLoopBackedDetection: """Missing a loop-backed object leaves the deadlock window open.""" From d1e407dd382431b28b56b68e40c68de49e439c92 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 08:39:24 +0200 Subject: [PATCH 09/13] Fix the pause tests on platforms without aiohttp or threads - test_remote_h5_handle_pauses_gc built a real HTTP filesystem to decide loop-backedness; wasm and free-threaded CPython have no aiohttp, so it silently took the no-pause branch. Stand in a fake async filesystem. - Mark the deadlock property test as concurrency; it starts a thread, which wasm cannot do. --- tests/test_utils/test_gc_pause.py | 1 + tests/test_utils/test_io_utils.py | 3 +++ 2 files changed, 4 insertions(+) diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py index 3e97f2dcc..49ef96ebb 100644 --- a/tests/test_utils/test_gc_pause.py +++ b/tests/test_utils/test_gc_pause.py @@ -50,6 +50,7 @@ def _open_paused(fileobj, constructor=lambda fileobj, **kwargs: object()): class TestDeadlockProperty: """The pause must actually prevent the h5py/loop-thread deadlock.""" + @pytest.mark.concurrency def test_pause_prevents_loop_thread_deadlock(self): """A collection on the loop thread cannot wedge a lock-holding reader. diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 43ccfaa8b..77d0afee3 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -447,6 +447,9 @@ def _open(_self, _mode, **kwargs): def test_remote_h5_handle_pauses_gc(self, monkeypatch): """Automatic collection stays paused while a remote handle is open.""" path = UPath("http://example.com/gc-pause.h5") + # Stand in for the real filesystem, which needs aiohttp; some + # platforms DASCore supports (wasm, free-threaded) do not have it. + monkeypatch.setattr(type(path), "fs", property(lambda _self: _FakeAsyncFS())) monkeypatch.setattr(type(path), "open", lambda *a, **k: _DummyHandle()) monkeypatch.setattr( H5Reader, From d5bda4c02ecae2296b538ee8e4ce7ffebcff7f86 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 09:36:08 +0200 Subject: [PATCH 10/13] Address CodeRabbit review - Say collection resumes when the last remote handle closes, not the handle. - Restore the safety-valve deadline through monkeypatch. - Report which part of the collection state a test changed. --- docs/changelog.qmd | 2 +- tests/conftest.py | 8 ++++++-- tests/test_utils/test_gc_pause.py | 6 +++--- 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/docs/changelog.qmd b/docs/changelog.qmd index f74785d37..8d0cbf752 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -77,4 +77,4 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f - Removed deprecated compatibility parameters `gauge_multiple`, `lag`, and `notch`; use `step_multiple`, select on lag coordinates after correlation, and `invert`, respectively. - DASCore now requires Python 3.11 or later (was 3.10, which was never covered by CI and reaches end of life in October 2026). - Writing HDF5 to a remote path no longer commits a partial file when the write fails. A remote HDF5 write buffers into a temp file and uploads it on close, so an error mid-write previously uploaded whatever had been written so far; the temp file is now discarded instead. Other remote writers stream directly and are unchanged. -- Reading an HDF5 file from an async fsspec backend (HTTP, S3, ...) pauses Python's automatic garbage collection for the lifetime of the open handle. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the handle closes; cyclic garbage from all threads accumulates until then. Local paths and synchronous backends such as `memory://` are unaffected. See [Working with Remote Patches](tutorial/remote_patches.qmd). +- Reading an HDF5 file from an async fsspec backend (HTTP, S3, ...) pauses Python's automatic garbage collection while at least one such handle is open. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the last such handle closes; cyclic garbage from all threads accumulates until then. Local paths and synchronous backends such as `memory://` are unaffected. See [Working with Remote Patches](tutorial/remote_patches.qmd). diff --git a/tests/conftest.py b/tests/conftest.py index 7e78e83bf..e0b8b901b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -170,14 +170,18 @@ def gc_pause_is_not_leaked(): was_enabled = gc.isenabled() yield depth = remote_io._gc_pause_depth - if not depth and gc.isenabled() == was_enabled: + is_enabled = gc.isenabled() + if not depth and is_enabled == was_enabled: return remote_io._gc_pause_depth = 0 if was_enabled: gc.enable() else: gc.disable() - pytest.fail(f"test left the remote-read gc pause held (depth={depth})") + pytest.fail( + f"test changed collection state: pause depth={depth}, " + f"gc enabled={is_enabled}, expected={was_enabled}" + ) @pytest.fixture(scope="session", autouse=True) diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py index 49ef96ebb..4f28d5fc0 100644 --- a/tests/test_utils/test_gc_pause.py +++ b/tests/test_utils/test_gc_pause.py @@ -142,14 +142,14 @@ def test_leaked_handle_resumes(self): gc.collect() assert gc.isenabled() - def test_stranded_cyclic_handle_is_recovered(self): + def test_stranded_cyclic_handle_is_recovered(self, monkeypatch): """A handle leaked inside a cycle is healed by the next remote open.""" holder = {} handle = _open_paused(_FakeRemoteFile()) holder["handle"], holder["self"] = handle, holder # unreachable cycle del handle, holder assert not gc.isenabled() - remote_io._gc_collect_after = 0.0 # the valve is rate limited + monkeypatch.setattr(remote_io, "_gc_collect_after", 0.0) # rate limited pause_gc() resume_gc() assert gc.isenabled() @@ -163,7 +163,7 @@ def test_interrupted_safety_collect_takes_no_pause(self, monkeypatch): enable=gc.enable, ) monkeypatch.setattr(remote_io, "gc", fake_gc) - remote_io._gc_collect_after = 0.0 # the valve is rate limited + monkeypatch.setattr(remote_io, "_gc_collect_after", 0.0) # rate limited with pytest.raises(KeyboardInterrupt): pause_gc() assert gc.isenabled() From 30e94ccc55206f134f265dfe50257feb45faa78a Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 09:47:19 +0200 Subject: [PATCH 11/13] Tolerate a probe that refuses an attribute in _is_loop_backed --- dascore/utils/hdf5.py | 13 +++++++------ tests/test_utils/test_gc_pause.py | 9 +++++++++ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index dcc141e97..fe729b081 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -145,14 +145,15 @@ def _is_loop_backed(resource) -> bool: """ while resource is not None: try: - fs = getattr(resource, "fs", None) + if getattr(getattr(resource, "fs", None), "async_impl", False): + return True + wrapped = getattr(resource, "raw", None) except Exception: - # A UPath whose backend is not installed raises here; the open - # which follows reports that properly, so just skip the pause. + # Probing can fail rather than return nothing: a UPath whose + # backend is not installed raises from ``fs``, and a wrapper + # can refuse an attribute with something other than + # AttributeError. Either way the open below reports it. return False - if getattr(fs, "async_impl", False): - return True - wrapped = getattr(resource, "raw", None) resource = None if wrapped is resource else wrapped return False diff --git a/tests/test_utils/test_gc_pause.py b/tests/test_utils/test_gc_pause.py index 4f28d5fc0..bfa0fdfee 100644 --- a/tests/test_utils/test_gc_pause.py +++ b/tests/test_utils/test_gc_pause.py @@ -261,6 +261,15 @@ def fs(self): assert not _is_loop_backed(_MissingBackend()) + def test_refused_attribute_is_not_loop_backed(self): + """A wrapper may refuse an attribute with more than AttributeError.""" + + class _Refusing: + def __getattr__(self, name): + raise io.UnsupportedOperation(name) + + assert not _is_loop_backed(_Refusing()) + def test_sync_filesystem_is_not_loop_backed(self): """A synchronous fsspec filesystem needs no pause.""" From adf35fcc0224a37d4f28bec4b3c736de83f1aa25 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 12:01:43 +0200 Subject: [PATCH 12/13] Reference h5py's own documentation of the file-object deadlock h5py documents this failure mode and names disabling collection as one of its two mitigations; say so in the tutorial, the changelog, and pause_gc. --- dascore/utils/remote_io.py | 6 ++++++ docs/changelog.qmd | 2 +- docs/tutorial/remote_patches.qmd | 4 +++- 3 files changed, 10 insertions(+), 2 deletions(-) diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 5a1a90a5a..76dac6581 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -68,6 +68,12 @@ def pause_gc() -> None: threads deadlock. Disabling automatic collection for the handle's lifetime closes that window; reference counting still frees non-cyclic garbage. + h5py documents this deadlock, and disabling collection as one of its two + mitigations, under "Python file-like objects": + https://docs.h5py.org/en/stable/high/file.html#python-file-like-objects + Its other mitigation, avoiding reference cycles which keep h5py objects + alive, is not available to us: the cycle can be anywhere in the process. + Calls nest; every ``pause_gc`` needs one ``resume_gc``. ``_ManagedH5pyFile`` pairs them with ``close``/``__del__``. An interrupted ``pause_gc`` still leaves the depth consistent with the pauses it took, so a caller which diff --git a/docs/changelog.qmd b/docs/changelog.qmd index de528ea8f..17536de7a 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -76,4 +76,4 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f - Removed deprecated compatibility parameters `gauge_multiple`, `lag`, and `notch`; use `step_multiple`, select on lag coordinates after correlation, and `invert`, respectively. - DASCore now requires Python 3.11 or later (was 3.10, which was never covered by CI and reaches end of life in October 2026). - Writing HDF5 to a remote path no longer commits a partial file when the write fails. A remote HDF5 write buffers into a temp file and uploads it on close, so an error mid-write previously uploaded whatever had been written so far; the temp file is now discarded instead. Other remote writers stream directly and are unchanged. -- Reading an HDF5 file from an async fsspec backend (HTTP, S3, ...) pauses Python's automatic garbage collection while at least one such handle is open. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the last such handle closes; cyclic garbage from all threads accumulates until then. Local paths and synchronous backends such as `memory://` are unaffected. See [Working with Remote Patches](tutorial/remote_patches.qmd). +- Reading an HDF5 file from an async fsspec backend (HTTP, S3, ...) pauses Python's automatic garbage collection while at least one such handle is open. h5py holds a process-global lock while waiting on fsspec's event-loop thread, and a collection on that thread deallocating an h5py object deadlocks against it. Reference counting still frees non-cyclic garbage, and collection resumes when the last such handle closes; cyclic garbage from all threads accumulates until then. Local paths and synchronous backends such as `memory://` are unaffected. This is the mitigation [h5py recommends](https://docs.h5py.org/en/stable/high/file.html#python-file-like-objects) for the deadlock. See [Working with Remote Patches](tutorial/remote_patches.qmd). diff --git a/docs/tutorial/remote_patches.qmd b/docs/tutorial/remote_patches.qmd index f51e22618..c88930ca3 100644 --- a/docs/tutorial/remote_patches.qmd +++ b/docs/tutorial/remote_patches.qmd @@ -141,6 +141,8 @@ The same distinction applies to spools: While a remote HDF5 handle is open, DASCore pauses Python's automatic garbage collection for the whole process, then restores it when the last such handle closes. -This avoids a deadlock: h5py holds a process-global lock while waiting on the event loop thread [fsspec](https://filesystem-spec.readthedocs.io/) uses for each remote fetch, and an automatic collection running on that thread needs the same lock to deallocate a dead h5py object. +This avoids a deadlock which h5py documents under [Python file-like objects](https://docs.h5py.org/en/stable/high/file.html#python-file-like-objects): h5py serializes its low-level calls behind a process-global lock, holds that lock while a file-like object services a read, and needs the same lock to deallocate a dead h5py object. A remote read is serviced by the event loop thread [fsspec](https://filesystem-spec.readthedocs.io/) runs for each async backend, so a cyclic collection landing on that thread waits for a lock the reader will not release until the read completes. + +Of the mitigations h5py suggests, DASCore takes the second: temporarily disabling collection. The first, avoiding reference cycles which keep h5py objects alive, is not something a library can guarantee, since the cycle may be anywhere in the process. Reference counting is unaffected, so most objects are still freed immediately; only cyclic garbage accumulates, and only until the handle closes. Reads of local files and of synchronous backends such as `memory://` are unaffected. From 3ea0a2420d878d8e70083a070015e8ae09a58e00 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Wed, 12 Aug 2026 15:21:50 +0200 Subject: [PATCH 13/13] Address review: drop the matrix comment, correct the get_format comment --- .github/workflows/runtests.yml | 2 -- dascore/io/core.py | 6 +++++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/.github/workflows/runtests.yml b/.github/workflows/runtests.yml index 8ad9dc169..f59e5d614 100644 --- a/.github/workflows/runtests.yml +++ b/.github/workflows/runtests.yml @@ -154,8 +154,6 @@ jobs: strategy: fail-fast: false matrix: - # The same OS list as test_code; the deadlock these tests cover was - # platform-agnostic, so every supported platform should exercise it. os: ${{ fromJson(needs.setup.outputs.os-matrix) }} # Keep remote-IO coverage visible without blocking unrelated changes. diff --git a/dascore/io/core.py b/dascore/io/core.py index e0edaf874..d67dbe4be 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -881,7 +881,11 @@ def _wrapper(*args, _pre_cast=False, **kwargs): if new_resource is not None and new_resource is not resource: with suppress(Exception): release_handle(new_resource, abort=True) - # get_format can't raise; it must return False instead. + # get_format reports "not my format" by returning False rather + # than raising, so an ordinary Exception becomes False here. + # Everything else propagates, including a BaseException raised + # inside get_format: the catch is only this wide so the cleanup + # above runs on a KeyboardInterrupt, not to swallow one. if fun_name != "get_format" or not isinstance(e, Exception): raise out = False