How to wait for a list of async function calls in Rust?

user_6389445 0
Viewed 6275

I have a list of async functions that I want to execute concurrently and then wait for all them to finish. The working code I have right now is:

async fn start_consumers(&self) {
    for consumer in &self.consumers {
        consumer.consume().await;
    }
}

This is not quite accurate as the functions are executed serially. I am looking for something like join!, but which works on a dynamic vector. With that I should be able to write something like:

async fn start_consumers(&self) {
    let mut v = Vec::new();
    for consumer in &self.consumers {
        consumer.consume();
    }
    join!(v);
}

Right now join! supports only tuples. I am looking for an alternative for that. Something similar to Promise.all() in JavaScript.

async-await rust concurrency
5 Answers

The futures crate has a join_all function which allows for waiting on a collection of futures:

use futures::future::join_all;

async fn start_consumers(&self) {
    let mut v = Vec::new();
    for consumer in &self.consumers {
        v.push(consumer.consume());
    }
    join_all(v).await;
 }
user_63894450

I also asked a similar question on the same day, but in my case I had a Result wrapped in a Future. So instead of join_all I had to use try_join_all

user_33584990

join_all/try_join_all do the trick, but the output is a Vec that collects the results of the futures. In the modified example from above, the combined future produces a Vec<()>, which does not result in allocations, and even operations extending this vector should be optimized to nothing in release builds.

Even in cases when you do need the outputs, it may be worthwhile to process them as they come asynchronously in a stream, rather than wait for all of them collected. For that you can use FuturesOrdered or FuturesUnordered, depending on whether you care about preserving the order of the original futures in the outputs yielded by the stream, or rather prefer receiving the outputs in the order of their completion. FuturesUnordered does not require buffering of the results and may complete faster than a FuturesOrdered composed of the same futures would do.

user_22171510

The easiest way to do this is to use an mpsc channel where, instead of sending messages, you wait for the channel to be closed, which happens when every sender has been dropped.

See example with tokio is here.

user_18781800

If you want to await/join all your Futures from a normal synchronous function, and don't care about their results, you can write this:

futures::executor::block_on(async{futures::join!(future1, future2, future3, ...)});

You can use this macro block_all for more ergonomical usage:

macro_rules! block_all {
    ($($future:expr),*) => {{
        futures::executor::block_on(async{futures::join!($($future),*)})   
    }};
}

Usage:

block_all!(future1, future2, ...);
user_108237380
Related