持續性、Latch 與 Barrier (Continuations, Latches and Barriers)

Overview Table

並發技術規範與 experimental::future

• 並發技術規範在 std::experimental 命名空間提供新版 promise 與 packaged_task:與 std 版的差別在於回傳 std::experimental::future(<experimental/future>),具備持續性能力。
• std::experimental 中的語法/語義,納入 C++ 標準後可能有所不同。

then():為 future 添加持續性

用 std::future 處理「結果就緒後的下一步」必須以 wait()/get()/wait_for() 阻塞等待,程式碼複雜。持續性的意義是「完事俱備,只等資料」——future 就緒時自動執行後續處理:

std::experimental::future<int> fut = find_the_answer();
auto fut2 = fut.then(find_the_question);  // 就緒時自動調用
assert(!fut.valid());                     // 原 future 失效
assert(fut2.valid());                     // 新 future 持有持續性的結果

• 存入的值只能檢索一次:.then() 之後對原 fut 的操作非法,.then() 回傳新 future 持有持續性結果。
• 持續性不能自行傳參:運行庫會傳入就緒的 future 本身(而非解引用後的值)——若 find_the_answer 回傳 int,持續性函式簽名須為 std::string find_the_questionfuture<int>)。原因:future 中可能存的是值或例外;傳 future 讓持續性函式自行 .get(,例外便由持續性處理或沿鏈傳播,而非由運行庫決定。
• 持續性在哪個執行緒上執行未指定——實作有自由(執行緒池等)。

spawn_async(代碼 4.17,與 std::async 等價的手工版):

std::thread t([p = std::move(p), f = std::decay_t<Func>(func)]() mutable {
    try { p.set_value_at_thread_exit(f()); }
    catch (...) { p.set_exception_at_thread_exitcurrent_exception(); }
});
t.detach();
return res;   // res = p.get_future()

• 用 set_value_at_thread_exit / set_exception_at_thread_exit:確保 future 就緒前 thread_local 變數已清理完畢。

持續性連接鏈(4.4.4)

使用者登入流程(驗證 → 取帳戶資訊 → 更新畫面)的四個版本:

return backend.async_authenticate_user(username, password)
    .thenfuture<user_id> id {
        return backend.async_request_current_info(id.get());   // 回傳 future → 自動展開
    })
    .thenfuture<user_data> info {
        try { update_display(info.get()); }
        catch exception& e) { display_error(e; }        // 全鏈例外統一處理
    });

• future 展開 (future-unwrapping):持續性函式回傳 future<T> 時,.then() 的結果型別是 future<T> 而不是 future<future<T>>——異步函式鏈因此不會阻塞。
• 例外沿鏈傳播:上游任一環拋出 → 下游呼叫 .get() 時拋出,最末端的 catch 可統一處理。
• 支援 C++14 泛型 Lambda 時,參數可寫 auto。

std::experimental::shared_future 也支援持續性,差異:

• 可以有多個持續性(單一持續性會使多執行緒同時添加時產生競爭條件)。
• 持續性函式的參數必須是 shared_future;但持續性回傳的值仍是 experimental::future(目前無法共享)。

when_all:等待全部就緒(4.4.5)

用 std::async 收集多個 future(代碼 4.22)的缺點:收集任務佔用一條執行緒,對每個未就緒的 f.get() 反覆「喚醒 → 發現未就緒 → 再休眠」,增加上下文切換與額外開銷。

std::experimental::when_all(first, last) 回傳新 future:全部傳入的 future 就緒時才就緒,搭配 .then 無阻塞收集:

return std::experimental::when_all(results.begin(), results.end())
    .then([](auto ready_results) {   // future<vector<experimental::future<ChunkResult>>>
        std::vector<ChunkResult> v;
        for (auto& f : ready_results.get())
            v.push_back(f.get());    // 不會阻塞:此刻全部已就緒
        return gather_results(v);
    });

when_any:等待第一個就緒(4.4.6)

情境:多個任務平行搜尋資料子集,找到任一符合值即可;第一個完成者觸發後續處理,其他任務透過共享的 std::atomic<bool> done_flag 提早停止。

• when_any 回傳 future<when_any_result<Collection>>;when_any_result 含兩個成員:futures(整個集合)與 index(觸發就緒者的索引)。
• 遞迴持續性模式(DoneCheck 函式物件):
  1. results.futures[results.index].get() 取就緒者的結果
  2. 找到 → final_result->set_value(...)
  3. 沒找到 → 從集合移除該 future;若還有剩 → 對剩餘再呼叫 when_anymove(*this));空了 → set_exception("Not found"
• when_all / when_any 皆有迭代器範圍版與可變參數版;可變參數版的結果存 tuple。futures 以值傳遞 → 需顯式 std::move 傳入或用臨時變數。

latch:一次性計數同步(4.4.8)

latch(鎖存器)是輕量級同步物件:計數器減到 0 時進入就緒態,之後保持就緒直到銷毀——適合同步「一組一次性事件」。

• 不在乎是誰遞減:同一執行緒可減多次、多執行緒各減一次皆可。

std::experimental::latch done(thread_count);
for (unsigned i = 0; i < thread_count; ++i)
    threads.push_back(std::async(std::launch::async, [&, i] {
        data[i] = make_data(i);
        done.count_down();   // 資料就緒即遞減,不必等整個任務結束
        do_more_stuff();
    }));
done.wait();                 // 只等「資料就緒」,不等任務全部完成
process_data(data, thread_count);

• Lambda [&, i]:i 以值捕獲(迴圈計數器,引用捕獲會導致資料競爭與未定義行為),其餘以引用捕獲。
• count_down 的呼叫與 wait 的返回同步 (synchronizes-with) → 主執行緒讀取 data 是安全的(底層記憶體順序見 05-Memory-Model-and-Atomics/03-Synchronizes-With-and-Happens-Before)。

barrier:可重複的組同步(4.4.9)

barrier(柵欄)用於一組執行緒間的可重複同步:<experimental/barrier> 提供 std::experimental::barrier(簡單、低開銷)與 std::experimental::flex_barrier(靈活、開銷較大)。

• 建構時指定組內執行緒數;每個執行緒完成本週期工作後呼叫 arrive_and_wait() 到達柵欄並等待;最後一個到達時全體釋放、柵欄重置,進入下一週期。
• 每個執行緒每週期只能到達柵欄一次(latch 則不限)。
• arrive_and_drop():本執行緒明確退出組 → 下一週期需要到達的執行緒數減 1。

代碼 4.26 的流水線模式(來源 source → 平行處理 → 輸出 sink):

每個週期(處理一個 data_block):
  thread 0                 thread 1..N-1
  讀取 + 切塊(串行)          (等待)
     │                          │
     ├── sync.arrive_and_wait() ┤   ◀ 同步點 1:確保資料切好才開工
     │                          │
  process(chunks[0])       process(chunks[i])   ◀ 平行處理各自區塊
     │                          │
     ├── sync.arrive_and_wait() ┤   ◀ 同步點 2:確保全部寫入 result
     │                          │
  sink.write_data(result)   (等待 / 下一週期)   ◀ 串行部分只在 0 號執行緒

flex_barrier:帶完成階段(4.4.10)

std::experimental::flex_barrier 多一個建構參數:完成函式——所有執行緒到達柵欄後,由其中一個執行緒執行它(天然的串行區)。

• 完成函式的回傳值決定下一週期的參與執行緒數:−1 → 維持不變;0 或其他數值 → 指定為該數(可增可減)。
• 代碼 4.27:把「輸出結果 + 切下一塊資料」放進完成函式 → 工作執行緒主迴圈只剩平行部分 + 單一同步點:

std::experimental::flex_barrier sync(num_threads, [&] {
    sink.write_datamove(result);
    split_source();   // 準備下一輪資料(在 0 號執行緒等單一執行緒上跑)
    return -1;        // 下一週期執行緒數不變
});
// 工作執行緒:while (!source.done()) { result.set_chunk(..., process(chunks[i])); sync.arrive_and_wait(); }

• 應用:流水線——初始與收尾階段所需執行緒數少於主處理階段,可動態調整參與數。

Exam/Test Patterns

Related Notes

• 04-Synchronizing-Operations/01-Condition-Variables-and-Thread-Safe-Queue
• 04-Synchronizing-Operations/02-Futures-and-Asynchronous-Tasks
• 04-Synchronizing-Operations/03-Waiting-with-Time-Limits
• 04-Synchronizing-Operations/04-Functional-and-Message-Passing-Styles
• 04-Synchronizing-Operations/Practice-Synchronizing-Operations
• 05-Memory-Model-and-Atomics/03-Synchronizes-With-and-Happens-Before
• 08-Designing-Concurrent-Code/01-Dividing-Work-Between-Threads