如何通过 C++11 的 CAS 实现 Valois 的队列

Posted 2023-02-16

【中文标题】如何通过 C++11 的 CAS 实现 Valois 的队列

【英文标题】：How to implement Valois' Queue by C++11's CAS

【发布时间】：2018-03-17 03:19:57

【问题描述】：

我想通过 C++11 中提供的原子 CAS 来实现 JD Valois 的论文实现无锁队列。

例如，Valois 的算法定义了一个Enqueue 函数：

Enqueue(x) 
    q = new Record();
    q->value = x;
    q->next = NULL;

    do 
        p = tail;
     while( ! CAS(p->next, NULL, q) ); // 1

    CAS(tail, p, q); // 2

我写了这样的代码

struct Node 
    T * val;
    Node * next;
;
std::atomic<Node *> head, tail;

void enqueue(T * t) 
    Node * nd = new Node();
    nd->val = t;
    nd->next = nullptr;

    std::atomic<Node *> p;
    do 
        p = tail; // 1
     while ( /* CAS1 */  ! std::atomic_compare_exchange_weak(&(p->next), nullptr, nd) );
    /* CAS2 */ std::atomic_compare_exchange_weak(&tail, p.load(), nd)

然后发现CAS1和CAS2这两个cas函数用错了。一方面，p->next 不是std::atomic 的类型，另一方面，atomic_compare_exchange_weak 的第二个参数expected 需要一个指针。在问题Why do C++11 CAS operations take two pointer parameters? 中，cas 函数会将*expected 设置为当前值，这会导致nullptr 的取消引用。而且stmt 1,p = tail也失败了，因为atomic的operator=被删除了。那么如何根据 JD Valois 的论文实现无锁队列呢？

【参考方案1】：

您正确描述了所有问题。下一步只是修复代码。

struct Node 
    T * val;
    std::atomic<Node *> next;
;
std::atomic<Node *> head, tail;

void enqueue(T * t) 
    Node * nd = new Node();
    nd->val = t;
    nd->next = nullptr;

    Node *p, *next;
    do 
        p = tail.load(); // 1
        next = nullptr;
     while (/* CAS1 */ !std::atomic_compare_exchange_weak(&p->next, &next, nd));
    /* CAS2 */ std::atomic_compare_exchange_weak(&tail, &p, nd);