Hacktoberfest 2026: los issues que los mantenedores marcaron para octubre, abiertos y aptos para principiantes. Explorar issues de Hacktoberfest

dask unstable; multiple race condition errors: P2PConsistencyError, RuntimeError, FutureCancelledError ...

Abierto
  • ¿Algo no está bien?
#9,163 5 comentarios 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
5/5
Tiempo estimado
Más de una semana
Aptitud para principiantes
20/100
Tipo de issue
Error
Claridad
Necesita aclaración
Estado de actividad
Estancado
Stack tecnológico
python
Área
distributed-systems

Línea de trabajo

Comienza convirtiendo el flujo de trabajo reportado de xarray.Dataset.to_zarr() en un reproductor autónomo y completa los detalles que faltan sobre Python, el sistema operativo, la instalación y el entorno. Aísla la ruta de Dask shuffle/rechunk y determina cuáles de los errores reportados P2PConsistencyError, RuntimeError, FutureCancelledError, de tiempo de espera de conexión y de deserialización se pueden reproducir; el trabajo estará terminado cuando haya un fallo acotado con una prueba de regresión adecuada o una causa raíz claramente identificada.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

bug needs info needs reproducer

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 loop

I'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 here

Anything else we need to know?:

Environment:

  • Dask version: 2025.11.0
  • Python version:
  • Operating System:
  • Install method (conda, pip, source):
Lenguaje dominante
Python
Estrellas
1.7k
Forks
778
Sin PR fusionados en 30 d

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de dask/distributed

Todos los issues de dask/distributed

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.