> Denchik   (18.08.17 00:10) [35]

> ... поковыряться в реализации TThread.

При такой схеме главному потоку нужен цикл обработки сообщений, который придется писать самому (раз в главном потоке нет окон, то и нет готового цикла). Написать такой цикл несложно, но в какое место TThread его вставить?

(Поручик, молчать!)

То есть, если главный поток не является первичным потоком всей программы, а порожден от TThread, то возникает эта проблема. Поэтому главный поток (а заодно, пожалуй, и рабочие потоки) я написал бы без TThread, на чистом WinAPI. На "Королевстве" есть статься "Сплэш - показываем красиво", а в ней есть пример потока с циклом выборки сообщений.

И еще - нужно ли ограничивать число рабочих потоков числом ядер? Пусть ими рулит система, она сама разберется.

> Задача: максимально быстро просчитать md5 списка фалов.


Вопрос: можно ли сделать задачу в один поток просто циклом? Не важно, в основном потоке или в дополнительном. Если да - то есть замечательные либы распараллеливания почти любого кода. Я у себя достаточно часто использую. Если код оптимизированный под потоки (либо не использует общих ресурсов, либо защищает их) то распараллелить получается сразу несколькими строчками лекговесного кода. Ничего не надо создавать, разрушать, ожидать.
Мне видится, что это вполне возможно в данной ситуации: создали список файлов, многопоточно их обработали, пошли дальше разбираться.
Ускорение на чтении + парсинге файлов с помощью библиотеки может получится в 6-7 раз на 8ми ядрах, у меня как раз так.

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

так что все волшебные распараллеливатели почти любого кода здесь сдуются

> Вопрос: можно ли сделать задачу в один поток просто циклом?
если да, то рекомендую так и оставить в одном потоке... смысл(ускорение) потоков есть только там где есть ожидания в расчетах, если же, как тут, весь расчет выполняется практически в фоне от чтения данных... - нафиг потоки.

4 ядра фигня, у меня 16 и все одно выгоды от распараллеливания подобного нет, вот если бы было 16 дисков по одному на каждый файл, и пусть даже всего 1 ядро, вот тут уже можно было поспорить/побороться, а так нет.

Но загрузка CPU  на 25% откуда-то берется же. А это немало.

то что показывает диспетчер задач вовсе не обязательно нагрузка... не полезные действия. в конце концов даже пустой цикл типа
 i:= 1;
 while i < 10000000000000 do
   i:= i + 1;
грузит систему полностью, а вставь туда sleep
 i:= 1;
 while i < 10000000000000 do begin
   i:= i + 1;
   sleep(1);
 end;
и нет нагрузки. как так? ;)

кстати то, что есть WaitForMultipleObjects и при этом грузится все ядро на 100% это больше показатель кривого кода... т.к. ожидание выполнений потоков  оно типа как sleep, в смысле должно распределять нагрузку, а другие потоки/процессы должны как то "смягчать" вот это вот забивание ядра работой. (один делает на 100%, другой,третий...десятый на 0%, в среднем диспетчер должен показать меньше). ИМХО конечно.

> sniknik ©   (18.08.17 14:43) [47]

То, что sleep снижает показания - нормально. Это же СРЕДНЯЯ загрузка за интервал времени.

Но я не об этом. Если бы тормоза были ТОЛЬКО из-за файловых операций, то CPU так не грузился бы. А он грузится - значит, тормоза есть еще из-за чего-то. И если это "что-то" вынести в потоки, нагрузив CPU на всю катушку, то выигрыш в итоговом времени должен быть. Конечно, из рабочих потоков надо убрать все Wait'ы.

> Юрий Зотов

> То есть, если главный поток не является первичным потоком всей программы, а порожден от TThread, то возникает эта проблема. Поэтому главный поток (а заодно, пожалуй, и рабочие потоки) я написал бы без TThread, на чистом WinAPI

Так и есть, главный поток не является первичным потоком программы. Сдаётся мне если лезть в потоки на чистом WinApi, то там вообще утопия будет по времени)..

А в чём сложности сделать самому цикл обработки сообщений? Может какие-то подводные камни? Или может вообще есть под рукой какой-то простенький пример (лучше тысячи слов)?

> И еще - нужно ли ограничивать число рабочих потоков числом ядер?

Это из практических тестов получился результат. Даже если например с 4мя ядрами сделать 5 потоков, то система резко начинает притормаживать, без выигрыша в плане производительности моих потоков. Именно CPU ресурсов не хватает.

Вот ещё вспомнил, народ где-то рекомендовал вообще привязывать потоки к ядрам через SetThreadAffinityMask, т.к. винда вроде не сильно грамотно умеет этим управлять. Сам не проверял.. Не знаю насколько оправдано это.
Кста, писали вроде в винде если правой кнопкой тыкнуть на процессе в диспетчере задач, выбрать Задать сходство, поставить галки на против только одно ЦП, то как раз для процесса и применяется SetThreadAffinityMask.

> Вопрос: можно ли сделать задачу в один поток просто циклом? Не важно, в основном потоке или в дополнительном.

Сама задача очень простая, просто считать md5 для фалйов из подготовленного списка. Поэтому вполне можно вернуться к старому коду, который делал это всё в одном потоке, но с низкой производительностью. Вопрос только что дальше с этим делать)?

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

А что за либы такие, подскажите пожалуйста? Может пример опять же под рукой какой-то есть?

> sniknik

Не совсем понял, имеется ввиду задача именно расчёта md5? Если это копирование файлов, тут никаких сомнений, что наращивать надо не количество ядер, а количество дисков, а вот с md5 ситация немного другая.

> Но загрузка CPU  на 25% откуда-то берется же.

25% в моём случае это дно ядро занимается этой задачей, поэтому 25% (всего 4 ядра). Я бы давно остановился, если бы наращивание ядер не дало бы прироста по времени обсчёта. Но оно стало, грубо говоря в 3 раза меньше, на 4х ядерном CPU. Есть подозрение в 3 раза меньше а не в 4 стало, из-за не эффективности кода.. На stackoverflow этот вопрос поднимался не раз, и при распараллеливании, народ получает прирост примерно пропорциональный количеству ядер.

> кстати то, что есть WaitForMultipleObjects и при этом грузится все ядро на 100% это больше показатель кривого кода...

Не исключаю этого.. Когда приступал к задаче, почитал доки форумы и пришёл к выводу WaitForMultipleObjects это то что нужно и это прямая замена своим выдумкам из счётчиков, по которым решаем, что все потоки завершили работу. Но судя по результату это не так) Может кто-то на пальцах расскажет, какое в принципе ему применение и для каких задач используется WaitForMultipleObjects?

> Если бы тормоза были ТОЛЬКО из-за файловых операций, то CPU так не грузился бы.

Полностью с этим согласен. Даже результаты тестов где-то выше приводил

> если лезть в потоки на чистом WinApi,
>  то там вообще утопия будет по времени).

Не зная, что имелось в виду под словом "утопия", но ничего страшного точно не будет. В конечном счете, дельфишный TThread все равно работает  через WinAPI.

> А в чём сложности сделать самому цикл обработки сообщений?

Уже говорилось. СДЕЛАТЬ цикл несложно, но в какое место TThread его ВСТАВЛЯТЬ?

> есть под рукой какой-то простенький пример

См. ссылку выше. Выкиньте из файла SplashDLL.dpr все, что относится к окну - получите готовый (и даже уже работающий) скелет.

> для каких задач используется WaitForMultipleObjects?

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

Для Вашей задачи это не подходит, потому что главный поток должен принять данные, а в спящем состоянии он этого сделать не может. Правда, есть вариант - создать буфер данных и пусть рабочие потоки пишут в этот буфер. Тогда WaitForMultipleObjects Вам подойдет.

он подойдет если у нас четыре файла, и четыре потока.

или если файлов больше чем потоков,
но нам пофик сколько времени потратится на все файлы.

Я бы посоветовал повнимательней отнестись к реализации самого хеширования. Нормальная производительность md5 - порядка сотен мегабайт в секунду, так что должно бы упираться в hdd, а не в cpu. У Вас какая производительность получается - в MiB/s/core?

Подразобрался с примером. Начал переделывать. Упёрся в проблему.

У меня есть экземпляр TThread, назовём его Т1. Т1 в свою очередь может запускаться из: формы VCL, консольного приложения.

Есть класс, наследник TObject, назовём его О1, который и предназначен для обсчёта md5 и он же создаёт потоки для параллельного просчёта.

Экземпляр O1, создаётся либо в T1.Execute, либо его создаёт моя же виндовая служба.

Главное условие, чтобы класс O1 был полностью самостоятельным и не зависел от класса из которого он создаётся.

В идеале порождённые параллельные рабочие потоки, должны слать сообщения о том, что закончили обсчёт очередной партии фалов, в экземпляр O1. А вот как это организовать нет понимания. Ещё раз повторюсь, лезть в код Т1 нельзя, т.к. класс используется не только в одном юните в котором описан Т1. А плодить одинаковый код в нескольких юнитах вообще не айс..

Вопрос, как правильно организовать приём сообщений от потока в классе О1? Создавать ещё один поток, целью которого будет исключительно приём сообщений от потоков, которые непосредственно считают md5?

> У Вас какая производительность получается - в MiB/s/core?

Цифры по размерам файлов не записал.. Если времени хватит переделать, то что начал, обязательно протестирую ещё раз. О результатах отпишусь.

> Denchik   (18.08.17 20:29) [55]

> Вопрос, как правильно организовать приём сообщений от потока
> в классе О1? Создавать ещё один поток, целью которого будет
> исключительно приём сообщений от потоков,

Пуркуа не па?

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

Вот пример постмортальной обработки данных потоков:


--
Regards, LVT.

Да, ну и пример

type
 TForm1 = class(TForm)
   Button2: TButton;
   Button1: TButton;
   Button3: TButton;
   procedure Button2Click(Sender: TObject);
   procedure Button1Click(Sender: TObject);
   procedure Button3Click(Sender: TObject);
 private
   { Private declarations }
 public
   { Public declarations }
 end;

var
 Form1: TForm1;

implementation

{$R *.dfm}

uses
 thexproc;

var
 teps: TThreadExitProcs;

type
 TMyThread2 = class(TThread)
   procedure Execute; override;
 end;

procedure AsyncDispose(AData: DWord);
begin
 OutputDebugString(PChar(Format('%8.8x-', [AData])));
 Dispose(Pointer(AData));
end;

procedure TMyThread2.Execute;
var
 pData: PDWord;
begin
 New(pData);
 OutputDebugString(PChar(Format('%p+', [pData])));
 teps.RegThreadExitProc(DWord(pData), AsyncDispose);
 pData^ := Random(10000);
 Sleep (pData^);
end;

procedure TForm1.Button1Click(Sender: TObject);
begin
 teps := TThreadExitProcs.Create;
end;

procedure TForm1.Button2Click(Sender: TObject);
begin
 TMyThread2.Create(False).FreeOnTerminate := True;
end;

procedure TForm1.Button3Click(Sender: TObject);
begin
 teps.Free;
end;

--
Regards, LVT.

> Leonid Troyanovsky

Очень сложно понять с моим багажом знаний, но спасибо за пример!

Сделал черновой вариант по совету Юрия Зотова. Прошу прокомментировать уважаемых гуру возможные проблемы кода:


unit uThreadMultiMd5Ex;

interface

{$I Sources\defines.inc}

uses Classes, SysUtils, Windows,  Dialogs, Messages, uCleanerTypes, uTypes;

const
 THREAD_READY=WM_USER+20;
 MD5_READY=WM_USER+21;
 MD5_REQUIRED=WM_USER+22;

type
 TToWorkerData=record
   AFullFileName: string;
 end;
 PToWorkerData=^TToWorkerData;

 TFromWorkerData=record
   AFullFileName,
   AFileMd5: string;
   ThreadId: TThreadId;
 end;
 PFromWorkerData=^TFromWorkerData;

 TThreadMultiMd5Ex = class(TThread)
 private
   FRHashFileProc: Trhash_file;
   FFileHashesList: TFileHashesList;

   procedure SetFileHashesList(AValue: TFileHashesList);
 protected
   procedure Execute; override;
 public
   constructor Create(CreateSuspended: Boolean; ARHashFileProc: Trhash_file);
   destructor Destroy; override;

   property FileHashesList: TFileHashesList read FFileHashesList write SetFileHashesList;
 end;

 TThreadMultiMd5Worker = class(TThread)
 private
   FParThreadId: TThreadId;
   FRHashFileProc: Trhash_file;
 protected
   procedure Execute; override;
 public
   constructor Create(CreateSuspended: Boolean; AParThreadId: TThreadId; ARHashFileProc: Trhash_file);
   destructor Destroy; override;
 end;

implementation

uses uCoreCommon, uCoreUtils;

//------------------------------------------------------------------------------

constructor TThreadMultiMd5Ex.Create(CreateSuspended: Boolean; ARHashFileProc: Trhash_file);
begin
 inherited Create(CreateSuspended);
 FRHashFileProc:=ARHashFileProc;
 FFileHashesList:=TFileHashesList.Create;
end;

//------------------------------------------------------------------------------

destructor TThreadMultiMd5Ex.Destroy;
begin
 FreeAndNil(FFileHashesList);
end;

//------------------------------------------------------------------------------

procedure TThreadMultiMd5Ex.SetFileHashesList(AValue: TFileHashesList);
begin
 if Assigned(AValue) then
   FFileHashesList.Assign(AValue);
end;

//------------------------------------------------------------------------------

procedure TThreadMultiMd5Ex.Execute;
var
 Fn, Md5Str, m: string;
 LParamBuf, WParamBuf: TCopyDataStruct;
 ThrList: array of TThreadMultiMd5Worker;
 ThrCnt, i, j: integer;
 ToWorkerData: PToWorkerData;
 FromWorkerData: PFromWorkerData;
 Msg: TMsg;
 AFileHashesItem: TFileHashesItem;
begin
 ThrCnt:=4;
 SetLength(ThrList, ThrCnt);

 for i := 0 to ThrCnt-1 do
 begin
   ThrList[i]:=TThreadMultiMd5Worker.Create(True, ThreadId, FRHashFileProc);
   ThrList[i].Start;
 end;

 j := 0;
 while j < FFileHashesList.Count do
 begin

   while not Terminated do
   begin
     if PeekMessage(Msg, 0, THREAD_READY, MD5_READY, PM_REMOVE) then
     begin
       FromWorkerData:=PFromWorkerData(Msg.lParam);
       if not Assigned(FromWorkerData) then
         WrLog('ALARMA!!! MD5_READY Msg.lParam=nil')
       else
       begin
         case Msg.message of
           MD5_READY:
             begin
               WrLog('MD5_READY Catched with file name: '+FromWorkerData^.AFullFileName+', md5: '+FromWorkerData^.AFileMd5+' from thread id: '+IntToStr(FromWorkerData^.ThreadId));
               AFileHashesItem:=FFileHashesList.ItemByFullFileName(FromWorkerData^.AFullFileNam e);
               if AFileHashesItem<>nil then
                 AFileHashesItem.FileMd5:=FromWorkerData^.AFileMd5;
             end;
           THREAD_READY:
             begin
               WrLog('THREAD_READY Catched');
             end;
         end;

         New(ToWorkerData);
         ToWorkerData^.AFullFileName:=FFileHashesList[j].FilePath+FFileHashesList[j].FileName;
         PostThreadMessage(FromWorkerData^.ThreadId, MD5_REQUIRED, 0, LParam(ToWorkerData));
         Dispose(FromWorkerData);

         Inc(j);
         Break;
       end;
     end;
   end;
 end;

end;

//------------------------------------------------------------------------------

constructor TThreadMultiMd5Worker.Create(CreateSuspended: Boolean; AParThreadId: TThreadId; ARHashFileProc: Trhash_file);
begin
 inherited Create(CreateSuspended);
 FRHashFileProc:=ARHashFileProc;
 FParThreadId:=AParThreadId;
end;

//------------------------------------------------------------------------------

destructor TThreadMultiMd5Worker.Destroy;
begin
 //
end;

//------------------------------------------------------------------------------

procedure TThreadMultiMd5Worker.Execute;
var
 Msg: TMsg;
 FromWorkerData: PFromWorkerData;
 ToWorkerData: PToWorkerData;
 Fn: string;
begin
 New(FromWorkerData);
 FromWorkerData^.ThreadId:=ThreadId;
 FromWorkerData^.AFullFileName:='';
 FromWorkerData^.AFileMd5:='';

 PostThreadMessage(FParThreadID, THREAD_READY, 0, LParam(FromWorkerData));

 while not Terminated do
 begin
   if PeekMessage(Msg, 0, MD5_REQUIRED, MD5_REQUIRED, PM_REMOVE) then
   begin
     ToWorkerData:=PToWorkerData(Msg.LParam);
     if not Assigned(ToWorkerData) then
       WrLog('ALARMA!!! MD5_REQUIRED Msg.lParam=nil')
     else
     begin
       WrLog('MD5_REQUIRED Catched with file name: '+ToWorkerData^.AFullFileName);
       Fn:=ToWorkerData^.AFullFileName;
       Dispose(ToWorkerData);

       New(FromWorkerData);
       FromWorkerData^.ThreadId:=ThreadId;
       FromWorkerData^.AFullFileName:=Fn;
       FromWorkerData^.AFileMd5:=GetFileHashEx(FRHashFileProc, Fn);

       PostThreadMessage(FParThreadID, MD5_READY, 0, LParam(FromWorkerData));
     end;
   end;
 end;

end;

//------------------------------------------------------------------------------

end.


Алгоритм получился такой:
1. Запускаем главный поток TThreadMultiMd5Ex, предварительно определив для него OnTerminate
2. TThreadMultiMd5Ex.Execute запускает 4 рабочих потока
3. Ждёт пока любой из рабочих потоков пришлёт THREAD_READY и свой ThreadId
4. Как только THREAD_READY или MD5_READY получен, TThreadMultiMd5Ex выбирает из FFilesListHashes очередной элемент, шлёт рабочему потоку через PostThreadMessage путь к очередному файлу для которого считаем md5. Если был получен MD5_READY, а не просто THREAD_READY, тогда сохраняет результат просчёта
5. Так до тех пор пока не кончится список FFilesListHashes

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

Код сыровать, ещё не сделал проверки успешности PostThreadMessage, так же проверки если по какой-то причине пришли MD5_READY или MD5_REQUIRED, а данных в lParam нет (хотя с трудом могу представить в каком случае это может произойти).

о, оказалось и вцл форма там вполне может быть если не капризы проходят.

итого. универсальный случай для консоли и гуи.
делается класс хоть от tobject, хоть от tcomponent

методы : addFile() addFiles()
свойство : maxThreads, threadsStillRunning
события : onHashCalced(FileName, HashVal)
               onAllDone()

внутри : AllocateHwnd  которое будет принимать WM_USER + XXX

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

в onHashCalced получаем очередной результат, обрабатываем оставшуюся очередь

если это консоль, то крутим цикл со слипом а если гуи то все и так понятно.

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