C++并发：设计无锁数据结构

只要摆脱锁，实现支持安全并发访问的数据结构，就有可能解决大粒度锁影响并发程度以及错误的加锁方式导致死锁的问题。这种数据结构称为无锁数据结构。

在了解本文时，务必读懂内存次序章节。

在设计无锁数据结构时，需要极为小心谨慎，因为它们的正确实现相当不容易。导致代码出错的情形难以复现。

1 定义和推论

算法和数据结构中，只要采用了互斥，条件变量或future进行同步操作，就称之为阻塞型算法和阻塞型数据结构。

如果应用程序调用某些库函数，发起调用的线程便会暂停运行，即在函数的调用点阻塞，等到另一线程完成某项相关操作，阻塞才会解除，前者才会继续运行。这种库函数的调用被命名为阻塞型调用。

1.1 非阻塞型数据结构

在实践中，我们需要参考下列定义，根据适用的条款，分辨该型别/函数属于哪一类：

1 无阻碍：假定其他线程全部暂停，则目标线程将在有限步骤内完成自己的操作。

2 无锁：如果多个线程共同操作一份数据，那么在有限步骤内，其中某一线程能够完成自己的操作。

3 免等：在某份数据上，每个线程经过有限步骤就能完成自己的操作，即使该份数据同时被其他多个线程所操作。

绝大多数时候无障碍算法并不切实有用，因为其他线程全部暂停这对于一个项目来说是一个非常匪夷所思的场景。

1.2 无锁数据结构

免等和无锁数据结构能够避免线程受饿问题，也就是说，两个并发执行的线程，其中一个按部就班的执行操作，另一个总是在错误的时机开始执行操作，导致被迫中止，反复开始，试图完成操作。

1.3 免等的数据结构

是具备额外功能的无锁数据结构，如果它被多个线程访问，不论其他线程上发生了什么，每个线程都能在有限的步骤内完成自己的操作。若多个线程之间存在冲突，导致某算法无限制地反复尝试执行操作，那它就是免等算法（比如使用while循环进行的一些操作，在里面执行比较-交换操作）。

1.4 无锁数据结构的优点和缺点

1.4.1 优点

本质上，使用无锁数据结构的首要原因是：最大限度地实现并发。

基于锁的实现往往导致线程需要阻塞，在无锁数据结构上，总是存在某个线程能执行下一步操作。免等数据结构则完全无需等待，但是难以实现，很容易写成自旋锁。

还有一点是，代码健壮性。

假设数据结构的写操作受锁保护，如果线程在持锁期间终止，那么该数据结构仅完成了部分改动，且此后无从修补。但是，若某线程操作无锁数据结构时意外终结，则丢失的数据仅限于它持有的部分，其他数据依然完好，能被别的线程进行处理（不会阻塞等待锁）

1.4.2 缺点。需要注意的地方

1.4.2.1 不变量相关

力求保持不变量成立，或选取别的可以一直成立的不变量作为替代。

1.4.2.2 留心内存次序约束

1.4.2.3 数据修改使用原子操作

1.4.2.4 就其他线程所见，各项修改步骤次序正确

1.4.2.5 避免活锁

假设两个线程同时更改同一份数据结构，若它们所做的改动都导致对方从头开始操作，那双方就会反复循环，不断重试，这种现象称为活锁。它的出现与否完全取决于现成的调度次序，故往往只会短暂存在。因此它们仅降低程序性能，不会导致严重问题。也因此可能会导致提高了操作同一个数据结构的并发程度，缩短了单个线程因等待消耗的时间，却降低了整体的性能。

1.4.2.6 缓存乒乓

并且，如果多个线程访问相同的原子变量，硬件必须在线程之间同步数据，这还会造成缓存乒乓现象，导致严重的性能损耗。

2 无锁数据结构范例

无锁数据结构依赖原子操作和内存次序约束（作用是令其他线程按正确的内存次序见到数据操作的过程），默认内存次序std::memory_order_seq_cst最易于分析和推理（全部该次序的操作形成确定且唯一的总序列）

2.1 实现线程安全的无锁栈

需要保证：

1 一旦某线程将一项数据加入栈容器，就能立即安全地被另一线程取出

2 只有唯一一个线程能获取该项数据

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;node* next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::atomic<node*> head;
public:void push(T const& data) {node* const new_node = new node(data);new_node->next = head.load();while (!head.compare_exchange_weak(new_node->next, new_node));}std::shared_ptr<T> pop() {node* old_head = head.load();while(old_head && !head.compare_exchange_weak(old_head, old_head->next));return old_head ? old_head->data : std::shared_ptr<T>();}
};

比较交换操作：如果head指针与第一个参数new_node->next所存储值相同，将head改为指向第二个参数new_node，compare_exchange_weak返回true。如果head指针与第一个参数new_node->next所存储值不同，表示head指针被其他线程修改过，第一个参数new_node->next就被更新成head指针的当前值，并且compare_exchange_weak返回false，让循环继续。

上述代码虽然是无锁实现，但是却是非免等的，如果compare_exchange_weak总是false，理论上push和pop中的while循环要持续进行。

2.2 制止内存泄漏：在无锁数据结构中管理内存

本质问题是：若要删除某节点，必须先行确认，其他线程并未持有指向该节点的指针。

对于上述实现，若有多个线程同时调用pop，需要采取措施判断何时删除节点。

可以维护一个等待删除链表。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;node* next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::atomic<node*> head;std::atomic<unsigned> threads_in_pop;std::atomic<node *> to_be_deleted;static void delete_nodes(node* nodes) {while (nodes) {node* next = nodes->next;delete nodes;nodes = next;}}void try_reclaim(node* old_head) {if (threads_in_pop == 1) {node* nodes_to_delete = to_be_deleted.exchange(nullptr);if (!--threads_in_pop) {delete_nodes(nodes_to_delete);} else if (nodes_to_delete) {chain_pending_nodes(nodes_to_delete);}delete old_head;} else {chain_pending_node(old_head);--threads_in_pop;}}void chain_pending_nodes(node* nodes) {node* last = nodes;while (node* const next = last->next) {last = next;}chain_pending_nodes(nodes, last);}void chain_pending_nodes(node* first, node* last) {last->next = to_be_deleted;while (!to_be_deleted.compare_exchange_weak(last->next, first));}void chain_pending_node(node* n) {chain_pending_nodes(n, n);}  
public:void push(T const& data) {node* const new_node = new node(data);new_node->next = head.load();while (!head.compare_exchange_weak(new_node->next, new_node));}std::shared_ptr<T> pop() {++threads_in_pop;node* old_head = head.load();while(old_head && !head.compare_exchange_weak(old_head, old_head->next));std::shared_ptr<T> res;if (old_head) {res.swap(old_head->data);}try_reclaim(old_head);return res;}
};

2.3 运用风险指针检测无法回收的节点

术语“风险指针”是一种技法，得名缘由是：若某节点仍被其他线程指涉，而我们依然删除它，此举便成了“冒险”动作。删除目标节点后，别的线程还持有指向它的引用，还通过这一引用对其进行访问，便会导致程序产生未定义行为。

上述机制产生的基本思想：假设当前线程要访问某对象，而它却即将被别的线程删除，那就让当前线程设置指涉目标对象的风险指针，以通知其他线程删除该对象将产生实质风险。若程序不再需要该对象，风险指针被清零。

#include <atomic>
#include <thread>
unsigned const max_hazard_pointers=100;
struct hazard_pointer {std::atomic<std::thread::id> id;std::atomic<void*> pointer;
};hazard_pointer hazard_pointers[max_hazard_pointers];class hp_owner {hazard_pointer* hp;public:hp_owner(hp_owner const&)=delete;hp_owner operator=(hp_owner const&)=delete;hp_owner(): hp(nullptr) {for (unsigned i = 0; i < max_hazard_pointers; ++i) {std::thread::id old_id;if (hazard_pointers[i].id.compare_exchange_strong(old_id, std::this_thread::get_id())) {hp = &hazard_pointers[i];break;}}if (!hp) {throw std::runtime_error("No hazard");}}std::atomic<void*>& get_pointer() {return hp->pointer;}~hp_owner() {hp->pointer.store(nullptr);hp->id.store(std::thread::id());}
};std::atomic<void*>& get_hazard_pointer_for_current_thread() {thread_local static hp_owner hazard;return hazard.get_pointer();
}

2.4 借引用计数检测正在使用中的节点

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;node* next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::shared_ptr<node> head;
public:void push(T const& data) {std::shared_ptr<node> const new_node = std::make_shared<node>(data);new_node->next = std::atomic_load(&head);while (!std::atomic_compare_exchange_weak(&head, &new_node->next, new_node));}std::shared_ptr<T> pop() {std::shared_ptr<node> old_head = std::atomic_load(&head);while (old_head && std::atomic_compare_exchange_weak(&head, &old_head, std::atomic_load(&old_head->next)));if (old_head) {std::atomic_store(&old_head->next, std::shared_ptr<node>());return old_head->next;}return std::shared_ptr<T>();}~lock_free_stack() {while (pop());}
};

引用计数针对各个节点分别维护一个计数器，随时了解访问它的线程数目。

std::shared_ptr的引用计数在这里无法借鉴，因为他的原子特性不一定通过无锁机制实现，若强行按照无锁方式实现该指针类的原子操作，很可能造成额外开销。

2.4.1 std::experimental::atomic_shared_ptr<T>

std::shared_ptr无法结合std::atomic<>使用，原因是std::shared_ptr<T>并不具备平实拷贝语义。但是std::experimental::atomic_shared_ptr<T>支持，因此可以正确的处理引用计数，同时令操作原子化。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node {std::shared_ptr<T> data;std::experimental::atomic_shared_ptr<node> next;node(T const& data_) : data(std::make_shared<T>(data_)) {}};std::experimental::atomic_shared_ptr<node> head;
public:void push(T const& data) {std::shared_ptr<node> const new_node = std::make_shared<node>(data);new_node->next = std::atomic_load(&head);while (!std::atomic_compare_exchange_weak(&head, &new_node->next, new_node));}std::shared_ptr<T> pop() {std::shared_ptr<node> old_head = std::atomic_load(&head);while (old_head && std::atomic_compare_exchange_weak(&head, &old_head, std::atomic_load(&old_head->next)));if (old_head) {std::atomic_store(&old_head->next, std::shared_ptr<node>());return old_head->data;}return std::shared_ptr<T>();}~lock_free_stack() {while (pop());}
};

2.4.2 内、外部计数器进行引用计数

一种经典的实现是，使用两个计数器：内、外部计数器各一。两个计数器之和即为节点的总引用数目。

外部计数器与节点的指针组成结构体，每当指针被读取，外部计数器自增。

内部计数器位于节点之中，随着节点读取完成自减。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node;struct counted_node_ptr {int external_count;node* ptr;};struct node {std::shared_ptr<T> data;std::atomic<int> internal_count;counted_node_ptr next;node(T const& data_) : data(std::make_shared<T>(data_)), internal_count(0) {}};std::atomic<counted_node_ptr> head;void increase_head_count(counted_node_ptr& old_counter) {counted_node_ptr new_counter;do {new_counter = old_counter;++new_counter.external_count;} while (!head.compare_exchange_strong(old_counter, new_counter));old_counter.external_count = new_counter.external_count;}
public:std::shared_ptr<T> pop() {counted_node_ptr old_head = head.load();for (;;) {increase_head_count(old_head);node* const ptr = old_head.ptr;if (!ptr) {return std::shared_ptr<T>();}if (head.compare_exchange_strong(old_head, ptr->next)) {std::shared_ptr<T> res;res.swap(ptr->data);int const count_increase = old_head.external_count - 2;if (ptr->internal_count.fetch_add(count_increase) == -count_increase) {delete ptr;}return res;} else if (ptr->internal_count.fetch_sub(1) == 1) {delete ptr;}}}void push(T const& data) {counted_node_ptr new_node;new_node.ptr = new node(data);new_node.external_count = 1; // head指针本身算作一个外部引用new_node.ptr->next = head.load();while (!head.compare_exchange_weak(new_node.ptr->next, new_node));}~lock_free_stack() {while (pop());}
};

结构体counted_node_ptr的尺寸足够小，如果硬件平台支持双字比较-交换操作，那么std::atomic<counted_node_ptr>就属于无锁数据。若不支持，那么std::atomic<>涉及的结构体的尺寸过大，无法直接通过原子指令操作，便会采用互斥来保证操作原子化。使得“无锁”数据结构和算法成为基于锁的实现。

如果想要缩小结构体counted_node_ptr的尺寸，可以采取另一种方法替代：假定在硬件平台上，指针型别有空余的位。（例如，硬件寻址空间只有48位，指针型别的大小是64位），他们可以用来放置计数器，借此将counted_node_ptr结构体缩成单个机器字长。

使用分离引用计数的原因：我们通过外部引用计数的自增来保证，在访问目标节点的过程中，其指针依然安全有效。（先自增，再读取，被指涉后自增的值保护了节点不被删除）

具体流程详解见《cpp 并发实战》p236

以上实例使用的内存次序是std::memory_order_seq_cst。同步开销较大，下面对于内存次序进行优化。

2.5 为无锁栈容器施加内存模型

需要先确认各项操作哪些存在内存次序关系。

1 next指针只是一个未被原子化的普通对象，所以为了安全读取其值，存储操作必须再载入操作发生之前，前者由压入数据的线程执行，后者由弹出数据的线程执行。

#include <atomic>
#include <memory>template<typename T>
class lock_free_stack {
private:struct node;struct counted_node_ptr {int external_count;node* ptr;};struct node {std::shared_ptr<T> data;std::atomic<int> internal_count;counted_node_ptr next;node(T const& data_) : data(std::make_shared<T>(data_)), internal_count(0) {}};std::atomic<counted_node_ptr> head;void increase_head_count(counted_node_ptr& old_counter) {counted_node_ptr new_counter;do {new_counter = old_counter;++new_counter.external_count;} while (!head.compare_exchange_strong(old_counter, new_counter, std::memory_order_acquire, std::memory_order_relaxed));old_counter.external_count = new_counter.external_count;}
public:std::shared_ptr<T> pop() {counted_node_ptr old_head = head.load(std::memory_order_relaxed);for (;;) {increase_head_count(old_head);node* const ptr = old_head.ptr;if (!ptr) {return std::shared_ptr<T>();}if (head.compare_exchange_strong(old_head, ptr->next, std::memory_order_relaxed)) {std::shared_ptr<T> res;res.swap(ptr->data);int const count_increase = old_head.external_count - 2;if (ptr->internal_count.fetch_add(count_increase, std::memory_order_relaxed) == -count_increase) { // ?delete ptr;}return res;} else if (ptr->internal_count.fetch_add(-1, std::memory_order_relaxed) == 1) {ptr->internal_count.load(std::memory_order_acquire);delete ptr;}}}void push(T const& data) {counted_node_ptr new_node;new_node.ptr = new node(data);new_node.external_count = 1;new_node.ptr->next = head.load(std::memory_order_relaxed);while (!head.compare_exchange_weak(new_node.ptr->next, new_node, std::memory_order_release, std::memory_order_relaxed));}~lock_free_stack() {while (pop());}
};

这里的push，唯一的原子操作是compare_exchange_weak，如果需要在线程间构成先行关系，则代码需要一项释放操作，因此compare_exchange_weak必须采用std::memory_order_release或者更严格的内存次序。

若compare_exchange_weak执行失败，则指针head和new_node均无变化，代码继续执行，这种情况下使用memory_order_relaxed即可。

（没懂，为什么是这两个次序）

这里的pop，访问next指针前进行了额外操作。也就是先调用了increase_head_count()，该操作收到memory_order_acquire或者更严格的内存次序约束。因为这里通过原子操作获取的head指针旧值访问next指针。对原子操作失败的情况则使用宽松次序。

因为push中的存储行为是释放操作，pop，increase_head这里的是获取操作，因此存储行为和载入操作同步，构成先行关系。因此，对于push中的成员指针ptr的存储操作先行发生，然后pop才会在increase_head_count()中访问ptr->next，代码符合线程安全。

（push中的head.load不影响上述内存次序关系的分析）

剩余的很多没看懂，在《cpp并发实战》p240，有空多看看。

2.6 实现线程安全的无锁队列

对于队列，其pop和push分别访问不同的部分。

#include <atomic>
#include <memory>
#include <mutex>template<typename T>
class lock_free_queue {
private:struct node {std::shared_ptr<T> data;node* next;node() : next(nullptr) {}};std::atomic<node*> head;std::atomic<node*> tail;std::unique_ptr<node> pop_head() {node* const old_head = head.load();if (head.get() == tail) {return nullptr;}head.store(old_head->next);return old_head;}public:lock_free_queue() : head(new node), tail(head.load()) {}lock_free_queue(const lock_free_queue& other) = delete;lock_free_queue& operator=(const lock_free_queue& other) = delete;~lock_free_queue() {while (node* const old_head = head.load()) {head.store(old_head->next);delete old_head;}}std::shared_ptr<T> pop() {node* old_head = pop_head();if (!old_head) {return std::shared_ptr<T>();}std::shared_ptr<T> const res(old_head->data);delete old_head;return res;}void push(T new_value) {std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));node* p = new node;node* const old_tail = tail.load();old_tail->data.swap(new_data);old_tail->next = p;tail.store(p);}
};

tail指针的存储操作store（push的最后一句）和载入操作load（pop_head的if里面的条件判断）存在同步。

但是若是由多个线程进行操作，上述代码就不可行。

其余细节具体实现见书。