Parallel LINQ (PLINQ) в C#. Обработка ошибок и отмена запроса (метод WithCancellation)

При выполнении параллельных запросов в PLINQ также могут возникать различные исключения, которые необходимо каким-либо образом обрабатывать. При выполнении параллельных запросов PLINQ разбивает исходную последовательность на части, каждая из которых обрабатывается в отдельном потоке. При этом, если в каких-либо потоках возникают исключения, то все они собираются в одном объекте исключений типа AggregateException.

Обработка исключений в PLINQ

Рассмотрим следующий пример:

static void Main(string[] args)
{
    int[] ints = { 1, 2, 3, 4, 5, 6, 7, 8, 9, 0 };

    var query = ints.AsParallel().Select(i => 2 / i);
    try
    {
        query.ForAll(Console.WriteLine);    
    }
    catch (AggregateException ex) 
    { 
        Console.WriteLine(ex.Message);
    } 
    
}

Здесь мы создали запрос PLINQ в котором вычисляется значение » два поделить на элемент массива ints«. Так как в массиве присутствует значение 0, то это неизбежно должно привести к исключению типа System.DivideByZeroException как только очередной поток начнет вычислять выражение «2/0». Далее, в блоке try...catch мы выполняем запрос, используя метод ForAll PLINQ.

Теперь, если мы запустим приложение без отладки (Ctrl+F5), то увидим в консоли сообщение следующего содержания:

Если же приложение запустится в режиме отладки, то выполнение прервется на следующей строке:

Если мы продолжим выполнение приложение, то в консоль также выведется сообщение об ошибке.

Отмена параллельного запроса

Вполне возможно, что параллельный запрос может выполняться достаточно продолжительное время и нам потребуется механизм его отмены. Чтобы иметь возможность отмены параллельного запроса нам необходимо использовать метод WithCancellation(), передав ему в качестве параметра токен CancellationToken.

static void Main(string[] args)
{
    
    CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

    Task task = Task.Run(() =>
    {
        Task.Delay(100);
        cancellationTokenSource.Cancel(); //отменяем выполнение параллельного запроса  
    });


    int[] ints = { 1, 2, 3, 4, 5, 6, 7, 8, 9, 0 };

    var query = ints.AsParallel().Select(i => 2 / i).WithCancellation(cancellationTokenSource.Token);
    try
    {
        query.ForAll(Console.WriteLine);
    }
    catch (AggregateException ex)
    {
        Console.WriteLine(ex.Message);
    }
    catch (OperationCanceledException cancel)
    { 
        Console.WriteLine("Запрос прерван");
    }
    finally
    {
        cancellationTokenSource.Dispose();
    }
    
}

Для отмены операции мы запускаем отдельную задачу, которая через 100 мс. вызывает отмену (в приложении с графическим интерфейсом мы могли бы, например, «повесить» вызов сancellationTokenSource.Cancel() на клик по кнопке) . При отмене операции вызывается исключение типа OperationCanceledException , которое мы также перехватываем в блоке try..catch. Так как ещё до отмены задачи может быть вызвано исключение AggregateException, то его мы также обрабатываем.

Итого

Исключение, генерируемые в отдельных потоках при выполнении параллельного запроса PLINQ собираются в объекте AggregateException. Для обработки исключений PLINQ мы используем блоки try..catch. Чтобы отменить выполнение параллельного запроса мы должны использовать в запросе метод WithCancellation() в которые передается токен CancellationToken.