Копия викизеркало

Это зеркало вики с pascal.chat. Правка, история и вход — там.

Parallel procedures/ru

Deutsch · English · Русский · 日本語

Общие сведения

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

Параллельные процедуры и методы часто встречаются в параллельных алгоритмах, и некоторые языки программирования предоставляют встроенную поддержку для них (например, OpenMP в gcc). См. "поддержка OpenMP" для ознакомления с планами добавления таких языковых функций в FPC. Эти возможности, будучи встроенными в язык, могут сэкономить время на ввод кода и позволяют компилятору создавать код с меньшими затратами. С другой стороны, существует множество способов преобразования однопоточного фрагмента кода в параллельный. Однопоточный подход к программированию часто замедляет код. Для получения хороших результатов необходимо указать некоторые параметры, которые компилятор не может определить самостоятельно. Для ознакомления с примерами, посетите следующие ресурсы: / OpenMP и / OpenCL. Если вам нужны параллельные алгоритмы, то MTProcs поможет в их реализации.

Добавление MTProcs

Модуль mtprocs.pas является частью пакета multithreadprocslaz.lpk. Он не зависим от других пакетов и требует лишь наличия FPC >= 2.6.0.

Вы можете найти его исходники на sourceforge:

pascal
svn co https://lazarus-ccr.svn.sourceforge.net/svnroot/lazarus-ccr/components/multithreadprocs multithreadprocs

Или в Lazarus components/multithreadprocs.

Как обычно, откройте пакет multithreadprocslaz.lpk в IDE один раз, что бы он узнал путь. Для использования пакета в вашем проекте, выберите: Package / Open recent package / .../multithreadprocslaz.lpk / More / add to project

Простой пример

Ниже приведён пример, который ничего не делает, но демонстрирует, как выглядит параллельная процедура и как она вызывается:

pascal
program Test;

{$mode objfpc}{$H+}

uses
  {$IFDEF UNIX}
  cthreads, cmem,
  {$ENDIF}
  MTProcs;

// простая параллельная процедура
procedure DoSomethingParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
var
  i: Integer;
begin
  writeln(Index);
  for i:=1 to Index*1000000 do ; // делаем некую работу
end;

begin
  ProcThreadPool.DoParallel(@DoSomethingParallel,1,5,nil); // адрес, стартовый индекс, конечный индекс, дополнительные данные
end.

На выходе будет что-то вроде этого:

pascal
2
3
1
4
5

Тут следует дать несколько коротких заметок. Подробности позже.

Особенности

Выполнение процедуры параллельно означает:

pascal
ProcThreadPool.MaxThreadCount := 8;

Накладные расходы, замедление

Накладные расходы сильно зависят от системы (количества и типа ядер, тип общей памяти, скорость критических разделов, размер кэша). Вот некоторые общие советы:

Накладные расходы на многопоточность, которые не зависят от модулей MTProcs, а просто являются результатом современных компьютерных архитектур:

Установка максимального числа потоков для процедуры

Вы можете указать максимальное число потоков для процедуры в пятом параметре.

pascal
begin
  ProcThreadPool.DoParallel(@DoSomethingParallel,1,10,nil,2); // адрес, начальный_индекс, конечный_индекс, 
      // необязательный параметр: данные (в данном случае: nil), необязательный параметр: максимальное число потоков (в данном случае: 2)
end.

Данная опция может быть полезна когда потоки работают с одними и теми же данными, и слишком большое количество потоков будет создавать много конфликтов с кэшем, что в итоге приведет к их замедлению или когда алгоритм использует много WaitForIndex, так что фактически могут работать только несколько потоков. Затем потоки могут использоваться для других задач.

Дождитесь индексации/выполнения по порядку

Иногда результат текущего индекса зависит от результата предыдущего индекса. Например, обычная задача - сначала вычислить блок 5, а затем объединить результат с результатом блока 3. Используйте для этого метод WaitForIndex:

pascal
procedure DoSomethingParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
begin
  ... ?????????????????? ?????????? ?????????? 'Index' ...
  if Index=5 then begin
    if not Item.WaitForIndex(3) then exit;
    ... ?????????????????? ...
  end;
end;

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

Существует расширенная функция WaitForIndexRange, ожидающая весь диапазон индекса:

pascal
if not Item.WaitForIndexRange(3,5) then exit; // ожидаем для значений индекса 3,4 и 5

Исключения

Если в одном из потоков возникает исключение, то другие потоки завершают работу нормально, но не запускают новый индекс. Пул ожидает завершения всех потоков, а затем вызовет исключение. Вот почему вы можете использовать try..except как всегда:

pascal
try
  ...
  ProcThreadPool.DoParallel(...);
  ...
except
  On E: Exception do ...
end;

Если есть несколько исключений, будет вызвано только первое исключение. Чтобы обрабатывать все исключения, добавьте в свой параллельный метод команду try..except.

Синхронизация

Если вы хотите вызвать функцию в основном потоке, например, чтобы обновить какой-либо элемент графического интерфейса, вы можете использовать метод класса TThread.Synchronize. Он принимает в качестве аргументов текущий TThread и адрес метода. Начиная с версии 1.2 mtprocs предоставляет переменную потока CurrentThread, которая упрощает синхронизацию:

pascal
TThread.Synchronize(CurrentThread,@YourMethod);

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

Если вы создаете своих собственных потомков TThread, вы должны установить переменную в своем методе Execute. Например:

pascal
procedure TYourThread.Execute;
begin
  CurrentThread:=Self;
  ...????????????????...
end;

Пример: Параллельный цикл

В этом примере шаг за шагом объясняется, как преобразовать цикл в параллельную процедуру. В примере вычисляется максимальное количество целочисленного массива BigArray.

Исходный цикл

pascal
type
  TArrayOfInteger = array of integer;

function FindMaximum(BigArray: TArrayOfInteger): integer;
var
  i: PtrInt;
begin
  Result:=BigArray[0];
  for i:=1 to length(BigArray)-1 do begin
    if Result<BigArray[i] then Result:=BigArray[i];
  end;
end;

Разделение работы

Работа должна быть равномерно распределена по n потокам. Для этого BigArray разбивается на блоки одинакового размера, и внешний цикл проходит по каждому блоку. Обычно n - это количество процессоров/ядер в системе. MTProcs имеет несколько служебных функций для вычисления размера и количества блоков:

pascal
function FindMaximum(BigArray: TArrayOfInteger): integer;
var
  BlockCount, BlockSize: PtrInt;
  i: PtrInt;
  Index: PtrInt;
  BlockStart, BlockEnd: PtrInt;
begin
  Result:=BigArray[0];
  ProcThreadPool.CalcBlockSize(length(BigArray),BlockCount,BlockSize);
  for Index:=0 to BlockCount-1 do begin
    Item.CalcBlock(Index,BlockSize,length(BigArray),BlockStart,BlockEnd);
    for i:=BlockStart to BlockEnd do begin
      if Result<BigArray[i] then Result:=BigArray[i];
    end;
  end;
end;

Добавленные строки можно использовать для любого цикла. Со временем можно будет написать инструмент для автоматизации этого.

Теперь работа разбита на более мелкие части. Теперь части должны стать более независимыми.

Локальные и общие переменные

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

Вот решение, заменяющее переменную Result на массив, который в конце объединяется:

pascal
function FindMaximum(BigArray: TArrayOfInteger): integer;
var
  // общие переменные
  BlockCount, BlockSize: PtrInt;
  BlockMax: PPtrInt;
  // локальные переменные
  i: PtrInt;
  Index: PtrInt;
  BlockStart, BlockEnd: PtrInt;
begin
  ProcThreadPool.CalcBlockSize(length(BigArray),BlockCount,BlockSize);
  BlockMax:=AllocMem(BlockCount*SizeOf(PtrInt)); // выделение памяти для локальных переменных
  // вычисляем максимум для каждого блока
  for Index:=0 to BlockCount-1 do begin
    // вычисляем максимум блока
    Item.CalcBlock(Index,BlockSize,length(BigArray),BlockStart,BlockEnd);
    BlockMax[Index]:=BigArray[BlockStart];
    for i:=BlockStart to BlockEnd do begin
      if BlockMax[Index]<BigArray[i] then BlockMax[Index]:=BigArray[i];
    end;
  end;
  // вычисляем максимум всех блоков
  // (это лучшее решение, если у вас сотни потоков)
  Result:=BlockMax[0];
  for Index:=1 to BlockCount-1 do
    Result:=Max(Result,BlockMax[Index]);

  FreeMem(BlockMax);
end;

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

DoParallel

Последний шаг - переместить внутренний цикл в подпрограмму и заменить цикл вызовом DoParallelNested.

pascal
...
{$ModeSwitch nestedprocvars}
uses mtprocs;
...
function TMainForm.FindMaximum(BigArray: TArrayOfInteger): integer;
var
  BlockCount, BlockSize: PtrInt;
  BlockMax: PPtrInt;

  procedure FindMaximumParallel(Index: PtrInt; Data: Pointer;
                                Item: TMultiThreadProcItem);
  var
    i: integer;
    BlockStart, BlockEnd: PtrInt;
  begin
    // вычисляем максимум блоков
    Item.CalcBlock(Index,BlockSize,length(BigArray),BlockStart,BlockEnd);
    BlockMax[Index]:=BigArray[BlockStart];
    for i:=BlockStart to BlockEnd do
      if BlockMax[Index]<BigArray[i] then BlockMax[Index]:=BigArray[i];
  end;
var
  Index: PtrInt;
begin
  // разделяем работу на блоки одинакового размера
  ProcThreadPool.CalcBlockSize(length(BigArray),BlockCount,BlockSize);
  // выделяем память для локальных/потоковых переменных
  BlockMax:=AllocMem(BlockCount*SizeOf(PtrInt));
  // вычисляем максимум для каждого блока
  ProcThreadPool.DoParallelNested(@FindMaximumParallel,0,BlockCount-1);
  // вычисляем максимум всех блоков
  Result:=BlockMax[0];
  for Index:=1 to BlockCount-1 do
    Result:=Max(Result,BlockMax[Index]);
  FreeMem(BlockMax);
end;

В основном это было копирование и вставка, так что снова это можно было автоматизировать.

Пример: параллельная сортировка

Модуль mtputils содержит функцию ParallelSortFPList, которая использует mtprocs для параллельной сортировки TFPList. Должна быть указана функция сравнения.

pascal
procedure ParallelSortFPList(List: TFPList; const Compare: TListSortCompare; MaxThreadCount: integer = 0; const OnSortPart: TSortPartEvent = nil);

Эта функция использует параллельный алгоритм MergeSort. Параметр MaxThreadCount передается DoParallel. 0 означает использование системы по умолчанию.

При желании вы можете предоставить свою собственную функцию сортировки (OnSortPart) для сортировки части каждого отдельного потока. Например, вы можете сортировать блоки через QuickSort, которые затем объединяются. Тогда у вас будет параллельный QuickSort. См. TFPList.Sort для примера реализации QuickSort.

Использование вложенных процедур

pascal
procedure DoSomething(Value: PtrInt);
var
  p: array[1..2] of Pointer;

  procedure SubProc(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
  begin
    p[Index]:=Pointer(Value); // возможен доступ к локальным переменным и параметрам!
  end;

var
  i: Integer;
begin
  ProcThreadPool.DoParallelNested(@SubProc,1,2);
end;

Это может сохранить много времени на рефакторинг и сделать код параллельно работающих процедур более читаемым.

См. также

From the Free Pascal wiki, page "Parallel procedures/ru", as it stood on 2020-09-01 after a change by Zoltanleo, out of the wiki's dump of 2024-05-26. Copyright stays with the author who wrote it, according to the old wiki's Licensing page. If you have concerns about this material, message the admin.

Версия 1, последнее изменение: WikiArchive, .