ChannelReader.ReadAllAsync(CancellationToken) 实际上并未在迭代中取消

问题描述

我一直在研究一项功能，该功能可将通道中的耗时工作排队，并在那里使用例如迭代通道await foreach(var item in channel.Reader.ReadAllAsync(cancellationToken)) {...}

I've been working on a feature that queues time consuming work in a channel, and there I iterate the channel using e.g.await foreach(var item in channel.Reader.ReadAllAsync(cancellationToken)) {...}

我期望当通过 cancellationToken 请求取消时，ReadAllAsync 会在取消后的第一次迭代中抛出.

I was expecting that when cancellation is requested through that cancellationToken, ReadAllAsync would throw on the first iteration that follows the cancellation.

在我看来，情况并非如此.循环继续，直到处理完所有项目，然后然后抛出一个OperationCanceledException.

As it seems to me, that is not the case. The loop continues until all items are processed, and then it throws an OperationCanceledException.

至少可以说，这看起来有点奇怪.来自 ChannelReader 的 github repo 可以看到取消令牌标有 [EnumeratorCancellation] 属性，因此应该将其传递给状态机围绕 yield return item; 生成(如果我错了，请纠正我).

This looks a bit strange, to say the least. From ChannelReader's github repo one could see that the cancellation token is marked with the [EnumeratorCancellation] attribute, and so it should be passed to the state machine generated around yield return item; (please correct me if I'm wrong).

我的问题是，这是 ReadAllAsync(CancellationToken) 的(有点)正常行为，还是我遗漏了什么?

My question is, is this a (somewhat) normal behavior of ReadAllAsync(CancellationToken), or am I missing something?

这是一个简单的测试代码，演示了该问题(在 dotnetfiddle 上尝试):

Here is a simple test code that demonstrates the issue (try it on dotnetfiddle):

var channel = Channel.CreateUnbounded<int>();
for (int i = 1; i <= 10; i++) channel.Writer.TryWrite(i);
int itemsRead = 0;
var cts = new CancellationTokenSource();
try
{
    await foreach (var i in channel.Reader.ReadAllAsync(cts.Token))
    {
        Console.WriteLine($"Read item: {i}. Requested cancellation: " +
            $"{cts.Token.IsCancellationRequested}");

        if (++itemsRead > 4 && !cts.IsCancellationRequested)
        {
            Console.WriteLine("Cancelling...");
            cts.Cancel();
        }
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine($"Operation cancelled. Items read: {itemsRead}");
}

这是上面的输出.请注意项目提取在中途取消后如何继续:

Here is the output from the above. Note how item fetching continues after it should have been cancelled in the middle:

Read item: 1. Requested cancellation: False
Read item: 2. Requested cancellation: False
Read item: 3. Requested cancellation: False
Read item: 4. Requested cancellation: False
Read item: 5. Requested cancellation: False
Cancelling...
Read item: 6. Requested cancellation: True
Read item: 7. Requested cancellation: True
Read item: 8. Requested cancellation: True
Read item: 9. Requested cancellation: True
Read item: 10. Requested cancellation: True
Operation cancelled. Items read: 10

推荐答案

对于必须研究相同问题的人来说，更多的是更新而不是实际答案:现在在关于 ChannelReader.ReadAllAsync 的文档中指定了这种行为() 这里

More of an update than actual answer, to whoever has had to research the same matter: this behavior is now specified in the documentation regarding ChannelReader.ReadAllAsync() here

描述取消令牌效果的第二句话(如果数据可以立即读取，那么即使在请求取消之后也可能产生该数据.")已添加为提出这个话题的结果.

The second sentence describing the cancellation token's effect ("If data is immediately ready for reading, then that data may be yielded even after cancellation has been requested.") has been added as a result of raising this subject.

感谢所有参与的人！

这篇关于ChannelReader.ReadAllAsync(CancellationToken) 实际上并未在迭代中取消的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！