为什么使用Iterator :: map生成线程不能并行运行线程?

问题描述 投票:0回答:1

我在Rust中编写了一个简单的多线程应用程序,用于将数字从1添加到x。 (我知道有一个公式,但重点是在Rust中编写一些多线程代码,而不是得到结果。)它工作正常,但在我将它重构为更具功能性的风格而不是命令式之后,多线程没有更多的加速。在检查CPU使用情况时,似乎只有一个核心用于我的4核/ 8线程CPU。原始代码的CPU使用率为790%,而重构版本只有99%。

原始代码:

use std::thread;

fn main() {
    let mut handles: Vec<thread::JoinHandle<u64>> = Vec::with_capacity(8);

    const thread_count: u64 = 8;
    const batch_size: u64 = 20000000;

    for thread_id in 0..thread_count {
        handles.push(thread::spawn(move || {
            let mut sum = 0_u64;

            for i in thread_id * batch_size + 1_u64..(thread_id + 1) * batch_size + 1_u64 {
                sum += i;
            }

            sum
        }));
    }

    let mut total_sum = 0_u64;

    for handle in handles.into_iter() {
        total_sum += handle.join().unwrap();
    }
    println!("{}", total_sum);
}

重构的代码:

use std::thread;

fn main() {
    const THREAD_COUNT: u64 = 8;
    const BATCH_SIZE: u64 = 20000000;

    // spawn threads that calculate a part of the sum
    let handles = (0..THREAD_COUNT).map(|thread_id| {
        thread::spawn(move ||
            // calculate the sum of all numbers from assigned to this thread
            (thread_id * BATCH_SIZE + 1 .. (thread_id + 1) * BATCH_SIZE + 1)
                .fold(0_u64,|sum, number| sum + number))
    });

    // add the parts of the sum together to get the total sum
    let sum = handles.fold(0_u64, |sum, handle| sum + handle.join().unwrap());

    println!("{}", sum);
}

程序的输出是相同的(12800000080000000),但重构版本慢了5-6倍。

似乎迭代器被懒惰地评估了。如何强制评估整个迭代器?我试图将它收集到[thread::JoinHandle<u64>; THREAD_COUNT as usize]类型的数组中,但我得到以下错误:

  --> src/main.rs:14:7
   |
14 |     ).collect::<[thread::JoinHandle<u64>; THREAD_COUNT as usize]>();
   |       ^^^^^^^ a collection of type `[std::thread::JoinHandle<u64>; 8]` cannot be built from `std::iter::Iterator<Item=std::thread::JoinHandle<u64>>`
   |
   = help: the trait `std::iter::FromIterator<std::thread::JoinHandle<u64>>` is not implemented for `[std::thread::JoinHandle<u64>; 8]`

收集到一个向量确实有效,但这似乎是一个奇怪的解决方案,因为大小是预先知道的。有没有比使用矢量更好的方法?

multithreading functional-programming rust
1个回答
4
投票

Rust中的迭代器是惰性的,因此在handles.fold尝试访问迭代器的相应元素之前,您的线程不会启动。基本上会发生什么:

  1. handles.fold尝试访问迭代器的第一个元素。
  2. 第一个线程启动。
  3. handles.fold称其闭包,将handle.join()称为第一个线程。
  4. handle.join等待第一个线程结束。
  5. handles.fold尝试访问迭代器的第二个元素。
  6. 第二个线程启动。
  7. 等等。

在折叠结果之前,您应该将控制柄收集到矢量中:

let handles: Vec<_> = (0..THREAD_COUNT)
    .map(|thread_id| {
        thread::spawn(move ||
            // calculate the sum of all numbers from assigned to this thread
            (thread_id * BATCH_SIZE + 1 .. (thread_id + 1) * BATCH_SIZE + 1)
                .fold(0_u64,|sum, number| sum + number))
    })
    .collect();

或者您可以使用像Rayon这样提供并行迭代器的包。

© www.soinside.com 2019 - 2024. All rights reserved.