Récupérer les résultats

Attendre un ou plusieurs pipelines, avec ou sans sortie

Rien ne s'exécute vraiment tant qu'on n'attend pas : ces méthodes bloquent jusqu'à la fin du pipeline puis libèrent ses ressources. .wait_for_result() renvoie une liste plate. .wait_for_results([autres]) attend ce pipeline et ceux de other_pipes (typiquement les branches d'un split) et renvoie une liste par pipeline, dans l'ordre donné. .wait_for_completion() attend sans rien collecter, pour les étapes à effet de bord (écriture disque, base de données).

Signature

Pipeline.wait_for_result(debug_graph=None) -> list[InPacket]
Pipeline.wait_for_results(other_pipes=None, debug_graph=None) -> list[list[InPacket]]
Pipeline.wait_for_completion(other_pipes=None, debug_graph=None) -> None

python

import os
from typing import Optional, Tuple
from olympipe import Pipeline

def parity(x: int) -> Tuple[Optional[int], Optional[int]]:
    return (x, None) if x % 2 == 0 else (None, x)

evens, odds = Pipeline(range(10)).split(parity)

# Une liste par pipeline, dans l'ordre donné
squares, negatives = evens.task(lambda x: x * x).wait_for_results(
    [odds.task(lambda x: -x)]
)
print(squares, negatives)

# Effets de bord uniquement : rien n'est collecté
def save(x: int) -> None:
    with open(f"/tmp/olympipe_{x}.txt", "w") as f:
        f.write(str(x))

Pipeline(range(3)).task(save).wait_for_completion()
print(sorted(f for f in os.listdir("/tmp") if f.startswith("olympipe_")))

Sortie

[0, 4, 16, 36, 64] [-1, -3, -5, -7, -9]
['olympipe_0.txt', 'olympipe_1.txt', 'olympipe_2.txt']

Comment ça fonctionne

Pendant l'attente, Olympipe surveille tous les workers : une exception ou la mort d'un processus lève PipelineExecutionError dans l'appelant (voir la page Exécution & robustesse). Le paramètre debug_graph est décrit sur la page Débogage.

Précédent Limite & budget de temps Suivant Débogage