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

本文关键字:什么 问题 代码 | 更新日期: 2023-09-27 18:00:18

我正在读的一本书是莫里斯·赫利希和尼尔·沙维特的《多处理器编程的艺术》。在它中,有一个"无等待"队列,它(经过一点语言调整)在线程环境中进行功能测试和逻辑测试。至少,即使分布在5个线程中的10000000个项目和逻辑检查也没有冲突。

(如果无法获取项目,我编辑了队列以返回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.

以及无等待队列(之前的代码包含在注释中,以防我做出任何突破性的更改):

/// <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;
    }
}

那么:这行吗?如果没有,为什么?如果是的话,还有什么可预见的问题吗?

谢谢。

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

尝试编写无锁的多线程代码很难,您应该把它留给比您或我更了解它的人(并使用例如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);

    现在,两个线程都排队到相同的位置,之后的位置没有改变。这是错误的。

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

排队

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

排队

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