使用非重叠向量块，并合并结果

Question

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

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

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

当然，您可以使用 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);
}

使用非重叠向量块，并合并结果

Consume non-overlapping vector chunks, and combine results

parallel-processing

multithreading

rust