C++并发：在线程间共享数据

1 线程间共享数据的问题

1.1 条件竞争

条件竞争：在并发编程中：操作由两个或多个线程负责，它们争先让线程执行各自的操作，而结果取决于它们执行的相对次序，这样的情况就是条件竞争。

诱发恶性条件竞争的典型场景是，要完成一项操作，却需要改动两份或多份不同的数据，而它们只能用单独的指令改动，当其中的一份数据完成改动时，别的线程有可能不期而访。并且由于这样的场景出现的时间窗口小，因此一般很难复现场景定位。

1.2 防止恶性条件竞争

有如下方法：

1 采取保护措施包装数据结构，确保中间状态只对执行改动的线程可见。

2 修改设计，由一连串不可拆分的改动完成数据变更，每个改动都维持不变量不被破坏。这通常称为无锁编程，难以正确编写。如果从事这一层面的开发，就要探究内存模型的细节，以及区分每个线程能够看到什么数据集。

3 修改数据结构来当作事务处理。

2 用互斥保护共享数据

访问一个数据结构前，先锁住与数据相关的互斥，访问结束后再解锁互斥。C++线程库保证了，一旦有线程锁住了某个互斥，若其他线程试图再给他加锁，需要等待。

互斥也可能带来某些问题，比如死锁，对数据的过保护和欠保护。

2.1 std::mutex

C++中使用std::mutex的实例来构造互斥。

可以通过成员函数lock()对其加锁，unlock()进行解锁。但是并不推荐直接调用成员函数，原因是这样需要记住在函数以外的每条代码路径都要调用unlock()，包括异常退出的路径。

取而代之，C++便准库提供了模板std::lock_guard<>，针对互斥类融合实现了RAII：在构造时加锁，在析构时解锁，从而保证互斥总被正确解锁。

#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();
}

C++17支持了模板参数推导，使得上述实现可以写成如下样式。并且引入了std::scoped_lock，他是增强版的lock_guard

std::lock_guard guard(some_mutex);std::scoped_guard guard(some_mutex);

2.2 指针和引用打破互斥保护

如果成员函数返回指针或引用，指向受保护的数据，那么即便成员函数全部按良好、有序的方式锁定互斥，仍然会无济于事。

只要存在任何能访问该指针和引用的代码，它就可以访问受保护的共享数据，而无需锁定互斥。因此，利用互斥保护共享数据，需要谨慎设计程序接口，从而保证互斥已先行锁定，再对受保护的共享数据进行访问。

2.3 组织和编排代码以保护共享数据

我们除了要防止成员函数向调用者传出指针或者引用，还要注意成员函数内部调用的别的函数，也不要向这些函数传递指针或者引用。

#include <mutex>
#include <string>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();
}

比如上述代码，malicious_function方法将被互斥锁保护的data_wrapper中的some_data的引用赋值给外面的unprotected，导致互斥保护被打破，在外面可直接通过unprotected进行操作。

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

#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&&);template <class... Args> void emplace(Args&&... args);
};

上述实现会导致条件竞争，也就是empty和size的结果不可信，因为在函数返回后，其他线程不再受限，可能马上会有新元素入栈或者出栈。

线程1	线程2
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);

这样，当一个栈只有一个元素的时候，第二个pop的线程会导致未定义行为。

并且，当我们复制vector时，如果vector中的元素数量巨大，可能导致因为资源不足造成的内存分配失败。pop函数的定义是，返回栈顶元素的值，并且将其从栈顶移除。因此，只有在栈被改动之后，弹出的元素才返回给调用者，然而在向调用者复制数据的过程中，有可能抛出异常。万一弹出的元素已经从栈上移除，但是复制不成功，就会造成数据丢失。

2.4.1 消除竞争

2.4.1.1 传入引用

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

优点：pop的元素在外部容器白村了生命周期

缺点：如果要调用pop，还要先闯将一个别的容器。

2.4.1.2 提供不抛出异常的拷贝构造函数，或不抛出异常的移动构造函数

这样虽然安全，但是效果并不理想。栈容器的用途会受限。

2.4.1.3 返回指针，指向弹出元素

优点：指针可以自由的复制，不会抛出异常。

缺点：指向的对象仍然在内存中，需要额外的内存管理，可以使用shared_ptr。

2.4.1.4 结合1，2或者1，3

2.4.1.5 线程安全的栈容器

#include <exception>
#include <memory>
#include <mutex>
#include <stack>struct empty_stack: std::exception {const char* what() const throw();
};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(std::move(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.pop();data.pop();}bool empty() const {std::lock_guard<std::mutex> lock(m);return data.empty();}
};

2.5 死锁：问题和解决方法

防范死锁的建议通常是，始终按照相同的顺序对两个互斥加锁。

C++标准提供了std::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;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);}
};

std::adopt_lock对象指明了互斥已被锁住，即互斥上有锁存在。std::lock_guard实例据此接收锁的归属权，不会在构造函数内试图另行加锁。

无论是正常返回还是异常退出，std::lock_guard都保证了互斥全都正确解锁。

另外，lock()对lhs.m或rhs.m进行加锁，这一函数调用可能导致抛出异常。

C++17还提供了全新的特性std::scoped_lock<>。它和std::lock_guard<>完全等价。只不过前者是可变参数模板，接收各种互斥型别作为模板参数列表，还能以多个互斥对象作为构造函数的参数列表。

void swap(X& lhs, X& rhs)
{if (&lhs==&rhs)return;std::scoped_lock guard(lhs.m, rhs.m);swap(lhs.some_detail, rhs.some_detail);
}

使用新特性实现如上，并且上述代码还是用了类模板参数推导（C++17）。使用std::scoped_lock将lock和lock_guard合并为一句，降低出错概率。

2.6 防范死锁的补充准则

即使没有牵涉锁，也会发生死锁现象。假定有两个线程，各自关联了std::thread实例，若同时在对方的std::thread实例上调用join，那么就能制造出死锁现象。

防范死锁的最终准则：只要另一个线程有可能正在等待当前线程，那么当前线程不要反过来等待他。

2.6.1 避免嵌套锁

假如已经持有锁，就不要试图获取第二个锁。这样保证每个线程最多只持有一个锁，仅锁的使用本身不可能导致锁。

但是还存在其他可能引起死锁的场景（比如多个线程彼此等待），操作多个互斥锁很可能是最常见的死锁诱因。如果真的需要获取多个锁，应使用lock函数，单独的调用动作一次获取全部锁来避免死锁。

2.6.2 一旦持锁，就须避免调用由用户提供的程序接口

若程序接口由用户自行实现，则我们无从得知它到底会做什么，可能会试图获取锁。这样便可能违反避免嵌套锁的准则，可能发生死锁。

不过有时候这个情况难以避免，因此在需要调用用户提供的程序接口时，要遵守2.6.3准则。

2.6.3 依从固定顺序获取锁

如果多个锁是绝对必要的，却无法通过std::lock()在一步操作内全部获取，我们只能退而求其次，在每个线程内部依从固定顺序获取这些锁。

也可以同时给这些互斥加锁。

或者对于双向链表来说，规定遍历的方向，让线程总是必须先锁住A，再锁住B，也可以防范死锁。

2.6.4 按层级加锁

锁的层级划分就是按照特定的方式规定加锁次序，在运行期据此查验加锁操作是否遵从预设规则。若某个线程已对低层级互斥加锁，则不准它再对高层级互斥加锁。不过这种模式C++标准库尚未提供支持，自行实现如下：

#include <limits.h>
#include <mutex>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() {if (this_thread_hierarchy_value!=hierarchy_value) {throw std::logic_error("mutex hierarchy violated");}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);hierarchical_mutex high_level_mutex(10000);
hierarchical_mutex low_level_mutex(5000);
hierarchical_mutex other_mutex(6000);int do_low_level_stuff();
int low_level_func() {std::lock_guard<hierarchical_mutex> lk(low_level_mutex);return do_low_level_stuff();
}int high_level_stuff(int some_param);
int high_level_func() {std::lock_guard<hierarchical_mutex> lk(low_level_mutex);high_level_stuff(low_level_func());
}void thread_a() {high_level_func();
}int 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();
}

2.6.5 将准则推广到锁操作之外

如果要等待线程，那就值得针对线程规定层级，使得每个线程仅等待层级更低的线程。

有一种简单的方法可以实现这种机制：让同一个函数启动全部线程，且汇合工作也由之负责。

2.7 std::unique_lock<>

它与std::lock_guard<>一样，也是一个依据互斥作为参数的类模板，并且使用RAII手法管理锁。

std::unique_lock<>放宽了不变量的成立条件，因此更灵活一些。std::unique_lock对象不一定始终占有与之关联的互斥。

其构造函数接收两个参数：互斥锁和lock实例。

可以传入std::adopt_lock实例，借此指明std::unique_lock对象管理互斥上的锁。

也可以传入std::defer_lock实例，从而使互斥再完成构造时处于无锁状态，等以后有需要时才在std::unique_lock对象（不是互斥对象）上调用lock()而获取锁，或者把std::unique_lock对象交给std::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;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::unique_lock<std::mutex> lock_b(rhs.m, std::defer_lock);std::lock(lock_a, lock_b);swap(lhs.some_detail, rhs.some_detail);}
};

因为std::unique_lock类具有成员lock()，try_lock()，unlock()，所以它的实例可以传给lock()函数。std::unique_lock底层与目标互斥关联，此互斥也有这三个同名函数，因此上述函数调用转由它们执行。

std::unique_lock实例还有一个内部标志，随着函数的执行而更新，表明关联的互斥目前是否正被该类的实例占据。这个标志可以通过owns_lock()查询。

不过，最好还是用C++17提供的变参模板类std::scoped_lock，除非必须用std::unique_lock类进行某些操作，如转移锁的归属权。

2.8 在不同作用域之间转移互斥归属权

转移有一种用途：准许函数锁定互斥，然后把互斥的归属权转移给函数调用者，好让他在同一个锁的保护下执行其他操作。

std::unique_lock实例可以在被销毁前解锁，这意味着，在执行流程的任意分支上，若某个锁没必要继续持有，则可解锁。这对应用程序的性能来说很重要。

2.9 按适合的粒度加锁

持锁期间应避免任何耗时的操作，如：I/O操作加锁将毫无必要地阻塞其他线程（文件操作通常比内存慢几百上千倍）

这种情况可以用std::unique_lock处理：假设代码不再需要访问共享数据，那就调用unlock()解锁，若需要重新访问，就调用lock()加锁。

3 保护共享数据的其他工具

3.1 仅在初始化过程中保护共享数据

#include <memory>
#include <mutex>
std::shared_ptr<some_resource> resource_ptr;
std::mutex resource_mutex;
void foo() {std::unique_lock<std::mutex> lk(resource_mutex);if (!resource_ptr) {resource_ptr.reset(new some_resource);}lk.unlock();resource_ptr->do_something();
}

上述代码迫使多个线程循序运行，问题较大。为此，其中一个备受诟病的改进就是双重检验锁定模式。

3.1.1 双重检验锁定模式

void undefined_behaviour_with_double_checked_locking() {if (!resource_ptr) {std::lock_guard<std::mutex> lk(resource_mutex);if (!resource_ptr) {resource_ptr.reset(new some_resource);}}resource_ptr->do_something();
}

这种模式的思路如下：

在无锁的条件下读取指针，如果为空，获取锁。

为了避免在一个线程进入第一个判断后，其他线程已经获取锁并且已经为resource_ptr赋值，因此在获取锁和赋值之间加入第二个判断，保证不会重复初始化，让resource_ptr被赋值两次。

但是这种思路可能导致恶性竞争：

在第一个判断，一个线程想要读取resource_ptr的值的时候，另一个线程可能已经获取锁并且正在为resource_ptr执行写操作。但是这个时候可能new some_resource还未生效被无视，导致前一个线程虽然能在第一个判断时了解到resource_ptr不为空，但是他后面又会走到resource_ptr->do_something()，这个时候使用了一个中间态的数据执行do_somthing()。就产生了读写操作的不同步。也就是数据竞争，是条件竞争的一种。

3.1.2 std::once_flag和std::call_once函数

用来专门处理上述数据竞争的情况：

令所有线程共同调用std::call_once()函数，确保在该调用返回时，指针初始化由其中某一线程安全且唯一地完成（通过合适的同步机制）。

必要的同步数据由std::once_flag实例存储，每个std::once_flag实例对应一次不同的初始化。

#include <memory>
#include <mutex>
std::shared_ptr<some_resource> resource_ptr;
std::once_flag resource_flag;
void init_resource() {resource_ptr.reset(new some_resource);
}void foo() {std::call_once(resource_flag, init_resource);resource_ptr->do_something();
}

上述代码中的call_once函数包含两个对象，需要初始化数据（some_resource）和once_flag对象，两者的作用域都完整涵盖了它们所属的名字空间。

3.1.2.2 对于类的数据成员的初始化

#include <mutex>class X {
private:connection_info connection_details;connection_handle connection;std::once_flag connection_init_flag;void open_connection() {connection = connection_manager.open(connection_details);}
public:X(connection_info const& connection_details_) :connection_details(connection_details_) {}void send_data(data_packet const& data) {std::call_once(connection_init_flag, &X::open_connection, this);connection.send_data(data);}data_packet receive_data() {std::call_once(connection_init_flag, &X::open_connection, this);return connection.receive_data();}
};

上述代码中的初始化在send或者receive中进行，由于这个时候传入call_once的是类成员，因此需要传入this指针作为附加参数。

std::once_flag不可复制也不可移动，这点与std::mutex类似。

3.1.2.3 使用静态变量代替成员变量进行初始化

class my_class;my_class& get_my_class_instance()
{static my_class instance;return instance;
}

把局部变量声明成静态数据，在C++11之后，规定静态数据初始化只会在某一线程单独发生，不会出现多个线程都认为自己应当为其赋值的情况，在初始化完成前，其他线程不会越过静态数据的声明而运行。某些类的代码只需要用到唯一一个全局实例，这种情况下可以用静态成员代替std::call_once。来让多个线程可以安全地调用上述方法。

3.2 保护很少更新的数据结构：std::shared_mutex和std::shared_timed_mutex

C++17标准库提供了两种新的互斥：std::shared_mutex和std::shared_timed_mutex。

C++14标准库只有std::shared_timed_mutex。

C++11标准库都没有。可以使用Boost库。

std::shared_mutex相较于std::shared_timed_mutex，后者支持更多的操作，前者在某些平台上可能会到来额外性能收益。

利用std::shared_mutex实施同步操作：

更新操作：使用std::lock_guard<std::shared_mutex>或者std::unique_lock<std::shared_mutex>锁定代替std::mutex.

共享锁：std::shared_lock<std::shared_mutex>，实现共享访问。C++14引入，工作原理是RAII过程。假设它被某些线程持有，要等线程全部释放共享锁，其他线程才能访问排他锁。如果任一线程持有排他锁，其他线程无法获取共享锁以及排他锁，直至排他锁被释放。

#include <map>
#include <string>
#include <mutex>
#include <shared_mutex>class dnc_entry;
class dns_cache {std::map<std::string, dns_entry> entries;mutable std::shared_mutex entry_mutex;
public:dnc_entry find_entry(std::string const& domain) const {std::shared_lock<std::shared_mutex> lk(entry_mutex);std::map<std::string, dns_entry>::const_iterator const it = entries.find(domain);return (it == entries.end()) ? dns_entry() : it->second;}void update_or_add_entry(std::string const& domain, dns_entry const& dns_details) {std::lock_guard<std::shared_mutex> lk(entry_mutex);entries[domain] = dns_details;}
};

3.3 递归加锁std::recursive_mutex

某些场景需要让线程在同一互斥上多次加锁，而无需解锁。为此提供了std::recursive_mutex。

其允许同一线程对某互斥多次加锁，必须先释放全部锁，才能让另一个线程获取锁。

也是通过std::lock_guard<std::recursive_mutex>或者std::unique_lock<std::recursive_mutex>

递归互斥常常用于这样的情形：

每个公有函数都需要先锁住互斥，才进行操作，但是当共有函数调用共有函数时，使用std::mutex就会有问题，因此这个时候使用递归互斥。但是一般不建议这样做，因为这可能意味着设计有问题。

4 小结

4.1 几种互斥锁

std::mutex
std::shared_mutex	假设它被某些线程持有，要等线程全部释放共享锁，其他线程才能访问排他锁。如果任一线程持有排他锁，其他线程无法获取共享锁以及排他锁，直至排他锁被释放，性能更优
std::shared_timed_mutex	假设它被某些线程持有，要等线程全部释放共享锁，其他线程才能访问排他锁。如果任一线程持有排他锁，其他线程无法获取共享锁以及排他锁，直至排他锁被释放，支持操作更多
std::recursive_mutex	其允许同一线程对某互斥多次加锁，必须先释放全部锁，才能让另一个线程获取锁

4.2 几种加锁实例

std::lock_guard	依据互斥作为参数的类模板，并且使用RAII手法管理锁
std::scoped_lock<>	它和std::lock_guard<>完全等价。只不过前者是可变参数模板，接收各种互斥型别作为模板参数列表，同时锁住多个
std::unique_lock<>	它与std::lock_guard<>一样，除了std::unique_lock对象不一定始终占有与之关联的互斥

4.3 其他

std::adopt_lock	传入std::adopt_lock实例，指明std::unique_lock对象管理互斥上的锁。
std::defer_lock	传入std::defer_lock实例，从而使互斥再完成构造时处于无锁状态，等以后有需要时才在std::unique_lock对象（不是互斥对象）上调用lock()而获取锁，或者把std::unique_lock对象交给std::lock()函数加锁。
std::call_once()	令所有线程共同调用std::call_once()函数，确保在该调用返回时，指针初始化由其中某一线程安全且唯一地完成（通过合适的同步机制）。
std::once_flag	必要的同步数据由std::once_flag实例存储，每个std::once_flag实例对应一次不同的初始化。