執行緒池 (Thread Pools)

Overview Table

執行緒池基本概念

多數系統中,替每個任務都建立一條執行緒並不切實際。執行緒池 (thread pool) 提供固定數量的工作執行緒共用給所有任務(如同公司的共用車輛):提交的任務掛在任務佇列上,工作執行緒從佇列取出任務執行,完成後再取下一個。

• 關鍵設計問題:執行緒數量、高效的任務分配方式、是否需要等待任務完成
• 執行緒數通常取 std::thread::hardware_concurrency()
• 好處:程式碼只需切成最小工作單元提交,執行緒的生命週期與數量交給池管理

                    ┌────────────────────────────────┐
  submit(f) ──────► │ work_queue (thread_safe_queue) │
                    └───────────────┬────────────────┘
                            try_pop │ (取不到就 yield)
              ┌─────────────────────┼─────────────────────┐
              ▼                     ▼                     ▼
        ┌──────────┐         ┌──────────┐          ┌──────────┐
        │ worker 0 │         │ worker 1 │   ...    │ worker N │   N = hardware_concurrency()
        │ task()   │         │ task()   │          │ task()   │
        └──────────┘         └──────────┘          └──────────┘

最簡單的執行緒池(代碼 9.1)

成員:std::atomic_bool done、thread_safe_queue<std::function<void()>>(第 6 章的執行緒安全佇列)、std::vector<std::thread>、join_threads joiner(第 8 章的 RAII 匯入類)。

void worker_thread() {
    while (!done) {
        std::function<void()> task;
        if (work_queue.try_pop(task))
            task();                      // 取到任務就執行
        else
            std::this_thread::yield();   // 沒任務先讓出 CPU
    }
}

• submit() 把可呼叫物件包成 std::function<void()> 推入佇列;任務無回傳值、不可等待
• 建構函式用 try-catch 包住執行緒啟動:啟動失敗時設定 done 再重新拋出,已啟動的執行緒由 joiner 匯入
• 解構函式只需設 done=true,join_threads 保證所有 worker 結束後池才銷毀

可等待任務:function_wrapper 與新版 submit()(代碼 9.2)

想等待任務完成並取得結果,需要 submit() 回傳 std::future<> —— 把任務包進 std::packaged_task<>。但 std::packaged_task<> 只可移動、不可複製,而 std::function<> 要求可複製建構的函式物件,所以佇列不能再用 std::function<void()>。

解法:自製 function_wrapper —— 一個帶 operator() 的型別擦除 (type erasure) 類:

• 內部 impl_base 有虛擬函式 call();模板衍生類 impl_type<F> 持有 F f,以 std::unique_ptr<impl_base> 保存
• 提供移動建構/移動指派;複製建構與複製指派全部 =delete
• 只需處理「無參數、無回傳」的呼叫,故只是一次簡單的虛擬函式呼叫

template <typename FunctionType>
std::future<typename std::result_of<FunctionType()>::type>
submit(FunctionType f) {
    typedef typename std::result_of<FunctionType()>::type result_type;
    std::packaged_task<result_type()> taskmove(f);
    std::future<result_type> res(task.get_future());
    work_queue.pushmove(task);   // packaged_task 不可複製,只能 move
    return res;
}

• std::result_of<FunctionTypetype 推導 f() 的回傳型別(C++17 起改用 std::invoke_result_t
• 順序:先 get_future() 取得 future → 再 std::move 推入佇列 → 回傳 future
• 異常安全:任務中拋出的異常會存入 future,呼叫端 get() 時重新拋出

用執行緒池改寫 parallel_accumulate(代碼 9.3)

unsigned long const block_size = 25;
unsigned long const num_blocks = (length + block_size - 1) / block_size;
// 每塊 submit 一個任務,最後在 futures[i].get() 累加

• 與第 8 章版本最大差異:工作量依塊數 num_blocks 劃分,而非執行緒數。塊夠小,硬體執行緒越多就有越多塊並發執行,可擴展性 (scalability) 更好
• 不再需要手動打包任務、管理 std::thread 物件與匯入 —— 全交給池
• 異常經由 submit() 回傳的 future 傳遞;若因異常提前離開,池的解構函式會丟棄未完成任務、等待 worker 收工

等待相依任務:run_pending_task() 避免死結(代碼 9.4–9.5)

以平行快速排序為例:每層遞迴把「小於中軸」的半邊 submit() 給池,自己排另一半、再等待 future。問題:池只有固定數量執行緒,當所有 worker 都在等待別的資料塊排序、卻沒有執行緒能去執行那些任務時,就死結了。

解法:在 thread_pool 新增 run_pending_task() —— 把 worker_thread() 主迴圈中「取一個任務執行,否則 yield」的那一次動作抽出來,讓等待中的執行緒也能幫忙消化佇列:

while seconds(0) ==
       std::future_status::timeout) {
    pool.run_pending_task();   // 等待期間不閒著,幫忙執行佇列上的任務
}

• sorter<T> 類模板包裝 thread_pool,do_sort() 用 std::binddo_sort, this, std::move(new_lower_chunk) 提交子任務;std::move 傳遞資料塊,移動比複製便宜
• 執行緒管理邏輯全部移入池中,比第 8 章手寫執行緒 + 待排序堆疊的版本簡潔許多

Exam/Test Patterns

Related Notes

• 09-Advanced-Thread-Management/02-Work-Stealing-and-Queue-Contention
• 09-Advanced-Thread-Management/03-Interrupting-Threads
• 09-Advanced-Thread-Management/Practice-Advanced-Thread-Management
• 04-Synchronizing-Operations/02-Futures-and-Asynchronous-Tasks
• 06-Lock-Based-Data-Structures/02-Thread-Safe-Stack-and-Queue
• 08-Designing-Concurrent-Code/05-Parallel-Algorithm-Case-Studies