IAsynceNumerator.Current在枚举器集合未强制转换到列表时返回null

Question

第一个函数设计成使linq能够安全地并行执行lambda函数（即使是异步void函数）。

因此您可以执行collection.asParallel（）.forAllAsync(asyncx=>awaitx.action)。

第二个函数的设计目的是使您能够并行组合和执行多个IAsynceNumerable，并尽可能快地返回它们的结果。

我有以下代码：

    public static async Task ForAllAsync<TSource>(
        this ParallelQuery<TSource> source, 
        Func<TSource, Task> selector,
        int? maxDegreeOfParallelism = null)
    {
        int maxAsyncThreadCount = maxDegreeOfParallelism ?? Math.Min(System.Environment.ProcessorCount, 128);
        using SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);

        IEnumerable<Task> tasks = source.Select(async input =>
        {
            await throttler.WaitAsync().ConfigureAwait(false);
            
            try
            {
                await selector(input).ConfigureAwait(false);
            }
            finally
            {
                throttler.Release();
            }
        });

        await Task.WhenAll(tasks).ConfigureAwait(true);
    }

    public static async IAsyncEnumerable<T> ForAllAsync<TSource, T>(
        this ParallelQuery<TSource> source,
        Func<TSource, IAsyncEnumerable<T>> selector,
        int? maxDegreeOfParallelism = null,
        [EnumeratorCancellation]CancellationToken cancellationToken = default) 
        where T : new()
    {
        IEnumerable<(IAsyncEnumerator<T>, bool)> enumerators = 
            source.Select(x => (selector.Invoke(x).GetAsyncEnumerator(cancellationToken), true)).ToList();

        while (enumerators.Any())
        {
            await enumerators.AsParallel()
                .ForAllAsync(async e => e.Item2 = (await e.Item1.MoveNextAsync()), maxDegreeOfParallelism)
                .ConfigureAwait(false);
            foreach (var enumerator in enumerators)
            {
                yield return enumerator.Item1.Current;
            }
            enumerators = enumerators.Where(e => e.Item2);
        }
    }

如果我从第二个函数中删除“toList（）”，yield return将开始返回null，因为Enumerator.item1。尽管Enumerator.item2（MoveNextAsync（）的结果）为true，但Current倾向于为null。

为什么？

Answer 1

这是一个经典的缓期执行案例。每次在未物化的IEnumerable<>上调用求值方法时，它都要物化IEnumerable。在本例中，这是重新调用选择器并创建等待GetAsyncEnumerator调用的任务的新实例。

通过调用.toList()可以实现IEnumerable。如果没有它，物化将与对.any()、对forAllAsync()的调用以及您的foreach循环一起发生。

同样的行为可以像这样最低限度地再现：

var enumerable = new[] { 1 }.Select(_ => Task.Delay(10));
await Task.WhenAll(enumerable);
Console.WriteLine(enumerable.First().IsCompleted); // False
enumerable = enumerable.ToList();
await Task.WhenAll(enumerable);
Console.WriteLine(enumerable.First().IsCompleted); // True

在对enumerable.first()的第一次调用中，我们得到的任务实例与我们在它之前等待的任务实例不同。

在第二个调用中，我们使用了相同的实例，因为任务已经物化为一个列表。