Параллельные вычисления в Lazarus (модуль MTProcs)

В Delphi библиотека для параллельных вычислений (Parallel Programming Library (PPL)) появилась в версии Delphi XE7. В принципе, неплохая библиотека, однако серьезно с ней я так и не познакомился. А недавно в Lazarus довелось познакомиться с MTProcs с помощью которого можно достаточно просто и быстро организовать параллельное выполнение процедур и методов в приложении. Ниже, я рассмотрю основные моменты по работе с MTProcs.

Установка пакета multithreadprocslaz.lpk

Чтобы воспользоваться MTProcs необходимо установить пакет multithreadprocslaz.lpk и добавить необходимые зависимости в ваш проект.

На момент написания этой заметки в Lazarus была версия пакета 1.2.1. Устанавливается этот пакет также как и любые другие пакеты Lazarus (например, как вот в этой статье).

Чтобы добавить зависимость в проект можно:

1. Зайти в «Проект — Инспектор проекта»

2. В «Инспекторе проекта» выбрать «Добавить — Новая зависимость»

3. Найти пакет multithreadprocslaz.lpk в списке и нажать «Ок»

Теперь в «Инспекторе проекта» Lazarus появится необходимая зависимость и можно будет воспользоваться возможностями MTProcs

Использование MTProcs в Lazarus для параллельных вычислений

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

Для проверки является ли число простым нам потребуется вот такая функция:

function IsPrime (n: Integer): boolean;
var i: integer;
begin
   Result:=true;
   for i := 2 to n - 1 do
      if (n mod i) = 0 then
        Exit(False);
end;

Классический способ поиска простых чисел (без использования параллельных вычислений) может быть таким:

const Max = 100000;
var i:integer;
begin
  Total:=0;
  for i:=1 to Max do
    if IsPrime(i) then
      inc(Total);//счётчик простых чисел
  ShowMessage('Количество простых чисел: '+Total.ToString);
end;

Теперь попробуем сделать вычисление количества простых чисел параллельным. Делается это следующим образом:

unit Unit1;
 
{$mode objfpc}{$H+}
 
interface
 
uses
  ..., mtprocs;//подключаем модуль MTProcs
 
var
  Total: integer;
 
implementation
 
function IsPrime (n: Integer): boolean;
var i: integer;
begin
   Result:=true;
   for i := 2 to n - 1 do
      if (n mod i) = 0 then
        Exit(False);
end;
 
//параллельная процедура
procedure DoIsPrimeParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
begin
  if IsPrime(Index) then
    InterlockedIncrement(Total)
end;
 
//вычисляем простые числа в заданном диапазоне
procedure DoParallel;
const Max = 100000;
begin
  Total:=0;
  ProcThreadPool.DoParallel(@DoIsPrimeParallel,1,Max,nil);//запускаем параллельную процедуру
  ShowMessage('Количество простых чисел: '+Total.ToString);
end;
end.

Рассмотрим подробнее приведенный выше код. Итак, чтобы запустить параллельные вычисления мы использовали метод DoParallel объекта ProcThreadPool из модуля MTProcs. В нашем случае, метод DoParallel имеет следующие параметры:

procedure DoParallel(const AProc: TMTProcedure; StartIndex, EndIndex: PtrInt; Data: Pointer = nil; MaxThreads: PtrInt = 0); inline;

• AProc: TMTProcedure — параллельная процедура (в нашем случае — это была  процедура DoIsPrimeParallel.
• StartIndex, EndIndex: PtrInt — начальный и конечный индексы, которые будут выполняться параллельной  процедурой.
• Data: Pointer — указатель на  какую-либо структуру. Этот параметр необязательный и его можно использовать для чего угодно.
• MaxThreads: PtrInt — максимальное количество потоков для выполнения параллельной процедуры. Если параметр равен нулю, то ProcThreadPool подберет оптимальное количество потоков, исходя из конфигурации вашего компьютера.

Имеются также следующие версии DoParallel:

procedure DoParallel(const AMethod: TMTMethod; StartIndex, EndIndex: PtrInt; Data: Pointer = nil; MaxThreads: PtrInt = 0); inline;
procedure DoParallel(const AProc: TMTProcedure; StartIndex, EndIndex: PtrInt; Data: Pointer = nil; MaxThreads: PtrInt = 0); inline;
procedure DoParallelNested(const ANested: TMTNestedProcedure; StartIndex, EndIndex: PtrInt; Data: Pointer = nil; MaxThreads: PtrInt = 0); inline;

Тип параллельной процедуры также описан в  модуле MTProcs и выглядит  следующим образом:

TMTProcedure = procedure(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);

• Index: PtrInt — определяет какую часть работы выполняет вызов процедуры;
• Data: Pointer — то же самое, что было передано в четвертом параметре для DoParallel
• Item: TMultiThreadProcItem — объект, позволяющие получить доступ к прочим возможностям пула потоков, например, для выполнения ожидания какого-либо потока.

Установка максимального числа потоков

Установить  максимальное количество потоков при работе с MTProcs можно тремя способами:

1. Дать ProcThreadPool самому выбрать оптимальное количество потоков
2. Задать максимальное количество потоков в вызове DoParallel (последний параметр)
3. Задать максимальное количество потоков, используя свойство ProcThreadPool:

property MaxThreadCount: PtrInt read FMaxThreadCount write SetMaxThreadCount;

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

property ThreadCount: PtrInt read FThreadCount;

В случае, если вы оставляете выбор максимального количества потоков за объектом ProcThreadPool, то при запуске параллельной процедуры вызывается метод из модуля MTPCPU:

function GetSystemThreadCount: integer;

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

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

const Max = 1000000;
      cStr = 'Количество простых чисел: %d. Время выполнения операции %d мс.';
var   t_start, t_end: Int64;
begin
  Total:=0;
  t_start:=GetTickCount64;
    ProcThreadPool.DoParallel(@DoIsPrimeParallel,1,Max,nil);
  t_end:=GetTickCount64;
  ShowMessageFmt(cStr, [Total, t_end-t_start]);
end;

Результат работы при таком вызове DoParallel (с использованием дефолтного значения количества потоков) — 15 969 мс.

Попробуем увеличить количество потоков в два раза дефолтного значения:

const Max = 1000000;
      cStr = 'Количество простых чисел: %d. Время выполнения операции %d мс.';
var   t_start, t_end: Int64;
begin
  Total:=0;
  ProcThreadPool.MaxThreadCount:=2*GetSystemThreadCount;//увеличиваем количество потоков
  t_start:=GetTickCount64;
    ProcThreadPool.DoParallel(@DoIsPrimeParallel,1,Max,nil);
  t_end:=GetTickCount64;
  ShowMessageFmt(cStr, [Total, t_end-t_start]);
end;

Результат получился следующий — 15 891 мс. Чуть быстрее, но не прямо пропорционально количеству потоков. Кстати, та же самая операция без параллельных процедур занимает время 94 719 мс.

Так что, в принципе, выбор максимального количества потоков можно оставить за самим объектом ProcThreadPool.

Ожидание индекса и выполнение операций по порядку

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

procedure DoSomethingParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
begin
//если предполагается выполнение блока данных с индексом 5
//который зависит от блока с индексом 3
  if Index=5 then begin
    //если произошла ошибка при вычислении индекса 3
    if not Item.WaitForIndex(3) then 
      exit; 
    //вычисление индекса прошло успешно - проводим свои вычисление 
    ....
  end;
end;

Метод WaitForIndex принимает в качестве аргумента индекс, который ниже текущего индекса или диапазона. Если метод возвращает true, то все работает как ожидалось. Если метод возвращает false, то исключение произошло в одном из других потоков. Существует расширенная функция WaitForIndexRange, которая ожидает весь диапазон индекса:

function WaitForIndexRange(StartIndex, EndIndex: PtrInt): boolean;

Например, такой вызов:

WaitForIndexRange(2, 5)

приведет к тому, что поток будет ожидать выполнение следующих индексов: 2, 3, 4, 5.

Вообще, на мой взгляд, хоть методы WaitForIndex/WaitForIndexRange и могут пригодиться при работе с MTProcs, но при проектировании стоит все-таки стараться избегать случаев, когда эти методы потребуется, так как частое их использование может не только не ускорить, но и «повешать» вашу программу, так как потоки будут постоянно ожидать друг друга.

Синхронизация потоков

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

TThread.Synchronize(AThread: TThread; AMethod: TThreadMethod)

Этот метод принимает в качестве аргументов текущий TThread и адрес метода. В MTprocs содержится потоковая переменная CurrentThread, которая упрощает синхронизацию и вызов Synchronize может быть, например, таким:

TThread.Synchronize (CurrentThread, @UpdateGUI);

Это опубликует событие в главной очереди событий и будет ждать, пока основной поток не выполнит ваш метод.

Общее впечатление

Конечно, выше рассмотрено не всё, что касается MTProcs в Lazarus. Например, вычисление оптимального размера блоков для выполнения, примеры сортировки списков и т.д. Кому интересно, обо всём этом можно почитать в wiki Lazarus.

Что же касается моего впечатления от использования MTProcs в, так сказать, «полу боевых условиях», то, в принципе, свою задачу ProcThreadPool, как пул потоков выполнил нормально, а использование основных возможностей MTProcs не вызвало никаких серьезных затруднений.

Хотелось бы узнать у вас: в каких случаях вы предпочтете использование MTProcs в Lazarus написанию и использованию собственных потоков (TThread)?