threadtimer.pas

由delphi实现的bt下载器示例程序

unit ThreadTimer;

interface

//{$DEFINE DEBUGMSG}

uses
  FastShareMem, Windows, SysUtils, Classes, Messages, MemPools{$IFDEF DEBUGMSG}, DebugUnit{$ENDIF};

const
  WSM_TIMER         =  WM_USER + 390;
  WSM_USER_FUNC     =  WM_USER + 391;
  WSM_SET_FUNC      =  WM_USER + 392;
  WSM_DEL_FUNC      =  WM_USER + 393;

  MAX_THREADJOB     =  1024;
  MAX_THREAD_CNT    =  6;

  IOCP_TIMERJOB     =  Cardinal(-2);

type
{****** Window Handle ******}
  TTimerWnd = class
  public
    Wnd: Cardinal;
    Thrd: Cardinal;
    MsgQueue: array of TWndMethod;
    QCount: Integer;
    constructor Create;
    destructor Destroy; override;
    procedure WndProc(var Message: TMessage);
    procedure AddMsgProc(Proc: TWndMethod);
    procedure RemoveMsgProc(Proc: TWndMethod);
  end;

{****** Thread timer ******}
  TThreadTimerCallback = procedure (WParam, LParam: Integer) of object;
  TThreadTimerCallbackProc = procedure (Sender: Pointer; Wparam, LParm: Integer);

  TIOCPCallback = procedure (oSelf: TObject; Key: Integer; Len: Integer;
                            OV: Pointer; ErrCode: DWORD);

  TTimerInfo = record
    Interval: Cardinal;
    NextTick: Cardinal;
    CallBack: Pointer;
    Sender: Pointer;
    Param: Integer;
    Param2: Integer;
    RunInThrd: LongBool;
    Ref: Integer;
  end;
  PTimerInfo = ^TTimerInfo;

  TIOCPOverlapped = record
    OV: TOverlapped;
    CallBack: Pointer;
    oSelf: TObject;
    BufLen: Integer;
    Buf: Pointer;
    Reserved1, Reserved2: Cardinal;
  end;
  PIOCPOverlapped = ^TIOCPOverlapped;

  TThreadTimer = class(TThread)
  protected
    procedure Execute; override;
  public
    WaitEvent: Cardinal;
    //Lock: TRTLCriticalSection;
    //Jobs: array of PTimerInfo;
    //JobCnt: Integer;
    Jobs: TFIFOBuffer;
    constructor Create;
    function AddJob(WParam, LParam: Integer; Interval: Cardinal;
               CallBack: TThreadTimerCallback; RunOnce: Boolean;
               RunInThread: Boolean = false): Cardinal; overload;
    function AddJob(WParam, LParam: Integer; Interval: Cardinal;
               CallBack: TThreadTimerCallbackProc; Sender: Pointer; RunOnce: Boolean;
               RunInThread: Boolean = false): Cardinal; overload;
    function AddJobAnyway(WParam, LParam: Integer; Interval: Cardinal;
               CallBack, Sender: Pointer; RunOnce: Boolean;
               RunInThread: Boolean = false): Cardinal; overload;
    procedure deleteJob(Handle: Cardinal);
    procedure Stop;
  end;

  TWorkerThread = class(TThread)
  protected
    procedure Execute; override;
  public
    WaitEvent: Cardinal;
    JobCnt: Integer;
    constructor Create;
    procedure Stop; virtual;
  end;

  TWorkerThreadNT = class(TWorkerThread)
  protected
    procedure Execute; override;
  public
    GlobalCPort: Cardinal;
    constructor Create(Pt: Cardinal);
    procedure Stop; override;
  end;

  TWorkerThreadPool = class
    Pool: array of TWorkerThread;
    GlobalCPort: Cardinal;
    constructor Create(Cnt: Integer);
    destructor Destroy; override;
    function AllocThread: TWorkerThread;
    function PostJob(Func: Pointer; Data: Cardinal): Cardinal;
  end;

procedure RegisterMsgHandle(WndProc: TWndMethod);
procedure UnregisterMsgHandle(WndProc: TWndMethod);
function IsFree: Boolean;
procedure SetGlobalVal(Index: Integer; Val: Cardinal);
function GetGlobalVal(Index: Integer): Cardinal;
function AllocLock: PRTLCriticalSection;
procedure DeallocLock(Lck: PRTLCriticalSection);
function AllocTempFile(Path, Pref, Ext: string; IncludePath: Boolean): string;
function RandomStr(Len: Integer): string;
procedure GetInts(var p: PChar; Dst: Pointer; IntLen: Integer; IntCnt: Integer;
    Negative: Boolean);
function Buf2Str(const buf; Len: Integer): string;
function StrToHex(s: string): string;
function BufToHex(const buf; Len: Integer): string;

procedure RegisterTimer(var Handle: Cardinal; WParam, LParam: Integer; Interval: Cardinal;
          Callback, Sender: Pointer; RunOnce, RunInThread: Boolean); overload;
procedure RegisterTimer(var Handle: Cardinal; WParam, LParam: Integer; Interval: Cardinal;
          Callback: TThreadTimerCallback; RunOnce, RunInThread: Boolean); overload;
procedure UnregisterTimer(var Handle: Cardinal);

procedure SockCallback(const dwError, cbTransferred : DWORD;
                        const lpOverlapped : PIOCPOverlapped;
                        const dwFlags : DWORD); stdcall;

function MakeBuffer(Bufs: array of const): string;

var
  GlobalTimer: TThreadTimer;
  GlobalHwnd: Cardinal;
  GlobalThreadPool: TWorkerThreadPool;

implementation

{$I ShareGlobals.inc}

var
  TimerWnd: TTimerWnd;
  ThreadRef: Integer = 0;
  TimerInfos: TFixedMemPool;

function IsFree: Boolean;
begin
  result := fastsharemem.ThisFree;
end;

procedure SetGlobalVal(Index: Integer; Val: Cardinal);
begin
  fastsharemem.SetGlobalVal(Index, Val);
end;

function GetGlobalVal(Index: Integer): Cardinal;
begin
  result := fastsharemem.GetGlobalVal(Index);
end;

procedure RegisterMsgHandle(WndProc: TWndMethod);
begin
  TimerWnd.AddMsgProc(wndproc);
end;

procedure UnregisterMsgHandle(WndProc: TWndMethod);
begin
  TimerWnd.RemoveMsgProc(WndProc);
end;

var
  CreateProcess: Cardinal = 0;

procedure InitAll;
begin
  IsMultiThread := True;
  if thisfree or (getglobalval(GLOBAL_TIMERINFO) = 0) then
  begin
    createprocess := getcurrentprocess;
    TimerInfos := TFixedMemPool.Create(sizeof(TTimerInfo), MAX_THREADJOB);
    TimerWnd := TTimerWnd.Create;
    globalhwnd := timerwnd.Wnd;
    GlobalTimer := TThreadTimer.Create;
    //setlength(GlobalThreadPool, MAX_THREAD_CNT);
    GlobalThreadPool := TWorkerThreadPool.Create(MAX_THREAD_CNT);
    fastsharemem.SetGlobalVal(GLOBAL_TIMERINFO, cardinal(TimerInfos));
    fastsharemem.SetGlobalVal(GLOBAL_TIMERWND, Cardinal(TimerWnd));
    fastsharemem.SetGlobalVal(GLOBAL_TIMER, cardinal(GlobalTimer));
    fastsharemem.setGlobalVal(GLOBAL_THREADPOOL, Cardinal(GlobalThreadPool));
  end
  else begin
    TimerInfos := TFixedMemPool(fastsharemem.GetGlobalVal(GLOBAL_TIMERINFO));
    TimerWnd := TTimerWnd(fastsharemem.GetGlobalVal(GLOBAL_TIMERWND));
    GlobalHwnd := TimerWnd.Wnd;
    GlobalTimer := TThreadTimer(fastsharemem.GetGlobalVal(GLOBAL_TIMER));
    GlobalThreadPool := pointer(fastsharemem.GetGlobalVal(GLOBAL_THREADPOOL));
  end;
end;

procedure UninitTimerWnd;
begin
  if getcurrentprocess = createprocess then
  begin
    try
      GlobalTimer.Stop;
      TimerWnd.Free;
      globalthreadpool.Free;
      TimerInfos.Free;
    except
    {$IFDEF DEBUGMSG}
    on e: exception do
      LogDbgMsg('UninitTimerWnd error: '+e.Message);
    {$ENDIF}
    end;
    fastsharemem.SetGlobalVal(GLOBAL_TIMERINFO, 0);
    fastsharemem.SetGlobalVal(GLOBAL_TIMERWND, 0);
    fastsharemem.SetGlobalVal(GLOBAL_TIMER, 0);
    fastsharemem.SetGlobalVal(GLOBAL_CPORT, 0);
    fastsharemem.setGlobalVal(GLOBAL_THREADPOOL, 0);
  end;
end;

function AllocInfo: PTimerInfo;
begin
  //result := TimerInfos.Get;
  result := allocmem(sizeof(TTimerInfo));
end;

procedure DeallocInfo(p: PTimerInfo);
begin
  if interlockeddecrement(p^.ref) < 0 then
    //TimerInfos.Put(p);
    dispose(p);
end;

procedure DeleteLock(Sender: Pointer; Param, Param2: Integer);
begin
  try
    deletecriticalsection(PRTLCriticalsection(param)^);
    dispose(PRTLCriticalSection(param));
  except
  end;
end;

function AllocLock: PRTLCriticalSection;
begin
  result := allocmem(sizeof(TRTLCriticalsection));
  initializecriticalsection(result^);
end;

procedure DeallocLock(Lck: PRTLCriticalSection);
begin
  globaltimer.AddJob(Integer(lck), 0, 50, DeleteLock, nil, true, true);
end;

function RandomStr(Len: Integer): string;
const
  ValidChrs: string = '_0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ';
var
  i: Integer;
begin
  setlength(result, len);
  for i := 1 to length(result) do
    result[i] := validchrs[random(length(validchrs))+1];
end;

function AllocTempFile(Path, Pref, Ext: string; IncludePath: Boolean): string;
var
  s: array [0..MAX_PATH+1] of Char;
  ps: string;
begin
  if (path<>'') and not directoryexists(path) then
    forcedirectories(path)
  else if path = '' then
  begin
    gettemppath(MAX_PATH, pchar(@s));
    path := s;
  end;
  gettempfilename(pchar(path), pchar(pref), 0, pchar(@s));
  ps := s;
  result := ps;
  if comparetext(ext, '.tmp')<>0 then
    repeat
      ext := randomstr(1)+ext;
      result := changefileext(ps, ext);
    until renamefile(pchar(ps), pchar(result));
  if not includepath then result := extractfilename(result);
end;

procedure GetInts(var p: PChar; Dst: Pointer; IntLen: Integer; IntCnt: Integer;
    Negative: Boolean);
var
  v: Integer;
  Neg: Boolean;
begin
  v := 0;
  Neg := false;
  while not (p^ in [#0, #9, #13, '-', '0'..'9']) do inc(p);
  while not (p^ in [#0, #9, #13]) and (IntCnt>0) do
  begin
    case p^ of
      '0'..'9':
      begin
        v := v * 10 + byte(p^)-48;
        inc(p);
      end;
    else
      if negative and (p^='-') then
      begin
        Neg := true;
        inc(p);
        v := 0;
        while not (p^ in [#0, #9, #13, '0'..'9']) do
          inc(p);
      end
      else begin
        if neg then v := -v;
        move(v, dst^, intlen);
        inc(p);
        v := 0;
        neg := false;
        dst := pointer(integer(dst)+intlen);
        dec(intcnt);
        while not (p^ in [#0, #9, #13, '-', '0'..'9']) do
          inc(p);
      end;
    end;
  end;
end;

function Buf2Str(const buf; Len: Integer): string;
begin
  setstring(result, pchar(@buf), len);
end;

function StrToHex(s: string): string;
var
  i: Integer;
begin
  result := '';
  for i := 1 to length(s) do
    result := result + inttohex(byte(s[i]), 2)+' ';
end;

function BufToHex(const buf; Len: Integer): string;
var
  i: Integer;
  p: PByte;
begin
  result := '';
  p := @Buf;
  for i := 1 to Len do
  begin
    result := result + inttohex(p^, 2)+' ';
    inc(p);
  end;
end;

procedure RegisterTimer(var Handle: Cardinal; WParam, LParam: Integer; Interval: Cardinal;
          Callback, Sender: Pointer; RunOnce, RunInThread: Boolean); overload;
begin
  if handle <> 0 then
    globaltimer.deleteJob(handle);
  handle := globaltimer.AddJobanyway(WParam, LParam, interval, callback, sender,
            runonce, runinthread);
end;

procedure RegisterTimer(var Handle: Cardinal; WParam, LParam: Integer; Interval: Cardinal;
          Callback: TThreadTimerCallback; RunOnce, RunInThread: Boolean); overload;
begin
  if handle <> 0 then
    globaltimer.deleteJob(handle);
  handle := globaltimer.AddJob(WParam, LParam, interval, callback, runonce, runinthread);
end;

procedure UnregisterTimer(var Handle: Cardinal);
begin
  if handle <> 0 then
  begin
    globaltimer.deleteJob(handle);
    handle := 0;
  end;
end;


type
  TThrdCallback = procedure (Sender: Pointer); stdcall;

procedure ProcessTimer(Sender:PTimerInfo); stdcall;
begin
  try
    if assigned(sender.CallBack) then
    try
      TThreadTimerCallbackProc(sender.callback)(sender.Sender, sender.Param, sender.Param2);
    except
    {$IFDEF DEBUGMSG}
    on e: exception do
      LogDbgMsg('ProcessTimer error: '+e.Message);
    {$ENDIF}
    end;
    InterlockedDecrement(sender.Ref);
    if sender.Interval = 0 then
      deallocinfo(sender);
  except
  {$IFDEF DEBUGMSG}
  on e: exception do
    LogDbgMsg('ProcessTimer error: '+e.Message);
  {$ENDIF}
  end;
end;

procedure SockCallback(const dwError, cbTransferred : DWORD;
                        const lpOverlapped : PIOCPOverlapped;
                        const dwFlags : DWORD); stdcall;
begin
  try
    with lpOverlapped^ do
    begin
      //if dwError = 0 then dwError := Handle;
      TIOCPCallback(Callback)(oSelf, 0, cbTransferred, lpOverlapped, dwError);
    end;
  except
  end;
end;

function MakeBuffer(Bufs: array of const): string;
var
  i: Integer;
  len, l: Integer;
  Flg, IsP: Boolean;

begin
{ calc result length }
  Flg := false;
  len := 0;
  l := 0;
  for i := 0 to high(bufs) do
    with bufs[i] do
    begin
      if flg then
      begin
        if VType = vtInteger then
          l := VInteger
        else
          l := Length(String(Bufs[i-1].VPointer)) and $FFFFFFFC - 4;
        inc(len, l);
        flg := false;
        if VType = vtInteger then continue;
      end;
      case VType of
        vtInteger    : l := 4;
        vtBoolean    ,
        vtChar       : l := 1;
        vtExtended   : l := sizeof(Extended);
        vtString     : l := byte(VString^[0])+1;
        vtPointer    : Flg := True;
        vtPChar      : l := strlen(VPChar);
        vtWideChar   : l := 2;
        vtAnsiString : l := length(string(VAnsiString));
        vtCurrency   : l := sizeof(Currency);
        vtVariant    : l := length(string(VVariant^));
        vtInterface  : l := length(string(VInterface^));
        vtInt64      : l := sizeof(Int64);
      else
        l := 0;
      end;
      if not flg and (l > 0) then
        inc(len, l);
    end;
  if Flg then
  begin
    l := length(string(Bufs[high(bufs)].VPointer)) and $FFFFFFFC - 4;
    inc(Len, l);
  end;

  setlength(result, len);

{ copy buffers }
  len := 0;
  flg := false;
  //l := 0;
  for i := 0 to high(bufs) do
    with Bufs[i] do
    begin
      IsP := True;
      if Flg then
      begin
        if VType = vtInteger then
          l := VInteger
        else l := length(string(Bufs[i-1].VPointer)) and $FFFFFFFC - 4;
        if l > 0 then
        begin
          move(Bufs[i-1].VPointer^, result[Len+1], l);
          inc(Len, l);
        end;
        Flg := false;
        if VType = vtInteger then continue;
      end;
      case VType of
        vtInteger    :
        begin
          l := 4;
          isp := false;
        end;
        vtBoolean    ,
        vtChar       :