Tokio Patterns

Tokio Patterns

Runtime Setup

// Multi-threaded (default for servers)
#[tokio::main]
async fn main() { run().await; }

// Single-threaded (embed in sync code)
fn main() {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async { run().await });
}

Task Spawning and JoinSet

use tokio::task::JoinSet;
use std::sync::Arc;
use tokio::sync::Semaphore;

// Bounded concurrency with JoinSet + Semaphore
async fn bounded_fanout(ids: Vec<u64>, concurrency: usize) -> Vec<Data> {
    let sem = Arc::new(Semaphore::new(concurrency));
    let mut set = JoinSet::new();

    for id in ids {
        let sem = sem.clone();
        set.spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            fetch(id).await.unwrap()
        });
    }

    let mut results = Vec::new();
    while let Some(Ok(data)) = set.join_next().await {
        results.push(data);
    }
    results
}

// Detached background task
fn spawn_cleanup(state: AppState) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
        loop {
            ticker.tick().await;
            cleanup_expired(&state).await;
        }
    });
}

Time and Intervals

use tokio::time::{sleep, interval, timeout, Duration, MissedTickBehavior};

// Periodic interval (does not drift)
async fn heartbeat() {
    let mut ticker = interval(Duration::from_secs(30));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
    loop {
        ticker.tick().await;
        send_heartbeat().await;
    }
}

// Timeout on any future
async fn fetch_with_timeout() -> anyhow::Result<Data> {
    timeout(Duration::from_secs(10), fetch_data())
        .await
        .map_err(|_| anyhow::anyhow!("fetch timed out"))?
}

select! Patterns

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

// Event loop with cancellation
async fn event_loop(
    mut events: mpsc::Receiver<Event>,
    cancel: CancellationToken,
) {
    loop {
        tokio::select! {
            biased; // check cancel first every iteration
            _ = cancel.cancelled() => {
                tracing::info!("shutting down");
                break;
            }
            Some(event) = events.recv() => {
                handle_event(event).await;
            }
            else => break,
        }
    }
}

// Merge two channels
async fn merge_sources(mut rx1: mpsc::Receiver<String>, mut rx2: mpsc::Receiver<String>) {
    loop {
        tokio::select! {
            Some(msg) = rx1.recv() => process(msg),
            Some(msg) = rx2.recv() => process(msg),
            else => break,
        }
    }
}

Graceful Shutdown

use tokio::signal;
use tokio_util::sync::CancellationToken;

async fn run_server(state: AppState) -> anyhow::Result<()> {
    let token = CancellationToken::new();

    let worker = tokio::spawn({
        let token = token.clone();
        let state = state.clone();
        async move { background_worker(state, token).await }
    });

    let server = tokio::spawn({
        let token = token.clone();
        async move {
            let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
            axum::serve(listener, create_router(state))
                .with_graceful_shutdown(token.cancelled_owned())
                .await
        }
    });

    // Wait for Ctrl+C or SIGTERM
    tokio::select! {
        _ = signal::ctrl_c() => tracing::info!("received Ctrl+C"),
    }

    token.cancel();
    let (sr, wr) = tokio::join!(server, worker);
    sr??; wr?;
    Ok(())
}

Sync Primitives

use tokio::sync::{OnceCell, Notify, Semaphore};

// Lazy async initialization (singleton)
static DB: OnceCell<DbPool> = OnceCell::const_new();

async fn get_db() -> &'static DbPool {
    DB.get_or_init(|| async { DbPool::connect("postgres://...").await.unwrap() }).await
}

// Notify: signal without data
async fn wait_for_work(notify: std::sync::Arc<Notify>) {
    notify.notified().await;
    println!("woken up");
}

CPU-Bound Work

// Never block the async runtime with CPU-heavy work
async fn process_image(data: Vec<u8>) -> anyhow::Result<Vec<u8>> {
    tokio::task::spawn_blocking(move || {
        // runs on a dedicated blocking thread pool
        encode_image(data)
    })
    .await?
}

Common Anti-Patterns

• std::thread::sleep in async code — use tokio::time::sleep
• std::sync::Mutex held across .await — use tokio::sync::Mutex or release before awaiting
• Unbounded task spawning — bound concurrency with Semaphore or JoinSet
• Dropping JoinHandle without awaiting — detached tasks; panics become silent
• select! without else branch — add else => break when channels can close