Algoritmi paralleli e policy di esecuzione
Sfrutta la STL parallela del C++17 per parallelizzare ordinamento, trasformazione e riduzione con le policy di esecuzione. Comprendi quando il parallelismo conviene, la legge di Amdahl e la progettazione pratica di un thread pool.
STL parallela — parallelismo facile
Il C++17 ha aggiunto le policy di esecuzione alla maggior parte delle funzioni di
È il modo più semplice per aggiungere parallelismo in C++: prendi la tua chiamata esistente a std::sort, std::transform o std::for_each, anteponi std::execution::par e lascia che sia l'implementazione a fare il resto. Naturalmente devi assicurarti che le tue operazioni siano sicure per l'esecuzione concorrente (nessun data race nei predicati, nessuno stato condiviso modificabile).
Le policy di esecuzione spiegate
Esistono tre policy di esecuzione standard, definite in
#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::accumulate
std::accumulate elabora gli elementi rigorosamente da sinistra a destra e non può essere parallelizzata. std::reduce (C++17) permette all'implementazione di elaborare gli elementi in qualsiasi ordine e di combinare i risultati parziali — rendendo possibile l'esecuzione parallela. Perché reduce produca risultati corretti, l'operazione deve essere associativa e commutativa.
#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;
}Quando usi le policy di esecuzione parallele, il callable (predicato, funzione di trasformazione, operatore di riduzione) che passi all'algoritmo può essere invocato concorrentemente da più thread. Se il tuo callable accede a uno stato condiviso modificabile, hai un data race:
int count = 0;
std::for_each(std::execution::par, v.begin(), v.end(),
[&count](int x) {
if (x > 0) ++count; // DATA RACE! Più thread incrementano count
});Usa std::atomic, accumulatori thread-local o, meglio ancora, std::reduce / std::count_if, che gestiscono l'aggregazione in sicurezza.
La legge di Amdahl — quando il parallelismo conviene
La legge di Amdahl afferma che lo speedup massimo ottenibile dal parallelismo è limitato dalla frazione sequenziale del tuo programma. Se il 10% del lavoro deve essere svolto in sequenza, allora anche con infiniti thread lo speedup massimo è 10x.
Speedup = 1 / (S + P/N), dove S è la frazione sequenziale, P è la frazione parallela e N è il numero di thread.
Il parallelismo ha anche un suo overhead: creazione dei thread, sincronizzazione, contesa sulla cache e false sharing. Su insiemi di dati piccoli, l'overhead può superare il beneficio. Regole pratiche:
- Ordinamento: lo std::sort parallelo conviene tipicamente sopra i ~100.000 elementi.
- Transform/for_each: conviene sopra i ~10.000 elementi, con lavoro non banale per elemento.
- Reduce: conviene sopra i ~100.000 elementi oppure con operazioni costose.
- Misura sempre — profila con e senza std::execution::par.
Panoramica pratica di un thread pool
Lo standard C++ non fornisce un thread pool (nemmeno in C++23), ma il pattern è semplice: un insieme fisso di thread worker preleva i task da una coda condivisa. Ecco uno schema minimale.
#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;
}Per gli algoritmi standard preferisci std::execution::par al parallelismo scritto a mano — l'implementazione conosce il tuo hardware. Verifica sempre che il tuo callable sia privo di data race e di effetti collaterali. Usa std::reduce al posto di std::accumulate quando ti serve il parallelismo. Fai benchmark con dimensioni di dati realistiche: per input piccoli la versione sequenziale può essere più veloce. Ricorda che par_unseq vieta di chiamare qualsiasi cosa acquisisca un lock (inclusa, su alcune piattaforme, l'allocazione di memoria).
- Le policy di esecuzione del C++17 (
seq,par,par_unseq) aggiungono parallelismo agli algoritmi standard con modifiche minime al codice std::reduceè il sostituto distd::accumulateadatto al parallelismo — la tua operazione deve essere associativa e commutativa- La legge di Amdahl limita lo speedup: la porzione sequenziale del tuo codice diventa il collo di bottiglia
- Gli algoritmi paralleli hanno un overhead — convengono solo con insiemi di dati sufficientemente grandi o con lavoro costoso per elemento
- Assicurati sempre che i callable passati agli algoritmi paralleli siano privi di data race
- I thread pool riutilizzano un insieme fisso di thread per evitare il costo di crearne uno nuovo per ogni piccolo task
Quiz — Metti alla prova le tue conoscenze
(15 XP)1. Perché `std::accumulate` non può essere usata con le policy di esecuzione?
2. Quale vincolo impone `std::execution::par_unseq` al callable?
3. Secondo la legge di Amdahl, qual è lo speedup massimo se il 20% di un programma è sequenziale?