diff --git a/Src/Myc.Core.Futures.pas b/Src/Myc.Core.Futures.pas new file mode 100644 index 0000000..024d52a --- /dev/null +++ b/Src/Myc.Core.Futures.pas @@ -0,0 +1,80 @@ +unit Myc.Core.Futures; + +interface + +uses + System.SysUtils, + Myc.Signals, Myc.TaskManager, Myc.Futures; + +type + // Abstract base class for IMycFuture implementations. + TMycFuture = class abstract( TInterfacedObject, IMycFuture ) + protected + function GetResult: T; virtual; abstract; + function GetDone: IMycState; virtual; abstract; + end; + + // Concrete future implementation triggered by an initial state, executing a TFunc. + TMycInitStateFuncFuture = class( TMycFuture ) + private + FInit: TMycSubscription; + FDone: IMycLatch; + FResult: T; + protected + function GetResult: T; override; + function GetDone: IMycState; override; + public + // Creates a future that executes AProc after AGate is set, using ATaskFactory. + constructor Create( const ATaskManager: IMycTaskManager; const AGate: IMycState; AProc: TFunc ); + destructor Destroy; override; + end; + +implementation + +{ TMycInitStateFuncFuture } + +constructor TMycInitStateFuncFuture.Create( const ATaskManager: IMycTaskManager; const AGate: IMycState; AProc: TFunc ); +begin + inherited Create; + + FDone := TMycLatch.CreateLatch( 1 ); + + // Subscribe the job execution to AGate. + // The job will run when AGate notifies the subscriber returned by Run. + FInit := ATaskManager.CreateTask( + AGate, + procedure + begin + 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) + end; + end ); +end; + +destructor TMycInitStateFuncFuture.Destroy; +begin + Assert( FDone.State.IsSet ); + + FInit.Unsubscribe; + inherited Destroy; +end; + +function TMycInitStateFuncFuture.GetDone: IMycState; +begin + Result := FDone.State; +end; + +function TMycInitStateFuncFuture.GetResult: T; +begin + Assert( FDone.State.IsSet, 'Result is not yet available.' ); + Result := FResult; +end; + +end. diff --git a/Src/Myc.Core.Tasks.pas b/Src/Myc.Core.Tasks.pas index b103df8..c272f85 100644 --- a/Src/Myc.Core.Tasks.pas +++ b/Src/Myc.Core.Tasks.pas @@ -4,55 +4,55 @@ interface uses System.SysUtils, System.Classes, System.SyncObjs, - Myc.Core.Atomic, Myc.Signals; // Myc.Signals now requires direct use of TMycLatch static members + Myc.Core.Atomic, Myc.TaskManager, Myc.Signals; // Myc.Signals now requires direct use of TMycLatch static members type - IMycTaskFactory = interface - {$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): IMycState; - - // 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): IMycSubscriber; - - 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) - // will be re-raised in the calling (main) thread. - procedure Teardown; - - // 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(State: IMycState); - - // Gets the number of worker threads. - property ThreadCount: Integer read GetThreadCount; - end; + IMycTaskFactory = interface( IMycTaskManager ) + {$region 'property access'} + // Retrieves the number of worker threads. + function GetThreadCount: Integer; + {$endregion} - TMycTaskFactory = class(TInterfacedObject, IMycTaskFactory) + // 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): IMycState; + + // 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): IMycSubscriber; + + 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) + // will be re-raised in the calling (main) thread. + procedure Teardown; + + // 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(State: IMycState); + + // Gets the number of worker threads. + property ThreadCount: Integer read GetThreadCount; + end; + + TMycTaskFactory = class(TInterfacedObject, IMycTaskManager, IMycTaskFactory) type ETaskException = class(Exception) end; private @@ -67,6 +67,8 @@ type FWorkStack: TMycAtomicStack; // Stack of pending jobs FWorkThreads: TArray; // Array of worker threads procedure WorkerThread; + class var + FTaskManagerLock: Integer; protected procedure ExecuteJob; @@ -81,10 +83,14 @@ type procedure HandleException; procedure EnqueueJob(const Job: TProc); function Run(Job: TProc): IMycSubscriber; + function CreateTask(const Gate: IMycState; const Proc: TProc): TMycSubscription; procedure WaitFor(State: IMycState); procedure Teardown; function InMainThread: Boolean; function InWorkerThread: Boolean; + + class procedure AquireTaskManager; + class procedure ReleaseTaskManager; end; implementation @@ -175,6 +181,20 @@ begin Result.NameThreadForDebugging(DbgName); // Set thread name for debugging end; +function TMycTaskFactory.CreateTask(const Gate: IMycState; const Proc: TProc): + TMycSubscription; +begin + if not Assigned(Gate) or Gate.IsSet then + begin + EnqueueJob( Proc ); + end + else + begin + // The job will run when Gate notifies the subscriber returned by Run. + Result := Gate.Subscribe( Run( Proc ) ); + end; +end; + function TMycTaskFactory.CreateThread(const Proc: TProc): IMycState; var res: IMycLatch; @@ -261,6 +281,20 @@ 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): IMycSubscriber; begin if FTerminated <> 0 then @@ -361,7 +395,6 @@ end; function TMycPendingJob.Notify: Boolean; var - newCount: Integer; P: TProc; begin PPointer(@P)^ := TInterlocked.Exchange( PPointer(@FJob)^, nil ); diff --git a/Src/Myc.Futures.pas b/Src/Myc.Futures.pas index 455d655..4bf4c7f 100644 --- a/Src/Myc.Futures.pas +++ b/Src/Myc.Futures.pas @@ -4,7 +4,7 @@ interface uses System.SysUtils, - Myc.Signals, Myc.Core.Tasks; + Myc.Signals, Myc.TaskManager; type // Represents the eventual result of an asynchronous operation. @@ -19,91 +19,58 @@ type property Done: IMycState read GetDone; end; - // Abstract base class for IMycFuture implementations. - TMycFuture = class abstract(TInterfacedObject, IMycFuture) - protected - function GetResult: T; virtual; abstract; - function GetDone: IMycState; virtual; abstract; - end; + Future = record + strict private + FFuture: IMycFuture; + class constructor CreateClass; + class destructor DestroyClass; + constructor Create(const AFuture: IMycFuture); + public + class function Construct(const Proc: TFunc): Future; overload; static; + class function Construct(const Gate: IMycState; const Proc: TFunc): Future; overload; static; - // Concrete future implementation triggered by an initial state, executing a TFunc. - TMycInitStateFuncFuture = class(TMycFuture) - private - FInit: TMycSubscription; - FDone: IMycLatch; - FResult: T; - protected - function GetResult: T; override; - function GetDone: IMycState; override; - public - // Creates a future that executes AProc after AInitState is set, using ATaskFactory. - constructor Create(const ATaskFactory: IMycTaskFactory; const AInitState: - IMycState; AProc: TFunc); - // Cleans up resources, primarily by unsubscribing from the initial state. - destructor Destroy; override; + function Chain(const Proc: TFunc): Future; end; implementation -{ TMycInitStateFuncFuture } +uses + Myc.Core.Futures, Myc.Core.Tasks; -constructor TMycInitStateFuncFuture.Create(const ATaskFactory: - IMycTaskFactory; const AInitState: IMycState; AProc: TFunc); +constructor Future.Create(const AFuture: IMycFuture); begin - inherited Create; - Assert(Assigned(AInitState), 'AInitState must be assigned.'); - - FDone := TMycLatch.CreateLatch(1); - - // Subscribe the job execution to AInitState. - // The job will run when AInitState notifies the subscriber returned by Run. - // AStartCount for Run is 1, meaning it waits for one Notify from AInitState's subscription. - FInit := AInitState.Subscribe( - ATaskFactory.Run( - - procedure - begin - 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) - end; - end - ) - ); + FFuture := AFuture; end; -destructor TMycInitStateFuncFuture.Destroy; +class constructor Future.CreateClass; begin - Assert( FInit.State.IsSet ); - - // Explicitly unsubscribe from the initial state to ensure cleanup. - FInit.Unsubscribe; - inherited Destroy; + TMycTaskFactory.AquireTaskManager; end; - -function TMycInitStateFuncFuture.GetDone: IMycState; +class destructor Future.DestroyClass; begin - Result := FDone.State; + TMycTaskFactory.ReleaseTaskManager; end; -function TMycInitStateFuncFuture.GetResult: T; +function Future.Chain(const Proc: TFunc): Future; begin - if not FDone.State.IsSet then - begin - // Design decision: Caller must ensure future is done. - raise Exception.Create('Result is not yet available for the future.'); - end; + var Cap := FFuture; - // Returns FResult. If AProc failed, FResult is Default(T) and the exception - // was raised to be handled by the TaskFactory. - Result := FResult; + Result := Future.Construct( FFuture.Done, + function: S + begin + Result := Proc( Cap.Result ); + end ); +end; + +class function Future.Construct(const Proc: TFunc): Future; +begin + Result.Create( TMycInitStateFuncFuture.Create( TaskManager, nil, Proc ) ); +end; + +class function Future.Construct(const Gate: IMycState; const Proc: TFunc): Future; +begin + Result.Create( TMycInitStateFuncFuture.Create( TaskManager, Gate, Proc ) ); end; end. diff --git a/Src/Myc.Signals.pas b/Src/Myc.Signals.pas index ba0271c..b418daa 100644 --- a/Src/Myc.Signals.pas +++ b/Src/Myc.Signals.pas @@ -21,10 +21,11 @@ type FState: IMycState; FTag: TMycNotifyList.TTag; // Tag identifying this subscription within the notifier list. function GetState: IMycState; - constructor Create(const AState: IMycState; ATag: TMycNotifyList.TTag); public + constructor Create(const AState: IMycState; ATag: TMycNotifyList.TTag); // Unsubscribes from the associated signal. procedure Unsubscribe; + class operator Initialize(out Dest: TMycSubscription); property State: IMycState read GetState; end; @@ -154,44 +155,44 @@ uses System.SyncObjs; // For TInterlocked type - // TMycNullState implements a "null object" pattern for IMycState. - // This state is always considered "set". - TMycNullState = class(TInterfacedObject, IMycState) - protected - // IMycState - function GetIsSet: Boolean; - function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; - procedure Unsubscribe(Tag: TMycNotifyList.TTag); - public - constructor Create; - end; - -{ TMycNullState } - -constructor TMycNullState.Create; -begin - inherited Create; -end; - -function TMycNullState.GetIsSet: Boolean; -begin - Result := true; // Null state is always set. -end; - -function TMycNullState.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; -begin - // Since the state is always set, notify the subscriber immediately. - if Assigned(Subscriber) then - begin - Subscriber.Notify; - end; - // No ongoing subscription is necessary or meaningful for a null state. - Result.Create(Self, nil); // Return an empty subscription. -end; - -procedure TMycNullState.Unsubscribe(Tag: TMycNotifyList.TTag); -begin - // Unsubscribing from a null state has no effect. + // TMycNullState implements a "null object" pattern for IMycState. + // This state is always considered "set". + TMycNullState = class(TInterfacedObject, IMycState) + protected + // IMycState + function GetIsSet: Boolean; + function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; + procedure Unsubscribe(Tag: TMycNotifyList.TTag); + public + constructor Create; + end; + +{ TMycNullState } + +constructor TMycNullState.Create; +begin + inherited Create; +end; + +function TMycNullState.GetIsSet: Boolean; +begin + Result := true; // Null state is always set. +end; + +function TMycNullState.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; +begin + // Since the state is always set, notify the subscriber immediately. + if Assigned(Subscriber) then + begin + Subscriber.Notify; + end; + // No ongoing subscription is necessary or meaningful for a null state. + Result.Create(Self, nil); // Return an empty subscription. +end; + +procedure TMycNullState.Unsubscribe(Tag: TMycNotifyList.TTag); +begin + // Unsubscribing from a null state has no effect. end; { TMycSubscription } @@ -206,17 +207,20 @@ end; procedure TMycSubscription.Unsubscribe; begin - if Assigned(FState) then - begin - FState.Unsubscribe(FTag); - FTag := nil; - FState := nil; - end; + FState.Unsubscribe(FTag); + FState := TMycState.Null; + FTag := nil; end; function TMycSubscription.GetState: IMycState; begin - Result := FState; + Result := FState +end; + +class operator TMycSubscription.Initialize(out Dest: TMycSubscription); +begin + Dest.FState := TMycState.Null; + Dest.FTag := nil; end; { TMycLatch } @@ -301,7 +305,6 @@ function TMycLatch.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription var alreadySet: Boolean; begin - Result.Create(TMycState.Null, nil); // Default to an empty subscription. if not Assigned(Subscriber) then exit; @@ -409,7 +412,6 @@ end; function TMycDirty.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; begin - Result.Create(TMycState.Null, nil); // Default to an empty subscription if not Assigned(Subscriber) then exit; diff --git a/Src/Myc.TaskManager.pas b/Src/Myc.TaskManager.pas new file mode 100644 index 0000000..c46ee00 --- /dev/null +++ b/Src/Myc.TaskManager.pas @@ -0,0 +1,19 @@ +unit Myc.TaskManager; + +interface + +uses + System.SysUtils, + Myc.Signals; + +type + IMycTaskManager = interface + function CreateTask( const Gate: IMycState; const Proc: TProc ): TMycSubscription; + end; + +var + TaskManager: IMycTaskManager; + +implementation + +end. diff --git a/Test/MycTests.dpr b/Test/MycTests.dpr index 6e03a57..e1a53fc 100644 --- a/Test/MycTests.dpr +++ b/Test/MycTests.dpr @@ -24,7 +24,9 @@ uses TestTasks in 'TestTasks.pas', TestSignals_Dirty in 'TestSignals_Dirty.pas', Myc.Futures in '..\Src\Myc.Futures.pas', - TestFutures in 'TestFutures.pas'; + TestFutures in 'TestFutures.pas', + Myc.TaskManager in '..\Src\Myc.TaskManager.pas', + Myc.Core.Futures in '..\Src\Myc.Core.Futures.pas'; { keep comment here to protect the following conditional from being removed by the IDE when adding a unit } {$IFNDEF TESTINSIGHT} diff --git a/Test/MycTests.dproj b/Test/MycTests.dproj index 6f57883..1b9ad3b 100644 --- a/Test/MycTests.dproj +++ b/Test/MycTests.dproj @@ -124,6 +124,8 @@ $(PostBuildEvent)]]> + + Base @@ -1129,6 +1131,15 @@ $(PostBuildEvent)]]> True True + + + Winapi;System.Win;System;Data;REST;Xml;Vcl;FMX + + + + + + 12 diff --git a/Test/TestFutures.pas b/Test/TestFutures.pas index ff2315e..dd667a4 100644 --- a/Test/TestFutures.pas +++ b/Test/TestFutures.pas @@ -8,7 +8,7 @@ uses System.SyncObjs, Myc.Signals, // For IMycLatch, TMycLatch, IMycState, IMycSubscriber [cite: 62, 68, 53, 45] Myc.Core.Tasks, // For IMycTaskFactory, TMycTaskFactory [cite: 97, 113] - Myc.Futures; // For IMycFuture, TMycInitStateFuncFuture + Myc.Futures, Myc.Core.Futures; // For IMycFuture, TMycInitStateFuncFuture type // Helper class to notify a latch when its Notify method is called. @@ -207,12 +207,8 @@ procedure TTestMycInitStateFuncFuture.Test_GetResult_BeforeDone_RaisesException; var LFuture: IMycFuture; LInitLatch: IMycLatch; - LExpectedExceptionRaised: Boolean; -const - CExpectedExceptionMessage = 'Result is not yet available for the future.'; begin - LExpectedExceptionRaised := False; - LInitLatch := TMycLatch.CreateLatch(1); // Create a latch that is not yet set [cite: 77, 212, 330] + LInitLatch := TMycLatch.CreateLatch(1); LFuture := TMycInitStateFuncFuture.Create(FTaskFactory, LInitLatch.State, function: Integer @@ -223,19 +219,21 @@ begin Assert.IsFalse(LFuture.Done.IsSet, 'Future should not be marked as done initially.'); - try - LFuture.GetResult; - except - on E: Exception do + Assert.WillRaise( + procedure begin - Assert.AreEqual(CExpectedExceptionMessage, E.Message, 'Exception message mismatch.'); - LExpectedExceptionRaised := True; - end; - end; - Assert.IsTrue(LExpectedExceptionRaised, 'Calling GetResult before done should raise an exception.'); + LFuture.GetResult; + end ); - LInitLatch.Notify; // [cite: 80, 215, 333] - FTaskFactory.WaitFor(LFuture.Done); // + LInitLatch.Notify; + + FTaskFactory.WaitFor(LFuture.Done); + + Assert.WillNotRaise( + procedure + begin + LFuture.GetResult; + end ); end; procedure TTestMycInitStateFuncFuture.Test_FanIn_OneFutureWaitsForMultipleOthers;