Parallel foreach with asynchronous lambda

Question

I would like to handle a collection in parallel  but I m having trouble implementing it and I m therefore hoping for some help   The trouble arises if I want to call a method marked async in C   within the lambda of the parallel loop  For example   var bag   new ConcurrentBag lt object gt     Parallel ForEach myCollection  async item   gt         some pre stuff   var response   await GetData item     bag Add response        some post stuff   var count   bag Count    The problem occurs with the count being 0  because all the threads created are effectively just background threads and the Parallel ForEach call doesn t wait for completion  If I remove the async keyword  the method looks like this   var bag   new ConcurrentBag lt object gt     Parallel ForEach myCollection  item   gt         some pre stuff   var responseTask   await GetData item     responseTask Wait      var response   responseTask Result    bag Add response        some post stuff   var count   bag Count    It works  but it completely disables the await cleverness and I have to do some manual exception handling    Removed for brevity    How can I implement a Parallel ForEach loop  that uses the await keyword within the lambda  Is it possible   The prototype of the Parallel ForEach method takes an Action lt T gt  as parameter  but I want it to wait for my asynchronous lambda

User · Answer

In the accepted answer the ConcurrentBag is not required. Here's an implementation without it:

var tasks = myCollection.Select(GetData).ToList();
await Task.WhenAll(tasks);
var results = tasks.Select(t => t.Result);

Any of the "// some pre stuff" and "// some post stuff" can go into the GetData implementation (or another method that calls GetData)

Aside from being shorter, there's no use of an "async void" lambda, which is an anti pattern.

User · Answer

You can use the ParallelForEachAsync extension method from AsyncEnumerator NuGet Package   using Dasync Collections   var bag   new ConcurrentBag lt object gt     await myCollection ParallelForEachAsync async item   gt         some pre stuff   var response   await GetData item     bag Add response        some post stuff    maxDegreeOfParallelism  10   var count   bag Count

User · Answer

The following is set to work with IAsyncEnumerable but can be modified to use IEnumerable by just changing the type and removing the  quot await quot  on the foreach  It s far more appropriate for large sets of data than creating countless parallel tasks and then awaiting them all      public static async Task ForEachAsyncConcurrent lt T gt  this IAsyncEnumerable lt T gt  enumerable  Func lt T  Task gt  action  int maxDegreeOfParallelism  int  boundedCapacity   null                ActionBlock lt T gt  block   new ActionBlock lt T gt              action              new ExecutionDataflowBlockOptions                             MaxDegreeOfParallelism   maxDegreeOfParallelism                BoundedCapacity   boundedCapacity    maxDegreeOfParallelism   3                          await foreach  T item in enumerable                       await block SendAsync item  ConfigureAwait false                      block Complete            await block Completion

User · Answer

My lightweight implementation of ParallelForEach async  Features   Throttling  max degree of parallelism   Exception handling  aggregation exception will be thrown at completion   Memory efficient  no need to store the list of tasks     public static class AsyncEx       public static async Task ParallelForEachAsync lt T gt  this IEnumerable lt T gt  source  Func lt T  Task gt  asyncAction  int maxDegreeOfParallelism   10                var semaphoreSlim   new SemaphoreSlim maxDegreeOfParallelism           var tcs   new TaskCompletionSource lt object gt             var exceptions   new ConcurrentBag lt Exception gt             bool addingCompleted   false           foreach  T item in source                        await semaphoreSlim WaitAsync                asyncAction item  ContinueWith t   gt                                semaphoreSlim Release                     if  t Exception    null                                        exceptions Add t Exception                                      if  Volatile Read ref addingCompleted   amp  amp  semaphoreSlim CurrentCount    maxDegreeOfParallelism                                        tcs TrySetResult null                                                        Volatile Write ref addingCompleted  true           await tcs Task          if  exceptions Count  gt  0                        throw new AggregateException exceptions                      Usage example  await Enumerable Range 1  10000  ParallelForEachAsync async  i    gt        var data   await GetData i      maxDegreeOfParallelism  100

User · Answer

I ve created an extension method for this which makes use of SemaphoreSlim and also allows to set maximum degree of parallelism           lt summary gt          Concurrently Executes async actions for each item of  lt see cref  IEnumerable lt typeparamref name  T   gt           lt  summary gt           lt typeparam name  T  gt Type of IEnumerable lt  typeparam gt           lt param name  enumerable  gt instance of  lt see cref  IEnumerable lt typeparamref name  T   gt    gt  lt  param gt           lt param name  action  gt an async  lt see cref  Action    gt  to execute lt  param gt           lt param name  maxDegreeOfParallelism  gt Optional  An integer that represents the maximum degree of parallelism          Must be grater than 0 lt  param gt           lt returns gt A Task representing an async operation lt  returns gt           lt exception cref  ArgumentOutOfRangeException  gt If the maxActionsToRunInParallel is less than 1 lt  exception gt      public static async Task ForEachAsyncConcurrent lt T gt           this IEnumerable lt T gt  enumerable          Func lt T  Task gt  action          int  maxDegreeOfParallelism   null                if  maxDegreeOfParallelism HasValue                        using  var semaphoreSlim   new SemaphoreSlim                  maxDegreeOfParallelism Value  maxDegreeOfParallelism Value                                 var tasksWithThrottler   new List lt Task gt                      foreach  var item in enumerable                                           Increment the number of currently running tasks and wait if they are more than limit                      await semaphoreSlim WaitAsync                         tasksWithThrottler Add Task Run async      gt                                                await action item  ContinueWith res   gt                                                           action is completed  so decrement the number of currently running tasks                             semaphoreSlim Release                                                                                               Wait for all tasks to complete                  await Task WhenAll tasksWithThrottler ToArray                                     else                       await Task WhenAll enumerable Select item   gt  action item                       Sample Usage   await enumerable ForEachAsyncConcurrent      async item   gt                await SomeAsyncMethod item              5

User · Answer

With SemaphoreSlim you can achieve parallelism control      var bag   new ConcurrentBag lt object gt     var maxParallel   20  var throttler   new SemaphoreSlim initialCount  maxParallel   var tasks   myCollection Select async item   gt      try          await throttler WaitAsync         var response   await GetData item        bag Add response         finally          throttler Release            await Task WhenAll tasks   var count   bag Count

User · Answer

For a more simple solution  not sure if the most optimal   you can simply nest Parallel ForEach inside a Task - as such var options   new ParallelOptions   MaxDegreeOfParallelism   5   Task Run      gt        Parallel ForEach myCollection  options  item   gt                DoWork item            The ParallelOptions will do the throttlering for you  out of the box  I am using it in a real world scenario to run a very long operations in the background  These operations are called via HTTP and it was designed not to block the HTTP call while the long operation is running   Calling HTTP for long background operation  Operation starts at the background  User gets status ID which can be used to check the status using another HTTP call  The background operation update its status   That way  the CI CD call does not timeout because of long HTTP operation  rather it loops the status every x seconds without blocking the process

User · Answer

If you just want simple parallelism  you can do this   var bag   new ConcurrentBag lt object gt     var tasks   myCollection Select async item   gt         some pre stuff   var response   await GetData item     bag Add response        some post stuff     await Task WhenAll tasks   var count   bag Count    If you need something more complex  check out Stephen Toub s ForEachAsync post

[c#] Parallel foreach with asynchronous lambda

Examples related to c#

Examples related to async-await

Examples related to task-parallel-library

Examples related to parallel.foreach