鍍金池/ 教程/ C/ 3.2 使用互斥量保護共享數(shù)據(jù)
3.4 本章總結(jié)
6.3 基于鎖設(shè)計更加復雜的數(shù)據(jù)結(jié)構(gòu)
6.1 為并發(fā)設(shè)計的意義何在?
5.2 <code>C++</code>中的原子操作和原子類型
A.7 自動推導變量類型
2.1 線程管理的基礎(chǔ)
8.5 在實踐中設(shè)計并發(fā)代碼
2.4 運行時決定線程數(shù)量
2.2 向線程函數(shù)傳遞參數(shù)
第4章 同步并發(fā)操作
2.3 轉(zhuǎn)移線程所有權(quán)
8.3 為多線程性能設(shè)計數(shù)據(jù)結(jié)構(gòu)
6.4 本章總結(jié)
7.3 對于設(shè)計無鎖數(shù)據(jù)結(jié)構(gòu)的指導建議
關(guān)于這本書
A.1 右值引用
2.6 本章總結(jié)
D.2 &lt;condition_variable&gt;頭文件
A.6 變參模板
6.2 基于鎖的并發(fā)數(shù)據(jù)結(jié)構(gòu)
4.5 本章總結(jié)
A.9 本章總結(jié)
前言
第10章 多線程程序的測試和調(diào)試
5.4 本章總結(jié)
第9章 高級線程管理
5.1 內(nèi)存模型基礎(chǔ)
2.5 識別線程
第1章 你好,C++的并發(fā)世界!
1.2 為什么使用并發(fā)?
A.5 Lambda函數(shù)
第2章 線程管理
4.3 限定等待時間
D.3 &lt;atomic&gt;頭文件
10.2 定位并發(fā)錯誤的技術(shù)
附錄B 并發(fā)庫的簡單比較
5.3 同步操作和強制排序
A.8 線程本地變量
第8章 并發(fā)代碼設(shè)計
3.3 保護共享數(shù)據(jù)的替代設(shè)施
附錄D C++線程庫參考
第7章 無鎖并發(fā)數(shù)據(jù)結(jié)構(gòu)設(shè)計
D.7 &lt;thread&gt;頭文件
D.1 &lt;chrono&gt;頭文件
4.1 等待一個事件或其他條件
A.3 默認函數(shù)
附錄A 對<code>C++</code>11語言特性的簡要介紹
第6章 基于鎖的并發(fā)數(shù)據(jù)結(jié)構(gòu)設(shè)計
封面圖片介紹
7.2 無鎖數(shù)據(jù)結(jié)構(gòu)的例子
8.6 本章總結(jié)
8.1 線程間劃分工作的技術(shù)
4.2 使用期望等待一次性事件
8.4 設(shè)計并發(fā)代碼的注意事項
D.5 &lt;mutex&gt;頭文件
3.1 共享數(shù)據(jù)帶來的問題
資源
9.3 本章總結(jié)
10.3 本章總結(jié)
10.1 與并發(fā)相關(guān)的錯誤類型
D.4 &lt;future&gt;頭文件
3.2 使用互斥量保護共享數(shù)據(jù)
9.1 線程池
1.1 何謂并發(fā)
9.2 中斷線程
4.4 使用同步操作簡化代碼
A.2 刪除函數(shù)
1.3 C++中的并發(fā)和多線程
1.4 開始入門
第5章 C++內(nèi)存模型和原子類型操作
消息傳遞框架與完整的ATM示例
8.2 影響并發(fā)代碼性能的因素
7.1 定義和意義
D.6 &lt;ratio&gt;頭文件
A.4 常量表達式函數(shù)
7.4 本章總結(jié)
1.5 本章總結(jié)
第3章 線程間共享數(shù)據(jù)

3.2 使用互斥量保護共享數(shù)據(jù)

當程序中有共享數(shù)據(jù),肯定不想讓其陷入條件競爭,或是不變量被破壞。那么,將所有訪問共享數(shù)據(jù)結(jié)構(gòu)的代碼都標記為互斥豈不是更好?這樣任何一個線程在執(zhí)行這些代碼時,其他任何線程試圖訪問共享數(shù)據(jù)結(jié)構(gòu),就必須等到那一段代碼執(zhí)行結(jié)束。于是,一個線程就不可能會看到被破壞的不變量,除非它本身就是修改共享數(shù)據(jù)的線程。

當訪問共享數(shù)據(jù)前,使用互斥量將相關(guān)數(shù)據(jù)鎖住,再當訪問結(jié)束后,再將數(shù)據(jù)解鎖。線程庫需要保證,當一個線程使用特定互斥量鎖住共享數(shù)據(jù)時,其他的線程想要訪問鎖住的數(shù)據(jù),都必須等到之前那個線程對數(shù)據(jù)進行解鎖后,才能進行訪問。這就保證了所有線程能看到共享數(shù)據(jù),而不破壞不變量。

互斥量是C++中一種最通用的數(shù)據(jù)保護機制,但它不是“銀彈”;精心組織代碼來保護正確的數(shù)據(jù)(見3.2.2節(jié)),并在接口內(nèi)部避免競爭條件(見3.2.3節(jié))是非常重要的。但互斥量自身也有問題,也會造成死鎖(見3.2.4節(jié)),或是對數(shù)據(jù)保護的太多(或太少)(見3.2.8節(jié))。

3.2.1 C++中使用互斥量

C++中通過實例化std::mutex創(chuàng)建互斥量,通過調(diào)用成員函數(shù)lock()進行上鎖,unlock()進行解鎖。不過,不推薦實踐中直接去調(diào)用成員函數(shù),因為調(diào)用成員函數(shù)就意味著,必須記住在每個函數(shù)出口都要去調(diào)用unlock(),也包括異常的情況。C++標準庫為互斥量提供了一個RAII語法的模板類std::lock_guard,其會在構(gòu)造的時候提供已鎖的互斥量,并在析構(gòu)的時候進行解鎖,從而保證了一個已鎖的互斥量總是會被正確的解鎖。下面的程序清單中,展示了如何在多線程程序中,使用std::mutex構(gòu)造的std::lock_guard實例,對一個列表進行訪問保護。std::mutexstd::lock_guard都在<mutex>頭文件中聲明。

清單3.1 使用互斥量保護列表

#include <list>
#include <mutex>
#include <algorithm>

std::list<int> some_list;    // 1
std::mutex some_mutex;    // 2

void add_to_list(int new_value)
{
  std::lock_guard<std::mutex> guard(some_mutex);    // 3
  some_list.push_back(new_value);
}

bool list_contains(int value_to_find)
{
  std::lock_guard<std::mutex> guard(some_mutex);    // 4
  return std::find(some_list.begin(),some_list.end(),value_to_find) != some_list.end();
}

清單3.1中有一個全局變量①,這個全局變量被一個全局的互斥量保護②。add_to_list()③和list_contains()④函數(shù)中使用std::lock_guard<std::mutex>,使得這兩個函數(shù)中對數(shù)據(jù)的訪問是互斥的:list_contains()不可能看到正在被add_to_list()修改的列表。

雖然某些情況下,使用全局變量沒問題,但在大多數(shù)情況下,互斥量通常會與保護的數(shù)據(jù)放在同一個類中,而不是定義成全局變量。這是面向?qū)ο笤O(shè)計的準則:將其放在一個類中,就可讓他們聯(lián)系在一起,也可對類的功能進行封裝,并進行數(shù)據(jù)保護。在這種情況下,函數(shù)add_to_list和list_contains可以作為這個類的成員函數(shù)?;コ饬亢鸵Wo的數(shù)據(jù),在類中都需要定義為private成員,這會讓訪問數(shù)據(jù)的代碼變的清晰,并且容易看出在什么時候?qū)コ饬可湘i。當所有成員函數(shù)都會在調(diào)用時對數(shù)據(jù)上鎖,結(jié)束時對數(shù)據(jù)解鎖,那么就保證了數(shù)據(jù)訪問時不變量不被破壞。

當然,也不是總是那么理想,聰明的你一定注意到了:當其中一個成員函數(shù)返回的是保護數(shù)據(jù)的指針或引用時,會破壞對數(shù)據(jù)的保護。具有訪問能力的指針或引用可以訪問(并可能修改)被保護的數(shù)據(jù),而不會被互斥鎖限制?;コ饬勘Wo的數(shù)據(jù)需要對接口的設(shè)計相當謹慎,要確?;コ饬磕苕i住任何對保護數(shù)據(jù)的訪問,并且不留后門。

3.2.2 精心組織代碼來保護共享數(shù)據(jù)

使用互斥量來保護數(shù)據(jù),并不是僅僅在每一個成員函數(shù)中都加入一個std::lock_guard對象那么簡單;一個迷失的指針或引用,將會讓這種保護形同虛設(shè)。不過,檢查迷失指針或引用是很容易的,只要沒有成員函數(shù)通過返回值或者輸出參數(shù)的形式向其調(diào)用者返回指向受保護數(shù)據(jù)的指針或引用,數(shù)據(jù)就是安全的。如果你還想往祖墳上刨,就沒這么簡單了。在確保成員函數(shù)不會傳出指針或引用的同時,檢查成員函數(shù)是否通過指針或引用的方式來調(diào)用也是很重要的(尤其是這個操作不在你的控制下時)。函數(shù)可能沒在互斥量保護的區(qū)域內(nèi),存儲著指針或者引用,這樣就很危險。更危險的是:將保護數(shù)據(jù)作為一個運行時參數(shù),如同下面清單中所示那樣。

清單3.2 無意中傳遞了保護數(shù)據(jù)的引用

class some_data
{
  int a;
  std::string b;
public:
  void do_something();
};

class data_wrapper
{
private:
  some_data data;
  std::mutex m;
public:
  template<typename Function>
  void process_data(Function func)
  {
    std::lock_guard<std::mutex> l(m);
    func(data);    // 1 傳遞“保護”數(shù)據(jù)給用戶函數(shù)
  }
};

some_data* unprotected;

void malicious_function(some_data& protected_data)
{
  unprotected=&protected_data;
}

data_wrapper x;
void foo()
{
  x.process_data(malicious_function);    // 2 傳遞一個惡意函數(shù)
  unprotected->do_something();    // 3 在無保護的情況下訪問保護數(shù)據(jù)
}

例子中process_data看起來沒有任何問題,std::lock_guard對數(shù)據(jù)做了很好的保護,但調(diào)用用戶提供的函數(shù)func①,就意味著foo能夠繞過保護機制將函數(shù)malicious_function傳遞進去②,在沒有鎖定互斥量的情況下調(diào)用do_something()。

這段代碼的問題在于根本沒有保護,只是將所有可訪問的數(shù)據(jù)結(jié)構(gòu)代碼標記為互斥。函數(shù)foo()中調(diào)用unprotected->do_something()的代碼未能被標記為互斥。這種情況下,C++線程庫無法提供任何幫助,只能由程序員來使用正確的互斥鎖來保護數(shù)據(jù)。從樂觀的角度上看,還是有方法可循的:切勿將受保護數(shù)據(jù)的指針或引用傳遞到互斥鎖作用域之外,無論是函數(shù)返回值,還是存儲在外部可見內(nèi)存,亦或是以參數(shù)的形式傳遞到用戶提供的函數(shù)中去。

雖然這是在使用互斥量保護共享數(shù)據(jù)時常犯的錯誤,但絕不僅僅是一個潛在的陷阱而已。下一節(jié)中,你將會看到,即便是使用了互斥量對數(shù)據(jù)進行了保護,條件競爭依舊可能存在。

3.2.3 發(fā)現(xiàn)接口內(nèi)在的條件競爭

因為使用了互斥量或其他機制保護了共享數(shù)據(jù),就不必再為條件競爭所擔憂嗎?并不是,你依舊需要確定數(shù)據(jù)受到了保護?;叵胫半p鏈表的例子,為了能讓線程安全地刪除一個節(jié)點,需要確保防止對這三個節(jié)點(待刪除的節(jié)點及其前后相鄰的節(jié)點)的并發(fā)訪問。如果只對指向每個節(jié)點的指針進行訪問保護,那就和沒有使用互斥量一樣,條件競爭仍會發(fā)生——除了指針,整個數(shù)據(jù)結(jié)構(gòu)和整個刪除操作需要保護。這種情況下最簡單的解決方案就是使用互斥量來保護整個鏈表,如清單3.1所示。

盡管鏈表的個別操作是安全的,但不意味著你就能走出困境;即使在一個很簡單的接口中,依舊可能遇到條件競爭。例如,構(gòu)建一個類似于std::stack結(jié)構(gòu)的棧(清單3.3),除了構(gòu)造函數(shù)和swap()以外,需要對std::stack提供五個操作:push()一個新元素進棧,pop()一個元素出棧,top()查看棧頂元素,empty()判斷棧是否是空棧,size()了解棧中有多少個元素。即使修改了top(),使其返回一個拷貝而非引用(即遵循了3.2.2節(jié)的準則),對內(nèi)部數(shù)據(jù)使用一個互斥量進行保護,不過這個接口仍存在條件競爭。這個問題不僅存在于基于互斥量實現(xiàn)的接口中,在無鎖實現(xiàn)的接口中,條件競爭依舊會產(chǎn)生。這是接口的問題,與其實現(xiàn)方式無關(guān)。

清單3.3 std::stack容器的實現(xiàn)

template<typename T,typename Container=std::deque<T> >
class stack
{
public:
  explicit stack(const Container&);
  explicit stack(Container&& = Container());
  template <class Alloc> explicit stack(const Alloc&);
  template <class Alloc> stack(const Container&, const Alloc&);
  template <class Alloc> stack(Container&&, const Alloc&);
  template <class Alloc> stack(stack&&, const Alloc&);

  bool empty() const;
  size_t size() const;
  T& top();
  T const& top() const;
  void push(T const&);
  void push(T&&);
  void pop();
  void swap(stack&&);
};

雖然empty()和size()可能在被調(diào)用并返回時是正確的,但其的結(jié)果是不可靠的;當它們返回后,其他線程就可以自由地訪問棧,并且可能push()多個新元素到棧中,也可能pop()一些已在棧中的元素。這樣的話,之前從empty()和size()得到的結(jié)果就有問題了。

特別地,當棧實例是非共享的,如果棧非空,使用empty()檢查再調(diào)用top()訪問棧頂部的元素是安全的。如下代碼所示:

stack<int> s;
if (! s.empty()){    // 1
  int const value = s.top();    // 2
  s.pop();    // 3
  do_something(value);
}

以上是單線程安全代碼:對一個空棧使用top()是未定義行為。對于共享的棧對象,這樣的調(diào)用順序就不再安全了,因為在調(diào)用empty()①和調(diào)用top()②之間,可能有來自另一個線程的pop()調(diào)用并刪除了最后一個元素。這是一個經(jīng)典的條件競爭,使用互斥量對棧內(nèi)部數(shù)據(jù)進行保護,但依舊不能阻止條件競爭的發(fā)生,這就是接口固有的問題。

怎么解決呢?問題發(fā)生在接口設(shè)計上,所以解決的方法也就是改變接口設(shè)計。有人會問:怎么改?在這個簡單的例子中,當調(diào)用top()時,發(fā)現(xiàn)棧已經(jīng)是空的了,那么就拋出異常。雖然這能直接解決這個問題,但這是一個笨拙的解決方案,這樣的話,即使empty()返回false的情況下,你也需要異常捕獲機制。本質(zhì)上,這樣的改變會讓empty()成為一個多余函數(shù)。

當仔細的觀察過之前的代碼段,就會發(fā)現(xiàn)另一個潛在的條件競爭在調(diào)用top()②和pop()③之間。假設(shè)兩個線程運行著前面的代碼,并且都引用同一個棧對象s。這并非罕見的情況,當為性能而使用線程時,多個線程在不同的數(shù)據(jù)上執(zhí)行相同的操作是很平常的,并且共享同一個??梢詫⒐ぷ鞣謹偨o它們。假設(shè),一開始棧中只有兩個元素,這時任一線程上的empty()和top()都存在競爭,只需要考慮可能的執(zhí)行順序即可。

當棧被一個內(nèi)部互斥量所保護時,只有一個線程可以調(diào)用棧的成員函數(shù),所以調(diào)用可以很好地交錯,并且do_something()是可以并發(fā)運行的。在表3.1中,展示一種可能的執(zhí)行順序。

表3.1 一種可能執(zhí)行順序

Thread A Thread B
if (!s.empty);
if(!s.empty);
int const value = s.top();
int const value = s.top();
s.pop();
do_something(value); s.pop();
do_something(value);

當線程運行時,調(diào)用兩次top(),棧沒被修改,所以每個線程能得到同樣的值。不僅是這樣,在調(diào)用top()函數(shù)調(diào)用的過程中(兩次),pop()函數(shù)都沒有被調(diào)用。這樣,在其中一個值再讀取的時候,雖然不會出現(xiàn)“寫后讀”的情況,但其值已被處理了兩次。這種條件競爭,比未定義的empty()/top()競爭更加嚴重;雖然其結(jié)果依賴于do_something()的結(jié)果,但因為看起來沒有任何錯誤,就會讓這個Bug很難定位。

這就需要接口設(shè)計上有較大的改動,提議之一就是使用同一互斥量來保護top()和pop()。Tom Cargill[1]指出當一個對象的拷貝構(gòu)造函數(shù)在棧中拋出一個異常,這樣的處理方式就會有問題。在Herb Sutter[2]看來,這個問題可以從“異常安全”的角度完美解決,不過潛在的條件競爭,可能會組成一些新的組合。

說一些大家沒有意識到的問題:假設(shè)有一個stack<vector<int>>,vector是一個動態(tài)容器,當你拷貝一個vetcor,標準庫會從堆上分配很多內(nèi)存來完成這次拷貝。當這個系統(tǒng)處在重度負荷,或有嚴重的資源限制的情況下,這種內(nèi)存分配就會失敗,所以vector的拷貝構(gòu)造函數(shù)可能會拋出一個std::bad_alloc異常。當vector中存有大量元素時,這種情況發(fā)生的可能性更大。當pop()函數(shù)返回“彈出值”時(也就是從棧中將這個值移除),會有一個潛在的問題:這個值被返回到調(diào)用函數(shù)的時候,棧才被改變;但當拷貝數(shù)據(jù)的時候,調(diào)用函數(shù)拋出一個異常會怎么樣? 如果事情真的發(fā)生了,要彈出的數(shù)據(jù)將會丟失;它的確從棧上移出了,但是拷貝失敗了!std::stack的設(shè)計人員將這個操作分為兩部分:先獲取頂部元素(top()),然后從棧中移除(pop())。這樣,在不能安全的將元素拷貝出去的情況下,棧中的這個數(shù)據(jù)還依舊存在,沒有丟失。當問題是堆空間不足,應(yīng)用可能會釋放一些內(nèi)存,然后再進行嘗試。

不幸的是,這樣的分割卻制造了本想避免或消除的條件競爭。幸運的是,我們還有的別的選項,但是使用這些選項是要付出代價的。

選項1: 傳入一個引用

第一個選項是將變量的引用作為參數(shù),傳入pop()函數(shù)中獲取想要的“彈出值”:

std::vector<int> result;
some_stack.pop(result);

大多數(shù)情況下,這種方式還不錯,但有明顯的缺點:需要構(gòu)造出一個棧中類型的實例,用于接收目標值。對于一些類型,這樣做是不現(xiàn)實的,因為臨時構(gòu)造一個實例,從時間和資源的角度上來看,都是不劃算。對于其他的類型,這樣也不總能行得通,因為構(gòu)造函數(shù)需要的一些參數(shù),在代碼的這個階段不一定可用。最后,需要可賦值的存儲類型,這是一個重大限制:即使支持移動構(gòu)造,甚至是拷貝構(gòu)造(從而允許返回一個值),很多用戶自定義類型可能都不支持賦值操作。

選項2:無異常拋出的拷貝構(gòu)造函數(shù)或移動構(gòu)造函數(shù)

對于有返回值的pop()函數(shù)來說,只有“異常安全”方面的擔憂(當返回值時可以拋出一個異常)。很多類型都有拷貝構(gòu)造函數(shù),它們不會拋出異常,并且隨著新標準中對“右值引用”的支持(詳見附錄A,A.1節(jié)),很多類型都將會有一個移動構(gòu)造函數(shù),即使他們和拷貝構(gòu)造函數(shù)做著相同的事情,它也不會拋出異常。一個有用的選項可以限制對線程安全的棧的使用,并且能讓棧安全的返回所需的值,而不會拋出異常。

雖然安全,但非可靠。盡管能在編譯時可使用std::is_nothrow_copy_constructiblestd::is_nothrow_move_constructible類型特征,讓拷貝或移動構(gòu)造函數(shù)不拋出異常,但是這種方式的局限性太強。用戶自定義的類型中,會有不拋出異常的拷貝構(gòu)造函數(shù)或移動構(gòu)造函數(shù)的類型, 那些有拋出異常的拷貝構(gòu)造函數(shù),但沒有移動構(gòu)造函數(shù)的類型往往更多(這種情況會隨著人們習慣于C++11中的右值引用而有所改變)。如果這些類型不能被存儲在線程安全的棧中,那將是多么的不幸。

、選項3:返回指向彈出值的指針

第三個選擇是返回一個指向彈出元素的指針,而不是直接返回值。指針的優(yōu)勢是自由拷貝,并且不會產(chǎn)生異常,這樣你就能避免Cargill提到的異常問題了。缺點就是返回一個指針需要對對象的內(nèi)存分配進行管理,對于簡單數(shù)據(jù)類型(比如:int),內(nèi)存管理的開銷要遠大于直接返回值。對于選擇這個方案的接口,使用std::shared_ptr是個不錯的選擇;不僅能避免內(nèi)存泄露(因為當對象中指針銷毀時,對象也會被銷毀),而且標準庫能夠完全控制內(nèi)存分配方案,也就不需要new和delete操作。這種優(yōu)化是很重要的:因為堆棧中的每個對象,都需要用new進行獨立的內(nèi)存分配,相較于非線程安全版本,這個方案的開銷相當大。

選項4:“選項1 + 選項2”或 “選項1 + 選項3”

對于通用的代碼來說,靈活性不應(yīng)忽視。當你已經(jīng)選擇了選項2或3時,再去選擇1也是很容易的。這些選項提供給用戶,讓用戶自己選擇對于他們自己來說最合適,最經(jīng)濟的方案。

例:定義線程安全的堆棧

清單3.4中是一個接口沒有條件競爭的堆棧類定義,它實現(xiàn)了選項1和選項3:重載了pop(),使用一個局部引用去存儲彈出值,并返回一個std::shared_ptr<>對象。它有一個簡單的接口,只有兩個函數(shù):push()和pop();

清單3.4 線程安全的堆棧類定義(概述)

#include <exception>
#include <memory>  // For std::shared_ptr<>

struct empty_stack: std::exception
{
  const char* what() const throw();
};

template<typename T>
class threadsafe_stack
{
public:
  threadsafe_stack();
  threadsafe_stack(const threadsafe_stack&);
  threadsafe_stack& operator=(const threadsafe_stack&) = delete; // 1 賦值操作被刪除

  void push(T new_value);
  std::shared_ptr<T> pop();
  void pop(T& value);
  bool empty() const;
};

削減接口可以獲得最大程度的安全,甚至限制對棧的一些操作。棧是不能直接賦值的,因為賦值操作已經(jīng)刪除了①(詳見附錄A,A.2節(jié)),并且這里沒有swap()函數(shù)。棧可以拷貝的,假設(shè)棧中的元素可以拷貝。當棧為空時,pop()函數(shù)會拋出一個empty_stack異常,所以在empty()函數(shù)被調(diào)用后,其他部件還能正常工作。如選項3描述的那樣,使用std::shared_ptr可以避免內(nèi)存分配管理的問題,并避免多次使用new和delete操作。堆棧中的五個操作,現(xiàn)在就剩下三個:push(), pop()和empty()(這里empty()都有些多余)。簡化接口更有利于數(shù)據(jù)控制,可以保證互斥量將一個操作完全鎖住。下面的代碼將展示一個簡單的實現(xiàn)——封裝std::stack<>的線程安全堆棧。

清單3.5 擴充(線程安全)堆棧

#include <exception>
#include <memory>
#include <mutex>
#include <stack>

struct empty_stack: std::exception
{
  const char* what() const throw() {
    return "empty stack!";
  };
};

template<typename T>
class threadsafe_stack
{
private:
  std::stack<T> data;
  mutable std::mutex m;

public:
  threadsafe_stack()
    : data(std::stack<T>()){}

  threadsafe_stack(const threadsafe_stack& other)
  {
    std::lock_guard<std::mutex> lock(other.m);
    data = other.data; // 1 在構(gòu)造函數(shù)體中的執(zhí)行拷貝
  }

  threadsafe_stack& operator=(const threadsafe_stack&) = delete;

  void push(T new_value)
  {
    std::lock_guard<std::mutex> lock(m);
    data.push(new_value);
  }

  std::shared_ptr<T> pop()
  {
    std::lock_guard<std::mutex> lock(m);
    if(data.empty()) throw empty_stack(); // 在調(diào)用pop前,檢查棧是否為空

    std::shared_ptr<T> const res(std::make_shared<T>(data.top())); // 在修改堆棧前,分配出返回值
    data.pop();
    return res;
  }

  void pop(T& value)
  {
    std::lock_guard<std::mutex> lock(m);
    if(data.empty()) throw empty_stack();

    value=data.top();
    data.pop();
  }

  bool empty() const
  {
    std::lock_guard<std::mutex> lock(m);
    return data.empty();
  }
};

堆??梢钥截悺截悩?gòu)造函數(shù)對互斥量上鎖,再拷貝堆棧。構(gòu)造函數(shù)體中①的拷貝使用互斥量來確保復制結(jié)果的正確性,這樣的方式比成員初始化列表好。

之前對top()和pop()函數(shù)的討論中,惡性條件競爭已經(jīng)出現(xiàn),因為鎖的粒度太小,需要保護的操作并未全覆蓋到。不過,鎖住的顆粒過大同樣會有問題。還有一個問題,一個全局互斥量要去保護全部共享數(shù)據(jù),在一個系統(tǒng)中存在有大量的共享數(shù)據(jù)時,因為線程可以強制運行,甚至可以訪問不同位置的數(shù)據(jù),抵消了并發(fā)帶來的性能提升。在第一版為多處理器系統(tǒng)設(shè)計Linux內(nèi)核中,就使用了一個全局內(nèi)核鎖。雖然這個鎖能正常工作,但在雙核處理系統(tǒng)的上的性能要比兩個單核系統(tǒng)的性能差很多,四核系統(tǒng)就更不能提了。太多請求去競爭占用內(nèi)核,使得依賴于處理器運行的線程沒有辦法很好的工作。隨后修正的Linux內(nèi)核加入了一個細粒度鎖方案,因為少了很多內(nèi)核競爭,這時四核處理系統(tǒng)的性能就和單核處理的四倍差不多了。

使用多個互斥量保護所有的數(shù)據(jù),細粒度鎖也有問題。如前所述,當增大互斥量覆蓋數(shù)據(jù)的粒度時,只需要鎖住一個互斥量。但是,這種方案并非放之四海皆準,比如:互斥量正在保護一個獨立類的實例;這種情況下,鎖的狀態(tài)的下一個階段,不是離開鎖定區(qū)域?qū)㈡i定區(qū)域還給用戶,就是有獨立的互斥量去保護這個類的全部實例。當然,這兩種方式都不理想。

一個給定操作需要兩個或兩個以上的互斥量時,另一個潛在的問題將出現(xiàn):死鎖。與條件競爭完全相反——不同的兩個線程會互相等待,從而什么都沒做。

3.2.4 死鎖:問題描述及解決方案

試想有一個玩具,這個玩具由兩部分組成,必須拿到這兩個部分,才能夠玩。例如,一個玩具鼓,需要一個鼓錘和一個鼓才能玩。現(xiàn)在有兩個小孩,他們都很喜歡玩這個玩具。當其中一個孩子拿到了鼓和鼓錘時,那就可以盡情的玩耍了。當另一孩子想要玩,他就得等待另一孩子玩完才行。再試想,鼓和鼓錘被放在不同的玩具箱里,并且兩個孩子在同一時間里都想要去敲鼓。之后,他們就去玩具箱里面找這個鼓。其中一個找到了鼓,并且另外一個找到了鼓錘?,F(xiàn)在問題就來了,除非其中一個孩子決定讓另一個先玩,他可以把自己的那部分給另外一個孩子;但當他們都緊握著自己所有的部分而不給予,那么這個鼓誰都沒法玩。

現(xiàn)在沒有孩子去爭搶玩具,但線程有對鎖的競爭:一對線程需要對他們所有的互斥量做一些操作,其中每個線程都有一個互斥量,且等待另一個解鎖。這樣沒有線程能工作,因為他們都在等待對方釋放互斥量。這種情況就是死鎖,它的最大問題就是由兩個或兩個以上的互斥量來鎖定一個操作。

避免死鎖的一般建議,就是讓兩個互斥量總以相同的順序上鎖:總在互斥量B之前鎖住互斥量A,就永遠不會死鎖。某些情況下是可以這樣用,因為不同的互斥量用于不同的地方。不過,事情沒那么簡單,比如:當有多個互斥量保護同一個類的獨立實例時,一個操作對同一個類的兩個不同實例進行數(shù)據(jù)的交換操作,為了保證數(shù)據(jù)交換操作的正確性,就要避免數(shù)據(jù)被并發(fā)修改,并確保每個實例上的互斥量都能鎖住自己要保護的區(qū)域。不過,選擇一個固定的順序(例如,實例提供的第一互斥量作為第一個參數(shù),提供的第二個互斥量為第二個參數(shù)),可能會適得其反:在參數(shù)交換了之后,兩個線程試圖在相同的兩個實例間進行數(shù)據(jù)交換時,程序又死鎖了!

很幸運,C++標準庫有辦法解決這個問題,std::lock——可以一次性鎖住多個(兩個以上)的互斥量,并且沒有副作用(死鎖風險)。下面的程序清單中,就來看一下怎么在一個簡單的交換操作中使用std::lock。

清單3.6 交換操作中使用std::lock()std::lock_guard

// 這里的std::lock()需要包含<mutex>頭文件
class some_big_object;
void swap(some_big_object& lhs,some_big_object& rhs);
class X
{
private:
  some_big_object some_detail;
  std::mutex m;
public:
  X(some_big_object const& sd):some_detail(sd){}

  friend void swap(X& lhs, X& rhs)
  {
    if(&lhs==&rhs)
      return;
    std::lock(lhs.m,rhs.m); // 1
    std::lock_guard<std::mutex> lock_a(lhs.m,std::adopt_lock); // 2
    std::lock_guard<std::mutex> lock_b(rhs.m,std::adopt_lock); // 3
    swap(lhs.some_detail,rhs.some_detail);
  }
};

首先,檢查參數(shù)是否是不同的實例,因為操作試圖獲取std::mutex對象上的鎖,所以當其被獲取時,結(jié)果很難預料。(一個互斥量可以在同一線程上多次上鎖,標準庫中std::recursive_mutex提供這樣的功能。詳情見3.3.3節(jié))。然后,調(diào)用std::lock()①鎖住兩個互斥量,并且兩個std:lock_guard實例已經(jīng)創(chuàng)建好②③。提供std::adopt_lock參數(shù)除了表示std::lock_guard對象可獲取鎖之外,還將鎖交由std::lock_guard對象管理,而不需要std::lock_guard對象再去構(gòu)建新的鎖。

這樣,就能保證在大多數(shù)情況下,函數(shù)退出時互斥量能被正確的解鎖(保護操作可能會拋出一個異常),也允許使用一個簡單的“return”作為返回。還有,需要注意的是,當使用std::lock去鎖lhs.m或rhs.m時,可能會拋出異常;這種情況下,異常會傳播到std::lock之外。當std::lock成功的獲取一個互斥量上的鎖,并且當其嘗試從另一個互斥量上再獲取鎖時,就會有異常拋出,第一個鎖也會隨著異常的產(chǎn)生而自動釋放,所以std::lock要么將兩個鎖都鎖住,要不一個都不鎖。

雖然std::lock可以在這情況下(獲取兩個以上的鎖)避免死鎖,但它沒辦法幫助你獲取其中一個鎖。這時,不得不依賴于開發(fā)者的紀律性(譯者:也就是經(jīng)驗),來確保你的程序不會死鎖。這并不簡單:死鎖是多線程編程中一個令人相當頭痛的問題,并且死鎖經(jīng)常是不可預見的,因為在大多數(shù)時間里,所有工作都能很好的完成。不過,也一些相對簡單的規(guī)則能幫助寫出“無死鎖”的代碼。

3.2.5 避免死鎖的進階指導

雖然鎖是產(chǎn)生死鎖的一般原因,但也不排除死鎖出現(xiàn)在其他地方。無鎖的情況下,僅需要每個std::thread對象調(diào)用join(),兩個線程就能產(chǎn)生死鎖。這種情況下,沒有線程可以繼續(xù)運行,因為他們正在互相等待。這種情況很常見,一個線程會等待另一個線程,其他線程同時也會等待第一個線程結(jié)束,所以三個或更多線程的互相等待也會發(fā)生死鎖。為了避免死鎖,這里的指導意見為:當機會來臨時,不要拱手讓人。以下提供一些個人的指導建議,如何識別死鎖,并消除其他線程的等待。

避免嵌套鎖

第一個建議往往是最簡單的:一個線程已獲得一個鎖時,再別去獲取第二個。如果能堅持這個建議,因為每個線程只持有一個鎖,鎖上就不會產(chǎn)生死鎖。即使互斥鎖造成死鎖的最常見原因,也可能會在其他方面受到死鎖的困擾(比如:線程間的互相等待)。當你需要獲取多個鎖,使用一個std::lock來做這件事(對獲取鎖的操作上鎖),避免產(chǎn)生死鎖。

避免在持有鎖時調(diào)用用戶提供的代碼

第二個建議是次簡單的:因為代碼是用戶提供的,你沒有辦法確定用戶要做什么;用戶程序可能做任何事情,包括獲取鎖。你在持有鎖的情況下,調(diào)用用戶提供的代碼;如果用戶代碼要獲取一個鎖,就會違反第一個指導意見,并造成死鎖(有時,這是無法避免的)。當你正在寫一份通用代碼,例如3.2.3中的棧,每一個操作的參數(shù)類型,都在用戶提供的代碼中定義,就需要其他指導意見來幫助你。

使用固定順序獲取鎖

當硬性條件要求你獲取兩個以上(包括兩個)的鎖,并且不能使用std::lock單獨操作來獲取它們;那么最好在每個線程上,用固定的順序獲取它們獲取它們(鎖)。3.2.4節(jié)中提到一種當需要獲取兩個互斥量時,避免死鎖的方法:關(guān)鍵是如何在線程之間,以一定的順序獲取鎖。一些情況下,這種方式相對簡單。比如,3.2.3節(jié)中的?!總€棧實例中都內(nèi)置有互斥量,但是對數(shù)據(jù)成員存儲的操作上,棧就需要帶調(diào)用用戶提供的代碼。雖然,可以添加一些約束,對棧上存儲的數(shù)據(jù)項不做任何操作,對數(shù)據(jù)項的處理僅限于棧自身。這會給用戶提供的棧增加一些負擔,但是一個容器很少去訪問另一個容器中存儲的數(shù)據(jù),即使發(fā)生了也會很明顯,所以這對于通用棧來說并不是一個特別沉重的負擔。

其他情況下,這就不會那么簡單了,例如:3.2.4節(jié)中的交換操作,這種情況下你可能同時鎖住多個互斥量(但是有時不會發(fā)生)。當回看3.1節(jié)中那個鏈表連接例子時,將會看到列表中的每個節(jié)點都會有一個互斥量保護。為了訪問列表,線程必須獲取他們感興趣節(jié)點上的互斥鎖。當一個線程刪除一個節(jié)點,它必須獲取三個節(jié)點上的互斥鎖:將要刪除的節(jié)點,兩個鄰接節(jié)點(因為他們也會被修改)。同樣的,為了遍歷鏈表,線程必須保證在獲取當前節(jié)點的互斥鎖前提下,獲得下一個節(jié)點的鎖,要保證指向下一個節(jié)點的指針不會同時被修改。一旦下一個節(jié)點上的鎖被獲取,那么第一個節(jié)點的鎖就可以釋放了,因為沒有持有它的必要性了。

這種“手遞手”鎖的模式允許多個線程訪問列表,為每一個訪問的線程提供不同的節(jié)點。但是,為了避免死鎖,節(jié)點必須以同樣的順序上鎖:如果兩個線程試圖用互為反向的順序,使用“手遞手”鎖遍歷列表,他們將執(zhí)行到列表中間部分時,發(fā)生死鎖。當節(jié)點A和B在列表中相鄰,當前線程可能會同時嘗試獲取A和B上的鎖。另一個線程可能已經(jīng)獲取了節(jié)點B上的鎖,并且試圖獲取節(jié)點A上的鎖——經(jīng)典的死鎖場景。

當A、C節(jié)點中的B節(jié)點正在被刪除時,如果有線程在已獲取A和C上的鎖后,還要獲取B節(jié)點上的鎖時,當一個線程遍歷列表的時候,這樣的情況就可能發(fā)生死鎖。這樣的線程可能會試圖首先鎖住A節(jié)點或C節(jié)點(根據(jù)遍歷的方向),但是后面就會發(fā)現(xiàn),它無法獲得B上的鎖,因為線程在執(zhí)行刪除任務(wù)的時候,已經(jīng)獲取了B上的鎖,并且同時也獲取了A和C上的鎖。

這里提供一種避免死鎖的方式,定義遍歷的順序,所以一個線程必須先鎖住A才能獲取B的鎖,在鎖住B之后才能獲取C的鎖。這將消除死鎖發(fā)生的可能性,在不允許反向遍歷的列表上。類似的約定常被用來建立其他的數(shù)據(jù)結(jié)構(gòu)。

使用鎖的層次結(jié)構(gòu)

雖然,這對于定義鎖的順序,的確是一個特殊的情況,但鎖的層次的意義在于提供對運行時約定是否被堅持的檢查。這個建議需要對你的應(yīng)用進行分層,并且識別在給定層上所有可上鎖的互斥量。當代碼試圖對一個互斥量上鎖,在該層鎖已被低層持有時,上鎖是不允許的。你可以在運行時對其進行檢查,通過分配層數(shù)到每個互斥量上,以及記錄被每個線程上鎖的互斥量。下面的代碼列表中將展示兩個線程如何使用分層互斥。

清單3.7 使用層次鎖來避免死鎖

hierarchical_mutex high_level_mutex(10000); // 1
hierarchical_mutex low_level_mutex(5000);  // 2

int do_low_level_stuff();

int low_level_func()
{
  std::lock_guard<hierarchical_mutex> lk(low_level_mutex); // 3
  return do_low_level_stuff();
}

void high_level_stuff(int some_param);

void high_level_func()
{
  std::lock_guard<hierarchical_mutex> lk(high_level_mutex); // 4
  high_level_stuff(low_level_func()); // 5
}

void thread_a()  // 6
{
  high_level_func();
}

hierarchical_mutex other_mutex(100); // 7
void do_other_stuff();

void other_stuff()
{
  high_level_func();  // 8
  do_other_stuff();
}

void thread_b() // 9
{
  std::lock_guard<hierarchical_mutex> lk(other_mutex); // 10
  other_stuff();
}

thread_a()⑥遵守規(guī)則,所以它運行的沒問題。另一方面,thread_b()⑨無視規(guī)則,因此在運行的時候肯定會失敗。thread_a()調(diào)用high_level_func(),讓high_level_mutex④上鎖(其層級值為10000①),為了獲取high_level_stuff()的參數(shù)對互斥量上鎖,之后調(diào)用low_level_func()⑤。low_level_func()會對low_level_mutex上鎖,這就沒有問題了,因為這個互斥量有一個低層值5000②。

thread_b()運行就不會順利了。首先,它鎖住了other_mutex⑩,這個互斥量的層級值只有100⑦。這就意味著,超低層級的數(shù)據(jù)已被保護。當other_stuff()調(diào)用high_level_func()⑧時,就違反了層級結(jié)構(gòu):high_level_func()試圖獲取high_level_mutex,這個互斥量的層級值是10000,要比當前層級值100大很多。因此hierarchical_mutex將會產(chǎn)生一個錯誤,可能會是拋出一個異常,或直接終止程序。在層級互斥量上產(chǎn)生死鎖,是不可能的,因為互斥量本身會嚴格遵循約定順序,進行上鎖。這也意味,當多個互斥量在是在同一級上時,不能同時持有多個鎖,所以“手遞手”鎖的方案需要每個互斥量在一條鏈上,并且每個互斥量都比其前一個有更低的層級值,這在某些情況下無法實現(xiàn)。

例子也展示了另一點,std::lock_guard<>模板與用戶定義的互斥量類型一起使用。雖然hierarchical_mutex不是C++標準的一部分,但是它寫起來很容易;一個簡單的實現(xiàn)在列表3.8中展示出來。盡管它是一個用戶定義類型,它可以用于std::lock_guard<>模板中,因為它的實現(xiàn)有三個成員函數(shù)為了滿足互斥量操作:lock(), unlock() 和 try_lock()。雖然你還沒見過try_lock()怎么使用,但是其使用起來很簡單:當互斥量上的鎖被一個線程持有,它將返回false,而不是等待調(diào)用的線程,直到能夠獲取互斥量上的鎖為止。在std::lock()的內(nèi)部實現(xiàn)中,try_lock()會作為避免死鎖算法的一部分。

列表3.8 簡單的層級互斥量實現(xiàn)

class hierarchical_mutex
{
  std::mutex internal_mutex;

  unsigned long const hierarchy_value;
  unsigned long previous_hierarchy_value;

  static thread_local unsigned long this_thread_hierarchy_value;  // 1

  void check_for_hierarchy_violation()
  {
    if(this_thread_hierarchy_value <= hierarchy_value)  // 2
    {
      throw std::logic_error(“mutex hierarchy violated”);
    }
  }

  void update_hierarchy_value()
  {
    previous_hierarchy_value=this_thread_hierarchy_value;  // 3
    this_thread_hierarchy_value=hierarchy_value;
  }

public:
  explicit hierarchical_mutex(unsigned long value):
      hierarchy_value(value),
      previous_hierarchy_value(0)
  {}

  void lock()
  {
    check_for_hierarchy_violation();
    internal_mutex.lock();  // 4
    update_hierarchy_value();  // 5
  }

  void unlock()
  {
    this_thread_hierarchy_value=previous_hierarchy_value;  // 6
    internal_mutex.unlock();
  }

  bool try_lock()
  {
    check_for_hierarchy_violation();
    if(!internal_mutex.try_lock())  // 7
      return false;
    update_hierarchy_value();
    return true;
  }
};
thread_local unsigned long
     hierarchical_mutex::this_thread_hierarchy_value(ULONG_MAX);  // 8

這里重點是使用了thread_local的值來代表當前線程的層級值:this_thread_hierarchy_value①。它被初始化為最大值⑧,所以最初所有線程都能被鎖住。因為其聲明中有thread_local,所以每個線程都有其拷貝副本,這樣線程中變量狀態(tài)完全獨立,當從另一個線程進行讀取時,變量的狀態(tài)也完全獨立。參見附錄A,A.8節(jié),有更多與thread_local相關(guān)的內(nèi)容。

所以,第一次線程鎖住一個hierarchical_mutex時,this_thread_hierarchy_value的值是ULONG_MAX。由于其本身的性質(zhì),這個值會大于其他任何值,所以會通過check_for_hierarchy_vilation()②的檢查。在這種檢查方式下,lock()代表內(nèi)部互斥鎖已被鎖?、?。一旦成功鎖住,你可以更新層級值了⑤。

當你現(xiàn)在鎖住另一個hierarchical_mutex時,還持有第一個鎖,this_thread_hierarchy_value的值將會顯示第一個互斥量的層級值。第二個互斥量的層級值必須小于已經(jīng)持有互斥量檢查函數(shù)②才能通過。

現(xiàn)在,最重要的是為當前線程存儲之前的層級值,所以你可以調(diào)用unlock()⑥對層級值進行保存;否則,就鎖不住任何互斥量(第二個互斥量的層級數(shù)高于第一個互斥量),即使線程沒有持有任何鎖。因為保存了之前的層級值,只有當持有internal_mutex③,且在解鎖內(nèi)部互斥量⑥之前存儲它的層級值,才能安全的將hierarchical_mutex自身進行存儲。這是因為hierarchical_mutex被內(nèi)部互斥量的鎖所保護著。

try_lock()與lock()的功能相似,除了在調(diào)用internal_mutex的try_lock()⑦失敗時,不能持有對應(yīng)鎖,所以不必更新層級值,并直接返回false。

雖然是運行時檢測,但是它沒有時間依賴性——不必去等待那些導致死鎖出現(xiàn)的罕見條件。同時,設(shè)計過程需要去拆分應(yīng)用,互斥量在這樣的情況下可以消除可能導致死鎖的可能性。這樣的設(shè)計練習很有必要去做一下,即使你之后沒有去做,代碼也會在運行時進行檢查。

超越鎖的延伸擴展

如我在本節(jié)開頭提到的那樣,死鎖不僅僅會發(fā)生在鎖之間;死鎖也會發(fā)生在任何同步構(gòu)造中(可能會產(chǎn)生一個等待循環(huán)),因此這方面也需要有指導意見,例如:要去避免獲取嵌套鎖等待一個持有鎖的線程是一個很糟糕的決定,因為線程為了能繼續(xù)運行可能需要獲取對應(yīng)的鎖。類似的,如果去等待一個線程結(jié)束,它應(yīng)該可以確定這個線程的層級,這樣一個線程只需要等待比起層級低的線程結(jié)束即可??梢杂靡粋€簡單的辦法去確定,以添加的線程是否在同一函數(shù)中被啟動,如同在3.1.2節(jié)和3.3節(jié)中描述的那樣。

當代碼已經(jīng)能規(guī)避死鎖,std::lock()std::lock_guard能組成簡單的鎖覆蓋大多數(shù)情況,但是有時需要更多的靈活性。在這些情況,可以使用標準庫提供的std::unique_lock模板。如std::lock_guard,這是一個參數(shù)化的互斥量模板類,并且它提供很多RAII類型鎖用來管理std::lock_guard類型,可以讓代碼更加靈活。

3.2.6 std::unique_lock——靈活的鎖

std::unqiue_lock使用更為自由的不變量,這樣std::unique_lock實例不會總與互斥量的數(shù)據(jù)類型相關(guān),使用起來要比std:lock_guard更加靈活。首先,可將std::adopt_lock作為第二個參數(shù)傳入構(gòu)造函數(shù),對互斥量進行管理;也可以將std::defer_lock作為第二個參數(shù)傳遞進去,表明互斥量應(yīng)保持解鎖狀態(tài)。這樣,就可以被std::unique_lock對象(不是互斥量)的lock()函數(shù)的所獲取,或傳遞std::unique_lock對象到std::lock()中。清單3.6可以輕易的轉(zhuǎn)換為清單3.9,使用std::unique_lockstd::defer_lock①,而非std::lock_guardstd::adopt_lock。代碼長度相同,幾乎等價,唯一不同的就是:std::unique_lock會占用比較多的空間,并且比std::lock_guard稍慢一些。保證靈活性要付出代價,這個代價就是允許std::unique_lock實例不帶互斥量:信息已被存儲,且已被更新。

清單3.9 交換操作中std::lock()std::unique_lock的使用

class some_big_object;
void swap(some_big_object& lhs,some_big_object& rhs);
class X
{
private:
  some_big_object some_detail;
  std::mutex m;
public:
  X(some_big_object const& sd):some_detail(sd){}
  friend void swap(X& lhs, X& rhs)
  {
    if(&lhs==&rhs)
      return;
    std::unique_lock<std::mutex> lock_a(lhs.m,std::defer_lock); // 1 
    std::unique_lock<std::mutex> lock_b(rhs.m,std::defer_lock); // 1 std::def_lock 留下未上鎖的互斥量
    std::lock(lock_a,lock_b); // 2 互斥量在這里上鎖
    swap(lhs.some_detail,rhs.some_detail);
  }
};

列表3.9中,因為std::unique_lock支持lock(), try_lock()和unlock()成員函數(shù),所以能將std::unique_lock對象傳遞到std::lock()②。這些同名的成員函數(shù)在低層做著實際的工作,并且僅更新std::unique_lock實例中的標志,來確定該實例是否擁有特定的互斥量,這個標志是為了確保unlock()在析構(gòu)函數(shù)中被正確調(diào)用。如果實例擁有互斥量,那么析構(gòu)函數(shù)必須調(diào)用unlock();但當實例中沒有互斥量時,析構(gòu)函數(shù)就不能去調(diào)用unlock()。這個標志可以通過owns_lock()成員變量進行查詢。

可能如你期望的那樣,這個標志被存儲在某個地方。因此,std::unique_lock對象的體積通常要比std::lock_guard對象大,當使用std::unique_lock替代std::lock_guard,因為會對標志進行適當?shù)母禄驒z查,就會做些輕微的性能懲罰。當std::lock_guard已經(jīng)能夠滿足你的需求,那么還是建議你繼續(xù)使用它。當需要更加靈活的鎖時,最好選擇std::unique_lock,因為它更適合于你的任務(wù)。你已經(jīng)看到一個遞延鎖的例子,另外一種情況是鎖的所有權(quán)需要從一個域轉(zhuǎn)到另一個域。

3.2.7 不同域中互斥量所有權(quán)的傳遞

std::unique_lock實例沒有與自身相關(guān)的互斥量,一個互斥量的所有權(quán)可以通過移動操作,在不同的實例中進行傳遞。某些情況下,這種轉(zhuǎn)移是自動發(fā)生的,例如:當函數(shù)返回一個實例;另些情況下,需要顯式的調(diào)用std::move()來執(zhí)行移動操作。從本質(zhì)上來說,需要依賴于源值是否是左值——一個實際的值或是引用——或一個右值——一個臨時類型。當源值是一個右值,為了避免轉(zhuǎn)移所有權(quán)過程出錯,就必須顯式移動成左值。std::unique_lock是可移動,但不可賦值的類型。附錄A,A.1.1節(jié)有更多與移動語句相關(guān)的信息。

一種使用可能是允許一個函數(shù)去鎖住一個互斥量,并且將所有權(quán)移到調(diào)用者上,所以調(diào)用者可以在這個鎖保護的范圍內(nèi)執(zhí)行額外的動作。

下面的程序片段展示了:函數(shù)get_lock()鎖住了互斥量,然后準備數(shù)據(jù),返回鎖的調(diào)用函數(shù):

std::unique_lock<std::mutex> get_lock()
{
  extern std::mutex some_mutex;
  std::unique_lock<std::mutex> lk(some_mutex);
  prepare_data();
  return lk;  // 1
}
void process_data()
{
  std::unique_lock<std::mutex> lk(get_lock());  // 2
  do_something();
}

lk在函數(shù)中被聲明為自動變量,它不需要調(diào)用std::move(),可以直接返回①(編譯器負責調(diào)用移動構(gòu)造函數(shù))。process_data()函數(shù)直接轉(zhuǎn)移std::unique_lock實例的所有權(quán)②,調(diào)用do_something()可使用的正確數(shù)據(jù)(數(shù)據(jù)沒有受到其他線程的修改)。

通常這種模式會用于已鎖的互斥量,其依賴于當前程序的狀態(tài),或依賴于傳入返回類型為std::unique_lock的函數(shù)(或以參數(shù)返回)。這樣的用法不會直接返回鎖,不過網(wǎng)關(guān)類的一個數(shù)據(jù)成員可用來確認已經(jīng)對保護數(shù)據(jù)的訪問權(quán)限進行上鎖。這種情況下,所有的訪問都必須通過網(wǎng)關(guān)類:當你想要訪問數(shù)據(jù),需要獲取網(wǎng)關(guān)類的實例(如同前面的例子,通過調(diào)用get_lock()之類函數(shù))來獲取鎖。之后你就可以通過網(wǎng)關(guān)類的成員函數(shù)對數(shù)據(jù)進行訪問。當完成訪問,可以銷毀這個網(wǎng)關(guān)類對象,將鎖進行釋放,讓別的線程來訪問保護數(shù)據(jù)。這樣的一個網(wǎng)關(guān)類可能是可移動的(所以他可以從一個函數(shù)進行返回),在這種情況下鎖對象的數(shù)據(jù)必須是可移動的。

std::unique_lock的靈活性同樣也允許實例在銷毀之前放棄其擁有的鎖??梢允褂胾nlock()來做這件事,如同一個互斥量:std::unique_lock的成員函數(shù)提供類似于鎖定和解鎖互斥量的功能。std::unique_lock實例在銷毀前釋放鎖的能力,當鎖沒有必要在持有的時候,可以在特定的代碼分支對其進行選擇性的釋放。這對于應(yīng)用性能來說很重要,因為持有鎖的時間增加會導致性能下降,其他線程會等待這個鎖的釋放,避免超越操作。

3.2.8 鎖的粒度

3.2.3節(jié)中,已經(jīng)對鎖的粒度有所了解:鎖的粒度是一個擺手術(shù)語(hand-waving term),用來描述通過一個鎖保護著的數(shù)據(jù)量大小。一個細粒度鎖(a fine-grained lock)能夠保護較小的數(shù)據(jù)量,一個粗粒度鎖(a coarse-grained lock)能夠保護較多的數(shù)據(jù)量。選擇粒度對于鎖來說很重要,為了保護對應(yīng)的數(shù)據(jù),保證鎖有能力保護這些數(shù)據(jù)也很重要。我們都知道,在超市等待結(jié)賬的時候,正在結(jié)賬的顧客突然意識到他忘了拿蔓越莓醬,然后離開柜臺去拿,并讓其他的人都等待他回來;或者當收銀員,準備收錢時,顧客才去翻錢包拿錢,這樣的情況都會讓等待的顧客很無奈。當每個人都檢查了自己要拿的東西,且能隨時為拿到的商品進行支付,那么的每件事都會進行的很順利。

這樣的道理同樣適用于線程:如果很多線程正在等待同一個資源(等待收銀員對自己拿到的商品進行清點),當有線程持有鎖的時間過長,這就會增加等待的時間(別等到結(jié)賬的時候,才想起來蔓越莓醬沒拿)。在可能的情況下,鎖住互斥量的同時只能對共享數(shù)據(jù)進行訪問;試圖對鎖外數(shù)據(jù)進行處理。特別是做一些費時的動作,比如:對文件的輸入/輸出操作進行上鎖。文件輸入/輸出通常要比從內(nèi)存中讀或?qū)懲瑯娱L度的數(shù)據(jù)慢成百上千倍,所以除非鎖已經(jīng)打算去保護對文件的訪問,要么執(zhí)行輸入/輸出操作將會將延遲其他線程執(zhí)行的時間,這很沒有必要(因為文件鎖阻塞住了很多操作),這樣多線程帶來的性能效益會被抵消。

std::unique_lock在這種情況下工作正常,在調(diào)用unlock()時,代碼不需要再訪問共享數(shù)據(jù);而后當再次需要對共享數(shù)據(jù)進行訪問時,就可以再調(diào)用lock()了。下面代碼就是這樣的一種情況:

void get_and_process_data()
{
  std::unique_lock<std::mutex> my_lock(the_mutex);
  some_class data_to_process=get_next_data_chunk();
  my_lock.unlock();  // 1 不要讓鎖住的互斥量越過process()函數(shù)的調(diào)用
  result_type result=process(data_to_process);
  my_lock.lock(); // 2 為了寫入數(shù)據(jù),對互斥量再次上鎖
  write_result(data_to_process,result);
}

不需要讓鎖住的互斥量越過對process()函數(shù)的調(diào)用,所以可以在函數(shù)調(diào)用①前對互斥量手動解鎖,并且在之后對其再次上鎖②。

這能表示只有一個互斥量保護整個數(shù)據(jù)結(jié)構(gòu)時的情況,不僅可能會有更多對鎖的競爭,也會增加鎖持鎖的時間。較多的操作步驟需要獲取同一個互斥量上的鎖,所以持有鎖的時間會更長。成本上的雙重打擊也算是為向細粒度鎖轉(zhuǎn)移提供了雙重激勵和可能。

如同上面的例子,鎖不僅是能鎖住合適粒度的數(shù)據(jù),還要控制鎖的持有時間,以及什么操作在執(zhí)行的同時能夠擁有鎖。一般情況下,執(zhí)行必要的操作時,盡可能將持有鎖的時間縮減到最小。這也就意味有一些浪費時間的操作,比如:獲取另外一個鎖(即使你知道這不會造成死鎖),或等待輸入/輸出操作完成時沒有必要持有一個鎖(除非絕對需要)。

清單3.6和3.9中,交換操作需要鎖住兩個互斥量,其明確要求并發(fā)訪問兩個對象。假設(shè)用來做比較的是一個簡單的數(shù)據(jù)類型(比如:int類型),將會有什么不同么?int的拷貝很廉價,所以可以很容易的進行數(shù)據(jù)復制,并且每個被比較的對象都持有該對象的鎖,在比較之后進行數(shù)據(jù)拷貝。這就意味著,在最短時間內(nèi)持有每個互斥量,并且你不會在持有一個鎖的同時再去獲取另一個。下面的清單中展示了一個在這樣情景中的Y類,并且展示了一個相等比較運算符的等價實現(xiàn)。

列表3.10 比較操作符中一次鎖住一個互斥量

class Y
{
private:
  int some_detail;
  mutable std::mutex m;
  int get_detail() const
  {
    std::lock_guard<std::mutex> lock_a(m);  // 1
    return some_detail;
  }
public:
  Y(int sd):some_detail(sd){}

  friend bool operator==(Y const& lhs, Y const& rhs)
  {
    if(&lhs==&rhs)
      return true;
    int const lhs_value=lhs.get_detail();  // 2
    int const rhs_value=rhs.get_detail();  // 3
    return lhs_value==rhs_value;  // 4
  }
};

在這個例子中,比較操作符首先通過調(diào)用get_detail()成員函數(shù)檢索要比較的值②③,函數(shù)在索引值時被一個鎖保護著①。比較操作符會在之后比較索引出來的值④。注意:雖然這樣能減少鎖持有的時間,一個鎖只持有一次(這樣能消除死鎖的可能性),這里有一個微妙的語義操作同時對兩個鎖住的值進行比較。

列表3.10中,當操作符返回true時,那就意味著在這個時間點上的lhs.some_detail與在另一個時間點的rhs.some_detail相同。這兩個值在讀取之后,可能會被任意的方式所修改;兩個值會在②和③處進行交換,這樣就會失去比較的意義。等價比較可能會返回true,來表明這兩個值時相等的,實際上這兩個值相等的情況可能就發(fā)生在一瞬間。這樣的變化要小心,語義操作是無法改變一個問題的比較方式:當你持有鎖的時間沒有達到整個操作時間,就會讓自己處于條件競爭的狀態(tài)。

有時,只是沒有一個合適粒度級別,因為并不是所有對數(shù)據(jù)結(jié)構(gòu)的訪問都需要同一級的保護。這個例子中,就需要尋找一個合適的機制,去替換std::mutex


[1] Tom Cargill, “Exception Handling: A False Sense of Security,” in C++ Report 6, no. 9 (November–December 1994). Also available at http://www.informit.com/content/images/020163371x/supplements/Exception_Handling_Article.html.

[2] Herb Sutter, Exceptional C++: 47 Engineering Puzzles, Programming Problems, and Solutions (Addison Wesley Pro-fessional, 1999).