Actors, Worker Pools, and Pipelines

This lesson covers four ways to split work across tasks with channels: an actor that owns state, a worker pool that shares a queue, a pipeline of stages, and a single-flight call that shares one result among many callers. Examples in this lesson are not machine-checked (they are marked rust,ignore); their output follows from the code as written and is independent of scheduling.

The lesson uses bounded mpsc channels (mpsc Channels), oneshot replies (oneshot, broadcast, and watch), and JoinHandle (Spawning Tasks).

The actor pattern

An actor is a task that owns some state and changes it only in response to messages. No other task touches the state, so it needs no lock. An actor has three parts:

• A message type, usually an enum with one variant per operation.
• A task that loops on rx.recv() and handles one message at a time.
• A handle, a small cloneable struct that wraps the mpsc::Sender and offers ordinary async methods.

A message that needs an answer carries a oneshot::Sender. The actor sends the reply through it, and the caller awaits the matching oneshot::Receiver. Messages from one sender are handled in the order sent.

Example 1: a counter actor

The task ends when recv returns None, which happens after every Sender is dropped. Dropping all handles is the shutdown signal, and awaiting the task's JoinHandle waits until the actor has finished its last message.

use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;

enum Command {
    Add(u64),
    Get { reply: oneshot::Sender<u64> },
}

#[derive(Clone)]
struct CounterHandle {
    tx: mpsc::Sender<Command>,
}

impl CounterHandle {
    fn spawn() -> (CounterHandle, JoinHandle<()>) {
        let (tx, mut rx) = mpsc::channel(16);
        let task = tokio::spawn(async move {
            let mut total = 0;
            while let Some(command) = rx.recv().await {
                match command {
                    Command::Add(n) => total += n,
                    Command::Get { reply } => {
                        let _ = reply.send(total);
                    }
                }
            }
            println!("counter actor stopped at {total}");
        });
        (CounterHandle { tx }, task)
    }

    async fn add(&self, n: u64) {
        self.tx.send(Command::Add(n)).await.unwrap();
    }

    async fn get(&self) -> u64 {
        let (reply, answer) = oneshot::channel();
        self.tx.send(Command::Get { reply }).await.unwrap();
        answer.await.unwrap()
    }
}

#[tokio::main]
async fn main() {
    let (handle, task) = CounterHandle::spawn();

    let other = handle.clone();
    let worker = tokio::spawn(async move {
        for _ in 0..3 {
            other.add(10).await;
        }
    });
    worker.await.unwrap();

    handle.add(5).await;
    println!("total: {}", handle.get().await);

    drop(handle);
    task.await.unwrap();
}

total: 35
counter actor stopped at 35

The worker's three Add messages are queued before handle.add(5), because the program awaits the worker first. get returns 35 only after the actor handled all four additions, since the channel keeps messages in order.

The channel is bounded (16 here). When the actor falls behind, send waits, and callers slow down. This is backpressure (Backpressure and Load Shedding). An unbounded channel would let the queue grow without limit.

Example 2: supervision with JoinHandle

A panic inside a task stays inside that task. Awaiting its JoinHandle returns Err(JoinError), and is_panic() tells a panic apart from a cancellation (Cancellation, abort, and Panics). A supervisor is a task that starts a child, awaits its handle, and decides what to do: restart it, stop, or report.

A panicking actor drops its receiver, and queued messages would be lost. To keep them, the supervisor owns the receiver inside a tokio::sync::Mutex and each incarnation of the actor locks it. The Tokio mutex does not poison on panic, so the next incarnation can lock it.

use std::sync::Arc;
use tokio::sync::{Mutex, mpsc};

enum Msg {
    Work(u32),
    Crash,
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(8);
    tx.send(Msg::Work(1)).await.unwrap();
    tx.send(Msg::Crash).await.unwrap();
    tx.send(Msg::Work(2)).await.unwrap();
    drop(tx);

    let rx = Arc::new(Mutex::new(rx));
    let mut incarnation = 0;
    loop {
        incarnation += 1;
        let rx = rx.clone();
        let task = tokio::spawn(async move {
            let mut rx = rx.lock().await;
            while let Some(msg) = rx.recv().await {
                match msg {
                    Msg::Work(n) => println!("incarnation {incarnation}: work {n}"),
                    Msg::Crash => panic!("crash requested"),
                }
            }
        });
        match task.await {
            Ok(()) => {
                println!("actor stopped normally");
                break;
            }
            Err(error) if error.is_panic() => println!("actor panicked, restarting"),
            Err(_) => break,
        }
    }
}

incarnation 1: work 1
actor panicked, restarting
incarnation 2: work 2
actor stopped normally

The panic message itself goes to standard error, so it does not appear in the output. The Crash message was taken from the queue before the panic, so incarnation 2 continues with Work(2). A real supervisor also limits restarts and waits between them (backoff), or a crashing message could loop forever.

Example 3: a worker pool with a shared receiver

A worker pool is a fixed number of tasks that take jobs from one queue. Two channels connect it to the rest of the program: jobs go out to the workers (fan-out), and results come back into one channel (fan-in).

mpsc::Receiver has one owner, so the pool wraps it in Arc<Mutex<_>>. A worker locks the mutex only while it waits for the next job. The lock is released when the recv finishes, and the job runs without holding it. Another worker can then wait for the following job.

use std::sync::Arc;
use tokio::sync::{Mutex, mpsc};

#[tokio::main]
async fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u32>(8);
    let (out_tx, mut out_rx) = mpsc::channel::<(u32, u32)>(8);
    let job_rx = Arc::new(Mutex::new(job_rx));

    for _ in 0..3 {
        let job_rx = job_rx.clone();
        let out_tx = out_tx.clone();
        tokio::spawn(async move {
            loop {
                let job = job_rx.lock().await.recv().await;
                match job {
                    Some(n) => out_tx.send((n, n * n)).await.unwrap(),
                    None => break,
                }
            }
        });
    }
    drop(out_tx);

    for n in 1..=6 {
        job_tx.send(n).await.unwrap();
    }
    drop(job_tx);

    let mut results = Vec::new();
    while let Some(result) = out_rx.recv().await {
        results.push(result);
    }
    results.sort();
    println!("{results:?}");
}

[(1, 1), (2, 4), (3, 9), (4, 16), (5, 25), (6, 36)]

Which worker handles which job depends on scheduling, so the program sorts the results before printing. Shutdown follows the channels: dropping job_tx makes each worker's recv return None, the workers exit and drop their out_tx clones, and out_rx returns None once the last clone is gone. The main function keeps no out_tx of its own (drop(out_tx)), or the loop over results would never end.

The pool size limits parallel work. For CPU-bound jobs, run them with spawn_blocking inside the worker (Blocking and CPU-Bound Work).

Example 4: a pipeline

A pipeline is a chain of stages, each a task, connected by bounded channels. Each stage reads from the channel before it and writes to the one after it. A slow stage fills its input channel, which makes the stage before it wait, all the way back to the source.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx1, mut rx1) = mpsc::channel::<u32>(2);
    let (tx2, mut rx2) = mpsc::channel::<u32>(2);

    let source = tokio::spawn(async move {
        for n in 1..=5 {
            tx1.send(n).await.unwrap();
        }
    });
    let doubler = tokio::spawn(async move {
        while let Some(n) = rx1.recv().await {
            tx2.send(n * 2).await.unwrap();
        }
    });

    let mut total = 0;
    while let Some(n) = rx2.recv().await {
        println!("sink got {n}");
        total += n;
    }
    println!("total: {total}");

    source.await.unwrap();
    doubler.await.unwrap();
}

sink got 2
sink got 4
sink got 6
sink got 8
sink got 10
total: 30

Each stage has one task, so the order of values is kept. The end of the stream travels forward: the source finishes and drops tx1, the doubler's recv returns None, it finishes and drops tx2, and the sink's loop ends. A stage that needs more throughput can become a worker pool (Example 3), but then the order of its output is no longer guaranteed.

Example 5: single-flight with Shared

Single-flight means that when many callers ask for the same expensive result at the same time, one of them does the work and the others wait for its answer. futures::future::Shared supports this. It wraps a future so the wrapper can be cloned, and every clone gets the same output once the future completes. The output type must be Clone. Clones are polled by their owners, and the underlying future runs once.

use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::time::sleep;

static LOADS: AtomicUsize = AtomicUsize::new(0);

async fn load_config() -> String {
    LOADS.fetch_add(1, Ordering::SeqCst);
    sleep(Duration::from_millis(50)).await;
    "config-v1".to_string()
}

#[tokio::main]
async fn main() {
    let shared = load_config().boxed().shared();

    let a = tokio::spawn(shared.clone());
    let b = tokio::spawn(shared.clone());
    let c = tokio::spawn(shared);

    println!("{} {} {}", a.await.unwrap(), b.await.unwrap(), c.await.unwrap());
    println!("loads: {}", LOADS.load(Ordering::SeqCst));
}

config-v1 config-v1 config-v1
loads: 1

.boxed() turns the future into a boxed, Send future so it has a nameable type. A Shared future finishes once and then keeps returning the stored output, so it fits one load. It cannot refresh: a new load needs a new Shared.

For a value that changes over time, use a watch channel instead. One task refreshes the value and sends it, and readers borrow the latest value or wait for changed() (oneshot, broadcast, and watch).

Shutting down these patterns

For a signal-driven stop with a drain deadline, see Graceful Shutdown.

Interview points

• An actor owns its state in one task and receives messages over an mpsc channel. This avoids locks on the state and keeps message order per sender.
• Request/response uses a oneshot::Sender inside the message. If the actor drops it without replying, the caller's await returns an error.
• Bounded channels give backpressure. Unbounded channels grow until memory runs out.
• A task's panic is reported only through its JoinHandle. Supervisors call is_panic() on the JoinError.
• A worker pool over one mpsc::Receiver needs Arc<Mutex<Receiver>>. Work assignment and result order are not deterministic.
• Channel closure carries the shutdown signal: drop the senders and receivers see None.
• Shared runs a future once for many awaiters and needs a Clone output. A watch channel fits values that change.

Summary

• An actor is a handle, a task, an mpsc channel, and oneshot replies for calls that need an answer.
• Dropping all senders ends the actor loop. Awaiting the JoinHandle waits for it to finish.
• A supervisor awaits a child's JoinHandle and restarts the child after a panic. Keep the receiver outside the child to keep queued messages.
• A worker pool shares one receiver through Arc<Mutex<_>> and returns results on a second channel (fan-out and fan-in).
• A pipeline connects stages with bounded channels. Backpressure flows backward, and end-of-stream flows forward.
• futures::future::Shared gives single-flight loading of one value.