如何在不使用跟踪句柄的情况下等待未知数量的 Rust 线​​程完成？

Posted 2023-03-17

【中文标题】如何在不使用跟踪句柄的情况下等待未知数量的 Rust 线​​程完成？

【英文标题】：How can I wait for an unknown number of Rust threads to complete without using tracking handles?

【发布时间】：2017-12-08 13:04:37

【问题描述】：

有什么好的方法可以调整这个Barrier 示例来处理两个差异：

事先不知道项目的数量（例如，在将大文件分成几行的情况下）

不跟踪线程句柄（例如，在下面的示例中不使用 handles 向量）。这样做的动机是增加额外的开销。

示例代码：

use std::sync::Arc, Barrier;
use std::thread;

let mut handles = Vec::with_capacity(10);
let barrier = Arc::new(Barrier::new(10));
for _ in 0..10 
    let c = barrier.clone();
    handles.push(thread::spawn(move|| 
        // do some work
        c.wait();
    ));

// Wait for other threads to finish.
for handle in handles 
    handle.join().unwrap();

代码 sn-p 稍微改编自 Barrier docs。

我首先想到的是（如果可能的话）改变Barrier 的内在价值；但是，API 不提供对Barrier 结构的num_threads 属性的可变访问。

另一个想法是不使用Barrier，而是使用AtomicUsize 编写自定义逻辑。

我愿意在 Rust 中学习最符合人体工程学/惯用的方法。

【问题讨论】：

是在问如何在不跟踪线程同步机制的情况下同步线程？如果您想稍后加入它们，则必须将这些句柄保留在某个地方。由于向量是可调整大小的，那么制作和保持可变数量的句柄和屏障实例有什么问题？

【参考方案1】：

您可以在 atomic 上使用 spinlock 来等待所有线程退出。当然，您可以将Arc<AtomicUsize> 传递给每个线程，而不是使用静态原子。

Ordering::SeqCst 可能太强了，但并发编程很难，我不确定如何放宽这种排序。

虽然可以这样做，但创建线程的成本可能会使像这样的微优化相形见绌。另外值得考虑的是，忙等待会降低程序的性能。

use std::panic;
use std::sync::atomic::AtomicUsize, Ordering, ATOMIC_USIZE_INIT;
use std::thread;
use std::time::Duration;

static GLOBAL_THREAD_COUNT: AtomicUsize = ATOMIC_USIZE_INIT;

fn main() 
    for i in 0..10 
        // mark that the thread is about to run
        // we need to do it in the main thread to prevent spurious exits
        GLOBAL_THREAD_COUNT.fetch_add(1, Ordering::SeqCst);
        thread::spawn(move|| 
            // We need to catch panics to reliably signal exit of a thread
            let result = panic::catch_unwind(move || 
                // do some work
                println!("-th thread reporting", i+1);
            );
            // process errors
            match result 
                _ => 
            
            // signal thread exit
            GLOBAL_THREAD_COUNT.fetch_sub(1, Ordering::SeqCst);
        );
    
    // Wait for other threads to finish.
    while GLOBAL_THREAD_COUNT.load(Ordering::SeqCst) != 0 
        thread::sleep(Duration::from_millis(1)); 
    

Playground link

【参考方案2】：

让每个线程将其结果（或只是 Thread::current）发送到等待父线程正在消费的多生产者、单消费者、通道。标准库为此提供了mpsc channnel ()。它的 CPU 密集度比自旋锁少 很多。