MSDN 上 SimplePriorityQueue 示例中的严重错误

Question

我需要使用并发优先级队列，并且我正在考虑调整 MSDN 上 How to: Add Bounding and Blocking Functionality to a Collection 教程中给出的 SimplePriorityQueue<TPriority, TValue> 示例。然而，我对上述样本似乎存在的错误的严重性感到惊讶。有人可以验证这些问题是否真的存在吗？

1) 在TryAdd 和ToArray 之间存在竞争危险，这可能导致ArgumentException 从后者抛出。 TryAdd 方法首先将一个项目添加到内部队列，然后递增m_count 计数器。另一方面，ToArray 首先初始化一个大小为m_count 的新数组，然后将内部队列复制到该数组中。如果在执行ToArray 时调用TryAdd，则ToArray 最终可能会尝试复制比它在数组中分配的空间更多的项目，从而导致CopyTo 调用抛出ArgumentException。

private ConcurrentQueue<KeyValuePair<int, TValue>>[] _queues;
private int m_count;

// ...

// IProducerConsumerCollection members
public bool TryAdd(KeyValuePair<int, TValue> item)
{
    _queues[item.Key].Enqueue(item);
    Interlocked.Increment(ref m_count);
    return true;
}    

public int Count
{
    get { return m_count; }
}

public KeyValuePair<int, TValue>[] ToArray()
{
    KeyValuePair<int, TValue>[] result;

    lock (_queues)
    {
        result = new KeyValuePair<int, TValue>[this.Count];
        // *** context switch here; new item gets added ***
        int index = 0;
        foreach (var q in _queues)
        {
            if (q.Count > 0)
            {
                q.CopyTo(result, index);   // *** ArgumentException ***
                index += q.Count;
            }
        }
        return result;
    }
}

2) GetEnumerator 方法中存在另一个竞争风险：内部队列的更新之间没有同步。

public IEnumerator<KeyValuePair<int, TValue>> GetEnumerator()
{
    for (int i = 0; i < priorityCount; i++)
    {
        foreach (var item in _queues[i])
            yield return item;
    }
}

考虑以下代码 sn-p，它从队列中取出一个项目并以递增的优先级重新添加它：

if (queue.TryTake(out item) && item.Key < maxPriority - 1)
    queue.TryAdd(new KeyValuePair<int, string>(item.Key + 1, item.Value))

如果上面的 sn-p 与枚举同时运行，人们会期望该项目最多出现一次，无论是原始优先级还是递增优先级 - 或者可能根本不出现。人们不会期望该项目在两个优先级中出现两次。但是，由于GetEnumerator 会按顺序迭代其内部队列，因此它不能防止这种跨队列的排序不一致。

3) 公共Count 属性可以返回陈旧值，因为它读取共享m_count 字段时没有任何内存栅栏。如果消费者在不生成自己的内存栅栏的循环中访问此属性，如下所示，尽管项目已被其他线程添加到队列，但他们可能会陷入无限循环。

while (queue.Count == 0) 
{ }

在其他几篇文章中讨论了读取共享变量时对内存栅栏的需求：

• How to correctly read an Interlocked.Increment'ed int field?
• Reading an int that's updated by Interlocked on other threads
• Do concurrent interlocked and reads require a memory barrier or locking?

4) 在_queues 数组的初始化和SimplePriorityQueue 构造函数的完成之间没有内存屏障。当另一个线程上的外部消费者调用TryAdd 并访问_queues 在其初始化完成之前（或在其内存缓存中显示为已完成），可能会出现竞争危险。这将在我关于constructors and memory barriers 的另一个问题中进一步讨论。

5) TryTake 和 ToArray 通过使用 lock 关键字受到保护。除了不足（由于上面讨论的错误）之外，这也违背了设计并发集合的整个目的。鉴于它的缺点，我认为最好的方法是将内部 ConcurrentQueue 结构降级为普通的 Queue，在各处添加锁，然后开始将其视为非并发但线程安全的结构。

Answer 1

我认为这取决于您将如何使用该课程。如果您将自己限制在 TryAdd 和 TryTake（这是 BlockingCollection<T> 所依赖的两个主要内容），那么您应该没有任何问题，并且您将拥有一个非常快的锁定最小优先级队列。

如果您开始使用Count、CopyTo 或任何其他方法，您可能会遇到您指出的问题。