ITask - значения параметров для анонимной процедуры

Мои 5 минут после ужина пищеварительный :)

program Project1;

{$APPTYPE CONSOLE}
{$R *.res}

uses
  System.SysUtils,
  System.SyncObjs,
  System.Threading;

type
  Pmyrec = ^Tmyrec;
  Tmyrec = record
    vet: array of integer;
    total: integer;
    average: integer;
  end;

// (1) ===> here is the major procedure that populates the dyn. array and
// calculates other two fields  :  Tmyrec.total  and Tmyrec.avg

procedure ProcA(const pin, pfin: integer; const Prec: Pmyrec);
var
  vind: integer;
begin
  for vind := pin to pfin do
    begin
      Prec.vet[vind] := vind * 10;
      TInterlocked.Add(Prec.total, Prec.vet[vind]);     // sum all array values
    end;
end;

// (2)  Here iTask is created and calls ProcA

function CreateTask(first, last: integer; const Pmyrec: Pmyrec): ITask;
begin
  Result := TTask.Create(
    procedure
    begin
      ProcA(first, last, Pmyrec)
    end);
end;

procedure Test;
var
  Recarray: Tmyrec;
  Ptasks: array of ITask;
  vind, indtask, vslice: integer;
  vfirst, vlast, vthreads, vsize: integer;
begin

  vthreads := 4;
  vsize := 16;

  SetLength(Ptasks, vthreads);
  SetLength(Recarray.vet, vsize);

  // Initialize the array , just to check after iTask execution
  for vind := low(Recarray.vet) to high(Recarray.vet) do
    Recarray.vet[vind] := -33;

  // initialize the sum and average field just to check after iTask execution
  Recarray.total := -1;
  Recarray.average := -2;

  // portion of array to scan for each iTask
  vslice := Length(Recarray.vet) div vthreads;

  for indtask := low(Ptasks) to high(Ptasks) do
    begin
      vfirst := indtask * vslice;
      vlast := (indtask + 1) * vslice - 1;

      if (Length(Recarray.vet) mod vthreads <> 0) and (indtask = high(Ptasks)) then vlast := high(Recarray.vet);

      Ptasks[indtask] := CreateTask(vfirst, vlast, @Recarray);
    end;

  // Starting all Tasks
  for indtask := low(Ptasks) to high(Ptasks) do
    Ptasks[indtask].Start;

  // Waits for all Tasks been concluded
  TTask.WaitForAll(Ptasks);
  Recarray.average := Trunc(Recarray.total / Length(Recarray.vet)); // calculates the average


  // (3) Here it is listed the array contents and it is ok
  for vind := low(Recarray.vet) to high(Recarray.vet) do
      Writeln(' Array position  : ' + Format('%.3d', [vind]) + '   content  : ' + Recarray.vet[vind].tostring);

  Writeln(' =========================================================');

// (4) Here is is listed fields recarray.total and recarray.avg and they were not
// processed inside the iTask .    I expected to see the computed values for those fields

  Writeln(' Array sum   : ' + Format('%.0d', [Recarray.total]) + '    Array average   : ' + Format('%5.2n', [Recarray.average * 1.0]));

end;

begin
  Test;
  Readln;
end.

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

1. Как заполнить целочисленные поля в записи?

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

Посмотрим, что здесь происходит.

function CreateTask(first, last: integer; var pmyrec: myrec): ITask;
var
  mylocalrec: myrec;
begin
  mylocalrec := pmyrec;

Поскольку mylocalrec и pmyrec являются записями (типами значений), их содержимое будет занимать два разных места в памяти. Вышеуказанное назначение скопирует содержаниеpmyrec в mylocalrec.

Это будет эквивалентно следующему:

mylocalrec.vet := pmyrec.vet;
mylocalrec.total := pmyrec.total;
mylocalrec.average := pmyrec.average;

Поскольку total и average также являются типами значений, их содержимое будет скопировано, и с этого момента любое изменение любого из этих целочисленных полей в mylocalrec не повлияет на исходный pmyrec. Вот почему ваш код не работает.

Почему назначение поля динамического массива vet из pmyrec в mylocalrec работает?

Поскольку динамические массивы являются ссылочными типами, при присвоении одной переменной другой копируется только значение ссылки (указатель), а не фактическое содержимое. Обе переменные vet будут указывать на тот же массив, который вы выделили перед запуском ваших задач.

Чтобы решить вышеуказанную проблему, вам необходимо передать какой-либо ссылочный тип вместо типа значения. Самое простое - объявить тип указателя записи и передать его.

type
  myrec = record
    vet: array of integer;
    total: integer;
    average: integer;
  end;

  pmyrec = ^myrec;


function CreateTask(first, last: integer; rec: pmyrec): ITask;
var
  mylocalrec: pmyrec;
begin
  mylocalrec := rec;
  Result := TTask.Create(
    procedure
    begin
      ProcA(first, last, mylocalrec^)
    end);
end;

...
Ptasks[indtask] := CreateTask(vfirst, vlast, @Recarray);

2. Как решить проблему многопоточности?

После решения исходной проблемы у вас возникнет проблема с потоками. Чтение и запись одной и той же области памяти из нескольких потоков небезопасно. Полученные вами результаты могут быть неверными. Хотя вы можете запускать свой код тысячу раз, и все значения могут быть правильными, рано или поздно вы столкнетесь с ситуацией, когда этого не произойдет.

Заполнение динамического массива в вашем случае безопасно, потому что каждый поток работает с отдельной частью массива, и массив не перераспределяется (его размер не изменяется) во время выполнения ваших задач. Эта часть вашего кода является потокобезопасной.

Расчет total и average не является потокобезопасным.

Вы должны синхронизировать такой код с основным потоком - в этом случае все чтение и запись будут выполняться из основного потока, и вы получите правильные результаты. Или вам нужно запустить такой код после того, как все задачи будут выполнены. Что больше подходит для конкретного случая.

procedure ProcA(const pin, pfin: integer; prec: pmyrec);
var
  vind: integer;
  total: integer;
begin
  total := 0;
  for vind := pin to pfin do
    begin
      prec.vet[vind] := vind * 10;
      total := total + prec.vet[vind];        // sum all array values
    end;

  TThread.Synchronize(nil,
    procedure
    begin
      prec.total := prec.total + total;
      prec.average := Trunc(prec.total / Length(prec.vet)); // calculates the average
    end);
end;

Однако синхронизация с основным потоком внутри задач вызовет взаимоблокировку с методом TTask.WaitForAll, который в вашем случае также запускается из основного потока. Чтобы решить эту проблему, вам также необходимо запустить весь метод Test из другого потока.

  TTask.Run(
    procedure
    begin
      Test;
    end);

И когда мы сложим все эти части вместе, полный код будет:

program ProjectTest;

{$APPTYPE CONSOLE}
{$R *.res}

uses
  System.SysUtils,
  System.Classes,
  System.Threading;

type
  myrec = record
    vet: array of integer;
    total: integer;
    average: integer;
  end;

  pmyrec = ^myrec;

// (1) ===> here is the major procedure that populates the dyn. array and
// calculates other two fields  :  myrec.total  and myrec.avg

procedure ProcA(const pin, pfin: integer; prec: pmyrec);
var
  vind: integer;
  total: integer;
begin
  total := 0;
  for vind := pin to pfin do
    begin
      prec.vet[vind] := vind * 10;
      total := total + prec.vet[vind];        // sum all array values
    end;

  TThread.Synchronize(nil,
    procedure
    begin
      prec.total := prec.total + total;
      prec.average := Trunc(prec.total / Length(prec.vet)); // calculates the average
    end);
end;

// (2)  Here iTask is created and calls ProcA

function CreateTask(first, last: integer; rec: pmyrec): ITask;
var
  mylocalrec: pmyrec;
begin
  mylocalrec := rec;
  Result := TTask.Create(
    procedure
    begin
      ProcA(first, last, mylocalrec);
    end);
end;

procedure Test;
var
  Recarray: myrec;
  Ptasks: array of ITask;
  vind, indtask, vslice: integer;
  vfirst, vlast, vthreads, vsize: integer;
begin

  vthreads := 4;
  vsize := 16;

  SetLength(Ptasks, vthreads);
  SetLength(Recarray.vet, vsize);

  // Initialize the array , just to check after iTask execution
  for vind := low(Recarray.vet) to high(Recarray.vet) do
    Recarray.vet[vind] := -33;

  // initialize the sum and average field just to check after iTask execution
  Recarray.total := -1;
  Recarray.average := -2;

  // portion of array to scan for each iTask
  vslice := Length(Recarray.vet) div vthreads;

  for indtask := low(Ptasks) to high(Ptasks) do
    begin
      vfirst := indtask * vslice;
      vlast := (indtask + 1) * vslice - 1;

      if (Length(Recarray.vet) mod vthreads <> 0) and (indtask = high(Ptasks)) then vlast := high(Recarray.vet);

      Ptasks[indtask] := CreateTask(vfirst, vlast, @Recarray);
    end;

  // Starting all Tasks
  for indtask := low(Ptasks) to high(Ptasks) do
    Ptasks[indtask].Start;

  // Waits for all Tasks been concluded
  TTask.WaitForAll(Ptasks);

  // (3) Here it is listed the array contents and it is ok
  for vind := low(Recarray.vet) to high(Recarray.vet) do
      Writeln(' Array position  : ' + Format('%.3d', [vind]) + '   content  : ' + Recarray.vet[vind].tostring);

  Writeln(' =========================================================');

// (4) Here is is listed fields recarray.total and recarray.avg and they were not
// processed inside the iTask .    I expected to see the computed values for those fields

  Writeln(' Array sum   : ' + Format('%.0d', [Recarray.total]) + '    Array average   : ' + Format('%5.2n', [Recarray.average * 1.0]));

end;

begin
  TTask.Run(
    procedure
    begin
      Test;
    end);
end.