From 095708beed1d460d3219b252b9b2b0cd462d82bd Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sat, 25 Jul 2026 20:22:54 +0200 Subject: [PATCH 1/5] ENH: synchronize remote IO, add free-threaded CI and concurrency docs Final part of the series superseding #763. Remote file cache: - One download lock per cached resource, so unrelated files still download at the same time while callers wanting the same file take turns. The locks are never removed, which lets a cache clear hold all of them without racing new ones into existence. - clear_remote_file_cache holds every download lock, and refuses to run from inside a materialization (thread-local depth guard). - A failed download publishes nothing, so the next caller retries it. - The lru_cache on _materialize_remote_file goes away in favor of the locks; the policy checks move into _download_to_cache. IOResourceManager gets an instance lock so each required type is opened exactly once. Its source property is no longer memoized into the handle cache, where close_all could have closed the source itself. Adds a standalone free-threaded CI job (3.14t, PYTHON_GIL=0, asserting the GIL really is off) which runs the suite on the core dependencies, and documents the concurrency contract the series establishes in the parallelization recipe. --- .github/workflows/test_free_threaded.yml | 86 +++++++++++++++ dascore/utils/io.py | 42 ++++--- dascore/utils/remote_io.py | 133 ++++++++++++++++------- docs/recipes/parallelization.qmd | 30 ++++- tests/test_utils/test_io_utils.py | 125 +++++++++++++++++++++ 5 files changed, 363 insertions(+), 53 deletions(-) create mode 100644 .github/workflows/test_free_threaded.yml diff --git a/.github/workflows/test_free_threaded.yml b/.github/workflows/test_free_threaded.yml new file mode 100644 index 000000000..c1a8211f7 --- /dev/null +++ b/.github/workflows/test_free_threaded.yml @@ -0,0 +1,86 @@ +# Run the test suite on a free-threaded (no-GIL) CPython build. +name: TestFreeThreaded +on: + push: + branches: + - master + - dev + pull_request: + branches: + - master + - dev + paths: + - 'dascore/**' + - 'tests/**' + - 'pyproject.toml' + - '.github/workflows/test_free_threaded.yml' + +# This job only reads the repository. +permissions: + contents: read + +concurrency: + group: TestFreeThreaded-${{ github.event.pull_request.number || github.run_id }} + cancel-in-progress: true + +env: + CACHE_NUMBER: 1 + MPLBACKEND: Agg + QT_QPA_PLATFORM: offscreen + # Keep the GIL off even if an extension module asks for it back, so a + # dependency cannot quietly turn this job into an ordinary one. + PYTHON_GIL: '0' + +jobs: + free_thread: + runs-on: ubuntu-latest + timeout-minutes: 60 + + # only run if CI isn't turned off + if: github.event_name == 'push' || !contains(github.event.pull_request.labels.*.name, 'no_ci') + + steps: + - uses: actions/checkout@v5 + with: + fetch-tags: 'true' + fetch-depth: '0' + persist-credentials: false + + - uses: actions/setup-python@v5 + with: + python-version: '3.14t' + + # Only the core dependencies: the optional ones are not all built for + # free-threading yet, and their tests skip when they are missing. + - name: install dascore + run: | + python -m pip install --upgrade pip + pip install -e . pytest pytest-timeout + + - name: confirm the GIL is disabled + run: | + python -c " + import sys + print(sys.version) + assert not sys._is_gil_enabled(), 'the GIL is enabled; this job proves nothing' + " + + - name: export test data cache env + env: + INPUT_CACHE_NUMBER: ${{ env.CACHE_NUMBER }} + RUNNER_OS: ${{ runner.os }} + run: python .github/scripts/export_test_data_cache_env.py >> "$GITHUB_ENV" + + - name: restore test data cache + id: restore-test-data + uses: actions/cache/restore@v5 + with: + path: ${{ env.DATA_CACHE_PATH }} + key: ${{ env.DATA_CACHE_KEY }} + + - name: prime test data cache + if: steps.restore-test-data.outputs.cache-hit != 'true' + run: python .github/scripts/cache_test_data.py + + - name: run test suite without the GIL + run: python -m pytest tests -m "not network" -q --timeout=900 diff --git a/dascore/utils/io.py b/dascore/utils/io.py index df111f14f..e41e29f13 100644 --- a/dascore/utils/io.py +++ b/dascore/utils/io.py @@ -8,6 +8,7 @@ from functools import cache from inspect import isfunction, ismethod from pathlib import Path +from threading import RLock from typing import Any, get_type_hints import numpy as np @@ -18,7 +19,6 @@ from dascore.exceptions import PatchConversionError from dascore.utils.misc import ( _maybe_make_parent_directory, - cached_method, iterate, optional_import, ) @@ -210,16 +210,25 @@ def get_handle_from_resource(uri, required_type): class IOResourceManager: - """A class for managing opening/closing files.""" + """ + A class for managing opening/closing files. + + One manager serves one IO operation. Creating and closing its + resources is synchronized, so concurrent callers share a single + handle per type; a handle it hands back is not itself safe to use + from several threads at once. + """ def __init__(self, source: Any): self._source = source self._cache = {} + self._lock = RLock() @property - @cached_method def source(self): """Get the source of the IO manager.""" + # Not cached: the walk is a couple of isinstance checks, and + # memoizing it into _cache would let close_all close the source. source = self._source # this handles IO managers derived from other IO managers; # effectively, we need to go back to the original, non-io manager source @@ -228,7 +237,7 @@ def source(self): return source def get_resource(self, required_type: RequiredType) -> RequiredType: - """Get the requested resource.""" + """Get the requested resource, opening each handle exactly once.""" # no required type, just return source of manager. if required_type is None: return self.source @@ -238,21 +247,24 @@ def get_resource(self, required_type: RequiredType) -> RequiredType: if isinstance(self._source, self.__class__): return self._source.get_resource(required_type) required_type = _get_required_type(required_type) - if required_type not in self._cache: - source = _resolve_resource(self._source, required_type) - out = get_handle_from_resource(source, required_type) - self._cache[required_type] = out - return self._cache[required_type] + with self._lock: + if required_type not in self._cache: + source = _resolve_resource(self._source, required_type) + out = get_handle_from_resource(source, required_type) + self._cache[required_type] = out + return self._cache[required_type] def close_all(self): """Close any open file handles.""" - for handle in self._cache.values(): - getattr(handle, "close", lambda: None)() + with self._lock: + for handle in self._cache.values(): + getattr(handle, "close", lambda: None)() def clear_cache(self): """Close and forget any cached resources so they can be reopened fresh.""" - self.close_all() - self._cache.clear() + with self._lock: + self.close_all() + self._cache.clear() def __enter__(self): """Entering context manager.""" @@ -263,7 +275,9 @@ def __exit__(self, exc_type, exc_val, exc_tb): self.close_all() def __del__(self): - self.close_all() + # __del__ can run on a partially built manager (eg if __init__ raised). + if getattr(self, "_lock", None) is not None: + self.close_all() def patch_to_xarray(patch: PatchType): diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 185016694..3296a2b85 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -7,15 +7,16 @@ import shutil import tempfile import warnings -from contextlib import contextmanager +from contextlib import ExitStack, contextmanager from contextvars import ContextVar -from functools import lru_cache from hashlib import sha256 from pathlib import Path +from threading import RLock, local from dascore.compat import UPath from dascore.config import get_config 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"} @@ -24,11 +25,30 @@ "only reading this file from the beginning is supported", ) _REMOTE_RESOURCE_CACHE: dict[str, UPath] = {} +# One lock per cached resource, so unrelated downloads still run at the +# same time. Entries are never removed: the dict is bounded by the number +# of distinct remote resources a session touches, and keeping them lets a +# cache clear hold every lock without racing new ones into existence. +_REMOTE_KEY_LOCKS: dict[tuple[str, Path], RLock] = {} +# Guards the two dicts above. Always acquired before a download lock and +# never while holding one. +_REMOTE_CACHE_LOCK = RLock() +# Counts this thread's in-progress materializations (see clear). +_REMOTE_CACHE_LOCAL = local() _REMOTE_CACHE_SCOPE: ContextVar[str] = ContextVar( "remote_cache_scope", default="default" ) +@_reinit_after_fork +def _reinit_remote_cache_locks(): + """Install fresh remote-cache locks; see _reinit_after_fork.""" + global _REMOTE_CACHE_LOCK, _REMOTE_KEY_LOCKS, _REMOTE_CACHE_LOCAL + _REMOTE_CACHE_LOCK = RLock() + _REMOTE_KEY_LOCKS = {} + _REMOTE_CACHE_LOCAL = local() + + @contextmanager def remote_cache_scope(scope: str): """Temporarily set the current remote-cache policy scope.""" @@ -66,10 +86,19 @@ def normalize_remote_id(path) -> str: serialized = json.dumps(storage_options, sort_keys=True, default=str) options_suffix = f"#{sha256(serialized.encode()).hexdigest()}" remote_id = f"{resource}{options_suffix}" - _REMOTE_RESOURCE_CACHE[remote_id] = resource + with _REMOTE_CACHE_LOCK: + _REMOTE_RESOURCE_CACHE[remote_id] = resource return remote_id +def _get_download_lock(key: tuple[str, Path]) -> RLock: + """Return the download lock for one cached resource, creating it once.""" + with _REMOTE_CACHE_LOCK: + if (lock := _REMOTE_KEY_LOCKS.get(key)) is None: + lock = _REMOTE_KEY_LOCKS[key] = RLock() + return lock + + def _get_remote_cache_dir(remote_id: str) -> Path: # pragma: no cover """Return the cache directory for a normalized remote identifier.""" return get_remote_cache_path() / sha256(remote_id.encode()).hexdigest() @@ -114,11 +143,22 @@ def _warn_remote_cache_download(resource: UPath, local_path: Path): def clear_remote_file_cache(): - """Remove all locally cached remote files and memoized paths.""" - shutil.rmtree(get_remote_cache_path(), ignore_errors=True) - get_remote_cache_path().mkdir(parents=True, exist_ok=True) - _materialize_remote_file.cache_clear() - _REMOTE_RESOURCE_CACHE.clear() + """ + Remove all locally cached remote files and memoized paths. + + Clearing is a single-writer operation: it holds every download lock, + so materializations already in flight finish before their artifacts + are removed, and later ones download again. + """ + if getattr(_REMOTE_CACHE_LOCAL, "materializing", 0): + msg = "Cannot clear the remote file cache from inside a materialization." + raise RuntimeError(msg) + with _REMOTE_CACHE_LOCK, ExitStack() as stack: + for lock in _REMOTE_KEY_LOCKS.values(): + stack.enter_context(lock) + shutil.rmtree(get_remote_cache_path(), ignore_errors=True) + get_remote_cache_path().mkdir(parents=True, exist_ok=True) + _REMOTE_RESOURCE_CACHE.clear() def _download_remote_file(path, local_path: Path): @@ -146,12 +186,16 @@ def _download_remote_file(path, local_path: Path): tmp_path.unlink(missing_ok=True) -@lru_cache -def _materialize_remote_file( # pragma: no cover - remote_id: str, cache_root: Path -) -> Path: - """Materialize one remote resource to a stable local cache path.""" - resource = _REMOTE_RESOURCE_CACHE.get(remote_id) +def _materialize_remote_file(remote_id: str, cache_root: Path) -> Path: + """ + Materialize one remote resource to a stable local cache path. + + Callers wanting the same resource take turns on its download lock: + the first downloads, the rest find the published file. A download + which fails publishes nothing, so the next caller retries it. + """ + with _REMOTE_CACHE_LOCK: + resource = _REMOTE_RESOURCE_CACHE.get(remote_id) if resource is None: resource = coerce_to_upath(remote_id.split("#", maxsplit=1)[0]) local_path = ( @@ -159,33 +203,46 @@ def _materialize_remote_file( # pragma: no cover / sha256(remote_id.encode()).hexdigest() / _safe_remote_name(resource) ) - if not local_path.exists(): - config = get_config() - scope = get_remote_cache_scope() - if scope == "metadata" and not config.allow_remote_cache_for_metadata: - msg = ( - "Remote metadata access requires a local cached file for " - f"{_redact_remote_resource(resource)}, " - "but DASCore does not download remote files during " - "`scan()` or public `get_format()` by default. Set " - "`allow_remote_cache_for_metadata=True` to opt in to metadata-time " - "remote caching." - ) - raise RemoteCacheError(msg) - if scope != "metadata" and not config.allow_remote_cache: - msg = ( - f"Remote caching is disabled, but DASCore needs a local file for " - f"{_redact_remote_resource(resource)}. " - "Set `allow_remote_cache=True` to permit downloading " - "remote files into the local cache." - ) - raise RemoteCacheError(msg) - if config.warn_on_remote_cache: - _warn_remote_cache_download(resource, local_path) - _download_remote_file(resource, local_path) + # Record the depth so a cache clear attempted from inside a download + # (eg from a warning handler) raises rather than deleting live state. + depth = getattr(_REMOTE_CACHE_LOCAL, "materializing", 0) + _REMOTE_CACHE_LOCAL.materializing = depth + 1 + try: + with _get_download_lock((remote_id, cache_root)): + if not local_path.exists(): + _download_to_cache(resource, local_path) + finally: + _REMOTE_CACHE_LOCAL.materializing = depth return local_path +def _download_to_cache(resource: UPath, local_path: Path) -> None: + """Apply the remote-cache policy, then download the resource.""" + config = get_config() + scope = get_remote_cache_scope() + if scope == "metadata" and not config.allow_remote_cache_for_metadata: + msg = ( + "Remote metadata access requires a local cached file for " + f"{_redact_remote_resource(resource)}, " + "but DASCore does not download remote files during " + "`scan()` or public `get_format()` by default. Set " + "`allow_remote_cache_for_metadata=True` to opt in to metadata-time " + "remote caching." + ) + raise RemoteCacheError(msg) + if scope != "metadata" and not config.allow_remote_cache: + msg = ( + f"Remote caching is disabled, but DASCore needs a local file for " + f"{_redact_remote_resource(resource)}. " + "Set `allow_remote_cache=True` to permit downloading " + "remote files into the local cache." + ) + raise RemoteCacheError(msg) + if config.warn_on_remote_cache: + _warn_remote_cache_download(resource, local_path) + _download_remote_file(resource, local_path) + + def ensure_local_file(resource) -> Path: """Return a stable local path for one resource for the current session.""" if is_pathlike(resource) and is_local_path(resource): diff --git a/docs/recipes/parallelization.qmd b/docs/recipes/parallelization.qmd index 34dbb9efb..beb134a12 100644 --- a/docs/recipes/parallelization.qmd +++ b/docs/recipes/parallelization.qmd @@ -25,10 +25,38 @@ executor = ProcessPoolExecutor() spool.map(my_patch_processing_function, client=executor) ``` -The `ThreadPoolExecutor` from the same module will also work, but due to python's GIL may not provide much of a speed-up. +The `ThreadPoolExecutor` from the same module also works. On a free-threaded (no-GIL) build of CPython threads run patch processing in parallel; on a standard build the GIL limits the speed-up to whatever the work releases it for, which for DASCore is mostly file reading and NumPy/SciPy computation. There are two downsides to this approach. First, if the patches aren't chunked adequately it may exhaust the available memory. Second, it will only work on a single machine. The next section presents a more scalable option. +# Thread Safety + +DASCore's test suite runs on a free-threaded CPython build with the GIL disabled, and the guarantees below are what that supports. They apply equally to threads on a standard build. + +## What threads may share + +Patches are immutable, so any number of threads may read the same patch. Spools may also be read concurrently: iterating, selecting, chunking, indexing and taking their length are safe from several threads at once, because the underlying metadata caches are synchronized internally. + +[`Spool.get_contents`](`dascore.core.spool.BaseSpool.get_contents`) hands back a dataframe you own. Mutating it never changes the spool, so each thread may modify its own copy freely. Arrays reached through coordinates go the other way: they are shared and read-only, so copy one before writing to it. + +The process-wide registries take care of themselves. The file-format (`FiberIO`) registry, the method-namespace registry and the [pint](https://pint.readthedocs.io) unit registry are each guarded by a single lock, so first use from several threads at once is safe. Loading the plugins for one format is serialized: the first thread to need a format loads it while the others wait, and none of them can observe a partially registered format. + +Remote files are cached per resource. Threads asking for the same remote file take turns, so it downloads once and the rest use the result; different files download at the same time. + +## What threads may not share + +State-changing calls follow a single-writer model. Updating a directory spool, adding patches to an in-memory spool and [`clear_remote_file_cache`](`dascore.utils.remote_io.clear_remote_file_cache`) should each run from one thread, with no other thread reading the same object meanwhile. DASCore keeps its own structures consistent through such a change, but it does not make a reader see a coherent before-or-after snapshot of it. + +An open file handle belongs to one IO operation. DASCore opens each resource once per operation and closes it when the operation ends; a handle obtained from one is not safe to read from two threads at once. Prefer giving each thread its own patch or spool to read, rather than sharing a handle. + +Third-party `FiberIO` plugins are responsible for their own thread safety. DASCore synchronizes the registry that holds them and serializes their import, but a plugin which keeps mutable state of its own must guard it. + +## Configuration in threads + +Configuration has two tiers, described in [runtime configuration](../tutorial/configuration.qmd). `set_config` changes the process-wide base and is visible from every thread. `config_context` overrides the config for the current context only, and a newly started thread begins with a fresh context, so a scoped override does not automatically reach threads started inside it. + +[`Spool.map`](`dascore.core.spool.BaseSpool.map`) handles this for you: it binds the configuration active when `map` is called and re-applies it inside each worker, for thread pools and process pools alike. When starting threads yourself, either set the configuration permanently before starting them or re-apply the scoped override inside each thread. + # MPI4Py This section shows how to use the "mpi4py" library to parallelize dascore code. diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index c241c9061..8ddab2794 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 threading from contextlib import closing from io import BufferedReader, BufferedWriter, BytesIO, StringIO, TextIOBase from pathlib import Path @@ -1163,3 +1164,127 @@ def test_example_event(self, event_patch_2): st = patch.io.to_obspy() assert isinstance(st, obspy.Stream) assert len(st) == len(patch.get_coord("distance")) + + +class TestRemoteCacheConcurrency: + """The remote cache serializes per resource, not globally.""" + + @pytest.fixture(autouse=True) + def isolated_cache(self, tmp_path, permanent_config): + """ + Give each test its own cache directory. + + Uses the permanent config: worker threads start with a fresh + context, so a scoped config_context override would not reach them. + """ + with permanent_config( + remote_cache_dir=tmp_path / "remote_cache", warn_on_remote_cache=False + ): + clear_remote_file_cache() + yield + clear_remote_file_cache() + + def _memory_file(self, name: str) -> UPath: + """Write a small file into the in-memory filesystem.""" + path = UPath(f"memory://dascore/concurrent/{name}") + with path.open("wb") as fi: + fi.write(b"dascore" * 64) + return path + + def test_racing_callers_download_once(self, monkeypatch, run_in_threads): + """Callers wanting one resource agree on the path and download it once.""" + resource = self._memory_file("shared.bin") + downloads = [] + original = remote_io._download_remote_file + + def _counted(path, local_path): + downloads.append(local_path) + return original(path, local_path) + + monkeypatch.setattr(remote_io, "_download_remote_file", _counted) + results = run_in_threads(lambda _: ensure_local_file(resource)) + assert len({str(x) for x in results}) == 1 + assert len(downloads) == 1 + assert results[0].exists() + + def test_distinct_resources_are_not_serialized(self, monkeypatch, run_in_threads): + """Unrelated downloads run at once; one global lock would time out here.""" + resources = [self._memory_file(f"file_{i}.bin") for i in range(4)] + barrier = threading.Barrier(len(resources), timeout=30) + original = remote_io._download_remote_file + + def _synchronized(path, local_path): + # Every download has to be in flight together to get past this. + barrier.wait() + return original(path, local_path) + + monkeypatch.setattr(remote_io, "_download_remote_file", _synchronized) + results = run_in_threads(lambda index: ensure_local_file(resources[index])) + assert all(x.exists() for x in results) + + def test_failed_download_is_retried(self, monkeypatch): + """A failed download publishes nothing, so the next caller retries.""" + resource = self._memory_file("flaky.bin") + calls = [] + original = remote_io._download_remote_file + + def _fail_once(path, local_path): + calls.append(local_path) + if len(calls) == 1: + raise OSError("download failed") + return original(path, local_path) + + monkeypatch.setattr(remote_io, "_download_remote_file", _fail_once) + with pytest.raises(OSError, match="download failed"): + ensure_local_file(resource) + assert ensure_local_file(resource).exists() + assert len(calls) == 2 + + def test_clear_from_inside_materialization_raises(self, monkeypatch): + """Clearing the cache mid-download would delete live state.""" + resource = self._memory_file("clearing.bin") + + def _clear_while_downloading(path, local_path): + clear_remote_file_cache() + + monkeypatch.setattr( + remote_io, "_download_remote_file", _clear_while_downloading + ) + with pytest.raises(RuntimeError, match="inside a materialization"): + ensure_local_file(resource) + + def test_unregistered_remote_id_is_coerced(self): + """An id missing from the resource cache is rebuilt from the id itself.""" + resource = self._memory_file("unregistered.bin") + remote_io._REMOTE_RESOURCE_CACHE.clear() + cache_root = remote_io._normalize_cache_root(remote_io.get_remote_cache_path()) + local_path = remote_io._materialize_remote_file(str(resource), cache_root) + assert local_path.exists() + + def test_fork_handler_replaces_held_locks(self): + """Locks held at fork time are replaced so the child cannot deadlock.""" + old_lock = remote_io._REMOTE_CACHE_LOCK + try: + with old_lock: + remote_io._reinit_remote_cache_locks() + new_lock = remote_io._REMOTE_CACHE_LOCK + assert new_lock.acquire(blocking=False) + new_lock.release() + assert new_lock is not old_lock + assert not remote_io._REMOTE_KEY_LOCKS + finally: + remote_io._REMOTE_CACHE_LOCK = old_lock + + +class TestIOResourceManagerConcurrency: + """One manager hands every caller the same handle per type.""" + + def test_racing_callers_share_one_handle(self, tmp_path, run_in_threads): + """get_resource opens each required type exactly once.""" + path = tmp_path / "concurrent_resource.bin" + path.write_bytes(b"dascore") + with IOResourceManager(path) as man: + handles = run_in_threads(lambda _: man.get_resource(BinaryReader)) + assert len({id(x) for x in handles}) == 1 + assert not handles[0].closed + assert handles[0].closed From 832c5f11f07e09608724dce2d117e59273d1e6d6 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sat, 25 Jul 2026 20:49:13 +0200 Subject: [PATCH 2/5] Memoize materialized remote paths again Dropping the lru_cache made every resolution of an already-cached remote file re-hash the id, rebuild the path, take two locks and stat the file: 12.2us -> 35.1us per call, none of which the download lock needs to protect. Keep a dict of published paths, checked before that work and emptied by clear_remote_file_cache, which puts the warm path back at 12.8us. Only successful downloads are recorded, so a failed one is still retried by the next caller. --- dascore/utils/remote_io.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 3296a2b85..94aa6eab5 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -25,6 +25,10 @@ "only reading this file from the beginning is supported", ) _REMOTE_RESOURCE_CACHE: dict[str, UPath] = {} +# Resources already materialized this session, so the common case (the +# file is present) costs one dict lookup rather than a hash, a path +# build and a stat. Emptied by clear_remote_file_cache. +_REMOTE_MATERIALIZED: dict[tuple[str, Path], Path] = {} # One lock per cached resource, so unrelated downloads still run at the # same time. Entries are never removed: the dict is bounded by the number # of distinct remote resources a session touches, and keeping them lets a @@ -159,6 +163,7 @@ def clear_remote_file_cache(): shutil.rmtree(get_remote_cache_path(), ignore_errors=True) get_remote_cache_path().mkdir(parents=True, exist_ok=True) _REMOTE_RESOURCE_CACHE.clear() + _REMOTE_MATERIALIZED.clear() def _download_remote_file(path, local_path: Path): @@ -194,7 +199,11 @@ def _materialize_remote_file(remote_id: str, cache_root: Path) -> Path: the first downloads, the rest find the published file. A download which fails publishes nothing, so the next caller retries it. """ + key = (remote_id, cache_root) with _REMOTE_CACHE_LOCK: + # Already materialized: skip hashing, path building and the stat. + if (published := _REMOTE_MATERIALIZED.get(key)) is not None: + return published resource = _REMOTE_RESOURCE_CACHE.get(remote_id) if resource is None: resource = coerce_to_upath(remote_id.split("#", maxsplit=1)[0]) @@ -208,11 +217,14 @@ def _materialize_remote_file(remote_id: str, cache_root: Path) -> Path: depth = getattr(_REMOTE_CACHE_LOCAL, "materializing", 0) _REMOTE_CACHE_LOCAL.materializing = depth + 1 try: - with _get_download_lock((remote_id, cache_root)): + with _get_download_lock(key): if not local_path.exists(): _download_to_cache(resource, local_path) finally: _REMOTE_CACHE_LOCAL.materializing = depth + # Only published paths are memoized, so a failed download is retried. + with _REMOTE_CACHE_LOCK: + _REMOTE_MATERIALIZED[key] = local_path return local_path From c2b6844867bcb8572eb50cf6d5c03acd361b3eab Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sat, 25 Jul 2026 21:15:08 +0200 Subject: [PATCH 3/5] Derive the remote cache path in one place The materializer and the cached-path probe each built cache_root / sha256(remote_id) / name themselves, so they had to agree by inspection or one would download a file the other could not find. Both now call one helper. Deletes _get_remote_cache_dir, which had no callers. Its "pragma: no cover" is why that went unnoticed: coverage cannot report dead code it has been told to ignore. Also drops the pragma from _annotate_handle_path, which the non-network suite does reach. The remaining pragmas in hdf5.py, chunk_plan.py and io/core.py were checked the same way and are still needed. --- dascore/utils/io.py | 2 +- dascore/utils/remote_io.py | 26 +++++++++++++------------- 2 files changed, 14 insertions(+), 14 deletions(-) diff --git a/dascore/utils/io.py b/dascore/utils/io.py index e41e29f13..2fc6cdd65 100644 --- a/dascore/utils/io.py +++ b/dascore/utils/io.py @@ -69,7 +69,7 @@ def _resolve_resource(resource, required_type): return resource -def _annotate_handle_path(handle, resource): # pragma: no cover +def _annotate_handle_path(handle, resource): """Attach lightweight source-path metadata to a remote handle when absent.""" path_str = str(resource) # This is intentionally a small compatibility hack for readers that still diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 94aa6eab5..8a5e020f6 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -103,16 +103,22 @@ def _get_download_lock(key: tuple[str, Path]) -> RLock: return lock -def _get_remote_cache_dir(remote_id: str) -> Path: # pragma: no cover - """Return the cache directory for a normalized remote identifier.""" - return get_remote_cache_path() / sha256(remote_id.encode()).hexdigest() - - def _normalize_cache_root(cache_root: Path | str) -> Path: """Return one normalized cache-root path.""" return Path(cache_root).expanduser() +def _remote_cache_local_path(remote_id: str, cache_root: Path, resource: UPath) -> Path: + """ + Return the local path one remote resource materializes to. + + The materializer and the cached-path probe must agree on this, or one + would download a file the other cannot find. + """ + digest = sha256(remote_id.encode()).hexdigest() + return cache_root / digest / _safe_remote_name(resource) + + def _redact_remote_resource(resource: UPath | str) -> str: """Return a minimally identifying label for remote-cache messages.""" path = coerce_to_upath(resource) @@ -207,11 +213,7 @@ def _materialize_remote_file(remote_id: str, cache_root: Path) -> Path: resource = _REMOTE_RESOURCE_CACHE.get(remote_id) if resource is None: resource = coerce_to_upath(remote_id.split("#", maxsplit=1)[0]) - local_path = ( - cache_root - / sha256(remote_id.encode()).hexdigest() - / _safe_remote_name(resource) - ) + local_path = _remote_cache_local_path(remote_id, cache_root, resource) # Record the depth so a cache clear attempted from inside a download # (eg from a warning handler) raises rather than deleting live state. depth = getattr(_REMOTE_CACHE_LOCAL, "materializing", 0) @@ -277,9 +279,7 @@ def _get_cached_local_file(resource) -> Path | None: remote = coerce_to_upath(resource) cache_root = _normalize_cache_root(get_remote_cache_path()) remote_id = normalize_remote_id(remote) - local_path = ( - cache_root / sha256(remote_id.encode()).hexdigest() / _safe_remote_name(remote) - ) + local_path = _remote_cache_local_path(remote_id, cache_root, remote) return local_path if local_path.exists() else None From cec6d183917a4f419becd7a1c12ad1ee0bebb562 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sun, 26 Jul 2026 06:30:27 +0200 Subject: [PATCH 4/5] Use one cross-reference form for Spool.map in the recipe The file mixed dascore.BaseSpool.map with the dascore.core.spool.BaseSpool form used by its other links and elsewhere in the docs. Both resolve, but only one form should appear. --- docs/recipes/parallelization.qmd | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/recipes/parallelization.qmd b/docs/recipes/parallelization.qmd index beb134a12..ecc930194 100644 --- a/docs/recipes/parallelization.qmd +++ b/docs/recipes/parallelization.qmd @@ -7,7 +7,7 @@ execute: This recipe shows a few strategies to parallelize "embarrassingly parallel" spool processing workflows. # Processes and Threads -[dascore.Spool.map](`dascore.BaseSpool.map`) is the easiest way to process patches in a spool in parallel. Here is an example using the Python standard library module [concurrent.futures](https://docs.python.org/3/library/concurrent.futures.html): +[`Spool.map`](`dascore.core.spool.BaseSpool.map`) is the easiest way to process patches in a spool in parallel. Here is an example using the Python standard library module [concurrent.futures](https://docs.python.org/3/library/concurrent.futures.html): ```{python} from concurrent.futures import ProcessPoolExecutor From 74ab6bce92aaf7973e4d86539ee63c73af3c4d96 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Sun, 26 Jul 2026 07:02:54 +0200 Subject: [PATCH 5/5] Reduce the remote cache to locks that actually hold Two adversarial reviews of the previous design both reproduced the same defect: the memo was published outside the download lock, so the fence clear_remote_file_cache put up did not cover the step that mattered. A clear landing in that window left a memo entry pointing at a deleted file, permanently, for the rest of the session. One review also produced a hard deadlock: a download hook re-entering materialization for another resource took the management lock while holding a download lock, which inverts the order clear_remote_file_cache acquires them in. Rather than add a generation counter to defend a guarantee the docs already say is unsupported, this drops the machinery that was buying it: the management lock, the ExitStack over every download lock, the thread-local depth guard, and the hand-rolled memo. What remains is one lock per resource, created with setdefault, plus the lru_cache the base branch used. That is enough for the property this PR is actually for: two threads never download the same file at once, unrelated downloads still run together, and a failed download is retried because lru_cache does not memoize exceptions. Clearing is now documented as unsynchronized, matching both the base branch's behavior and what the concurrency docs already required of callers. Measured: an already-cached remote resolution is 12.0us against the base branch's 12.1us, so the memo the previous commit added is not needed. --- .github/workflows/test_free_threaded.yml | 8 +++ dascore/utils/io.py | 4 +- dascore/utils/remote_io.py | 87 +++++++----------------- docs/recipes/parallelization.qmd | 2 +- tests/test_utils/test_io_utils.py | 37 +++------- 5 files changed, 44 insertions(+), 94 deletions(-) diff --git a/.github/workflows/test_free_threaded.yml b/.github/workflows/test_free_threaded.yml index c1a8211f7..8828b7b6e 100644 --- a/.github/workflows/test_free_threaded.yml +++ b/.github/workflows/test_free_threaded.yml @@ -14,6 +14,7 @@ on: - 'tests/**' - 'pyproject.toml' - '.github/workflows/test_free_threaded.yml' + - '.github/scripts/**' # This job only reads the repository. permissions: @@ -82,5 +83,12 @@ jobs: if: steps.restore-test-data.outputs.cache-hit != 'true' run: python .github/scripts/cache_test_data.py + - name: save test data cache + if: steps.restore-test-data.outputs.cache-hit != 'true' + uses: actions/cache/save@v5 + with: + path: ${{ env.DATA_CACHE_PATH }} + key: ${{ env.DATA_CACHE_KEY }} + - name: run test suite without the GIL run: python -m pytest tests -m "not network" -q --timeout=900 diff --git a/dascore/utils/io.py b/dascore/utils/io.py index 2fc6cdd65..8042eff17 100644 --- a/dascore/utils/io.py +++ b/dascore/utils/io.py @@ -275,9 +275,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): self.close_all() def __del__(self): - # __del__ can run on a partially built manager (eg if __init__ raised). - if getattr(self, "_lock", None) is not None: - self.close_all() + self.close_all() def patch_to_xarray(patch: PatchType): diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 8a5e020f6..72bc6ba5b 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -7,11 +7,12 @@ import shutil import tempfile import warnings -from contextlib import ExitStack, contextmanager +from contextlib import contextmanager from contextvars import ContextVar +from functools import lru_cache from hashlib import sha256 from pathlib import Path -from threading import RLock, local +from threading import RLock from dascore.compat import UPath from dascore.config import get_config @@ -25,20 +26,10 @@ "only reading this file from the beginning is supported", ) _REMOTE_RESOURCE_CACHE: dict[str, UPath] = {} -# Resources already materialized this session, so the common case (the -# file is present) costs one dict lookup rather than a hash, a path -# build and a stat. Emptied by clear_remote_file_cache. -_REMOTE_MATERIALIZED: dict[tuple[str, Path], Path] = {} -# One lock per cached resource, so unrelated downloads still run at the -# same time. Entries are never removed: the dict is bounded by the number -# of distinct remote resources a session touches, and keeping them lets a -# cache clear hold every lock without racing new ones into existence. +# One lock per cached resource, so two threads never download the same +# file at once while unrelated downloads still run together. Entries are +# never removed; the dict is bounded by the resources a session touches. _REMOTE_KEY_LOCKS: dict[tuple[str, Path], RLock] = {} -# Guards the two dicts above. Always acquired before a download lock and -# never while holding one. -_REMOTE_CACHE_LOCK = RLock() -# Counts this thread's in-progress materializations (see clear). -_REMOTE_CACHE_LOCAL = local() _REMOTE_CACHE_SCOPE: ContextVar[str] = ContextVar( "remote_cache_scope", default="default" ) @@ -46,11 +37,9 @@ @_reinit_after_fork def _reinit_remote_cache_locks(): - """Install fresh remote-cache locks; see _reinit_after_fork.""" - global _REMOTE_CACHE_LOCK, _REMOTE_KEY_LOCKS, _REMOTE_CACHE_LOCAL - _REMOTE_CACHE_LOCK = RLock() + """Install fresh download locks; see _reinit_after_fork.""" + global _REMOTE_KEY_LOCKS _REMOTE_KEY_LOCKS = {} - _REMOTE_CACHE_LOCAL = local() @contextmanager @@ -90,19 +79,10 @@ def normalize_remote_id(path) -> str: serialized = json.dumps(storage_options, sort_keys=True, default=str) options_suffix = f"#{sha256(serialized.encode()).hexdigest()}" remote_id = f"{resource}{options_suffix}" - with _REMOTE_CACHE_LOCK: - _REMOTE_RESOURCE_CACHE[remote_id] = resource + _REMOTE_RESOURCE_CACHE[remote_id] = resource return remote_id -def _get_download_lock(key: tuple[str, Path]) -> RLock: - """Return the download lock for one cached resource, creating it once.""" - with _REMOTE_CACHE_LOCK: - if (lock := _REMOTE_KEY_LOCKS.get(key)) is None: - lock = _REMOTE_KEY_LOCKS[key] = RLock() - return lock - - def _normalize_cache_root(cache_root: Path | str) -> Path: """Return one normalized cache-root path.""" return Path(cache_root).expanduser() @@ -156,20 +136,14 @@ def clear_remote_file_cache(): """ Remove all locally cached remote files and memoized paths. - Clearing is a single-writer operation: it holds every download lock, - so materializations already in flight finish before their artifacts - are removed, and later ones download again. + Clearing is a single-writer operation: it is not synchronized against + materializations running at the same time, so run it while nothing + else is reading remote files. """ - if getattr(_REMOTE_CACHE_LOCAL, "materializing", 0): - msg = "Cannot clear the remote file cache from inside a materialization." - raise RuntimeError(msg) - with _REMOTE_CACHE_LOCK, ExitStack() as stack: - for lock in _REMOTE_KEY_LOCKS.values(): - stack.enter_context(lock) - shutil.rmtree(get_remote_cache_path(), ignore_errors=True) - get_remote_cache_path().mkdir(parents=True, exist_ok=True) - _REMOTE_RESOURCE_CACHE.clear() - _REMOTE_MATERIALIZED.clear() + shutil.rmtree(get_remote_cache_path(), ignore_errors=True) + get_remote_cache_path().mkdir(parents=True, exist_ok=True) + _REMOTE_RESOURCE_CACHE.clear() + _materialize_remote_file.cache_clear() def _download_remote_file(path, local_path: Path): @@ -197,36 +171,23 @@ def _download_remote_file(path, local_path: Path): tmp_path.unlink(missing_ok=True) +@lru_cache def _materialize_remote_file(remote_id: str, cache_root: Path) -> Path: """ Materialize one remote resource to a stable local cache path. Callers wanting the same resource take turns on its download lock: - the first downloads, the rest find the published file. A download - which fails publishes nothing, so the next caller retries it. + the first downloads, the rest find the published file. Failures are + not memoized, so the next caller retries the download. """ - key = (remote_id, cache_root) - with _REMOTE_CACHE_LOCK: - # Already materialized: skip hashing, path building and the stat. - if (published := _REMOTE_MATERIALIZED.get(key)) is not None: - return published - resource = _REMOTE_RESOURCE_CACHE.get(remote_id) + resource = _REMOTE_RESOURCE_CACHE.get(remote_id) if resource is None: resource = coerce_to_upath(remote_id.split("#", maxsplit=1)[0]) local_path = _remote_cache_local_path(remote_id, cache_root, resource) - # Record the depth so a cache clear attempted from inside a download - # (eg from a warning handler) raises rather than deleting live state. - depth = getattr(_REMOTE_CACHE_LOCAL, "materializing", 0) - _REMOTE_CACHE_LOCAL.materializing = depth + 1 - try: - with _get_download_lock(key): - if not local_path.exists(): - _download_to_cache(resource, local_path) - finally: - _REMOTE_CACHE_LOCAL.materializing = depth - # Only published paths are memoized, so a failed download is retried. - with _REMOTE_CACHE_LOCK: - _REMOTE_MATERIALIZED[key] = local_path + # setdefault is atomic, so every caller gets the same lock object. + with _REMOTE_KEY_LOCKS.setdefault((remote_id, cache_root), RLock()): + if not local_path.exists(): + _download_to_cache(resource, local_path) return local_path diff --git a/docs/recipes/parallelization.qmd b/docs/recipes/parallelization.qmd index ecc930194..67fc92352 100644 --- a/docs/recipes/parallelization.qmd +++ b/docs/recipes/parallelization.qmd @@ -31,7 +31,7 @@ There are two downsides to this approach. First, if the patches aren't chunked a # Thread Safety -DASCore's test suite runs on a free-threaded CPython build with the GIL disabled, and the guarantees below are what that supports. They apply equally to threads on a standard build. +DASCore's core test suite — no optional dependencies, no network tests — runs on a free-threaded CPython build with the GIL disabled. The guarantees below are what that supports; they apply equally to threads on a standard build. ## What threads may share diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index 8ddab2794..f57de0452 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -1220,7 +1220,7 @@ def _synchronized(path, local_path): monkeypatch.setattr(remote_io, "_download_remote_file", _synchronized) results = run_in_threads(lambda index: ensure_local_file(resources[index])) - assert all(x.exists() for x in results) + assert all(x is not None and x.exists() for x in results) def test_failed_download_is_retried(self, monkeypatch): """A failed download publishes nothing, so the next caller retries.""" @@ -1240,19 +1240,6 @@ def _fail_once(path, local_path): assert ensure_local_file(resource).exists() assert len(calls) == 2 - def test_clear_from_inside_materialization_raises(self, monkeypatch): - """Clearing the cache mid-download would delete live state.""" - resource = self._memory_file("clearing.bin") - - def _clear_while_downloading(path, local_path): - clear_remote_file_cache() - - monkeypatch.setattr( - remote_io, "_download_remote_file", _clear_while_downloading - ) - with pytest.raises(RuntimeError, match="inside a materialization"): - ensure_local_file(resource) - def test_unregistered_remote_id_is_coerced(self): """An id missing from the resource cache is rebuilt from the id itself.""" resource = self._memory_file("unregistered.bin") @@ -1261,19 +1248,15 @@ def test_unregistered_remote_id_is_coerced(self): local_path = remote_io._materialize_remote_file(str(resource), cache_root) assert local_path.exists() - def test_fork_handler_replaces_held_locks(self): - """Locks held at fork time are replaced so the child cannot deadlock.""" - old_lock = remote_io._REMOTE_CACHE_LOCK - try: - with old_lock: - remote_io._reinit_remote_cache_locks() - new_lock = remote_io._REMOTE_CACHE_LOCK - assert new_lock.acquire(blocking=False) - new_lock.release() - assert new_lock is not old_lock - assert not remote_io._REMOTE_KEY_LOCKS - finally: - remote_io._REMOTE_CACHE_LOCK = old_lock + def test_reinit_drops_download_locks(self): + """The fork hook drops locks a dead thread may have been holding.""" + resource = self._memory_file("forked.bin") + ensure_local_file(resource) + assert remote_io._REMOTE_KEY_LOCKS + old_locks = remote_io._REMOTE_KEY_LOCKS + remote_io._reinit_remote_cache_locks() + assert not remote_io._REMOTE_KEY_LOCKS + assert remote_io._REMOTE_KEY_LOCKS is not old_locks class TestIOResourceManagerConcurrency: