PLINQ Операторы и методы

Перечесление ParallelExecutionMode

В некоторых операциях PLINQ может решить, что запрос лучше выполнять последовательно, для того что бы этого происходило, используется метод WithExecutionMode(). Метод WithExecutionMode() принимает значение из перечисления ParallelExecutionMode. Этот перечисление может принимать два вида значения:

Пример использования метода WithExecutionMode() продемонстрирован ниже:

int[] numbers = { 1, 2, 3, 7, 3, 8, 3, 55, 7, 9, 33, 45, 5, 68, 4, 34, 3, 1, 34, 63, 2, 54, 7, 4, 32, 7, 4, 3, 6, 87, 23 };
      var num = from i in numbers.AsParallel()
                   .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                      where i > 5 
                      select i;
    foreach (var s in num)
            {
                Console.WriteLine(s);
            }
            Console.ReadLine();

Метод WithDegreeOfParallelism

Метод WithDegreeOfParallelism() - ограничивает количество разделов, обрабатываемых параллельно. Этот метод принимает аргумент int, который устанавливает максимальное количество разделов, которые должны быть обработаны одновременно - это называется степенью параллелизма. Установка степени параллелизма не указывает PLINQ использовать столько разделов, а лишь задает верхний предел. PLINQ может решить использовать меньшее количество разделов, чем указано, или же, если метод WithExecutionMode() не использовался, может вообще выполнять запрос последовательно. Пример использования данного метода продемонстрирован ниже:

int[] numbers = { 1, 2, 3, 7, 3, 8, 3, 55, 7, 9, 33, 45, 5, 68, 4, 34, 3, 1, 34, 63, 2, 54, 7, 4, 32, 7, 4, 3, 6, 87, 23 };
            var num = from i in numbers.AsParallel()
                   .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
.WithDegreeOfParallelism(1)
                      where i > 5 
                      select i;
            foreach (var s in num)
            {
                Console.WriteLine(s);
            }
            Console.ReadLine();

В данном примере степень параллелизма равен 1, что означает что программа будет использовать не более одного процесса (ядра) и выполнит PLINQ запрос в 1 поток т.е последовательно. Использование данного метода может быть удобно, если необходимо ограничить влияние запроса на машину, которая должна выполнять и другие задачи.

Метод AsOrdered

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

public static ParallelQuery<TSource> AsOrdered<TSource>(
this ParallelQuery<TSource> source
)

public static ParallelQuery AsOrdered(
	this ParallelQuery source
)

Ниже продемонстрирован пример кода, в котором используется метод AsOrdered():

int[] numbers = { 1, 2, 3, 7, 3, 8, 3, 55, 7, 9, 33, 45, 5, 68, 4, 34, 3, 1, 34, 63, 2, 54, 7, 4, 32, 7, 4, 3, 6, 87, 23 };
            var num = from i in numbers.AsParallel()
                  .AsOrdered()
                      where i > 5 
                      select i;
            foreach (var s in num)
            {
                Console.WriteLine(s);
            }
Console.ReadLine();

Запуск кода дает показанные ниже результаты. Здесь видно, что порядок исходной последовательности предохранен:

Метод Range

Метод Range() - генерирует последовательность целых чисел в заданном диапазоне. Данный метод имеет следующий синтаксис при объявлении:

public static ParallelQuery<int> Range( 
         int start, int count)

где start - это значение первого целого числа для последовательности; count - количество генерируемых последовательных целых чисел.

Ниже приведен пример использования метода Range():

var numbers = ParallelEnumerable.Range(1, 100);
       var num = from i in numbers.AsOrdered()
                      where i >= 50
                      select i;
            foreach (var s in num)
            {
                Console.WriteLine(s);
            }
             Console.ReadLine();

В данном примере вызывается метод Range() для создания ParallelQuery<int>, содержащего диапазон целых чисел от 1 до 100 включительно. Данная последовательность служит основой для LINQ - запроса в дальнейшем.

Метод Repeat

Метод Repeat() - создает параллельную последовательность, содержащую одно повторяющееся значение. Данный метод имеет следующий синтаксис при объявлении:

public static ParallelQuery<TResult> Repeat<TResult>(
	TResult element,
	int count
)

где TResult - тип значения, которое будет повторяться в результирующей последовательности; element - элемент, который должен повторяться; count - количество его повторений в последовательности. Ниже приведен пример создания повторяющейся последовательности с использованием метода epeat():

var numbers = ParallelEnumerable.Repeat('a', 5);
           foreach (var s in numbers)
            {
                Console.WriteLine(s);
            }
             Console.ReadLine();

Как видно из примера (Рис. 12.5), создастся последовательность, в которой значение 'a' выведется на экран 5 раз.

Метод AsUnordered

Метод AsUnordered() - отменяет действие метода AsOrdered(). Используется в тех в запросах, которые состоят из множества частей, когда нужно упорядочить одну часть, но избежать накладных расходов по упорядочиванию результатов другой части.

Метод AsUnordered() имеет один тип перегрузки:

public static ParallelQuery<TSource> AsUnordered<TSource>(
	this ParallelQuery<TSource> source
) 

где Tsource - тип элементов последовательности source; source - входная последовательность.

Ниже продемонстрирован пример использование метода AsUnordered() в запросе PLINQ:

   int[] numbers = { 1, 2, 3, 7, 3, 8, 3, 55, 7, 9, 33, 45, 5, 68, 4, 34, 3, 1, 34, 63, 2, 54, 7, 4, 32, 7, 4, 3, 6, 87, 23 };
            var num = numbers.AsParallel().AsOrdered().Where(i=>i > 5).Take(10).AsUnordered().Where(i=>i > 7). Take(5).Select(i=>i);
                   
            foreach (var s in num)
            {
                Console.WriteLine(s);
            }
            Console.ReadLine(); 
         }

В данном примере сначала находятся числа, которые больше значения 5 (Where(i=>i > 5)) с сохранением порядка, после чего берутся первые 10 чисел (метод Take (10)) данной последовательности. В нашем случае это последовательность: "7,8,55,7,9,33,45,68,34,34". После чего к данной последовательности применяется метод AsUnordered(), это означает, что дальнейшие результаты PLINQ - запросов будут неупорядочены (Рис. 12.6).

Метод AsSequential

Метод AsSequential() - противоположен методу AsParallel(), т.е. метод преобразует ParallelQuery<TSource> в IEnumerable<T>, чтобы последующие операторы вызывали стандартные операторы запросов и выполнялись последовательно. Метод AsSequential() имеет один тип перегрузки:

public static IEnumerable<TSource> AsSequential<TSource>(
	this ParallelQuery<TSource> source
)

где Tsource - тип элементов последовательности source; source - ParallelQuery<TSource> для преобразования в интерфейс IEnumerable<T>.

Ниже продемонстрирован пример использование метода AsSequential() в запросе PLINQ:

int[] numbers = { 1, 2, 3, 7, 3, 8, 3, 55, 7, 9, 33, 45, 5, 68, 4, 34, 3, 1, 34, 63, 2, 54, 7, 4, 32, 7, 4, 3, 6, 87, 23 };
            var num = numbers.AsParallel()
                .Where(i=>i > 5).Take(10)
                .AsSequential().Where(i=>i > 7)
                .Select(i=>i);
                   
            foreach (var s in num)
            {
                Console.WriteLine(s);
            }
            Console.ReadLine();

Как видно из примера для второй части PLINQ - запроса используется метод AsSequential(), который переключает запрос с параллельного на последовательное выполнение. Использовать методы AsParallel() и AsSequential() в одном PLINQ - запросе, можно столько раз, сколько это необходимо.

Метод WithMergeOptions

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

public static ParallelQuery<TSource> WithMergeOptions<TSource>(
	this ParallelQuery<TSource> source,
	ParallelMergeOptions mergeOptions
)

где TSource - тип элементов последовательности source; source - объект ParallelQuery, свойство которого нужно установить; mergeOptions - параметры слияния, устанавливаемые для данного запроса.

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

Ниже приведен пример использования метода WithMergeOptions() со значением FullBuffered из перечисления ParallelMergeOptions:

var numbers = ParallelEnumerable.Range(1, 10)
        .WithMergeOptions(ParallelMergeOptions.FullyBuffered)
        .Select(i =>
        {
            Thread.Sleep(1000);
            return i;
        });
System.Diagnostics.Stopwatch sw = System.Diagnostics.Stopwatch.StartNew();
foreach (var i in numbers)
{
  Console.WriteLine("Значение: "+ i + " Время: " + sw.ElapsedMilliseconds);
}
Console.ReadLine();
}

В данном примере в операторе select запроса добавлено ожидание, чтобы перед обработкой каждого элемента созданной последовательности происходила задержка в 1 секунду (Thread.Sleep(1000)). После чего, с помощью оператора foreach результаты PLINQ-запроса и каждый элемент выводится на консоль. Для каждого элемента последовательности, выводится временная метка с использованием класса Stopwatch, чтобы приблизительно видеть, сколько времени проходит между получением результирующих элементов.

Такое время задержки (~5 сек), у каждого значения, связанно с тем, что результаты выводятся в консоль после завершения отработки PLINQ-запроса. Обратный же результат получится при использовании значения NotBuffered (отключенной буферизацией):

var numbers = ParallelEnumerable.Range(1, 10)
          .WithMergeOptions(ParallelMergeOptions.NotBuffered)
          .Select(i =>
          {
              Thread.Sleep(1000);
              return i;
          });
      System.Diagnostics.Stopwatch sw = System.Diagnostics.Stopwatch.StartNew();
      foreach (var i in numbers)
      {
          Console.WriteLine("Значение: "+ i + " Время: " + sw.ElapsedMilliseconds);
      }
      Console.ReadLine();
);
      }

Такой результат (Рис. 12.9) получается потому, что обработка PLINQ-запроса распределяется между двумя ядрами процессора; при этом элементы на консоль выводятся последовательно, а не одновременно при включенной буферизации.

Метод ForAll

Метод ForAll() - позволяет указывать действие, которое будет выполнено над каждым элементом исходной последовательности при выполнении запроса. У метода ForAll() один тип перегрузки:

public static void ForAll<TSource>(
	this ParallelQuery<TSource> source,
	Action<TSource> action
) 

где TSource - тип элементов последовательности source; source - объект ParallelQuery<TSource>, элементы которого обрабатываются действием action; action - действие, вызываемое для каждого элемента.

Ниже приведен пример использования метода ForAll():

int[] numbers = { 1, 2, 3, 7, 3, 8, 3, 55, 7, 9, 33, 45, 5, 68, 4, 34, 3, 1, 34, 63, 2, 54, 7, 4, 32, 7, 4, 3, 6, 87, 23 };
         numbers.AsParallel().Where(i => i > 5).Select(i => Math.Sqrt(i)).ForAll(Console.WriteLine);
         Console.ReadLine();

Объединение и перечисление результатов не является "дорогой" операцией, поэтому использование метода ForAll() дает значительные преимущества только при большом количестве элементов, обработка которых выполняется очень быстро.