multithreading - Rust 中的线程安全可变非拥有指针？

Question

我正在尝试并行化我拥有的算法。这是我将如何用 C++ 编写它的草图:

void thread_func(std::vector<int>& results, int threadid) {
   results[threadid] = threadid;
}

std::vector<int> foo() {
  std::vector<int> results(4);

  for(int i = 0; i < 4; i++)
  {
     spawn_thread(thread_func, results, i);
  }

  join_threads();

  return results;
}

这里的要点是每个线程都有一个对它不拥有的共享可变对象的引用。这在 Rust 中似乎很难做到。我应该尝试根据(我在这里猜测)Mutex、Cell 和 &mut 来拼凑它，还是有更好的我应该遵循的模式？

Answer 1

正确的方法是使用Arc<Mutex<...>>或者，例如 Arc<RWLock<...>> . Arc是一个基于共享所有权的并发安全指针，指向不可变数据，Mutex/RWLock引入同步内部可变性。您的代码将如下所示:

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

fn thread_func(results: Arc<Mutex<Vec<i32>>>, thread_id: i32) {
    let mut results = results.lock().unwrap();
    results[thread_id as usize] = thread_id;
}

fn foo() -> Arc<Mutex<Vec<i32>>> {
    let results = Arc::new(Mutex::new(vec![0; 4]));

    let guards: Vec<_> = (0..4).map(|i| {
        let results = results.clone();
        thread::spawn(move || thread_func(results, i))
    }).collect();

    for guard in guards {
        guard.join();
    }

    results
}

很遗憾，这需要您返回 Arc<Mutex<Vec<i32>>>从函数中，因为没有办法“展开”值。另一种方法是在返回之前克隆载体。

但是，使用像 scoped_threadpool 这样的 crate (它的方法最近才变得合理；类似的东西可能会进入标准库，而不是现在不推荐使用的 thread::scoped() 函数，这是不安全的)它可以以更好的方式完成:

extern crate scoped_threadpool;

use scoped_threadpool::Pool;

fn thread_func(result: &mut i32, thread_id: i32) {
    *result = thread_id;
}

fn foo() -> Vec<i32> {
    let results = vec![0; 4];
    let mut pool = Pool::new(4);

    pool.scoped(|scope| {
        for (i, e) in results.iter_mut().enumerate() {
            scope.execute(move || thread_func(e, i as i32));
        }
    });

    results
}

如果您的 thread_func需要访问整个矢量，但是，没有同步就无法逃脱，因此您需要 Mutex ，你仍然会遇到展开问题:

extern crate scoped_threadpool;

use std::sync::Mutex;

use scoped_threadpool::Pool;

fn thread_func(results: &Mutex<Vec<u32>>, thread_id: i32) {
    let mut results = results.lock().unwrap();
    result[thread_id as usize] = thread_id;
}

fn foo() -> Vec<i32> {
    let results = Mutex::new(vec![0; 4]);
    let mut pool = Pool::new(4);

    pool.scoped(|scope| {
        for i in 0..4 {
            scope.execute(move || thread_func(&results, i));
        }
    });

    results.lock().unwrap().clone()
}

但至少你不需要任何 Arc在这里。还有 execute()方法是 unsafe如果您使用稳定的编译器，因为它没有相应的修复程序来确保它的安全。根据其构建脚本，它在所有高于 1.4.0 的编译器版本上都是安全的。