Async & Tokio

async fn greet() -> String {
    "hello".to_string()
}

// is equivalent to:
fn greet2() -> impl std::future::Future<Output = String> {
    async { "hello".to_string() }
}

Calling greet() runs nothing: it creates a future, a value that represents a computation not yet done.

async fn write_log() {
    println!("written");
}

async fn example() {
    write_log();          // NOTHING happens. Warning: unused implementer of `Future`
    write_log().await;    // now the function runs
}

Rust futures are lazy: unlike JavaScript promises, they only start when you await them (or hand them to spawn).

async fn handle() {
    let u = read_user().await;     // the task is suspended; the thread serves other tasks
    let o = read_orders().await;
}

This code is still sequential: read_orders only starts after read_user. The difference from synchronous code: while waiting, the thread handles other requests. To run both at the same time: section 6.

let future = async {
    let a = 1;
    a + 1
};

let x = 5;
let with_capture = async move { x * 2 };   // move: the block owns x

let f = async |n: u32| n + 1;              // async closure (Rust 1.85+)

[dependencies]
tokio = { version = "1", features = ["full"] }

full enables everything; a library or an optimised binary only picks what it needs:

#[tokio::main]
async fn main() {
    println!("inside the runtime");
}

// is equivalent to:
fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("inside the runtime");
        });
}

#[tokio::main(flavor = "current_thread")]        // a single thread: lighter, for a CLI tool
async fn main() {}

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]   // chosen number of threads
async fn main2() {}

By default, the multi-thread runtime creates one thread per core.

fn tool() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let r = rt.block_on(async { 42 });
}

block_on inside a runtime that is already running panics ("Cannot start a runtime from within a runtime"): use it only at the boundary, in a main or a synchronous tool.

#[tokio::main]
async fn main() {
    let task = tokio::spawn(async {
        // runs in parallel with main, possibly on another thread
        21 * 2
    });

    let result = task.await.unwrap();      // JoinHandle: wait for the end and get the value
    println!("{result}");
}

• spawn starts the task immediately, without waiting for an .await.
• It returns a JoinHandle<T>; awaiting it gives Result<T, JoinError> (Err if the task panicked or was cancelled).
• The task must be Send + 'static: it can change thread and outlive the function that launches it (course send-sync).

tokio::spawn(async {
    send_email().await;     // nobody waits for the result
});

If the JoinHandle is ignored, the task keeps running anyway. But a panic or an error in the task goes unnoticed: log errors inside it.

use std::sync::Arc;

let config = Arc::new(String::from("prod"));
for i in 0..3 {
    let config = Arc::clone(&config);      // one copy of the pointer per task
    tokio::spawn(async move {
        println!("task {i}: {config}");
    });
}

use tokio::task::JoinSet;

let mut group = JoinSet::new();
for id in 1..=5 {
    group.spawn(async move { id * 10 });
}

while let Some(r) = group.join_next().await {   // in order of COMPLETION, not launch
    println!("{}", r.unwrap());
}

A JoinSet keeps the tasks together: join_next returns results as they come, and all the remaining tasks are cancelled if the JoinSet is dropped. This is the safe way to launch a variable number of tasks.

async fn bad() {
    std::thread::sleep(std::time::Duration::from_secs(1));   // blocks the whole THREAD
    let h = argon2_hash("password");                          // 100 ms computation: blocks too
}

A runtime thread that blocks cannot move any other task forward. With 4 threads and 4 requests hashing a password, the whole server is frozen.

async fn good() {
    // 1. Wait: the async version
    tokio::time::sleep(std::time::Duration::from_secs(1)).await;

    // 2. Heavy computation or blocking library: a dedicated thread
    let hash = tokio::task::spawn_blocking(|| argon2_hash("password"))
        .await
        .unwrap();
}

spawn_blocking runs the closure on a separate set of threads, meant for blocking.

async fn user() -> String { "ada".into() }
async fn orders() -> Vec<u32> { vec![1, 2] }

async fn page() {
    // Sequential: 20 ms + 30 ms = 50 ms
    let u = user().await;
    let o = orders().await;

    // Simultaneous: max(20, 30) = 30 ms
    let (u, o) = tokio::join!(user(), orders());
}

join! moves both futures forward in the same task, in turns: no new thread, no Send required.

async fn a() -> Result<u32, String> { Ok(1) }
async fn b() -> Result<u32, String> { Err("failure".into()) }

async fn combine() -> Result<u32, String> {
    let (x, y) = tokio::try_join!(a(), b())?;   // stops at the first error
    Ok(x + y)
}

use std::time::Duration;

async fn fast() -> &'static str { "fast" }
async fn slow() -> &'static str { tokio::time::sleep(Duration::from_secs(5)).await; "slow" }

async fn race() {
    tokio::select! {
        r = fast() => println!("{r}"),
        r = slow() => println!("{r}"),       // never printed: this future is CANCELLED
    }
}

As soon as one branch finishes, the other futures are dropped: their work in progress is abandoned (section 12).

use tokio::sync::mpsc;

async fn run_loop(mut rx: mpsc::Receiver<String>, mut stop: tokio::sync::oneshot::Receiver<()>) {
    loop {
        tokio::select! {
            Some(msg) = rx.recv() => println!("message: {msg}"),
            _ = &mut stop => {
                println!("stop requested");
                break;
            }
        }
    }
}

&mut stop: we borrow the future instead of consuming it, so it can be reused on each turn.

use futures::future::join_all;

async fn load(id: u32) -> u32 { id }

async fn load_all() -> Vec<u32> {
    let futures = (1..=10).map(load);
    join_all(futures).await                 // all at the same time, results in order
}

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

async fn timing() {
    sleep(Duration::from_millis(100)).await;

    let mut tick = interval(Duration::from_secs(60));
    for _ in 0..3 {
        tick.tick().await;           // the first tick is immediate
        println!("every minute");
    }

    match timeout(Duration::from_secs(2), slow_operation()).await {
        Ok(value) => println!("{value}"),
        Err(_) => println!("too long: giving up"),
    }
}

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(100);    // capacity 100: beyond that, send waits

    for i in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("job {i}")).await.unwrap();
        });
    }
    drop(tx);                                            // otherwise rx would never end

    while let Some(msg) = rx.recv().await {              // None when all senders are dropped
        println!("{msg}");
    }
}

The capacity limits memory: when the queue is full, send().await waits for the consumer to catch up. This is backpressure. unbounded_channel has no limit, so it can grow memory without end if the consumer is slower.

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

enum Command {
    Read { key: String, reply: oneshot::Sender<Option<String>> },
}

async fn handler(mut rx: mpsc::Receiver<Command>) {
    let mut data = std::collections::HashMap::new();
    data.insert("a".to_string(), "1".to_string());
    while let Some(Command::Read { key, reply }) = rx.recv().await {
        let _ = reply.send(data.get(&key).cloned());
    }
}

async fn client(tx: mpsc::Sender<Command>) {
    let (reply_tx, reply_rx) = oneshot::channel();
    tx.send(Command::Read { key: "a".into(), reply: reply_tx }).await.unwrap();
    let value = reply_rx.await.unwrap();
}

This is the "actor" pattern: a single task owns the state, the others send it commands. No lock is needed.

use tokio::sync::{broadcast, watch};

let (tx, mut rx1) = broadcast::channel::<String>(16);
let mut rx2 = tx.subscribe();
tx.send("to everyone".into()).unwrap();          // rx1 and rx2 each receive the message

let (cfg_tx, mut cfg_rx) = watch::channel("v1".to_string());
cfg_tx.send("v2".into()).unwrap();          // readers see the latest value
// cfg_rx.changed().await: wait for the next change
let current = cfg_rx.borrow().clone();

use std::sync::{Arc, Mutex};

#[derive(Clone)]
struct State { visits: Arc<Mutex<u64>> }

async fn handler(state: State) {
    *state.visits.lock().unwrap() += 1;      // lock taken and released with no .await in between
}

A std::sync::Mutex is the right choice as long as the lock is never held across an .await: it is faster than an async lock.

use std::sync::Arc;
use tokio::sync::Mutex;

async fn write(file: Arc<Mutex<tokio::fs::File>>, line: &[u8]) {
    use tokio::io::AsyncWriteExt;
    let mut f = file.lock().await;           // wait for the lock without blocking the thread
    f.write_all(line).await.unwrap();        // lock held during the write: required here
}

use std::sync::{Arc, RwLock};

let cache: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
let reading = cache.read().unwrap().len();       // several simultaneous readers
cache.write().unwrap().push("x".into());         // a single writer

use std::sync::atomic::{AtomicU64, Ordering};
static REQUESTS: AtomicU64 = AtomicU64::new(0);
REQUESTS.fetch_add(1, Ordering::Relaxed);

The details of the sharing primitives are in the concurrence course.

// Launching 10,000 requests at once: the remote service refuses, memory explodes
// for url in urls { tokio::spawn(download(url)); }

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

async fn download(id: u32) -> u32 { id }

async fn all(ids: Vec<u32>) -> Vec<u32> {
    let limit = Arc::new(Semaphore::new(10));    // at most 10 at the same time
    let mut group = JoinSet::new();
    for id in ids {
        let permit = Arc::clone(&limit).acquire_owned().await.unwrap();   // waits for a free slot
        group.spawn(async move {
            let r = download(id).await;
            drop(permit);                          // frees the slot
            r
        });
    }
    let mut results = Vec::new();
    while let Some(r) = group.join_next().await {
        results.push(r.unwrap());
    }
    results
}

A Semaphore holds N permits; each task takes one and gives it back by dropping it. The 11th waits for a slot to free up.

use futures::stream::{self, StreamExt};

async fn load(id: u32) -> u32 { id * 2 }

async fn example() {
    // Like an iterator: map, filter, collect... but with .await
    let doubled: Vec<u32> = stream::iter(1..=5).then(load).collect().await;

    // At most 4 simultaneous loads, results in order of arrival
    let fast: Vec<u32> = stream::iter(1..=100)
        .map(load)
        .buffer_unordered(4)
        .collect()
        .await;

    // Go through them one by one
    let mut s = stream::iter(vec!["a", "b"]);
    while let Some(x) = s.next().await {
        println!("{x}");
    }
}

// 1. Transaction: nothing is committed until commit is reached
async fn safe_transfer(db: &sea_orm::DatabaseConnection) -> Result<(), sea_orm::DbErr> {
    use sea_orm::TransactionTrait;
    let txn = db.begin().await?;
    // debit(&txn).await?; credit(&txn).await?;
    txn.commit().await          // cancelled before: the transaction is rolled back on drop
}

// 2. Detach the work that must run to the end
async fn handler() {
    tokio::spawn(async {
        // keeps going even if the client disconnects
    });
}

use tokio::signal;

#[tokio::main]
async fn main() {
    let (stop_tx, _) = tokio::sync::broadcast::channel::<()>(1);

    let mut stop_rx = stop_tx.subscribe();
    let worker = tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
                    println!("working");
                }
                _ = stop_rx.recv() => {
                    println!("worker stops cleanly");
                    break;
                }
            }
        }
    });

    signal::ctrl_c().await.unwrap();          // wait for Ctrl+C (or SIGTERM with signal::unix)
    let _ = stop_tx.send(());                // notify all tasks
    worker.await.unwrap();                   // wait for them to finish
}

use futures::stream::{self, StreamExt};
use std::sync::Arc;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;

async fn profile(id: u32) -> String { format!("profile {id}") }

// 1. Sequential: simple, slow (20 x the latency)
async fn v1() -> Vec<String> {
    let mut r = Vec::new();
    for id in 1..=20 {
        r.push(profile(id).await);
    }
    r
}

// 2. join_all: everything at once in the same task, order kept, no limit
async fn v2() -> Vec<String> {
    futures::future::join_all((1..=20).map(profile)).await
}

// 3. JoinSet: separate tasks, on several threads, order of arrival
async fn v3() -> Vec<String> {
    let mut g = JoinSet::new();
    for id in 1..=20 {
        g.spawn(profile(id));
    }
    let mut r = Vec::new();
    while let Some(p) = g.join_next().await {
        r.push(p.unwrap());
    }
    r
}

// 4. Stream with a limit: at most 5 at a time, order kept
async fn v4() -> Vec<String> {
    stream::iter(1..=20).map(profile).buffered(5).collect().await
}

// 5. JoinSet + Semaphore: separate tasks and a limit
async fn v5() -> Vec<String> {
    let limit = Arc::new(Semaphore::new(5));
    let mut g = JoinSet::new();
    for id in 1..=20 {
        let l = Arc::clone(&limit);
        g.spawn(async move {
            let _p = l.acquire().await.unwrap();
            profile(id).await
        });
    }
    let mut r = Vec::new();
    while let Some(p) = g.join_next().await {
        r.push(p.unwrap());
    }
    r
}

// runique/src/middleware/session/session_db.rs (excerpt)
pub fn spawn_cleanup(&self, period: tokio::time::Duration) {
    let db = self.db.clone();
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(period);
        loop {
            interval.tick().await;
            let now = chrono::Utc::now().naive_utc();
            // deletes expired sessions; any error is logged
            // Entity::delete_many().filter(Column::ExpiresAt.lt(now)).exec(&*db).await
        }
    });
}

loop + interval in a detached task: the cleanup runs for the whole life of the server. The connection is cloned before the move (section 4.3), and errors are logged inside the task, since nobody waits for its result (section 4.2).

// runique/src/forms/fields/media.rs (excerpt)
let path = filename.to_string();
let read = tokio::task::spawn_blocking(move || read_image_dimensions(&path)).await;

Reading the dimensions of an uploaded image is a synchronous disk and decoding operation: it goes through spawn_blocking so it does not block the thread that handles other requests (section 5).

// runique/src/auth/password.rs (excerpt)
// Everything that depends on the account runs in the background, its
// lookup and the token's write included: the page answers the same, in
// the same time, for every address.
tokio::spawn(async move {
    let Some(user) = BuiltinUserEntity::find_by_email(&engine.db, &email).await else {
        return;
    };
    // ... token creation, email sending ...
});

On the "forgot password" page, if the response waited for the account lookup and the email sending, its duration would reveal whether the address exists (an unknown address would answer faster). By handing all this work to a detached task, the page answers immediately and the same way for every address. This is a protection against timing attacks, obtained with a simple spawn.

#[tokio::main] async fn main() {}                       #[tokio::test] async fn t() {}
let h = tokio::spawn(async move { .. }); h.await?;      tokio::task::spawn_blocking(|| ..).await?
tokio::join!(a(), b())     tokio::try_join!(a(), b())?   tokio::select! { x = a() => .., _ = b() => .. }
let mut g = JoinSet::new(); g.spawn(f); g.join_next().await
tokio::time::sleep(d).await     interval(d).tick().await  timeout(d, f).await
mpsc::channel(n)   oneshot::channel()   broadcast::channel(n)   watch::channel(v)
Semaphore::new(n).acquire().await     stream::iter(v).map(f).buffered(n).collect().await
signal::ctrl_c().await   axum::serve(..).with_graceful_shutdown(..)