Rust Async Programming: Parallelism with join! vs select!

#webdev #programming #rust #backend

When executing only one Future, you can directly use .await inside an async function async fn or an async code block async {}. However, when multiple Futures need to be executed concurrently, directly using .await will block concurrent tasks until a specific Future completes—effectively executing them serially. The futures crate provides many useful tools for executing Futures concurrently, such as the join! and select! macros.

Note: The futures::future module provides a range of functions for operating on Futures (much more comprehensive than the macros). See:

https://docs.rs/futures/latest/futures/index.html#modules

https://docs.rs/futures/latest/futures/future/index.html#functions

The `join!` Macro

The join! macro allows waiting for the completion of multiple different Futures simultaneously and can execute them concurrently.

Let’s first look at two incorrect examples using .await:

struct Book;
struct Music;

async fn enjoy_book() -> Book { /* ... */ Book }
async fn enjoy_music() -> Music { /* ... */ Music }

// Incorrect version 1: Executes tasks sequentially inside the async function instead of concurrently
async fn enjoy1_book_and_music() -> (Book, Music) {
    // Actually executes sequentially inside the async function
    let book = enjoy_book().await; // await triggers blocking execution
    let music = enjoy_music().await; // await triggers blocking execution
    (book, music)
}

// Incorrect version 2: Also sequential execution inside the async function instead of concurrently
async fn enjoy2_book_and_music() -> (Book, Music) {
    // Actually executes sequentially inside the async function
    let book_future = enjoy_book(); // async functions are lazy and don't execute immediately
    let music_future = enjoy_music(); // async functions are lazy and don't execute immediately
    (book_future.await, music_future.await)
}

The two examples above may appear to execute asynchronously, but in fact, you must finish reading the book before you can listen to the music. That is, the tasks inside the async function are executed sequentially (one after the other), not concurrently.

This is because in Rust, Futures are lazy—they only start running when .await is called. And because the two await calls occur in order in the code, they are executed sequentially.

To correctly execute two Futures concurrently, let’s try the futures::join! macro:

use futures::join;

// Using `join!` returns a tuple containing the values output by each Future once it completes.
async fn enjoy_book_and_music() -> (Book, Music) {
    let book_fut = enjoy_book();
    let music_fut = enjoy_music();
    // The join! macro must wait until all managed Futures are completed before it itself completes
    join!(book_fut, music_fut)
}

fn main() {
    futures::executor::block_on(enjoy_book_and_music());
}

If you want to run multiple async tasks in an array concurrently, you can use the futures::future::join_all method.

The `try_join!` Macro

Since join! must wait until all of the Futures it manages have completed, if you want to stop the execution of all Futures immediately when any one of them fails, you can use try_join!—especially useful when the Futures return Result.

Note: All Futures passed to try_join! must have the same error type. If the error types differ, you can use the map_err and err_into methods from the futures::future::TryFutureExt module to convert the errors:

use futures::{
    future::TryFutureExt,
    try_join,
};

struct Book;
struct Music;

async fn get_book() -> Result<Book, ()> { /* ... */ Ok(Book) }
async fn get_music() -> Result<Music, String> { /* ... */ Ok(Music) }

/**
 * All Futures passed to try_join! must have the same error type.
 * If the error types differ, consider using map_err or err_into
 * from the futures::future::TryFutureExt module to convert them.
 */
async fn get_book_and_music() -> Result<(Book, Music), String> {
    let book_fut = get_book().map_err(|()| "Unable to get book".to_string());
    let music_fut = get_music();
    // If any Future fails, try_join! stops all execution immediately
    try_join!(book_fut, music_fut)
}

async fn get_into_book_and_music() -> (Book, Music) {
    get_book_and_music().await.unwrap()
}

fn main() {
    futures::executor::block_on(get_into_book_and_music());
}

The `select!` Macro

The join! macro only allows you to process results after all Futures have completed. In contrast, the select! macro waits on multiple Futures, and as soon as any one of them completes, it can be handled immediately:

use futures::{
    future::FutureExt, // for `.fuse()`
    pin_mut,
    select,
};

async fn task_one() { /* ... */ }
async fn task_two() { /* ... */ }

/**
 * Race mode: runs t1 and t2 concurrently.
 * Whichever finishes first, the function ends and the other task is not waited on.
 */
async fn race_tasks() {
    // .fuse() enables the Future to implement the FusedFuture trait
    let t1 = task_one().fuse();
    let t2 = task_two().fuse();

    // pin_mut macro gives the Futures the Unpin trait
    pin_mut!(t1, t2);

    // Use select! to wait on multiple Futures and handle whichever completes first
    select! {
        () = t1 => println!("Task 1 finished first"),
        () = t2 => println!("Task 2 finished first"),
    }
}

The code above runs t1 and t2 concurrently. Whichever finishes first will trigger its corresponding println! output. The function will then end without waiting for the other task to complete.

Note: Requirements for `select!` – `FusedFuture + Unpin`

Using select! requires the Futures to implement both FusedFuture and Unpin, which are achieved via the .fuse() method and the pin_mut! macro.

The .fuse() method enables a Future to implement the FusedFuture trait.
The pin_mut! macro allows the Future to implement the Unpin trait.

Note: select! requires two trait bounds: FusedStream + Unpin:

Unpin: Since select doesn’t consume the ownership of the Future, it accesses them via mutable reference. This allows the Future to be reused if it hasn’t completed after select finishes.

FusedFuture: Once a Future completes, select should no longer poll it. “Fuse” means short-circuiting—the Future will return Poll::Pending immediately if polled again after finishing.

Only by implementing FusedFuture can select! work correctly within a loop. Without it, a completed Future might still be polled continuously by select.

For Stream, a slightly different trait called FusedStream is used. By calling .fuse() (or implementing it manually), a Stream becomes a FusedStream, allowing you to call .next() or .try_next() on it and receive a Future that implements FusedFuture.

use futures::{
    stream::{Stream, StreamExt, FusedStream},
    select,
};

async fn add_two_streams() -> u8 {
    // mut s1: impl Stream<Item = u8> + FusedStream + Unpin,
    // mut s2: impl Stream<Item = u8> + FusedStream + Unpin,

    // The `.fuse()` method enables Stream to implement the FusedStream trait
    let s1 = futures::stream::once(async { 10 }).fuse();
    let s2 = futures::stream::once(async { 20 }).fuse();

    // The pin_mut macro allows Stream to implement the Unpin trait
    pin_mut!(s1, s2);

    let mut total = 0;

    loop {
        let item = select! {
            x = s1.next() => x,
            x = s2.next() => x,
            complete => break,
            default => panic!(), // This branch will never run because `Future`s are prioritized first, then `complete`
        };
        if let Some(next_num) = item {
            total += next_num;
        }
    }
    println!("add_two_streams, total = {total}");
    total
}

fn main() {
    executor::block_on(add_two_streams());
}

Note: The select! macro also supports the default and complete branches:

complete branch: Runs only when all Futures and Streams have completed. It’s often used with a loop to ensure all tasks are finished.

default branch: If none of the Futures or Streams are in a Ready state, this branch is executed immediately.

Recommended Utilities for use with `select!`

When using the select! macro, two particularly useful functions/types are:

Fuse::terminated() function: Used to construct an empty Future (already implements FusedFuture) in a select loop, and later populate it as needed.
FuturesUnordered type: Allows a Future to have multiple copies, all of which can run concurrently.

use futures::{
    future::{Fuse, FusedFuture, FutureExt},
    stream::{FusedStream, FuturesUnordered, Stream, StreamExt},
    pin_mut,
    select,
};

async fn future_in_select() {
    // Create an empty Future that already implements FusedFuture
    let fut = Fuse::terminated();
    // Create a FuturesUnordered container which can hold multiple concurrent Futures
    let mut async_tasks: FuturesUnordered<Pin<Box<dyn Future<Output = i32>>>> = FuturesUnordered::new();
    async_tasks.push(Box::pin(async { 1 }));

    pin_mut!(fut);

    let mut total = 0;
    loop {
        select! {
            // select_next_some: processes only the Some(_) values from the stream and ignores None
            num = async_tasks.select_next_some() => {
                println!("first num is {num} and total is {total}");
                total += num;
                println!("total is {total}");
                if total >= 10 { break; }
                // Check if fut has terminated
                if fut.is_terminated() {
                    // Populate new future when needed
                    fut.set(async { 1 }.fuse());
                }
            },
            num = fut => {
                println!("second num is {num} and total is {total}");
                total += num;
                println!("now total is {total}");
                async_tasks.push(Box::pin(async { 1 }));
            },
            complete => break,
            default => panic!(),
        };
    }

    println!("total finally is {total}");
}

fn main() {
    executor::block_on(future_in_select());
}

Summary

The futures crate provides many practical tools for executing Futures concurrently, including:

join! macro: Runs multiple different Futures concurrently and waits until all of them complete before finishing. This can be understood as a must-complete-all concurrency model.
try_join! macro: Runs multiple different Futures concurrently, but if any one of them returns an error, it immediately stops executing all Futures. This is useful when Futures return Result and early exit is needed—a fail-fast concurrency model.
select! macro: Runs multiple different Futures concurrently, and as soon as any one of them completes, it can be immediately processed. This can be thought of as a race concurrency model.
Requirements for using select!: FusedFuture + Unpin, which can be implemented via the .fuse() method and pin_mut! macro.