最佳化多執行緒程式時,為了減少非必要阻塞(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。假設執行時有二個節點 x
與 y
。執行緒 A 呼叫 swap(x, y)
而執行緒 B 呼叫 swap(y, x)
。再假設兩者同時執行到 lhs_lock
各自鎖定 x
與 y
。接著,當兩者執行 rhs_lock
的時候就會發現 y
與 x
分別被另一個執行緒鎖定,因此產生 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 的四個前提:
- Mutual Execution -- 有一個資源無法被共享。此例為
x.m
與y.m
。 - Hold and Wait -- 有一個執行緒在佔有一個資源的時候,嘗試獲取另一個執行緒擁有的資源。此例為二個執行緒的
lhs_guard
與rhs_lock
建構式。 - No Preemption -- 一個資源只能被擁有者主動釋放。此例的
x.m
與y.m
只能由擁有者釋放。 - Circular Dependency -- 各個執行緒「佔有的資源」與「想獲取的資源」有循環依賴。此例的
x.m
與y.m
有可能會循環依賴。
其中第一項與第三項是 std::mutex
本身的特性,因此我們只能從第二項或第四項著手。
以 std::scoped_lock 鎖定多個 Mutex
C++ 17 提供 std::scoped_lock
作為 std::lock_guard
的替代品。std::scoped_lock
能讓一個執行緒一次鎖定多個 Mutex。std::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::lock
和 std::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_lock
或 std::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
如果你喜歡這篇文章,請追蹤羅根學習筆記粉絲專頁。最新文章都會在粉絲專頁發佈通知。