Olympipe

Zero-boilerplate parallel pipelines for Python.

📖 Documentation & examples: datatoy.fr/en/olympipe

Turn any iterator into a multi-process pipeline with a single line. Each .task() runs in its own process, bypassing the GIL — no multiprocessing.Pool, no queues, no plumbing.

from olympipe import Pipeline

results = (
    Pipeline(range(1000))
    .task(heavy_compute, count=8)   # 8 parallel workers
    .filter(lambda x: x > 0)
    .batch(32)
    .wait_for_result()
)

Installation

pip install olympipe

Python 3.9 or later.

Why Olympipe?

• Chainable API — compose steps like pandas or streams
• True multiprocessing — each step is a separate process, GIL-free
• PyTorch-safe — auto-detects torch and switches to spawn context to avoid CUDA/DataLoader deadlocks
• Fast with big payloads — anything over 8 KB travels through shared memory instead of the pipe, roughly twice the throughput and no way for a dying worker to leave a half-written item behind
• Fails loudly — an unhandled error in a worker raises in the caller with the worker's traceback, and a worker that dies fails the run instead of truncating it
• Type-safe — fully typed, pipeline errors caught at IDE time
• Batteries included — split/gather, time-windows, disk caching, HTTP server source

Basic usage

from olympipe import Pipeline

results = (
    Pipeline(range(20))
    .task(lambda x: x * 2)           # transform
    .filter(lambda x: x % 4 == 0)    # keep even multiples of 4
    .batch(3)                         # group into lists of 3
    .wait_for_result()
)
# [[0, 8, 16], [24, 32, 40]]

Parallel workers

Scale any step horizontally by passing count=N. Workers share the same input queue and output queue — no manual coordination needed.

import time
from olympipe import Pipeline

def slow_io(url: str) -> bytes:
    # simulate network call
    time.sleep(0.5)
    return b"data"

# 200 URLs processed with 20 concurrent workers
results = (
    Pipeline(my_urls)
    .task(slow_io, count=20)
    .wait_for_result()
)

Machine learning inference

Olympipe shines for ML workloads: saturate your GPU(s) with a pipeline that loads data in parallel, batches it, runs inference, and post-processes results — all without a single thread/process management line.

import torch
from torch import nn
from olympipe import Pipeline

class Classifier:
    """One instance per worker process — each gets its own model copy."""

    def __init__(self, device: str = "cuda"):
        self.device = device
        self.model = nn.Sequential(nn.Linear(512, 128), nn.ReLU(), nn.Linear(128, 10))
        self.model.load_state_dict(torch.load("weights.pt"))
        self.model.eval().to(device)

    def predict(self, batch: list) -> list:
        with torch.no_grad():
            x = torch.stack(batch).to(self.device)
            return self.model(x).argmax(dim=1).tolist()


def load_and_preprocess(path: str) -> torch.Tensor:
    # CPU-bound preprocessing, runs in parallel with GPU inference
    img = load_image(path)
    return preprocess(img)


predictions = (
    Pipeline(image_paths)
    .task(load_and_preprocess, count=4)   # 4 CPU workers loading/preprocessing
    .batch(64)                             # feed GPU in batches of 64
    .class_task(Classifier, Classifier.predict, ["cuda"])  # 1 GPU worker
    .explode(lambda x: x)                 # flatten back to individual predictions
    .wait_for_result()
)

Note: Olympipe detects torch at startup and automatically uses spawn context instead of fork, preventing CUDA deadlocks. Set OLYMPIPE_FORCE_FORK=1 to override.

Stateful workers

Use .class_task() when each worker needs persistent state (model weights, DB connection, accumulator…). The class is instantiated once per worker.

from olympipe import Pipeline

class RunningStats:
    def __init__(self):
        self.n = 0
        self.total = 0.0

    def update(self, x: float) -> dict:
        self.n += 1
        self.total += x
        return {"n": self.n, "mean": self.total / self.n}

results = Pipeline(sensor_stream).class_task(RunningStats, RunningStats.update).wait_for_result()

Split & gather

Branch your pipeline into independent streams and merge them back.

from typing import Optional, Tuple
from olympipe import Pipeline

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

evens, odds = Pipeline(range(20)).split(route, n=2)

results = (
    evens.task(lambda x: x * 10)          # multiply evens
         .gather(odds.task(lambda x: -x)) # negate odds, then merge
         .wait_for_result()
)

HTTP server pipeline

Turn an HTTP endpoint into a pipeline source. Each incoming request becomes a packet; downstream tasks process and respond.

import socket
from olympipe import Pipeline
from olympipe.helpers.server import send_json_response

def handle_request(pair):
    conn: socket.socket
    data: dict
    conn, data = pair

    result = heavy_computation(data["payload"])
    send_json_response(conn, {"result": result})
    return result

Pipeline.server(
    [("POST", "/compute", lambda body: body)],
    port=8000,
    inactivity_timeout=60.0,
).task(handle_request, count=4).wait_for_completion()

LLM inference server (HuggingFace)

Serve a HuggingFace model as an HTTP API. The model loads once per worker process; requests are batched and processed in parallel while the server stays responsive.

import socket
from olympipe import Pipeline
from olympipe.helpers.server import send_json_response


class LLMWorker:
    """Loaded once per worker — keeps the model in GPU memory across requests."""

    def __init__(self, model_name: str = "mistralai/Mistral-7B-Instruct-v0.3"):
        from transformers import pipeline as hf_pipeline

        self.pipe = hf_pipeline(
            "text-generation",
            model=model_name,
            device_map="auto",      # spreads across available GPUs
            torch_dtype="auto",
        )

    def generate(self, pair: tuple) -> tuple:
        conn: socket.socket
        data: dict
        conn, data = pair

        prompt = data.get("prompt", "")
        max_new_tokens = data.get("max_new_tokens", 256)

        output = self.pipe(prompt, max_new_tokens=max_new_tokens, do_sample=False)
        generated = output[0]["generated_text"][len(prompt):]

        send_json_response(conn, {"response": generated, "model": self.pipe.model.name_or_path})
        return {"prompt": prompt, "response": generated}


Pipeline.server(
    [("POST", "/generate", lambda body: body)],
    port=8000,
).class_task(LLMWorker, LLMWorker.generate, ["mistralai/Mistral-7B-Instruct-v0.3"]).wait_for_completion()

Call it:

curl -X POST http://localhost:8000/generate \
     -H "Content-Type: application/json" \
     -d '{"prompt": "Explain transformers in one sentence:", "max_new_tokens": 128}'

Tip: Scale to multiple GPUs by passing count=N to .class_task() — each worker gets its own model replica on a separate device.

Step caching

Cache intermediate results to disk. Reruns skip already-computed steps automatically — great for iterating on the end of a slow pipeline.

import tempfile
from olympipe import Pipeline

def expensive_step_1(x: int) -> int:
    time.sleep(1)  # simulate heavy computation
    return x ** 2

def expensive_step_2(x: int) -> int:
    time.sleep(1)
    return x + 1

with tempfile.TemporaryDirectory() as cache_dir:
    # First run: computes everything, writes .pkl files
    results = (
        Pipeline(range(100))
        .cached_task(expensive_step_1, cache_dir=cache_dir)
        .cached_task(expensive_step_2, cache_dir=cache_dir)
        .uncache()
        .wait_for_result()
    )

    # Second run: instant — reads from disk, skips all computation
    results_again = (
        Pipeline(range(100))
        .cached_task(expensive_step_1, cache_dir=cache_dir)
        .cached_task(expensive_step_2, cache_dir=cache_dir)
        .uncache()
        .wait_for_result()
    )

Temporal batching (streams)

Group items arriving within a time window — ideal for audio frames, sensor feeds, or any real-time stream.

import time
from olympipe import Pipeline

def slow_producer(x: int) -> int:
    time.sleep(0.05)
    return x

results = (
    Pipeline(range(100))
    .task(slow_producer)
    .temporal_batch(0.5)     # collect items for 500ms, then emit as a list
    .task(lambda batch: sum(batch))
    .wait_for_result()
)

API reference

fork or spawn

fork is 35-60% faster than spawn here, so olympipe uses it wherever it is safe, and keeps its own side of the bargain: the whole graph is built before any data starts flowing, so every worker is forked from a single-threaded parent.

It switches to spawn when forking would not be safe: torch is installed (a CUDA context cannot survive a fork, and a forked DataLoader worker deadlocks), fork is unavailable, or your process already has threads — their locks would be inherited by the worker in whatever state they were in. Threads started after import olympipe cannot be detected in time, so a worker forked next to one warns instead; set OLYMPIPE_FORCE_SPAWN=1 in that case.

The torch case is checked against a real device rather than assumed: a tiny model runs on the GPU inside spawned workers, and the same run with fork forced dies with the driver's own Cannot re-initialize CUDA in forked subprocess. Those two tests need a GPU and are deselected everywhere else, so ask for them by name:

uv run pytest -m gpu   # needs a CUDA device and a CUDA build of torch

Environment variables

The start method is chosen when olympipe is imported. OLYMPIPE_FORCE_FORK / OLYMPIPE_FORCE_SPAWN have to be set before that import: worker processes are instances of the context's Process class, which is fixed at import time.

Monitoring and early feeding

wait_for_result() / wait_for_completion() start the source when the graph is already built. A monitor that attaches consumers after the source exists should call start() itself — it is idempotent and, on any pipeline of the graph, starts the root and every root merged in by gather():

from olympipe import Pipeline

source = Pipeline(iter(sensor()))
source.start()                  # feeder runs; blocks when the queue is full
source.task(decode).start()     # same feeder, second call is a no-op

Pipeline.server() starts feeding on its own, because it has to be listening before the rest of the graph exists.

To inspect queues without setting OLYMPIPE_MONITORING or patching MonitorableQueue, call enable_monitoring() before building the pipeline:

import olympipe
from olympipe import Pipeline

olympipe.enable_monitoring(inspect_messages=True)

pipe = Pipeline([1, 2, 3]).task(lambda x: x + 1)
# pipe.queues maps pid -> MonitorableQueue
# peek_at / peek_all see payloads, including ones a spawned worker put()

inspect_messages=True keeps a mirror of the payloads currently in the queue. The choice is stored on each queue and pickled into workers, so a spawn worker — which re-imports olympipe and would otherwise reset the class defaults — still fills the mirror. inspect_messages=False (the default) only registers queues and keeps the fast counters.

The same switches are constructor arguments, inherited by every later stage: Pipeline(data, monitoring=True, inspect_messages=True).

MonitorableQueue.fast_mode = False and skip_mirror_payload = False still work when set before the queue is created. OLYMPIPE_MONITORING=1 still registers queues on its own.