MycLib/Src/Myc.Core.Tasks.pas

unit Myc.Core.Tasks;

interface

uses
    System.SysUtils, System.Classes, System.SyncObjs,
    Myc.Core.Atomic, Myc.TaskManager, Myc.Signals; // Myc.Signals now requires direct use of TMycLatch static members

type
    IMycTaskFactory = interface( IMycTaskManager )
        {$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(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)
        // will be re-raised in the calling (main) thread.
        procedure Teardown;

        // Gets the number of worker threads.
        property ThreadCount: Integer read GetThreadCount;
    end;

    TMycTaskFactory = class(TInterfacedObject, IMycTaskManager, 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 System.SyncObjs.TSemaphore for WaitFor
        FWorkGate: TSemaphore; // System.SyncObjs.TSemaphore to signal available work
        FWorkStack: TMycAtomicStack<TProc>; // Stack of pending jobs
        FWorkThreads: TArray<TThread>; // Array of worker threads
        procedure WorkerThread;
    class var
        FTaskManagerLock: Integer;

    protected
        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;
        procedure EnqueueJob(const Job: TProc);
        function Run(Job: TProc): IMycSubscriber;
        function CreateTask(const Gate: IMycState; const Proc: TProc): TMycSubscription;
        procedure WaitFor(State: IMycState);
        procedure Teardown;
        function InMainThread: Boolean;
        function InWorkerThread: Boolean;

        // Initialize global TaskManager
        class procedure AquireTaskManager;
        class procedure ReleaseTaskManager;
    end;

implementation

uses
    Myc.Core.Signals;

type
    TMycPendingJob = class(TInterfacedObject, IMycSubscriber)
    private
        [volatile]
        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(OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc);
        property Owner: TMycTaskFactory read FOwner;
    end;

    TMycTaskWait = class sealed(TInterfacedObject, IMycSubscriber)
    private
        [volatile]
        FSemaphore: TSemaphore; // System.SyncObjs.TSemaphore to be released on notification

    protected
        function Notify: Boolean;

    public
        constructor Create(SemaphoreToSignal: TSemaphore);
    end;

{ TMycTaskFactory }

constructor TMycTaskFactory.Create;
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 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].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}
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;
    {$WARN SYMBOL_PLATFORM ON}
    {$ENDIF MSWINDOWS}

    if DbgName <> '' then
        Result.NameThreadForDebugging(DbgName); // Set thread name for debugging
end;

function TMycTaskFactory.CreateTask(const Gate: IMycState; const Proc: TProc):
    TMycSubscription;
begin
    if not Assigned(Gate) or Gate.IsSet then
    begin
        EnqueueJob( Proc );
    end
    else
    begin
        // The job will run when Gate notifies the subscriber returned by Run.
        Result := Gate.Subscribe( Run( Proc ) );
    end;
end;

function TMycTaskFactory.CreateThread(const Proc: TProc): IMycState;
var
    res: IMycLatch;
    capturedProc: TProc;
begin
    // Create a latch that will be signaled when the proc finishes
    res := TMycLatch.CreateLatch(1); // Changed to use direct static call on TMycLatch

    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 via the latch's Notify method
            end;
        end).Start;

    // Return the state interface of the latch
    exit(res.State);
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
        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 Exception(ex);
    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;

class procedure TMycTaskFactory.AquireTaskManager;
begin
    if not Assigned(TaskManager) then
        TaskManager := TMycTaskFactory.Create;
    inc( FTaskManagerLock );
end;

class procedure TMycTaskFactory.ReleaseTaskManager;
begin
    dec( FTaskManagerLock );
    if FTaskManagerLock = 0 then
        TaskManager := nil;
end;

function TMycTaskFactory.Run(Job: TProc): IMycSubscriber;
begin
    if FTerminated <> 0 then
        raise ETaskException.Create('Task factory terminated');

    if not Assigned(Job) then
        exit(TMycLatch.Null); // Already correct, uses direct static property on TMycLatch

    // Create a pending job
    Result := TMycPendingJob.Create(Self, Job);
end;

procedure TMycTaskFactory.Teardown;
var
    i: Integer;
    s: TSemaphore; // This is System.SyncObjs.TSemaphore from FWaitSemaphores
    job: TProc;
begin
    HandleException; // Process any pending exception before shutdown

    if TInterlocked.Exchange(FTerminated, 1) <> 0 then
        exit; // Ensure teardown runs only once

    for i := 0 to High(FWorkThreads) do
        FWorkGate.Release;

    TSpinWait.SpinUntil(
        function: Boolean
        begin
            Result := FThreadsRunning = 0;
        end);

    Assert(FThreadsRunning = 0);

    repeat
        job := FWorkStack.Pop();
    until not Assigned(job);

    s := FWaitSemaphores.Pop;
    while s <> nil do
    begin
        s.Free;
        s := FWaitSemaphores.Pop;
    end;
end;

procedure TMycTaskFactory.WaitFor(State: IMycState);
var
    lock: TSemaphore; // This is System.SyncObjs.TSemaphore
begin
    if State.IsSet then
        exit;

    if InWorkerThread then
        raise ETaskException.Create('Waiting not allowed within tasks');

    try
        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
        HandleException;
    end;
end;

procedure TMycTaskFactory.ExecuteJob;
var
    job: TProc;
begin
    job := FWorkStack.Pop();

    if Assigned(job) then
    begin
        try
            job();
        except
            if FException = nil then
                FException := AcquireExceptionObject;
        end;
    end;
end;

{ TMycPendingJob }

constructor TMycPendingJob.Create(OwnerTaskFactory: TMycTaskFactory; const
    JobProc: TProc);
begin
    inherited Create;
    FOwner := OwnerTaskFactory;
    FJob := JobProc;
end;

function TMycPendingJob.Notify: Boolean;
var
    P: TProc;
begin
    PPointer(@P)^ := TInterlocked.Exchange( PPointer(@FJob)^, nil );
    Result := Assigned(P);
    if Result then
    begin
        if Assigned(FOwner) then
            FOwner.EnqueueJob(P);
         P := nil;
    end;
end;

{ TMycTaskWait }

constructor TMycTaskWait.Create(SemaphoreToSignal: TSemaphore); // Parameter is System.SyncObjs.TSemaphore
begin
    inherited Create;
    FSemaphore := SemaphoreToSignal; // Field is System.SyncObjs.TSemaphore
end;

function TMycTaskWait.Notify: Boolean;
var
    s: TSemaphore; // Local var is System.SyncObjs.TSemaphore
begin
    s := TInterlocked.Exchange<TSemaphore>(FSemaphore, nil);
    if s <> nil then
        s.Release;
    Result := false;
end;

initialization
    IsMultiThread := true;
end.