这条逻辑/代码链有什么明显的问题吗？

Question

我正在阅读的一本书是“多处理器编程的艺术，作者 Maurice Herlihy 和 Nir ​​Shavit”。在其中，有一个“无等待”队列（经过一些语言适应后）在线程环境中完美地测试和逻辑功能 - 至少，即使分布在 5 个线程中的 10,000,000 个项目也没有冲突，并且逻辑检查。

（我编辑了队列以在无法获取项目时返回false，而不是抛出异常。代码如下）。

但是，它有一个警告；队列不能增长。粗略的逻辑检查表明，如果不锁定队列，它就无法增长 - 这在某种程度上否定了拥有无锁队列的意义。

因此，目的是创建一个可以增长的无锁（或至少无饥饿锁）队列。

所以：如果我们本质上给每个线程自己的共享队列，以一种不矛盾的方式（并且接受这个问题已经解决的可能性很高，并且为了边做边学更好）：

        WaitFreeQueue<Queue<int>> queues = new WaitFreeQueue<Queue<int>>(threadCount);

        // Dequeue a queue, enqueue an item, enqueue the queue.

        // Dequeue a queue, dequeue an item, enqueue the queue.

还有免等待队列（以前的代码包含在 cmets 中，以防我做出任何重大更改）：

/// <summary>
/// A wait-free queue for use in threaded environments.
/// Significantly adapted from "The Art of Multiprocessor Programming by Maurice Herlihy and Nir Shavit".
/// </summary>
/// <typeparam name="T">The type of item in the queue.</typeparam>
public class WaitFreeQueue<T>
{
    /// <summary>
    /// The index to dequeue from.
    /// </summary>
    protected int head;
    /// <summary>
    /// The index to queue to.
    /// </summary>
    protected int tail;
    /// <summary>
    /// The array to queue in.
    /// </summary>
    protected T[] items;


    /// <summary>
    /// The number of items queued.
    /// </summary>
    public int Count
    {
        get { return tail - head; }
    }


    /// <summary>
    /// Creates a new wait-free queue.
    /// </summary>
    /// <param name="capacity">The capacity of the queue.</param>
    public WaitFreeQueue(int capacity)
    {
        items = new T[capacity];
        head = 0; tail = 0;
    }


    /// <summary>
    /// Attempts to enqueue an item.
    /// </summary>
    /// <param name="value">The item to enqueue.</param>
    /// <returns>Returns false if there was no room in the queue.</returns>
    public bool Enqueue(T value)
    {
        if (tail - head == items.Length)
            // throw new IndexOutOfRangeException();
            return false;
        items[tail % items.Length] = value;
        System.Threading.Interlocked.Increment(ref tail);
        return true;
        // tail++;
    }


    /// <summary>
    /// Attempts to dequeue an item.
    /// </summary>
    /// <param name="r">The variable to dequeue to.</param>
    /// <returns>Returns true if there was an available item to dequeue.</returns>
    public bool Dequeue(out T r)
    {
        if (tail - head == 0)
        // throw new InvalidOperationException("No more items.");
        { r = default(T); return false; }
        r = items[head % items.Length];
        System.Threading.Interlocked.Increment(ref head);
        // head++;
        // return r;
        return true;
    }
}

那么：这行得通吗？如果不是，为什么？如果是这样，是否还有任何可预见的问题？

谢谢。

Answer 1

尝试编写无锁多线程代码很难，你应该把它留给比你或我更了解它的人（并使用例如ConcurrentQueue<T>），或者根本不编写它（并使用锁），如果可能的话。

话虽如此，您的代码存在几个问题：

1. 这不是队列。如果threadCount 为 2，您将项目 1、2 和 3 一个接一个地入队，然后出列，您将获得项目 2！

2. 你不能像你一样先使用一个字段的值然后调用Interlocked.Increment()。想象一下这样的事情：

   1. 在线程 1 上：items[tail % items.Length] = value;
   2. 在线程 2 上：items[tail % items.Length] = value;
   3. 在线程 2 上：Interlocked.Increment(ref head);
   4. 在线程 1 上：Interlocked.Increment(ref head);

   现在，两个线程都排在同一个位置，之后的位置没有改变。这是错误的。

Answer 2

您发布的代码对于入队和出队都不是线程安全的。

入队

• 您有一个空队列
• head = tail = 0
• 您有 2 个线程试图入队。
• 第一个线程执行Enqueue 并在items[tail % items.Length] = value; 之后但在执行联锁增量之前被中断。它现在已将值写入 item[0] 但未递增 tail
• 第二个线程进入Enqueue并执行该方法。现在tail 仍然是0，因此前一个线程写入的值被替换，因为第二个线程也将一个值写入item[0]。
• 两个线程都完成了执行。
• 项目计数正确，但您会丢失数据。

出队

• 假设您在队列中有两个项目item[0] 和items[1]
• head = 0 和 tail = 2
• 您有两个线程试图出队。
• 第一个线程进入Dequeue 并在r = items[head % items.Length]; 之后但在执行互锁增量之前被中断。
• 现在第二个线程进入Dequeue，将返回与head相同的项目，仍然为0
• 之后，您的队列将是空的，但您将阅读一项两次，一项根本没有阅读。