Expose the remote HDF5 knobs and announce the gc pause

dascore/config.py

@@ -137,7 +137,27 @@ class DascoreConfig(BaseModel):
     )
     remote_hdf5_block_size: int = Field(
         default=5_242_880,
-        description="Block size in bytes for remote HDF5 access on tuned protocols.",
+        gt=0,
+        description=(
+            "Block size in bytes for remote HDF5 access on tuned protocols. "
+            "Zero would make fsspec return a non-seekable streaming file and "
+            "download the whole thing, so it is rejected."
+        ),
+    )
+    remote_hdf5_max_blocks: int = Field(
+        default=8,
+        gt=0,
+        description=(
+            "Blocks each open HTTP HDF5 handle may keep cached. Retained memory "
+            "is this times `remote_hdf5_block_size`."
+        ),
+    )
+    warn_on_gc_pause: bool = Field(
+        default=True,
+        description=(
+            "Warn the first time DASCore pauses automatic garbage collection "
+            "for a remote HDF5 read."
+        ),
     )
 
     @field_validator(

dascore/io/core.py

@@ -79,7 +79,11 @@
 from dascore.utils.paths import coerce_to_local_path, coerce_to_upath, is_local_path
 from dascore.utils.plugins import get_entry_point_loaders
 from dascore.utils.progress import track
-from dascore.utils.remote_io import get_remote_cache_scope, remote_cache_scope
+from dascore.utils.remote_io import (
+    get_remote_cache_scope,
+    remote_cache_scope,
+    suppress_gc_pause_warning,
+)
 
 # What the scan dispatchers accept: one resource or patch, or an
 # iterable of them (`_iterate_scan_inputs` flattens its input with
@@ -749,7 +753,11 @@ def _get_format(
         See [`dascore.io.core.get_format`](`dascore.io.core.get_format`)
         for docs.
         """
-        with IOResourceManager(path) as man:
+        # Probing must not announce a remote gc pause: the resource is not
+        # known to be HDF5 yet, and under warnings-as-errors the warning
+        # would be caught by the robustness handler below and read as
+        # "wrong format", silently skipping the reader which does match.
+        with IOResourceManager(path) as man, suppress_gc_pause_warning():
             path = man.source
             if isinstance(path, UPath):
                 exists = path.exists()

dascore/utils/hdf5.py

@@ -316,17 +316,26 @@ def _get_open_kwargs(resource: UPath) -> dict[str, object]:
         protocol = getattr(resource, "protocol", None)
         if protocol not in remote_hdf5_tuned_protocols:
             return {}
+        # One snapshot: config is swappable, and reading the size and the
+        # block count separately could pair one setting with the other's
+        # replacement, for a cap neither configuration asked for.
+        config = get_config()
         # 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}
+        out = {"block_size": 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}}
+        # every jump. A few blocks keep both ends resident; the cap bounds
+        # what one open handle retains.
+        max_blocks = config.remote_hdf5_max_blocks
+        return out | {
+            "cache_type": "blockcache",
+            "cache_options": {"maxblocks": max_blocks},
+        }
 
     @classmethod
     def get_handle(cls, resource):

dascore/utils/remote_io.py

@@ -37,16 +37,66 @@
 _REMOTE_CACHE_SCOPE: ContextVar[str] = ContextVar(
     "remote_cache_scope", default="default"
 )
+_SUPPRESS_GC_PAUSE_WARNING: ContextVar[bool] = ContextVar(
+    "suppress_gc_pause_warning", default=False
+)
 
 _gc_pause_lock = threading.Lock()
 _gc_pause_depth = 0
 _gc_was_enabled = False
 _gc_collect_after = 0.0
+_gc_pause_warned = False
 # 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
 
 
+@contextmanager
+def suppress_gc_pause_warning():
+    """
+    Do not announce the gc pause for the duration of the block.
+
+    Wrapped around format detection. There the warning is both wrong and
+    dangerous: it claims an HDF5 read before h5py has decided the resource
+    is HDF5, and a filter which turns warnings into errors would make the
+    probe read as "wrong format", silently skipping the right reader.
+    """
+    token = _SUPPRESS_GC_PAUSE_WARNING.set(True)
+    try:
+        yield
+    finally:
+        _SUPPRESS_GC_PAUSE_WARNING.reset(token)
+
+
+def _warn_gc_pause_once() -> None:
+    """
+    Say once per process that a remote read is pausing collection.
+
+    The pause is process-global and invisible otherwise, so a program whose
+    memory grows, or which finds ``gc.isenabled()`` False, has no way to
+    connect either to a remote read. Warning once keeps a spool over many
+    remote files from repeating it.
+    """
+    global _gc_pause_warned
+    if _SUPPRESS_GC_PAUSE_WARNING.get():
+        return
+    # Claimed under the lock so two openers cannot both warn.
+    with _gc_pause_lock:
+        if _gc_pause_warned or not get_config().warn_on_gc_pause:
+            return
+        _gc_pause_warned = True
+    # Warned outside it: a filter turning this into an error must not
+    # propagate while the lock is held.
+    msg = (
+        "Reading remote HDF5 pauses Python's automatic garbage collection "
+        "until the last such handle closes, which avoids a deadlock between "
+        "h5py's global lock and collection on fsspec's event-loop thread. "
+        "Reference counting is unaffected, but cyclic garbage accumulates "
+        "meanwhile. Set `warn_on_gc_pause=False` to silence this warning."
+    )
+    warnings.warn(msg, UserWarning, stacklevel=4)
+
+
 def _claim_safety_collect() -> bool:
     """Return True when this caller wins the rate-limited safety collection."""
     global _gc_collect_after
@@ -103,6 +153,10 @@ def pause_gc() -> None:
                 gc.disable()
         finally:
             _gc_pause_depth += 1
+    # Last, so a warning filter which raises cannot escape before the pause is
+    # accounted for. The caller resumes on any failure, and would otherwise
+    # release a pause this call never took -- stranding another live handle.
+    _warn_gc_pause_once()
 
 
 def resume_gc() -> None:
@@ -125,13 +179,15 @@ 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, _gc_collect_after
+    global _gc_pause_depth, _gc_pause_lock, _gc_collect_after, _gc_pause_warned
     _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.
+    # Neither the parent's rate limit nor its warning describes this process;
+    # a pool worker which pauses collection should say so itself.
     _gc_collect_after = 0.0
+    _gc_pause_warned = False
 
 
 @_reinit_after_fork

docs/tutorial/configuration.qmd

@@ -48,6 +48,7 @@ with config_context(patch_history="disabled"):
 ```
 
 For remote IO settings such as `allow_remote_cache`,
-`allow_remote_cache_for_metadata`, and `warn_on_remote_cache`, plus examples
-of metadata-only vs read-time behavior, see
+`allow_remote_cache_for_metadata`, `warn_on_remote_cache`,
+`remote_hdf5_block_size`, `remote_hdf5_max_blocks`, and `warn_on_gc_pause`,
+plus examples of metadata-only vs read-time behavior, see
 [Working with Remote Patches](remote_patches.qmd).

docs/tutorial/remote_patches.qmd

@@ -146,3 +146,55 @@ This avoids a deadlock which h5py documents under [Python file-like objects](htt
 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.
+
+DASCore warns the first time it pauses collection in a process, since the effect is otherwise invisible. Silence it with `warn_on_gc_pause=False` once you know it applies to your workflow.
+
+```python
+import dascore as dc
+from dascore.config import config_context
+
+with config_context(warn_on_gc_pause=False):
+    patch = dc.read("http://example.com/data/prodml_2.1.h5")[0]
+```
+
+## Tuning Remote HDF5 Transfers
+
+Opening an HDF5 file over HTTP is dominated by h5py's metadata probe, which alternates between the file's header and footer. DASCore reads through a block cache so both ends stay resident rather than being refetched on every jump. Two settings control it:
+
+- `remote_hdf5_block_size` — bytes fetched per block (5 MiB by default)
+- `remote_hdf5_max_blocks` — blocks one open handle may keep (8 by default)
+
+Their product is roughly the memory one open handle retains, so ~40 MiB by default. The cache fetches one block per request, so the block size trades bytes against round trips. Which way to move it depends on what you are doing, because scanning and reading pull in opposite directions.
+
+### Scanning many files
+
+Scanning reads only metadata — kilobytes, from two distant regions of each file, which is then closed before the cache is reused. Big blocks spend megabytes to deliver kilobytes, so shrink both:
+
+```python
+import dascore as dc
+from dascore.config import config_context
+
+urls = [f"http://example.com/data/file_{i}.h5" for i in range(100)]
+
+with config_context(remote_hdf5_block_size=262_144, remote_hdf5_max_blocks=4):
+    df = dc.scan_to_df(urls)
+```
+
+That retains 1 MiB per handle rather than 40 MiB. Do not shrink too far: on a high-latency link the extra round trips cost more than the bytes saved.
+
+### Reading whole patches
+
+Reading pulls large contiguous ranges, so bigger blocks mean fewer requests for the same bytes. The LRU earns little on a single pass, so trade blocks for size:
+
+```python
+with config_context(remote_hdf5_block_size=16_777_216, remote_hdf5_max_blocks=2):
+    patch = dc.read("http://example.com/data/prodml_2.1.h5")[0]
+```
+
+If you read whole files, especially more than once, let DASCore materialize them locally instead — see [Remote Cache Settings](#remote-cache-settings). One sequential download beats any streaming pattern.
+
+### Many handles at once
+
+The figure is per *open* handle, so concurrent reads multiply it: eight files in parallel at the defaults holds ~320 MiB. Lower `remote_hdf5_max_blocks` first, since evictions cost nothing for a scan or a single pass.
+
+Both settings apply to the protocols DASCore tunes; `remote_hdf5_max_blocks` reaches only HTTP and HTTPS, as S3 uses a read-ahead cache which takes the block size alone.

tests/test_utils/test_gc_pause.py

@@ -6,12 +6,17 @@
 import io
 import os
 import threading
+import warnings
 from contextlib import suppress
 from types import SimpleNamespace
 
 import pytest
+from upath import UPath
 
+import dascore as dc
+import dascore.utils.hdf5 as hdf5_module
 import dascore.utils.remote_io as remote_io
+from dascore.config import config_context
 from dascore.utils.hdf5 import (
     _is_loop_backed,
     _ManagedH5pyFile,
@@ -235,6 +240,115 @@ def _pause_and_resume(_index):
         assert gc.isenabled()
 
 
+class TestPauseWarning:
+    """The pause is invisible otherwise, so it must announce itself once."""
+
+    @pytest.fixture(autouse=True)
+    def _unwarned(self, monkeypatch):
+        """Start each test as though nothing had warned yet."""
+        monkeypatch.setattr(remote_io, "_gc_pause_warned", False)
+
+    def test_warns_once_per_process(self):
+        """A spool over many remote files must not repeat the warning."""
+        with pytest.warns(UserWarning, match="pauses Python's automatic garbage"):
+            pause_gc()
+        resume_gc()
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")  # a second warning would raise
+            pause_gc()
+            resume_gc()
+
+    def test_can_be_silenced(self):
+        """`warn_on_gc_pause=False` turns the warning off."""
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")
+            with config_context(warn_on_gc_pause=False):
+                pause_gc()
+                resume_gc()
+
+    def test_raising_filter_cannot_steal_a_live_pause(self):
+        """A warning turned into an error must not release someone else's pause.
+
+        The caller resumes on any failure from pause_gc, so a warning which
+        raises before the depth moves would release a pause this call never
+        took, re-enabling collection under a handle still open.
+        """
+        with config_context(warn_on_gc_pause=False):
+            pause_gc()  # stand in for a handle already open
+        try:
+            with pytest.raises(UserWarning), warnings.catch_warnings():
+                warnings.simplefilter("error")
+                pause_gc()
+            resume_gc()  # what _open_h5_fileobj does on failure
+            assert not gc.isenabled(), "the live pause was stolen"
+            assert remote_io._gc_pause_depth == 1
+        finally:
+            resume_gc()
+
+    @pytest.mark.concurrency
+    @pytest.mark.skipif(not hasattr(os, "fork"), reason="requires fork")
+    # Forking a multi-threaded process can wedge the child, and the repo sets
+    # no global timeout, so bound it rather than hang the job for an hour.
+    @pytest.mark.timeout(30)
+    def test_fork_rearms_the_warning(self):
+        """A pool worker which pauses collection should say so itself.
+
+        Forks for real rather than calling the reset hook, so this also
+        pins that the hook is registered with os.register_at_fork.
+        """
+        with pytest.warns(UserWarning, match="pauses Python's automatic garbage"):
+            pause_gc()
+        resume_gc()
+        assert remote_io._gc_pause_warned
+        read_fd, write_fd = os.pipe()
+        pid = os.fork()
+        if pid == 0:  # the child reports what it inherited, then leaves
+            os.close(read_fd)
+            os.write(write_fd, b"1" if remote_io._gc_pause_warned else b"0")
+            os._exit(0)
+        os.close(write_fd)
+        inherited = os.read(read_fd, 1)
+        os.close(read_fd)
+        os.waitpid(pid, 0)
+        assert inherited == b"0", "the child inherited the parent's warned flag"
+
+
+class TestProbeSuppression:
+    """Format detection must not be disturbed by the pause warning."""
+
+    def test_probe_survives_warnings_as_errors(
+        self, random_patch, tmp_path, monkeypatch
+    ):
+        """A warning raised while probing must not read as "wrong format".
+
+        `_get_format` treats any exception from a probe as "not my format",
+        so a filter turning the warning into an error would silently skip
+        the reader which does match. `file_format` pins the probe to that
+        reader; without it the once-per-process warning lands on an earlier
+        miss and the failure hides.
+        """
+        path = tmp_path / "probe.h5"
+        dc.write(random_patch, path, "dasdae")
+        # Make a local file take the loop-backed branch, which pauses.
+        monkeypatch.setattr(hdf5_module, "_is_loop_backed", lambda _resource: True)
+        monkeypatch.setattr(remote_io, "_gc_pause_warned", False)
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", UserWarning)
+            out = dc.get_format(UPath(path), file_format="DASDAE")
+        assert out == ("DASDAE", "1")
+
+    def test_probe_does_not_warn(self, random_patch, tmp_path, monkeypatch):
+        """Probing claims no HDF5 read; the resource may not even be one."""
+        path = tmp_path / "quiet.h5"
+        dc.write(random_patch, path, "dasdae")
+        monkeypatch.setattr(hdf5_module, "_is_loop_backed", lambda _resource: True)
+        monkeypatch.setattr(remote_io, "_gc_pause_warned", False)
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            dc.get_format(UPath(path), file_format="DASDAE")
+        assert not [x for x in caught if "automatic garbage" in str(x.message)]
+
+
 class TestLoopBackedDetection:
     """Missing a loop-backed object leaves the deadlock window open."""