Параллельные алгоритмы и исполнительная политика
Использование параллельного STL C++17 для параллелизации сортировки, преобразования и сокращения с политикой исполнения. Поймите, когда параллелизм помогает, закон Амдала и практический дизайн бассейна потоков.
Параллельный STL — легкий параллелизм
C++17 добавил политики исполнения к большинству функций
Это самый простой способ добавить параллелизм в C++: возьмите существующий вызов std::sort, std::transform или std::for_each, предоставьте std::execution::par и позвольте реализации сделать все остальное. Конечно, вы должны убедиться, что ваши операции безопасны для одновременного выполнения (без гонок данных в предикатах, без общего изменяемого состояния).
Политика исполнения объяснена
В
#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 vs. std::acumulate
std::accumulate обрабатывает элементы строго слева направо и не может быть распараллелен. std::reduce (C++17) позволяет осуществлять обработку элементов в любом порядке** и комбинировать частичные результаты, обеспечивая параллельное выполнение. Операция должна быть ассоциативной и коммутативной для reduce для получения правильных результатов.
#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;
}При использовании параллельных политик исполнения callable (предикат, функция преобразования, оператор восстановления), который вы передаете алгоритму, может одновременно вызываться из нескольких потоков. Если вы получаете доступ к общему изменяемому состоянию, у вас есть гонка данных:
int count = 0;
std::for_each(std::execution::par, v.begin(), v.end(),
[&count](int x) {
if (x > 0) ++count; // DATA RACE! Multiple threads increment count
});Используйте std::atomic, резьбо-локальные аккумуляторы или, еще лучше, используйте std::reduce / std::count_if, которые безопасно обрабатывают агрегацию.
Закон Амдала — Когда параллелизм помогает
Закон Амдала гласит, что максимальное ускорение от параллелизма ограничено последовательной фракцией вашей программы. Если 10% работы нужно делать последовательно, то даже при бесконечных нитях максимальное ускорение 10х.
Скорость = 1/ (S + P/N), где S — последовательная дробь, P — параллельная дробь, а N — число потоков.
Параллелизм также имеет накладные расходы : создание потоков, синхронизация, борьба с кэшем и ложный обмен. Для небольших наборов данных накладные расходы могут превышать выгоду. Правила большого пальца:
- Сорт : Параллельный std::sort обычно помогает выше ~100K элементов.
Трансформация/для каждого: Преимущество выше ~ 10K элементов с нетривиальной работой на элемент.
Уменьшите : Преимущества выше ~100K элементов или с дорогостоящими операциями.
measure — профиль с std::execution::par и без него.
Практический обзор Thread Pool
Стандарт C++ не обеспечивает пул потоков (как у C++23), но шаблон прост: фиксированный набор рабочих потоков вытягивает задачи из общей очереди. Вот минимальный эскиз.
#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 XP)1. Почему `std::accumulate` не может быть использован с политикой исполнения?
2. Какое ограничение размещает `std::execution::par_unseq` на телефоне?
3. Согласно закону Амдала, каково максимальное ускорение, если 20% программы последовательны?