Datatoy Logo

Exécution & robustesse (2.x)

Erreurs, processus, mémoire partagée et variables d'environnement

Depuis la 2.x, un pipeline échoue bruyamment : une exception non gérée dans un worker lève PipelineExecutionError dans l'appelant (le wait_*), avec le traceback du worker dans e.errors (le worker l'imprime aussi au passage). Un worker qui meurt sans exception (segfault, OOM killer, os._exit) est détecté et fait aussi échouer le run, au lieu de le tronquer silencieusement. Pour tolérer les erreurs item par item, utilisez .task(fn, on_error=...) (voir Tâches parallèles).

Signature

class PipelineExecutionError(RuntimeError):
    errors: list[str]   # one message (with traceback) per failure
python 
import os
from olympipe import Pipeline, PipelineExecutionError

def parse(s: str) -> int:
    return int(s)

try:
    Pipeline(["1", "2", "oops"]).task(parse).wait_for_result()
except PipelineExecutionError as e:
    print(type(e).__name__, "-", len(e.errors), "error(s)")
    print(e.errors[0].splitlines()[0])

def crash(x: int) -> int:
    if x == 3:
        os._exit(1)          # simule un segfault / un kill OOM
    return x

try:
    Pipeline(range(10)).task(crash).wait_for_result()
except PipelineExecutionError as e:
    print(e.errors[0])
Sortie
TaskPipe[parse]: ValueError: invalid literal for int() with base 10: 'oops'
Traceback (most recent call last):        # imprimé par le worker
  ...
PipelineExecutionError - 1 error(s)
TaskPipe[parse]: ValueError: invalid literal for int() with base 10: 'oops'
TaskPipe[crash] exited unexpectedly with code 1

Comment ça fonctionne

Pendant l'attente, les erreurs signalées par les workers sont lues en continu et les codes de sortie des processus sont vérifiés toutes les 50 ms environ.

Ordre des résultats

Avec count>1, plusieurs workers consomment la même file : les résultats sortent dans l'ordre où ils finissent, pas dans l'ordre d'entrée. Avec un seul worker par étape (et sans split/gather), les files sont FIFO et l'ordre est conservé. Si l'ordre compte, transportez un index avec l'item et triez à la fin.

python
import random
import time
from olympipe import Pipeline

def jitter(x: int) -> int:
    time.sleep(random.random() / 20)
    return x

print(Pipeline(range(10)).task(jitter, count=4).wait_for_result())
print(Pipeline(range(10)).task(jitter).wait_for_result())   # count=1 : ordre conservé
Sortie
[0, 2, 3, 1, 4, 5, 8, 7, 6, 9]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

fork ou spawn, et la garde __main__

Olympipe choisit la méthode de démarrage des workers, par ordre de priorité : OLYMPIPE_FORCE_FORK=1 → fork ; OLYMPIPE_FORCE_SPAWN=1 → spawn ; torch installé → spawn (un contexte CUDA ne survit pas à un fork) ; fork indisponible (Windows, macOS par défaut) → spawn ; votre processus a déjà des threads au moment de import olympipe → spawn ; sinon fork, plus rapide. olympipe.get_mp_context() indique la méthode retenue. En spawn, Python réexécute votre script dans chaque worker : Olympipe neutralise le pipeline pendant cette relance (pas de récursion), mais le reste du code de niveau module (print, écriture de fichiers…) est réexécuté. Avec la garde if __name__ == "__main__":, ce code ne s'exécute que dans le processus principal.

python
from olympipe import Pipeline

def double(x: int) -> int:
    return 2 * x

if __name__ == "__main__":
    print("loading script")
    print(Pipeline(range(5)).task(double).wait_for_result())
Sortie
$ OLYMPIPE_FORCE_SPAWN=1 python guarded.py
loading script
[0, 2, 4, 6, 8]

# même script sans la garde : le code de niveau module est réexécuté dans chaque worker
$ OLYMPIPE_FORCE_SPAWN=1 python unguarded.py
loading script
loading script
[]
loading script
[]
[0, 2, 4, 6, 8]

Mémoire partagée, barre de progression et variables d'environnement

Les items de plus de 8 Ko une fois sérialisés passent par la mémoire partagée au lieu du pipe entre processus (les petits items restent sur le pipe). La source affiche une barre de progression tqdm sur stderr ; TQDM_DISABLE=1 la masque. Variables reconnues : OLYMPIPE_FORCE_SPAWN=1 (toujours spawn), OLYMPIPE_FORCE_FORK=1 (toujours fork, même là où il est désactivé automatiquement), OLYMPIPE_DISABLE_SHM=1 (gros items via le pipe plutôt que la mémoire partagée), OLYMPIPE_MONITORING=1 (chaque file est enregistrée et inspectable via pipeline.queues).

Précédent Décorateurs