Build an async job pipeline with fibre

You'll build a small Tokio program with three producers, a pool of workers and one collector. The producers push jobs into a bounded mpmc queue, the workers process them and an mpsc channel carries the results to the collector. The program shuts down on its own once the producers finish.

fibre = "0"

Set up the project#

Start a new binary crate:

cargo new fibre-pipeline
cd fibre-pipeline

Add fibre, Tokio and futures-util to Cargo.toml. The pipeline only needs Tokio's single-threaded runtime, its macros and its timers. futures-util provides StreamExt, which you'll use on the result channel at the end.

[dependencies]
fibre = "0"
tokio = { version = "1", features = ["macros", "rt", "time"] }
futures-util = "0.3"

The program uses the current_thread runtime throughout. Everything runs on one thread, so the output comes out in the same order every time and you can compare it line for line with what's shown here.

Send jobs to a worker#

The job queue is an mpmc channel. That's fibre's many-to-many channel: any number of tasks can send into it and any number can receive from it. Each job goes to exactly one receiver. You start with one of each. Replace src/main.rs with this:

use fibre::mpmc;

#[derive(Debug)]
struct Job {
    id: u32,
    text: String,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (job_tx, job_rx) = mpmc::bounded_async::<Job>(8);

    let worker = tokio::spawn(async move {
        while let Ok(job) = job_rx.recv().await {
            let words = job.text.split_whitespace().count();
            println!("worker: job {} has {words} words", job.id);
        }
        println!("worker: queue closed, exiting");
    });

    let lines = ["the quick brown fox", "jumps over", "the lazy dog"];
    for (id, text) in lines.into_iter().enumerate() {
        let job = Job { id: id as u32, text: text.to_string() };
        job_tx.send(job).await.expect("worker is running");
    }
    drop(job_tx);

    worker.await.unwrap();
}

mpmc::bounded_async(8) returns an AsyncSender and an AsyncReceiver for a queue that holds at most 8 jobs. send returns a future that resolves once the job is in the queue. It fails only if every receiver is gone. recv resolves with the next job. Once every sender has been dropped and the queue is empty, it resolves with Err(RecvError::Disconnected) instead.

That error is how the worker knows to stop. The drop(job_tx) after the loop drops the last sender. The worker drains the three jobs that are already queued and then its while let loop ends. Run it:

$ cargo run
worker: job 0 has 4 words
worker: job 1 has 2 words
worker: job 2 has 3 words
worker: queue closed, exiting

Add more producers#

Senders on an mpmc channel are Clone, so each producer gets its own clone. Give Job a field for the producer's name and replace main with a version that spawns three producer tasks:

#[derive(Debug)]
struct Job {
    producer: &'static str,
    id: u32,
    text: String,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (job_tx, job_rx) = mpmc::bounded_async::<Job>(8);

    let inputs = [
        ("alpha", ["the quick brown fox", "jumps over"]),
        ("beta", ["hello world", "fibre channels in rust"]),
        ("gamma", ["one two", "three four five"]),
    ];

    for (producer, lines) in inputs {
        let job_tx = job_tx.clone();
        tokio::spawn(async move {
            for (id, text) in lines.into_iter().enumerate() {
                let job = Job { producer, id: id as u32, text: text.to_string() };
                if job_tx.send(job).await.is_err() {
                    break;
                }
            }
        });
    }
    drop(job_tx);

    let worker = tokio::spawn(async move {
        while let Ok(job) = job_rx.recv().await {
            let words = job.text.split_whitespace().count();
            println!("worker: {} #{} has {words} words", job.producer, job.id);
        }
        println!("worker: queue closed, exiting");
    });

    worker.await.unwrap();
}

Each producer owns its clone of the sender. That clone is dropped when the producer's task finishes. The original job_tx still has to be dropped by hand right after the clones are made. If you leave it alive, the channel always has one sender left and the worker waits forever for a job that never arrives.

A producer that sees send fail stops early instead of panicking. That happens only if the workers have all gone away. There's nothing useful to do with the job at that point.

worker: alpha #0 has 4 words
worker: alpha #1 has 2 words
worker: beta #0 has 2 words
worker: beta #1 has 4 words
worker: gamma #0 has 2 words
worker: gamma #1 has 3 words
worker: queue closed, exiting

The queue holds 8 jobs and there are only 6, so each producer sends both of its jobs as soon as it runs.

See backpressure#

A bounded queue makes fast producers wait for a slow consumer. To see it, shrink the queue to 2, make the worker take 100ms per job and have each producer print a line when a job gets in. The helper ms rounds the elapsed time to the nearest 100ms so the numbers stay stable from run to run. Here is the whole file:

use fibre::mpmc;
use std::time::Duration;
use tokio::time::{sleep, Instant};

#[derive(Debug)]
struct Job {
    producer: &'static str,
    id: u32,
    text: String,
}

fn ms(start: Instant) -> u128 {
    (start.elapsed().as_millis() + 50) / 100 * 100
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (job_tx, job_rx) = mpmc::bounded_async::<Job>(2);
    let start = Instant::now();

    let inputs = [
        ("alpha", ["the quick brown fox", "jumps over"]),
        ("beta", ["hello world", "fibre channels in rust"]),
        ("gamma", ["one two", "three four five"]),
    ];

    for (producer, lines) in inputs {
        let job_tx = job_tx.clone();
        tokio::spawn(async move {
            for (id, text) in lines.into_iter().enumerate() {
                let job = Job { producer, id: id as u32, text: text.to_string() };
                if job_tx.send(job).await.is_err() {
                    break;
                }
                println!("{:>4}ms {producer} queued #{id}", ms(start));
            }
        });
    }
    drop(job_tx);

    let worker = tokio::spawn(async move {
        while let Ok(job) = job_rx.recv().await {
            sleep(Duration::from_millis(100)).await;
            let words = job.text.split_whitespace().count();
            println!("{:>4}ms worker: {} #{} has {words} words", ms(start), job.producer, job.id);
        }
        println!("{:>4}ms worker: queue closed, exiting", ms(start));
    });

    worker.await.unwrap();
}

   0ms alpha queued #0
   0ms alpha queued #1
   0ms beta queued #0
 100ms worker: alpha #0 has 4 words
 100ms gamma queued #0
 200ms worker: alpha #1 has 2 words
 200ms beta queued #1
 300ms worker: beta #0 has 2 words
 300ms gamma queued #1
 400ms worker: gamma #0 has 2 words
 500ms worker: beta #1 has 4 words
 600ms worker: gamma #1 has 3 words
 600ms worker: queue closed, exiting

At time zero three jobs get in: two fill the queue and the worker has already pulled alpha #0 out, which frees one slot for beta #0. After that every producer's send sits waiting until the worker finishes a job and takes the next one, so from then on exactly one new job enters the queue every 100ms. Nothing is dropped and memory stays flat no matter how many jobs the producers have lined up.

If you'd rather shed load than wait, use try_send. It never waits. When the queue is full it returns TrySendError::Full with the job inside, so you can log it, retry it later or throw it away. Swap the producer loop body for this and add use fibre::error::TrySendError; at the top:

match job_tx.try_send(job) {
    Ok(()) => println!("{:>4}ms {producer} queued #{id}", ms(start)),
    Err(TrySendError::Full(job)) => println!("{:>4}ms {producer} dropped #{}, queue full", ms(start), job.id),
    Err(_) => break,
}

   0ms alpha queued #0
   0ms alpha queued #1
   0ms beta dropped #0, queue full
   0ms beta dropped #1, queue full
   0ms gamma dropped #0, queue full
   0ms gamma dropped #1, queue full
 100ms worker: alpha #0 has 4 words
 200ms worker: alpha #1 has 2 words
 200ms worker: queue closed, exiting

The catch-all arm covers TrySendError::Closed, which means the receivers are gone. Put the blocking send back before moving on.

Run a pool of workers#

Receivers on an mpmc channel are Clone too. Every clone pulls from the same queue and each job goes to whichever worker asks first, so adding workers is a matter of cloning the receiver. Remove the println! from the producer loop so the output stays short, then replace the single worker with two:

    let mut workers = Vec::new();
    for worker in 0..2 {
        let job_rx = job_rx.clone();
        workers.push(tokio::spawn(async move {
            while let Ok(job) = job_rx.recv().await {
                sleep(Duration::from_millis(100)).await;
                let words = job.text.split_whitespace().count();
                println!("{:>4}ms worker {worker}: {} #{} has {words} words", ms(start), job.producer, job.id);
            }
            println!("{:>4}ms worker {worker}: queue closed, exiting", ms(start));
        }));
    }
    drop(job_rx);

    for handle in workers {
        handle.await.unwrap();
    }

 100ms worker 0: alpha #0 has 4 words
 100ms worker 1: alpha #1 has 2 words
 200ms worker 0: beta #0 has 2 words
 200ms worker 1: beta #1 has 4 words
 300ms worker 0: gamma #0 has 2 words
 300ms worker 0: queue closed, exiting
 300ms worker 1: gamma #1 has 3 words
 300ms worker 1: queue closed, exiting

Two workers finish the six jobs in 300ms instead of 600ms. The drop(job_rx) after the loop matters for the same reason as drop(job_tx): the original receiver would otherwise stay open and keep the channel alive with nobody reading from it.

Collect results over mpsc#

Printing from inside the workers is fine for a demo, but a real pipeline hands results to the next stage. Many workers feeding one collector is the pattern mpsc is built for. Its receiver is not Clone, which guarantees there's only ever one consumer. Its async receiver implements futures::Stream, so the collector can read it with StreamExt::next.

This is the finished program. Replace all of src/main.rs with it:

use fibre::{mpmc, mpsc};
use futures_util::StreamExt;
use std::time::Duration;
use tokio::time::sleep;

#[derive(Debug)]
struct Job {
    producer: &'static str,
    id: u32,
    text: String,
}

#[derive(Debug)]
struct Done {
    worker: usize,
    producer: &'static str,
    id: u32,
    words: usize,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (job_tx, job_rx) = mpmc::bounded_async::<Job>(2);
    let (done_tx, mut done_rx) = mpsc::bounded_async::<Done>(16);

    let inputs = [
        ("alpha", ["the quick brown fox", "jumps over"]),
        ("beta", ["hello world", "fibre channels in rust"]),
        ("gamma", ["one two", "three four five"]),
    ];

    for (producer, lines) in inputs {
        let job_tx = job_tx.clone();
        tokio::spawn(async move {
            for (id, text) in lines.into_iter().enumerate() {
                let job = Job { producer, id: id as u32, text: text.to_string() };
                if job_tx.send(job).await.is_err() {
                    break;
                }
            }
        });
    }
    drop(job_tx);

    for worker in 0..2 {
        let job_rx = job_rx.clone();
        let done_tx = done_tx.clone();
        tokio::spawn(async move {
            while let Ok(job) = job_rx.recv().await {
                sleep(Duration::from_millis(100)).await;
                let words = job.text.split_whitespace().count();
                let done = Done { worker, producer: job.producer, id: job.id, words };
                if done_tx.send(done).await.is_err() {
                    break;
                }
            }
            println!("worker {worker}: queue closed, exiting");
        });
    }
    drop(job_rx);
    drop(done_tx);

    let mut total = 0;
    while let Some(done) = done_rx.next().await {
        println!("worker {}: {} #{} has {} words", done.worker, done.producer, done.id, done.words);
        total += done.words;
    }
    println!("collector: all workers gone, {total} words total");
}

worker 0: alpha #0 has 4 words
worker 1: alpha #1 has 2 words
worker 0: beta #0 has 2 words
worker 1: beta #1 has 4 words
worker 0: queue closed, exiting
worker 1: queue closed, exiting
worker 0: gamma #0 has 2 words
worker 1: gamma #1 has 3 words
collector: all workers gone, 17 words total

Each worker gets a clone of done_tx and main drops the original. next() returns None only when every worker has exited and the result channel is empty, so the collector loop is also how main waits for the whole pipeline. There are no join handles to track.

Shut down in order#

The order of the last output lines shows how shutdown works. Both workers print "queue closed" before the collector prints the gamma results. The workers sent those results and exited. The results sat in the mpsc buffer until the collector got to them. When the senders go away, the receiver still drains everything already buffered before it reports the end.

The whole shutdown is driven by dropping senders, one stage at a time:

producers finish      -> last job_tx dropped
workers drain queue   -> recv() returns Err, workers exit, last done_tx dropped
collector drains      -> next() returns None, main returns

You don't need a cancellation token or a shutdown message for this. Each stage finishes its buffered work and then closes the next stage by dropping its sender.

Dropping from the receiving side behaves differently. A receiver that is dropped or closed stops at once and doesn't drain what's left in the buffer. Once every receiver is gone, the next send on that channel fails. That's why the producers and workers here break out of their loops on a send error rather than unwrapping it. For a clean stop, always close a pipeline from the sending end.

Where to go next#

The rest of fibre is covered in the channels guide at https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md. It covers spsc for one-to-one pipes, spmc for broadcasting every message to every consumer, spmc::topic for publish-subscribe, oneshot for request and reply, rendezvous channels and the batch API. The same guide shows how to mix sync threads and async tasks on one channel with to_sync() and to_async().

The full API, including every error type, is on docs.rs at https://docs.rs/fibre.