diff --git a/.github/workflows/test_free_threaded.yml b/.github/workflows/test_free_threaded.yml new file mode 100644 index 000000000..8828b7b6e --- /dev/null +++ b/.github/workflows/test_free_threaded.yml @@ -0,0 +1,94 @@ +# 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' + - '.github/scripts/**' + +# 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: 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 df111f14f..8042eff17 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, ) @@ -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 @@ -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.""" diff --git a/dascore/utils/remote_io.py b/dascore/utils/remote_io.py index 185016694..72bc6ba5b 100644 --- a/dascore/utils/remote_io.py +++ b/dascore/utils/remote_io.py @@ -12,10 +12,12 @@ from functools import lru_cache from hashlib import sha256 from pathlib import Path +from threading import RLock 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 +26,22 @@ "only reading this file from the beginning is supported", ) _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 +# never removed; the dict is bounded by the resources a session touches. +_REMOTE_KEY_LOCKS: dict[tuple[str, Path], RLock] = {} _REMOTE_CACHE_SCOPE: ContextVar[str] = ContextVar( "remote_cache_scope", default="default" ) +@_reinit_after_fork +def _reinit_remote_cache_locks(): + """Install fresh download locks; see _reinit_after_fork.""" + global _REMOTE_KEY_LOCKS + _REMOTE_KEY_LOCKS = {} + + @contextmanager def remote_cache_scope(scope: str): """Temporarily set the current remote-cache policy scope.""" @@ -70,16 +83,22 @@ def normalize_remote_id(path) -> str: return remote_id -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) @@ -114,11 +133,17 @@ def _warn_remote_cache_download(resource: UPath, local_path: Path): def clear_remote_file_cache(): - """Remove all locally cached remote files and memoized paths.""" + """ + Remove all locally cached remote files and memoized paths. + + 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. + """ 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() + _materialize_remote_file.cache_clear() def _download_remote_file(path, local_path: Path): @@ -147,45 +172,52 @@ def _download_remote_file(path, local_path: Path): @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.""" +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. Failures are + not memoized, so the next caller retries the download. + """ 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) - ) - 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) + local_path = _remote_cache_local_path(remote_id, cache_root, resource) + # 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 +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): @@ -208,9 +240,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 diff --git a/docs/recipes/parallelization.qmd b/docs/recipes/parallelization.qmd index 34dbb9efb..67fc92352 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 @@ -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 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 + +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 399b379a3..a2a353ee9 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 @@ -1212,3 +1213,110 @@ 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 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.""" + 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_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_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: + """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