|
|
Регистрация | << Правила форума >> | FAQ | Пользователи | Календарь | Поиск | Сообщения за сегодня | Все разделы прочитаны |
|
Опции темы | Поиск в этой теме | Опции просмотра |
#1
|
|||
|
|||
Как правильно передать динамический массив в поток и обновить его при изменении
Доброго Всем.
Собственно ситуация программа для статистики. Делается выборка из блока данных по определенным правилам. есть тип запись обьедененная в массив обьявляю так: Код:
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)) в принципе для моей задачи подходит.... |
#2
|
|||
|
|||
Забыл добавить -
специально удалил все попытки синхронизации. все только ухудшалось. Родитель и хозяин OutFDDataSect - TForm, все изменения В OutFDDataSect делает он. Поток при RESUME только читает из него данные и выставляет флаг FindResult. после чего засыпает и ждет другого запуска на поиск. При изменении OutFDDataSect (TForm) нужно обновить RuleArr в потоке. Вобщем нужен алгоритм параллельного поиска в массиве. |
#3
|
||||
|
||||
Код:
Move((Pointer(DataArr[NumBlock,NumData].Data))^, n1arrB[0], SectLen); Move((Pointer(RuleArr[NumBlock,NumRule].Data))^, n2arrB[0], SectLen); Цитата:
Код:
type TSomeData = record Data:TBytes; end; TForm1 = class(TForm) Button1: TButton; procedure Button1Click(Sender: TObject); private { Private declarations } FParam:array of array of TSomeData; public { Public declarations } end; var Form1: TForm1; implementation {$R *.dfm} procedure TForm1.Button1Click(Sender: TObject); var i: Integer; j: Integer; somedata:RawByteString; begin somedata:='0123456789'; SetLength(FParam,10); for i := Low(FParam) to High(FParam) do begin SetLength(FParam[i],10); for j := Low(FParam[i]) to High(FParam[i]) do begin SetLength(FParam[i,j].Data,10); Move(somedata[1],FParam[i,j].Data[0],10); // <=== смотреть сюда end; end; somedata:=' '; //чтоб не париться с длинной. Move(FParam[9,9].Data[0],somedata[1],10); ShowMessage(somedata); end; end. Код:
FindTR1.Resume; // запускаем поток tmp1 := 0; while FindTR1.Suspended = false do begin // ждем окончания работы потока Не понятного больше чем понятного. Ну или ты передаёшь не всю картину происходящего. Некоторые программисты настолько ленивы, что сразу пишут рабочий код. Если вас наказали ни за что - радуйтесь: вы ни в чем не виноваты. |
#4
|
|||
|
|||
Код:
var 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; ..... 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); Aristarh Dark из 24 байт делается 3 int64 и xor а после выводятся в 24 байта. работает намного быстрее чем прямой перебор. толи в регистр 24 раза загружать, толи 3 .... Но за код спасибо. более красиво получается. один поток сделан чтобы убедиться что все работает правильно. А уж потом заботливо раскладывать самому себе грабли с делением на потоки. Для меня задача - как обновить динамический массив в потоке. при условии что поток в suspend. получить правильный и стабильный результат - а потом уже исполнять танец на граблях в в виде распараллеливания и деспечера потоков. |
#5
|
|||
|
|||
мдя. мыши плакали, кололись - но упорно продолжали грысть кактус или век живи - век учись. сказад ежик слезая с кактуса....
итак костьль найден. пошговый дамп массива из формы и потока показал что данные доходят правильно. результат в потоке тоже верный. Проблема оказалась в необходимой задержке для остаканивания системы и потока после Resume. после введения sleep(10) все заработало. результат повторяемый и совпадает с первым алгоритмом на 100%. Костыль: Код:
FindTR1.Resume; sleep(10); // собственно сам костыль. 10 минмал время принятия по стаканчику для выхода из suspend. while FindTR1.Suspended = false do begin |
#6
|
||||
|
||||
Цитата:
Некоторые программисты настолько ленивы, что сразу пишут рабочий код. Если вас наказали ни за что - радуйтесь: вы ни в чем не виноваты. |
#7
|
|||
|
|||
мдя Или "песнь о Великом человеке"
В детстве Били не любили. Что-бы Билли не побили Просто небыло и дня... итак ответ на собственный вопрос пошговый дамп массива из формы и потока показал что данные доходят правильно. результат в потоке тоже верный. Проблема оказалась в необходимой задержке для остаканивания системы и потока после 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; |
#8
|
|||
|
|||
Собственно решение найдено, но не все решено.
Огромное спасибо Всем за корректный пинок, на тему - кашу в голове нужно перемешивать, иначе - пригорает. Вообщем траблы решены и вполне корректно - без костылей типа 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. |