try_join! on parallel tasks

[zaszi]

I'm currently building a system service that runs several never-ending tasks concurrently. It is important that when a single task fails, the error is bubbled up to the caller (and again bubbled up to the main function of the program).

When I implement this concurrently with Tokio's try_join! macro, this works just fine. Consider the following code:

#[tokio::main]
pub async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
    // Create async tasks
    let task1 = async { // Loop forever };
    let task2 = async { // Loop forever };
    let task3 = async { // Loop forever };
    let task4 = async { // Loop forever };
    let task5 = async { // Loop forever };

    // Join async tasks together
    let res = tokio::try_join!(
        task1,
        task2,
        task3,
        task4,
        task5,
    );

    // Handle error
    if let Err(e) = res { return Err(e) };
    Ok(())

}

However, as soon as I try to run things in parallel I run into problems. According to the documentation oftry_join!, I should do this:

#[tokio::main]
pub async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
    // Create async tasks
    let task1 = async { // Loop forever };
    let task2 = async { // Loop forever };
    let task3 = async { // Loop forever };
    let task4 = async { // Loop forever };
    let task5 = async { // Loop forever };

    // Join async tasks together
    let res = tokio::try_join!(
        tokio::spawn(task1),
        tokio::spawn(task2),
        tokio::spawn(task3),
        tokio::spawn(task4),
        tokio::spawn(task5),
    );

    // Handle error
    if let Err(e) = res { return Err(e) };
    Ok(())

}

But this poses two problems. First I get a type error I'm not sure I entirely understand.

#[tokio::main]
^^^^^^^^^^^^^^ expected struct `std::boxed::Box`, found struct `tokio::task::JoinError`

If I match the result and simply discard it as Ok(()) the program does run. However, the second problem is more insidious: if a single task fails, the other tasks are not stopped, and consequentially errors are no longer bubbled up to the caller as the function hasn't ended yet.

I've been playing with this for some time and searched through documentation and examples, but thus far I haven't figured out the correct approach here. To stop the remaining tasks it seems I could use Abortable from the futures crate, but that still leaves the type issue that I'm unsure on how to handle.

Could someone point me in the right direction here? How do I implement this correctly using Tokio?

[Darksonn]

Okay so first, the error

#[tokio::main]
^^^^^^^^^^^^^^ expected struct `std::boxed::Box`, found struct `tokio::task::JoinError`

is due to this line:

if let Err(e) = res { return Err(e) };

The type of e here is not a Box<dyn Error>, so the return fails. You can do return Err(e.into()); to convert the error to a box.

As for cancellation of tasks, first not that tokio::spawn wraps the thing in a Result in case the spawned task panics, so if the tasks also return Results, you will want some sort of flatten that turns Result<Result<T, E1>, E2> into a Result<T, E> to properly notice the inner errors.

As for cancelling tasks, you can use a broadcast channel combined with select! to cancel many with a single message.

tokio::select! {
    _ = do_the_stuff() => {},
    _ = broadcast.recv() => {},
}

[zaszi]

@Darksonn Thank you for clarifying. You put me on the right track.

select! was the correct answer here. I no longer need to end tasks manually as I can just have the entire runtime end, and errors can be bubbled up to the caller upon first failure as I originally intended.

I still needed to flatten the nested Result I had due to the use of tokio::spawn as you suggested, and while there are prettier options (I should took a deeper look at combinators), the following code does work.

For posterity, this spawns a set of never-ending tasks not merely concurrently but also in parallel. Tasks return Result<(), Box<dyn Error + Send + Sync>> and tokio:spawn wraps this in another result, which is then flattened in a match statement. The select! macro returns that result of the first task that returns (i.e. encounters an error, as they should never end) and passes it back to the function caller. The runtime ends here as well, so all tasks end the moment an error is returned.

#[tokio::main]
pub async fn run() -> Result<(), Box<dyn Error + Send + Sync>> {
    // Create async tasks
    let task1 = async { // Loop forever };
    let task2 = async { // Loop forever };
    let task3 = async { // Loop forever };
    let task4 = async { // Loop forever };
    let task5 = async { // Loop forever };

    // Join async tasks together
    tokio::select! {
        r = tokio::spawn(task1) => {
            match r {
                Ok(s) => s,
                Err(e) => Err(e.into()),
            }
        },
        r = tokio::spawn(task2) => {
            match r {
                Ok(s) => s,
                Err(e) => Err(e.into()),
            }
        },
        // and so on
        ...
    }
}