Enter your search text in the box above

Select a result to preview

Future 與非同步任務 (Futures and Asynchronous Tasks)

#cpp-concurrency #concept #synchronization #future #promise #async #packaged-task

Overview Table

概念 重點
std::future<T> 一次性事件的結果;只能移動;get() 只能取一次;就緒後不可重置
std::async 啟動非同步任務並回傳 future;可用 std::launch 指定啟動策略
std::packaged_task<Sig> 將 future 綁定到函式/可調用物件;本身可調用、可入佇列 → 任務調度器/執行緒池
std::promise<T> 最底層:顯式 set_value() / set_exception() 設定結果
例外與 future 任務拋出的例外存入 future,get() 時重拋;未設值即銷毀 → broken_promise
std::shared_future 可拷貝;每個執行緒持有自己的副本即可安全存取,無需加鎖

future:一次性事件的等待

如同在機場等待「登機廣播」:等待期間可以做其他事(看書、喝咖啡),但真正等的是那個一次性事件。C++ 標準庫以 future 為這種事件建模,皆宣告於 <future>:

型別 類比 特性
std::future<> std::unique_ptr 只能與一個事件關聯;只能移動
std::shared_future<> std::shared_ptr 多個實例可關聯同一事件;可拷貝
future 物件本身不提供同步存取

多個執行緒存取同一個 future 物件時,必須用互斥鎖等機制保護;但多個執行緒各自存取自己的 std::shared_future 副本(即使指向同一結果)則不需同步。

std::async 與啟動策略

std::thread 沒有接收回傳值的機制;std::async 啟動非同步任務並回傳持有結果的 std::future(代碼 4.6):

std::future<int> the_answer = std::async(find_the_answer_to_ltuae);
do_other_stuff();
std::cout << the_answer.get();   // 阻塞直到就緒,並取回結果

啟動策略以額外的第一個參數 std::launch 指定:

策略 行為
std::launch::async 必須在獨立的新執行緒上執行
std::launch::deferred 延遲到 wait() / get() 被呼叫時,才在呼叫端執行緒上執行
deferred | async(預設) 由實作自行選擇兩者之一
deferred 的任務可能永不執行

預設策略允許實作選擇延遲執行;若函式呼叫被延遲,且從未對該 future 呼叫 wait() 或 get(),這個函式就永遠不會執行。需要保證併發時,必須明確指定 std::launch::async。

std::packaged_task<>:future 與任務綁定

std::packaged_task<> 將 future 綁定到函式或可調用物件——調用 packaged_task 就會調用底層函式,回傳值作為非同步結果存入 future。

在執行緒間傳遞任務(GUI 範例,代碼 4.9 核心):

std::deque<std::packaged_task<void()>> tasks;  // 受互斥鎖保護的任務佇列
template<typename Func>
std::future<void> post_task_for_gui_thread(Func f) {
    std::packaged_task<void()> task(f);        // 打包任務
    std::future<void> res = task.get_future(); // 先取得 future
    std::lock_guard<std::mutex> lk(m);
    tasks.push_backmove(task);          // packaged_task 只能移動
    return res;                                // 呼叫端可等待任務完成
}

GUI 執行緒在迴圈中從佇列取出 task 並執行 task();任務完成時對應的 future 變為就緒。

std::promise<>:顯式設值

當任務無法表達成單一函式呼叫、或結果來自多處時,使用最底層的 std::promise<T>:

if (connection->has_incoming_data()) {
    data_packet data = connection->incoming();
    std::promise<payload_type>& p = connection->get_promise(data.id);
    p.set_value(data.payload);   // 對應的 future 立即就緒
}

將例外存入 future

同步呼叫會拋出例外的函式,改成非同步後例外也要能傳給呼叫端:

來源 例外如何進入 future
std::async 函式拋例外 → 存入 future、狀態置為就緒,get() 時重拋
std::packaged_task 打包函式被調用時拋例外 → 同上
std::promise 顯式呼叫 set_exception()(而非 set_value()),通常在 catch 區塊中
違背 promise 未設值/未調用任務即銷毀 promise 或 packaged_task → future 存入 std::future_error,錯誤碼 std::future_errc::broken_promise 
try {
    some_promise.set_value(calculate_value());
} catch (...) {
    some_promise.set_exceptioncurrent_exception();  // 捕獲當前例外存入
}
兩個細節

  1. 重拋的例外是原始物件還是拷貝,標準未指定,各編譯器/庫實作不同。
  2. 書中的 std::copy_exception() 是標準化前的名稱,正式進入標準後為 std::make_exception_ptr()。

std::shared_future:多執行緒等待

三種建立方式:

std::shared_future<int> sfmove(f));            // 1 顯式移動;之後 !f.valid(
std::shared_future<std::string> sf2(p.get_future()); // 2 右值隱式轉移所有權
auto sf3 = p.get_future().share();                   // 3 share() + auto 型別推導

              ┌─ std::async(f)            ── 最省事:啟動非同步任務取回傳值
future 來源 ──┼─ std::packaged_task<Sig>  ── 打包任務:自行決定何時、在哪個執行緒執行
              └─ std::promise<T>          ── 完全手動:顯式 set_value / set_exception
                        │
                 get_future()
                        ▼
               std::future<T> ──move / share()──▶ std::shared_future<T>
               (只移動,get 一次)                        (可拷貝,每執行緒一份副本)

Exam/Test Patterns

情境/關鍵字 答案
「std::thread 拿不到回傳值」 std::async + future.get()
「async 的任務一直沒執行」 預設策略含 deferred(延遲到 get/wait);要保證新執行緒須 std::launch::async
「執行緒池/任務調度的任務單元」 std::packaged_task(future 綁定可調用物件)
「結果來自多處、不能包成函式呼叫」 std::promise 顯式 set_value
「非同步任務拋出例外」 例外存入 future,get() 時重拋
「promise 沒設值就銷毀」 std::future_error,錯誤碼 std::future_errc::broken_promise
「多執行緒等待同一結果」 std::shared_future,每執行緒一份副本(拷貝)
「future 的 get() 想呼叫兩次」 不行:std::future 結果只能取一次 → 改用 shared_future