Tokio in Practice

Tokio is the most widely used async runtime: a multi-threaded scheduler, a reactor for network I/O and timers, and async versions of channels, mutexes, files and processes. Axum, SQLx, reqwest and tonic all build on it.

[dependencies]
tokio = { version = "1", features = ["full"] }

full is convenient while learning. Libraries should enable only what they use (rt-multi-thread, macros, net, time, sync, ...).

The runtime and tasks

#[tokio::main] turns async fn main into a normal main that builds a runtime and blocks on your future. tokio::spawn starts a task — a lightweight green thread that the scheduler runs concurrently with others. It returns a JoinHandle you can .await for the result.

use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let handles: Vec<_> = (1..=3u64)
        .map(|i| {
            tokio::spawn(async move {
                sleep(Duration::from_millis(10 * i)).await;
                i * i
            })
        })
        .collect();

    let mut total = 0;
    for handle in handles {
        total += handle.await.expect("task panicked");
    }
    println!("total = {total}");
}

Expected output:

total = 14

handle.await returns Result<T, JoinError>; the error means the task panicked or was cancelled. Spawned tasks must be Send + 'static, so move owned data in rather than borrowing from the spawner.

Running futures concurrently: join! and select!

tokio::join! polls several futures at the same time on the current task and waits for all of them. tokio::select! waits for the first to finish and drops (cancels) the rest.

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

async fn fetch(name: &str, ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await;
    format!("{name} done")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (a, b) = tokio::join!(fetch("users", 200), fetch("orders", 200));
    println!("{a}, {b}");
    println!("concurrent: {}", start.elapsed() < Duration::from_millis(350));

    tokio::select! {
        v = fetch("fast", 20) => println!("winner: {v}"),
        v = fetch("slow", 500) => println!("winner: {v}"),
    }

    match timeout(Duration::from_millis(50), fetch("report", 1_000)).await {
        Ok(v) => println!("{v}"),
        Err(_) => println!("report timed out"),
    }
}

Expected output:

users done, orders done
concurrent: true
winner: fast done
report timed out

Cancellation in async Rust means dropping the future: it simply stops being polled at its current .await. Code that must not be interrupted halfway (for example, writing half a record) needs care when used inside select! or timeout.

join! gives concurrency, not parallelism — both futures run on one task. Use tokio::spawn (or a JoinSet for a dynamic number of tasks) when work should run in parallel across the runtime's worker threads.

Channels between tasks

tokio::sync::mpsc is a bounded, async multi-producer single-consumer channel. send().await waits when the buffer is full, which gives you backpressure for free. The receiver sees None once every sender is dropped.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(8);

    for id in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("worker {id} finished")).await.unwrap();
        });
    }
    drop(tx); // keep only the clones, so the channel closes when they finish

    let mut messages = Vec::new();
    while let Some(msg) = rx.recv().await {
        messages.push(msg);
    }
    messages.sort(); // arrival order is not deterministic
    println!("{messages:?}");
}

Expected output:

["worker 0 finished", "worker 1 finished", "worker 2 finished"]

Tokio also provides oneshot (a single reply, handy for request/response), broadcast (every receiver gets every message) and watch (latest value only, good for config or shutdown signals).

Blocking work: spawn_blocking

Never run blocking or CPU-heavy code directly in a task. spawn_blocking moves a closure onto a separate thread pool meant for blocking work:

fn checksum(data: &[u8]) -> u64 {
    data.iter().map(|&b| b as u64).sum()
}

#[tokio::main]
async fn main() {
    let data = vec![7u8; 1_000_000];
    let sum = tokio::task::spawn_blocking(move || checksum(&data))
        .await
        .expect("blocking task panicked");
    println!("{sum}");
}

Expected output:

7000000

Common mistakes

• Calling std::thread::sleep or a blocking client inside async code. Use tokio::time::sleep and async libraries, or spawn_blocking.
• Holding a std::sync::MutexGuard across an .await in a spawned task. The guard is not Send, so the task isn't either:

error: future cannot be sent between threads safely

Drop the guard before awaiting (scope it in a block), or use tokio::sync::Mutex if the lock genuinely must be held across .await.

• Borrowing local data in tokio::spawn. The task may outlive the caller; move owned values or Arcs in with async move.
• Forgetting to drop the original tx, so rx.recv() waits forever.
• Using #[tokio::main] without the macros and rt-multi-thread features.

Practice

Spawn five tasks. Task i (1 to 5) sleeps i * 10 ms and sends i over an mpsc channel. In main, receive everything with an overall 1-second timeout and print the sum.

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, timeout};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(5);
    for i in 1..=5u64 {
        let tx = tx.clone();
        tokio::spawn(async move {
            sleep(Duration::from_millis(i * 10)).await;
            let _ = tx.send(i).await;
        });
    }
    drop(tx);

    let collect = async {
        let mut sum = 0;
        while let Some(n) = rx.recv().await {
            sum += n;
        }
        sum
    };

    match timeout(Duration::from_secs(1), collect).await {
        Ok(sum) => println!("sum = {sum}"),
        Err(_) => println!("timed out"),
    }
}