Cached filesystem not concurrency-safe?

[jwodder]

We have a program that uses fsspec to mount a cached HTTP filesystem as a FUSE mount; while this process runs in the background, another traverses the FUSE mount and inspects multiple files in parallel. Unfortunately, the FUSE process keeps hitting errors in the following form:

Uncaught exception from FUSE operation open, returning errno.EINVAL.
Traceback (most recent call last):
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 734, in _wrapper
    return func(*args, **kwargs) or 0
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 834, in open
    fi.fh = self.operations('open', path.decode(self.encoding),
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/datalad_fuse/fuse_.py", line 75, in __call__
    return super(DataLadFUSE, self).__call__(op, self.root + path, *args)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 1076, in __call__
    return getattr(self, op)(*args)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/datalad_fuse/fuse_.py", line 203, in open
    fsspec_file = self._adapter.open(path)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/datalad_fuse/fsspec.py", line 243, in open
    return dsap.open(relpath, mode=mode, encoding=encoding, errors=errors)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/datalad_fuse/fsspec.py", line 169, in open
    return self.fs.open(url, mode, **kwargs)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 406, in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/spec.py", line 1034, in open
    f = self._open(
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 406, in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 345, in _open
    self.save_cache()
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 406, in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 161, in save_cache
    cached_files = pickle.load(f)
EOFError: Ran out of input

I suspect the cause is that the cached filesystem is not safe for concurrent access, with the result that multiple actions on the filesystem are causing the cache file to be read by one procedure while another writes to it. Indeed, there seems to be no locking in the cached filesystem code.

[martindurant]

You may well be right. The caching metadat should be concurrent safe (i.e., async), but not thread safe. The writing of the file could be protected by a lock, or could use filesystem mv to become atomic (which only works on the local filesystem)

[yarikoptic]

Thank you @martindurant for the quick response. Do you think you would implement one of those mechanisms in some near future to make fsspec thread safe in that regard?

[martindurant]

It is a reasonable request, but I don't think anyone has such plans right now.

[yarikoptic]

ok, if you were to do it -- would threading lock solution be considered? (I think making mv atomic might be half a solution since it might still require thread locking to avoid parallel modification and thus either conflicts or just loosing some data from one of the racing threads)

[martindurant]

The cache metadata should be conservative, so that races lead to occasional unnecessary re-fetches of data that were cached but the info not persisted. That's OK, if it is relatively rare.

The problem with the lock, is that it will still be vulnerable to multiple processes using the same cache directory.

[yarikoptic]

ok, @jwodder - could you please implement a mv based solution for storing the cache and see if it would resolve our dandi/dandisets-healthstatus#1 ?

[jwodder]

The save_cache() method already seems to use a mv-based approach:

filesystem_spec/fsspec/implementations/cached.py

Lines 187 to 191 in acad158

	fd2, fn2 = tempfile.mkstemp()
	with open(fd2, "wb") as f:
	pickle.dump(cache, f)
	self._mkcache()
	move(fn2, fn)

The only other method that writes to the cache is clear_expired_cache(), but I don't think that's used here.

[yarikoptic]

I wonder if that move is actually atomic?! the fn is fn = os.path.join(self.storage[-1], "cache") and thus would reside at self.storage, and in our case somewhere under .git/ within that repo. tempfile.mkstemp would generate in TEMPDIR which could be on a different filesystem/storage, and that is the case in our case.
Thus move would entail copying across filesystems: I do not think (although didn't check) it would bother creating some temp other file near destination of copy first to rename later. Thus, and in any case, e.g. for performance reasons, it might be worth making fn2 to be something like fn2 = f'{fn}-{randomstring}' and then placing it all into try/except so in case of error to if os.path.exists(fn2): unlink(fn2). This way move can be atomic within filesystem.

[martindurant]

Yes, I agree with that pattern - it's exactly what I've had to do elsewhere recently. This is what happens when you develop on a machine with exactly one disk :)

[yarikoptic]

Yes, I agree with that pattern - it's exactly what I've had to do elsewhere recently. This is what happens when you develop on a machine with exactly one disk :)

so seems like it begs for some reusbale helper like

with atomic_write(path, "wb") as f:
    # do whatever for that f handle

which would take care about all the renames etc at the target path. @jwodder could you please code this up to be used in the code block you pointed out, send a PR with it, and see if that would be sufficient to resolve our issue?

[jwodder]

Now the FUSEd cached filesystem is failing with:

Uncaught exception from FUSE operation getattr, returning errno.EINVAL.
Traceback (most recent call last):
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 734, in _wrapper
    return func(*args, **kwargs) or 0
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 774, in getattr
    return self.fgetattr(path, buf, None)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 1027, in fgetattr
    attrs = self.operations('getattr', self._decode_optional_path(path), fh)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/datalad_fuse/fuse_.py", line 75, in __call__
    return super(DataLadFUSE, self).__call__(op, self.root + path, *args)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fuse.py", line 1076, in __call__
    return getattr(self, op)(*args)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/methodtools.py", line 72, in __call__
    return self.__call__(*args, **kwargs)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/datalad_fuse/fuse_.py", line 166, in getattr
    fsspec_file.close()
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 381, in <lambda>
    f.close = lambda: self.close_and_update(f, close)
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 444, in <lambda>
    return lambda *args, **kw: getattr(type(self), item).__get__(self)(
  File "/home/dandi/cronlib/dandisets-healthstatus/venv/lib/python3.8/site-packages/fsspec/implementations/cached.py", line 394, in close_and_update
    c = self.cached_files[-1][path]
KeyError: 'https://api.dandiarchive.org/api/assets/1355f0fc-2926-406a-b6a5-77cc0681c54f/download'

[martindurant]

huh, that file exists, but is big. Let's remove that test, I don't remember why it's around. Perhaps we can think of some other very static URL that's likely to be around for a long time.

[jwodder]

@martindurant Did you mean to comment on a different issue? My comment has nothing to do with fsspec's tests.