c# - 如何强制 IAsyncEnumerable 遵守 CancellationToken

Question

我有一个生成 IAsyncEnumerable<int> 的异步迭代器方法(数字流)，每 200 毫秒一个数字。此方法的调用者使用流，但希望在 1000 毫秒后停止枚举。所以一个CancellationTokenSource被使用， token 被传递为 WithCancellation 的参数扩展方法。但是 token 不受尊重。枚举一直持续到消耗完所有数字:

static async IAsyncEnumerable<int> GetSequence()
{
    for (int i = 1; i <= 10; i++)
    {
        await Task.Delay(200);
        yield return i;
    }
}

var cts = new CancellationTokenSource(1000);
await foreach (var i in GetSequence().WithCancellation(cts.Token))
{
    Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} > {i}");
}

输出:

12:55:17.506 > 1
12:55:17.739 > 2
12:55:17.941 > 3
12:55:18.155 > 4
12:55:18.367 > 5
12:55:18.570 > 6
12:55:18.772 > 7
12:55:18.973 > 8
12:55:19.174 > 9
12:55:19.376 > 10

预期输出是 TaskCanceledException在 5 号之后发生。看来我误解了 WithCancellation 是什么实际上是在做。该方法只是将提供的标记传递给迭代器方法，如果该方法接受一个标记。否则，与方法一样 GetSequence()在我的示例中， token 被忽略。我想我的解决方案是手动询问枚举主体内的 token :

var cts = new CancellationTokenSource(1000);
await foreach (var i in GetSequence())
{
    cts.Token.ThrowIfCancellationRequested();
    Console.WriteLine($"{DateTime.Now:HH:mm:ss.fff} > {i}");
}

这很简单而且效果很好。但无论如何，我想知道是否可以创建一个扩展方法来实现我对 WithCancellation 的预期。要做的是，在随后的枚举中烘焙 token 。这是所需方法的签名:

public static IAsyncEnumerable<T> WithEnforcedCancellation<T>(
    this IAsyncEnumerable<T> source, CancellationToken cancellationToken)
{
    // Is it possible?
}

Answer 1

IAsyncEnumerable 通过 EnumeratorCancellation 属性明确提供此机制:

static async IAsyncEnumerable<int> GetSequence([EnumeratorCancellation] CancellationToken ct = default) {
    for (int i = 1; i <= 10; i++) {
        ct.ThrowIfCancellationRequested();
        await Task.Delay(200);    // or `Task.Delay(200, ct)` if this wasn't an example
        yield return i;
    }
}

事实上，如果您为方法提供了 CancellationToken 参数，但没有添加属性，编译器会发出警告。

请注意，传递给 .WithCancellation 的 token 将覆盖传递给该方法的任何本地 token 。 specs有这方面的详细信息。

当然，这仍然只有在枚举实际接受 CancellationToken 时才有效——但取消只有在协作完成时才真正有效的事实对于任何 async 都是正确的工作。 Yeldar's answer有利于将某些取消措施“强制”到不支持它的枚举中，但首选解决方案应该是修改枚举以支持自行取消——编译器会尽一切努力帮助您。