# Tokio Patterns

> When to activate: tokio async runtime, tasks, spawn, select, join, channels, time, interval, timeout, signal handling, graceful shutdown

- Skill: `mattakushi432/tokio-patterns` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/tokio-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/tokio-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/tokio-patterns

---


# Tokio Patterns

## Runtime Setup

```rust
// 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

```rust
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

```rust
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

```rust
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

```rust
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

```rust
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

```rust
// 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