| Новости | Поиск | 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 15:02:00) [101]

>  [10] panov ©   (14.10.04 14:59)


> После выполнения выделенной строки выполнение поточной функции
> начнется немедленно.

ну бог в помощь :)


> CREATE_SUSPENDED


а этааа что?

---

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

>Суслик ©   (14.10.04 14:54) [8]
Привожу код TThread.Create:

constructor TThread.Create(CreateSuspended: Boolean);
begin
 inherited Create;
 AddThread;
 FSuspended := CreateSuspended;
 FCreateSuspended := CreateSuspended;
 FHandle := BeginThread(nil, 0, @ThreadProc, Pointer(Self), CREATE_SUSPENDED, FThreadID);
 if FHandle = 0 then
   raise EThread.CreateResFmt(@SThreadCreateError, [SysErrorMessage(GetLastError)]);
end;


После выполнения выделенной строки выполнение поточной функции начнется немедленно.

---

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

> К сожалению, для реализации стандартных потоков(TThread)
> в Delphi нельзя сделать Rerasing

Что значит нельзя?
Используешь AcquireExceptionObject и делаешь reraise как угодно.
См. исходный код classes для класса tthread.

---

panov   (2004-10-14 14:54:00) [104]

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

Замечание-вопрос 2
Почему в TThreadExecuter.Execute "глушится" exception? Не лучше ли бы было его как-то обработать?

Замечание 3.


Скорее всего, обработка будет Exception будет добавлена.
К сожалению, для реализации стандартных потоков(TThread) в Delphi нельзя сделать Rerasing(хотя у меня есть сторонняя реализация от y-soft, которую я обязательно разберу).

Замечание 4.

Хорошее замечание, исправлю.

---

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

>  [5] panov ©   (14.10.04 14:49)


> Так делать нельзя.

Ты не прав. Именно так и надо делать в дельфи 6. Не веришь? См. исходный код TThread.Create, AfterContruction и FCreateSuspended

---

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

> Вопрос:
> После выполнения конструктора из TThreadPool проходит некоторое
> время, прежде чем будет создана очередь сообщений для потока.
> После M := TThreadPool.Create(5) поток использовать нельзя
> до создания очереди сообщений.

Я бы сделал так.
1) Инстанцировал TThreadPool например, через метод класса TThreadPool. Например так

class function TThreadPool.Counsturct: TTHreadPool;
begin
   Result := TThreadPool.Create(...);
   SomeEvent.WaitFor;
end;

2) Событие SomeEvent устанавливал бы в signaled в TThreadPool.Execute после создания очереди потока.

---

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

Замечание-вопрос 2
Почему в TThreadExecuter.Execute "глушится" exception? Не лучше ли бы было его как-то обработать?

Замечание 3.
Аналог первого замечания для TThreadPool.Create. Хотя, здесь эта маловероятна ошибка.

Замечание 4.
В destructor TThreadPool.Destroy почему пишешь

for i := 0 to FListQueue.Count-1 do Dispose(PQueueRec(FListQueue[0]));
а нет

if Assigned(FListQueue) then for i := 0 to FListQueue.Count-1 do Dispose(PQueueRec(FListQueue[0]));

?

пока все - дальше времени нет смотреть :)

---

panov   (2004-10-14 14:49:00) [108]

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

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


Так делать нельзя.

После inherited Create(False);немедленно стартует метод Execute, соответственно, остаются непроинициализированными переменные

FreeOnTerminate := True;
FThreadFree := True;
FThreadIdOwner := aOwnerThread;
FNumThread := aNumThread;


и метод Execute будет обращаться к еще неинициализированным переменным.

---

panov   (2004-10-14 14:46:00) [109]

Вопрос:
После выполнения конструктора из TThreadPool проходит некоторое время, прежде чем будет создана очередь сообщений для потока.
После M := TThreadPool.Create(5)поток использовать нельзя до создания очереди сообщений.

Как лучше обработать эту ситуацию?
Сейчас временно в конструкторе присваивается высокий приоритет tpTimeCritical, а в методе Execute возвращается в исходное состояние.

---

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

Замечание 1.


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


В зависимости от версии дельфи делать Resume в конструкторе не очень хорошо. Если поток отработает быстро, то будет ошибка в AfterContructrion, который в дельфи 6 выглядит так

procedure TThread.AfterConstruction;
begin
 if not FCreateSuspended then
   Resume;
end;

Я бы написал просто так

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

---

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


{ TThreadPool }

constructor TThreadPool.Create(aMaxThreads: Integer);
begin
 inherited Create(True);
//Непонятно пока, как сделать с наименьшими потерями возможность
//полностью инициализироваться потоку(успеть создать очередь
//сообщений), так как высокий приоритет не гарантирует этого
 Priority := tpTimeCritical;
 FMaxThreads := aMaxThreads;       //Максимальное число потоков в пуле
 FreeOnTerminate := True;
 FListQueue := TList.Create;       //Очередь заданий на выполнение
 FSemaphore := CreateSemaphore(nil,FMaxThreads,FMaxThreads,nil);
 Resume;
end;

destructor TThreadPool.Destroy;
var
 i: Integer;
begin
{ TODO -cВажно : Добавить обработку ошибок }
//Чистим очередь заданий
 for i := 0 to FListQueue.Count-1 do Dispose(PQueueRec(FListQueue[0]));
 FListQueue.Free;
//Очистка пула потоков
 for i := 0 to Length(FPool)-1 do FPool[i].Terminate;
 CloseHandle(FSemaphore);
 inherited;
end;

procedure TThreadPool.Execute;
var
 Msg: TMsg;                        //Приемник сообщений
 p: Pointer;                       //Фиктивный указатель
 RC: DWORD;                        //Код возврата
begin
{ TODO -cВажно : Защитить код try..finally..end }
 PeekMessage(Msg,0,0,0,pm_NoRemove); //Создание очереди сообщений
//Возвращаем нормальный приоритет - очередь сообщений создана
 Priority := tpNormal;
 p := nil;
 while not Terminated do
 begin
//Ждем появления любого сообщения в очереди сообщений
   rc := MsgWaitForMultipleObjects(0,p,False,10,QS_ALLPOSTMESSAGE);
{ TODO -cВажно : Обработать ошибку }
   if rc=WAIT_FAILED then
   begin
     Terminate;
     Continue;
   end;

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

   if (rc=WAIT_OBJECT_0) or (PeekMessage(Msg,0,0,0,pm_NoRemove)) then
   begin
     GetMessage(Msg,0,0,0);
     if Msg.message=WM_QUIT then
     begin
       Terminate;
       Continue;
     end;

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

     if Msg.message=WM_AddThread then
     begin
       JobAdd(Pointer(Msg.wParam),Pointer(Msg.lParam)); //Добавить задание в очередь
       while StartThreadFromQueue do; //Передать в пул на выполнение
       Continue;
     end;
   end;
 end;

end;

procedure TThreadPool.JobAdd(const aProc: TThreadProc;const aParm: Pointer);
var
 JobRec: PQueueRec;
begin
 New(JobRec);
 JobRec^.rProc := aProc;
 JobRec^.rParm := aParm;
 FListQueue.Add(JobRec);
end;

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;
       if Length(FPool)<FMaxThreads then //В пуле еще не максимальное
       begin                             //количество потоков?
         i := Length(FPool);
         SetLength(FPool,i+1);           //Добавляем поток в пул
         FPool[i] := TThreadExecuter.Create(Self.ThreadID,i);
       end;
     end;
   end;
end;

procedure TThreadPool.AddThread(aProc: TThreadProc; aParm: Pointer);
begin
//Пересылаем в очередь сообщений сообщение о новом задании
 PostThreadMessage(Self.ThreadID,WM_AddThread,Integer(@aProc),Integer(aParm));
end;

initialization
//Регистрируем сообщение об окончании обработки задания
 WM_EndThreadExecuter := RegisterWindowMessage('TThreadPoolWM_EndThreadExecuter');
//Регистрируем сообщение о появлении нового задания на обработку
 WM_AddThread := RegisterWindowMessage('TThreadPoolWM_AddThread');
finalization

end.


Вызов происходит следующим образом(Создавал для тестирования):

type
 PParmRec=^TParmRec;
 TParmRec=record
   n: Integer;
   TimeOut: integer;
 end;

procedure Exec1(aParm: pointer);
var
   Parm: PParmRec;
begin
 Parm := aParm;
 Sleep(Parm.TimeOut);
 Form1.lb.Items.Add('End '+IntToStr(Parm^.n)+'/'+IntToStr(Parm^.TimeOut));
 Dispose(Parm);
end;

procedure TForm1.Button2Click(Sender: TObject);
var
   Parm: PParmRec;
   i: Integer;
begin
 Randomize;
 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;
end;


---

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

Наверх