diff --git a/AuraTrader/AuraTrader.dproj b/AuraTrader/AuraTrader.dproj index 78cf432..e0cbb98 100644 --- a/AuraTrader/AuraTrader.dproj +++ b/AuraTrader/AuraTrader.dproj @@ -4,7 +4,7 @@ 20.3 FMX True - Release + Debug Win64 AuraTrader 3 diff --git a/AuraTrader/MainForm.fmx b/AuraTrader/MainForm.fmx index cbf3bbe..0c1bc72 100644 --- a/AuraTrader/MainForm.fmx +++ b/AuraTrader/MainForm.fmx @@ -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 diff --git a/AuraTrader/MainForm.pas b/AuraTrader/MainForm.pas index 7c3d693..28d22b8 100644 --- a/AuraTrader/MainForm.pas +++ b/AuraTrader/MainForm.pas @@ -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.CreateWriteable(width); - var done := + Result := FServer.ProcessData( Symbol, FTerminate.Signal, @@ -246,10 +246,11 @@ begin currPathData.Value := TObjectRef.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( diff --git a/Src/Myc.Core.Futures.pas b/Src/Myc.Core.Futures.pas index 26801b1..143c77e 100644 --- a/Src/Myc.Core.Futures.pas +++ b/Src/Myc.Core.Futures.pas @@ -25,7 +25,7 @@ type TMycGateFuncFuture = class(TMycFuture) private - FDone: TLatch.ILatch; + FDone: TState; FResult: T; protected function GetValue: T; override; @@ -74,26 +74,21 @@ constructor TMycGateFuncFuture.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.Destroy; @@ -103,12 +98,12 @@ end; function TMycGateFuncFuture.GetDone: TState; begin - Result := FDone.State; + Result := FDone; end; function TMycGateFuncFuture.GetValue: T; begin - Assert(FDone.State.IsSet, 'Result is not yet available.'); + Assert(FDone.IsSet, 'Result is not yet available.'); Result := FResult; end; diff --git a/Src/Myc.Core.Tasks.pas b/Src/Myc.Core.Tasks.pas index c198f45..6581f13 100644 --- a/Src/Myc.Core.Tasks.pas +++ b/Src/Myc.Core.Tasks.pas @@ -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; diff --git a/Src/Myc.Futures.pas b/Src/Myc.Futures.pas index d52d4e4..b86f3b4 100644 --- a/Src/Myc.Futures.pas +++ b/Src/Myc.Futures.pas @@ -44,7 +44,7 @@ type class operator Implicit(const A: TFuture): IFuture; overload; class function Construct(const Proc: TFunc): TFuture; overload; static; - class function Construct(const Gate: TState.IState; const Proc: TFunc): TFuture; overload; static; + class function Construct(const Gate: TState; const Proc: TFunc): TFuture; overload; static; class property Null: IFuture read FNull; @@ -94,23 +94,21 @@ end; function TFuture.Chain(const Proc: TProcConst): 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.Chain(const Proc: TFunc): TFuture; begin - var Cap := FFuture; - Result := TFuture.Construct(FFuture.Done, function: S begin Result := Proc(Cap.Value); end); + var cFuture := FFuture; + Result := TFuture.Construct(cFuture.Done, function: S begin Result := Proc(cFuture.Value); end); end; function TFuture.Chain(const Proc: TFuncConst): TFuture; begin - var Cap := FFuture; - Result := TFuture.Construct(FFuture.Done, function: S begin Result := Proc(Cap.Value); end); + var cFuture := FFuture; + Result := TFuture.Construct(cFuture.Done, function: S begin Result := Proc(cFuture.Value); end); end; class function TFuture.Construct(const Proc: TFunc): TFuture; @@ -118,7 +116,7 @@ begin Result := TMycGateFuncFuture.Create(TaskManager, nil, Proc); end; -class function TFuture.Construct(const Gate: TState.IState; const Proc: TFunc): TFuture; +class function TFuture.Construct(const Gate: TState; const Proc: TFunc): TFuture; begin Result := TMycGateFuncFuture.Create(TaskManager, Gate, Proc); end; diff --git a/Src/Myc.Signals.pas b/Src/Myc.Signals.pas index 0361807..b8e851f 100644 --- a/Src/Myc.Signals.pas +++ b/Src/Myc.Signals.pas @@ -81,8 +81,6 @@ type class function All(const States: TArray): TState; static; class function Any(const States: TArray): 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; diff --git a/Src/Myc.TaskManager.pas b/Src/Myc.TaskManager.pas index de2a680..01dea84 100644 --- a/Src/Myc.TaskManager.pas +++ b/Src/Myc.TaskManager.pas @@ -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; // 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; +begin + var cProc: TFunc := 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; diff --git a/Src/Myc.Test.Signals.Dirty.pas b/Src/Myc.Test.Signals.Dirty.pas index b8fa49e..0f95c07 100644 --- a/Src/Myc.Test.Signals.Dirty.pas +++ b/Src/Myc.Test.Signals.Dirty.pas @@ -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). diff --git a/Src/Myc.Test.Signals.Latch.pas b/Src/Myc.Test.Signals.Latch.pas index 582da80..59ee3fe 100644 --- a/Src/Myc.Test.Signals.Latch.pas +++ b/Src/Myc.Test.Signals.Latch.pas @@ -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.'); diff --git a/Src/Myc.Test.Trade.DataPoint.pas b/Src/Myc.Test.Trade.DataPoint.pas index 77cd56f..6dc76c8 100644 --- a/Src/Myc.Test.Trade.DataPoint.pas +++ b/Src/Myc.Test.Trade.DataPoint.pas @@ -13,7 +13,7 @@ type [TestFixture] TTestDataArray = class(TObject) private - FArray: IDataArray; + FArray: TDataSeries; procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); public [Setup] @@ -67,7 +67,6 @@ type [TestFixture] TTestDataSeries = class(TObject) private - FArray: IDataArray; FSeries: TDataSeries; procedure SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); public @@ -111,7 +110,7 @@ implementation procedure TTestDataArray.Setup; begin - FArray := TDataSeries.CreateArray(0); + FArray := TDataSeries.CreateWriteable(0); end; procedure TTestDataArray.Teardown; @@ -125,7 +124,7 @@ var DataPoint: TDataPoint; baseTime: TDateTime; begin - FArray := TDataSeries.CreateArray(MaxLookBack); + FArray := TDataSeries.CreateWriteable(MaxLookBack); baseTime := EncodeDate(2020, 7, 7); // Add data points with increasing timestamps. @@ -410,14 +409,12 @@ end; procedure TTestDataSeries.Setup; begin - FArray := TDataSeries.CreateArray(0); - FSeries := FArray; + FSeries := TDataSeries.CreateWriteable(0); end; procedure TTestDataSeries.Teardown; begin FSeries := Default(TDataSeries); - FArray := nil; end; procedure TTestDataSeries.SetupSeriesWithData(ItemCount: Integer = 10; MaxLookBack: Int64 = 0); @@ -426,15 +423,14 @@ var DataPoint: TDataPoint; baseTime: TDateTime; begin - FArray := TDataSeries.CreateArray(MaxLookBack); - FSeries := FArray; + FSeries := TDataSeries.CreateWriteable(MaxLookBack); baseTime := EncodeDate(2020, 7, 7); // Add data points with increasing timestamps. for i := 0 to ItemCount - 1 do begin DataPoint := TDataPoint.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.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.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.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'); diff --git a/Test/TestCoreFutures.pas b/Test/TestCoreFutures.pas index 65061cc..cfef253 100644 --- a/Test/TestCoreFutures.pas +++ b/Test/TestCoreFutures.pas @@ -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.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.'); diff --git a/Test/TestTasks.pas b/Test/TestTasks.pas index 36e982e..9edfc54 100644 --- a/Test/TestTasks.pas +++ b/Test/TestTasks.pas @@ -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(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;