Add locks around fsspec operations

datalad_fuse/fuse_.py

@@ -150,7 +150,8 @@ def getattr(self, path, fh=None):
                 # We should just fabricate stats from the key here or not even
                 # bother???!
                 lgr.debug("File not already open")
-                fsspec_file = self._adapter.open(path)
+                with self.rwlock:
+                    fsspec_file = self._adapter.open(path)
                 to_close = True
             if fsspec_file:
                 if isinstance(fsspec_file, io.BufferedIOBase):
@@ -163,7 +164,8 @@ def getattr(self, path, fh=None):
                         fsspec_file, timestamp=self._adapter.get_commit_datetime(path)
                     )
                 if to_close:
-                    fsspec_file.close()
+                    with self.rwlock:
+                        fsspec_file.close()
             else:
                 # TODO: although seems to be logical -- seems to cause logging etc
                 # lgr.error("ENOENTing %s %s", path, fh)
@@ -200,7 +202,8 @@ def open(self, path, flags):
             else:
                 # write/create
                 raise FuseOSError(EROFS)
-            fsspec_file = self._adapter.open(path)
+            with self.rwlock:
+                fsspec_file = self._adapter.open(path)
             lgr.debug("Counter = %d", self._counter)
             # TODO: threadlock ?
             self._fhdict[self._counter] = fsspec_file  # self.fs.open(fn, mode)
@@ -219,8 +222,9 @@ def read(self, _path, size, offset, fh):
             # must be open already and we must have mapped it to fsspec file
             # TODO: check for path to correspond?
             f = self._fhdict[fh]
-            f.seek(offset)
-            return f.read(size)
+            with self.rwlock:
+                f.seek(offset)
+                return f.read(size)
 
     def opendir(self, path):
         lgr.debug("opendir(path=%r)", path)
@@ -259,7 +263,8 @@ def release(self, path, fh):
             #  files, so we need to provide some proper use of lru_cache
             #  to have not recently used closed
             if f is not None and not f.closed:
-                f.close()
+                with self.rwlock:
+                    f.close()
         return 0
 
     def readlink(self, path):

datalad_fuse/tests/conftest.py

@@ -18,6 +18,26 @@
 
 lgr = logging.getLogger("datalad.fuse.tests")
 
+# We need large files that exceed the blocksize of 1 MiB
+BIG_URLS = [
+    # (dataset path, URL, SHA256 digest)
+    (
+        "APL.pdf",
+        "http://www.softwarepreservation.org/projects/apl/Books/APROGRAMMING%20LANGUAGE",
+        "c65ccc2a97cdb6042641847112dc6e4d4d6e75fdedcd476fdd61f855711bbaf4",
+    ),
+    (
+        "gameboy.pdf",
+        "https://archive.org/download/GameBoyProgManVer1.1/GameBoyProgManVer1.1.pdf",
+        "5263e6c1f5fa51fc6813d2ed71c738c887ec78554eef633ad72c6285f7ff9197",
+    ),
+    (
+        "libpython3.10-stdlib_3.10.4-3_i386.deb",
+        "http://nyc3.clouds.archive.ubuntu.com/ubuntu/pool/main/p/python3.10/libpython3.10-stdlib_3.10.4-3_i386.deb",
+        "e79c1416ec792b61ad9770f855bf6889e57be5f6511ea814d81ef5f9b1a3eec9",
+    ),
+]
+
 
 @pytest.fixture(autouse=True)
 def capture_all_logs(caplog):
@@ -163,3 +183,12 @@ def superdataset(served_files, request, tmp_home, tmp_path_factory):  # noqa: U1
     sub = ds.create(dspath / "sub")
     initdataset(sub, served_files, request.param == "remote")
     return (ds, {os.path.join("sub", df.path): df.content for df in served_files})
+
+
+@pytest.fixture
+def big_url_dataset(tmp_home, tmp_path_factory):  # noqa: U100
+    workpath = tmp_path_factory.mktemp("big_url_dataset")
+    ds = Dataset(workpath / "ds").create()
+    for path, url, _ in BIG_URLS:
+        ds.repo.add_url_to_file(path, url, options=["--relaxed"])
+    yield (ds, {path: digest for path, _, digest in BIG_URLS})

datalad_fuse/tests/test_fuse.py

@@ -1,4 +1,6 @@
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
+import hashlib
 import os.path
 from pathlib import Path
 import subprocess
@@ -231,3 +233,23 @@ def test_fuse_git_status(tmp_path):
             universal_newlines=True,
         )
         assert r.stdout == ""
+
+
+def test_parallel_access(tmp_path, big_url_dataset):
+    ds, data_files = big_url_dataset
+    with fusing(ds.path, tmp_path) as mount:
+        with ThreadPoolExecutor() as pool:
+            futures = {
+                pool.submit(sha256_file, mount / path): dgst
+                for path, dgst in data_files.items()
+            }
+            for fut in as_completed(futures.keys()):
+                assert fut.result() == futures[fut]
+
+
+def sha256_file(path):
+    dgst = hashlib.sha256()
+    with open(path, "rb") as fp:
+        for chunk in iter(lambda: fp.read(65535), b""):
+            dgst.update(chunk)
+    return dgst.hexdigest()