Skip to content

Commit 7352f59

Browse files
committed
Merge branch 'claude/validate-stream-via-datalad-fuse' into claude/datalad-fuse-remfile
Resolve the import conflict, and check the type of the file object wrapped by the one that pauses garbage collection in the remfile test. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr
2 parents 5dfd63e + 0be1c28 commit 7352f59

2 files changed

Lines changed: 82 additions & 2 deletions

File tree

‎dandi/support/datalad_fuse.py‎

Lines changed: 59 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,13 @@
3232
from dataclasses import dataclass, field
3333
from datetime import datetime
3434
from functools import cache
35+
import gc
3536
from itertools import chain
3637
import logging
3738
import os
3839
from pathlib import Path
3940
import subprocess
41+
import threading
4042
from typing import IO, Any, cast
4143

4244
from ..misctypes import Readable
@@ -84,6 +86,62 @@ def annex_initialized(directory: Path) -> bool:
8486
return r.returncode == 0 and bool(r.stdout.strip())
8587

8688

89+
_gc_lock = threading.Lock()
90+
_gc_pauses = 0
91+
_gc_was_enabled = False
92+
93+
94+
class _GCPausedFile:
95+
"""
96+
A file object streamed by datalad-fuse, during whose lifetime automatic
97+
garbage collection is disabled
98+
99+
h5py holds a global lock while it reads from a Python file object, and the
100+
read waits for another thread (fsspec's I/O thread, or an HTTP server in
101+
the same process) to fetch the data. If a garbage collection ran in that
102+
thread meanwhile, finalizers of h5py objects (e.g., of an unreferenced
103+
``NWBHDF5IO``) would wait for the same lock, deadlocking the process.
104+
"""
105+
106+
def __init__(self, fp: Any) -> None:
107+
global _gc_pauses, _gc_was_enabled
108+
with _gc_lock:
109+
if _gc_pauses == 0:
110+
# Collect pending garbage now, in this thread, which does not
111+
# hold h5py's lock yet
112+
gc.collect()
113+
_gc_was_enabled = gc.isenabled()
114+
gc.disable()
115+
_gc_pauses += 1
116+
self._fp = fp
117+
self._paused = True
118+
119+
def __getattr__(self, name: str) -> Any:
120+
return getattr(self._fp, name)
121+
122+
def __enter__(self) -> _GCPausedFile:
123+
return self
124+
125+
def __exit__(self, *_exc: Any) -> None:
126+
self.close()
127+
128+
def __del__(self) -> None:
129+
if self._paused:
130+
self.close()
131+
132+
def close(self) -> None:
133+
global _gc_pauses
134+
try:
135+
self._fp.close()
136+
finally:
137+
with _gc_lock:
138+
if self._paused:
139+
self._paused = False
140+
_gc_pauses -= 1
141+
if _gc_pauses == 0 and _gc_was_enabled:
142+
gc.enable()
143+
144+
87145
@dataclass
88146
class AnnexedReadableFile(Readable):
89147
"""
@@ -103,7 +161,7 @@ class AnnexedReadableFile(Readable):
103161
adapter: Any = field(repr=False, compare=False)
104162

105163
def open(self) -> IO[bytes]:
106-
return cast("IO[bytes]", self.adapter.open(self.filepath, "rb"))
164+
return cast("IO[bytes]", _GCPausedFile(self.adapter.open(self.filepath, "rb")))
107165

108166
def get_size(self) -> int:
109167
return self.size

‎dandi/support/tests/test_datalad_fuse.py‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
from __future__ import annotations
22

3+
import gc
34
from pathlib import Path
45
import shutil
56

@@ -56,6 +57,26 @@ def test_annexed_readable(
5657
assert server.ranged_requests > 0
5758

5859

60+
@pytest.mark.ai_generated
61+
def test_annexed_readable_pauses_gc(
62+
streamed_nwb: tuple[Path, RangeHTTPServer],
63+
) -> None:
64+
# Automatic garbage collection is disabled while any streamed file is open
65+
# (to avoid deadlocks with h5py's lock; see `_GCPausedFile`)
66+
nwb, _ = streamed_nwb
67+
r = get_annexed_readable(nwb)
68+
assert r is not None
69+
assert gc.isenabled()
70+
with r.open() as fp:
71+
assert not gc.isenabled()
72+
with r.open() as fp2:
73+
assert fp2.read(8) == b"\x89HDF\r\n\x1a\n"
74+
assert not gc.isenabled()
75+
fp.seek(1)
76+
assert fp.read(3) == b"HDF"
77+
assert gc.isenabled()
78+
79+
5980
@pytest.mark.ai_generated
6081
def test_annexed_readable_none(
6182
streamed_nwb: tuple[Path, RangeHTTPServer], simple2_nwb: Path, tmp_path: Path
@@ -111,7 +132,8 @@ def test_annexed_readable_remfile(
111132
assert r is not None
112133
fp = r.open()
113134
try:
114-
assert isinstance(fp, RemfileWrapper)
135+
# (wrapped to pause garbage collection while it is open)
136+
assert isinstance(fp._fp, RemfileWrapper) # type: ignore[attr-defined]
115137
with h5py.File(fp, "r") as h5:
116138
assert h5.attrs["nwb_version"]
117139
finally:

0 commit comments

Comments
 (0)