C++并发：设计基于锁的并发数据结构

（上一篇主要讲了底层逻辑，比较晦涩难懂，这一章讲解数据结构设计）

对于并发场景，数据保护的形式主要如下：

1 采用独立互斥和外部锁

2 专门为并发访问自行设计的数据结构

1 并发设计的内涵

互斥保护数据结构的方式是明令阻止真正的并发访问。以上行为称为串行化：每个线程轮流访问受互斥保护的数据，他们只能先后串行依次访问，而非并发访问。

因此在设计数据结构时必须深思熟虑，力求实现真正的并发访问。

相对而言，某些数据结构更有潜质支持真正的并发访问，但万变不离其宗：保护的范围越小，需要的串行化操作就越少，并发程度就越高。

1.1 设计并发数据结构的指引

在设计支持并发访问的数据结构时，需要考虑两点：

1 确保访问安全，通常，构造函数和析构函数都要按排他的方式执行。在构造前和析构后不会有访问发生。赋值，内部数据互换，拷贝构造等操作与其他操作一起并发调用是否安全，以及这些函数是否要求使用者保证以排他的方式访问。

2 实现真正的并发访问，可以考虑一下问题：

能否限制锁的作用域，从而让操作的某些部分在锁的保护意外执行

数据结构内部的不同部分能否采用不同的互斥

是否所有操作都需要相同程度的保护

能否通过简单地改动，提高数据结构的并发程度，为并发操作增加机会，而不影响操作语义。

这些问题全部都归结为一个核心问题：如何才能只保留最必要的串行操作，最大限度地实现并发。

2 基于锁的并发数据结构（注意使用多个互斥容易死锁）

2.1 采用锁实现线程安全的栈容器

#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>(std::move(data.top())));data.pop();return res;}void pop(T& value) {std::lock_guard<std::mutex> lock(m);if (data.empty()) {throw empty_stack();}value = std::move(data.top());data.pop();}bool empty() const {std::lock_guard<std::mutex> lock(m);return data.empty();}
};

2.2 采用锁和条件变量实现线程安全的队列容器

template<typename T>
class threadsafe_queue {
private:mutable std::mutex mut;std::queue<T> data_queue;std::condition_variable data_cond;public:threadsafe_queue() {}void push(T new_value) {std::lock_guard<std::mutex> lk(mut);data_queue.push(std::move(new_value));data_cond.notify_one();}void wait_and_pop(T& value) {std::unique_lock<std::mutex> lk(mut);data_cond.wait(lk, [this]{return !data_queue.empty();});value = std::move(data_queue.front());data_queue.pop();}std::shared_ptr<T> wait_and_pop() {std::unique_lock<std::mutex> lk(mut);data_cond.wait(lk, [this]{return !data_queue.empty();});std::shared_ptr<T> res(std::make_shared<T>(std::move(data_queue.front())));data_queue.pop();return res;}bool try_pop(T& value) {std::lock_guard<std::mutex> lk(mut);if (data_queue.empty()) {return false;}value = std::move(data_queue.front());data_queue.pop();return true;}std::shared_ptr<T> try_pop() {std::lock_guard<std::mutex> lk(mut);if (data_queue.empty()) {return std::shared_ptr<T>();}std::shared_ptr<T> res(std::make_shared<T>(std::move(data_queue.front())));data_queue.pop();return res;}bool empty() const {std::lock_guard<std::mutex> lk(mut);return data_queue.empty();}
};

与2.1的不同之处在于对于notify_one()的使用。支持了等待-通知功能。弹出的功能不再需要连续地对empty进行判断，只需要使用条件变量进行wait，并且等待期间由于互斥已经加锁，因此数据仍然受到保护（使得pop和empty等操作不会有数据竞争）。

2.2.1 使用std::shared_ptr避免唤醒线程时抛出异常导致无线程唤醒

为了避免之前的代码在wait_and_pop时抛出异常（比如新指针shared_ptr在构建时抛出了异常），导致不会有任何线程被唤醒，因此这里选择了一个处理方法：使用std::shared_ptr<>，将其初始化语句移到push()的调用处，令队列容器改为存储std::shared_ptr<>，而不再直接存储数据的值。这样从内部std::queue<>复制std::shared_ptr实例的操作不会抛出异常，所以此时wait_and_pop是异常安全的。

#include <condition_variable>
#include <mutex>
#include <queue>template<typename T>
class threadsafe_queue {
private:mutable std::mutex mut;std::queue<std::shared_ptr<T>> data_queue;std::condition_variable data_cond;public:threadsafe_queue() {}void wait_and_pop(T& value) {std::unique_lock<std::mutex> lk(mut);data_cond.wait(lk, [this]{return !data_queue.empty();});value = std::move(*data_queue.front());data_queue.pop();}bool try_pop(T& value) {std::lock_guard<std::mutex> lk(mut);if (data_queue.empty()) {return false;}value = std::move(*data_queue.front());data_queue.pop();return true;}std::shared_ptr<T> wait_and_pop() {std::unique_lock<std::mutex> lk(mut);data_cond.wait(lk, [this]{return !data_queue.empty();});std::shared_ptr<T> res = data_queue.front();data_queue.pop();return res;}std::shared_ptr<T> try_pop() {std::lock_guard<std::mutex> lk(mut);if (data_queue.empty()) {return std::shared_ptr<T>();}std::shared_ptr<T> res = data_queue.front();data_queue.pop();return res;}void push(T new_value) {std::shared_ptr<T> data(std::make_shared<T>(std::move(new_value)));std::lock_guard<std::mutex> lk(mut);data_queue.push(data);data_cond.notify_one();}bool empty() const {std::lock_guard<std::mutex> lk(mut);return data_queue.empty();}
};

将创建shared_ptr需要的内存分配的步骤从锁保护的范围移出，还有一个好处就是，内存分配是一个开销很大的操作，将他放在所保护的范围之外，减少了锁持有的时长。

缺点是，都是由唯一的互斥保护整个数据结构，它所支持的并发程度因此受限。事实上首先的原因是每次只允许一个线程操作队列数据。我们对于std::queue这一个整体施加了保护。

2.3 采用精细粒度的锁和条件变量实现线程安全的队列容器

在队列和栈的实现代码中，我们以queue为最小内部结构进行了数据保护，但是这样的粒度其实是较大的，在这里我们使用单向链表来实现队列，减小粒度。

#include <memory>template<typename T>
class queue {
private:struct node {std::shared_ptr<T> data;std::unique_ptr<node> next;};std::unique_ptr<node> head;node* tail;public:queue() : head(new node), tail(head.get()) {}queue(const queue& other) = delete;queue& operator=(const queue& other) = delete;std::shared_ptr<T> try_pop() {if (head.get()==tail) {return std::shared_ptr<T>();}std::shared_ptr<T> const res(head->data);std::unique_ptr<node> const old_head = std::move(head);head = std::move(old_head->next);return res;}void push(T new_value) {std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));std::unique_ptr<node> p(new node);tail->data = new_data;node* const new_tail = p.get();tail->next = std::move(p);tail = new_tail;}
};

head是头节点，不保有业务数据，保证了在push的时候不会出现head和tail指向同一业务节点造成的数据冲突。也就是使用了虚位节点（head对应的节点），导致head和tail不再同时指向相同的节点（pop处理head，push处理tail），因此为head和tail各用一个互斥提供了可能。

下面提供了精细粒度的线程安全队列：

#include <memory>
#include <mutex>template<typename T>
class threadsafe_queue {
private:struct node {std::shared_ptr<T> data;std::unique_ptr<node> next;};std::mutex head_mutex;std::unique_ptr<node> head;std::mutex tail_mutex;node* tail;node* get_tail() {std::lock_guard<std::mutex> tail_lock(tail_mutex);return tail;}std::unique_ptr<node> pop_head() {std::lock_guard<std::mutex> head_lock(head_mutex);if (head.get() == get_tail()) {return nullptr;}std::unique_ptr<node> old_head = std::move(head);head = std::move(old_head->next);return old_head;}public:threadsafe_queue() : head(new node), tail(head.get()) {}threadsafe_queue(const threadsafe_queue& other) = delete;threadsafe_queue& operator=(const threadsafe_queue& other) = delete;std::shared_ptr<T> try_pop() {std::unique_ptr<node> old_head = pop_head();return old_head ? old_head->data : std::shared_ptr<T>();}void push(T new_value) {std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));std::unique_ptr<node> p(new node);node* const new_tail = p.get();std::lock_guard<std::mutex> tail_lock(tail_mutex);tail->data = new_data;tail->next = std::move(p);tail = new_tail;}
};

虽然保证了head和tail不会指向相同的节点，但是对于push和try_pop操作，它们仍然能对相同的节点进行操作。因此这里的实现在try_pop上也给tail互斥加上了锁，来保证不同线程执行push和try_pop操作，能够有确定的先后顺序。

并且tail的互斥锁在head互斥之内，保证了不会有死锁的发生，以及tail和head的指向均为正确的尾节点，头节点。

这样实现的方法将很多内存分配步骤移到了锁外，让上锁的范围更精细。

2.3.1 等待数据弹出

#include <condition_variable>
#include <memory>
#include <mutex>template<typename T>
class threadsafe_queue {
private:struct node {std::shared_ptr<T> data;std::unique_ptr<node> next;};std::mutex head_mutex;std::unique_ptr<node> head;std::mutex tail_mutex;node* tail;std::condition_variable data_cond;node* get_tail() {std::lock_guard<std::mutex> tail_lock(tail_mutex);return tail;}std::unique_ptr<node> pop_head() {std::unique_ptr<node> old_head = std::move(head);head = std::move(old_head->next);return old_head;}std::unique_lock<std::mutex> wait_for_data() {std::unique_lock<std::mutex> head_lock(head_mutex);data_cond.wait(head_lock, [&]{return head.get() != get_tail();});return std::move(head_lock);}std::unique_ptr<node> wait_pop_head() {std::unique_lock<std::mutex> head_lock(wait_for_data());return pop_head();}std::unique_ptr<node> wait_pop_head(T& value) {std::unique_lock<std::mutex> head_lock(wait_for_data());value = std::move(*head->data);return pop_head();}std::unique_ptr<node> try_pop_head() {std::lock_guard<std::mutex> head_lock(head_mutex);if (head.get() == get_tail()) {return std::unique_ptr<node>();}return pop_head();}std::unique_ptr<node> try_pop_head(T& value) {std::lock_guard<std::mutex> head_lock(head_mutex);if (head.get() == get_tail()) {return std::unique_ptr<node>();}value = std::move(*head->data);return pop_head();}public:std::shared_ptr<T> wait_and_pop() {std::unique_ptr<node> const old_head = wait_pop_head();return old_head->data;}void wait_and_pop(T& value) {std::unique_ptr<node> const old_head = wait_pop_head(value);}std::shared_ptr<T> try_pop() {std::unique_ptr<node> old_head = try_pop_head();return old_head ? old_head->data : std::shared_ptr<T>();}bool try_pop(T& value) {std::unique_ptr<node> const old_head = try_pop_head(value);return old_head != nullptr;}bool empty() {std::lock_guard<std::mutex> head_lock(head_mutex);return head.get() == get_tail();}void push(T new_value) {std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));std::unique_ptr<node> p(new node);{node* const new_tail = p.get();std::lock_guard<std::mutex> tail_lock(tail_mutex);tail->data = new_data;tail->next = std::move(p);tail = new_tail;}data_cond.notify_one();}
};

这里加上了基于条件变量实现的等待和通知，避免了外部循环调用等待互斥。

3 设计更复杂的基于锁的并发数据结构

查找表（lookup table），又称字典，也是cpp里面的各种map，从并发的角度来讲，std::map<>接口中最大问题是迭代器。有一种思路是通过迭代器访问容器内部，即便有其他线程访问并改动容器，迭代器所提供的访问依然安全。

但是如果这个时候一个线程要删除某个元素，而它却正被迭代器引用，就容易触发问题。所以本章先将迭代器从线程安全的map接口中剔除。

3.1 采用锁编写线程安全的查找表

#include <bitset>
#include <list>
#include <shared_mutex>
#include <vector>template<typename Key, typename Value, typename Hash=std::hash<Key>>
class threadsafe_lookup_table {
private:class bucket_type {private:typedef std::pair<Key, Value> bucket_value;typedef std::list<bucket_value> bucket_iterator;bucket_data data;mutable std::shared_mutex mutex;bucket_iterator find_entry_for(Key const& key) const {return std::find_if(data.begin(), data.end(), [&](bucket_value const& item){return item.first == key;});}public:Value value_for(Key const& key, Value const& default_value) const {std::shared_lock<std::shared_mutex> lock(mutex);bucket_iterator const found_entry = find_entry_for(key);return (found_entry == data.end()) ? default_value : found_entry->second;}void add_or_update_mapping(Key const& key, Value const& value) {std::unique_lock<std::shared_mutex> lock(mutex);bucket_iterator const found_entry = find_entry_for(key);if (found_entry == data.end()) {data.push_back(bucket_value(key, value));} else {found_entry->second = value;}}void remove_mapping(Key const& key) {std::unique_lock<std::shared_mutex> lock(mutex);bucket_iterator const found_entry = find_entry_for(key);if (found_entry != data.end()) {data.erase(found_entry);}}};std::vector<std::unique_ptr<bucket_type>> buckets;Hash hasher;bucket_type& get_bucket(Key const& key) const {std::size_t const bucket_index = hasher(key) % buckets.size();return *buckets[bucket_index];}
public:typedef Key key_type;typedef Value mapped_type;typedef Hash hash_type;threadsafe_lookup_table(unsigned num_buckets=19, Hash const& hasher_ = Hash()): buckets(num_buckets), hasher(hasher_) {for (unsigned i = 0; i < num_buckets; ++i) {buckets[i].reset(new bucket_type);}}threadsafe_lookup_table(threadsafe_lookup_table const& other)=delete;threadsafe_lookup_table operator=(threadsafe_lookup_table const& other)=delete;Value value_for(Key const& key, Value const& default_value = Value()) const {return get_bucket(key).value_for(key, default_value);}void add_or_update_mapping(Key const& key, Value const& value) {get_bucket(key).add_or_update_mapping(key, value);}void remove_mapping(Key const& key) {get_bucket(key).remove_mapping(key);}
};

使用散列表，以桶为单位加锁。

#include <bitset>
#include <list>
#include <map>
#include <shared_mutex>
#include <vector>template<typename Key, typename Value, typename Hash=std::hash<Key>>
class threadsafe_lookup_table {
private:class bucket_type {private:typedef std::pair<Key, Value> bucket_value;typedef std::list<bucket_value> bucket_iterator;bucket_data data;mutable std::shared_mutex mutex;bucket_iterator find_entry_for(Key const& key) const {return std::find_if(data.begin(), data.end(), [&](bucket_value const& item){return item.first == key;});}public:Value value_for(Key const& key, Value const& default_value) const {std::shared_lock<std::shared_mutex> lock(mutex);bucket_iterator const found_entry = find_entry_for(key);return (found_entry == data.end()) ? default_value : found_entry->second;}void add_or_update_mapping(Key const& key, Value const& value) {std::unique_lock<std::shared_mutex> lock(mutex);bucket_iterator const found_entry = find_entry_for(key);if (found_entry == data.end()) {data.push_back(bucket_value(key, value));} else {found_entry->second = value;}}void remove_mapping(Key const& key) {std::unique_lock<std::shared_mutex> lock(mutex);bucket_iterator const found_entry = find_entry_for(key);if (found_entry != data.end()) {data.erase(found_entry);}}};std::vector<std::unique_ptr<bucket_type>> buckets;Hash hasher;bucket_type& get_bucket(Key const& key) const {std::size_t const bucket_index = hasher(key) % buckets.size();return *buckets[bucket_index];}
public:typedef Key key_type;typedef Value mapped_type;typedef Hash hash_type;threadsafe_lookup_table(unsigned num_buckets=19, Hash const& hasher_ = Hash()): buckets(num_buckets), hasher(hasher_) {for (unsigned i = 0; i < num_buckets; ++i) {buckets[i].reset(new bucket_type);}}threadsafe_lookup_table(threadsafe_lookup_table const& other)=delete;threadsafe_lookup_table operator=(threadsafe_lookup_table const& other)=delete;Value value_for(Key const& key, Value const& default_value = Value()) const {return get_bucket(key).value_for(key, default_value);}void add_or_update_mapping(Key const& key, Value const& value) {get_bucket(key).add_or_update_mapping(key, value);}void remove_mapping(Key const& key) {get_bucket(key).remove_mapping(key);}std::map<Key, Value> get_map() const {std::vector<std::unique_lock<std::shared_mutex>> locks;for (unsigned i = 0; i < buckets.size(); ++i) {locks.push_back(std::unique_lock<std::shared_mutex>(buckets[i]->mutex));}std::map<Key, Value> res;for (unsigned i = 0; i < buckets.size(); ++i) {for (bucket_iterator it = buckets[i]->data.begin(); it != buckets[i]->data.end();++it) {res.insert(*it);}}return res;}
};

添加数据快照功能，这是一项锁住全部桶的操作，只要每次都按照相同的顺序对每个桶加锁，就没有机会发生死锁（这里指两次快照功能加锁顺序一致）。

3.2 采用多种锁编写线程安全的链表

链表若要具备精细粒度的锁操作，基本思想是让每个节点都具备自己的互斥。

#include <memory>
#include <mutex>template<typename T>
class threadsafe_list {struct node {std::mutex m;std::shared_ptr<T> data;std::unique_ptr<node> next;node() : next() {}node(T const& value) : data(std::make_shared<T>(value)) {}};node head;
public:threadsafe_list() {}~threadsafe_list() {std::remove_if([](node const&) {return true;});}threadsafe_list(threadsafe_list const& other) = delete;threadsafe_list& operator=(threadsafe_list const& other) = delete;void push_front(T const& value) {std::unique_ptr<node> new_node(new node(value));std::lock_guard<std::mutex> lk(head.m);new_node->next = std::move(new_node);}template<typename Function>void for_each(Function f) {node* current = &head;std::unique_lock<std::mutex> lk(head.m);while (node* const next = current->next.get()) {std::unique_lock<std::mutex> next_lk(next->m);lk.unlock();f(*next->data);current=next;lk = std::move(next_lk);}}template<typename Predicate>std::shared_ptr<T> find_first_if(Predicate p) {node* current = &head;std::unique_lock<std::mutex> lk(head.m);while (node* const next = current->next.get()) {std::unique_lock<std::mutex> next_lk(next->m);lk.unlock();if (p(*next->data)) {return next->data;}current=next;lk = std::move(next_lk);}return std::shared_ptr<T>();}template<typename Predicate>void remove_if(Predicate p) {node* current = &head;std::unique_lock<std::mutex> lk(head.m);while (node* const next = current->next.get()) {std::unique_lock<std::mutex> next_lk(next->m);if (p(*next->data)) {std::unique_ptr<node> old_next = std::move(current->next);current->next = std::move(next->next);next_lk.unlock();} else {lk.unlock();current=next;lk = std::move(next_lk);}}}
};

添加Predicate，Function，对应传入的函数对象，以T为入参，处理和断言。比较好理解。