multithreading - 使用非重叠向量 block ，并合并结果

Question

我正在尝试通过使用线程来加速大型向量的昂贵计算。我的函数使用一个向量，计算一个新值向量(它不聚合，但必须保留输入顺序)，然​​后返回它。但是，我正在努力弄清楚如何生成线程、为每个线程分配向量切片，​​然后收集并组合结果。

// tunable
const NUMTHREADS: i32 = 4;

fn f(val: i32) -> i32 {
    // expensive computation
    let res = val + 1;
    res

}

fn main() {
    // choose an odd number of elements
    let orig = (1..14).collect::<Vec<i32>>();
    let mut result: Vec<Vec<i32>> = vec!();
    let mut flat: Vec<i32> = Vec::with_capacity(orig.len());
    // split into slices
    for chunk in orig.chunks(orig.len() / NUMTHREADS as usize) {
        result.push(
            chunk.iter().map(|&digit|
                f(digit)).collect()
            );
    };
    // flatten result vector
    for subvec in result.iter() {
        for elem in subvec.iter() {
            flat.push(elem.to_owned());
        }
    }
    println!("Flattened result: {:?}", flat);
}

线程计算应该发生在 for chunk… 和 //flatten … 之间，但我找不到许多生成 x 线程、分配 block 的简单示例顺序地，并将新计算的向量从线程中返回到容器中，这样它就可以被展平。我是否必须将 orig.chunks() 包裹在 Arc 中，然后在循环中手动抓取每个 block ？我必须将 f 传递到每个线程吗？我是否必须使用 B 树来确保输入和输出顺序匹配？我可以只使用 simple_parallel 吗？ ？

Answer 1

好吧，这是不稳定的理想应用程序 thread::scoped() :

#![feature(scoped)]
use std::thread::{self, JoinGuard};

// tunable
const NUMTHREADS: i32 = 4;

fn f(val: i32) -> i32 {
    // expensive computation
    let res = val + 1;
    res
}

fn main() {
    // choose an odd number of elements
    let orig: Vec<i32> = (1..14).collect();

    let mut guards: Vec<JoinGuard<Vec<i32>>> = vec!();

    // split into slices
    for chunk in orig.chunks(orig.len() / NUMTHREADS as usize) {
        let g = thread::scoped(move || chunk.iter().cloned().map(f).collect());
        guards.push(g);
    };

    // collect the results
    let mut result: Vec<i32> = Vec::with_capacity(orig.len());
    for g in guards {
        result.extend(g.join().into_iter());
    }

    println!("Flattened result: {:?}", result);
}

它不稳定并且不太可能以这种形式稳定下来，因为它有一个固有的缺陷(您可以找到更多信息 here )。据我所知，simple_parallel 只是这种方法的扩展 - 它隐藏了 JoinGuards 的摆弄，也可以在稳定的 Rust 中使用(可能与一些 不安全ty，我相信)。但是，正如其文档所建议的那样，不建议将其用于一般用途。

当然，您可以使用 thread::spawn()，但是您需要克隆每个 block 以便将其移动到每个线程中:

use std::thread::{self, JoinHandle};

// tunable
const NUMTHREADS: i32 = 4;

fn f(val: i32) -> i32 {
    // expensive computation
    let res = val + 1;
    res
}

fn main() {
    // choose an odd number of elements
    let orig: Vec<i32> = (1..14).collect();

    let mut guards: Vec<JoinHandle<Vec<i32>>> = vec!();

    // split into slices
    for chunk in orig.chunks(orig.len() / NUMTHREADS as usize) {
        let chunk = chunk.to_owned();
        let g = thread::spawn(move || chunk.into_iter().map(f).collect());
        guards.push(g);
    };

    // collect the results
    let mut result: Vec<i32> = Vec::with_capacity(orig.len());
    for g in guards {
        result.extend(g.join().unwrap().into_iter());
    }

    println!("Flattened result: {:?}", result);
}