Brummel/MycLib
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

TaskManager.RunTask() & TFuture.Chain(Proc)

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-24 20:17:04 +02:00
parent a65a5f2b0a
commit 13c41d01b5
8 changed files with 120 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Core.Futures.pas
+16 -19
View File
@@ -25,14 +25,13 @@ type
TMycGateFuncFuture<T> = class(TMycFuture<T>)
private
FInit: TSignal.TSubscription;
FDone: TLatch.ILatch;
FResult: T;
protected
function GetValue: T; override;
function GetDone: TState; override;
public
constructor Create(const ATaskManager: IMycTaskManager; const AGate: TState.IState; AProc: TFunc<T>);
constructor Create(const ATaskManager: TTaskManager; const AGate: TState.IState; AProc: TFunc<T>);
destructor Destroy; override;
end;
@@ -71,7 +70,7 @@ end;
{ TMycGateFuncFuture<T> }
constructor TMycGateFuncFuture<T>.Create(const ATaskManager: IMycTaskManager; const AGate: TState.IState; AProc: TFunc<T>);
constructor TMycGateFuncFuture<T>.Create(const ATaskManager: TTaskManager; const AGate: TState.IState; AProc: TFunc<T>);
begin
inherited Create;
@@ -79,28 +78,26 @@ begin
// Subscribe the job execution to AGate.
// The job will run when AGate notifies the subscriber returned by Run.
FInit :=
ATaskManager.CreateTask(
AGate,
procedure
begin
ATaskManager.RunTask(
AGate,
procedure
begin
try
try
try
Self.FResult := AProc();
except
Self.FResult := Default(T); // Set result to Default(T) on error
raise; // Re-raise for TaskFactory to handle
end;
finally
Self.FDone.Notify; // Signal that this future is done (successfully or with error)
Self.FResult := AProc();
except
Self.FResult := Default(T); // Set result to Default(T) on error
raise; // Re-raise for TaskFactory to handle
end;
end
);
finally
Self.FDone.Notify; // Signal that this future is done (successfully or with error)
end;
end
);
end;
destructor TMycGateFuncFuture<T>.Destroy;
begin
FInit.Unsubscribe;
inherited Destroy;
end;
Src/Myc.Core.Tasks.pas
+14 -28
View File
@@ -11,7 +11,7 @@ uses
Myc.Signals;
type
IMycTaskFactory = interface(IMycTaskManager)
IMycThreadPool = interface(TTaskManager.IMycTaskManager)
{$region 'property access'}
// Retrieves the number of worker threads.
function GetThreadCount: Integer;
@@ -49,7 +49,7 @@ type
property ThreadCount: Integer read GetThreadCount;
end;
TMycTaskFactory = class(TInterfacedObject, IMycTaskManager, IMycTaskFactory)
TMycTaskFactory = class(TInterfacedObject, TTaskManager.IMycTaskManager, IMycThreadPool)
type
ETaskException = class(Exception)
end;
@@ -65,9 +65,7 @@ type
FWorkStack: TMycAtomicStack<TProc>; // Stack of pending jobs
FWorkThreads: TArray<TThread>; // Array of worker threads
procedure WorkerThread;
class var
FTaskManagerLock: Integer;
class constructor CreateClass;
protected
procedure ExecuteJob;
function GetThreadCount: Integer;
@@ -81,16 +79,12 @@ type
procedure HandleException;
procedure EnqueueJob(const Job: TProc);
function Run(Job: TProc): TSignal.ISubscriber;
function CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
procedure WaitFor(const State: TState.IState);
procedure RunTask(const Gate: TState; const Proc: TProc);
procedure WaitFor(const State: TState);
procedure Teardown;
function InMainThread: Boolean;
function InWorkerThread: Boolean;
// Initialize global TaskManager
class procedure AquireTaskManager;
class procedure ReleaseTaskManager;
property ThreadsRunning: Integer read FThreadsRunning;
property WorkThreads: TArray<TThread> read FWorkThreads;
end;
@@ -147,6 +141,10 @@ begin
FWaitSemaphores.Push(TSemaphore.Create(nil, 0, 1, ''));
end;
class constructor TMycTaskFactory.CreateClass;
begin
end;
destructor TMycTaskFactory.Destroy;
begin
Teardown; // Perform cleanup and stop threads
@@ -188,7 +186,7 @@ begin
Result.NameThreadForDebugging(DbgName); // Set thread name for debugging
end;
function TMycTaskFactory.CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
procedure TMycTaskFactory.RunTask(const Gate: TState; const Proc: TProc);
begin
if Gate.IsSet then
begin
@@ -197,7 +195,7 @@ begin
else
begin
// The job will run when Gate notifies the subscriber returned by Run.
Result := Gate.Signal.Subscribe(Run(Proc));
Gate.Signal.Subscribe(Run(Proc));
end;
end;
@@ -289,20 +287,6 @@ begin
exit(false); // Not found in worker thread list
end;
class procedure TMycTaskFactory.AquireTaskManager;
begin
if not Assigned(TaskManager) then
TaskManager := TMycTaskFactory.Create;
inc(FTaskManagerLock);
end;
class procedure TMycTaskFactory.ReleaseTaskManager;
begin
dec(FTaskManagerLock);
if FTaskManagerLock = 0 then
TaskManager := nil;
end;
function TMycTaskFactory.Run(Job: TProc): TSignal.ISubscriber;
begin
if FTerminated <> 0 then
@@ -345,7 +329,7 @@ begin
end;
end;
procedure TMycTaskFactory.WaitFor(const State: TState.IState);
procedure TMycTaskFactory.WaitFor(const State: TState);
var
lock: TSemaphore; // This is System.SyncObjs.TSemaphore
begin
@@ -431,4 +415,6 @@ end;
initialization
IsMultiThread := true;
TaskManager := TMycTaskFactory.Create;
end.
Src/Myc.Futures.pas
+9 -7
View File
@@ -23,6 +23,7 @@ type
end;
TFuncConst<S, TResult> = reference to function(const Arg1: S): TResult;
TProcConst<S, TResult> = reference to function(const Arg1: S): TResult;
{$REGION 'private'}
strict private
@@ -30,8 +31,6 @@ type
FNull: IFuture;
class constructor CreateClass;
class destructor DestroyClass;
private
FFuture: IFuture;
function GetDone: TState.IState; inline;
@@ -53,6 +52,7 @@ type
function Chain<S>(const Proc: TFunc<T, S>): TFuture<S>; overload;
function Chain<S>(const Proc: TFuncConst<T, S>): TFuture<S>; overload;
function Chain(const Proc: TProcConst<T, TState>): TState; overload;
function WaitFor: T;
property Done: TState.IState read GetDone;
@@ -62,8 +62,7 @@ type
implementation
uses
Myc.Core.Futures,
Myc.Core.Tasks;
Myc.Core.Futures;
constructor TFuture<T>.Create(const AFuture: IFuture);
begin
@@ -74,13 +73,16 @@ end;
class constructor TFuture<T>.CreateClass;
begin
TMycTaskFactory.AquireTaskManager;
FNull := TMycNullFuture<T>.Create;
end;
class destructor TFuture<T>.DestroyClass;
function TFuture<T>.Chain(const Proc: TProcConst<T, TState>): TState;
begin
TMycTaskFactory.ReleaseTaskManager;
var Done := TLatch.CreateLatch(1);
Result := Done.State;
var future := FFuture;
TaskManager.RunTask(future.Done, procedure begin Proc(future.Value).Subscribe(Done); end);
end;
function TFuture<T>.Chain<S>(const Proc: TFunc<T, S>): TFuture<S>;
Src/Myc.Signals.pas
+15 -10
View File
@@ -233,8 +233,9 @@ end;
constructor TState.Create(const AState: IState);
begin
if Assigned(AState) then
FState := AState;
FState := AState;
if not Assigned(FState) then
FState := Null;
end;
class constructor TState.ClassCreate;
@@ -304,8 +305,9 @@ end;
constructor TEvent.Create(const AEvent: IEvent);
begin
if Assigned(AEvent) then
FEvent := AEvent;
FEvent := AEvent;
if not Assigned(FEvent) then
FEvent := Null;
end;
class constructor TEvent.ClassCreate;
@@ -357,8 +359,9 @@ end;
constructor TFlag.Create(const AFlag: IFlag);
begin
if Assigned(AFlag) then
FFlag := AFlag;
FFlag := AFlag;
if not Assigned(FFlag) then
FFlag := Null;
end;
class function TFlag.CreateFlag(Init: Boolean = false): TFlag;
@@ -412,8 +415,9 @@ end;
constructor TLatch.Create(const ALatch: TLatch.ILatch);
begin
if Assigned(ALatch) then
FLatch := ALatch;
FLatch := ALatch;
if not Assigned(FLatch) then
FLatch := Null;
end;
class function TLatch.CreateLatch(Count: Integer): ILatch;
@@ -500,8 +504,9 @@ end;
constructor TSignal.Create(const ASignal: ISignal);
begin
if Assigned(FSignal) then
FSignal := ASignal;
FSignal := ASignal;
if not Assigned(FSignal) then
FSignal := Null;
end;
function TSignal.Subscribe(Subscriber: ISubscriber): TSubscription;
Src/Myc.TaskManager.pas
+52 -13
View File
@@ -7,27 +7,41 @@ uses
Myc.Signals;
type
IMycTaskManager = interface
// TOD Dokumentation
function CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
TTaskManager = record
type
IMycTaskManager = interface
procedure RunTask(const Gate: TState; const Proc: TProc);
procedure WaitFor(const State: TState);
end;
private
FTaskManager: IMycTaskManager;
public
constructor Create(const ATaskManager: IMycTaskManager);
class operator Implicit(const A: IMycTaskManager): TTaskManager; overload;
class operator Implicit(const A: TTaskManager): IMycTaskManager; overload;
// Run a task when the gate is opened.
procedure RunTask(const Gate: TState; const Proc: TProc); inline;
// Waits for the operation associated with State to complete.
// Must not be called from a worker thread of this factory.
// After waiting, or if the state is already set, any first stored exception
// from any worker thread of this factory will be re-raised in the calling (main) thread.
// Raises ETaskException if called from within a worker thread.
procedure WaitFor(const State: TState.IState);
procedure WaitFor(const State: TState); inline;
end;
var
TaskManager: IMycTaskManager;
TaskManager: TTaskManager;
procedure SetupTaskManagerMock;
implementation
uses
System.Generics.Collections;
System.Generics.Collections,
Myc.Core.Tasks;
type
TMycExecMock = class(TInterfacedObject, TSignal.ISubscriber)
@@ -38,11 +52,11 @@ type
function Notify: Boolean;
end;
TMycTaskManagerMock = class(TInterfacedObject, IMycTaskManager)
TMycTaskManagerMock = class(TInterfacedObject, TTaskManager.IMycTaskManager)
public
// IMycTaskManager
function CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
procedure WaitFor(const State: TState.IState);
procedure RunTask(const Gate: TState; const Proc: TProc);
procedure WaitFor(const State: TState);
constructor Create;
end;
@@ -72,19 +86,44 @@ begin
inherited Create;
end;
function TMycTaskManagerMock.CreateTask(const Gate: TState; const Proc: TProc): TSignal.TSubscription;
procedure TMycTaskManagerMock.RunTask(const Gate: TState; const Proc: TProc);
begin
Result := Gate.Signal.Subscribe(TMycExecMock.Create(Proc));
Gate.Signal.Subscribe(TMycExecMock.Create(Proc));
end;
procedure TMycTaskManagerMock.WaitFor(const State: TState.IState);
procedure TMycTaskManagerMock.WaitFor(const State: TState);
begin
Assert(State.IsSet);
end;
procedure SetupTaskManagerMock;
begin
TaskManager := TMycTaskManagerMock.Create;
TaskManager.Create(TMycTaskManagerMock.Create);
end;
constructor TTaskManager.Create(const ATaskManager: IMycTaskManager);
begin
FTaskManager := ATaskManager;
end;
procedure TTaskManager.RunTask(const Gate: TState; const Proc: TProc);
begin
FTaskManager.RunTask(Gate, Proc);
end;
procedure TTaskManager.WaitFor(const State: TState);
begin
FTaskManager.WaitFor(State);
end;
class operator TTaskManager.Implicit(const A: IMycTaskManager): TTaskManager;
begin
Result.Create(A);
end;
class operator TTaskManager.Implicit(const A: TTaskManager): IMycTaskManager;
begin
Result := A.FTaskManager;
end;
end.
Src/Myc.Trade.DataStream.pas
+11 -85
View File
@@ -172,29 +172,6 @@ type
function ParseFileName(const FileName: string): TAuraDataFile; override;
end;
// Implements a data stream that reads from Aura-specific historical data files.
TAuraFileLoader<T: record> = class(TInterfacedObject)
type
TDataProc = reference to procedure(const Values: TArray<TDataPoint<T>>);
private
class procedure LoadFile(
DataServer: TAuraDataServer<T>;
currFile: TAuraDataFile;
Terminated: TState;
Done: TLatch;
Proc: TDataProc
);
public
class function LoadData(
DataServer: TAuraDataServer<T>;
const Symbol: String;
const Terminate: TSignal;
const Proc: TDataProc
): TState;
end;
implementation
uses
@@ -413,12 +390,9 @@ begin
var capProc := Proc;
var terminated := TFlag.CreateObserver(Terminate).State;
var firstFile := FindFirstFile(Symbol);
var done := TLatch.CreateLatch(1);
Result := done.State;
TaskManager.CreateTask(firstFile.Done, procedure begin LoadFile(firstFile.Value, terminated, capProc).Subscribe(done); end);
Result :=
FindFirstFile(Symbol)
.Chain(function(const FirstFile: TAuraDataFile): TState begin Result := LoadFile(FirstFile, terminated, capProc); end);
end;
function TAuraDataServer<T>.LoadDataFile(const DataFile: TAuraDataFile): TFuture<TArray<TDataPoint<T>>>;
@@ -431,19 +405,16 @@ begin
if not currFile.IsValid or Terminated.IsSet then
exit(TState.Null);
var done := TLatch.CreateLatch(1);
Result := done.State;
var data := FCachedFiles.GetOrAdd(currFile.GetFullFileName);
TaskManager.CreateTask(
data.Done,
procedure
begin
Proc(data.Value, Terminated);
LoadFile(currFile.GetNextFile, Terminated, Proc).Subscribe(done);
end
);
Result :=
data.Chain(
function(const Data: TArray<TDataPoint<T>>): TState
begin
Proc(Data, Terminated);
Result := LoadFile(currFile.GetNextFile, Terminated, Proc);
end
);
end;
{ TAuraFileStream<T> }
@@ -792,49 +763,4 @@ begin
end;
end;
class function TAuraFileLoader<T>.LoadData(
DataServer: TAuraDataServer<T>;
const Symbol: String;
const Terminate: TSignal;
const Proc: TDataProc
): TState;
begin
var capProc := Proc;
var terminated := TFlag.CreateObserver(Terminate).State;
var firstFile := DataServer.FindFirstFile(Symbol);
var done := TLatch.CreateLatch(1);
TaskManager.CreateTask(firstFile.Done, procedure begin LoadFile(DataServer, firstFile.Value, terminated, done, capProc); end);
Result := done.State;
end;
class procedure TAuraFileLoader<T>.LoadFile(
DataServer: TAuraDataServer<T>;
currFile: TAuraDataFile;
Terminated: TState;
Done: TLatch;
Proc: TDataProc
);
begin
if not currFile.IsValid or Terminated.IsSet then
begin
Done.Notify;
exit;
end;
var data := DataServer.LoadDataFile(currFile);
TaskManager.CreateTask(
data.Done,
procedure
begin
Proc(data.Value);
LoadFile(DataServer, currFile.GetNextFile, Terminated, Done, Proc);
end
);
end;
end.
Test/TestCoreFutures.pas
+2 -2
View File
@@ -27,7 +27,7 @@ type
[IgnoreMemoryLeaks(true)]
TTestMycGateFuncFuture = class(TObject)
private
FTaskFactory: IMycTaskFactory;
FTaskFactory: IMycThreadPool;
FProcExecutionCount: Integer; // Counter for side effects of AProc
FSharedCounter: Integer; // For Fan-Out test side effects
public
@@ -156,7 +156,7 @@ end;
procedure TTestMycGateFuncFuture.Test_ExceptionInProc_HandledAsPlanned;
var
LFuture: TFuture<Integer>.IFuture;
LLocalTaskFactory: IMycTaskFactory;
LLocalTaskFactory: IMycThreadPool;
LInitStateAsState: TState.IState;
LResultValue: Integer;
LExpectedExceptionRaisedByFactory: Boolean;
Test/TestTasks.pas
+1 -1
View File
@@ -17,7 +17,7 @@ type
[IgnoreMemoryLeaks(true)]
TMycTaskFactoryTests = class(TObject)
private
FFactory: IMycTaskFactory;
FFactory: IMycThreadPool;
public
[Setup]
procedure Setup;