33.4. Обработка исключений и отмена выполнения задач

Создание параллельных приложений требует особых подходов при обработке исключительных ситуаций и прерывании работы задач. Исключительные ситуации могут возникнуть одновременно в разных потоках, а их обработка может выполняться в отдельном потоке. Задачи могут образовывать цепочку выполнения, а, значит, отмена одной задачи должна вести к отмене всех следующих задач цепочки.

В библиотеке параллельных расширений используются следующие принципы работы с исключительными ситуациями:

– при возникновении исключения в задаче (как созданной явно, так и порождённой неявно, например, методом Parallel.For()) это исключение обрабатывается средствами библиотеки (если перехват не был предусмотрен программистом) и направляется в ту задачу, которая ожидает завершения данной;

– при параллельном возникновении нескольких исключений все они собираются в единое исключение System.AggregateException, которое переправляется далее по цепочке вызовов задач;

– если возникла в точности одна исключительная ситуация, то на её основе будет создан объект AggregateException в целях единообразной обработки всех исключительных ситуаций.

Исключительные ситуации типа AggregateException могут возникать при работе со следующими конструкциями библиотеки параллельных расширений:

1. Класс Task – исключения, возникшие в теле задачи, будут повторно возбуждены в месте вызова метода Wait() данной задачи. Кроме того, исключение доступно через свойство Exception объекта Task.

2. Класс Task<T> – исключения, возникшие в теле задачи, будут повторно возбуждены в месте вызова метода Wait() или в месте обращения к экземплярному свойству Task<T>.Result.

3. Класс Parallel – исключения могут возникнуть в параллельно исполняемых итерациях циклов Parallel.For() и Parallel.ForEach() или в параллельных блоках кода при работе с Parallel.Invoke().

4. PLINQ – из-за отложенного характера исполнения запросов, исключения обычно возникают на этапе перебора элементов, полученных по запросу.

В библиотеке параллельных расширений применяется особый подход для выполнения отмены задач. Отметим, что многие методы задач, упоминавшиеся выше, принимают в качестве аргумент значение типа CancellationToken. Это так называемый токен отмены – своеобразный маркер того, что задачу можно отменить. Класс System.Threading.CancellationTokenSource содержит свойство Token для получения токенов отмены и метод Cancel() для отмены выполнения всех задач, использующих общий токен.

В следующем фрагменте кода демонстрируется типичный сценарий использования токенов отмены: вначале создаётся CancellationToken, затем его токен назначается задачам, а потом вызывается метод Cancel(), прерывающий выполнение всех задач:

// создаём токен и используем его в двух задачах

var token = new CancellationTokenSource();

// используем токен в двух задачах

new Task(SomeMethod, token.Token).Start();

Task.Factory.StartNew(OtherMethod, token.Token);

// в нужный момент отменяем обе задачи

token.Cancel();

33.5. Коллекции, поддерживающие параллелизм

Библиотека параллельных расширений содержит набор классов, представляющих коллекции с различным уровнем поддержки параллелизма. Указанные классы сосредоточены в пространстве имён System.Collections.Concurrent.

Класс BlockingCollection<T> является реализацией шаблона «поставщик-потребитель». Этот класс реализует интерфейсы IEnumerable<T>, ICollection, IDisposable и имеет собственные элементы, описанные в табл. 21.

Таблица 21

Элементы класса BlockingCollection<T>

Имя элемента	Описание
Add()	Добавляет элемент в коллекцию
AddToAny()	Статический метод, который добавляет элемент в любую из указанных BlockingCollection<T>
CompleteAdding()	После вызова этого метода добавление элементов невозможно
GetConsumingEnumerable()	Возвращает перечислитель, который перебирает элементы с их одновременным удалением из коллекции
Take()	Получает элемент и удаляет его из коллекции. Если коллекция пуста, и у коллекции был вызван метод CompleteAdding(), генерируется исключение
TakeFromAny()	Статический метод, который получает элемент из любой указанной BlockingCollection<T>
TryAdd()	Пытается добавить элемент в коллекцию, в случае успеха возвращает true. Дополнительно может быть задан временной интервал и токен отмены
TryAddToAny()	Статический метод, который пытается добавить элемент в любую из указанных коллекций
TryTake()	Пытается получить элемент (с удалением из коллекции), в случае успеха возвращает true
TryTakeFromAny()	Статический метод, который пытается получить элемент из любой указанной BlockingCollection<T>
BoundedCapacity	Свойство возвращает максимальное число элементов, которое можно добавить в коллекцию без блокировки поставщика (данный параметр может быть указан при вызове конструктора BlockingCollection<T>)
IsAddingCompleted	Возвращает true, если вызывался CompleteAdding()
IsCompleted	Возвращает true, если вызывался CompleteAdding(), и коллекция пуста

Продемонстрируем работу с BlockingCollection<T>, используя десять задач в качестве поставщика и одну в качестве потребителя:

BlockingCollection<int> bc = new BlockingCollection<int>();

for (int producer = 0; producer < 10; producer++)

{

Task.Factory.StartNew(() =>

{

Random rand = new Random();

for (int i = 0; i < 5; i++)

{

Thread.Sleep(200);

bc.Add(rand.Next(100));

}

});

}

var consumer = Task.Factory.StartNew(() =>

{

foreach (var item in bc.GetConsumingEnumerable())

{

Console.WriteLine(item);

}

});

consumer.Wait();

Классы ConcurrentQueue<T>, ConcurrentStack<T>, ConcurrentBag<T> и ConcurrentDictionary<T> – это потокобезопасные классы для представления очереди, стека, неупорядоченного набора объектов и словаря. Предполагается, что данные классы будут использоваться в качестве ресурсов, разделяемых между потоками, вместо обычных классов-коллекций. Отличительная особенность данных коллекций – наличие Try-методов для получения (изменения) элементов. Такие методы удобны, так как исключают предварительные проверки существования и необходимость использования в клиентском коде секции lock.

// используем объект типа ConcurrentStack<T>

T item;

if (concurrentStack.TryPop(out item)) // пытаемся извлечь элемент

{

UseData(item);

}