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

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

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

3.2.1 C++中使用互斥量

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

清單3.1 使用互斥量保護(hù)列表

#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中有一個(gè)全局變量①，這個(gè)全局變量被一個(gè)全局的互斥量保護(hù)②。add_to_list()③和list_contains()④函數(shù)中使用std::lock_guard<std::mutex>，使得這兩個(gè)函數(shù)中對(duì)數(shù)據(jù)的訪問(wèn)是互斥的：list_contains()不可能看到正在被add_to_list()修改的列表。

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

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

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

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

清單3.2 無(wú)意中傳遞了保護(hù)數(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 傳遞“保護(hù)”數(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 傳遞一個(gè)惡意函數(shù)
  unprotected->do_something();    // 3 在無(wú)保護(hù)的情況下訪問(wèn)保護(hù)數(shù)據(jù)
}

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

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

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

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

因?yàn)槭褂昧嘶コ饬炕蚱渌麢C(jī)制保護(hù)了共享數(shù)據(jù)，就不必再為條件競(jìng)爭(zhēng)所擔(dān)憂嗎？并不是，你依舊需要確定數(shù)據(jù)受到了保護(hù)?；叵胫半p鏈表的例子，為了能讓線程安全地刪除一個(gè)節(jié)點(diǎn)，需要確保防止對(duì)這三個(gè)節(jié)點(diǎn)(待刪除的節(jié)點(diǎn)及其前后相鄰的節(jié)點(diǎn))的并發(fā)訪問(wèn)。如果只對(duì)指向每個(gè)節(jié)點(diǎn)的指針進(jìn)行訪問(wèn)保護(hù)，那就和沒(méi)有使用互斥量一樣，條件競(jìng)爭(zhēng)仍會(huì)發(fā)生——除了指針，整個(gè)數(shù)據(jù)結(jié)構(gòu)和整個(gè)刪除操作需要保護(hù)。這種情況下最簡(jiǎn)單的解決方案就是使用互斥量來(lái)保護(hù)整個(gè)鏈表，如清單3.1所示。

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

清單3.3 std::stack容器的實(shí)現(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)用并返回時(shí)是正確的，但其的結(jié)果是不可靠的；當(dāng)它們返回后，其他線程就可以自由地訪問(wèn)棧，并且可能push()多個(gè)新元素到棧中，也可能pop()一些已在棧中的元素。這樣的話，之前從empty()和size()得到的結(jié)果就有問(wèn)題了。

特別地，當(dāng)棧實(shí)例是非共享的，如果棧非空，使用empty()檢查再調(diào)用top()訪問(wèn)棧頂部的元素是安全的。如下代碼所示：

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

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

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

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

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

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

當(dāng)線程運(yùn)行時(shí)，調(diào)用兩次top()，棧沒(méi)被修改，所以每個(gè)線程能得到同樣的值。不僅是這樣，在調(diào)用top()函數(shù)調(diào)用的過(guò)程中(兩次)，pop()函數(shù)都沒(méi)有被調(diào)用。這樣，在其中一個(gè)值再讀取的時(shí)候，雖然不會(huì)出現(xiàn)“寫后讀”的情況，但其值已被處理了兩次。這種條件競(jìng)爭(zhēng)，比未定義的empty()/top()競(jìng)爭(zhēng)更加嚴(yán)重；雖然其結(jié)果依賴于do_something()的結(jié)果，但因?yàn)榭雌饋?lái)沒(méi)有任何錯(cuò)誤，就會(huì)讓這個(gè)Bug很難定位。

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

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

不幸的是，這樣的分割卻制造了本想避免或消除的條件競(jìng)爭(zhēng)。幸運(yùn)的是，我們還有的別的選項(xiàng)，但是使用這些選項(xiàng)是要付出代價(jià)的。

選項(xiàng)1： 傳入一個(gè)引用

第一個(gè)選項(xiàng)是將變量的引用作為參數(shù)，傳入pop()函數(shù)中獲取想要的“彈出值”：

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

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

選項(xiàng)2：無(wú)異常拋出的拷貝構(gòu)造函數(shù)或移動(dòng)構(gòu)造函數(shù)

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

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

、選項(xiàng)3：返回指向彈出值的指針

第三個(gè)選擇是返回一個(gè)指向彈出元素的指針，而不是直接返回值。指針的優(yōu)勢(shì)是自由拷貝，并且不會(huì)產(chǎn)生異常，這樣你就能避免Cargill提到的異常問(wèn)題了。缺點(diǎn)就是返回一個(gè)指針需要對(duì)對(duì)象的內(nèi)存分配進(jìn)行管理，對(duì)于簡(jiǎn)單數(shù)據(jù)類型(比如：int)，內(nèi)存管理的開銷要遠(yuǎn)大于直接返回值。對(duì)于選擇這個(gè)方案的接口，使用std::shared_ptr是個(gè)不錯(cuò)的選擇；不僅能避免內(nèi)存泄露(因?yàn)楫?dāng)對(duì)象中指針銷毀時(shí)，對(duì)象也會(huì)被銷毀)，而且標(biāo)準(zhǔn)庫(kù)能夠完全控制內(nèi)存分配方案，也就不需要new和delete操作。這種優(yōu)化是很重要的：因?yàn)槎褩Ｖ械拿總€(gè)對(duì)象，都需要用new進(jìn)行獨(dú)立的內(nèi)存分配，相較于非線程安全版本，這個(gè)方案的開銷相當(dāng)大。

選項(xiàng)4：“選項(xiàng)1 + 選項(xiàng)2”或 “選項(xiàng)1 + 選項(xiàng)3”

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

例：定義線程安全的堆棧

清單3.4中是一個(gè)接口沒(méi)有條件競(jìng)爭(zhēng)的堆棧類定義，它實(shí)現(xiàn)了選項(xiàng)1和選項(xiàng)3：重載了pop()，使用一個(gè)局部引用去存儲(chǔ)彈出值，并返回一個(gè)std::shared_ptr<>對(duì)象。它有一個(gè)簡(jiǎn)單的接口，只有兩個(gè)函數(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;
};

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

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

#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ù)對(duì)互斥量上鎖，再拷貝堆棧。構(gòu)造函數(shù)體中①的拷貝使用互斥量來(lái)確保復(fù)制結(jié)果的正確性，這樣的方式比成員初始化列表好。

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

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

一個(gè)給定操作需要兩個(gè)或兩個(gè)以上的互斥量時(shí)，另一個(gè)潛在的問(wèn)題將出現(xiàn)：死鎖。與條件競(jìng)爭(zhēng)完全相反——不同的兩個(gè)線程會(huì)互相等待，從而什么都沒(méi)做。

3.2.4 死鎖：?jiǎn)栴}描述及解決方案

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

現(xiàn)在沒(méi)有孩子去爭(zhēng)搶玩具，但線程有對(duì)鎖的競(jìng)爭(zhēng)：一對(duì)線程需要對(duì)他們所有的互斥量做一些操作，其中每個(gè)線程都有一個(gè)互斥量，且等待另一個(gè)解鎖。這樣沒(méi)有線程能工作，因?yàn)樗麄兌荚诘却龑?duì)方釋放互斥量。這種情況就是死鎖，它的最大問(wèn)題就是由兩個(gè)或兩個(gè)以上的互斥量來(lái)鎖定一個(gè)操作。

避免死鎖的一般建議，就是讓兩個(gè)互斥量總以相同的順序上鎖：總在互斥量B之前鎖住互斥量A，就永遠(yuǎn)不會(huì)死鎖。某些情況下是可以這樣用，因?yàn)椴煌幕コ饬坑糜诓煌牡胤?。不過(guò)，事情沒(méi)那么簡(jiǎn)單，比如：當(dāng)有多個(gè)互斥量保護(hù)同一個(gè)類的獨(dú)立實(shí)例時(shí)，一個(gè)操作對(duì)同一個(gè)類的兩個(gè)不同實(shí)例進(jìn)行數(shù)據(jù)的交換操作，為了保證數(shù)據(jù)交換操作的正確性，就要避免數(shù)據(jù)被并發(fā)修改，并確保每個(gè)實(shí)例上的互斥量都能鎖住自己要保護(hù)的區(qū)域。不過(guò)，選擇一個(gè)固定的順序(例如，實(shí)例提供的第一互斥量作為第一個(gè)參數(shù)，提供的第二個(gè)互斥量為第二個(gè)參數(shù))，可能會(huì)適得其反：在參數(shù)交換了之后，兩個(gè)線程試圖在相同的兩個(gè)實(shí)例間進(jìn)行數(shù)據(jù)交換時(shí)，程序又死鎖了！

很幸運(yùn)，C++標(biāo)準(zhǔn)庫(kù)有辦法解決這個(gè)問(wèn)題，std::lock——可以一次性鎖住多個(gè)(兩個(gè)以上)的互斥量，并且沒(méi)有副作用(死鎖風(fēng)險(xiǎn))。下面的程序清單中，就來(lái)看一下怎么在一個(gè)簡(jiǎn)單的交換操作中使用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ù)是否是不同的實(shí)例，因?yàn)椴僮髟噲D獲取std::mutex對(duì)象上的鎖，所以當(dāng)其被獲取時(shí)，結(jié)果很難預(yù)料。(一個(gè)互斥量可以在同一線程上多次上鎖，標(biāo)準(zhǔn)庫(kù)中std::recursive_mutex提供這樣的功能。詳情見3.3.3節(jié))。然后，調(diào)用std::lock()①鎖住兩個(gè)互斥量，并且兩個(gè)std:lock_guard實(shí)例已經(jīng)創(chuàng)建好②③。提供std::adopt_lock參數(shù)除了表示std::lock_guard對(duì)象可獲取鎖之外，還將鎖交由std::lock_guard對(duì)象管理，而不需要std::lock_guard對(duì)象再去構(gòu)建新的鎖。

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

雖然std::lock可以在這情況下(獲取兩個(gè)以上的鎖)避免死鎖，但它沒(méi)辦法幫助你獲取其中一個(gè)鎖。這時(shí)，不得不依賴于開發(fā)者的紀(jì)律性(譯者：也就是經(jīng)驗(yàn))，來(lái)確保你的程序不會(huì)死鎖。這并不簡(jiǎn)單：死鎖是多線程編程中一個(gè)令人相當(dāng)頭痛的問(wèn)題，并且死鎖經(jīng)常是不可預(yù)見的，因?yàn)樵诖蠖鄶?shù)時(shí)間里，所有工作都能很好的完成。不過(guò)，也一些相對(duì)簡(jiǎn)單的規(guī)則能幫助寫出“無(wú)死鎖”的代碼。

3.2.5 避免死鎖的進(jìn)階指導(dǎo)

雖然鎖是產(chǎn)生死鎖的一般原因，但也不排除死鎖出現(xiàn)在其他地方。無(wú)鎖的情況下，僅需要每個(gè)std::thread對(duì)象調(diào)用join()，兩個(gè)線程就能產(chǎn)生死鎖。這種情況下，沒(méi)有線程可以繼續(xù)運(yùn)行，因?yàn)樗麄冋诨ハ嗟却?。這種情況很常見，一個(gè)線程會(huì)等待另一個(gè)線程，其他線程同時(shí)也會(huì)等待第一個(gè)線程結(jié)束，所以三個(gè)或更多線程的互相等待也會(huì)發(fā)生死鎖。為了避免死鎖，這里的指導(dǎo)意見為：當(dāng)機(jī)會(huì)來(lái)臨時(shí)，不要拱手讓人。以下提供一些個(gè)人的指導(dǎo)建議，如何識(shí)別死鎖，并消除其他線程的等待。

避免嵌套鎖

第一個(gè)建議往往是最簡(jiǎn)單的：一個(gè)線程已獲得一個(gè)鎖時(shí)，再別去獲取第二個(gè)。如果能堅(jiān)持這個(gè)建議，因?yàn)槊總€(gè)線程只持有一個(gè)鎖，鎖上就不會(huì)產(chǎn)生死鎖。即使互斥鎖造成死鎖的最常見原因，也可能會(huì)在其他方面受到死鎖的困擾(比如：線程間的互相等待)。當(dāng)你需要獲取多個(gè)鎖，使用一個(gè)std::lock來(lái)做這件事(對(duì)獲取鎖的操作上鎖)，避免產(chǎn)生死鎖。

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

第二個(gè)建議是次簡(jiǎn)單的：因?yàn)榇a是用戶提供的，你沒(méi)有辦法確定用戶要做什么；用戶程序可能做任何事情，包括獲取鎖。你在持有鎖的情況下，調(diào)用用戶提供的代碼；如果用戶代碼要獲取一個(gè)鎖，就會(huì)違反第一個(gè)指導(dǎo)意見，并造成死鎖(有時(shí)，這是無(wú)法避免的)。當(dāng)你正在寫一份通用代碼，例如3.2.3中的棧，每一個(gè)操作的參數(shù)類型，都在用戶提供的代碼中定義，就需要其他指導(dǎo)意見來(lái)幫助你。

使用固定順序獲取鎖

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

其他情況下，這就不會(huì)那么簡(jiǎn)單了，例如：3.2.4節(jié)中的交換操作，這種情況下你可能同時(shí)鎖住多個(gè)互斥量(但是有時(shí)不會(huì)發(fā)生)。當(dāng)回看3.1節(jié)中那個(gè)鏈表連接例子時(shí)，將會(huì)看到列表中的每個(gè)節(jié)點(diǎn)都會(huì)有一個(gè)互斥量保護(hù)。為了訪問(wèn)列表，線程必須獲取他們感興趣節(jié)點(diǎn)上的互斥鎖。當(dāng)一個(gè)線程刪除一個(gè)節(jié)點(diǎn)，它必須獲取三個(gè)節(jié)點(diǎn)上的互斥鎖：將要?jiǎng)h除的節(jié)點(diǎn)，兩個(gè)鄰接節(jié)點(diǎn)(因?yàn)樗麄円矔?huì)被修改)。同樣的，為了遍歷鏈表，線程必須保證在獲取當(dāng)前節(jié)點(diǎn)的互斥鎖前提下，獲得下一個(gè)節(jié)點(diǎn)的鎖，要保證指向下一個(gè)節(jié)點(diǎn)的指針不會(huì)同時(shí)被修改。一旦下一個(gè)節(jié)點(diǎn)上的鎖被獲取，那么第一個(gè)節(jié)點(diǎn)的鎖就可以釋放了，因?yàn)闆](méi)有持有它的必要性了。

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

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

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

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

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

清單3.7 使用層次鎖來(lái)避免死鎖

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ī)則，所以它運(yùn)行的沒(méi)問(wèn)題。另一方面，thread_b()⑨無(wú)視規(guī)則，因此在運(yùn)行的時(shí)候肯定會(huì)失敗。thread_a()調(diào)用high_level_func()，讓high_level_mutex④上鎖(其層級(jí)值為10000①)，為了獲取high_level_stuff()的參數(shù)對(duì)互斥量上鎖，之后調(diào)用low_level_func()⑤。low_level_func()會(huì)對(duì)low_level_mutex上鎖，這就沒(méi)有問(wèn)題了，因?yàn)檫@個(gè)互斥量有一個(gè)低層值5000②。

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

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

列表3.8 簡(jiǎn)單的層級(jí)互斥量實(shí)現(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

這里重點(diǎn)是使用了thread_local的值來(lái)代表當(dāng)前線程的層級(jí)值：this_thread_hierarchy_value①。它被初始化為最大值⑧，所以最初所有線程都能被鎖住。因?yàn)槠渎暶髦杏衪hread_local，所以每個(gè)線程都有其拷貝副本，這樣線程中變量狀態(tài)完全獨(dú)立，當(dāng)從另一個(gè)線程進(jìn)行讀取時(shí)，變量的狀態(tài)也完全獨(dú)立。參見附錄A，A.8節(jié)，有更多與thread_local相關(guān)的內(nèi)容。

所以，第一次線程鎖住一個(gè)hierarchical_mutex時(shí)，this_thread_hierarchy_value的值是ULONG_MAX。由于其本身的性質(zhì)，這個(gè)值會(huì)大于其他任何值，所以會(huì)通過(guò)check_for_hierarchy_vilation()②的檢查。在這種檢查方式下，lock()代表內(nèi)部互斥鎖已被鎖?、堋Ｒ坏┏晒︽i住，你可以更新層級(jí)值了⑤。

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

現(xiàn)在，最重要的是為當(dāng)前線程存儲(chǔ)之前的層級(jí)值，所以你可以調(diào)用unlock()⑥對(duì)層級(jí)值進(jìn)行保存；否則，就鎖不住任何互斥量(第二個(gè)互斥量的層級(jí)數(shù)高于第一個(gè)互斥量)，即使線程沒(méi)有持有任何鎖。因?yàn)楸４媪酥暗膶蛹?jí)值，只有當(dāng)持有internal_mutex③，且在解鎖內(nèi)部互斥量⑥之前存儲(chǔ)它的層級(jí)值，才能安全的將hierarchical_mutex自身進(jìn)行存儲(chǔ)。這是因?yàn)閔ierarchical_mutex被內(nèi)部互斥量的鎖所保護(hù)著。

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

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

超越鎖的延伸擴(kuò)展

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

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

3.2.6 std::unique_lock——靈活的鎖

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

清單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中，因?yàn)?code>std::unique_lock支持lock(), try_lock()和unlock()成員函數(shù)，所以能將std::unique_lock對(duì)象傳遞到std::lock()②。這些同名的成員函數(shù)在低層做著實(shí)際的工作，并且僅更新std::unique_lock實(shí)例中的標(biāo)志，來(lái)確定該實(shí)例是否擁有特定的互斥量，這個(gè)標(biāo)志是為了確保unlock()在析構(gòu)函數(shù)中被正確調(diào)用。如果實(shí)例擁有互斥量，那么析構(gòu)函數(shù)必須調(diào)用unlock()；但當(dāng)實(shí)例中沒(méi)有互斥量時(shí)，析構(gòu)函數(shù)就不能去調(diào)用unlock()。這個(gè)標(biāo)志可以通過(guò)owns_lock()成員變量進(jìn)行查詢。

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

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

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

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

下面的程序片段展示了：函數(shù)get_lock()鎖住了互斥量，然后準(zhǔn)備數(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ù)中被聲明為自動(dòng)變量，它不需要調(diào)用std::move()，可以直接返回①(編譯器負(fù)責(zé)調(diào)用移動(dòng)構(gòu)造函數(shù))。process_data()函數(shù)直接轉(zhuǎn)移std::unique_lock實(shí)例的所有權(quán)②，調(diào)用do_something()可使用的正確數(shù)據(jù)(數(shù)據(jù)沒(méi)有受到其他線程的修改)。

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

std::unique_lock的靈活性同樣也允許實(shí)例在銷毀之前放棄其擁有的鎖?？梢允褂胾nlock()來(lái)做這件事，如同一個(gè)互斥量：std::unique_lock的成員函數(shù)提供類似于鎖定和解鎖互斥量的功能。std::unique_lock實(shí)例在銷毀前釋放鎖的能力，當(dāng)鎖沒(méi)有必要在持有的時(shí)候，可以在特定的代碼分支對(duì)其進(jìn)行選擇性的釋放。這對(duì)于應(yīng)用性能來(lái)說(shuō)很重要，因?yàn)槌钟墟i的時(shí)間增加會(huì)導(dǎo)致性能下降，其他線程會(huì)等待這個(gè)鎖的釋放，避免超越操作。

3.2.8 鎖的粒度

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

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

std::unique_lock在這種情況下工作正常，在調(diào)用unlock()時(shí)，代碼不需要再訪問(wèn)共享數(shù)據(jù)；而后當(dāng)再次需要對(duì)共享數(shù)據(jù)進(jìn)行訪問(wèn)時(shí)，就可以再調(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 不要讓鎖住的互斥量越過(guò)process()函數(shù)的調(diào)用
  result_type result=process(data_to_process);
  my_lock.lock(); // 2 為了寫入數(shù)據(jù)，對(duì)互斥量再次上鎖
  write_result(data_to_process,result);
}

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

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

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

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

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

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
  }
};

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

列表3.10中，當(dāng)操作符返回true時(shí)，那就意味著在這個(gè)時(shí)間點(diǎn)上的lhs.some_detail與在另一個(gè)時(shí)間點(diǎn)的rhs.some_detail相同。這兩個(gè)值在讀取之后，可能會(huì)被任意的方式所修改；兩個(gè)值會(huì)在②和③處進(jìn)行交換，這樣就會(huì)失去比較的意義。等價(jià)比較可能會(huì)返回true，來(lái)表明這兩個(gè)值時(shí)相等的，實(shí)際上這兩個(gè)值相等的情況可能就發(fā)生在一瞬間。這樣的變化要小心，語(yǔ)義操作是無(wú)法改變一個(gè)問(wèn)題的比較方式：當(dāng)你持有鎖的時(shí)間沒(méi)有達(dá)到整個(gè)操作時(shí)間，就會(huì)讓自己處于條件競(jìng)爭(zhēng)的狀態(tài)。

有時(shí)，只是沒(méi)有一個(gè)合適粒度級(jí)別，因?yàn)椴⒉皇撬袑?duì)數(shù)據(jù)結(jié)構(gòu)的訪問(wèn)都需要同一級(jí)的保護(hù)。這個(gè)例子中，就需要尋找一個(gè)合適的機(jī)制，去替換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).