并行算法与执行策略

利用 C++17 并行 STL 来并行化排序、变换和归约，使用执行策略。了解并行何时有帮助、阿姆达尔定律以及实用的线程池设计。

并行 STL ── 简单的并行化

C++17 为大多数函数添加了执行策略。通过将策略作为第一个参数传递，你可以告诉实现并行化该算法 ── 可能使用多线程、SIMD 或两者 ── 而无需更改任何其他代码。

这是向 C++ 添加并行性最简单的方式：将你现有的 std::sort、std::transform 或 std::for_each 调用前面加上 std::execution::par，让实现来完成其余工作。当然，你必须确保你的操作对并发执行是安全的（谓词中没有数据竞争，没有共享可变状态）。

执行策略详解

中定义了三种标准执行策略。每种策略对并行性和向量化提供不同的保证。

execution_policies.cpp

#include <algorithm>
#include <execution>
#include <vector>
#include <iostream>
#include <chrono>

int main() {
    std::vector<int> data(10'000'000);
    std::iota(data.begin(), data.end(), 0);

    // Sequential — same as no policy
    auto t1 = std::chrono::high_resolution_clock::now();
    std::sort(std::execution::seq, data.begin(), data.end(),
              std::greater<>{});
    auto t2 = std::chrono::high_resolution_clock::now();

    // Parallel — may use multiple threads
    std::sort(std::execution::par, data.begin(), data.end());
    auto t3 = std::chrono::high_resolution_clock::now();

    // Parallel + vectorized — threads AND SIMD
    std::sort(std::execution::par_unseq, data.begin(), data.end(),
              std::greater<>{});
    auto t4 = std::chrono::high_resolution_clock::now();

    auto ms = [](auto a, auto b) {
        return std::chrono::duration_cast<std::chrono::milliseconds>(b - a).count();
    };
    std::cout << "seq: " << ms(t1, t2) << "ms\n";
    std::cout << "par: " << ms(t2, t3) << "ms\n";
    std::cout << "par_unseq: " << ms(t3, t4) << "ms\n";
    return 0;
}

std::reduce 与 std::accumulate

std::accumulate 严格按从左到右的顺序处理元素，无法被并行化。std::reduce（C++17）允许实现以任意顺序处理元素并组合部分结果 ── 从而支持并行执行。操作必须是结合律和交换律的，reduce 才能产生正确的结果。

reduce_demo.cpp

#include <numeric>
#include <execution>
#include <vector>
#include <iostream>

int main() {
    std::vector<double> data(1'000'000, 1.0);

    // Sequential accumulate — always left-to-right
    double sum1 = std::accumulate(data.begin(), data.end(), 0.0);

    // Parallel reduce — may split and combine in any order
    double sum2 = std::reduce(std::execution::par,
                              data.begin(), data.end(), 0.0);

    // Transform-reduce: parallel dot product
    std::vector<double> a(1'000'000, 2.0);
    std::vector<double> b(1'000'000, 3.0);
    double dot = std::transform_reduce(
        std::execution::par,
        a.begin(), a.end(), b.begin(),
        0.0);  // default: multiply + add

    std::cout << "accumulate: " << sum1 << "\n";
    std::cout << "reduce:     " << sum2 << "\n";
    std::cout << "dot product: " << dot << "\n";  // 6'000'000
    return 0;
}

常见陷阱

使用并行执行策略时，你传给算法的可调用对象（谓词、变换函数、归约运算符）可能会从多个线程并发调用。如果你的可调用对象访问了共享的可变状态，就会产生数据竞争：

int count = 0;
std::for_each(std::execution::par, v.begin(), v.end(),
    [&count](int x) {
        if (x > 0) ++count;  // 数据竞争！多个线程递增 count
    });

使用 std::atomic、线程局部累加器，或者更好的做法是使用 std::reduce / std::count_if 来安全地处理聚合。

阿姆达尔定律 ── 并行何时有帮助

阿姆达尔定律指出，并行化带来的最大加速比受到程序中串行部分的限制。如果 10% 的工作必须串行完成，那么即使有无限多的线程，最大加速比也只有 10 倍。

加速比 = 1 / (S + P/N)，其中 S 是串行部分的比例，P 是并行部分的比例，N 是线程数。

并行化也有开销：线程创建、同步、缓存争用和伪共享。对于小数据集，开销可能超过收益。经验法则：

- 排序：并行 std::sort 通常在超过约 100K 个元素时才有帮助。
- Transform/for_each：在每个元素有大量工作的情况下，超过约 10K 个元素时受益。
- Reduce：在超过约 100K 个元素或操作开销大时受益。
- 始终测量 ── 对比使用和不使用 std::execution::par 的性能。

实用线程池概述

C++ 标准（截至 C++23）不提供线程池，但其模式很简单：一组固定的工作线程从共享队列中拉取任务。以下是一个最小化的框架。

thread_pool.cpp

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <functional>
#include <vector>

class ThreadPool {
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stop_ = false;

public:
    explicit ThreadPool(size_t n) {
        for (size_t i = 0; i < n; ++i) {
            workers_.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock lock(mtx_);
                        cv_.wait(lock, [this] {
                            return stop_ || !tasks_.empty();
                        });
                        if (stop_ && tasks_.empty()) return;
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }
                    task();  // run outside the lock
                }
            });
        }
    }

    void enqueue(std::function<void()> task) {
        {
            std::lock_guard lock(mtx_);
            tasks_.push(std::move(task));
        }
        cv_.notify_one();
    }

    ~ThreadPool() {
        {
            std::lock_guard lock(mtx_);
            stop_ = true;
        }
        cv_.notify_all();
        for (auto& w : workers_) w.join();
    }
};

int main() {
    ThreadPool pool(4);
    for (int i = 0; i < 20; ++i) {
        pool.enqueue([i] {
            std::cout << "Task " << i << " on thread "
                      << std::this_thread::get_id() << "\n";
        });
    }
    // pool destructor waits for all tasks
    return 0;
}

最佳实践

对于标准算法，优先使用 std::execution::par 而非手动编写并行化代码 ── 实现了解你的硬件。始终验证你的可调用对象没有数据竞争和副作用。当需要并行时使用 std::reduce 代替 std::accumulate。使用真实数据量进行基准测试；对于小输入，串行可能更快。记住 par_unseq 禁止调用任何获取锁的操作（在某些平台上包括内存分配）。

关键要点

C++17 执行策略（seq、par、par_unseq）以最少的代码更改为标准算法添加并行性
std::reduce 是 std::accumulate 的并行友好替代品 ── 你的操作必须满足结合律和交换律
阿姆达尔定律限制加速比：程序的串行部分成为瓶颈
并行算法有开销 ── 仅对足够大的数据集或每个元素开销大的工作有益
始终确保传给并行算法的可调用对象没有数据竞争
线程池重用一组固定的线程，以避免为许多小任务创建线程的开销

测验 — 检验你的知识

(15 经验值)

1. 为什么 `std::accumulate` 不能与执行策略一起使用？

2. `std::execution::par_unseq` 对可调用对象施加了什么约束？

3. 根据阿姆达尔定律，如果程序的 20% 是串行的，最大加速比是多少？