Repository: fsspec/filesystem_spec
Pull request #1755 (2024-11-21 19:02:19)

Ensure reference FS wraps any sync filesystems
This PR guards against the possibility (and likelihood, as `LocalFileSystem` is the default value here) that non-async filesystems are used in `ReferenceFileSystem`. This is important, as `ReferenceFileSystem` is, itself, an `AsyncFileSystem` and it is currently failing to provide functional methods required by that interface (cf. https://github.com/fsspec/filesystem_spec/blob/master/fsspec/implementations/reference.py#L809)

martindurant: You are thinking of a referenceFS called from an async context?
moradology: Test failures here are pretty strange. Don't think these changes should have any effect on `test_async.py` - certainly nothing obvious
martindurant: Those tests seem to be counting the number of coroutines, but now you are wrapping the coroutines inside coroutines? Maybe there's a more rigorous way to count.
martindurant: Actually this is the error. The stuff about async is in the cleanup stage for those coroutines (which ought to be awaited in the test even when they fail, but this doesn't cause a failure)
```
m = <fsspec.implementations.memory.MemoryFileSystem object at 0x7f1b6cc6dc10>

    def test_fss_has_defaults(m):
        fs = fsspec.filesystem("reference", fo={})
        assert None in fs.fss
    
        fs = fsspec.filesystem("reference", fo={}, remote_protocol="memory")
>       assert fs.fss[None].protocol == "memory"
E       AssertionError: assert 'abstract' == 'memory'
E         
E         - memory
E         + abstract
```
martindurant: To fix the async thing (if you wish to try), probably in this block from `_run_coros_in_chunks`, all remaining coroutines should be cancelled and awaited
```
        try:
            return await asyncio.wait_for(coro, timeout=timeout), i
        except Exception as e:
            if not return_exceptions:
                # <- cancel pending coroutines
                raise
            return e, i
```
(or maybe try/except at `await asyncio.wait(..)` a few lines down)
moradology: The remaining s3fs test failure is pretty strange! Looks to me like conda is installing the wrong version? Came across this: https://github.com/eWaterCycle/ewatercycle/pull/468

Not sure of the best way to test it, unfortunately, as this is pretty specific to CI. Looks like we may be able to simply pin the version over in s3fs (or maybe here?): `pyopenssl>=24.0.0`

Patch:
diff --git a/fsspec/implementations/asyn_wrapper.py b/fsspec/implementations/asyn_wrapper.py
index 0d46265..43604a2 100644
--- a/fsspec/implementations/asyn_wrapper.py
+++ b/fsspec/implementations/asyn_wrapper.py
@@ -45,26 +45,27 @@ class AsyncFileSystemWrapper(AsyncFileSystem):
     def __init__(self, sync_fs, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.asynchronous = True
-        self.fs = sync_fs
+        self.sync_fs = sync_fs
+        self.protocol = self.sync_fs.protocol
         self._wrap_all_sync_methods()
 
     @property
     def fsid(self):
-        return f"async_{self.fs.fsid}"
+        return f"async_{self.sync_fs.fsid}"
 
     def _wrap_all_sync_methods(self):
         """
         Wrap all synchronous methods of the underlying filesystem with asynchronous versions.
         """
-        for method_name in dir(self.fs):
+        for method_name in dir(self.sync_fs):
             if method_name.startswith("_"):
                 continue
 
-            attr = inspect.getattr_static(self.fs, method_name)
+            attr = inspect.getattr_static(self.sync_fs, method_name)
             if isinstance(attr, property):
                 continue
 
-            method = getattr(self.fs, method_name)
+            method = getattr(self.sync_fs, method_name)
             if callable(method) and not asyncio.iscoroutinefunction(method):
                 async_method = async_wrapper(method, obj=self)
                 setattr(self, f"_{method_name}", async_method)
diff --git a/fsspec/implementations/reference.py b/fsspec/implementations/reference.py
index 5c5372c..60fb5d5 100644
--- a/fsspec/implementations/reference.py
+++ b/fsspec/implementations/reference.py
@@ -20,6 +20,7 @@ except ImportError:
 from fsspec.asyn import AsyncFileSystem
 from fsspec.callbacks import DEFAULT_CALLBACK
 from fsspec.core import filesystem, open, split_protocol
+from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
 from fsspec.utils import isfilelike, merge_offset_ranges, other_paths
 
 logger = logging.getLogger("fsspec.reference")
@@ -757,6 +758,10 @@ class ReferenceFileSystem(AsyncFileSystem):
             self.fss[remote_protocol] = fs
 
         self.fss[None] = fs or filesystem("file")  # default one
+        # Wrap any non-async filesystems to ensure async methods are available below
+        for k, f in self.fss.items():
+            if not f.async_impl:
+                self.fss[k] = AsyncFileSystemWrapper(f)
 
     def _cat_common(self, path, start=None, end=None):
         path = self._strip_protocol(path)


Test patch:
diff --git a/fsspec/implementations/tests/test_reference.py b/fsspec/implementations/tests/test_reference.py
index 0063d11..dd9ff68 100644
--- a/fsspec/implementations/tests/test_reference.py
+++ b/fsspec/implementations/tests/test_reference.py
@@ -458,7 +458,8 @@ def test_fss_has_defaults(m):
     assert fs.fss["memory"].protocol == "memory"
 
     fs = fsspec.filesystem("reference", fs=m, fo={})
-    assert fs.fss[None] is m
+    # Default behavior here wraps synchronous filesystems to enable the async API
+    assert fs.fss[None].sync_fs is m
 
     fs = fsspec.filesystem("reference", fs={"memory": m}, fo={})
     assert fs.fss["memory"] is m