Asynchronous parallel ForEach implemented using Task.Run

Question

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 => Task.Run(() => body(s), token))); more efficient while allowing me to cancel running tasks.

public static async Task ForEach<T>(ICollection<T> source, Func<T, Task> body, CancellationToken token )
{
  // create the list of tasks we will be running
  var tasks = new List<Task>(source.Count);
  try
  {
    // and add them all at once.
    tasks.AddRange(source.Select(s => Task.Run(() => 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 => tt.IsFaulted).Select(tu => tu.Exception).ToList();

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

Any suggestions where I could improve performance and the creation of tasks?

JAD · Accepted Answer · 2019-09-13 12:29:47Z

2

Task.WhenAll also accepts Enumerable<Task> as argument. So your lines with tasks can be simplified:

var tasks = source.Select(x => Task.Run(() => 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 => tt.IsFaulted).Select(tu => tu.Exception).ToList();

    // we are back in our own thread
    if (errors.Count > 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<T>(ICollection<T> seq, Func<T, CancellationToken, Task> body, CancellationToken token)

or

Task ForEach<T>(ICollection<T> seq, Func<T, Task> 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.

edited Sep 13, 2019 at 12:29

answered Sep 13, 2019 at 12:14

JAD

2,9592 gold badges16 silver badges30 bronze badges

\$\begingroup\$ Thanks for the info on Task.Run I have to admit, I never knew that and thanks for the Throw that might never get caught. While I want the body to control the token I also want my ForEach to manage it, I might had a delay task and a WhenAny \$\endgroup\$

FFMG
– FFMG

2019-09-14 04:00:40 +00:00
Commented Sep 14, 2019 at 4:00
\$\begingroup\$ Also, why did you move the source.Select( ... ) out of the try/catch? \$\endgroup\$

FFMG
– FFMG

2019-09-14 04:02:40 +00:00
Commented Sep 14, 2019 at 4:02
\$\begingroup\$ @FFMG multiple reasons: 1) It makes it more convenient, since we both need the collection of tasks in the try, as the catch block. Moving the declaration out, makes that possible. 2) There is no clear reason to put it in the try block. From the usage here,try is supposed to catch any exceptions the tasks might throw during their execution. If those tasks throw exceptions at some point during their execution, they will be caught by the created Task object (the objects in tasks), and stored until the result is accessed. In this example, the exceptions will be thrown by (cont.) \$\endgroup\$

JAD
– JAD

2019-09-15 09:51:28 +00:00
Commented Sep 15, 2019 at 9:51
\$\begingroup\$ the statement that awaits the tasks: await Task.WhenAll(tasks).ConfigureAwait(false);, which is within the try block. 3) If the source.Select statement throws for some other reason than task failure or cancellation, like either source or body being null, you would also catch them in your catch block, which is not something you want, since the catch block currently would swallow those exceptions. \$\endgroup\$

JAD
– JAD

2019-09-15 09:53:21 +00:00
Commented Sep 15, 2019 at 9:53

Add a comment |

Stack Exchange Network

Asynchronous parallel ForEach implemented using Task.Run

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Asynchronous parallel ForEach implemented using Task.Run

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions