简单的无锁栈 c++11答案

【问题标题】：Simple lock free stack c++11简单的无锁栈 c++11
【发布时间】：2015-01-01 01:24:19
【问题描述】：

我已经看到了几个过于复杂（在我看来很明显）在 C++ 中实现无锁堆栈（使用像 here 这样的标签），我想出了一个我认为是简单且仍然有效的实现。由于我在任何地方都找不到这个实现（我已经看到 Push 函数的实现类似于我所做的但不是 Pop），我猜测它在某种程度上是不正确的（很可能在 ABA 案例中失败）：

template<typename Data>
struct Element
{
  Data mData;
  Element<Data>* mNext;
};

template<typename Data>
class Stack
{
 public:
  using Obj = Element<Data>;
  std::atomic<Obj*> mHead;

  void Push(Obj *newObj)
  {
    newObj->mNext = mHead.load();
    //Should I be using std::memory_order_acq_rel below??
    while(!mHead.compare_exchange_weak(newObj->mNext, newObj));
  }

  Obj* Pop()
  {
    Obj* old_head = mHead.load();
    while (1)
    {
      if (old_head == nullptr)
        return nullptr;
      //Should I be using std::memory_order_acq_rel below??
      if(mHead.compare_exchange_weak(old_head, old_head->mNext)) ///<<< CL1
        return old_head;
    }
  }
};

我假设 Push 和 Pop 的调用者将负责内存分配和释放。另一种选择是使上述 Push 和 Pop 私有方法，并具有新的公共函数来处理内存分配并在内部调用这些函数。我相信这个实现中最棘手的部分是我用“CL1”标记的那一行。我认为它是正确的并且在 ABA 案例中仍然有效的原因如下：

假设 ABA 案例确实发生了。这意味着“CL1”处的 mHead 将等于 old_head 但他们都指向的对象实际上与我将 mHead 分配给它时最初指向的那个 old_head 不同。但是，我认为即使它是一个不同的对象，我们仍然可以，因为我们知道它是一个有效的“头”。 old_head 指向与 mHead 相同的对象，因此它是堆栈的有效头，这意味着 old_head->mNext 是有效的下一个头。所以，更新 mHead 为 old_head->mNext 还是正确的！

总结一下：

如果 mHead != old_head（另一个线程抢占我们并更改了 mHead）-> old_head 被更新为新的 mHead，我们再次开始循环。
[NON-ABA] 如果 mHead == old_head -> 简单情况，将 mHead 更新为 old_head->next (==mHead->mNext) 并返回 old_head。
[ABA] 如果 mHead == old_head -> 按上述说明工作。

那么，我的实现是否有效？我错过了什么？

【问题讨论】：

您缺少一个测试工具，其中有几十个线程不断推送和弹出（以及某种检测错误的方法）。
@DietmarKühl compare_exchange_weak 在比较失败时将其第一个参数的值更新为观察值，因此循环newObj->mNext 的每次行程都设置为mHead 的值。首先将其设置为 mHead.load() 是一个非常小的优化，nullptr 将同样正确，但即使没有争用，也需要额外的循环。
@Casey：感谢您指出这一点！我可能最好不要评论无锁代码，因为我没有实际经验（即，我删除了我之前的评论......）。
这不应该去代码审查吗？

标签： c++ multithreading c++11 stack lock-free

【解决方案1】：

一般来说，如果数据类型是按位规则的，则实现可能是正确的，如果只是半规则（或相等不是按位），则不可避免地会受到 ABA 的影响。

如果数据类型是按位规则的，那么 CAS 操作就足够了，例如对于整数。 ABA 不是问题，因为 A 等于 A，故事结束。

Harris 算法似乎允许非按位正则类型提供集合实现，但以牺牲无限性能为代价，更糟糕的是，它引发了如何分配列表节点的问题。这意味着要解决的核心问题是提供一个 O(1) 无锁分配器，并且这样的数据结构不存在，因为所涉及的指针只是半正则的。特别是，当指针仅将字节数组标识为内存块时，它是规则的，但当它标识链表的节点时，它不再是规则的，因为指针相等并不能确保嵌入的下一个指针相等。

没有可能的解决方法。我怀疑 DCAS 会有所帮助，我也怀疑将恒定时间限制放宽到线性也有。

对于不可抢占的线程，自旋锁将工作，只要它保护的关键部分是有限的时间。您可以在 MacOS 和 RT-Linux 上获得这些。

【讨论】：

【解决方案2】：

如果需要跨平台，这个lfstack可以做跨平台构建，是c native内置的

例子：-

int* int_data;
lfstack_t mystack;

if (lfstack_init(&mystack) == -1)
    return -1;

/** Wrap This scope in other threads **/
int_data = (int*) malloc(sizeof(int));
assert(int_data != NULL);
*int_data = i++;
/*PUSH*/
 while (lfstack_push(&mystack, int_data) == -1) {
    printf("ENQ Full ?\n");
}

/** Wrap This scope in other threads **/
/*POP*/
while  ( (int_data = lfstack_pop(&mystack)) == NULL) {
    printf("POP EMPTY ..\n");
}

// printf("%d\n", *(int*) int_data );
free(int_data);
/** End **/

lfstack_destroy(&mystack);

【讨论】：

【解决方案3】：

ABA 在以下情况下发生：

线程 A 读取 old_head->mNext 并在调用 compare_exchange_weak 之前阻塞。
线程 B 弹出当前节点，推送一些其他节点，然后将原始节点推回堆栈。
线程 A 解除阻塞，成功完成 compare_exchange_weak，因为 mHead 具有相同的值，但将旧的 mNext 值存储为新的 mHead。

See this answer for more details，您同时遇到问题 #2（mNext 上的数据竞争）和问题 #3 (ABA)。

【讨论】：