Распараллеливание циклов. Класс Parallel

Короткие и длинные итерации

Рассмотрим цикл, допускающий распараллеливание:

for(int i = 0; i < n; i++) { body }

Возникает естественный вопрос, сколько потоков следует создавать для оптимального выполнения этого цикла? Если создавать n потоков, каждый из которых будет выполнять одну итерацию, то время выполнения цикла $$T_с$$ можно представить следующим образом:

Здесь $$n \cdot T_r$$ это накладные расходы, связанные с созданием и удалением потоков, а $$T_b$$ - это максимальное время выполнения одной итерации. Для хорошо сбалансированных итераций можно полагать, что $$T_b$$ - это время выполнения одной итерации.

Накладные расходы зависят от n - длины цикла. Кажется, что для больших значений n накладные расходы могут быть слишком велики. Кажется, что наилучшие результаты могут быть достигнуты, когда n сопоставимо с числом ядер компьютера. Каждая итерация выполняется в отдельном потоке, каждый поток выполняется отдельным ядром процессора. Значит ли это, что на четырехядерном компьютере, на котором я работаю, распараллеливать имеет смысл циклы длиной не более 10? Если это утверждение верно, то как перейти к короткому циклу, когда n велико?

Это нетрудно сделать, используя классический прием разделения длинного цикла на два цикла. Область изменения индекса [0, n-1] разбивается на группы (сегменты) и вначале идет внешний цикл по числу групп, а внутренний цикл идет по элементам группы. По сути это означает применение сегментного алгоритма, описанного в предыдущих главах. Рассмотрим цикл:

for(int i = 0; i < n; i++) { body(i) }

Область изменения индекса цикла разобьём на p сегментов. Заменим наш цикл двумя циклами:

int p = 10;
  int m = n / p;            
  int start = 0, finish = 0;
      //Внешний цикл распараллеливается
      Parallel.For(0, p, (j) =>
         {
           start = j * m;
           finish = (start + m < n) ? start + m : n;
           //Внутрений цикл удлиняет итерацию внешнего цикла
           for (int i = start; i < finish; i++)
           {
             body(i);
            }
          });

При замене цикла for циклом Parallel.For возникает естественный вопрос, что лучше иметь ли короткий параллельный цикл и длинные итерации или длинный параллельный цикл с короткими итерациями?

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

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

/// <summary>
        /// Разбиение на два цикла 
        /// Применяем Parallel.For к внешнему циклу 
        /// </summary>
        static void Sample6P()
        {
            int[] temp = new int[n];
            int p = 100;
            int m = n / p;            
            int start = 0, finish = 0;
            //Внешний цикл распараллеливается
            Parallel.For(0, p, (j) =>
            {
                start = j * m;
                finish = (start + m < n) ? start + m : n;
                //Внутрений цикл удлиняет итерацию внешнего цикла
                for (int i = start; i < finish; i++)
                {
                    x[i] = Fx(i);
                    temp[i] = Fs(i);
                }
            });
            S = 0;
            for (int i = 0; i < n; i++)
                S = S + temp[i];
        }

Как ведет себя Parallel.For в этой ситуации? Вот результаты теста:

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

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

Построим простой класс для работы с матрицами. Для наших исследований ограничимся рассмотрением квадратных матриц. Вот общая часть нашего класса:

/// <summary>
    /// Работа с квадратными матрицами
    /// </summary>
    class MultMatr
    {
        //Размер матриц
        int n;
        //Квадратные матрицы [n, n]
        int[,] A, B, C;
        public int dc, dab;
        Random rnd = new Random();
        public MultMatr(int n)
        {
            this.n = n;
            A = new int[n, n];
            B = new int[n, n];
            C = new int[n, n];
            dc = dab = 0;
            
        }
        public void Init_AB()
        {
            for (int i = 0; i < n; i++)
            {
                A[i, i] = rnd.Next(1, 10);
                B[i, i] = rnd.Next(1, 10);
            }
        }

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

/// <summary>
        /// Классический последовательный 
        /// алгоритм умножения матриц
        /// </summary>
        public void MultS()
        {
            for (int i = 0; i < n; i++)
                for (int j = 0; j < n; j++)
                {
                   C[i, j] = 0;
                    for (int k = 0; k < n; k++)
                        C[i, j] += A[i, k] * B[k, j];
                }
        }

Вот версия, где распараллеливается только внешний цикл:

/// <summary>
        /// Умножение матриц
        /// Распараллеливание внешнего цикла
        /// </summary>
        public void MultOuterFor()
        {
            Parallel.For(0, n, (i) =>
                {
                    for (int j = 0; j < n; j++)
                    {
                        C[i, j] = 0;
                        for (int k = 0; k < n; k++)
                            C[i, j] += A[i, k] * B[k, j];
                    }
                });
        }

В следующей версии распараллеливаются два цикла:

/// <summary>
        /// Умножение матриц
        /// Распараллеливание внешнего 
        /// и внутреннего цикла
        /// </summary>
        public void MultOuterInnerFor()
        {
            Parallel.For(0, n, (i) =>
            {
                Parallel.For(0, n, (j) =>
                {
                    C[i, j] = 0;
                    for (int k = 0; k < n; k++)
                        C[i, j] += A[i, k] * B[k, j];
                });
            });
        }

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

/// <summary>
        /// Умножение матриц
        /// Распараллеливание длинного цикла
        /// </summary>
        public void MultLongCicle()
        {
            Parallel.For(0, n * n, Mult);
        }

Мы выполнили свертку двух циклов в один. Метод Mult задает итерацию, выполняемую на каждом шаге цикла. Методу передается параметр цикла и итерации, согласно семантике Parallel.For, могут выполняться в произвольном порядке и параллельно.

Метод Mult имеет вид:

void Mult(int q)
        {
            int i = 0, j = 0;
            i = q / n; j = q - i * n;
            {
                C[i, j] = 0;
                for (int k = 0; k < n; k++)
                    C[i, j] += A[i, k] * B[k, j];
            }
        }

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

public void Check()
        {
            dc = dab = 0;
            for (int i = 0; i < n; i++)
            {
                dc += C[i, i];
                dab += A[i,i] * B[i, i];
            }
        }

А теперь построим тест, позволяющий выяснить эффективность различных версий умножения матриц:

static void TestMM()
        {
            MultMatr mm = new MultMatr(n);
            mm.Init_AB();            
            T = MyTimer(mm.MultS);
            Console.WriteLine("Последовательный алгоритм умножения матриц");
            Console.WriteLine("T =" + T);
            mm.Check();             
            Console.WriteLine("Results: " + mm.dc + " : " + mm.dab);           
            T = MyTimer(mm.MultOuterFor);
            Console.WriteLine("Распараллелен внешний цикл");
            Console.WriteLine("T =" + T);
            mm.Check();
            Console.WriteLine("Results: " + mm.dc + " : " + mm.dab);
            T = MyTimer(mm.MultOuterInnerFor);
            Console.WriteLine("Распараллелен внешний и внутренний цикл");
            Console.WriteLine("T =" + T);
            mm.Check();
            Console.WriteLine("Results: " + mm.dc + " : " + mm.dab);
            T = MyTimer(mm.MultLongCicle);
            Console.WriteLine("Распараллелен длинный цикл");
            Console.WriteLine("T =" + T);
            mm.Check();
            Console.WriteLine("Results: " + mm.dc + " : " + mm.dab);
        }

Метод MyTimer измеряет время работы метода, переданного ему в качестве параметра. Реализация его стандартна:

static long MyTimer(VV par)
        {
            DateTime start, finish;
            start = DateTime.Now;
              par();
            finish = DateTime.Now;
            return(finish - start).Ticks;
        }

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

При таких размерах распараллеливание не имеет особого смысла, поскольку с работой прекрасно справляется и последовательный алгоритм. Но, заметьте, и параллельные версии не приводят к заметным накладным расходам.

Увеличим размер перемножаемых матриц в 10 раз (n = 100), объем вычислений при этом увеличится в 1000 раз. Посмотрим, как это скажется на результатах:

Нетрудно видеть, что компьютер легко справляется с такими вычислениями. Практически время вычислений измеряется миллисекундами, что сравнимо с ошибкой измерения таймера. Но уже в этом случае заметно некоторое превосходство параллельных алгоритмов над последовательным. Какие либо заключения о том, какая из параллельных версий является предпочтительной пока делать рано. Увеличим размер перемножаемых матриц еще в 10 раз (n = 1000). Вот как выглядят результаты:

На моем компьютере с 4-мя ядрами все параллельные версии обеспечивают 4-х кратное ускорение. Лучшие результаты показывает версия, в которой распараллеливается только внешний цикл. Но, заметьте, и другие версии показывают сравнимые результаты. Проведем еще один заключительный эксперимент, увеличив размер матриц до 2000. Время вычислений теперь уже будет измеряться десятками секунд для параллельных версий, а для последовательного алгоритма счет пойдет на минуты. Вот результаты теста:

И здесь параллельные версии показывают 4-х кратное ускорение в сравнении с последовательным алгоритмом. Лучшей является версия с распараллеливанием одного внешнего цикла.

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

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

Управление циклом при распараллеливании

Напомню, метод Parallel.For имеет множество реализаций и по синтаксису является функцией, возвращающей значение типа ParallelLoopResult. До сих пор метод вызывался как оператор, и возвращаемое значение никак не использовалось. Попробуем разобраться, как и в каких ситуациях следует использовать значение, возвращаемое методом.

Рассмотрим обычный оператор цикла for. Возможны следующие ситуации при его выполнении:

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

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

Если одна или несколько итераций завершаются преждевременно, то и здесь кроме завершения текущей итерации ничего предпринимать не нужно, поскольку другие итерации запускаются автоматически.

Если на итерации с номером i выполнился метод break, то цикл необходимо завершить. Для параллельного выполнения это означает завершение всех уже запущенных итераций, запуск на выполнение всех еще не запущенных итераций с номерами, меньшими i. Запуск итераций с номерами, большими i, отменяется, если они конечно еще не были запущены. Заметьте, поскольку одновременно выполняются несколько итераций, то break может выполняться не один раз. Если например break выполнился на итерациях с номерами 10 и 20, то запускать нужно итерации, номера которых меньше 10.

Если на итерации с номером i возникла исключительная ситуация, то цикл необходимо завершить, а ситуацию обработать. Поскольку одновременно выполняются несколько итераций, то исключительные ситуации могут возникать на разных итерациях. Поэтому выбрасывается агрегирующая исключительная ситуация Aggregation Exception, которая содержит информацию обо всех возникших исключительных ситуациях. Следует предусмотреть catch блок, обрабатывающий эту ситуацию.

Как же выполнить оператор break при параллельном исполнении? Для этого нужно использовать перегруженную версию Parallel.For, в которой методу, выполняющему итерацию, помимо индекса цикла передается дополнительный параметр класса ParallelLoopState. Объект этого класса может вызывать метод Break, который и реализует стратегию прерывания для параллельного выполнения. Анализируя возвращаемое значение метода Parallel.For, можно узнать минимальный номер итерации, на которой впервые выполнялось прерывание (break). Помимо метода Break объект класса ParallelLoopState может вызывать и метод Stop, который также завершает все выполняемые итерации, но, в отличие от Break, не вызывает на исполнение итерации с меньшими номерами. Вызов Stop означает остановить выполнение. В этом случае теряет смысл понятие итерации с минимальным номером, прервавшей выполнение.

Давайте рассмотрим пример на тему прерывания исполнения. В программировании есть знаменитая проблема, связанная с целыми числами, которую иногда называют проблемой "3x + 1", а сами преобразуемые числа называют числами - градинами. Суть ее в следующем: если взять любое целое число и проводить над ним в цикле достаточно простые преобразования, получая новое число, то результат обязательно сойдется к единице. Строго обосновать этот факт, доказав завершаемость достаточно простого цикла, пока никому не удалось.

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

/// <summary>
        /// Проблема 3x + 1
        /// Поиск медленно сходящихся чисел-градин
        /// </summary>
        static void Break_Stop_Test()
        {            
            ParallelLoopResult res;
            res = Parallel.For(2, n, Grad);
            Console.WriteLine(res.LowestBreakIteration.ToString());
            Console.WriteLine(res.IsCompleted.ToString());
        }

Значение, возвращаемое методом Parallel.For, представляет структуру ParallelLoopResult, у которой два поля. Булевское поле IsCompleted возвращает значение true, если все итерации закончились без происшествий, и false, в противном случае. Если на одной или нескольких итерациях выполнялся break, то поле LowestBreakIteration вернет значение минимальной итерации.

Приведу текст метода Grad, выполняющего итерацию:

static void Grad(int i, ParallelLoopState pls)
        {
            int N = i;
            const int m = 150;
            int k =0;
            while (N != 1 )
            {
                if (N % 2 == 0)
                    N = N / 2;
                else
                    N = 3 * N + 1;
                k++;
               // if (k == m) pls.Stop();  
                if (k == m) pls.Break();                              
            }
        }

Методу передаются два параметра. Объект pls класса ParallelLoopState позволяет в нужный момент вызвать метод Break или метод Stop. В данном исследовании нас интересует именно Break, чтобы обнаружить самое маленькое число, на котором впервые достигается заданный предел. Поэтому возможность вызова метода Stop показана, но закомментирована.

У объектов класса ParallelLoopState есть ряд полезных свойств, позволяющих разным итерациям цикла обмениваться информацией о возникающих событиях. Но мы не будем демонстрировать эти возможности.

Наш пример позволяет обнаружить первое число, которое после проведения m шагов цикла while все еще не сошлось к единице. Вот результаты выполнения нашего теста:

Число 703 - это первое число, для которого требуется не менее 150 преобразований, чтобы оно сошлось к единице. Цикл был прерван, не все его итерации были выполнены, на что указывает значение false свойства IsCompleted.

Управление исключительными ситуациями при распараллеливании

Рассмотрим теперь пример, отражающий организацию обработки исключительных ситуаций, которые могут возникать в ходе выполнения параллельного цикла. Будем моделировать работу прибора, управляющего температурным режимом. Предполагается, что числа, появляющиеся на итерации, задают значение температуры. На итерации могут возникать исключения двух моделируемых нами типов. Если полученное значение температуры слишком велико, то "выбрасывается" исключение, свидетельствующее о "высокой температуре". В другой ситуации "выбрасывается" исключение, свидетельствующее о "низкой температуре".

Вот как устроен наш тест:

/// <summary>
        /// Выбрасывание исключений и их обработка
        /// </summary>
        static void HLTest()
        {
            ParallelLoopResult res = new ParallelLoopResult();
            try
            {
                res = Parallel.For(0, n, Temperature);
            }
            catch (LowTemperatureException e)
            {
                Console.WriteLine(e.Message);
            }
            catch (HighTemperatureException e)
            {
                Console.WriteLine(e.Message);
            }

            catch (AggregateException ae)
            {
                Console.WriteLine(ae.Message);
                ae.Handle((x) =>
                {
                    if (x is HighTemperatureException)
                    {
                      Console.WriteLine(
                  "Агрегированное сообщение: Высокая температура");
                        return true;
                    }
                    else
                        if (x is LowTemperatureException)
                        {
                          Console.WriteLine(
                      "Агрегированное сообщение: Низкая температура");
                            return true;
                        }
                        else return false;
                });
            }
            finally
            {
                Console.WriteLine(res.IsCompleted.ToString());
                Console.WriteLine(res.LowestBreakIteration.ToString());
            }
        }

Метод Parallel.For помещается в try-блок. После try-блока следуют три catch-обработчика исключительной ситуации. Заметьте, первые два бесполезны, несмотря на то, что они пытаются перехватить фактически возникающие исключения. Дело в том, что при параллельном выполнении все исключения перехватываются и собираются в одно агрегированное исключение AggregateException, которое и перехватывает специальный catch-обработчик.

У объектов типа AggregateException есть замечательный метод Handle, позволяющий получить все исключения, возникшие на параллельно исполняемых итерациях цикла. Эти исключения можно тут же проанализировать и выполнить их обработку. Наш пример демонстрирует этот способ обработки исключений.

Давайте рассмотрим метод, выполняющий итерацию, в ходе которой и могут появляться исключения того или иного типа:

static void Temperature(int i, ParallelLoopState pls)
        {
            const int m = 1000;
          //моделируем показания прибора
            int N = rnd.Next(-m, m);
            if (N > 777) throw new HighTemperatureException(
                      "Высокая температура");
            if (N < -777) throw new LowTemperatureException(
                      "Низкая температура");            
        }

Пример прост и в комментариях не нуждается. Для полноты картины приведем классы, описывающие моделируемые нами исключения:

class HighTemperatureException : Exception
        {
            public HighTemperatureException() { }
            public HighTemperatureException(string message): base(message){ }
            public HighTemperatureException(string message, 
                Exception e) : base(message, e) { }
        }
        class LowTemperatureException : Exception
        {
            public LowTemperatureException() { }
            public LowTemperatureException(string message) : base(message){ }
            public LowTemperatureException(string message,
                Exception e)
                : base(message, e) { }
        }

В завершение приведу результаты одного сеанса выполнения теста:

Оператор Parallel.ForEach

Принципиально, все, что было сказано о распараллеливании циклов с использованием оператора Parallel.For, относится и к оператору (методу) Parallel.ForEach. Разница такая же, как и между обычными операторами for и foreach. Оператор ForEach позволяет распараллелить обработку элементов некоторой коллекции - массивов, списков, словарей, - предоставляя возможность обработки каждого элемента в отдельном потоке.

Для оператора Parallel.ForEach сохраняется ограничение, характерное для его прототипа - обычного оператора foreach, - элемент коллекции можно использовать только для чтения, но не для его изменения. Оператор в некотором порядке предоставляет элемент за элементом из коллекции. Если быть точным, то элементы выбираются из некоторого буфера, создаваемого при работе с коллекцией. Предоставляемый элемент программист может изменять, но эти изменения никак не отразятся на элементах самой коллекции, поскольку предоставляется локальный объект и все изменения носят локальный характер. При завершении метода локальный элемент перестает существовать, и все изменения пропадают вместе с самим элементом. Ситуация аналогична передаче методу параметра значимого типа, заданного без описателя ref или out. Для такого входного параметра создается локальная копия, существующая только на время выполнения метода.

В цикле Parallel.ForEach можно создавать элементы новой коллекции, но нельзя модифицировать коллекцию, предоставляемую оператором цикла.

Метод Parallel.ForEach перегружен, мы ограничимся рассмотрением его простейшей версии, имеющей следующий синтаксис:

public static ParallelLoopResult ForEach<TSource>(
  IEnumerable<TSource> source,  Action<TSource> body)

Метод представляет функцию, возвращающую тот же результат, что и метод Parallel.For.

У метода в этой реализации два аргумента, - первый представляет коллекцию, по элементам которой выполняется цикл, второй - метод, которому представляется элемент коллекции.

Корректная работа предполагает, что итерации независимы, так что обработка каждого элемента коллекции может выполняться независимо, параллельно и в любом порядке.

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

Начнем с создания класса Robot. Вот общая часть этого класса:

/// <summary>
    /// Класс, моделирующий работу с коллекцией роботов
    /// </summary>
    public class Robots
    {
        //число роботов
        int n;
        //число характеристик робота                   
       int m;
       //коллекция роботов                   
       public List<StructR> robots;
       //коллекция,создаваемая при обработке
       public List<StructR> results;
       //глобальный ключ закрытия критической секции
       object locker = new object();    
        Random rnd = new Random();
        /// <summary>
        /// Конструктор
        /// </summary>
        /// <param name="n">число роботов</param>
        public Robots(int n)
        {
            this.n = n;
            m = 5;
            robots = new List<StructR>(n);
            results = new List<StructR>(n);           
        }

Пояснения смысла полей класса даны в комментариях к проекту. Коллекция роботов задана списком, элементы которого представляют структуру StructR, характеризующую робота. В нашей модели она проста:

/// <summary>
    /// Характеристика робота
    /// </summary>
    public struct StructR
    {
        string id;
        int[] marks;
        double average_ball;
        public string Id
        {
            get { return id; }
            set { id = value; } 
        }
        public int[] Marks
        {
            get { return marks; }
            set { marks = value; }
        }
        public double Average_ball
        {
            get { return average_ball; }
            set { average_ball = value; }
        }
    }

Здесь, поле id идентифицирует робота, marks - это его характеристики, а average_ball - это характеристика, которую необходимо вычислить в результате обработки оценок marks.

Добавим теперь в класс Robot метод Init, позволяющий моделировать создание коллекции. Поскольку создание каждого робота можно вести независимо, то используем распараллеливание и в работе этого метода:

/// <summary>
        /// Инициализация списка robots
        /// </summary>
        public void Init()
        {  
            try
            {
               ParallelLoopResult res = Parallel.For(0, n, Init_Robots);
               if (!res.IsCompleted)
                   Console.WriteLine("Ошибки при инициализации списка");
               else
                   Console.WriteLine("все итерации завершились нормально");
            }
            catch (AggregateException ae)
            {
                Console.WriteLine(ae.Message);
                ae.Handle((x) =>
                    {
                        Console.WriteLine(x.Message);
                        return true;
                    });
            }
        }

Заметьте, на этапе разработки проекта лучше использовать полную форму с получением результата работы цикла, с обработкой возможных исключительных ситуаций. Затем, когда проект отлажен, будучи уверенным в корректности инициализации, можно обойтись одной строчкой:

Parallel.For(0, n, Init_Robots);

Метод Init_Robots - это метод, выполняемый на каждой итерации, которому передается индекс итерации цикла:

/// <summary>
        /// Создание робота
        /// </summary>
        /// <param name="k">индекс итерации цикла
        /// В методе не используется</param>
         void Init_Robots (int k)
         {
            StructR sr = new StructR();
            string id = "R" + rnd.Next(1, 1000);
            int[] marks = new int[m];
            for (int i = 0; i < m; i++)
            {
                marks[i] = rnd.Next(5, 25);
            }
            sr.Id = id;
            sr.Marks = marks;
            lock (locker)
            {
                robots.Add(sr);                       
            }             
        }

Здесь создается объект sr типа StructR и этот объект, моделирующий робота, добавляется в коллекцию (список) роботов. Поскольку список является общим ресурсом, то добавление нового элемента коллекции помещается в критическую секцию, закрываемую ключом locker.

Добавим теперь в класс Robot метод, позволяющий проводить параллельную обработку созданной коллекции. В этом методе используем конструкцию Parallel.For:

/// <summary>
        /// Обработка коллекции robots
        /// </summary>
        public void CountAverage()
        {
            Parallel.ForEach(robots, CAV);
        }

Для ясности понимания вызова я не стал приводить полную форму с получением результата и обработкой возможной исключительной ситуации. Она такая же, как и для оператора Parallel.For.

Итак, мы видим, что при вызове метода Parallel.ForEach ему передается коллекция robots и метод CAV, обрабатывающий элемент коллекции. Вот как выглядит этот метод в нашем случае:

void CAV(StructR sr)
        {            
            sr.Average_ball = 0;
            for (int i = 0; i < m; i++)
                sr.Average_ball += sr.Marks[i];
            sr.Average_ball = Math.Round(sr.Average_ball / m, 2);            
            lock (locker)
            {
                results.Add(sr);
            }
        }

Наша цель состоит в том, чтобы изменить значения поля в структуре, характеризующей робота. Конечно, хотелось бы, чтобы значение поля sr.Average_ball, вычисляемое в цикле, изменялось бы непосредственно для каждого элемента обрабатываемой коллекции robots, но, как уже говорилось, конструкция ForEach этого не позволяет. Поэтому в методе создается новая коллекция results, аналогичная коллекции robots, отличающаяся заполненным полем Average_ball. И здесь, критическая секция, в которой ведется работа с общим ресурсом, закрывается ключом locker.

В завершение нашей модели добавим еще один метод, позволяющий выбрать одного из роботов коллекции. В данном случае "лучшим" будем признавать робота с наибольшим баллом Average_ball. Хотя вычисление максимума также может быть распараллелено, о чем говорилось в предыдущих главах, но мы ограничимся последовательным алгоритмом, в котором используется обычный оператор foreach:

/// <summary>
        /// Нахождение лучшего
        /// </summary>
        /// <returns>робот с максимальным баллом</returns>
        public StructR Best()
        {
            StructR best = results.ElementAt(0);
            foreach (StructR item in results)
                if (item.Average_ball > best.Average_ball)
                    best = item;
            return best;
        }

Приведу теперь процедуру Main консольного проекта, оживляющую нашу модель и приводящую ее в действие:

class Program
    { 
        static int n = 15;
        static Robots my = new Robots(n);
        static void Main(string[] args)
        {
            my.Init(); 
            my.CountAverage();
            PrintRobots();
            StructR best = my.Best();
            Console.WriteLine("Лучший робот");
            PrintOne(best);
        }
        static void PrintRobots()
        {
            StructR item;
            if( n < 20)
                for (int i = 0; i < n; i++)
                {
                    item = my.results.ElementAt(i);
                        PrintOne(item);
                }
        }
        static void PrintOne(StructR one)
        {
            Console.WriteLine("ID : {0} Оценка : {1}",
                one.Id, one.Average_ball);
        }
    }

Осталось привести результаты работы:

Об одной "классической" ошибке при параллельном программировании

К приведенному выше проекту можно высказать несколько упреков, указав на характерные ошибки программирования. Вот некоторые из них:

Можно указать и на другие ошибки, нарушающие стиль программирования. Я иногда сознательно иду на подобные нарушения для краткости текста и ясности изложения. Но на одной ошибке, которую я сделал в ходе разработки проекта, хочу остановиться подробнее, поскольку, полагаю, она является типичной ошибкой тех, кто начинает работать с параллельными программами. В методах Init_Robots и CAV мне понадобилось закрывать критическую секцию объектом locker:

lock(locker){<критическая секция>}

Где нужно объявлять объект locker? Создавая методы Init_Robots и CAV, я там же объявил и объект locker, представляющий ключ, закрывающий секцию. Когда я начал отладку проекта, то для небольших значений n все работало прекрасно. Но при параллельных вычислениях отладка на "малых" примерах, широко применяемая в последовательном программировании, мало что дает, - ошибки проявляются на "больших" данных. Уже при n, больших 100, из-за гонки данных стали теряться элементы коллекции. Причина в том, что объявленный ключ представлял локальную переменную. В результате каждый поток открывал критическую секцию своим ключом, что и приводило к гонке данных.

Ключ должен быть глобальной переменной - полем класса, как это сделано в нашем проекте. Помните об этом.

Метод Parallel.Invoke

Нам осталось рассмотреть третий и последний метод класса Parallel - метод Invoke. Это самый простой по синтаксису и по семантике метод. У него всего две реализации. Синтаксис основной реализации имеет вид:

public static void Parallel.Invoke(params Action[] actions)

Поскольку параметр actions объявлен с описателем params, то фактический параметр может представлять список с произвольным числом элементов. Каждый элемент этого списка задает имя метода, сигнатура которого удовлетворяет делегату Action. Никакие параметры методу не передаются. Если же методу необходимо передать информацию, то всегда можно использовать тот факт, что любой вызов метода всегда имеет цель - объект, вызывающий метод. Поэтому методу Invoke в качестве аргумента можно передавать некоторый объект, вызывающий метод без параметров. Вся необходимая информация, как входная, так и выходная, создаваемая вызываемым методом, передается через поля объекта.

Все методы, переданные при вызове Invoke, могут выполняться параллельно и в произвольном порядке. Что происходит, когда один из методов заканчивает свою работу нормальным образом или, не дай бог, в результате выполнения исключительной ситуации? Все остальные методы продолжат свою работу, в том числе будут запущены на выполнение все еще не начавшие работать методы, ждущие своей очереди. Метод Invoke заканчивает свое выполнение только тогда, когда закончат работу все методы из переданного ему списка. Все возникшие исключительные ситуации собираются в одну ситуацию - AggregateException, - которая и выбрасывается в этом случае по завершении работы метода Invoke. Естественно, необходимо предусматривать обработчик этой ситуации.

Рассмотрим пример, моделирующий ситуацию с вызовом Invoke. Вспомним сказку о Золушке, которая не могла поехать на бал, поскольку злая мачеха, поручила ей столько работ, что сама Золушка, работая последовательно, никак не могла бы с ними справиться до начала бала. Благодаря фее, появились дополнительные процессоры, выполняющие работы параллельно. В результате, все работы были выполнены, а Золушка смогла попасть на бал и очаровать принца. Вот модель этой сказочной истории:

/// <summary>
    /// Работы Золушки
    /// </summary>
    class Zolushka_Jobs
    {
        delegate void Job();
        Job[] jobs;
        string[] messages;
       Prince prince = new Prince("Гарри");
        public Zolushka_Jobs()
        {
            const int N = 5;
            messages = new string[N];
            jobs = new Job[N];
            jobs[0] = Job_One;
            jobs[1] = Job_Two;
            jobs[2] = Job_Three;
            jobs[3] = Job_Four;
            jobs[4] = prince.Love;
        }
        public string[] Messages
        {
            get { return messages; }
        }

Добавим теперь в класс методы, моделирующие работы Золушки:

void Job_One()
        {
            messages[0] = "Пол вымыт!";
        }
        void Job_Two()
        {
            messages[1] = "Посуда вымыта, вычищена и сверкает!";
        }
        void Job_Three()
        {
            messages[2] = "Фасоль отсортирована!";
        }
        void Job_Four()
        {           
            messages[3] = "Принц очарован!";
        }

Следующий метод запускает параллельное выполнение всех работ:

public void Make_All_Jobs()
        {            
            Parallel.Invoke(Job_One, Job_Two, Job_Three, Job_Four,
                                             prince.Love);            
           // Parallel.For(0, 5, Make); 
            messages[4] = prince.Message;        
        }

Обратите внимание, первые четыре метода, передаваемые Invoke, это методы класса Zolushka. Для них целью является текущий объект этого класса (this). Но параллельно будет выполняться и метод другого класса - метод Love из класса Prince, вызываемый объектом prince этого класса. Это демонстрирует возможность в одном вызове Invoke параллельно исполнять методы разных классов.

Метод Invoke аналогичен параллельному циклу, его можно заменить оператором For. Можно создать массив работ jobs, как это сделано в конструкторе класса Zolushka, а затем запустить работы на выполнение, выполнив вызов:

Parallel.For(0, 5, Make);

Этот вызов включен в процедуру Make_All_Jobs. Он закомментирован, но вызовы Invoke и For дают эквивалентные результаты.

Приведу текст метода Make:

void Make(int i)
        {
            jobs[i]();            
        }

Для полноты картины приведу текст класса Prince с методом Love:

class Prince
    {
        string name;
        string message;
        public string Message
        {
            get { return message; }
        }
        public Prince(string name)
        {
            this.name = name; 
        }
        public void Love()
        {
            message = String.Format("Я, принц {0}, влюбился в Золушку!",
                                                     name);
        }
    }

А вот текст главного метода Main, оживляющего нашу модель:

static void Main(string[] args)
        {
            Zolushka_Jobs zj = new Zolushka_Jobs();
            zj.Make_All_Jobs();
            for (int i = 0; i < zj.Messages.Length; i++)
                Console.WriteLine(zj.Messages[i]);
        }

При запуске проекта на консоль выводятся следующие результаты:

Вычисление интеграла и Parallel.For

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

public void ParallelIntegralWithParallelFor()
        {
            Parallel.For(0, p, EvalIntegral);
            result = 0;
            for (int i = 0; i < p; i++)
            {
                result += results[i];
            }
        }

Это самое простое и самое понятное решение. Цикл состоит из одной строчки. В цикле вызывается метод EvalIntegral, которому передается индекс итерации цикла. В методе EvalIntegral формируются границы интервала интегрирования, и вызывается последовательный метод, осуществляющий интегрирование на заданном отрезке. Вот код этого метода:

void EvalIntegral(int i)
        {
            double dx = (b - a) / p;            
            double start, finish;
            start = a + i * dx;
            finish = start + dx;
            DefiniteIntegral(start, finish, out results[i]);
        }

Работа с классом NewIntegral, вызов различных методов этого класса, оценка времени, необходимого для вычислений тем или иным методом, осуществляется в интерфейсном проекте, представляющем традиционный Windows Forms проект. Не буду останавливаться подробно на его описании. На следующем рисунке показан вид этого проекта, содержащего результаты работы каждого из рассматриваемых методов и то время, которое каждый из методов затратил на вычисление интеграла:

В этом эксперименте все параллельные методы в 4-5 раз эффективнее последовательных методов. Лучший результат в это раз показал метод, использующий потоки. Заметьте, при сравнительно большом числе потоков этот метод работает хорошо при длинных итерациях, как в данном случае. Тем не менее, для распараллеливания циклов следует применять метод Parallel.For как наиболее простой, интуитивно понятный метод, соответствующий привычному для нас оператору for.

Цикл while и Parallel.For

До сих пор, говоря о распараллеливании циклов, мы рассматривали исключительно цикл типа for. Более общей формой цикла является форма с циклом while:

while (B) { body }

Как распараллелить такой цикл, когда заголовок цикла не определяет число итераций, требуемых для завершения цикла? Пример на эту тему у нас уже встречался, когда мы рассматривали числа-градины. Там же, по существу, дано и решение возникающей проблемы. Решение основано на возможности использования оператора break в параллельно выполняемых итерациях цикла. Условие выхода (B) проверяется в ходе выполнения итерации и при его истинности осуществляется прерывание выполнения исполняемых итераций. При этом обеспечивается возможность выяснения наименьшего индекса итерации, для которого выполняется условие выхода. Подробная семантика процесса прерывания уже описана в этой главе. Давайте рассмотрим схему замены цикла while параллельным циклом Parallel.For. Она выглядит следующим образом:

ParallelLoopResult res;
  //Параллельный запуск итераций
   res = Parallel.For(0, N, body);
  //минимальный индекс итерации, на которой выполняется условие завершения          
  int index = res.LowestBreakIteration;
  if (res.IsCompleted)
  //выход по достижению максимума итераций
  else 
  //выход по условию цикла while

Тело цикла оформляется как метод, которому передаются два параметра - индекс текущей итерации и параметр класса ParallelLoopState:

void body(int i, ParallelLoopState pls)
{
  //начальная часть тела цикла
  …
  //проверка условия выхода
  if (B)
    pls.Break();
  //завершающая часть тела цикла
  …
}

Остается вопрос, требующий решения, - как задать параметр N. Чаще всего, известно максимально возможное число итераций. Например, в классической задаче поиска по образцу, использующей цикл while, число итераций не может превышать числа элементов массива. В других задачах это число выбирается из каких-то внешних предположений, как например в задаче о числах-градинах, нас интересовал ответ для некоторого фиксированного интервала чисел. Например, для плохо сходящихся процессов разумно задавать некоторое максимальное число итераций, при достижении которого процесс прекращается. Так что число N в этой схеме - это число, ограничивающее максимальное число итераций. Наша схема всегда позволяет выяснить, как закончился цикл - либо по достижению максимума итераций, либо по достижению истинности условия B на итерации. В последнем случае известен минимальный индекс итерации, на котором это условие стало истинным.

Подводя итог, можно сказать, что циклы while также могут быть достаточно просто распараллелены. Как всегда, главная проблема состоит в обеспечении независимости итераций цикла. Ответственность за решение этой проблемы несет программист.