The problem of parallelizing asynchronous operations has been solved with the introduction of the Parallel.ForEachAsync
API in .NET 6, but people who are using older .NET platforms might still need a decent substitute. A solid base for implementing one is the ActionBlock<T>
component from the TPL Dataflow library. This is part of the standard .NET libraries (.NET Core and .NET 5+), and available as a NuGet package for the .NET Framework. Here is how it can be used:
public static Task Parallel_ForEachAsync<T>(IEnumerable<T> source,
int maxDegreeOfParallelism, Func<T, Task> action)
{
var options = new ExecutionDataflowBlockOptions();
options.MaxDegreeOfParallelism = maxDegreeOfParallelism;
var block = new ActionBlock<T>(action, options);
foreach (var item in source) block.Post(item);
block.Complete();
return block.Completion;
}
This solution enumerates eagerly the supplied IEnumerable<T>
, and sends immediately all of its elements to the ActionBlock<T>
. Then it signals that no more items are going to be sent to the block, and returns the Completion
property of the block. The Completion
is a Task
that will complete when the block has processed all the items with the specified MaxDegreeOfParallelism
, or when an exception has occurred while processing an item (and after all pending operations have completed). This should be sufficient for more applications, but the resulting behavior is not identical with the behavior of the Parallel.ForEachAsync
. Here are the differences:
- The enumeration of the
source
enumerable happens synchronously, on the current thread. This is OK if the source
is a materialized collection like a List<T>
, but not if it's a deferred enumerable like for example a BlockingCollection<T>.GetConsumingEnumerable
. Not only the current thread might be blocked, but also any exception thrown during the enumeration will be propagated synchronously instead of being wrapped in the resulting Task
. On the contrary the Parallel.ForEachAsync
takes items from the source
enumerable one at a time, not all at once.¹
- An exception thrown during the enumeration of the
source
will result in all work already scheduled in the ActionBlock<T>
to become fire-and-forget.
- Any
OperationCanceledException
s thrown by the action
are suppressed. That's how the TPL Dataflow library behaves by design.
The first two issues are quite serious, and indicate that the type of the source
parameter should be something else than IEnumerable<T>
, like for example an IReadOnlyCollection<T>
or an IList<T>
. These interfaces are typically not implemented by deferred collections. The third issue is less serious, but this too can be annoying in some scenarios.
Below is a more sophisticated implementation that attempts to be a complete substitute of the Parallel.ForEachAsync
API, both in the signature and the behavior. It might be useful in advanced scenarios, where the simple implementation above is not sufficient.
public static Task Parallel_ForEachAsync<TSource>(IEnumerable<TSource> source,
ParallelOptions parallelOptions,
Func<TSource, CancellationToken, Task> body)
{
if (source == null) throw new ArgumentNullException("source");
if (parallelOptions == null) throw new ArgumentNullException("parallelOptions");
if (body == null) throw new ArgumentNullException("body");
var options = new ExecutionDataflowBlockOptions()
{
MaxDegreeOfParallelism = parallelOptions.MaxDegreeOfParallelism,
CancellationToken = parallelOptions.CancellationToken,
TaskScheduler = parallelOptions.TaskScheduler
};
// Immitate the default degree of parallelism of the Parallel.ForEachAsync.
if (options.MaxDegreeOfParallelism == DataflowBlockOptions.Unbounded)
options.MaxDegreeOfParallelism = Environment.ProcessorCount;
options.BoundedCapacity = options.MaxDegreeOfParallelism;
var block = new ActionBlock<TSource>(item =>
{
return body(item, options.CancellationToken)?.ContinueWith(t =>
{
// Fix TPL Dataflow's OperationCanceledException-swallowing behavior.
if (t.IsCanceled && !options.CancellationToken.IsCancellationRequested)
return Task.FromException(new TaskCanceledException(t));
return t;
}, CancellationToken.None, TaskContinuationOptions.DenyChildAttach |
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default).Unwrap();
}, options);
Task enumerationTask = Task.Factory.StartNew(async () =>
{
// Feed the block with data, allowing for backpressure, and funnel any
// exception to the ActionBlock.
try
{
foreach (TSource item in source)
if (!await block.SendAsync(item)) // Continue on captured context
break; // The block rejected the item
block.Complete();
}
catch (Exception ex) { ((IDataflowBlock)block).Fault(ex); }
}, CancellationToken.None, TaskCreationOptions.DenyChildAttach,
options.TaskScheduler).Unwrap();
return enumerationTask.ContinueWith(t =>
{
Debug.Assert(t.Status == TaskStatus.RanToCompletion);
if (t.Status != TaskStatus.RanToCompletion) return t; // Should never happen
return block.Completion;
}, CancellationToken.None, TaskContinuationOptions.DenyChildAttach |
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default).Unwrap();
}
The only difference is that the body
parameter is of type Func<TSource, CancellationToken, Task>
instead of Func<TSource, CancellationToken, ValueTask>
. This is because value-tasks are a relatively recent feature, and are not available in .NET Framework.
¹ This is not documented, and the behavior could change in the future, but currently (.NET 7) this is how it works.
Parallel.ForEachAsync
API.