Top.Mail.Ru
Мастера DELPHI, Delphi programming community Титульная страница Поиск, карта сайта Написать письмо 
| Новости | Поиск | FAQ | Статьи | Книги | Новости VCL |
| Форумы ЧАТ | Орешник | KOL и MCK |
Форумы -» Основная («Начинающим»)/ Базы / WinAPI / Компоненты / Сети / Media / Игры / Corba и COM / KOL / FreePascal / .Net / Прочее / RSDN.ORG

« предыдущая ветвь | форум | следующая ветвь »

Пул потоков.


Страницы:   1 2 3 4 5 6

panov ©   (2004-10-14 14:34:00) 

Создал класс для работы с пулом потоков.
Прошу высказать замечания.
В следующих постингах задам вопросы к коду.


unit uTThreadPool;

interface

uses
 Classes, windows, messages;

var
 WM_EndThreadExecuter: DWORD;      //Сообщение об окончании обработки
 WM_AddThread: DWORD;              //Сообщение о добавлении элемента в очередь

type
//Тип процедуры для выполнения в отдельном потоке
 TThreadProc=procedure(aParm: pointer);

//Структура для передачи параметров
 PQueueRec=^TQueueRec;
 TQueueRec=record
   rProc: TThreadProc;
   rParm: Pointer;
 end;

//Элемент пула потоков
 TThreadExecuter=class(TThread)
 private
   FThreadIdOwner: THandle;        //Родительский поток TThreadPool
   FThreadFree: Boolean;           //Поток свободен/занят работой
   FExecProc: TThreadProc;         //Адрес процуедуры для выполнения
   FParm: Pointer;                 //Указатель на параметр,
                                   //Предаваемый в процедуру FExecProc
   FNumThread: Integer;            //Нмер потока в пуле
 protected
   procedure Execute; override;
 public
//В конструктор передается ThreadId родительского потока TThreadPool
   constructor Create(const aOwnerThread: THandle;const aNumThread: Integer);
 end;

//Менеджер пула потоков TThreadExecuter
 TThreadPool = class(TThread)
 private
{ TODO -cВажно :
Добавить код для проверки работоспособности потоков из пула
Зависшие убивать по таймауту }
   FSemaphore: THandle;            //Семафор-счетчик потоков
   FMaxThreads: Integer;           //Максимальное число потоков в пуле
   FListQueue: TList;              //Очередь заданий на выполнение
   FPool: array of TThreadExecuter; //Пул потоков
   procedure JobAdd(const aProc: TThreadProc;const aParm: Pointer); //Добавление
                                                                    //заданий в очередь
   function StartThreadFromQueue: Boolean; //Передача задания из очереди в пул
                                           //на выполнение
 protected
   procedure Execute; override;
 public
//В конструктор передается максимальное число потоков
   constructor Create(aMaxThreads: Integer);
   destructor Destroy;override;
//Добавление заданий из внешних потоков в очередь потоков
//Параметры:
//  aProc - адрес процедуры для выполнения
//  aParm - адрес памяти, по которуму нахлодятся параметры
//          для выполнения прцедуры
   procedure AddThread(aProc:TThreadProc;aParm:Pointer);
 end;

implementation

uses Unit1;

{ TThreadExecuter }

constructor TThreadExecuter.Create(const aOwnerThread: THandle;const aNumThread: Integer);
begin
 inherited Create(True);
 FreeOnTerminate := True;
 FThreadFree := True;
 FThreadIdOwner := aOwnerThread;
 FNumThread := aNumThread;
 Resume;
end;

procedure TThreadExecuter.Execute;
begin
 while not Terminated do
 begin
   Suspend;                        //Засыпаем сразу
   try
     if Assigned(FExecProc) then FExecProc(FParm); //Выполнение переданной
                                                   //процедуры
   except
   end;
//Передача в родительский поток сообщения о завершении обработки
   PostThreadMessage(FThreadIdOwner,WM_EndThreadExecuter,FNumThread,0);
   if Terminated then break;
 end;
end;

Суслик   (2004-10-14 16:07:00) [81]

И вообще - на фига тут FSemaphore: он же используется в контексте одного потока?

Суслик   (2004-10-14 16:03:00) [82]


>  [28] panov ©   (14.10.04 16:01)

предлагаю следующее.

1. Учесть те замечания в которых я оказался прав.
2. Реализовать синхронизацию в целях обеспечения своевременности создания очереди.
3. Выложить все в полном виде.

Тогда продолжим разговор. Ок?

panov   (2004-10-14 16:01:00) [83]

>Суслик ©   (14.10.04 15:58) [26]
С чего это? TThreadExecuter уже создан. С чего ему не успевать что-то делать.

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

Суслик   (2004-10-14 16:01:00) [84]


>  [27] panov ©   (14.10.04 15:59)


> Функция StartThreadFromQueue должна возвращать True в единственном
> случае - при передаче задания на выполнение.
> В остальных случаях возвращается False.

рассмотрел :)) Виноват:)

panov   (2004-10-14 15:59:00) [85]

>Суслик ©   (14.10.04 15:55) [25]

Функция StartThreadFromQueue должна возвращать True в единственном случае - при передаче задания на выполнение.
В остальных случаях возвращается False.

Суслик   (2004-10-14 15:58:00) [86]


> Та же самая ситуация возникает, что и с TThreadPoool: поток
> не успевает начать выполнять метод TThreadExecuter.Execute

С чего это? TThreadExecuter уже создан. С чего ему не успевать что-то делать.


> А какая разница, сколько раз Resume вызывать?

Согласен, что в данном случае никакой.

Суслик   (2004-10-14 15:55:00) [87]


>  [23] panov ©   (14.10.04 15:44)


> Действительно, не обрабатывался WAIT_TIMEOUT;

Я про это не говорил - пропустил.

Ошибка то осталась.
WAIT_OBJECT_0:      
 begin
      for i :=0 to Length(FPool)-1 do
      begin
        if FPool[i].FThreadFree then
        begin
          FPool[i].FThreadFree := False;
          FPool[i].FExecProc := PQueueRec(FListQueue[0]).rProc;
          FPool[i].FParm := PQueueRec(FListQueue[0]).rParm;
          Dispose(PQueueRec(FListQueue[0]));
          FListQueue.Delete(0);
          Result := True;
          FPool[i].Resume;    
          Exit;
        end;
      end;
      NewThread;
      здесь не задается Result в случае занятости всех потоков, НЕдостижения максимума потоков и наличия заданий, а должен т.к. иначе при определенных условиях не будет правильно отрабатывать while StartThreadFromQueue do;     end;

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

Все же, есть способы объходиться без ОЯ. Например см. класс TMultiReadExclusiveWriteSynchronizer. Там реализована очень скорострельная синхронизация БЕЗ объектов ядра. Если интересно смотри его.

panov   (2004-10-14 15:48:00) [88]

>Суслик ©   (14.10.04 15:42) [22]
А какая разница, сколько раз Resume вызывать?

По поводу [14] и [15] -
if FListQueue.Count>0 then    //В очереди есть задания
   begin
     while StartThreadFromQueue do; //Передать в пул на выполнение
   end;

Та же самая ситуация возникает, что и с TThreadPoool: поток не успевает начать выполнять метод TThreadExecuter.Execute

panov   (2004-10-14 15:44:00) [89]

>Суслик ©   (14.10.04 15:39) [21]

Я понимаю, что можно использовать объекты ядра для ожидания.
Но нет ли более быстрого решения?
Создание и использование таких объектов занимает время и ресурсы.
Мне кажется, что для единовременного использования это неоправданно.

>Суслик ©   (14.10.04 15:31) [17]

Спасибо.
Действительно, не обрабатывался WAIT_TIMEOUT;
Изменил, теперь выглядит так:

function TThreadPool.StartThreadFromQueue: Boolean;
var
   RC: Integer;
   i: Integer;
begin
 Result := False;
 if FListQueue.Count=0 then Exit; //? ??????? ??? ??????? - ?????
 RC := WaitForSingleObject(FSemaphore,0);
 case RC of
   WAIT_OBJECT_0:              
     begin
       for i :=0 to Length(FPool)-1 do
       begin
         if FPool[i].FThreadFree then
         begin
           FPool[i].FThreadFree := False;
           FPool[i].FExecProc := PQueueRec(FListQueue[0]).rProc;
           FPool[i].FParm := PQueueRec(FListQueue[0]).rParm;
           Dispose(PQueueRec(FListQueue[0]));
           FListQueue.Delete(0);
           Result := True;
           FPool[i].Resume;    
           Exit;
         end;
       end;
       NewThread;
     end;
   WAIT_TIMEOUT:
     begin
       NewThread;
     end;
   end;
end;
procedure TThreadPool.NewThread;
var
 i: Integer;
begin
  if Length(FPool)<FMaxThreads then
  begin                            
    i := Length(FPool);
    SetLength(FPool,i+1);          
    FPool[i] := TThreadExecuter.Create(Self.ThreadID,i);
  end;
end;

Суслик   (2004-10-14 15:42:00) [90]


> [19] panov ©   (14.10.04 15:35)

Плохо.
А если поток уже делом занят?
Будет двойной вызов resume.

Лучше пользоваться event'ом.

Суслик   (2004-10-14 15:39:00) [91]


> [20] Суслик ©   (14.10.04 15:37)

пояснения

var
  E: TEvent;

// Главный поток
pooler := TPool.Create();
E.WaitFor;

// Поток-элемент пулера
procedure TThreadExecutor.Execute;
begin
   Создание очереди
   E.SetEvent;
   ...
end;

Все.

Суслик   (2004-10-14 15:37:00) [92]


>  [18] panov ©   (14.10.04 15:33)

Посмотри на ответ [6].
Там приведено общее решение: в главном потоке ждать TEvent, который будет установлен после создания очереди.

panov   (2004-10-14 15:35:00) [93]

>Суслик ©   (14.10.04 15:28) [16]
С терминированием - спасибо, упустил из вида.

 for i := 0 to Length(FPool)-1 do
 begin
   FPool[i].Terminate;
   FPool[i].Resume;
 end;

panov   (2004-10-14 15:33:00) [94]

Метод с inherited(False) не приводит к нужному результату.

Строка
 PostThreadMessage(Self.ThreadID,WM_AddThread,Integer(@aProc),Integer(aParm));
Выполняется раньше, чем код в Execute
PeekMessage(Msg,0,0,0,pm_NoRemove);

В результате выполнения

M := TThreadPool.Create(5);

for i := 0 to 100 do
begin
  New(Parm);
  Parm.n := i;
  Parm.TimeOut := (Random(10)+1)*30;
  M.AddThread(Exec1,Parm);
end;


т.е. обращение к методу AddThread начинается раньше, чем стартует метод Execute.

Суслик   (2004-10-14 15:31:00) [95]

Еще

в function TThreadPool.StartThreadFromQueue: Boolean;
в случае занятости делом всех потоков и недостижения макс. кол-ва потоков не верен будет результат: result будет равен false! Т.е. не все потоки будут запущены.

Суслик   (2004-10-14 15:28:00) [96]

Еще

destructor TThreadPool.Destroy;
var
  i: Integer;
begin
  ...
  //Очистка пула потоков
  for i := 0 to Length(FPool)-1 do FPool[i].Terminate;  ...
end;

О какой очистке пула может идти речь, если потоки в нерабочем состоянии находятся в режиме suspended.

Я бы лучше не делал потокам resume\suspend, а сажал их на ожидаение tevent

panov   (2004-10-14 15:27:00) [97]

>Суслик ©   (14.10.04 15:14) [14]

Спасибо за совет, добавлю
while StartThreadFromQueue do;
вместо if FListQueue>0  while...

Суслик   (2004-10-14 15:14:00) [98]

Я бы сюда

    if Msg.message=WM_EndThreadExecuter then
    begin
      FPool[Msg.wParam].FThreadFree := True; //Закончилось выполнение
                                             //задания в пуле
      ReleaseSemaphore(FSemaphore,1,nil);    //Увеличить счетчик свободных
                                             //потоков в пуле
      Continue;
    end;

перед (или вместо) continue добавил бы вот этот кусочек

    if FListQueue.Count>0 then    //В очереди есть задания
    begin
      while StartThreadFromQueue do; //Передать в пул на выполнение
    end;

Это позволит не ждать 10 мс в случае если в очереди есть необработанные запросы.

Суслик   (2004-10-14 15:09:00) [99]


>  [12] panov ©   (14.10.04 15:07)

из всего, что я сказал, скорее всего был не прав в этой фразе "Именно так и надо делать в дельфи 6." Тут надо еще подумать.

Но, то, что resume в коснтурторе делать нехорошо в Д6 это точно. Спасибо sha. Навел меня как-то на эту мысль.

---------------
Как решение обозначенной в вопросе проблемы с событием? Оно точно поможет, я всегда так делаю.

panov   (2004-10-14 15:07:00) [100]

>Суслик ©   (14.10.04 15:02) [11]

С AfterConstruction сейчас проверю

Страницы:   1 2 3 4 5 6

Наверх

Титульная страница Поиск, карта сайта Написать письмо