使用非重叠的矢量块,并组合结果

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线程的简单例子,顺序分配块,并将新计算的向量从线程中返回到容器中,以便可以展平.我必须包裹orig.chunks()在Arc,并手动抓住每个块的循环？我是否必须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);
}

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

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

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);
}