From cc05e3e2e10aa992b2f33fd20a8e61526f615603 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 20:46:50 +0000 Subject: [PATCH 01/15] Read missing annexed content with datalad-fuse, cached by annex key Add `dandi.support.datalad_fuse.AnnexedReadableFile`, a `Readable` for a locked annexed file whose content is not present locally, streamed with the adapter of datalad-fuse (no FUSE mount), which asks git-annex where the content is. It is available with the new `datalad` extra. Cache NWB metadata of locked annexed files, streamed or local, under their git-annex key paired with their path, with fscacher 0.5.0's `annex_key_fingerprint`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- .github/workflows/run-tests.yml | 7 ++ dandi/metadata/nwb.py | 3 +- dandi/pynwb_utils.py | 25 +++- dandi/support/datalad_fuse.py | 147 +++++++++++++++++++++++ dandi/support/tests/test_datalad_fuse.py | 97 +++++++++++++++ dandi/tests/fixtures.py | 123 +++++++++++++++++++ dandi/tests/skip.py | 5 + pyproject.toml | 8 +- 8 files changed, 410 insertions(+), 5 deletions(-) create mode 100644 dandi/support/datalad_fuse.py create mode 100644 dandi/support/tests/test_datalad_fuse.py diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 948b34353..97c9ab948 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -73,6 +73,9 @@ jobs: - os: ubuntu-latest python: '3.11' mode: lowest-deps + - os: ubuntu-latest + python: '3.12' + mode: datalad steps: - name: Set up environment @@ -123,6 +126,10 @@ jobs: git+https://github.com/hdmf-dev/hdmf \ git+https://github.com/hdmf-dev/hdmf-zarr + - name: Install git-annex and datalad-fuse (for streaming annexed content) + if: matrix.mode == 'datalad' + run: pip install git-annex ".[datalad]" + - name: Create NFS filesystem if: matrix.mode == 'nfs' run: | diff --git a/dandi/metadata/nwb.py b/dandi/metadata/nwb.py index fe7820f2a..43a33d264 100644 --- a/dandi/metadata/nwb.py +++ b/dandi/metadata/nwb.py @@ -16,6 +16,7 @@ from ..misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, Readable from ..pynwb_utils import ( _get_pynwb_metadata, + annex_fingerprint, get_neurodata_types, get_nwb_version, ignore_benign_pynwb_warnings, @@ -28,7 +29,7 @@ # Disable this for clean hacking -@metadata_cache.memoize_path +@metadata_cache.memoize_path(custom_fingerprint=annex_fingerprint) def get_metadata( path: str | Path | Readable, digest: Digest | None = None ) -> dict[str, Any]: diff --git a/dandi/pynwb_utils.py b/dandi/pynwb_utils.py index 297df05f3..da6e4d835 100644 --- a/dandi/pynwb_utils.py +++ b/dandi/pynwb_utils.py @@ -23,7 +23,7 @@ import warnings import dandischema -from fscacher import PersistentCache +from fscacher import PersistentCache, annex_key_fingerprint import h5py import hdmf import numpy as np @@ -73,6 +73,25 @@ ) +def annex_fingerprint(source: Any) -> tuple[str, str] | None: + """ + Fingerprint of ``source`` for ``PersistentCache.memoize_path`` + + Pass it as ``custom_fingerprint`` to cache the results of a function of a + local path or a `Readable` under the git-annex key of a locked annexed + file, paired with its path (see `fscacher.annex_key_fingerprint`), rather + than under ``stat()``: the content of an `AnnexedReadableFile`, which is + not present locally, cannot be ``stat()``-ed. Anything else is handled as + without this. + """ + # Avoid circular import: + from .support.datalad_fuse import AnnexedReadableFile + + if isinstance(source, AnnexedReadableFile): + source = source.filepath + return cast("tuple[str, str] | None", annex_key_fingerprint(source)) + + def _sanitize_nwb_version( v: Any, filename: str | Path | None = None, @@ -194,7 +213,7 @@ def get_neurodata_types_to_modalities_map() -> dict[str, str]: return ndtypes -@metadata_cache.memoize_path +@metadata_cache.memoize_path(custom_fingerprint=annex_fingerprint) def get_neurodata_types(filepath: str | Path | Readable) -> list[str]: with open_readable(filepath) as fp, h5py.File(fp, "r") as h5file: all_pairs = _scan_neurodata_types(h5file) @@ -808,7 +827,7 @@ def copy_nwb_file(src: str | Path, dest: str | Path) -> str: return str(dest) -@metadata_cache.memoize_path +@metadata_cache.memoize_path(custom_fingerprint=annex_fingerprint) def nwb_has_external_links(filepath: str | Path | Readable) -> bool: with open_readable(filepath) as f, h5py.File(f, "r") as fp: visited = set() diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py new file mode 100644 index 000000000..8460039f3 --- /dev/null +++ b/dandi/support/datalad_fuse.py @@ -0,0 +1,147 @@ +""" +Reading the content of annexed files of git-annex_ repositories (such as +DataLad_ datasets) that is not present locally, with datalad-fuse_. + +In a git-annex repository, a (locked) annexed file is a symbolic link into +:file:`.git/annex/objects/`, which is broken until the content is fetched. +datalad-fuse can mount such a repository with FUSE so that the content is +streamed on demand, but its adapter, which the mount is built on, also opens +the files directly with no mount. It asks git-annex where the content is +(``git annex whereis``) and streams it over HTTP(S) with fsspec_. That adapter +is used here, so that the files of a DataLad Dandiset can be read (e.g., for +their metadata, or to be validated) without downloading the (possibly +terabytes of) data. + +This requires datalad-fuse (``pip install "dandi[datalad]"``), which requires +DataLad and git-annex. + +.. _git-annex: https://git-annex.branchable.com +.. _DataLad: https://www.datalad.org +.. _datalad-fuse: https://github.com/datalad/datalad-fuse +.. _fsspec: https://github.com/fsspec/filesystem_spec +""" + +from __future__ import annotations + +import atexit +from dataclasses import dataclass, field +from datetime import datetime +from functools import cache +import logging +import os +from pathlib import Path +import subprocess +from typing import IO, Any, cast + +from ..misctypes import Readable + +lgr = logging.getLogger("dandi.support.datalad_fuse") + + +@cache +def get_adapter() -> Any: + """ + The datalad-fuse adapter shared by all files, rooted at the filesystem + root so that it serves files of any dataset, and closed at exit + """ + # Optional dependency: + from datalad_fuse.fsspec import FsspecAdapter + + # caching=False: do not keep the streamed blocks on disk, in the dataset + adapter = FsspecAdapter(Path(os.path.abspath(os.sep)), caching=False) + atexit.register(adapter.__exit__, None, None, None) + return adapter + + +@cache +def annex_initialized(directory: Path) -> bool: + """ + Whether ``directory`` is in a Git repository in which git-annex is + initialized. (DataLad would otherwise initialize it, modifying the + repository.) + """ + try: + r = subprocess.run( + ["git", "-C", str(directory), "config", "--get", "annex.uuid"], + capture_output=True, + text=True, + ) + except FileNotFoundError: + return False + return r.returncode == 0 and bool(r.stdout.strip()) + + +@dataclass +class AnnexedReadableFile(Readable): + """ + A `Readable` for a (locked) annexed file whose content is not present + locally, which is streamed with datalad-fuse (see the module docstring). + + Instances are obtained by calling `get_annexed_readable()`. + """ + + #: The absolute path of the (broken) symbolic link to the file's content + filepath: Path + + #: The git-annex key of the file + key: str + + #: The size of the content, as recorded in the key + size: int + + #: The first URL that the content may be streamed from (for reporting; + #: the adapter tries the others if it cannot be read from that one) + url: str + + adapter: Any = field(repr=False, compare=False) + + def open(self) -> IO[bytes]: + return cast("IO[bytes]", self.adapter.open(self.filepath, "rb")) + + def get_size(self) -> int: + return self.size + + def get_mtime(self) -> datetime | None: + return None + + def get_filename(self) -> str: + return self.filepath.name + + def __str__(self) -> str: + return str(self.filepath) + + +def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: + """ + Return an `AnnexedReadableFile` for streaming the content of the annexed + file at ``path``, or `None` if datalad-fuse is not installed, ``path`` is + not a (locked) annexed file whose content is missing in a repository in + which git-annex is initialized, its key does not record the size of the + content, or git-annex knows no URL to stream it from + """ + filepath = Path(os.path.abspath(path)) + if not filepath.is_symlink() or filepath.exists(): + # Not a broken symbolic link + return None + try: + from datalad_fuse.fsspec import FileState + except ImportError: + return None + if not annex_initialized(filepath.parent): + return None + try: + dsap, relpath = get_adapter().resolve_dataset(filepath) + state, key = dsap.get_file_state(relpath) + except Exception as e: + # E.g., a repository without any commit + lgr.debug("%s: Cannot be read with datalad-fuse: %s", filepath, e) + return None + if state is not FileState.NO_CONTENT or key is None or key.size is None: + return None + url = next(dsap.get_urls(str(key)), None) + if url is None: + lgr.debug("%s: git-annex knows no URL for key %s", filepath, key) + return None + return AnnexedReadableFile( + filepath=filepath, key=str(key), size=key.size, url=url, adapter=get_adapter() + ) diff --git a/dandi/support/tests/test_datalad_fuse.py b/dandi/support/tests/test_datalad_fuse.py new file mode 100644 index 000000000..6dc31495a --- /dev/null +++ b/dandi/support/tests/test_datalad_fuse.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from pathlib import Path +import shutil + +from fscacher import PersistentCache +import h5py +import pytest + +from ..datalad_fuse import AnnexedReadableFile, get_annexed_readable +from ...pynwb_utils import annex_fingerprint, get_neurodata_types +from ...tests.fixtures import RangeHTTPServer, make_git_annex_dandiset +from ...tests.skip import mark + +pytestmark = mark.skipif_no_git_annex + + +@pytest.fixture +def streamed_nwb( + range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path +) -> tuple[Path, RangeHTTPServer]: + """ + An annexed copy of ``simple2_nwb`` in a git-annex repository, whose content + is not present but registered at a URL of the HTTP server + """ + pytest.importorskip("datalad_fuse.fsspec") + server, base_url = range_http_server + shutil.copy(simple2_nwb, tmp_path / "served" / "content.nwb") + make_git_annex_dandiset( + tmp_path / "ds", + {"sub-01/sub-01.nwb": (simple2_nwb, f"{base_url}/content.nwb")}, + ) + nwb = tmp_path / "ds" / "sub-01" / "sub-01.nwb" + assert not nwb.exists() + return nwb, server + + +@pytest.mark.ai_generated +def test_annexed_readable( + streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path +) -> None: + nwb, server = streamed_nwb + r = get_annexed_readable(nwb) + assert isinstance(r, AnnexedReadableFile) + assert r.key.startswith(f"SHA256E-s{simple2_nwb.stat().st_size}--") + assert r.url.endswith("/content.nwb") + assert r.get_size() == simple2_nwb.stat().st_size + assert r.get_filename() == "sub-01.nwb" + fp = r.open() + try: + # Random access over HTTP is enough for h5py to read the file + with h5py.File(fp, "r") as h5: + assert h5.attrs["nwb_version"] + finally: + fp.close() + assert server.ranged_requests > 0 + + +@pytest.mark.ai_generated +def test_annexed_readable_none( + streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path +) -> None: + nwb, _ = streamed_nwb + # The content is present + (nwb.parent / "regular.nwb").write_bytes(b"content") + assert get_annexed_readable(nwb.parent / "regular.nwb") is None + # Not in a Git repository + assert get_annexed_readable(simple2_nwb) is None + # A broken symbolic link in a repository without git-annex initialized + plain = tmp_path / "plain" + plain.mkdir() + shutil.copytree(nwb.parent, plain / "sub-01", symlinks=True) + shutil.copytree(nwb.parent.parent / ".git", plain / ".git") + (plain / ".git" / "config").write_text("[core]\n\tbare = false\n") + assert get_annexed_readable(plain / "sub-01" / "sub-01.nwb") is None + + +@pytest.mark.ai_generated +def test_annexed_readable_cached_by_key( + streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path +) -> None: + nwb, server = streamed_nwb + cache = PersistentCache(path=tmp_path / "cache") + neurodata_types = cache.memoize_path(custom_fingerprint=annex_fingerprint)( + get_neurodata_types.__wrapped__ + ) + r = get_annexed_readable(nwb) + assert r is not None + expected = get_neurodata_types.__wrapped__(simple2_nwb) + assert neurodata_types(r) == expected + requests = server.ranged_requests + assert requests > 0 + # Served from the cache, without the content being read again: by the + # readable, and by the path of the (broken) link itself + assert neurodata_types(get_annexed_readable(nwb)) == expected + assert neurodata_types(nwb) == expected + assert server.ranged_requests == requests diff --git a/dandi/tests/fixtures.py b/dandi/tests/fixtures.py index 9067b0937..bf5fea55b 100644 --- a/dandi/tests/fixtures.py +++ b/dandi/tests/fixtures.py @@ -3,12 +3,16 @@ from collections.abc import Callable, Iterator from dataclasses import dataclass, field, replace from datetime import datetime, timezone +from functools import partial +from http import HTTPStatus +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer import logging import os from pathlib import Path import re import shutil from subprocess import DEVNULL, check_output, run +import threading from time import sleep from typing import Any, Literal from uuid import uuid4 @@ -363,6 +367,125 @@ def fixture( return fixture +def make_git_annex_dandiset(dandiset: Path, files: dict[str, tuple[Path, str]]) -> None: + """ + Make the directory ``dandiset`` (created if needed, with a minimal + :file:`dandiset.yaml`) a git-annex repository looking like a clone of a + DataLad Dandiset whose annexed content has not been fetched: for each + ``relpath: (source, url)`` item of ``files``, a copy of ``source`` is + annexed at ``relpath`` with ``url`` registered for it, and then dropped. + This needs git-annex. + """ + env = { + **os.environ, + "GIT_AUTHOR_NAME": "DANDI tests", + "GIT_AUTHOR_EMAIL": "tests@dandiarchive.org", + "GIT_COMMITTER_NAME": "DANDI tests", + "GIT_COMMITTER_EMAIL": "tests@dandiarchive.org", + } + + def git(*args: str) -> str: + return check_output( + ["git", "-C", str(dandiset), *args], env=env, text=True + ).strip() + + dandiset.mkdir(parents=True, exist_ok=True) + (dandiset / dandiset_metadata_file).write_text( + "identifier: '000027'\nname: Test\ndescription: Test dandiset\n" + ) + git("init", "-q") + git("annex", "init", "-q") + git("config", "annex.backend", "SHA256E") + git("add", dandiset_metadata_file) + for relpath, (source, _) in files.items(): + (dandiset / relpath).parent.mkdir(parents=True, exist_ok=True) + shutil.copy(source, dandiset / relpath) + git("annex", "add", "-q", *files) + for relpath, (_, url) in files.items(): + key = git("annex", "lookupkey", relpath) + git("annex", "registerurl", key, url) + git("commit", "-q", "-m", "Add files") + git("annex", "drop", "-q", "--force", *files) + + +class RangeHTTPServer(ThreadingHTTPServer): + """An HTTP server counting the range requests it served""" + + ranged_requests: int = 0 + + +class RangeHTTPRequestHandler(SimpleHTTPRequestHandler): + """ + A handler serving files with support for ``HEAD`` and single-range ``GET`` + requests, like S3 does + """ + + def log_message(self, format: str, *args: Any) -> None: + pass + + def _send_headers( + self, status: HTTPStatus, length: int, content_range: str | None = None + ) -> None: + self.send_response(status) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Content-Length", str(length)) + self.send_header("Accept-Ranges", "bytes") + if content_range is not None: + self.send_header("Content-Range", content_range) + self.end_headers() + + def do_HEAD(self) -> None: + path = self.translate_path(self.path) + if not os.path.isfile(path): + self.send_error(HTTPStatus.NOT_FOUND) + return + self._send_headers(HTTPStatus.OK, os.path.getsize(path)) + + def do_GET(self) -> None: + path = self.translate_path(self.path) + if not os.path.isfile(path): + self.send_error(HTTPStatus.NOT_FOUND) + return + size = os.path.getsize(path) + start, end = 0, size - 1 + if m := re.fullmatch(r"bytes=(\d+)-(\d*)", self.headers.get("Range", "")): + start = int(m[1]) + end = min(int(m[2]) if m[2] else size - 1, size - 1) + assert isinstance(self.server, RangeHTTPServer) + self.server.ranged_requests += 1 + self._send_headers( + HTTPStatus.PARTIAL_CONTENT, + end - start + 1, + f"bytes {start}-{end}/{size}", + ) + else: + self._send_headers(HTTPStatus.OK, size) + with open(path, "rb") as fp: + fp.seek(start) + self.wfile.write(fp.read(end - start + 1)) + + +@pytest.fixture() +def range_http_server(tmp_path: Path) -> Iterator[tuple[RangeHTTPServer, str]]: + """ + Serve the files in ``tmp_path / "served"`` over HTTP with support for range + requests; yields the server and its base URL + """ + served = tmp_path / "served" + served.mkdir() + server = RangeHTTPServer( + ("127.0.0.1", 0), partial(RangeHTTPRequestHandler, directory=str(served)) + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server, f"http://127.0.0.1:{server.server_address[1]}" + finally: + server.shutdown() + server.server_close() + thread.join() + + def _make_subdirs_dandisets(path: Path) -> None: for bids_dataset_path in path.iterdir(): if bids_dataset_path.is_dir(): diff --git a/dandi/tests/skip.py b/dandi/tests/skip.py index 54d864b76..9e30b195d 100644 --- a/dandi/tests/skip.py +++ b/dandi/tests/skip.py @@ -125,6 +125,10 @@ def no_git(): return "Git not installed", shutil.which("git") is None +def no_git_annex(): + return "git-annex not installed", shutil.which("git-annex") is None + + # ### END MODIFIED CODE @@ -162,6 +166,7 @@ def on_windows(): no_docker_commands, no_docker_engine, no_git, + no_git_annex, no_network, # no_singularity, no_ssh, diff --git a/pyproject.toml b/pyproject.toml index 7a6ba7b9b..3fcdedd51 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,7 +53,7 @@ dependencies = [ "dandischema >= 0.12.0, != 0.13.0, < 0.15.0", "etelemetry >= 0.2.2", "fasteners >= 0.19", - "fscacher >= 0.3.0", + "fscacher >= 0.5.0", # Floor raised to what `py3-lowest` actually resolves: modern pynwb pins # hdmf >= 4, so declared floor is now the tested floor. "hdmf >= 4.1.0", @@ -120,6 +120,11 @@ test = [ # 5.0 switched to urllib3.connection.VerifiedHTTPSConnection (urllib3 2.x compat) "vcrpy >= 5.0.0", ] +# For reading the content of annexed files of DataLad datasets without +# fetching it (requires git-annex) +datalad = [ + "datalad-fuse >= 0.5.1", +] tools = [ "boto3", ] @@ -215,6 +220,7 @@ follow_imports = "skip" module = [ "bidsschematools.*", "click_didyoumean.*", + "datalad_fuse.*", "etelemetry.*", "fasteners.*", "fscacher.*", From e1db5a801b362ca11ac0b8ea56cb2d2ad6eacc05 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 21:17:17 +0000 Subject: [PATCH 02/15] Move annex_fingerprint next to AnnexedReadableFile; trim field comments annex_fingerprint is not specific to NWB files, and living in dandi.support.datalad_fuse it needs no deferred import. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/metadata/nwb.py | 2 +- dandi/pynwb_utils.py | 22 ++------------- dandi/support/datalad_fuse.py | 34 ++++++++++++++++-------- dandi/support/tests/test_datalad_fuse.py | 4 +-- 4 files changed, 28 insertions(+), 34 deletions(-) diff --git a/dandi/metadata/nwb.py b/dandi/metadata/nwb.py index 43a33d264..a773475d6 100644 --- a/dandi/metadata/nwb.py +++ b/dandi/metadata/nwb.py @@ -16,13 +16,13 @@ from ..misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, Readable from ..pynwb_utils import ( _get_pynwb_metadata, - annex_fingerprint, get_neurodata_types, get_nwb_version, ignore_benign_pynwb_warnings, metadata_cache, nwb_has_external_links, ) +from ..support.datalad_fuse import annex_fingerprint from ..utils import find_parent_directory_containing lgr = get_logger() diff --git a/dandi/pynwb_utils.py b/dandi/pynwb_utils.py index da6e4d835..c49293f57 100644 --- a/dandi/pynwb_utils.py +++ b/dandi/pynwb_utils.py @@ -23,7 +23,7 @@ import warnings import dandischema -from fscacher import PersistentCache, annex_key_fingerprint +from fscacher import PersistentCache import h5py import hdmf import numpy as np @@ -42,6 +42,7 @@ metadata_nwb_subject_fields, ) from .misctypes import Readable +from .support.datalad_fuse import annex_fingerprint from .utils import get_module_version, is_url from .validate._types import ( Origin, @@ -73,25 +74,6 @@ ) -def annex_fingerprint(source: Any) -> tuple[str, str] | None: - """ - Fingerprint of ``source`` for ``PersistentCache.memoize_path`` - - Pass it as ``custom_fingerprint`` to cache the results of a function of a - local path or a `Readable` under the git-annex key of a locked annexed - file, paired with its path (see `fscacher.annex_key_fingerprint`), rather - than under ``stat()``: the content of an `AnnexedReadableFile`, which is - not present locally, cannot be ``stat()``-ed. Anything else is handled as - without this. - """ - # Avoid circular import: - from .support.datalad_fuse import AnnexedReadableFile - - if isinstance(source, AnnexedReadableFile): - source = source.filepath - return cast("tuple[str, str] | None", annex_key_fingerprint(source)) - - def _sanitize_nwb_version( v: Any, filename: str | Path | None = None, diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py index 8460039f3..a33c84fe9 100644 --- a/dandi/support/datalad_fuse.py +++ b/dandi/support/datalad_fuse.py @@ -33,6 +33,8 @@ import subprocess from typing import IO, Any, cast +from fscacher import annex_key_fingerprint + from ..misctypes import Readable lgr = logging.getLogger("dandi.support.datalad_fuse") @@ -74,25 +76,19 @@ def annex_initialized(directory: Path) -> bool: @dataclass class AnnexedReadableFile(Readable): """ - A `Readable` for a (locked) annexed file whose content is not present - locally, which is streamed with datalad-fuse (see the module docstring). + A `Readable` for a (locked) annexed file at ``filepath`` (a broken symbolic + link) whose content is not present locally, which is streamed with + datalad-fuse (see the module docstring). ``url`` is only the first URL + known for the content, for reporting: if it cannot be read, the adapter + tries the others. Instances are obtained by calling `get_annexed_readable()`. """ - #: The absolute path of the (broken) symbolic link to the file's content filepath: Path - - #: The git-annex key of the file key: str - - #: The size of the content, as recorded in the key size: int - - #: The first URL that the content may be streamed from (for reporting; - #: the adapter tries the others if it cannot be read from that one) url: str - adapter: Any = field(repr=False, compare=False) def open(self) -> IO[bytes]: @@ -111,6 +107,22 @@ def __str__(self) -> str: return str(self.filepath) +def annex_fingerprint(source: Any) -> tuple[str, str] | None: + """ + Fingerprint of ``source`` for ``PersistentCache.memoize_path`` + + Pass it as ``custom_fingerprint`` to cache the results of a function of a + local path or a `Readable` under the git-annex key of a locked annexed + file, paired with its path (see `fscacher.annex_key_fingerprint`), rather + than under ``stat()``: the content of an `AnnexedReadableFile`, which is + not present locally, cannot be ``stat()``-ed. Anything else is handled as + without this. + """ + if isinstance(source, AnnexedReadableFile): + source = source.filepath + return cast("tuple[str, str] | None", annex_key_fingerprint(source)) + + def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: """ Return an `AnnexedReadableFile` for streaming the content of the annexed diff --git a/dandi/support/tests/test_datalad_fuse.py b/dandi/support/tests/test_datalad_fuse.py index 6dc31495a..6f594eb07 100644 --- a/dandi/support/tests/test_datalad_fuse.py +++ b/dandi/support/tests/test_datalad_fuse.py @@ -7,8 +7,8 @@ import h5py import pytest -from ..datalad_fuse import AnnexedReadableFile, get_annexed_readable -from ...pynwb_utils import annex_fingerprint, get_neurodata_types +from ..datalad_fuse import AnnexedReadableFile, annex_fingerprint, get_annexed_readable +from ...pynwb_utils import get_neurodata_types from ...tests.fixtures import RangeHTTPServer, make_git_annex_dandiset from ...tests.skip import mark From 428acafda4511bd62b53c02dd598d77f232bc495 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 21:18:07 +0000 Subject: [PATCH 03/15] Note URL order and proxy behavior of datalad-fuse As documented in datalad/datalad-fuse#138. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/datalad_fuse.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py index a33c84fe9..cfe71d895 100644 --- a/dandi/support/datalad_fuse.py +++ b/dandi/support/datalad_fuse.py @@ -12,6 +12,10 @@ their metadata, or to be validated) without downloading the (possibly terabytes of) data. +The URLs are tried in the order git-annex lists them; for a DataLad Dandiset, +that is the DANDI Archive's API download URL (which redirects to S3) before the +direct S3 URL. datalad-fuse does not use HTTP proxy environment variables. + This requires datalad-fuse (``pip install "dandi[datalad]"``), which requires DataLad and git-annex. @@ -49,7 +53,8 @@ def get_adapter() -> Any: # Optional dependency: from datalad_fuse.fsspec import FsspecAdapter - # caching=False: do not keep the streamed blocks on disk, in the dataset + # caching=False: do not keep the streamed blocks on disk, in the dataset. + # The root and all paths passed to the adapter must be absolute. adapter = FsspecAdapter(Path(os.path.abspath(os.sep)), caching=False) atexit.register(adapter.__exit__, None, None, None) return adapter From 55df8f7ad9e2d5dd4386d41698e66ac9ca8f6cac Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 21:26:15 +0000 Subject: [PATCH 04/15] Import fscacher lazily in dandi.support.datalad_fuse fscacher imports joblib, which imports numpy, so importing it with the module made the CLI (via dandi.validate in the stacked PR) import numpy, failing test_no_heavy_imports. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/datalad_fuse.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py index cfe71d895..54c27a18c 100644 --- a/dandi/support/datalad_fuse.py +++ b/dandi/support/datalad_fuse.py @@ -37,8 +37,6 @@ import subprocess from typing import IO, Any, cast -from fscacher import annex_key_fingerprint - from ..misctypes import Readable lgr = logging.getLogger("dandi.support.datalad_fuse") @@ -123,6 +121,10 @@ def annex_fingerprint(source: Any) -> tuple[str, str] | None: not present locally, cannot be ``stat()``-ed. Anything else is handled as without this. """ + # Avoid heavy import (fscacher imports joblib, which imports numpy) when + # this module is imported, e.g., by the CLI: + from fscacher import annex_key_fingerprint + if isinstance(source, AnnexedReadableFile): source = source.filepath return cast("tuple[str, str] | None", annex_key_fingerprint(source)) From fec927ff9bb22c2b9797b5a9283bba2b8a60f214 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 02:40:19 +0000 Subject: [PATCH 05/15] Disable automatic GC while a streamed annexed file is open h5py holds its global lock while reading from a Python file object, and the read waits for another thread to fetch the data over HTTP (fsspec's I/O thread, or an HTTP server in the same process as in the tests). A garbage collection in that thread could then run the finalizer of an unreferenced h5py-backed object (e.g., hdmf's HDF5IO.__del__), which waits for the same lock: a deadlock, as seen in CI in test_validate_stream, ended by pytest-timeout with a segfault. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/datalad_fuse.py | 60 +++++++++++++++++++++++- dandi/support/tests/test_datalad_fuse.py | 21 +++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py index 54c27a18c..1dfaf117c 100644 --- a/dandi/support/datalad_fuse.py +++ b/dandi/support/datalad_fuse.py @@ -31,10 +31,12 @@ from dataclasses import dataclass, field from datetime import datetime from functools import cache +import gc import logging import os from pathlib import Path import subprocess +import threading from typing import IO, Any, cast from ..misctypes import Readable @@ -76,6 +78,62 @@ def annex_initialized(directory: Path) -> bool: return r.returncode == 0 and bool(r.stdout.strip()) +_gc_lock = threading.Lock() +_gc_pauses = 0 +_gc_was_enabled = False + + +class _GCPausedFile: + """ + A file object streamed by datalad-fuse, during whose lifetime automatic + garbage collection is disabled + + h5py holds a global lock while it reads from a Python file object, and the + read waits for another thread (fsspec's I/O thread, or an HTTP server in + the same process) to fetch the data. If a garbage collection ran in that + thread meanwhile, finalizers of h5py objects (e.g., of an unreferenced + ``NWBHDF5IO``) would wait for the same lock, deadlocking the process. + """ + + def __init__(self, fp: Any) -> None: + global _gc_pauses, _gc_was_enabled + with _gc_lock: + if _gc_pauses == 0: + # Collect pending garbage now, in this thread, which does not + # hold h5py's lock yet + gc.collect() + _gc_was_enabled = gc.isenabled() + gc.disable() + _gc_pauses += 1 + self._fp = fp + self._paused = True + + def __getattr__(self, name: str) -> Any: + return getattr(self._fp, name) + + def __enter__(self) -> _GCPausedFile: + return self + + def __exit__(self, *_exc: Any) -> None: + self.close() + + def __del__(self) -> None: + if self._paused: + self.close() + + def close(self) -> None: + global _gc_pauses + try: + self._fp.close() + finally: + with _gc_lock: + if self._paused: + self._paused = False + _gc_pauses -= 1 + if _gc_pauses == 0 and _gc_was_enabled: + gc.enable() + + @dataclass class AnnexedReadableFile(Readable): """ @@ -95,7 +153,7 @@ class AnnexedReadableFile(Readable): adapter: Any = field(repr=False, compare=False) def open(self) -> IO[bytes]: - return cast("IO[bytes]", self.adapter.open(self.filepath, "rb")) + return cast("IO[bytes]", _GCPausedFile(self.adapter.open(self.filepath, "rb"))) def get_size(self) -> int: return self.size diff --git a/dandi/support/tests/test_datalad_fuse.py b/dandi/support/tests/test_datalad_fuse.py index 6f594eb07..d75b20c32 100644 --- a/dandi/support/tests/test_datalad_fuse.py +++ b/dandi/support/tests/test_datalad_fuse.py @@ -1,5 +1,6 @@ from __future__ import annotations +import gc from pathlib import Path import shutil @@ -56,6 +57,26 @@ def test_annexed_readable( assert server.ranged_requests > 0 +@pytest.mark.ai_generated +def test_annexed_readable_pauses_gc( + streamed_nwb: tuple[Path, RangeHTTPServer], +) -> None: + # Automatic garbage collection is disabled while any streamed file is open + # (to avoid deadlocks with h5py's lock; see `_GCPausedFile`) + nwb, _ = streamed_nwb + r = get_annexed_readable(nwb) + assert r is not None + assert gc.isenabled() + with r.open() as fp: + assert not gc.isenabled() + with r.open() as fp2: + assert fp2.read(8) == b"\x89HDF\r\n\x1a\n" + assert not gc.isenabled() + fp.seek(1) + assert fp.read(3) == b"HDF" + assert gc.isenabled() + + @pytest.mark.ai_generated def test_annexed_readable_none( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path From 7e62bcc2653b68901a7c0a8f7cfd325d0507b2d5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 22:00:54 +0000 Subject: [PATCH 06/15] Read streamed annexed content in small cached blocks datalad-fuse opens the files with fsspec's defaults: 5 MiB of read-ahead, discarded whenever a read jumps elsewhere. h5py reads many small pieces scattered over an NWB file, so each open downloaded about as much as the whole file (64 MB for the neurodata types of a 16 MB file in the new test), and dandi opens each file several times during validation. Read in cached blocks of 256 KiB instead (1.3 MB for the same file). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/datalad_fuse.py | 22 ++++++++++++++ dandi/support/tests/test_datalad_fuse.py | 37 ++++++++++++++++++++++++ dandi/tests/fixtures.py | 5 +++- 3 files changed, 63 insertions(+), 1 deletion(-) diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py index 1dfaf117c..4629e9f0f 100644 --- a/dandi/support/datalad_fuse.py +++ b/dandi/support/datalad_fuse.py @@ -188,6 +188,27 @@ def annex_fingerprint(source: Any) -> tuple[str, str] | None: return cast("tuple[str, str] | None", annex_key_fingerprint(source)) +#: Size of the blocks in which streamed content is read and cached +STREAM_BLOCK_SIZE = 256 * 1024 + + +def _read_in_small_blocks(dsap: Any) -> None: + """ + Make the files of a dataset adapter be read in cached blocks of + `STREAM_BLOCK_SIZE` bytes rather than with fsspec's default: h5py reads + many small pieces scattered over the file, and with the default (5 MiB of + read-ahead, discarded whenever a read jumps elsewhere) reading an NWB file + downloads about as much as the whole file each time it is opened + """ + # Avoid heavy import when this module is imported: + from fsspec.implementations.http import HTTPFileSystem + + fs = getattr(dsap, "fs", None) + if isinstance(fs, HTTPFileSystem): + fs.block_size = STREAM_BLOCK_SIZE + fs.cache_type = "blockcache" + + def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: """ Return an `AnnexedReadableFile` for streaming the content of the annexed @@ -208,6 +229,7 @@ def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: return None try: dsap, relpath = get_adapter().resolve_dataset(filepath) + _read_in_small_blocks(dsap) state, key = dsap.get_file_state(relpath) except Exception as e: # E.g., a repository without any commit diff --git a/dandi/support/tests/test_datalad_fuse.py b/dandi/support/tests/test_datalad_fuse.py index d75b20c32..8e58bc036 100644 --- a/dandi/support/tests/test_datalad_fuse.py +++ b/dandi/support/tests/test_datalad_fuse.py @@ -1,11 +1,15 @@ from __future__ import annotations +from datetime import datetime, timezone import gc from pathlib import Path import shutil from fscacher import PersistentCache import h5py +from hdmf.backends.hdf5 import H5DataIO +import numpy as np +from pynwb import NWBHDF5IO, NWBFile, TimeSeries import pytest from ..datalad_fuse import AnnexedReadableFile, annex_fingerprint, get_annexed_readable @@ -77,6 +81,39 @@ def test_annexed_readable_pauses_gc( assert gc.isenabled() +@pytest.mark.ai_generated +def test_annexed_readable_reads_little( + range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path +) -> None: + """Reading an NWB file's structure downloads only a small part of it""" + pytest.importorskip("datalad_fuse.fsspec") + server, base_url = range_http_server + nwbfile = NWBFile("streamed", "id", datetime(2026, 1, 1, tzinfo=timezone.utc)) + for i in range(4): + nwbfile.add_acquisition( + TimeSeries( + name=f"ts{i}", + data=H5DataIO(np.arange(1_000_000, dtype="f4"), chunks=(10_000,)), + unit="V", + rate=1.0, + ) + ) + local = tmp_path / "local.nwb" + with NWBHDF5IO(local, "w") as io: + io.write(nwbfile) + size = local.stat().st_size + assert size > 15_000_000 + shutil.copy(local, tmp_path / "served" / "content.nwb") + make_git_annex_dandiset( + tmp_path / "ds", {"sub-01/sub-01.nwb": (local, f"{base_url}/content.nwb")} + ) + r = get_annexed_readable(tmp_path / "ds" / "sub-01" / "sub-01.nwb") + assert r is not None + assert get_neurodata_types.__wrapped__(r) == get_neurodata_types.__wrapped__(local) + # With fsspec's default 5 MiB read-ahead, most of the file was downloaded + assert 0 < server.bytes_served < size / 10 + + @pytest.mark.ai_generated def test_annexed_readable_none( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path diff --git a/dandi/tests/fixtures.py b/dandi/tests/fixtures.py index bf5fea55b..e69808e81 100644 --- a/dandi/tests/fixtures.py +++ b/dandi/tests/fixtures.py @@ -409,9 +409,10 @@ def git(*args: str) -> str: class RangeHTTPServer(ThreadingHTTPServer): - """An HTTP server counting the range requests it served""" + """An HTTP server counting the range requests and the bytes it served""" ranged_requests: int = 0 + bytes_served: int = 0 class RangeHTTPRequestHandler(SimpleHTTPRequestHandler): @@ -460,6 +461,8 @@ def do_GET(self) -> None: ) else: self._send_headers(HTTPStatus.OK, size) + assert isinstance(self.server, RangeHTTPServer) + self.server.bytes_served += end - start + 1 with open(path, "rb") as fp: fp.seek(start) self.wfile.write(fp.read(end - start + 1)) From c47d267f484b00cb9b8f23531199ab6545ecf5e3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 22:26:56 +0000 Subject: [PATCH 07/15] Fetch consecutive missing blocks of streamed content in one request fsspec's "blockcache" fetches each block of a read with its own request, so a large read (e.g., of a dataset's data, by NWB Inspector) took one round trip per 256 KiB, i.e., a couple of MB/s however fast the connection. Use instead a block cache that fetches each run of consecutive missing blocks with a single request, and keeps up to 512 blocks (128 MiB) per open file rather than 32, which h5py's scattered reads of metadata kept evicting and fetching again. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/datalad_fuse.py | 76 +++++++++++++++++++++++- dandi/support/tests/test_datalad_fuse.py | 63 +++++++++++++++++++- 2 files changed, 137 insertions(+), 2 deletions(-) diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py index 4629e9f0f..3765860b8 100644 --- a/dandi/support/datalad_fuse.py +++ b/dandi/support/datalad_fuse.py @@ -191,6 +191,79 @@ def annex_fingerprint(source: Any) -> tuple[str, str] | None: #: Size of the blocks in which streamed content is read and cached STREAM_BLOCK_SIZE = 256 * 1024 +#: Maximum number of blocks cached for each open streamed file +STREAM_MAX_BLOCKS = 512 + +#: Name of the fsspec cache defined by `_coalescing_block_cache()` +COALESCING_BLOCK_CACHE = "dandi-coalescing-blockcache" + + +@cache +def _coalescing_block_cache() -> type: + """ + Define (once, so as not to import fsspec when this module is imported) and + register with fsspec a cache that, like fsspec's ``"blockcache"``, keeps + the least recently used blocks of a file, but fetches the consecutive + blocks missing for a read with a single request rather than with one + request per block, which would make reading a large dataset take one + round trip per `STREAM_BLOCK_SIZE` bytes + """ + from collections import OrderedDict + + from fsspec.caching import BaseCache, register_cache + + class CoalescingBlockCache(BaseCache): # type: ignore[misc] + name = COALESCING_BLOCK_CACHE + + def __init__( + self, + blocksize: int, + fetcher: Any, + size: int, + maxblocks: int = STREAM_MAX_BLOCKS, + ) -> None: + super().__init__(blocksize, fetcher, size) + self.maxblocks = maxblocks + self.blocks: OrderedDict[int, bytes] = OrderedDict() + + def _fetch(self, start: int | None, end: int | None) -> bytes: + if start is None: + start = 0 + if end is None or end > self.size: + end = self.size + if start >= end: + return b"" + first = start // self.blocksize + last = (end - 1) // self.blocksize + numbers = range(first, last + 1) + missing = [n for n in numbers if n not in self.blocks] + # Fetch each run of consecutive missing blocks in one request + i = 0 + while i < len(missing): + j = i + while j + 1 < len(missing) and missing[j + 1] == missing[j] + 1: + j += 1 + run_start = missing[i] * self.blocksize + run_end = min((missing[j] + 1) * self.blocksize, self.size) + data = self.fetcher(run_start, run_end) + for n in range(missing[i], missing[j] + 1): + offset = (n - missing[i]) * self.blocksize + self.blocks[n] = data[offset : offset + self.blocksize] + i = j + 1 + out = [] + for n in numbers: + self.blocks.move_to_end(n) + out.append(self.blocks[n]) + # Evict the least recently used blocks, but not those just read + while len(self.blocks) > max(self.maxblocks, len(numbers)): + self.blocks.popitem(last=False) + data = b"".join(out) + offset = first * self.blocksize + return data[start - offset : end - offset] + + register_cache(CoalescingBlockCache, clobber=True) + return CoalescingBlockCache + def _read_in_small_blocks(dsap: Any) -> None: """ @@ -206,7 +279,8 @@ def _read_in_small_blocks(dsap: Any) -> None: fs = getattr(dsap, "fs", None) if isinstance(fs, HTTPFileSystem): fs.block_size = STREAM_BLOCK_SIZE - fs.cache_type = "blockcache" + _coalescing_block_cache() + fs.cache_type = COALESCING_BLOCK_CACHE def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: diff --git a/dandi/support/tests/test_datalad_fuse.py b/dandi/support/tests/test_datalad_fuse.py index 8e58bc036..00a0674a0 100644 --- a/dandi/support/tests/test_datalad_fuse.py +++ b/dandi/support/tests/test_datalad_fuse.py @@ -12,7 +12,12 @@ from pynwb import NWBHDF5IO, NWBFile, TimeSeries import pytest -from ..datalad_fuse import AnnexedReadableFile, annex_fingerprint, get_annexed_readable +from ..datalad_fuse import ( + AnnexedReadableFile, + _coalescing_block_cache, + annex_fingerprint, + get_annexed_readable, +) from ...pynwb_utils import get_neurodata_types from ...tests.fixtures import RangeHTTPServer, make_git_annex_dandiset from ...tests.skip import mark @@ -114,6 +119,62 @@ def test_annexed_readable_reads_little( assert 0 < server.bytes_served < size / 10 +@pytest.mark.ai_generated +def test_coalescing_block_cache() -> None: + pytest.importorskip("fsspec") + content = bytes(range(256)) * 40 + requests: list[tuple[int, int]] = [] + + def fetcher(start: int, end: int) -> bytes: + requests.append((start, end)) + return content[start:end] + + c = _coalescing_block_cache()(100, fetcher, len(content), maxblocks=5) + # The blocks missing for a read are fetched with one request + assert c._fetch(150, 420) == content[150:420] + assert requests == [(100, 500)] + # Cached blocks are not fetched again, and runs of missing blocks around + # them are fetched with one request each + assert c._fetch(30, 720) == content[30:720] + assert requests[1:] == [(0, 100), (500, 800)] + # Reads at the end of the content, and beyond + assert c._fetch(10_190, 20_000) == content[10_190:] + assert requests[3:] == [(10_100, 10_240)] + assert c._fetch(20_000, 20_100) == b"" + # Only the 5 most recently used blocks are kept (but at least the blocks + # of the last read) + assert sorted(c.blocks) == [5, 6, 7, 101, 102] + requests.clear() + assert c._fetch(0, 10_240) == content + assert requests == [(0, 500), (800, 10_100)] + assert len(c.blocks) == 103 + + +@pytest.mark.ai_generated +def test_annexed_readable_large_read( + range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path +) -> None: + """A large read of streamed content is fetched with one request""" + pytest.importorskip("datalad_fuse.fsspec") + server, base_url = range_http_server + local = tmp_path / "local.bin" + content = np.random.default_rng(0).bytes(10_000_000) + local.write_bytes(content) + shutil.copy(local, tmp_path / "served" / "content.bin") + make_git_annex_dandiset( + tmp_path / "ds", {"sub-01/data.bin": (local, f"{base_url}/content.bin")} + ) + r = get_annexed_readable(tmp_path / "ds" / "sub-01" / "data.bin") + assert r is not None + with r.open() as fp: + fp.seek(123) + assert fp.read(9_000_000) == content[123:9_000_123] + assert server.ranged_requests == 1 + fp.seek(5_000_000) + assert fp.read(100) == content[5_000_000:5_000_100] + assert server.ranged_requests == 1 + + @pytest.mark.ai_generated def test_annexed_readable_none( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path From d384a130319882c53f98c6018f24d6d929d75452 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:08:09 +0000 Subject: [PATCH 08/15] Stream annexed content with remfile directly, without datalad-fuse Rather than through datalad-fuse's adapter (and fsspec), the content of a locked annexed file whose content is missing is now streamed by dandi with remfile, from the URLs git-annex knows for its key: - The key, and the size of the content, are taken from the target of the (broken) symbolic link, and the URLs from `git annex whereis --json --key`, run only in repositories in which git-annex is initialized (as any `git annex` command would initialize it). - The content is read from the first URL answering a range request, with remfile, which reads in the calling thread, in requests sized for HDF5's access pattern: reading the structure of a 16 MB NWB file downloads 0.6 MB. No fsspec I/O thread means no deadlock with h5py's lock, so garbage collection is no longer paused (`_GCPausedFile`), and the coalescing fsspec cache is gone. - The module is renamed `dandi.support.annex_stream`, and the `datalad` extra (datalad-fuse) is replaced by a `streaming` extra (remfile); the CI job is renamed accordingly. - The tests' HTTP server runs in a separate process (dandi.tests. range_http_server), so that no thread of the tests' process serving requests can run h5py finalizers while h5py's lock is held. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- .github/workflows/run-tests.yml | 8 +- dandi/metadata/nwb.py | 2 +- dandi/pynwb_utils.py | 2 +- dandi/support/annex_stream.py | 273 +++++++++++++++ dandi/support/datalad_fuse.py | 320 ------------------ ...t_datalad_fuse.py => test_annex_stream.py} | 200 ++++++----- dandi/tests/fixtures.py | 94 +---- dandi/tests/range_http_server.py | 129 +++++++ pyproject.toml | 8 +- 9 files changed, 539 insertions(+), 497 deletions(-) create mode 100644 dandi/support/annex_stream.py delete mode 100644 dandi/support/datalad_fuse.py rename dandi/support/tests/{test_datalad_fuse.py => test_annex_stream.py} (55%) create mode 100644 dandi/tests/range_http_server.py diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 97c9ab948..cb2384d88 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -75,7 +75,7 @@ jobs: mode: lowest-deps - os: ubuntu-latest python: '3.12' - mode: datalad + mode: annex-streaming steps: - name: Set up environment @@ -126,9 +126,9 @@ jobs: git+https://github.com/hdmf-dev/hdmf \ git+https://github.com/hdmf-dev/hdmf-zarr - - name: Install git-annex and datalad-fuse (for streaming annexed content) - if: matrix.mode == 'datalad' - run: pip install git-annex ".[datalad]" + - name: Install git-annex and remfile (for streaming annexed content) + if: matrix.mode == 'annex-streaming' + run: pip install git-annex ".[streaming]" - name: Create NFS filesystem if: matrix.mode == 'nfs' diff --git a/dandi/metadata/nwb.py b/dandi/metadata/nwb.py index a773475d6..24db09d52 100644 --- a/dandi/metadata/nwb.py +++ b/dandi/metadata/nwb.py @@ -22,7 +22,7 @@ metadata_cache, nwb_has_external_links, ) -from ..support.datalad_fuse import annex_fingerprint +from ..support.annex_stream import annex_fingerprint from ..utils import find_parent_directory_containing lgr = get_logger() diff --git a/dandi/pynwb_utils.py b/dandi/pynwb_utils.py index c49293f57..b0dda71c0 100644 --- a/dandi/pynwb_utils.py +++ b/dandi/pynwb_utils.py @@ -42,7 +42,7 @@ metadata_nwb_subject_fields, ) from .misctypes import Readable -from .support.datalad_fuse import annex_fingerprint +from .support.annex_stream import annex_fingerprint from .utils import get_module_version, is_url from .validate._types import ( Origin, diff --git a/dandi/support/annex_stream.py b/dandi/support/annex_stream.py new file mode 100644 index 000000000..afb3b5862 --- /dev/null +++ b/dandi/support/annex_stream.py @@ -0,0 +1,273 @@ +""" +Reading the content of annexed files of git-annex_ repositories (such as +DataLad_ datasets) that is not present locally, by streaming it. + +In a git-annex repository, a (locked) annexed file is a symbolic link into +:file:`.git/annex/objects/`, which is broken until the content is fetched. Its +target names the git-annex key of the content, which records the size of the +content, and git-annex knows URLs for the key (``git annex whereis``). The +content is read from those URLs with remfile_, which fetches only the parts of +a file that are read, in requests sized for the access pattern of HDF5-based +files (such as NWB), so that the files of a DataLad Dandiset can be read (e.g., +for their metadata, or to be validated) without downloading the (possibly +terabytes of) data. + +The URLs are tried in the order git-annex lists them; for a DataLad Dandiset, +that is the DANDI Archive's API download URL (which redirects to S3) before the +direct S3 URL. + +This requires git-annex and remfile (``pip install "dandi[streaming]"``). No +``git annex`` command is run in a repository in which git-annex is not +initialized (e.g., a fresh ``git clone``), as it would initialize git-annex +there, modifying the repository. + +.. _git-annex: https://git-annex.branchable.com +.. _DataLad: https://www.datalad.org +.. _remfile: https://github.com/flatironinstitute/remfile +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from functools import cache +import io +import json +import logging +import os +from pathlib import Path +import re +import subprocess +from typing import IO, Any, cast + +from ..misctypes import Readable + +lgr = logging.getLogger("dandi.support.annex_stream") + +#: Seconds to wait for a server to start answering a request +REQUEST_TIMEOUT = 60 + + +@cache +def annex_initialized(directory: Path) -> bool: + """ + Whether ``directory`` is in a Git repository in which git-annex is + initialized. (Any ``git annex`` command would otherwise initialize it, + modifying the repository.) + """ + try: + r = subprocess.run( + ["git", "-C", str(directory), "config", "--get", "annex.uuid"], + capture_output=True, + text=True, + ) + except FileNotFoundError: + return False + return r.returncode == 0 and bool(r.stdout.strip()) + + +def annex_key_size(key: str) -> int | None: + """ + The size of the content of a git-annex key, as recorded in the key (e.g., + ``SHA256E-s1234--.nwb``), or `None` if it is not + """ + fields = key.split("--", 1)[0].split("-") + for f in fields[1:]: + if m := re.fullmatch(r"s(\d+)", f): + return int(m[1]) + return None + + +def annex_urls(directory: Path, key: str) -> list[str]: + """ + The http(s) URLs git-annex knows for ``key`` (in the order ``git annex + whereis`` lists them), in the repository at ``directory``, in which + git-annex must be initialized + """ + r = subprocess.run( + ["git", "-C", str(directory), "annex", "whereis", "--json", "--key", key], + capture_output=True, + text=True, + ) + try: + info = json.loads(r.stdout) + except ValueError: + lgr.debug("git annex whereis --key %s failed: %s", key, r.stderr.strip()) + return [] + urls: list[str] = [] + for location in info.get("whereis", []) + info.get("untrusted", []): + for url in location.get("urls", []): + if url.lower().startswith(("http://", "https://")) and url not in urls: + urls.append(url) + return urls + + +class RemoteFile(io.RawIOBase): + """ + A seekable, read-only binary file object reading the content at ``url`` + (of ``size`` bytes) with remfile, which fetches it in the reading thread + """ + + def __init__(self, url: str, size: int) -> None: + super().__init__() + # Optional dependency: + import remfile + + self.url = url + self.size = size + # Passing the size spares a request for it + self._fp = remfile.File(url, _size=size) + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return True + + def readinto(self, buffer: Any) -> int: + view = memoryview(buffer).cast("B") + n = min(len(view), max(self.size - self._fp.tell(), 0)) + if n == 0: + return 0 + data = self._fp.read(n) + view[: len(data)] = data + return len(data) + + def seek(self, offset: int, whence: int = io.SEEK_SET) -> int: + self._fp.seek(offset, whence) + return int(self._fp.tell()) + + def tell(self) -> int: + return int(self._fp.tell()) + + def close(self) -> None: + if not self.closed: + self._fp.close() + super().close() + + +def _responds(url: str) -> bool: + """Whether a range request for the first byte at ``url`` succeeds""" + # Avoid importing requests when this module is imported: + import requests + + try: + with requests.get( + url, + headers={"Range": "bytes=0-0"}, + stream=True, + timeout=REQUEST_TIMEOUT, + ) as r: + return r.status_code in (200, 206) + except requests.RequestException as e: + lgr.debug("%s: Request failed: %s", url, e) + return False + + +@dataclass +class AnnexedReadableFile(Readable): + """ + A `Readable` for a (locked) annexed file at ``filepath`` (a broken symbolic + link) whose content is not present locally, which is streamed from the + first of ``urls`` (the http(s) URLs git-annex knows for the content) that + can be read (see the module docstring). + + Instances are obtained by calling `get_annexed_readable()`. + """ + + filepath: Path + key: str + size: int + urls: list[str] + _url: str | None = field(default=None, init=False, repr=False, compare=False) + + @property + def url(self) -> str: + """The URL the content is read from (once it was opened), else the first""" + return self._url if self._url is not None else self.urls[0] + + def open(self) -> IO[bytes]: + if self._url is None: + for url in self.urls: + if _responds(url): + self._url = url + break + lgr.debug("%s: Cannot read %s", self.filepath, url) + else: + raise OSError( + f"{self.filepath}: None of the URLs of its content can be read:" + f" {', '.join(self.urls)}" + ) + return cast("IO[bytes]", RemoteFile(self._url, self.size)) + + def get_size(self) -> int: + return self.size + + def get_mtime(self) -> datetime | None: + return None + + def get_filename(self) -> str: + return self.filepath.name + + def __str__(self) -> str: + return str(self.filepath) + + +def annex_fingerprint(source: Any) -> tuple[str, str] | None: + """ + Fingerprint of ``source`` for ``PersistentCache.memoize_path`` + + Pass it as ``custom_fingerprint`` to cache the results of a function of a + local path or a `Readable` under the git-annex key of a locked annexed + file, paired with its path (see `fscacher.annex_key_fingerprint`), rather + than under ``stat()``: the content of an `AnnexedReadableFile`, which is + not present locally, cannot be ``stat()``-ed. Anything else is handled as + without this. + """ + # Avoid heavy import (fscacher imports joblib, which imports numpy) when + # this module is imported, e.g., by the CLI: + from fscacher import annex_key_fingerprint + + if isinstance(source, AnnexedReadableFile): + source = source.filepath + return cast("tuple[str, str] | None", annex_key_fingerprint(source)) + + +def can_stream() -> bool: + """Whether the requirements for streaming (remfile) are installed""" + try: + import remfile # noqa: F401 + except ImportError: + return False + return True + + +def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: + """ + Return an `AnnexedReadableFile` for streaming the content of the annexed + file at ``path``, or `None` if remfile is not installed, ``path`` is not a + (locked) annexed file whose content is missing in a repository in which + git-annex is initialized, its key does not record the size of the content, + or git-annex knows no http(s) URL to stream it from + """ + filepath = Path(os.path.abspath(path)) + if not filepath.is_symlink() or filepath.exists(): + # Not a broken symbolic link + return None + if not can_stream(): + return None + target = Path(os.readlink(filepath)) + if ".git/annex/objects" not in target.as_posix(): + return None + key = target.name + size = annex_key_size(key) + if size is None: + lgr.debug("%s: The key %s does not record the size", filepath, key) + return None + if not annex_initialized(filepath.parent): + return None + urls = annex_urls(filepath.parent, key) + if not urls: + lgr.debug("%s: git-annex knows no URL for key %s", filepath, key) + return None + return AnnexedReadableFile(filepath=filepath, key=key, size=size, urls=urls) diff --git a/dandi/support/datalad_fuse.py b/dandi/support/datalad_fuse.py deleted file mode 100644 index 3765860b8..000000000 --- a/dandi/support/datalad_fuse.py +++ /dev/null @@ -1,320 +0,0 @@ -""" -Reading the content of annexed files of git-annex_ repositories (such as -DataLad_ datasets) that is not present locally, with datalad-fuse_. - -In a git-annex repository, a (locked) annexed file is a symbolic link into -:file:`.git/annex/objects/`, which is broken until the content is fetched. -datalad-fuse can mount such a repository with FUSE so that the content is -streamed on demand, but its adapter, which the mount is built on, also opens -the files directly with no mount. It asks git-annex where the content is -(``git annex whereis``) and streams it over HTTP(S) with fsspec_. That adapter -is used here, so that the files of a DataLad Dandiset can be read (e.g., for -their metadata, or to be validated) without downloading the (possibly -terabytes of) data. - -The URLs are tried in the order git-annex lists them; for a DataLad Dandiset, -that is the DANDI Archive's API download URL (which redirects to S3) before the -direct S3 URL. datalad-fuse does not use HTTP proxy environment variables. - -This requires datalad-fuse (``pip install "dandi[datalad]"``), which requires -DataLad and git-annex. - -.. _git-annex: https://git-annex.branchable.com -.. _DataLad: https://www.datalad.org -.. _datalad-fuse: https://github.com/datalad/datalad-fuse -.. _fsspec: https://github.com/fsspec/filesystem_spec -""" - -from __future__ import annotations - -import atexit -from dataclasses import dataclass, field -from datetime import datetime -from functools import cache -import gc -import logging -import os -from pathlib import Path -import subprocess -import threading -from typing import IO, Any, cast - -from ..misctypes import Readable - -lgr = logging.getLogger("dandi.support.datalad_fuse") - - -@cache -def get_adapter() -> Any: - """ - The datalad-fuse adapter shared by all files, rooted at the filesystem - root so that it serves files of any dataset, and closed at exit - """ - # Optional dependency: - from datalad_fuse.fsspec import FsspecAdapter - - # caching=False: do not keep the streamed blocks on disk, in the dataset. - # The root and all paths passed to the adapter must be absolute. - adapter = FsspecAdapter(Path(os.path.abspath(os.sep)), caching=False) - atexit.register(adapter.__exit__, None, None, None) - return adapter - - -@cache -def annex_initialized(directory: Path) -> bool: - """ - Whether ``directory`` is in a Git repository in which git-annex is - initialized. (DataLad would otherwise initialize it, modifying the - repository.) - """ - try: - r = subprocess.run( - ["git", "-C", str(directory), "config", "--get", "annex.uuid"], - capture_output=True, - text=True, - ) - except FileNotFoundError: - return False - return r.returncode == 0 and bool(r.stdout.strip()) - - -_gc_lock = threading.Lock() -_gc_pauses = 0 -_gc_was_enabled = False - - -class _GCPausedFile: - """ - A file object streamed by datalad-fuse, during whose lifetime automatic - garbage collection is disabled - - h5py holds a global lock while it reads from a Python file object, and the - read waits for another thread (fsspec's I/O thread, or an HTTP server in - the same process) to fetch the data. If a garbage collection ran in that - thread meanwhile, finalizers of h5py objects (e.g., of an unreferenced - ``NWBHDF5IO``) would wait for the same lock, deadlocking the process. - """ - - def __init__(self, fp: Any) -> None: - global _gc_pauses, _gc_was_enabled - with _gc_lock: - if _gc_pauses == 0: - # Collect pending garbage now, in this thread, which does not - # hold h5py's lock yet - gc.collect() - _gc_was_enabled = gc.isenabled() - gc.disable() - _gc_pauses += 1 - self._fp = fp - self._paused = True - - def __getattr__(self, name: str) -> Any: - return getattr(self._fp, name) - - def __enter__(self) -> _GCPausedFile: - return self - - def __exit__(self, *_exc: Any) -> None: - self.close() - - def __del__(self) -> None: - if self._paused: - self.close() - - def close(self) -> None: - global _gc_pauses - try: - self._fp.close() - finally: - with _gc_lock: - if self._paused: - self._paused = False - _gc_pauses -= 1 - if _gc_pauses == 0 and _gc_was_enabled: - gc.enable() - - -@dataclass -class AnnexedReadableFile(Readable): - """ - A `Readable` for a (locked) annexed file at ``filepath`` (a broken symbolic - link) whose content is not present locally, which is streamed with - datalad-fuse (see the module docstring). ``url`` is only the first URL - known for the content, for reporting: if it cannot be read, the adapter - tries the others. - - Instances are obtained by calling `get_annexed_readable()`. - """ - - filepath: Path - key: str - size: int - url: str - adapter: Any = field(repr=False, compare=False) - - def open(self) -> IO[bytes]: - return cast("IO[bytes]", _GCPausedFile(self.adapter.open(self.filepath, "rb"))) - - def get_size(self) -> int: - return self.size - - def get_mtime(self) -> datetime | None: - return None - - def get_filename(self) -> str: - return self.filepath.name - - def __str__(self) -> str: - return str(self.filepath) - - -def annex_fingerprint(source: Any) -> tuple[str, str] | None: - """ - Fingerprint of ``source`` for ``PersistentCache.memoize_path`` - - Pass it as ``custom_fingerprint`` to cache the results of a function of a - local path or a `Readable` under the git-annex key of a locked annexed - file, paired with its path (see `fscacher.annex_key_fingerprint`), rather - than under ``stat()``: the content of an `AnnexedReadableFile`, which is - not present locally, cannot be ``stat()``-ed. Anything else is handled as - without this. - """ - # Avoid heavy import (fscacher imports joblib, which imports numpy) when - # this module is imported, e.g., by the CLI: - from fscacher import annex_key_fingerprint - - if isinstance(source, AnnexedReadableFile): - source = source.filepath - return cast("tuple[str, str] | None", annex_key_fingerprint(source)) - - -#: Size of the blocks in which streamed content is read and cached -STREAM_BLOCK_SIZE = 256 * 1024 - -#: Maximum number of blocks cached for each open streamed file -STREAM_MAX_BLOCKS = 512 - -#: Name of the fsspec cache defined by `_coalescing_block_cache()` -COALESCING_BLOCK_CACHE = "dandi-coalescing-blockcache" - - -@cache -def _coalescing_block_cache() -> type: - """ - Define (once, so as not to import fsspec when this module is imported) and - register with fsspec a cache that, like fsspec's ``"blockcache"``, keeps - the least recently used blocks of a file, but fetches the consecutive - blocks missing for a read with a single request rather than with one - request per block, which would make reading a large dataset take one - round trip per `STREAM_BLOCK_SIZE` bytes - """ - from collections import OrderedDict - - from fsspec.caching import BaseCache, register_cache - - class CoalescingBlockCache(BaseCache): # type: ignore[misc] - name = COALESCING_BLOCK_CACHE - - def __init__( - self, - blocksize: int, - fetcher: Any, - size: int, - maxblocks: int = STREAM_MAX_BLOCKS, - ) -> None: - super().__init__(blocksize, fetcher, size) - self.maxblocks = maxblocks - self.blocks: OrderedDict[int, bytes] = OrderedDict() - - def _fetch(self, start: int | None, end: int | None) -> bytes: - if start is None: - start = 0 - if end is None or end > self.size: - end = self.size - if start >= end: - return b"" - first = start // self.blocksize - last = (end - 1) // self.blocksize - numbers = range(first, last + 1) - missing = [n for n in numbers if n not in self.blocks] - # Fetch each run of consecutive missing blocks in one request - i = 0 - while i < len(missing): - j = i - while j + 1 < len(missing) and missing[j + 1] == missing[j] + 1: - j += 1 - run_start = missing[i] * self.blocksize - run_end = min((missing[j] + 1) * self.blocksize, self.size) - data = self.fetcher(run_start, run_end) - for n in range(missing[i], missing[j] + 1): - offset = (n - missing[i]) * self.blocksize - self.blocks[n] = data[offset : offset + self.blocksize] - i = j + 1 - out = [] - for n in numbers: - self.blocks.move_to_end(n) - out.append(self.blocks[n]) - # Evict the least recently used blocks, but not those just read - while len(self.blocks) > max(self.maxblocks, len(numbers)): - self.blocks.popitem(last=False) - data = b"".join(out) - offset = first * self.blocksize - return data[start - offset : end - offset] - - register_cache(CoalescingBlockCache, clobber=True) - return CoalescingBlockCache - - -def _read_in_small_blocks(dsap: Any) -> None: - """ - Make the files of a dataset adapter be read in cached blocks of - `STREAM_BLOCK_SIZE` bytes rather than with fsspec's default: h5py reads - many small pieces scattered over the file, and with the default (5 MiB of - read-ahead, discarded whenever a read jumps elsewhere) reading an NWB file - downloads about as much as the whole file each time it is opened - """ - # Avoid heavy import when this module is imported: - from fsspec.implementations.http import HTTPFileSystem - - fs = getattr(dsap, "fs", None) - if isinstance(fs, HTTPFileSystem): - fs.block_size = STREAM_BLOCK_SIZE - _coalescing_block_cache() - fs.cache_type = COALESCING_BLOCK_CACHE - - -def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: - """ - Return an `AnnexedReadableFile` for streaming the content of the annexed - file at ``path``, or `None` if datalad-fuse is not installed, ``path`` is - not a (locked) annexed file whose content is missing in a repository in - which git-annex is initialized, its key does not record the size of the - content, or git-annex knows no URL to stream it from - """ - filepath = Path(os.path.abspath(path)) - if not filepath.is_symlink() or filepath.exists(): - # Not a broken symbolic link - return None - try: - from datalad_fuse.fsspec import FileState - except ImportError: - return None - if not annex_initialized(filepath.parent): - return None - try: - dsap, relpath = get_adapter().resolve_dataset(filepath) - _read_in_small_blocks(dsap) - state, key = dsap.get_file_state(relpath) - except Exception as e: - # E.g., a repository without any commit - lgr.debug("%s: Cannot be read with datalad-fuse: %s", filepath, e) - return None - if state is not FileState.NO_CONTENT or key is None or key.size is None: - return None - url = next(dsap.get_urls(str(key)), None) - if url is None: - lgr.debug("%s: git-annex knows no URL for key %s", filepath, key) - return None - return AnnexedReadableFile( - filepath=filepath, key=str(key), size=key.size, url=url, adapter=get_adapter() - ) diff --git a/dandi/support/tests/test_datalad_fuse.py b/dandi/support/tests/test_annex_stream.py similarity index 55% rename from dandi/support/tests/test_datalad_fuse.py rename to dandi/support/tests/test_annex_stream.py index 00a0674a0..beef5dfcc 100644 --- a/dandi/support/tests/test_datalad_fuse.py +++ b/dandi/support/tests/test_annex_stream.py @@ -1,9 +1,10 @@ from __future__ import annotations from datetime import datetime, timezone -import gc +import io from pathlib import Path import shutil +import subprocess from fscacher import PersistentCache import h5py @@ -12,14 +13,15 @@ from pynwb import NWBHDF5IO, NWBFile, TimeSeries import pytest -from ..datalad_fuse import ( +from ..annex_stream import ( AnnexedReadableFile, - _coalescing_block_cache, annex_fingerprint, + annex_key_size, get_annexed_readable, ) from ...pynwb_utils import get_neurodata_types -from ...tests.fixtures import RangeHTTPServer, make_git_annex_dandiset +from ...tests.fixtures import make_git_annex_dandiset +from ...tests.range_http_server import RangeHTTPServer from ...tests.skip import mark pytestmark = mark.skipif_no_git_annex @@ -33,7 +35,7 @@ def streamed_nwb( An annexed copy of ``simple2_nwb`` in a git-annex repository, whose content is not present but registered at a URL of the HTTP server """ - pytest.importorskip("datalad_fuse.fsspec") + pytest.importorskip("remfile") server, base_url = range_http_server shutil.copy(simple2_nwb, tmp_path / "served" / "content.nwb") make_git_annex_dandiset( @@ -45,6 +47,20 @@ def streamed_nwb( return nwb, server +@pytest.mark.ai_generated +@pytest.mark.parametrize( + "key,size", + [ + ("SHA256E-s1234--abc.nwb", 1234), + ("MD5E-s0--d41d8cd98f00b204e9800998ecf8427e", 0), + ("SHA256E-s99-S10-C2--abc.nwb", 99), + ("URL--https&c%%example.org%x.nwb", None), + ], +) +def test_annex_key_size(key: str, size: int | None) -> None: + assert annex_key_size(key) == size + + @pytest.mark.ai_generated def test_annexed_readable( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path @@ -56,34 +72,90 @@ def test_annexed_readable( assert r.url.endswith("/content.nwb") assert r.get_size() == simple2_nwb.stat().st_size assert r.get_filename() == "sub-01.nwb" - fp = r.open() - try: + with r.open() as fp: # Random access over HTTP is enough for h5py to read the file with h5py.File(fp, "r") as h5: assert h5.attrs["nwb_version"] - finally: - fp.close() assert server.ranged_requests > 0 @pytest.mark.ai_generated -def test_annexed_readable_pauses_gc( - streamed_nwb: tuple[Path, RangeHTTPServer], +def test_annexed_readable_file_object( + range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path ) -> None: - # Automatic garbage collection is disabled while any streamed file is open - # (to avoid deadlocks with h5py's lock; see `_GCPausedFile`) - nwb, _ = streamed_nwb - r = get_annexed_readable(nwb) + pytest.importorskip("remfile") + _, base_url = range_http_server + content = bytes(range(256)) * 1000 + local = tmp_path / "local.bin" + local.write_bytes(content) + shutil.copy(local, tmp_path / "served" / "content.bin") + make_git_annex_dandiset( + tmp_path / "ds", {"data.bin": (local, f"{base_url}/content.bin")} + ) + r = get_annexed_readable(tmp_path / "ds" / "data.bin") assert r is not None - assert gc.isenabled() with r.open() as fp: - assert not gc.isenabled() - with r.open() as fp2: - assert fp2.read(8) == b"\x89HDF\r\n\x1a\n" - assert not gc.isenabled() - fp.seek(1) - assert fp.read(3) == b"HDF" - assert gc.isenabled() + assert fp.readable() and fp.seekable() + assert fp.read(8) == content[:8] + assert fp.seek(1000) == 1000 + assert fp.read(3) == content[1000:1003] + assert fp.tell() == 1003 + assert fp.seek(-5, io.SEEK_END) == len(content) - 5 + # Reads stop at the end of the content + assert fp.read(100) == content[-5:] + assert fp.read(100) == b"" + fp.seek(0) + assert fp.read() == content + assert fp.closed + + +@pytest.mark.ai_generated +def test_annexed_readable_url_fallback( + range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path +) -> None: + """The content is read from the first of its URLs that can be read""" + pytest.importorskip("remfile") + _, base_url = range_http_server + shutil.copy(simple2_nwb, tmp_path / "served" / "content.nwb") + make_git_annex_dandiset( + tmp_path / "ds", + { + "sub-01.nwb": ( + simple2_nwb, + [f"{base_url}/missing.nwb", f"{base_url}/content.nwb"], + ) + }, + ) + found = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") + assert found is not None + # (in the order in which git-annex lists them, which is not the order in + # which they were registered) + assert sorted(found.urls) == [f"{base_url}/content.nwb", f"{base_url}/missing.nwb"] + r = AnnexedReadableFile( + filepath=found.filepath, + key=found.key, + size=found.size, + urls=[f"{base_url}/missing.nwb", f"{base_url}/content.nwb"], + ) + assert r.url == f"{base_url}/missing.nwb" + with r.open() as fp, h5py.File(fp, "r") as h5: + assert h5.attrs["nwb_version"] + assert r.url == f"{base_url}/content.nwb" + + +@pytest.mark.ai_generated +def test_annexed_readable_no_url_readable( + range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path +) -> None: + pytest.importorskip("remfile") + _, base_url = range_http_server + make_git_annex_dandiset( + tmp_path / "ds", {"sub-01.nwb": (simple2_nwb, f"{base_url}/missing.nwb")} + ) + r = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") + assert r is not None + with pytest.raises(OSError, match="None of the URLs"): + r.open() @pytest.mark.ai_generated @@ -91,7 +163,7 @@ def test_annexed_readable_reads_little( range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path ) -> None: """Reading an NWB file's structure downloads only a small part of it""" - pytest.importorskip("datalad_fuse.fsspec") + pytest.importorskip("remfile") server, base_url = range_http_server nwbfile = NWBFile("streamed", "id", datetime(2026, 1, 1, tzinfo=timezone.utc)) for i in range(4): @@ -104,8 +176,8 @@ def test_annexed_readable_reads_little( ) ) local = tmp_path / "local.nwb" - with NWBHDF5IO(local, "w") as io: - io.write(nwbfile) + with NWBHDF5IO(local, "w") as io_: + io_.write(nwbfile) size = local.stat().st_size assert size > 15_000_000 shutil.copy(local, tmp_path / "served" / "content.nwb") @@ -115,64 +187,16 @@ def test_annexed_readable_reads_little( r = get_annexed_readable(tmp_path / "ds" / "sub-01" / "sub-01.nwb") assert r is not None assert get_neurodata_types.__wrapped__(r) == get_neurodata_types.__wrapped__(local) - # With fsspec's default 5 MiB read-ahead, most of the file was downloaded assert 0 < server.bytes_served < size / 10 -@pytest.mark.ai_generated -def test_coalescing_block_cache() -> None: - pytest.importorskip("fsspec") - content = bytes(range(256)) * 40 - requests: list[tuple[int, int]] = [] - - def fetcher(start: int, end: int) -> bytes: - requests.append((start, end)) - return content[start:end] - - c = _coalescing_block_cache()(100, fetcher, len(content), maxblocks=5) - # The blocks missing for a read are fetched with one request - assert c._fetch(150, 420) == content[150:420] - assert requests == [(100, 500)] - # Cached blocks are not fetched again, and runs of missing blocks around - # them are fetched with one request each - assert c._fetch(30, 720) == content[30:720] - assert requests[1:] == [(0, 100), (500, 800)] - # Reads at the end of the content, and beyond - assert c._fetch(10_190, 20_000) == content[10_190:] - assert requests[3:] == [(10_100, 10_240)] - assert c._fetch(20_000, 20_100) == b"" - # Only the 5 most recently used blocks are kept (but at least the blocks - # of the last read) - assert sorted(c.blocks) == [5, 6, 7, 101, 102] - requests.clear() - assert c._fetch(0, 10_240) == content - assert requests == [(0, 500), (800, 10_100)] - assert len(c.blocks) == 103 - - -@pytest.mark.ai_generated -def test_annexed_readable_large_read( - range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path -) -> None: - """A large read of streamed content is fetched with one request""" - pytest.importorskip("datalad_fuse.fsspec") - server, base_url = range_http_server - local = tmp_path / "local.bin" - content = np.random.default_rng(0).bytes(10_000_000) - local.write_bytes(content) - shutil.copy(local, tmp_path / "served" / "content.bin") - make_git_annex_dandiset( - tmp_path / "ds", {"sub-01/data.bin": (local, f"{base_url}/content.bin")} +def annex_uuid(repo: Path) -> str: + r = subprocess.run( + ["git", "-C", str(repo), "config", "--get", "annex.uuid"], + capture_output=True, + text=True, ) - r = get_annexed_readable(tmp_path / "ds" / "sub-01" / "data.bin") - assert r is not None - with r.open() as fp: - fp.seek(123) - assert fp.read(9_000_000) == content[123:9_000_123] - assert server.ranged_requests == 1 - fp.seek(5_000_000) - assert fp.read(100) == content[5_000_000:5_000_100] - assert server.ranged_requests == 1 + return r.stdout.strip() @pytest.mark.ai_generated @@ -185,13 +209,15 @@ def test_annexed_readable_none( assert get_annexed_readable(nwb.parent / "regular.nwb") is None # Not in a Git repository assert get_annexed_readable(simple2_nwb) is None - # A broken symbolic link in a repository without git-annex initialized - plain = tmp_path / "plain" - plain.mkdir() - shutil.copytree(nwb.parent, plain / "sub-01", symlinks=True) - shutil.copytree(nwb.parent.parent / ".git", plain / ".git") - (plain / ".git" / "config").write_text("[core]\n\tbare = false\n") - assert get_annexed_readable(plain / "sub-01" / "sub-01.nwb") is None + # A broken symbolic link in a clone in which git-annex is not initialized, + # which is not initialized by trying + clone = tmp_path / "clone" + subprocess.run( + ["git", "clone", "-q", str(nwb.parent.parent), str(clone)], check=True + ) + assert annex_uuid(clone) == "" + assert get_annexed_readable(clone / "sub-01" / "sub-01.nwb") is None + assert annex_uuid(clone) == "" @pytest.mark.ai_generated diff --git a/dandi/tests/fixtures.py b/dandi/tests/fixtures.py index e69808e81..90b69d792 100644 --- a/dandi/tests/fixtures.py +++ b/dandi/tests/fixtures.py @@ -3,16 +3,12 @@ from collections.abc import Callable, Iterator from dataclasses import dataclass, field, replace from datetime import datetime, timezone -from functools import partial -from http import HTTPStatus -from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer import logging import os from pathlib import Path import re import shutil from subprocess import DEVNULL, check_output, run -import threading from time import sleep from typing import Any, Literal from uuid import uuid4 @@ -32,6 +28,7 @@ import requests import zarr +from .range_http_server import RangeHTTPServer from .skip import skipif from .. import get_logger from ..cli.cmd_organize import organize @@ -367,14 +364,16 @@ def fixture( return fixture -def make_git_annex_dandiset(dandiset: Path, files: dict[str, tuple[Path, str]]) -> None: +def make_git_annex_dandiset( + dandiset: Path, files: dict[str, tuple[Path, str | list[str]]] +) -> None: """ Make the directory ``dandiset`` (created if needed, with a minimal :file:`dandiset.yaml`) a git-annex repository looking like a clone of a DataLad Dandiset whose annexed content has not been fetched: for each - ``relpath: (source, url)`` item of ``files``, a copy of ``source`` is - annexed at ``relpath`` with ``url`` registered for it, and then dropped. - This needs git-annex. + ``relpath: (source, urls)`` item of ``files``, a copy of ``source`` is + annexed at ``relpath`` with ``urls`` (one or a list) registered for it, and + then dropped. This needs git-annex. """ env = { **os.environ, @@ -401,92 +400,27 @@ def git(*args: str) -> str: (dandiset / relpath).parent.mkdir(parents=True, exist_ok=True) shutil.copy(source, dandiset / relpath) git("annex", "add", "-q", *files) - for relpath, (_, url) in files.items(): + for relpath, (_, urls) in files.items(): key = git("annex", "lookupkey", relpath) - git("annex", "registerurl", key, url) + for url in [urls] if isinstance(urls, str) else urls: + git("annex", "registerurl", key, url) git("commit", "-q", "-m", "Add files") git("annex", "drop", "-q", "--force", *files) -class RangeHTTPServer(ThreadingHTTPServer): - """An HTTP server counting the range requests and the bytes it served""" - - ranged_requests: int = 0 - bytes_served: int = 0 - - -class RangeHTTPRequestHandler(SimpleHTTPRequestHandler): - """ - A handler serving files with support for ``HEAD`` and single-range ``GET`` - requests, like S3 does - """ - - def log_message(self, format: str, *args: Any) -> None: - pass - - def _send_headers( - self, status: HTTPStatus, length: int, content_range: str | None = None - ) -> None: - self.send_response(status) - self.send_header("Content-Type", "application/octet-stream") - self.send_header("Content-Length", str(length)) - self.send_header("Accept-Ranges", "bytes") - if content_range is not None: - self.send_header("Content-Range", content_range) - self.end_headers() - - def do_HEAD(self) -> None: - path = self.translate_path(self.path) - if not os.path.isfile(path): - self.send_error(HTTPStatus.NOT_FOUND) - return - self._send_headers(HTTPStatus.OK, os.path.getsize(path)) - - def do_GET(self) -> None: - path = self.translate_path(self.path) - if not os.path.isfile(path): - self.send_error(HTTPStatus.NOT_FOUND) - return - size = os.path.getsize(path) - start, end = 0, size - 1 - if m := re.fullmatch(r"bytes=(\d+)-(\d*)", self.headers.get("Range", "")): - start = int(m[1]) - end = min(int(m[2]) if m[2] else size - 1, size - 1) - assert isinstance(self.server, RangeHTTPServer) - self.server.ranged_requests += 1 - self._send_headers( - HTTPStatus.PARTIAL_CONTENT, - end - start + 1, - f"bytes {start}-{end}/{size}", - ) - else: - self._send_headers(HTTPStatus.OK, size) - assert isinstance(self.server, RangeHTTPServer) - self.server.bytes_served += end - start + 1 - with open(path, "rb") as fp: - fp.seek(start) - self.wfile.write(fp.read(end - start + 1)) - - @pytest.fixture() def range_http_server(tmp_path: Path) -> Iterator[tuple[RangeHTTPServer, str]]: """ Serve the files in ``tmp_path / "served"`` over HTTP with support for range - requests; yields the server and its base URL + requests, from a separate process; yields the server and its base URL """ served = tmp_path / "served" served.mkdir() - server = RangeHTTPServer( - ("127.0.0.1", 0), partial(RangeHTTPRequestHandler, directory=str(served)) - ) - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() + server = RangeHTTPServer(served) try: - yield server, f"http://127.0.0.1:{server.server_address[1]}" + yield server, server.url finally: - server.shutdown() - server.server_close() - thread.join() + server.stop() def _make_subdirs_dandisets(path: Path) -> None: diff --git a/dandi/tests/range_http_server.py b/dandi/tests/range_http_server.py new file mode 100644 index 000000000..67ab07090 --- /dev/null +++ b/dandi/tests/range_http_server.py @@ -0,0 +1,129 @@ +""" +An HTTP server supporting range requests (like S3), run in a separate process +for tests streaming files from it, and counting the range requests and bytes +it served + +It runs in a separate process so that no thread of the tests' process serves +requests: a garbage collection in such a thread, while the thread reading a +file over HTTP holds h5py's global lock waiting for it, would deadlock if it +ran finalizers of h5py objects. +""" + +from __future__ import annotations + +from functools import partial +from http import HTTPStatus +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +import multiprocessing +from multiprocessing.sharedctypes import Synchronized +import os +from pathlib import Path +import re +from typing import Any + + +class _Server(ThreadingHTTPServer): + ranged_requests: Synchronized + bytes_served: Synchronized + + +class _Handler(SimpleHTTPRequestHandler): + """ + A handler serving files with support for ``HEAD`` and single-range ``GET`` + requests + """ + + server: _Server + + def log_message(self, format: str, *args: Any) -> None: + pass + + def _send_headers( + self, status: HTTPStatus, length: int, content_range: str | None = None + ) -> None: + self.send_response(status) + self.send_header("Content-Type", "application/octet-stream") + self.send_header("Content-Length", str(length)) + self.send_header("Accept-Ranges", "bytes") + if content_range is not None: + self.send_header("Content-Range", content_range) + self.end_headers() + + def do_HEAD(self) -> None: + path = self.translate_path(self.path) + if not os.path.isfile(path): + self.send_error(HTTPStatus.NOT_FOUND) + return + self._send_headers(HTTPStatus.OK, os.path.getsize(path)) + + def do_GET(self) -> None: + path = self.translate_path(self.path) + if not os.path.isfile(path): + self.send_error(HTTPStatus.NOT_FOUND) + return + size = os.path.getsize(path) + start, end = 0, size - 1 + if m := re.fullmatch(r"bytes=(\d+)-(\d*)", self.headers.get("Range", "")): + start = int(m[1]) + end = min(int(m[2]) if m[2] else size - 1, size - 1) + with self.server.ranged_requests.get_lock(): + self.server.ranged_requests.value += 1 + self._send_headers( + HTTPStatus.PARTIAL_CONTENT, + end - start + 1, + f"bytes {start}-{end}/{size}", + ) + else: + self._send_headers(HTTPStatus.OK, size) + with self.server.bytes_served.get_lock(): + self.server.bytes_served.value += end - start + 1 + with open(path, "rb") as fp: + fp.seek(start) + self.wfile.write(fp.read(end - start + 1)) + + +def _serve( + directory: str, + ranged_requests: Synchronized, + bytes_served: Synchronized, + port: Any, +) -> None: + server = _Server(("127.0.0.1", 0), partial(_Handler, directory=directory)) + server.ranged_requests = ranged_requests + server.bytes_served = bytes_served + port.put(server.server_address[1]) + server.serve_forever() + + +class RangeHTTPServer: + """ + Serve the files in ``directory`` over HTTP, with support for range + requests, from a separate process (until `stop()` is called) at `url` + """ + + def __init__(self, directory: Path) -> None: + ctx = multiprocessing.get_context("spawn") + self._ranged_requests: Synchronized = ctx.Value("q", 0) + self._bytes_served: Synchronized = ctx.Value("q", 0) + port = ctx.Queue() + self._process = ctx.Process( + target=_serve, + args=(str(directory), self._ranged_requests, self._bytes_served, port), + daemon=True, + ) + self._process.start() + self.url = f"http://127.0.0.1:{port.get(timeout=60)}" + + @property + def ranged_requests(self) -> int: + """The number of range requests served""" + return int(self._ranged_requests.value) + + @property + def bytes_served(self) -> int: + """The number of bytes served""" + return int(self._bytes_served.value) + + def stop(self) -> None: + self._process.terminate() + self._process.join() diff --git a/pyproject.toml b/pyproject.toml index 3fcdedd51..e74a03f04 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -120,10 +120,10 @@ test = [ # 5.0 switched to urllib3.connection.VerifiedHTTPSConnection (urllib3 2.x compat) "vcrpy >= 5.0.0", ] -# For reading the content of annexed files of DataLad datasets without +# For streaming the content of annexed files of DataLad datasets without # fetching it (requires git-annex) -datalad = [ - "datalad-fuse >= 0.5.1", +streaming = [ + "remfile", ] tools = [ "boto3", @@ -220,7 +220,6 @@ follow_imports = "skip" module = [ "bidsschematools.*", "click_didyoumean.*", - "datalad_fuse.*", "etelemetry.*", "fasteners.*", "fscacher.*", @@ -232,6 +231,7 @@ module = [ "nwbinspector.*", "pynwb.*", "pyout.*", + "remfile.*", "semantic_version.*", "vcr.*", "zarr.*", From 43cfe05922c6f0c2ce67df90376bffc0fc4a05e5 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:54:26 +0000 Subject: [PATCH 09/15] Cover the fallbacks of annex streaming in tests, and the test HTTP server The HTTP server for streaming tests now exits normally when stopped, so that coverage saves the data of its process. Test that annexed content is not streamed when it cannot be (no remfile, not annexed content, no size in the key, no URL), and that URLs which fail to connect are skipped. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/tests/test_annex_stream.py | 57 +++++++++++++++++++++++- dandi/tests/range_http_server.py | 9 ++++ 2 files changed, 64 insertions(+), 2 deletions(-) diff --git a/dandi/support/tests/test_annex_stream.py b/dandi/support/tests/test_annex_stream.py index beef5dfcc..a49281a3a 100644 --- a/dandi/support/tests/test_annex_stream.py +++ b/dandi/support/tests/test_annex_stream.py @@ -5,6 +5,7 @@ from pathlib import Path import shutil import subprocess +import sys from fscacher import PersistentCache import h5py @@ -16,7 +17,10 @@ from ..annex_stream import ( AnnexedReadableFile, annex_fingerprint, + annex_initialized, annex_key_size, + annex_urls, + can_stream, get_annexed_readable, ) from ...pynwb_utils import get_neurodata_types @@ -72,6 +76,8 @@ def test_annexed_readable( assert r.url.endswith("/content.nwb") assert r.get_size() == simple2_nwb.stat().st_size assert r.get_filename() == "sub-01.nwb" + assert r.get_mtime() is None + assert str(r) == str(nwb) with r.open() as fp: # Random access over HTTP is enough for h5py to read the file with h5py.File(fp, "r") as h5: @@ -152,8 +158,16 @@ def test_annexed_readable_no_url_readable( make_git_annex_dandiset( tmp_path / "ds", {"sub-01.nwb": (simple2_nwb, f"{base_url}/missing.nwb")} ) - r = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") - assert r is not None + found = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") + assert found is not None + # Neither a URL the server does not serve, nor one at which no server + # listens, can be read + r = AnnexedReadableFile( + filepath=found.filepath, + key=found.key, + size=found.size, + urls=found.urls + ["http://127.0.0.1:1/missing.nwb"], + ) with pytest.raises(OSError, match="None of the URLs"): r.open() @@ -220,6 +234,45 @@ def test_annexed_readable_none( assert annex_uuid(clone) == "" +@pytest.mark.ai_generated +def test_annexed_readable_none_not_streamable( + streamed_nwb: tuple[Path, RangeHTTPServer], monkeypatch: pytest.MonkeyPatch +) -> None: + nwb, _ = streamed_nwb + ds = nwb.parent.parent + objects = Path(".git", "annex", "objects", "Xx", "Yy") + # A broken symbolic link which is not to annexed content + (ds / "other.nwb").symlink_to("missing.nwb") + assert get_annexed_readable(ds / "other.nwb") is None + # The key does not record the size of the content + key = "URL--https&c%%example.org%x.nwb" + (ds / "nosize.nwb").symlink_to(objects / key / key) + assert get_annexed_readable(ds / "nosize.nwb") is None + # git-annex knows no URL for the key + key = "SHA256E-s5--" + "0" * 64 + ".nwb" + (ds / "nourl.nwb").symlink_to(objects / key / key) + assert annex_urls(ds, key) == [] + assert get_annexed_readable(ds / "nourl.nwb") is None + # remfile is not installed + monkeypatch.setitem(sys.modules, "remfile", None) + assert not can_stream() + assert get_annexed_readable(nwb) is None + + +@pytest.mark.ai_generated +def test_annex_initialized_no_git( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("PATH", str(tmp_path / "nonexistent")) + assert not annex_initialized(tmp_path) + + +@pytest.mark.ai_generated +def test_annex_urls_whereis_fails(tmp_path: Path) -> None: + # Not a Git repository + assert annex_urls(tmp_path, "SHA256E-s5--" + "0" * 64) == [] + + @pytest.mark.ai_generated def test_annexed_readable_cached_by_key( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path diff --git a/dandi/tests/range_http_server.py b/dandi/tests/range_http_server.py index 67ab07090..9d1af9fe2 100644 --- a/dandi/tests/range_http_server.py +++ b/dandi/tests/range_http_server.py @@ -19,6 +19,8 @@ import os from pathlib import Path import re +import signal +import sys from typing import Any @@ -82,12 +84,19 @@ def do_GET(self) -> None: self.wfile.write(fp.read(end - start + 1)) +def _exit(_signum: int, _frame: Any) -> None: + sys.exit(0) + + def _serve( directory: str, ranged_requests: Synchronized, bytes_served: Synchronized, port: Any, ) -> None: + # Exit normally when terminated by `RangeHTTPServer.stop()`, so that exit + # handlers run (e.g., saving coverage data when testing with coverage) + signal.signal(signal.SIGTERM, _exit) server = _Server(("127.0.0.1", 0), partial(_Handler, directory=directory)) server.ranged_requests = ranged_requests server.bytes_served = bytes_served From 91c40e8d4aba7ea4157cfae0f6d05d7b43790304 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 19:41:05 +0000 Subject: [PATCH 10/15] Stream annexed content with datalad-fuse's adapter and its remfile backend Rather than dandi reading the keys and URLs of annexed files from git-annex and streaming them with remfile itself, the content is again streamed with datalad-fuse's adapter, now that it has a remfile backend and can be told not to initialize git-annex (datalad/datalad-fuse#131, and CodyCBakerPhD/datalad-fuse's remfile-dev branch, which CI installs until those are released): - The adapter (RemoteFilesystemAdapter, with caching=False and init_annex=False) gives the state and key of a file, the URLs git-annex knows for it, and opens it from the first URL that can be read, with remfile for HDF5-based files such as NWB. dandi only wraps the file object it returns as a raw binary file, for h5py. - git-annex being initialized is still checked first (`git config annex.uuid`), so that a DataLad dataset is not loaded for each broken symbolic link of a fresh clone. - The `streaming` extra is again a `datalad` extra, requiring datalad-fuse[remfile], and the CI job is named `datalad-fuse`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- .github/workflows/run-tests.yml | 15 +- dandi/support/annex_stream.py | 217 +++++++++-------------- dandi/support/tests/test_annex_stream.py | 88 +++------ pyproject.toml | 9 +- 4 files changed, 130 insertions(+), 199 deletions(-) diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index cb2384d88..b7f2e2cd2 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -75,7 +75,7 @@ jobs: mode: lowest-deps - os: ubuntu-latest python: '3.12' - mode: annex-streaming + mode: datalad-fuse steps: - name: Set up environment @@ -126,9 +126,16 @@ jobs: git+https://github.com/hdmf-dev/hdmf \ git+https://github.com/hdmf-dev/hdmf-zarr - - name: Install git-annex and remfile (for streaming annexed content) - if: matrix.mode == 'annex-streaming' - run: pip install git-annex ".[streaming]" + # TODO: install a release once datalad-fuse has the remfile backend and the + # `init_annex` option of its adapter (datalad/datalad-fuse#131). The + # branch is fetched with datalad-fuse's tags, which its version is derived + # from. + - name: Install git-annex and datalad-fuse (for streaming annexed content) + if: matrix.mode == 'datalad-fuse' + run: | + git clone -q -b remfile-dev https://github.com/CodyCBakerPhD/datalad-fuse /tmp/datalad-fuse + git -C /tmp/datalad-fuse fetch -q --tags https://github.com/datalad/datalad-fuse + pip install git-annex "/tmp/datalad-fuse[remfile]" ".[datalad]" - name: Create NFS filesystem if: matrix.mode == 'nfs' diff --git a/dandi/support/annex_stream.py b/dandi/support/annex_stream.py index afb3b5862..e9bb8c462 100644 --- a/dandi/support/annex_stream.py +++ b/dandi/support/annex_stream.py @@ -1,42 +1,45 @@ """ Reading the content of annexed files of git-annex_ repositories (such as -DataLad_ datasets) that is not present locally, by streaming it. +DataLad_ datasets) that is not present locally, by streaming it with +datalad-fuse_. In a git-annex repository, a (locked) annexed file is a symbolic link into -:file:`.git/annex/objects/`, which is broken until the content is fetched. Its -target names the git-annex key of the content, which records the size of the -content, and git-annex knows URLs for the key (``git annex whereis``). The -content is read from those URLs with remfile_, which fetches only the parts of -a file that are read, in requests sized for the access pattern of HDF5-based -files (such as NWB), so that the files of a DataLad Dandiset can be read (e.g., -for their metadata, or to be validated) without downloading the (possibly -terabytes of) data. - -The URLs are tried in the order git-annex lists them; for a DataLad Dandiset, -that is the DANDI Archive's API download URL (which redirects to S3) before the -direct S3 URL. - -This requires git-annex and remfile (``pip install "dandi[streaming]"``). No -``git annex`` command is run in a repository in which git-annex is not -initialized (e.g., a fresh ``git clone``), as it would initialize git-annex -there, modifying the repository. +:file:`.git/annex/objects/`, which is broken until the content is fetched. +datalad-fuse can mount such a repository with FUSE so that the content is +streamed on demand, but its adapter, which the mount is built on, also opens +the files directly, with no mount: it asks git-annex where the content is +(``git annex whereis``) and reads it over HTTP(S), from the first of its URLs +that can be read, with remfile_ for HDF5-based files (such as NWB). remfile +fetches only the parts of a file that are read, in requests sized for the +access pattern of HDF5, so that the files of a DataLad Dandiset can be read +(e.g., for their metadata, or to be validated) without downloading the +(possibly terabytes of) data. + +For a DataLad Dandiset, git-annex lists the DANDI Archive's API download URL +(which redirects to S3) before the direct S3 URL. + +This requires datalad-fuse with its remfile backend (``pip install +"dandi[datalad]"``), which requires DataLad and git-annex. No ``git annex`` +command is run in a repository in which git-annex is not initialized (e.g., a +fresh ``git clone``), as it would initialize git-annex there, modifying the +repository. .. _git-annex: https://git-annex.branchable.com .. _DataLad: https://www.datalad.org +.. _datalad-fuse: https://github.com/datalad/datalad-fuse .. _remfile: https://github.com/flatironinstitute/remfile """ from __future__ import annotations +import atexit from dataclasses import dataclass, field from datetime import datetime from functools import cache import io -import json import logging import os from pathlib import Path -import re import subprocess from typing import IO, Any, cast @@ -44,16 +47,32 @@ lgr = logging.getLogger("dandi.support.annex_stream") -#: Seconds to wait for a server to start answering a request -REQUEST_TIMEOUT = 60 + +@cache +def get_adapter() -> Any: + """ + The datalad-fuse adapter shared by all files, rooted at the filesystem + root so that it serves files of any dataset, and closed at exit + """ + # Optional dependency: + from datalad_fuse.adapter import RemoteFilesystemAdapter + + # caching=False: do not keep the streamed content on disk, in the dataset. + # init_annex=False: do not initialize git-annex in a dataset in which it is + # not. The root and all paths passed to the adapter must be absolute. + adapter = RemoteFilesystemAdapter( + Path(os.path.abspath(os.sep)), caching=False, init_annex=False + ) + atexit.register(adapter.__exit__, None, None, None) + return adapter @cache def annex_initialized(directory: Path) -> bool: """ Whether ``directory`` is in a Git repository in which git-annex is - initialized. (Any ``git annex`` command would otherwise initialize it, - modifying the repository.) + initialized. (This is checked before handing a file to datalad-fuse, + which would otherwise load the dataset with DataLad only to find out.) """ try: r = subprocess.run( @@ -66,57 +85,16 @@ def annex_initialized(directory: Path) -> bool: return r.returncode == 0 and bool(r.stdout.strip()) -def annex_key_size(key: str) -> int | None: - """ - The size of the content of a git-annex key, as recorded in the key (e.g., - ``SHA256E-s1234--.nwb``), or `None` if it is not - """ - fields = key.split("--", 1)[0].split("-") - for f in fields[1:]: - if m := re.fullmatch(r"s(\d+)", f): - return int(m[1]) - return None - - -def annex_urls(directory: Path, key: str) -> list[str]: - """ - The http(s) URLs git-annex knows for ``key`` (in the order ``git annex - whereis`` lists them), in the repository at ``directory``, in which - git-annex must be initialized - """ - r = subprocess.run( - ["git", "-C", str(directory), "annex", "whereis", "--json", "--key", key], - capture_output=True, - text=True, - ) - try: - info = json.loads(r.stdout) - except ValueError: - lgr.debug("git annex whereis --key %s failed: %s", key, r.stderr.strip()) - return [] - urls: list[str] = [] - for location in info.get("whereis", []) + info.get("untrusted", []): - for url in location.get("urls", []): - if url.lower().startswith(("http://", "https://")) and url not in urls: - urls.append(url) - return urls - - -class RemoteFile(io.RawIOBase): +class StreamedFile(io.RawIOBase): """ - A seekable, read-only binary file object reading the content at ``url`` - (of ``size`` bytes) with remfile, which fetches it in the reading thread + A seekable, read-only binary file object around a file object returned by + datalad-fuse's adapter, for consumers (such as h5py) that read with + ``readinto()`` """ - def __init__(self, url: str, size: int) -> None: + def __init__(self, fp: Any) -> None: super().__init__() - # Optional dependency: - import remfile - - self.url = url - self.size = size - # Passing the size spares a request for it - self._fp = remfile.File(url, _size=size) + self._fp = fp def readable(self) -> bool: return True @@ -126,16 +104,12 @@ def seekable(self) -> bool: def readinto(self, buffer: Any) -> int: view = memoryview(buffer).cast("B") - n = min(len(view), max(self.size - self._fp.tell(), 0)) - if n == 0: - return 0 - data = self._fp.read(n) + data = self._fp.read(len(view)) view[: len(data)] = data return len(data) def seek(self, offset: int, whence: int = io.SEEK_SET) -> int: - self._fp.seek(offset, whence) - return int(self._fp.tell()) + return int(self._fp.seek(offset, whence)) def tell(self) -> int: return int(self._fp.tell()) @@ -146,31 +120,14 @@ def close(self) -> None: super().close() -def _responds(url: str) -> bool: - """Whether a range request for the first byte at ``url`` succeeds""" - # Avoid importing requests when this module is imported: - import requests - - try: - with requests.get( - url, - headers={"Range": "bytes=0-0"}, - stream=True, - timeout=REQUEST_TIMEOUT, - ) as r: - return r.status_code in (200, 206) - except requests.RequestException as e: - lgr.debug("%s: Request failed: %s", url, e) - return False - - @dataclass class AnnexedReadableFile(Readable): """ A `Readable` for a (locked) annexed file at ``filepath`` (a broken symbolic - link) whose content is not present locally, which is streamed from the - first of ``urls`` (the http(s) URLs git-annex knows for the content) that - can be read (see the module docstring). + link) whose content is not present locally, which is streamed with + datalad-fuse (see the module docstring). ``url`` is only the first URL + known for the content, for reporting: if it cannot be read, datalad-fuse + tries the others. Instances are obtained by calling `get_annexed_readable()`. """ @@ -178,27 +135,11 @@ class AnnexedReadableFile(Readable): filepath: Path key: str size: int - urls: list[str] - _url: str | None = field(default=None, init=False, repr=False, compare=False) - - @property - def url(self) -> str: - """The URL the content is read from (once it was opened), else the first""" - return self._url if self._url is not None else self.urls[0] + url: str + adapter: Any = field(repr=False, compare=False) def open(self) -> IO[bytes]: - if self._url is None: - for url in self.urls: - if _responds(url): - self._url = url - break - lgr.debug("%s: Cannot read %s", self.filepath, url) - else: - raise OSError( - f"{self.filepath}: None of the URLs of its content can be read:" - f" {', '.join(self.urls)}" - ) - return cast("IO[bytes]", RemoteFile(self._url, self.size)) + return cast("IO[bytes]", StreamedFile(self.adapter.open(self.filepath, "rb"))) def get_size(self) -> int: return self.size @@ -234,8 +175,12 @@ def annex_fingerprint(source: Any) -> tuple[str, str] | None: def can_stream() -> bool: - """Whether the requirements for streaming (remfile) are installed""" + """ + Whether the requirements for streaming are installed: datalad-fuse with + its adapter, and remfile + """ try: + import datalad_fuse.adapter # noqa: F401 import remfile # noqa: F401 except ImportError: return False @@ -245,10 +190,10 @@ def can_stream() -> bool: def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: """ Return an `AnnexedReadableFile` for streaming the content of the annexed - file at ``path``, or `None` if remfile is not installed, ``path`` is not a - (locked) annexed file whose content is missing in a repository in which - git-annex is initialized, its key does not record the size of the content, - or git-annex knows no http(s) URL to stream it from + file at ``path``, or `None` if datalad-fuse (with remfile) is not + installed, ``path`` is not a (locked) annexed file whose content is missing + in a repository in which git-annex is initialized, its key does not record + the size of the content, or git-annex knows no URL to stream it from """ filepath = Path(os.path.abspath(path)) if not filepath.is_symlink() or filepath.exists(): @@ -256,18 +201,26 @@ def get_annexed_readable(path: str | Path) -> AnnexedReadableFile | None: return None if not can_stream(): return None - target = Path(os.readlink(filepath)) - if ".git/annex/objects" not in target.as_posix(): + if not annex_initialized(filepath.parent): return None - key = target.name - size = annex_key_size(key) - if size is None: - lgr.debug("%s: The key %s does not record the size", filepath, key) + from datalad_fuse.adapter import FileState + + try: + dsap, relpath = get_adapter().resolve_dataset(filepath) + state, key = dsap.get_file_state(relpath) + except Exception as e: + # E.g., a repository without any commit + lgr.debug("%s: Cannot be read with datalad-fuse: %s", filepath, e) return None - if not annex_initialized(filepath.parent): + if state is not FileState.NO_CONTENT or key is None: return None - urls = annex_urls(filepath.parent, key) - if not urls: + if key.size is None: + lgr.debug("%s: The key %s does not record the size", filepath, key) + return None + url = next(dsap.get_urls(str(key)), None) + if url is None: lgr.debug("%s: git-annex knows no URL for key %s", filepath, key) return None - return AnnexedReadableFile(filepath=filepath, key=key, size=size, urls=urls) + return AnnexedReadableFile( + filepath=filepath, key=str(key), size=key.size, url=url, adapter=get_adapter() + ) diff --git a/dandi/support/tests/test_annex_stream.py b/dandi/support/tests/test_annex_stream.py index a49281a3a..aeba8d44d 100644 --- a/dandi/support/tests/test_annex_stream.py +++ b/dandi/support/tests/test_annex_stream.py @@ -18,8 +18,6 @@ AnnexedReadableFile, annex_fingerprint, annex_initialized, - annex_key_size, - annex_urls, can_stream, get_annexed_readable, ) @@ -31,6 +29,11 @@ pytestmark = mark.skipif_no_git_annex +@pytest.fixture(autouse=True) +def require_streaming() -> None: + pytest.importorskip("datalad_fuse.adapter") + + @pytest.fixture def streamed_nwb( range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path @@ -39,7 +42,6 @@ def streamed_nwb( An annexed copy of ``simple2_nwb`` in a git-annex repository, whose content is not present but registered at a URL of the HTTP server """ - pytest.importorskip("remfile") server, base_url = range_http_server shutil.copy(simple2_nwb, tmp_path / "served" / "content.nwb") make_git_annex_dandiset( @@ -51,20 +53,6 @@ def streamed_nwb( return nwb, server -@pytest.mark.ai_generated -@pytest.mark.parametrize( - "key,size", - [ - ("SHA256E-s1234--abc.nwb", 1234), - ("MD5E-s0--d41d8cd98f00b204e9800998ecf8427e", 0), - ("SHA256E-s99-S10-C2--abc.nwb", 99), - ("URL--https&c%%example.org%x.nwb", None), - ], -) -def test_annex_key_size(key: str, size: int | None) -> None: - assert annex_key_size(key) == size - - @pytest.mark.ai_generated def test_annexed_readable( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path @@ -89,16 +77,15 @@ def test_annexed_readable( def test_annexed_readable_file_object( range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path ) -> None: - pytest.importorskip("remfile") _, base_url = range_http_server content = bytes(range(256)) * 1000 - local = tmp_path / "local.bin" + local = tmp_path / "local.h5" local.write_bytes(content) - shutil.copy(local, tmp_path / "served" / "content.bin") + shutil.copy(local, tmp_path / "served" / "content.h5") make_git_annex_dandiset( - tmp_path / "ds", {"data.bin": (local, f"{base_url}/content.bin")} + tmp_path / "ds", {"data.h5": (local, f"{base_url}/content.h5")} ) - r = get_annexed_readable(tmp_path / "ds" / "data.bin") + r = get_annexed_readable(tmp_path / "ds" / "data.h5") assert r is not None with r.open() as fp: assert fp.readable() and fp.seekable() @@ -120,7 +107,6 @@ def test_annexed_readable_url_fallback( range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path ) -> None: """The content is read from the first of its URLs that can be read""" - pytest.importorskip("remfile") _, base_url = range_http_server shutil.copy(simple2_nwb, tmp_path / "served" / "content.nwb") make_git_annex_dandiset( @@ -132,43 +118,25 @@ def test_annexed_readable_url_fallback( ) }, ) - found = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") - assert found is not None - # (in the order in which git-annex lists them, which is not the order in - # which they were registered) - assert sorted(found.urls) == [f"{base_url}/content.nwb", f"{base_url}/missing.nwb"] - r = AnnexedReadableFile( - filepath=found.filepath, - key=found.key, - size=found.size, - urls=[f"{base_url}/missing.nwb", f"{base_url}/content.nwb"], - ) - assert r.url == f"{base_url}/missing.nwb" + r = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") + assert r is not None + # Whichever of the URLs git-annex lists first, the content is read + assert r.url in (f"{base_url}/missing.nwb", f"{base_url}/content.nwb") with r.open() as fp, h5py.File(fp, "r") as h5: assert h5.attrs["nwb_version"] - assert r.url == f"{base_url}/content.nwb" @pytest.mark.ai_generated def test_annexed_readable_no_url_readable( range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path ) -> None: - pytest.importorskip("remfile") _, base_url = range_http_server make_git_annex_dandiset( tmp_path / "ds", {"sub-01.nwb": (simple2_nwb, f"{base_url}/missing.nwb")} ) - found = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") - assert found is not None - # Neither a URL the server does not serve, nor one at which no server - # listens, can be read - r = AnnexedReadableFile( - filepath=found.filepath, - key=found.key, - size=found.size, - urls=found.urls + ["http://127.0.0.1:1/missing.nwb"], - ) - with pytest.raises(OSError, match="None of the URLs"): + r = get_annexed_readable(tmp_path / "ds" / "sub-01.nwb") + assert r is not None + with pytest.raises(OSError, match="Could not open"): r.open() @@ -177,7 +145,6 @@ def test_annexed_readable_reads_little( range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path ) -> None: """Reading an NWB file's structure downloads only a small part of it""" - pytest.importorskip("remfile") server, base_url = range_http_server nwbfile = NWBFile("streamed", "id", datetime(2026, 1, 1, tzinfo=timezone.utc)) for i in range(4): @@ -251,12 +218,19 @@ def test_annexed_readable_none_not_streamable( # git-annex knows no URL for the key key = "SHA256E-s5--" + "0" * 64 + ".nwb" (ds / "nourl.nwb").symlink_to(objects / key / key) - assert annex_urls(ds, key) == [] assert get_annexed_readable(ds / "nourl.nwb") is None - # remfile is not installed - monkeypatch.setitem(sys.modules, "remfile", None) - assert not can_stream() - assert get_annexed_readable(nwb) is None + # A repository without any commit, which datalad-fuse cannot handle + empty = ds.parent / "empty" + subprocess.run(["git", "init", "-q", str(empty)], check=True) + subprocess.run(["git", "-C", str(empty), "annex", "init", "-q"], check=True) + (empty / "sub-01.nwb").symlink_to(objects / key / key) + assert get_annexed_readable(empty / "sub-01.nwb") is None + # datalad-fuse, or remfile, is not installed + for module in ["datalad_fuse.adapter", "remfile"]: + with monkeypatch.context() as m: + m.setitem(sys.modules, module, None) + assert not can_stream() + assert get_annexed_readable(nwb) is None @pytest.mark.ai_generated @@ -267,12 +241,6 @@ def test_annex_initialized_no_git( assert not annex_initialized(tmp_path) -@pytest.mark.ai_generated -def test_annex_urls_whereis_fails(tmp_path: Path) -> None: - # Not a Git repository - assert annex_urls(tmp_path, "SHA256E-s5--" + "0" * 64) == [] - - @pytest.mark.ai_generated def test_annexed_readable_cached_by_key( streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path diff --git a/pyproject.toml b/pyproject.toml index e74a03f04..6ab975579 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -121,9 +121,11 @@ test = [ "vcrpy >= 5.0.0", ] # For streaming the content of annexed files of DataLad datasets without -# fetching it (requires git-annex) -streaming = [ - "remfile", +# fetching it (requires git-annex). +# TODO: require the first release with the remfile backend and the `init_annex` +# option of the adapter (datalad/datalad-fuse#131) +datalad = [ + "datalad-fuse[remfile] >= 0.6.0", ] tools = [ "boto3", @@ -220,6 +222,7 @@ follow_imports = "skip" module = [ "bidsschematools.*", "click_didyoumean.*", + "datalad_fuse.*", "etelemetry.*", "fasteners.*", "fscacher.*", From 92d7226be5d156fa46a0af7b91af9805b43136af Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 20:21:52 +0000 Subject: [PATCH 11/15] Pin the `datalad` extra to the datalad-fuse commit with the remfile backend Until datalad-fuse releases its remfile backend and the adapter's `init_annex` option (datalad/datalad-fuse#131), the `datalad` extra requires them from CodyCBakerPhD/datalad-fuse at a fixed commit, so that `pip install "dandi[datalad]"` gives working streaming, and CI installs just `.[datalad]`. This must be replaced by a release requirement before merging, as PyPI rejects distributions with direct URL dependencies. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- .github/workflows/run-tests.yml | 9 +-------- pyproject.toml | 7 ++++--- 2 files changed, 5 insertions(+), 11 deletions(-) diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index b7f2e2cd2..7ee5117e8 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -126,16 +126,9 @@ jobs: git+https://github.com/hdmf-dev/hdmf \ git+https://github.com/hdmf-dev/hdmf-zarr - # TODO: install a release once datalad-fuse has the remfile backend and the - # `init_annex` option of its adapter (datalad/datalad-fuse#131). The - # branch is fetched with datalad-fuse's tags, which its version is derived - # from. - name: Install git-annex and datalad-fuse (for streaming annexed content) if: matrix.mode == 'datalad-fuse' - run: | - git clone -q -b remfile-dev https://github.com/CodyCBakerPhD/datalad-fuse /tmp/datalad-fuse - git -C /tmp/datalad-fuse fetch -q --tags https://github.com/datalad/datalad-fuse - pip install git-annex "/tmp/datalad-fuse[remfile]" ".[datalad]" + run: pip install git-annex ".[datalad]" - name: Create NFS filesystem if: matrix.mode == 'nfs' diff --git a/pyproject.toml b/pyproject.toml index 6ab975579..60fb9e2eb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -122,10 +122,11 @@ test = [ ] # For streaming the content of annexed files of DataLad datasets without # fetching it (requires git-annex). -# TODO: require the first release with the remfile backend and the `init_annex` -# option of the adapter (datalad/datalad-fuse#131) +# TODO: before merging, require the first release with the remfile backend and +# the `init_annex` option of the adapter (datalad/datalad-fuse#131) instead: +# PyPI rejects distributions with direct URL dependencies. datalad = [ - "datalad-fuse[remfile] >= 0.6.0", + "datalad-fuse[remfile] @ git+https://github.com/CodyCBakerPhD/datalad-fuse@7183bf439504495be25924a90902673a92b53894", ] tools = [ "boto3", From 98b09ce369e1dbe2e31534baea9be274393640a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 21:09:11 +0000 Subject: [PATCH 12/15] Always read HDF5-based files with datalad-fuse's remfile backend Pass `backends="remfile,fsspec"` to the adapter rather than leaving it to a dataset's `datalad.fusefs.backends` configuration: a dataset configured to use fsspec would otherwise have NWB files read by fsspec, which fetches in another thread, where a garbage collection running h5py finalizers while h5py's lock is held waiting for the data would deadlock. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/annex_stream.py | 10 +++++++++- dandi/support/tests/test_annex_stream.py | 21 +++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/dandi/support/annex_stream.py b/dandi/support/annex_stream.py index e9bb8c462..2cf63c8a0 100644 --- a/dandi/support/annex_stream.py +++ b/dandi/support/annex_stream.py @@ -58,10 +58,18 @@ def get_adapter() -> Any: from datalad_fuse.adapter import RemoteFilesystemAdapter # caching=False: do not keep the streamed content on disk, in the dataset. + # backends: always read HDF5-based files (such as NWB) with remfile, + # whatever a dataset configures (datalad.fusefs.backends), as it reads in + # the calling thread: h5py holds a global lock while it reads, and fsspec, + # which reads in another thread, could deadlock with h5py finalizers run by + # a garbage collection there. # init_annex=False: do not initialize git-annex in a dataset in which it is # not. The root and all paths passed to the adapter must be absolute. adapter = RemoteFilesystemAdapter( - Path(os.path.abspath(os.sep)), caching=False, init_annex=False + Path(os.path.abspath(os.sep)), + caching=False, + backends="remfile,fsspec", + init_annex=False, ) atexit.register(adapter.__exit__, None, None, None) return adapter diff --git a/dandi/support/tests/test_annex_stream.py b/dandi/support/tests/test_annex_stream.py index aeba8d44d..482de7288 100644 --- a/dandi/support/tests/test_annex_stream.py +++ b/dandi/support/tests/test_annex_stream.py @@ -73,6 +73,27 @@ def test_annexed_readable( assert server.ranged_requests > 0 +@pytest.mark.ai_generated +def test_annexed_readable_remfile_whatever_configured( + streamed_nwb: tuple[Path, RangeHTTPServer], +) -> None: + """NWB files are read with remfile even if a dataset configures fsspec""" + from datalad_fuse.remfile import RemfileWrapper + + nwb, _ = streamed_nwb + ds = nwb.parent.parent + subprocess.run( + ["git", "-C", str(ds), "config", "datalad.fusefs.backends", "fsspec"], + check=True, + ) + r = get_annexed_readable(nwb) + assert r is not None + with r.open() as fp: + assert isinstance(fp._fp, RemfileWrapper) + with h5py.File(fp, "r") as h5: + assert h5.attrs["nwb_version"] + + @pytest.mark.ai_generated def test_annexed_readable_file_object( range_http_server: tuple[RangeHTTPServer, str], tmp_path: Path From 08e238ae00c92db39818841b775a6bd7a22fad73 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 21:09:44 +0000 Subject: [PATCH 13/15] Type-check the test that NWB files are read with remfile Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/tests/test_annex_stream.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/dandi/support/tests/test_annex_stream.py b/dandi/support/tests/test_annex_stream.py index 482de7288..809622fa6 100644 --- a/dandi/support/tests/test_annex_stream.py +++ b/dandi/support/tests/test_annex_stream.py @@ -16,6 +16,7 @@ from ..annex_stream import ( AnnexedReadableFile, + StreamedFile, annex_fingerprint, annex_initialized, can_stream, @@ -89,6 +90,7 @@ def test_annexed_readable_remfile_whatever_configured( r = get_annexed_readable(nwb) assert r is not None with r.open() as fp: + assert isinstance(fp, StreamedFile) assert isinstance(fp._fp, RemfileWrapper) with h5py.File(fp, "r") as h5: assert h5.attrs["nwb_version"] From 06f96ebb5e04c61a990d8f8257786d9470e0329a Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 21:59:58 +0000 Subject: [PATCH 14/15] Require only datalad-fuse's remfile backend (datalad/datalad-fuse#131) Pin the datalad extra to the head commit of datalad/datalad-fuse#131 rather than a fork combining it with an init_annex option of the adapter: dandi hands files to the adapter only in repositories in which git-annex is initialized, so the adapter never initializes it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/support/annex_stream.py | 6 +++--- pyproject.toml | 8 ++++---- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/dandi/support/annex_stream.py b/dandi/support/annex_stream.py index 2cf63c8a0..95e5c831c 100644 --- a/dandi/support/annex_stream.py +++ b/dandi/support/annex_stream.py @@ -63,13 +63,13 @@ def get_adapter() -> Any: # the calling thread: h5py holds a global lock while it reads, and fsspec, # which reads in another thread, could deadlock with h5py finalizers run by # a garbage collection there. - # init_annex=False: do not initialize git-annex in a dataset in which it is - # not. The root and all paths passed to the adapter must be absolute. + # The root and all paths passed to the adapter must be absolute. (Files + # are only handed to the adapter in repositories in which git-annex is + # initialized: see `annex_initialized()`.) adapter = RemoteFilesystemAdapter( Path(os.path.abspath(os.sep)), caching=False, backends="remfile,fsspec", - init_annex=False, ) atexit.register(adapter.__exit__, None, None, None) return adapter diff --git a/pyproject.toml b/pyproject.toml index 60fb9e2eb..c83746e96 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -122,11 +122,11 @@ test = [ ] # For streaming the content of annexed files of DataLad datasets without # fetching it (requires git-annex). -# TODO: before merging, require the first release with the remfile backend and -# the `init_annex` option of the adapter (datalad/datalad-fuse#131) instead: -# PyPI rejects distributions with direct URL dependencies. +# TODO: before merging, require the first release with the remfile backend +# (datalad/datalad-fuse#131) instead: PyPI rejects distributions with direct URL +# dependencies. datalad = [ - "datalad-fuse[remfile] @ git+https://github.com/CodyCBakerPhD/datalad-fuse@7183bf439504495be25924a90902673a92b53894", + "datalad-fuse[remfile] @ git+https://github.com/datalad/datalad-fuse@b4dbee656eeb8ed54b1396c8d7e5c8b369b64a4a", ] tools = [ "boto3", From b75c824d248d7abf498722f4c5dca43234f14d9b Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 02:26:53 +0000 Subject: [PATCH 15/15] Disable automatic GC while a remote asset is open, to avoid a deadlock h5py holds its global lock while it reads from a Python file object, and a read of a RemoteReadableAsset waits for fsspec's IO thread to fetch the data. If a garbage collection runs in that thread meanwhile, the finalizer of an unreferenced object closing an HDF5 file (e.g., hdmf's HDMFIO.__del__ -> HDF5IO.close() -> h5py's File.close()) waits for the same lock, and neither thread can proceed. The aiohttp timeouts cannot end this, as the event loop thread itself is blocked. This is the hang of test_nwb2asset_remote_asset (#1450), and it made test_reextract_metadata time out in the lowest-deps CI job of #1955. RemoteReadableAsset.open() now wraps fsspec's file object in a _GCPausedFile, which disables automatic garbage collection while any such file is open (and restores the previous state once all are closed). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/misctypes.py | 84 +++++++++++++++++++++++++++++++---- dandi/tests/test_dandiapi.py | 21 +++++++++ dandi/tests/test_misctypes.py | 44 ++++++++++++++++++ 3 files changed, 141 insertions(+), 8 deletions(-) create mode 100644 dandi/tests/test_misctypes.py diff --git a/dandi/misctypes.py b/dandi/misctypes.py index 5ecc25486..df2f4d3fc 100644 --- a/dandi/misctypes.py +++ b/dandi/misctypes.py @@ -11,9 +11,11 @@ from dataclasses import dataclass from datetime import datetime from fnmatch import fnmatchcase +import gc import os.path from pathlib import Path -from typing import IO, TypeVar, cast +import threading +from typing import IO, Any, TypeVar, cast from dandischema.models import DigestType @@ -355,15 +357,22 @@ def open(self) -> IO[bytes]: # (or minio in tests) causes fsspec's background IO thread to block # forever, which in turn blocks the calling thread in # threading.Event.wait() — see https://github.com/fsspec/filesystem_spec/issues/1666 + # + # Automatic garbage collection is disabled while the file is open, as + # it could otherwise deadlock fsspec's IO thread (see `_GCPausedFile`). return cast( IO[bytes], - fsspec.open( - self.url, - mode="rb", - client_kwargs={ - "timeout": ClientTimeout(total=120, sock_read=60, sock_connect=30) - }, - ).open(), + _GCPausedFile( + fsspec.open( + self.url, + mode="rb", + client_kwargs={ + "timeout": ClientTimeout( + total=120, sock_read=60, sock_connect=30 + ) + }, + ).open() + ), ) def get_size(self) -> int: @@ -377,3 +386,62 @@ def get_filename(self) -> str: def __str__(self) -> str: return self.url + + +#: Protects `_gc_pauses` and `_gc_was_enabled` +_gc_lock = threading.Lock() +#: The number of open `_GCPausedFile` instances +_gc_pauses = 0 +#: Whether automatic garbage collection was enabled before the first open +#: `_GCPausedFile` disabled it +_gc_was_enabled = False + + +class _GCPausedFile: + """ + A file object during whose lifetime automatic garbage collection is + disabled + + h5py holds a global lock while it reads from a Python file object, and the + read of a `RemoteReadableAsset` waits for fsspec's IO thread to fetch the + data. If a garbage collection ran in that thread meanwhile, the finalizer + of an unreferenced object closing an HDF5 file (e.g., hdmf's + ``HDMFIO.__del__``) would wait for the same lock, and neither thread could + proceed (). + """ + + def __init__(self, fp: IO[bytes]) -> None: + global _gc_pauses, _gc_was_enabled + self._fp = fp + self._paused = False + with _gc_lock: + if _gc_pauses == 0: + _gc_was_enabled = gc.isenabled() + gc.disable() + _gc_pauses += 1 + self._paused = True + + def __getattr__(self, name: str) -> Any: + return getattr(self._fp, name) + + def __enter__(self) -> _GCPausedFile: + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + def __del__(self) -> None: + if self._paused: + self.close() + + def close(self) -> None: + global _gc_pauses + try: + self._fp.close() + finally: + with _gc_lock: + if self._paused: + self._paused = False + _gc_pauses -= 1 + if _gc_pauses == 0 and _gc_was_enabled: + gc.enable() diff --git a/dandi/tests/test_dandiapi.py b/dandi/tests/test_dandiapi.py index 8da28852e..670cb743a 100644 --- a/dandi/tests/test_dandiapi.py +++ b/dandi/tests/test_dandiapi.py @@ -2,6 +2,7 @@ import builtins from datetime import datetime, timezone +import gc import json import logging from pathlib import Path @@ -1107,6 +1108,26 @@ def test_asset_as_readable_open(new_dandiset: SampleDandiset, tmp_path: Path) -> fp.close() +@pytest.mark.ai_generated +def test_asset_as_readable_open_pauses_gc( + new_dandiset: SampleDandiset, tmp_path: Path +) -> None: + # Automatic garbage collection is disabled while a remote asset is open, so + # that it cannot run in fsspec's IO thread while h5py holds its lock + # (https://github.com/dandi/dandi-cli/issues/1450) + p = tmp_path / "foo.txt" + p.write_bytes(b"This is test text.\n") + d = new_dandiset.dandiset + d.upload_raw_asset(p, {"path": "foo.txt"}) + (asset,) = d.get_assets() + assert isinstance(asset, RemoteBlobAsset) + assert gc.isenabled() + with asset.as_readable().open() as fp: + assert not gc.isenabled() + assert fp.read() == b"This is test text.\n" + assert gc.isenabled() + + @pytest.mark.parametrize( ("instance_name", "expected_env_var_name"), [ diff --git a/dandi/tests/test_misctypes.py b/dandi/tests/test_misctypes.py new file mode 100644 index 000000000..27fc0136d --- /dev/null +++ b/dandi/tests/test_misctypes.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import gc +from io import BytesIO + +import pytest + +from ..misctypes import _GCPausedFile + + +@pytest.mark.ai_generated +def test_gc_paused_file() -> None: + # Automatic garbage collection is disabled while any wrapped file is open + assert gc.isenabled() + with _GCPausedFile(BytesIO(b"0123456789")) as fp: + assert not gc.isenabled() + fp2 = _GCPausedFile(BytesIO(b"abc")) + assert fp2.read(2) == b"ab" + fp2.close() + assert not gc.isenabled() + fp.seek(2) + assert fp.read(3) == b"234" + assert fp.closed + assert gc.isenabled() + # Closing again changes nothing + fp.close() + assert gc.isenabled() + # Dropping an unclosed file re-enables it too + fp3 = _GCPausedFile(BytesIO(b"")) + assert not gc.isenabled() + del fp3 + assert gc.isenabled() + + +@pytest.mark.ai_generated +def test_gc_paused_file_gc_disabled() -> None: + # If automatic garbage collection was disabled already, it stays disabled + gc.disable() + try: + with _GCPausedFile(BytesIO(b"")): + assert not gc.isenabled() + assert not gc.isenabled() + finally: + gc.enable()