寻找对我的线程安全、无锁队列实现的批评

Question

所以，经过一番研究，我写了一个队列。它使用固定大小的缓冲区，因此它是一个循环队列。它必须是线程安全的，我试图让它无锁。我想知道它有什么问题，因为这些事情我自己很难预测。

这是标题：

template <class T>
class LockFreeQueue
{
public:
    LockFreeQueue(uint buffersize) : buffer(NULL), ifront1(0), ifront2(0), iback1(0), iback2(0), size(buffersize) { buffer = new atomic <T>[buffersize]; }
    ~LockFreeQueue(void) { if (buffer) delete[] buffer; }

    bool pop(T* output);
    bool push(T input);

private:
    uint incr(const uint val)
        {return (val + 1) % size;}

    atomic <T>* buffer;
    atomic <uint> ifront1, ifront2, iback1, iback2;
    uint size;
};

下面是实现：

template <class T>
bool LockFreeQueue<T>::pop(T* output)
{
    while (true)
    {
        /* Fetch ifront and store it in i. */
        uint i = ifront1;

        /* If ifront == iback, the queue is empty. */
        if (i == iback2)
            return false;

        /* If i still equals ifront, increment ifront, */
        /* Incrememnting ifront1 notifies pop() that it can read the next element. */
        if (ifront1.compare_exchange_weak(i, incr(i)))
        {
            /* then fetch the output. */
            *output = buffer[i];
            /* Incrememnting ifront2 notifies push() that it's safe to write. */
            ++ifront2;
            return true;
        }

        /* If i no longer equals ifront, we loop around and try again. */
    }
}

template <class T>
bool LockFreeQueue<T>::push(T input)
{
    while (true)
    {
        /* Fetch iback and store it in i. */
        uint i = iback1;

        /* If ifront == (iback +1), the queue is full. */
        if (ifront2 == incr(i))
            return false;

        /* If i still equals iback, increment iback, */
        /* Incrememnting iback1 notifies push() that it can write a new element. */
        if (iback1.compare_exchange_weak(i, incr(i)))
        {
            /* then store the input. */
            buffer[i] = input;
            /* Incrementing iback2 notifies pop() that it's safe to read. */
            ++iback2;
            return true;
        }

        /* If i no longer equals iback, we loop around and try again. */
    }
}

编辑：我基于 cmets 对代码进行了一些重大修改（感谢 KillianDS 和 n.m.！）。最重要的是，ifront 和 iback 现在是 ifront1、ifront2、iback1 和 iback2。 push() 现在将递增 iback1，通知其他推送线程他们可以安全地写入下一个元素（只要它未满），写入元素，然后递增 iback2。 iback2 是 pop() 检查的所有内容。 pop() 做同样的事情，但使用 ifrontn 索引。

现在，我又一次陷入了“这应该工作......”的陷阱，但我对形式证明或类似的东西一无所知。至少这一次，我想不出它可能会失败的潜在方式。任何建议都值得赞赏，除了“停止尝试编写无锁代码”。

Answer 1

不，它不是线程安全的 - 如果发生事件，请考虑以下顺序：

1. 第一个线程在pop 中完成if (ifront.compare_exchange_weak(i, incr(i))) 并通过调度程序进入睡眠状态。
2. 第二个线程调用 push size 次（刚好足以使 ifront 等于第一个线程中 i 的值）。
3. 第一个线程唤醒。

在这种情况下 pop buffer[i] 将包含最后推送的值，这是错误的。

Answer 2

考虑环绕时存在一些问题，但我认为您的代码的主要问题是它可能会从缓冲区中弹出无效值。

考虑一下：

ifront = iback = 0

Push 被调用并且 CAS 增加 iback 0 -> 1 的值。但是，线程现在在分配缓冲区 [0] 之前停止。

ifront = 0，iback = 1

现在调用 Pop。 CAS 增加 ifront 1 -> 1 并在分配之前读取 buffer[0]。

一个陈旧或无效的值被弹出。

PS。因此，一些研究要求使用 DCAS 或 TCAS（Di 和 Tri CAS）。

Answer 3

处理无锁数据结构的正确方法是编写一个半正式的证明，证明您的设计可以在伪代码中运行。你不应该问“这个无锁代码是线程安全的”，而是“我证明这个无锁代码是线程安全的有什么错误吗？”

只有在您获得伪代码设计有效的正式证明后，您才会尝试实现它。这通常会暴露出必须小心处理的垃圾收集等问题。

您的代码应该是 cmets 中的形式证明和伪代码，其中穿插相对不重要的实现。

验证您的代码是否正确包括理解伪代码、检查证明，然后检查您的代码是否无法映射到您的伪代码和证明。

直接获取代码并尝试检查它是否无锁是不切实际的。证明是正确设计这类东西的重要部分，实际代码是次要的，因为证明是困难的部分。

并且在和当你完成了以上所有，并让其他人验证它，你必须把你的代码通过实际测试，看看你是否有盲点和漏洞，或者不理解您的并发原语，或者您的并发原语中是否存在错误。

如果您对编写半正式证明来设计代码不感兴趣，那么您不应该手动滚动无锁算法和数据结构并将它们放入生产代码中。

确定一堆代码是否“线程安全”是把所有工作量都交给其他人了。您需要有一个论据，为什么您的代码“是线程安全的”以这样一种方式排列，以便其他人尽可能容易地找到其中的漏洞。如果你的论点为什么你的代码“线程安全”的排列方式使得更难找到漏洞，那么即使没有人能在你的代码中发现漏洞，也不能假定你的代码是线程安全的。

您在上面发布的代码是一团糟。它包含注释掉的代码，没有正式的不变量，没有证明行，没有强烈描述为什么它是线程安全的，并且通常不会尝试以一种易于发现的方式将自己显示为线程安全缺陷。因此，即使找不到任何错误，理性的读者也不会认为代码是线程安全的。