Poor performance of async IO in a multithreaded env

Thanks for the great work!
I experience poor performance of IO bound tasks in a multithreaded environment.

Dask based example:

import fsspec

import dask.array as da
import xarray as xr
from dask.distributed import Client

def test(path):
    fs_map = fsspec.get_mapper(path)

    ar_dsk = da.random.random(size=(200_000, 200_000), chunks=(1_000, 1_000))
    ar_xr = xr.Dataset(data_vars=(dict(foo=(("x", "y"), ar_dsk))))
    ar_xr.to_zarr(fs_map, mode="w")

if __name__ == "__main__":
    # this saturates just one core ("IO loop")
    c = Client(processes=False)
    test("gs://foo/bar)

afaiu the current asyc design, some methods are coroutines. Let's take cat for example, _cat is the underlying coroutine, there might be multiple paths to cat, and those will happen concurrently. There is a single thread/loop per AsyncFileSystem (let's call it IO loop loop thread + default executor), that runs all coroutines per an instance of AsyncFileSystem (since GCS instances are cached that's usually more than a single instance). If a regular thread calls fs.cat, that runs maybe_sync, which effectively hands over the _cat to the IO loop and the thread waits on the results in the sync method. This works okish in a single threaded environment, not so much in a multithreaded environment like Dask, where it could oversaturate the "IO loop", effectively single threading the IO (with slight improvement given the async logic).

The issue seems to be that in the code above dask is using threadpool for tasks (in my case 16 threads), each task is mostly IO bound, so most of the work is being actually done by a single fsspec's "IO loop", whilst most of the worker threads just wait. The CPU saturation is very poor.

To validate this theory, if I use a process pool for dask cluster (each process with 1 worker thread) (via say: Client(processes=True, threads_per_worker=1)), the CPU saturation is much better since each worker process/thread has its own "IO loop" (separate processes and separate instances of FS).

This example issue should be reproducible with multprocessing as well.

Thinking about solutions: loop thread could hand over the IO work to executor (via run_in_executor), or worker threads should handle their own loop (if possible). Please let know what you think.

See an example of partial stacktrace dump:

• Thread-1 is the loop thread (performing IO)
• ThreadPoolExecutor-0_0 afaiu is the default executor (doing nothing)
• Dask-Worker-Threads'-6291-0 is one of many Dask threads (waiting on results from Thread-1)

Thread 6672 (active): "Thread-1"
    loads (json/__init__.py:343)
    _call (gcsfs/core.py:503)
    _cat_file (gcsfs/core.py:826)
    _run (asyncio/events.py:81)
    _run_once (asyncio/base_events.py:1859)
    run_forever (asyncio/base_events.py:570)
    run (threading.py:870)
    _bootstrap_inner (threading.py:932)
    _bootstrap (threading.py:890)
Thread 6343 (idle): "ThreadPoolExecutor-0_0"
    _worker (concurrent/futures/thread.py:78)
    run (threading.py:870)
    _bootstrap_inner (threading.py:932)
    _bootstrap (threading.py:890)
Thread 6350 (idle): "Dask-Worker-Threads'-6291-0"
    wait (threading.py:306)
    wait (threading.py:558)
    sync (fsspec/asyn.py:68)
    maybe_sync (fsspec/asyn.py:100)
    cat (fsspec/asyn.py:226)
    getitems (fsspec/mapping.py:89)
    _chunk_getitems (zarr/core.py:1666)
    _get_selection (zarr/core.py:1033)
    _get_basic_selection_nd (zarr/core.py:739)
    get_basic_selection (zarr/core.py:696)
    __getitem__ (zarr/core.py:571)
    __getitem__ (xarray/backends/zarr.py:56)
    __array__ (xarray/core/indexing.py:560)
    asarray (numpy/core/_asarray.py:83)
    __array__ (xarray/core/indexing.py:495)
    asarray (numpy/core/_asarray.py:83)
    getter (dask/array/core.py:102)
    apply_function (distributed/worker.py:3411)
    run (distributed/_concurrent_futures_thread.py:65)
    _worker (distributed/threadpoolexecutor.py:55)
    run (threading.py:870)
    _bootstrap_inner (threading.py:932)
    _bootstrap (threading.py:890)

Start with the async entry points in fsspec/asyn.py, especially sync, maybe_sync, and cat, then reproduce the Dask example with threaded and process-based clients. Trace how work reaches the IO loop and compare concurrency and CPU saturation; the issue is done when an agreed design removes the reported multithreaded bottleneck and the example demonstrates the improvement.