select!: Racing Futures

tokio::select! waits on several futures at once, runs the handler of the first one that completes, and drops the rest. This lesson covers its syntax, how to use it in a loop, and the rule that decides whether dropping a future loses data: cancel safety. Examples in this lesson are not machine-checked (they are marked rust,ignore); their output is derived from the timelines in the text.

Examples use tokio::time::pause(), which makes the clock jump forward whenever every task is idle, so sleeps finish at once and in a fixed order (explained in Time). They also use mpsc::channel, a queue that sends values between tasks, whose recv waits for the next value and returns None after all senders are dropped (explained in mpsc Channels).

Example 1: the smallest race

Each branch has the form pattern = future => handler. select! polls all the futures on the current task. When one completes, its handler runs and select! evaluates to the handler's value. The other futures are dropped, which cancels them (Cancellation).

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

struct Noisy(&'static str);

impl Drop for Noisy {
    fn drop(&mut self) {
        println!("dropped {}", self.0);
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    tokio::time::pause();
    let winner = tokio::select! {
        _ = async {
            let _guard = Noisy("slow");
            sleep(Duration::from_secs(5)).await;
        } => "slow",
        _ = async {
            let _guard = Noisy("fast");
            sleep(Duration::from_secs(1)).await;
        } => "fast",
    };
    println!("winner: {winner}");
}

dropped fast
dropped slow
winner: fast

The fast block finishes and drops its own guard. select! then drops the slow block, which drops its guard while it is suspended at the .await. All handlers must produce the same type, because select! is one expression.

All branches run inside one task, one at a time as each is polled. select! gives concurrency, not parallelism. Work that needs its own thread of execution goes into tokio::spawn.

Example 2: patterns, else, and biased

The pattern on the left can refuse a value. If the future's output does not match the pattern, that branch is disabled for the rest of this select! call and the other branches keep waiting. When every branch is disabled, select! runs the else branch. Without an else branch, that situation panics.

By default, when several branches are ready at the same moment, select! picks one at random, so no branch starves the others. Adding biased; as the first line makes it check the branches from top to bottom.

use tokio::sync::mpsc;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx_a, mut rx_a) = mpsc::channel::<u32>(4);
    let (tx_b, mut rx_b) = mpsc::channel::<&str>(4);
    tx_a.send(1).await.unwrap();
    tx_a.send(2).await.unwrap();
    tx_b.send("x").await.unwrap();
    drop(tx_a);
    drop(tx_b);

    loop {
        tokio::select! {
            biased;
            Some(n) = rx_a.recv() => println!("a: {n}"),
            Some(s) = rx_b.recv() => println!("b: {s}"),
            else => {
                println!("both closed");
                break;
            }
        }
    }
}

a: 1
a: 2
b: x
both closed

With biased;, channel a is drained before b. Once rx_a.recv() returns None, the pattern Some(n) fails and that branch is disabled for the call. The same happens to b, and then else runs.

A precondition adds , if condition after the future: Some(n) = rx.recv(), if open => .... The condition is evaluated before polling. When it is false, the branch is disabled for that call and its future is not polled. Preconditions switch branches off after their work is done, as with a flag that records "this channel closed".

Example 3: loops, &mut futures, and pin!

A select! in a loop creates its branch futures again on every iteration. A sleep(...) written inside the macro starts a new countdown each time, so the deadline never arrives while other branches keep firing. To keep one deadline across iterations, create the future once, pin it, and pass &mut to the branch.

tokio::pin! pins a value to the stack, a requirement for polling a future by reference (Pin is explained in Polling, Wakers, and a Tiny Executor).

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

#[tokio::main(flavor = "current_thread")]
async fn main() {
    tokio::time::pause();
    let (tx, mut rx) = mpsc::channel::<u32>(8);
    tokio::spawn(async move {
        for n in 1..=3 {
            tx.send(n).await.unwrap();
            sleep(Duration::from_millis(100)).await;
        }
    });

    let deadline = sleep(Duration::from_millis(250));
    tokio::pin!(deadline);

    loop {
        tokio::select! {
            Some(n) = rx.recv() => println!("got {n}"),
            _ = &mut deadline => {
                println!("deadline reached");
                break;
            }
        }
    }
}

got 1
got 2
got 3
deadline reached

The sender sends at 0, 100, and 200 ms. The deadline fires at 250 ms, measured from the single sleep call before the loop. When the deadline branch runs, the handler leaves the loop, so the finished sleep is never polled again.

A JoinHandle needs no pin. It can be raced with &mut handle and awaited later if another branch wins.

Example 4: a cancel-safety bug and its fix

A future is cancel safe when dropping it before completion loses no data and leaves no half-finished state. The state that lives inside a dropped future is gone. In a loop or a retry, that matters for any future that has already consumed input.

read_pair below takes two values from a channel. When the timeout wins after the first value arrived, the first value is dropped with the future.

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

async fn read_pair(rx: &mut mpsc::Receiver<u32>) -> Option<(u32, u32)> {
    let a = rx.recv().await?;
    let b = rx.recv().await?;
    Some((a, b))
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    tokio::time::pause();
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        tx.send(1).await.unwrap();
        sleep(Duration::from_millis(100)).await;
        tx.send(2).await.unwrap();
    });

    tokio::select! {
        pair = read_pair(&mut rx) => println!("pair: {pair:?}"),
        _ = sleep(Duration::from_millis(60)) => println!("timeout"),
    }
    println!("next: {:?}", rx.recv().await);
}

timeout
next: Some(2)

The value 1 was received, held in the local a, and dropped. Nothing tells the program it is gone. The fix moves the partial state to the caller, which owns it across attempts:

async fn read_pair(
    rx: &mut mpsc::Receiver<u32>,
    first: &mut Option<u32>,
) -> Option<(u32, u32)> {
    if first.is_none() {
        *first = Some(rx.recv().await?);
    }
    let b = rx.recv().await?;
    Some((first.take()?, b))
}

// in main:
let mut first = None;
loop {
    tokio::select! {
        pair = read_pair(&mut rx, &mut first) => {
            println!("pair: {pair:?}");
            break;
        }
        _ = sleep(Duration::from_millis(60)) => println!("timeout"),
    }
}

timeout
pair: Some((1, 2))

The first attempt stores 1 in first before it is dropped. The second attempt starts from the stored value, and 2 arrives at 100 ms, before the second 60 ms timeout ends at 120 ms. Cancel-safe futures keep no such state. mpsc::Receiver::recv is one of them, so recv alone can go in any select! branch.

Cancel safety of common APIs

The statements below come from the "Cancel safety" sections of the Tokio 1.x documentation (tokio 1.53 source) and of tokio-stream 0.1. "Safe" means the docs say that if another branch wins, no data was consumed or lost. A method marked "queue place" is safe for data, but its docs say that cancelling loses the caller's position in the fair queue.

The table has gaps on purpose. A method that is missing has no statement in the docs, so treat it as unverified until the docs or the source say otherwise. A sleep or timeout holds no data. Dropping one only cancels the timer.

Two rules follow from the table. Methods that read a fixed number of bytes into a buffer usually keep no state between polls, so a partial read is lost. Methods that only wait for a value or a slot in a queue are safe. For a not-safe method, run it outside select! (in its own task, with the result sent over a channel), or move its state into the caller as in Example 4.

Example 5: timeouts and join!

A timeout is a race between the work and a sleep. The branch order does not matter here, because only one can finish first.

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

async fn fetch() -> &'static str {
    sleep(Duration::from_secs(2)).await;
    "data"
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    tokio::time::pause();
    let result = tokio::select! {
        value = fetch() => Ok(value),
        _ = sleep(Duration::from_secs(1)) => Err("timed out"),
    };
    println!("{result:?}");

    let (a, b) = tokio::join!(fetch(), fetch());
    println!("{a} {b}");
}

Err("timed out")
data data

tokio::time::timeout(duration, future) expresses the same race and returns a Result with an Elapsed error (Time). Write the select! form when the timeout is one of several events, as in Example 3.

Interview points

• select! polls all branches on one task and cancels the losers by dropping them.
• A dropped future loses its local state. A future is cancel safe when that loss is harmless.
• recv on mpsc, broadcast, and oneshot, accept, and tick are cancel safe. read_exact, write_all, read_line, and multi-byte read_u32 are not.
• mpsc::Sender::send is cancel safe in the sense that nothing is sent, but the value is dropped. reserve avoids that.
• With no biased;, ready branches are picked at random. biased; gives a fixed top-to-bottom order, which can starve later branches.
• In a loop, create long-lived futures (a deadline, a JoinHandle) once and pass &mut (pinned with tokio::pin! if the type is not Unpin).
• A false precondition or a failed pattern disables a branch for that call. If all branches are disabled and there is no else, select! panics.

Summary

• select! returns the value of the first handler to run and drops every other branch.
• Branches are pattern = future, if precondition => handler. else handles the case where every branch is disabled.
• biased; switches from random choice to top-to-bottom order.
• Reuse a future across loop iterations with &mut future after tokio::pin!.
• Cancel safety is a property of each future. The table lists the documented status of common Tokio APIs.
• Keep state that must survive cancellation outside the future.
• select! races futures and cancels the rest. join! waits for all of them.