diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 948b34353..7ee5117e8 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-fuse 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-fuse' + 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..24db09d52 100644 --- a/dandi/metadata/nwb.py +++ b/dandi/metadata/nwb.py @@ -22,13 +22,14 @@ metadata_cache, nwb_has_external_links, ) +from ..support.annex_stream import annex_fingerprint from ..utils import find_parent_directory_containing lgr = get_logger() # 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/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/pynwb_utils.py b/dandi/pynwb_utils.py index 297df05f3..b0dda71c0 100644 --- a/dandi/pynwb_utils.py +++ b/dandi/pynwb_utils.py @@ -42,6 +42,7 @@ metadata_nwb_subject_fields, ) from .misctypes import Readable +from .support.annex_stream import annex_fingerprint from .utils import get_module_version, is_url from .validate._types import ( Origin, @@ -194,7 +195,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 +809,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/annex_stream.py b/dandi/support/annex_stream.py new file mode 100644 index 000000000..95e5c831c --- /dev/null +++ b/dandi/support/annex_stream.py @@ -0,0 +1,234 @@ +""" +Reading the content of annexed files of git-annex_ repositories (such as +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. +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 logging +import os +from pathlib import Path +import subprocess +from typing import IO, Any, cast + +from ..misctypes import Readable + +lgr = logging.getLogger("dandi.support.annex_stream") + + +@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. + # 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. + # 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", + ) + 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. (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( + ["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()) + + +class StreamedFile(io.RawIOBase): + """ + 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, fp: Any) -> None: + super().__init__() + self._fp = fp + + def readable(self) -> bool: + return True + + def seekable(self) -> bool: + return True + + def readinto(self, buffer: Any) -> int: + view = memoryview(buffer).cast("B") + data = self._fp.read(len(view)) + view[: len(data)] = data + return len(data) + + def seek(self, offset: int, whence: int = io.SEEK_SET) -> int: + return int(self._fp.seek(offset, whence)) + + def tell(self) -> int: + return int(self._fp.tell()) + + def close(self) -> None: + if not self.closed: + self._fp.close() + super().close() + + +@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, datalad-fuse + 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]", StreamedFile(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)) + + +def can_stream() -> bool: + """ + 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 + 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 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(): + # Not a broken symbolic link + return None + if not can_stream(): + return None + if not annex_initialized(filepath.parent): + return None + 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 state is not FileState.NO_CONTENT or key is None: + return None + 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=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 new file mode 100644 index 000000000..809622fa6 --- /dev/null +++ b/dandi/support/tests/test_annex_stream.py @@ -0,0 +1,286 @@ +from __future__ import annotations + +from datetime import datetime, timezone +import io +from pathlib import Path +import shutil +import subprocess +import sys + +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 ..annex_stream import ( + AnnexedReadableFile, + StreamedFile, + annex_fingerprint, + annex_initialized, + can_stream, + get_annexed_readable, +) +from ...pynwb_utils import get_neurodata_types +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 + + +@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 +) -> 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 + """ + 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" + 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: + assert h5.attrs["nwb_version"] + 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, StreamedFile) + 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 +) -> None: + _, base_url = range_http_server + content = bytes(range(256)) * 1000 + local = tmp_path / "local.h5" + local.write_bytes(content) + shutil.copy(local, tmp_path / "served" / "content.h5") + make_git_annex_dandiset( + tmp_path / "ds", {"data.h5": (local, f"{base_url}/content.h5")} + ) + 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() + 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""" + _, 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"], + ) + }, + ) + 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"] + + +@pytest.mark.ai_generated +def test_annexed_readable_no_url_readable( + range_http_server: tuple[RangeHTTPServer, str], simple2_nwb: Path, tmp_path: Path +) -> None: + _, 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="Could not open"): + r.open() + + +@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""" + 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) + assert 0 < server.bytes_served < size / 10 + + +def annex_uuid(repo: Path) -> str: + r = subprocess.run( + ["git", "-C", str(repo), "config", "--get", "annex.uuid"], + capture_output=True, + text=True, + ) + return r.stdout.strip() + + +@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 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 +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 get_annexed_readable(ds / "nourl.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 +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_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..90b69d792 100644 --- a/dandi/tests/fixtures.py +++ b/dandi/tests/fixtures.py @@ -28,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 @@ -363,6 +364,65 @@ def fixture( return fixture +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, 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, + "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, (_, urls) in files.items(): + key = git("annex", "lookupkey", relpath) + 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) + + +@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, from a separate process; yields the server and its base URL + """ + served = tmp_path / "served" + served.mkdir() + server = RangeHTTPServer(served) + try: + yield server, server.url + finally: + server.stop() + + 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/range_http_server.py b/dandi/tests/range_http_server.py new file mode 100644 index 000000000..9d1af9fe2 --- /dev/null +++ b/dandi/tests/range_http_server.py @@ -0,0 +1,138 @@ +""" +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 +import signal +import sys +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 _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 + 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/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/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() diff --git a/pyproject.toml b/pyproject.toml index 7a6ba7b9b..c83746e96 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,14 @@ test = [ # 5.0 switched to urllib3.connection.VerifiedHTTPSConnection (urllib3 2.x compat) "vcrpy >= 5.0.0", ] +# 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 +# (datalad/datalad-fuse#131) instead: PyPI rejects distributions with direct URL +# dependencies. +datalad = [ + "datalad-fuse[remfile] @ git+https://github.com/datalad/datalad-fuse@b4dbee656eeb8ed54b1396c8d7e5c8b369b64a4a", +] tools = [ "boto3", ] @@ -215,6 +223,7 @@ follow_imports = "skip" module = [ "bidsschematools.*", "click_didyoumean.*", + "datalad_fuse.*", "etelemetry.*", "fasteners.*", "fscacher.*", @@ -226,6 +235,7 @@ module = [ "nwbinspector.*", "pynwb.*", "pyout.*", + "remfile.*", "semantic_version.*", "vcr.*", "zarr.*",