olympipe-rs

Parallel data pipeline library for Rust, inspired by olympipe.

Each stage runs in its own thread (or several threads for task(..., count)), communicating via bounded crossbeam channels (capacity 4 by default, matching olympipe). Back-pressure is automatic: a slow downstream stage will slow down upstream producers.

Use case: ML inference pipelines

A common bottleneck in machine learning workloads is keeping the GPU busy. olympipe-rs makes it straightforward to overlap CPU-bound preprocessing and postprocessing with GPU inference in separate threads:

use olympipe_rs::Pipeline;

Pipeline::new(image_paths)
    .task(|path| load_and_preprocess(path), 4)  // 4 CPU threads decode & normalise
    .task(|tensor| run_inference(tensor), 1)     // 1 GPU thread, always fed
    .task(|output| postprocess(output), 4)       // 4 CPU threads decode results
    .for_each(|result| save(result));

Because each stage runs independently and channels act as bounded buffers, the GPU thread is never starved waiting for preprocessing, and preprocessed batches don't pile up unboundedly in memory.

Quick start

[dependencies]
olympipe-rs = "0.1"

use olympipe_rs::Pipeline;

let result = Pipeline::new(0..100)
    .filter(|x| x % 2 == 0)
    .task(|x: i32| x * x, 4)   // 4 parallel workers
    .limit(10)
    .collect();

API

Constructors

Stages (consume self, return a new Pipeline)

Terminal methods

Split / Gather

use olympipe_rs::Pipeline;

let (evens, odds) = Pipeline::new(0..10i32).split(|x| {
    if x % 2 == 0 { (Some(x), None) } else { (None, Some(x)) }
});

// Both branches can be consumed sequentially (split uses unbounded channels
// internally so the router thread never blocks).
let mut e = evens.collect();
let mut o = odds.collect();
e.sort(); o.sort();

assert_eq!(e, vec![0, 2, 4, 6, 8]);
assert_eq!(o, vec![1, 3, 5, 7, 9]);

let p1 = Pipeline::new(vec![1i32, 2]);
let p2 = Pipeline::new(vec![3i32, 4]);
let mut result = p1.gather(vec![p2]).collect();
result.sort();
assert_eq!(result, vec![1, 2, 3, 4]);

Error handling

Worker panics are caught via std::panic::catch_unwind, logged to stderr, and treated as a clean shutdown for that worker (the output channel is dropped, propagating disconnection downstream — equivalent to olympipe's default behavior).

For recoverable errors in a stage, use task_or:

Pipeline::new(vec!["1", "two", "3"])
    .task_or(
        |s| s.parse::<i32>().map_err(|e| e),
        |s, _err| { eprintln!("skipping {s}"); None },
    )
    .collect(); // [1, 3]

Design notes