Collecting results

Wait for one or several pipelines, with or without output

Nothing really runs until you wait: these methods block until the pipeline ends, then release its resources. .wait_for_result() returns a flat list. .wait_for_results([others]) waits for this pipeline and those in other_pipes (typically the branches of a split) and returns one list per pipeline, in the given order. .wait_for_completion() waits without collecting anything, for side-effect steps (disk writes, databases).

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)

# One list per pipeline, in the order given
squares, negatives = evens.task(lambda x: x * x).wait_for_results(
    [odds.task(lambda x: -x)]
)
print(squares, negatives)

# Side effects only: nothing is collected
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_")))

Output

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

How it works

While waiting, Olympipe watches every worker: an exception or a dead process raises PipelineExecutionError in the caller (see the Runtime & robustness page). The debug_graph parameter is covered on the Debugging page.

Previous Limit & time budget Next Debugging