unit uThreadPool;

{   aPool.AddRequest(TMyRequest.Create(RequestParam1, RequestParam2, ...)); }

interface
uses
  Windows,
  Classes;

// 是否記錄日志
// {$DEFINE NOLOGS}

type
  TCriticalSection = class(TObject)
  protected
    FSection: TRTLCriticalSection;
  public
    constructor Create;
    destructor Destroy; override;
    // 進入臨界區
    procedure Enter;
    // 離開臨界區
    procedure Leave;
    // 嘗試進入
    function TryEnter: Boolean;
  end;

type
  // 儲存請求數據的基本類
  TWorkItem = class(TObject)
  public
    // 是否有重復任務
    function IsTheSame(DataObj: TWorkItem): Boolean; virtual;
    // 如果 NOLOGS 被定義，則禁用。
    function TextForLog: string; virtual;
  end;

type
  TThreadsPool = class;

  //線程狀態
  TThreadState = (tcsInitializing, tcsWaiting, tcsGetting, tcsProcessing,
    tcsProcessed, tcsTerminating, tcsCheckingDown);
  // 工作線程僅用于線程池內， 不要直接創建并調用它。
  TProcessorThread = class(TThread)
  private
    // 創建線程時臨時的Event對象, 阻塞線程直到初始化完成
    hInitFinished: THandle;
    // 初始化出錯信息
    sInitError: string;
    // 記錄日志
    procedure WriteLog(const Str: string; Level: Integer = 0);
  protected
    // 線程臨界區同步對像
    csProcessingDataObject: TCriticalSection;
    // 平均處理時間
    FAverageProcessing: Integer;
    // 等待請求的平均時間
    FAverageWaitingTime: Integer;
    // 本線程實例的運行狀態
    FCurState: TThreadState;
    // 本線程實例所附屬的線程池
    FPool: TThreadsPool;
    // 當前處理的數據對像。
    FProcessingDataObject: TWorkItem;
    // 線程停止 Event, TProcessorThread.Terminate 中開綠燈
    hThreadTerminated: THandle;
    uProcessingStart: DWORD;
    // 開始等待的時間, 通過 GetTickCount 取得。
    uWaitingStart: DWORD;
    // 計算平均工作時間
    function AverageProcessingTime: DWORD;
    // 計算平均等待時間
    function AverageWaitingTime: DWORD;
    procedure Execute; override;
    function IamCurrentlyProcess(DataObj: TWorkItem): Boolean;
    // 轉換枚舉類型的線程狀態為字串類型
    function InfoText: string;
    // 線程是否長時間處理同一個請求？（已死掉？）
    function IsDead: Boolean;
    // 線程是否已完成當成任務
    function isFinished: Boolean;
    // 線程是否處于空閑狀態
    function isIdle: Boolean;
    // 平均值校正計算。
    function NewAverage(OldAvg, NewVal: Integer): Integer;
  public
    Tag: Integer;
    constructor Create(APool: TThreadsPool);
    destructor Destroy; override;
    procedure Terminate;
  end;

  // 線程初始化時觸發的事件
  TProcessorThreadInitializing = procedure(Sender: TThreadsPool; aThread:
    TProcessorThread) of object;
  // 線程結束時觸發的事件
  TProcessorThreadFinalizing = procedure(Sender: TThreadsPool; aThread:
    TProcessorThread) of object;
  // 線程處理請求時觸發的事件
  TProcessRequest = procedure(Sender: TThreadsPool; WorkItem: TWorkItem;
    aThread: TProcessorThread) of object;
  TEmptyKind = (
    ekQueueEmpty, //任務被取空后
    ekProcessingFinished // 最后一個任務處理完畢后
    );
  // 任務隊列空時觸發的事件
  TQueueEmpty = procedure(Sender: TThreadsPool; EmptyKind: TEmptyKind) of
    object;

  TThreadsPool = class(TComponent)
  private
    csQueueManagment: TCriticalSection;
    csThreadManagment: TCriticalSection;
    FProcessRequest: TProcessRequest;
    FQueue: TList;
    FQueueEmpty: TQueueEmpty;
    // 線程超時閥值
    FThreadDeadTimeout: DWORD;
    FThreadFinalizing: TProcessorThreadFinalizing;
    FThreadInitializing: TProcessorThreadInitializing;
    // 工作中的線程
    FThreads: TList;
    // 執行了 terminat 發送退出指令, 正在結束的線程.
    FThreadsKilling: TList;
    // 最少, 最大線程數
    FThreadsMax: Integer;
    // 最少, 最大線程數
    FThreadsMin: Integer;
    // 池平均等待時間
    function PoolAverageWaitingTime: Integer;
    procedure WriteLog(const Str: string; Level: Integer = 0);
  protected
    FLastGetPoint: Integer;
    // Semaphore， 統計任務隊列
    hSemRequestCount: THandle;
    // Waitable timer. 每30觸發一次的時間量同步
    hTimCheckPoolDown: THandle;
    // 線程池停機（檢查并清除空閑線程和死線程）
    procedure CheckPoolDown;
    // 清除死線程，并補充不足的工作線程
    procedure CheckThreadsForGrow;
    procedure DoProcessed;
    procedure DoProcessRequest(aDataObj: TWorkItem; aThread: TProcessorThread);
      virtual;
    procedure DoQueueEmpty(EmptyKind: TEmptyKind); virtual;
    procedure DoThreadFinalizing(aThread: TProcessorThread); virtual;
    // 執行事件
    procedure DoThreadInitializing(aThread: TProcessorThread); virtual;
    // 釋放 FThreadsKilling 列表中的線程
    procedure FreeFinishedThreads;
    // 申請任務
    procedure GetRequest(out Request: TWorkItem);
    // 清除死線程
    procedure KillDeadThreads;
  public
    constructor Create(AOwner: TComponent); override;
    destructor Destroy; override;
    // 就進行任務是否重復的檢查, 檢查發現重復就返回 False
    function AddRequest(aDataObject: TWorkItem; CheckForDoubles: Boolean =
      False): Boolean; overload;
    // 轉換枚舉類型的線程狀態為字串類型
    function InfoText: string;
  published
    // 線程處理任務時觸發的事件
    property OnProcessRequest: TProcessRequest read FProcessRequest write
      FProcessRequest;
    // 任務列表為空時解發的事件
    property OnQueueEmpty: TQueueEmpty read FQueueEmpty write FQueueEmpty;
    // 線程結束時觸發的事件
    property OnThreadFinalizing: TProcessorThreadFinalizing read
      FThreadFinalizing write FThreadFinalizing;
    // 線程初始化時觸發的事件
    property OnThreadInitializing: TProcessorThreadInitializing read
      FThreadInitializing write FThreadInitializing;
    // 線程超時值（毫秒), 如果處理超時，將視為死線程
    property ThreadDeadTimeout: DWORD read FThreadDeadTimeout write
      FThreadDeadTimeout default 0;
    // 最大線程數
    property ThreadsMax: Integer read FThreadsMax write FThreadsMax default 1;
    // 最小線程數
    property ThreadsMin: Integer read FThreadsMin write FThreadsMin default 0;
  end;

type
  //日志記志函數
  TLogWriteProc = procedure(
    const Str: string; //日志
    LogID: Integer = 0;
    Level: Integer = 0 //Level = 0 - 跟蹤信息, 10 - 致命錯誤
    );

var
  WriteLog: TLogWriteProc; // 如果存在實例就寫日志

implementation
uses
  SysUtils;

// 儲存請求數據的基本類
{
********************************** TWorkItem ***********************************
}

function TWorkItem.IsTheSame(DataObj: TWorkItem): Boolean;
begin
  Result := False;
end; { TWorkItem.IsTheSame }

function TWorkItem.TextForLog: string;
begin
  Result := 'Request';
end; { TWorkItem.TextForLog }

{
********************************* TThreadsPool *********************************
}

constructor TThreadsPool.Create(AOwner: TComponent);
var
  DueTo: Int64;
begin
{$IFNDEF NOLOGS}
  WriteLog('創建線程池', 5);
{$ENDIF}
  inherited;
  csQueueManagment := TCriticalSection.Create;
  FQueue := TList.Create;
  csThreadManagment := TCriticalSection.Create;
  FThreads := TList.Create;
  FThreadsKilling := TList.Create;
  FThreadsMin := 0;
  FThreadsMax := 1;
  FThreadDeadTimeout := 0;
  FLastGetPoint := 0;
  //
  hSemRequestCount := CreateSemaphore(nil, 0, $7FFFFFFF, nil);

  DueTo := -1;
  //可等待的定時器（只用于Window NT4或更高）
  hTimCheckPoolDown := CreateWaitableTimer(nil, False, nil);

  if hTimCheckPoolDown = 0 then // Win9x不支持
    // In Win9x number of thread will be never decrised
    hTimCheckPoolDown := CreateEvent(nil, False, False, nil)
  else
    SetWaitableTimer(hTimCheckPoolDown, DueTo, 30000, nil, nil, False);
end; { TThreadsPool.Create }

destructor TThreadsPool.Destroy;
var
  n, i: Integer;
  Handles: array of THandle;
begin
{$IFNDEF NOLOGS}
  WriteLog('線程池銷毀', 5);
{$ENDIF}
  csThreadManagment.Enter;

  SetLength(Handles, FThreads.Count);
  n := 0;
  for i := 0 to FThreads.Count - 1 do
    if FThreads[i] <> nil then
    begin
      Handles[n] := TProcessorThread(FThreads[i]).Handle;
      TProcessorThread(FThreads[i]).Terminate;
      Inc(n);
    end;

  csThreadManagment.Leave;  // lixiaoyu 添加于 2009.1.6，如沒有此行代碼無法成功釋放正在執行中的工作者線程，死鎖。

  WaitForMultipleObjects(n, @Handles[0], True, 30000);  // 等待工作者線程執行終止  lixiaoyu 注釋于 2009.1.6

  csThreadManagment.Enter;  // lixiaoyu 添加于 2009.1.6 再次進入鎖定，并釋放資源
  for i := 0 to FThreads.Count - 1 do
    TProcessorThread(FThreads[i]).Free;
  FThreads.Free;
  FThreadsKilling.Free;
  csThreadManagment.Free;

  csQueueManagment.Enter;
  for i := FQueue.Count - 1 downto 0 do
    TObject(FQueue[i]).Free;
  FQueue.Free;
  csQueueManagment.Free;

  CloseHandle(hSemRequestCount);
  CloseHandle(hTimCheckPoolDown);
  inherited;
end; { TThreadsPool.Destroy }

function TThreadsPool.AddRequest(aDataObject: TWorkItem; CheckForDoubles:
  Boolean = False): Boolean;
var
  i: Integer;
begin
{$IFNDEF NOLOGS}
  WriteLog('AddRequest(' + aDataObject.TextForLog + ')', 2);
{$ENDIF}
  Result := False;
  csQueueManagment.Enter;
  try
    // 如果 CheckForDoubles = TRUE
    // 則進行任務是否重復的檢查
    if CheckForDoubles then
      for i := 0 to FQueue.Count - 1 do
        if (FQueue[i] <> nil)
          and aDataObject.IsTheSame(TWorkItem(FQueue[i])) then
          Exit; // 發現有相同的任務

    csThreadManagment.Enter;
    try
      // 清除死線程，并補充不足的工作線程
      CheckThreadsForGrow;

      // 如果 CheckForDoubles = TRUE
      // 則檢查是否有相同的任務正在處理中
      if CheckForDoubles then
        for i := 0 to FThreads.Count - 1 do
          if TProcessorThread(FThreads[i]).IamCurrentlyProcess(aDataObject) then
            Exit; // 發現有相同的任務

    finally
      csThreadManagment.Leave;
    end;

    //將任務加入隊列
    FQueue.Add(aDataObject);

    //釋放一個同步信號量
    ReleaseSemaphore(hSemRequestCount, 1, nil);
{$IFNDEF NOLOGS}
    WriteLog('釋放一個同步信號量)', 1);
{$ENDIF}
    Result := True;
  finally
    csQueueManagment.Leave;
  end;
{$IFNDEF NOLOGS}
  //調試信息
  WriteLog('增加一個任務(' + aDataObject.TextForLog + ')', 1);
{$ENDIF}
end; { TThreadsPool.AddRequest }

{
函 數 名：TThreadsPool.CheckPoolDown
功能描述：線程池停機（檢查并清除空閑線程和死線程）
輸入參數：無
返 回 值: 無
創建日期：2006.10.22 11:31
修改日期：2006.
作    者：Kook
附加說明：
}

procedure TThreadsPool.CheckPoolDown;
var
  i: Integer;
begin
{$IFNDEF NOLOGS}
  WriteLog('TThreadsPool.CheckPoolDown', 1);
{$ENDIF}
  csThreadManagment.Enter;
  try
{$IFNDEF NOLOGS}
    WriteLog(InfoText, 2);
{$ENDIF}
    // 清除死線程
    KillDeadThreads;
    // 釋放 FThreadsKilling 列表中的線程
    FreeFinishedThreads;

    // 如果線程空閑，就終止它
    for i := FThreads.Count - 1 downto FThreadsMin do
      if TProcessorThread(FThreads[i]).isIdle then
      begin
        //發出終止命令
        TProcessorThread(FThreads[i]).Terminate;
        //加入待清除隊列
        FThreadsKilling.Add(FThreads[i]);
        //從工作隊列中除名
        FThreads.Delete(i);
        //todo: ？？
        Break;
      end;
  finally
    csThreadManagment.Leave;
  end;
end; { TThreadsPool.CheckPoolDown }

{
函 數 名：TThreadsPool.CheckThreadsForGrow
功能描述：清除死線程，并補充不足的工作線程
輸入參數：無
返 回 值: 無
創建日期：2006.10.22 11:31
修改日期：2006.
作    者：Kook
附加說明：
}

procedure TThreadsPool.CheckThreadsForGrow;
var
  AvgWait: Integer;
  i: Integer;
begin
  {
    New thread created if:
    新建線程的條件：
      1. 工作線程數小于最小線程數
      2. 工作線程數小于最大線程數 and 線程池平均等待時間 < 100ms(系統忙)
      3. 任務大于工作線程數的4倍
  }

  csThreadManagment.Enter;
  try
    KillDeadThreads;
    if FThreads.Count < FThreadsMin then
    begin
{$IFNDEF NOLOGS}
      WriteLog('工作線程數小于最小線程數', 4);
{$ENDIF}
      for i := FThreads.Count to FThreadsMin - 1 do
      try
        FThreads.Add(TProcessorThread.Create(Self));
      except
        on e: Exception do

          WriteLog(
            'TProcessorThread.Create raise: ' + e.ClassName + #13#10#9'Message: '
            + e.Message,
            9
            );
      end
    end
    else if FThreads.Count < FThreadsMax then
    begin
{$IFNDEF NOLOGS}
      WriteLog('工作線程數小于最大線程數 and 線程池平均等待時間 < 100ms', 3);
{$ENDIF}
      AvgWait := PoolAverageWaitingTime;
{$IFNDEF NOLOGS}
      WriteLog(Format(
        'FThreads.Count (%d)<FThreadsMax(%d), AvgWait=%d',
        [FThreads.Count, FThreadsMax, AvgWait]),
        4
        );
{$ENDIF}

      if AvgWait < 100 then
      try
        FThreads.Add(TProcessorThread.Create(Self));
      except
        on e: Exception do
          WriteLog(
            'TProcessorThread.Create raise: ' + e.ClassName +
            #13#10#9'Message: ' + e.Message,
            9
            );
      end;
    end;
  finally
    csThreadManagment.Leave;
  end;
end; { TThreadsPool.CheckThreadsForGrow }

procedure TThreadsPool.DoProcessed;
var
  i: Integer;
begin
  if (FLastGetPoint < FQueue.Count) then
    Exit;
  csThreadManagment.Enter;
  try
    for i := 0 to FThreads.Count - 1 do
      if TProcessorThread(FThreads[i]).FCurState in [tcsProcessing] then
        Exit;
  finally
    csThreadManagment.Leave;
  end;
  DoQueueEmpty(ekProcessingFinished);
end; { TThreadsPool.DoProcessed }

procedure TThreadsPool.DoProcessRequest(aDataObj: TWorkItem; aThread:
  TProcessorThread);
begin
  if Assigned(FProcessRequest) then
    FProcessRequest(Self, aDataObj, aThread);
end; { TThreadsPool.DoProcessRequest }

procedure TThreadsPool.DoQueueEmpty(EmptyKind: TEmptyKind);
begin
  if Assigned(FQueueEmpty) then
    FQueueEmpty(Self, EmptyKind);
end; { TThreadsPool.DoQueueEmpty }

procedure TThreadsPool.DoThreadFinalizing(aThread: TProcessorThread);
begin
  if Assigned(FThreadFinalizing) then
    FThreadFinalizing(Self, aThread);
end; { TThreadsPool.DoThreadFinalizing }

procedure TThreadsPool.DoThreadInitializing(aThread: TProcessorThread);
begin