c++ - 无锁堆栈 : visibility issue when checking hazard pointers during pop()?

Question

我正在阅读 Anthony William 的 C++ Concurrency in Action。第 7 章描述了开发无锁堆栈的过程，并说明了使无锁编程变得困难的常见问题。具体来说，第 7.2.3 节(使用危险指针检测无法回收的节点)描述了如何使用危险指针来避免数据竞争并确保其他线程不会删除 一个节点仍然被另一个线程引用。

此代码是该章中说明的 pop() 的迭代之一:

std::shared_ptr<T> pop()
{
  std::atomic<void*>& hp = get_hazard_pointer_for_current_thread();
  node* 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<T> 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;
}

我对这个片段有疑问:

    if(outstanding_hazard_pointers_for(old_head))
    {
      reclaim_later(old_head);
    }
    else
    {
      delete old_head;
    }

危险指针的目的是确保 old_head 在没有其他线程可能仍在使用它时被删除。 outstanding_hazard_pointers_for 的建议实现如下:

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];

bool outstanding_hazard_pointers_for(void* p)
{
  for(unsigned i=0; i < max_hazard_pointers; ++i)
  {
    if(hazard_pointers[i].pointer.load() == p)
    {
      return true;
    }
  }

  return false;
}

基本上，扫描冒险指针数组以检查是否存在指向要查找的节点的指针。我想知道为什么这个操作确实是安全的。执行原子 load()，即使使用顺序一致的顺序，load() 也可能加载过时值。因此，可能找不到 p，并且 pop() 将删除仍在使用的节点。

假设发生以下情况:

• 线程 A 开始执行 pop() 并在执行之前刚好被抢占:

  while(old_head &&
    !head.compare_exchange_strong(old_head,old_head->next));
  

  因此，线程 A 将当前头视为 old_head，并将其保存到其风险指针中。 old_head 将在线程唤醒并尝试通过调用 head.compare_exchange_strong(old_head, old_head->next) 弹出头部时取消引用。

• 线程 B 开始调用 pop() 直到

  if(outstanding_hazard_pointers_for(old_head))
  

  old_head 将是堆栈的当前头部，即线程 A 引用为 old_head 的同一节点。线程 B 将不会 删除 old_head 当且仅当线程 A 的危险指针上的 load() 返回线程 A 存储的最新值。

基本上:我想知道线程 B 是否可以 load() 旧值而不是最新值。换句话说，我不确定为什么它必须返回由线程 A 的 (old_node) 设置的值。

这个推理的缺陷在哪里？我找不到关于为什么另一个线程上的 hp.store(old_head) 会发生在 hazard_pointers[i].pointer.load() 之前的理由。

Answer 1

我回答我自己的问题有两个原因:我认为我接受的答案不是很清楚，并且 JJ15k's comment证实了这种印象。

基本上关键是观察另一个线程是否通过 if(outstanding_hazard_pointers_for(old_head)) 和看到相同的old_head被另一个在执行前被抢占的线程看到 while(old_head && !head.compare_exchange_strong(old_head, old_head->next)) , 它必须执行 head.compare_exchange_strong(old_head, old_head->next)与相同old_head .但是然后(假设 < 表示先发生关系):

thread A: hp.store(old_head)     < 
thread A: old_head = head.load() < 
thread B: head.compare_exchange_strong(old_head, old_head->next)

请记住，线程 B 看到的是相同 old_head我们加载了第一条指令并将它的值交换为 old_head->next .我们仍然在 head.load() 中看到相同的值，这就是线程 A hp.store(old_head) 的原因发生在线程 B 之前 compare_exchange_strong .

所以即将检查是否可以删除危险指针中包含的头部的线程有看到old_head .还要注意 old_head = head.load() 发挥的基础作用(以及包含那些乍一看似乎多余的语句的循环)。没有那个load操作，store 之间不会有发生之前的关系。的 old_head进入hp和 compare_exchange_strong .

我希望这能回答您的问题。