Конструкция Parallel.For

Иногда в циклах, используемых в программах, итерации являются независимыми; т.е., ни в одной итерации не используются результаты работы предыдущих итераций. Посредством класса System.Threading.Parallel такие циклы могут быть преобразованы в циклы, в которых каждая итерация потенциально может быть выполнена параллельно при наличии достаточного количества доступных ядер (процессоров). Заметим, что параллельное исполнение циклов, итерации которых не являются независимыми, может привести к некорректным результатам. В этом случае необходимо либо пересмотреть структуру цикла, либо использовать примитивы синхронизации и\или потокобезопасные структуры данных.

В PFX существует несколько вариантов метода For, исполняющихся параллельно. Наиболее часто применяемым является For(Int32, Int32, Action<Int32>). Здесь первый параметр задает начальный индекс, второй параметр - конечный индекс, и третий параметр - делегат, определяющий действие, которое будет выполняться на каждой итерации.

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

for (int i = 0; i < N; i++)
{
  results[i] = Compute(i);
}

C помощью PFX и класса System.Threading.Parallel можно распараллелить этот цикл следующим образом:

using System.Threading;
…
Parallel.For(0, N, delegate(int i)
{
results[i] = Compute(i); 
});

Для упрощения кода можно использовать синтаксис лямбда-выражений, введенный в C# 3.0 (Visual Studio 2008):

using System.Threading;
…
Parallel.For(0, N, i =>
{
results[i] = Compute(i); 
});

Теперь итерации этого цикла могут быть выполнены параллельно.

Распараллеливание циклов foreach выполняется аналогичным образом. Рассмотрим цикл на С#:

foreach(MyClass с in data) 
{
  Compute(c);
}

C помощью PFX он распараллеливается следующим образом:

Parallel.Foreach(data, delegate(MyClass с)) 
{
  Compute(c);
}

Упрощенный код с использованием лямбда-выражений выглядит так:

Parallel.Foreach(data, с => 
{
  Compute(c);
});

Использование foreach, как правило, менее эффективно, чем for, поскольку несколько потоков должны иметь доступ к общей обрабатываемой структуре данных, например, списку. Однако, реализация Parallel.ForEach достаточно интеллектуальна, и в ней доступ из нескольких потоков к общей структуре данных происходит достаточно эффективно.

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

void  MultiplyMatrices(int size, double[,] m1, double[,] m2, double[,] result)
{
  for (int i = 0; i < size; i++)
  {
    for (int j = 0; j < size; j++)
    {
      result[i,j] = 0;
      for (int k = 0; k < size; k++)
{
  result[i,j] += m1[i,k] * m2[k,j];
}
}
  }
}

Распараллеливание заключается в простой замене внешнего цикла For циклом Parallel.For:

void  MultiplyMatrices(int size, double[,] m1, double[,] m2, double[,] result)
{
  Parallel.For (0; size; i =>
  {
    for (int j = 0; j < size; j++)
    {
      result[i,j] = 0;
      for (int k = 0; k < size; k++)
{
  result[i,j] += m1[i,k] * m2[k,j];
}
}
  });
}

Также можно выполнить распараллеливание внутреннего (по j) цикла, но только для матриц достаточно большого размера. Распараллеливание внешнего цикла дает достаточную степень параллельности, чтобы получить от нее эффект. Необдуманное использование Parallel.For может нанести ущерб производительности, поэтому внутренний цикл оставлен последовательным.

Еще один пример применения Parallel.ForEach - перечисление всех файлов изображений в директории и обработка каждого из них:

foreach(string imagePath in Directory.GetFiles(path, "*.jpg"))
{
  ProcessImage(imagePath);
}

Параллельный вариант выглядит следующим образом:

Parallel.ForEach(Directory.GetFiles(path, "*.jpg"), imagePath =>

{
  ProcessImage(imagePath);
});

Семинарское занятие № 2. Вариант Parallel.For с локальными состояниями

Конструкции Parallel.For/ForEach имеют несколько перегруженных (overloaded) вариантов, с помощью которых можно организовать передачу информации между итерациями цикла, исполняющимися в одном потоке. Вспомним, что при выполнении Parallel.For/ForEach каждый рабочий поток выполняет, в общем случае, несколько итераций. Например, при общем количестве итераций 100, на 4-хядерной машине, обычно будет запущено 4 потока, каждый из которых выполнит 25 итераций. Обычно, передача информации (например, подсчет какого-либо значения) между итерациями, исполняющимися в одном потоке, нужна для вычисления некоторого сводного значения по всем итерациям цикла.

Одним из вариантов перегруженной конструкции Parallel.For для поддержки вычислений такого рода выглядит так:

public static void For<TLocal> (
    int fromInclusive, int toExclusive, 
    Func<TLocal> threadLocalInit, 
    Action<int, ParallelState<TLocal> > body, 
    Action<TLocal> threadLocalFinally);

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

Результат каждой итерации может быть записан в объект класса ParallelState<TLocal> который является одним из встроенных классов библиотеки PFX), передаваемый в качестве параметра в делегат body. Делегат body может обновлять специальное поле (а точнее, свойство) ThreadLocalState, которое имеет объект ParallelState<TLocal>, и которое имеет тип TLocal, и значение этого поля будет доступно следующим итерациям, исполняющимся в данном потоке. После того, как поток завершил исполнение своих итераций, в рамках его же, будет вызван делегат threadLocalFinally, которому в качестве параметра будет передано значение ThreadLocalState.

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

Parallel.For(0, N, () => new NonThreadSafeData(), (i,loop)= >
{
    UseData(loop.ThreadLocalState);
});

(Здесь не использован делегат threadLocalFinally из вышеприведенной сигнатуры).

В данном примере для каждого нового потока в рамках Parallel.For будет создан свой объект класса NonThreadSafeData (т.е., в данном случае TLocal = NonThreadSafeData ). Таким образом, объект NonThreadSafeData будет передаваться между итерациями одного потока, что позволит потоку безопасно использовать этот объект. При этом, таких объектов будет создано ровно столько, сколько потоков будет запущено. Отметим также, что локальная переменная loop в приведенном выше примере имеет тип ParallelState<TLocal> .

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

int total = 0;
Parallel.ForEach(data, ()=>0, (elem,i,loop)= >
{
    loop.ThreadLocalState += Process(elem);
},
partial => Interlocked.Add(ref total, partial));

Отметим, что в данном случае локальная переменная partial имеет тип TLocal, который, в данном примере, есть тип $$int$$ (см. делегат threadLocalInit ).

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

class MultipleValues { public int Total, Count; }

int total = 0, count = 0;
Parallel.ForEach(data, 
    ()=>new MultipleValues { Total=0, Count=0 }, 
    (elem,i,loop)= >
{
    loop.ThreadLocalState.Total += Process(elem);
    loop.ThreadLocalState.Count++;
},
partial => {
    Interlocked.Add(ref total, partial.Total);
    Interlocked.Add(ref count, partial.Count);
});

Задания:

Реализуйте подсчет суммы элементов массива с использованием предлагаемого подхода, кроме того, подсчитайте, сколько потоков было запущено при исполнении Parallel.For/ForEach в вашей программе.

Изучите понятие "анонимный класс" языка C# и выясните можно ли использовать анонимные классы вместо явного определения класса MultipleValues в приведенном выше примере,и если нельзя, то почему.