是否有一个 API 可以让 N 个线程（或 N 个线程上的 N 个闭包）竞赛完成？

Question

给定几个以Output 值完成的线程，我如何获得第一个产生的Output？理想情况下，仍然能够在稍后按照它们的生成顺序获取剩余的Outputs，并记住某些线程可能会或可能不会终止。

例子：

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?
}

我可以使用共享队列/通道/类似的方法自己解决问题，但是是否有内置 API 或现有库可以更优雅地为我解决这个用例？

我正在寻找这样的 API：

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 是我能找到的最接近的，但是 a）它只运行 2 个闭包而不是任意数量的闭包，并且 b）线程池 w/ 工作窃取方法不会对我的用例来说，有一些可能会永远运行的闭包。）

可以使用来自 How to check if a thread has finished in Rust? 的指针来解决这个用例，就像可以使用 MPSC 通道来解决这个用例一样，但是在这里我需要一个干净的 API 来竞争 n 线程（或失败） , n n 线程上的闭包）。

Answer 1

没有，没有这样的 API。

您已经看到了多种解决问题的方法：

• 使用channels
• 使用a CondVar
• 使用futures

有时在编程时，您必须超越将预制块粘在一起。这应该是编程的一个乐趣部分。我鼓励你接受它。使用可用的组件创建您理想的 API 并将其发布到 crates.io。

---

我真的不明白频道版本有什么可怕的：

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);
    }
}

Answer 2

这些问题可以通过使用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)

我并不是说这是最简单的方法。每次子线程完成时，条件变量都会唤醒主线程。该列表可以显示每个线程的状态，如果一个（即将）完成，则可以加入。