JoinSet and Structured Concurrency

A JoinSet owns a group of spawned tasks. It returns their results in the order they finish and aborts every task still running when it is dropped. This lesson builds up from one JoinSet to bounded parallelism, error policies, and the rules that keep spawned tasks from outliving the code that started them.

API	What it does
`JoinSet::new`, `spawn`, `spawn_blocking`	create a set, add an async task or a blocking closure
`join_next`, `try_join_next`	take the next finished result, waiting or not waiting
`len`, `is_empty`	count tasks still in the set
`abort_all`, `shutdown`, `detach_all`	cancel, cancel and wait, or release the tasks
`AbortHandle` (returned by `spawn`)	abort one task in the set
`Semaphore` + `JoinSet`	limit how many tasks run at once
`TaskTracker` (`tokio-util`)	count tasks and wait for all of them, without owning results

All time-based examples run on Tokio's paused clock: #[tokio::main(flavor = "current_thread", start_paused = true)] starts a single-threaded runtime whose clock jumps forward when every task is waiting on a timer (explained in Time: sleep, timeout, and interval). Sleeps therefore cost no real time, and the output is the same on every run.

Example 1: The smallest JoinSet

spawn adds a task. join_next waits for one task to finish and returns Some(Result<T, JoinError>). When the set is empty it returns None, which ends the loop.

Rust

use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for i in 1..=3 {
        set.spawn(async move { i * 10 });
    }

    let mut results = Vec::new();
    while let Some(joined) = set.join_next().await {
        results.push(joined.unwrap());
    }
    results.sort();
    println!("{results:?}");
}

Output

[10, 20, 30]

Every task in one set returns the same type T. The Result carries a JoinError when a task panicked or was aborted (Cancellation, abort, and Panics covers JoinError). The example sorts the results because the finishing order of tasks that never wait is not something to depend on.

Example 2: Results arrive in completion order

Three tasks sleep for 30, 10, and 20 milliseconds. join_next returns them in the order they finish, not the order they were spawned.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    for (name, ms) in [("slow", 30), ("fast", 10), ("medium", 20)] {
        set.spawn(async move {
            sleep(Duration::from_millis(ms)).await;
            name
        });
    }

    while let Some(joined) = set.join_next().await {
        println!("finished: {}", joined.unwrap());
    }
}

Output

finished: fast
finished: medium
finished: slow

A Vec<JoinHandle<T>> awaited in order would print slow first, because the handle for slow is first in the vector, and every later result waits behind it. JoinSet reports each result as soon as it exists.

Example 3: Vec of handles compared with JoinSet

The same three tasks, once with a vector of JoinHandles and once with a JoinSet. The program prints the moment each result becomes visible to the caller, using the paused clock.

Rust

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, Instant};

async fn work(ms: u64) -> u64 {
    sleep(Duration::from_millis(ms)).await;
    ms
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let start = Instant::now();
    let handles: Vec<_> = [30, 10, 20].into_iter().map(|ms| tokio::spawn(work(ms))).collect();
    for handle in handles {
        let value = handle.await.unwrap();
        println!("handles: got {value} at {} ms", start.elapsed().as_millis());
    }

    let start = Instant::now();
    let mut set = JoinSet::new();
    for ms in [30, 10, 20] {
        set.spawn(work(ms));
    }
    while let Some(joined) = set.join_next().await {
        let value = joined.unwrap();
        println!("set:     got {value} at {} ms", start.elapsed().as_millis());
    }
}

Output

handles: got 30 at 30 ms
handles: got 10 at 30 ms
handles: got 20 at 30 ms
set:     got 10 at 10 ms
set:     got 20 at 20 ms
set:     got 30 at 30 ms

With handles, the results for 10 and 20 are held back until 30 has been received. With the set, each result is available at the time the task finished. A caller that acts on each result (writes it, forwards it, starts follow-up work) starts that work earlier.

Example 4: Tag results to recover their input

Completion order loses the position of each input. Return the input, or its index, together with the result.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let delays = [30u64, 10, 20];
    let mut set = JoinSet::new();
    for (index, ms) in delays.into_iter().enumerate() {
        set.spawn(async move {
            sleep(Duration::from_millis(ms)).await;
            (index, format!("result-{ms}"))
        });
    }

    let mut slots: Vec<Option<String>> = vec![None; delays.len()];
    while let Some(joined) = set.join_next().await {
        let (index, value) = joined.unwrap();
        println!("index {index} finished");
        slots[index] = Some(value);
    }
    println!("{slots:?}");
}

Output

index 1 finished
index 2 finished
index 0 finished
[Some("result-30"), Some("result-10"), Some("result-20")]

The vector slots is in spawn order even though the tasks finished in another order. The same tagging works with any key, such as a URL or a job id.

Example 5: len, is_empty, and try_join_next

len counts the tasks still in the set. A task leaves the set when join_next returns its result. try_join_next returns immediately: None when no task has finished yet.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    set.spawn(async {
        sleep(Duration::from_millis(10)).await;
        "a"
    });
    set.spawn(async {
        sleep(Duration::from_millis(20)).await;
        "b"
    });
    println!("len = {}, empty = {}", set.len(), set.is_empty());
    println!("try_join_next right away: {:?}", set.try_join_next().map(|r| r.unwrap()));

    let first = set.join_next().await.unwrap().unwrap();
    println!("joined {first}, len = {}", set.len());
    let second = set.join_next().await.unwrap().unwrap();
    println!("joined {second}, len = {}", set.len());
    println!("join_next on an empty set: {:?}", set.join_next().await.map(|r| r.unwrap()));
    println!("empty = {}", set.is_empty());
}

Output

len = 2, empty = false
try_join_next right away: None
joined a, len = 1
joined b, len = 0
join_next on an empty set: None
empty = true

try_join_next is useful in a loop that must not wait, for example one that also has to poll a timer or a channel. Tasks spawned into the set start running at once, as with tokio::spawn; the set does not delay them.

Example 6: Dropping a JoinSet aborts its tasks

When a JoinSet is dropped, every task still in it is aborted. A guard value inside each task prints a line when the task's future is dropped, which shows the abort.

Rust

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

struct Guard(&'static str);

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    {
        let mut set = JoinSet::new();
        for name in ["one", "two"] {
            set.spawn(async move {
                let _guard = Guard(name);
                sleep(Duration::from_secs(60)).await;
                println!("{name} finished");
            });
        }
        // Let both tasks start and reach their sleep.
        sleep(Duration::from_millis(1)).await;
        println!("leaving the scope");
    }
    // The aborted tasks are cleaned up when the runtime next runs them.
    sleep(Duration::from_millis(1)).await;
    println!("done");
}

Output

leaving the scope
one dropped
two dropped
done

Neither task printed finished. Contrast this with tokio::spawn: a dropped JoinHandle detaches the task and it keeps running. A JoinSet ties the lifetime of its tasks to its own lifetime. This is the property that makes it a tool for structured concurrency (Example 14).

Example 7: abort_all and cancelled results

abort_all requests cancellation of every task but does not remove them. The tasks stay in the set until join_next returns their (cancelled) results.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    for i in 0..3 {
        set.spawn(async move {
            sleep(Duration::from_secs(60)).await;
            i
        });
    }
    sleep(Duration::from_millis(1)).await;

    set.abort_all();
    println!("after abort_all: len = {}", set.len());

    let mut cancelled = 0;
    while let Some(joined) = set.join_next().await {
        match joined {
            Ok(v) => println!("finished with {v}"),
            Err(e) if e.is_cancelled() => cancelled += 1,
            Err(e) => println!("other error: {e}"),
        }
    }
    println!("cancelled = {cancelled}, len = {}", set.len());
}

Output

after abort_all: len = 3
cancelled = 3, len = 0

abort_all marks the tasks. A task stops at its next .await (Cancellation, abort, and Panics). A task that has already finished before the abort keeps its Ok result.

Example 8: shutdown

shutdown is abort_all followed by join_next in a loop until the set is empty. It ignores panics from the tasks. When it returns, every task has stopped and dropped its state.

Rust

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

struct Guard(u32);

impl Drop for Guard {
    fn drop(&mut self) {
        println!("guard {} released", self.0);
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    for i in 0..3 {
        set.spawn(async move {
            let _guard = Guard(i);
            sleep(Duration::from_secs(60)).await;
        });
    }
    sleep(Duration::from_millis(1)).await;

    println!("calling shutdown");
    set.shutdown().await;
    println!("shutdown returned, len = {}", set.len());
}

Output

calling shutdown
guard 0 released
guard 1 released
guard 2 released
shutdown returned, len = 0

All three guards are released before shutdown returns. Dropping the set aborts the tasks too, but a drop cannot wait, so the cleanup happens later. Use shutdown when the code after it depends on the tasks being gone.

Example 9: detach_all

detach_all removes the tasks from the set without aborting them. They keep running in the background and their results are discarded.

Rust

use std::time::Duration;
use tokio::sync::oneshot;
use tokio::task::JoinSet;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, rx) = oneshot::channel();
    let mut set = JoinSet::new();
    set.spawn(async move {
        sleep(Duration::from_millis(10)).await;
        tx.send("background task finished").unwrap();
    });

    set.detach_all();
    println!("len after detach_all = {}", set.len());
    drop(set);

    println!("{}", rx.await.unwrap());
}

Output

len after detach_all = 0
background task finished

The set is empty and dropped, yet the task still ran to completion and sent its message. Detaching gives up both the results and the automatic abort, so it is the choice for work that must finish regardless of the owner (see Graceful Shutdown for tracking such work).

Example 10: AbortHandle for one task

spawn returns an AbortHandle. It aborts that one task and leaves the rest of the set alone. It does not give access to the result; join_next still does.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    let keep = set.spawn(async {
        sleep(Duration::from_millis(10)).await;
        "kept"
    });
    let cancel = set.spawn(async {
        sleep(Duration::from_millis(10)).await;
        "cancelled"
    });

    cancel.abort();
    println!("keep.is_finished() = {}", keep.is_finished());

    while let Some(joined) = set.join_next().await {
        match joined {
            Ok(v) => println!("ok: {v}"),
            Err(e) => println!("error: cancelled = {}", e.is_cancelled()),
        }
    }
}

Output

keep.is_finished() = false
error: cancelled = true
ok: kept

The aborted task reports a cancelled JoinError; the other task returns its value.

Example 11: A panic stays inside its task

A panic in one task becomes an Err(JoinError) for that task. The other tasks in the set continue and the set stays usable.

Rust

use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    set.spawn(async { 1 });
    set.spawn(async {
        if true {
            panic!("task failed");
        }
        2
    });
    set.spawn(async { 3 });

    let mut ok = Vec::new();
    let mut panics = 0;
    while let Some(joined) = set.join_next().await {
        match joined {
            Ok(v) => ok.push(v),
            Err(e) if e.is_panic() => panics += 1,
            Err(_) => {}
        }
    }
    ok.sort();
    println!("ok = {ok:?}, panics = {panics}");
}

Output

ok = [1, 3], panics = 1

The panic message goes to standard error and is not part of the program's output. JoinSet::join_all is the shortcut that collects every result into a Vec, but it panics if any task failed and cancels the rest, so it fits only when a failure should stop the caller too.

Example 12: Error policy, collect every error

Tasks that return Result have two levels of failure: the task can fail (JoinError) or the work can fail (the inner Err). This example keeps going after a failure and reports everything at the end.

Rust

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

async fn fetch(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(u64::from(id) * 10)).await;
    if id % 2 == 0 {
        Err(format!("item {id} failed"))
    } else {
        Ok(format!("item {id} ok"))
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    for id in 1..=5 {
        set.spawn(fetch(id));
    }

    let mut successes = Vec::new();
    let mut failures = Vec::new();
    while let Some(joined) = set.join_next().await {
        match joined.expect("task panicked") {
            Ok(v) => successes.push(v),
            Err(e) => failures.push(e),
        }
    }
    println!("successes: {successes:?}");
    println!("failures:  {failures:?}");
}

Output

successes: ["item 1 ok", "item 3 ok", "item 5 ok"]
failures:  ["item 2 failed", "item 4 failed"]

Collect-all fits batch work where each item is independent and a partial result is useful.

Example 13: Error policy, fail fast

Fail fast stops at the first error and cancels the remaining work. shutdown does the cancelling and waits for the tasks to stop.

Rust

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

struct Guard(u32);

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

async fn fetch(id: u32) -> Result<u32, String> {
    let _guard = Guard(id);
    sleep(Duration::from_millis(u64::from(id) * 10)).await;
    if id == 2 {
        Err(format!("item {id} failed"))
    } else {
        Ok(id)
    }
}

async fn run_all() -> Result<Vec<u32>, String> {
    let mut set = JoinSet::new();
    for id in 1..=4 {
        set.spawn(fetch(id));
    }

    let mut done = Vec::new();
    while let Some(joined) = set.join_next().await {
        match joined.expect("task panicked") {
            Ok(v) => done.push(v),
            Err(e) => {
                println!("first error: {e}");
                set.shutdown().await;
                return Err(e);
            }
        }
    }
    Ok(done)
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    println!("result: {:?}", run_all().await);
}

Output

item 1 future dropped
item 2 future dropped
first error: item 2 failed
item 3 future dropped
item 4 future dropped
result: Err("item 2 failed")

Item 1 finished normally. Items 3 and 4 were still sleeping when item 2 failed, and their futures were dropped before run_all returned. Fail fast fits work where one failure makes the other results useless, such as loading all parts of one configuration. Error Handling in Async Code covers error types and retries in depth.

Example 14: Structured concurrency, tasks do not outlive their scope

Structured concurrency is the rule that a group of concurrent tasks starts and ends inside one function scope, so when the function returns, none of its tasks are still running. The function below creates a JoinSet, waits for every task, and returns only then.

Rust

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

async fn sum_squares(numbers: Vec<u64>) -> u64 {
    let mut set = JoinSet::new();
    for n in numbers {
        set.spawn(async move {
            sleep(Duration::from_millis(n)).await;
            n * n
        });
    }
    let mut total = 0;
    while let Some(joined) = set.join_next().await {
        total += joined.unwrap();
    }
    total
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    println!("start");
    let total = sum_squares(vec![3, 1, 2]).await;
    println!("total = {total}");
    println!("no task from sum_squares is still running");
}

Output

start
total = 14
no task from sum_squares is still running

The rules that follow from this shape:

Rule	Reason
Keep the `JoinSet` in a local variable of the function that spawned the tasks.	Returning or dropping the function ends the set.
Wait for the results, or call `shutdown`, before returning.	The caller can rely on "the function returned, so its work is finished".
Let dropping cancel. If the function's own future is dropped (for example by a timeout), the set is dropped and the tasks abort.	Cancelling a caller then cancels its children.
Propagate errors to the caller instead of logging and continuing.	The caller decides the policy.
Use `tokio::spawn` with a detached handle only for work that is meant to outlive the caller.	Detached tasks have no owner that can cancel or wait for them.

Example 15: Cancelling the caller cancels the children

Because dropping the set aborts its tasks, wrapping the function in a timeout cancels the whole group when the timeout expires.

Rust

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, timeout};

struct Guard(u32);

impl Drop for Guard {
    fn drop(&mut self) {
        println!("child {} cancelled", self.0);
    }
}

async fn parent() -> u32 {
    let mut set = JoinSet::new();
    for id in 0..3 {
        set.spawn(async move {
            let _guard = Guard(id);
            sleep(Duration::from_secs(10)).await;
            id
        });
    }
    let mut total = 0;
    while let Some(joined) = set.join_next().await {
        total += joined.unwrap();
    }
    total
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let result = timeout(Duration::from_secs(1), parent()).await;
    println!("timed out: {}", result.is_err());
    // Give the aborted children a turn to drop their state.
    sleep(Duration::from_millis(1)).await;
    println!("done");
}

Output

timed out: true
child 0 cancelled
child 1 cancelled
child 2 cancelled
done

timeout dropped the future returned by parent. That dropped the local set, which aborted the three children. Tasks spawned with tokio::spawn and ignored would have kept running for ten seconds.

Example 16: Bounded parallelism with a Semaphore

A JoinSet alone runs every spawned task at once. To cap the number running, each task first takes a permit from a Semaphore (Semaphore, Notify, Barrier, and OnceCell). The program counts how many tasks run at the same time and prints the maximum.

Rust

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let limit = 3;
    let semaphore = Arc::new(Semaphore::new(limit));
    let running = Arc::new(AtomicUsize::new(0));
    let max_running = Arc::new(AtomicUsize::new(0));

    let mut set = JoinSet::new();
    for id in 0..10 {
        let semaphore = semaphore.clone();
        let running = running.clone();
        let max_running = max_running.clone();
        set.spawn(async move {
            let _permit = semaphore.acquire_owned().await.unwrap();
            let now = running.fetch_add(1, Ordering::SeqCst) + 1;
            max_running.fetch_max(now, Ordering::SeqCst);
            sleep(Duration::from_millis(10)).await;
            running.fetch_sub(1, Ordering::SeqCst);
            id
        });
    }

    let mut finished = 0;
    while let Some(joined) = set.join_next().await {
        joined.unwrap();
        finished += 1;
    }
    println!("tasks finished: {finished}");
    println!("limit: {limit}, max running at once: {}", max_running.load(Ordering::SeqCst));
}

Output

tasks finished: 10
limit: 3, max running at once: 3

All ten tasks exist from the start, but seven of them wait at acquire_owned. The permit is released when _permit is dropped at the end of the task. This form is short, and its cost is that the waiting tasks still occupy memory. For a very large job list, use the next example.

Example 17: Bounded parallelism by refilling the set

Keep at most limit tasks in the set. Before spawning the next job, wait for one task to finish if the set is full. Only limit tasks exist at any moment.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let limit = 3;
    let mut set = JoinSet::new();
    let mut max_len = 0;
    let mut results = Vec::new();

    for id in 0..10u64 {
        if set.len() >= limit {
            results.push(set.join_next().await.unwrap().unwrap());
        }
        set.spawn(async move {
            sleep(Duration::from_millis(10 * (id % 3 + 1))).await;
            id
        });
        max_len = max_len.max(set.len());
    }
    while let Some(joined) = set.join_next().await {
        results.push(joined.unwrap());
    }

    println!("max tasks alive: {max_len}");
    println!("results in completion order: {results:?}");
}

Output

max tasks alive: 3
results in completion order: [0, 1, 3, 2, 4, 6, 5, 7, 9, 8]

The job list could be an iterator that produces millions of items: at most three tasks and three results exist at once. The loop spawns a new job each time one finishes, so the set stays full while work remains.

Example 18: spawn_blocking in a JoinSet

JoinSet::spawn_blocking runs a synchronous closure on the blocking pool (Blocking and CPU-Bound Work) and returns its result through the same join_next loop.

Rust

use tokio::task::JoinSet;

fn checksum(data: &[u8]) -> u32 {
    data.iter().fold(0u32, |acc, b| acc.wrapping_mul(31).wrapping_add(u32::from(*b)))
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for id in 0..4u8 {
        set.spawn_blocking(move || {
            let data = vec![id; 1000];
            (id, checksum(&data))
        });
    }

    let mut results = Vec::new();
    while let Some(joined) = set.join_next().await {
        results.push(joined.unwrap());
    }
    results.sort();
    for (id, sum) in results {
        println!("block {id}: checksum {sum}");
    }
}

Output

block 0: checksum 0
block 1: checksum 3994340992
block 2: checksum 3693714688
block 3: checksum 3393088384

A blocking closure cannot be aborted once it has started. abort_all stops closures that have not begun running yet, and shutdown waits for the running ones to return.

Example 19: A JoinSet keeps every result until it is joined

A JoinSet stores the return value of each finished task until join_next takes it. A program that spawns tasks forever and never joins them accumulates results. This example shows the count with len.

Rust

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

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let mut set = JoinSet::new();
    for i in 0..1000u32 {
        set.spawn(async move { vec![i; 4] });
    }
    // Let every task run to completion. Nobody has called join_next yet.
    sleep(Duration::from_millis(1)).await;
    println!("tasks still counted by the set: {}", set.len());

    let mut taken = 0;
    while set.try_join_next().is_some() {
        taken += 1;
    }
    println!("results taken with try_join_next: {taken}");
    println!("len afterwards: {}", set.len());
}

Output

tasks still counted by the set: 1000
results taken with try_join_next: 1000
len afterwards: 0

All 1000 tasks had finished, yet the set still held 1000 results. For long-running servers that spawn one task per connection and do not need the results, the TaskTracker in the next example avoids that build-up.

Example 20: TaskTracker, count and wait without results

TaskTracker comes from the tokio-util crate (cargo add tokio-util --features rt). It counts spawned tasks and lets code wait until all of them have exited. It does not store results, does not need &mut self to spawn, and can be cloned into other tasks. close marks that no more spawning is planned, and wait returns once the tracker is closed and empty.

Rust

use std::time::Duration;
use tokio::time::sleep;
use tokio_util::task::TaskTracker;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let tracker = TaskTracker::new();
    for i in 0..3u64 {
        tracker.spawn(async move {
            sleep(Duration::from_millis(10 * (i + 1))).await;
            println!("task {i} done");
        });
    }
    println!("tracked: {}, closed: {}", tracker.len(), tracker.is_closed());

    tracker.close();
    println!("closed: {}", tracker.is_closed());
    tracker.wait().await;
    println!("all tasks exited, tracked: {}", tracker.len());
}

Output

tracked: 3, closed: false
closed: true
task 0 done
task 1 done
task 2 done
all tasks exited, tracked: 0

wait does nothing until close has been called, even when the tracker is empty. That lets a program spawn tasks from several places and decide once, at shutdown time, that the set is complete.

Example 21: TaskTracker does not abort on drop

Unlike a JoinSet, dropping a TaskTracker leaves its tasks running. The tracker only counts them.

Rust

use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time::sleep;
use tokio_util::task::TaskTracker;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, rx) = oneshot::channel();
    let tracker = TaskTracker::new();
    tracker.spawn(async move {
        sleep(Duration::from_millis(10)).await;
        tx.send("still ran after the tracker was dropped").unwrap();
    });
    drop(tracker);
    println!("{}", rx.await.unwrap());
}

Output

still ran after the tracker was dropped

	`JoinSet`	`TaskTracker`
Holds task results	yes, until `join_next`	no
Dropping it	aborts the tasks	leaves the tasks running
Spawning needs	`&mut self`	`&self`, and clones share the same tracker
Waiting	`join_next` in a loop	`close` then `wait`
Cancel a task	`abort_all`, `shutdown`, `AbortHandle`	none built in (pair with a `CancellationToken`, Graceful Shutdown)

Example 22: A preview of buffer_unordered

The futures crate offers a third way to run a bounded number of jobs. buffer_unordered(n) turns a stream (a sequence of values that arrive over time, taught in Streams) of futures into a stream of their results, running at most n futures at once and yielding results in completion order.

Rust

use futures::stream::{self, StreamExt};
use std::time::Duration;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let jobs = stream::iter(0..6u64).map(|id| async move {
        sleep(Duration::from_millis(10 * (id % 3 + 1))).await;
        id
    });

    let mut results = jobs.buffer_unordered(2);
    let mut order = Vec::new();
    while let Some(id) = results.next().await {
        order.push(id);
    }
    println!("completion order with 2 at a time: {order:?}");
}

Output

completion order with 2 at a time: [0, 1, 3, 2, 4, 5]

The futures run inside the calling task, not as separate spawned tasks. They therefore share one task's CPU time and stop when the stream is dropped. JoinSet spawns real tasks that can run on several worker threads. Choose buffer_unordered when the jobs are I/O waits inside one task, and JoinSet when jobs need parallel CPU time or independent panic isolation.

Interview points

A JoinSet returns task results in completion order through join_next, and returns None when it is empty.
Dropping a JoinSet aborts all its tasks. Dropping a JoinHandle or a TaskTracker does not.
abort_all only requests cancellation. shutdown cancels and waits until every task has stopped.
detach_all releases the tasks so they run on without an owner and without their results being kept.
A panic in one task becomes a JoinError for that task. The other tasks continue.
JoinSet::join_all panics on the first failed task; loop over join_next to apply another policy.
Bounded parallelism: a Semaphore acquired inside each task, or a refill loop that keeps set.len() at or below the limit. The refill loop keeps only limit tasks alive.
Structured concurrency: tasks start and finish inside one scope; cancelling the scope cancels the tasks.
A JoinSet keeps each result until it is joined. A TaskTracker keeps none.
TaskTracker::wait returns only after close has been called and the tracker is empty.

Summary

JoinSet::spawn adds a task and returns an AbortHandle; join_next yields Option<Result<T, JoinError>>.
Results arrive in completion order. Tag results with an index to recover input order.
len and is_empty count tasks not yet joined; try_join_next does not wait.
Drop aborts, abort_all requests cancellation, shutdown cancels and waits, detach_all releases.
spawn_blocking on a set runs synchronous closures on the blocking pool; started closures cannot be aborted.
Error policy is chosen by the join loop: collect every result, or call shutdown on the first error.
Semaphore limits running tasks; the refill loop limits existing tasks.
buffer_unordered(n) (from futures) runs n futures at a time inside one task.
TaskTracker from tokio-util counts tasks, is cloneable, does not abort on drop, and supports close and wait.