ForEachAsync with Result

by

Now that the Parallel.ForEachAsync API has become part of the standard libraries (.NET 6), it makes sense to implement a variant that returns a Task<TResult[]>, based on this API. Here is an implementation:

/// <summary>
/// Executes a foreach loop on an enumerable sequence, in which iterations may run
/// in parallel, and returns the results of all iterations in the original order.
/// </summary>
public static Task<TResult[]> ForEachAsync<TSource, TResult>(
    IEnumerable<TSource> source,
    ParallelOptions parallelOptions,
    Func<TSource, CancellationToken, ValueTask<TResult>> body)
{
    ArgumentNullException.ThrowIfNull(source);
    ArgumentNullException.ThrowIfNull(parallelOptions);
    ArgumentNullException.ThrowIfNull(body);
    List<TResult> results = new();
    if (source.TryGetNonEnumeratedCount(out int count)) results.Capacity = count;
    IEnumerable<(TSource, int)> withIndexes = source.Select((x, i) => (x, i));
    return Parallel.ForEachAsync(withIndexes, parallelOptions, async (entry, ct) =>
    {
        (TSource item, int index) = entry;
        TResult result = await body(item, ct).ConfigureAwait(false);
        lock (results)
        {
            while (results.Count <= index) results.Add(default);
            results[index] = result;
        }
    }).ContinueWith(t =>
    {
        TaskCompletionSource<TResult[]> tcs = new();
        switch (t.Status)
        {
            case TaskStatus.RanToCompletion:
                lock (results) tcs.SetResult(results.ToArray()); break;
            case TaskStatus.Faulted:
                tcs.SetException(t.Exception.InnerExceptions); break;
            case TaskStatus.Canceled:
                tcs.SetCanceled(new TaskCanceledException(t).CancellationToken); break;
            default: throw new UnreachableException();
        }
        Debug.Assert(tcs.Task.IsCompleted);
        return tcs.Task;
    }, default, TaskContinuationOptions.DenyChildAttach |
        TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default).Unwrap();
}

This implementation supports all the options and the functionality of the Parallel.ForEachAsync overload that has an IEnumerable<T> as source. Its behavior in case of errors and cancellation is identical. The results are arranged in the same order as the associated elements in the source sequence.

More Related Contents:

  1. How to limit the amount of concurrent async I/O operations?
  2. Best way in .NET to manage queue of tasks on a separate (single) thread
  3. WaitAll vs WhenAll
  4. Catch an exception thrown by an async void method
  5. Running multiple async tasks and waiting for them all to complete
  6. Run two async tasks in parallel and collect results in .NET 4.5
  7. Calling async method synchronously
  8. Why does this async action hang when I try and access the Result property of my Task?
  9. Task sequencing and re-entracy
  10. Enforce an async method to be called once
  11. What’s the difference between returning void and returning a Task?
  12. .NET Asynchronous stream read/write
  13. Best way to convert callback-based async method to awaitable task
  14. Async/Await – is it *concurrent*?
  15. How to make BackgroundWorker return an object
  16. Process files concurrently and asynchronously
  17. Executing tasks in parallel
  18. Calling async methods from non-async code
  19. Use Task.Run() in synchronous method to avoid deadlock waiting on async method?
  20. A pattern for self-cancelling and restarting task
  21. What is the correct way to cancel an async operation that doesn’t accept a CancellationToken?
  22. How can I capture the value of an outer variable inside a lambda expression?
  23. Converting async/await to Task.ContinueWith
  24. How to throttle multiple asynchronous tasks?
  25. Is it possible to get successful results from a Task.WhenAll when one of the tasks fails? [duplicate]
  26. TPL Dataflow, whats the functional difference between Post() and SendAsync()?
  27. TPL Dataflow, guarantee completion only when ALL source data blocks completed
  28. How to implement Task.WhenAny() with a predicate
  29. ConcurrentDictionary GetOrAdd async
  30. Calling TaskCompletionSource.SetResult in a non blocking manner

Leave a Comment