Обработка исключений при использовании PFX

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

Основные принципы работы с исключительными ситуациями при использовании библиотеки PFX состоят в следующем:

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

Имея представление о потенциальных местах, где могут возникнуть исключительные ситуации, при использовании параллельных конструкций библиотеки PFX, можно дать некоторые рекомендации о способах их перехвата и обработки. Общий принцип перехвата исключений в параллельной программе часто совпадает с аналогичным подходом для последовательных программ: перехватывать исключительные ситуации нужно, по возможности, как можно ближе к месту их возникновения. Например, если программа представляет собой целую иерархию задач, то, если есть возможность перехвата исключений в теле задач самого низкого уровня, то именно там и нужно применить механизм на базе конструкций try-catch, оставляя для главной задачи (корня иерархии) только функции управления исключительными ситуациями для задач более низкого уровня.

Например, при параллельной обработке совокупности изображений некоторые исключительные ситуации имеет смысл обработать на уровне рабочих потоков, а некоторые ? передать на более высокий уровень приложения:

public static void ProcessImages(string path)
{
      Parallel.ForEach(GetImageFiles(path), imageFilePath =>
      {
          try
          {
               Bitmap bmp = new Bitmap(imageFilePath);
               ProcessImage(bmp);
           }
          catch (UnauthorizedAccessException uae)
          {
               HandleUnauthorizedAccessException(uae);
          }
          catch (FileNotFoundException fnfe)
          {
               HandleFileNotFoundExceptions(fnfe);
           }
       });
}

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

При исполнении параллельной конструкции можно воспользоваться другим подходом, при котором все исключительные ситуации перехватываются и собираются в единую структуру данных в теле параллельной конструкции, а после окончания её исполнения эта структура передается, при необходимости, на более высокий уровень приложения. Это позволяет избежать досрочного прекращения исполнения параллельной конструкции при возникновении исключительной ситуации в одном из рабочих потоков:

var exceptions = new ConcurrentStack<Exception>();
Parallel.For(0, N, i=>
{
      try
      {
          Process(i);
      }
      catch(Exception exc) { exceptions.Push(exc); }
}

if (!exceptions.IsEmpty) throw new AggregateException(exceptions);

Семинарское занятие № 8. Модификация concurrent-структур данных во время перечисления их элементов

В состав PFX входят так называемые координационные структуры данных, среди которых имеются структуры для безопасного применения в многопоточной среде такие как, например, ConcurrentQueue<T> и СoncurrentStack<T>.

PFX-аналоги классов, содержащихся в пространстве имен System.Collections, имеют, во многом, сходные с ними свойства. Так, например, для объектов обоих видов классов допускается перечисление элементов этих объектов с помощью цикла foreach. Однако, стоит обратить вниманте, что для объектов обычных классов-коллекций запрещена модификация этих объектов в процессе перечисления. Например, выполнение кода, приведенного ниже, вызовет исключительную ситуацию InvalidOperationException: "Collection was modified after the enumerator was instantiated. ":

var q = new Queue<int> (new[] { 1, 2, 3, 4, 5 }); 
foreach (var item in q) 
{ 
    if (item <= 5) q.Enqueue(item * 6); 
}

Если же изменить тип переменной q c Queue<int> на тип concurrent-коллекции ConcurrentQueue<int>:

var q = new ConcurrentQueue<int> (new[] { 1, 2, 3, 4, 5 }); 
foreach (var item in q) 
{ 
    if (item <= 5) q.Enqueue(item * 6); 
}

то выполнение такого кода уже не вызовет исключительной ситуации, а в после выполнения цикла foreach, согласно правил, зафиксированных в текущей реализации библиотеки PFX, в очереди будет находиться 10 элементов.

Такое изменение свойств объектов-коллекций было специально принято в PFX, поскольку параллельная обработка (в т.ч., параллельное перечисление и изменение) concurrent-структур данных является основным вариантом их использования.

Задача 1.

Перечислите элементы, которые будут содержаться в очереди, после выполнения следующего кода:

var q = new ConcurrentQueue<int> (new[] { 1, 2, 3, 4, 5 }); 
foreach (var item in q) 
{ 
    if (item <= 5) q.Enqueue(item * 6); 
}

Сохранение порядка возвращаемых значений при параллельных вычислениях

При параллельной обработке, сохранение порядка значений, возвращаемых программой, часто требует от программиста написания дополнительного кода, сохраняющего этот порядок. Представим себе последовательное приложение, которое реализует рендеринг (порождение) кадров, допустим, видеофильма и последующую их запись в видеофайл:

for (int i = 0; i < numberOfFrames; i++)
{
    var frame = GenerateFrame(i);
    WriteToMovie(frame);
}

Этот же алгоритм можно реализовать и с помощью средств LINQ:

var frames = from i in Enumerable.Range(0, numberOfFrames)
             select GenerateFrame(i);
foreach (var frame in frames) WriteToMovie(frame);

Выполнение метода GenerateFrame может занимать достаточно много времени, поэтому имеет смысл распараллелить генерацию видеокадров. Для примера, предположим, что исполнение метода GenerateFrame в отдельном потоке является безопасным (т.е., использование общих данных, если таковые имеются, одновременно с другими потоками происходит корректно), и что порождение очередного кадра не зависит от предыдущих кадров. При параллельной генерации отдельных кадров, необходимо, тем не менее, предусмотреть средства блокировки при записи полученных кадров из разных потоков в один и тот же видеофайл:

using (ManualResetEvent mre = new ManualResetEvent(false))
{
    int count = numberOfFrames;
    object obj = new object();
    for (int i = 0; i < numberOfFrames; i++)
    {
        ThreadPool.QueueUserWorkItem(state =>
        {
            var frame = GenerateFrame((int)state);
            lock(obj) WriteToMovie(frame);
 
            if (Interlocked.Decrement(ref count) == 0) 
                mre.Set();

        }, i );
    }
    mre.WaitOne();
}

Легко видеть, что полученное параллельное решение имеет серьезный недостаток - оно не обеспечивает правильный порядок сгенерированных кадров в результирующем видеофайле. Сохранить порядок возвращаемых значений можно, например, с использованием массива событий ManualResetEvent:

var frames = new Bitmap[numberOfFrames];
var events = (from i in Enumerable.Range(0, numberOfFrames)
              select new ManualResetEvent(false)).ToArray();

for (int i = 0; i < numberOfFrames; i++)
{
    ThreadPool.QueueUserWorkItem(state =>
    {
        int frameNum = (int)state;
        frames[frameNum] = GenerateFrame(frameNum);
        events[frameNum].Set();
    }, i);
}
 
for (int i = 0; i < numberOfFrames; i++)
{
    events[i].WaitOne();
    WriteToMovie(frames[i]);
}

Представленный выше код является корректным - он обеспечивает правильный порядок кадров в результирующем видеофайле, но не эффективным. Неэффективность состоит в том, что на каждый генерируемый кадр заводится отдельное событие ManualResetEvent, которое реализуется через исполнение некоторого кода из ядра операционной системы. Замен событий можно воспользоваться классом Future<T> для реализации того же алгоритма:

var frames = (from i in Enumerable.Range(0, numberOfFrames)
              select Future.Create(() => GenerateFrame(i))).
             ToArray();
foreach (var frame in frames) WriteToMovie(frame.Value);

В приведенном выше фрагменте, вместо массива событий создается массив объектов класса Future<T>. Исполнение соответствующих им делегатов происходит параллельно, а при последовательной записи кадров в видеофайл происходит ожидание завершения работы соответствующего делегата. Код стал более кратким и более эффективным.

Возможен еще один вариант решения этой задачи, в котором не используются средства LINQ, но в добавок к Future<T> задействована очередь:

var frames = new Queue<Future<Bitmap> > ();
for (int i = 0; i < numberOfFrames; i++)
{
    var num = i;
    frames.Enqueue(Future.Create(() => GenerateFrame(num)));
}
while (frames.Count > 0) WriteToMovie(frames.Dequeue().Value);

Наконец, для решения этой задачи средствами PLINQ, можно применить опцию PreserveOrdering перечисления ParallelQueryOptions:

var frames = from i in Enumerable.Range(0, numberOfFrames).
    AsParallel(ParallelQueryOptions.PreserveOrdering)
             select GenerateFrame(i);
foreach (var frame in frames) WriteToMovie(frame);

Задача 1.

Пусть дан массив A, содержащий натуральные числа, большие 1. Напишите параллельную программу, которая записывает в другой массив B все простые числа из массива A, сохраняя их относительный порядок в исходном массиве. Т.е., если массив A содержит, например, числа 14,13,21,25,31,20,17, то массив B должен содержать числа 13, 31, 17 в указанном порядке.