added Futures

Src/Myc.Core.Tasks.pas

@@ -8,17 +8,47 @@ uses
 
 type
     IMycTaskFactory = interface
-        {$region 'property access'}
-        function GetThreadCount: Integer;
-        {$endregion}
-        function CreateThread(const Proc: TProc): IMycState;
-        function InMainThread: Boolean;
-        function InWorkerThread: Boolean;
-        function Run(StartCount: Integer; Job: TProc): IMycSubscriber;
-        procedure Teardown;
-        procedure WaitFor(State: IMycState);
-        property ThreadCount: Integer read GetThreadCount;
-    end;
+        {$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(StartCount: Integer; Job: TProc): IMycSubscriber;
+
+        // 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, IMycTaskFactory)
     type
@@ -58,7 +88,7 @@ type
 implementation
 
 type
-    TMyc2PendingJob = class(TInterfacedObject, IMycSubscriber)
+    TMycPendingJob = class(TInterfacedObject, IMycSubscriber)
     private
         [volatile]
         FCount: Integer; // Countdown counter
@@ -73,7 +103,7 @@ type
         property Owner: TMycTaskFactory read FOwner;
     end;
 
-    TMyc2TaskWait = class sealed(TInterfacedObject, IMycSubscriber)
+    TMycTaskWait = class sealed(TInterfacedObject, IMycSubscriber)
     private
         [volatile]
         FSemaphore: TSemaphore; // System.SyncObjs.TSemaphore to be released on notification
@@ -88,8 +118,6 @@ type
 { TMycTaskFactory }
 
 constructor TMycTaskFactory.Create;
-var
-    i: Integer;
 begin
     inherited Create;
 
@@ -98,24 +126,31 @@ begin
     FWorkGate := TSemaphore.Create(nil, 0, MaxInt, ''); // Initial count is 0, workers will wait
 
     SetLength(FWorkThreads, TThread.ProcessorCount); // Create one thread per processor core
-    for i := 0 to High(FWorkThreads) do
+    for var i := 0 to High(FWorkThreads) do
     begin
         FWorkThreads[i] := CreateAnonymousThread('Thread pool [' + IntToStr(i) + ']', WorkerThread);
         TInterlocked.Increment(FThreadsRunning); // Increment active thread counter
-        FWorkThreads[i].Start;
+        FWorkThreads[i].FreeOnTerminate := false;
     end;
+    for var i := 0 to High(FWorkThreads) do
+        FWorkThreads[i].Start;
 end;
 
 destructor TMycTaskFactory.Destroy;
 begin
     Teardown; // Perform cleanup and stop threads
 
+    for var i := High(FWorkThreads) downto 0 do
+        FWorkThreads[i].Free;
+
     FWorkGate.Free;
+    FWorkStack.Clear;
     inherited Destroy;
 end;
 
 class function TMycTaskFactory.CreateAnonymousThread(const DbgName: String; const Proc: TProc): TThread;
 {$IFDEF MSWINDOWS}
+{$WARN SYMBOL_PLATFORM OFF}
 var
     prio: TThreadPriority;
 {$ENDIF MSWINDOWS}
@@ -123,7 +158,6 @@ begin
     Result := TThread.CreateAnonymousThread(Proc);
 
     {$IFDEF MSWINDOWS}
-    {$WARN SYMBOL_PLATFORM OFF}
     // Set priority lower than MainThread priority if created from main thread
     if TThread.CurrentThread.ThreadID = MainThreadID then
     begin
@@ -199,7 +233,7 @@ begin
     begin
         ex := AtomicExchange(Pointer(FException), nil); // Atomically retrieve and clear the stored exception
         if ex <> nil then
-            raise ex; // Re-raise the exception in the main thread
+            raise Exception(ex);
     end;
 end;
 
@@ -238,7 +272,7 @@ begin
     end
     else
         // Create a pending job that waits for StartCount notifications
-        Result := TMyc2PendingJob.Create(StartCount, Self, Job);
+        Result := TMycPendingJob.Create(StartCount, Self, Job);
 end;
 
 procedure TMycTaskFactory.Teardown;
@@ -285,17 +319,19 @@ begin
     if InWorkerThread then
         raise ETaskException.Create('Waiting not allowed within tasks');
 
-    lock := FWaitSemaphores.Pop();
-    if lock = nil then
-        lock := TSemaphore.Create(nil, 0, 1, '');
     try
-        var subscription := State.Subscribe(TMyc2TaskWait.Create(lock));
-        lock.Acquire;
+        lock := FWaitSemaphores.Pop();
+        if lock = nil then
+            lock := TSemaphore.Create(nil, 0, 1, '');
+        try
+            var subscription := State.Subscribe(TMycTaskWait.Create(lock));
+            lock.Acquire;
+        finally
+            FWaitSemaphores.Push(lock);
+        end;
     finally
-        FWaitSemaphores.Push(lock);
+        HandleException;
     end;
-
-    HandleException;
 end;
 
 procedure TMycTaskFactory.ExecuteJob;
@@ -315,9 +351,9 @@ begin
     end;
 end;
 
-{ TMyc2PendingJob }
+{ TMycPendingJob }
 
-constructor TMyc2PendingJob.Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc);
+constructor TMycPendingJob.Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc);
 begin
     inherited Create;
     Assert(StartCountValue > 0);
@@ -326,7 +362,7 @@ begin
     FJob := JobProc;
 end;
 
-function TMyc2PendingJob.Notify: Boolean;
+function TMycPendingJob.Notify: Boolean;
 var
     newCount: Integer;
 begin
@@ -345,15 +381,15 @@ begin
     Result := newCount > 0;
 end;
 
-{ TMyc2TaskWait }
+{ TMycTaskWait }
 
-constructor TMyc2TaskWait.Create(SemaphoreToSignal: TSemaphore); // Parameter is System.SyncObjs.TSemaphore
+constructor TMycTaskWait.Create(SemaphoreToSignal: TSemaphore); // Parameter is System.SyncObjs.TSemaphore
 begin
     inherited Create;
     FSemaphore := SemaphoreToSignal; // Field is System.SyncObjs.TSemaphore
 end;
 
-function TMyc2TaskWait.Notify: Boolean;
+function TMycTaskWait.Notify: Boolean;
 var
     s: TSemaphore; // Local var is System.SyncObjs.TSemaphore
 begin