Is there an API to race N threads (or N closures on N threads) to completion?

Question

Given several threads that complete with an Output value, how do I get the first Output that's produced? Ideally while still being able to get the remaining Outputs later in the order they're produced, and bearing in mind that some threads may or may not terminate.

Example:

struct Output(i32);

fn main() {
    let mut spawned_threads = Vec::new();

    for i in 0..10 {
        let join_handle: ::std::thread::JoinHandle<Output> = ::std::thread::spawn(move || {
            // pretend to do some work that takes some amount of time
            ::std::thread::sleep(::std::time::Duration::from_millis(
                (1000 - (100 * i)) as u64,
            ));
            Output(i) // then pretend to return the `Output` of that work
        });
        spawned_threads.push(join_handle);
    }

    // I can do this to wait for each thread to finish and collect all `Output`s
    let outputs_in_order_of_thread_spawning = spawned_threads
        .into_iter()
        .map(::std::thread::JoinHandle::join)
        .collect::<Vec<::std::thread::Result<Output>>>();

    // but how would I get the `Output`s in order of completed threads?
}

I could solve the problem myself using a shared queue/channels/similar, but are there built-in APIs or existing libraries which could solve this use case for me more elegantly?

I'm looking for an API like:

fn race_threads<A: Send>(
    threads: Vec<::std::thread::JoinHandle<A>>
) -> (::std::thread::Result<A>, Vec<::std::thread::JoinHandle<A>>) {
    unimplemented!("so far this doesn't seem to exist")
}

(Rayon's join is the closest I could find, but a) it only races 2 closures rather than an arbitrary number of closures, and b) the thread pool w/ work stealing approach doesn't make sense for my use case of having some closures that might run forever.)

It is possible to solve this use case using pointers from How to check if a thread has finished in Rust? just like it's possible to solve this use case using an MPSC channel, however here I'm after a clean API to race n threads (or failing that, n closures on n threads).

Answer 1

These problems can be solved by using a condition variable:

use std::sync::{Arc, Condvar, Mutex};

#[derive(Debug)]
struct Output(i32);

enum State {
    Starting,
    Joinable,
    Joined,
}

fn main() {
    let pair = Arc::new((Mutex::new(Vec::new()), Condvar::new()));
    let mut spawned_threads = Vec::new();

    let &(ref lock, ref cvar) = &*pair;
    for i in 0..10 {
        let my_pair = pair.clone();
        let join_handle: ::std::thread::JoinHandle<Output> = ::std::thread::spawn(move || {
            // pretend to do some work that takes some amount of time
            ::std::thread::sleep(::std::time::Duration::from_millis(
                (1000 - (100 * i)) as u64,
            ));

            let &(ref lock, ref cvar) = &*my_pair;
            let mut joinable = lock.lock().unwrap();
            joinable[i] = State::Joinable;
            cvar.notify_one();
            Output(i as i32) // then pretend to return the `Output` of that work
        });
        lock.lock().unwrap().push(State::Starting);
        spawned_threads.push(Some(join_handle));
    }

    let mut should_stop = false;
    while !should_stop {
        let locked = lock.lock().unwrap();
        let mut locked = cvar.wait(locked).unwrap();

        should_stop = true;
        for (i, state) in locked.iter_mut().enumerate() {
            match *state {
                State::Starting => {
                    should_stop = false;
                }
                State::Joinable => {
                    *state = State::Joined;
                    println!("{:?}", spawned_threads[i].take().unwrap().join());
                }
                State::Joined => (),
            }
        }
    }
}

(playground link)

I'm not claiming this is the simplest way to do it. The condition variable will awake the main thread every time a child thread is done. The list can show the state of each thread, if one is (about to) finish, it can be joined.

Answer 2

No, there is no such API.

You've already been presented with multiple options to solve your problem:

• Use channels
• Use a CondVar
• Use futures

Sometimes when programming, you have to go beyond sticking pre-made blocks together. This is supposed to be a fun part of programming. I encourage you to embrace it. Go create your ideal API using the components available and publish it to crates.io.

---

I really don't see what's so terrible about the channels version:

use std::{sync::mpsc, thread, time::Duration};

#[derive(Debug)]
struct Output(i32);

fn main() {
    let (tx, rx) = mpsc::channel();

    for i in 0..10 {
        let tx = tx.clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis((1000 - (100 * i)) as u64));
            tx.send(Output(i)).unwrap();
        });
    }
    // Don't hold on to the sender ourselves
    // Otherwise the loop would never terminate
    drop(tx);

    for r in rx {
        println!("{:?}", r);
    }
}