Ensure reference FS wraps any sync filesystems

[moradology]

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]

The issue arises here, I believe: https://github.com/zarr-developers/zarr-python/blob/main/src/zarr/storage/remote.py#L219-L225 Here, the _cat_file method is explicitly called and it looks as though the default local filesystem is not playing nicely in that case

[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.

[moradology]

OK, interesting. Perhaps I'm missing something crucial about how that might be happening, because the failures in test_async.py don't seem like they'd implicate ReferenceFileSystem or the AsyncFileSystemWrapper

[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)