Asynchronous parallel ForEach implemented using Task.Run I wrote a simple implementation of aysnc parallel.ForEach( ... )

All it really does is create a list of tasks and wait for them all to complete and aggregate the exceptions if any.

But to use CancellationToken I added Task.Run but I am worried that this is not the best way of stopping a task from running.

I could add some shortcuts in case the token cannot be cancelled, empty collections and so on.

But those are minor tweaks.

Anything else I could do to make tasks.AddRange(source.Select(s =&gt; Task.Run(() =&gt; body(s), token))); more efficient while allowing me to cancel running tasks.

public static async Task ForEach&lt;T&gt;(ICollection&lt;T&gt; source, Func&lt;T, Task&gt; body, CancellationToken token )
{
  // create the list of tasks we will be running
  var tasks = new List&lt;Task&gt;(source.Count);
  try
  {
    // and add them all at once.
    tasks.AddRange(source.Select(s =&gt; Task.Run(() =&gt; body(s), token)));

    // execute it all.
    await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false);

    // throw if we are done here.
    token.ThrowIfCancellationRequested();
  }
  catch
  {
    // find the error(s) that might have happened.
    var errors = tasks.Where(tt =&gt; tt.IsFaulted).Select(tu =&gt; tu.Exception).ToList();

    // we are back in our own thread
    if (errors.Count &gt; 0)
    {
      throw new AggregateException(errors);
    }
  }
}


Any suggestions where I could improve performance and the creation of tasks?
Task.WhenAll also accepts Enumerable&lt;Task&gt; as argument. So your lines with tasks can be simplified:

var tasks = source.Select(x =&gt; Task.Run(() =&gt; body(s), token));
try 
{
    await Task.WhenAll(tasks).ConfigureAwait(false);
    token.ThrowIfCancellationRequested();
}
catch
{
    // find the error(s) that might have happened.
    var errors = tasks.Where(tt =&gt; tt.IsFaulted).Select(tu =&gt; tu.Exception).ToList();

    // we are back in our own thread
    if (errors.Count &gt; 0)
    {
        throw new AggregateException(errors);
    }
}


If you end up throwing because of the line token.ThrowIfCancellationRequested(), after all the tasks have completed (so none of the tasks themselves faulted), I think you end up swallowing the exception.



You provide the CancellationToken to Task.Run. The only thing this does is check whether the token has been cancelled before the tasks starts running. If you want to affect the task itself, you need to pass the token into the method body as well. Additionally, the implementation of body then needs to actually do something with the token (like checking whether it is cancelled), before cancelling will have any use.

So consider the following possible APIs:

Task ForEach&lt;T&gt;(ICollection&lt;T&gt; seq, Func&lt;T, CancellationToken, Task&gt; body, CancellationToken token)


or 

Task ForEach&lt;T&gt;(ICollection&lt;T&gt; seq, Func&lt;T, Task&gt; body)


Not much in between will have much use, assuming it's the function body that you want to cancel.

For some additional reading, see here.