Stop Parallel.ForEachAsync-CodePudding

In C#, I am interested in stopping a Parallel.ForEachAsync loop (considering the differences between Stop and Break); for Parallel.ForEach I can do the following:

Parallel.ForEach(items, (item, state) =>
{
    if (cancellationToken.IsCancellationRequested)
    {
        state.Stop();
        return;
    }

    // some process on the item
    Process(item);
});

However, since I have a process that needs to be executed asynchronously, I switched to using Parallel.ForEachAsync. The ForEachAsync does not have the Stop() method, I'm able to break the loop as the following, but I'm wondering if this the most effective way of breaking the look (in other words, the loop needs to stop ASAP when it receives the cancellation request).

await Parallel.ForEachAsync(items, async (item, state) =>
{
    if (cancellationToken.IsCancellationRequested)
    {
        return;
    }

    // some async process on the item
    await ProcessAsync(item);
});

CodePudding user response：

The Parallel.ForEachAsync lambda has a CancellationToken as its second parameter. This token is supplied by the API, it's not the same token that you have passed in the ParallelOptions. You can forward this token to any asynchronous method that you invoke inside the lambda. If you invoke non-cancelable methods, then the best you can do is to call the ThrowIfCancellationRequested at strategic places inside the lambda:

var cts = new CancellationTokenSource();
var options = new ParallelOptions() { CancellationToken = cts.Token };

try
{
    await Parallel.ForEachAsync(items, options, async (item, ct) =>
    {
        //...
        ct.ThrowIfCancellationRequested();
        //...
        await ProcessAsync(item, ct);
        //...
        ct.ThrowIfCancellationRequested();
        //...
    });
}
catch (OperationCanceledException ex)
{
    // ...
}

CodePudding user response：

You're going to need something like this:

await Parallel.ForEachAsync(items, async (item, state) =>
{
    await ProcessAsync(item, cancellationToken);
});


async Task ProcessAsync(string item, CancellationToken ct)
{
    while (!ct.IsCancellationRequested)
    {
        //Process
    }
}