ENH: synchronize remote IO, add free-threaded CI and concurrency docs

.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

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."""

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