How to stop multiple Parallel.For loops with CancellationToken

Question

I'm developing an app with C# and WPF. I'm using 3 nested Parallel.For loops as shown below. When I Cancel() the token, loops are starting to throw ImportAbortedException BUT, I can't catch ImportAbortedException. What I cought is AggregateException

What I want is, stop all the Parallel.Fors and catch the ImportAbortedException and do some other stuff.

Here is the code.

    private int _loopCount1 = 100;
    private int _loopCount2 = 200;
    private int _loopCount3 = 10000;
    private CancellationToken _cToken;
    private CancellationTokenSource _cSource;
    private void Init()
    {
        _cSource = new CancellationTokenSource();
        _cToken = new CancellationToken();
        _cToken = _cSource.Token;

        try
        {
            DoTheWork();
        }
        catch (ImportAbortedException)
        {
            /// 
        }
        catch (Exception)
        {

        }
    }

    private void StopAllLoops()
    {
        _cSource.Cancel();
    }

    private void DoTheWork()
    {
        Parallel.For(0, _loopCount1, i =>
        {
            if (CheckIfCanceled())
                throw new ImportAbortedException("process aborted!");

            // do a few calculations here.

            Parallel.For(0, _loopCount2, j =>
            {
                if (CheckIfCanceled())
                    throw new ImportAbortedException("process aborted!");

                // do a few calculations here.

                Parallel.For(0, _loopCount3, k =>
                {
                    if (CheckIfCanceled())
                        throw new ImportAbortedException("process aborted!");

                    // do some other process here.
                });
            });
        });
    }

    private bool CheckIfCanceled()
    {
        return _cToken.IsCancellationRequested;
    }

Answer 1

I would avoid using Parallel.For entirely and use Microsoft's Reactive Framework (NuGet "Rx-Main" & "Rx-WPF"). You can use it to neatly handle all of your parallel processing and you can marshall results back tot he UI thread.

Your code would look like this:

private IDisposable DoTheWork()
{
    var query =
        from i in Observable.Range(0, _loopCount1)
        from x in Observable.Start(() => SomeCalculation1(i))
        from j in Observable.Range(0, _loopCount2)
        from y in Observable.Start(() => SomeCalculation2(i, j))
        from k in Observable.Range(0, _loopCount3)
        from z in Observable.Start(() => SomeCalculation3(i, j, k))
        select new { x, y, z };

    return
        query
            .ObserveOnDispatcher()
            .Subscribe(w =>
            {
                /* Do something with w.x, w.y, w.z */
            });
}

You would call it like this:

var subscription = DoTheWork();

And to cancel, you simply do this:

subscription.Dispose();

It's all multi-threaded, UI safe, and can be easily cancelled.

Answer 2

The ParallelOptions attribute of the Parallel.For has a CancellationToken property you can pass it, so when the cancellation token is canceled the parallel for is stopped and a OperationCanceledException is produced.

See MSDN ParallelOptions