How to implement an efficient WhenEach that streams an IAsyncEnumerable of task results?

by

By using code from this article, you can implement the following:

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;
}

Then change your WhenEach to call the 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;
    }
}

Then you can call your WhenEach as per usual

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

I did some rudimentary benchmarking with 10k tasks and performed 5 times better in terms of speed.

More Related Contents:

  1. How to limit the amount of concurrent async I/O operations?
  2. Nesting await in Parallel.ForEach [duplicate]
  3. Best practice to call ConfigureAwait for all server-side code
  4. Async/await vs BackgroundWorker
  5. WaitAll vs WhenAll
  6. Running multiple async tasks and waiting for them all to complete
  7. Calling async method synchronously
  8. Parallel.ForEach and async-await [duplicate]
  9. StaTaskScheduler and STA thread message pumping
  10. What’s the difference between Task.Start/Wait and Async/Await?
  11. Does the use of async/await create a new thread?
  12. Why does this async action hang when I try and access the Result property of my Task?
  13. Regarding usage of Task.Start() , Task.Run() and Task.Factory.StartNew()
  14. Task sequencing and re-entracy
  15. Synchronously waiting for an async operation, and why does Wait() freeze the program here
  16. How to use non-thread-safe async/await APIs and patterns with ASP.NET Web API?
  17. Difference between await and ContinueWith
  18. Waiting for async/await inside a task
  19. If my interface must return Task what is the best way to have a no-operation implementation?
  20. multiple awaits vs Task.WaitAll – equivalent?
  21. Using async without await?
  22. Executing tasks in parallel
  23. Difference between returning and awaiting a Task in an async method [duplicate]
  24. Use Task.Run() in synchronous method to avoid deadlock waiting on async method?
  25. A pattern for self-cancelling and restarting task
  26. How to correctly write Parallel.For with async methods
  27. SynchronizationLockException on Monitor.Exit when using await
  28. The lack of non-capturing Task.Yield forces me to use Task.Run, why follow that?
  29. Create a completed Task
  30. How to construct a Task without starting it?

Leave a Comment