Intermittent deadlock when reading file with gcsfs

Not sure whether to post here or under gcsfs but seemed potentially related to #565: when reading a bunch of Parquet files with Dask (distributed scheduler), we occasionally (say one in ten thousand, or maybe one hundred thousand?) get a worker that deadlocks at the following:

...
File "/usr/local/lib/python3.10/site-packages/pandas/io/parquet.py", line 493, in read_parquet return impl.read(
File "/usr/local/lib/python3.10/site-packages/pandas/io/parquet.py", line 240, in read result = self.api.parquet.read_table(
File "/usr/local/lib/python3.10/site-packages/pyarrow/parquet/__init__.py", line 2780, in read_table dataset = _ParquetDatasetV2(
File "/usr/local/lib/python3.10/site-packages/pyarrow/parquet/__init__.py", line 2368, in __init__ [fragment], schema=schema or fragment.physical_schema,
File "/usr/local/lib/python3.10/site-packages/pyarrow/fs.py", line 420, in open_input_file return PythonFile(self.fs.open(path, mode="rb"), mode="r")
File "/usr/local/lib/python3.10/site-packages/fsspec/spec.py", line 1034, in open f = self._open(
File "/usr/local/lib/python3.10/site-packages/gcsfs/core.py", line 1201, in _open return GCSFile(
File "/usr/local/lib/python3.10/site-packages/gcsfs/core.py", line 1321, in __init__ super().__init__(
File "/usr/local/lib/python3.10/site-packages/fsspec/spec.py", line 1382, in __init__ self.size = self.details["size"]
File "/usr/local/lib/python3.10/site-packages/fsspec/spec.py", line 1395, in details self._details = self.fs.info(self.path)
File "/usr/local/lib/python3.10/site-packages/fsspec/asyn.py", line 111, in wrapper return sync(self.loop, func, *args, **kwargs)
File "/usr/local/lib/python3.10/site-packages/fsspec/asyn.py", line 84, in sync if event.wait(1):
File "/usr/local/lib/python3.10/threading.py", line 607, in wait signaled = self._cond.wait(timeout)
File "/usr/local/lib/python3.10/threading.py", line 324, in wait gotit = waiter.acquire(True, timeout)

The dask worker is healthy and responsive, so I don't believe it's an issue with dask per se (reproducing w/o dask is a bit impractical due to the scale of the job).

This currently happening with dask and distributed==2022.9.1, fsspec and gcsfs==2022.7.1. I was going to try upgrading to see if it helps but I didn't see anything in the changelogs that looked like it would be related (#565 was closed over a year ago so I'm sure we have that fix). We only see this bug every couple of weeks but I can keep an eye out next time if there's any more data that would be helpful to grab from a live example.

Start with the stack trace through fsspec/spec.py, fsspec/asyn.py, and gcsfs/core.py, then try to reproduce the deadlock using the reported Dask distributed Parquet workload. Done means identifying a reproducible cause and confirming a fix with a regression test or equivalent evidence.