# Build an async job pipeline with fibre

**For:** Rust developers who know the basics of Tokio.

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.

Before you start: you need Rust with Cargo. fibre comes from crates.io as `fibre = "0"`.

## Set up the project

Start a new binary crate:

```sh
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.

```toml title="Cargo.toml"
[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:

```rust title="src/main.rs"
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:

```shell-session
$ 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:

```rust title="src/main.rs"
#[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.

```text
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:

```rust title="src/main.rs"
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();
}
```

```text
   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:

```rust title="src/main.rs"
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,
}
```

```text
   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:

```rust title="src/main.rs"
    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();
    }
```

```text
 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:

```rust title="src/main.rs"
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");
}
```

```text
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:

```text
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](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](https://docs.rs/fibre).