dask: Dask cluster hangs while processing file using s3fs

We’re making some progress on this in https://github.com/h5py/h5py/issues/2019, https://github.com/h5netcdf/h5netcdf/pull/117, and linked issues.

There might be some fixes possible at the fsspec level, but for now it seems like updating libraries to avoid circular references if possible is the best path forward. Either way, I don’t think there’s anything Dask can do, so I’m closing this.

@sharkinsspatial below are the steps I used to reproduce on a single machine. I’m running in AWS us-east-1, the same region as the S3 dataset. It’s not consistent, for me the worker process deadlocks about 50% of the time.

First start a scheduler and single worker with s3fs debugging enabled:

export S3FS_LOGGING_LEVEL=DEBUG
dask-scheduler &   
dask-worker --nthreads 1 localhost:8786

Then run the following python script in a different terminal:

import xarray as xr
import s3fs
import dask
import time
import json
from dask.distributed import Client

@dask.delayed
def s3open(path):
    fs = s3fs.S3FileSystem(anon=True, default_fill_cache=False)
    return fs.open(path)

client = Client('localhost:8786')

# Opening a single file is usually enough
files_mapper = [s3open('s3://era5-pds/2020/12/data/sea_surface_temperature.nc')]
ds = xr.open_mfdataset(files_mapper, engine='h5netcdf',
                       chunks={'lon':200,'lat':200,'time0':720},
                       concat_dim='time0',
                       combine='nested',
                       coords='minimal',
                       compat='override',
                       parallel=True)

# A short pause here makes the deadlock more likely for some reason
time.sleep(1)
# Read dataset into worker memory
ds = client.persist(ds)

# Output the stats and worker call stack every 5 seconds to monitor progress
while(1):
    print(json.dumps(client.scheduler_info(),indent=2))
    print(json.dumps(client.call_stack(),indent=2))
    time.sleep(5)

Let it run for a little while, if the worker stops displaying s3fs debug logs while there are still queued tasks you’ve likely got it deadlocked. You may need to run it a few times before it happens.

There is a similar example here: https://github.com/lsterzinger/pangeo-cloud-L2-satellite/tree/main/dask_troubleshooting

I’ve tracked this down, at least in my case, to global locking performed in the h5py library which is used by the h5netcdf engine in xarray.

I managed to reproduce the issue semi-reliably on a single worker and hooked up gdb to the hung fsspecIO thread. The stack trace shows the python garbage collector attempting to acquire a lock during clean-up of a h5py object.

This is the sequence of events:

1. Dask worker thread acquires the global h5py lock through h5py Dataset.__getitem__ as you can see in the call stack from my earlier comment (although sometimes it occurs via a different method, they all acquire the lock)
2. The request is passed to fsspec and s3fs as a read
3. An S3 range request is submitted as a coroutine to the fsspecIO thread’s event loop in fsspec.sync.
4. Sometimes python garbage collection occurs during the fsspecIO thread’s execution
5. If the garbage collector running in the fsspecIO thread decides to clean up a h5py object, it calls ObjectID.__dealloc__ and attempts to acquire the global h5py lock
6. The two threads are now deadlocked and the worker hangs indefinitely

I’m not sure what the best fix might be, but a few ideas that came to mind:

• A single-threaded mode for fsspec, that could be enabled through config when h5py is in use
• More of a work-around, disabling garbage collection during the fsspec.sync() execution, or perhaps an explicit call to gc.collect() before handing off to the co-routine might work (I haven’t tested though)
• Set timeouts on all fsspec.sync() calls
• More granular object-level locking in the h5py library (likely infeasible)

Is there anything else that could be done within dask or fsspec to avoid the deadlock?

Here’s the stack trace: fsspecIO-stacktrace.txt