IAsyncEnumerator.Current 在未将枚举数集合强制转换为 List 时返回 null

Posted 2023-04-12

【中文标题】IAsyncEnumerator.Current 在未将枚举数集合强制转换为 List 时返回 null

【英文标题】：IAsyncEnumerator.Current returns null when enumerators collection is not casted to a List

【发布时间】：2021-07-15 19:27:02

【问题描述】：

第一个函数旨在使 linq 能够安全地并行执行 lambda 函数（甚至是 async void 函数）。

所以你可以做 collection.AsParallel().ForAllASync(async x => await x.Action)。

第二个函数旨在使您能够并行组合和执行多个 IAsyncEnumerables 并尽快返回它们的结果。

我有以下代码：

    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.Current 往往为 null，尽管 enumerator.Item2（MoveNextAsync() 的结果）为真。

为什么？

【参考方案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() 的第一次调用中，我们最终得到了一个与我们在它之前的行中等待的不同的任务实例。

在第二次调用中，我们使用的是同一个实例，因为任务已经具体化为一个列表。