Async parallel foreach

[RoccoDevs]

There are situations where a asynchronous parallel foreach is needed instead of the already existing Parallel.ForEach() and while it's not rocket science to implement yourself I think it would be a nice feature to be implemented in corefx. This should be seen as the asynchronous alternative to the existing Parallel.ForEach(). And should take IEnumerable<T> as well as IAsyncEnumerable<T>.

There are examples on this issue already here is one: https://devblogs.microsoft.com/pfxteam/implementing-a-simple-foreachasync-part-2/

        /// <summary>
        ///     Executes a foreach asynchronously.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="source">The source.</param>
        /// <param name="dop">The degrees of parallelism.</param>
        /// <param name="body">The body.</param>
        /// <returns></returns>
        public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
        {
            return Task.WhenAll(
                from partition in System.Collections.Concurrent.Partitioner.Create(source).GetPartitions(dop)
                select Task.Run(async delegate
                {
                    using (partition)
                    {
                        while (partition.MoveNext())
                            await body(partition.Current);
                    }
                }));
        }

Proposed API

namespace System.Threading.Tasks
{
    public static class Parallel
    {
        public static Task ForEachAsync<T>(IAsyncEnumerable<T> source, Func<T, ValuteTask> body);
        public static Task ForEachAsync<T>(IAsyncEnumerable<T> source, ParallelOptions parallelOptions, Func<T, CancellationToken, ValueTask> body);

        public static Task ForEachAsync<T>(IEnumerable<T> source, Func<T, ValuteTask> body);
        public static Task ForEachAsync<T>(IEnumerable<T> source, ParallelOptions parallelOptions, Func<T, CancellationToken, ValuteTask> body);
    }
}

Closed questions

While I consider these questions to be closed feel free to continue discussion on them.

• Should DOP be automatic or manual at all times?
  DOP Will have a default, just like parallel linq. However documentation should make it clear that this default can be wrong in various cases. Such as detailed by @GSPP in various comments.
• Should it accept IEnumerable?
  Yes it should
• Should the partitioner concept be supported at all?
  No we should ensure the specified DOP is adhered to exactly. Creating batches can lead to a smaller than expected effective DOP.
• Should we process the source items sequentially or make no attempt?
  By default we should try to process source items sequentially to improve data locality. Reason being: HDD read performance will increase drastically. Furthermore cache usage will be improved. It should however be made very clear to people in docs, that the order cannot be guaranteed
• Should there be an alternative to Parallel.For?
  There should be one but I'll leave that outside of this issue.
• Should the funcs return ValueTask instead of Task?
  Yes ValueTask is more accomodating than Task (cheaper to convert Task to ValueTask than the other way arround). Furthermore it should result in a peformance increase.
• Should TaskCreationOptions.DenyChildAttach be specified?
  Yes this should be specified see This comment for more info
• Should there be an overload without CancellationToken in the Func?
  No there shouldn't be. It would have little to no benefit, but would cause a lot of implications (ambiguity, analysis becomes harder)

[ichensky]

ForEachAsync method can have to many different implementations. Some of users would like to create their own Partitioner<TSource> class, others do not need it in the ForEachAsync like method, and some of users would like to have their own absolutly specific logic in Task.Run(async delegate ..... block.

[svick]

@ichensky Compare this with the synchronous Parallel.ForEach:

• Some users would like to use Partitioner, and there's an overload for that.
• Some users will need completely custom logic, and so they can't use Parallel.ForEach.

But Parallel.ForEach still exists and is useful. I think the situation with Parallel.ForEachAsync is similar: we could create a set of overloads that are useful in the common cases. And that would still be useful, even if it wouldn't cover all cases.

[RoccoDevs]

@ichensky I fully agree with the way @svick explains it.
I would simply see this as the asynchronous version of Parallel.ForEach.

A set of overloads for the most common cases is needed for sure!

[Clockwork-Muse]

Hmm. Probably wants to be based on dotnet/corefx#32640 ?

[svick]

@Clockwork-Muse Possibly. I think Parallel.ForEachAsync that somehow accepts IEnumerable<T> is important. One that accepts IAsyncEnumerable<T> would be useful too. If there was a simple built-in way to convert from IEnumerable<T> to IAsyncEnumerable<T>, then overloads that directly accept IEnumerable<T> might not be necessary. Does something like that exist or is it proposed?

[RoccoDevs]

@svick Yes accepting IAsyncEnumerable<T> sounds good!

[ichensky]

Usually when making ForEachAsync, for ex. over list of files and inside we create task that Download/Process these files, it is useful pass in custom function ForEachAsync param like maximum possible count of Tasks, that can executes at one time, because of internet bandwidth, depending on other services and so on. Time to time it is needed, that ForEachAsync not just Wait until all Tasks finished but also return result of all those tasks.

[WOLVIE97]

Kool..

[GSPP]

This API is urgently needed. In particular, there is no way currently to process a set of "items" in parallel and in an async way. Parallel and PLINQ (.AsParallel()) cannot be used for this. I believe Dataflow can be used but it's awkward for something this simple.

The source for the code in the opening post is https://blogs.msdn.microsoft.com/pfxteam/2012/03/05/implementing-a-simple-foreachasync-part-2/.

I am very active on Stack Overflow and I see people needing this all the time. People then use very bad workarounds such as starting all items in parallel at the same time, then WhenAll them. So they start 10000 HTTP calls and wonder why it performs so poorly. Or, they execute items in batches. This can be much slower because as the batch completes item by item the effective DOP decreases. Or, they write very awkward looping code with collections of tasks and weird waiting schemes.

I needed this API multiple times myself. So I developed a solution based on the code in this article. I made the following enhancements which the framework should also have:

1. A CancellationToken should be taken as an argument. When it is signaled no further loop iterations should be started.
2. When an exception is thrown by one item the loop should be cancelled. When all items are done executing all exceptions should be thrown together in the form of an AggregateException.
3. The worker delegate should receive a CancellationToken. This is a token that combines the externally passed token and the cancellation that can come from cancellation due to an exception.
4. The partitioner should not do any batching. It must be ensured that the specified degree of parallelism is adhered to exactly. For IO, it is important to use the DOP specified by the caller. Batching can lead to the effective DOP being smaller than expected.
5. Since we are starting tasks we should think about whether a TaskScheduler should be taken as an argument. The caller might want to do a lot of CPU work so he might want to schedule that. Alternatively, the caller can switch to a different scheduler as part of the delegate he passes in. So we might not want to natively support TaskScheduler.
6. In my implementation I specified TaskCreationOptions.DenyChildAttach for those tasks. I'm not entirely sure about this.
7. Do we need to think about the SynchronizationContext? As the code stands user code will always be called on the thread pool with no sync context.

---

Should the DOP be automatic? In my opinion, we have no choice but to force the caller to specify an explicit DOP. Usually, user code will perform IO as part of the loop (that's likely the reason for using async). The system has no way of knowing the right DOP for IO. Different devices (disk, SSD, web service) have vastly different characteristics. Also, the DOP might be intentionally low in order to not overload the system being called. IMO, no auto-tuning is possible. We cannot even make the parameter optional and default to the CPU count. The CPU count is unrelated to the ideal IO DOP! This is why IO on the thread pool can explode so radically. The auto-tuning heuristics can get into a state where they add unbounded amounts of threads (see these issues: https://github.com/dotnet/coreclr/issues/1754, https://github.com/dotnet/coreclr/issues/2184).

[tarekgh]

@tomesendam @GSPP could you please add the full proposal according to the first step in the doc https://github.com/dotnet/corefx/blob/master/Documentation/project-docs/api-review-process.md. Thanks.

[RoccoDevs]

@tarekgh I will try when I find the time to work on this!

[MgSam]

I agree this is urgently needed for the reasons @GSPP lists.

I'd also add it's urgently needed because Parallel.ForEach() is a pit of failure when writing async code. It's more than happy to accept an async lambda that returns a Task. Only it doesn't do what the user expects (which is only continue once the async functions within have all fully executed). Instead it will immediately complete, leaving the user's code silently broken and waiting for a potentially hard-to-track-down issue at runtime.

I'd go so far as to say that Parallel.ForEach should maybe throw an exception when passed a Task returning method- or at the very least VS should ship with an analyzer that points out this is an anti-pattern.

Even then, this only solves half the problem. Users need a proper async API to call as an alternative.

[yaakov-h]

Parallel.ForEach should maybe throw an exception when passed a Task returning method

It's not Task-returning though. It's async void as a lambda.

[tarekgh]

This is unlikely will be in v3.0. Please apply the request I mentioned in my comment https://github.com/dotnet/corefx/issues/34233#issuecomment-455354106 to be able to look at it as whole and move forward.

[MihaZupan]

@tomesendam if you want I can open the api proposal issue

I think that a base parallel foreach should accept a Func that takes a CancellationToken.

Func<T, CancellationToken, Task> asyncBody;

The same CT that is passed in ParallelOptions can then be passed along.

There could be other overloads, not accepting a CT for the body that would call the "base" overload, discarding the CT.

So in essence:

Task<ParallelLoopResult> ForEachAsync<T>(IAsyncEnumerable<T> source, ParallelOptions parallelOptions, Func<T, Task> asyncBody)
{
    return ForEachAsync(source, parallelOptions, (work, ct) => asyncBody(work));
}

Task<ParallelLoopResult> ForEachAsync<T>(IAsyncEnumerable<T> source, ParallelOptions parallelOptions, Func<T, CancellationToken, Task> asyncBody);