Наработки пула потоков

[Delphi_ProGer]

Здраствуйте форумчане!
Мог бы кто-нибудь поделиться своим пулом потоков(статическим)?

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

[3D Hunter]

Я бы делал так:
1. Главный класс пула потоков. В конструкторе число созданных потоков. Метод взятия свободного потока с указанием процедуры для Execute, которую поток должен делать. Метод сбора всех отработавших потоков обратно в пул.
2. Класс рабочего потока.

Ну и обрабатывать внештатные ситуации, сервисные функции (текущее число заняных потоков, процент использования пула и т.п.)

[Человек_Борща]

Вот же описание обьектного пула: http://ru.wikipedia.org/wiki/Объектный_пул
Создать можно на основе TList из contrs.pas

[Delphi_ProGer]

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

unit ThreadPool;
interface
uses Classes, ExtCtrls, dialogs,windows;
 
Type
TDWorkProc = procedure();
TDOnDoneEvent = procedure(ThreadId: Word);
TDOnErrorEvent = procedure(ThreadId: Word);
 
TDThread = class(TThread)
    private
      HasJob: Boolean;
      Param: Pointer;
    protected
      procedure Execute; override;
    public
      OnWorkProc: TDWorkProc;
      OnDone: TDOnDoneEvent;
      OnError: TDOnErrorEvent;
end;
 
TDPool = class
    private
     Threads: array of TDThread;
     TaskList: TList;
     fPriority: TThreadPriority;
     fSize: Word;
     OnWork: TDWorkProc;
    function GetPriority: TThreadPriority;
    function GetSize: Word;
    procedure SetPriority(const Value: TThreadPriority);
    procedure SetSize(const Value: Word);
    function GetJobsCount: Word;
    function UpdatePool(Sender: TObject): Boolean;
 
    protected
 
    public
     Suspended: Boolean;
     OnWorkDone: TDOnDoneEvent;
     OnError: TDOnErrorEvent;
     property Priority: TThreadPriority read GetPriority write SetPriority;
     property Size: Word read GetSize write SetSize default 20;
     Procedure Start;
     Procedure Pause;
     Procedure Stop;
     Constructor Create(aSize: word=20);
     Destructor Destroy; override;
     Procedure AddTask(ThreadProc: TDWorkProc);
     Procedure DeleteTask(Index: Integer);
 
  end;
{*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*_*}
 
implementation
 
uses SysUtils;
Var
  ActiveJobs: Word=0;
  aSelf: TObject;
 
{ TPool }
 
Procedure TDPool.AddTask(ThreadProc: TDWorkProc);
var i: Word;
begin
  if (ActiveJobs=Size)or(Suspended=True) then //якщо всі потоки працюють,
    begin                 //чи пул призупинений, то...
      TaskList.Add(@ThreadProc);//додамо у чергу робіт!
    end
  else
    begin
     for i:=0 to Size-1 do
       if not(Threads[i].HasJob) then
       begin
         Threads[i].OnWorkProc:=ThreadProc;
         Threads[i].OnDone:=OnWorkDone;
         Threads[i].OnError:=OnError;
         Threads[i].HasJob:=True;
         Threads[i].Resume;
         Break;
       end;
     Inc(ActiveJobs);
    end;
end;
 
constructor TDPool.Create(aSize: word=20);
var i:word;
begin
  aSelf := Self;
  Size := aSize;
  ActiveJobs:=0;
  SetLength(Threads, Size);
  for i:=0 to Size-1 do
  begin
    Threads[i]:=TDThread.Create(True);
    Threads[i].Priority:=tpLowest;
  end;
  TaskList:=TList.Create;
  //*
  Suspended:=False;
end;
 
Procedure TDPool.DeleteTask(Index: Integer);
begin
 
end;
 
destructor TDPool.Destroy;
begin
  Stop;
  FreeAndNil(TaskList);
end;
 
function TDPool.GetJobsCount: Word;
begin
  Result := ActiveJobs;
end;
 
function TDPool.GetPriority: TThreadPriority;
begin
  Result:= fPriority;
end;
 
function TDPool.GetSize: Word;
begin
  Result := fSize;
end;
 
function TDPool.UpdatePool(Sender: TObject): Boolean;
var i: Word;
begin
  //update the thread's jobs
with TaskList do
begin
  if Count>0 then
    for i := 0 to Size-1 do
      if not(Threads[i].HasJob) then
       begin
         Threads[i].OnWorkProc:=Items[Count-1];
         Threads[i].OnDone:=OnWorkDone;
         Threads[i].OnError:=OnError;
         Threads[i].HasJob:=True;
         Delete(Count-1);
         Pack;
         Break;
         ResumeThread((aSelf as TDThread).Handle);
       end
 
  else
 ResumeThread((aSelf as TDThread).Handle);
end;
end;
 
procedure TDPool.Pause;
var i: Word;
begin
  for i:=0 to Size-1 do
    Threads[i].Suspend;
end;
 
 
procedure TDPool.SetPriority(const Value: TThreadPriority);
var i: Word;
begin
  for i:=Low(Threads) to High(Threads) do
    Threads[i].Priority:=Value;
end;
 
procedure TDPool.SetSize(const Value: Word);
begin
  if Value>500 then
    fSize := 500
  else
  if Value=0 then
    fSize := 10
  else
    fSize := Value;
end;
 
procedure TDPool.Start;
var i: Word;
begin
 for i:=0 to Size-1 do
   if (Threads[i].HasJob) and (Threads[i].Suspended) then
     Threads[i].Resume;
end;
 
procedure TDPool.Stop;
var i: Word;
begin
 for i:=0 to Size-1 do
   if (Threads[i].HasJob) and not(Threads[i].Suspended) then
     Threads[i].Terminate;
end;
 
{ TDThread }
procedure TDThread.Execute;
begin
  inherited;
while not Terminated do
  begin
    if Assigned(OnWorkProc) then
  try
    OnWorkProc;
    if Assigned(OnDone) then
    OnDone(Self.ThreadID);
    Self.HasJob := False;
  except
    if Assigned(OnError) then
      OnError(Self.ThreadID);
  end;
    Dec(ActiveJobs);
    OnWorkProc:=nil;
    (aSelf as TDPool).UpdatePool(aSelf);
    Sleep(100);
  end;
end;
 
end.

[Delphi_ProGer]

Сообщение от Человек_Борща

Создать можно на основе TList из contrs.pas

что за contrs?? у меня нет такого. Delphi 7

[Человек_Борща]

contrs в общем как-то так. Этот модуль есть в delphi 7. Точнее посмотрите в справке. Там сложно-стрёмное имя.

У пула потоков есть 2 списка:
1. Активные потоки: те которые заняты работой.
2. Ожидающие потоки: те которые ожидают работу.
Сам пул потоков это TObject. Его списки это TList.

Кроме того, у вас вообще не пул потоков. В пуле потоков, число объектов постоянно. Завершенные потоки должны обнулиться, но не уничтожится, и переместиться в список ожидающих потоков.