概述

于是，你有一个类似于上一节中链表那样的共享数据结构，你想要保护它免于竞争条件以及可能因此产生的不变量损坏。如果你可以将所有访问该数据结构的代码块标记为互斥的（mutually exclusive)，岂不是很好？如果任何线程运行了其中之一，所有其他试图访问此数据结构的线程就必须一直等到第一个线程完成。这就使得线程不可能看到损坏的不变量，除非它是进行修改的线程。

嗯，这并非无稽之谈——它正是使用称为互斥元（mutex, mutual exclusion）的同步原语所能得到的。在访问共享数据结构之前，锁定（lock）与该数据相关的互斥元，当访问数据结构完成后，解锁（unlock）该互斥元。线程库会确保一旦一个线程已经锁定某个互斥元，所有其他试图锁定相同互斥元的线程必须等待，直到成功锁定了该互斥元的线程解锁此互斥元。这就确保所有线程看到共享数据自洽的一面，不带有任何损坏的不变量。

互斥元是C++中最常见的数据保护机制，但并非灵丹妙药。精心组织代码来保护正确的数据以及避免接口中固有的竞争条件也是很重要的。互斥元也伴随着自己的问题，以死锁(deadlock)和保护过多或过少数据的形式。接下来，让我们从基础开始。

使用C++中的互斥元

在C++中，通过构造std::mutex的实例创建互斥元，调用成员函数lock()来锁定它，调用成员函数unlock()来解锁它。然而，直接调用成员函数是不推荐的做法，因为这意味着你必须记住在离开函数的每条代码路径上都调用unlock()，包括由于异常所导致的在内。作为替代，标准C++库提供了
std::lock_guard类模板，实现了互斥元的RAII惯用语法，它在构造时锁定所给的互斥元，在析构时将互斥元解锁，从而保证被锁定的互斥元始终被正确解锁。清单3.1的代码展示了如何使用std::mutex保护一个可被多个线程访问的列表，连同std::lock_guard。二者都声明于<mutex>头文件中。

//清单3.1 用互斥元保护列表
#include <list>
#include <mutex>
#include <algorithm>

std::list<int> some_list; //❶
std::mutex some_mutex; //❷

void add_to_list(int new_value)
{
	std::lock_guard<std::mutex> guard(some_mutex); //❸
	some_list.push_back(new_value);
}
bool list_contains(int value_to_find)
{
	std::lock_guard<std::mutex> guard(some_mutex); //❹
	return std::find(some_list.begin(), some_list.end(), value_to_find)
		!= some_list.end();
}

#include <iostream>

int main()
{
	add_to_list(42);
	std::cout << "contains(1)=" << list_contains(1) << ", contains(42)=" << list_contains(42) << std::endl;
}

在清单3.1中，有一个全局变量❶，它被相应的std::mutex的全局实例保护。在add_to_list()❸以及list_contains()❹中对std::lock_guard<std::mutex>的使用意味着这些函数中的访问是互斥的，list_contains()将无法在add_to_list()进行修改的半途中看到该列表。

尽管这种全局变量的使用偶尔也是恰当的，在大多数情况下，不用全局变量，而是在类中将互斥元和受保护的数据组织在一起，是很普遍的。这是一个标准的面向对象应用程序设计规则，通过将它们放在一个类中，清楚地标记他们是相关的，还可以封装函数以及强制保护。在这种情况下，函数add_to_list和list_contains将成为类的成员函数，互斥元和受保护的数据都作为类的private成员，使其更容易鉴别哪些代码可以访问数据，哪些代码需要锁定互斥元。如果类的所有成员函数在访问任意其他数据成员之前锁定互斥元，并且在操作完成时解锁，则数据对于所有的访问者都被很好地保护了。

其实，并不完全是那样，你将敏锐地发现，如果其中一个成员函数返回对受保护数据的指针或引用，那么所有成员函数都以良好顺序的方式锁定互斥元也是没关系的，因为你已在保护中捅了一个大窟隆。能够访问（并可能修改）该指针或引用的任意代码现在可以访问受保护的数据而无需锁定该互斥元。因此使用互斥元保护数据需要仔细设计接口，以确保在有任意对受保护的数据进行访问之前，互斥元已被锁定，且不留后门。

为保护共享数据精心组织代码

如你所见，用互斥元保护数据并不只是像在每个成员函数中拍进一个std::lock_guard对象那样容易，一个迷路的指针或引用，所有的保护都将白费。在一个层面上，检查迷路的指针或引用是容易的，只要没有一个成员函数通过其返回值或输出参数，返回受保护数据的指针或引用给其调用者，数据就安全了。如果更深入一些，它没有那么直观——远远没有。除了检查成员函数没有向其调用者传出指针和引用，检查它们没有向其调用的不在你掌控之下的函数传入这种指针和引用，也是很重要的。这同样危险，那些函数可能将指针和引用存储在某个地方，将来可以脱离互斥元的保护而被使用。在这方面特别危险的是，函数是通过函数参数或其他方式在运行时提供的，如清单3.2所示。

//清单3.2 意外地传出对受保护数据的引用
#include <mutex>

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); //❶传递"受保护的"数据到用户提供的函数
	}
};

some_data* unprotected;

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

data_wrapper x;

void foo()
{
	x.process_data(malicious_function); //❷传入一个恶意函数
	unprotected->do_something(); //❸对受保护的数据进行未受保护的访问
}

int main()
{
	foo();
}

在这个例子中，process_data中的代码看起来挺无害，受到std::lock_guard很好地保护，但对用户提供的函数 func的调用❶意味着foo可以传入malicious_function❷来绕过保护，然后无需锁定互斥元即可调用do_something ()❸。

从根本上说，这个代码的问题在于它没有完成你所设置的内容，标记所有访问该数据结构的代码为互斥的(mutually exclusive)。在这个例子中，忽略了foo()中调用unprotected->do_something()的代码。不幸的是，这部分问题不是C++线程库所能帮助你的，而这取决于作为程序员的你，去锁定正确的互斥元来保护你的数据。想想好的一面，你有了一个可遵循的准则，它会在这些情况下帮助你：不要将对受保护数据的指针和引用传递到锁的范围之外，无论是通过从函数中返回它们、将其存放在外部可见的内存中，还是作为参数传递给用户提供的函数。

虽然这是在试图使用互斥元来保护共享数据时常犯的错误，但这绝非唯一可能的隐患。在下一节中你会看到，可能仍然会有竞争条件，即便当数据被互斥元保护着。

发现接口中固有的竞争条件

仅仅因为使用了互斥元或其他机制来保护共享数据，未必会免于竞争条件，你仍然需要确定保护了适当的数据。再次考虑双向链表的例子。为了让线程安全地删除节点，你需要确保已阻止对三个结点的并发访问。要删除的节点及其两边的结点。如果你分别保护访问每个节点的指针，就不会比未使用互斥元的代码更好，因为竞争条件仍会发生——需要保护的不是个别步骤中的个别结点，而是整个删除操作中的整个数据结构。这种情况下最简单的解决办法，就是用单个互斥元保护整个列表，如清单3.1中所示。

仅仅因为在列表上的个别操作是安全的，你还没有摆脱困境。你仍然会遇到竞争条件，即便是一个非常简单的接口。考虑像std::stack 容器适配器这样的堆栈数据结构，如清单3.3中所示。除了构造函数和swap()，对std::stack你只有五件事情可以做：可以push()一个新元素入栈、pop()一个元素出栈、读top()元素、检查它是否empty()以及读取元素数量——-堆栈的size()。如果更改top()使得它返回一个副本，而不是引用，同时用互斥元保护内部数据，该接口依然固有地受制于竞争条件。这个问题对基于互斥元的实现并不是独一无二的。它是一个接口问题，因此对于无锁实现仍然会发生竞争条件。

//清单3.3 std::stack容器适配器的接口
#include <deque>
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&&);
};

int main()
{}

这里的问题是empty()的结果和size()不可靠。虽然它们可能在被调用时是正确的，一旦它们返回，在调用了empty()或size()的线程可以使用该信息之前，其他线程可以自由地访问堆栈，并且可能push()新元素入栈或pop()已有的元素出栈。

特别地，如果该stack实例是非共享的，如果栈非空，检查empty()并调用top()访问顶部元素是安全的，如下所示。

stack<int> s;
if(!s.empty()) //❶
{ 
    int const value = s.top(); //❷
    s.pop(); //❸
    do_something(value);
}

它不仅在单线程代码中是安全的，预计为：在空堆栈上调用top()是未定义的行为。对于共享的stack对象，这个调用序列不再安全，因为在调用empty()❶和调用top()❷之间可能有来自另一个线程的pop()调用，删除最后一个元素。因此，这是一个典型的竞争条件，为了保护栈的内容而在内部使用互斥元，却并未能将其阻止，这就是接口的影响。

怎么解决呢？发生这个问题是接口设计的后果，所以解决办法就是改变接口。然而，这仍然回避了问题，要作出什么样的改变？在最简单的情况下，你只要声明top()在调用时如果栈中没有元素则引发异常。虽然这直接解决了问题，但它使编程变得更麻烦，因为现在你得能捕捉异常，即使对empty()的调用返回false。这基本上使得empty()的调用变得纯粹多余。

如果你仔细看看前面的代码片段，还有另一个可能的竞争条件，但这一次是在调用top()❷和调用pop()❸之间。考虑运行着前面代码片段的两个线程，它们都引用着同一个 stack 对象s。这并非罕见的情形，当为了性能而使用线程时，有数个线程在不同的数据上运行相同的代码是很常见的，并且一个共享的stack对象非常适合用来在它们之间分隔工作。假设一开始栈里有两个元素，那么你不用担心在任一线程上的empty()和top()之间的竞争，只需考虑可能的执行模式。

如果栈从内部被互斥元保护，只有一个线程可以在任何时间运行栈的成员函数，那么这些调用就能得以很好地交错，而对do_something()的调用可以同时运行。一个可能的执行正如表3.1所示。

如你所见，如果这些是仅有的在运行的线程，在两次调用top()修改该栈之间没有任何东西，所以这两个线程将看到相同的值。不仅如此，在pop()的两次调用之间没有对top()的调用。因此，栈上的两个值其中一个还没被读取就被丢弃了，而另一个被处理了两次。这是另一种竞争条件，远比 empty()/top()竞争的未定义行为更槽糕。从来没有任何明显的错误发生，同时错误造成的后果可能和诱因差距甚远，尽管他们明显取决于do_something()到底做什么。

这要求对接口进行更加激进的改变，在互斥元的保护下结合对top()和pop()两者的调用。Tom Cargill指出，如果栈上对象的拷贝构造函数能够引发异常，结合调用可能会导致问题。从 Herb Sutter的异常安全的观点来看，这个问题被处理得较为全面，但潜在的竞争条件为这一结合带来了新的东西。

对于那些尚未意识到这个问题的人，考虑一下 stack<vector<int>>。现在，vector是一个动态大小的容器，所以当你复制vector时，为了复制其内容，库就必须从堆中分配更多的内存。如果系统负载过重，或有明显的资源约束，此次内存分配就可能失败，于是 vector的拷贝构造函数可能引发 std::bad_alloc异常。如果vector中含有大量的元素的话则尤其可能。如果pop()函数被定义为返回出栈值，并且从栈中删除它，就会有潜在的问题。仅在栈被修改后，出栈值才返回给调用者，但复制数据以返回给调用者的过程可能会引发异常。如果发生这种情况，刚从栈中出栈的数据会丢失，它已经从栈中被删除了，但该复制却没成功！std::stack接口的设计者笼统地将操作一分为二。获取顶部的元素(top())，然后将其从栈中删除(pop())，以致于你无法安全地复制数据，它将留在栈上。如果问题是堆内存不足，也许应用程序可以释放一些内存，然后再试一次。

不幸的是，这种划分正是你在消除竞争条件中试图去避免的！值得庆幸的是，还有替代方案，但他们并非无代价的。

选项1：传入引用

第一个选项是把你希望接受出栈值的变量的引用，作为参数传递给对pop()的调用。

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

这在很多情况下都适用，但它有个明显的缺点，要求调用代码在调用之前先构造一个该栈值类型的实例，以便将其作为目标传入。对于某些类型而言这是行不通的，因为构造一个实例在时间和资源方面是非常昂贵的。对于其他类型，这并不总是可能的，因为构造函数需要参数，而在代码的这个位置不一定可用。最后，它要求所存储的类型是可赋值的。这是一个重要的限制。许多用户定义的类型不支持赋值，尽管它们可能支持移动构造函数，或者甚至是拷贝构造函数（从而允许通过值来返回）。

选项2：要求不引发异常的拷贝构造函数或移动构造函数

对于有返回值的pop()而言只有一个异常安全问题，就是以值进行的返回可能引发异常。许多类型具有不引发异常的拷贝构造函数，并且在C++标准中有了新的右值引用的支持(参见附录A中A.1节)，越来越多的类型将不会引发异常的移动构造函数，即便他们的拷贝构造函数会如此。一个有效的选择，就是把对线程安全堆栈的使用，限制在能够安全地通过值来返回且不引发异常的类型之内。

虽然这样安全了，但并不理想。尽管你可以在编译时使用std::is_nothrow_copy_constructible和std::is_nothrow_move_constructible类型特征，来检测一个不引发异常的拷贝或移动构造函数的存在，但这却很受限制。相比于具有不能引发异常的拷贝和/或移动构造函数的类型，有更多的用户定义类型具有能够引发异常的拷贝构造函数且没有移动构造函数（尽管这会随着人们习惯了C++11中对右值引用的支持而改变)。如果这种类型不能被存储在你的线程安全堆栈中，是不幸的。

选项3：返回指向出栈项的指针

第三个选择是返回一个指向出栈项的指针，而非通过值来返回该项。其优点是指针可以被自由地复制而不会引发异常，这样你就避免了Cargill 的异常问题。其缺点是，返回一个指针时需要一种手段来管理分配给对象的内存，对于像整数这样简单的类型，这种内存管理的成本可能会超过仅通过值来返回该类型。对于任何使用此选项的接口，std::shared_ptr 会是指针类型的一个好的选择。它不仅避免了内存泄漏，因为一旦最后一个指针被销毁则该对象也会被销毁，并且库可以完全控制内存分配方案且不必使用new和delete。对于优化用途来说这是很重要的，要求用new分别分配堆栈中的每一个对象，会比原来非线程安全的版本带来大得多的开销。

选项4：同时提供选项1以及2或3

灵活性永远不应被排除在外，特别是在通用的代码中。如果你选择选项2或3，那么同时提供选项1也是相对容易的，这也为你的代码的用户提供了选择的能力，为了很小的额外成本，哪个选项对他们是最适合的。

一个线程安全堆栈的示范定义

清单3.4展示了在接口中没有竞争条件的栈的类定义，实现了选项1和3。pop()有两个重载，一个接受存储该值的位置的引用，另一个返回std::shared_ptr<>。它具有一个简单的接口，只有两个函数，push()和pop()。

//清单3.4 一个线程安全栈的概要类定义
#include <exception>
#include <memory>

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; //❶赋值运算符被删除了

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

int main()
{}

通过削减接口，你考虑到了最大的安全性，甚至对整个堆栈的操作都受到限制。栈本身不能被赋值，因为赋值运算符被删除❶，而且也没有swap()函数。然而，它可以被复制，假设栈的元素可以被复制。如果栈是空的，pop()函数引发一个empty_stack 异常，所以即使在调用empty()后栈被修改，一切仍将正常工作。正如选项3的描述中提到的，如果需要，std::shared_ptr的使用允许栈来处理内存分配问题，同时避免对new和delete 的过多调用。五个堆栈操作现在变成三个，push()、pop()和empty()。甚至empty()都是多余的。接口的简化可以更好地控制数据。你可以确保互斥元为了操作的整体而被锁定。清单3.5展现了一个简单的实现，一个围绕std::stack<>的封装器。

//清单3.5 一个线程安全栈的详细类定义
#include <exception>
#include <stack>
#include <mutex>
#include <memory>

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(){}
    threadsafe_stack(const threadsafe_stack& other)
    {
        std::lock_guard<std::mutex> lock(other.m);
        data=other.data; //❶在构造函数体中执行复制
    }
    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(); //在试着出栈值的时候检查是否为空
        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();
    }
};

int main()
{
    threadsafe_stack<int> si;
    si.push(5);
    si.pop();
    if(!si.empty())
    {
        int x;
        si.pop(x);
    }
    
}

这个栈的实现实际上是可复制的（copyable）——源对象中的拷贝构造函数锁定互斥元，然后复制内部栈。你在构造函数体中进行复制❶而不是成员初始化列表，以确保互斥元被整个副本持有。

top()和pop()的讨论表明，接口中有问题的竞争条件基本上因为锁定的粒度过小而引起。保护没有覆盖期望操作的整体。互斥元的问题也可以由锁定的粒度过大而引起，极端情况是单个的全局互斥元保护所有共享的数据。在一个有大量共享数据的系统中，这可能会消除并发的所有性能优势，因为线程被限制为每次只能运行一个，即便是在他们访问数据的不同部分的时候。被设计为处理多处理器系统的Linux内核的第一个版本，使用了单个全局内核锁。虽然这也能工作，但却意味着一个双处理器系统通常比两个单处理器系统的性能更差，四个处理器系统的性能远远没有四个单处理器系统的性能好。有太多对内核的竞争，因此在更多处理器上运行的线程无法进行有效的工作。Linux内核的后续版本已经转移到一个更细粒度的锁定方案，因而四个处理器的系统性能更接近理想的单处理器系统的4倍，因为竞争少得多。

细粒度锁定方案的一个问题，就是有时为了保护操作中的所有数据，需要不止一个互斥元。如前所述，有时要做的正确的事情是增加被互斥元所覆盖的数据粒度，以使得只需要一个互斥元被锁定。然而，这有时是不可取的，例如互斥元保护着一个类的各个实例。在这种情况下，在下个级别进行锁定，将意味着要么将锁丢给用户，要么就让单个互斥元保护该类的所有实例，这些都不甚理想。

如果对于一个给定的操作你最终需要锁定两个或更多的互斥元，还有另一个潜在的问题潜伏在侧：死锁（deadlock)。这几乎是竞争条件的反面，两个线程不是在竞争成为第一，而是每一个都在等待另外一个，因而都不会有任何进展。

死锁：问题和解决方案

试想一下，你有一个由两部分组成的玩具，并且你需要两个部分一起玩——例如，玩具鼓和鼓槌。现在，假设你有两个小孩，他们两人都喜欢玩它。如果其中一人同时得到鼓和鼓槌，那这个孩子就可以高兴地玩鼓，直到厌烦。如果另一个孩子想要玩，就得等，不管这让他多不爽。现在想象一下，鼓和鼓槌被（分别）埋在玩具箱里，你的孩子同时都决定玩它们，于是他们去翻玩具箱。其中一个发现了鼓，而另一个发现了鼓槌。现在他们被困住了，除非一人让另一人玩，不然每个人都会赖着他已有的东西，并要求另一人将另一部分给自己，否则就都玩不成。

现在想象一下，你没有抢玩具的孩子，但却有争夺互斥元的线程。一对线程中的每一个都需要同时锁定两个互斥元来执行一些操作，并且每个线程都拥有了一个互斥元，同时等待另外一个。线程都无法继续，因为每个线程都在等待另一个释放其互斥元。这种情景称为死锁(deadlock)，它是在需要锁定两个或更多互斥元以执行操作时的最大问题。

为了避免死锁，常见的建议是始终使用相同的顺序锁定这两个互斥元。如果你总是在互斥元B之前锁定互斥元A，那么你永远不会死锁。有时候这是很直观的，因为互斥元服务于不同的目的，但其他时候却并不那么简单，比如当互斥元分别保护相同类的各个实例时。例如，考虑同一个类的两个实例之间的数据交换操作，为了确保数据被正确地交换，而不受并发修改的影响，两个实例上的互斥元都必须被锁定。然而，如果选择了一个固定的顺序（例如，作为第一个参数提供的实例的互斥元，然后是作为第二个参数所提供的实力的互斥元），可能适得其反：它表示两个线程尝试通过交换参数，而在相同的两个实例之间交换数据，你将产生死锁。

幸运的是，C++标准库中的std::lock可以解决这一问题——std::lock函数可以同时锁定两个或更多的互斥元，而没有死锁的风险。清单3.6中的例子展示了如何使用它来完成简单的交换操作。

//清单3.6 在交换操作中使用std::lock()和std::lock_guard
#include <mutex>

class some_big_object
{};

void swap(some_big_object& lhs,some_big_object& rhs)
{}

class X
{
private:
    some_big_object some_detail;
    mutable 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); //❶
        std::lock_guard<std::mutex> lock_a(lhs.m,std::adopt_lock); //❷
        std::lock_guard<std::mutex> lock_b(rhs.m,std::adopt_lock); //❸
        swap(lhs.some_detail,rhs.some_detail);
    }
};

int main()
{}

首先，检查参数以确保它们是不同的实例，因为试图在你已经锁定了的std::mutex上获取锁，是未定义的行为。（允许同一线程多重锁定的互斥元类型为std::recursive_mutex）然后，调用std::lock()锁定这两个互斥元❶，同时构造两个std::lock_guard 的实例❷❶，每个实例对应一个互斥元。额外提供一个参数std::adopt_lock给互斥元，告知std::lock_guard对象该互斥元已被锁定，并且它们只应沿用互斥元上已有锁的所有权，而不是试图在构造函数中锁定互斥元。

这就确保了通常在受保护的操作可能引发异常的情况下，函数退出时正确地解锁互斥元，这也考虑到了简单返回。此外，值得一提的是，在对std::lock的调用中锁定lhs.m抑或是rhs.m都可能引发异常，在这种情况下，该异常被传播出std::lock。如果std::lock已经成功地在一个互斥元上获取了锁，当它试图在另一个互斥元上获取锁的时候，就会引发异常，前一个互斥元将会自动释放。std::lock 提供了关于锁定给定的互斥元的全或无的语义。

尽管std::lock能够帮助你在需要同时获得两个或更多锁的情况下避免死锁，但是如果要分别获取锁，就没用了。在这种情况下，你必须依靠你作为开发人员的戒律，以确保不会得到死锁。这谈何容易，死锁是在编写多线程代码时遇到的最令人头疼的问题之一，而且往往无法预测，大部分时间内一切都工作正常。然而，有一些相对简单的规则可以帮助你写出无死锁的代码。

避免死锁的进一步指南

死锁并不仅仅产生于锁定，虽然这是最常见的诱因。你可以通过两个线程来制造死锁，不用锁定，只需令每个线程在std::thread对象上为另一线程调用join()。在这种情况下，两个线程都无法取得进展，因为正等着另一个线程完成，就像孩子们争夺他们的玩具。这种简单的循环可以发生在任何地方，一个线程等待另一个线程执行一些动作而另一个线程同时又在等待第一个线程，而且这不仅限于两个线程，三个或更多线程的循环也会导致死锁。避免死锁的准则全都可以归结为一个思路，如果有另外一个线程有可能在等待你，那你就别等它。这个独特的准则为识别和消除别的线程等待你的可能性提供了方法。

避免嵌套锁

第一个思路是最简单的，如果你已经持有一个锁,就别再获取锁。如果你坚持这个准则，光凭使用锁是不可能导致死锁的，因为每个线程仅持有一个锁。你仍然会从其他事情（像是线程相互等待）中得到死锁，但是互斥元锁定可能死锁最常见的诱因。如果需要获取多个锁，为了避免死锁，就以std::lock的单个动作来实行。

在持有锁时，避免调用用户提供的代码

第一个思路是最简单的，如果你已经持有一个锁，就别再获取锁。如果你坚持这个准则，光凭使用锁是不可能导致死锁的，因为每个线程仅持有一个锁。你仍然会从其他事情（像是线程相互等待）中得到死锁，但是互斥元锁定可能死锁最常见的诱因。如果需要获取多个锁，为了避免死锁，就以std::lock的单个动作来实行。

以固定顺序获取锁

如果你绝对需要获取两个或更多的锁，并且不能以std::lock的单个操作取得，次优的做法是在每个线程中以相同的顺序获取它们。我曾谈及此点，是作为在获取两个互斥元时避免死锁的方法，关键是要以一种在线程间相一致的方法来定义其顺序。在某些情况下，这是相对简单的。例如，看一看之前中的堆栈——互斥元在每个栈实例的内部，但对于存储在栈中的数据项的操作，则需要调用用户提供的代码。然而，你可以添加约束，对于存储在栈中的数据项的操作，都不应对栈本身进行任何操作。这样就增加了栈的使用者的负担，但是将数据存储在一个容器中来访问该容器是很罕见的，并且一旦发生就会十分明显，因此这并不是一个很难承受的负担。

在别的情况下，可能就不那么直观，就像在之前你所看到的交换操作那样。至少在这种情况下，你可以同时锁定这些互斥元，但并不总是可能的。如果你回顾一下链表的例子，你会看到一种保护链表的可能性，就是让每个结点都有一个互斥元。然后，为了访问这个链表，线程必须获取它们感兴趣的每个结点上的锁。对于一个删除某项的线程，它就必须获得三个结点上的锁，要删除的结点以及它两边的结点，因为它们全都要以某种方式进行修改。同样地，为了遍历链表，线程在获取序列中下一个结点上的锁的时候，必须保持当前结点上的锁，以确保指向下一结点的指针在此期间不被修改。一旦获取到下一个结点上的锁，就可以释放前面结点上的锁，因为它已经没用了。

这种逐节向上的锁定方式允许多线程访问链表，前提是每个线程访问不同的结点。然而，为了避免死锁，必须始终以相同的顺序锁定结点。如果两个线程试图用逐节锁定的方式以相反的顺序遍历链表，它们就会在链表中间产生相互死锁。如果结点A和B在链表中相邻，一个方向上的线程会试图保持锁定结点A，并尝试获取结点B上的锁。而另一个方向上的线程会保持锁定结点B，并且尝试获得结点A上的锁——死锁的典型情况。

同样地，当删除位于结点A和C之间的结点B时，如果该线程在获取结点A和C上的锁之前获取B上的锁，它就有可能与遍历链表的线程产生死锁。这样的线程会试图首先锁定A或C（取决于遍历的方向），但是它接下来会发现无法获得结点B上的锁，因为正在进行删除操作的线程持有了结点B上的锁，并试图获得结点A和C上的锁。

在这里防止死锁的一个办法是定义遍历的顺序，让线程必须始终在锁定B之前锁定A、在锁定C之前锁定B。该方法以禁止反向遍历为代价来消除产生死锁的可能。对于其他数据结构，常常会建立类似的约定。

使用锁层次

虽然这实际上是定义锁定顺序的一个特例，但锁层次能够提供一种方法，来检查在运行时是否遵循了约定。其思路是将应用程序分层，并且确认所有能够在任意给定的层级上被锁定的互斥元。当代码试图锁定一个互斥元时，如果它在较低层已经持有锁定，那么就不允许它锁定该互斥元。通过给每一个互斥元分配层号，并记录下每个线程都锁定了哪些互斥元，你就可以在运行时进行检查了。清单3.7列出了两个线程使用层次互斥元的例子。

//清单3.7 使用锁层次来避免死锁
#include <mutex>

class hierarchical_mutex
{
public:
    explicit hierarchical_mutex(unsigned level)
    {}
    
    void lock()
    {}
    void unlock()
    {}
};


hierarchical_mutex high_level_mutex(10000); //❶
hierarchical_mutex low_level_mutex(5000); //❷

int do_low_level_stuff()
{
    return 42;
}


int low_level_func()
{
    std::lock_guard<hierarchical_mutex> lk(low_level_mutex); //❸
    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); //❹
    high_level_stuff(low_level_func()); //❺
}

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

hierarchical_mutex other_mutex(100); //❼
void do_other_stuff()
{}


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

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

int main()
{}

thread_a()❻遵守了规则，所以它运行良好。另一方面，thread_b()❾无视了规则，因此将在运行时失败。thread_a()调用high_level_func()，它锁定了high_level_mutex❹(具有层次值10000❶）并接着使用这个锁定了的互斥元调用low_level_func()❺，以获得high_level_stuff()的参数。low_level_func()接着锁定了low_level_mutex❽，但是没关系，因为该互斥元具有较低的层次值5000❷。

在另一方面thread_b()却不妥。刚开始，它锁定了other_mutex❿，它具有的层次值仅为100❿。这意味着它应该是保护着超低级别的数据。当other_stuff()调用high_level_func()❽时，就会违反层次。high_level_func()试图获取值为10000的high_level_mutex，大大超过100的当前层次值。因此，hierarchical_mutex可能通过引发异常或终止程序来报错。层次互斥元之间的死锁是不可能出现的，因为互斥元本身实行了锁定顺序。这还意味着如果两个锁在层次中处于相同级别，你就不能同时持有它们，因此逐节锁定的方案要求链条中的每个互斥元具有比前一个互斥元更低的层次值，在某些情况下这可能是不切实际的。

这个例子也展现了另外一点，带有用户定义的互斥元类型的std::lock_guard<>模板的使用。hierarchical_mutex不是标准的一部分，但易于编写。清单3.8中展示了一个简单的实现。即便它是个用户定义的类型，但是可以用于std::lock_guard<>，这是因为它实现了满足互斥元概念所需要的三个成员函数:lock()、unlock()和try_lock()。你还没有见过直接使用try_lock()，但它是相当简单的。如果互斥元上的锁已被另一个线程持有，则返回false，而非一直等到调用线程可以获取该互斥元上的锁。try_lock()也可以在std::lock()内部，作为避免死锁算法的一部分来使用。

//清单3.8 简单的分层次互斥元
#include <mutex>
#include <stdexcept>

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; //❶

    void check_for_hierarchy_violation()
    {
        if(this_thread_hierarchy_value <= hierarchy_value) //❷
        {
            throw std::logic_error("mutex hierarchy violated");
        }
    }
    void update_hierarchy_value()
    {
        previous_hierarchy_value=this_thread_hierarchy_value; //❸
        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(); //❹
        update_hierarchy_value(); //❺
    }
    void unlock()
    {
        this_thread_hierarchy_value=previous_hierarchy_value; //❻
        internal_mutex.unlock();
    }
    bool try_lock()
    {
        check_for_hierarchy_violation();
        if(!internal_mutex.try_lock()) //❼
            return false;
        update_hierarchy_value();
        return true;
    }
};
thread_local unsigned long
    hierarchical_mutex::this_thread_hierarchy_value(ULONG_MAX);       

int main()
{
    hierarchical_mutex m1(42);
    hierarchical_mutex m2(2000);
    
}

这里的关键是使用thread_local 的值来表示当前线程的层次值：this_thread_hierarchy_value❶。它被初始化为最大值❽，所以在刚开始的时候任意互斥元都可以被锁定。由于它被声明为thread_local，每个线程都有属于自己的副本，所以在一个线程中该变量的状态，完全独立于从另一个线程中读取的该变量状态。

因此，当线程第一次锁定hierarchical_mutex的实例时，this_thread_hierarchy_value 的值为ULONG_MAX。就其本质而言，ULONG_MAX比其他任意值都大，所以通过了check_for_hierarchy_violation()❷中的检查。在检查通过之后，lock ()代理内部的互斥元用以实际锁定❹。一旦该锁定成功，就可以更新层次值❺。

现在如果在持有第一个hierarchical_mutex上的锁的同时，锁定另一个hierarchical_mutex，则this_thread_hierarchy_value的值反映的是第一个互斥元的层次值。为了通过检查❷，第二个互斥元的层次值必须小于已经持有的互斥元的层次值。

现在，保存当前线程之前的层次值是很重要的，这样才能在unlock()中恢复它❻；否则，你就无法再次锁定一个具有更高层次值的互斥元，即便该线程并没有持有任何锁。因为只有当你持有internal_mutex时才能保存之前的层次值❸，并在解锁该内部互斥元之前释放它❻，你可以安全地将其存储在hierarchical_mutex自身中，因为它被内部互斥元上的锁安全地保护。

try_lock()和 lock()工作原理相同，只是，如果在internal_mutex上调用try_lock ()失败❼，那么你就无法拥有这个锁，所以不能更新层次值，并且返回false而不是true。

虽然检测是在运行时间检查，但它至少不依赖于时间——你不必去等待能够导致死锁出现的罕见情况发生。此外，需要以这种方式划分应用程序和互斥元的设计流程，可以在写入代码之前帮助消除许多可能导致死锁的原因。即使你还没有到达实际编写运行时间检测的那一步，进行设计练习仍然是值得的。

将这些设计准则扩展到锁之外

正如我在本节开始时提到的，死锁不只是出现于锁定中，它可以发生在任何可以导致循环等待的同步结构中。因此，扩展上面所述的准则来涵盖那些情况也是值得的。举个例子，正如你应该尽量避免获取嵌套锁那样，在持有锁时等待一个线程是坏主意，因为该线程可能需要获取这个锁以继续运行。类似地，如果你正要等待一个线程完成，指定线程层次结构可能也是值得的，这样线程就只需要等待低层次上的线程。一个简单的做到这一点的方法，就是确保你的线程在启动它们的同一个函数中被结合。

一旦你设计了代码来避免死锁，std::lock()和std::lock_guard涵盖了大多数简单锁定的情况，但有时却需要更大的灵活性。在那种情况下，标准库提供了std::unique_lock模板。与std::lock_guard类似，std::unique_lock 是在互斥元类型上进行参数化的类模板，并且它也提供了与std::lock_guard 相同的RAII 风格锁管理，但是更加灵活。

用std::unique_lock灵活锁定

通过松弛不变量，std::unique_lock 比std::lock_guard提供了更多的灵活性，一个std::unique_lock实例并不总是拥有与之相关联的互斥元。首先，就像你可以把 std::adopt_lock 作为第二参数传递给构造函数，以便让锁对象来管理互斥元上的锁那样，你也可以把 std::defer_lock 作为第二参数传递，来表示该互斥元在构造时应保持未被锁定。这个锁就可以在这之后通过在std::unique_lock对象（不是互斥元）上调用lock()，或是通过将std::unique_lock 对象本身传递给std::lock()来获取。使用std::unique_lock 和std::defer_lock❶，而不是
std::lock_guard和std::adopt_lock，能够很容易地将清单3.6写成清单3.9中所示的那样。这段代码具有相同的行数，并且本质上是等效的，除了一个小问题，std::unique_lock占用更多空间并且使用起来比std::lock_guard 略慢。允许std::unique_lock实例不拥有互斥元的灵活性是有代价的，这条信息必须被存储，并且必须被更新。

//清单3.9 在交换操作中使用std::lock()和std::unique_lock
#include <mutex>

class some_big_object
{};

void swap(some_big_object& lhs,some_big_object& rhs)
{}

class X
{
private:
    some_big_object some_detail;
    mutable 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); //❶std::defer_lock保留互斥元为未锁定
        std::unique_lock<std::mutex> lock_b(rhs.m,std::defer_lock);
        std::lock(lock_a,lock_b); //❷互斥元在这里被锁定
        swap(lhs.some_detail,rhs.some_detail);
    }
};

int main()
{}

在清单3.9中，std::unique_lock 对象能够被传递给std::lock()❷，因为std:: unique_lock提供了lock()、try_lock()和unlock()三个成员函数。它们会转发给底层互斥元上同名的成员函数去做实际的工作，并且只是更新在std::unique_lock实例内部的一个标识，来表示该实例当前是否拥有此互斥元。为了确保unlock()在析构函数中被正确调用，这个标识是必需的。如果该实例确已拥有此互斥元，则析构函数必须调用unlock()，并且，如果该实例并未拥有此互斥元，则析构函数绝不能调用unlock()。可以通过调用owns_lock()成员函数来查询这个标识。

如你所想，这个标识必须被存储在某个地方。因此，std::unique_lock 对象的大小通常大于std::lock_guard对象，并且相比于std::lock_guard，使用std::unique_lock的时候，会有些许性能损失，因为需要对标识进行相应的更新或检查。如果 std::lock_guard足以满足需求，我会建议优先使用它。也就是说，还有一些使用std::unique_lock更适合于手头任务的情况，因为你需要利用额外的灵活性。一个例子就是延迟锁定，正如你已经看到的；另一种情况是锁的所有权需要从一个作用域转移到另一个作用域。

在作用域之间转移锁的所有权

因为std::unique_lock实例并没有拥有与其相关的互斥元，所以通过四处移动(moving)实例，互斥元的所有权可以在实例之间进行转移。在某些情况下这种转移是自动的，比如从函数中返回一个实例，而在其他情况下，你必须通过调用std::move()来显式实现。从根本上说,这取决于源是否为左值(lvalue）——实变量或对实变量的引用——或者是右值(rvalue)———某种临时量。如果源为右值，则所有权转移是自动的，而对于左值，所有权转移必须显式地完成，以避免从变量中意外地转移了所有权。std:unique_lock就是可移动(movable)但不可复制(copyable)的类型的例子。

一种可能的用法，是允许函数锁定一个互斥元，并将此锁的所有权转移给调用者，于是调用者接下来可以在同一个锁的保护下执行额外的操作。下面的代码片段展示了这样的例子：函数get_lock()锁定了互斥元，然后在将锁返回给调用者之前准备数据。

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

因为lk是在函数内声明的自动变量，它可以被直接返回❶而无需调用std::move()，编译器负责调用移动构造函数，process_data()函数可以直接将所有权转移到它自己的std::unique_lock 实例❷，并且对do_something()的调用能够依赖被正确准备了的数据，而无需另一个线程在此期间去修改数据。
通常使用这种模式，是在待锁定的互斥元依赖于程序的当前状态，或者依赖于传递给返回std::unique_lock对象的函数的参数的地方。这种用法之一，就是并不直接返回锁，但是使用一个网关类的数据成员，以确保正确锁定了对受保护的数据的访问。这种情况下，所有对该数据的访问都通过这个网关类，当你想要访问数据时，就获取这个网关类的实例（通过调用类似于前面例子中的get_lock()函数)，它会获取锁。然后，你可以通过网关对象的成员函数来访问数据。在完成后，销毁网关对象，从而释放锁，并允许其他线程访问受保护的数据。这样的网关对象很可能是可移动的（因此它可以从函数返回)，在这种情况下，锁对象的数据成员也需要是可移动的。

std::unique_lock的灵活性同样允许实例在被销毁之前撤回它们的锁。你可以使用unlock()成员函数来实现，就像对于互斥元那样，std::unique_lock支持与互斥元一样的用来锁定和解锁的基本成员函数集合，这是为了让它可以用于通用函数，比如std::lock。在std::unique_lock实例被销毁之前释放锁的能力，意味着你可以有选择地在特定的代码分支释放锁，如果很显然不再需要这个锁，这对于应用程序的性能可能很重要。持有锁的时间比所需时间更长，会导致性能下降，因为其他等待该锁的线程，被阻止运行超过了所需的时间。

锁定在恰当的粒度

锁粒度是我在之前曾提到过的，锁粒度是一个文字术语，用来描述由单个锁所保护的数据量。细粒度锁保护着少量的数据，粗粒度锁保护着的大量的数据。选择一个足够粗的锁粒度，来确保所需的数据都被保护是很重要的，不仅如此，同样重要的是，确保只在真正需要锁的操作中持有锁。我们都知道，带着满满一车杂货在超市排队结账，只因为正在结账的人突然意识到自己忘了一些小红莓酱，然后就跑去找，而让大家都等着，或者收银员已经准备好收钱，顾客才开始在自己的手提包里翻找钱包，是很令人抓狂的。如果每个人去结账时都拿到了他们想要的，并准备好了适当的支付方式，一切都更容易进行。

这同样适用于线程，如果多个线程正等待着同一个资源（收银台的收银员），然后，如果任意线程持有锁的时间比所需时间长，就会增加等待所花费的总时间（不要等到你已经到了收银台才开始寻找小红莓酱）。如果可能，仅在实际访问共享数据的时候锁定互斥元，尝试在锁的外面做任意的数据处理。特别地，在持有锁时，不要做任何确实很耗时的活动，比如文件 IO。文件 IO通常比从内存中读取或写人相同大小的数据量要慢上数百倍（如果不是数千倍）。因此，除非这个锁是真的想保护对文件的访问，否则在持有锁时进行IO会不必要地延迟其他线程（因为它们在等待获取锁时会阻塞），潜在地消除了使用多线程带来的性能提升。

std::unique_lock 在这种情况下运作良好，因为能够在代码不再需要访问共享数据时调用unlock()，然后在代码中又需要访问时再次调用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(); //❶在对process()的调用中不需要锁定互斥元
     result_type result = process(data_to_process);
     my_lock.unlock(); //❷重新锁定互斥元以回写结果
     write_result(data_to_process.result);
}

在调用process()过程中不需要锁定互斥元，所以手动地将其在调用前解锁❶，并在之后再次锁定❷。

希望这是显而易见的，如果你让一个互斥锁保护整个数据结构，不仅可能会有更多的对锁的竞争，锁被持有的时间也可能会减少。更多的操作步骤会需要在同一个互斥元上的锁，所以锁必须被持有更长的时间。这种成本上的双重打击，也是尽可能走向细粒度锁定的双重激励。

如这个例子所示，锁定在恰当的粒度不仅关乎锁定的数据量；这也是关系到锁会被持有多长时间，以及在持有锁时执行哪些操作。一般情况下，只应该以执行要求的操作所需的最小可能时间而去持有锁。这也意味着耗时的操作，比如获取另一个锁（即便你知道它不会死锁）或是等待I/O完成，都不应该在持有锁的时候去做，除非绝对必要。

在清单3.6和清单3.9中，需要锁定两个互斥锁的操作是交换操作，这显然需要并发访问两个对象。假设取而代之，你试图去比较仅为普通int的简单数据成员。这会有区别吗？int可以轻易被复制，所以你可以很容易地为每个待比较的的对象复制其数据，同时只用持有该对象的锁，然后比较已复制数值。这意味着你在每个互斥元上持有锁的时间最短，并且你也没有在持有一个锁的时候去锁定另外一个。清单3.10展示了这样的一个类Y，以及相等比较运算符的示例实现。

//清单3.10 在比较运算符中每次锁定一个互斥元
#include <mutex>
class Y
{
private:
    int some_detail;
    mutable std::mutex m;

    int get_detail() const
    {
        std::lock_guard<std::mutex> lock_a(m); //❶
        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(); //❷
        int const rhs_value=rhs.get_detail(); //❸
        return lhs_value==rhs_value; //❹
    }
};

int main()
{}

在这种情况下，比较运算符首先通过调用 get_detail()成员函数获取要进行比较的值❷、❸。此函数在获取值的同时用一个锁来保护它❷。比较运算符接着比较获取到的值❹。但是请注意，这同样会减少锁定的时间，而且每次只持有一个锁（从而消除了死锁的可能性），与同时持有两个锁相比，这巧妙地改变了操作的语义。在清单3.10中，如果运算符返回 true，意味着rhs.some_detail在一个时间点的值与rhs.some_detail 在另一个时间点的值相等。这两个值能够在两次读取之中以任何方式改变。例如，这两个值可能在0和R之间进行了交换，从而使这个比较变得毫无意义。这个相等比较可能会返回true来表示值是相等的，即使这两个值在某个瞬间从未真正地相等过。因此，当进行这样的改变时小心注意是很重要的，操作的语义不能以有问题的方式而被改变：如果你不能在操作的整个持续时间中持有所需的锁，你就把自己暴露在竞争条件中。

有时，根本就没有一个合适的粒度级别，因为并非所有的对数据结构的访问都要求同样级别的保护。在这种情况下，使用替代机制来代替普通的std::mutex可能才是恰当的。