rtcthrpool.pas

Delphi快速开发Web Server

  end;

procedure CloseThreadPool;
  var
    Work:TRtcWorkerThread;
    wrk,i:longword;
    havetowait:boolean;
    haveto_removecallbacks:boolean;
  begin
  havetowait:=False;

  CSThread.Enter;
  try
    haveto_removecallbacks:=InsideCallback=0;

    if assigned(ThreadPool) then
      begin
      havetowait:=True;
      while ThreadPool.Count>0 do
        begin
        wrk:=ThreadPool.search_min(i);
        ThreadPool.remove(wrk);
        Work:=TRtcWorkerThread(wrk);

        Work.PostQuit;
        end;

      Garbage(ThreadPool);
      Garbage(FreePool);
      end;

    // ------------
    if assigned(ThrList) then
      begin
      havetowait:=True;

      Garbage(ThrList);
      Garbage(WaitList);
      Garbage(WorkList);
      Garbage(Message_Quit);
      end;
    // ------------
  finally
    CSThread.Leave;
    end;

  if havetowait then CSOpen.WaitFor(10000);

  if haveto_removecallbacks then
    RemoveThreadCallbacks;

  Sleep(10);
  end;

{ TRtcThread }

constructor TRtcThread.Create;
  begin
  inherited;
  MsgList:=TXList.Create(128);
  FInfo:=TRtcInfo.Create;

  CSThread.Enter;
  try
    if not assigned(ThrList) then
      begin
      Message_Quit:=TRtcQuitMessage.Create;

      ThrList:=tBinList.Create(128);
      WaitList:=tBinList.Create(128);
      WorkList:=tBinList.Create(128);
      Thr_LastExec:=0;
      end;

    ThrList.insert(longword(self),1);
  finally
    CSThread.Leave;
    end;
  end;

destructor TRtcThread.Destroy;
  var
    o:TObject;
  begin
  CSThread.Enter;
  try
    if assigned(ThrList) then
      if ThrList.search(longword(self))>0 then
        begin
        if WaitList.search(longword(self))>0 then
          WaitList.remove(longword(self));
        if WorkList.search(longword(self))>0 then
          WorkList.remove(longword(self));
        ThrList.remove(longword(self));
        end;
  finally
    CSThread.Leave;
    end;

  while MsgList.Count>0 do
    begin
    o:=TObject(MsgList.Last);
    MsgList.removeLast;
    if o<>Message_Quit then
      Kill(o);
    end;
  MsgList.Free;

  FInfo.Free;

  inherited;
  end;

class function TRtcThread.Lock(me: TObject): boolean;
  begin
  Result:=False;
  CSThread.Enter;
  try
    if assigned(ThrList) then
      if ThrList.search(longword(me))>0 then
        Result:=True;
  finally
    if not Result then
      CSThread.Leave;
    end;
  end;

class procedure TRtcThread.UnLock;
  begin
  CSThread.Leave;
  end;

class procedure TRtcThread.Stop(me:TObject);
  begin
  CSThread.Enter;
  try
    if assigned(ThrList) then
      if ThrList.search(longword(me))>0 then
        TRtcThread.PostJob(me,Message_Quit,True,True);
  finally
    CSThread.Leave;
    end;
  end;

class function TRtcThread.PostJob(me:TObject; Job: TObject; HighPriority:boolean=False; AutoResume:boolean=False):boolean;
  var
    MyThr:TRtcWorkerThread;
    o:TObject;
  begin
  Result:=False;
  CSThread.Enter;
  try
    if assigned(ThrList) then
      begin
      if ThrList.search(longword(me))>0 then
        with TRtcThread(me) do begin
        if not Quitting then
          begin
          Result:=True;

          if Job=Message_Quit then
            begin
            while MsgList.Count>0 do
              begin
              o:=TObject(MsgList.Last);
              MsgList.removeLast;
              if o<>Message_Quit then
                Kill(o);
              end;
            end;

          if HighPriority then
            MsgList.addFirst(longword(Job))
          else
            MsgList.addLast(longword(Job));

          if AutoResume then
            Paused:=False;

          if not Paused then
            if PutWork(me) then // Post Thread to waiting list. If thread was idle, start executing next waiting thread.
              begin
              myThr:=GetThread;
              if assigned(myThr) then
                myThr.PostWork(GetWork);
              end;
          end;
        end;
      end;
  finally
    CSThread.Leave;
    end;
  end;

class function TRtcThread.GetJob(me:TObject): TObject;
  begin
  CSThread.Enter;
  try
    if assigned(ThrList)then
      begin
      if ThrList.search(longword(me))>0 then
        with TRtcThread(me) do begin
        if MsgList.Count>0 then
          begin
          Result:=TObject(MsgList.First);
          MsgList.removeFirst;
          end
        else
          Result:=nil;
        end
      else
        Result:=nil;
      end
    else
      Result:=nil;
  finally
    CSThread.Leave;
    end;
  end;

procedure TRtcThread.Pause;
  begin
  CSThread.Enter;
  try
    if assigned(ThrList) then
      if ThrList.search(longword(self))>0 then
        if not Paused and not Quitting then
          begin
          Paused:=True;
          if (WorkList.search(longword(self))=0) then // not currently working
            if WaitList.search(longword(self))>0 then // waiting for execution
              WaitList.remove(longword(self)); //remove from waiting list
          end;
  finally
    CSThread.Leave;
    end;
  end;

procedure TRtcThread.Resume;
  var
    myThr:TRtcWorkerThread;
  begin
  CSThread.Enter;
  try
    if assigned(ThrList) then
      if ThrList.search(longword(self))>0 then
        if Paused then
          begin
          Paused:=False;
          if (WorkList.search(longword(self))=0) then // not currently working
            if MsgList.Count>0 then
              if PutWork(self) then // add to waiting list
                begin
                myThr:=GetThread;
                if assigned(myThr) then
                  myThr.PostWork(GetWork);
                end;
          end;
  finally
    CSThread.Leave;
    end;
  end;

procedure TRtcThread.Idle;
  begin
  CSThread.Enter;
  try
    DoneWork(self);
    if not Paused and (MsgList.Count>0) then
      PutWork(self);
  finally
    CSThread.Leave;
    end;
  end;

procedure TRtcThread.Sync(Event: TRtcSyncEvent);
  begin
  if assigned(FThr) then
    TRtcWorkerThread(FThr).Sync(Event)
  else
    raise Exception.Create('No thread assigned.');
  end;

procedure TRtcThread.Run(Thr:TRtcWorkerThread);
  var
    Job:TObject;
    freed:boolean;
  begin
  Job:=GetJob(self);

  if Job=nil then
    Exit
  else if Job=Message_Quit then
    begin
    try
      Free;
    except
      on E:Exception do
        if LOG_THREAD_EXCEPTIONS then
          Log('RtcThread.Free',E);
      end;
    end
  else
    begin
    try
      freed:=Work(Job); // Work() has to free the 'Job' object.
    except
      on E:Exception do
        begin
        if LOG_THREAD_EXCEPTIONS then
          try
            Log('Work('+Job.ClassName+')',E);
          except
            Log('Work(undefined_Job)',E);
            end;
        try
          Free;
        except
          on E:Exception do
            if LOG_THREAD_EXCEPTIONS then
              Log('RtcThread.Free',E);
          end;
        Exit;
        end;
      end;
    try
      if not freed then Idle;
    except
      on E:Exception do
        if LOG_THREAD_EXCEPTIONS then
          Log('RtcThread.Idle',E);
      end;
    end;
  end;

function TRtcThread.Work(Job: TObject):boolean;
  begin
  Result:=False;
  if Job is TRtcJob then
    TRtcJob(Job).Run(self)
  else
    raise Exception.Create('Error!! Unknown Job class: '+Job.ClassName);
  end;

procedure TRtcThread.Kill(Job: TObject);
  begin
  if Job is TRtcJob then
    TRtcJob(Job).Kill;
  end;

function TRtcThread.ThreadID: Cardinal;
  begin
  if assigned(FThr) then
    Result:=FThr.ThreadID
  else
    Result:=0;
  end;

{ TRtcWorkerThread }

constructor TRtcWorkerThread.Create(CreateSuspended: boolean);
  begin
  CSThread.Enter;
  try
    Inc(OpenCnt);
    if OpenCnt=1 then
      CSOpen.ResetEvent;
  finally
    CSThread.Leave;
    end;

  inherited Create(True);
  Priority:=tpLower;
  FreeOnTerminate:=False;
  Run:=TRtcEvent.Create(False,False);
  if not CreateSuspended then Resume;
  end;

destructor TRtcWorkerThread.Destroy;
  begin
  Run.Free;
  inherited;
  end;

procedure TRtcWorkerThread.Execute;
  begin
  try
    DoAfterThreadStart;
  except
    end;

  try
    while Run.WaitFor(INFINITE)=wr_Signaled do
      begin
      if Msg then
        begin
        Msg:=False;
        try
          Work.Run(self);
        except
          on E:Exception do
            if LOG_THREAD_EXCEPTIONS then
              Log('Work.Run',E);
            // ignore exceptions (do not want to kill this thread)
          end;
        if not ReturnThread(self) then
          Break;
        end
      else
        Break;
      end;
  except
    on E:Exception do
      if LOG_THREAD_EXCEPTIONS then
        Log('RtcWorkThread.Execute',E);
    end;
  ClosingThread(self);

  try
    DoBeforeThreadStop;
  except
    end;
  end;

procedure TRtcWorkerThread.MySyncEvent;
  begin
  FEvent;
  end;

procedure TRtcWorkerThread.PostWork(Thr: TRtcThread);
  begin
  Msg:=True;
  Work:=Thr;
  Work.FThr:=self;
  Run.SetEvent;
  end;

procedure TRtcWorkerThread.PostQuit;
  begin
  Run.SetEvent;
  end;

procedure TRtcWorkerThread.Sync(Event: TRtcSyncEvent);
  begin
  FEvent:=Event;
  {$IFDEF FPC}
  Synchronize(@MySyncEvent);
  {$ELSE}
  Synchronize(MySyncEvent);
  {$ENDIF}
  end;

initialization
{$IFDEF CLR}
MainThr := System.Threading.Thread.CurrentThread;
{$ELSE}
MainThrID:=GetCurrentThreadID;
{$ENDIF}

ThreadCallbackCount:=0;
SetLength(ThreadCallbacks,0);
InsideCallback:=0;

CSThread:=TRtcCritSec.Create;
OpenCnt:=0;
CSOpen:=TRtcEvent.Create(True,True);

finalization
CloseThreadPool;

Garbage(CSOpen);
Garbage(CSThread);
end.