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

refactoring tasks

Browse Source
This commit is contained in:
Michael Schimmel
2025-06-02 01:14:33 +02:00
parent 63484cd495
commit 86e2729a52
3 changed files with 30 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Src/Myc.Core.Tasks.pas
+21 -28
View File
@@ -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<TSemaphore> 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 }
Src/Myc.Futures.pas
+2 -1
View File
@@ -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<T>.GetDone: IMycState;
begin
Result := FDone.State;
Test/TestTasks.pas
+7 -47
View File
@@ -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<IMycSubscriber>(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