Parallel tasks
Scale workers with a single parameter
.task(fn, count=N) runs fn in N separate processes sharing the same input queue. Each worker runs in its own process, bypassing the Python GIL — ideal for CPU-bound work and I/O calls.
Signature
Pipeline.task(task, count=1, on_error=None) -> Pipeline[OutPacket]from olympipe import Pipeline
import time
def slow_compute(x: int) -> int:
time.sleep(0.01) # simulate heavy computation
return x ** 2
# 1 sequential worker
results = Pipeline(range(100)).task(slow_compute).wait_for_result()
# 8 parallel workers — ~8x faster
results = (
Pipeline(range(100))
.task(slow_compute, count=8)
.wait_for_result()
)⚡ Performance
100 items, CPU work (0.01s/item)
How it works
Olympipe spawns N child processes via multiprocessing. Each item from the source is picked up by the first available worker — no manual coordination, no Lock, no Pool.
on_error: handle errors item by item
Signature: .task(task, count=1, on_error=None). If fn raises, on_error(item, exc) is called in the worker: returning None drops the item, any other value replaces it. Without on_error, the error fails the pipeline (PipelineExecutionError).
from olympipe import Pipeline
def parse(s: str) -> int:
return int(s)
def fallback(item: str, exc: Exception):
print(f"skip {item!r}: {exc}")
return None # None = drop the item, any other value replaces it
print(Pipeline(["1", "2", "oops", "4"]).task(parse, on_error=fallback).wait_for_result())
print(Pipeline(["1", "x"]).task(parse, on_error=lambda item, exc: -1).wait_for_result())skip 'oops': invalid literal for int() with base 10: 'oops'
[1, 2, 4]
[1, -1]Related examples
Bulk image resizing
Resize 10 000 images in parallel using Pillow.
Parallel web scraping
Download and parse 1 000 HTML pages with httpx.
CSV file processing
Read and transform thousands of CSV files in parallel.