c++ - 无锁双向链表的原子操作

Question

我正在写一个基于这些论文的无锁双向链表:

《基于引用计数的高效可靠的无锁内存回收》Anders Gidenstam，IEEE 成员(member)，Marina Papatriantafilou，H˚ akan Sundell 和 Philippas Tsigas

“无锁双端队列和双向链表”Håkan Sundell，Philippas Tsigas

对于这个问题我们可以搁置第一篇paper。

在这篇论文中，他们使用一种巧妙的方式在一个词中存储一个删除标志和一个指针。(更多信息 here)

论文中这部分的伪代码:

union Link
    : word
    (p,d): {pointer to Node, boolean} 

structure Node
    value: pointer to word
    prev: union Link
    next: union Link

还有我上面伪代码的代码:

template< typename NodeT >
struct LockFreeLink
{
public:
    typedef NodeT NodeType;

private:

protected:
    std::atomic< NodeT* > mPointer;

public:
    bcLockFreeLink()
    {
        std::atomic_init(&mPointer, nullptr);
    }
    ~bcLockFreeLink() {}

    inline NodeType* getNode() const throw()
    {
        return std::atomic_load(&mPointer, std::memory_order_relaxed);
    }
    inline std::atomic< NodeT* >* getAtomicNode() const throw()
    {
        return &mPointer;
    }
};

struct Node : public LockFreeNode
{
    struct Link : protected LockFreeLink< Node >
    {
        static const int dMask = 1;
        static const int ptrMask = ~dMask;

        Link() { } throw()
        Link(const Node* pPointer, bcBOOL pDel = bcFALSE) throw()
        { 
            std::atomic_init(&mPointer, (reinterpret_cast<int>(pPointer) | (int)pDel)); 
        }

        Node* pointer() const throw() 
        { 
            return reinterpret_cast<Node*>(
                std::atomic_load(&data, std::memory_order_relaxed) & ptrMask); 
        }
        bool del() const throw() 
        { 
            return std::atomic_load(&data, std::memory_order_relaxed) & dMask; 
        }
        bool compareAndSwap(const Link& pExpected, const Link& pNew) throw() 
        { 
            Node* lExpected = std::atomic_load(&pExpected.mPointer, std::memory_order_relaxed);
            Node* lNew = std::atomic_load(&pNew.mPointer, std::memory_order_relaxed);

            return std::atomic_compare_exchange_strong_explicit(
                &mPointer,
                &lExpected,
                lNew,
                std::memory_order_relaxed,
                std::memory_order_relaxed); 
        }

        bool operator==(const Link& pOther) throw() 
        { 
            return std::atomic_load(data, std::memory_order_relaxed) == 
                std::atomic_load(pOther.data, std::memory_order_relaxed); 
        }
        bool operator!=(const Link& pOther) throw() 
        { 
            return !operator==(pOther); 
        }
    };

    Link mPrev;
    Link mNext;
    Type mData;

    Node() {};
    Node(const Type& pValue) : mData(pValue) {};
};

在这篇论文中有这个函数可以将链接的删除标记设置为true:

procedure SetMark(link: pointer to pointer to Node)
    while true do
       node = *link;
       if node.d = true or CAS(link, node, (node.p, true)) then break;

我的这个函数的代码:

void _setMark(Link* pLink)
{
    while (bcTRUE)
    {
        Link lOld = *pLink;
        if(pLink->del() || pLink->compareAndSwap(lOld, Link(pLink->pointer(), bcTRUE)))
            break;
    }
}

但我的问题出在 compareAndSwap 函数中，我必须在其中比较和交换三个原子变量。有关问题的信息是 here

(实际上比较和交换函数中的 new 变量并不重要，因为它是线程本地的)

现在我的问题是:我如何编写 compareAndSwap 函数来比较和交换三个原子变量，或者我哪里出错了？

(不好意思问了这么久)

编辑:

类似的问题在内存管理器论文中:

function CompareAndSwapRef(link:pointer to pointer toNode,
old:pointer toNode, new:pointer toNode):boolean
    if CAS(link,old,new) then
        if new=NULL then
            FAA(&new.mmref,1);
            new.mmtrace:=false;
    if old=NULLthen FAA(&old.mmref,-1);
    return true;
return false; 

这里我必须再次比较和交换三个原子变量。(请注意，我的参数是 Link 类型，我必须比较和交换 Link 的 mPointer)

Answer 1

除非你可以让你正在比较/交换的三个数据项适合两个指针大小的元素，否则你不能用比较和交换来做到这一点(当然不能在 x86 上，我还没有听说过任何其他具有这种东西的机器体系结构)。

如果您依赖于存储在(至少)与偶数字节地址对齐的地址上的数据，则可以在删除元素时使用按位或设置最低位。过去，人们一直使用地址的上半部分来存储额外的数据，但至少在 x86-64 中，这是不可能的，因为地址的上半部分必须是“规范的”，这意味着任何地址位高于“可用限制”(由处理器架构定义，目前为 48 位)，必须与可用限制的最高位相同(因此，与位 47 相同)。

编辑:这部分代码完全符合我的描述:

    static const int dMask = 1;
    static const int ptrMask = ~dMask;

    Link() { } throw()
    Link(const Node* pPointer, bcBOOL pDel = bcFALSE) throw()
    { 
        std::atomic_init(&mPointer, (reinterpret_cast<int>(pPointer) | (int)pDel)); 
    }

    Node* pointer() const throw() 
    { 
        return reinterpret_cast<Node*>(
            std::atomic_load(&data, std::memory_order_relaxed) & ptrMask); 
    }

它使用最低位来存储pDel 标志。

您应该能够通过使用 cmpxchg16b 的形式(在 x86 上)对双链表执行此操作。在 Windows 系统中，这将是 _InterlockedCompareExchange128 .在 gcc 中(对于 Unix 类型的操作系统，例如 Linux/MacOS)，您需要首先从两个指针构造一个 int128。如果您正在编译 32 位代码，您可能需要为 Windows 和 Unix 操作系统创建一个 64 位 int。