Tasks.pas

unit Tasks;
{
  г======================================================================¬
  ¦  Tasks.pas v2 — Task Parallel Library for Delphi 7                  ¦
  ¦======================================================================¦
  ¦  NEW in v2:                                                          ¦
  ¦   • TCancellationToken  — cooperative cancellation flag             ¦
  ¦   • TCancellationTokenSource — manages the token                    ¦
  ¦   • TTaskCancelProc/Method — callbacks that receive the token       ¦
  ¦   • ETaskCancelled — transitions task to tsCancelled (not tsFailed) ¦
  ¦   • ITask.Cancel          — manual cancellation                     ¦
  ¦   • ITask.CancelAfter(ms) — timeout cancellation                    ¦
  ¦======================================================================¦
  ¦  COOPERATIVE CANCELLATION PRINCIPLE                                  ¦
  ¦                                                                      ¦
  ¦  Windows threads cannot be forcibly stopped without risk            ¦
  ¦  of data corruption. Therefore:                                      ¦
  ¦                                                                      ¦
  ¦  • Primitive tasks (TTaskProc) — Cancel only prevents               ¦
  ¦    start (if the task is still in the queue)                        ¦
  ¦                                                                      ¦
  ¦  • Tasks with TTaskCancelProc — receive a token and MUST            ¦
  ¦    periodically call AToken.ThrowIfCancellationRequested (or check  ¦
  ¦    AToken.IsCancellationRequested)                                  ¦
  L======================================================================-
}

interface

uses
  Windows, SysUtils, Classes;

type


   { ------------------------------------------------------------------ }
  {  Continuation options                                               }
  { ------------------------------------------------------------------ }
  TContinuationOption = (
    coAlways,          // always (default)
    coOnlyOnSuccess,   // only on tsCompleted
    coOnlyOnFailed,    // only on tsFailed
    coOnMainThread     // on main thread (Synchronize)
  );
  TContinuationOptions = set of TContinuationOption;

const
  TASK_POOL_SIZE_AUTO  = 0;
  TASK_PUMP_INTERVAL   = 15;   // ms: CheckSynchronize interval during Wait


  coDefault       : TContinuationOptions = [coAlways];
  coSuccessOnly   : TContinuationOptions = [coOnlyOnSuccess];
  coFailedOnly    : TContinuationOptions = [coOnlyOnFailed];
  coMainAlways    : TContinuationOptions = [coAlways,        coOnMainThread];
  coMainOnSuccess : TContinuationOptions = [coOnlyOnSuccess, coOnMainThread];
  coMainOnFailed  : TContinuationOptions = [coOnlyOnFailed,  coOnMainThread];
type
  { ------------------------------------------------------------------ }
  {  Forward                                                            }
  { ------------------------------------------------------------------ }
  TCancellationToken = class;

  { ------------------------------------------------------------------ }
  {  Cooperative cancellation exception                                 }
  {  Throw via AToken.ThrowIfCancellationRequested                      }
  {  The task will catch it and go to tsCancelled (not tsFailed)        }
  { ------------------------------------------------------------------ }
  ETaskCancelled = class(Exception)
  public
    constructor Create;
  end;

  { ------------------------------------------------------------------ }
  {  Task state                                                         }
  { ------------------------------------------------------------------ }
  TTaskState = (
    tsPending,    // waiting in the pool queue
    tsRunning,    // executing
    tsCompleted,  // completed successfully
    tsFailed,     // completed with an exception
    tsCancelled   // cancelled (before start or during execution)
  );






  { ------------------------------------------------------------------ }
  {  Callback types                                                    }
  { ------------------------------------------------------------------ }
  { Standard (no internal cancellation support) }
  TTaskProc         = procedure(AData: Pointer);
  TTaskMethod       = procedure(AData: Pointer) of object;
  { Cooperative cancellation: token is passed inside }
  TTaskCancelProc   = procedure(AData: Pointer; AToken: TCancellationToken);
  TTaskCancelMethod = procedure(AData: Pointer; AToken: TCancellationToken) of object;

  { ------------------------------------------------------------------ }
  {  ITask                                                              }
  { ------------------------------------------------------------------ }
  ITask = interface
    ['{9A3F7B2E-4C51-4D8A-B6E3-12F09A8B4C7D}']
    procedure Wait;
    function  ContinueWith(AProc  : TTaskProc;
                           AData  : Pointer;
                           AOptions: TContinuationOptions): ITask; overload;
    function  ContinueWith(AMethod: TTaskMethod;
                           AData  : Pointer;
                           AOptions: TContinuationOptions): ITask; overload;
    { --- Cancellation --- }
    procedure Cancel;
    procedure CancelAfter(ATimeoutMs: Cardinal);
    { --- WinAPI event for WaitAll/WaitAny --- }
    function  GetWaitHandle: THandle;
    { --- State --- }
    function  GetState       : TTaskState;
    function  GetErrorMessage: string;
    function  GetIsCompleted : Boolean;
    function  GetIsSuccess   : Boolean;
    function  GetIsFailed    : Boolean;
    function  GetIsCancelled : Boolean;
    property  State        : TTaskState read GetState;
    property  ErrorMessage : string     read GetErrorMessage;
    property  IsCompleted  : Boolean    read GetIsCompleted;
    property  IsSuccess    : Boolean    read GetIsSuccess;
    property  IsFailed     : Boolean    read GetIsFailed;
    property  IsCancelled  : Boolean    read GetIsCancelled;
    property  WaitHandle   : THandle    read GetWaitHandle;
  end;

  TThreadPool = class; // forward

  { ================================================================== }
  {  TCancellationToken                                                 }
  {  Thread-safe cancellation flag with optional auto-timer             }
  { ================================================================== }
  TCancellationToken = class
  private
    FCancelled   : LongBool;     // read/written via Interlocked
    FTimerThread : TObject;      // TCancelTimerThread (forward-compat)
    FCS          : TRTLCriticalSection;  // protects FTimerThread
  public
    constructor Create;
    destructor  Destroy; override;

    { Was cancellation requested? }
    function  IsCancellationRequested: Boolean;
    { Throws ETaskCancelled if cancellation was requested }
    procedure ThrowIfCancellationRequested;
    { Immediate cancellation }
    procedure Cancel;
    { Cancellation after ATimeoutMs milliseconds (replaces previous timer) }
    procedure CancelAfter(ATimeoutMs: Cardinal);

    { Singleton "never cancelled" — do not free! }
    class function None: TCancellationToken;
  end;

  { ================================================================== }
  {  TCancellationTokenSource                                           }
  {  C#-style: Source manages, Token observes                          }
  {                                                                     }
  {  Source := TCancellationTokenSource.Create;                        }
  {  Source.CancelAfter(3000);                                          }
  {  T := TTask.Run(@MyProc, Data, Source.Token);                      }
  {  ...                                                                }
  {  Source.Free;  // only after T.Wait !                              }
  { ================================================================== }
  TCancellationTokenSource = class
  private
    FToken: TCancellationToken;  // owns it
  public
    constructor Create;
    constructor CreateWithTimeout(ATimeoutMs: Cardinal);
    destructor  Destroy; override;
    procedure Cancel;
    procedure CancelAfter(ATimeoutMs: Cardinal);
    function  GetToken: TCancellationToken;
    function  GetIsCancellationRequested: Boolean;
    property  Token                  : TCancellationToken read GetToken;
    property  IsCancellationRequested: Boolean            read GetIsCancellationRequested;
  end;

  TTask = class;

  TTasksArray=array of TTask;

  { ================================================================== }
  {  TTask                                                              }
  { ================================================================== }
  TTask = class(TInterfacedObject, ITask)
  private
    { --- Callbacks --- }
    FProc         : TTaskProc;
    FMethod       : TTaskMethod;
    FCancelProc   : TTaskCancelProc;    // NEW
    FCancelMethod : TTaskCancelMethod;  // NEW
    FData         : Pointer;
    FPool         : TThreadPool;
    { --- State --- }
    FState        : TTaskState;
    FErrorMsg     : string;
    { --- Synchronization --- }
    FDoneEvent    : THandle;
    FCS           : TRTLCriticalSection;
    { --- Continuations --- }
    FConts        : TTasksArray;
    FContOpts     : array of TContinuationOptions;
    FContCount    : Integer;
    { --- Cancellation --- }
    FToken        : TCancellationToken;  // NEW: weak reference (sometimes owned)
    FOwnsToken    : Boolean;             // NEW: True -> free in Destroy
    FCancelFlag   : LongBool;            // NEW: set via Cancel()
    { --- Marshalling to UI --- }
    FRunOnMain    : Boolean;
  private
    procedure InitFields(AData: Pointer; APool: TThreadPool);
    function  AddContinuation(ACont: TTask; AOptions: TContinuationOptions): ITask;
    function  IsExternalCancellationRequested: Boolean;  // NEW
    procedure InternalExecute;
    procedure Complete(const AErrMsg: string);
    procedure CancelSelf;
    procedure ScheduleContinuations;
    function  ShouldRunCont(AOptions: TContinuationOptions;
                            AParentState: TTaskState): Boolean;
    { ITask }
    function  GetState       : TTaskState;
    function  GetErrorMessage: string;
    function  GetIsCompleted : Boolean;
    function  GetIsSuccess   : Boolean;
    function  GetIsFailed    : Boolean;
    function  GetIsCancelled : Boolean;
    function  GetWaitHandle  : THandle;
  public
    { --- Constructors (standard) --- }
    constructor Create(AProc  : TTaskProc;
                       AData  : Pointer; APool: TThreadPool); overload;
    constructor Create(AMethod: TTaskMethod;
                       AData  : Pointer; APool: TThreadPool); overload;
    { --- Constructors: standard callback + external token ---------- }
    {  Token is NOT passed to the callback, but affects start/timeout }
    constructor Create(AProc  : TTaskProc;   AData: Pointer;
                       AToken : TCancellationToken;
                       APool  : TThreadPool); overload;
    constructor Create(AMethod: TTaskMethod; AData: Pointer;
                       AToken : TCancellationToken;
                       APool  : TThreadPool); overload;
    { --- Constructors: cancel-aware callback + token ----------------- }
    {  AOwnsToken=True -> TTask will free the token when destroyed    }
    constructor Create(ACancelProc  : TTaskCancelProc;   AData: Pointer;
                       AToken       : TCancellationToken;
                       AOwnsToken   : Boolean;
                       APool        : TThreadPool); overload;
    constructor Create(ACancelMethod: TTaskCancelMethod; AData: Pointer;
                       AToken       : TCancellationToken;
                       AOwnsToken   : Boolean;
                       APool        : TThreadPool); overload;
    destructor Destroy; override;

    { ITask }
    procedure Wait;
    function  ContinueWith(AProc  : TTaskProc;
                           AData  : Pointer;
                           AOptions: TContinuationOptions): ITask; overload;
    function  ContinueWith(AMethod: TTaskMethod;
                           AData  : Pointer;
                           AOptions: TContinuationOptions): ITask; overload;
    procedure Cancel;
    procedure CancelAfter(ATimeoutMs: Cardinal);

    property  State        : TTaskState read GetState;
    property  ErrorMessage : string     read GetErrorMessage;
    property  IsCompleted  : Boolean    read GetIsCompleted;
    property  IsSuccess    : Boolean    read GetIsSuccess;
    property  IsFailed     : Boolean    read GetIsFailed;
    property  IsCancelled  : Boolean    read GetIsCancelled;
    property  WaitHandle   : THandle    read GetWaitHandle;

    { ——— FACTORY METHODS ——— }

    { Basic }
    class function Run(AProc  : TTaskProc;
                       AData  : Pointer = nil): ITask; overload;
    class function Run(AMethod: TTaskMethod;
                       AData  : Pointer = nil): ITask; overload;
    { With custom pool }
    class function Run(AProc  : TTaskProc;
                       AData  : Pointer;
                       APool  : TThreadPool): ITask; overload;
    class function Run(AMethod: TTaskMethod;
                       AData  : Pointer;
                       APool  : TThreadPool): ITask; overload;
    { Standard callback + external token (only pre-start cancellation) }
    class function Run(AProc  : TTaskProc;
                       AData  : Pointer;
                       AToken : TCancellationToken): ITask; overload;
    class function Run(AMethod: TTaskMethod;
                       AData  : Pointer;
                       AToken : TCancellationToken): ITask; overload;
    { Cancel-aware callback + token (full cooperative cancellation) }
    class function Run(ACancelProc  : TTaskCancelProc;
                       AData        : Pointer;
                       AToken       : TCancellationToken): ITask; overload;
    class function Run(ACancelMethod: TTaskCancelMethod;
                       AData        : Pointer;
                       AToken       : TCancellationToken): ITask; overload;
    class function Run(ACancelProc  : TTaskCancelProc;
                       AData        : Pointer;
                       AToken       : TCancellationToken;
                       APool        : TThreadPool): ITask; overload;
    class function Run(ACancelMethod: TTaskCancelMethod;
                       AData        : Pointer;
                       AToken       : TCancellationToken;
                       APool        : TThreadPool): ITask; overload;

    { ——— AGGREGATES ——— }
    class procedure WaitAll(const ATasks: array of ITask);
    class function  WaitAny(const ATasks: array of ITask): Integer;
  end;

  { ================================================================== }
  {  TTaskQueue — thread-safe queue (circular buffer)                  }
  { ================================================================== }
  TTaskQueue = class
  private
    FBuf      : TTasksArray;
    FHead     : Integer;
    FTail     : Integer;
    FCapacity : Integer;
    FCount    : Integer;
    FCS       : TRTLCriticalSection;
    FSem      : THandle;
    FDone     : Boolean;
    procedure Grow;
  public
    constructor Create(ACapacity: Integer = 64);
    destructor  Destroy; override;
    procedure   Enqueue(ATask: TTask);
    function    Dequeue: TTask;
    procedure   Terminate;
  end;

  { ================================================================== }
  {  TWorkerThread                                                      }
  { ================================================================== }
  TWorkerThread = class(TThread)
  private
    FQueue    : TTaskQueue;
    FSyncTask : TTask;
    procedure DoRunOnMain;
  protected
    procedure Execute; override;
  public
    constructor Create(AQueue: TTaskQueue);
  end;

  { ================================================================== }
  {  TThreadPool                                                        }
  { ================================================================== }
  TThreadPool = class
  private
    FQueue   : TTaskQueue;
    FWorkers : array of TWorkerThread;
    FCount   : Integer;
  public
    constructor Create(AWorkerCount: Integer = TASK_POOL_SIZE_AUTO);
    destructor  Destroy; override;
    procedure   Enqueue(ATask: TTask);
    property    WorkerCount: Integer read FCount;
    class function  Default: TThreadPool;
    class procedure ReleaseDefault;
  end;

implementation

{ ======================================================
  Global state
  ====================================================== }
var
  GMainThreadID : Cardinal;
  GDefaultPool  : TThreadPool = nil;
  GPoolCS       : TRTLCriticalSection;
  GNoneToken    : TCancellationToken = nil;   // singleton "never cancelled"

{ ======================================================
  Utilities
  ====================================================== }
function IsMainThread: Boolean;
begin
  Result := GetCurrentThreadId = GMainThreadID;
end;

procedure SmartWaitSingle(AEvent: THandle);
begin
  if IsMainThread then
    while WaitForSingleObject(AEvent, TASK_PUMP_INTERVAL) = WAIT_TIMEOUT do
      CheckSynchronize
  else
    WaitForSingleObject(AEvent, INFINITE);
end;

{ ======================================================
  ETaskCancelled
  ====================================================== }
constructor ETaskCancelled.Create;
begin
  inherited Create('Task was cancelled');
end;

{ ======================================================
  TCancelTimerThread (internal)
  Waits FDelay ms; if not interrupted — cancels the token
  ====================================================== }
type
  TCancelTimerThread = class(TThread)
  private
    FToken  : TCancellationToken;
    FDelay  : Cardinal;
    FAbort  : THandle;             // early timer abort event
  protected
    procedure Execute; override;
  public
    constructor Create(AToken: TCancellationToken; ADelay: Cardinal);
    destructor  Destroy; override;
    procedure   AbortTimer;
  end;

constructor TCancelTimerThread.Create(AToken: TCancellationToken;
  ADelay: Cardinal);
begin
  FToken          := AToken;
  FDelay          := ADelay;
  FAbort          := CreateEvent(nil, False, False, nil);
  FreeOnTerminate := False;
  inherited Create(False);
end;

destructor TCancelTimerThread.Destroy;
begin
  CloseHandle(FAbort);
  inherited;
end;

procedure TCancelTimerThread.AbortTimer;
begin
  SetEvent(FAbort);
end;

procedure TCancelTimerThread.Execute;
begin
  // Wait FDelay ms; if FAbort is signaled earlier — exit quietly
  if WaitForSingleObject(FAbort, FDelay) = WAIT_TIMEOUT then
    if not Terminated then
      FToken.Cancel;
end;

{ ======================================================
  TCancellationToken
  ====================================================== }
constructor TCancellationToken.Create;
begin
  inherited Create;
  FCancelled   := False;
  FTimerThread := nil;
  InitializeCriticalSection(FCS);
end;

destructor TCancellationToken.Destroy;
var
  T: TCancelTimerThread;
begin
  // Cancel the timer under lock, wait outside
  EnterCriticalSection(FCS);
  T            := TCancelTimerThread(FTimerThread);
  FTimerThread := nil;
  LeaveCriticalSection(FCS);

  if T <> nil then
  begin
    T.AbortTimer;
    T.WaitFor;
    T.Free;
  end;

  DeleteCriticalSection(FCS);
  inherited;
end;

function TCancellationToken.IsCancellationRequested: Boolean;
begin
  // LongBool on x86 is guaranteed to be read atomically
  Result := FCancelled;
end;

procedure TCancellationToken.ThrowIfCancellationRequested;
begin
  if FCancelled then
    raise ETaskCancelled.Create;
end;

procedure TCancellationToken.Cancel;
begin
  // Interlocked write; no CS needed for a simple flag
  InterlockedExchange(Integer(FCancelled), Integer(True));
end;

procedure TCancellationToken.CancelAfter(ATimeoutMs: Cardinal);
var
  OldThread: TCancelTimerThread;
  NewThread: TCancelTimerThread;
begin
  if ATimeoutMs = 0 then
  begin
    Cancel;
    Exit;
  end;

  // Create new timer; stop the old one
  NewThread := TCancelTimerThread.Create(Self, ATimeoutMs);

  EnterCriticalSection(FCS);
  OldThread    := TCancelTimerThread(FTimerThread);
  FTimerThread := NewThread;
  LeaveCriticalSection(FCS);

  if OldThread <> nil then
  begin
    OldThread.AbortTimer;
    OldThread.WaitFor;
    OldThread.Free;
  end;
end;

class function TCancellationToken.None: TCancellationToken;
begin
  // Initialised in the initialization section — thread-safe without CS
  Result := GNoneToken;
end;

{ ======================================================
  TCancellationTokenSource
  ====================================================== }
constructor TCancellationTokenSource.Create;
begin
  inherited Create;
  FToken := TCancellationToken.Create;
end;

constructor TCancellationTokenSource.CreateWithTimeout(ATimeoutMs: Cardinal);
begin
  Create;
  CancelAfter(ATimeoutMs);
end;

destructor TCancellationTokenSource.Destroy;
begin
  FToken.Free;
  inherited;
end;

procedure TCancellationTokenSource.Cancel;
begin
  FToken.Cancel;
end;

procedure TCancellationTokenSource.CancelAfter(ATimeoutMs: Cardinal);
begin
  FToken.CancelAfter(ATimeoutMs);
end;

function TCancellationTokenSource.GetToken: TCancellationToken;
begin
  Result := FToken;
end;

function TCancellationTokenSource.GetIsCancellationRequested: Boolean;
begin
  Result := FToken.IsCancellationRequested;
end;

{ ======================================================
  TTaskQueue
  ====================================================== }
constructor TTaskQueue.Create(ACapacity: Integer);
begin
  inherited Create;
  FCapacity := ACapacity;
  SetLength(FBuf, FCapacity);
  FHead  := 0;
  FTail  := 0;
  FCount := 0;
  FDone  := False;
  InitializeCriticalSection(FCS);
  FSem := CreateSemaphore(nil, 0, MaxInt, nil);
end;

destructor TTaskQueue.Destroy;
begin
  CloseHandle(FSem);
  DeleteCriticalSection(FCS);
  inherited;
end;

procedure TTaskQueue.Grow;
var
  NewBuf: TTasksArray;
  NewCap: Integer;
  i     : Integer;
begin
  NewCap := FCapacity * 2;
  SetLength(NewBuf, NewCap);
  for i := 0 to FCount - 1 do
    NewBuf[i] := FBuf[(FHead + i) mod FCapacity];
  FBuf      :=  NewBuf;
  FHead     := 0;
  FTail     := FCount;
  FCapacity := NewCap;
end;

procedure TTaskQueue.Enqueue(ATask: TTask);
begin
  ATask._AddRef;
  EnterCriticalSection(FCS);
  try
    if FCount = FCapacity then Grow;
    FBuf[FTail] := ATask;
    FTail       := (FTail + 1) mod FCapacity;
    Inc(FCount);
  finally
    LeaveCriticalSection(FCS);
  end;
  ReleaseSemaphore(FSem, 1, nil);
end;

function TTaskQueue.Dequeue: TTask;
begin
  WaitForSingleObject(FSem, INFINITE);
  EnterCriticalSection(FCS);
  try
    if FDone and (FCount = 0) then
    begin
      Result := nil;
      Exit;
    end;
    if FCount > 0 then
    begin
      Result        := FBuf[FHead];
      FBuf[FHead]   := nil;
      FHead         := (FHead + 1) mod FCapacity;
      Dec(FCount);
    end
    else
      Result := nil;
  finally
    LeaveCriticalSection(FCS);
  end;
end;

procedure TTaskQueue.Terminate;
begin
  EnterCriticalSection(FCS);
  FDone := True;
  LeaveCriticalSection(FCS);
  ReleaseSemaphore(FSem, 512, nil);
end;

{ ======================================================
  TWorkerThread
  ====================================================== }
constructor TWorkerThread.Create(AQueue: TTaskQueue);
begin
  FQueue          := AQueue;
  FSyncTask       := nil;
  FreeOnTerminate := False;
  inherited Create(False);
end;

procedure TWorkerThread.DoRunOnMain;
begin
  FSyncTask.InternalExecute;
end;

procedure TWorkerThread.Execute;
var
  Task: TTask;
begin
  while not Terminated do
  begin
    Task := FQueue.Dequeue;
    if Task = nil then Break;
    try
      if Task.FRunOnMain then
      begin
        FSyncTask := Task;
        Synchronize(DoRunOnMain);
        FSyncTask := nil;
      end
      else
        Task.InternalExecute;
    finally
      Task._Release;
    end;
  end;
end;

{ ======================================================
  TThreadPool
  ====================================================== }
constructor TThreadPool.Create(AWorkerCount: Integer);
var
  SI: TSystemInfo;
  i : Integer;
begin
  inherited Create;
  if AWorkerCount <= 0 then
  begin
    GetSystemInfo(SI);
    AWorkerCount := Integer(SI.dwNumberOfProcessors) * 2;
    if AWorkerCount < 2  then AWorkerCount := 2;
    if AWorkerCount > 32 then AWorkerCount := 32;
  end;
  FCount  := AWorkerCount;
  FQueue  := TTaskQueue.Create;
  SetLength(FWorkers, FCount);
  for i := 0 to FCount - 1 do
    FWorkers[i] := TWorkerThread.Create(FQueue);
end;

destructor TThreadPool.Destroy;
var
  i: Integer;
begin
  FQueue.Terminate;
  for i := 0 to FCount - 1 do
  begin
    FWorkers[i].Terminate;
    FWorkers[i].WaitFor;
    FWorkers[i].Free;
  end;
  SetLength(FWorkers, 0);
  FQueue.Free;
  inherited;
end;

procedure TThreadPool.Enqueue(ATask: TTask);
begin
  FQueue.Enqueue(ATask);
end;

class function TThreadPool.Default: TThreadPool;
begin
  if GDefaultPool = nil then
  begin
    EnterCriticalSection(GPoolCS);
    try
      if GDefaultPool = nil then
        GDefaultPool := TThreadPool.Create;
    finally
      LeaveCriticalSection(GPoolCS);
    end;
  end;
  Result := GDefaultPool;
end;

class procedure TThreadPool.ReleaseDefault;
begin
  EnterCriticalSection(GPoolCS);
  try
    FreeAndNil(GDefaultPool);
  finally
    LeaveCriticalSection(GPoolCS);
  end;
end;

{ ======================================================
  TTask — common initialisation
  ====================================================== }
procedure TTask.InitFields(AData: Pointer; APool: TThreadPool);
begin
  FProc         := nil;
  FMethod       := nil;
  FCancelProc   := nil;
  FCancelMethod := nil;
  FData         := AData;
  FPool         := APool;
  FState        := tsPending;
  FErrorMsg     := '';
  FContCount    := 0;
  FRunOnMain    := False;
  FToken        := nil;
  FOwnsToken    := False;
  FCancelFlag   := false;
  InitializeCriticalSection(FCS);
  FDoneEvent    := CreateEvent(nil, True, False, nil);
end;

{ ======================================================
  TTask — constructors
  ====================================================== }

{ Standard (no token) }
constructor TTask.Create(AProc: TTaskProc; AData: Pointer; APool: TThreadPool);
begin
  inherited Create;
  InitFields(AData, APool);
  FProc := AProc;
end;

constructor TTask.Create(AMethod: TTaskMethod; AData: Pointer; APool: TThreadPool);
begin
  inherited Create;
  InitFields(AData, APool);
  FMethod := AMethod;
end;

{ Standard callback + external token }
constructor TTask.Create(AProc: TTaskProc; AData: Pointer;
  AToken: TCancellationToken; APool: TThreadPool);
begin
  inherited Create;
  InitFields(AData, APool);
  FProc  := AProc;
  FToken := AToken;   // do not own
end;

constructor TTask.Create(AMethod: TTaskMethod; AData: Pointer;
  AToken: TCancellationToken; APool: TThreadPool);
begin
  inherited Create;
  InitFields(AData, APool);
  FMethod := AMethod;
  FToken  := AToken;
end;

{ Cancel-aware callback + token }
constructor TTask.Create(ACancelProc: TTaskCancelProc; AData: Pointer;
  AToken: TCancellationToken; AOwnsToken: Boolean; APool: TThreadPool);
begin
  inherited Create;
  InitFields(AData, APool);
  FCancelProc := ACancelProc;
  FToken      := AToken;
  FOwnsToken  := AOwnsToken;
end;

constructor TTask.Create(ACancelMethod: TTaskCancelMethod; AData: Pointer;
  AToken: TCancellationToken; AOwnsToken: Boolean; APool: TThreadPool);
begin
  inherited Create;
  InitFields(AData, APool);
  FCancelMethod := ACancelMethod;
  FToken        := AToken;
  FOwnsToken    := AOwnsToken;
end;

{ ======================================================
  TTask — destructor
  ====================================================== }
destructor TTask.Destroy;
var
  i: Integer;
begin
  for i := 0 to FContCount - 1 do
  begin
    FConts[i]._Release;
    FConts[i] := nil;
  end;
  SetLength(FConts,    0);
  SetLength(FContOpts, 0);

  // If the task created or took ownership of the token — free it
  if FOwnsToken and (FToken <> nil) then
    FreeAndNil(FToken);

  CloseHandle(FDoneEvent);
  DeleteCriticalSection(FCS);
  inherited;
end;

{ ======================================================
  TTask — execution logic
  ====================================================== }

function TTask.IsExternalCancellationRequested: Boolean;
begin
  Result := ((FCancelFlag) ) or
            ((FToken <> nil) and (FToken.IsCancellationRequested));
end;

procedure TTask.InternalExecute;
begin
  { -- Check cancellation BEFORE starting work -- }
  if IsExternalCancellationRequested then
  begin
    CancelSelf;
    Exit;
  end;

  EnterCriticalSection(FCS);
  FState := tsRunning;
  LeaveCriticalSection(FCS);

  try
    { Cancel-aware callbacks receive the token directly }
    if Assigned(FCancelProc) then
      FCancelProc(FData, FToken)
    else if Assigned(FCancelMethod) then
      FCancelMethod(FData, FToken)
    else if Assigned(FProc) then
      FProc(FData)
    else if Assigned(FMethod) then
      FMethod(FData);

    Complete('');

  except
    on E: ETaskCancelled do
      CancelSelf;                  // cooperative cancellation > tsCancelled
    on E: Exception do
      Complete(E.Message);         // error > tsFailed
  end;
end;

procedure TTask.Complete(const AErrMsg: string);
begin
  EnterCriticalSection(FCS);
  FErrorMsg := AErrMsg;
  if AErrMsg = '' then
    FState := tsCompleted
  else
    FState := tsFailed;
  LeaveCriticalSection(FCS);
  SetEvent(FDoneEvent);
  ScheduleContinuations;
end;

procedure TTask.CancelSelf;
begin
  EnterCriticalSection(FCS);
  FState := tsCancelled;
  LeaveCriticalSection(FCS);
  SetEvent(FDoneEvent);
  ScheduleContinuations;
end;

function TTask.ShouldRunCont(AOptions: TContinuationOptions;
  AParentState: TTaskState): Boolean;
begin
  Result := False;
  if coAlways in AOptions then
  begin
    Result := AParentState in [tsCompleted, tsFailed];
    Exit;
  end;
  case AParentState of
    tsCompleted: Result := coOnlyOnSuccess in AOptions;
    tsFailed   : Result := coOnlyOnFailed  in AOptions;
    // tsCancelled > continuation is not run (regardless of filter)
  end;
end;

procedure TTask.ScheduleContinuations;
var
  i         : Integer;
  ParentState: TTaskState;
  Count     : Integer;
  SnapConts : TTasksArray;
  SnapOpts  : array of TContinuationOptions;
begin
  EnterCriticalSection(FCS);
  try
    ParentState := FState;
    Count       := FContCount;
    if Count = 0 then Exit;
    SetLength(SnapConts, Count);
    SetLength(SnapOpts,  Count);
    for i := 0 to Count - 1 do
    begin
      SnapConts[i] := FConts[i];
      SnapOpts[i]  := FContOpts[i];
      FConts[i]._AddRef;
    end;
  finally
    LeaveCriticalSection(FCS);
  end;
  try
    for i := 0 to Count - 1 do
      if ShouldRunCont(SnapOpts[i], ParentState) then
      begin
        SnapConts[i].FRunOnMain := coOnMainThread in SnapOpts[i];
        FPool.Enqueue(SnapConts[i]);
      end
      else
        SnapConts[i].CancelSelf;
  finally
    for i := 0 to Count - 1 do
      SnapConts[i]._Release;
  end;
end;

{ ======================================================
  TTask — cancellation
  ====================================================== }

procedure TTask.Cancel;
var
  Token: TCancellationToken;
begin
  // 1) Set the flag — the task will not start if still in the queue
  InterlockedExchange(Integer(FCancelFlag), Integer(True));

  // 2) If there is a token — cancel it (gives a chance to a running task)
  EnterCriticalSection(FCS);
  Token := FToken;
  LeaveCriticalSection(FCS);
  if Token <> nil then
    Token.Cancel;
end;

procedure TTask.CancelAfter(ATimeoutMs: Cardinal);
var
  Token: TCancellationToken;
begin
  // If no token — create our own
  EnterCriticalSection(FCS);
  try
    if FToken = nil then
    begin
      FToken     := TCancellationToken.Create;
      FOwnsToken := True;
    end;
    Token := FToken;
  finally
    LeaveCriticalSection(FCS);
  end;
  Token.CancelAfter(ATimeoutMs);
end;

{ ======================================================
  TTask — continuations
  ====================================================== }

function TTask.AddContinuation(ACont: TTask;
  AOptions: TContinuationOptions): ITask;
var
  AlreadyDone : Boolean;
  CurrentState: TTaskState;
begin
  Result := ACont;   // _AddRef via interface: RC>1
  EnterCriticalSection(FCS);
  try
    CurrentState := FState;
    AlreadyDone  := CurrentState in [tsCompleted, tsFailed, tsCancelled];
    if not AlreadyDone then
    begin
      if FContCount >= Length(FConts) then
      begin
        SetLength(FConts,    FContCount + 8);
        SetLength(FContOpts, FContCount + 8);
      end;
      FConts   [FContCount] := ACont;
      FContOpts[FContCount] := AOptions;
      Inc(FContCount);
      ACont._AddRef;
    end;
  finally
    LeaveCriticalSection(FCS);
  end;
  if AlreadyDone then
  begin
    if ShouldRunCont(AOptions, CurrentState) then
    begin
      ACont.FRunOnMain := coOnMainThread in AOptions;
      FPool.Enqueue(ACont);
    end
    else
      ACont.CancelSelf;
  end;
end;

function TTask.ContinueWith(AProc: TTaskProc; AData: Pointer;
  AOptions: TContinuationOptions): ITask;
begin
  Result := AddContinuation(TTask.Create(AProc, AData, FPool), AOptions);
end;

function TTask.ContinueWith(AMethod: TTaskMethod; AData: Pointer;
  AOptions: TContinuationOptions): ITask;
begin
  Result := AddContinuation(TTask.Create(AMethod, AData, FPool), AOptions);
end;

{ ======================================================
  TTask — Wait and properties
  ====================================================== }

procedure TTask.Wait;
begin
  SmartWaitSingle(FDoneEvent);
end;

function TTask.GetState: TTaskState;
begin
  EnterCriticalSection(FCS);
  Result := FState;
  LeaveCriticalSection(FCS);
end;

function TTask.GetErrorMessage: string;
begin
  EnterCriticalSection(FCS);
  Result := FErrorMsg;
  LeaveCriticalSection(FCS);
end;

function TTask.GetIsCompleted: Boolean;
begin
  Result := GetState in [tsCompleted, tsFailed, tsCancelled];
end;

function TTask.GetIsSuccess: Boolean;
begin
  Result := GetState = tsCompleted;
end;

function TTask.GetIsFailed: Boolean;
begin
  Result := GetState = tsFailed;
end;

function TTask.GetIsCancelled: Boolean;
begin
  Result := GetState = tsCancelled;
end;

function TTask.GetWaitHandle: THandle;
begin
  Result := FDoneEvent;
end;

{ ======================================================
  TTask — factory methods
  ====================================================== }

class function TTask.Run(AProc: TTaskProc; AData: Pointer): ITask;
begin
  Result := Run(AProc, AData, TThreadPool.Default);
end;

class function TTask.Run(AMethod: TTaskMethod; AData: Pointer): ITask;
begin
  Result := Run(AMethod, AData, TThreadPool.Default);
end;

class function TTask.Run(AProc: TTaskProc; AData: Pointer;
  APool: TThreadPool): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(AProc, AData, APool);
  Result := T;
  APool.Enqueue(T);
end;

class function TTask.Run(AMethod: TTaskMethod; AData: Pointer;
  APool: TThreadPool): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(AMethod, AData, APool);
  Result := T;
  APool.Enqueue(T);
end;

{ Standard callback + external token }
class function TTask.Run(AProc: TTaskProc; AData: Pointer;
  AToken: TCancellationToken): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(AProc, AData, AToken, TThreadPool.Default);
  Result := T;
  TThreadPool.Default.Enqueue(T);
end;

class function TTask.Run(AMethod: TTaskMethod; AData: Pointer;
  AToken: TCancellationToken): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(AMethod, AData, AToken, TThreadPool.Default);
  Result := T;
  TThreadPool.Default.Enqueue(T);
end;

{ Cancel-aware callback + token }
class function TTask.Run(ACancelProc: TTaskCancelProc; AData: Pointer;
  AToken: TCancellationToken): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(ACancelProc, AData, AToken, False, TThreadPool.Default);
  Result := T;
  TThreadPool.Default.Enqueue(T);
end;

class function TTask.Run(ACancelMethod: TTaskCancelMethod; AData: Pointer;
  AToken: TCancellationToken): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(ACancelMethod, AData, AToken, False, TThreadPool.Default);
  Result := T;
  TThreadPool.Default.Enqueue(T);
end;

class function TTask.Run(ACancelProc: TTaskCancelProc; AData: Pointer;
  AToken: TCancellationToken; APool: TThreadPool): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(ACancelProc, AData, AToken, False, APool);
  Result := T;
  APool.Enqueue(T);
end;

class function TTask.Run(ACancelMethod: TTaskCancelMethod; AData: Pointer;
  AToken: TCancellationToken; APool: TThreadPool): ITask;
var
  T: TTask;
begin
  T      := TTask.Create(ACancelMethod, AData, AToken, False, APool);
  Result := T;
  APool.Enqueue(T);
end;

{ ======================================================
  TTask — aggregates
  ====================================================== }

class procedure TTask.WaitAll(const ATasks: array of ITask);
const
  MaxBatch = MAXIMUM_WAIT_OBJECTS;
var
  Handles: array[0..MAXIMUM_WAIT_OBJECTS - 1] of THandle;
  Total, Base, Batch, j: Integer;
  Res: DWORD;
begin
  Total := High(ATasks) - Low(ATasks) + 1;
  if Total = 0 then Exit;
  Base := 0;
  while Base < Total do
  begin
    Batch := Total - Base;
    if Batch > MaxBatch then Batch := MaxBatch;
    for j := 0 to Batch - 1 do
      Handles[j] := ATasks[Low(ATasks) + Base + j].WaitHandle;
    if IsMainThread then
      repeat
        Res := WaitForMultipleObjects(Batch, @Handles[0], True, TASK_PUMP_INTERVAL);
        if Res = WAIT_TIMEOUT then CheckSynchronize else Break;
      until False
    else
      WaitForMultipleObjects(Batch, @Handles[0], True, INFINITE);
    Inc(Base, Batch);
  end;
end;

class function TTask.WaitAny(const ATasks: array of ITask): Integer;
var
  Handles: array[0..MAXIMUM_WAIT_OBJECTS - 1] of THandle;
  Count, i: Integer;
  Res: DWORD;
begin
  Result := -1;
  Count  := High(ATasks) - Low(ATasks) + 1;
  if Count = 0 then Exit;
  if Count > MAXIMUM_WAIT_OBJECTS then Count := MAXIMUM_WAIT_OBJECTS;
  for i := 0 to Count - 1 do
    Handles[i] := ATasks[Low(ATasks) + i].WaitHandle;
  if IsMainThread then
    repeat
      Res := WaitForMultipleObjects(Count, @Handles[0], False, TASK_PUMP_INTERVAL);
      if Res = WAIT_TIMEOUT then CheckSynchronize else Break;
    until False
  else
    Res := WaitForMultipleObjects(Count, @Handles[0], False, INFINITE);
  if (Res >= WAIT_OBJECT_0) and (Res < WAIT_OBJECT_0 + DWORD(Count)) then
    Result := Low(ATasks) + Integer(Res - WAIT_OBJECT_0);
end;

{ ======================================================
  Initialisation / finalisation
  ====================================================== }
initialization
  GMainThreadID := GetCurrentThreadId;
  InitializeCriticalSection(GPoolCS);
  GNoneToken := TCancellationToken.Create;  // Lives until finalisation

finalization
  TThreadPool.ReleaseDefault;
  FreeAndNil(GNoneToken);
  DeleteCriticalSection(GPoolCS);

end.