如何在Rust中安全实现C++常用的完美并行处理模式？

问题背景 #

你在C++中常用的跨步并行处理伪代码如下：

void kernel(inputs, outputs, start, stride) {
  for (size_t i = start; i < length(inputs); i+=stride) {
    outputs[i] = process(inputs[i]);
  }
}
void run(inputs, number_of_threads) {
  threads = []
  vector outputs(length(inputs))
  for (int i = 0; i < number_of_threads; i++ {
    threads.push(thread(kernel, inputs, &outputs, i, number_of_threads));
  }
  for t in threads {
    t.join()
  }
return outputs
} 

该模式属于完美并行场景，无竞态条件，但受限于Rust的所有权模型，需要安全实现多线程对输出数组的可变访问。

安全实现方案 #

方法1：手动拆分可变切片（无第三方依赖） #

由于每个线程仅操作不重叠的索引范围，我们可以手动将输出数组拆分为多个独立的可变切片，每个线程持有自己的切片，完全符合Rust的借用规则，无需同步原语。

示例代码：

use std::thread;
use std::sync::Arc;

// 替换为你的实际处理逻辑
fn process<T>(input: T) -> T {
    input
}

fn kernel<T>(inputs: &[T], outputs: &mut [T], start: usize, stride: usize) {
    for i in (start..inputs.len()).step_by(stride) {
        outputs[i] = process(inputs[i]);
    }
}

fn run<T: Send + Clone + 'static>(inputs: &[T], number_of_threads: usize) -> Vec<T> {
    let mut outputs = vec![inputs[0].clone(); inputs.len()];
    let mut threads = Vec::with_capacity(number_of_threads);
    let inputs_arc = Arc::from(inputs);

    // 拆分输出为不重叠的可变切片
    let mut chunks = outputs.chunks_mut(inputs.len() / number_of_threads);
    for i in 0..number_of_threads {
        // 处理最后一个线程的剩余元素
        let chunk = if i == number_of_threads - 1 {
            chunks.into_remainder()
        } else {
            chunks.next().unwrap()
        };
        let start = i * (inputs.len() / number_of_threads);
        let stride = number_of_threads;
        let inputs_clone = inputs_arc.clone();

        threads.push(thread::spawn(move || {
            kernel(&inputs_clone, chunk, start, stride);
        }));
    }

    // 等待所有线程完成
    for thread in threads {
        thread.join().unwrap();
    }

    outputs
}

说明：用Arc包裹输入避免大数组克隆，chunks_mut保证切片不重叠，编译器会自动验证安全性。

方法2：使用Rayon库（简洁高效） #

Rayon是Rust生态成熟的并行计算库，封装了线程管理和切片拆分逻辑，一行代码即可实现需求，是最推荐的方案。

首先在Cargo.toml添加依赖：

[dependencies]
rayon = "1.7"

实现代码：

use rayon::prelude::*;

fn process<T>(input: T) -> T {
    input
}

fn run<T: Send + Clone + 'static>(inputs: &[T], number_of_threads: usize) -> Vec<T> {
    // 配置全局线程池（可选，不配置则使用默认CPU核心数）
    rayon::ThreadPoolBuilder::new()
        .num_threads(number_of_threads)
        .build_global()
        .unwrap();

    // 并行遍历输入并处理，自动分配线程
    inputs.par_iter()
        .map(|&x| process(x))
        .collect()
}

说明：Rayon会自动处理负载均衡和线程调度，代码简洁且安全，适合绝大多数并行场景。

方法3：使用UnsafeCell（不推荐） #

如果需要模拟C++的原始共享模式，可以用UnsafeCell配合Arc，但需要手动保证无竞态（编译器无法验证），存在风险，仅作了解：

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

fn process<T>(input: T) -> T {
    input
}

fn kernel<T>(inputs: &[T], outputs: &Arc<UnsafeCell<Vec<T>>>, start: usize, stride: usize) {
    // 手动获取可变引用，必须确保线程间无索引重叠
    let outputs = unsafe { &mut *outputs.get() };
    for i in (start..inputs.len()).step_by(stride) {
        outputs[i] = process(inputs[i]);
    }
}

fn run<T: Send + Clone + 'static>(inputs: &[T], number_of_threads: usize) -> Vec<T> {
    let outputs = Arc::new(UnsafeCell::new(vec![inputs[0].clone(); inputs.len()]));
    let mut threads = Vec::with_capacity(number_of_threads);
    let inputs_arc = Arc::from(inputs);

    for i in 0..number_of_threads {
        let outputs_clone = outputs.clone();
        let inputs_clone = inputs_arc.clone();
        threads.push(thread::spawn(move || {
            kernel(&inputs_clone, &outputs_clone, i, number_of_threads);
        }));
    }

    for thread in threads {
        thread.join().unwrap();
    }

    // 取出内部Vec（此时Arc引用计数已降为1）
    Arc::try_unwrap(outputs).unwrap().into_inner()
}

警告：此方法依赖开发者手动保证线程安全，一旦逻辑出错会导致未定义行为，优先推荐前两种安全方案。

内容的提问来源于stack exchange，提问作者orbita