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那样的包(其方法最近才能发出声音;类似于它可能会进入标准库而不是现在已弃用的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的所有编译器版本上都是安全的.