C++并发编程之std::partial

C++并发编程之std::partial_sum的并行版本

在C++中，std::partial_sum 是一个用于计算前缀和的算法，它将输入范围中的每个元素替换为其前缀和。为了提高性能，我们可以设计并实现一个并行版本的 std::partial_sum，以便在多核处理器上并行执行前缀和计算。基本思想是将输入范围划分为多个子范围，每个子范围由一个单独的线程处理，并在所有线程完成后进行合并。

基本思想

任务划分：将输入范围中的元素划分为多个子范围，每个子范围由一个线程处理。
线程执行：每个线程独立计算其子范围的前缀和。为了确保最终结果的正确性，每个子范围的前缀和计算需要考虑到前一个子范围的最后一个元素的前缀和。
合并结果：在所有线程完成其任务后，主线程负责合并各个子范围的前缀和结果，确保整个输入范围的前缀和计算是正确的。

实现代码

我们可以使用 C++11 的 std::thread 来实现并行版本的 std::partial_sum。为了简化实现，我们可以使用 std::vector 来管理线程，并使用 std::mutex 来确保对共享数据的访问是线程安全的。

#include <iostream>
#include <vector>
#include <thread>
#include <algorithm>
#include <iterator>
#include <mutex>// 并行版本的 std::partial_sum
template<typename Iterator, typename OutputIterator>
OutputIterator parallel_partial_sum(Iterator first, Iterator last, OutputIterator result) {const unsigned long length = std::distance(first, last);// 如果没有元素，直接返回 resultif (length == 0) {return result;}// 获取系统支持的并发线程数const unsigned long max_threads = std::thread::hardware_concurrency();const unsigned long num_threads = std::min(max_threads != 0 ? max_threads : 2, length);// 每个线程处理的元素数量const unsigned long block_size = length / num_threads;std::vector<std::thread> threads(num_threads - 1);std::vector<typename Iterator::value_type> block_sums(num_threads, 0);std::mutex block_sums_mutex;// 启动线程for (unsigned long i = 0; i < num_threads - 1; ++i) {Iterator block_start = first + i * block_size;Iterator block_end = block_start + block_size;threads[i] = std::thread([block_start, block_end, result, i, &block_sums, &block_sums_mutex, block_size]() {*result = *block_start;typename Iterator::value_type sum = *block_start;for (Iterator it = block_start + 1; it != block_end; ++it) {sum += *it;*++result = sum;}std::lock_guard<std::mutex> lock(block_sums_mutex);block_sums[i] = sum;});}// 主线程处理最后一个块Iterator block_start = first + (num_threads - 1) * block_size;Iterator block_end = last;*result = *block_start;typename Iterator::value_type sum = *block_start;for (Iterator it = block_start + 1; it != block_end; ++it) {sum += *it;*++result = sum;}std::lock_guard<std::mutex> lock(block_sums_mutex);block_sums[num_threads - 1] = sum;// 等待所有线程完成std::for_each(threads.begin(), threads.end(), std::mem_fn(&std::thread::join));// 合并结果OutputIterator final_result = result;for (unsigned long i = 1; i < num_threads; ++i) {*final_result += block_sums[i - 1];++final_result;}for (unsigned long i = 1; i < num_threads; ++i) {for (unsigned long j = 1; j < block_size; ++j) {*final_result += block_sums[i - 1];++final_result;}}return result + std::distance(first, last);
}int main() {std::vector<int> input = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};std::vector<int> output(input.size());// 使用并行版本的 std::partial_sumparallel_partial_sum(input.begin(), input.end(), output.begin());// 输出结果for (const auto& value : output) {std::cout << value << " ";}std::cout << std::endl;return 0;
}

代码说明

任务划分：
- length 是输入范围中的元素总数。
- max_threads 是系统支持的并发线程数，num_threads 是我们实际使用的线程数（不超过元素数量）。
- block_size 是每个线程处理的元素数量。
线程执行：
- 我们创建了一个 std::vector<std::thread> 来存储所有线程。
- 每个线程独立计算其子范围的前缀和，并将最后一个元素的前缀和存储在 block_sums 中。为了确保 block_sums 的访问是线程安全的，我们使用了 std::mutex。
合并结果：
- 主线程通过 std::thread::join 等待所有子线程完成。
- 主线程遍历 block_sums，对每个子范围的前缀和进行调整，确保整个输入范围的前缀和计算是正确的。