Algoritmi paraleli și politici de execuție
Utilizați STL paralel C++17 pentru a paraleliza sortarea, transformarea și reducerea cu politicile de execuție. Înțelegeți când paralelismul ajută, legea lui Amdahl și designul practic al piscinei cu fire.
Parallel STL — Paralelism ușor
C++17 a adăugat politici de execuție la majoritatea funcțiilor
Acesta este cel mai simplu mod de a adăuga paralelism la C++: preluați apelul dvs. existentstd::sort,std::transformsaustd::for_each, adăugațistd::execution::parînainte și lăsați implementarea să facă restul. Desigur, trebuie să vă asigurați că operațiunile dvs. sunt sigure pentru execuția concomitentă (fără curse de date în predicate, fără stare mutabilă partajată).
Politicile de execuție explicate
Există trei politici standard de execuție definite în
#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::cumulate
std::accumulateprocesează elemente strict de la stânga la dreapta și nu pot fi paralelizate.std::reduce(C++17) permite implementării să proceseze elemente în orice ordine și să combine rezultate parțiale - permițând execuția paralelă. Operația trebuie să fie asociativă și comutativă pentru careducesă producă rezultate corecte.
#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;
}Când utilizați politici de execuție paralelă, callable (predicat, funcție de transformare, operator de reducere) pe care îl transmiteți algoritmului poate fi invocat simultan din mai multe fire. Dacă apelabilul dvs. accesează starea mutabilă partajată, aveți o cursă de date:
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
});Use std::atomic, thread-local accumulators, or better yet, use std::reduce / std::count_if care gestionează agregarea în siguranță.
Legea lui Amdahl - Când paralelismul ajută
Legea lui Amdahl afirmă că accelerarea maximă din paralelism este limitată de fracția secvențială a programului tău. Dacă 10% din muncă trebuie făcută secvenţial, atunci chiar şi cu fire infinite, viteza maximă este de 10x.
Speedup = 1 / (S + P/N), unde S este fracția secvențială, P este fracția paralelă și N este numărul de fire.
Paralelismul are, de asemenea, overhead: crearea de fire, sincronizare, disputa în cache și partajare falsă. Pentru seturi de date mici, costul general poate depăși beneficiul. Reguli de bază:
- Sortează:std::sortparalel ajută de obicei peste ~100.000 de elemente.
- Transform/for_each: beneficii de peste ~10.000 de elemente cu lucru netrivial per element.
- Reduceți: Beneficii peste ~100.000 de elemente sau cu operațiuni costisitoare.
- Întotdeauna măsurați — profil cu și fărăstd::execution::par.
Prezentare generală a grupului de fire practice
Standardul C++ nu oferă un pool de fire (de la C++23), dar modelul este simplu: un set fix de fire de lucru extrage sarcini dintr-o coadă partajată. Iată o schiță minimă.
#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;
}Preferațistd::execution::parfață de paralelismul rulat manual pentru algoritmii standard - implementarea vă cunoaște hardware-ul. Verificați întotdeauna că telefonul apelabil nu conține curse de date și efecte secundare. Utilizațistd::reduceîn loc destd::accumulatecând aveți nevoie de paralelism. Benchmark cu dimensiuni realiste ale datelor; pentru intrări mici, secvențial poate fi mai rapid. Amintiți-vă căpar_unseqinterzice apelarea oricărui lucru care obține o blocare (inclusiv alocarea de memorie pe unele platforme).
- Politicile de execuție C++17 (
seq,par,par_unseq) adaugă paralelism la algoritmii standard cu modificări minime de cod std::reduceeste înlocuitorul prietenos cu paralele pentrustd::accumulate— operațiunea dvs. trebuie să fie asociativă și comutativă- Legea lui Amdahl limitează accelerarea: porțiunea secvențială a codului tău devine blocaj
- Algoritmii paraleli au supraîncărcare - beneficii doar pentru seturi de date suficient de mari sau pentru lucrări costisitoare pe element
- Asigurați-vă întotdeauna că apelabilele transmise algoritmilor paraleli sunt lipsite de curse de date
- Pool-urile de fire reutiliza un set fix de fire pentru a evita costul creării de fire pentru multe sarcini mici
Test — Verifică-ți cunoștințele
(15 XP)1. De ce nu poate fi folosit`std::accumulate`cu politici de execuție?
2. Ce constrângere pune`std::execution::par_unseq`asupra apelabilului?
3. Conform legii lui Amdahl, care este viteza maximă dacă 20% dintr-un program este secvenţial?