如何实现高效的 WhenEach 来传输 IAsyncEnumerable 任务结果？

我正在尝试使用以下提供的新工具更新我的工具集C# 8 https://learn.microsoft.com/en-us/dotnet/csharp/whats-new/csharp-8，一种似乎特别有用的方法是Task.WhenAll https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.task.whenall返回一个IAsyncEnumerable https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.iasyncenumerable-1。此方法应在任务结果可用时立即对其进行流式传输，因此将其命名为WhenAll没有多大意义。WhenEach听起来更合适。该方法的签名是：

public static IAsyncEnumerable<TResult> WhenEach<TResult>(Task<TResult>[] tasks);

这个方法可以这样使用：

var tasks = new Task<int>[]
{
    ProcessAsync(1, 300),
    ProcessAsync(2, 500),
    ProcessAsync(3, 400),
    ProcessAsync(4, 200),
    ProcessAsync(5, 100),
};

await foreach (int result in WhenEach(tasks))
{
    Console.WriteLine($"Processed: {result}");
}

static async Task<int> ProcessAsync(int result, int delay)
{
    await Task.Delay(delay);
    return result;
}

预期输出：

已处理：5
已处理：4
已处理：1
已处理：3
已处理：2

我设法使用该方法编写了一个基本实现Task.WhenAny https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.task.whenany循环中，但是这种方法有一个问题：

public static async IAsyncEnumerable<TResult> WhenEach<TResult>(
    Task<TResult>[] tasks)
{
    var hashSet = new HashSet<Task<TResult>>(tasks);
    while (hashSet.Count > 0)
    {
        var task = await Task.WhenAny(hashSet).ConfigureAwait(false);
        yield return await task.ConfigureAwait(false);
        hashSet.Remove(task);
    }
}

问题是性能。这Task.WhenAny方法必须监视所有提供的任务的完成，并且它通过附加和分离延续来实现，因此在循环中重复调用它会导致 O(n²) 计算复杂度。我的幼稚实现很难处理 10,000 个任务。在我的机器上，开销将近 10 秒。我希望该方法的性能几乎与内置方法一样高Task.WhenAll，可以轻松处理数十万个任务。我怎样才能改善WhenEach使其正常运行的方法？

通过使用代码this https://devblogs.microsoft.com/pfxteam/processing-tasks-as-they-complete/文章中，您可以执行以下操作：

public static Task<Task<T>>[] Interleaved<T>(IEnumerable<Task<T>> tasks)
{
   var inputTasks = tasks.ToList();

   var buckets = new TaskCompletionSource<Task<T>>[inputTasks.Count];
   var results = new Task<Task<T>>[buckets.Length];
   for (int i = 0; i < buckets.Length; i++)
   {
       buckets[i] = new TaskCompletionSource<Task<T>>();
       results[i] = buckets[i].Task;
   }

   int nextTaskIndex = -1;
   Action<Task<T>> continuation = completed =>
   {
       var bucket = buckets[Interlocked.Increment(ref nextTaskIndex)];
       bucket.TrySetResult(completed);
   };

   foreach (var inputTask in inputTasks)
       inputTask.ContinueWith(continuation, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);

   return results;
}

然后改变你的WhenEach打电话给Interleaved code

public static async IAsyncEnumerable<TResult> WhenEach<TResult>(Task<TResult>[] tasks)
{
    foreach (var bucket in Interleaved(tasks))
    {
        var t = await bucket;
        yield return await t;
    }
}

然后你可以打电话给你的WhenEach像往常一样

await foreach (int result in WhenEach(tasks))
{
    Console.WriteLine($"Processed: {result}");
}

我对 10k 任务做了一些基本的基准测试，速度提高了 5 倍。