How to use CancellationTokenSource in Parallel.ForEachAsync

Question

Having this handler :

public async Task<Result> Handle(MyQuery request, CancellationToken cancellationToken)
{
     var cancellationTokenSource = new CancellationTokenSource();

     await Parallel.ForEachAsync(myList, async (objectId, _) =>
     {
         var result = await _service.GetObject(objectId);

         if (result.Any())
         {
             cancellationTokenSource.Cancel();
         }
     });

     if (cancellationTokenSource.IsCancellationRequested) return Result.Fail("Error message.");

     return Result.Ok();
}

This works, but was wondering if I'm using CancellationTokenSource correct here?

Answer 1

You are using CancellationTokenSource just as a boolean variable, if the goal is to cancel the batch when some condition is met then the usage is completely wrong, you need to pass token to the ForEachAsync call (and ideally handle cancellation token in the body). Something along these lines:

try
{
    await Parallel.ForEachAsync(myList, cancellationTokenSource.Token, async (objectId, ct) =>
    {
        // pass the cancellation token
        var result = await _service.GetObject(objectId, ct);

        if (result.Any())
        {
            cancellationTokenSource.Cancel();
        }
    });
}
catch (OperationCanceledException e)
{
    Result.Fail("Error message.");
}

And as @Magnus correctly points out - you should use the cancellation token passed to your Handle method (CancellationToken cancellationToken) also:

using var cancellationTokenSource = 
    CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
// ...

See also:

• Using Parallel.ForEachAsync
• The need for two cancellation tokens in .NET 6 Parallel.ForEachAsync?