MycLib/Src/Myc.Core.Tasks.pas

unit Myc.Core.Tasks;

interface

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

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

    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; // System.SyncObjs.TSemaphore 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;
    inherited Destroy;
end;

class function TMycTaskFactory.CreateAnonymousThread(const DbgName: String; const Proc: TProc): TThread;
{$IFDEF MSWINDOWS}
var
    prio: TThreadPriority;
{$ENDIF MSWINDOWS}
begin
    Result := TThread.CreateAnonymousThread(Proc);

    {$IFDEF MSWINDOWS}
    {$WARN SYMBOL_PLATFORM OFF}
    // 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.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 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(TMycLatch.Null); // Already correct, uses direct static property on TMycLatch

    if StartCount <= 0 then
    begin
        EnqueueJob(Job); // Enqueue immediately if no countdown needed
        Result := TMycLatch.Null; // Already correct
    end
    else
        // Create a pending job that waits for StartCount notifications
        Result := TMyc2PendingJob.Create(StartCount, 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');

    lock := FWaitSemaphores.Pop();
    if lock = nil then
        lock := TSemaphore.Create(nil, 0, 1, '');
    try
        var subscription := State.Subscribe(TMyc2TaskWait.Create(lock));
        lock.Acquire;
    finally
        FWaitSemaphores.Push(lock);
    end;

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

{ TMyc2PendingJob }

constructor TMyc2PendingJob.Create(StartCountValue: Integer; OwnerTaskFactory: TMycTaskFactory; const JobProc: TProc);
begin
    inherited Create;
    Assert(StartCountValue > 0);
    FCount := StartCountValue;
    FOwner := OwnerTaskFactory;
    FJob := JobProc;
end;

function TMyc2PendingJob.Notify: Boolean;
var
    newCount: Integer;
begin
    newCount := TInterlocked.Decrement(FCount);
    if newCount < -MaxInt div 2 then
        TInterlocked.Increment(FCount);

    if newCount = 0 then
    begin
        if Assigned(FOwner) and Assigned(FJob) then
        begin
            FOwner.EnqueueJob(FJob);
            FJob := nil;
        end;
    end;
    Result := newCount > 0;
end;

{ TMyc2TaskWait }

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

function TMyc2TaskWait.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.