Sign in Join
Wiki copytemporary

Parallel procedures/ru: what changed

Revision 0 to 1. Removed lines have a minus sign; added lines have a plus sign.

Title: Parallel procedures/ru

OldNewLine
1+# Общие сведения
2+
3+На этой странице описывается параллельное выполнение отдельных процедур с помощью модуля MTProcs, что упрощает параллельное выполнение процедур и реализацию параллельных алгоритмов.
4+
5+Параллельные процедуры и методы часто встречаются в параллельных алгоритмах, и некоторые языки программирования предоставляют встроенную поддержку для них (например, OpenMP в gcc). См. "[[OpenMP support|поддержка OpenMP]]" для ознакомления с планами добавления таких языковых функций в FPC. Эти возможности, будучи встроенными в язык, могут сэкономить время на ввод кода и позволяют компилятору создавать код с меньшими затратами. С другой стороны, существует множество способов преобразования однопоточного фрагмента кода в параллельный. Однопоточный подход к программированию часто замедляет код. Для получения хороших результатов необходимо указать некоторые параметры, которые компилятор не может определить самостоятельно. Для ознакомления с примерами, посетите следующие ресурсы: [/ OpenMP](http://openmp.org/wp) и [/ OpenCL](http://www.khronos.org/opencl). Если вам нужны параллельные алгоритмы, то MTProcs поможет в их реализации.
6+
7+# Добавление MTProcs
8+
9+Модуль **mtprocs.pas** является частью пакета **multithreadprocslaz.lpk**. Он не зависим от других пакетов и требует лишь наличия FPC \>= 2.6.0.
10+
11+Вы можете найти его исходники на sourceforge:
12+
13+ svn co https://lazarus-ccr.svn.sourceforge.net/svnroot/lazarus-ccr/components/multithreadprocs multithreadprocs
14+
15+Или в Lazarus *components/multithreadprocs*.
16+
17+Как обычно, откройте пакет multithreadprocslaz.lpk в IDE один раз, что бы он узнал путь. Для использования пакета в вашем проекте, выберите: *Package / Open recent package / .../multithreadprocslaz.lpk / More / add to project*
18+
19+# Простой пример
20+
21+Ниже приведён пример, который ничего не делает, но демонстрирует, как выглядит параллельная процедура и как она вызывается:
22+
23+``` pascal
24+program Test;
25+
26+{$mode objfpc}{$H+}
27+
28+uses
29+ {$IFDEF UNIX}
30+ cthreads, cmem,
31+ {$ENDIF}
32+ MTProcs;
33+
34+// простая параллельная процедура
35+procedure DoSomethingParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
36+var
37+ i: Integer;
38+begin
39+ writeln(Index);
40+ for i:=1 to Index*1000000 do ; // делаем некую работу
41+end;
42+
43+begin
44+ ProcThreadPool.DoParallel(@DoSomethingParallel,1,5,nil); // адрес, стартовый индекс, конечный индекс, дополнительные данные
45+end.
46+```
47+
48+На выходе будет что-то вроде этого:
49+
50+ 2
51+ 3
52+ 1
53+ 4
54+ 5
55+
56+Тут следует дать несколько коротких заметок. Подробности позже.
57+
58+- Многопоточность в UNIX требует модуль **cthreads** как объяснено в [[Multithreaded Application Tutorial/ru|учебнике по многопоточным Приложениям]].
59+- Для повышения скорости **cmem** рекомендуется использовать диспетчер кучи, хотя в этом примере это не имеет значения.
60+- Параллельная процедура **DoSomethingParallel** получает некие фиксированные и предопределенные параметры.
61+- **Index** определяет, какую часть работы должен сделать этот вызов.
62+- **Data** - это указатель, который был дан в *ProcThreadPool.DoParallel* как четвертый параметр. Это необязательно и вы можете использовать его для чего угодно.
63+- **Item** может использоваться для доступа к некоторым более сложным функциям пула потоков.
64+- **ProcThreadPool.DoParallel** работает как обычный вызов процедуры. Он завершится, когда он полностью будет выполнен - это означает, что все потоки должны завершить свою работу.
65+- Выходные данные показывают типичное многопоточное поведение: порядок вызовов не определен. Несколько запусков могут привести к разным порядкам чисел.
66+
67+# Особенности
68+
69+Выполнение процедуры параллельно означает:
70+
71+- процедура или метод выполняется с индексом, идущим от произвольного StartIndex к произвольному EndIndex.
72+- Один или несколько потоков выполняют эти индексы параллельно. Например, если Индекс выполняется от 1 до 10 и в пуле доступно 3 потока, то 3 потока будут одновременно выполнять три разных индекса. Каждый раз, когда поток завершает один вызов (один индекс), он выделяет следующий индекс и выполняет его. Результатом может быть: поток 1 выполняет индекс 3,5,7, поток 2 выполняет 1,6,8,9 и поток 3 работает 2,4,10.
73+- Количество потоков может меняться во время выполнения, и нет никакой гарантии минимума потоков. В худшем случае весь индекс будет выполняться одним потоком.
74+- Максимальное количество потоков инициализируется для хорошей работы с учётом текущей системы. Однако, его можно изменить вручную через:
75+
76+``` pascal
77+ProcThreadPool.MaxThreadCount := 8;
78+```
79+
80+- Вы можете задать максимальное количество потоков для каждой процедуры.
81+- Параллельная процедура (или метод) может вызывать рекурсивно параллельные процедуры (или методы).
82+- Потоки используются повторно, это означает, что они не уничтожаются и не создаются для каждого индекса, но существует глобальный пул потоков. На двухъядерном процессоре будет два потока, выполняющих всю работу-основной поток и один дополнительный поток в пуле.
83+
84+# Накладные расходы, замедление
85+
86+Накладные расходы сильно зависят от системы (количества и типа ядер, тип общей памяти, скорость критических разделов, размер кэша). Вот некоторые общие советы:
87+
88+- Каждый блок работы (index) должен занимать не менее нескольких миллисекунд.
89+- Накладные расходы не зависят от рекурсивных уровней параллельных процедур.
90+
91+Накладные расходы на многопоточность, которые не зависят от модулей MTProcs, а просто являются результатом современных компьютерных архитектур:
92+
93+- Как только создается один поток, ваша программа становится многопоточной, и менеджеры памяти должны использовать критические разделы, что замедляет ее работу. Таким образом, даже если вы ничего не сделаете с потоком, ваша программа может стать медленнее.
94+- Менеджер кучи `cmem` в некоторых системах намного быстрее при многопоточности. В моих тестах, особенно на системах Intel и особенно под OS X, разница в скорости может быть более 10 раз.
95+- Строки и интерфейсы подсчитываются глобально. Для каждого случая доступа нужна критическая секция. Таким образом, обработка строк в нескольких потоках вряд ли увеличит скорость работы программы. Вместо них используйте PChars.
96+- Каждый фрагмент работы (индекс) должен работать с дизъюнктивной частью памяти, чтобы избежать перекрестных обновлений кеша.
97+- Не работайте с большим объемом памяти. В некоторых системах одного потока достаточно, чтобы заполнить скорость шины памяти. По достижении максимальной скорости шины памяти любой последующий поток будет замедляться, а не ускоряться.
98+
99+# Установка максимального числа потоков для процедуры
100+
101+Вы можете указать максимальное число потоков для процедуры в пятом параметре.
102+
103+``` pascal
104+begin
105+ ProcThreadPool.DoParallel(@DoSomethingParallel,1,10,nil,2); // адрес, начальный_индекс, конечный_индекс,
106+ // необязательный параметр: данные (в данном случае: nil), необязательный параметр: максимальное число потоков (в данном случае: 2)
107+end.
108+```
109+
110+Данная опция может быть полезна когда потоки работают с одними и теми же данными, и слишком большое количество потоков будет создавать много конфликтов с кэшем, что в итоге приведет к их замедлению или когда алгоритм использует много WaitForIndex, так что фактически могут работать только несколько потоков. Затем потоки могут использоваться для других задач.
111+
112+# Дождитесь индексации/выполнения по порядку
113+
114+Иногда результат текущего индекса зависит от результата предыдущего индекса. Например, обычная задача - сначала вычислить блок 5, а затем объединить результат с результатом блока 3. Используйте для этого метод WaitForIndex:
115+
116+``` pascal
117+procedure DoSomethingParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
118+begin
119+ ... вычисляем номер блока 'Index' ...
120+ if Index=5 then begin
121+ if not Item.WaitForIndex(3) then exit;
122+ ... вычисляем ...
123+ end;
124+end;
125+```
126+
127+WaitForIndex принимает в качестве аргумента индекс, который меньше текущего индекса или диапазона. Если он вернет `true`, все работает, как ожидалось. Если он возвращает `false`, то в одном из других потоков произошло исключение.
128+
129+Существует расширенная функция WaitForIndexRange, ожидающая весь диапазон индекса:
130+
131+``` pascal
132+if not Item.WaitForIndexRange(3,5) then exit; // ожидаем для значений индекса 3,4 и 5
133+```
134+
135+# Исключения
136+
137+Если в одном из потоков возникает исключение, то другие потоки завершают работу нормально, но не запускают новый индекс. Пул ожидает завершения всех потоков, а затем вызовет исключение. Вот почему вы можете использовать `try..except` как всегда:
138+
139+``` pascal
140+try
141+ ...
142+ ProcThreadPool.DoParallel(...);
143+ ...
144+except
145+ On E: Exception do ...
146+end;
147+```
148+
149+Если есть несколько исключений, будет вызвано только первое исключение. Чтобы обрабатывать все исключения, добавьте в свой параллельный метод команду `try..except`.
150+
151+# Синхронизация
152+
153+Если вы хотите вызвать функцию в основном потоке, например, чтобы обновить какой-либо элемент графического интерфейса, вы можете использовать метод класса `TThread.Synchronize`. Он принимает в качестве аргументов текущий TThread и адрес метода. Начиная с версии 1.2 `mtprocs` предоставляет переменную потока `CurrentThread`, которая упрощает синхронизацию:
154+
155+``` pascal
156+TThread.Synchronize(CurrentThread,@YourMethod);
157+```
158+
159+Это разместит событие в основной очереди событий и дождется, пока основной поток выполнит ваш метод. Имейте в виду, что голая программа fpc не имеет очереди событий. Это есть в программе LCL или fpgui.
160+
161+Если вы создаете своих собственных потомков `TThread`, вы должны установить переменную в своем методе `Execute`. Например:
162+
163+``` pascal
164+procedure TYourThread.Execute;
165+begin
166+ CurrentThread:=Self;
167+ ...работаем...
168+end;
169+```
170+
171+# Пример: Параллельный цикл
172+
173+В этом примере шаг за шагом объясняется, как преобразовать цикл в параллельную процедуру. В примере вычисляется максимальное количество целочисленного массива *BigArray*.
174+
175+## Исходный цикл
176+
177+``` pascal
178+type
179+ TArrayOfInteger = array of integer;
180+
181+function FindMaximum(BigArray: TArrayOfInteger): integer;
182+var
183+ i: PtrInt;
184+begin
185+ Result:=BigArray[0];
186+ for i:=1 to length(BigArray)-1 do begin
187+ if Result<BigArray[i] then Result:=BigArray[i];
188+ end;
189+end;
190+```
191+
192+## Разделение работы
193+
194+Работа должна быть равномерно распределена по n потокам. Для этого BigArray разбивается на блоки одинакового размера, и внешний цикл проходит по каждому блоку. Обычно n - это количество процессоров/ядер в системе. `MTProcs` имеет несколько служебных функций для вычисления размера и количества блоков:
195+
196+``` pascal
197+function FindMaximum(BigArray: TArrayOfInteger): integer;
198+var
199+ BlockCount, BlockSize: PtrInt;
200+ i: PtrInt;
201+ Index: PtrInt;
202+ BlockStart, BlockEnd: PtrInt;
203+begin
204+ Result:=BigArray[0];
205+ ProcThreadPool.CalcBlockSize(length(BigArray),BlockCount,BlockSize);
206+ for Index:=0 to BlockCount-1 do begin
207+ Item.CalcBlock(Index,BlockSize,length(BigArray),BlockStart,BlockEnd);
208+ for i:=BlockStart to BlockEnd do begin
209+ if Result<BigArray[i] then Result:=BigArray[i];
210+ end;
211+ end;
212+end;
213+```
214+
215+Добавленные строки можно использовать для любого цикла. Со временем можно будет написать инструмент для автоматизации этого.
216+
217+Теперь работа разбита на более мелкие части. Теперь части должны стать более независимыми.
218+
219+## Локальные и общие переменные
220+
221+Для каждой используемой переменной в цикле вы должны решить, является ли она общей переменной, используемой всеми потоками, или каждый поток использует свою собственную локальную переменную. Общие переменные `BlockCount` и `BlockSize` только читаются и не меняются, поэтому для них не требуется никакой работы. Но общая переменная, такая как `Result`, будет изменяться всеми потоками. Это может быть достигнуто либо с помощью синхронизации (например, критической секции), которая выполняется медленно, либо каждый поток использует локальную копию, и эти локальные переменные позже объединяются.
222+
223+Вот решение, заменяющее переменную `Result` на массив, который в конце объединяется:
224+
225+``` pascal
226+function FindMaximum(BigArray: TArrayOfInteger): integer;
227+var
228+ // общие переменные
229+ BlockCount, BlockSize: PtrInt;
230+ BlockMax: PPtrInt;
231+ // локальные переменные
232+ i: PtrInt;
233+ Index: PtrInt;
234+ BlockStart, BlockEnd: PtrInt;
235+begin
236+ ProcThreadPool.CalcBlockSize(length(BigArray),BlockCount,BlockSize);
237+ BlockMax:=AllocMem(BlockCount*SizeOf(PtrInt)); // выделение памяти для локальных переменных
238+ // вычисляем максимум для каждого блока
239+ for Index:=0 to BlockCount-1 do begin
240+ // вычисляем максимум блока
241+ Item.CalcBlock(Index,BlockSize,length(BigArray),BlockStart,BlockEnd);
242+ BlockMax[Index]:=BigArray[BlockStart];
243+ for i:=BlockStart to BlockEnd do begin
244+ if BlockMax[Index]<BigArray[i] then BlockMax[Index]:=BigArray[i];
245+ end;
246+ end;
247+ // вычисляем максимум всех блоков
248+ // (это лучшее решение, если у вас сотни потоков)
249+ Result:=BlockMax[0];
250+ for Index:=1 to BlockCount-1 do
251+ Result:=Max(Result,BlockMax[Index]);
252+
253+ FreeMem(BlockMax);
254+end;
255+```
256+
257+Этот подход прост и может быть автоматизирован. Однако для этого процесса потребуются подсказки от программиста.
258+
259+## DoParallel
260+
261+Последний шаг - переместить внутренний цикл в подпрограмму и заменить цикл вызовом `DoParallelNested`.
262+
263+``` pascal
264+...
265+{$ModeSwitch nestedprocvars}
266+uses mtprocs;
267+...
268+function TMainForm.FindMaximum(BigArray: TArrayOfInteger): integer;
269+var
270+ BlockCount, BlockSize: PtrInt;
271+ BlockMax: PPtrInt;
272+
273+ procedure FindMaximumParallel(Index: PtrInt; Data: Pointer;
274+ Item: TMultiThreadProcItem);
275+ var
276+ i: integer;
277+ BlockStart, BlockEnd: PtrInt;
278+ begin
279+ // вычисляем максимум блоков
280+ Item.CalcBlock(Index,BlockSize,length(BigArray),BlockStart,BlockEnd);
281+ BlockMax[Index]:=BigArray[BlockStart];
282+ for i:=BlockStart to BlockEnd do
283+ if BlockMax[Index]<BigArray[i] then BlockMax[Index]:=BigArray[i];
284+ end;
285+var
286+ Index: PtrInt;
287+begin
288+ // разделяем работу на блоки одинакового размера
289+ ProcThreadPool.CalcBlockSize(length(BigArray),BlockCount,BlockSize);
290+ // выделяем память для локальных/потоковых переменных
291+ BlockMax:=AllocMem(BlockCount*SizeOf(PtrInt));
292+ // вычисляем максимум для каждого блока
293+ ProcThreadPool.DoParallelNested(@FindMaximumParallel,0,BlockCount-1);
294+ // вычисляем максимум всех блоков
295+ Result:=BlockMax[0];
296+ for Index:=1 to BlockCount-1 do
297+ Result:=Max(Result,BlockMax[Index]);
298+ FreeMem(BlockMax);
299+end;
300+```
301+
302+В основном это было копирование и вставка, так что снова это можно было автоматизировать.
303+
304+# Пример: параллельная сортировка
305+
306+Модуль `mtputils` содержит функцию `ParallelSortFPList`, которая использует `mtprocs` для параллельной сортировки `TFPList`. Должна быть указана функция сравнения.
307+
308+``` pascal
309+procedure ParallelSortFPList(List: TFPList; const Compare: TListSortCompare; MaxThreadCount: integer = 0; const OnSortPart: TSortPartEvent = nil);
310+```
311+
312+Эта функция использует параллельный алгоритм `MergeSort`. Параметр `MaxThreadCount` передается `DoParallel`. 0 означает использование системы по умолчанию.
313+
314+При желании вы можете предоставить свою собственную функцию сортировки (`OnSortPart`) для сортировки части каждого отдельного потока. Например, вы можете сортировать блоки через `QuickSort`, которые затем объединяются. Тогда у вас будет **параллельный QuickSort**. См. `TFPList.Sort` для примера реализации `QuickSort`.
315+
316+# Использование вложенных процедур
317+
318+``` pascal
319+procedure DoSomething(Value: PtrInt);
320+var
321+ p: array[1..2] of Pointer;
322+
323+ procedure SubProc(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
324+ begin
325+ p[Index]:=Pointer(Value); // возможен доступ к локальным переменным и параметрам!
326+ end;
327+
328+var
329+ i: Integer;
330+begin
331+ ProcThreadPool.DoParallelNested(@SubProc,1,2);
332+end;
333+```
334+
335+Это может сохранить много времени на рефакторинг и сделать код параллельно работающих процедур более читаемым.
336+
337+# См. также
338+
339+- [[Multithreaded Application Tutorial/ru|Руководство по разработке многопоточных приложений]]
340+- [[OpenCL/ru|OpenCL]]
341+
342+---
343+*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](/wiki-contact).*

Back to history