Datatoy Logo

Runtime & robustness (2.x)

Errors, processes, shared memory and environment variables

Since 2.x, a pipeline fails loudly: an unhandled exception in a worker raises PipelineExecutionError in the caller (the wait_*), with the worker's traceback in e.errors (the worker also prints it as it fails). A worker that dies without an exception (segfault, OOM killer, os._exit) is detected and also fails the run, instead of silently truncating it. To tolerate errors item by item, use .task(fn, on_error=...) (see Parallel tasks).

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)          # simulates a segfault / OOM kill
    return x

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

How it works

While waiting, errors reported by workers are read continuously and process exit codes are checked roughly every 50 ms.

Result order

With count>1, several workers consume the same queue: results come out in the order they finish, not in input order. With a single worker per step (and no split/gather), queues are FIFO and order is kept. If order matters, carry an index with the item and sort at the end.

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: order kept
Output
[0, 2, 3, 1, 4, 5, 8, 7, 6, 9]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

fork or spawn, and the __main__ guard

Olympipe picks the worker start method, in priority order: OLYMPIPE_FORCE_FORK=1 → fork; OLYMPIPE_FORCE_SPAWN=1 → spawn; torch installed → spawn (a CUDA context cannot survive a fork); fork unavailable (Windows, macOS default) → spawn; your process already has threads when olympipe is imported → spawn; otherwise fork, which is faster. olympipe.get_mp_context() tells which method was chosen. Under spawn, Python re-executes your script in every worker: Olympipe neutralises the pipeline during that replay (no recursion), but the rest of the module-level code (prints, file writes…) runs again. With an if __name__ == "__main__": guard, that code only runs in the main process.

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())
Output
$ OLYMPIPE_FORCE_SPAWN=1 python guarded.py
loading script
[0, 2, 4, 6, 8]

# same script without the guard: module-level code re-runs in each spawned worker
$ OLYMPIPE_FORCE_SPAWN=1 python unguarded.py
loading script
loading script
[]
loading script
[]
[0, 2, 4, 6, 8]

Shared memory, progress bar and environment variables

Items larger than 8 KB once serialized travel through shared memory instead of the inter-process pipe (small items stay on the pipe). The source shows a tqdm progress bar on stderr; TQDM_DISABLE=1 hides it. Recognised variables: OLYMPIPE_FORCE_SPAWN=1 (always spawn), OLYMPIPE_FORCE_FORK=1 (always fork, even where it is auto-disabled), OLYMPIPE_DISABLE_SHM=1 (large items through the pipe instead of shared memory), OLYMPIPE_MONITORING=1 (every queue is registered and inspectable via pipeline.queues).

Previous Decorators