Как правильно передать динамич. массив в поток и обновить его при изменении

[stalkernet]

Доброго Всем.

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

есть тип запись обьедененная в массив обьявляю так:

type
 TDataSection = record
   Data: TBytes;
   MD5: RecMD5uint64;
   PRuleN: uint32;
   NumBlock: byte;
   BtLen: word; //byte;
   SectUse: byte;
 end;

type
  TForm1 = class(TForm)
....

  public
   FDDataSect: TArray<TArray<TDataSection>>;
   OutFDDataSect: TArray<TArray<TDataSection>>;

есть поток в нем обьявляю так

type
  TFindTr = class(TThread)
  private
    { Private declarations }
  public
   DataArr: TArray<TArray<TDataSection>>;
   RuleArr: TArray<TArray<TDataSection>>;

// исполняемая часть
procedure TFindTr.Execute;

 while CloseTR = 0 do begin // закрыть поток по внешнему требованию
  StatWork := 1;
  StopFind := 0;
  for NumRule := FStart to FEnd do begin
   Move((Pointer(DataArr[NumBlock,NumData].Data))^, n1arrB[0], SectLen);
   Move((Pointer(RuleArr[NumBlock,NumRule].Data))^, n2arrB[0], SectLen);
   NOutarr[0] := N1arr[0] xor N2arr[0];
   NOutarr[1] := N1arr[1] xor N2arr[1];
   NOutarr[2] := N1arr[2] xor N2arr[2];
   Move(nOutarrB, (Pointer(OutxorArr))^, SectLen);

   PacLenSect := 0;
   BsectSet := [];
    for nBt := 0 to SectLen - 1 do begin
     if OutxorArr[nBt] in BSectSet = false then begin
      include(BSectSet, OutxorArr[nBt]);
      inc(PacLenSect, 1);
     end;
    end;

//   првоверка на условия
    if (PacLenSect <= MinLenUse) then begin
     FindResult := 1;
     break;
    end;
    if StopFind = 1 then break;
  end;
  suspend; // ждем обновления данных из формы.
end;

обработка данных в форме

 // create potok
 x := 0;
 FindTR1 := TFindTr.Create(true);
 FindTR1.Priority := tpNormal;
 FindTR1.DataArr := FDDataSect;
 FindTR1.RuleArr := OutFDDataSect;

 for n := 0 to Ndata -1 do begin
 FindTR1.NumData := n;  // задаем блок для обработки из иассива FDDataSect [x,n]
 FindTR1.Resume; // запускаем поток

  tmp1 := 0;
  while FindTR1.Suspended = false do begin // ждем окончания работы потока
    if (tmp1 mod 300) = 0 then begin
      tmp1 := 0;
      application.ProcessMessages;
    end;
   inc(tmp1, 1);
  end;

  if (FindTR1.FindResult = 1) then begin // если блок найдем в масиве OutFDDataSect 
   inc(NMinLen, 1);
   FindTR1.StatWork := 0;
  end;

// если блок ненайдем в масиве OutFDDataSect добовляем блок FDDataSect[x,n]

  if (FindTR1.FindResult = 0) and (n > 0) then begin 
   OutFDLen := length(OutFDDataSect[x]);
   Setlength(OutFDDataSect[x], OutFDlen + 1);
   OutFDDataSect[x, OutFDLen] := FDDataSect[x, n];
   inc(allFindNMinLen, 1);
   Nrule := length(OutFDDataSect[x]);
   FindTR1.FEnd := Nrule - 1;
  end;

Теперь собственно грабли.
Все Вычесления проходят в форме - результат правильный. повторяемость 100% - но ОЧЕНЬ медленно.
вычисления проходят в потоке БЕЗ ИЗМЕНЕНИЯ массива OutFDDataSect - результат правильный. повторяемость 100%
вычисления проходят в потоке С ИЗМЕНЕНИЯ массива OutFDDataSect - и тут начинаются грабли.
правильного результата нет. часто дожодит до зависания потока намертво.
при дебаге выяснилось что изменения длинны OutFDDataSect не доходят до потока.

Вобщем чтото сделал не правильно, при попытке выяснить с помощь гугля и глубокого курения RTFM где именно результата не дало.
Причем такоеже обьявление прекрасно и стабильно работает в паралелльной сортировке массива OutFDDataSect методом инжекции.

Вобще буду рад любым подсказкам, желательно с кодом.

Ужк начал посматривать в сторону стандартной библиотеки System.Threading (Parallel Programming Library (PPL)) в принципе для моей задачи подходит....

[stalkernet]

Забыл добавить -
специально удалил все попытки синхронизации. все только ухудшалось. Родитель и хозяин OutFDDataSect - TForm, все изменения В OutFDDataSect делает он. Поток при RESUME только читает из него данные и выставляет флаг FindResult. после чего засыпает и ждет другого запуска на поиск. При изменении OutFDDataSect (TForm) нужно обновить RuleArr в потоке. Вобщем нужен алгоритм параллельного поиска в массиве.

[jillitil]

Вы создали единственную нитку в форме и ждёте её же завершения прямо в процедуре формы?
Вам бы написать книгу "Искусство программирования: Как довести производительность Core I9 до уровня Ардуинки."

• Узнать кол-во ядер. Примитивней всего переменная окружения NUMBER_OF_PROCESSORS говорит об этом или вручную пользователь вводит в окне настроек программы.
• Создать кол-во ниток по кол-ву ядер.
• Разделить данные для обработки на блоки равные кол-ву ниток или по уместить блоки в кэш (это потом освоите).
• Скормить каждой нитке блок.
• Ждать ответа от нитки о завершении обработки и передать ей следующий блок. Делать через "события", а не опросом (polling) в цикле, как вы это сделали. Для этого добавить нитке вручную св-во/а типа DataDoneGiveMeNext(Sender: TObject); и дёргать их из TFindTr.Execute когда нитка что-то важное сделала. Логику придумайте сами.
• Вести учёт кол-ва рабочих ниток. Когда последняя завершит работу, вывести готовые данные куда следует.

[evg_m]

Ждать ответа от нитки о завершении обработки и передать ей следующий блок. Делать через "события", а не опросом (polling) в цикле, как вы это сделали. Для этого добавить нитке вручную св-во/а типа DataDoneGiveMeNext(Sender: TObject); и дёргать их из TFindTr.Execute когда нитка что-то важное сделала. Логику придумайте сами.

http://programmersforum.ru/showthread.php?t=327167
http://programmersforum.ru/showthread.php?t=210495

[stalkernet]

jillitil

один поток сделан чтобы убедиться что все работает правильно. А уж потом заботливо раскладывать самому себе грабли с делением на потоки.
Для меня задача - как обновить динамический массив в потоке. при условии что поток в suspend. получить правильный и стабильный результат - а потом уже исполнять танец на граблях в в виде распараллеливания и деспечера потоков.

[stalkernet]

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

итак костьль найден.
пошговый дамп массива из формы и потока показал что данные доходят правильно. результат в потоке тоже верный. Проблема оказалась в необходимой задержке для остаканивания системы и потока после Resume. после введения sleep(10) все заработало. результат повторяемый и совпадает с первым алгоритмом на 100%.

Костыль:

FindTR1.Resume;
 sleep(10); // собственно сам костыль. 10 минмал время принятия по стаканчику для выхода из suspend.
 while FindTR1.Suspended = false do begin

для проверки основного алгоритма сойдет. для нормальной работы софта - нет.

и jillitil незачем пинать АTMeg-у. АБДурино профонация. напишите архиватор работающий на ATMega32-16М. а потом уж критикуйте. Сделано лично мной еще в начале 2000-х. и требовал всего 1К памяти. и успевал обработать 10К данных в секунду.

[stalkernet]

мдя Или "песнь о Великом человеке"
В детстве Били не любили.
Что-бы Билли не побили
Просто небыло и дня...

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

В прцессе поиска среди плагиата созданного из одной статьи обратил внимание на TTask & TParallel.For
результат

TParallel.For - ну вот наконец-то перенесли аналог из FORTRAN-a - а нет. не туто было....
Нормально запустить не удалось. точнее заработало но так медленно что линейный алгоритм сильно обгонял.

TTask - заработало. код ниже. паралельность - только в таком виде работает. Сильно зависит от загрузки системы... ускорение от 2 до 2.8 раз.
Вобщем если нет желания заморачиватся с диспечером - можно использовать.

для проверки основного алгоритма сойдет. для нормальной работы софта - в принципе тоже.

Резюме по PPL Lib.
Ни о какой паралельности не и речи. Распределенные вычесления - да.(каждому потоку своя копия данных). Работа с общим блоком - только последовательно.... Вся прелесть паралельности теряется на копировании данных в поток...
Вобщем что этой библиотекой хотели сказать индусы так и осталось тайной, покрытой матом.

Отдельное Спасибо MBo за пример с потоками. По факту - раскуриваю книжку и буду разбиратся с примером. хотелосьбы получить стабильное ускорение процесса от 2.9 и выше.

Собственно рабочий код с TTask

// WT, WTS и start/stoptime для подсчета времени - заменить как кому нравится......

uses System.Threading, System.SyncObjs, System.IOUtils;

type RecMD5uint64 = record
  MD5Hi: uint64;
  MD5Lo: uint64;
end;
type
 TDataSection = record
   Data: TBytes;
   MD5: RecMD5uint64;
   PRuleN: uint32;
   NumBlock: byte;
   BtLen: word; //byte;
   SectUse: byte;
 end;

 ..........

// TForm1 //
  public
   FDDataSect: TArray<TArray<TDataSection>>;
   OutFDDataSect: TArray<TArray<TDataSection>>;
   Ts2OutFDDataSect: TArray<TArray<TDataSection>>;
   FindResult: integer;
   TasksEnd: TBytes;

 ...........

procedure CreateTasksV2(DataArr, RuleArr:TArray<TArray<TDataSection>>; min, max: uint32; Pn: uint32; PNTask: byte; PtasksEnd: TBytes; var Wtasks: TArray<ITask>);
begin
 Wtasks[PNTask] := TTask.Create(procedure()
  var
    PNumRule: uint32;
    PnBt: uint32;
    N1arr: array[0..2] of uint64;   // calc xor sect
   N2arr: array[0..2] of uint64;
   NOutarr: array[0..2] of uint64;
   n1arrB: array[0..23] of byte absolute n1arr;
   n2arrB: array[0..23] of byte absolute n2arr;
    nOutarrB: array[0..23] of byte absolute nOutarr;
   BsectSet: set of byte;
   OutxorArr: TBytes;
   PacLenSect: uint32;
   Px: uint32;
   SectLen: uint32;
   MinLenUse: uint32;
  begin
   Px := 0;
   SectLen := 24;
   MinLenUse := 12;
   for PNumRule := min to max do begin
    setlength(OutxorArr, 24);
    Move((Pointer(DataArr[Px,Pn].Data))^, n1arrB[0], 24);
    Move((Pointer(RuleArr[Px,PNumRule].Data))^, n2arrB[0], 24);
    NOutarr[0] := N1arr[0] xor N2arr[0];
    NOutarr[1] := N1arr[1] xor N2arr[1];
    NOutarr[2] := N1arr[2] xor N2arr[2];
    Move(nOutarrB, (Pointer(OutxorArr))^, 24);
    //   tmp1 :=  CalcSectByteLen3T4(tmp);
   PacLenSect := 0;
   BsectSet := [];
    for PnBt := 0 to SectLen - 1 do begin
     if OutxorArr[PnBt] in BSectSet = false then begin
      include(BSectSet, OutxorArr[PnBt]);
      inc(PacLenSect, 1);
     end;
    end;

    if (PacLenSect <= MinLenUse) then begin
       PtasksEnd[PNTask] := 1;
       if PNTask = 1 then begin
        PacLenSect := 1;
       end;
    end;
   end;
  end);
end;

procedure TForm1.Button7Click(Sender: TObject);
var
 Tasks: TArray<ITask>;
 task: ITask;
 FStart, FEnd: uint32;
 FSectCount, FSectLen: uint32;
 FTrNum: uint32;
 FIn: TMemoryStream;
 NLoadRule, x, n: uint32;
 wt, wts, oldwts: RecTime;
 Ndata, Nrule: uint32;
 nresfind, nresnofind: uint32;
 s: string;
 NumRule: uint32;
 SectLen: uint32;
 OutFDLen: uint32;
 MinLenUse: uint32;
 NMinLen, allFindNMinLen: uint32;
 PacLen: uint32;
 FS: tstringlist;

 TaskRes: uint32;
 TasksecLen, TasksecLenEnd: uint32;
 BTasksEnd: uint32;
 Zt: uint32;
begin

 starttime(wt);
 x := 0;
 SectLen := 24;
 MinLenUse := 12;
 NMinLen := 0;
 allFindNMinLen := 0;
 setlength(FDDataSect, 0);     // clear
 setlength(OutFDDataSect, 0);      // clear
 FIn := TMemoryStream.Create;
 FIn.LoadFromFile(Memo1.Lines[0]);
 FIn.Seek(0, soBeginning);
 NLoadRule := FIn.Size div 24;
 setlength(FDDataSect, x + 1);
 setlength(FDDataSect[x], NLoadRule);
 for n := 0 to NLoadRule - 1 do begin
   setlength(FDDataSect[x,n].Data, 24);
   FIn.ReadData(FDDataSect[x,n].Data, 24);
 end;
 FIn.Free;

   setlength(OutFDDataSect, x + 1);
  OutFDLen := length(OutFDDataSect[x]);
   if OutFDLen = 0 then begin
     Setlength(OutFDDataSect[x], OutFDlen + 1);
     OutFDDataSect[x, OutFDLen] := FDDataSect[x, 0];
   end;
  Ndata := length(FDDataSect[x]);
  PBar1.Max := Ndata;
  TaskRes := 0;
  starttime(wts);
 for n := 0 to Ndata - 1 do begin // WORK

 NLoadRule := length(OutFDDataSect[x]);
 setlength(Tasks, 0);
 setlength(TasksEnd, 0);

// создаем потоки
 if NLoadRule <= 1000 then begin // 1 potok
  setlength(Tasks, 1);
  setlength(TasksEnd, 1);
 for BTasksEnd := 0 to Length(tasksEnd) - 1 do tasksEnd[BTasksEnd] := 0;
// procedure CreateTasksV2(DataArr, RuleArr:TArray<TArray<TDataSection>>; min, max: uint32; Pn: uint32; PNTask: byte;
//  var PtasksEnd: TBytes; var Wtasks: TArray<ITask>);
  CreateTasksV2(FDDataSect, OutFDDataSect, 0, NLoadRule -1, n, 0, TasksEnd, tasks);
 end;

 if NLoadRule > 1000 then begin
  setlength(Tasks, 4);
  setlength(TasksEnd, 4);
 for BTasksEnd := 0 to Length(tasksEnd) - 1 do tasksEnd[BTasksEnd] := 0;
  TasksecLen := NLoadRule div 4;
  TasksecLenEnd := NLoadRule - TasksecLen;
  CreateTasksV2(FDDataSect, OutFDDataSect, 0, TasksecLen - 1, n, 0, TasksEnd, Tasks);
  CreateTasksV2(FDDataSect, OutFDDataSect, TasksecLen, (TasksecLen * 2) - 1, n, 1, TasksEnd, Tasks);
  CreateTasksV2(FDDataSect, OutFDDataSect, (TasksecLen * 2), (TasksecLen * 3) - 1 , n, 2, TasksEnd, Tasks);
  CreateTasksV2(FDDataSect, OutFDDataSect, (TasksecLen * 3), NLoadRule - 1 , n, 3, TasksEnd, Tasks);
 end;

 FindResult := 0;
  for BTasksEnd := 0 to Length(tasksEnd) - 1 do tasksEnd[BTasksEnd] := 0;
  for task in tasks do
 task.Start;
 FindResult := 0;
 // for BTasksEnd := 0 to Length(tasksEnd) - 1 do tasksEnd[BTasksEnd] := 0;
   //Ждём выполнение всех задач.
 TTask.WaitForAll(tasks);
  FindResult := 0;
  for BTasksEnd := 0 to Length(tasksEnd) - 1 do begin
   if tasksEnd[BTasksEnd] = 1 then FindResult := 1;;
  end;
   if (FindResult > 0) then begin
   inc(NMinLen, 1);
  end;
  if (FindResult = 0) and (n > 0) then begin
   OutFDLen := length(OutFDDataSect[x]);
   Setlength(OutFDDataSect[x], OutFDlen + 1);
   OutFDDataSect[x, OutFDLen] := FDDataSect[x, n];
   inc(allFindNMinLen, 1);
  end;

    if (n mod 1000) = 0 then begin
   PBar1.Position := n;
   stoptime(wts);
   oldwts.start := wts.stop - wts.start;
   s := DecodeRecTime(oldwts);
   oldwts.stop := oldwts.start;

   memo1.Lines.Add(n.ToString +'   '+ s +'   '+ DecodeRecTime(wts));
   application.ProcessMessages;

   starttime(wts);
  end;
 end;
   stoptime(wt);
  memo1.Lines.Add(Zt.ToString + '  Dup  '+ NMinLen.ToString +'  Add  '+ allFindNMinLen.ToString +'  '+ DecodeRecTime(wt));
  Memo1.Lines.SaveToFile('LogTime');
end;

[stalkernet]

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

Вообщем траблы решены и вполне корректно - без костылей типа Sleep. Правда 1 вопрос остался.
Итак то о чем забывают написать в мануалах. Опять-же - это мое понимание.

MsgWaitForMultipleObjects - во время ожидания обработка списка MSG не ведется. Если ожидается больше 1 MSG, в обработку передается только одно. какое из них ????? и обязатально application.ProcessMessages; после выхода. иначе обработка списка не начнется.

EventStatus := 1;
WaitForMultipleObjects - если присвоение и постановка в ожидание происходит меньше какого-то крит. минимума внутри SystemTick - значение не доежает до интерфейсной секции потока. Решения не нашел. костль тапа sleep ломается.

Код рабочий. Потоки контролируемые. ожидание не морозит форму. Лишнего процессорного времени не требует. Кому надо можно поставить потоки - каждому потоку по своему ядру. Проверен на 10 потоках в течении 12 часов. Ошибок не было. Наверно мало гонял. )))
По факту - написать оказалось проще и быстрее. даже с учетом граблей. больше времени ушло на понимание что-же авторы мануалов пытальсь сказать. Особенно в своих примерах. Осталось прикрутить Break потоков на выполнение дурной работы. Будут вопросы по коду - задавайте. Смогу отвечу.

Собственно код - на форме 1 кнопка и мемо. названия стандартные. FunTime - LIB для замера времени. можно выкинуть.

форма

const
  THR_MSG = WM_USER + 666;
  stop_MSG = WM_USER + 100;

  private
   TrWorkArr: TArray<THFind>;
   MainFEventHandles: THandle;
    { Private declarations }
  public
   EvHandles: TArray<Tarray<THandle>>;
   TrWorkResArr: TArray<uint32>;
   SyncOK: uint32;
   EndFlg: uint32;
   ThrCnt: byte;
   MList: TStringlist;
   procedure THRMSG(var MSG: TMessage); message THR_MSG;
   procedure THRMSGstop(var MSG: TMessage); message stop_MSG;
    { Public declarations }
  end;

implementation

{$R *.dfm}

procedure TForm1.THRMSG(var MSG: TMessage);
begin
  TrWorkResArr[MSG.WParam] := MSG.LParam;
   MList.Add('Np  ' + MSG.WParam.ToString +' FinRes  '+ MSG.LParam.ToString);
end;

procedure TForm1.THRMSGstop(var MSG: TMessage);
begin
   MList.Add('Np stop ' + MSG.WParam.ToString +' Sleep  '+ MSG.LParam.ToString);
   inc(EndFlg, 1);
end;

procedure TForm1.Button1Click(Sender: TObject);
var
 i, n: uint32;
 WaitHandle: TArray<THandle>;
 waitres: integer;
 WT: RecTime;
begin
 ThrCnt := 4;
 memo1.Clear;
 MList := TStringlist.Create;
 MainFEventHandles := CreateEvent(nil, True, False, nil);
  setlength(EvHandles, ThrCnt);
  setlength(TrWorkArr, ThrCnt);
  setlength(TrWorkResArr, ThrCnt);
  setlength(WaitHandle, ThrCnt);
 for I := 0 to ThrCnt - 1 do begin
  TrWorkResArr[i] := 0;
  setlength(EvHandles[i], 2);
  EvHandles[i, 0] := CreateEvent(nil, True, False, nil);
  EvHandles[i, 1] := CreateEvent(nil, True, False, nil);

  TrWorkArr[i] := THFind.Create(true);
  TrWorkArr[i].FreeOnTerminate := true;
  TrWorkArr[i].Priority := tpNormal;
  TrWorkArr[i].MainHandle := MainFEventHandles;
  TrWorkArr[i].FWinHandle := Form1.Handle;
  TrWorkArr[i].FResumeHandle := EvHandles[i, 1];
  TrWorkArr[i].FStopHandle := EvHandles[i, 0];
  TrWorkArr[i].FIdx := i;
  WaitHandle[i] := TrWorkArr[i].Handle;
 end;
 starttime(wt);
 for n := 0 to 10 do begin
  for I := 0 to ThrCnt - 1 do begin
   if TrWorkArr[i].Suspended = true then TrWorkArr[i].Resume
    else
      SetEvent(EvHandles[i, 1]);
  end;
 try
  EndFlg := 0;
  SyncOk := 0;
  while EndFlg <> ThrCnt do begin
   MsgWaitForMultipleObjects(0, TrWorkArr, false, INFINITE,  QS_SENDMESSAGE);
   application.ProcessMessages;
  end;
  MList.Add('');
  except
//   ShowMessage('n ' + n.ToString);
  end;
 end;
 stoptime(wt);
 memo1.Lines := MList;
 memo1.Lines.Add('OK');
 Memo1.Lines.Add(DecodeRecTime(wt));
  for I := 0 to ThrCnt - 1 do begin
    SetEvent(EvHandles[i, 0]);
  end;
 MList.Free;
end;
end.

Поток

unit TTHFind;

interface

uses
  Winapi.Windows, Winapi.Messages, System.Classes, System.Generics.Collections, math;

const
  THR_MSG = WM_USER + 666;
  stop_MSG = WM_USER + 100;


type
  ThFind = class(TThread)
  private
   FEventHandles: array [0 .. 1] of THandle;   // 0: immediate stop; 1: resume treatment
    { Private declarations }
  public
   FIdx: integer;
   EventStatus: uint32;
   FWinHandle, FResumeHandle, FStopHandle: THandle;
   MainHandle: THandle;
   FindResult: byte;
  protected
    procedure Execute; override;
  end;

implementation

{ ThFind }

procedure ThFind.Execute;
var
   WaitRes, n,sm: Integer;
begin
   FEventHandles[0] := FStopHandle;
   FEventHandles[1] := FResumeHandle;
   sm := 0;
  repeat
   EventStatus := 0;
   FindResult := 0;
  if FIdx = 0 then sm := RandomRange(100, 200);
  if FIdx = 1 then sm := RandomRange(1000, 2000);
  if FIdx > 1 then sm := RandomRange(10, 100);
    sleep(sm);
   FindResult := 1;
   SendMessage(FWinHandle, stop_MSG, FIdx, sm);
   sendMessage(FWinHandle, THR_MSG, FIdx, FindResult);
   EventStatus := 1;
   WaitRes := WaitForMultipleObjects(2, @FEventHandles, false, INFINITE); // in end work
   ResetEvent(FEventHandles[0]);
   ResetEvent(FEventHandles[1]);
    inc(sm, 1);
  until WaitRes = WAIT_OBJECT_0; // ImmediateStop
 free;
end;

end.