How to tokio::join multiple tasks?

Question

Imagine that some futures are stored in a Vec whose length are runtime-determined, you are supposed to join these futures concurrently, what should you do?

Obviously, by the example in the document of tokio::join, manually specifying each length the Vec could be, like 1, 2, 3, ... and dealing with respectable case should work.

extern crate tokio;

let v = Vec::new();
v.push(future_1);

// directly or indirectly you push many futures to the vector
 
v.push(future_N);

// to join these futures concurrently one possible way is 

if v.len() == 0 {}
if v.len() == 1 { join!(v.pop()); }
if v.len() == 2 { join!(v.pop(), v.pop() ); }
// ...

And I also noticed that tokio::join! take a list as parameter in the document, when I use syntax like

tokio::join!(v);

or something like

tokio::join![ v ] /  tokio::join![ v[..] ] / tokio::join![ v[..][..] ]

it just doesn't work

And here comes the question that is there any doorway to join these futures more efficient or should I miss something against what the document says?

`tokio::join!` takes a variable number of async expressions, not a single slice / vec of async values. — Masklinn, Aug 26 '20 at 05:50
misinterpreting the document of tokio I made in trying to solve this myself. I assume that a block of async code(async function or async expression or whatever) will be treated as futures in rust. And my problem is how join these length-varied ' async expression' concurrently in a more clear way. — bruceyuan, Aug 26 '20 at 12:05
Duplicate of [How to wait for a list of async function calls in rust?](https://stackoverflow.com/questions/63326882/how-to-wait-for-a-list-of-async-function-calls-in-rust) — Herohtar, Aug 26 '20 at 15:53

score 23 · Accepted Answer · answered Aug 26 '20 at 04:10

23

You can use futures::future::join_all to "merge" your collection of futures together into a single future, that resolves when all of the subfutures resolve.

answered Aug 26 '20 at 04:10

Colonel Thirty Two

23,953
8
45
85

thanks. Solved my problem. I didn't find the symmetrical one "join_any(...)", though. Is it somewhere else? Or absent... ? – zertyz Feb 18 '22 at 00:36
found https://docs.rs/tokio/latest/tokio/macro.select.html but it cancels the other branches when the first completes... – zertyz Feb 18 '22 at 00:46

mzabaluev · Answer 2 · 2022-12-16T06:48:32.767

join_all and try_join_all, as well as more versatile FuturesOrdered and FuturesUnordered utilities from the same crate futures, are executed as a single task. This is probably fine if the constituent futures are not often concurrently ready to perform work, but if you want to make use of CPU parallelism with the multi-threaded runtime, consider spawning the individual futures as separate tasks and waiting on the tasks to finish.

Tokio 1.21.0 or later: JoinSet

With recent Tokio releases, you can use JoinSet to get the maximum flexibility, including the ability to abort all tasks. The tasks in the set are also aborted when JoinSet is dropped.

use tokio::task::JoinSet;

let mut set = JoinSet::new();

for fut in v {
    set.spawn(fut);
}

while let Some(res) = set.join_next().await {
    let out = res?;
    // ...
}

Older API

Spawn tasks with tokio::spawn and wait on the join handles:

use futures::future;

// ...

let outputs = future::try_join_all(v.into_iter().map(tokio::spawn)).await?;

You can also use the FuturesOrdered and FuturesUnordered combinators to process the outputs asynchronously in a stream:

use futures::stream::FuturesUnordered;
use futures::prelude::*;

// ...

let mut completion_stream = v.into_iter()
    .map(tokio::spawn)
    .collect::<FuturesUnordered<_>>();
while let Some(res) = completion_stream.next().await {
    // ...    
}

One caveat with waiting for tasks this way is that the tasks are not cancelled when the future (e.g. an async block) that has spawned the task and possibly owns the returned JoinHandle gets dropped. The JoinHandle::abort method needs to be used to explicitly cancel the task.

score 4 · Answer 3 · answered Aug 12 '22 at 15:13

A full example:

#[tokio::main]
async fn main() {
    let tasks = (0..5).map(|i| tokio::spawn(async move {
        sleep(Duration::from_secs(1)).await; // simulate some work
        i * 2
    })).collect::<FuturesUnordered<_>>();

    let result = futures::future::join_all(tasks).await;
    println!("{:?}", result); // [Ok(8), Ok(6), Ok(4), Ok(2), Ok(0)]
}

Playground

How to tokio::join multiple tasks?

3 Answers3

Tokio 1.21.0 or later: JoinSet

Older API

Linked