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

这是标题:

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. */
    }
}

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

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

最佳答案

接近无锁数据结构的正确方法是编写一个半正式的证明，证明您的设计在伪代码中有效。您不应该问“这个无锁代码线程安全吗”，而是“我证明这个无锁代码是线程安全的证据有任何错误吗？”

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

你的代码应该是正式的证明和注释中的伪代码，中间穿插着相对不重要的实现。

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

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

并且在和同时您已经完成了上述所有操作，并让其他人对其进行了验证，您必须对您的代码进行实际测试以查看您是否一个盲点和一个洞，或者不理解你的并发原语，或者你的并发原语中有错误。

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

确定一堆代码是否“是线程安全的”是将所有工作负载都推给其他人。你需要争论为什么你的代码“是线程安全的”以这样一种方式排列，以便其他人尽可能容易地找到其中的漏洞。如果您论证为什么您的代码“是线程安全的”，其排列方式使得更难找到漏洞，则您的代码不能被假定为线程安全的，即使没有人可以在您的代码中发现漏洞。

您上面发布的代码一团糟。它包含注释掉的代码，没有正式的不变量，没有行的证明，没有对为什么它是线程安全的强有力的描述，并且通常不会试图以一种容易被发现的方式来证明自己是线程安全的缺陷。因此，任何理性的读者都不会认为代码是线程安全的，即使他们找不到其中的任何错误。

关于c++ - 寻找对我的线程安全、无锁队列实现的批评，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/21561155/