Step 5 of 7

Async channels: tokio::sync::mpsc

Tasks pass messages through channels too, but std::sync::mpsc is the wrong tool in async code: its recv() blocks the thread. Tokio's channels, in tokio::sync, await instead:

• mpsc::channel(capacity) creates a bounded channel. tx.send(value).await waits while the channel is full (backpressure, as in module 32), and returns Err if the receiver is gone.
• rx.recv().await returns an Option: Some(message), or None once the channel is empty and every Sender has been dropped. So the receiving loop is while let Some(msg) = rx.recv().await. The receiver must be declared mut.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    for service in ["auth", "cart", "search"] {
        let tx = tx.clone();
        tokio::spawn(async move {
            for n in 1..=2 {
                tx.send(format!("{service} event {n}")).await.unwrap();
            }
        });
    }
    drop(tx);

    let mut events = Vec::new();
    while let Some(event) = rx.recv().await {
        events.push(event);
    }
    events.sort();
    for event in &events {
        println!("{event}");
    }
}

auth event 1
auth event 2
cart event 1
cart event 2
search event 1
search event 2

Everything from module 32 still applies: clone the Sender for each producer, drop the original so the loop can end, and sort messages from several producers before printing.

The other Tokio channels

Actors: a task that owns its state

A very common design in async services: instead of sharing a Mutex, one task owns the state, and others send it commands over an mpsc channel. A command that needs an answer carries a oneshot::Sender for the reply. No locks at all, and the state is only ever touched by one task:

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

enum Command {
    Deposit(u64),
    Balance(oneshot::Sender<u64>),
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(16);
    let bank = tokio::spawn(async move {
        let mut balance = 0;
        while let Some(command) = rx.recv().await {
            match command {
                Command::Deposit(amount) => balance += amount,
                Command::Balance(reply) => {
                    let _ = reply.send(balance);
                }
            }
        }
    });

    tx.send(Command::Deposit(50)).await.unwrap();
    tx.send(Command::Deposit(25)).await.unwrap();
    let (reply_tx, reply_rx) = oneshot::channel();
    tx.send(Command::Balance(reply_tx)).await.unwrap();
    println!("balance: {}", reply_rx.await.unwrap());

    drop(tx);
    bank.await.unwrap();
}

balance: 75

reply.send(...) on a oneshot isn't async (there's never a queue to wait for). let _ = ignores the Result: if the asker stopped waiting, there's nobody to tell.

Common mistakes

• Using std::sync::mpsc in async code: its recv() blocks a worker thread.
• Forgetting mut on the receiver: error[E0596]: cannot borrow `rx` as mutable, as it is not declared as mutable.
• Forgetting to drop the original Sender: rx.recv().await never returns None, and the loop waits forever.
• unbounded_channel everywhere: if the consumer is slower than the producers, memory grows without limit. Prefer a bounded channel.

Your task

Spawn a producer task that sends each amount in orders over a Tokio channel with capacity 2. The main task receives them with while let Some(...) = rx.recv().await, prints each one, and then prints the total:

order: 120
order: 80
order: 45
order: 300
total: 545

Challenge 2

Fill in the method that waits for the next message. The program should print received 3 alerts: cpu, disk, memory.

Challenge 3

Write an actor: async fn visitor_log(mut rx: mpsc::Receiver<Cmd>) owns a set of visitor names and handles commands until the channel closes:

• Cmd::Visit(name) records a visit (the same name twice counts once).
• Cmd::Count(reply) sends the number of different visitors so far through the oneshot reply.

The function returns when every sender is gone. The hidden tests spawn it as a task and talk to it.

Previous: Tasks: tokio::spawn and JoinHandle Next: select!: the first one to finish wins