Tokio / Chapter 2 of 6

Tasks, channels, and shared state

Use tokio::spawn, select!, mpsc, and oneshot to coordinate async tasks. Examples cover shared state, timeouts, and cancellation.

7 topicsUpdated 22 Oct 2026

05 / 26

Spawn tasks and wait for results#

Use tasks for independent work. Keep the JoinHandle when the result or completion matters.

tokio::spawn schedules a future and returns a JoinHandle<T>. Awaiting that handle returns Result<T, JoinError>; a panic or cancellation becomes a JoinError. Dropping a handle detaches the task: it keeps running until it finishes or the runtime shuts down.

RustOwn data with async move

use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), tokio::task::JoinError> {
    let name = String::from("invoice");
    let handle = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(50)).await;
        format!("{name} ready") // task owns name
    });

    let message = handle.await?; // handle may report a panic
    println!("{message}");
    Ok(())
}

RustWait for many tasks with JoinSet

use tokio::task::JoinSet;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut tasks = JoinSet::new();
    for job in 0..5 {
        tasks.spawn(async move { (job, job * job) });
    }
    while let Some(result) = tasks.join_next().await {
        let (job, answer) = result?;
        println!("job {job}: {answer}");
    }
    Ok(())
}

Structured cleanup

A JoinSet aborts its remaining tasks when dropped. For ordinary handles, use explicit abort() then await the handle if you need cancellation to complete.

06 / 26

join!, try_join!, and select!#

Wait for all operations with join!, propagate errors with try_join!, or respond to the first event with select!.

RustBoth must finish

async fn load(id: u32) -> Result<u32, &'static str> {
    if id == 0 { Err("bad id") } else { Ok(id * 10) }
}

#[tokio::main]
async fn main() -> Result<(), &'static str> {
    let (a, b) = tokio::try_join!(load(1), load(2))?;
    println!("{}", a + b);
    Ok(())
}

RustFirst event wins

use std::time::Duration;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(8);
    tx.send("ready".into()).await.unwrap();
    tokio::select! {
        Some(message) = rx.recv() => println!("{message}"),
        _ = tokio::time::sleep(Duration::from_secs(2)) => {
            println!("timed out");
        }
    }
}

Cancellation safety

When one select! branch wins, other branch futures are dropped. Receiving with mpsc::Receiver::recv is cancellation safe; read_exact and write_all can lose partial progress if recreated in a loop. select! is concurrent within one task; it does not spawn tasks.

07 / 26

mpsc: bounded work queues#

Multiple producers send jobs to one receiver. A bounded channel applies backpressure when the consumer falls behind.

The capacity limits queued items; send().await waits for space. Clone senders for more producers. recv().await returns None after all senders have dropped and queued messages have been consumed. One receiver can distribute work by spawning tasks after receiving.

RustTwo producers, one worker

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(16);
    for producer in 0..2 {
        let tx = tx.clone();
        tokio::spawn(async move {
            for job in 0..3 {
                if tx.send(producer * 10 + job).await.is_err() {
                    break; // receiver was dropped
                }
            }
        });
    }
    drop(tx); // release main's sender or recv never returns None
    while let Some(job) = rx.recv().await {
        println!("process {job}");
    }
}

Bounded by default

mpsc::unbounded_channel() uses a synchronous send but has no queue limit. Avoid it if producers can outpace the consumer indefinitely.

08 / 26

oneshot: request and reply#

Send exactly one result back to a caller, often combined with an mpsc request queue.

The sender uses synchronous send(value); the receiver uses .await. A dropped sender makes the receiver return an error. This is an easy way to give each queued request its own response.

RustWorker with a per-request reply

use tokio::sync::{mpsc, oneshot};

struct Request {
    value: u64,
    reply: oneshot::Sender<u64>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel::<Request>(8);
    let worker = tokio::spawn(async move {
        while let Some(Request { value, reply }) = rx.recv().await {
            let _ = reply.send(value * value);
        }
    });

    let (reply_tx, reply_rx) = oneshot::channel();
    tx.send(Request { value: 12, reply: reply_tx }).await
        .map_err(|_| std::io::Error::new(
            std::io::ErrorKind::BrokenPipe, "worker closed"
        ))?;
    println!("answer: {}", reply_rx.await?);
    drop(tx);
    worker.await?;
    Ok(())
}

09 / 26

Shared state: Arc, Mutex, RwLock#

Share one value with Arc; choose a lock based on what happens while holding it.

Arc<T> lets tasks own references to the same allocation. Use std::sync::Mutex for short, non-async operations and release its guard before any .await. Tokio’s Mutex is appropriate when the guard truly must survive an await. RwLock allows several readers or one writer.

RustShort critical section

use std::sync::{Arc, Mutex};
use tokio::task::JoinSet;

#[tokio::main]
async fn main() -> Result<(), tokio::task::JoinError> {
    let counts = Arc::new(Mutex::new(vec![0u32; 4]));
    let mut tasks = JoinSet::new();
    for index in 0..4 {
        let counts = Arc::clone(&counts);
        tasks.spawn(async move {
            // Guard is released at the end of this scope.
            { counts.lock().unwrap()[index] += 1; }
            tokio::task::yield_now().await;
        });
    }
    while let Some(result) = tasks.join_next().await { result?; }
    println!("{:?}", counts.lock().unwrap());
    Ok(())
}

RustAsync RwLock for shared configuration

use std::sync::Arc;
use tokio::sync::RwLock;

#[tokio::main]
async fn main() {
    let config = Arc::new(RwLock::new(String::from("v1")));
    let reader = Arc::clone(&config);
    let task = tokio::spawn(async move {
        let snapshot = reader.read().await.clone();
        println!("using {snapshot}");
    });
    *config.write().await = "v2".into();
    task.await.unwrap();
}

Do not hold locks around slow work

If I/O or a long calculation needs to happen, copy the needed data under the lock, release it, then do the work. An awaited lock guard can serialize unrelated tasks.

10 / 26

Sleep, timeout, and intervals#

Tokio timers suspend tasks. A timeout stops waiting and drops its inner future when time expires.

RustTimeout around an operation

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

#[tokio::main]
async fn main() {
    let result = timeout(Duration::from_millis(30), async {
        sleep(Duration::from_secs(1)).await;
        42
    }).await;
    assert!(result.is_err()); // Elapsed
}

RustPeriodic task with missed-tick policy

use std::time::Duration;
use tokio::time::{interval, MissedTickBehavior};

#[tokio::main]
async fn main() {
    let mut ticks = interval(Duration::from_secs(5));
    ticks.set_missed_tick_behavior(MissedTickBehavior::Skip);
    for _ in 0..3 {
        ticks.tick().await; // first tick is immediate
        println!("refresh");
    }
}

Deadline detail

timeout checks elapsed time before polling the inner future. Long synchronous work without yielding can exceed the deadline. interval defaults to burst behavior after missed ticks; choose Skip or Delay when bursts are undesirable.

11 / 26

Cancellation and task lifetimes#

Cancellation is usually cooperative: a task notices a signal and returns after cleanup.

Dropping a future stops it at its next suspension point. JoinHandle::abort() requests cancellation and handle.await confirms completion. A CancellationToken (from tokio-util) lets many tasks choose their own cleanup path.

TOMLCargo.toml addition

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.7", features = ["rt"] }

RustRequest cancellation and await cleanup

use std::time::Duration;
use tokio_util::sync::CancellationToken;

#[tokio::main]
async fn main() -> Result<(), tokio::task::JoinError> {
    let token = CancellationToken::new();
    let worker_token = token.clone();
    let worker = tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = worker_token.cancelled() => break,
                _ = tokio::time::sleep(Duration::from_secs(1)) => {
                    println!("heartbeat");
                }
            }
        }
        println!("flush buffered work here");
    });
    token.cancel();
    worker.await?;
    Ok(())
}

Aborts are not cleanup hooks

abort() drops a task when it next yields; Rust destructors run, but async cleanup code cannot run after an abrupt abort. Blocking work in spawn_blocking generally cannot be stopped by abort once running.