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<TSemaphore>; // Pool of semaphores for WaitFor
        FWorkGate: TSemaphore; // Semaphore to signal available work to worker threads
        FWorkStack: TMycAtomicStack<TProc>; // Stack of pending jobs
        FWorkThreads: TArray<TThread>; // Array of worker threads
        procedure WorkerThread;

    protected
        procedure EnqueueJob(const Job: TProc);
        procedure ExecuteJob;
        function GetThreadCount: Integer;
        property WaitSemaphores: TMycAtomicStack<TSemaphore> 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<TSemaphore>(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.