数据结构的选择是解决方案的重要组成部分，当然并行程序也不例外。如果一种数据结构可以被多个线程所访问，要不就是绝对不变(其值不会发生变化，并且不需同步)，要不程序就要对数据结构进行正确的设计，以确保其能在多线程环境下能够(正确的)同步。

  • 一种选择是使用独立的互斥量，其可以锁住需要保护的数据。这种方式是线程轮流访问被保护的数据。这是对数据进行串行的访问，而非并发。
  • 另一种选择是设计一种能够并发访问的数据结构。减少保护区域，减少序列化操作，就能提升并发访问的能力。

设计并发数据结构意味着，多个线程可以并发的访问这个数据结构，线程可对这个数据结构做相同或不同的操作，并且每一个线程都能在自己域中看到该数据结构。多线程环境下，无数据丢失和损毁，所有的数据需要维持原样，且无条件竞争。这样的数据结构，称之为“线程安全”的数据结构。

1 设计建议

如何保证数据结构是线程安全的：

  • 确保无线程能够看到修改数据结构的“不变量”时的状态。
  • 小心会引起条件竞争的接口，提供完整操作的函数，而非操作步骤。
  • 注意数据结构的行为是否会产生异常，从而确保“不变量”的状态。
  • 将死锁的概率降到最低。使用数据结构时，需要限制锁的范围，避免嵌套锁的存在。

作为一个数据结构的设计者，设计数据结构时考虑以下问题：

  • 在锁范围中进行操作，是否允许在锁外执行？
  • 数据结构中不同的区域能是否被不同的互斥量所保护？
  • 所有操作都需要同级互斥量保护吗？
  • 能否对数据结构进行简单的修改，以增加并发访问的概率，且不影响操作语义？

2 基于锁的并发数据结构

基于锁的并发数据结构，需要确保访问线程持有锁的时间最短，对于只有一个互斥量的数据结构来说，这十分困难。需要保证数据不被锁之外的操作所访问，并且还要保证不会在结构上产生条件竞争。使用多个互斥量来保护数据结构中不同的区域时，问题会暴露的更加明显，当操作需要获取多个互斥锁时，就有可能产生死锁。所以在设计时，使用多个互斥量时需要格外小心。

2.1 线程安全栈——互斥锁

  1. /*
  2. 基于锁的线程安全栈
  3. */
  4. #include <exception>
  5. //自定义异常
  6. struct empty_stack : std::exception
  7. {
  8.     const char *what() const throw();
  9. };
  11. template <typename T>
  12. class threadsafe_stack
  13. {
  14. private:
  15.     std::stack<T> data; //封装的标准库stack
  16.     //互斥量m用于保证线程安全，对每个成员函数进行加锁保护，保证在同一时间只有一个线程可以访问数据
  17.     mutable std::mutex m; //mutable表示可变的，const的反义词
  19. public:
  20.     threadsafe_stack() {}
  21.     //拷贝构造函数
  22.     threadsafe_stack(const threadsafe_stack &other)
  23.     {
  24.         std::lock_guard<std::mutex> lock(other.m);
  25.         data = other.data;
  26.     }
  27.     //不允许自动生成赋值构造函数，标记为delete
  28.     threadsafe_stack &operator=(const threadsafe_stack &) = delete;
  30.     void push(T new_value)
  31.     {
  32.         std::lock_guard<std::mutex> lock(m);
  33.         data.push(std::move(new_value)); // 1
  34.     }
  35.     std::shared_ptr<T> pop()
  36.     {
  37.         std::lock_guard<std::mutex> lock(m);
  38.         if (data.empty())
  39.             throw empty_stack(); // 2
  40.         std::shared_ptr<T> const res(std::make_shared<T>(std::move(data.top()))); // 3
  41.         data.pop();                                      // 4
  42.         return res;
  43.     }
  44.     void pop(T &value)
  45.     {
  46.         std::lock_guard<std::mutex> lock(m);
  47.         if (data.empty())
  48.             throw empty_stack();
  49.         value = std::move(data.top()); // 5
  50.         data.pop();                    // 6
  51.     }
  52.     bool empty() const
  53.     {
  54.         std::lock_guard<std::mutex> lock(m);
  55.         return data.empty();
  56.     }
  57. };

所有成员函数都使用std::lock_guard<>保护数据，所以栈成员函数才是“线程安全”的。当然，构造与析构函数不是“线程安全”的，不过也没关系，因为构造与析构只有一次。即使在多线程下，并发调用的成员函数也是安全的(因为使用锁)。
序列化线程会隐性的限制程序性能，这就是栈争议声最大的地方：当一个线程在等待锁时，就会无所事事。这样的实现会限制栈的实现方式，线程等待时会浪费宝贵的资源去检查数据，或要求用户编写外部等待和提示的代码。

2.2 线程安全队列

锁和条件变量

  1. /*
  2. 基于锁和条件变量的线程安全队列
  3. */
  4. template <typename T>
  5. class threadsafe_queue
  6. {
  7. private:
  8.     //互斥量m用于保证线程安全，对每个成员函数进行加锁保护，保证在同一时间只有一个线程可以访问数据
  9.     mutable std::mutex mut;
  10.     std::queue<std::shared_ptr<T>> data_queue; //使用std::shared_ptr实例，而不是直接使用数据的值
  11.     std::condition_variable data_cond;         //条件变量
  13. public:
  14.     threadsafe_queue() {}
  16.     void push(T new_value)
  17.     {
  18.         std::shared_ptr<T> data(std::make_shared<T>(std::move(new_value))); // 5
  19.         std::lock_guard<std::mutex> lk(mut);
  20.         data_queue.push(data);
  21.         data_cond.notify_one(); //通知一个等待条件变量的线程
  22.     }
  24.     void wait_and_pop(T &value)
  25.     {
  26.         std::unique_lock<std::mutex> lk(mut);
  27.         //使用条件变量等待条件达成，比循环调用empty()要好很多
  28.         data_cond.wait(lk, [this] { return !data_queue.empty(); });
  29.         value = std::move(*data_queue.front()); // 1
  30.         data_queue.pop();
  31.     }
  33.     bool try_pop(T &value)
  34.     {
  35.         std::lock_guard<std::mutex> lk(mut);
  36.         if (data_queue.empty())
  37.             return false;
  38.         value = std::move(*data_queue.front()); // 2
  39.         data_queue.pop();
  40.         return true;
  41.     }
  43.     std::shared_ptr<T> wait_and_pop()
  44.     {
  45.         std::unique_lock<std::mutex> lk(mut);
  46.         data_cond.wait(lk, [this] { return !data_queue.empty(); });
  47.         std::shared_ptr<T> res = data_queue.front(); // 3
  48.         data_queue.pop();
  49.         return res;
  50.     }
  52.     std::shared_ptr<T> try_pop()
  53.     {
  54.         std::lock_guard<std::mutex> lk(mut);
  55.         if (data_queue.empty())
  56.             return std::shared_ptr<T>();
  57.         std::shared_ptr<T> res = data_queue.front(); // 4
  58.         data_queue.pop();
  59.         return res;
  60.     }
  62.     bool empty() const
  63.     {
  64.         std::lock_guard<std::mutex> lk(mut);
  65.         return data_queue.empty();
  66.     }
  67. };

std::shared_ptr<>持有数据的好处：新实例分配结束时，不会被锁在push()⑤当中。因为内存分配需要在性能上付出很高的代价(性能较低)，所以使用std::shared_ptr<>对队列的性能有很大的提升，其减少了互斥量持有的时间，允许其他线程在分配内存的同时，对队列进行其他的操作。
如同栈的例子，使用互斥量保护整个数据结构，不过会限制队列对并发的支持；虽然，多线程可能被队列中的各种成员函数所阻塞，但仍有一个线程能在任意时间内进行工作。不过，这种限制是因为在实现中使用了std::queue<>；因为使用标准容器的原因，数据处于保护中。要对数据结构实现进行具体的控制，需要提供更多细粒度锁，来完成更高级的并发。

细粒度锁和条件变量

为了使用细粒度锁，需要看一下队列内部的组成结构，并且将一个互斥量与每个数据相关联。

  1. //基于细粒度锁和条件变量的自定义线程安全队列
  2. #ifndef THREADSAFE_QUEUE2
  3. #define THREADSAFE_QUEUE2
  4. #include <memory>
  5. #include <mutex>
  6. #include <condition_variable>
  7. //类定义
  8. template <typename T>
  9. class threadsafe_queue
  10. {
  11. private:
  12.     struct node
  13.     {
  14.         std::shared_ptr<T> data;
  15.         std::unique_ptr<node> next;
  16.     };
  18.     std::mutex head_mutex;
  19.     std::unique_ptr<node> head;
  20.     std::mutex tail_mutex;
  21.     node *tail;
  22.     std::condition_variable data_cond;
  24. private:
  25.     node *get_tail()
  26.     {
  27.         std::lock_guard<std::mutex> tail_lock(tail_mutex);
  28.         return tail;
  29.     }
  31.     std::unique_ptr<node> pop_head() // 1
  32.     {
  33.         std::unique_ptr<node> old_head = std::move(head);
  34.         head = std::move(old_head->next);
  35.         return old_head;
  36.     }
  38.     std::unique_lock<std::mutex> wait_for_data() // 2
  39.     {
  40.         std::unique_lock<std::mutex> head_lock(head_mutex);
  41.         data_cond.wait(head_lock, [&] { return head.get() != get_tail(); });
  42.         return std::move(head_lock); // 3
  43.     }
  45.     std::unique_ptr<node> wait_pop_head()
  46.     {
  47.         std::unique_lock<std::mutex> head_lock(wait_for_data()); // 4
  48.         return pop_head();
  49.     }
  51.     std::unique_ptr<node> wait_pop_head(T &value)
  52.     {
  53.         std::unique_lock<std::mutex> head_lock(wait_for_data()); // 5
  54.         value = std::move(*head->data);
  55.         return pop_head();
  56.     }
  58.     std::unique_ptr<node> try_pop_head()
  59.     {
  60.         std::lock_guard<std::mutex> head_lock(head_mutex);
  61.         if (head.get() == get_tail())
  62.         {
  63.             return std::unique_ptr<node>();
  64.         }
  65.         return pop_head();
  66.     }
  68.     std::unique_ptr<node> try_pop_head(T &value)
  69.     {
  70.         std::lock_guard<std::mutex> head_lock(head_mutex);
  71.         if (head.get() == get_tail())
  72.         {
  73.             return std::unique_ptr<node>();
  74.         }
  75.         value = std::move(*head->data);
  76.         return pop_head();
  77.     }
  79. public:
  80.     threadsafe_queue() : head(new node), tail(head.get()) {}
  81.     threadsafe_queue(const threadsafe_queue &other) = delete;
  82.     threadsafe_queue &operator=(const threadsafe_queue &other) = delete;
  84.     void push(T new_value)
  85.     {
  86.         std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));
  87.         std::unique_ptr<node> p(new node);
  88.         {
  89.             std::lock_guard<std::mutex> tail_lock(tail_mutex);
  90.             tail->data = new_data;
  91.             node *const new_tail = p.get();
  92.             tail->next = std::move(p);
  93.             tail = new_tail;
  94.         }
  95.         data_cond.notify_one();
  96.     }
  98.     std::shared_ptr<T> wait_and_pop()
  99.     {
  100.         std::unique_ptr<node> const old_head = wait_pop_head();
  101.         return old_head->data;
  102.     }
  104.     void wait_and_pop(T &value)
  105.     {
  106.         std::unique_ptr<node> const old_head = wait_pop_head(value);
  107.     }
  109.     std::shared_ptr<T> try_pop()
  110.     {
  111.         std::unique_ptr<node> old_head = try_pop_head();
  112.         return old_head ? old_head->data : std::shared_ptr<T>();
  113.     }
  115.     bool try_pop(T &value)
  116.     {
  117.         std::unique_ptr<node> const old_head = try_pop_head(value);
  118.         return old_head;
  119.     }
  121.     bool empty()
  122.     {
  123.         std::lock_guard<std::mutex> head_lock(head_mutex);
  124.         return (head.get() == get_tail());
  125.     }
  126. };
  127. #endif

3 更复杂的数据结构

3.1 线程安全的链表

用细粒度锁最初的想法，是为了让链表每个节点都拥有一个互斥量。当链表很长时，就会使用有很多的互斥量！
这样的好处是对于链表中每一个独立的部分，都能实现真实的并发：其真正感兴趣的是对持有的节点群进行上锁，并且在移动到下一个节点的时，对当前节点进行释放。

  1. /*线程安全链表——支持迭代器
  2. */
  3. #ifndef THREADSAFE_LIST
  4. #define THREADSAFE_LIST
  5. #include <memory>
  6. #include <mutex>
  7. template <typename T>
  8. class threadsafe_list
  9. {
  10.     struct node // 1
  11.     {
  12.         std::mutex m;
  13.         std::shared_ptr<T> data;
  14.         std::unique_ptr<node> next;
  15.         node() : // 2
  16.                  next()
  17.         {
  18.         }
  20.         node(T const &value) : // 3
  21.                                data(std::make_shared<T>(value))
  22.         {
  23.         }
  24.     };
  26.     node head;
  28. public:
  29.     threadsafe_list()
  30.     {
  31.     }
  33.     ~threadsafe_list()
  34.     {
  35.         remove_if([](node const &) { return true; });
  36.     }
  38.     threadsafe_list(threadsafe_list const &other) = delete;
  39.     threadsafe_list &operator=(threadsafe_list const &other) = delete;
  41.     void push_front(T const &value)
  42.     {
  43.         std::unique_ptr<node> new_node(new node(value)); // 4
  44.         std::lock_guard<std::mutex> lk(head.m);
  45.         new_node->next = std::move(head.next); // 5
  46.         head.next = std::move(new_node);       // 6
  47.     }
  49.     template <typename Function>
  50.     void for_each(Function f) // 7
  51.     {
  52.         node *current = &head;
  53.         std::unique_lock<std::mutex> lk(head.m);       // 8
  54.         while (node *const next = current->next.get()) // 9
  55.         {
  56.             std::unique_lock<std::mutex> next_lk(next->m); // 10
  57.             lk.unlock();                                   // 11
  58.             f(*next->data);                                // 12
  59.             current = next;
  60.             lk = std::move(next_lk); // 13
  61.         }
  62.     }
  64.     template <typename Predicate>
  65.     std::shared_ptr<T> find_first_if(Predicate p) // 14
  66.     {
  67.         node *current = &head;
  68.         std::unique_lock<std::mutex> lk(head.m);
  69.         while (node *const next = current->next.get())
  70.         {
  71.             std::unique_lock<std::mutex> next_lk(next->m);
  72.             lk.unlock();
  73.             if (p(*next->data)) // 15
  74.             {
  75.                 return next->data; // 16
  76.             }
  77.             current = next;
  78.             lk = std::move(next_lk);
  79.         }
  80.         return std::shared_ptr<T>();
  81.     }
  83.     template <typename Predicate>
  84.     void remove_if(Predicate p) // 17
  85.     {
  86.         node *current = &head;
  87.         std::unique_lock<std::mutex> lk(head.m);
  88.         while (node *const next = current->next.get())
  89.         {
  90.             std::unique_lock<std::mutex> next_lk(next->m);
  91.             if (p(*next->data)) // 18
  92.             {
  93.                 std::unique_ptr<node> old_next = std::move(current->next);
  94.                 current->next = std::move(next->next);
  95.                 next_lk.unlock();
  96.             } // 20
  97.             else
  98.             {
  99.                 lk.unlock(); // 21
  100.                 current = next;
  101.                 lk = std::move(next_lk);
  102.             }
  103.         }
  104.     }
  105. };
  106. #endif

3.2 线程安全map

假设有固定数量的桶，每个桶都有一个键值(关键特性)，以及散列函数。这就意味着你可以安全的对每个桶上锁。当再次使用互斥量(支持多读者单作者)时，就能将并发访问的可能性增加N倍，这里N是桶的数量。
对于键值的操作，需要有合适的函数。C++标准库提供std::hash<>模板。

  1. /*
  2. 线程安全hashmap
  3. 在构造函数中指定构造桶的数量。默认为19个（哈希表在有质数个桶时，工作效率最高）。
  4. 每一个桶都会被一个std::shared_mutex①实例锁保护，来允许并发读取，或对每一个桶，只有一个线程对其进行修改。
  5. */
  6. #ifndef THREADSAFE_MAP
  7. #define THREADSAFE_MAP
  8. #include <shared_mutex>
  9. #include <list>
  10. #include <vector>
  11. #include <map>
  12. #include <algorithm>
  13. template <typename Key, typename Value, typename Hash = std::hash<Key>>
  14. class threadsafe_lookup_table
  15. {
  16. private:
  17.     class bucket_type
  18.     {
  19.     private:
  20.         typedef std::pair<Key, Value> bucket_value;
  21.         typedef std::list<bucket_value> bucket_data;
  22.         typedef typename bucket_data::iterator bucket_iterator;
  24.         bucket_data data;
  25.         mutable std::shared_mutex mutex; // 1
  27.         bucket_iterator find_entry_for(Key const &key) const // 2
  28.         {
  29.             return std::find_if(data.begin(), data.end(),
  30.                                 [&](bucket_value const &item) { return item.first == key; });
  31.         }
  33.     public:
  34.         Value value_for(Key const &key, Value const &default_value) const
  35.         {
  36.             std::shared_lock<std::shared_mutex> lock(mutex); // 3
  37.             bucket_iterator const found_entry = find_entry_for(key);
  38.             return (found_entry == data.end()) ? default_value : found_entry->second;
  39.         }
  41.         void add_or_update_mapping(Key const &key, Value const &value)
  42.         {
  43.             std::unique_lock<std::shared_mutex> lock(mutex); // 4
  44.             bucket_iterator const found_entry = find_entry_for(key);
  45.             if (found_entry == data.end())
  46.             {
  47.                 data.push_back(bucket_value(key, value));
  48.             }
  49.             else
  50.             {
  51.                 found_entry->second = value;
  52.             }
  53.         }
  55.         void remove_mapping(Key const &key)
  56.         {
  57.             std::unique_lock<std::shared_mutex> lock(mutex); // 5
  58.             bucket_iterator const found_entry = find_entry_for(key);
  59.             if (found_entry != data.end())
  60.             {
  61.                 data.erase(found_entry);
  62.             }
  63.         }
  64.     };
  66.     std::vector<std::unique_ptr<bucket_type>> buckets; // 6
  67.     Hash hasher;
  69.     bucket_type &get_bucket(Key const &key) const // 7
  70.     {
  71.         std::size_t const bucket_index = hasher(key) % buckets.size();
  72.         return *buckets[bucket_index];
  73.     }
  75. public:
  76.     typedef Key key_type;
  77.     typedef Value mapped_type;
  79.     typedef Hash hash_type;
  80.     threadsafe_lookup_table(unsigned num_buckets = 19, Hash const &hasher_ = Hash())
  81.         : buckets(num_buckets), hasher(hasher_)
  82.     {
  83.         for (unsigned i = 0; i < num_buckets; ++i)
  84.         {
  85.             buckets[i].reset(new bucket_type);
  86.         }
  87.     }
  89.     threadsafe_lookup_table(threadsafe_lookup_table const &other) = delete;
  90.     threadsafe_lookup_table &operator=(
  91.         threadsafe_lookup_table const &other) = delete;
  93.     Value value_for(Key const &key,
  94.                     Value const &default_value = Value()) const
  95.     {
  96.         return get_bucket(key).value_for(key, default_value); // 8
  97.     }
  99.     void add_or_update_mapping(Key const &key, Value const &value)
  100.     {
  101.         get_bucket(key).add_or_update_mapping(key, value); // 9
  102.     }
  104.     void remove_mapping(Key const &key)
  105.     {
  106.         get_bucket(key).remove_mapping(key); // 10
  107.     }
  109.     std::map<Key, Value> get_map() const
  110.     {
  111.         std::vector<std::unique_lock<std::shared_mutex>> locks;
  112.         for (unsigned i = 0; i < buckets.size(); ++i)
  113.         {
  114.             locks.push_back(std::unique_lock<std::shared_mutex>(buckets[i].mutex));
  115.         }
  116.         std::map<Key, Value> res;
  117.         for (unsigned i = 0; i < buckets.size(); ++i)
  118.         {
  119.             for (auto it = buckets[i].data.begin();
  120.                  it != buckets[i].data.end();
  121.                  ++it)
  122.             {
  123.                 res.insert(*it);
  124.             }
  125.         }
  126.         return res;
  127.     }
  128. };
  129. #endif