Tokio Patterns: select!, JoinSet & Friends

Async Fundamentals covered the core: futures, .await, spawn, and join!. Real services need more: waiting for a dynamic number of tasks, racing a request against a timeout, limiting how many calls are in flight, running something every few seconds, and shutting all of it down cleanly. This lesson covers the Tokio patterns for each of these.

Task	Rust / Tokio
starting async work	`tokio::spawn(async { ... })`
dependent async steps	plain code and `.await` in sequence
waiting for a list of futures	`futures::future::join_all(futures)`
racing futures	`tokio::select!`, `futures::future::select_ok`
collecting results as tasks finish	`tokio::task::JoinSet`
structured concurrency	`JoinSet` (tasks are aborted when it's dropped)
periodic work	`tokio::time::interval`
limiting concurrency	`tokio::sync::Semaphore`
error handling	`Result` + `match` / `?`

1. Dependent async steps are just code

The biggest simplification: dependent async steps are ordinary sequential statements.

Rust

use std::time::Duration;

async fn fetch_user_id(email: &str) -> Result<u32, String> {
    tokio::time::sleep(Duration::from_millis(10)).await;
    if email.ends_with("@example.com") { Ok(42) } else { Err(format!("unknown {email}")) }
}

async fn fetch_orders(user_id: u32) -> Result<Vec<u32>, String> {
    tokio::time::sleep(Duration::from_millis(10)).await;
    Ok(vec![user_id * 10, user_id * 10 + 1])
}

async fn order_summary(email: &str) -> Result<String, String> {
    let id = fetch_user_id(email).await?; // thenCompose
    let orders = fetch_orders(id).await?; // thenCompose
    let label = format!("{} orders", orders.len()); // thenApply
    Ok(format!("{email}: {label} {orders:?}"))
}

#[tokio::main]
async fn main() {
    println!("{:?}", order_summary("ana@example.com").await);
    println!("{:?}", order_summary("bob@nowhere.org").await);
}

Output

Ok("ana@example.com: 2 orders [420, 421]")
Err("unknown bob@nowhere.org")

2. `join_all`: wait for a list of futures

tokio::join! needs a fixed number of futures written out in code. For a Vec of futures, use futures::future::join_all, which returns the results in input order.

Rust

use futures::future::join_all;
use std::time::{Duration, Instant};

async fn fetch_price(sku: &str) -> (String, u32) {
    let ms = 20 * sku.len() as u64; // longer names take longer
    tokio::time::sleep(Duration::from_millis(ms)).await;
    (sku.to_string(), sku.len() as u32 * 100)
}

#[tokio::main]
async fn main() {
    let skus = ["keyboard", "mouse", "monitor", "cable"];
    let start = Instant::now();

    let prices = join_all(skus.iter().map(|s| fetch_price(s))).await;

    println!("{prices:?}");
    println!("concurrent (under 250 ms): {}", start.elapsed() < Duration::from_millis(250));
}

Output

[("keyboard", 800), ("mouse", 500), ("monitor", 700), ("cable", 500)]
concurrent (under 250 ms): true

Like join!, join_all runs everything inside the current task. That's ideal for a few dozen I/O calls; for CPU work or thousands of futures, spawn tasks (JoinSet, below).

3. `try_join_all`: stop at the first error

try_join_all takes futures that return Result and fails fast: as soon as one returns Err, the others are dropped (cancelled) and the error is returned.

Rust

use futures::future::try_join_all;
use std::time::Duration;

async fn validate(file: &str) -> Result<usize, String> {
    tokio::time::sleep(Duration::from_millis(10)).await;
    if file.ends_with(".csv") { Ok(file.len()) } else { Err(format!("{file}: not a CSV file")) }
}

#[tokio::main]
async fn main() {
    let good = ["jan.csv", "feb.csv"];
    let bad = ["jan.csv", "notes.txt", "mar.csv"];

    println!("{:?}", try_join_all(good.iter().map(|f| validate(f))).await);
    println!("{:?}", try_join_all(bad.iter().map(|f| validate(f))).await);
}

Output

Ok([7, 7])
Err("notes.txt: not a CSV file")

4. `JoinSet`: a dynamic group of tasks

JoinSet is a collection of spawned tasks. join_next() returns results as tasks finish. Here the fastest mirror reports first.

Rust

use std::time::Duration;
use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for (mirror, ms) in [("eu", 150), ("us", 50), ("asia", 100)] {
        set.spawn(async move {
            tokio::time::sleep(Duration::from_millis(ms)).await;
            format!("{mirror} responded after {ms} ms")
        });
    }

    while let Some(result) = set.join_next().await {
        println!("{}", result.unwrap());
    }
}

Output

us responded after 50 ms
asia responded after 100 ms
eu responded after 150 ms

5. `JoinSet` is structured concurrency

When a JoinSet is dropped, all its unfinished tasks are aborted. Tasks can't leak past the scope that owns the set.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let finished = Arc::new(AtomicU32::new(0));

    {
        let mut set = JoinSet::new();
        for ms in [10, 500, 600] {
            let finished = Arc::clone(&finished);
            set.spawn(async move {
                tokio::time::sleep(Duration::from_millis(ms)).await;
                finished.fetch_add(1, Ordering::Relaxed);
            });
        }
        // Wait for just the first task, then leave the scope.
        set.join_next().await;
    } // `set` dropped: the two slow tasks are aborted

    tokio::time::sleep(Duration::from_millis(700)).await;
    println!("tasks that finished: {}", finished.load(Ordering::Relaxed));
}

Output

tasks that finished: 1

Compare tokio::spawn, whose tasks keep running after their handle is dropped. Prefer JoinSet when tasks belong to one operation.

6. `JoinSet` with errors: fail fast and cancel the rest

A common pattern: run a batch of tasks, and if one fails, cancel the others (abort_all) and report the error.

Rust

use std::time::Duration;
use tokio::task::JoinSet;

async fn upload(part: u32) -> Result<u32, String> {
    tokio::time::sleep(Duration::from_millis(20 * part as u64)).await;
    if part == 3 { Err(format!("part {part}: checksum mismatch")) } else { Ok(part) }
}

async fn upload_all(parts: u32) -> Result<Vec<u32>, String> {
    let mut set = JoinSet::new();
    for p in 1..=parts {
        set.spawn(upload(p));
    }
    let mut done = Vec::new();
    while let Some(joined) = set.join_next().await {
        match joined.expect("upload task panicked") {
            Ok(p) => done.push(p),
            Err(e) => {
                set.abort_all(); // cancel parts still uploading
                return Err(format!("{e} (after {} parts succeeded)", done.len()));
            }
        }
    }
    done.sort();
    Ok(done)
}

#[tokio::main]
async fn main() {
    println!("{:?}", upload_all(2).await);
    println!("{:?}", upload_all(6).await);
}

Output

Ok([1, 2])
Err("part 3: checksum mismatch (after 2 parts succeeded)")

7. `select!`: race futures, keep the winner

tokio::select! waits on several futures at once and runs the branch of whichever finishes first. The other futures are dropped, which cancels them.

Rust

use std::time::Duration;
use tokio::time::sleep;

async fn from_cache() -> &'static str {
    sleep(Duration::from_millis(20)).await;
    "cached profile"
}

async fn from_database() -> &'static str {
    sleep(Duration::from_millis(200)).await;
    "fresh profile"
}

#[tokio::main]
async fn main() {
    tokio::select! {
        profile = from_cache() => println!("cache won: {profile}"),
        profile = from_database() => println!("database won: {profile}"),
    }
}

Output

cache won: cached profile

8. `select!` as a timeout

A sleep branch turns select! into a timeout, with room for different handling per outcome. For a single future, tokio::time::timeout is shorter; select! shines when there are several things to wait for.

Rust

use std::time::Duration;
use tokio::time::sleep;

async fn generate_report(ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await;
    format!("report ({ms} ms)")
}

async fn report_or_placeholder(ms: u64) -> String {
    tokio::select! {
        report = generate_report(ms) => report,
        _ = sleep(Duration::from_millis(100)) => String::from("report is still being generated, check back later"),
    }
}

#[tokio::main]
async fn main() {
    println!("{}", report_or_placeholder(30).await);
    println!("{}", report_or_placeholder(500).await);
}

Output

report (30 ms)
report is still being generated, check back later

9. `select!` in a loop: an event loop

The most common use: a loop that waits for whichever event comes next (a message, a timer tick, a shutdown signal). Here a worker processes jobs from a channel, prints a heartbeat on a timer, and exits on a shutdown signal. (Channels are covered in detail in the next lesson.)

Rust

use std::time::Duration;
use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<u32>(16);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    let worker = tokio::spawn(async move {
        let mut heartbeat = tokio::time::interval(Duration::from_millis(40));
        let (mut processed, mut beats) = (0, 0);
        loop {
            tokio::select! {
                Some(job) = job_rx.recv() => processed += job,
                _ = heartbeat.tick() => beats += 1,
                _ = &mut stop_rx => break,
            }
        }
        (processed, beats > 0)
    });

    for job in 1..=5 {
        job_tx.send(job).await.unwrap();
    }
    tokio::time::sleep(Duration::from_millis(100)).await;
    stop_tx.send(()).unwrap();

    let (processed, had_heartbeats) = worker.await.unwrap();
    println!("processed sum: {processed}, heartbeats seen: {had_heartbeats}");
}

Output

processed sum: 15, heartbeats seen: true

Two details: the Some(job) = ... pattern disables that branch if the channel closes (the pattern doesn't match None), and &mut stop_rx polls the receiver by reference so it isn't moved into the first loop iteration.

10. `biased;`: fixed priority

By default select! picks randomly among branches that are ready at the same time, so no branch starves. biased; checks branches top to bottom, which is useful when one should always win, like a shutdown signal.

Rust

#[tokio::main]
async fn main() {
    let mut picks = Vec::new();
    for _ in 0..5 {
        // Both branches are immediately ready.
        let pick = tokio::select! {
            biased;
            _ = async {} => "first",
            _ = async {} => "second",
        };
        picks.push(pick);
    }
    println!("{picks:?}");
}

Output

["first", "first", "first", "first", "first"]

11. A `select!` branch with an `else`

If every branch's pattern fails to match (for example, all channels are closed), select! runs the else branch. Without an else, it would panic.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (fast_tx, mut fast) = mpsc::channel(4);
    let (slow_tx, mut slow) = mpsc::channel(4);
    fast_tx.send("f1").await.unwrap();
    slow_tx.send("s1").await.unwrap();
    drop((fast_tx, slow_tx)); // close both channels

    let mut received = Vec::new();
    loop {
        tokio::select! {
            Some(m) = fast.recv() => received.push(m),
            Some(m) = slow.recv() => received.push(m),
            else => break, // both channels closed and drained
        }
    }
    received.sort();
    println!("{received:?}");
}

Output

["f1", "s1"]

12. `select_ok`: the first successful result

select! returns whichever future finishes first, even if it failed. futures::future::select_ok returns the first Ok, and only fails if all of them fail. Good for querying redundant replicas.

Rust

use futures::future::select_ok;
use std::time::Duration;

async fn query_replica(name: &'static str, ms: u64, healthy: bool) -> Result<String, String> {
    tokio::time::sleep(Duration::from_millis(ms)).await;
    if healthy { Ok(format!("{name}: 42 rows")) } else { Err(format!("{name}: connection refused")) }
}

#[tokio::main]
async fn main() {
    // The fastest replica fails; the next fastest healthy one wins.
    let replicas = vec![
        Box::pin(query_replica("replica-a", 10, false)),
        Box::pin(query_replica("replica-b", 50, true)),
        Box::pin(query_replica("replica-c", 90, true)),
    ];
    match select_ok(replicas).await {
        Ok((rows, _still_running)) => println!("{rows}"),
        Err(last_error) => println!("all failed, last: {last_error}"),
    }

    let all_down = vec![Box::pin(query_replica("a", 10, false)), Box::pin(query_replica("b", 20, false))];
    println!("{:?}", select_ok(all_down).await.map(|(r, _)| r));
}

Output

replica-b: 42 rows
Err("b: connection refused")

select_ok needs futures of one type that can be moved around, so each is wrapped in Box::pin.

13. Periodic work with `interval`

tokio::time::interval(period) yields a tick every period. The first tick completes immediately.

Rust

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

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut ticker = interval(Duration::from_millis(50));

    let mut ticks = 0;
    for _ in 1..=4 {
        ticker.tick().await;
        ticks += 1;
    }
    // Ticks at 0, 50, 100 and 150 ms.
    println!("{ticks} ticks, the last one at least 150 ms after the first: {}", start.elapsed().as_millis() >= 150);
}

Output

4 ticks, the last one at least 150 ms after the first: true

If a tick is late because the task was busy, interval by default fires the missed ticks immediately to catch up (MissedTickBehavior::Burst). ticker.set_missed_tick_behavior(MissedTickBehavior::Skip) or Delay changes that.

14. An overall deadline across steps

A per-call timeout doesn't bound the total time of several steps. Compute one deadline up front and give each step what's left of it with timeout_at.

Rust

use std::time::Duration;
use tokio::time::{Instant, sleep, timeout_at};

async fn step(name: &str, ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await;
    format!("{name} done")
}

async fn checkout(budget_ms: u64) -> Result<Vec<String>, String> {
    let deadline = Instant::now() + Duration::from_millis(budget_ms);
    let mut log = Vec::new();
    for (name, ms) in [("reserve stock", 40), ("charge card", 60), ("send receipt", 40)] {
        let result = timeout_at(deadline, step(name, ms)).await;
        log.push(result.map_err(|_| format!("deadline exceeded during {name:?} (completed: {log:?})"))?);
    }
    Ok(log)
}

#[tokio::main]
async fn main() {
    println!("{:?}", checkout(300).await);
    println!("{:?}", checkout(120).await);
}

Output

Ok(["reserve stock done", "charge card done", "send receipt done"])
Err("deadline exceeded during \"send receipt\" (completed: [\"reserve stock done\", \"charge card done\"])")

15. Limiting concurrency with a `Semaphore`

Spawning 1,000 requests at once can overwhelm the service you call. A Semaphore with N permits allows at most N tasks into a section at a time. The permit is released when it's dropped.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let limit = Arc::new(Semaphore::new(3));
    let in_flight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));

    let mut set = JoinSet::new();
    for id in 0..20 {
        let (limit, in_flight, peak) = (Arc::clone(&limit), Arc::clone(&in_flight), Arc::clone(&peak));
        set.spawn(async move {
            let _permit = limit.acquire().await.unwrap(); // waits if 3 are already running
            let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
            peak.fetch_max(now, Ordering::SeqCst);
            tokio::time::sleep(Duration::from_millis(10)).await; // the rate-limited call
            in_flight.fetch_sub(1, Ordering::SeqCst);
            id
        });
    }

    let mut finished = 0;
    while set.join_next().await.is_some() {
        finished += 1;
    }
    println!("finished {finished} calls, never more than {} at once", peak.load(Ordering::SeqCst));
}

Output

finished 20 calls, never more than 3 at once

16. `acquire_owned` for permits that move into tasks

acquire() borrows the semaphore, so the permit can't outlive the Arc reference in scope. acquire_owned() takes an Arc<Semaphore> and returns an owned permit that can move into a spawned task. Acquiring before spawning also means you never create more tasks than permits.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    let pool = Arc::new(Semaphore::new(2)); // e.g. 2 database connections
    let mut handles = Vec::new();

    for query in ["q1", "q2", "q3", "q4"] {
        let permit = Arc::clone(&pool).acquire_owned().await.unwrap(); // waits here, not inside the task
        handles.push(tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(20)).await;
            drop(permit); // connection returned
            format!("{query} ok")
        }));
    }

    for h in handles {
        println!("{}", h.await.unwrap());
    }
    println!("permits available again: {}", pool.available_permits());
}

Output

q1 ok
q2 ok
q3 ok
q4 ok
permits available again: 2

17. Hedged requests

To cut tail latency, send a request; if it hasn't answered within a short delay, send a second copy to another server and take whichever answers first. select! with a delayed second branch expresses this directly.

Rust

use std::time::Duration;
use tokio::time::sleep;

async fn call(server: &str, ms: u64) -> String {
    sleep(Duration::from_millis(ms)).await;
    format!("answer from {server} in {ms} ms")
}

async fn hedged(primary_ms: u64, backup_ms: u64) -> String {
    let hedge_after = Duration::from_millis(50);
    tokio::select! {
        a = call("primary", primary_ms) => a,
        b = async {
            sleep(hedge_after).await; // only start the backup if the primary is slow
            call("backup", backup_ms).await
        } => format!("{b} (hedged)"),
    }
}

#[tokio::main]
async fn main() {
    println!("{}", hedged(20, 30).await); // primary is fast: no hedge needed
    println!("{}", hedged(400, 30).await); // primary is slow: backup wins
}

Output

answer from primary in 20 ms
answer from backup in 30 ms (hedged)

18. Non-`Send` futures can't be spawned

tokio::spawn may move a task between worker threads, so the future must be Send. Holding an Rc (or a RefCell borrow, or a std::sync::MutexGuard) across an .await makes the future non-Send:

Rust✗ does not compile

use std::rc::Rc;

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        let cache = Rc::new(vec![1, 2, 3]);
        tokio::task::yield_now().await; // `cache` is alive across this await
        cache.len()
    });
    println!("{}", handle.await.unwrap());
}

Output

error: future cannot be sent between threads safely
   --> src/main.rs:5:18
    |
  5 |       let handle = tokio::spawn(async {
    |  __________________^
  6 | |         let cache = Rc::new(vec![1, 2, 3]);
  7 | |         tokio::task::yield_now().await; // `cache` is alive across this await
  8 | |         cache.len()
  9 | |     });
    | |______^ future created by async block is not `Send`
    |
    = help: within `{async block@src/main.rs:5:31: 5:36}`, the trait `Send` is not implemented for `Rc<Vec<i32>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:7:34
    |
  6 |         let cache = Rc::new(vec![1, 2, 3]);
    |             ----- has type `Rc<Vec<i32>>` which is not `Send`
  7 |         tokio::task::yield_now().await; // `cache` is alive across this await
    |                                  ^^^^^ await occurs here, with `cache` maybe used later
note: required by a bound in `tokio::spawn`
   --> ~/.cargo/registry/src/.../tokio-1.53.1/src/task/spawn.rs:176:21
    |
174 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
    |            ----- required by a bound in this function
175 |     where
176 |         F: Future + Send + 'static,
    |                     ^^^^ required by this bound in `spawn`

Fixes, in order of preference: use Arc instead of Rc; make sure the value is dropped before the .await (put it in an inner block); or run the task on a LocalSet (next example).

19. `LocalSet` for tasks that stay on one thread

spawn_local runs a task on the current thread only, so it may hold non-Send values. It must run inside a LocalSet. Useful for libraries that use Rc internally, or single-threaded designs.

Rust

use std::cell::RefCell;
use std::rc::Rc;
use tokio::task::LocalSet;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let history = Rc::new(RefCell::new(Vec::new()));
    let local = LocalSet::new();

    local
        .run_until(async {
            let mut handles = Vec::new();
            for page in ["/home", "/cart", "/checkout"] {
                let history = Rc::clone(&history);
                handles.push(tokio::task::spawn_local(async move {
                    tokio::task::yield_now().await; // Rc held across .await: fine on a LocalSet
                    history.borrow_mut().push(page);
                }));
            }
            for h in handles {
                h.await.unwrap();
            }
        })
        .await;

    println!("{:?}", history.borrow());
}

Output

["/checkout", "/cart", "/home"]

All three tasks run on the current thread, so sharing an Rc<RefCell<...>> is safe. (The order they ran in is the LocalSet scheduler's choice, not the spawn order.)

20. Background task with a final result

A long-lived background task (a cache refresher, a metrics flusher) often returns a summary when it's asked to stop. Keep its JoinHandle, signal it, and await the handle.

Rust

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

async fn refresher(mut stop: oneshot::Receiver<()>) -> u32 {
    let mut refreshes = 0;
    let mut every = tokio::time::interval(Duration::from_millis(20));
    loop {
        tokio::select! {
            _ = every.tick() => refreshes += 1, // refresh the cache
            _ = &mut stop => return refreshes,
        }
    }
}

#[tokio::main]
async fn main() {
    let (stop_tx, stop_rx) = oneshot::channel();
    let task = tokio::spawn(refresher(stop_rx));

    tokio::time::sleep(Duration::from_millis(110)).await; // the app runs...
    stop_tx.send(()).unwrap();

    let refreshes = task.await.unwrap();
    println!("refresher stopped after a few refreshes: {}", (3..=8).contains(&refreshes));
}

Output

refresher stopped after a few refreshes: true

21. `JoinError` vs your own errors

Awaiting a JoinHandle<Result<T, E>> gives Result<Result<T, E>, JoinError>: the outer error means the task panicked or was aborted, the inner one is your application error. Flatten them into one error type at the boundary.

Rust

use tokio::task::JoinError;

#[derive(Debug)]
enum JobError {
    Failed(String),
    Crashed(String),
}

impl From<JoinError> for JobError {
    fn from(e: JoinError) -> Self {
        JobError::Crashed(if e.is_panic() { String::from("task panicked") } else { String::from("task cancelled") })
    }
}

async fn run_job(input: i32) -> Result<i32, JobError> {
    let handle = tokio::spawn(async move {
        if input < 0 {
            return Err(JobError::Failed(format!("negative input {input}")));
        }
        if input == 0 {
            panic!("division by zero");
        }
        Ok(100 / input)
    });
    handle.await? // `?` converts JoinError; the inner Result is returned as is
}

#[tokio::main]
async fn main() {
    for input in [4, -1, 0] {
        println!("{input}: {:?}", run_job(input).await);
    }
}

Output

4: Ok(25)
-1: Err(Failed("negative input -1"))
0: Err(Crashed("task panicked"))

22. Scatter-gather with partial results

Ask many sources, wait up to a deadline, and use whatever arrived. JoinSet plus timeout around join_next does it; tasks still running at the deadline are aborted when the set is dropped.

Rust

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{Instant, timeout_at};

async fn quote(airline: &'static str, ms: u64, price: u32) -> (&'static str, u32) {
    tokio::time::sleep(Duration::from_millis(ms)).await;
    (airline, price)
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    set.spawn(quote("AirOne", 30, 420));
    set.spawn(quote("SkyJet", 60, 380));
    set.spawn(quote("SlowAir", 900, 250)); // cheapest, but too slow

    let deadline = Instant::now() + Duration::from_millis(200);
    let mut quotes = Vec::new();
    while let Ok(Some(res)) = timeout_at(deadline, set.join_next()).await {
        quotes.push(res.unwrap());
    }
    quotes.sort_by_key(|&(_, price)| price);

    println!("quotes in time: {quotes:?}");
    println!("best: {:?}", quotes.first());
    println!("still running (will be aborted): {}", set.len());
}

Output

quotes in time: [("SkyJet", 380), ("AirOne", 420)]
best: Some(("SkyJet", 380))
still running (will be aborted): 1

23. Rate limiting with an interval

To send at most N requests per second, wait for an interval tick before each request. Unlike a Semaphore (which limits requests in flight), this limits the rate.

Rust

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

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut limiter = interval(Duration::from_millis(25)); // 40 requests/second
    limiter.set_missed_tick_behavior(MissedTickBehavior::Delay);

    let mut sent = Vec::new();
    for id in 1..=5 {
        limiter.tick().await;
        sent.push(id); // send request `id`
    }

    let elapsed = start.elapsed();
    println!("sent {sent:?}");
    println!("5 requests spread over at least 100 ms: {}", elapsed >= Duration::from_millis(100));
}

Output

sent [1, 2, 3, 4, 5]
5 requests spread over at least 100 ms: true

24. Realistic: a concurrent crawler

A crawler over a fake in-memory site: start from one page, fetch linked pages concurrently with a JoinSet, cap concurrency with a Semaphore, and track visited pages so each is fetched once.

Rust

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;

type Site = HashMap<&'static str, Vec<&'static str>>;

async fn fetch(site: &Site, url: &'static str) -> Vec<&'static str> {
    tokio::time::sleep(Duration::from_millis(10)).await; // network
    site.get(url).cloned().unwrap_or_default()
}

async fn crawl(site: Arc<Site>, start: &'static str, max_concurrent: usize) -> Vec<&'static str> {
    let limit = Arc::new(Semaphore::new(max_concurrent));
    let mut visited = HashSet::from([start]);
    let mut set = JoinSet::new();

    let spawn_fetch = |set: &mut JoinSet<_>, url: &'static str| {
        let (site, limit) = (Arc::clone(&site), Arc::clone(&limit));
        set.spawn(async move {
            let _permit = limit.acquire().await.unwrap();
            fetch(&site, url).await
        });
    };

    spawn_fetch(&mut set, start);
    while let Some(links) = set.join_next().await {
        for link in links.unwrap() {
            if visited.insert(link) {
                spawn_fetch(&mut set, link);
            }
        }
    }

    let mut pages: Vec<_> = visited.into_iter().collect();
    pages.sort();
    pages
}

#[tokio::main]
async fn main() {
    let site: Site = HashMap::from([
        ("/", vec!["/about", "/blog", "/shop"]),
        ("/blog", vec!["/blog/rust", "/blog/java", "/"]),
        ("/blog/rust", vec!["/blog/async", "/shop"]),
        ("/shop", vec!["/shop/cart", "/about"]),
        ("/shop/cart", vec!["/checkout"]),
    ]);
    let pages = crawl(Arc::new(site), "/", 3).await;
    println!("{} pages: {pages:?}", pages.len());
}

Output

9 pages: ["/", "/about", "/blog", "/blog/async", "/blog/java", "/blog/rust", "/checkout", "/shop", "/shop/cart"]

25. Realistic: health checks with per-service timeouts

Check every dependency concurrently, each with its own timeout, and produce a report. A slow service shows up as "timeout" instead of stalling the whole check.

Rust

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

#[derive(Debug)]
enum Health {
    Up(u64),
    Down(String),
    Timeout,
}

async fn probe(latency_ms: u64, ok: bool) -> Result<u64, String> {
    sleep(Duration::from_millis(latency_ms)).await;
    if ok { Ok(latency_ms) } else { Err(String::from("HTTP 503")) }
}

async fn check(name: &'static str, latency_ms: u64, ok: bool) -> (&'static str, Health) {
    let health = match timeout(Duration::from_millis(100), probe(latency_ms, ok)).await {
        Ok(Ok(ms)) => Health::Up(ms),
        Ok(Err(e)) => Health::Down(e),
        Err(_) => Health::Timeout,
    };
    (name, health)
}

#[tokio::main]
async fn main() {
    let report = join_all([
        check("postgres", 12, true),
        check("redis", 3, true),
        check("payments-api", 40, false),
        check("search", 800, true),
    ])
    .await;

    for (name, health) in &report {
        println!("{name:<13} {health:?}");
    }
    let all_up = report.iter().all(|(_, h)| matches!(h, Health::Up(_)));
    println!("overall: {}", if all_up { "healthy" } else { "degraded" });
}

Output

postgres      Up(12)
redis         Up(3)
payments-api  Down("HTTP 503")
search        Timeout
overall: degraded

26. Realistic: a cache with request coalescing

When many requests ask for the same missing key at once, only one should hit the backend; the others wait for its result. A tokio::sync::OnceCell per key does this: the first caller runs the initializer, everyone else awaits it.

Rust

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use tokio::sync::{Mutex, OnceCell};

#[derive(Default)]
struct Cache {
    entries: Mutex<HashMap<String, Arc<OnceCell<String>>>>,
    backend_calls: AtomicU32,
}

impl Cache {
    async fn get(&self, key: &str) -> String {
        let cell = {
            let mut entries = self.entries.lock().await;
            Arc::clone(entries.entry(key.to_string()).or_default())
        }; // map lock released before the slow call
        cell.get_or_init(|| async {
            self.backend_calls.fetch_add(1, Ordering::Relaxed);
            tokio::time::sleep(Duration::from_millis(50)).await; // slow backend
            format!("value-of-{key}")
        })
        .await
        .clone()
    }
}

#[tokio::main]
async fn main() {
    let cache = Arc::new(Cache::default());
    let mut handles = Vec::new();
    for i in 0..10 {
        let cache = Arc::clone(&cache);
        let key = if i % 2 == 0 { "user:1" } else { "user:2" };
        handles.push(tokio::spawn(async move { cache.get(key).await }));
    }
    let mut values: Vec<String> = Vec::new();
    for h in handles {
        values.push(h.await.unwrap());
    }
    println!("10 requests, backend calls: {}", cache.backend_calls.load(Ordering::Relaxed));
    println!("first values: {:?}", &values[..2]);
}

Output

10 requests, backend calls: 2
first values: ["value-of-user:1", "value-of-user:2"]

The map is protected by tokio::sync::Mutex, the async-aware mutex from the next lesson.

27. Realistic: graceful shutdown of a worker pool

A pool of workers processes jobs until told to stop. On shutdown, workers finish their current job, then exit; the coordinator waits for all of them with a JoinSet. A watch channel broadcasts the shutdown flag to every worker.

Rust

use std::time::Duration;
use tokio::sync::{mpsc, watch};
use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u32>(100);
    let job_rx = std::sync::Arc::new(tokio::sync::Mutex::new(job_rx));
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    let mut workers = JoinSet::new();
    for id in 0..3 {
        let (job_rx, mut shutdown) = (std::sync::Arc::clone(&job_rx), shutdown_rx.clone());
        workers.spawn(async move {
            let mut done = 0;
            loop {
                let job = tokio::select! {
                    biased;
                    _ = shutdown.changed() => break,
                    job = async { job_rx.lock().await.recv().await } => job,
                };
                let Some(job) = job else { break };
                tokio::time::sleep(Duration::from_millis(job as u64)).await; // finish the current job
                done += 1;
            }
            (id, done)
        });
    }

    for ms in [5, 10, 5, 10, 5, 10] {
        job_tx.send(ms).await.unwrap();
    }
    tokio::time::sleep(Duration::from_millis(100)).await;
    shutdown_tx.send(true).unwrap();

    let mut total = 0;
    while let Some(res) = workers.join_next().await {
        total += res.unwrap().1;
    }
    println!("all workers stopped; jobs completed: {total}");
}

Output

all workers stopped; jobs completed: 6

Summary

Dependent async steps are sequential .awaits; ? propagates errors.
join_all / try_join_all wait for a list of futures in one task; JoinSet spawns a group of tasks, yields results as they finish, and aborts leftovers when dropped.
select! races futures and cancels the losers: use it for timeouts, event loops, shutdown, and hedging. biased; fixes the order; else handles "all branches disabled".
select_ok returns the first success among several.
interval runs periodic work; timeout_at with one deadline bounds several steps.
Semaphore limits concurrency; an interval limits rate.
Futures given to tokio::spawn must be Send; use Arc, drop non-Send values before .await, or use LocalSet/spawn_local.

Next: Async Channels & Locks: Tokio's mpsc, oneshot, broadcast, and watch channels, and when to use tokio::sync::Mutex.

1. Dependent async steps are just code

2. join_all: wait for a list of futures

3. try_join_all: stop at the first error

4. JoinSet: a dynamic group of tasks

5. JoinSet is structured concurrency

6. JoinSet with errors: fail fast and cancel the rest

7. select!: race futures, keep the winner

8. select! as a timeout

9. select! in a loop: an event loop

10. biased;: fixed priority

11. A select! branch with an else

12. select_ok: the first *successful* result

13. Periodic work with interval

14. An overall deadline across steps

15. Limiting concurrency with a Semaphore

16. acquire_owned for permits that move into tasks

17. Hedged requests

18. Non-Send futures can't be spawned

19. LocalSet for tasks that stay on one thread

20. Background task with a final result

21. JoinError vs your own errors

22. Scatter-gather with partial results

23. Rate limiting with an interval

24. Realistic: a concurrent crawler

25. Realistic: health checks with per-service timeouts

26. Realistic: a cache with request coalescing

27. Realistic: graceful shutdown of a worker pool

Summary

2. `join_all`: wait for a list of futures

3. `try_join_all`: stop at the first error

4. `JoinSet`: a dynamic group of tasks

5. `JoinSet` is structured concurrency

6. `JoinSet` with errors: fail fast and cancel the rest

7. `select!`: race futures, keep the winner

8. `select!` as a timeout

9. `select!` in a loop: an event loop

10. `biased;`: fixed priority

11. A `select!` branch with an `else`

12. `select_ok`: the first successful result

13. Periodic work with `interval`

15. Limiting concurrency with a `Semaphore`

16. `acquire_owned` for permits that move into tasks

18. Non-`Send` futures can't be spawned

19. `LocalSet` for tasks that stay on one thread

21. `JoinError` vs your own errors