rust - 如何进行数据并行图像处理？

Question

我正在尝试用 Rust 编写一个简单的数据并行图像处理过滤器。我已经使用以下代码在单线程中工作。

/// an example of a very simple filter
fn modular_filter_chunk(input: &[u16], slice_width: usize, slice_height: usize, mod_value: u16, output: &mut[u16]) {
    let size = slice_width*slice_height;
    for i in 0..size {
        output[i] = input[i] % mod_value;
    }
}

fn modular_filter_multi(input: &Vec<u16>, width: usize, height: usize, slice_num: usize, mod_value: u16, output: &mut Vec<u16>) {
    // divide image vertically to slices
    let height_per_slice = height / slice_num;
    let size_per_chunk = height_per_slice * width;
    let in_itr = input.chunks(size_per_chunk);
    let out_itr = output.chunks_mut(size_per_chunk);
    for (input, output) in in_itr.zip(out_itr) {
        modular_filter_chunk(input, width, height_per_slice, mod_value, output);
    }
}

fn main() {
    let width: usize = 1024;
    let height: usize = 1024;
    let input = vec![1234; width*height];
    let mut output = vec![0; width*height];
    modular_filter_multi(&input, width, height, 4, 73, &mut output);
}

现在我想并行处理 for 循环，但我想不出一个简单的方法来执行此操作。我尝试像下面这样更改 for 循环，但无法通过编译错误。

let mut handles = Vec::new();
for (input, output) in in_itr.zip(out_itr) {
    let h = std::thread::spawn(move || {
        modular_filter_chunk(input, width, height_per_slice, mod_value, output);
    });
    handles.push(h);
}
for handle in handles {
    handle.join().unwrap();
}

编译错误信息

src\main.rs:25:21: 25:43 error: cannot infer an appropriate lifetime for lifetime parameter 'a in function call due to c
onflicting requirements
src\main.rs:25  let in_itr = input.chunks(size_per_chunk);
                                   ^~~~~~~~~~~~~~~~~~~~~~
src\main.rs:27:25: 27:44 note: first, the lifetime cannot outlive the method call at 27:24...
src\main.rs:27  for (input, output) in in_itr.zip(out_itr) {
                                       ^~~~~~~~~~~~~~~~~~~
note: in expansion of for loop expansion
src\main.rs:27:2: 29:3 note: expansion site
src\main.rs:27:25: 27:31 note: ...so that method receiver is valid for the method call
src\main.rs:27  for (input, output) in in_itr.zip(out_itr) {
                                       ^~~~~~
note: in expansion of for loop expansion
src\main.rs:27:2: 29:3 note: expansion site
src\main.rs:25:15: 25:20 note: but, the lifetime must be valid for the expression at 25:14...
src\main.rs:25  let in_itr = input.chunks(size_per_chunk);
                             ^~~~~
src\main.rs:25:15: 25:20 note: ...so that pointer is not dereferenced outside its lifetime
src\main.rs:25  let in_itr = input.chunks(size_per_chunk);
                             ^~~~~
error: aborting due to previous error
Could not compile `rust_multithread`.

我应该如何更改我的代码以使过滤器并行工作？

Answer 1

让我们看看 thread::spawn 的签名:

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    where F: FnOnce() -> T,
          F: Send + 'static,
          T: Send + 'static

这说明 spawn 接受一个实现了 FnOnce 的类型 F(将被恰好调用一次)并且当被调用时将返回一些类型T。 F 和 T 类型必须实现 Send 并且必须至少具有生命周期 'static。

Send bound 将可能的类型限制为“能够跨线程边界传输的类型”，'static 意味着该类型中的任何引用必须在程序的整个 生命周期内存在，从 main 开始之前到 main 退出之后。

这解释了您收到的错误消息:您的任何引用都不能保证在程序的整个生命周期内都有效。事实上，他们肯定不会活那么久。

当您产生一个线程时，新线程不再与产生它的线程有任何具体联系。新线程很可能比父线程长寿!如果您在父线程死后尝试使用引用，这将导致非常糟糕的事情发生。

如您所见，thread::scoped 提供了一个解决方案。作用域线程需要在它们包含的任何引用过期之前加入。 但是，作用域线程不稳定是有原因的:它们是unsound in the presence of reference cycles .有一个 RFC把它带回来，但它周围有一些深层次的细节，所以它被推迟了。

作为hamstergene points out , 你可以使用 Arc在稳定的 Rust 中安全地共享不可变数据。您需要使用 Mutex共享您的可变输出缓冲区。您可以看出为什么人们对恢复 thread::scoped 感到兴奋!