平行演算法案例研究 (Parallel Algorithm Case Studies)

#cpp-concurrency #concept #concurrent-design #data-partitioning #exception-safety

Overview Table

案例	核心挑戰	關鍵技術
平行版 `std::for_each`	無回傳值,但異常仍要傳給呼叫者	`packaged_task<void(void)>` + `future<void>`;或 `std::async` 遞迴對半劃分
平行版 `std::find`	找到就該停——如何提前中斷其他執行緒	`std::atomic<bool>` done flag + `std::promise` 存結果/異常
平行版 `std::partial_sum`	第 i 個結果依賴前 i 個元素,塊間有依賴	法一:逐段傳播(promise/future 流水線);法二:barrier + 2 的冪距離

本節把 8.1–8.4 的設計考量(劃分方式、效能因素、資料存取模式、異常安全、可擴展性)套用到三個標準庫演算法的平行實作上。示範用途,非最前沿實作;生產環境優先用 C++17 平行演算法(第 10 章)或專家庫(如 Intel TBB)。

平行版 std::for_each(代碼 8.7、8.8)

對範圍中每個元素套用同一函數;平行版不保證處理順序。元素數事先已知 → 事前塊狀劃分(8.1.1),以 hardware_concurrency() 決定執行緒數,連續區塊順便避免偽共享(8.2.3)。

與 parallel_accumulate 的差異:不需要回傳值——但工作執行緒的異常仍要轉移,所以用 std::packaged_task<void(void)> + std::future<void>。
lambda 把 std::for_each(block_start, block_end, f) 包進任務;join_threads RAII 保證匯合(異常安全)。
futures[i].get() 只為了檢索異常,沒有值可取;若不在乎傳遞異常,這一步可省略。

std::async 遞迴版(代碼 8.8)更精簡——執行時對半劃分,不必事先算執行緒數:

template<typename Iterator,typename Func>
void parallel_for_each(Iterator first,Iterator last,Func f){
  unsigned long const length=std::distance(first,last);
  if(!length) return;
  unsigned long const min_per_thread=25;
  if(length<(2*min_per_thread)){ std::for_each(first,last,f); return; }
  Iterator const mid_point=first+length/2;
  std::future<void> first_half=
      std::async(&parallel_for_each<Iterator,Func>,first,mid_point,f);
  parallel_for_each(mid_point,last,f);  // 後半直接遞迴
  first_half.get();                     // 只為傳播異常
}

兩版取捨

手動版可精準控制執行緒數;std::async 版由標準庫調度、不會超額申請,且 future 解構會等待執行緒 → 天然異常安全(同代碼 8.5 的邏輯,見 08-Designing-Concurrent-Code/04-Exception-Safety-and-Scalability)。

平行版 std::find(代碼 8.9、8.10)

std::find 的特性:找到第一個符合就可以停,不必處理所有元素(同類還有 std::equal、std::any_of)。演算法屬性直接影響平行設計——若不中斷其他執行緒,平行版可能比串行還慢:串行版找到就停,平行版每條執行緒卻把自己那 1/N 全部掃完。

提前結束:atomic<bool> done flag

每處理一個元素就檢查一次 done flag;有執行緒找到 → 設定結果 → 設 done flag → 其他執行緒看到後停止。
未處理的資料保持原樣;多數情況比串行快。代價:每次迭代 load 原子變數是慢操作,會拖慢每條執行緒。

 執行緒1: [e0 ..e24]  逐元素檢查 done ──────────┐
 執行緒2: [e25..e49]  找到 e30! ── set_value ──▶ done_flag = true
 執行緒3: [e50..e74]  檢查 done ◀───────────────┤
 執行緒4: [e75..e99]  檢查 done ◀───────────────┘ → 看到 true,提前返回

為何選 std::promise 而非 packaged_task

選項	異常行為	適用
future 陣列 + `std::packaged_task`	各執行緒異常各自保存;其他執行緒繼續查找	想容忍個別異常、繼續搜完
單一 `std::promise`	第一個結果或異常直接勝出並停止全部	行為最接近 `std::find` → 書中選擇

工作執行緒(find*element)的異常處理:catch(...) → set_exceptioncurrent_exception()) + 設 done flag;若 promise 已被別人設定,set*\*會再拋 → 內層catch(... 吞掉即可。
多執行緒同時找到/拋異常會競爭 promise——這是良性競爭條件:先設定者就是「第一個返回者」。

值不在範圍內的死等陷阱

沒有任何執行緒設定 promise 時,future.get() 會永遠等待。必須先等全部執行緒匯合(join),再檢查 done flag:未設定 → 回傳 last;有設定 → 才呼叫 result.get_future().get() 取值或重拋異常。

std::async 遞迴版(代碼 8.10)

done flag 以 std::ref 沿遞迴傳遞(從主入口 parallel_find 傳入 parallel_find_impl)。
小塊直接迴圈查找(邊查邊檢查 done);大塊找中點,後半 std::async、前半直接遞迴;前半回傳 mid_point 表示沒找到 → 才 async_result.get()。
整個函數包 try/catch:catch(...){ done=true; throw; } —— 一有異常就設 done,所有分支快速終止;不加 try/catch 也正確,只是要等所有執行緒做完。
future 解構會等待對應執行緒 → 遞迴中拋異常也不會有執行緒洩漏;非同步端的異常由 get() 傳播。

語義差異(平行化的必要屬性)

平行演算法不能保證按順序處理元素。parallel_find 即使前段有匹配,也可能回傳範圍後段的匹配——與 std::find「回傳第一個」的順序語義不同。元素必須可獨立處理,演算法才能平行化;有順序依賴就不能直接併發。

平行版 std::partial_sum(代碼 8.11–8.13)

std::partial_sum 把序列就地換成前綴和:[1,2,3,4,5] → [1,3,6,10,15]。困難在於第 i 個結果依賴前面全部元素——不能把任務切塊後完全獨立計算。兩種解法對應兩種硬體形態。

法一:逐段傳播(代碼 8.11)——處理器少、元素多

各塊先獨立算部分和,再把前一塊的末值傳給下一塊全體元素:

 原始:            [1 2 3 | 4 5 6 | 7 8 9]
 步1 各塊獨立求和: [1 3 6 | 4 9 15 | 7 15 24]
 步2 塊1末值6→塊2: [1 3 6 | 10 15 21 | 7 15 24]
 步3 塊2末值21→塊3:[1 3 6 | 10 15 21 | 28 36 45]

流水線關鍵:拿到前塊末值後,先更新本塊最後一個元素並立即 set_value 到 promise → 下一塊可馬上開工;再回頭用 std::for_each 更新本塊其餘元素 → 平行度最大化。
塊間以 std::promise<value_type>future<value_type>(前塊末值串接;第一塊沒有 previous future,最後一塊不需要 promise。
異常處理:catch(...) 存入 promise → 下一塊 get() 時重拋;最後一塊直接重拋(該塊由主執行緒執行)。
不能用 std::async 改寫:任務中途要等前塊結果,所有任務必須真正同時執行,延遲執行會死等。
成本:每執行緒約 N/k 次運算 + N/k 次結果傳遞 → 總工作量 O(N)。

法二:2 的冪距離相加(代碼 8.13)——大規模並行(處理器 ≥ 元素)

處理器多於元素時,法一的塊狀劃分毫無意義。改為每步與距離 stride 的元素相加,stride 每步翻倍:

 初始:    [1  2  3  4  5  6  7  8  9]
 步1 d=1: [1  3  5  7  9 11 13 15 17]
 步2 d=2: [1  3  6 10 14 18 22 26 30]
 步3 d=4: [1  3  6 10 15 21 28 36 44]
 步4 d=8: [1  3  6 10 15 21 28 36 45]   ← 共 log2(N) 步完成

每步每個處理器處理一個元素(SIMD 式:單指令-多資料流);log₂(N) 步完成,但總操作量 O(N log N) > 法一的 O(N)。
選擇準則:處理單元少 → 法一;大規模並行系統 → 法二(每處理器每步都有活幹,不必等待傳遞)。
每步之間所有執行緒必須同步——先做完的不能搶跑,否則會改掉別人下一步要讀的資料 → 需要 barrier(柵欄)。

自製 barrier(代碼 8.12)與其演進

三個成員:count(座位數)、spaces(剩餘座位)、generation(第幾輪):

void wait(){
  unsigned const my_generation=generation;
  if(!--spaces){ spaces=count; ++generation; } // 最後一人:重置座位、放行本輪
  else
    while(generation==my_generation)
      std::this_thread::yield();               // 自旋等待,讓出 CPU
}

元素 i 只需要 stride ≤ i 的那幾步就定案 → 先完工的執行緒呼叫 done_waiting():遞減 count(改為 std::atomic<unsigned>)並遞減 spaces,之後輪次不再等它;最後一個完成者不需等待。
雙緩衝:每步從「原始資料或 buffer」其中一邊讀、寫到另一邊(依 step 奇偶交替)→ 讀與寫由不同位置承擔,無競爭條件;結束時依 update_source 決定是否把 buffer 值寫回原始資料。

簡易 barrier 的限制與異常安全

自旋等待不適合長時間等待;呼叫 wait() 的執行緒超過 count 個就會失效;全域同步在大規模架構上開銷巨大——barrier 狀態的 cache line 會乒乓(見 08-Designing-Concurrent-Code/02-Factors-Affecting-Performance)。支援併發 TS 時可用 std::experimental::barrier(第 4 章)。
代碼 8.13 不是異常安全的:process_element 拋異常會直接 std::terminate。修補方式:如 parallel_find 用 std::promise 存異常,或用 mutex 保護的 std::exception_ptr。

Exam/Test Patterns

情境/關鍵字	答案
「for_each 沒有回傳值,還需要 future 嗎」	需要:`future<void>` 只為轉移工作執行緒的異常,`get()` 檢索異常;不在乎可省略
「find 類演算法平行化的關鍵」	提前中斷:`atomic<bool>` done flag,每個元素檢查一次;不中斷可能比串行慢
「parallel_find 為何用 promise 不用 packaged_task」	promise = 第一個結果/異常勝出、全部停止,行為最接近 `std::find`;packaged_task 讓其他執行緒繼續找
「要找的值不在範圍內,get() 卡死」	promise 永遠不會被設定 → 先 join 全部執行緒、檢查 done flag,未設定回傳 `last`
「多執行緒同時找到,誰贏」	對 promise 的良性競爭條件:先 set 者作為「第一個」返回
「partial_sum 兩種平行法怎麼選」	逐段傳播 O(N) → 處理器少;2 的冪距離 O(N log N)、log₂N 步 → 大規模並行(處理器 ≥ 元素)
「法二為何需要 barrier + 雙緩衝」	每步全體同步,防止搶跑改壞資料;雙緩衝讓讀寫分離、無競爭條件
「barrier 的 done_waiting() 用途」	先完工的執行緒退出同步圈(遞減 count),其餘輪次不再等它;最後完成者不等待
「parallel_find 與 std::find 結果可能不同?」	可能:平行版不保證回傳最前面的匹配——順序語義是平行化犧牲的代價