Fails with uvloop

It looks like fsspec fails when used with uvloop, as is sometimes used with higher performing Dask workloads. Here is a non-minimal-but-reproducible error with Dask.

from dask.distributed import Client
import dask.dataframe as dd

client = Client()
df = dd.read_parquet("s3://sra-pub-sars-cov2-metadata-us-east-1/v2/tax_analysis/", storage_options={"anon": True})
df.compute()

---------------------------------------------------------------------
AttributeError                      Traceback (most recent call last)
<ipython-input-1-841b4d3cd8fb> in <module>
      3 
      4 client = Client()
----> 5 df = dd.read_parquet("s3://sra-pub-sars-cov2-metadata-us-east-1/v2/tax_analysis/", storage_options={"anon": True})
      6 df.compute()

~/workspace/dask/dask/dataframe/io/parquet/core.py in read_parquet(path, columns, filters, categories, index, storage_options, engine, gather_statistics, split_row_groups, read_from_paths, chunksize, **kwargs)
    305         index = [index]
    306 
--> 307     read_metadata_result = engine.read_metadata(
    308         fs,
    309         paths,

~/workspace/dask/dask/dataframe/io/parquet/arrow.py in read_metadata(cls, fs, paths, categories, index, gather_statistics, filters, split_row_groups, read_from_paths, **kwargs)
    505             split_row_groups,
    506             gather_statistics,
--> 507         ) = cls._gather_metadata(
    508             paths,
    509             fs,

~/workspace/dask/dask/dataframe/io/parquet/arrow.py in _gather_metadata(cls, paths, fs, split_row_groups, gather_statistics, filters, index, dataset_kwargs)
   1676 
   1677         # Step 1: Create a ParquetDataset object
-> 1678         dataset, base, fns = _get_dataset_object(paths, fs, filters, dataset_kwargs)
   1679         if fns == [None]:
   1680             # This is a single file. No danger in gathering statistics

~/workspace/dask/dask/dataframe/io/parquet/arrow.py in _get_dataset_object(paths, fs, filters, dataset_kwargs)
   1636         #       existence of _metadata.  Listing may be much more
   1637         #       expensive in storage systems like S3.
-> 1638         allpaths = fs.glob(paths[0] + fs.sep + "*")
   1639         allpaths, base, fns = _sort_and_analyze_paths(allpaths, fs)
   1640         dataset = pq.ParquetDataset(paths[0], filesystem=fs, filters=filters, **kwargs)

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/spec.py in glob(self, path, **kwargs)
    523             depth = None if "**" in path else path[ind + 1 :].count("/") + 1
    524 
--> 525         allpaths = self.find(root, maxdepth=depth, withdirs=True, detail=True, **kwargs)
    526         # Escape characters special to python regex, leaving our supported
    527         # special characters in place.

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py in wrapper(*args, **kwargs)
    119     def wrapper(*args, **kwargs):
    120         self = obj or args[0]
--> 121         return maybe_sync(func, self, *args, **kwargs)
    122 
    123     return wrapper

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py in maybe_sync(func, self, *args, **kwargs)
     98         if inspect.iscoroutinefunction(func):
     99             # run the awaitable on the loop
--> 100             return sync(loop, func, *args, **kwargs)
    101         else:
    102             # just call the blocking function

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py in sync(loop, func, callback_timeout, *args, **kwargs)
     69     if error[0]:
     70         typ, exc, tb = error[0]
---> 71         raise exc.with_traceback(tb)
     72     else:
     73         return result[0]

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py in f()
     53             if callback_timeout is not None:
     54                 future = asyncio.wait_for(future, callback_timeout)
---> 55             result[0] = await future
     56         except Exception:
     57             error[0] = sys.exc_info()

~/miniconda/envs/play/lib/python3.8/site-packages/s3fs/core.py in _find(self, path, maxdepth, withdirs, detail)
    537             raise ValueError("Cannot traverse all of S3")
    538         if maxdepth:
--> 539             return super().find(
    540                 bucket + "/" + key, maxdepth=maxdepth, withdirs=withdirs, detail=detail
    541             )

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/spec.py in find(self, path, maxdepth, withdirs, **kwargs)
    432         out = dict()
    433         detail = kwargs.pop("detail", False)
--> 434         for _, dirs, files in self.walk(path, maxdepth, detail=True, **kwargs):
    435             if withdirs:
    436                 files.update(dirs)

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/spec.py in walk(self, path, maxdepth, **kwargs)
    381         detail = kwargs.pop("detail", False)
    382         try:
--> 383             listing = self.ls(path, detail=True, **kwargs)
    384         except (FileNotFoundError, IOError):
    385             return [], [], []

~/miniconda/envs/play/lib/python3.8/site-packages/s3fs/core.py in ls(self, path, detail, refresh, **kwargs)
    991         """
    992         path = self._strip_protocol(path).rstrip("/")
--> 993         files = maybe_sync(self._ls, self, path, refresh=refresh)
    994         if not files:
    995             files = maybe_sync(self._ls, self, self._parent(path), refresh=refresh)

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py in maybe_sync(func, self, *args, **kwargs)
     91         if inspect.iscoroutinefunction(func):
     92             # run coroutine while pausing this one (because we are within async)
---> 93             return _run_until_done(func(*args, **kwargs))
     94         else:
     95             # make awaitable which then calls the blocking function

~/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py in _run_until_done(coro)
     29     runner = loop.create_task(coro)
     30     while not runner.done():
---> 31         loop._run_once()
     32     asyncio.tasks._current_tasks[loop] = task
     33     return runner.result()

AttributeError: 'Loop' object has no attribute '_run_once'

In [2]: debug
> /home/mrocklin/miniconda/envs/play/lib/python3.8/site-packages/fsspec/asyn.py(31)_run_until_done()
     29     runner = loop.create_task(coro)
     30     while not runner.done():
---> 31         loop._run_once()
     32     asyncio.tasks._current_tasks[loop] = task
     33     return runner.result()

ipdb> pp loop
<uvloop.Loop running=True closed=False debug=False>
ipdb> pp loop

Start with fsspec/asyn.py, especially _run_until_done and maybe_sync, and reproduce the supplied Dask parquet example with uvloop. Done means the example no longer raises AttributeError: 'Loop' object has no attribute '_run_once'.