FuturesUnordered: An Efficient Way to Manage Multiple Futures in Rust

FuturesUnordered: An Efficient Way to Manage Multiple Futures in Rust

FuturesUnordered is the life of the async party.

In this post, we’re going to explore futures::stream::FuturesUnordered, a mighty and efficient tool for juggling multiple asynchronous tasks concurrently. This utility allows us to poll multiple futures in a non-blocking manner, automatically handling the completion of individual tasks and yielding their results as they become available.

Let’s create a few asynchronous functions that simulate different tasks and use FuturesUnordered to run them concurrently. First, add the required dependencies to your Cargo.toml file:

[dependencies]
tokio = { version = "1.0", features = ["full"] }
futures = "0.3"

Next, create the asynchronous functions:

use std::time::Duration;
use tokio::time::sleep;// Define the async functions that return a String.
async fn task_one() -> String {
    sleep(Duration::from_secs(3)).await;
    "Task one completed".to_owned()
}
async fn task_two() -> String {
    sleep(Duration::from_secs(1)).await;
    "Task two completed".to_owned()
}
async fn task_three() -> String {
    sleep(Duration::from_secs(2)).await;
    "Task three completed".to_owned()
}

We defined three asynchronous functions task_one, task_two, and task_three, each of which simulates a task with varying durations using tokio::time::sleep. Now, we'll use FuturesUnordered to run these tasks concurrently:

use futures::stream::FuturesUnordered;
use futures::StreamExt;
use std::pin::Pin;
use std::future::Future;#[tokio::main]
async fn main() {
    // Create a FuturesUnordered struct with the same type for all futures: Box<dyn Future<Output = String>>.
    let mut tasks = FuturesUnordered::<Pin<Box<dyn Future<Output = String>>>>::new();
    // Use Box::pin to create a pinned, heap-allocated version of each future.
    // This ensures that the memory location of the future does not change once it starts executing.
    tasks.push(Box::pin(task_one()));
    tasks.push(Box::pin(task_two()));
    tasks.push(Box::pin(task_three()));
    while let Some(result) = tasks.next().await {
        println!("{}", result);
    }
}

When you run this example, you’ll see that the tasks are executed concurrently, and their results are printed as they complete. The output may vary in order, as it depends on the completion times of each task:

Task two completed
Task three completed
Task one completed

In this example, we defined three asynchronous functions task_one, task_two, and task_three, each of which simulates a task with varying durations using tokio::time::sleep.

We created a FuturesUnordered instance called tasks and pushed our three tasks into it.

How does it work?

The theory behind futures::stream::FuturesUnordered lies in its implementation, which combines efficient task management and Rust's asynchronous programming model. Here's an overview of how it works under the hood:

1. FuturesUnordered is a data structure that holds a set of futures. Internally, it maintains two linked lists of futures — one for futures that are ready to be polled and another for futures that are not yet ready.
2. When you add a future to FuturesUnordered using push(), it is initially placed in the list of not-ready futures. The future is then registered with an async runtime to be notified when it's ready to make progress (i.e. when it's ready to be polled).
3. When a future becomes ready, it’s moved from the not-ready list to the ready list. This happens as a result of the waker associated with the future being notified by the async runtime.
4. The FuturesUnordered struct itself implements the Stream trait. When you call next() on a FuturesUnordered instance, it polls the futures in the ready list.
5. The next().await call will resolve to a completed future from the ready list when one is available. If there are no ready futures, it will asynchronously wait until a future becomes ready.
6. When a future is polled and completed, it’s removed from the ready list, and the result is returned by next().await. If the future is not completed, it remains on the ready list, and another future may be polled in the meantime. This process ensures that FuturesUnordered efficiently handles multiple futures concurrently, leveraging the available resources and avoiding blocking.
7. If a future is canceled or dropped, it will be removed from its respective list (ready or not ready), ensuring that any resources associated with the future are freed.
8. The FuturesUnordered stream continues to return values as long as there are futures in the ready or not-ready lists. Once all futures have been completed or canceled, the FuturesUnordered stream will return None, indicating that the stream has ended.

End

FuturesUnordered is the life of the async party. It makes sure no future is left behind, and every future gets its moment in the spotlight when the time is right.

We’ve seen how FuturesUnordered maintains two lists, one for the futures that are ready to be polled, and another for those still waiting for their time.

Utilizing the capabilities of FuturesUnordered, Rust applications can achieve improved performance, effectively manage concurrency, and maximize resource utilization. It is essential to acknowledge the significance of this remarkable tool and recognize that asynchronous programming in Rust can be both efficient and rewarding.

Happy coding!