How to spawn a task with task_local？

[lz1998]

When the http server receives a request, it will generate a log_id and save it in task_local. If it call other service, the log_id should be carried.

If I want to call several services concurrently, what should I do? Is there some method to spawn a task with the same task_local?

[Darksonn]

You would need to set the task local in the child task as well. Alternatively, you can use combinators such as tokio::join!, FuturesUnordered or StreamExt::buffer_unordered which don't actually spawn independent tasks, so task locals remain active inside them.

[lz1998]

Is there some example of FuturesUnordered? There is no example on https://docs.rs.

[Darksonn]

Basically once you import StreamExt, you can call .next().await on the FuturesUnordered. That method will run everything inside the collection simultaneously until any one of them completes, then return the output of the first future to complete. Usually next is called in a loop. The collection also has a push method that lets you add new futures to the collection.

[lz1998]

This works.

    let mut all_futures = FuturesUnordered::new();
    for i in 1..10 {
        let j = i;
        all_futures.push(async move {
            tracing::info!("{} aaa",j);
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            tracing::info!("{} bbb",j);
            tokio::time::sleep(std::time::Duration::from_millis(1)).await;
        });
    }
    loop {
        match all_futures.next().await {
            Some(result) => {
                tracing::info!("end");
            }
            None => {
                println!("Done!");
                break;
            }
        }
    }

However, when I replace loop with join, an error occurs.

    let mut all_futures = FuturesUnordered::new();
    for i in 1..10 {
        let j = i;
        all_futures.push(async move {
            tracing::info!("{} aaa",j);
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            tracing::info!("{} bbb",j);
            tokio::time::sleep(std::time::Duration::from_millis(1)).await;
        });
    }
    futures::future::join_all(all_futures).await;

error[E0277]: the trait bound `FuturesUnordered<impl futures::Future>: IntoIterator` is not satisfied
   --> src/lib.rs:25:31
    |
25  |     futures::future::join_all(all_futures).await;
    |     ------------------------- ^^^^^^^^^^^ the trait `IntoIterator` is not implemented for `FuturesUnordered<impl futures::Future>`
    |     |
    |     required by a bound introduced by this call
    |
    = help: the following implementations were found:
              <&'a FuturesUnordered<Fut> as IntoIterator>
              <&'a mut FuturesUnordered<Fut> as IntoIterator>
              <FuturesUnordered<Fut> as IntoIterator>

[Darksonn]

If you're using join_all, you're not supposed to also use FuturesUnoredered and vice-versa. Use just one of them.

You can write the loop as this:

while let Some(result) = all_futures.next().await {
    ...
}