Concurrency and async Rust

Threads and scoped threads, Send and Sync, channels, shared state, async/await, the Tokio runtime, tasks, timeouts and cancellation.

Threads, scoped threads and channels

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

fn main() {
    // scoped threads: can borrow local data, all joined at the end of the scope
    let data = vec![1, 2, 3, 4];
    let total = std::thread::scope(|s| {
        let (a, b) = data.split_at(2);
        let h1 = s.spawn(|| a.iter().sum::<i32>());
        let h2 = s.spawn(|| b.iter().sum::<i32>());
        h1.join().unwrap() + h2.join().unwrap()
    });
    println!("{total} {:?}", data);

    // a channel is the safest way to move data between threads
    let (tx, rx) = mpsc::channel::<Job>();
    for id in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(Job { id, payload: id * 10 }).unwrap();
        });
    }
    drop(tx);                       // close the channel: the loop then ends

    let mut done = 0;
    for job in rx {                 // blocks until a value or the channel closes
        done += job.payload;
    }
    println!("{done}");
    thread::sleep(Duration::from_millis(0));
}

struct Job {
    id: u32,
    payload: u32,
}

• thread::scope removes the need for Arc when the data outlives the spawn site: the borrow is checked, and the scope waits for every thread before returning.
• Send means a value can be moved to another thread; Sync means a shared reference to it can be used from another thread. Rc is neither, Arc is both when its contents are.
• Clone the sender per thread and drop the original when finished. The receiving loop ends when every sender is gone, which is the cleanest shutdown signal in the standard library.
• mpsc is many-producer, single-consumer. For multiple consumers use a channel from crossbeam-channel or a Mutex<VecDeque>, or move to async tasks.
• Threads are for CPU-bound work and blocking APIs. A thread per request does not scale to tens of thousands of idle connections; that is what async is for.

Async, await and the runtime

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

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // concurrent, not sequential: both futures start before the await
    let (a, b) = tokio::join!(fetch("a"), fetch("b"));
    println!("{} {}", a, b);

    // a timeout wraps any future and cancels it on expiry
    match timeout(Duration::from_millis(50), sleep(Duration::from_secs(5))).await {
        Ok(()) => println!("slept"),
        Err(_) => println!("timed out and the future was dropped"),
    }

    // spawn a task on the runtime: it must be Send + 'static
    let handle = tokio::spawn(async move {
        sleep(Duration::from_millis(1)).await;
        42
    });
    println!("{}", handle.await?);

    // select takes whichever branch finishes first and drops the others
    tokio::select! {
        v = fetch("c") => println!("fetched {v}"),
        _ = sleep(Duration::from_millis(10)) => println!("gave up waiting"),
    }
    Ok(())
}

async fn fetch(name: &str) -> usize {
    sleep(Duration::from_millis(5)).await;
    name.len()
}

• An async fn returns a future that does nothing until it is awaited or spawned. Building it is free; polling it is the work.
• A future is a state machine the compiler generates, and dropping it is cancellation. That is why timeout works with no cooperation from the callee — the future is simply discarded.
• tokio::join! runs futures concurrently on one task and waits for all; tokio::spawn puts work on the runtime so it continues after the current task yields; select! takes the first to complete.
• Never block inside an async task. std::thread::sleep, a synchronous file read or a CPU loop with no yield stalls the executor thread and every task scheduled on it. Use tokio::task::spawn_blocking for blocking work.
• A spawned task must be Send and 'static, so it cannot borrow local variables. Clone an Arc into the closure instead, or use a scoped task if the runtime offers one.

Shared async state and shutdown

use std::sync::Arc;
use tokio::sync::{Mutex, RwLock, broadcast};

#[derive(Clone)]
struct App {
    cache: Arc<RwLock<HashMap<String, String>>>,
    events: broadcast::Sender<String>,
}

impl App {
    async fn put(&self, k: String, v: String) {
        self.cache.write().await.insert(k.clone(), v.clone());
        let _ = self.events.send(format!("updated {k}"));
    }
}

async fn serve(app: App, mut shutdown: tokio::signal::unix::Signal) {
    let mut sub = app.events.subscribe();
    loop {
        tokio::select! {
            // cancellation: the signal wins and the loop exits
            _ = shutdown.recv() => break,
            Ok(msg) = sub.recv() => println!("event: {msg}"),
        }
    }
    println!("shutting down cleanly");
}

// HashMap is only used for the snippet above
use std::collections::HashMap;

• Use tokio::sync::Mutex when the lock is held across an .await. The standard std::sync::Mutex guard is not Send, so holding it across an await will not compile.
• For a short critical section with no await inside, std::sync::Mutex is faster and simpler. Take the lock, copy what you need, release it, then await.
• broadcast fans one message out to many subscribers, watch keeps only the latest value for state, and mpsc is a queue with a single consumer. Choosing the wrong one is how shutdown logic gets complicated.
• Cancellation safety is a property of the future, not of select!: if a future is dropped mid-operation, partial work may be lost. Check the documentation for select! branches that perform I/O or write state.
• Bound your queues. An unbounded channel converts a slow consumer into unbounded memory growth; a bounded one applies backpressure, which is what you want.

FAQ

Related

Smart pointers and interior mutability File I/O, serde and everyday crates

Last refreshed 2026-09-18.