Forum > Other
TPowerThread
mercurhyo:
I just put here a template that I'am working on, as a case study around Pascal power
It compiles but it is unfinished yet. I will put a final version later.
Goal =
- Command to Threads with 'orders/answers records' of data pushed on queues
- Exec methods in threads
- sync or queue callbacks to main thread after methods ends
- automaticaly THEN free 'orders/answers' records memory
--- Code: Pascal [+][-]window.onload = function(){var x1 = document.getElementById("main_content_section"); if (x1) { var x = document.getElementsByClassName("geshi");for (var i = 0; i < x.length; i++) { x[i].style.maxHeight='none'; x[i].style.height = Math.min(x[i].clientHeight+15,306)+'px'; x[i].style.resize = "vertical";}};} ---unit PowThread; { MERCVRHYO } {$mode objfpc}{$H+} interface uses Classes, SysUtils, contnrs, syncobjs; type TPowerThread = class; PThreadCmd = ^TThreadCmd; TPowCallback = procedure(ACmd: PThreadCmd; Sender: TPowerThread) of object; TThreadCmd = record msg: shortstring; param: longint; extra: Pointer; Done: boolean; // always true when TPowMethod finished wait: boolean; // false = queue callback - true = sync callback callback: TPowCallBack; // called in main thread, after method finished end; TPowMethod = procedure(ACmd: TThreadCmd) of object; { TPowCmdQueue } TPowCmdQueue = class(TQueue) private FLocker: TCriticalSection; FNotBusy: TEvent; FEvName: string; protected procedure ClearAll; virtual; procedure PushItem(AItem: Pointer); override; function PopItem: Pointer; override; function PeekItem: Pointer; override; public constructor Create; destructor Destroy; override; end; { TPowerThread } TPowerThread = class(TThread) private type PInternalCmd = ^TInternalCmd; TInternalCmd = record method: TPowMethod; rec: TThreadCmd; end; private var FCmdQ: TPowCmdQueue; FCBCmd: TThreadCmd; FCallBack: TPowCallback; protected procedure DoCallBack; // redefine DefHandler in derived classes as needed // DefHandler is called by PostMsg(nil, aShortString) procedure DefHandler(aCmd: TThreadCmd); virtual; public procedure Execute; override; constructor Create(CreateSuspended: boolean; const StackSize: SizeUInt = DefaultStackSize); destructor Destroy; override; // usr procs called from outside the thread procedure PostCmd(aMethod: TPowMethod; aCmd: PThreadCmd = nil); procedure PostMsg(aMethod: TPowMethod; aMsg: shortstring); // add here your TPowMethod and TPowCallback // or inherit class(TPowerThread) and add them // to public section // TPowMethods will be called inside the thread while // TPowCallback will be queued or synced inside the main thread end; var MainThreadID: TThreadID; implementation var EvNameFmt: string; EvRef: smallint; { TPowCmdQueue } procedure TPowCmdQueue.ClearAll; // in main thread: no need lockbegin while not (TQueue(Self).Peek = nil) do Dispose(PThreadCmd(TQueue(Self).Pop));end; procedure TPowCmdQueue.PushItem(AItem: Pointer);var c: TEvent;begin FLocker.Enter; c := TEvent.Create(nil, True, True, FEvName); if c.WaitFor(10) <> wrSignaled then begin c.Free; FLocker.Leave; raise ESyncObjectException.Create('Thread Cmdqueue error'); end; c.Free; FNotBusy.ResetEvent; inherited PushItem(AItem); FNotBusy.SetEvent; FLocker.Leave;end; function TPowCmdQueue.PopItem: Pointer;var c: TEvent;begin FLocker.Enter; c := TEvent.Create(nil, True, True, FEvName); if c.WaitFor(10) <> wrSignaled then begin c.Free; FLocker.Leave; raise ESyncObjectException.Create('Thread Cmdqueue error'); end; c.Free; FNotBusy.ResetEvent; Result := inherited PopItem; FNotBusy.SetEvent; FLocker.Leave;end; function TPowCmdQueue.PeekItem: Pointer;var c: TEvent;begin FLocker.Enter; c := TEvent.Create(nil, True, True, FEvName); if c.WaitFor(10) <> wrSignaled then begin c.Free; FLocker.Leave; raise ESyncObjectException.Create('Thread Cmdqueue error'); end; c.Free; FNotBusy.ResetEvent; Result := inherited PeekItem; FNotBusy.SetEvent; FLocker.Leave;end; constructor TPowCmdQueue.Create;begin FLocker := TCriticalSection.Create; Inc(EvRef); FEvName := Format(EvNameFmt, [EvRef]); FNotBusy := TEvent.Create(nil, True, True, FEvName); inherited;end; destructor TPowCmdQueue.Destroy;begin FNotBusy.Free; FLocker.Free; inherited Destroy;end; { TPowerThread } procedure TPowerThread.DoCallBack;begin FCallBack(@FCBCmd,Self);end; procedure TPowerThread.DefHandler(aCmd: TThreadCmd);begin aCmd.Done := True;end; procedure TPowerThread.Execute;var cmd: PInternalCmd;begin while not Terminated do begin if FCmdQ.Count > 0 then begin cmd := FCmdQ.Pop; if not Assigned(cmd^.method) then cmd^.method := @DefHandler; cmd^.method(cmd^.rec); cmd^.rec.Done := True; if Assigned(cmd^.rec.callback) then begin FCBCmd := cmd^.rec; FCallBack:=cmd^.rec.callback; if cmd^.rec.wait then Synchronize(@DoCallBack) else Queue(@DoCallBack); end; Dispose(cmd); end; Sleep(1); end;end; constructor TPowerThread.Create(CreateSuspended: boolean; const StackSize: SizeUInt);begin FCmdQ := TPowCmdQueue.Create; inherited Create(CreateSuspended, StackSize);end; destructor TPowerThread.Destroy;begin FCmdQ.ClearAll; FCmdQ.Free; inherited Destroy;end; procedure TPowerThread.PostCmd(aMethod: TPowMethod; aCmd: PThreadCmd);var c: PInternalCmd;begin if aCmd = nil then exit; new(c); FillChar(c, SizeOf(TInternalCmd), #0); c^.method := aMethod; if Assigned(aCmd) then c^.rec := aCmd^; FCmdQ.Push(c);end; procedure TPowerThread.PostMsg(aMethod: TPowMethod; aMsg: shortstring);var c: PInternalCmd;begin new(c); FillChar(c, SizeOf(TInternalCmd), #0); c^.method := aMethod; c^.rec.msg := aMsg; FCmdQ.Push(c);end; initialization EvNameFmt := 'EvPowThread_%.5d'; EvRef := 0; MainThreadID := TThread.CurrentThread.ThreadID;end.
mercurhyo:
I am not happy with WThread package, so I am working on an alternative (I hope crossplatform)
Faster (no variant craps)
shorter If possible
mercurhyo:
we'll see :D
mercurhyo:
1st post modified ->> DoCallBack done
mercurhyo:
Well, ... after some debugging and little corrections, I can put there a working PowerThread
--- Code: Pascal [+][-]window.onload = function(){var x1 = document.getElementById("main_content_section"); if (x1) { var x = document.getElementsByClassName("geshi");for (var i = 0; i < x.length; i++) { x[i].style.maxHeight='none'; x[i].style.height = Math.min(x[i].clientHeight+15,306)+'px'; x[i].style.resize = "vertical";}};} ---unit PowThread; { MERCVRHYO }{ version 0.3} {$mode objfpc}{$H+} interface uses Classes, SysUtils, contnrs, syncobjs; type TPowerThread = class; PThreadCmd = ^TThreadCmd; TPowCallback = procedure(ACmd: PThreadCmd; Sender: TPowerThread) of object; TThreadCmd = record msg: shortstring; param: longint; extra: Pointer; Done: boolean; // always true when TPowMethod finished wait: boolean; // false = queue callback - true = sync callback callback: TPowCallBack; // called in main thread, after method finished end; TPowMethod = procedure(ACmd: TThreadCmd) of object; { TPowCmdQueue } TPowCmdQueue = class(TQueue) private FLocker: TCriticalSection; FNotBusy: TEvent; FEvName: string; protected procedure ClearAll; virtual; procedure PushItem(AItem: Pointer); override; function PopItem: Pointer; override; function PeekItem: Pointer; override; public constructor Create; destructor Destroy; override; end; { TPowerThread } TPowerThread = class(TThread) private type PInternalCmd = ^TInternalCmd; TInternalCmd = record method: TPowMethod; rec: PThreadCmd; end; private var FCmdQ, FcbQ: TPowCmdQueue; FSyncCmd: PThreadCmd; protected procedure DoSyncCallBack; procedure DoAsyncCallBack; // redefine DefHandler in derived classes as needed // called when PostCmd(nil,aCmd) procedure DefHandler(aCmd: TThreadCmd); virtual; public procedure Execute; override; constructor Create(CreateSuspended: boolean; const StackSize: SizeUInt = DefaultStackSize); destructor Destroy; override; // usr proc called from outside the thread procedure PostCmd(aMethod: TPowMethod; var aCmd: TThreadCmd); // add here your TPowMethod and TPowCallback // or inherit class(TPowerThread) and add them // to public section // TPowMethods will be called inside the thread while // TPowCallback will be queued or synced inside the main thread end; var MainThreadID: TThreadID; implementation var EvNameFmt: string; EvRef: smallint; { TPowCmdQueue } procedure TPowCmdQueue.ClearAll; // in main thread: no need lockbegin while not (TQueue(Self).Peek = nil) do Dispose(PThreadCmd(TQueue(Self).Pop));end; procedure TPowCmdQueue.PushItem(AItem: Pointer);var c: TEvent;begin FLocker.Enter; c := TEvent.Create(nil, True, True, FEvName); if c.WaitFor(10) <> wrSignaled then begin c.Free; FLocker.Leave; raise ESyncObjectException.Create('Thread Cmdqueue error'); end; c.Free; FNotBusy.ResetEvent; inherited; FNotBusy.SetEvent; FLocker.Leave;end; function TPowCmdQueue.PopItem: Pointer;var c: TEvent;begin FLocker.Enter; c := TEvent.Create(nil, True, True, FEvName); if c.WaitFor(10) <> wrSignaled then begin c.Free; FLocker.Leave; raise ESyncObjectException.Create('Thread Cmdqueue error'); end; c.Free; FNotBusy.ResetEvent; Result := inherited; FNotBusy.SetEvent; FLocker.Leave;end; function TPowCmdQueue.PeekItem: Pointer;var c: TEvent;begin FLocker.Enter; c := TEvent.Create(nil, True, True, FEvName); if c.WaitFor(10) <> wrSignaled then begin c.Free; FLocker.Leave; raise ESyncObjectException.Create('Thread Cmdqueue error'); end; c.Free; FNotBusy.ResetEvent; Result := inherited; FNotBusy.SetEvent; FLocker.Leave;end; constructor TPowCmdQueue.Create;begin FLocker := TCriticalSection.Create; Inc(EvRef); FEvName := Format(EvNameFmt, [EvRef]); FNotBusy := TEvent.Create(nil, True, True, FEvName); inherited;end; destructor TPowCmdQueue.Destroy;begin FNotBusy.Free; FLocker.Free; inherited;end; { TPowerThread } procedure TPowerThread.DoSyncCallBack;begin FSyncCmd^.callback(FSyncCmd, Self);end; procedure TPowerThread.DoAsyncCallBack;var c: TThreadCmd;begin if FcbQ.Count > 0 then begin c := PThreadCmd(FcbQ.Pop)^; c.callback(@c, Self); end;end; procedure TPowerThread.DefHandler(aCmd: TThreadCmd);begin // nothing to doend; procedure TPowerThread.Execute;var cmd: PInternalCmd;begin while not Terminated do begin if FCmdQ.Count > 0 then begin cmd := FCmdQ.Pop; if not Assigned(cmd^.method) then cmd^.method := @DefHandler; cmd^.method(cmd^.rec^); cmd^.rec^.Done := True; if Assigned(cmd^.rec^.callback) then begin if cmd^.rec^.wait then begin New(FSyncCmd); FSyncCmd^ := cmd^.rec^; Synchronize(@DoSyncCallBack); Dispose(FSyncCmd); end else FcbQ.Push(@cmd^.rec); end; while FcbQ.Count > 0 do Queue(@DoAsyncCallBack); Dispose(cmd); end; Sleep(1); end;end; constructor TPowerThread.Create(CreateSuspended: boolean; const StackSize: SizeUInt);begin FCmdQ := TPowCmdQueue.Create; FcbQ := TPowCmdQueue.Create; inherited;end; destructor TPowerThread.Destroy;begin FCmdQ.ClearAll; FCmdQ.Free; FcbQ.Free; inherited;end; procedure TPowerThread.PostCmd(aMethod: TPowMethod; var aCmd: TThreadCmd);var c: PInternalCmd;begin new(c); FillChar(c^, SizeOf(TInternalCmd), #0); c^.method := aMethod; c^.rec:=@aCmd; FCmdQ.Push(c);end; initialization EvNameFmt := 'EvPowThread_%.5d'; EvRef := 0; MainThreadID := TThread.CurrentThread.ThreadID;end.
I started testing as follow, with synapse
--- Code: Pascal [+][-]window.onload = function(){var x1 = document.getElementById("main_content_section"); if (x1) { var x = document.getElementsByClassName("geshi");for (var i = 0; i < x.length; i++) { x[i].style.maxHeight='none'; x[i].style.height = Math.min(x[i].clientHeight+15,306)+'px'; x[i].style.resize = "vertical";}};} ---unit TestClient; {$mode objfpc}{$H+} interface uses Classes, SysUtils, blcksock, PowThread; const STK_SZ = $10000; type { TClientThread } TClientThread = class(TPowerThread) private FSock: TTCPBlockSocket; public // this will be general Handler parsing aCmd inside the worker thread procedure Handler(aCmd: TThreadCmd); end; { TPlug } TPlug = class(TComponent) private var FWorker: TClientThread; public constructor Create(AOwner: TComponent); override; destructor Destroy; override; end; implementation { TClientThread } procedure TClientThread.Handler(aCmd: TThreadCmd);begin // todo interpretation of aCmdend; { TPlug } constructor TPlug.Create(AOwner: TComponent);var cmd: TThreadCmd;begin inherited Create(AOwner); FWorker := TClientThread.Create(False, STK_SZ); // // testing // Sleep(10); FillChar(cmd,SizeOf(TThreadCmd),#0); cmd.msg:='testing';// POST JOB TO DO FWorker.PostCmd(@FWorker.Handler,cmd); // WAITFOR job done while not cmd.Done do Sleep(1);end; destructor TPlug.Destroy;begin FWorker.Free; inherited Destroy;end; end.
I used GDB with breakpoints inside the Worker thread and around the calling, all went good so far
Navigation
[0] Message Index
[#] Next page