iwantcoding.com
🔥 Daily 👥 Rooms 🏆 Top Log in Sign up

Tokio

By · Reviewed & last updated · Part of Rust Tutorial

Tokio is Rusts mainstream async runtime: a multi-threaded scheduler, an async I/O reactor, timers, channels, and a synchronisation toolkit. Pick it for any non-trivial async work — Axum, Reqwest, SQLx, and Tonic all build on it. Configure the runtime macro (#[tokio::main]) to choose multi-threaded vs current-thread.

Tokio runtime, tasks, channels, timeouts, JoinSet

EXAMPLE
// Cargo.toml
// [dependencies]
// tokio = { version = "1", features = ["full"] }

use std::time::Duration;
use tokio::{
    sync::{mpsc, Semaphore, oneshot},
    task::JoinSet,
    time,
};

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() -> anyhow::Result<()> {
    // 1) Spawn — fire and forget, returns a JoinHandle you can await later
    let handle = tokio::spawn(async {
        time::sleep(Duration::from_millis(50)).await;
        42
    });
    println!("task returned {}", handle.await?);

    // 2) Channels — mpsc (multi-producer single-consumer) with backpressure
    let (tx, mut rx) = mpsc::channel::<u32>(32);
    let producer = tokio::spawn(async move {
        for i in 0..10 {
            tx.send(i).await.unwrap();
        }
    });
    let consumer = tokio::spawn(async move {
        while let Some(n) = rx.recv().await {
            println!("got {n}");
        }
    });
    let _ = tokio::try_join!(producer, consumer);

    // 3) Oneshot — single-use reply channel, common for request/response
    let (tx, rx) = oneshot::channel::<&str>();
    tokio::spawn(async move {
        time::sleep(Duration::from_millis(50)).await;
        let _ = tx.send("done");
    });
    println!("oneshot: {}", rx.await?);

    // 4) Timeout — wrap any future
    match time::timeout(Duration::from_millis(20), slow()).await {
        Ok(v) => println!("got: {v}"),
        Err(_) => println!("timed out"),
    }

    // 5) Concurrency limit with Semaphore — pool around a finite resource
    let sem = std::sync::Arc::new(Semaphore::new(3));
    let mut joins = JoinSet::new();
    for i in 0..10 {
        let sem = sem.clone();
        joins.spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            time::sleep(Duration::from_millis(30)).await;
            i * 2
        });
    }
    let mut results = Vec::new();
    while let Some(r) = joins.join_next().await {
        results.push(r?);
    }
    results.sort();
    println!("results: {results:?}");

    // 6) CPU-bound work goes on the blocking pool, not the async pool
    let n = tokio::task::spawn_blocking(|| {
        (0..1_000_000_u64).fold(0u64, |a, x| a.wrapping_add(x.wrapping_mul(31)))
    })
    .await?;
    println!("sum = {n}");

    // 7) Graceful shutdown — cancel + drain remaining tasks
    let (cancel_tx, mut cancel_rx) = mpsc::channel::<()>(1);
    let worker = tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = time::sleep(Duration::from_millis(20)) => println!("tick"),
                _ = cancel_rx.recv() => { println!("shutdown"); break; }
            }
        }
    });
    time::sleep(Duration::from_millis(80)).await;
    let _ = cancel_tx.send(()).await;
    let _ = worker.await;

    Ok(())
}

async fn slow() -> &"static str {
    time::sleep(Duration::from_millis(100)).await;
    "ok"
}

Why it matters

JoinSet is the right abstraction when you spawn N tasks and need to consume their results as they finish. It avoids the boilerplate of carrying Vec>, gives you cancellation when dropped, and pairs naturally with a Semaphore to bound concurrency.

Tip: Tweak the snippet with Try it Yourself », then sit the quiz at the bottom of the page.

Example

Example
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    sleep(Duration::from_millis(500)).await;
    println!("woke up");
}
Try it Yourself »

Discussion

Loading…