From 86e2729a52afd7c3476f09c1988fb5d81c3bd819 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 2 Jun 2025 01:14:33 +0200 Subject: [PATCH] refactoring tasks --- Src/Myc.Core.Tasks.pas | 49 ++++++++++++++++---------------------- Src/Myc.Futures.pas | 3 ++- Test/TestTasks.pas | 54 ++++++------------------------------------ 3 files changed, 30 insertions(+), 76 deletions(-) diff --git a/Src/Myc.Core.Tasks.pas b/Src/Myc.Core.Tasks.pas index f50e7b1..b103df8 100644 --- a/Src/Myc.Core.Tasks.pas +++ b/Src/Myc.Core.Tasks.pas @@ -32,7 +32,9 @@ type // 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; + 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) @@ -67,7 +69,6 @@ type procedure WorkerThread; protected - procedure EnqueueJob(const Job: TProc); procedure ExecuteJob; function GetThreadCount: Integer; property WaitSemaphores: TMycAtomicStack read FWaitSemaphores; @@ -78,7 +79,8 @@ type class function CreateAnonymousThread(const DbgName: String; const Proc: TProc): TThread; function CreateThread(const Proc: TProc): IMycState; procedure HandleException; - function Run(StartCount: Integer; Job: TProc): IMycSubscriber; + procedure EnqueueJob(const Job: TProc); + function Run(Job: TProc): IMycSubscriber; procedure WaitFor(State: IMycState); procedure Teardown; function InMainThread: Boolean; @@ -91,7 +93,6 @@ type TMycPendingJob = class(TInterfacedObject, IMycSubscriber) private [volatile] - FCount: Integer; // Countdown counter FJob: TProc; // The job to execute when count reaches zero FOwner: TMycTaskFactory; // The task factory to enqueue the job on @@ -99,7 +100,7 @@ type function Notify: Boolean; public - constructor Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc); + constructor Create(OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc); property Owner: TMycTaskFactory read FOwner; end; @@ -200,6 +201,9 @@ end; procedure TMycTaskFactory.EnqueueJob(const Job: TProc); begin + if FTerminated <> 0 then + raise ETaskException.Create('Task factory terminated'); + if Assigned(Job) then begin FWorkStack.Push(Job); // Push job onto the lock-free stack @@ -257,7 +261,7 @@ begin exit(false); // Not found in worker thread list end; -function TMycTaskFactory.Run(StartCount: Integer; Job: TProc): IMycSubscriber; +function TMycTaskFactory.Run(Job: TProc): IMycSubscriber; begin if FTerminated <> 0 then raise ETaskException.Create('Task factory terminated'); @@ -265,14 +269,8 @@ begin if not Assigned(Job) then exit(TMycLatch.Null); // Already correct, uses direct static property on TMycLatch - if StartCount <= 0 then - begin - EnqueueJob(Job); // Enqueue immediately if no countdown needed - Result := TMycLatch.Null; // Already correct - end - else - // Create a pending job that waits for StartCount notifications - Result := TMycPendingJob.Create(StartCount, Self, Job); + // Create a pending job + Result := TMycPendingJob.Create(Self, Job); end; procedure TMycTaskFactory.Teardown; @@ -353,11 +351,10 @@ end; { TMycPendingJob } -constructor TMycPendingJob.Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc); +constructor TMycPendingJob.Create(OwnerTaskFactory: TMycTaskFactory; const + JobProc: TProc); begin inherited Create; - Assert(StartCountValue > 0); - FCount := StartCountValue; FOwner := OwnerTaskFactory; FJob := JobProc; end; @@ -365,20 +362,16 @@ end; function TMycPendingJob.Notify: Boolean; var newCount: Integer; + P: TProc; begin - newCount := TInterlocked.Decrement(FCount); - if newCount < -MaxInt div 2 then - TInterlocked.Increment(FCount); - - if newCount = 0 then + PPointer(@P)^ := TInterlocked.Exchange( PPointer(@FJob)^, nil ); + Result := Assigned(P); + if Result then begin - if Assigned(FOwner) and Assigned(FJob) then - begin - FOwner.EnqueueJob(FJob); - FJob := nil; - end; + if Assigned(FOwner) then + FOwner.EnqueueJob(P); + P := nil; end; - Result := newCount > 0; end; { TMycTaskWait } diff --git a/Src/Myc.Futures.pas b/Src/Myc.Futures.pas index 6df8f55..455d655 100644 --- a/Src/Myc.Futures.pas +++ b/Src/Myc.Futures.pas @@ -59,7 +59,7 @@ begin // 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(1, + ATaskFactory.Run( procedure begin @@ -87,6 +87,7 @@ begin inherited Destroy; end; + function TMycInitStateFuncFuture.GetDone: IMycState; begin Result := FDone.State; diff --git a/Test/TestTasks.pas b/Test/TestTasks.pas index cdc6435..634739d 100644 --- a/Test/TestTasks.pas +++ b/Test/TestTasks.pas @@ -32,8 +32,6 @@ type [Test] procedure TestRunDelayedJobExecutesAfterAllNotifies; [Test] - procedure TestRunDelayedJobDoesNotExecutePrematurely; - [Test] procedure TestWaitForAlreadySetState; [Test] procedure TestThreadInfoMethods; @@ -95,12 +93,12 @@ begin jobExecuted := False; jobCompletedLatch := TMycLatch.CreateLatch(1); // Changed from Signals.CreateLatch - FFactory.Run(0, + FFactory.Run( procedure begin jobExecuted := True; jobCompletedLatch.Notify; - end); + end).Notify; FFactory.WaitFor(jobCompletedLatch.State); Assert.IsTrue(jobExecuted, 'Immediate job should have executed'); @@ -111,13 +109,11 @@ var jobExecuted: Boolean; jobCompletedLatch: IMycLatch; subscriber: IMycSubscriber; - startCount: Integer; begin jobExecuted := False; - startCount := 2; jobCompletedLatch := TMycLatch.CreateLatch(1); // Changed from Signals.CreateLatch - subscriber := FFactory.Run(startCount, + subscriber := FFactory.Run( procedure begin jobExecuted := True; @@ -128,48 +124,12 @@ begin // Use TMycLatch.Null for comparison Assert.AreNotEqual(TMycLatch.Null, subscriber, 'Subscriber should not be TMycLatch.Null for StartCount > 0'); - subscriber.Notify; subscriber.Notify; FFactory.WaitFor(jobCompletedLatch.State); Assert.IsTrue(jobExecuted, 'Delayed job should execute after all notifications'); end; -procedure TMycTaskFactoryTests.TestRunDelayedJobDoesNotExecutePrematurely; -var - jobExecuted: Boolean; - jobCompleted: IMycLatch; - subscriber: IMycSubscriber; - startCount: Integer; - dummyWait: IMycLatch; -begin - jobExecuted := False; - startCount := 3; - jobCompleted := TMycLatch.CreateLatch(1); // Changed from Signals.CreateLatch - dummyWait := TMycLatch.CreateLatch(1); // Changed from Signals.CreateLatch - - - subscriber := FFactory.Run(startCount, - procedure - begin - jobExecuted := True; - jobCompleted.Notify; - end); - - Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job_P2'); - - subscriber.Notify; - Assert.IsFalse(jobExecuted, 'Job should not execute after one notification_P2'); - - subscriber.Notify; - Assert.IsFalse(jobExecuted, 'Job should not execute after two notifications_P2'); - - FFactory.Run(0, procedure begin dummyWait.Notify; end); - FFactory.WaitFor(dummyWait.State); - - Assert.IsFalse(jobExecuted, 'Job should still not have executed before third notify_P2'); -end; - procedure TMycTaskFactoryTests.TestWaitForAlreadySetState; var @@ -200,7 +160,7 @@ begin Assert.IsTrue(FFactory.InMainThread, 'Test method itself should be in main thread'); Assert.IsFalse(FFactory.InWorkerThread, 'Test method itself should not be in a factory worker thread'); - FFactory.Run(0, + FFactory.EnqueueJob( procedure begin inMainThreadInJob := FFactory.InMainThread; @@ -220,7 +180,7 @@ var begin jobDoneLatch := TMycLatch.CreateLatch(1); - FFactory.Run(0, + FFactory.EnqueueJob( procedure begin try @@ -255,7 +215,7 @@ begin Assert.WillRaise( procedure begin - FFactory.Run(0, procedure begin end); + FFactory.EnqueueJob(procedure begin end); end, TMycTaskFactory.ETaskException, 'Running a job on a torn-down factory should raise ETaskException' @@ -272,7 +232,7 @@ begin jobDoneLatch := TMycLatch.CreateLatch(1); dummyStateToWaitFor := TMycLatch.CreateLatch(1).State; - FFactory.Run(0, + FFactory.EnqueueJob( procedure begin try