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

TaskManager refactoring

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-25 15:00:51 +02:00
parent e2a262bc5a
commit 10b653e16f
13 changed files with 183 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AuraTrader/AuraTrader.dproj
+1 -1
View File
@@ -4,7 +4,7 @@
<ProjectVersion>20.3</ProjectVersion>
<FrameworkType>FMX</FrameworkType>
<Base>True</Base>
<Config Condition="'$(Config)'==''">Release</Config>
<Config Condition="'$(Config)'==''">Debug</Config>
<Platform Condition="'$(Platform)'==''">Win64</Platform>
<ProjectName Condition="'$(ProjectName)'==''">AuraTrader</ProjectName>
<TargetedPlatforms>3</TargetedPlatforms>
AuraTrader/MainForm.fmx
+8 -8
View File
@@ -3,7 +3,7 @@ object Form1: TForm1
Top = 0
Caption = 'Form1'
ClientHeight = 840
ClientWidth = 797
ClientWidth = 796
FormFactor.Width = 320
FormFactor.Height = 480
FormFactor.Devices = [Desktop]
@@ -12,7 +12,7 @@ object Form1: TForm1
DesignerMasterStyle = 0
object Panel1: TPanel
Align = Top
Size.Width = 797.00000000000000000
Size.Width = 796.00000000000000000
Size.Height = 529.00000000000000000
Size.PlatformDefault = False
TabOrder = 2
@@ -28,13 +28,13 @@ object Form1: TForm1
Anchors = [akLeft, akTop, akRight]
Position.X = 8.00000000000000000
Position.Y = 8.00000000000000000
Size.Width = 262.00000000000000000
Size.Width = 261.00000000000000000
Size.Height = 513.00000000000000000
Size.PlatformDefault = False
end
object SymbolsComboBox: TComboBox
Anchors = [akTop, akRight]
Position.X = 494.00000000000000000
Position.X = 493.00000000000000000
Position.Y = 8.00000000000000000
Size.Width = 217.00000000000000000
Size.Height = 22.00000000000000000
@@ -43,7 +43,7 @@ object Form1: TForm1
end
object LoadButton: TButton
Anchors = [akTop, akRight]
Position.X = 719.00000000000000000
Position.X = 718.00000000000000000
Position.Y = 8.00000000000000000
Size.Width = 47.00000000000000000
Size.Height = 22.00000000000000000
@@ -85,11 +85,11 @@ object Form1: TForm1
Touch.InteractiveGestures = [Pan, LongTap, DoubleTap]
DataDetectorTypes = []
Align = Client
Size.Width = 797.00000000000000000
Size.Width = 796.00000000000000000
Size.Height = 311.00000000000000000
Size.PlatformDefault = False
TabOrder = 1
Viewport.Width = 793.00000000000000000
Viewport.Height = 307.00000000000000000
Viewport.Width = 796.00000000000000000
Viewport.Height = 311.00000000000000000
end
end
AuraTrader/MainForm.pas
+5 -4
View File
@@ -220,12 +220,12 @@ begin
const width = 1000;
const incsize = 100;
TaskManager.RunTask(
var done := TaskManager.RunTask(
nil,
procedure
function: TState
begin
var arr := TDataSeries<TAskBidItem>.CreateWriteable(width);
var done :=
Result :=
FServer.ProcessData(
Symbol,
FTerminate.Signal,
@@ -246,10 +246,11 @@ begin
currPathData.Value := TObjectRef<TPathData>.Create(PathData);
end
);
FLoadDone := TState.All([FLoadDone, done]);
end
);
FLoadDone := TState.All([FLoadDone, done]);
LogMemo.AddIdleHandler(currLog.Changed, procedure begin LogMemo.Lines.Add(Symbol + ': ' + currLog.Value); end);
Path1.AddIdleHandler(
Src/Myc.Core.Futures.pas
+10 -15
View File
@@ -25,7 +25,7 @@ type
TMycGateFuncFuture<T> = class(TMycFuture<T>)
private
FDone: TLatch.ILatch;
FDone: TState;
FResult: T;
protected
function GetValue: T; override;
@@ -74,26 +74,21 @@ constructor TMycGateFuncFuture<T>.Create(const ATaskManager: TTaskManager; const
begin
inherited Create;
FDone := TLatch.CreateLatch(1);
// Subscribe the job execution to AGate.
// The job will run when AGate notifies the subscriber returned by Run.
ATaskManager.RunTask(
AGate,
procedure
begin
try
FDone :=
ATaskManager.RunTask(
AGate,
function: TState
begin
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)
end;
end
);
end
);
end;
destructor TMycGateFuncFuture<T>.Destroy;
@@ -103,12 +98,12 @@ end;
function TMycGateFuncFuture<T>.GetDone: TState;
begin
Result := FDone.State;
Result := FDone;
end;
function TMycGateFuncFuture<T>.GetValue: T;
begin
Assert(FDone.State.IsSet, 'Result is not yet available.');
Assert(FDone.IsSet, 'Result is not yet available.');
Result := FResult;
end;
Src/Myc.Core.Tasks.pas
+16 -48
View File
@@ -11,37 +11,22 @@ uses
Myc.Signals;
type
IMycThreadPool = interface(TTaskManager.IMycTaskManager)
IThreadPool = interface(TTaskManager.ITaskManager)
{$region 'property access'}
// Retrieves the number of worker threads.
function GetThreadCount: Integer;
{$endregion}
// Creates and starts a new, independent thread executing Proc.
// Returns IMycState to await thread completion.
// Exceptions within Proc are caught; the first one is stored
// and can be re-raised in the main thread via WaitFor or Teardown.
function CreateThread(const Proc: TProc): TState.IState;
// Returns True if called from the main application thread.
function InMainThread: Boolean;
// Returns True if called from one of the factory's worker threads.
function InWorkerThread: Boolean;
// Schedules Job for execution by a worker thread.
// If StartCount > 0, Job is deferred until Notify is called StartCount times on the returned IMycSubscriber.
// If StartCount <= 0, Job is enqueued immediately.
// Returns IMycSubscriber for deferred jobs, or TMycLatch.Null for immediate jobs or if Job is nil.
// Exceptions during Job execution are caught; the first one is stored
// and can be re-raised in the main thread via WaitFor or Teardown.
// Raises ETaskException if the factory is already terminated.
function Run(Job: TProc): TSignal.ISubscriber;
procedure EnqueueJob(const Job: TProc);
// Shuts down the task factory: terminates worker threads and cleans up resources.
// Before shutdown, any first stored exception from a worker thread (from Run or CreateThread tasks managed by this factory)
// Before shutdown, any first stored exception from a worker thread (from CreateThread tasks managed by this factory)
// will be re-raised in the calling (main) thread.
procedure Teardown;
@@ -49,7 +34,7 @@ type
property ThreadCount: Integer read GetThreadCount;
end;
TMycTaskFactory = class(TInterfacedObject, TTaskManager.IMycTaskManager, IMycThreadPool)
TMycTaskFactory = class(TInterfacedObject, TTaskManager.ITaskManager, IThreadPool)
type
ETaskException = class(Exception)
end;
@@ -75,10 +60,9 @@ type
constructor Create;
destructor Destroy; override;
class function CreateAnonymousThread(const DbgName: String; const Proc: TProc): TThread;
function CreateThread(const Proc: TProc): TState.IState;
procedure CreateThread(const Proc: TProc);
procedure HandleException;
procedure EnqueueJob(const Job: TProc);
function Run(Job: TProc): TSignal.ISubscriber;
procedure RunTask(const Gate: TState; const Proc: TProc);
procedure WaitFor(const State: TState);
procedure Teardown;
@@ -188,41 +172,37 @@ end;
procedure TMycTaskFactory.RunTask(const Gate: TState; const Proc: TProc);
begin
if not Assigned(Proc) then
exit;
if FTerminated <> 0 then
raise ETaskException.Create('Task factory terminated');
if Gate.IsSet then
begin
EnqueueJob(Proc);
end
else
begin
// The job will run when Gate notifies the subscriber returned by Run.
Gate.Signal.Subscribe(Run(Proc));
// Create a pending job
Gate.Signal.Subscribe(TMycPendingJob.Create(Self, Proc));
end;
end;
function TMycTaskFactory.CreateThread(const Proc: TProc): TState.IState;
var
res: TLatch.ILatch;
capturedProc: TProc;
procedure TMycTaskFactory.CreateThread(const Proc: TProc);
begin
// Create a latch that will be signaled when the proc finishes
res := TLatch.CreateLatch(1);
capturedProc := Proc; // Capture Proc for the anonymous method
var cProc := Proc; // Capture Proc for the anonymous method
CreateAnonymousThread(
'Thread',
procedure
begin
try
capturedProc(); // Execute the provided procedure
cProc(); // Execute the provided procedure
finally
capturedProc := nil; // Clear the captured proc
res.Notify; // Signal completion via the latch's Notify method
cProc := nil; // Clear the captured proc
end;
end)
.Start;
// Return the state interface of the latch
exit(res.State);
end;
procedure TMycTaskFactory.EnqueueJob(const Job: TProc);
@@ -287,18 +267,6 @@ begin
exit(false); // Not found in worker thread list
end;
function TMycTaskFactory.Run(Job: TProc): TSignal.ISubscriber;
begin
if FTerminated <> 0 then
raise ETaskException.Create('Task factory terminated');
if not Assigned(Job) then
exit(TLatch.Null); // Already correct, uses direct static property on TMycLatch
// Create a pending job
Result := TMycPendingJob.Create(Self, Job);
end;
procedure TMycTaskFactory.Teardown;
var
i: Integer;
Src/Myc.Futures.pas
+8 -10
View File
@@ -44,7 +44,7 @@ type
class operator Implicit(const A: TFuture<T>): IFuture; overload;
class function Construct(const Proc: TFunc<T>): TFuture<T>; overload; static;
class function Construct(const Gate: TState.IState; const Proc: TFunc<T>): TFuture<T>; overload; static;
class function Construct(const Gate: TState; const Proc: TFunc<T>): TFuture<T>; overload; static;
class property Null: IFuture read FNull;
@@ -94,23 +94,21 @@ end;
function TFuture<T>.Chain(const Proc: TProcConst<T, TState>): TState;
begin
var Done := TLatch.CreateLatch(1);
Result := Done.State;
var cFuture := FFuture;
var future := FFuture;
TaskManager.RunTask(future.Done, procedure begin Proc(future.Value).Subscribe(Done); end);
Result := TaskManager.RunTask(cFuture.Done, function: TState begin Result := Proc(cFuture.Value); end);
end;
function TFuture<T>.Chain<S>(const Proc: TFunc<T, S>): TFuture<S>;
begin
var Cap := FFuture;
Result := TFuture<S>.Construct(FFuture.Done, function: S begin Result := Proc(Cap.Value); end);
var cFuture := FFuture;
Result := TFuture<S>.Construct(cFuture.Done, function: S begin Result := Proc(cFuture.Value); end);
end;
function TFuture<T>.Chain<S>(const Proc: TFuncConst<T, S>): TFuture<S>;
begin
var Cap := FFuture;
Result := TFuture<S>.Construct(FFuture.Done, function: S begin Result := Proc(Cap.Value); end);
var cFuture := FFuture;
Result := TFuture<S>.Construct(cFuture.Done, function: S begin Result := Proc(cFuture.Value); end);
end;
class function TFuture<T>.Construct(const Proc: TFunc<T>): TFuture<T>;
@@ -118,7 +116,7 @@ begin
Result := TMycGateFuncFuture<T>.Create(TaskManager, nil, Proc);
end;
class function TFuture<T>.Construct(const Gate: TState.IState; const Proc: TFunc<T>): TFuture<T>;
class function TFuture<T>.Construct(const Gate: TState; const Proc: TFunc<T>): TFuture<T>;
begin
Result := TMycGateFuncFuture<T>.Create(TaskManager, Gate, Proc);
end;
Src/Myc.Signals.pas
-7
View File
@@ -81,8 +81,6 @@ type
class function All(const States: TArray<TState>): TState; static;
class function Any(const States: TArray<TState>): TState; static;
function Subscribe(Subscriber: TSignal.ISubscriber): TSignal.TSubscription; inline;
class property Null: IState read FNull;
property IsSet: Boolean read GetIsSet;
@@ -281,11 +279,6 @@ begin
Result := FState.IsSet;
end;
function TState.Subscribe(Subscriber: TSignal.ISubscriber): TSignal.TSubscription;
begin
Result := Signal.Subscribe(Subscriber);
end;
class operator TState.Implicit(const A: TState): IState;
begin
Result := A.FState;
Src/Myc.TaskManager.pas
+64 -17
View File
@@ -9,23 +9,29 @@ uses
type
TTaskManager = record
type
IMycTaskManager = interface
ITaskManager = interface
procedure CreateThread(const Proc: TProc);
procedure RunTask(const Gate: TState; const Proc: TProc);
procedure WaitFor(const State: TState);
end;
private
FTaskManager: IMycTaskManager;
FTaskManager: ITaskManager;
public
constructor Create(const ATaskManager: IMycTaskManager);
class operator Implicit(const A: IMycTaskManager): TTaskManager; overload;
class operator Implicit(const A: TTaskManager): IMycTaskManager; overload;
constructor Create(const ATaskManager: ITaskManager);
class operator Implicit(const A: ITaskManager): TTaskManager; overload;
class operator Implicit(const A: TTaskManager): ITaskManager; overload;
// Creates and starts a new, independent thread executing Proc.
// Returns a State to await thread completion.
function CreateThread(const Proc: TProc): TState;
// Run a task when the gate is opened.
procedure RunTask(const Gate: TState; const Proc: TProc); inline;
// Returns a State to await thread completion.
function RunTask(const Gate: TState; const Proc: TFunc<TState>): TState;
// Waits for the operation associated with State to complete.
// Must not be called from a worker thread of this factory.
// Must not be called from a task 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.
@@ -40,7 +46,6 @@ procedure SetupTaskManagerMock;
implementation
uses
System.Generics.Collections,
Myc.Core.Tasks;
type
@@ -52,13 +57,12 @@ type
function Notify: Boolean;
end;
TMycTaskManagerMock = class(TInterfacedObject, TTaskManager.IMycTaskManager)
TMycTaskManagerMock = class(TInterfacedObject, TTaskManager.ITaskManager)
public
// IMycTaskManager
constructor Create;
procedure RunTask(const Gate: TState; const Proc: TProc);
procedure WaitFor(const State: TState);
constructor Create;
procedure CreateThread(const Proc: TProc);
end;
{ TMycExecMock }
@@ -86,6 +90,11 @@ begin
inherited Create;
end;
procedure TMycTaskManagerMock.CreateThread(const Proc: TProc);
begin
Proc();
end;
procedure TMycTaskManagerMock.RunTask(const Gate: TState; const Proc: TProc);
begin
Gate.Signal.Subscribe(TMycExecMock.Create(Proc));
@@ -101,14 +110,52 @@ begin
TaskManager.Create(TMycTaskManagerMock.Create);
end;
constructor TTaskManager.Create(const ATaskManager: IMycTaskManager);
constructor TTaskManager.Create(const ATaskManager: ITaskManager);
begin
FTaskManager := ATaskManager;
end;
procedure TTaskManager.RunTask(const Gate: TState; const Proc: TProc);
function TTaskManager.CreateThread(const Proc: TProc): TState;
begin
FTaskManager.RunTask(Gate, Proc);
// Create a latch that will be signaled when the proc finishes
var done := TLatch.CreateLatch(1);
Result := done.State;
var cProc := Proc; // Capture Proc for the anonymous method
FTaskManager.CreateThread(
procedure
begin
try
cProc(); // Execute the provided procedure
finally
cProc := nil; // Clear the captured proc
done.Notify; // Signal completion via the latch's Notify method
end;
end
);
end;
function TTaskManager.RunTask(const Gate: TState; const Proc: TFunc<TState>): TState;
begin
var cProc: TFunc<TState> := Proc;
if not Assigned(@cProc) then
exit(TState.Null);
var Done := TLatch.CreateLatch(1);
Result := Done.State;
FTaskManager.RunTask(
Gate,
procedure
begin
try
cProc.Signal.Subscribe(Done);
except
Done.Notify;
raise;
end;
end
);
end;
procedure TTaskManager.WaitFor(const State: TState);
@@ -116,12 +163,12 @@ begin
FTaskManager.WaitFor(State);
end;
class operator TTaskManager.Implicit(const A: IMycTaskManager): TTaskManager;
class operator TTaskManager.Implicit(const A: ITaskManager): TTaskManager;
begin
Result.Create(A);
end;
class operator TTaskManager.Implicit(const A: TTaskManager): IMycTaskManager;
class operator TTaskManager.Implicit(const A: TTaskManager): ITaskManager;
begin
Result := A.FTaskManager;
end;
Src/Myc.Test.Signals.Dirty.pas
+26 -26
View File
@@ -196,7 +196,7 @@ begin
dirtyFlag := CreateAndPrepareDirtyFlag(True); // Start dirty
subscriber := TMockSubscriber.Create;
subscription := dirtyFlag.State.Subscribe(subscriber);
subscription := dirtyFlag.State.Signal.Subscribe(subscriber);
Assert.AreEqual(1, subscriber.NotifyCount, 'Subscriber should be notified immediately when subscribing to an already dirty flag.');
Assert.IsTrue(subscriber.WasCalled, 'Subscriber WasCalled should be true.');
@@ -211,7 +211,7 @@ begin
dirtyFlag := CreateAndPrepareDirtyFlag(False); // Start clean
subscriber := TMockSubscriber.Create;
subscription := dirtyFlag.State.Subscribe(subscriber);
subscription := dirtyFlag.State.Signal.Subscribe(subscriber);
Assert.AreEqual(0, subscriber.NotifyCount, 'Subscriber should NOT be notified immediately when subscribing to a clean flag.');
Assert.IsFalse(subscriber.WasCalled, 'Subscriber WasCalled should be false.');
@@ -225,7 +225,7 @@ var
begin
dirtyFlag := CreateAndPrepareDirtyFlag(False); // Start clean
subscriber := TMockSubscriber.Create;
subscription := dirtyFlag.State.Subscribe(subscriber);
subscription := dirtyFlag.State.Signal.Subscribe(subscriber);
Assert.AreEqual(0, subscriber.NotifyCount, 'Subscriber initially not notified.');
@@ -242,7 +242,7 @@ var
begin
dirtyFlag := CreateAndPrepareDirtyFlag(True); // Start dirty
subscriber := TMockSubscriber.Create;
subscription := dirtyFlag.State.Subscribe(subscriber); // Gets initial notification
subscription := dirtyFlag.State.Signal.Subscribe(subscriber); // Gets initial notification
Assert.AreEqual(1, subscriber.NotifyCount, 'Subscriber notified on subscribe to dirty flag.');
@@ -259,7 +259,7 @@ var
begin
dirtyFlag := CreateAndPrepareDirtyFlag(True); // Start dirty
subscriber := TMockSubscriber.Create;
subscription := dirtyFlag.State.Subscribe(subscriber); // Gets initial notification
subscription := dirtyFlag.State.Signal.Subscribe(subscriber); // Gets initial notification
Assert.AreEqual(1, subscriber.NotifyCount, 'Subscriber notified on subscribe.');
dirtyFlag.Reset; // Reset the flag
@@ -278,8 +278,8 @@ begin
dirtyFlag := CreateAndPrepareDirtyFlag(False); // Start clean
sub1 := TMockSubscriber.Create;
sub2 := TMockSubscriber.Create;
subscription1 := dirtyFlag.State.Subscribe(sub1);
subscription2 := dirtyFlag.State.Subscribe(sub2);
subscription1 := dirtyFlag.State.Signal.Subscribe(sub1);
subscription2 := dirtyFlag.State.Signal.Subscribe(sub2);
dirtyFlag.Notify; // Make flag dirty
@@ -295,7 +295,7 @@ var
begin
dirtyFlag := CreateAndPrepareDirtyFlag(False); // Start clean
subscriber := TMockSubscriber.Create;
subscription := dirtyFlag.State.Subscribe(subscriber);
subscription := dirtyFlag.State.Signal.Subscribe(subscriber);
subscription.Unsubscribe; // Explicitly unsubscribe
@@ -318,8 +318,8 @@ begin
flagB := CreateAndPrepareDirtyFlag(False); // B starts clean
subToB := TMockSubscriber.Create;
subscriptionMock_B := flagB.State.Subscribe(subToB);
subscriptionA_B := flagA.State.Subscribe(flagB); // B subscribes to A's state changes (B's Notify will be called)
subscriptionMock_B := flagB.State.Signal.Subscribe(subToB);
subscriptionA_B := flagA.State.Signal.Subscribe(flagB); // B subscribes to A's state changes (B's Notify will be called)
flagA.Notify; // Make A dirty. This should trigger B.Notify
@@ -339,9 +339,9 @@ begin
flagC := CreateAndPrepareDirtyFlag(False);
subToC := TMockSubscriber.Create;
subscriptionMock_C := flagC.State.Subscribe(subToC);
subscriptionC_B := flagB.State.Subscribe(flagC); // C subscribes to B
subscriptionB_A := flagA.State.Subscribe(flagB); // B subscribes to A
subscriptionMock_C := flagC.State.Signal.Subscribe(subToC);
subscriptionC_B := flagB.State.Signal.Subscribe(flagC); // C subscribes to B
subscriptionB_A := flagA.State.Signal.Subscribe(flagB); // B subscribes to A
flagA.Notify; // Make A dirty
@@ -362,8 +362,8 @@ begin
flagA := CreateAndPrepareDirtyFlag(False);
flagB := CreateAndPrepareDirtyFlag(False);
subToB := TMockSubscriber.Create;
subscriptionMock_B := flagB.State.Subscribe(subToB);
subscriptionA_B := flagA.State.Subscribe(flagB);
subscriptionMock_B := flagB.State.Signal.Subscribe(subToB);
subscriptionA_B := flagA.State.Signal.Subscribe(flagB);
flagA.Notify; // A dirty -> B dirty. subToB notified.
Assert.IsTrue(flagB.State.IsSet, 'Flag B should be dirty initially.');
@@ -391,8 +391,8 @@ begin
flagA := CreateAndPrepareDirtyFlag(False);
flagB := CreateAndPrepareDirtyFlag(False);
subToB := TMockSubscriber.Create;
subscriptionMock_B := flagB.State.Subscribe(subToB);
subscriptionA_B := flagA.State.Subscribe(flagB);
subscriptionMock_B := flagB.State.Signal.Subscribe(subToB);
subscriptionA_B := flagA.State.Signal.Subscribe(flagB);
flagA.Notify; // A dirty -> B dirty. subToB notified.
Assert.IsTrue(flagB.State.IsSet, 'Flag B should be dirty after A triggers.');
@@ -422,9 +422,9 @@ begin
flagC := CreateAndPrepareDirtyFlag(False);
subToC := TMockSubscriber.Create;
subscriptionMock_C := flagC.State.Subscribe(subToC);
subscriptionC_B := flagB.State.Subscribe(flagC);
subscriptionB_A := flagA.State.Subscribe(flagB);
subscriptionMock_C := flagC.State.Signal.Subscribe(subToC);
subscriptionC_B := flagB.State.Signal.Subscribe(flagC);
subscriptionB_A := flagA.State.Signal.Subscribe(flagB);
// Initial full propagation
flagA.Notify; // A dirty -> B dirty -> C dirty. subToC notified.
@@ -461,9 +461,9 @@ begin
flagC := CreateAndPrepareDirtyFlag(False);
subToC := TMockSubscriber.Create;
subscriptionMock_C := flagC.State.Subscribe(subToC);
subscriptionC_B := flagB.State.Subscribe(flagC);
subscriptionB_A := flagA.State.Subscribe(flagB);
subscriptionMock_C := flagC.State.Signal.Subscribe(subToC);
subscriptionC_B := flagB.State.Signal.Subscribe(flagC);
subscriptionB_A := flagA.State.Signal.Subscribe(flagB);
// All are clean. Trigger A.
flagA.Notify; // A dirty -> B dirty -> C dirty. subToC notified.
@@ -501,9 +501,9 @@ begin
flagC := CreateAndPrepareDirtyFlag(False);
subToC := TMockSubscriber.Create;
subscriptionMock_C := flagC.State.Subscribe(subToC);
subscriptionC_B := flagB.State.Subscribe(flagC);
subscriptionB_A := flagA.State.Subscribe(flagB);
subscriptionMock_C := flagC.State.Signal.Subscribe(subToC);
subscriptionC_B := flagB.State.Signal.Subscribe(flagC);
subscriptionB_A := flagA.State.Signal.Subscribe(flagB);
// 1. Trigger A
flagA.Notify; // A, B, C become dirty. subToC notified (count = 1).
Src/Myc.Test.Signals.Latch.pas
+24 -23
View File
@@ -182,7 +182,7 @@ var
begin
latch := TLatch.CreateLatch(1);
mockSub := TMockSubscriber.Create;
subscription := latch.State.Subscribe(mockSub); // Subscribing to the Latch's state
subscription := latch.State.Signal.Subscribe(mockSub); // Subscribing to the Latch's state
Assert.IsFalse(latch.State.IsSet, 'Latch should initially not be set.');
Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should not have been notified yet (before latch set).');
@@ -204,9 +204,9 @@ begin
mockSub2 := TMockSubscriber.Create;
mockSub3 := TMockSubscriber.Create;
sub1 := latch.State.Subscribe(mockSub1);
sub2 := latch.State.Subscribe(mockSub2);
sub3 := latch.State.Subscribe(mockSub3);
sub1 := latch.State.Signal.Subscribe(mockSub1);
sub2 := latch.State.Signal.Subscribe(mockSub2);
sub3 := latch.State.Signal.Subscribe(mockSub3);
latch.Notify; // Latch becomes set
@@ -223,7 +223,7 @@ var
begin
latch := TLatch.CreateLatch(2); // Count = 2
mockSub := TMockSubscriber.Create;
subscription := latch.State.Subscribe(mockSub);
subscription := latch.State.Signal.Subscribe(mockSub);
Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should not be notified upon subscription if latch is not set.');
latch.Notify; // Count becomes 1, still not set
@@ -238,7 +238,7 @@ var
begin
latch := TLatch.CreateLatch(1);
mockSub := TMockSubscriber.Create;
subscription := latch.State.Subscribe(mockSub);
subscription := latch.State.Signal.Subscribe(mockSub);
latch.Notify; // Sets latch, notifies MockSub. Subscribers are then unadvised by UnadviseAll.
Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub notify count should be 1 after latch set.');
@@ -255,7 +255,7 @@ var
begin
latch := TLatch.CreateLatch(1); // Create with count 1
mockSub1 := TMockSubscriber.Create;
subscription1 := latch.State.Subscribe(mockSub1); // Subscribe before set
subscription1 := latch.State.Signal.Subscribe(mockSub1); // Subscribe before set
latch.Notify; // Latch becomes set. mockSub1 is notified.
@@ -264,7 +264,8 @@ begin
// Attempt to subscribe mockSub2 after the latch is already set.
mockSub2 := TMockSubscriber.Create;
subscription2 := latch.State.Subscribe(mockSub2); // TMycLatch.Subscribe (via TMycAbstractFlag) notifies immediately if already set.
subscription2 :=
latch.State.Signal.Subscribe(mockSub2); // TMycLatch.Subscribe (via TMycAbstractFlag) notifies immediately if already set.
Assert.AreEqual(1, mockSub2.NotifyCount, 'MockSub2 should be notified immediately upon subscribing to an already set latch.');
@@ -285,9 +286,9 @@ begin
latchB := TLatch.CreateLatch(1); // latchB is an TSignal.ISubscriber
mockSubForB := TMockSubscriber.Create;
subHandle_Mock_listens_B := latchB.State.Subscribe(mockSubForB);
subHandle_Mock_listens_B := latchB.State.Signal.Subscribe(mockSubForB);
// LatchA's state is an IMycSignal. LatchB (as TSignal.ISubscriber) subscribes to it.
subHandle_B_listens_A := latchA.State.Subscribe(latchB);
subHandle_B_listens_A := latchA.State.Signal.Subscribe(latchB);
Assert.IsFalse(latchA.State.IsSet, 'LatchA initially not set.');
Assert.IsFalse(latchB.State.IsSet, 'LatchB initially not set.');
@@ -311,9 +312,9 @@ begin
latchC := TLatch.CreateLatch(1);
mockSubForC := TMockSubscriber.Create;
sub_Mock_C := latchC.State.Subscribe(mockSubForC);
sub_C_B := latchB.State.Subscribe(latchC); // LatchC subscribes to LatchB's state
sub_B_A := latchA.State.Subscribe(latchB); // LatchB subscribes to LatchA's state
sub_Mock_C := latchC.State.Signal.Subscribe(mockSubForC);
sub_C_B := latchB.State.Signal.Subscribe(latchC); // LatchC subscribes to LatchB's state
sub_B_A := latchA.State.Signal.Subscribe(latchB); // LatchB subscribes to LatchA's state
latchA.Notify; // Call Notify on LatchA
@@ -335,8 +336,8 @@ begin
latchB := TLatch.CreateLatch(1);
mockSubForB := TMockSubscriber.Create;
sub_Mock_B := latchB.State.Subscribe(mockSubForB);
sub_B_A := latchA.State.Subscribe(latchB);
sub_Mock_B := latchB.State.Signal.Subscribe(mockSubForB);
sub_B_A := latchA.State.Signal.Subscribe(latchB);
latchA.Notify; // First notify for LatchA (count becomes 1)
Assert.IsFalse(latchA.State.IsSet, 'LatchA should not be set after first notify of two.');
@@ -361,7 +362,7 @@ begin
latch := TLatch.CreateLatch(0); // Returns TMycLatch.Null which is set.
mockSub := TMockSubscriber.Create;
subscription := latch.State.Subscribe(mockSub); // TMycLatch.Subscribe notifies immediately if already set.
subscription := latch.State.Signal.Subscribe(mockSub); // TMycLatch.Subscribe notifies immediately if already set.
Assert.IsTrue(latch.State.IsSet, 'Latch IsSet property mismatch (should be true for Null latch).');
Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub should be notified immediately upon subscribing to an initially set latch.');
@@ -376,7 +377,7 @@ begin
latch := TLatch.CreateLatch(1);
mockSub := TMockSubscriber.Create;
subscription := latch.State.Subscribe(mockSub);
subscription := latch.State.Signal.Subscribe(mockSub);
subscription.Unsubscribe; // Explicitly call Unsubscribe on the record
latch.Notify; // Latch becomes set
@@ -399,7 +400,7 @@ begin
Assert.IsFalse(latch.State.IsSet, 'Latch should not be set yet after partial notifies.');
mockSub := TMockSubscriber.Create;
subscription := latch.State.Subscribe(mockSub);
subscription := latch.State.Signal.Subscribe(mockSub);
Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should not be notified on subscribe if latch not yet set.');
latch.Notify; // Count -> 0, Latch becomes set and notifies mockSub
@@ -419,8 +420,8 @@ begin
mockSub1 := TMockSubscriber.Create;
mockSub2 := TMockSubscriber.Create;
subscription1 := latch.State.Subscribe(mockSub1);
subscription2 := latch.State.Subscribe(mockSub2);
subscription1 := latch.State.Signal.Subscribe(mockSub1);
subscription2 := latch.State.Signal.Subscribe(mockSub2);
Assert.AreEqual(0, mockSub1.NotifyCount, 'MockSub1 initial NotifyCount should be 0.');
Assert.AreEqual(0, mockSub2.NotifyCount, 'MockSub2 initial NotifyCount should be 0.');
@@ -450,7 +451,7 @@ begin
try
begin // Inner block to control lifetime of 'subscription'
var subscription: TSignal.TSubscription; // Local to this block
subscription := latch.State.Subscribe(mockSub);
subscription := latch.State.Signal.Subscribe(mockSub);
latch.Notify; // Sets the latch, mockSub is notified, UnadviseAll is called internally.
Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub was notified when latch set.');
@@ -481,7 +482,7 @@ var
begin
latch := TLatch.CreateLatch(1);
mockSubInitial := TMockSubscriber.Create;
subscriptionInitial := latch.State.Subscribe(mockSubInitial);
subscriptionInitial := latch.State.Signal.Subscribe(mockSubInitial);
// Set the latch; mockSubInitial is notified, UnadviseAll is called.
latch.Notify;
@@ -489,7 +490,7 @@ begin
// Subscribe a new subscriber to the already set latch
mockSubNew := TMockSubscriber.Create;
subscriptionNew := latch.State.Subscribe(mockSubNew);
subscriptionNew := latch.State.Signal.Subscribe(mockSubNew);
// New subscriber should be notified immediately upon subscription to an already set latch
Assert.AreEqual(1, mockSubNew.NotifyCount, 'New subscriber should be notified immediately upon subscription.');
Src/Myc.Test.Trade.DataPoint.pas
+16 -20
View File
@@ -13,7 +13,7 @@ type
[TestFixture]
TTestDataArray = class(TObject)
private
FArray: IDataArray<TAskBidItem>;
FArray: TDataSeries<TAskBidItem>;
procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0);
public
[Setup]
@@ -67,7 +67,6 @@ type
[TestFixture]
TTestDataSeries = class(TObject)
private
FArray: IDataArray<TAskBidItem>;
FSeries: TDataSeries<TAskBidItem>;
procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0);
public
@@ -111,7 +110,7 @@ implementation
procedure TTestDataArray.Setup;
begin
FArray := TDataSeries<TAskBidItem>.CreateArray(0);
FArray := TDataSeries<TAskBidItem>.CreateWriteable(0);
end;
procedure TTestDataArray.Teardown;
@@ -125,7 +124,7 @@ var
DataPoint: TDataPoint<TAskBidItem>;
baseTime: TDateTime;
begin
FArray := TDataSeries<TAskBidItem>.CreateArray(MaxLookBack);
FArray := TDataSeries<TAskBidItem>.CreateWriteable(MaxLookBack);
baseTime := EncodeDate(2020, 7, 7);
// Add data points with increasing timestamps.
@@ -410,14 +409,12 @@ end;
procedure TTestDataSeries.Setup;
begin
FArray := TDataSeries<TAskBidItem>.CreateArray(0);
FSeries := FArray;
FSeries := TDataSeries<TAskBidItem>.CreateWriteable(0);
end;
procedure TTestDataSeries.Teardown;
begin
FSeries := Default(TDataSeries<TAskBidItem>);
FArray := nil;
end;
procedure TTestDataSeries.SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0);
@@ -426,15 +423,14 @@ var
DataPoint: TDataPoint<TAskBidItem>;
baseTime: TDateTime;
begin
FArray := TDataSeries<TAskBidItem>.CreateArray(MaxLookBack);
FSeries := FArray;
FSeries := TDataSeries<TAskBidItem>.CreateWriteable(MaxLookBack);
baseTime := EncodeDate(2020, 7, 7);
// Add data points with increasing timestamps.
for i := 0 to ItemCount - 1 do
begin
DataPoint := TDataPoint<TAskBidItem>.Create(baseTime + i, TAskBidItem.Create(i * 1.0, i * 1.0 + 0.1));
FArray.Add(DataPoint);
FSeries.Add(DataPoint);
end;
end;
@@ -445,7 +441,7 @@ var
begin
SetupSeriesWithData(15);
copy := FSeries.Copy;
copy := FSeries.Immutable;
Assert.AreEqual(FSeries.Count, copy.Count, 'Copy must have the same count as the original');
Assert.AreEqual(FSeries.TotalCount, copy.TotalCount, 'Copy must have the same total count as the original');
@@ -467,21 +463,21 @@ begin
SetupSeriesWithData(10);
// Create the copy
copy := FSeries.Copy;
copy := FSeries.Immutable;
originalCount := copy.Count;
originalTotalCount := copy.TotalCount;
// Modify the original series
lastTime := FArray[0].Time;
lastTime := FSeries[0].Time;
newPoint := TDataPoint<TAskBidItem>.Create(lastTime + 1, TAskBidItem.Create(99, 99.1));
FArray.Add(newPoint);
FSeries.Add(newPoint);
Assert.AreEqual(Int64(11), FArray.Count, 'Original array count should have increased');
Assert.AreEqual(Int64(11), FSeries.Count, 'Original array count should have increased');
// Verify the copy remains unchanged
Assert.AreEqual(originalCount, copy.Count, 'Copy count must not change after original is modified');
Assert.AreEqual(originalTotalCount, copy.TotalCount, 'Copy total count must not change after original is modified');
Assert.AreNotEqual(FArray[0].Time, copy[0].Time, 'Newest item in copy should not be the new item from original');
Assert.AreNotEqual(FSeries[0].Time, copy[0].Time, 'Newest item in copy should not be the new item from original');
end;
procedure TTestDataSeries.TestCopyRespectsMaxLookback;
@@ -495,16 +491,16 @@ begin
Assert.AreEqual(Int64(15), FSeries.Count, 'Pre-condition: Series count should be capped by MaxLookback');
Assert.AreEqual(Int64(20), FSeries.TotalCount, 'Pre-condition: Series total count should be 20');
copy := FSeries.Copy;
copy := FSeries.Immutable;
// Verify the copy reflects the MaxLookback state
Assert.AreEqual(Int64(15), copy.Count, 'Copy count should be the MaxLookback value of the original');
Assert.AreEqual(Int64(20), copy.TotalCount, 'Copy total count should be the total items added to the original');
// Modify original
lastTime := FArray[0].Time;
lastTime := FSeries[0].Time;
newPoint := TDataPoint<TAskBidItem>.Create(lastTime + 1, TAskBidItem.Create(99, 99.1));
FArray.Add(newPoint);
FSeries.Add(newPoint);
// Verify the copy is still unchanged
Assert.AreEqual(Int64(15), copy.Count, 'Copy count must remain unchanged after original is modified');
@@ -606,7 +602,7 @@ begin
// Use default setup, but add one item
testTime := EncodeDate(2025, 1, 1);
DataPoint := TDataPoint<TAskBidItem>.Create(testTime, TAskBidItem.Create(1.0, 2.0));
FArray.Add(DataPoint);
FSeries.Add(DataPoint);
Assert.AreEqual(Int64(1), FSeries.Count);
Assert.AreEqual(Int64(0), FSeries.IndexOf(testTime), 'IndexOf for exact single item should be 0');
Test/TestCoreFutures.pas
+4 -4
View File
@@ -27,7 +27,7 @@ type
[IgnoreMemoryLeaks(true)]
TTestMycGateFuncFuture = class(TObject)
private
FTaskFactory: IMycThreadPool;
FTaskFactory: IThreadPool;
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: IMycThreadPool;
LLocalTaskFactory: IThreadPool;
LInitStateAsState: TState.IState;
LResultValue: Integer;
LExpectedExceptionRaisedByFactory: Boolean;
@@ -288,7 +288,7 @@ begin
end
);
LSub1 := TLatchNotifierSubscriber.Create(LGateLatch);
Subscriptions[1] := LPrerequisiteFuture1.Done.Subscribe(LSub1); // Subscribe to PF1's completion [cite: 56, 188, 306]
Subscriptions[1] := LPrerequisiteFuture1.Done.Signal.Subscribe(LSub1); // Subscribe to PF1's completion [cite: 56, 188, 306]
// PrerequisiteFuture2
LInitStateP2 := TLatch.CreateLatch(1); // Controllable init state for PF2
@@ -303,7 +303,7 @@ begin
end
);
LSub2 := TLatchNotifierSubscriber.Create(LGateLatch);
Subscriptions[2] := LPrerequisiteFuture2.Done.Subscribe(LSub2); // Subscribe to PF2's completion
Subscriptions[2] := LPrerequisiteFuture2.Done.Signal.Subscribe(LSub2); // Subscribe to PF2's completion
// --- Execution & Verification ---
Assert.IsFalse(LMainFuture.Done.IsSet, 'MainFuture should not be done yet.');
Test/TestTasks.pas
+1 -72
View File
@@ -17,7 +17,7 @@ type
[IgnoreMemoryLeaks(true)]
TMycTaskFactoryTests = class(TObject)
private
FFactory: IMycThreadPool;
FFactory: IThreadPool;
public
[Setup]
procedure Setup;
@@ -27,12 +27,6 @@ type
[Test]
procedure TestFactoryCreationAndTeardown;
[Test]
procedure TestCreateThreadExecutesAndSignals;
[Test]
procedure TestRunImmediateJob;
[Test]
procedure TestRunDelayedJobExecutesAfterAllNotifies;
[Test]
procedure TestWaitForAlreadySetState;
[Test]
procedure TestThreadInfoMethods;
@@ -67,71 +61,6 @@ begin
Assert.IsNotNull(FFactory, 'Factory should be created');
end;
procedure TMycTaskFactoryTests.TestCreateThreadExecutesAndSignals;
var
executed: Boolean;
threadState: TState.IState;
procWrapper: TProc;
begin
executed := False;
procWrapper := procedure begin executed := True; end;
threadState := FFactory.CreateThread(procWrapper); // CreateThread in TaskFactory now uses TMycLatch.CreateLatch
Assert.IsNotNull(threadState, 'CreateThread should return a valid state object');
FFactory.WaitFor(threadState);
Assert.IsTrue(executed, 'Procedure in created thread should have executed');
end;
procedure TMycTaskFactoryTests.TestRunImmediateJob;
var
jobExecuted: Boolean;
jobCompletedLatch: TLatch.ILatch;
begin
jobExecuted := False;
jobCompletedLatch := TLatch.CreateLatch(1); // Changed from Signals.CreateLatch
FFactory
.Run(
procedure
begin
jobExecuted := True;
jobCompletedLatch.Notify;
end)
.Notify;
FFactory.WaitFor(jobCompletedLatch.State);
Assert.IsTrue(jobExecuted, 'Immediate job should have executed');
end;
procedure TMycTaskFactoryTests.TestRunDelayedJobExecutesAfterAllNotifies;
var
jobExecuted: Boolean;
jobCompletedLatch: TLatch.ILatch;
subscriber: TSignal.ISubscriber;
begin
jobExecuted := False;
jobCompletedLatch := TLatch.CreateLatch(1); // Changed from Signals.CreateLatch
subscriber :=
FFactory.Run(
procedure
begin
jobExecuted := True;
jobCompletedLatch.Notify;
end
);
Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job');
// Use TMycLatch.Null for comparison
Assert.AreNotEqual<TSignal.ISubscriber>(TLatch.Null, subscriber, 'Subscriber should not be TMycLatch.Null for StartCount > 0');
subscriber.Notify;
FFactory.WaitFor(jobCompletedLatch.State);
Assert.IsTrue(jobExecuted, 'Delayed job should execute after all notifications');
end;
procedure TMycTaskFactoryTests.TestWaitForAlreadySetState;
var
alreadySetLatch: TLatch.ILatch;