c# - ConcurrentBag 的预期用途和空作为终止条件

Question

"ConcurrentBag(T) is a thread-safe bag implementation, optimized for scenarios where the same thread will be both producing and consuming data stored in the bag." - MSDN

我有这个确切的用例(多个线程同时消费和生产)，但我需要能够及时有效地确定袋子何时永久变空(我的线程只根据消费的东西生产，而包在线程启动之前用单个元素快速启动)。

我很难找到一种没有全局锁的无竞争条件的有效方法来执行此操作。我相信引入全局锁会抵消使用大部分无锁 ConcurrentBag 的好处。

我的实际用例是“无序”(二进制)树遍历。我只需要访问每个节点，并为每个节点做一些非常简单的计算。我不关心访问它们的顺序。当所有节点都被访问时，算法应该终止。

int taskCount = Environment.ProcessorCount;
Task[] tasks = new Task[taskCount];
var bag = new ConcurrentBag<TreeNode>();
bag.Add(root);
for (int i = 0; i < taskCount; i++)
{
    int threadId = i;
    tasks[threadId] = new Task(() =>
    {
        while(???) // Putting bag.IsEmpty>0 here would be obviously wrong as some other thread could have removed the last node but not yet added the node's "children"
        {
            TreeNode node;
            bool success = bag.TryTake(out node);

            if (!success) continue; //This spinning is probably not very clever here, but I don't really mind it.

            // Placeholder: Do stuff with node

            if (node.Left != null) bag.Add(node.Left);
            if (node.Right != null) bag.Add(node.Right);
        }
    });
    tasks[threadId].Start();
}
Task.WaitAll(tasks);

那么如何为此添加一个有效的终止条件？我不介意当袋子快要变空时条件变得昂贵。

Answer 1

我以前遇到过这个问题。在检查队列之前，我让线程注册为处于等待状态。如果队列为空并且所有其他线程也在等待，我们就完成了。如果其他线程仍然很忙，那么黑客来了，休眠 10ms。我相信通过使用某种同步(可能是 Barrier)无需等待就可以解决这个问题。

代码是这样的:

string Dequeue()
{
    Interlocked.Increment(ref threadCountWaiting);
    try
    {
        while (true)
        {
            string result = queue.TryDequeue();
            if (result != null)
                return result;

            if (cancellationToken.IsCancellationRequested || threadCountWaiting == pendingThreadCount)
            {
                Interlocked.Decrement(ref pendingThreadCount);
                return null;
            }

            Thread.Sleep(10);
        }
    }
    finally
    {
        Interlocked.Decrement(ref threadCountWaiting);
    }
}

用 Barrier 替换 sleep 和计数器维护是可能的。我只是没有打扰，这已经够复杂了。

Interlocked 操作是可伸缩性瓶颈，因为它们基本上是使用硬件自旋锁实现的。所以你可能想在方法的开头插入一个快速路径:

            string result = queue.TryDequeue();
            if (result != null)
                return result;

大多数情况下，将采用快速路径。