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;

type
    IThreadPool = interface(TTaskManager.ITaskManager)
        {$region 'property access'}
        // Retrieves the number of worker threads.
        function GetThreadCount: Integer;
        {$endregion}

        // 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;

        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 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, TTaskManager.ITaskManager, IThreadPool)
        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 constructor CreateClass;
    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;
        procedure CreateThread(const Proc: TProc);
        procedure HandleException;
        procedure EnqueueJob(const Job: TProc);
        procedure RunTask(const Gate: TState; const Proc: TProc);
        procedure WaitFor(const State: TState);
        procedure Teardown;
        function InMainThread: Boolean;
        function InWorkerThread: Boolean;

        property ThreadsRunning: Integer read FThreadsRunning;
        property WorkThreads: TArray<TThread> read FWorkThreads;
    end;

implementation

type
    TMycPendingJob = class(TInterfacedObject, TSignal.ISubscriber)
    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, TSignal.ISubscriber)
    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;

    FWaitSemaphores.Push(TSemaphore.Create(nil, 0, 1, ''));
end;

class constructor TMycTaskFactory.CreateClass;
begin
end;

destructor TMycTaskFactory.Destroy;
begin
    Teardown; // Perform cleanup and stop threads

    for var i := High(FWorkThreads) downto 0 do
        FWorkThreads[i].Free;

    if Assigned(FException) then
        FException.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;

procedure TMycTaskFactory.RunTask(const Gate: TState; const Proc: TProc);
begin
    if not Assigned(Proc) then
        exit;

    if FTerminated <> 0 then
        raise ETaskException.Create('Task factory terminated');

    if Gate.IsSet then
    begin
        EnqueueJob(Proc);
    end
    else
    begin
        // Create a pending job
        Gate.Signal.Subscribe(TMycPendingJob.Create(Self, Proc));
    end;
end;

procedure TMycTaskFactory.CreateThread(const Proc: TProc);
begin
    var cProc := Proc; // Capture Proc for the anonymous method
    CreateAnonymousThread(
            'Thread',
            procedure
            begin
                try
                    cProc(); // Execute the provided procedure
                finally
                    cProc := nil; // Clear the captured proc
                end;
            end)
        .Start;
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;

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(const State: TState);
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.Signal.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;
    TaskManager := TMycTaskFactory.Create;

end.