issues

What happened?

When writing a dataset to zarr, we sometimes see that the last few tasks hang indefinitely with no cpu activity or data transfer activity (as seen from the daskui). Inspecting the daskui always shows we are waiting on a task with a name like ('store-map-34659153bd4dc964b4e5f380dacebdbe', 0, 1).

What did you expect to happen?

For all tasks to finish and all taking approximately the same amount of time to complete. I also would expect worker saturation to have some effect and kick in a queue with xarray and map_blocks to zarr, but I don't see this behavior. I do see this behavior with the dask counterpart (example in code the extra section below).

Minimal Complete Verifiable Example

```Python import os import xarray as xr import dask.array as da from distributed import Client

def main(): client = Client("...")

# task 1: make dataset
variables = [str(i) for i in range(36)]
dset = xr.Dataset(
    {
        v: (
            ("x", "y"),
            da.ones((50000, 50000), chunks=(4096, 4096), dtype="float32"),
        )
        for v in variables
    },
    coords={"x": range(50000), "y": range(50000)},
)

bucket = "..."
path = f"gs://{bucket}/tmp/test/0"
dset.to_zarr(os.path.join(path, "1"))

# task 2: "preprocess" the dataset with map_blocks
rt = xr.open_zarr(os.path.join(path, "1"))

def f(block):
    # I have found copying helps reproduce the issue
    # and is often happening anyways within functions we apply
    # with map blocks e.g. .rio.clip, .fillna, etc.
    block = block.copy().copy()
    return block

# first setup a target store
target_path = os.path.join(path, "2")
rt.to_zarr(target_path, compute=False)

# I could not reproduce without writing a region
region = {"x": slice(0, 20000), "y": slice(0, 10000)}
subset = rt.isel(**region)
subset.map_blocks(f, template=subset).to_zarr(target_path, region=region)

if name == "main": main()

```

MVCE confirmation

• [X] Minimal example — the example is as focused as reasonably possible to demonstrate the underlying issue in xarray.
• [X] Complete example — the example is self-contained, including all data and the text of any traceback.
• [X] Verifiable example — the example copy & pastes into an IPython prompt or Binder notebook, returning the result.
• [X] New issue — a search of GitHub Issues suggests this is not a duplicate.

Relevant log output

No response

Anything else we need to know?

My dask cluster is deployed with helm with (32) workers each with 1cpu and 4gb.

We often mitigate this issue (poorly) using environment variables to trigger worker restarts so long hanging tasks get rescheduled so the program can complete, but at the cost of the expected time to restart, which can slow down performance. "DASK_DISTRIBUTED__WORKER__LIFETIME__DURATION": "8 minutes", "DASK_DISTRIBUTED__WORKER__LIFETIME__STAGGER": "4 minutes", "DASK_DISTRIBUTED__WORKER__LIFETIME__RESTART": "True",

In the mre above, we write to_zarr with region often because we often have to throttle the number of tasks submitted to dask.

Possibly relevant links: - https://github.com/dask/distributed/issues/391 - https://github.com/fsspec/gcsfs/issues/379 - https://github.com/pydata/xarray/issues/4406

note: I have tried to minimize the example even more with array = da.random.random((250000, 250000), chunks=(4096, 4096)).astype("float32") def f(block): block = block.copy().copy() return block array.map_blocks(f).to_zarr("gs://...") but I am not seeing the issue. I am seeing worker saturation taking some effect, which might be throttling file operations to gcs causing me to wonder why map_blocks().to_zarr() does not see the worker saturation effects and if that would help alleviate issues gcsfs is probably seeing.

Environment