myo*_*dpa 5 c++ multithreading reference-counting lock-free
我正在阅读 C++ concurrency in action 2。它为无锁堆栈引入了拆分引用计数。
一种可能的技术涉及对每个节点使用不是一个而是两个引用计数：内部计数和外部计数。这些值的总和就是对节点的引用总数。外部计数与指向节点的指针一起保存，每次读取指针时都会增加。当阅读器完成节点时，它会减少内部计数。读取指针的简单操作将在完成时使外部计数增加一，内部计数减一。
上面的短语解释了拆分引用计数的简要概念。听起来外部计数总是增加而内部计数总是减少。
当不再需要外部计数/指针配对时（节点不再能从多个线程可访问的位置访问），内部计数增加外部计数值减一，并丢弃外部计数器。一旦内部计数等于零，就没有对节点的未完成引用，可以安全地删除它。使用原子操作更新共享数据仍然很重要。现在让我们看看一个无锁堆栈的实现，它使用这种技术来确保只有在安全的情况下才回收节点。
但是在上面的短语中，当节点不再可访问时，内部计数应该增加外部计数的值减一。我对这个解释感到非常困惑。
(1) 使用内部和外部计数的确切目的是什么？
(2) 为什么它需要两个引用计数而不是一个？
编辑：我从书中添加了下面的示例代码。
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:
~lock_free_stack() {
while (pop())
;
}
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();
while (!head.atomic_compare_exchange_weak(new_node.ptr->next, new_node,
std::memory_order_release,
std::memory_order_relaxed))
;
}
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_release) == -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;
}
}
}
};
分割引用计数减少了争用。在此实现中，添加external_count-2到internal_count分隔了引用计数的两个阶段：在第一阶段中，引用计数被分割，而在第二阶段中，引用计数internal_count仅存在。
请阅读下面的详细解释。
让我们从书中的无锁堆栈案例开始，为什么我们需要引用计数。如果该节点已被删除，我们需要阻止访问该next节点的字段head。head当不再使用时，我们要删除指向的节点。为了实现这一点，我们使用引用计数来延迟删除该节点。
重要的是，节点只能由头（当节点在列表中的第一个时）或前一个节点的下一个指针（当节点在列表的中间或末尾时）引用，但不能同时通过两者引用 -因为不存在头指向链表中间或末尾节点的情况。这就是为什么我们不需要从多个计数节点指针指向单独的控制块（如shared_ptr）。所以，我们可以将counter直接放入head指针中。
虽然我们需要保护对 only 指向的节点字段的访问head.ptr（实际上仅对字段next），但将计数器放入头指针是不够的。我们还必须在指针中放入计数器next- 以继续保护节点不被删除，即使该节点在其之前被推送和弹出。如果我们将节点 B 压入正在被其他线程 Tb 弹出的节点 A 之前，然后在线程 Tb 之前弹出节点 B，则我们将头指针计数器保存并恢复到下一个指针中。
增加计数器和加载当前指针应该是单个原子操作，以确保取消引用的指针的计数器增加。
我们将计数器分为内部计数器和外部计数器有两个原因：
当计数器递减时，我们不需要对计数器和指针进行原子操作。原子地改变计数器就足够了。如果我们有一个计数器，我们就必须像在increase_head_count 中那样在循环中递减指针。
使用两个原子计数器可以减少争用，但会增加代码复杂性、内存使用量，并且需要添加两个计数器进行检查。
为了保护节点不被删除，我们增加external_count。Internal_count 位于 counted 指针所指向的节点内部。
一旦在弹出操作期间通过compare_exchange操作将头指针移动到下一个节点，没有线程可以增加指针计数器，因为已经读取了前一个头的线程将失败compare_exchange操作，而尚未读取头的线程将永远不会读取此头，因为头已经移动到下一个节点。这就是不再需要外部计数/指针配对的原因。其值被添加到internal_count。此后，计数器不再拆分：我们现在使用单个计数器internal_count，它将所有计数器的增加和减少聚合在一个变量中。
我们可以将其视为两个阶段：在第一阶段，我们的计数器被拆分（分为external_count和internal_count），在第二阶段，我们的计数器被合并为一个变量——internal_count。在第一阶段，我们需要将external_count添加到internal_count以获取真实的计数器值。在第二阶段，我们必须仅使用internal_count，因为external_count有一些现在没有任何意义的旧值（正如书中所说，现在“外部计数器被丢弃”）。
当external_count与internal_count相加时，external_count值减2。为简单起见，该操作可以分为三个：
比较internal_count.fetch_add(external_count - 2)to的结果2 - external_count只是将结果internal_count 与零进行比较的原子方法。
在代码中，external_count在添加到internal_count之前先减2，而在书本中“内部计数增加了外部计数减1的值”。我认为这是文本中的一个错误（可能，作者最初计划将internal_count减1，与添加分开external_count-1）。