Пользовательские разделители для PLINQ и TPL

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

Виды секционирования

Существует множество способов секционирования источника данных. В наиболее эффективных подходах несколько потоков взаимодействуют для обработки исходной последовательности, а не физического разбиения её на несколько подпоследовательностей. Для массивов и других индексированных источников, таких как IList коллекции, где длина известна заранее, секционирование диапазонов является самым простым типом секционирования. Каждый поток получает уникальные начальные и конечные индексы, чтобы он мог обрабатывать свой диапазон источника без перезаписи или перезаписывания любым другим потоком. Единственными затратами, связанными с секционированием диапазона, является начальная работа по созданию диапазонов; после этого дополнительная синхронизация не требуется. Таким образом, он может обеспечить хорошую производительность, если рабочая нагрузка распределяется равномерно. Недостатком секционирования диапазона является то, что если один поток завершается рано, он не может помочь другим потокам завершить работу.

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

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

Секционаторы TPL также поддерживают динамическое число секций. Это означает, что они могут создавать разделы на лету, например, когда ForEach цикл порождает новую задачу. Эта функция позволяет партиционеру масштабироваться синхронно с циклом. Динамические разделители также по своей природе балансируют нагрузку. При создании настраиваемого ForEach секционатора необходимо поддерживать динамическое секционирование для использования из цикла.

Настройка секционаторов балансировки нагрузки для PLINQ

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

Как правило, балансировка нагрузки требует, чтобы секции запрашивали элементы относительно часто от секционатора. В отличие от этого, разделитель, выполняющий статическое секционирование, может назначать элементы для каждой секции одновременно, используя либо разделение по диапазону, либо разделение на блоки. Это требует меньше затрат, чем балансировка нагрузки, но выполнение может занять больше времени, если один поток получает значительно больше работы, чем другие. По умолчанию при передаче IList или массива PLINQ всегда использует секционирование диапазона без балансировки нагрузки. Чтобы включить балансировку нагрузки для PLINQ, используйте Partitioner.Create метод, как показано в следующем примере.

// Static partitioning requires indexable source. Load balancing
// can use any IEnumerable.
var nums = Enumerable.Range(0, 100000000).ToArray();

// Create a load-balancing partitioner. Or specify false for static partitioning.
Partitioner<int> customPartitioner = Partitioner.Create(nums, true);

// The partitioner is the query's data source.
var q = from x in customPartitioner.AsParallel()
        select x * Math.PI;

q.ForAll((x) =>
{
    ProcessData(x);
});

' Static number of partitions requires indexable source.
Dim nums = Enumerable.Range(0, 100000000).ToArray()

' Create a load-balancing partitioner. Or specify false For  Shared partitioning.
Dim customPartitioner = Partitioner.Create(nums, True)

' The partitioner is the query's data source.
Dim q = From x In customPartitioner.AsParallel()
        Select x * Math.PI

q.ForAll(Sub(x) ProcessData(x))

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

В следующей Create таблице перечислены доступные перегрузки метода. Эти распределители не ограничиваются использованием только с PLINQ или Task. Их также можно использовать с любой пользовательской параллельной конструкцией.

Настройка секционаторов статического диапазона для Parallel.ForEach

В цикле For тело цикла передаётся методу в качестве делегата. Стоимость вызова этого делегата примерно совпадает с вызовом виртуального метода. В некоторых сценариях тело параллельного цикла может быть достаточно небольшим, чтобы стоимость вызова делегата для каждой итерации цикла стала значительной. В таких ситуациях можно использовать одну из Create перегрузок для создания IEnumerable<T> секций диапазона по исходным элементам. Затем можно передать эту коллекцию ForEach диапазонов методу, текст которого состоит из регулярного for цикла. Преимуществом этого подхода является то, что стоимость вызова делегата взимается только один раз за диапазон, а не один раз на элемент. В следующем примере показан базовый шаблон.

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {

        // Source must be array or IList.
        var source = Enumerable.Range(0, 100000).ToArray();

        // Partition the entire source array.
        var rangePartitioner = Partitioner.Create(0, source.Length);

        double[] results = new double[source.Length];

        // Loop over the partitions in parallel.
        Parallel.ForEach(rangePartitioner, (range, loopState) =>
        {
            // Loop over each range element without a delegate invocation.
            for (int i = range.Item1; i < range.Item2; i++)
            {
                results[i] = source[i] * Math.PI;
            }
        });

        Console.WriteLine("Operation complete. Print results? y/n");
        char input = Console.ReadKey().KeyChar;
        if (input == 'y' || input == 'Y')
        {
            foreach(double d in results)
            {
                Console.Write("{0} ", d);
            }
        }
    }
}

Imports System.Threading.Tasks
Imports System.Collections.Concurrent

Module PartitionDemo

    Sub Main()
        ' Source must be array or IList.
        Dim source = Enumerable.Range(0, 100000).ToArray()

        ' Partition the entire source array. 
        ' Let the partitioner size the ranges.
        Dim rangePartitioner = Partitioner.Create(0, source.Length)

        Dim results(source.Length - 1) As Double

        ' Loop over the partitions in parallel. The Sub is invoked
        ' once per partition.
        Parallel.ForEach(rangePartitioner, Sub(range, loopState)

                                               ' Loop over each range element without a delegate invocation.
                                               For i As Integer = range.Item1 To range.Item2 - 1
                                                   results(i) = source(i) * Math.PI
                                               Next
                                           End Sub)
        Console.WriteLine("Operation complete. Print results? y/n")
        Dim input As Char = Console.ReadKey().KeyChar
        If input = "y"c Or input = "Y"c Then
            For Each d As Double In results
                Console.Write("{0} ", d)
            Next
        End If

    End Sub
End Module

Каждый поток в цикле получает собственный Tuple<T1,T2>, содержащий начальные и конечные значения индекса в указанном поддиапазоне. Внутренний for цикл использует значения fromInclusive и toExclusive для обхода массива или IList непосредственно.

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

Пользовательские разделители

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

Чтобы создать базовый пользовательский секционатор, наследуйте класс от System.Collections.Concurrent.Partitioner<TSource> и переопределите виртуальные методы, как описано в следующей таблице.

Если результаты должны быть сортируемыми или требуется индексированный доступ к элементам, тогда наследуйте от System.Collections.Concurrent.OrderablePartitioner<TSource> и переопределите его виртуальные методы, как описано в следующей таблице.

В следующей таблице приведены дополнительные сведения о том, как три типа секционаторов балансировки нагрузки реализуют класс OrderablePartitioner<TSource>.

Динамические разделы

Если вы планируете использовать секционировщик в методе ForEach , необходимо иметь возможность возвращать динамическое число секций. Это означает, что секционатор может предоставлять перечислитель для новой секции по запросу в любое время во время выполнения цикла. В основном каждый раз, когда цикл добавляет новую параллельную задачу, он запрашивает новую секцию для этой задачи. Если требуется, чтобы данные были упорядочиваемыми, необходимо наследовать от System.Collections.Concurrent.OrderablePartitioner<TSource>, чтобы каждому элементу в каждом разделе был назначен уникальный индекс.

Дополнительные сведения и пример см. в статье "Практическое руководство. Реализация динамических секций".

Контракт для секционаторов

При реализации пользовательского секционатора следуйте этим рекомендациям, чтобы обеспечить правильное взаимодействие с PLINQ и ForEach в TPL:

• Если GetPartitions вызывается с аргументом partitionsCount, равным нулю или меньше, вызовите ArgumentOutOfRangeException. Несмотря на то, что PLINQ и TPL не передают параметр partitionCount, равный 0, мы, тем не менее, рекомендуем принять меры предосторожности на случай такой возможности.

• GetPartitions и GetOrderablePartitions всегда должны возвращать partitionsCount число разделов. Если секционатор выходит из данных и не может создавать столько секций, сколько запрошено, метод должен возвращать пустой перечислитель для каждой оставшейся секции. В противном случае PLINQ и TPL будут выдавать InvalidOperationException.

• GetPartitions, GetOrderablePartitions, GetDynamicPartitions, и GetOrderableDynamicPartitions никогда не должны возвращаться null (Nothing в Visual Basic). Если они это сделают, PLINQ / TPL выдаст InvalidOperationException.

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

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

  • KeysOrderedInEachPartition: каждый раздел возвращает элементы с возрастающими ключевыми индексами.

  • KeysOrderedAcrossPartitions: для всех возвращаемых разделов ключевые индексы в секции i выше, чем ключевые индексы в разделе i-1.

  • KeysNormalized: все ключевые индексы монотонно увеличиваются без пробелов, начиная с нуля.

• Все индексы должны быть уникальными. Не может быть повторяющихся индексов. Если это правило не будет соблюдено, то выходной порядок может быть перепутан.

• Все индексы должны быть ненагтивными. Если это правило не соблюдается, то могут возникнуть исключения в PLINQ/TPL.

См. также

• Параллельное программирование
• Как реализовать динамические секции
• Практическое руководство. Реализация секционатора для статического секционирования