函數式與訊息傳遞風格 (Functional and Message-Passing Styles)

Overview Table

函數式程式設計與併發

函數式程式設計 (FP):函式結果只依賴傳入的參數,相同參數呼叫多少次都得到相同結果(如 sin、cos、sqrt);純函式不改變任何外部狀態。

• 第3章的結論:不修改共享資料,就沒有競爭條件,也就不需要互斥鎖——這是 FP 對併發最大的簡化(Haskell 中所有函式預設都是純函式)。
• C++11 讓 FP 風格更容易:Lambda 表達式、std::bind、auto 型別推導,而 future 是最後一塊拼圖——future 可在執行緒間傳遞,讓一個計算結果依賴另一個任務的結果,而不必顯式存取共享資料 → FP 風格併發 (FP-style concurrency)。

FP 快速排序:串行版(代碼 4.12)

介面是 FP 的:傳入 list、回傳新的已排序 list(不像 std::sort() 原地修改);內部以 splice() 移動節點避免大量拷貝。步驟:

1. result.splice(result.begin(), input, input.begin()) — 取第一個元素為 pivot 移入結果
2. T const& pivot = *result.begin() — 以引用避免拷貝
3. std::partition(input.begin(), input.end(), [&](T const& t){ return t < pivot; }) — 分成「小於 pivot」與「不小於 pivot」兩段,回傳分割點迭代器
4. lower_part.splice(...) — 小於 pivot 的節點移入新 list
5. 對 lower_part 與剩餘的 input 分別遞迴(用 std::move 傳入)
6. result.splice() 兩次:new_higher 接在 pivot 之後、new_lower 接在 pivot 之前

FP 快速排序:平行版(代碼 4.13)

純函式模式容易轉平行——兩半的排序互不相干,只需把其中一半交給另一個執行緒:

std::future<std::list<T>> new_lower(                        // 差異 1:async
    std::async(&parallel_quick_sort<T>, std::move(lower_part)));
auto new_highermove(input));     // 本執行緒遞迴
result.splice(result.end(), new_higher);
result.splice(result.begin(), new_lower.get());             // 差異 2:get() 等待並取回

input ──splice──▶ pivot(移入 result)
  │
  ├─ std::partition(t < pivot)
  │
  ├─ lower_part ──[std::async 另一執行緒]──▶ new_lower(future)
  └─ input(≥ pivot)──[本執行緒遞迴]───────▶ new_higher
                                                │
result = new_lower.get() ++ [pivot] ++ new_higher ◀─ splice 拼接

• get() 回傳右值引用,結果可移動。
• 執行緒數量倍增:每層遞迴 async 一次 → 遞迴 3 層約 8 執行緒;10 層(約 1000 個元素)可達 1024 個執行緒。
• 預設啟動策略給實作自由:任務超額(超過硬體併發數)時,運行庫可能改為同步執行新任務——在呼叫 get() 的執行緒上跑,避免任務傳遞開銷;明確指定 std::launch::async 則強制每任務一執行緒。行為依實作而異,建議查閱實作文件。

spawn_task:packaged_task + thread 的手工包裝(代碼 4.14)

template<typename F, typename A>
std::future<std::result_of_t<F(A&&)>> spawn_task(F&& f, A&& a) {
    using result_type = std::result_of_t<F(A&&)>;
    std::packaged_task<result_type(A&&)> taskmove(f);
    std::future<result_type> res(task.get_future());
    std::thread tmove(task), std::move(a);
    t.detach();               // 任務完成時 future 自動就緒
    return res;
}

• 本身沒有太多優勢(事實上會造成大規模任務超額),但可作為進階實作的鋪墊:改為「把任務加入佇列 + 執行緒池執行」(第9章)。
• std::async 更適合「已知所有任務」且要讓運行庫全權管理執行緒的情況。

CSP 與 Actor 模型(4.4.2)

CSP(Communicating Sequential Processes,通訊順序進程):

• 沒有共享資料;每個執行緒只根據收到的訊息獨立運行 → 每個執行緒就是一個狀態機:收到訊息時更新狀態,並可能向其他執行緒發出訊息。
• 被 Erlang 採納,也是 MPI(Message Passing Interface)高效能運算的基礎。
• C++ 執行緒共享位址空間,無法「真正」隔離 → 靠約定:由應用/庫的作者負責確保不共享資料;訊息佇列本身必須共享,但細節封裝在訊息傳遞庫中。

Actor 模型(參與者模式):系統中有多個獨立的參與者 (actor),各自運行在獨立執行緒上,互相發送訊息執行任務,除了透過訊息傳入的內容外不共享任何狀態。

ATM 狀態機範例(代碼 4.15、4.16)

為 ATM(自動提款機)分三個執行緒:處理實體機械、ATM 邏輯、與銀行通訊——互不共享資料,只互發訊息(如插卡時,機械執行緒發訊息給邏輯執行緒;邏輯執行緒發「吐鈔金額」訊息給機械執行緒)。

ATM 邏輯建模為狀態機(圖 4.3 簡化):

          卡片插入              PIN 輸入完成(滿4位)
┌──────────────────┐    ┌─────────────┐    ┌───────────────┐
│ waiting_for_card │──▶│ getting_pin │──▶│ verifying_pin │
└──────▲───────────┘    └──┬───▲──┬───┘    └──────┬────────┘
       │           digit/清除│   │  │取消          │
       │            (自身迴圈)└───┘  ▼           驗證失敗──▶ 退卡 ─┐
       │                     done_processing ◀── 取消          │
       │                            ▲          驗證通過        │
       └────── 退卡、等下一位 ◀─────┴── 取款:等銀行確認 ──────┘
                                        (吐鈔 / 餘額不足)

實作:每個狀態一個成員函式,狀態轉移 = 改變成員函式指標;主迴圈反覆呼叫當前狀態函式:

void atm::waiting_for_card() {
    interface_hardware.send(display_enter_card());
    incoming.wait()                              // 等待傳入訊息
        .handle<card_inserted>([&](card_inserted const& msg) {
            account = msg.account; pin = "";
            interface_hardware.send(display_enter_pin());
            state = &atm::getting_pin;           // 狀態轉移
        });
}
// run(): state = &atm::waiting_for_card; for(;;) (this->*state)();

• wait() 後可鏈接多個 handle<MsgType>()(如 getting_pin 處理 digit_pressed / clear_last_pressed / cancel_pressed 三種);型別不匹配的訊息會被丟棄,執行緒繼續等待,直到收到匹配型別的訊息。
• 收到訊息不一定轉移狀態:digit_pressed 只把數字加進 pin,滿 4 位才發 verify_pin 給銀行並轉入 verifying_pin。

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/05-Continuations-Latches-and-Barriers
• 04-Synchronizing-Operations/Practice-Synchronizing-Operations
• 12-Appendix/03-ATM-Message-Passing-Example
• 08-Designing-Concurrent-Code/01-Dividing-Work-Between-Threads
• 10-Parallel-Algorithms/01-Execution-Policies