MycLib/Src/Myc.TaskManager.pas

unit Myc.TaskManager;

interface

uses
    System.SysUtils,
    Myc.Signals,
    System.Contnrs;

type
    TTaskManager = record
        type
            ITaskManager = interface
                procedure CreateThread(const Proc: TProc);
                procedure RunTask(const Gate: TState; const Proc: TProc);
                procedure WaitFor(const State: TState);
                function InWorkerThread: Boolean;
            end;
    private
        FTaskManager: ITaskManager;

    public
        constructor Create(const ATaskManager: ITaskManager);
        class operator Implicit(const A: ITaskManager): TTaskManager; overload;
        class operator Implicit(const A: TTaskManager): ITaskManager; overload;

        // Creates and starts a new, independent thread executing Proc.
        // Returns a State to await thread completion.
        function CreateThread(const Proc: TProc): TState;

        // Synchronize from independent thread (created by CreateThread). This is a blocking function and it will
        // raise an exception, if it is called called from a task (created by RunTask).
        procedure Synchronize(const Proc: TProc);

        // true, if called from a task (created by RunTask)
        function InTask: Boolean;

        // Run a task when the gate is opened.
        // Returns a State to await thread completion.
        // No blocking operations are allowed whithin tasks!
        function RunTask(const Proc: TFunc<TState>): TState; overload;
        function RunTask(const Gate: TState; const Proc: TFunc<TState>): TState; overload;
        procedure RunTask(const Gate: TState; const Proc: TProc); overload;
        procedure RunTask(const Proc: TProc); overload;

        class function RunSequence(const Gate: TState; First, Count: Integer; const Proc: TFunc<Integer, TState>): TState; static;

        // Waits for the operation associated with State to complete.
        // Must not be called from a task of this factory.
        // After waiting, or if the state is already set, any first stored exception
        // from any worker thread of this factory will be re-raised in the calling (main) thread.
        // Raises ETaskException if called from within a worker thread.
        procedure WaitFor(const State: TState); inline;
    end;

    TQueue = record
    strict private
        [volatile]
        FQueueLock: Integer;
        FState: TState;
    public
        function Enqueue(const Func: TFunc<TState>): TState;
        class operator Initialize(out Dest: TQueue);
    end;

var
    TaskManager: TTaskManager;

procedure SetupTaskManagerMock;

implementation

uses
    System.Classes,
    Myc.Core.Tasks;

type
    TMycExecMock = class(TInterfacedObject, TSignal.ISubscriber)
    private
        FProc: TProc;
    public
        constructor Create(const AProc: TProc);
        function Notify: Boolean;
    end;

    TMycTaskManagerMock = class(TInterfacedObject, TTaskManager.ITaskManager)
    public
        constructor Create;
        procedure RunTask(const Gate: TState; const Proc: TProc);
        procedure WaitFor(const State: TState);
        procedure CreateThread(const Proc: TProc);
        function InWorkerThread: Boolean;
    end;

{ TMycExecMock }

constructor TMycExecMock.Create(const AProc: TProc);
begin
    inherited Create;
    FProc := AProc;
end;

function TMycExecMock.Notify: Boolean;
begin
    Result := Assigned(FProc);
    if Result then
    begin
        FProc();
        FProc := nil;
    end;
end;

{ TMycTaskManagerMock }

constructor TMycTaskManagerMock.Create;
begin
    inherited Create;
end;

procedure TMycTaskManagerMock.CreateThread(const Proc: TProc);
begin
    Proc();
end;

function TMycTaskManagerMock.InWorkerThread: Boolean;
begin
    Result := false;
end;

procedure TMycTaskManagerMock.RunTask(const Gate: TState; const Proc: TProc);
begin
    Gate.Signal.Subscribe(TMycExecMock.Create(Proc));
end;

procedure TMycTaskManagerMock.WaitFor(const State: TState);
begin
    Assert(State.IsSet);
end;

procedure SetupTaskManagerMock;
begin
    TaskManager.Create(TMycTaskManagerMock.Create);
end;

constructor TTaskManager.Create(const ATaskManager: ITaskManager);
begin
    FTaskManager := ATaskManager;
end;

function TTaskManager.CreateThread(const Proc: TProc): TState;
begin
    // Create a latch that will be signaled when the proc finishes
    var done := TLatch.CreateLatch(1);
    Result := done.State;

    var cProc := Proc; // Capture Proc for the anonymous method
    FTaskManager.CreateThread(
        procedure
        begin
            try
                cProc(); // Execute the provided procedure
            finally
                cProc := nil; // Clear the captured proc
                done.Notify; // Signal completion via the latch's Notify method
            end;
        end
    );
end;

function TTaskManager.InTask: Boolean;
begin
    Result := FTaskManager.InWorkerThread;
end;

procedure TTaskManager.Synchronize(const Proc: TProc);
begin
    if InTask then
        raise Exception.Create('Synchronization in tasks not allowed');

    var cProc := Proc;
    if Assigned(cProc) then
        TThread.Synchronize(
            nil,
            procedure
            begin
                if Assigned(cProc) then
                begin
                    cProc();
                    cProc := nil;
                end;
            end
        );
end;

class function TTaskManager.RunSequence(const Gate: TState; First, Count: Integer; const Proc: TFunc<Integer, TState>): TState;
begin
    if First >= Count then
        exit;

    var cProc: TFunc<Integer, TState> := Proc;
    Result := TaskManager.RunTask(Gate, function: TState begin Result := RunSequence(Proc(First), 1 + First, Count, cProc); end);
end;

function TTaskManager.RunTask(const Gate: TState; const Proc: TFunc<TState>): TState;
begin
    var cProc: TFunc<TState> := Proc;
    if not Assigned(@cProc) then
        exit(TState.Null);

    var Done := TLatch.CreateLatch(1);
    Result := Done.State;

    FTaskManager.RunTask(
        Gate,
        procedure
        begin
            try
                cProc.Signal.Subscribe(Done);
                cProc := nil;
            except
                Done.Notify;
                raise;
            end;
        end
    );
end;

function TTaskManager.RunTask(const Proc: TFunc<TState>): TState;
begin
    Result := RunTask(TState.Null, Proc);
end;

procedure TTaskManager.RunTask(const Proc: TProc);
begin
    RunTask(TState.Null, Proc);
end;

procedure TTaskManager.RunTask(const Gate: TState; const Proc: TProc);
begin
    FTaskManager.RunTask(Gate, Proc);
end;

procedure TTaskManager.WaitFor(const State: TState);
begin
    FTaskManager.WaitFor(State);
end;

class operator TTaskManager.Implicit(const A: ITaskManager): TTaskManager;
begin
    Result.Create(A);
end;

class operator TTaskManager.Implicit(const A: TTaskManager): ITaskManager;
begin
    Result := A.FTaskManager;
end;

function TQueue.Enqueue(const Func: TFunc<TState>): TState;
begin
    var state := TaskManager.RunTask(FState, Func);

    repeat
        if FQueueLock > 0 then
        begin
            YieldProcessor;
            continue;
        end;
    until AtomicCmpExchange(FQueueLock, 1, 0) = 0;

    try
        Result := FState;
        FState := state;
    finally
        AtomicDecrement(FQueueLock);
    end;
end;

class operator TQueue.Initialize(out Dest: TQueue);
begin
    Dest.FQueueLock := 0;
end;

end.