Введение в PLINQ

PLINQ (Parallel Language-Integrated Query) - параллельный интегрированный язык запросов - является параллельной реализацией LINQ. Отличие PLINQ от LINQ состоит в том, что запросы выполняются параллельно, используя, обычно, все доступные ядра/процессоры.

7.1 Использование PLINQ

PLINQ предназначен для параллельного выполнения LINQ-запросов и потому его реализация встроена в библиотеку System.Linq. PLINQ полностью поддерживает все операторы запросов, имеющиеся в .NET, и имеет минимальное влияние на существующую модель использования и исполнения LINQ-операторов.

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

private static bool IsDivisibleBy5 ( int p )
{
      //   Моделирование длительной обработки
      for ( int i = 0; i < 10000000; i++ ) {
     i++; i--;
      }
      return p % 5 == 0;
}

Следующий фрагмент программы с использование LINQ-операторов позволяет подсчитать количество чисел в интервале от 1 до 1000, которые делятся на 5:

IEnumerable<int> arr = Enumerable.Range ( 1, 1000 );
var q =
     from n in arr          
         where ISDivisibleBy5 ( n ) 
      select n;
List<int> list = q.ToList();
Console.WriteLine ( list.Count.ToString() );

Чтобы распараллелить этот запрос средствами PLINQ, достаточно применить к источнику данных (в данном случае, это перечислимый массив arr) extension-метод AsParallel():

IEnumerable<int> arr = Enumerable.Range ( 1, 1000 );
var q =
     from n in arr.AsParallel()          
         where ISDivisibleBy5 ( n ) 
      select n;
List<int> list = q.ToList();
Console.WriteLine ( list.Count.ToString() );

В действительности, LINQ-запросы, записанные в SQL-подобном синтаксисе, как в Примере 2, переводятся в серию вызовов extension-методов классов, содержащихся в System.Linq.Enumerable. Другими словами, запрос из Примера 2 может быть переписан в следующей эквивалентной форме:

IEnumerable<int> q = 
         Enumerable.Select<int,int> (
     Enumerable.Where<int> (          
         arr, delegate (int n) { return 
      IsDivisibleBy5 (n); }
         ),
         delegate (int n) { return n; }
);

Наряду с Enumerable, библиотека System.Linq содержит эквивалентный класс ParallelEnumerable, который содержит тот же набор методов, что и класс Enumerable, но предназначенных для параллельного исполнения.

Тогда параллельная реализация Примера 4 запишется следующим образом:

IParallelEnumerable<int> q = 
         ParallelEnumerable.Select<int,int> (
     ParallelEnumerable.Where<int> (          
         ParallelQuery.AsParallel<int> (arr),
         delegate (int n) { return 
                         IsDivisibleBy5 (n); }
         ),
         delegate (int n) { return n; }
);

Рассмотрим еще один (схематичный) запрос, записанный в синтаксисе LINQ:

IEnumerable<T> data = …;
var q =
     from x in data          
         where p ( x ) 
     orderby k ( x )
      select f ( x );
foreach ( var e in q )
       a ( e );

Распараллеливание исполнения этого запроса снова достигается применением extension-метода AsParallel():

IEnumerable<T> data = …;
var q =
     from x in data.AsParallel()          
         where p ( x ) 
     orderby k ( x )
      select f ( x );
foreach ( var e in q )
       a ( e );

И снова запрос из Примера 6 может быть эквивалентно переписан с использованием Enumerable:

IEnumerable<T> data = …;
var q = Enumerable.Select(
      Enumerable.OrderBy(
      Enumerable.Where(data, x => p(x)),
      x => k(x)),
          x => f(x));
foreach (var e in q)
     a(e);

Распараллелить запрос из Примера 8 легко:

IEnumerable<T> data = …;
var q = ParallelEnumerable.Select(
      ParallelEnumerable.OrderBy(
      ParallelEnumerable.Where(data.AsParallel(), x => p(x)),
      x => k(x)),
          x => f(x));
foreach (var e in q)
     a(e);

Запрос из Примера 8 можно переписать еще одним эквивалентным способом, который отличается тем, что в нем опущены явные указания типа Enumerable, что сокращает запрос в записи, но вызывает процедуру (неявного) вывода типов на этапе компиляции программы:

IEnumerable<T> data = …;
var q = data.Where(x => p(x)).Orderby(x => k(x)).Select(x => f(x));
foreach (var e in q) a(e);

Распараллеливание запроса из Примера 10 снова происходит применением метода AsParallel() к источнику данных:

IEnumerable<T> data = …;
var q = data.AsParallel().
          Where(x => p(x)).Orderby(x => k(x)).Select(x => f(x));
foreach (var e in q) a(e);

Семинарское занятие № 7. Параллельные шаблоны агрегирования в PLINQ

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

Понятие агрегирования в LINQ

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

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

В языке LINQ для вычисления операций агрегирования предусмотрена специальная функция-шаблон - функция Aggregate, реализованная, во-первых, в виде extension-метода (об extension-методах см., например, документ "C# Version 3.0 Specification", http://msdn.microsoft.com/en-us/library/ms364047(VS.80).aspx ), а, во-вторых, имеющая несколько перегруженных вариантов для применения в различных ситуациях. Ниже показан пример реализации одного из таких вариантов (код, обрабатывающий ошибки, опущен):

public static TResult Aggregate<TSource, TAccumulate, TResult> (
        this IEnumerable<TSource> source,
        TAccumulate seed,
        Func<TAccumulate, TSource, TAccumulate> func,
        Func<TAccumulate, TResult> resultSelector
    )
{

        TAccumulate accumulator = seed;
        foreach (TSource elem in source)
        {
            accumulator = func(accumulator, elem);
        }

        return resultSelector(accumulator);
}

Чтобы воспользоваться этой функцией, программист должен указать

В частности, с использованием данной функции сумму квадратов последовательности чисел можно вычислить следующим образом:

public static int SumSquares(IEnumerable<int> source)
{
 return source.Aggregate(0,(sum, x) => sum + x * x, (sum) => sum);
}

Параллельное вычисление операций агрегирования

Предположим, что вызов функции

SumSquares ( Enumerable.Range ( 1,4 ) )

происходит на машине с двумя ядрами (процессорами). Если предполагать, что при этом будут запущены два потока, то вполне возможно, что один поток будет вычислять сумму квадратов для $${1,4}$$, а второй поток - для $${3,2}$$. Таким образом, при последовательном и параллельном исполнении, шаги вычислений могут быть следующими:

Последовательное исполнение: $$((((0 + 1^2) + 2^2) + 3^2) + 4^2) = 30$$

Параллельное исполнение: $$(((0 + 1^2) + 4^2) + ((0 + 3^2) + 2^2)) = 30$$

Промежуточные результаты при параллельных операциях агрегирования

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

а функция объединения промежуточных результатов есть просто функция сложения

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

По этой причине, функция Aggregate в PLINQ имеет в качестве дополнительного параметра, по сравнению с аналогичной функцией в LINQ, функцию finalReduceFunc объединения (редукции) значений переменных-аккумуляторов:

public static TResult Aggregate<TSource, TAccumulate, TResult> (
    this IParallelEnumerable<TSource> source,
    TAccumulate seed,
    Func<TAccumulate,TSource,TAccumulate> intermediateReduceFunc,
    Func<TAccumulate, TAccumulate, TAccumulate> finalReduceFunc,
    Func<TAccumulate, TResult> resultSelector
)

Свойства операций параллельного агрегирования

В общем случае, не любая операция агрегирования может быть корректно вычислена с помощью шаблонов (функций Aggregate ), предлагаемых в PLINQ.

Для того, чтобы определить возможно ли корректное параллельное вычисление некоторой операции агрегирования, необходимо промоделировать шаги такого вычисления. Для этого нужно предположить, что исходная последовательность элементов произвольным образом переупорядочена и "разрезана" на несколько подпоследовательностей. Для каждой подпоследовательности, переменная-аккумулятор инициализируется значением seed и к элементам подпоследовательности применяется функция редукции. Затем, промежуточные значения, хранящиеся в переменных-аккумуляторах, объединяются с помощью функции finalReduceFunc. Если такой алгоритм всегда дает корректный результат, то это значит, что его можно вычислять с помощью шаблонов агрегирования PLINQ.

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

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

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

Оператор $$F(x,y) $$ является ассоциативным, если $$F(F(x,y),z) = F(x,F(y,z)) $$ и коммутативным, если $$F(x,y) = F(y,x) $$, для всех значений $$x,y,z$$. Например, оператор Max обладает обоими этими свойствами, так как: $$Max(x,y) = Max(y,x) $$ и $$Max(Max(x,y),z) = Max(x,Max(y,z)) $$. Оператор "- " не является ни ассоциативным, ни коммутативным.

Ниже приведена таблица с примерами различных по категориям операторов:

Таким образом, необходимым условием корректности параллельного вычисления операции агрегирования явялется коммутативность и ассоциативность функции редукции.

При использовании LINQ, программист может задавать в качестве начального значения ( seed ) переменной-аккумулятора любое значение. Например, вычислить "сумму квадратов плюс 5" можно с помощью следующей операции агрегирования:

public static int SumSquaresPlus5(IEnumerable<int> source)
{
 return source.Aggregate(5, (sum, x) => sum + x * x, (sum) => sum);
}

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

Также неверными, в общем случае, начальными значениями переменных-аккумуляторов будут значение, возвращаемое оператором default для некоторого типа T, или первый элемент исходной последовательности.

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

Пусть $$F()$$ - функция редукции, $$G()$$ - функция объединения промежуточных результатов (полученных отдельными потоками), а s есть начальное значение переменных-аккумуляторов. Тогда, чтобы определенная пользователем операция агрегирования могла быть корректно вычислена параллельно с использованием средств PLINQ, необходимо и достаточно чтобы:

Задача 1.

Реализуйте с помощью параллельной операции агрегирования вычисление среднеарифметического для последовательности целых чисел.

Задача 2.

Реализуйте с помощью параллельной операции агрегирования подсчет частоты встречаемости символов алфавита $$a \dots z$$ в заданной строке в этом алфавите. Все частоты символов должны быть сохранены в специальном массиве.

Задача 3.

В "Решето Эратосфена для нахождения простых чисел", параллельная реализация построена на основе конструкции Parallel.For. При этом, логика алгоритма допускает использование операций агрегирования. Измените реализацию алгоритма так, чтобы она использовала функцию Aggregate из PLINQ.