Cpp 无锁编程（C++ Concurrency in Action）

本文是 C++ Concurrency in Action 的总结

通过原子变量实现 spinlock

1. class spinlock_mutex {
2.   std::atomic_flag flag;
3. public:
4.   spinlock_mutex(): flag(ATOMIC_FLAG_INIT) {}
5.   void lock() {
6.     while (flag.test_and_set(std::memory_order_acquire));
7.   }
8.   void unlock() {
9.     flag.clear(std::memory_order_release);
10.   }
11. };

复制代码

lock 这里利用 acquire 内存序的缘故因由是 lock 须要确保临界区里的读写须要看到上一个线程在 unlock 之前的修改
unlock 利用 release 表现临界区里的修改须要对下一个持锁线程可见
这里值得留意的是这个着实不是无锁编程，固然没有利用 mutex ，固然它不会被 block （线程进入 D 状态）
lock-free 的意思是如果有多个线程在利用一个共享的数据，在有限的step内里，此中有一个线程肯定能完成这个利用。
从这个界说上看，我们的 spinlock 如果有一个线程 lock 后 panic 了，那其他的线程同样被壅闭无法继续完成相应的利用。因此不是 lock-free
但是这个例子着实用到了无锁编程常常利用到的原子变量和其对应的利用。这也分析了用原子变量不代表就是无锁
实现一个 lock-free stack

一个最简化版的 stack

1. template <typename T>
2. class LockFreeStack {
3. private:
4.   struct Node {
5.     T data_;
6.     Node* next;
8.     Node(T const& data) : data_(data) {}
9.   };
10.   std::atomic<Node*> head{nullptr};
12. public:
13.   void push(T const& data) {
14.     Node* node = new Node(data);
15.     node->next = head.load();
16.     while (!head.compare_exchange_weak(node->next, node));
17.   }
19.   bool pop(T& result) {
20.     Node* old_head = head.load();
21.     while (old_head && !head.compare_exchange_weak(old_head, old_head->next));
22.     if (!old_head) {
23.       return false;
24.     }
25.     result = old_head->data_;
26.     return true;
27.   }
28. };

复制代码

pop 这里不能仅靠 load ，由于 pop 须要确保差别线程拿到的值须要不一样，以是须要利用 compare_exchange_weak 来包管如果节点被其他线程取了以后重新获取节点。
这里尚有一点小标题，pop 出来的节点利用了拷贝数据来返回，但是拷贝过程中大概发生非常，那如许的话这个数据就会丢失。以是可以利用智能指针来处置惩罚这个标题。如下：

1. template <typename T>
2. class LockFreeStack {
3. private:
4.   struct Node {
5.     std::shared_ptr<T> data_;
6.     Node* next;
8.     Node(T const& data) : data_(std::make_shared<T>(data)) {}
9.   };
10.   std::atomic<Node*> head{nullptr};
12. public:
13.   void push(T const& data) {
14.     Node* node = new Node(data);
15.     node->next = head.load();
16.     while (!head.compare_exchange_weak(node->next, node));
17.   }
19.   std::shared_ptr<T> pop() {
20.     Node* old_head = head.load();
21.     while (old_head && !head.compare_exchange_weak(old_head, old_head->next));
22.     return old_head ? old_head->data_ : std::shared_ptr<T>{};
23.   }
24. };

复制代码

现在的代码仍旧有一个标题，pop 出节点后没有开释内存。由于在多线程环境下，你无法确定这个节点是否被其他线程持有，以是不能安全的删除。shared_ptr 只管理了 data，没有管理 Node
修复内存走漏

通过 pop 的线程数制止竞争

1. template <typename T>
2. class LockFreeStack {
3. private:
4.   struct Node {
5.           //...
6.   };
7.   std::atomic<Node*> head{nullptr};
8.   std::atomic<unsigned> threads_in_pop{0};
9.   std::atomic<Node*> to_be_deleted{nullptr};
10.   static void delete_nodes(Node* node) {
11.     while (node) {
12.       auto next = node->next;
13.       delete node;
14.       node = next;
15.     }
16.   }
18.   void chain_pending_nodes(Node* nodes) {
19.     Node* last = nodes;
20.     while (Node* const next = last->next) {
21.       last = next;
22.     }
23.     chain_pending_nodes(nodes, last);
24.   }
25.   void chain_pending_nodes(Node* first, Node* last) {
26.     last->next = to_be_deleted.load();
27.     while (!to_be_deleted.compare_exchange_weak(last->next, first));
28.   }
30.   void try_reclaim(Node* node) {
31.     if (threads_in_pop == 1) {
32.       Node* nodes_to_delete = to_be_deleted.exchange(nullptr);
33.       if (!--threads_in_pop) {
34.         delete_nodes(nodes_to_delete);
35.       } else if (nodes_to_delete) {
36.         chain_pending_nodes(nodes_to_delete);
37.       }
38.       delete node;
39.     } else {
40.       if (node) {
41.         chain_pending_nodes(node);
42.       }
43.       --threads_in_pop;
44.     }
45.   }
47. public:
48.   void push(T const& data) {
49.     //...
50.   }
52.   std::shared_ptr<T> pop() {
53.     threads_in_pop++;
54.     Node* old_head = head.load();
55.     while (old_head && !head.compare_exchange_weak(old_head, old_head->next));
56.     std::shared_ptr<T> res;
57.     if (old_head) {
58.       res.swap(old_head->data_);
59.     }
60.     try_reclaim(old_head);
61.     return res;
62.   }
63. };

复制代码

这个实现的焦点头脑是：如果没有其他线程在实行 pop，就可以安全的把节点删掉。
以是引入了 threads_in_pop 来监控有几个线程，进入 pop 就加一，脱离就减一。
如果 threads_in_pop == 1 分析是只有一个线程，可以安全删除，但须要留意的是在 delete_nodes 前须要再判定一次，由于这中央有大概有其他线程进入 pop 然后利用了。
threads_in_pop == 1 的时间可以安全删掉当前pop出的节点，由于其他线程背面肯定不会访问到这个节点。

这个方案有个标题，在低并发的时间还能工作，但是高并发下面大概会失效。现在的方案是监控 pop 利用是否有其他线程，可以思量监控节点是否有其他的线程在访问。
Hazard Pointer

hazard pointer 根本思绪是，如果一个线程要访问另一个线程大概想要删除的对象，它起首会设置一个伤害指针来指向该对象，告知另一个线程删除该对象确实会带来风险。一旦不再须要该对象，则扫除伤害指针。
Hazard Pointer 的工作原理可以压缩成三个利用：Protect（掩护）、Retire（退役）、Scan（扫描采取）。

• 掩护：线程声明我正在利用这个节点
  
• 退役：线程声明我已经把这个节点移除了，但是先不 free
  
• 扫描采取：查抄待删除链表内里的节点，如果没有线程在利用就开释
  

手写一个 hazard pointer

改进后的 pop 为例：

1.   std::shared_ptr<T> pop() {
2.     std::atomic<void*>& hp = get_hazard_pointer_for_current_thread();
3.     auto old_head = head.load();
4.     // protect
5.     do {
6.       Node* temp;
7.       do {
8.         temp = old_head;
9.         hp.store(old_head);
10.         old_head = head.load();
11.       } while (old_head != temp);
12.     } while (old_head &&
13.              !head.compare_exchange_strong(old_head, old_head->next));
14.     hp.store(nullptr);
15.     std::shared_ptr<T> res;
16.     if (old_head) {
17.       res.swap(old_head->data_);
18.       // retire
19.       if (outstanding_hazard_pointers_for(old_head)) {
20.         reclaim_later(old_head);
21.       } else {
22.         delete old_head;
23.       }
24.       // scan and delete
25.       delete_nodes_with_no_hazards();
26.     }
27.     return res;
28.   }

复制代码

获取 hazard pointer，hazard pointer 须要将线程和节点关联起来，而且生存到一个全部线程都可以访问到的地方:

1. unsigned const MAX_HAZARD_POINTERS = 100;
2. struct HazardPointer {
3.   std::atomic<std::thread::id> id;
4.   std::atomic<void*> pointer;
5. };
6. HazardPointer hazard_pointers[MAX_HAZARD_POINTERS];
7. class HpOwner {
8.   HazardPointer* hp;
10. public:
11.   HpOwner(HpOwner const&) = delete;
12.   HpOwner operator=(HpOwner const&) = delete;
13.   HpOwner() : hp(nullptr) {
14.     for (auto i = 0; i < MAX_HAZARD_POINTERS; i++) {
15.       std::thread::id old_id;
16.       if (hazard_pointers[i].id.compare_exchange_strong(
17.               old_id, std::this_thread::get_id())) {
18.         hp = &hazard_pointers[i];
19.         break;
20.       }
21.     }
22.     if (!hp) {
23.       throw std::runtime_error("No hazard pointers available");
24.     }
25.   }
26.   std::atomic<void*>& get_pointer() { return hp->pointer; }
27.   ~HpOwner() {
28.     hp->pointer.store(nullptr);
29.     hp->id.store(std::thread::id());
30.   }
31. };
33. std::atomic<void*>& get_hazard_pointer_for_current_thread() {
34.   thread_local static HpOwner hazard;
35.   return hazard.get_pointer();
36. }

复制代码

判定某个节点是否被其他线程持有：

1. bool outstanding_hazard_pointers_for(void* p) {
2.   for (auto i = 0; i < MAX_HAZARD_POINTERS; i++) {
3.     if (hazard_pointers[i].pointer.load() == p) {
4.       return true;
5.     }
6.   }
7.   return false;
8. }

复制代码

retire 这个节点，也就是把这个节点插入一个待删除的链表内里：

1. template <typename T>
2. void do_delete(void* p) {
3.   delete static_cast<T*>(p);
4. }
6. struct DataToReclaim {
7.   void* data;
8.   std::function<void(void*)> deleter;
9.   DataToReclaim* next;
10.   template <typename T>
11.   DataToReclaim(T* p) : data(p), deleter(&do_delete<T>), next(nullptr) {}
12.   ~DataToReclaim() { deleter(data); }
13. };
15. std::atomic<DataToReclaim*> nodes_to_reclaim;
16. void add_to_reclaim_list(DataToReclaim* node) {
17.   node->next = nodes_to_reclaim.load();
18.   while (!nodes_to_reclaim.compare_exchange_weak(node->next, node));
19. }
21. template <typename T>
22. void reclaim_later(T* data) {
23.   add_to_reclaim_list(new DataToReclaim(data));
24. }

复制代码

扫描采取：

1. void delete_nodes_with_no_hazards() {
2.   auto current = nodes_to_reclaim.exchange(nullptr);
3.   while (current) {
4.     auto next = current->next;
5.     if (!outstanding_hazard_pointers_for(current->data)) {
6.       delete current;
7.     } else {
8.       add_to_reclaim_list(current);
9.     }
10.     current = next;
11.   }
12. }

复制代码

这段代码是比力须要留意的：

1. //...
2.     do {
3.       Node* temp;
4.       do {
5.         temp = old_head;
6.         hp.store(old_head);
7.         old_head = head.load();
8.       } while (old_head != temp);
9.     } while (old_head &&
10.              !head.compare_exchange_strong(old_head, old_head->next));
11. //...

复制代码

这里有两层循环：1. 内层循环，作用是确保 hp 掩护的节点照旧当前的 head。在获取 old_head （temp = old_head）和 hp.store(old_head) 之间大概存在竞态 2. 外层节点就是一个很正常的 pop 节点所做的利用，如果发现 old_head 被修改了则须要重新去获取头节点。这里由于失败会重新处置惩罚内层循环，以是 spurious failure 的代价比力昂贵，在这里用了 exchange_strong
现在这个版本的实现着实有很大的性能标题。由于每一次 pop 都会去扫描待删除链表，扫描是须要占用 CPU 资源的，而且差别线程去访问一个全局的链表是有竞争的。以是有两种办理办法：1. 空间换时间，不须要每次都去扫描，只有当链表里的节点凌驾肯定命目才须要删除 2. 每个线程有自己的链表，不须要去竞争同一个。
Hazard Pointer 内存上界

Hazard Pointer 是有明白的内存上界的。在恣意时候，体系中未被采取的退役节点数最多为：N * (R + K * N) N 是线程数，K 是每个线程的 hazard pointer 数目， R 是触发 scan 的域值。
这个上界的寄义是：每个线程最多持有 R 个未被 scan 的退役节点，加上最多有 K×N 个节点被其他线程的 hazard pointer 掩护着无法开释。通常 R 被设置为 2×K×N，以是上界约为 N × 3KN = 3KN²。对于 64 线程、K=2 的 M-S queue 场景，上界是 3×2×64² = 24576 个节点。如果每个节点 64 字节，这是约 1.5 MB 的"耽误开释"开销——在绝大多数场景下完全可以担当。
Hazard Pointer 为什么是安全的

Safety Invariant: 节点 P 被开释，当且仅当：(1) P 在某个线程的 retired_list 中；且 (2) 实行 scan 时，P 不在任何线程的 hazard pointer 中。
条件 (1) 要求 P 已经被从数据结构中逻辑删除（否则不会被 retire）。条件 (2) 要求没有线程声明正在利用 P。这两个条件一起包管了：如果一个线程正在利用 P（它的 hazard pointer 中有 P），P 就不会被开释。
这个稳定量的证明焦点在于 protect-then-validate 协议：如果一个线程 T 通过 hp.store(P)+ 验证 P 仍在数据结构中这两步都乐成了，那么在 T 扫除它的 hazard pointer 之前，P 肯定不会被开释——由于任何实行 scan 的线程在网络 hazard pointer 时肯定会看到 T 的声明（acquire-release 语义包管了可见性）。
看一个场景：
场景设定：队列状态 Head → A → B → C，线程 T1 实行 dequeue（目的是摘掉 A），线程 T2 也实行 dequeue 而且还会触发 scan。最伤害的交错是：T1 正在掩护 A 的过程中，T2 已经把 A retire 了而且开始 scan。
时候T1 (dequeuer)T2 (dequeuer + scanner)A 的状态t0head = Head.load()

→ 得到 A-在队列中t1hp[0].store(A)-在队列中，被T1掩护t2-head = Head.load()

→ 得到 A在队列中，被T1掩护t3-hp[0].store(A)被T1和T2掩护t4-验证 head == Head → 通过同上t5-next = A->next

→ 得到 B同上t6-CAS(Head, A, B)

→ 乐成逻辑删除

，仍被T1和T2掩护t7-hp[0].clear(); hp[1].clear()逻辑删除，仅被T1掩护t8-retire(A)

→ A进入retired_list退役，仅被T1掩护t9-scan()

开始：网络全部 hazard pointer退役，仅被T1掩护t10-scan 读取 T1.hp[0] → 发现值为 A退役，scan知道被掩护t11-A 在 protected_set 中，跳过不开释退役，安全生存t12验证 head == Head → 失败-退役，仅被T1掩护t13hp[0].clear()

→ 重新开始-退役，无人掩护t14-（下次 scan 发现 A 无掩护）退役，无人掩护t15-delete A

→ 安全开释已开释t9-t10 是整个协议的安全关键。 在 t9，T2 的 scan 开始网络全部线程的 hazard pointer。它读取 T1 的 hp[0]，发现值为 A。这个读取利用 memory_order_acquire，共同 T1 在 t1 时 hp[0].store(A)利用的 memory_order_release，构成了一个 acquire-release 同步——T2 在 t10 看到的 A，肯定是 T1 在 t1 或之后存入的值，不会看到更早的 nullptr。完备的实现

1. #include #include #include #include #include unsigned const MAX_HAZARD_POINTERS = 100;
2. struct HazardPointer {
3.   std::atomic<std::thread::id> id;
4.   std::atomic<void*> pointer;
5. };
6. HazardPointer hazard_pointers[MAX_HAZARD_POINTERS];
7. class HpOwner {
8.   HazardPointer* hp;
10. public:
11.   HpOwner(HpOwner const&) = delete;
12.   HpOwner operator=(HpOwner const&) = delete;
13.   HpOwner() : hp(nullptr) {
14.     for (auto i = 0; i < MAX_HAZARD_POINTERS; i++) {
15.       std::thread::id old_id;
16.       if (hazard_pointers[i].id.compare_exchange_strong(
17.               old_id, std::this_thread::get_id())) {
18.         hp = &hazard_pointers[i];
19.         break;
20.       }
21.     }
22.     if (!hp) {
23.       throw std::runtime_error("No hazard pointers available");
24.     }
25.   }
26.   std::atomic<void*>& get_pointer() { return hp->pointer; }
27.   ~HpOwner() {
28.     hp->pointer.store(nullptr);
29.     hp->id.store(std::thread::id());
30.   }
31. };
33. std::atomic<void*>& get_hazard_pointer_for_current_thread() {
34.   thread_local static HpOwner hazard;
35.   return hazard.get_pointer();
36. }bool outstanding_hazard_pointers_for(void* p) {
37.   for (auto i = 0; i < MAX_HAZARD_POINTERS; i++) {
38.     if (hazard_pointers[i].pointer.load() == p) {
39.       return true;
40.     }
41.   }
42.   return false;
43. }template <typename T>
44. void do_delete(void* p) {
45.   delete static_cast<T*>(p);
46. }
48. struct DataToReclaim {
49.   void* data;
50.   std::function<void(void*)> deleter;
51.   DataToReclaim* next;
52.   template <typename T>
53.   DataToReclaim(T* p) : data(p), deleter(&do_delete<T>), next(nullptr) {}
54.   ~DataToReclaim() { deleter(data); }
55. };
57. std::atomic<DataToReclaim*> nodes_to_reclaim;
58. void add_to_reclaim_list(DataToReclaim* node) {
59.   node->next = nodes_to_reclaim.load();
60.   while (!nodes_to_reclaim.compare_exchange_weak(node->next, node));
61. }
63. template <typename T>
64. void reclaim_later(T* data) {
65.   add_to_reclaim_list(new DataToReclaim(data));
66. }void delete_nodes_with_no_hazards() {
67.   auto current = nodes_to_reclaim.exchange(nullptr);
68.   while (current) {
69.     auto next = current->next;
70.     if (!outstanding_hazard_pointers_for(current->data)) {
71.       delete current;
72.     } else {
73.       add_to_reclaim_list(current);
74.     }
75.     current = next;
76.   }
77. }template class LockFreeStack { private:  struct Node {    std::shared_ptr data_;    Node* next;    Node(T const& data) : data_(std::make_shared(data)) {}  };  std::atomic head{nullptr}; public:  void push(T const& data) {    Node* node = new Node(data);    node->next = head.load();    while (!head.compare_exchange_weak(node->next, node));  }  std::shared_ptr pop() {    std::atomic& hp = get_hazard_pointer_for_current_thread();    auto old_head = head.load();    do {      Node* temp;      do {        temp = old_head;        hp.store(old_head);        old_head = head.load();      } while (old_head != temp);    } while (old_head &&             !head.compare_exchange_strong(old_head, old_head->next));    hp.store(nullptr);    std::shared_ptr res;    if (old_head) {      res.swap(old_head->data_);      if (outstanding_hazard_pointers_for(old_head)) {        reclaim_later(old_head);      } else {        delete old_head;      }      delete_nodes_with_no_hazards();    }    return res;  }};int main() {  LockFreeStack stack;  constexpr int count = 10;  std::atomic popped{0};  std::thread t([&] {    for (int i = 0; i < count; ++i) {      stack.push(i);    }  });  std::thread t1([&] {    while (popped.load() < count) {      auto value = stack.pop();      if (value) {        popped.fetch_add(1);        std::cout internal_count.fetch_sub(1) == 1) {        delete ptr;      }    }  }};

复制代码

这里利用了两个计数，一个外部计数和一个内部计数。外部计数是和指向节点的指针绑定在一起的，而内部计数是在节点内部。外部计数负责“安全拿到节点”，内部计数负责“节点摘下后耽误采取”
increase_head_count(old_head); 起首取头节点，而且立即给头节点的 external_count 加 1，表现线程正在利用这个指针不能立即 delete。
if (head.compare_exchange_strong(old_head, ptr->next)) 如果当前 head 仍旧是我刚刚增长过引用计数的 old_head，就把 head 改成 ptr->next。
int const count_increase = old_head.external_count - 2; 这里减掉两个引用计数，分别是：1. 当火线程通过 increase_head_count 增长的 2. head 自己就持有的谁人外部引用
ptr->internal_count.fetch_add(count_increase) == -count_increase 剩下的外部引用要转移到内部去，如果加上后便是0就可以删除这个节点。
else if (ptr->internal_count.fetch_sub(1) == 1) 这个分支是节点被其他线程取走了，减去内部计数。 以是如果许多线程都CAS失败了，内部引用就大概为负数。比及末了乐成 pop 的线程把外部计数归并进 internal_count 则会抵消。
以是，内部计数为负数的本质是： 失败线程先把“我不消了”记到 internal_count 里；乐成线程稍后把 external_count 转进来，两边抵消。负数只是中央状态。
Epoch-Based Reclamation

TODO
实现一个 lock-free queue

TODO
lock-free 的利益和弊端

利益：

• 进步并发，如果有锁一些线程会被 block，无锁下每个线程照旧连续运行
  
• robustness，整个体系不受持锁线程的影响，就算持锁线程 panic 了，体系仍旧可以继续运行
  

弊端：

• 编程难度大。没有锁的掩护，你没办法制止多个线程同时访问某个数据，你必须自己来确保数据结构的 invariants 在并发下 仍旧创建
  
• 制止 live lock。死锁是两个线程相互期待卡死的环境，无锁编程不会出现这种环境，但是会出现 live lock. live lock 是两个线程没有卡死，都在积极处置惩罚数据，但是没办法继续往前推进。
  
• 低沉体系的性能（看起来特别反直觉）。原子变量利用的性能是小于非原子变量的。而且原子变量须要在多个线程之间同步，这对 cpu 缓存非常不友爱（由于线程大概在差别的cpu核上，以是同步就须要让cache invalid）。如果在竞争很低的环境下乃至不如 mutex（mutex 有快路径，可以制止进入 futex wait）
  

live lock 的例子：

1. template <typename T>
2. class LockFreeStack {
3. private:
4.   struct Node;
5.   struct CountedNodePtr {
6.     int external_count;
7.     Node* ptr;
8.   };
9.   struct Node {
10.     std::shared_ptr<T> data_;
11.     std::atomic<int> internal_count;
12.     CountedNodePtr next;
13.     Node(T const& data) : data_(std::make_shared<T>(data)), internal_count(0) {}
14.   };
15.   std::atomic<CountedNodePtr> head;
17.   void increase_head_count(CountedNodePtr& old_counter) {
18.     CountedNodePtr new_counter;
19.     do {
20.       new_counter = old_counter;
21.       new_counter.external_count++;
22.     } while(!head.compare_exchange_strong(old_counter, new_counter));
23.     old_counter.external_count = new_counter.external_count;
24.   }
25. public:
26.   LockFreeStack() : head(CountedNodePtr{0, nullptr}) {}
28.   void push(T const& data) {
29.     CountedNodePtr new_node;
30.     new_node.external_count = 1;
31.     new_node.ptr = new Node(data);
32.     new_node.ptr->next = head.load();
33.     while (!head.compare_exchange_weak(new_node.ptr->next, new_node));
34.   }
36.   std::shared_ptr<T> pop() {
37.     auto old_head = head.load();
38.     while(true) {
39.       increase_head_count(old_head);
40.       auto ptr = old_head.ptr;
41.       if (!ptr) {
42.         return std::shared_ptr<T>{};
43.       }
44.       if (head.compare_exchange_strong(old_head, ptr->next)) {
45.         std::shared_ptr<T> res;
46.         res.swap(ptr->data_);
47.         int const count_increase = old_head.external_count - 2;
48.         if (ptr->internal_count.fetch_add(count_increase) == -count_increase) {
49.           delete ptr;
50.         }
51.         return res;
52.       } else if (ptr->internal_count.fetch_sub(1) == 1) {
53.         delete ptr;
54.       }
55.     }
56.   }
57. };

复制代码

会发生：

1. // thread 1
2. mutex1.lock();
4. if (!mutex2.try_lock()) {
5.     mutex1.unlock();
6.     retry();
7. }
9. // thread 2
10. mutex1.lock();
12. if (!mutex2.try_lock()) {
13.     mutex1.unlock();
14.     retry();
15. }

复制代码

线程没有卡死，但是仍旧不停重试。

免责声明：如果侵犯了您的权益，请联系站长及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金.