【发布时间】：2010-10-09 22:46:45
【问题描述】：
我正在从事一个学术开源项目，现在我需要在 C# 中创建一个快速阻塞 FIFO 队列。我的第一个实现只是在读取器的信号量中包装了一个同步队列（带动态扩展），然后我决定以以下（理论上更快）的方式重新实现
public class FastFifoQueue<T>
{
private T[] _array;
private int _head, _tail, _count;
private readonly int _capacity;
private readonly Semaphore _readSema, _writeSema;
/// <summary>
/// Initializes FastFifoQueue with the specified capacity
/// </summary>
/// <param name="size">Maximum number of elements to store</param>
public FastFifoQueue(int size)
{
//Check if size is power of 2
//Credit: http://stackoverflow.com/questions/600293/how-to-check-if-a-number-is-a-power-of-2
if ((size & (size - 1)) != 0)
throw new ArgumentOutOfRangeException("size", "Size must be a power of 2 for this queue to work");
_capacity = size;
_array = new T[size];
_count = 0;
_head = int.MinValue; //0 is the same!
_tail = int.MinValue;
_readSema = new Semaphore(0, _capacity);
_writeSema = new Semaphore(_capacity, _capacity);
}
public void Enqueue(T item)
{
_writeSema.WaitOne();
int index = Interlocked.Increment(ref _head);
index %= _capacity;
if (index < 0) index += _capacity;
//_array[index] = item;
Interlocked.Exchange(ref _array[index], item);
Interlocked.Increment(ref _count);
_readSema.Release();
}
public T Dequeue()
{
_readSema.WaitOne();
int index = Interlocked.Increment(ref _tail);
index %= _capacity;
if (index < 0) index += _capacity;
T ret = Interlocked.Exchange(ref _array[index], null);
Interlocked.Decrement(ref _count);
_writeSema.Release();
return ret;
}
public int Count
{
get
{
return _count;
}
}
}
这是我们在教科书中找到的带有静态数组的经典 FIFO 队列实现。它旨在以原子方式递增指针，并且由于我无法使指针在达到（容量-1）时回到零，因此我计算模数。理论上，使用 Interlocked 与在进行增量之前锁定相同，并且由于存在信号量，因此可以有多个生产者/消费者进入队列，但一次只有一个能够修改队列指针。 首先，因为 Interlocked.Increment 先递增，然后返回，我已经明白我仅限于使用后递增值并从数组中的位置 1 开始存储项目。没问题，到了一定值我就回0了
它有什么问题？ 您不会相信，在重负载下运行时，有时队列会返回 NULL 值。我确定，重复一遍，我确定，没有方法将 null 排入队列。这绝对是正确的，因为我尝试在 Enqueue 中进行空检查以确保没有抛出任何错误。我用 Visual Studio 为此创建了一个测试用例（顺便说一句，我使用像 maaaaaaaany 人这样的双核 CPU）
private int _errors;
[TestMethod()]
public void ConcurrencyTest()
{
const int size = 3; //Perform more tests changing it
_errors = 0;
IFifoQueue<object> queue = new FastFifoQueue<object>(2048);
Thread.CurrentThread.Priority = ThreadPriority.AboveNormal;
Thread[] producers = new Thread[size], consumers = new Thread[size];
for (int i = 0; i < size; i++)
{
producers[i] = new Thread(LoopProducer) { Priority = ThreadPriority.BelowNormal };
consumers[i] = new Thread(LoopConsumer) { Priority = ThreadPriority.BelowNormal };
producers[i].Start(queue);
consumers[i].Start(queue);
}
Thread.Sleep(new TimeSpan(0, 0, 1, 0));
for (int i = 0; i < size; i++)
{
producers[i].Abort();
consumers[i].Abort();
}
Assert.AreEqual(0, _errors);
}
private void LoopProducer(object queue)
{
try
{
IFifoQueue<object> q = (IFifoQueue<object>)queue;
while (true)
{
try
{
q.Enqueue(new object());
}
catch
{ }
}
}
catch (ThreadAbortException)
{ }
}
private void LoopConsumer(object queue)
{
try
{
IFifoQueue<object> q = (IFifoQueue<object>)queue;
while (true)
{
object item = q.Dequeue();
if (item == null) Interlocked.Increment(ref _errors);
}
}
catch (ThreadAbortException)
{ }
}
一旦消费者线程得到一个空值，就会计算一个错误。 当使用 1 个生产者和 1 个消费者执行测试时，它成功了。当使用 2 个生产者和 2 个消费者或更多进行测试时，会发生灾难：甚至检测到 2000 次泄漏。我发现问题可能出在 Enqueue 方法中。根据设计合同，生产者只能写入空单元格 (null)，但是通过一些诊断修改我的代码，我发现有时生产者试图在非空单元格上写入，然后被“好”数据占用。
public void Enqueue(T item)
{
_writeSema.WaitOne();
int index = Interlocked.Increment(ref _head);
index %= _capacity;
if (index < 0) index += _capacity;
//_array[index] = item;
T leak = Interlocked.Exchange(ref _array[index], item);
//Diagnostic code
if (leak != null)
{
throw new InvalidOperationException("Too bad...");
}
Interlocked.Increment(ref _count);
_readSema.Release();
}
“太糟糕了”的异常经常发生。但是并发写入引发冲突太奇怪了，因为增量是原子的，并且写入器的信号量只允许与空闲数组单元一样多的写入器。
有人可以帮我吗？如果您能与我分享您的技能和经验，我将不胜感激。
谢谢。
【问题讨论】：
-
有什么理由不使用BlockingCollection<T>？
-
也许 Interlocked.Exchange 在阵列上的行为很有趣并打破了波动性？
-
@dtb：我从来不知道那门课。我的软件主要在 Mono 上工作（我没有提到，因为我不认为它很重要），据我所知 Mono 对 .NET 4 的支持不稳定（但我可能是错的）所以这就是为什么我没有导入 4.0 dll，我没有看到类。无论如何，我会看看它，谢谢 - @pst：这可能是 Windows 上的 .NET 错误吗？我还没有在 Mono 中测试过
-
锁定整个方法并使用 volatile 共享变量时问题是否仍然存在？
标签： c# concurrency queue fifo interlocked-increment