Потоки. Оптимальный вариант реализации

[gusluk]

Есть программа вычисляющая некоторую функцию, выглядит она примерно так:

Function Fx(a,b:real):real;
begin
Fx:=a+b;
end;

Function Fy(a,b:real):real;
begin
Fy:=a-b;
end;

Function Fz(a,b:real):real;
begin
Fz:=a*b;
end;

Function Fxyz:real;
begin
Fxyz:=Fx(1,2)*Fy(2,3)/Fz(3,4);
end;

Procedure CalcFxyz;
var i:integer;
begin
for i:=1 to 10 do form1.Memo1.Lines.Add(floattostr(Fxyz));
end;

Вычисления Fx, Fy, Fz занимают очень много времени, поэтому решил разбить это дело на 3 потока (каждая функция в своем потоке). Чтобы посчитать Fxyz необходимо дождаться пока завершатся предыдущие потоки. Решил это установкой счетчика, получилось вот что:

 type
   TMyThread = class(TThread)
   private
     { Private declarations }
   Public
   W:integer;

   protected
     procedure Execute; override;
   end;

var
  Form1: TForm1;
  T1: TMyThread;
  T2: TMyThread;
  T3: TMyThread;
  c:integer;
  Res1,Res2,Res3:real;
implementation

{$R *.dfm}

Function Fx(a,b:real):real;
begin
Fx:=a+b;
end;

Function Fy(a,b:real):real;
begin
Fy:=a-b;
end;

Function Fz(a,b:real):real;
begin
Fz:=a*b;
end;

Function Fxyz(x,y,z:real):real;
begin
Fxyz:=x*y/z;
end;

procedure TMyThread.Execute;
begin
 case W of
 1: Res1:=Fx(1,2);
 2: Res2:=Fy(2,3);
 3: Res3:=Fz(3,4);
 end;
 c:=c+1;
end;

Procedure CalcFxyz;
var i:integer;
begin
for i:=1 to 10 do
       begin
       c:=0;
       T1 := TMyThread.Create(False);
       T1.W:=1;
       T2 := TMyThread.Create(False);
       T2.W:=2;
       T3 := TMyThread.Create(False);
       T3.W:=3;
       T1.Priority := tpLowest;
       T2.Priority := tpLowest;
       T3.Priority := tpLowest;
       while c<3 do begin
       Application.ProcessMessages;
       Sleep(1);
       end;
       T1.Terminate;
       T2.Terminate;
       T3.Terminate;
       form1.Memo1.Lines.Add(floattostr(Fxyz(Res1,Res2,Res3)));
       end;
end;

Но кривость данной реализации не дает мне покоя. Интернет предлагает кучу вариантов решения подобных задач, но пробовать их все времени у меня не хватить. Так что может подскажите наиболее оптимальный с точки зрения скорости выполнения вариант?

[Stilet]

for i:=1 to 3 do       begin
       c:=0;
       With TMyThread.Create(True) do begin
        W:=i;
        Priority := tpLowest;
        OnTerminate:=OnTerm;
        FreeOnTerminate:=True;
        Resume;
       end;
end;

procedure OnTerm(Sender);
begin
 form1.Memo1.Lines.Add(floattostr(Fxyz(Res1,Res2,Res3)));
end;

Это как минимум.

OnTerm нужно описать как событие.
Ну и сама задумка мне представляется с граблями...

[gusluk]

Мне и самому задумка кажется корявой, но особого опыта в работе с потоками нет, потому мне и нужен пинок в правильную сторону.

[s-andriano]

Каждая функция (из трех) в своем потоке.
Поток при завершении записывает две переменные: результат и флаг.
После чего посылает основному окну сообщение.
Получив сообщение, основная программа проверяет все 3 флага. Если все установлены - берет из переменных результаты и считает последнюю функцию.

[veniside]

> разбить это дело на 3 потока

если проц не 4-х головый, то смысла нет. Количество потоков <= количество ядер. Проще всего это реализовать через семафор CreateSemaphore(), установленный на количество ядер. Тогда можно сделать так, что на 4-х ядерном проце у вас будет считаться 4 функций сразу, на 16-ти ядерном 16 и т.д. Т.е. максимальная масштабируемость.

Синхронизацию по окончанию работы обычно делают через ивенты CreateEvent(). Fxyz() ждёт ивенты от 3 потоков через WaitForMultiplyObjects().

И ещё, создание потока — тяжелая операция, а т.к. зависят они только от i (как я понимаю), то нет смысла прибивать поток, чтобы через секунду запустить его же с другим i.

Костыль с циклом Application.ProcessMessages(); действительно кривой. Проще засунуть это в отдельный поток, запустить его, и спокойно сидеть в гуе, ожидая событий от цикла.

[gusluk]

veniside, допустим у меня проц на два ядра, а потоков три, на сколько сильно они будут друг другу мешать? И еще, если не убивать поток после завершения, он просто запускается ф-цией Resume, или тут есть какая то хитрость?
В моем примере на входе Fx, Fy, Fz, стоят конкретные числа, на самом деле данные на все три ф-ции берутся из одного массива. Соответственно, чтоб небыло проблем с доступом, приходится делать синхронизацию, но в этом случае идет задержка при чтении, либо приходится копировать массив в каждый поток, что тоже занимает время. Как тут правильно сделать доступ к данным в массиве?

[veniside]

> на сколько сильно они будут друг другу мешать?

просто не будет выигрыша в производительности, но и падения сильного тоже не будет, так что можно оставить и 3. Просто это не масштабируемое решение, на 4-х ядерном не будет использоваться 1 ядро.


> он просто запускается ф-цией Resume, или тут есть какая то хитрость?

я бы построил тело потока так:

begin
  while (not AllJobDone) do
    //
    if WaitForNewJob() then
      doJob();
end;

Т.е. пока вся работа не сделана, поток ждёт новую задачу, выполняет её и уходит на новый цикл. Pasue()/Resume() тут не нужно (они, кстати, помечены как deprecated в последних версиях дельфи). WaitForNewJob() можно организовать на основе ивентов (CreateEvent/WaitForSingleObject), так что оно не будет занимать проц при ожидании.


> все три ф-ции берутся из одного массива ... Как тут правильно сделать доступ к данным в массиве?

они меняют что-то в этом массиве или только читают из него?

[gusluk]

они меняют что-то в этом массиве или только читают из него?

Меняют, результат расчетов Fxyz записывается в исходный массив. Т.е. пока она не рассчитана начинать расчеты остальных трех ф-ций нет смысла.

[veniside]

если Fx(), Fy(), Fz() активно меняют переданный им массив в любой точке, то их вобще нельзя параллельно запускать. Ну или синхронизировать каждую запись, что в общем бессмысленно, т.к. получится, кто первый, тот и наменял. И каждый раз первым может быть другой поток.

Думаю, что меняют они только определенную область, типа результат. При этом для Fx(), Fy(), Fz() эта область не пересекается. Иначе, см. выше.


> результат расчетов Fxyz записывается в исходный массив

если только Fxyz() меняет массив, то это другое дело. Она будет это делать только когда Fx(), Fy(), Fz() уже отработали текущее задание, т.е. нужна только одна синхронизация на окончание работы всех трёх функций (WaitForMultipleObjects).

[s-andriano]

Сообщение от veniside

> разбить это дело на 3 потока

если проц не 4-х головый, то смысла нет. Количество потоков <= количество ядер. Проще всего это реализовать через семафор CreateSemaphore(), установленный на количество ядер. Тогда можно сделать так, что на 4-х ядерном проце у вас будет считаться 4 функций сразу, на 16-ти ядерном 16 и т.д. Т.е. максимальная масштабируемость.

Чудес не бывает.
Если задачу удается распараллелить на 3 независимые части, то ядра, начиная с 4-го, будут "курить".
И сделать с этим ничего нельзя. Какими бы средствами мы это ни пытались.
Поэтому "смысла нет" - заявление, мягко говоря, опрометчивое.
Другое дело, цикл заставляет задуматься, что, возможно, допустимо сначала насчитать все 3 функции для всех вариантов, и только потом подводить все итоги одновременно. В этом случае я бы запустил сразу все 30 потоков. При этом периоды ожидания, когда два закончившихся потока ждут третий, минимизируется.
Вопрос лишь в том, допустимо ли это по условиям задачи.
И еще.
Если уж мы не ограничены 3-мя потоками, то и подгонять количество потоков под число ядер не вижу смысла - практика показывает, что сотню потоков процессор держит без сколько-нибудь заметного ущерба для производительности.

Синхронизацию по окончанию работы обычно делают через ивенты CreateEvent(). Fxyz() ждёт ивенты от 3 потоков через WaitForMultiplyObjects().

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

И ещё, создание потока — тяжелая операция, а т.к. зависят они только от i (как я понимаю), то нет смысла прибивать поток, чтобы через секунду запустить его же с другим i.

Затраты на поток по сравнению с секундой совершенно ничтожны.