C++ 17: 以 std::scoped_lock 避免 Dead Lock

最佳化多執行緒程式時,為了減少非必要阻塞(Blocking),常見作法是將一個 Mutex 依照保護對象拆分為多個 Mutex。如此一來,一個執行緒只需鎖定它需要的 Mutex,不必等待它不需要的 Mutex。有時我們會進一步實作「並行資料結構(Concurrent Data Structure)」,將 Mutex 的保護對象縮小到資料結構的節點,讓多個執行緒能同時讀寫一個資料結構。

然而,同時操作多個 Mutex 必需留意一些技術細節。舉例來說,以下程式碼有問題:

#include <mutex>

struct Node {
  std::mutex m;
  int data;
};

void swap(Node &lhs, Node &rhs) {
  if (&lhs == &rhs) return;
  std::lock_guard<std::mutex> lhs_lock(lhs.m);
  std::lock_guard<std::mutex> rhs_lock(rhs.m);
  std::swap(lhs.data, rhs.data);
}

大家看得出問題在哪裡嗎?

Dead Lock

上面的程式碼有可能讓 swap 的呼叫者不經意地產生 Dead Lock。假設執行時有二個節點 xy。執行緒 A 呼叫 swap(x, y) 而執行緒 B 呼叫 swap(y, x)。再假設兩者同時執行到 lhs_lock 各自鎖定 xy。接著,當兩者執行 rhs_lock 的時候就會發現 yx 分別被另一個執行緒鎖定,因此產生 Dead Lock:

執行緒 A 執行緒 B
Node x;  
Node y;  
swap(x, y); swap(y, x);
if (&lhs == &rhs) return; if (&lhs == &rhs) return;
std::lock_guard<std::mutex> lhs_lock(lhs.m);
// Lock x.m
std::lock_guard<std::mutex> lhs_lock(lhs.m);
// Lock y.m
std::lock_guard<std::mutex> rhs_lock(rhs.m);
// Wait for y.m
std::lock_guard<std::mutex> rhs_lock(rhs.m);
// Wait for x.m

複習一下 Dead Lock 的四個前提:

  1. Mutual Execution -- 有一個資源無法被共享。此例為 x.my.m
  2. Hold and Wait -- 有一個執行緒在佔有一個資源的時候,嘗試獲取另一個執行緒擁有的資源。此例為二個執行緒的 lhs_guardrhs_lock 建構式。
  3. No Preemption -- 一個資源只能被擁有者主動釋放。此例的 x.my.m 只能由擁有者釋放。
  4. Circular Dependency -- 各個執行緒「佔有的資源」與「想獲取的資源」有循環依賴。此例的 x.my.m 有可能會循環依賴。

其中第一項與第三項是 std::mutex 本身的特性,因此我們只能從第二項或第四項著手。

以 std::scoped_lock 鎖定多個 Mutex

C++ 17 提供 std::scoped_lock 作為 std::lock_guard 的替代品。std::scoped_lock 能讓一個執行緒一次鎖定多個 Mutexstd::scoped_lock 的實作可以任意決定 Mutex 的鎖定順序但是必須實作 Dead Lock 防護機制。

上面的例子可以改寫為:

void swap(Node &lhs, Node &rhs) {
  if (&lhs == &rhs) return;
  std::scoped_lock locks(lhs.m, rhs.m);
  std::swap(lhs.data, rhs.data);
}

備註std::scoped_lock 的建構式是 Variadic Template。它能接受任意數量的 Mutex,例如:std::scoped_lock guard(mutex1, mutex2, mutex3);

以 std::lock 函式鎖定多個 Mutex

在 C++ 11 我們必須使用 std::lock 函式搭配 std::lock_guard 類別。std::lockstd::scoped_lock 一樣能一次鎖定多個 Mutex 也有實作 Dead Lock 防護機制。改用 std::lock 鎖定 Mutex 時,我們必需先呼叫 std::lock 函式鎖定所有 Mutex 再以 std::adopt_lock 作為第二個參數讓 std::lock_guard 建構式跳過鎖定 Mutex 的步驟:

void swap(Node &lhs, Node &rhs) {
  if (&lhs == &rhs) return;
  std::lock(lhs.m, rhs.m);
  std::lock_guard<std::mutex> lhs_lock(lhs.m, std::adopt_lock);
  std::lock_guard<std::mutex> rhs_lock(rhs.m, std::adopt_lock);
  std::swap(lhs.data, rhs.data);
}

實作細節

最後,我想要簡單地介紹 std::scoped_lockstd::lock 是如何避免 Dead Lock。一個簡單的想法是以 try_lock 鎖定 Mutex。如果無法鎖定 Mutex,則釋放已經鎖定的 Mutex,讓其他執行緒先執行。除此之外,有些實作還會嘗試不同的鎖定順序。概念程式碼如下:

void lock_impl(std::mutex &x, std::mutex &y) {
  while (true) {
    if (x.try_lock()) {
      if (y.try_lock()) {
        return;
      }
      x.unlock();
    }

    sched_yield();  // implementation-defined

    if (y.try_lock()) {
      if (x.try_lock()) {
        return;
      }
      y.unlock();
    }

    sched_yield();  // implementation-defined
  }
}

雖然實際的實作為了「能接受多個參數」與「正確地處理例外」會更為複雜,但我知道的實作都是以這個想法為基礎。

參考資料

  • Standard for Programming Language C++, Working Draft N4687, Section 33.4.4 Locks and 33.4.5 Generic locking algorithms

Note

如果你喜歡這篇文章,請追蹤羅根學習筆記粉絲專頁。最新文章都會在粉絲專頁發佈通知。