From 95fddb01810fb448b097360751a859a0ebe75303 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 26 May 2025 00:27:09 +0200 Subject: [PATCH] TaskFactory added --- Src/Myc.Core.Atomic.pas | 37 +-- Src/Myc.Core.Notifier.pas | 115 ++++----- Src/Myc.Core.Tasks.pas | 382 +++++++++++++++++++++++++++++ Src/Myc.Signals.pas | 157 ++++++------ Test/MycTests.dpr | 10 +- Test/MycTests.dproj | 12 +- Test/TestNotifier.pas | 8 +- Test/TestNotifier_ChaosStress.pas | 6 +- Test/TestNotifier_Threading.pas | 2 +- Test/TestSList.pas | 52 ++-- Test/TestSignals.pas | 49 ---- Test/TestSignals_Semapore.pas | 390 ++++++++++++++++++++++++++++++ Test/TestTasks.pas | 323 +++++++++++++++++++++++++ 13 files changed, 1292 insertions(+), 251 deletions(-) create mode 100644 Src/Myc.Core.Tasks.pas delete mode 100644 Test/TestSignals.pas create mode 100644 Test/TestSignals_Semapore.pas create mode 100644 Test/TestTasks.pas diff --git a/Src/Myc.Core.Atomic.pas b/Src/Myc.Core.Atomic.pas index 9b3e894..390af62 100644 --- a/Src/Myc.Core.Atomic.pas +++ b/Src/Myc.Core.Atomic.pas @@ -56,6 +56,8 @@ type // Pushes an item onto the stack. procedure Push(const Data: T); inline; + procedure Clear; + // Allocates an aligned memory block for a stack item. function Alloc: PItem; inline; // Pushes a pre-allocated item pointer onto the SList. @@ -183,22 +185,8 @@ begin end; class operator TMycAtomicStack.Finalize(var Dest: TMycAtomicStack); -var - item: PItem; begin - // Use Dest to access fields and call instance methods - // on the record instance being finalized - item := Dest.PopPtr; // Call instance method PopPtr via Dest - while item <> nil do - begin - item.Data := Default(T); - FreeMemAligned(item); // Global procedure - item := Dest.PopPtr; // Call instance method PopPtr via Dest - end; - if Dest.FSList <> nil then // Check if FSList was initialized - begin - Dest.FSList.Free; // Call instance method Free on FSList via Dest - end; + Dest.Clear; end; function TMycAtomicStack.Alloc: PItem; @@ -209,6 +197,25 @@ begin Result.Data := Default(T); end; +procedure TMycAtomicStack.Clear; +var + item: PItem; +begin + // Use Dest to access fields and call instance methods + // on the record instance being finalized + item := PopPtr; // Call instance method PopPtr via Dest + while item <> nil do + begin + item.Data := Default(T); + FreeMemAligned(item); // Global procedure + item := PopPtr; // Call instance method PopPtr via Dest + end; + if FSList <> nil then // Check if FSList was initialized + begin + FSList.Free; // Call instance method Free on FSList via Dest + end; +end; + function TMycAtomicStack.Pop: T; begin // Instance method diff --git a/Src/Myc.Core.Notifier.pas b/Src/Myc.Core.Notifier.pas index 74786e5..aacf31c 100644 --- a/Src/Myc.Core.Notifier.pas +++ b/Src/Myc.Core.Notifier.pas @@ -1,49 +1,47 @@ unit Myc.Core.Notifier; - -interface - -uses - System.SysUtils, System.SyncObjs; - -type - // Low-level implementation for thread-safe multicast events. - // Implemented as a list referencing IInterface instances, with a locking mechanism for thread safety. - // Utilizes minimal memory. - // `Advise` adds an interface and returns a tag for constant-time removal by `Unadvise`. - // `UnadviseAll` removes all registered interfaces. - // After `Finalize` is called, the list is cleared, and no new interfaces can be added (closed state). - TMycNotifyList = record - type - TTag = Pointer; // Opaque tag used to identify a registered receiver for unsubscription. - - PItem = ^TItem; // Pointer to an internal list item. - TItem = record // Internal structure for storing a receiver and list linkage. - Next, Prev: PItem; // Pointers to the next and previous items in the doubly linked list. - Receiver: T; // The registered interface instance (the event sink). - end; - - strict private - FFirst: NativeUInt; // Stores the first receiver if no list is allocated, or acts as a combined lock and state field. - // Bit 0: Lock state (0 = locked, 1 = unlocked). - // Bit 1: Finalized state (0 = not finalized, 1 = finalized). - // Other bits (if not 0 and bit 0 is 1) can be a direct interface pointer if FList is nil. - FList: PItem; // Head of the linked list for additional receivers beyond the first one. - - class function AllocItem: PItem; static; inline; // Allocates and initializes memory for a new TItem. - class procedure FreeItem( Item: PItem ); static; inline; // Frees memory previously allocated for a TItem. - - public - procedure Create; // Initializes the notification list, preparing it for use. - procedure Destroy; // Cleans up all resources, including unadvising all receivers. Assumes no concurrent access. - function Advise(const Receiver: T): TTag; // Registers a receiver interface and returns an opaque tag for later unsubscription. - procedure Unadvise(Tag: TTag); // Unregisters a specific receiver using the tag obtained from Advise. - procedure UnadviseAll; // Unregisters all currently advised receivers. - procedure Finalize; // Clears all receivers and permanently prevents new interfaces from being added. - procedure Lock; inline; // Acquires an exclusive lock for thread-safe operations on the list. - procedure Release; inline;// Releases the previously acquired exclusive lock. - function IsLocked: Boolean; inline; // Checks if the list is currently locked by any thread. - function IsFinalized: Boolean; inline; // Checks if the list has been finalized and no longer accepts new receivers. - procedure Notify(Func: TPredicate); // Iterates through registered receivers and invokes the predicate; removes receiver if predicate returns false. + +interface + +uses + System.SysUtils, System.SyncObjs; + +type + // Low-level implementation for thread-safe multicast events. + // Implemented as a list referencing IInterface instances, with a locking mechanism for thread safety. + // Utilizes minimal memory. + // `Advise` adds an interface and returns a tag for constant-time removal by `Unadvise`. + // `UnadviseAll` removes all registered interfaces. + // After `Finalize` is called, the list is cleared, and no new interfaces can be added (closed state). + TMycNotifyList = record + type + TTag = Pointer; // Opaque tag used to identify a registered receiver for unsubscription. + + PItem = ^TItem; // Pointer to an internal list item. + TItem = record // Internal structure for storing a receiver and list linkage. + Next, Prev: PItem; // Pointers to the next and previous items in the doubly linked list. + Receiver: T; // The registered interface instance (the event sink). + end; + + strict private + FFirst: NativeUInt; // Stores the first receiver if no list is allocated, or acts as a combined lock and state field. + // Bit 0: Lock state (0 = locked, 1 = unlocked). + // Bit 1: Finalized state (0 = not finalized, 1 = finalized). + // Other bits (if not 0 and bit 0 is 1) can be a direct interface pointer if FList is nil. + FList: PItem; // Head of the linked list for additional receivers beyond the first one. + + class function AllocItem: PItem; static; inline; // Allocates and initializes memory for a new TItem. + class procedure FreeItem( Item: PItem ); static; inline; // Frees memory previously allocated for a TItem. + + public + procedure Create; // Initializes the notification list, preparing it for use. + procedure Destroy; // Cleans up all resources, including unadvising all receivers. Assumes no concurrent access. + function Advise(const Receiver: T): TTag; // Registers a receiver interface and returns an opaque tag for later unsubscription. + procedure Unadvise(Tag: TTag); // Unregisters a specific receiver using the tag obtained from Advise. + procedure UnadviseAll; // Unregisters all currently advised receivers. + procedure Lock; inline; // Acquires an exclusive lock for thread-safe operations on the list. + procedure Release; inline;// Releases the previously acquired exclusive lock. + function IsLocked: Boolean; inline; // Checks if the list is currently locked by any thread. + procedure Notify(Func: TPredicate); // Iterates through registered receivers and invokes the predicate; removes receiver if predicate returns false. end; implementation @@ -69,9 +67,6 @@ var begin Assert( IsLocked ); - if IsFinalized then - exit( 0 ); - if FFirst=0 then begin IInterface( FFirst ) := Receiver; @@ -108,13 +103,6 @@ begin Result := AllocMem( sizeof( TItem ) ); end; -procedure TMycNotifyList.Finalize; -begin - Assert( IsLocked ); - UnadviseAll; - FFirst := FFirst or 2; -end; - class procedure TMycNotifyList.FreeItem(Item: PItem); begin FreeMem( Item, sizeof( TItem ) ); @@ -125,20 +113,12 @@ begin Result := FFirst and 1 = 0; end; -function TMycNotifyList.IsFinalized: Boolean; -begin - Result := FFirst and 2 <> 0; -end; - procedure TMycNotifyList.Notify(Func: TPredicate); var Item, P: PItem; begin Assert( IsLocked ); - if IsFinalized then - exit; - if FFirst<>0 then if not Func( IInterface( FFirst ) ) then IInterface( FFirst ) := nil; @@ -165,15 +145,15 @@ var begin Assert( IsLocked ); - if IsFinalized then - exit; - if NativeUInt(Tag) = FFirst then begin IInterface( FFirst ) := nil; exit; end; + if FList = nil then + exit; + Item := PItem( Tag ); if Item = FList then @@ -193,9 +173,6 @@ procedure TMycNotifyList.UnadviseAll; begin Assert( IsLocked ); - if IsFinalized then - exit; - IInterface( FFirst ) := nil; while FList<>nil do Unadvise( TTag( FList ) ); diff --git a/Src/Myc.Core.Tasks.pas b/Src/Myc.Core.Tasks.pas new file mode 100644 index 0000000..5cba06a --- /dev/null +++ b/Src/Myc.Core.Tasks.pas @@ -0,0 +1,382 @@ +unit Myc.Core.Tasks; + +interface + +uses + System.SysUtils, System.Classes, System.SyncObjs, + Myc.Core.Atomic, Myc.Signals; + +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; + + TMycTaskFactory = class(TInterfacedObject, IMycTaskFactory) + type + ETaskException = class(Exception) end; + private + [volatile] + FException: TObject; // Holds the first exception object from a worker thread + [volatile] + FTerminated: Integer; // Flag to signal worker threads to terminate + [volatile] + FThreadsRunning: Integer; // Counter for active worker threads + FWaitSemaphores: TMycAtomicStack; // Pool of semaphores for WaitFor + FWorkGate: TSemaphore; // Semaphore to signal available work to worker threads + FWorkStack: TMycAtomicStack; // Stack of pending jobs + FWorkThreads: TArray; // Array of worker threads + procedure WorkerThread; + + protected + procedure EnqueueJob(const Job: TProc); + procedure ExecuteJob; + function GetThreadCount: Integer; + property WaitSemaphores: TMycAtomicStack read FWaitSemaphores; + + public + constructor Create; + destructor Destroy; override; + 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 WaitFor(State: IMycState); + procedure Teardown; + function InMainThread: Boolean; + function InWorkerThread: Boolean; + end; + +implementation + +type + TMyc2PendingJob = 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 + + protected + function Notify: Boolean; + + public + constructor Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc); + property Owner: TMycTaskFactory read FOwner; + end; + + TMyc2TaskWait = class sealed(TInterfacedObject, IMycSubscriber) + private + [volatile] + FSemaphore: TSemaphore; // Semaphore to be released on notification + + protected + function Notify: Boolean; + + public + constructor Create(SemaphoreToSignal: TSemaphore); + end; + +{ TMycTaskFactory } + +constructor TMycTaskFactory.Create; +var + i: Integer; +begin + inherited Create; + + Assert(InMainThread); // Ensure factory is created in the main thread + + 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 + begin + FWorkThreads[i] := CreateAnonymousThread('Thread pool [' + IntToStr(i) + ']', WorkerThread); + TInterlocked.Increment(FThreadsRunning); // Increment active thread counter + FWorkThreads[i].Start; + end; +end; + +destructor TMycTaskFactory.Destroy; +begin + Teardown; // Perform cleanup and stop threads + + FWorkGate.Free; + // FException is an acquired exception object, should not be freed manually. + // If FException <> nil here, it means an unhandled exception was not processed by HandleException. + // FException := nil; // Optionally nil it out if necessary for state clarity + + 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} +begin + Result := TThread.CreateAnonymousThread(Proc); + + {$IFDEF MSWINDOWS} + // Set priority lower than MainThread priority if created from main thread + if TThread.CurrentThread.ThreadID = MainThreadID then + begin + prio := TThread.CurrentThread.Priority; + if prio > Low(TThreadPriority) then + prio := TThreadPriority(Ord(prio) - 1); // Decrease priority by one step + + Result.Priority := prio; + end; + {$ENDIF MSWINDOWS} + + if DbgName <> '' then + Result.NameThreadForDebugging(DbgName); // Set thread name for debugging +end; + +function TMycTaskFactory.CreateThread(const Proc: TProc): IMycState; +var + res: IMycSemaphore; + capturedProc: TProc; +begin + res := Signals.CreateSemaphore(1); // Create a semaphore that will be signaled when the proc finishes + + capturedProc := Proc; // Capture Proc for the anonymous method + CreateAnonymousThread('Thread', + procedure + begin + try + capturedProc(); // Execute the provided procedure + finally + capturedProc := nil; // Clear the captured proc + res.Notify; // Signal completion + end; + end).Start; + + exit(res.State); // Return the state interface of the semaphore +end; + +procedure TMycTaskFactory.EnqueueJob(const Job: TProc); +begin + if Assigned(Job) then + begin + FWorkStack.Push(Job); // Push job onto the lock-free stack + FWorkGate.Release; // Signal a worker thread that a job is available + end; +end; + +procedure TMycTaskFactory.WorkerThread; +begin + try + repeat + FWorkGate.Acquire; // Wait for a job to be available or for termination + if FTerminated = 0 then // Check if termination has been requested + ExecuteJob; // Execute a job from the stack + until FTerminated <> 0; // Loop until termination is signaled + finally + TInterlocked.Decrement(FThreadsRunning); // Decrement active thread counter + end; +end; + +function TMycTaskFactory.GetThreadCount: Integer; +begin + Result := Length(FWorkThreads); +end; + +procedure TMycTaskFactory.HandleException; +var + ex: TObject; +begin + if InMainThread then + 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 + end; +end; + +function TMycTaskFactory.InMainThread: Boolean; +begin + Result := TThread.Current.ThreadID = MainThreadID; +end; + +function TMycTaskFactory.InWorkerThread: Boolean; +var + currentThreadId: TThreadID; + i: Integer; +begin + currentThreadId := TThread.Current.ThreadID; + if currentThreadId = MainThreadID then + exit(false); // Not a worker thread if it's the main thread + + for i := 0 to High(FWorkThreads) do + if FWorkThreads[i].ThreadID = currentThreadId then + exit(true); // Current thread is one of the worker threads + exit(false); // Not found in worker thread list +end; + +function TMycTaskFactory.Run(StartCount: Integer; Job: TProc): IMycSubscriber; +begin + if FTerminated <> 0 then + raise ETaskException.Create('Task factory terminated'); + + if not Assigned(Job) then + exit(Signals.Null); // Return a null subscriber if no job is provided + + if StartCount <= 0 then + begin + EnqueueJob(Job); // Enqueue immediately if no countdown needed + Result := Signals.Null; + end + else + // Create a pending job that waits for StartCount notifications + Result := TMyc2PendingJob.Create(StartCount, Self, Job); +end; + +procedure TMycTaskFactory.Teardown; +var + i: Integer; + s: TSemaphore; + job: TProc; // Variable to receive popped job to ensure it's processed/discarded +begin + HandleException; // Process any pending exception before shutdown + + if TInterlocked.Exchange(FTerminated, 1) <> 0 then + exit; // Ensure teardown runs only once + + // Release the work gate for each thread so they can wake up and terminate + for i := 0 to High(FWorkThreads) do + FWorkGate.Release; + + // Wait for all worker threads to finish + TSpinWait.SpinUntil( + function: Boolean + begin + Result := FThreadsRunning = 0; + end); + + Assert(FThreadsRunning = 0); // Verify all threads have terminated + + // Clear any remaining jobs from the work stack (these will not be executed) + repeat + job := FWorkStack.Pop(); + until not Assigned(job); + + + // Clear and free semaphores from the wait semaphore pool + s := FWaitSemaphores.Pop; + while s <> nil do + begin + s.Free; + s := FWaitSemaphores.Pop; + end; +end; + +procedure TMycTaskFactory.WaitFor(State: IMycState); +var + lock: TSemaphore; +begin + // If the state is already set, we are NOT allowed to wait. + if State.IsSet then + exit; + + if InWorkerThread then + raise ETaskException.Create('Waiting not allowed within tasks'); // Prevent deadlocks + + lock := FWaitSemaphores.Pop(); // Try to get a semaphore from the pool + if lock = nil then + lock := TSemaphore.Create(nil, 0, 1, ''); // Create new one if pool is empty (initial count 0) + try + // Subscribe a temporary waiter that will release the lock when the state is signaled + var subscription := State.Subscribe(TMyc2TaskWait.Create(lock)); + lock.Acquire; // Wait until the lock is released + finally + // The lock is now signaled (count 0 after Acquire). Push it back to the pool for reuse. + FWaitSemaphores.Push(lock); + end; + + HandleException; // Check for exceptions that might have occurred in worker threads during the wait +end; + +procedure TMycTaskFactory.ExecuteJob; +var + job: TProc; +begin + job := FWorkStack.Pop(); // Pop a job from the stack + + if Assigned(job) then + begin + try + job(); // Execute the job + except + // If no exception has been captured yet, capture this one + if FException = nil then + FException := AcquireExceptionObject; // Store the exception object to be raised in the main thread + // Else: an exception is already pending, this one is ignored to keep the first one. + end; + end; +end; + +{ TMyc2PendingJob } + +constructor TMyc2PendingJob.Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc); +begin + inherited Create; + Assert(StartCountValue > 0); // StartCount must be positive + FCount := StartCountValue; + FOwner := OwnerTaskFactory; + FJob := JobProc; +end; + +function TMyc2PendingJob.Notify: Boolean; +var + newCount: Integer; +begin + newCount := TInterlocked.Decrement(FCount); // Atomically decrement the counter + // Defend against extreme underflow, though ideally FCount should not go far below zero. + if newCount < -MaxInt div 2 then + TInterlocked.Increment(FCount); + + if newCount = 0 then // If counter reaches exactly zero + begin + // FOwner and FJob are captured by the class instance + if Assigned(FOwner) and Assigned(FJob) then // Ensure owner and job are still valid + begin + FOwner.EnqueueJob(FJob); // Enqueue the job + FJob := nil; // Clear the job reference to prevent re-enqueuing and allow ARC + end; + end; + Result := newCount > 0; // Returns true if further notifications are expected (count is still positive) +end; + +{ TMyc2TaskWait } + +constructor TMyc2TaskWait.Create(SemaphoreToSignal: TSemaphore); +begin + inherited Create; + FSemaphore := SemaphoreToSignal; +end; + +function TMyc2TaskWait.Notify: Boolean; +var + s: TSemaphore; +begin + // Atomically exchange FSemaphore with nil to ensure Release is called only once + s := TInterlocked.Exchange(FSemaphore, nil); + if s <> nil then + s.Release; // Release the semaphore, unblocking the waiting thread + Result := false; // No further notifications expected from this subscriber +end; + +initialization + IsMultiThread := true; // Mark the application as multi-threaded +end. diff --git a/Src/Myc.Signals.pas b/Src/Myc.Signals.pas index 998be29..885a51a 100644 --- a/Src/Myc.Signals.pas +++ b/Src/Myc.Signals.pas @@ -20,11 +20,12 @@ type FSignal: TMycSignal; FTag: TMycNotifyList.TTag; public - procedure Setup(ASignal: TMycSignal; ATag: TMycNotifyList.TTag); inline; - class operator Assign(var Dest: TMycSubscription; const [ref] Src: - TMycSubscription); class operator Initialize(out Dest: TMycSubscription); class operator Finalize(var Dest: TMycSubscription); + class operator Assign(var Dest: TMycSubscription; const [ref] Src: TMycSubscription); + + procedure Setup(ASignal: TMycSignal; ATag: TMycNotifyList.TTag); inline; + procedure Unsubscribe; end; IMycSignal = interface @@ -32,18 +33,9 @@ type end; TMycSignal = class abstract( TInterfacedObject, IMycSignal ) - strict private - FSubscribers: TMycNotifyList; - private - procedure Unsubscribe(Tag: TMycNotifyList.TTag); - protected - procedure Lock; - procedure Unlock; - procedure Finalize; - procedure Notify; public - function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; - property Subscribers: TMycNotifyList read FSubscribers; + function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; virtual; abstract; + procedure Unsubscribe(Tag: TMycNotifyList.TTag); virtual; abstract; end; IMycState = interface( IMycSignal ) @@ -61,18 +53,28 @@ type end; TMycSemaphore = class(TMycSignal, IMycSemaphore, IMycState) + strict private + FSubscribers: TMycNotifyList; private [volatile] FCount: Integer; function GetState: IMycState; function GetIsSet: Boolean; - protected - function Notify: Boolean; public constructor Create(ACount: Integer); + destructor Destroy; override; + function Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; override; + procedure Unsubscribe(Tag: TMycNotifyList.TTag); override; + function Notify: Boolean; end; Signals = record + strict private + class var + FNull: IMycSemaphore; + class constructor Create; + public class function CreateSemaphore(Count: Integer): IMycSemaphore; static; + class property Null: IMycSemaphore read FNull; end; implementation @@ -80,63 +82,23 @@ implementation uses System.SyncObjs; -procedure TMycSignal.Finalize; -begin - FSubscribers.Finalize; -end; - -procedure TMycSignal.Lock; -begin - FSubscribers.Lock; -end; - -procedure TMycSignal.Notify; -begin - FSubscribers.Notify( - function( Subscriber: IMycSubscriber ): Boolean - begin - Result := Subscriber.Notify; - end ); -end; - -function TMycSignal.Subscribe(const Subscriber: IMycSubscriber): TMycSubscription; -begin - if not Assigned(Subscriber) then - exit; - - FSubscribers.Lock; - try - Result.Setup( Self, FSubscribers.Advise( Subscriber ) ); - finally - FSubscribers.Release; - end; -end; - -procedure TMycSignal.Unlock; -begin - -end; - -procedure TMycSignal.Unsubscribe(Tag: TMycNotifyList.TTag); -begin - if Tag=nil then - exit; - - FSubscribers.Lock; - try - FSubscribers.Unadvise( Tag ); - finally - FSubscribers.Release; - end; -end; - procedure TMycSubscription.Setup(ASignal: TMycSignal; ATag: TMycNotifyList.TTag); begin + Unsubscribe; FSignal := ASignal; FTag := ATag; end; +procedure TMycSubscription.Unsubscribe; +begin + if FTag<>nil then + begin + FSignal.Unsubscribe( FTag ); + FTag := nil; + end; +end; + class operator TMycSubscription.Assign(var Dest: TMycSubscription; const [ref] Src: TMycSubscription); begin @@ -150,33 +112,42 @@ end; class operator TMycSubscription.Finalize(var Dest: TMycSubscription); begin - if Dest.FTag=nil then - exit; - Dest.FSignal.Unsubscribe( Dest.FTag ); + Dest.Unsubscribe; end; constructor TMycSemaphore.Create(ACount: Integer); begin inherited Create; FCount := ACount; + FSubscribers.Create; +end; + +destructor TMycSemaphore.Destroy; +begin + FSubscribers.Destroy; + inherited; end; function TMycSemaphore.Notify: Boolean; begin - Lock; + FSubscribers.Lock; try var n := TInterlocked.Decrement( FCount ); if n < -MaxInt div 2 then TInterlocked.Increment( FCount ); if n = 0 then - Notify; + FSubscribers.Notify( + function( Subscriber: IMycSubscriber ): Boolean + begin + Result := Subscriber.Notify; + end ); Result := n > 0; if not Result then - Finalize; + FSubscribers.UnadviseAll; finally - Release; + FSubscribers.Release; end; end; @@ -190,6 +161,46 @@ begin exit( Self ); end; +function TMycSemaphore.Subscribe(const Subscriber: IMycSubscriber): + TMycSubscription; +begin + if not Assigned(Subscriber) then + exit; + + Subscriber._AddRef; + try + FSubscribers.Lock; + try + if FCount <= 0 then + Subscriber.Notify + else + Result.Setup( Self, FSubscribers.Advise( Subscriber ) ); + finally + FSubscribers.Release; + end; + finally + Subscriber._Release; + end; +end; + +procedure TMycSemaphore.Unsubscribe(Tag: TMycNotifyList.TTag); +begin + if Tag=nil then + exit; + + FSubscribers.Lock; + try + FSubscribers.Unadvise( Tag ); + finally + FSubscribers.Release; + end; +end; + +class constructor Signals.Create; +begin + FNull := TMycSemaphore.Create( 0 ); +end; + { Signals } class function Signals.CreateSemaphore(Count: Integer): IMycSemaphore; diff --git a/Test/MycTests.dpr b/Test/MycTests.dpr index 77a3ce2..7066fed 100644 --- a/Test/MycTests.dpr +++ b/Test/MycTests.dpr @@ -17,10 +17,12 @@ uses DUnitX.TestFramework, TestNotifier in 'TestNotifier.pas', TestSList in 'TestSList.pas', - TestStack in 'TestStack.pas', - Myc.Core.Notifier in '..\Src\Myc.Core.Notifier.pas', - TestNotifier_Threading in 'TestNotifier_Threading.pas', - TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas'; + TestStack in 'TestStack.pas' {/TestNotifier_Threading in 'TestNotifier_Threading.pas',}, + TestNotifier_Threading in 'TestNotifier_Threading.pas' {/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',}, + TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas', + TestSignals_Semapore in 'TestSignals_Semapore.pas', + Myc.Core.Tasks in '..\Src\Myc.Core.Tasks.pas', + TestTasks in 'TestTasks.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 e8127c5..a860af3 100644 --- a/Test/MycTests.dproj +++ b/Test/MycTests.dproj @@ -107,10 +107,16 @@ - - - + +
/TestNotifier_Threading in 'TestNotifier_Threading.pas',
+ + +
/TestNotifier_ChaosStress in 'TestNotifier_ChaosStress.pas',
+ + + + Base diff --git a/Test/TestNotifier.pas b/Test/TestNotifier.pas index b53a4c7..c808486 100644 --- a/Test/TestNotifier.pas +++ b/Test/TestNotifier.pas @@ -114,8 +114,8 @@ begin FNotifier.Lock; Assert.IsTrue( FNotifier.IsLocked, 'Notifier should be locked.' ); - FNotifier.Finalize; // Internally, FFirst becomes 2 (locked & finalized state bits) if it was 0 after Lock - Assert.IsTrue( FNotifier.IsFinalized, 'Notifier should be finalized.' ); +// FNotifier.Finalize; // Internally, FFirst becomes 2 (locked & finalized state bits) if it was 0 after Lock +// Assert.IsTrue( FNotifier.IsFinalized, 'Notifier should be finalized.' ); // In a corrected Notifier, this call should not cause an AV and not call the predicate. FNotifier.Notify( dummyPredicate ); @@ -136,8 +136,8 @@ begin FNotifier.Lock; Assert.IsTrue( FNotifier.IsLocked, 'Notifier should be locked.' ); - FNotifier.Finalize; // FFirst becomes 2 internally - Assert.IsTrue( FNotifier.IsFinalized, 'Notifier should be finalized.' ); +// FNotifier.Finalize; // FFirst becomes 2 internally +// Assert.IsTrue( FNotifier.IsFinalized, 'Notifier should be finalized.' ); // In a corrected Notifier, this call should not cause an AV. FNotifier.UnadviseAll; diff --git a/Test/TestNotifier_ChaosStress.pas b/Test/TestNotifier_ChaosStress.pas index fd54a45..fb3d86f 100644 --- a/Test/TestNotifier_ChaosStress.pas +++ b/Test/TestNotifier_ChaosStress.pas @@ -211,7 +211,7 @@ begin begin FOwnerFixture.FNotifier.Lock; try - if not FOwnerFixture.FNotifier.IsFinalized then // Don't notify if finalized +// if not FOwnerFixture.FNotifier.IsFinalized then // Don't notify if finalized begin FOwnerFixture.FNotifier.Notify( function(Item: IMyStressTestInterface): Boolean @@ -272,7 +272,7 @@ begin try // UnadviseAll should be safe if the IsFinalized guard is present // and the FFirst state is not pathological. - if not FNotifier.IsFinalized then +// if not FNotifier.IsFinalized then begin FNotifier.UnadviseAll; end; @@ -358,7 +358,7 @@ begin try FNotifier.Lock; try - if not FNotifier.IsFinalized then +// if not FNotifier.IsFinalized then begin FNotifier.Notify( function(Item: IMyStressTestInterface): Boolean diff --git a/Test/TestNotifier_Threading.pas b/Test/TestNotifier_Threading.pas index 3d120f8..c275fee 100644 --- a/Test/TestNotifier_Threading.pas +++ b/Test/TestNotifier_Threading.pas @@ -169,7 +169,7 @@ begin try // UnadviseAll should be safe if the IsFinalized guard is present // and the FFirst state is not pathological. - if not FNotifier.IsFinalized then +// if not FNotifier.IsFinalized then begin FNotifier.UnadviseAll; end; diff --git a/Test/TestSList.pas b/Test/TestSList.pas index 6e3669e..f1e2ecc 100644 --- a/Test/TestSList.pas +++ b/Test/TestSList.pas @@ -5,9 +5,9 @@ interface uses DUnitX.TestFramework, Myc.Core.Atomic, // The unit to be tested. TSListEntry and PSListEntry are expected from here. - System.SysUtils, // For NativeUInt, FillChar, Format static class, etc. - System.SyncObjs, // TInterlocked - Winapi.Windows, // For general Windows API types if needed by Myc.Heap for TSListHeader. + System.SysUtils, // For NativeUInt, FillChar, Format static class, etc. + System.SyncObjs, // TInterlocked + Winapi.Windows, // For general Windows API types if needed by Myc.Heap for TSListHeader. System.Generics.Collections; // For TThreadedQueue, TDictionary type @@ -58,16 +58,16 @@ type implementation uses - System.Math, // For Random and Randomize - System.Classes, // For TThread - System.Types; // For TWaitResult + System.Math, // For Random and Randomize + System.Classes, // For TThread + System.Types; // For TWaitResult // --- Helper Implementations (static) --- class function TMycTestSList.CreateTestListItem(ID: Integer): PTestListItem; begin GetMemAligned(Result, SizeOf(TTestListItem)); - Assert.IsNotNull(Result, Format('Failed to allocate memory for TestListItem ID %d', [ID])); + Assert.IsNotNull(Result, 'Failed to allocate memory for TestListItem.'); // Replaced Format string FillChar(Result^, SizeOf(TTestListItem), 0); Result.ID := ID; end; @@ -128,18 +128,16 @@ begin GetMemAligned(ptr, sizeToAllocate); if sizeToAllocate = 0 then begin - Assert.IsTrue(True, Format('Zero-size allocation processed (Iteration %d). Pointer: %p', [i, ptr])); + Assert.IsTrue(True, 'Zero-size allocation processed.'); // Replaced Format string end else if ptr = nil then begin - Assert.Fail(Format('GetMemAligned returned nil for size %u (Iteration %d).', - [sizeToAllocate, i])); + Assert.Fail('GetMemAligned returned nil for non-zero size.'); // Replaced Format string end else begin Assert.AreEqual(System.NativeUInt(0), System.NativeUInt(ptr) and TestAlignmentMask, - Format('Pointer %p not 16-byte aligned. Size: %u, Iteration: %d. (Value AND Mask = %u)', - [ptr, sizeToAllocate, i, System.NativeUInt(ptr) and TestAlignmentMask])); + 'Pointer not 16-byte aligned.'); // Replaced Format string fillValue := Byte(i mod 256); System.FillChar(ptr^, sizeToAllocate, fillValue); @@ -149,8 +147,7 @@ begin begin if bytePtr[j] <> fillValue then begin - Assert.Fail(Format('Data corruption at offset %u for pointer %p. Size: %u, Iteration: %d. Expected %u, got %u.', - [j, ptr, sizeToAllocate, i, fillValue, bytePtr[j]])); + Assert.Fail('Data corruption detected.'); // Replaced Format string Break; end; end; @@ -212,18 +209,18 @@ begin begin items[i] := TMycTestSList.CreateTestListItem(200 + i); sListPtr.Push(@items[i].ListEntry); - Assert.AreEqual(i, sListPtr.QueryDepth, Format('Depth should be %d after %d pushes.', [i,i])); + Assert.AreEqual(i, sListPtr.QueryDepth, 'Depth incorrect after a push operation.'); // Replaced Format string end; Assert.AreEqual(NumItems, sListPtr.QueryDepth, 'Final depth after all pushes incorrect.'); for i := NumItems downto 1 do begin poppedInternalListEntry := sListPtr.Pop; - Assert.IsNotNull(poppedInternalListEntry, Format('Pop should return an item, not nil (iteration %d).', [i])); + Assert.IsNotNull(poppedInternalListEntry, 'Pop should return a non-nil item.'); // Replaced Format string poppedItem := PTestListItem(poppedInternalListEntry); Assert.AreEqual(NativeUInt(items[i]), NativeUInt(poppedItem), - Format('LIFO order violated. Expected item ID %d (original index %d), got ID %d.', [items[i].ID, i, poppedItem.ID])); + 'LIFO order violated.'); // Replaced Format string Assert.AreEqual(items[i].ID, poppedItem.ID, 'Popped item ID mismatch.'); - Assert.AreEqual(i - 1, sListPtr.QueryDepth, Format('Depth incorrect after pop %d.', [NumItems - i + 1])); + Assert.AreEqual(i - 1, sListPtr.QueryDepth, 'Depth incorrect after a pop operation.'); // Replaced Format string end; Assert.AreEqual(0, sListPtr.QueryDepth, 'Depth should be 0 after all items are popped.'); for i := 1 to NumItems do @@ -324,8 +321,7 @@ begin begin sharedSList.Push(@item.ListEntry); pushLogResult := pushedIDs.PushItem(itemID); // Use PushItem for TThreadedQueue - Assert.AreEqual(TWaitResult.wrSignaled, pushLogResult, - Format('Failed to push item ID %d to pushedIDs logging queue.', [itemID])); + Assert.AreEqual(TWaitResult.wrSignaled, pushLogResult, 'Failed to push item to pushedIDs logging queue.'); end; end else // Attempt to Pop @@ -336,12 +332,12 @@ begin item := PTestListItem(poppedInternalEntry); pushLogResult := poppedIDs.PushItem(item.ID); // Use PushItem for TThreadedQueue Assert.AreEqual(TWaitResult.wrSignaled, pushLogResult, - Format('Failed to push item ID %d to poppedIDs logging queue.', [item.ID])); + 'Failed to push item ID to poppedIDs logging queue.'); // Replaced Format string TMycTestSList.FreeTestListItem(item); end; end; if j mod (OperationsPerThread div 10) = 0 then // Occasional sleep - TThread.Sleep(1); // TThread.Sleep from System.Classes + TThread.Sleep(1); // TThread.Sleep from System.Classes end; end); threads[i].FreeOnTerminate := false; @@ -366,7 +362,7 @@ begin remainingItem := PTestListItem(remainingInternalEntry); waitResult := poppedIDs.PushItem(remainingItem.ID); // Use PushItem Assert.AreEqual(TWaitResult.wrSignaled, waitResult, - Format('Failed to push remaining item ID %d to poppedIDs logging queue during drain.', [remainingItem.ID])); + 'Failed to push remaining item ID to poppedIDs during drain.'); // Replaced Format string TMycTestSList.FreeTestListItem(remainingItem); end; @@ -374,8 +370,7 @@ begin Assert.AreEqual(pushedIDs.TotalItemsPushed, poppedIDs.TotalItemsPushed, // Compare total items PUSHED to each log queue - Format('Mismatch between total logged pushed items (%u) and total logged popped items (%u).', - [pushedIDs.TotalItemsPushed, poppedIDs.TotalItemsPushed])); + 'Mismatch in total logged pushed and popped items.'); // Replaced Format string pushedItemsFinal := TDictionary.Create; try @@ -393,8 +388,7 @@ begin // Drain poppedIDs queue using PopItem with timeout = 0 behavior while poppedIDs.PopItem(qz, currentPoppedID) = TWaitResult.wrSignaled do begin - Assert.IsTrue(pushedItemsFinal.ContainsKey(currentPoppedID), - Format('Item ID %d was logged as popped but never logged as pushed.', [currentPoppedID])); + Assert.IsTrue(pushedItemsFinal.ContainsKey(currentPoppedID), 'Item ID logged as popped but never as pushed.'); // Replaced Format string if pushedItemsFinal.ContainsKey(currentPoppedID) then // Re-check for safety before decrementing begin pushedItemsFinal.Items[currentPoppedID] := pushedItemsFinal.Items[currentPoppedID] - 1; @@ -405,9 +399,7 @@ begin // Verify that all pushed items were accounted for (counts should be zero) for currentPushedID in pushedItemsFinal.Keys do begin - Assert.AreEqual(0, pushedItemsFinal.Items[currentPushedID], - Format('Item ID %d count mismatch. Final count in reconciliation dictionary is not zero (is %d).', - [currentPushedID, pushedItemsFinal.Items[currentPushedID]])); + Assert.AreEqual(0, pushedItemsFinal.Items[currentPushedID], 'Item count mismatch. Final count in reconciliation dictionary is not zero.'); end; finally pushedItemsFinal.Free; diff --git a/Test/TestSignals.pas b/Test/TestSignals.pas deleted file mode 100644 index e10e7fb..0000000 --- a/Test/TestSignals.pas +++ /dev/null @@ -1,49 +0,0 @@ -unit TestSignals; - -interface - -uses - DUnitX.TestFramework, - Myc.Signals; - -type - [TestFixture] - TMycTest = class - public - [Setup] - procedure Setup; - [TearDown] - procedure TearDown; - // Sample Methods - // Simple single Test - [Test] - procedure Test1; - // Test with TestCase Attribute to supply parameters. - [Test] - [TestCase('TestA','1,2')] - [TestCase('TestB','3,4')] - procedure Test2(const AValue1 : Integer;const AValue2 : Integer); - end; - -implementation - -procedure TMycTest.Setup; -begin -end; - -procedure TMycTest.TearDown; -begin -end; - -procedure TMycTest.Test1; -begin -end; - -procedure TMycTest.Test2(const AValue1 : Integer;const AValue2 : Integer); -begin -end; - -initialization - TDUnitX.RegisterTestFixture(TMycTest); - -end. diff --git a/Test/TestSignals_Semapore.pas b/Test/TestSignals_Semapore.pas new file mode 100644 index 0000000..bd877c7 --- /dev/null +++ b/Test/TestSignals_Semapore.pas @@ -0,0 +1,390 @@ +// TestSignals_Semapore.pas +unit TestSignals_Semapore; + +interface + +uses + DUnitX.TestFramework, + Myc.Signals, // Unit under test + System.SysUtils, + System.StrUtils; // For BoolToStr (though not used in asserts anymore) + +type + // Helper class to mock a subscriber. + TMockSubscriber = class(TInterfacedObject, IMycSubscriber) + public + NotifyCount: Integer; + NotifyShouldReturn: Boolean; + WasCalled: Boolean; + + constructor Create(ShouldReturn: Boolean = True); + function Notify: Boolean; // IMycSubscriber implementation. + end; + + [TestFixture] + TTestMycSemaphore = class(TObject) + public + [Setup] + procedure Setup; + [TearDown] + procedure TearDown; + + // --- Category 1: Basic Counter and State Functionality (Parameterized) --- + [TestCase('InitialPositive', '1,False')] + [TestCase('InitialZero', '0,True')] + [TestCase('InitialNegative', '-1,True')] + [TestCase('InitialLargePositive', '5,False')] + procedure TestCreate_InitialState(InitialCount: Integer; ExpectedIsSet: Boolean); + + [TestCase('DecrementPositiveToZero', '1,False,True')] + [TestCase('DecrementPositiveToPositive', '2,True,False')] + [TestCase('DecrementZeroToNegative', '0,False,True')] + [TestCase('DecrementNegativeToNegative', '-1,False,True')] + procedure TestNotify_ReturnValueAndState_AfterOneNotify(InitialCount: Integer; ExpectedReturn: Boolean; ExpectedIsSet: Boolean); + + [Test] + procedure TestNotify_DecrementsCounter_StateChanges; + [Test] + procedure TestNotify_MultipleNotifies_CounterBecomesNegativeAndStaysSet; + + // --- Category 2: Notification to Subscribers --- + [Test] + procedure TestSubscriber_Single_IsNotifiedWhenSet; + [Test] + procedure TestSubscriber_Multiple_AreNotifiedWhenSet; + [Test] + procedure TestSubscriber_NotNotified_BeforeSet; + [Test] + procedure TestSubscriber_NotifiedOnlyOnce_AfterFinalization; + [Test] + procedure TestSubscribe_AfterFinalization_SubscriberNotNotified; + + // --- Category 3: Chaining and Cascading --- + [Test] + procedure TestChaining_A_Notifies_B; + [Test] + procedure TestCascade_A_Notifies_B_Notifies_C; + [Test] + procedure TestChaining_B_NeedsMultipleSignalsFromA_WithSubscriberOnB; + + // --- Category 4: Special Cases --- + [Test] + procedure TestInitiallySet_NotifiesNewSubscriberImmediately; + + [Test] + procedure TestUnsubscribe_SubscriberNotNotified; + + // --- Category 5: Signal Loss / Counter Integrity --- + [Test] + procedure TestMultipleTriggersNoSubscriber_StateCorrectForLaterSubscriber; + end; + +implementation + +{ TMockSubscriber } + +constructor TMockSubscriber.Create(ShouldReturn: Boolean = True); +begin + inherited Create; + NotifyCount := 0; + NotifyShouldReturn := ShouldReturn; + WasCalled := False; +end; + +function TMockSubscriber.Notify: Boolean; +begin + Inc(NotifyCount); + WasCalled := True; + Result := NotifyShouldReturn; +end; + +{ TTestMycSemaphore } + +procedure TTestMycSemaphore.Setup; +begin + // Per-test setup code can go here (if any). +end; + +procedure TTestMycSemaphore.TearDown; +begin + // Per-test teardown code can go here (if any). +end; + +procedure TTestMycSemaphore.TestCreate_InitialState(InitialCount: Integer; ExpectedIsSet: Boolean); +var + semaphore: IMycSemaphore; +begin + semaphore := Signals.CreateSemaphore(InitialCount); + Assert.AreEqual(ExpectedIsSet, semaphore.State.IsSet, 'Semaphore initial IsSet state mismatch.'); +end; + +procedure TTestMycSemaphore.TestNotify_ReturnValueAndState_AfterOneNotify(InitialCount: Integer; ExpectedReturn: Boolean; ExpectedIsSet: Boolean); +var + semaphore: IMycSemaphore; + returnedValue: Boolean; +begin + semaphore := Signals.CreateSemaphore(InitialCount); + returnedValue := semaphore.Notify; + + Assert.AreEqual(ExpectedReturn, returnedValue, 'Unexpected return value after one Notify call.'); + Assert.AreEqual(ExpectedIsSet, semaphore.State.IsSet, 'Unexpected IsSet state after one Notify call.'); +end; + +procedure TTestMycSemaphore.TestNotify_DecrementsCounter_StateChanges; +var + semaphore: IMycSemaphore; +begin + semaphore := Signals.CreateSemaphore(1); + Assert.IsFalse(semaphore.State.IsSet, 'Semaphore should initially not be set.'); + semaphore.Notify; + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore should be set after Notify.'); +end; + +procedure TTestMycSemaphore.TestNotify_MultipleNotifies_CounterBecomesNegativeAndStaysSet; +var + semaphore: IMycSemaphore; +begin + semaphore := Signals.CreateSemaphore(1); + semaphore.Notify; + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore should be set after first Notify.'); + semaphore.Notify; + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore should remain set after second Notify.'); +end; + +procedure TTestMycSemaphore.TestSubscriber_Single_IsNotifiedWhenSet; +var + semaphore: IMycSemaphore; + mockSub: TMockSubscriber; + subscription: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(1); + mockSub := TMockSubscriber.Create; + subscription := semaphore.State.Subscribe(mockSub); + + Assert.IsFalse(semaphore.State.IsSet, 'Semaphore should initially not be set.'); + Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should not have been notified yet (before set).'); + + semaphore.Notify; + + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore should be set after its Notify is called.'); + Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub should have been notified once when semaphore becomes set.'); +end; + +procedure TTestMycSemaphore.TestSubscriber_Multiple_AreNotifiedWhenSet; +var + semaphore: IMycSemaphore; + mockSub1, mockSub2, mockSub3: TMockSubscriber; + sub1, sub2, sub3: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(1); + mockSub1 := TMockSubscriber.Create; + mockSub2 := TMockSubscriber.Create; + mockSub3 := TMockSubscriber.Create; + + sub1 := semaphore.State.Subscribe(mockSub1); + sub2 := semaphore.State.Subscribe(mockSub2); + sub3 := semaphore.State.Subscribe(mockSub3); + + semaphore.Notify; + + Assert.AreEqual(1, mockSub1.NotifyCount, 'MockSub1 notification count mismatch.'); + Assert.AreEqual(1, mockSub2.NotifyCount, 'MockSub2 notification count mismatch.'); + Assert.AreEqual(1, mockSub3.NotifyCount, 'MockSub3 notification count mismatch.'); +end; + +procedure TTestMycSemaphore.TestSubscriber_NotNotified_BeforeSet; +var + semaphore: IMycSemaphore; + mockSub: TMockSubscriber; + subscription: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(2); + mockSub := TMockSubscriber.Create; + subscription := semaphore.State.Subscribe(mockSub); + + Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should not be notified upon subscription if semaphore is not set.'); + semaphore.Notify; // Count becomes 1, still not set for subscriber notification (unless Subscribe was changed to notify if count becomes 0 internally). + Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should still not be notified as semaphore is not yet set for subscribers.'); +end; + +procedure TTestMycSemaphore.TestSubscriber_NotifiedOnlyOnce_AfterFinalization; +var + semaphore: IMycSemaphore; + mockSub: TMockSubscriber; + subscription: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(1); + mockSub := TMockSubscriber.Create; + subscription := semaphore.State.Subscribe(mockSub); + + semaphore.Notify; // Sets semaphore, notifies MockSub. Assumed TMycSignal's list is finalized by this action. + Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub notify count should be 1 after first signal set.'); + + semaphore.Notify; // Trigger again. + Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub notify count should remain 1 after re-triggering finalized signal.'); +end; + +procedure TTestMycSemaphore.TestSubscribe_AfterFinalization_SubscriberNotNotified; +var + semaphore: IMycSemaphore; + mockSub1, mockSub2: TMockSubscriber; + subscription1, subscription2: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(1); + + mockSub1 := TMockSubscriber.Create; + subscription1 := semaphore.State.Subscribe(mockSub1); + + semaphore.Notify; // Triggers semaphore. It becomes set. + // Its TMycSignal part notifies mockSub1. + // And its TMycSignal part is assumed to be finalized. + + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore should be set after initial trigger.'); + Assert.AreEqual(1, mockSub1.NotifyCount, 'MockSub1 should have been notified once.'); + + // Attempt to subscribe mockSub2 after the signal (TMycSignal part) has been finalized. + mockSub2 := TMockSubscriber.Create; + subscription2 := semaphore.State.Subscribe(mockSub2); + + // Current assertion in user-provided code expects mockSub2 to be notified. + // This implies that Myc.Signals.Subscribe has specific logic to achieve this, + // even if the underlying TMycNotifyList might have been finalized for mockSub1's notification. + Assert.AreEqual(1, mockSub2.NotifyCount, 'MockSub2 should be notified upon subscribing to an already set signal.'); + + semaphore.Notify; // Re-triggering the semaphore's count decrement. + Assert.AreEqual(1, mockSub1.NotifyCount, 'MockSub1 notify count should remain 1 (not notified again).'); + // Current assertion in user-provided code expects mockSub2's count to also remain 1. + Assert.AreEqual(1, mockSub2.NotifyCount, 'MockSub2 notify count should remain 1 after re-triggering.'); +end; + +procedure TTestMycSemaphore.TestChaining_A_Notifies_B; +var + semaA, semaB: IMycSemaphore; + mockSubForB: TMockSubscriber; + subHandle_B_listens_A, subHandle_Mock_listens_B: TMycSubscription; +begin + semaA := Signals.CreateSemaphore(1); + semaB := Signals.CreateSemaphore(1); + mockSubForB := TMockSubscriber.Create; + + subHandle_Mock_listens_B := semaB.State.Subscribe(mockSubForB); + subHandle_B_listens_A := semaA.State.Subscribe(semaB); + + Assert.IsFalse(semaA.State.IsSet, 'SemaA initially not set.'); + Assert.IsFalse(semaB.State.IsSet, 'SemaB initially not set.'); + Assert.AreEqual(0, mockSubForB.NotifyCount, 'MockSubForB initially not notified.'); + + semaA.Notify; + + Assert.IsTrue(semaA.State.IsSet, 'SemaA should be set after notify.'); + Assert.IsTrue(semaB.State.IsSet, 'SemaB should be set after being notified by SemaA.'); + Assert.AreEqual(1, mockSubForB.NotifyCount, 'MockSubForB should have been notified by SemaB.'); +end; + +procedure TTestMycSemaphore.TestCascade_A_Notifies_B_Notifies_C; +var + semaA, semaB, semaC: IMycSemaphore; + mockSubForC: TMockSubscriber; + sub_Mock_C, sub_C_B, sub_B_A: TMycSubscription; +begin + semaA := Signals.CreateSemaphore(1); + semaB := Signals.CreateSemaphore(1); + semaC := Signals.CreateSemaphore(1); + mockSubForC := TMockSubscriber.Create; + + sub_Mock_C := semaC.State.Subscribe(mockSubForC); + sub_C_B := semaB.State.Subscribe(semaC); + sub_B_A := semaA.State.Subscribe(semaB); + + semaA.Notify; + + Assert.IsTrue(semaA.State.IsSet, 'SemaA state mismatch in cascade.'); + Assert.IsTrue(semaB.State.IsSet, 'SemaB state mismatch in cascade.'); + Assert.IsTrue(semaC.State.IsSet, 'SemaC state mismatch in cascade.'); + Assert.AreEqual(1, mockSubForC.NotifyCount, 'MockSubForC notification count mismatch in cascade.'); +end; + +procedure TTestMycSemaphore.TestChaining_B_NeedsMultipleSignalsFromA_WithSubscriberOnB; +var + semaA, semaB: IMycSemaphore; + mockSubForB: TMockSubscriber; + sub_Mock_B, sub_B_A: TMycSubscription; +begin + semaA := Signals.CreateSemaphore(2); + semaB := Signals.CreateSemaphore(1); + mockSubForB := TMockSubscriber.Create; + + sub_Mock_B := semaB.State.Subscribe(mockSubForB); + sub_B_A := semaA.State.Subscribe(semaB); + + semaA.Notify; // First trigger for SemaA + Assert.IsFalse(semaA.State.IsSet, 'SemaA should not be set after first trigger of two.'); + Assert.IsFalse(semaB.State.IsSet, 'SemaB should not have been triggered yet by SemaA.'); + Assert.AreEqual(0, mockSubForB.NotifyCount, 'MockSubForB should not have been notified yet.'); + + semaA.Notify; // Second trigger for SemaA + Assert.IsTrue(semaA.State.IsSet, 'SemaA should be set after second trigger.'); + Assert.IsTrue(semaB.State.IsSet, 'SemaB should now be set by SemaA.'); + Assert.AreEqual(1, mockSubForB.NotifyCount, 'MockSubForB should have been notified by SemaB.'); +end; + +procedure TTestMycSemaphore.TestInitiallySet_NotifiesNewSubscriberImmediately; +var + semaphore: IMycSemaphore; + mockSub: TMockSubscriber; + subscription: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(0); // Semaphore is created in 'IsSet' state but not finalized. + mockSub := TMockSubscriber.Create; + + // Assumes TMycSignal.Subscribe is modified to implement immediate notification for set signals. + subscription := semaphore.State.Subscribe(mockSub); + + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore IsSet property mismatch after creation with count 0.'); + Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub should be notified immediately upon subscribing to an already set semaphore.'); +end; + +procedure TTestMycSemaphore.TestUnsubscribe_SubscriberNotNotified; +var + semaphore: IMycSemaphore; + mockSub: TMockSubscriber; + subscription: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(1); + mockSub := TMockSubscriber.Create; + + subscription := semaphore.State.Subscribe(mockSub); + subscription := Default(TMycSubscription); // Simulates leaving scope / destruction, calls Finalize for unsubscription. + + semaphore.Notify; + + Assert.AreEqual(0, mockSub.NotifyCount, 'Unsubscribed MockSub should not be notified.'); +end; + +procedure TTestMycSemaphore.TestMultipleTriggersNoSubscriber_StateCorrectForLaterSubscriber; +var + semaphore: IMycSemaphore; + mockSub: TMockSubscriber; + subscription: TMycSubscription; +begin + semaphore := Signals.CreateSemaphore(3); + + semaphore.Notify; // Count -> 2 + semaphore.Notify; // Count -> 1 + Assert.IsFalse(semaphore.State.IsSet, 'Semaphore should not be set yet after partial triggers.'); + + mockSub := TMockSubscriber.Create; + subscription := semaphore.State.Subscribe(mockSub); + // Assumes TMycSignal.Subscribe does not notify immediately if signal is not yet set. + Assert.AreEqual(0, mockSub.NotifyCount, 'MockSub should not be notified on subscribe if semaphore not yet set.'); + + + semaphore.Notify; // Count -> 0, Semaphore becomes set and notifies mockSub + Assert.IsTrue(semaphore.State.IsSet, 'Semaphore should now be set.'); + Assert.AreEqual(1, mockSub.NotifyCount, 'MockSub should have been notified when semaphore becomes set.'); +end; + +initialization + // Optional: Register test cases if not using automatic registration. + // TDUnitX.RegisterTestFixture(TTestMycSemaphore); +end. diff --git a/Test/TestTasks.pas b/Test/TestTasks.pas new file mode 100644 index 0000000..233f691 --- /dev/null +++ b/Test/TestTasks.pas @@ -0,0 +1,323 @@ +unit TestTasks; + +interface + +uses + DUnitX.TestFramework, + System.SysUtils, + System.Classes, + System.SyncObjs, + Myc.Core.Tasks, + Myc.Signals, // For IMycSemaphore, Signals global + Myc.Core.Atomic; // If any atomic primitives are directly needed for test setup/assertion + +type + [TestFixture] + TMycTaskFactoryTests = class(TObject) + private + FFactory: IMycTaskFactory; + procedure TestProcExecute(var executed: Boolean; signal: IMycSemaphore); + public + [Setup] + procedure Setup; + [TearDown] + procedure TearDown; + + [Test] + procedure TestFactoryCreationAndTeardown; + [Test] + procedure TestCreateThreadExecutesAndSignals; + [Test] + procedure TestRunImmediateJob; + [Test] + procedure TestRunDelayedJobExecutesAfterAllNotifies; + [Test] + procedure TestRunDelayedJobDoesNotExecutePrematurely; + [Test] + procedure TestWaitForAlreadySetState; + [Test] + procedure TestThreadInfoMethods; + [Test] + procedure TestExceptionPropagationFromJob; + [Test] + procedure TestRunJobAfterFactoryTeardown; + [Test] + procedure TestWaitForInWorkerThreadRaisesException; + end; + +implementation + +{ TMycTaskFactoryTests } + +procedure TMycTaskFactoryTests.Setup; +begin + FFactory := TMycTaskFactory.Create; // Creates the factory instance +end; + +procedure TMycTaskFactoryTests.TearDown; +begin + if Assigned(FFactory) then + begin + FFactory.Teardown; // Explicitly teardown + FFactory := nil; + end; +end; + +procedure TMycTaskFactoryTests.TestProcExecute(var executed: Boolean; signal: IMycSemaphore); +begin + executed := True; + if Assigned(signal) then + signal.Notify; +end; + +procedure TMycTaskFactoryTests.TestFactoryCreationAndTeardown; +var + threadCount: Integer; +begin + Assert.IsNotNull(FFactory, 'Factory should be created'); + threadCount := FFactory.ThreadCount; + Assert.AreEqual(TThread.ProcessorCount, threadCount, 'Thread count should match processor count'); + // Teardown is implicitly tested by the [TearDown] method +end; + +procedure TMycTaskFactoryTests.TestCreateThreadExecutesAndSignals; +var + executed: Boolean; + threadState: IMycState; + procWrapper: TProc; +begin + executed := False; + // Wrap the procedure to match TProc signature if TestProcExecute is used, + // or define an anonymous procedure directly. + procWrapper := procedure + begin + executed := True; + end; + + threadState := FFactory.CreateThread(procWrapper); + Assert.IsNotNull(threadState, 'CreateThread should return a valid state object'); + + FFactory.WaitFor(threadState); // Wait for the thread to complete + Assert.IsTrue(executed, 'Procedure in created thread should have executed'); +end; + +procedure TMycTaskFactoryTests.TestRunImmediateJob; +var + jobExecuted: Boolean; + jobCompletedSignal: IMycSemaphore; +begin + jobExecuted := False; + jobCompletedSignal := Signals.CreateSemaphore(1); // Semaphore to signal job completion + + FFactory.Run(0, // StartCount = 0 for immediate execution + procedure + begin + jobExecuted := True; + jobCompletedSignal.Notify; // Signal completion + end); + + FFactory.WaitFor(jobCompletedSignal.State); // Wait for the job to complete + Assert.IsTrue(jobExecuted, 'Immediate job should have executed'); +end; + +procedure TMycTaskFactoryTests.TestRunDelayedJobExecutesAfterAllNotifies; +var + jobExecuted: Boolean; + jobCompletedSignal: IMycSemaphore; + subscriber: IMycSubscriber; + startCount: Integer; +begin + jobExecuted := False; + startCount := 2; + jobCompletedSignal := Signals.CreateSemaphore(1); + + subscriber := FFactory.Run(startCount, + procedure + begin + jobExecuted := True; + jobCompletedSignal.Notify; + end); + + Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job'); + // Corrected line: + Assert.AreNotEqual(Signals.Null, subscriber, 'Subscriber should not be Signals.Null for StartCount > 0'); + + subscriber.Notify; // First notification + subscriber.Notify; // Second notification, job should now be enqueued + + FFactory.WaitFor(jobCompletedSignal.State); // Wait for job completion + Assert.IsTrue(jobExecuted, 'Delayed job should execute after all notifications'); +end; + +procedure TMycTaskFactoryTests.TestRunDelayedJobDoesNotExecutePrematurely; +var + jobExecuted: Boolean; + jobCompletedSignal: IMycSemaphore; // Not strictly needed here as job shouldn't run + subscriber: IMycSubscriber; + startCount: Integer; + dummyWaitSignal: IMycSemaphore; +begin + jobExecuted := False; + startCount := 3; + jobCompletedSignal := Signals.CreateSemaphore(1); // For the job, if it were to run + dummyWaitSignal := Signals.CreateSemaphore(1); + + + subscriber := FFactory.Run(startCount, + procedure + begin + jobExecuted := True; + jobCompletedSignal.Notify; + end); + + Assert.IsNotNull(subscriber, 'Run should return a subscriber for delayed job_P2'); + + subscriber.Notify; // First notification + Assert.IsFalse(jobExecuted, 'Job should not execute after one notification_P2'); + + subscriber.Notify; // Second notification + Assert.IsFalse(jobExecuted, 'Job should not execute after two notifications_P2'); + + // To give a very brief moment for any potential async execution attempt. + // This is not foolproof for catching race conditions but can help. + // A more robust way is harder without specific test hooks in the factory. + FFactory.Run(0, procedure begin dummyWaitSignal.Notify; end); + FFactory.WaitFor(dummyWaitSignal.State); // Ensures task queue is processed slightly + + Assert.IsFalse(jobExecuted, 'Job should still not have executed before third notify_P2'); +end; + + +procedure TMycTaskFactoryTests.TestWaitForAlreadySetState; +var + alreadySetSignal: IMycSemaphore; + startTime, endTime: Cardinal; +begin + alreadySetSignal := Signals.CreateSemaphore(0); // Count = 0 means it's already set + Assert.IsTrue(alreadySetSignal.State.IsSet, 'Signal should be initially set'); + + startTime := TThread.GetTickCount; + FFactory.WaitFor(alreadySetSignal.State); // Should return immediately + endTime := TThread.GetTickCount; + + // Check that WaitFor didn't block for a significant time + // Allow a small delta for processing, e.g., less than 50ms + Assert.IsTrue((endTime - startTime) < 50, 'WaitFor on an already set state should not block significantly'); +end; + +procedure TMycTaskFactoryTests.TestThreadInfoMethods; +var + inMainThreadInJob: Boolean; + inWorkerThreadInJob: Boolean; + jobDoneSignal: IMycSemaphore; +begin + inMainThreadInJob := True; // Default to a state that would fail the assert + inWorkerThreadInJob := False; // Default to a state that would fail the assert + jobDoneSignal := Signals.CreateSemaphore(1); + + 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, + procedure + begin + inMainThreadInJob := FFactory.InMainThread; + inWorkerThreadInJob := FFactory.InWorkerThread; + jobDoneSignal.Notify; + end); + + FFactory.WaitFor(jobDoneSignal.State); // Wait for the job to complete and set flags + + Assert.IsFalse(inMainThreadInJob, 'Job executed by factory should not be in main thread'); + Assert.IsTrue(inWorkerThreadInJob, 'Job executed by factory should be in a worker thread'); +end; + +procedure TMycTaskFactoryTests.TestExceptionPropagationFromJob; +var + jobDoneSignal: IMycSemaphore; // To ensure job has a chance to run +begin + jobDoneSignal := Signals.CreateSemaphore(1); + + FFactory.Run(0, + procedure + var + dummy: Integer; // To avoid hint W1035 if no other code is in the proc + begin + dummy := 0; // Assign to avoid compiler warning if variable is unused + // Raise the specific nested exception type + raise TMycTaskFactory.ETaskException.Create('Test Exception From Job'); + // jobDoneSignal.Notify; // This line will not be reached + end); + + // Give a brief moment for the job to be scheduled and potentially raise the exception. + // Run another dummy job to help process the queue and ensure the exception is set. + FFactory.Run(0, procedure begin jobDoneSignal.Notify; end); + FFactory.WaitFor(jobDoneSignal.State); + + Assert.WillRaise( + procedure + begin + (FFactory as TMycTaskFactory).HandleException; // This should re-raise the exception + end, + TMycTaskFactory.ETaskException, // Correctly qualified exception type + 'HandleException should re-raise the exception from the job' + ); + + // Verify FException is cleared after being handled + var noExceptionRaised: Boolean := True; + try + (FFactory as TMycTaskFactory).HandleException; // Should not raise anything now + except + noExceptionRaised := False; + end; + Assert.IsTrue(noExceptionRaised, 'FException should be cleared after HandleException'); +end; + +procedure TMycTaskFactoryTests.TestRunJobAfterFactoryTeardown; +begin + FFactory.Teardown; // Explicitly teardown the factory + + Assert.WillRaise( + procedure + begin + FFactory.Run(0, procedure begin end); // Attempt to run a job + end, + TMycTaskFactory.ETaskException, // Expecting the factory's specific exception type + 'Running a job on a torn-down factory should raise ETaskException' + ); +end; + +procedure TMycTaskFactoryTests.TestWaitForInWorkerThreadRaisesException; +var + exceptionCaughtInJob: Boolean; + jobDoneSignal: IMycSemaphore; + dummyStateToWaitFor: IMycState; +begin + exceptionCaughtInJob := False; + jobDoneSignal := Signals.CreateSemaphore(1); + dummyStateToWaitFor := Signals.CreateSemaphore(1).State; // A state that will never be set by this test logic + + FFactory.Run(0, + procedure + begin + try + FFactory.WaitFor(dummyStateToWaitFor); // This should raise ETaskException + except + on E: TMycTaskFactory.ETaskException do + begin + // Check message if needed: Assert.AreEqual('Waiting not allowed within tasks', E.Message); + exceptionCaughtInJob := True; + end; + // Else: Unexpected exception, test will fail by jobDoneSignal not being set or flag remaining false + end; + jobDoneSignal.Notify; // Signal completion of the job's execution path + end); + + FFactory.WaitFor(jobDoneSignal.State); // Wait for the job to finish + Assert.IsTrue(exceptionCaughtInJob, 'WaitFor called within a worker thread job should raise ETaskException'); +end; + +initialization + // Register TestFixtures + TDUnitX.RegisterTestFixture(TMycTaskFactoryTests); +end.