ThreadPool.ProducerConsumer API Documentation

Overview#

ThreadPool.ProducerConsumer implements a thread pool backed by a fixed-size circular queue with built-in backpressure. Use it when task production can outpace consumption and you need predictable memory usage and overflow control.

For simpler use cases see ThreadPool.Simple.

uses
  {$IFDEF UNIX}cthreads,{$ENDIF}  // must be first on Unix-like systems
  ThreadPool.ProducerConsumer;

See the official FPC documentation on cthreads.

v0.9 observable Submit tasks, batches, ranges, and pending cancellation are documented in the ThreadPool.Tasks API. Bounded TrySubmit uses the same deadline rules as TryQueue.

---

Constructor#

constructor Create(AThreadCount: Integer = 0; AQueueSize: Integer = 1024);

Parameter	Default	Description
`AThreadCount`	`0` (uses CPU count)	Worker threads. Clamped: min 4, max 2× `ProcessorCount`
`AQueueSize`	`1024`	Circular queue capacity in work items

// Default: CPU-count threads, 1024-item queue
Pool := TProducerConsumerThreadPool.Create;

// Custom: 4 threads, 512-item queue
Pool := TProducerConsumerThreadPool.Create(4, 512);

---

Queue Methods#

All four overloads are thread-safe. A call waits only while the bounded queue is actually full. If its compatibility timeout expires, EQueueFullException is raised.

procedure Queue(AProcedure: TThreadProcedure);
procedure Queue(AMethod: TThreadMethod);
procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer);
procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer);

Pool.Queue(@MyProcedure);            // plain procedure
Pool.Queue(@MyObject.MyMethod);      // object method
Pool.Queue(@MyIndexedProc, 42);      // indexed procedure
Pool.Queue(@MyObject.MyMethod, 42);  // indexed method

Use the matching TryQueue overloads when queue saturation is expected. They return False instead of raising:

if not Pool.TryQueue(@MyProcedure, 50) then
  WriteLn('No queue space became available within 50 ms');

Timeouts are milliseconds. 0 does not wait and THREADPOOL_INFINITE waits without a deadline.

Task type signatures (from `ThreadPool.Types`)#

TThreadProcedure      = procedure;
TThreadMethod         = procedure of object;
TThreadProcedureIndex = procedure(index: Integer);
TThreadMethodIndex    = procedure(index: Integer) of object;

---

WaitForAll#

Blocks until every queued task has finished executing.

Pool.WaitForAll;

if not Pool.WaitForAll(250) then
  WriteLn('Tasks are still running');

WaitForAll does not close admission. Coordinate concurrent producers first, or use Shutdown to atomically stop admission before draining.

Always call before freeing the pool or any objects whose methods were queued.

---

Error Handling#

Two distinct error paths exist — handle both:

1. Queue-full errors (raised by `Queue`)#

EQueueFullException is raised synchronously on the calling thread when the queue stays full until the compatibility timeout expires. Catch it around each Queue call, not around WaitForAll.

try
  Pool.Queue(@MyProcedure);
except
  on E: EQueueFullException do
  begin
    // Queue is saturated. Options:
    // - wait and retry: Pool.WaitForAll; Pool.Queue(@MyProcedure);
    // - log and skip
    WriteLn('Queue full: ', E.Message);
  end;
end;

Always catch by type (EQueueFullException), not by message string.

2. Worker execution errors (read from `LastError`)#

Exceptions that occur inside a task are caught by the worker thread and stored in LastError. Check after WaitForAll.

Pool.WaitForAll;

if Pool.LastError <> '' then
begin
  // LastError holds the raw message of the most recent worker exception.
  WriteLn('Task error: ', Pool.LastError);
  Pool.ClearLastError;  // reset before reuse
end;

To inspect every task failure (not just the last), use the Errors collection added in v0.7.0 — oldest first, capped at MAX_STORED_ERRORS = 1000:

var
  Msg: string;
begin
  Pool.ClearErrors;
  // ... queue tasks ...
  Pool.WaitForAll;

  WriteLn(Pool.ErrorCount, ' task(s) failed:');
  for Msg in Pool.Errors do
    WriteLn('  - ', Msg);
  Pool.ClearErrors;  // resets the collection and LastError
end;

Or assign OnError to react the moment a task fails, instead of polling:

// Called synchronously from a worker thread — keep it short, bounded, and
// thread-safe.
Pool.OnError := @Handler.OnTaskError;

Exceptions raised by the handler are contained by the pool. They cannot terminate a worker or prevent task completion accounting.

Containment does not limit callback execution time. A task or OnError handler that blocks continues to occupy its worker, and can delay WaitForAll and Shutdown. Apply an application-level timeout or cancellation mechanism to operations that may block.

Full pattern#

var
  Pool: TProducerConsumerThreadPool;
begin
  Pool := TProducerConsumerThreadPool.Create;
  try
    try
      Pool.Queue(@RiskyOperation);
    except
      on E: EQueueFullException do
        WriteLn('Queue full: ', E.Message);
    end;

    Pool.WaitForAll;

    if Pool.LastError <> '' then
      WriteLn('Task failed: ', Pool.LastError);
  finally
    Pool.Free;
  end;
end;

---

Properties and Methods#

property ThreadCount: Integer;        // read-only; number of worker threads
property LastError: string;           // read-only; last worker exception message
property Errors: TStringArray;        // read-only; all captured messages (capped)
property ErrorCount: Integer;         // read-only; count of messages in Errors
property OnError: TThreadPoolErrorEvent; // fired (on a worker thread) per failed task
property QueueCount: Integer;             // read-only; queued items waiting for workers
property QueueCapacity: Integer;          // read-only; configured queue capacity
property QueueLoadFactor: Double;         // read-only; QueueCount / QueueCapacity
property BackpressureConfig: TBackpressureConfig; // thread-safe compatibility config
property WorkQueue: TThreadSafeQueue;     // legacy compatibility access; avoid in new code
property State: TThreadPoolState;      // accepting, draining, or stopped

function TryQueue(...; ATimeoutMS: Cardinal): Boolean; // four matching overloads
procedure WaitForAll; overload;
function WaitForAll(ATimeoutMS: Cardinal): Boolean; overload;
procedure Shutdown;
procedure ClearLastError;
procedure ClearErrors;                // clears both Errors and LastError

---

Backpressure Configuration compatibility#

v0.8.0 replaces load-threshold sleeps with event-driven not-full signalling. TBackpressureConfig remains available so existing source continues to compile. MaxAttempts and HighLoadDelay determine the compatibility wait used by the legacy Queue overloads; new code should express its deadline directly with TryQueue(..., TimeoutMS). Threshold and low/medium-delay fields are retained for source compatibility and no longer introduce sleeps.

type
  TBackpressureConfig = record
    LowLoadThreshold:    Double;   // Default: 0.5  — delay starts at 50% capacity
    MediumLoadThreshold: Double;   // Default: 0.7  — medium delay at 70%
    HighLoadThreshold:   Double;   // Default: 0.9  — max delay at 90%
    LowLoadDelay:        Integer;  // Default: 10 ms
    MediumLoadDelay:     Integer;  // Default: 50 ms
    HighLoadDelay:       Integer;  // Default: 100 ms
    MaxAttempts:         Integer;  // Default: 5 — raises EQueueFullException after this
  end;

Read and write the compatibility config directly through the pool:

var
  Config: TBackpressureConfig;
begin
  Config := Pool.BackpressureConfig;
  Config.MaxAttempts   := 3;
  Config.HighLoadDelay := 200;
  Pool.BackpressureConfig := Config;
end;

WorkQueue remains public so existing v0.x programs compile, but new code should not use it. Calling its mutating methods (TryEnqueue, TryDequeue, or Clear) bypasses the pool's completion accounting and can invalidate WaitForAll. Use QueueCount, QueueCapacity, QueueLoadFactor, and BackpressureConfig on the pool instead.

---

Debug Logging#

The DEBUG_LOG variable controls verbose output:

DEBUG_LOG := True; // opt in while diagnosing scheduler behaviour

Debug logging is disabled by default so normal workloads and benchmarks do not pay for synchronized console output.

---

Lifecycle (v0.8.0)#

Shutdown atomically stops admission, lets submissions already in progress finish enqueueing, drains all accepted tasks, wakes and joins every worker, and sets State to tpsStopped. It is idempotent and is called automatically by the destructor. Queueing after shutdown raises EThreadPoolShutdown.

---

Thread Count Rules#

Condition	Result
`AThreadCount <= 0`	Uses `TThread.ProcessorCount`
`AThreadCount < 4`	Raised to 4 (minimum enforced)
`AThreadCount > 2 × ProcessorCount`	Clamped to `2 × ProcessorCount`

Thread count is fixed after construction.

---

Common Pitfalls#

Catching EQueueFullException by message string#

// WRONG — the message includes dynamic content and will never match literally
on E: Exception do
  if E.Message = 'Queue is full' then ...

// CORRECT — catch by exception type
on E: EQueueFullException do ...

Freeing an object before WaitForAll#

// WRONG — worker thread may still be calling MyObject.MyMethod
Pool.Queue(@MyObject.MyMethod);
MyObject.Free;

// CORRECT
try
  Pool.Queue(@MyObject.MyMethod);
  Pool.WaitForAll;
finally
  MyObject.Free;
end;

Placing WaitForAll inside the EQueueFullException handler#

// WRONG — if Queue raises, WaitForAll is never reached, pool is freed while busy
try
  Pool.Queue(@MyProc);
  Pool.WaitForAll;   // skipped on exception
except
  on E: EQueueFullException do ...
end;
Pool.Free;

// CORRECT — separate concerns
try
  Pool.Queue(@MyProc);
except
  on E: EQueueFullException do
    WriteLn('Queue full: ', E.Message);
end;
Pool.WaitForAll;  // always reached
Pool.Free;

---

Advanced: Queue Management#

Queue multiple items and handle overflow per-item:

var
  Pool: TProducerConsumerThreadPool;
  i: Integer;
begin
  Pool := TProducerConsumerThreadPool.Create(4, 512);
  try
    for i := 1 to 2000 do
    begin
      try
        Pool.Queue(@MyProc);
      except
        on E: EQueueFullException do
        begin
          Pool.WaitForAll;   // let the queue drain
          Pool.Queue(@MyProc);  // retry
        end;
      end;
    end;

    Pool.WaitForAll;
  finally
    Pool.Free;
  end;
end;

---

Limitations#

Fixed queue capacity — no dynamic resizing
LastError holds only the most recent worker exception (use Errors to collect all; capped at MAX_STORED_ERRORS)
No task priorities or forced cancellation of running callbacks; v0.9 can cancel pending submitted work
No dynamic thread scaling after construction
Not suitable for real-time or UI-thread work

ThreadPool.ProducerConsumer API Documentation

Overview#

Constructor#

Queue Methods#

Task type signatures (from ThreadPool.Types)#

WaitForAll#

Error Handling#

1. Queue-full errors (raised by Queue)#

2. Worker execution errors (read from LastError)#

Full pattern#

Properties and Methods#

Backpressure Configuration compatibility#

Debug Logging#

Lifecycle (v0.8.0)#

Thread Count Rules#

Common Pitfalls#

Catching EQueueFullException by message string#

Freeing an object before WaitForAll#

Placing WaitForAll inside the EQueueFullException handler#

Advanced: Queue Management#

Limitations#

Task type signatures (from `ThreadPool.Types`)#

1. Queue-full errors (raised by `Queue`)#

2. Worker execution errors (read from `LastError`)#