c# - BlockingCollection(T).GetConsumingEnumerable() 如何抛出 OperationCanceledException？

Question

我正在使用 BlockingCollection 来实现任务调度程序，基本上:

public class DedicatedThreadScheduler : TaskScheduler, IDisposable
{
    readonly BlockingCollection<Task> m_taskQueue = new BlockingCollection<Task>();

    readonly Thread m_thread;


    public DedicatedThreadScheduler()
    {
        m_thread = new Thread(() =>
        {
            foreach (var task in m_taskQueue.GetConsumingEnumerable())
            {
                TryExecuteTask(task);
            }
            m_taskQueue.Dispose();
        });
        m_thread.Start();
    }

    public void Dispose()
    {
        m_taskQueue.CompleteAdding();
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return Thread.CurrentThread == m_thread && TryExecuteTask(task);
    }

    (...)
}

我只见过一次，无法重现，但在 foreach 的某个时刻(在 TryTakeWithNoTimeValidation 中)我得到了一个 OperationCanceledException。我不明白，因为我正在使用不带 CancellationToken 和 the documentation states 的重载它可能只会抛出 ObjectDisposedException。异常意味着什么？阻塞收集已完成？队列中的任务被取消了吗？

更新:调用堆栈如下所示:

mscorlib.dll!System.Threading.SemaphoreSlim.WaitUntilCountOrTimeout(int millisecondsTimeout, uint startTime, System.Threading.CancellationToken cancellationToken) + 0x36 bytes 
mscorlib.dll!System.Threading.SemaphoreSlim.Wait(int millisecondsTimeout, System.Threading.CancellationToken cancellationToken) + 0x178 bytes   
System.dll!System.Collections.Concurrent.BlockingCollection<System.Threading.Tasks.Task>.TryTakeWithNoTimeValidation(out System.Threading.Tasks.Task item, int millisecondsTimeout, System.Threading.CancellationToken cancellationToken, System.Threading.CancellationTokenSource combinedTokenSource) Line 710 + 0x25 bytes   C#
System.dll!System.Collections.Concurrent.BlockingCollection<System.Threading.Tasks.Task>.GetConsumingEnumerable(System.Threading.CancellationToken cancellationToken) Line 1677 + 0x18 bytes    C#

Answer 1

这是一个老问题，但我会为将来找到它的任何人添加完整的答案。 Eugene 提供的答案部分正确；当时您一定是在使用 Visual Studio 进行调试，该 Visual Studio 配置为在处理的框架异常时中断。

但是，您在 OperationCanceledException 上中断的真正原因是 BlockingCollection<T>.CompleteAdding() 的代码看起来像这样:

    public void CompleteAdding()
    {
        int num;
        this.CheckDisposed();
        if (this.IsAddingCompleted)
        {
            return;
        }
        SpinWait wait = new SpinWait();
    Label_0017:
        num = this.m_currentAdders;
        if ((num & -2147483648) != 0)
        {
            wait.Reset();
            while (this.m_currentAdders != -2147483648)
            {
                wait.SpinOnce();
            }
        }
        else if (Interlocked.CompareExchange(ref this.m_currentAdders, num | -2147483648, num) == num)
        {
            wait.Reset();
            while (this.m_currentAdders != -2147483648)
            {
                wait.SpinOnce();
            }
            if (this.Count == 0)
            {
                this.CancelWaitingConsumers();
            }
            this.CancelWaitingProducers();
        }
        else
        {
            wait.SpinOnce();
            goto Label_0017;
        }
    }

注意这些特定的行:

if (this.Count == 0)
{
    this.CancelWaitingConsumers();
}

哪个调用这个方法:

private void CancelWaitingConsumers()
{
    this.m_ConsumersCancellationTokenSource.Cancel();
}

所以即使您没有明确使用 CancellationToken在您的代码中，底层框架代码抛出一个 OperationCanceledException如果BlockingCollection CompleteAdding() 时为空叫做。它这样做是为了向 GetConsumingEnumerable() 发出信号方法退出。该异常由框架代码处理，如果您没有将调试器配置为拦截它，您将不会注意到它。

您无法复制它的原因是您调用了CompleteAdding()在你的Dispose()方法。因此，它是在 GC 的突发奇想下被调用的。