Repository navigation
dask unstable; multiple race condition errors: P2PConsistencyError, RuntimeError, FutureCancelledError ... #9163
Description
Hi,
I'm encountering repeated failures during xarray.Dataset.to_zarr() when using Dask Distributed.
The failures occur during the Dask shuffle/rechunk phase and appear to be caused by instability in the P2P shuffle as well as many other dask issues.
While processing a batch of NetCDF to zarr, I end up having to create a horrible code which has to catch all possible dask issues, so that, the cluster/client gets destroyed, recreated, and the failed batch of files reprocessed.
Even though recreating the cluster allows the batch to succeed on retry, it shouldn't be required.
Below are all the dask errors I have to catch in my code.
batch_is_processed = False
while not batch_is_processed:
try:
with self.lock:
ds.to_zarr(
self.store,
mode="w", # Overwrite mode for the first batch
write_empty_chunks=self.write_empty_chunks,
compute=True, # Compute the result immediately
consolidated=self.consolidated,
safe_chunks=self.safe_chunks,
align_chunks=self.align_chunks,
)
batch_is_processed = True
except (FutureCancelledError, P2PConsistencyError, RuntimeError) as e:
error_text = str(e)
SHUFFLE_KEYWORDS = [
"P2P",
"failed during transfer phase",
"failed during barrier phase",
"failed during shuffle phase",
"shuffle failure",
"No active shuffle",
"Unexpected error encountered during P2P",
]
CONNECTION_KEYWORDS = [
"Timed out trying to connect",
"CancelledError",
"Too many open files",
]
DESERIALISATION_KEYWORDS = [
"Error during deserialization",
"different environments",
]
# Determine if the error should trigger a retry (cluster reset)
retryable = any(
keyword in error_text
for keyword in (
SHUFFLE_KEYWORDS
+ CONNECTION_KEYWORDS
+ DESERIALISATION_KEYWORDS
)
)
# RuntimeError that is NOT retryable -> treat as normal exception
if isinstance(e, RuntimeError) and not retryable:
# Treat as a regular exception: fallback to individual processing
# code removed for simplification
else:
self._reset_cluster()
batch_is_processed = False
# ... the rest of the logic is in a while loopI'd like to highlight that my workers/schedulers are only using 50% of cpu/mem max. I've tried many different settings, and ended up having to create workers only with 1 thread if I don't want to run into more race conditions.
Errors Observed
P2PConsistencyError: No active shuffle with id='868df3e600f2968e0cd678a103d355d2' found
RuntimeError: P2P 43b3a653ad1c988df817f013d7314141 failed during transfer phase
OSError: Timed out trying to connect to tls://<worker-ip>:8786 after 30 s
asyncio.exceptions.CancelledError
...
Minimal Complete Verifiable Example:
# Put your MCVE code hereAnything else we need to know?:
Environment:
- Dask version: 2025.11.0
- Python version:
- Operating System:
- Install method (conda, pip, source):
Activity
jacobtomlinson commentedon Dec 10, 2025
Sorry you're having a tough time. We did just merge some zarr improvements today which might help with a few things, this should be released next week.
To look into this further we will need a reproducible example so that we can see these problems ourselves. It seems like you're seeing a variety of different problems on your system, so it might be best to break each one down into a separate issue with it's own reproducer.
cc @dcherian @d-v-b @melonora
melonora commentedon Dec 10, 2025
also might be good to know which zarr version you are using
lbesnard commentedon Dec 11, 2025
thanks @melonora
I've pinned a few packages to older versions as I experience a lot of regressions with packages such as xarray. Happy to try to update my zarr package and see
@jacobtomlinson , thanks, for background, I'm trying to convert all publicly available oceanographic data in Australia with this library I've been developing (https://github.com/aodn/aodn_cloud_optimised/) ! Using the full stack of libraries mentioned above has been challenging. I'd love to re-created a MVCE, but I don't even see where I can start when the behaviour is random. As mentioned, If I destroy a cluster when there is a bug, re-create it, and process the same batch of files, this batch works; and the cluster works fine until the next failure. It could be link to the NetCDF4 files I'm using which are moderately big (50-100mb) but I doubt.
lbesnard commentedon Dec 17, 2025
@jacobtomlinson
I did further testing with the latest version of zarr 3.1.5, and i now get errors I've never encountered before with
open_zarrReverting to using
zarr == 2.18.7removes this issue.Unfortunately, this is a common pattern I have with the dask/xarray/s3fs... ecosystem where I have to pin library versions and stay stuck because of common regressions.
melonora commentedon Dec 17, 2025
Could you please provide the code that causes this error? Just wondering what store is being used at the moment and also to see if I can try replicate it myself locally.