31. Tokio in practice: tasks, select!, timeouts, cancellation, shutdown

31.1 Tasks, JoinSet, and spawn_blocking

use std::time::Duration;
use tokio::task::JoinSet;

async fn check(host: &'static str) -> Result<u128, String> {
    let start = std::time::Instant::now();
    tokio::time::sleep(Duration::from_millis(host.len() as u64 * 10)).await; // stand-in for I/O
    if host.contains("bad") { return Err(format!("{host}: unreachable")); }
    Ok(start.elapsed().as_millis())
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for host in ["api.local", "db.local", "bad.local", "cache.local"] {
        set.spawn(async move { (host, check(host).await) }); // 'static + Send, like thread::spawn
    }
    while let Some(res) = set.join_next().await {            // results in COMPLETION order
        let (host, outcome) = res.expect("task panicked");
        println!("{host:<12} {outcome:?}");
    }

    // CPU-heavy or blocking work goes to a separate thread pool:
    let digest = tokio::task::spawn_blocking(|| (0..5_000_000u64).fold(0u64, |a, x| a.wrapping_add(x * x)))
        .await
        .unwrap();
    println!("digest {digest}");
}

This is Go's URL health-checker worker pool, without the WaitGroup.

31.2 Timeouts and select!

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

#[tokio::main]
async fn main() {
    // timeout() wraps any future; when time runs out, the inner future is DROPPED (cancelled).
    match timeout(Duration::from_millis(50), sleep(Duration::from_secs(10))).await {
        Ok(()) => println!("finished"),
        Err(_) => println!("timed out after 50ms"),
    }

    let (tx, mut rx) = mpsc::channel::<String>(8);
    tokio::spawn(async move {
        for i in 0..3 {
            tx.send(format!("msg {i}")).await.unwrap();
            sleep(Duration::from_millis(20)).await;
        }
    });

    let mut ticker = tokio::time::interval(Duration::from_millis(25));
    loop {
        tokio::select! {                      // wait on several futures; the first one ready wins
            Some(m) = rx.recv() => println!("got {m}"),
            _ = ticker.tick() => println!("tick"),
            else => break,                    // all branches disabled: channel closed
        }
        if rx.is_closed() && rx.is_empty() { break; }
    }
}

31.3 Cancellation is "drop the future"

In Go, cancellation is cooperative through ctx.Done() (Go §18), and a function that ignores ctx keeps running. In Rust, a future that is dropped stops at its current .await point and runs Drop for everything it holds. When timeout or select! picks another branch, the losing future is dropped. That is simpler than Go's model, but it has one sharp edge, cancellation safety:

// fragment
// BUG: if the timeout fires between the two writes, the peer sees half a message.
tokio::select! {
    _ = async { sock.write_all(header).await?; sock.write_all(body).await } => {}
    _ = sleep(deadline) => {}  // the other branch is dropped mid-write
}

The rule: any future you put in select! or timeout may stop at any .await. Keep partial state somewhere that survives, such as the connection's buffer, or encode and write the whole frame in one call. Tokio's documentation lists which methods are cancel-safe (for example, mpsc::Receiver::recv is; AsyncReadExt::read_exact is not).

31.4 Graceful shutdown

use std::time::Duration;
use tokio::sync::watch;
use tokio::task::JoinSet;

async fn worker(id: u32, mut shutdown: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            _ = shutdown.changed() => {
                println!("worker {id}: draining and exiting");
                return;
            }
            _ = tokio::time::sleep(Duration::from_millis(30)) => println!("worker {id}: tick"),
        }
    }
}

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel(false);
    let mut workers = JoinSet::new();
    for id in 0..2 { workers.spawn(worker(id, rx.clone())); }

    // In a real server: tokio::signal::ctrl_c().await, or a SIGTERM stream (Ch 40, 67).
    tokio::time::sleep(Duration::from_millis(70)).await;
    tx.send(true).unwrap();                           // tell everyone

    // Give them a deadline to finish, like Kubernetes' terminationGracePeriodSeconds.
    let drained = tokio::time::timeout(Duration::from_secs(5), async {
        while workers.join_next().await.is_some() {}
    }).await;
    println!("clean shutdown: {}", drained.is_ok());
}

tokio_util::sync::CancellationToken packages this pattern as a tree-shaped cancel signal, the closest equivalent to context.WithCancel.

31.5 The two async mistakes everyone makes

1. Blocking the executor. std::thread::sleep, std::fs::read of a large file, heavy CPU work, or a blocking database driver inside an async fn stalls every task on that worker thread. Use tokio::time::sleep, tokio::fs, or spawn_blocking.
2. Holding a std::sync::MutexGuard across .await. The guard is not Send, so tokio::spawn rejects the future with a long error that mentions "future cannot be sent between threads safely". Even when it compiles, holding a lock across a suspension point invites deadlocks. Copy what you need out of the guard and drop it before .await. Use tokio::sync::Mutex only when you really must hold a lock across .await.