From 35cd648aa657abb38134c2562cbdcd1e06ad4a36 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah Date: Mon, 13 Jul 2026 21:56:26 +1000 Subject: [PATCH 1/3] feat(repo): make thread pools event-driven and shutdown-safe - unify lifecycle and timeout contracts across both pools - replace worker polling with event-driven queues - contain task and error-handler exceptions - add deterministic shutdown and concurrency tests - add v0.7.0 comparison benchmarks - update documentation and release metadata for v0.8.0 --- .github/workflows/ci.yml | 12 + CHANGELOG.md | 45 ++ README.md | 95 ++-- benchmarks/README.md | 17 + benchmarks/ThreadPoolBenchmark.lpi | 54 +++ benchmarks/ThreadPoolBenchmark.lpr | 142 ++++++ docs/ThreadPool.ProducerConsumer-API.md | 70 ++- docs/ThreadPool.ProducerConsumer-Technical.md | 369 ++++----------- docs/ThreadPool.Simple-API.md | 40 +- docs/ThreadPool.Simple-Technical.md | 55 ++- docs/release-notes-v0.8.0.md | 63 +++ package/lazarus/threadpool_fp.lpk | 17 +- src/ThreadPool.ProducerConsumer.pas | 413 ++++++++++------- src/ThreadPool.Simple.pas | 419 +++++++++++------- src/threadpool.types.pas | 135 +++++- tests/ThreadPool.ProducerConsumer.Tests.pas | 335 +++++++++++--- tests/threadpool.simple.tests.pas | 190 +++++++- 17 files changed, 1694 insertions(+), 777 deletions(-) create mode 100644 benchmarks/README.md create mode 100644 benchmarks/ThreadPoolBenchmark.lpi create mode 100644 benchmarks/ThreadPoolBenchmark.lpr create mode 100644 docs/release-notes-v0.8.0.md diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4165494..bc952fc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,6 +40,18 @@ jobs: ./tests/TestRunner -a -p --format=plain fi + - name: Build benchmark + run: lazbuild --build-mode=Release benchmarks/ThreadPoolBenchmark.lpi + + - name: Run benchmark smoke test + shell: bash + run: | + if [ -f benchmarks/ThreadPoolBenchmark.exe ]; then + ./benchmarks/ThreadPoolBenchmark.exe + else + ./benchmarks/ThreadPoolBenchmark + fi + build-examples: name: Build examples (${{ matrix.os }}) runs-on: ${{ matrix.os }} diff --git a/CHANGELOG.md b/CHANGELOG.md index 167369a..c92acc5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,51 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). +## [0.8.0] - 2026-07-13 + +### Added + +- Shared lifecycle contract for both pools: `tpsAccepting` → `tpsDraining` → + `tpsStopped`, exposed through `State` +- Idempotent `Shutdown`, which stops admission, drains all accepted work, wakes + and joins workers, and rejects calls made from a pool worker to avoid self-wait +- Timeout-aware `WaitForAll(TimeoutMS): Boolean` on both pools +- Four timeout-aware `TryQueue(..., TimeoutMS): Boolean` overloads on both pools +- `EThreadPoolShutdown` for submissions attempted after shutdown begins +- Event-driven not-empty/not-full signalling for the bounded queue +- Event-driven, dynamically growing O(1) circular FIFO for the Simple pool +- Lifecycle, queue-timeout, callback-failure, concurrent admission/shutdown, + invalid-capacity, and draining regression tests +- Release benchmark covering 20,000-task bursts and idle queue-to-start latency + +### Changed + +- Existing `Queue` overloads remain source-compatible +- Both destructors now use the same draining shutdown behaviour +- Worker completion accounting runs independently of task and callback failures +- `OnError` exceptions are contained so they cannot terminate a worker or wedge + `WaitForAll` +- Producer backpressure waits only when the queue is full; load-threshold sleeps + are removed while `TBackpressureConfig` remains source-compatible +- Producer debug logging is disabled by default +- Test-suite runtime reduced from about 54 seconds to about 4 seconds on the + development machine while growing from 43 to 58 tests + +### Performance + +- Median of three identical-source Windows/FPC 3.2.2 `-O3` runs, logging disabled: + - Simple 20,000-task burst: 234 ms (v0.7.0) → 109 ms (v0.8.0), about 2.1× faster + - Producer burst: 3,953 ms → 172 ms, about 23.0× faster + - Simple idle latency: 7.9 ms → below the benchmark's 1 ms timer resolution + - Producer idle latency: 81.3 ms → below the 1 ms timer resolution + +### Compatibility notes + +- Queueing after `Shutdown` now raises `EThreadPoolShutdown`; v0.7.0's Simple + pool silently discarded such work +- Backpressure threshold and low/medium-delay fields no longer introduce sleeps; + new code should use `TryQueue` to state its timeout directly + ## [0.7.0] - 2026-07-11 ### Added diff --git a/README.md b/README.md index f58023a..98f37b9 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # 🚀 ThreadPool for Free Pascal -[![Version](https://img.shields.io/badge/version-0.7.0-8B5CF6.svg)](CHANGELOG.md) +[![Version](https://img.shields.io/badge/version-0.8.0-8B5CF6.svg)](CHANGELOG.md) [![License: MIT](https://img.shields.io/badge/License-MIT-1E3A8A.svg)](https://opensource.org/licenses/MIT) [![Free Pascal](https://img.shields.io/badge/Free%20Pascal-3.2.2+-3B82F6.svg)](https://www.freepascal.org/) [![Lazarus](https://img.shields.io/badge/Lazarus-4.0+-60A5FA.svg)](https://www.lazarus-ide.org/) @@ -47,6 +47,7 @@ A lightweight, easy-to-use thread pool implementation for Free Pascal. Simplify - [🏃 Quick Start](#-quick-start) - [Simple Thread Pool](#simple-thread-pool) - [Producer-Consumer Thread Pool](#producer-consumer-thread-pool) + - [Timeouts and shutdown (v0.8.0)](#timeouts-and-shutdown-v080) - [Error Handling Simple Thread Pool](#error-handling-simple-thread-pool) - [Error Handling Producer-Consumer Thread Pool](#error-handling-producer-consumer-thread-pool) - [Tips](#tips) @@ -76,7 +77,7 @@ This library provides two thread pool implementations, each with its own strengt uses ThreadPool.Simple; ``` - Global singleton instance for quick use -- Direct task execution +- Event-driven, dynamically growing FIFO work queue - Automatic thread count management - Best for simple parallel tasks - Lower memory overhead @@ -86,7 +87,8 @@ uses ThreadPool.Simple; uses ThreadPool.ProducerConsumer; ``` -A thread pool with fixed-size circular buffer (1024 items) and built-in backpressure handling: +A thread pool with a fixed-size circular buffer (1024 items) and event-driven +backpressure: - **Queue Management** - Fixed-size circular buffer for predictable memory usage @@ -94,17 +96,20 @@ A thread pool with fixed-size circular buffer (1024 items) and built-in backpres - Configurable capacity (default: 1024 items) - **Backpressure Handling** - - Load-based adaptive delays (10ms to 100ms) - - Automatic retry mechanism (up to 5 attempts) - - Throws EQueueFullException when retries exhausted + - Producers wait only while the queue is actually full + - `TryQueue(..., TimeoutMS)` reports saturation without raising + - Existing `Queue(...)` calls retain a bounded default wait and raise + `EQueueFullException` if that wait expires - **Monitoring & Debug** - - Thread-safe error capture with thread IDs - - Detailed debug logging (can be disabled) + - Thread-safe error capture + - Internal debug logging is disabled by default > [!WARNING] > -> While the system includes automatic retry mechanisms, it's recommended that users implement their own error handling strategies for scenarios where the queue remains full after all retry attempts. +> For sustained saturation, prefer `TryQueue(..., TimeoutMS)` and handle a +> `False` result explicitly. Legacy `Queue(...)` raises `EQueueFullException` +> when its compatibility wait expires. ### Shared Features @@ -118,16 +123,20 @@ A thread pool with fixed-size circular buffer (1024 items) and built-in backpres - Object methods: `Pool.Queue(@MyObject.MyMethod)` - Indexed variants: `Pool.Queue(@MyProc, Index)` -- **Thread Safety** +- **Lifecycle & Thread Safety — v0.8.0** - Built-in synchronization - - Safe resource sharing - - Protected error handling + - Event-driven workers; no polling sleeps + - `Shutdown` stops admission, drains accepted work, and joins workers + - Queueing after shutdown raises `EThreadPoolShutdown` + - `WaitForAll(TimeoutMS)` and `TryQueue(..., TimeoutMS)` use milliseconds; + zero means do not wait and `THREADPOOL_INFINITE` means no timeout - **Error Management** - Worker exceptions are caught automatically; the pool keeps running - `LastError` for the most recent failure - `Errors` collection captures **all** failed-task messages (capped, oldest dropped) — *v0.7.0* - Optional `OnError` callback fired per failed task — *v0.7.0* + - Exceptions raised by `OnError` are contained and cannot terminate workers > [!NOTE] > Thread count is determined by `TThread.ProcessorCount` at startup and remains fixed. See [Thread Management](#-thread-management) for details. @@ -198,6 +207,37 @@ begin end; ``` +### Timeouts and shutdown (v0.8.0) + +Both pools use the same lifecycle contract and timeout units: + +```pascal +// Existing source remains valid and waits indefinitely for completion. +Pool.Queue(@DoWork); +Pool.WaitForAll; + +// Timeout-aware wait: False means work is still running after 250 ms. +if not Pool.WaitForAll(250) then + WriteLn('Still working'); + +// ProducerConsumer is bounded; this waits at most 50 ms for queue space. +if not Pool.TryQueue(@DoWork, 50) then + WriteLn('Queue is saturated'); + +// Stop accepting work, drain every accepted task, then join all workers. +Pool.Shutdown; +``` + +Timeouts are milliseconds. `0` performs only an immediate check and +`THREADPOOL_INFINITE` waits without a deadline. `TSimpleThreadPool` is +unbounded, so its `TryQueue` timeout is accepted for API symmetry but queue +capacity cannot time out. After `Shutdown`, all `Queue` and `TryQueue` overloads +raise `EThreadPoolShutdown`. + +`WaitForAll` is not an admission barrier for unrelated producer threads. If +producers may still submit concurrently, coordinate them first or call +`Shutdown`, which closes admission before draining. + ### Error Handling Simple Thread Pool ```pascal @@ -310,13 +350,13 @@ Pool.OnError := @MyHandler.OnTaskError; > [!NOTE] > **Error Handling** -> - 🛡️ Exceptions are caught and stored with thread IDs +> - 🛡️ Exceptions are caught and stored without terminating workers > - ⚡ Pool continues operating after exceptions > - 🔄 Use ClearLastError to reset error state > > **Debugging** -> - 🔍 Error messages contain thread identification -> - 📝 Debug logging enabled by default (configurable) +> - 🔍 Task failures are collected without stopping workers +> - 📝 Internal debug logging is disabled by default > - 📊 Queue capacity monitoring available @@ -331,7 +371,7 @@ Need a thread pool? ``` **Use Simple Thread Pool when:** -- Direct task execution without queuing needed +- An unbounded, dynamically growing queue is appropriate - Task count is predictable and moderate - Low memory overhead is important - Global instance (GlobalThreadPool) convenience desired @@ -342,7 +382,7 @@ Need a thread pool? - Backpressure handling required - Queue overflow protection important - Need detailed execution monitoring -- Want configurable retry mechanisms +- Want bounded submission with explicit timeouts ### Queue overload reference @@ -500,6 +540,7 @@ All tasks completed successfully! - [ThreadPool.Simple Technical Details](docs/ThreadPool.Simple-Technical.md) - [ThreadPool.ProducerConsumer API Documentation](docs/ThreadPool.ProducerConsumer-API.md) - [ThreadPool.ProducerConsumer Technical Details](docs/ThreadPool.ProducerConsumer-Technical.md) +- [v0.8.0 Release Notes](docs/release-notes-v0.8.0.md) ## 🧪 Testing @@ -508,7 +549,8 @@ All tasks completed successfully! 3. Run `./TestRunner.exe -a -p --format=plain` to see the test results. 4. Ensure all tests pass to verify the library's functionality -May take up to 5 mins to run all tests. +The complete suite normally finishes in a few seconds. Release benchmarks live +in [`benchmarks/`](benchmarks/). ## 🧵 Thread Management @@ -521,14 +563,14 @@ May take up to 5 mins to run all tests. ### Implementation Characteristics **Simple Thread Pool** -- Direct task execution without queuing -- Continuous task processing -- Clean shutdown handling +- Dynamically growing O(1) FIFO queue +- Event-driven worker wakeups +- Draining shutdown **Producer-Consumer Thread Pool** - Fixed-size circular queue (1024 items by default, configurable) -- Backpressure handling with adaptive delays -- Graceful overflow management +- Event-driven not-empty/not-full signalling +- Timeout-aware bounded submission and draining shutdown ## ⚠️ Common Mistakes @@ -603,10 +645,9 @@ GlobalThreadPool.WaitForAll; ## 🚧 Planned/In Progress -- Performance & robustness pass — event-driven idle workers (remove the Simple pool's poll loop), shutdown review, stress/soak tests (planned for 0.8.0) -- Support for `procedure Queue(AMethod: TProc; AArgs: array of Const);` -- More comprehensive tests -- More examples +- Result-bearing task handles/futures +- Chunked `ParallelFor` helpers +- Additional platform and long-running soak coverage ## 🤝 Contributing diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000..6f728c1 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,17 @@ +# ThreadPool benchmark + +Build and run the release-mode benchmark: + +```bash +lazbuild --build-mode=Release benchmarks/ThreadPoolBenchmark.lpi +./benchmarks/ThreadPoolBenchmark +``` + +The same source compiles against v0.7.0 and v0.8.0. It reports: + +- completion time for a 20,000-task burst through each pool; +- average queue-to-start latency after workers have been idle. + +Run each version several times on the same otherwise-idle machine and compare +medians. Debug logging must be disabled in both versions so console I/O is not +included in the scheduler measurement. diff --git a/benchmarks/ThreadPoolBenchmark.lpi b/benchmarks/ThreadPoolBenchmark.lpi new file mode 100644 index 0000000..2aaae44 --- /dev/null +++ b/benchmarks/ThreadPoolBenchmark.lpi @@ -0,0 +1,54 @@ + + + + + + + + + + + + + <UseAppBundle Value="False"/> + <ResourceType Value="res"/> + </General> + <BuildModes> + <Item Name="Default" Default="True"/> + <Item Name="Release"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target><Filename Value="ThreadPoolBenchmark"/></Target> + <SearchPaths> + <OtherUnitFiles Value="..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <CodeGeneration> + <SmartLinkUnit Value="True"/> + <Optimizations><OptimizationLevel Value="3"/></Optimizations> + </CodeGeneration> + <Linking><LinkSmart Value="True"/></Linking> + </CompilerOptions> + </Item> + </BuildModes> + <RequiredPackages> + <Item><PackageName Value="FCL"/></Item> + </RequiredPackages> + <Units> + <Unit> + <Filename Value="ThreadPoolBenchmark.lpr"/> + <IsPartOfProject Value="True"/> + </Unit> + </Units> + </ProjectOptions> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target><Filename Value="ThreadPoolBenchmark"/></Target> + <SearchPaths> + <OtherUnitFiles Value="..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + </CompilerOptions> +</CONFIG> diff --git a/benchmarks/ThreadPoolBenchmark.lpr b/benchmarks/ThreadPoolBenchmark.lpr new file mode 100644 index 0000000..653d1e0 --- /dev/null +++ b/benchmarks/ThreadPoolBenchmark.lpr @@ -0,0 +1,142 @@ +program ThreadPoolBenchmark; + +{$mode objfpc}{$H+}{$J-} + +uses + {$IFDEF UNIX} + cthreads, + {$ENDIF} + Classes, SysUtils, SyncObjs, + ThreadPool.Simple, ThreadPool.ProducerConsumer; + +const + BURST_TASKS = 20000; + IDLE_SAMPLES = 10; + IDLE_PAUSE_MS = 120; + +var + Counter: Integer; + IdleEvent: TEvent; + SubmittedAt: QWord; + LastIdleLatency: QWord; + +procedure CountTask; +begin + InterlockedIncrement(Counter); +end; + +procedure IdleTask; +begin + LastIdleLatency := GetTickCount64 - SubmittedAt; + IdleEvent.SetEvent; +end; + +function BenchmarkSimpleBurst: QWord; +var + Pool: TSimpleThreadPool; + StartedAt: QWord; + I: Integer; +begin + Counter := 0; + Pool := TSimpleThreadPool.Create(4); + try + StartedAt := GetTickCount64; + for I := 1 to BURST_TASKS do + Pool.Queue(@CountTask); + Pool.WaitForAll; + Result := GetTickCount64 - StartedAt; + if Counter <> BURST_TASKS then + raise Exception.CreateFmt('Simple burst lost tasks: %d/%d', + [Counter, BURST_TASKS]); + finally + Pool.Free; + end; +end; + +function BenchmarkProducerBurst: QWord; +var + Pool: TProducerConsumerThreadPool; + StartedAt: QWord; + I: Integer; +begin + Counter := 0; + Pool := TProducerConsumerThreadPool.Create(4, 1024); + try + StartedAt := GetTickCount64; + for I := 1 to BURST_TASKS do + Pool.Queue(@CountTask); + Pool.WaitForAll; + Result := GetTickCount64 - StartedAt; + if Counter <> BURST_TASKS then + raise Exception.CreateFmt('Producer burst lost tasks: %d/%d', + [Counter, BURST_TASKS]); + finally + Pool.Free; + end; +end; + +function BenchmarkSimpleIdle: Double; +var + Pool: TSimpleThreadPool; + Total: QWord; + I: Integer; +begin + Total := 0; + Pool := TSimpleThreadPool.Create(4); + try + for I := 1 to IDLE_SAMPLES do + begin + Sleep(IDLE_PAUSE_MS); + IdleEvent.ResetEvent; + SubmittedAt := GetTickCount64; + Pool.Queue(@IdleTask); + if IdleEvent.WaitFor(2000) <> wrSignaled then + raise Exception.Create('Simple idle task timed out'); + Pool.WaitForAll; + Inc(Total, LastIdleLatency); + end; + Result := Total / IDLE_SAMPLES; + finally + Pool.Free; + end; +end; + +function BenchmarkProducerIdle: Double; +var + Pool: TProducerConsumerThreadPool; + Total: QWord; + I: Integer; +begin + Total := 0; + Pool := TProducerConsumerThreadPool.Create(4, 1024); + try + for I := 1 to IDLE_SAMPLES do + begin + Sleep(IDLE_PAUSE_MS); + IdleEvent.ResetEvent; + SubmittedAt := GetTickCount64; + Pool.Queue(@IdleTask); + if IdleEvent.WaitFor(2000) <> wrSignaled then + raise Exception.Create('Producer idle task timed out'); + Pool.WaitForAll; + Inc(Total, LastIdleLatency); + end; + Result := Total / IDLE_SAMPLES; + finally + Pool.Free; + end; +end; + +begin + IdleEvent := TEvent.Create(nil, True, False, ''); + try + WriteLn('burst_tasks=', BURST_TASKS); + WriteLn('idle_samples=', IDLE_SAMPLES); + WriteLn('simple_burst_ms=', BenchmarkSimpleBurst); + WriteLn('producer_burst_ms=', BenchmarkProducerBurst); + WriteLn('simple_idle_avg_ms=', FormatFloat('0.00', BenchmarkSimpleIdle)); + WriteLn('producer_idle_avg_ms=', FormatFloat('0.00', BenchmarkProducerIdle)); + finally + IdleEvent.Free; + end; +end. diff --git a/docs/ThreadPool.ProducerConsumer-API.md b/docs/ThreadPool.ProducerConsumer-API.md index 6ce86ac..181c125 100644 --- a/docs/ThreadPool.ProducerConsumer-API.md +++ b/docs/ThreadPool.ProducerConsumer-API.md @@ -45,9 +45,9 @@ Pool := TProducerConsumerThreadPool.Create(4, 512); ## Queue Methods -All four overloads are thread-safe. Each call may block briefly if the queue is -near capacity (adaptive backpressure delays apply). After the maximum retry -attempts are exhausted, `EQueueFullException` is raised. +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. ```pascal procedure Queue(AProcedure: TThreadProcedure); @@ -63,6 +63,17 @@ 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: + +```pascal +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`) ```pascal @@ -82,6 +93,14 @@ Blocks until every queued task has finished executing. Pool.WaitForAll; ``` +```pascal +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. --- @@ -93,7 +112,7 @@ 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 after all retry attempts. Catch it **around each `Queue` call**, +queue stays full until the compatibility timeout expires. Catch it **around each `Queue` call**, not around `WaitForAll`. ```pascal @@ -110,12 +129,6 @@ except end; ``` -The exception message format is: - -```text -Queue is full after N attempts (Capacity: M) -``` - Always catch by **type** (`EQueueFullException`), not by message string. ### 2. Worker execution errors (read from `LastError`) @@ -159,6 +172,9 @@ Or assign `OnError` to react the moment a task fails, instead of polling: Pool.OnError := @Handler.OnTaskError; ``` +Exceptions raised by the handler are contained by the pool. They cannot +terminate a worker or prevent task completion accounting. + ### Full pattern ```pascal @@ -195,18 +211,26 @@ property Errors: TStringArray; // read-only; all captured messages (cappe property ErrorCount: Integer; // read-only; count of messages in Errors property OnError: TThreadPoolErrorEvent; // fired (on a worker thread) per failed task property WorkQueue: TThreadSafeQueue; // access to queue for monitoring/config +property State: TThreadPoolState; // accepting, draining, or stopped -procedure WaitForAll; +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 +## Backpressure Configuration compatibility -When the queue load factor exceeds a threshold, `Queue` introduces a delay before -each retry attempt. This slows the producer instead of failing immediately. +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. ```pascal type @@ -238,15 +262,23 @@ end; ## Debug Logging -The constant `DEBUG_LOG` at the top of the unit controls verbose output: +The `DEBUG_LOG` variable controls verbose output: ```pascal -const - DEBUG_LOG = True; // set to False to silence all debug output +DEBUG_LOG := True; // opt in while diagnosing scheduler behaviour ``` -When enabled, each queue operation and worker event is logged to stdout with a -timestamp and thread ID. Disable for production use. +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`. --- diff --git a/docs/ThreadPool.ProducerConsumer-Technical.md b/docs/ThreadPool.ProducerConsumer-Technical.md index b60a405..b575e2e 100644 --- a/docs/ThreadPool.ProducerConsumer-Technical.md +++ b/docs/ThreadPool.ProducerConsumer-Technical.md @@ -2,328 +2,109 @@ ## Overview -`ThreadPool.ProducerConsumer` implements a thread pool using the -producer-consumer pattern. Tasks are placed into a fixed-size circular buffer by -the caller (producer) and removed for execution by worker threads (consumers). -When the buffer fills faster than workers can drain it, adaptive backpressure -slows the producer before raising `EQueueFullException`. +`ThreadPool.ProducerConsumer` uses a fixed-capacity circular FIFO buffer. Calls +to `Queue` or `TryQueue` produce work, and a fixed set of workers consume it. +v0.8.0 makes both sides event-driven: workers wait for a not-empty signal and +bounded producers wait for a not-full signal. ---- - -## Architecture - -### Class Structure - -```mermaid -classDiagram - class TThreadPoolBase { - <<abstract>> - +Create(threadCount: Integer) - +Queue(procedure) - +Queue(method) - +Queue(procedureIndex, index) - +Queue(methodIndex, index) - +WaitForAll() - +GetLastError() - +ClearLastError() - } - - class TProducerConsumerThreadPool { - -FThreads: TThreadList - -FWorkQueue: TThreadSafeQueue - -FCompletionEvent: TEvent - -FWorkItemCount: Integer - -FErrorLock: TCriticalSection - -FWorkItemLock: TCriticalSection - -FLocalThreadCount: Integer - +Create(threadCount, queueSize: Integer) - +Queue(procedure) - +Queue(method) - +Queue(procedureIndex, index) - +Queue(methodIndex, index) - +WaitForAll() - +WorkQueue: TThreadSafeQueue - +ThreadCount: Integer - +LastError: string - } - - class TThreadSafeQueue { - -FItems: array of IWorkItem - -FHead: Integer - -FTail: Integer - -FCount: Integer - -FCapacity: Integer - -FLock: TCriticalSection - -FBackpressureConfig: TBackpressureConfig - +Create(capacity: Integer) - +TryEnqueue(item: IWorkItem): Boolean - +TryDequeue(out item: IWorkItem): Boolean - +GetCount(): Integer - +LoadFactor: Double - +BackpressureConfig: TBackpressureConfig - +Clear() - } - - class TProducerConsumerWorkerThread { - -FThreadPool: TObject - +Create(threadPool: TObject) - #Execute() - } - - TThreadPoolBase <|-- TProducerConsumerThreadPool - TProducerConsumerThreadPool *-- TThreadSafeQueue - TProducerConsumerThreadPool *-- TProducerConsumerWorkerThread -``` - -### Component Interaction - -```mermaid -sequenceDiagram - participant App as Application - participant Pool as TProducerConsumerThreadPool - participant Queue as TThreadSafeQueue - participant Worker as TProducerConsumerWorkerThread - - App->>Pool: Queue(Task) - Pool->>Pool: Increment FWorkItemCount - Pool->>Queue: TryEnqueue(WorkItem) - alt Queue full after MaxAttempts - Queue-->>App: raise EQueueFullException - Pool->>Pool: Decrement FWorkItemCount - else Space available - Queue-->>Pool: True - Pool-->>App: return - Worker->>Queue: TryDequeue - Queue-->>Worker: WorkItem - Worker->>Worker: WorkItem.Execute - Worker->>Pool: Decrement FWorkItemCount - opt FWorkItemCount = 0 - Pool->>App: FCompletionEvent.SetEvent - end - end -``` - -### Thread Pool States - -```mermaid -stateDiagram-v2 - [*] --> Created: Create() - Created --> Ready: Initialize Threads - Ready --> Processing: Queue Task - Processing --> Ready: Task Complete - Processing --> QueueFull: Queue Full - QueueFull --> Processing: Queue Space Available - Ready --> Shutdown: Destroy - Processing --> Shutdown: Destroy - QueueFull --> Shutdown: Destroy - Shutdown --> [*]: Cleanup Complete -``` - -### Work Item Flow - -```mermaid -flowchart LR - A[Task] -->|Create| B[TProducerConsumerWorkItem] - B -->|TryEnqueue| C[TThreadSafeQueue] - C -->|TryDequeue| D[TProducerConsumerWorkerThread] - D -->|Execute| E[Complete] - D -->|Exception| F[SetLastError] - F --> G[Pool.LastError] -``` - ---- - -## Key Components +## Components ### TProducerConsumerThreadPool -The main class. It: - -- Creates and starts all worker threads at construction time -- Wraps each `Queue` call into a `TProducerConsumerWorkItem` and passes it to `TThreadSafeQueue.TryEnqueue` -- Tracks the number of in-flight work items (`FWorkItemCount`) under `FWorkItemLock` -- Signals `FCompletionEvent` when `FWorkItemCount` reaches zero (satisfying `WaitForAll`) -- Stores the most recent worker exception message in `FLastError` under `FErrorLock` +- Owns the queue, worker threads, completion event, and pending-work counter. +- Applies the lifecycle contract implemented by `TThreadPoolBase`. +- Wraps all four existing task signatures in reference-counted work items. +- Counts a task before it becomes visible to workers and decrements it in a + worker `finally` block. ### TThreadSafeQueue -A thread-safe circular buffer. Key properties: - -- **Capacity** is fixed at construction — no dynamic resizing -- **Head/tail pointers** wrap around modulo capacity: O(1) enqueue and dequeue -- `LoadFactor` = `FCount / FCapacity` — used by backpressure logic -- All operations are protected by a single `TCriticalSection` (`FLock`) +- Stores `IWorkItem` values in a fixed-size circular array. +- Uses head/tail indices for O(1) enqueue and dequeue. +- Protects array state with one critical section. +- Maintains manual-reset not-empty and not-full events under the same lock, so + wakeups cannot be lost between a state check and an event reset. +- Rejects capacities less than one. ### TProducerConsumerWorkerThread -Each worker: - -- Loops calling `TryDequeue`; executes the work item if one is available -- Calls `Sleep(100)` when the queue is empty to avoid busy-waiting -- On exception inside `Execute`, stores the message in `Pool.FLastError` and - decrements the work item counter normally (pool keeps running) -- Exits the loop when `Terminated` is set during pool destruction - ---- - -## Backpressure +- Blocks indefinitely on the queue's not-empty event while idle. +- Drains available work after being woken. +- Captures task exceptions through `TThreadPoolBase.SetLastError`. +- Releases the work item and updates completion state in `finally`, independent + of task or `OnError` callback failures. -Before each enqueue attempt, `ApplyBackpressure` reads `LoadFactor` and sleeps -for a configurable duration: +## Event-driven queue algorithm -| Load factor | Default delay | -| --- | --- | -| ≥ 50% (`LowLoadThreshold`) | 10 ms | -| ≥ 70% (`MediumLoadThreshold`) | 50 ms | -| ≥ 90% (`HighLoadThreshold`) | 100 ms | - -After `MaxAttempts` (default 5) failures, `EQueueFullException` is raised. - -Configure via `Pool.WorkQueue.BackpressureConfig`: - -```pascal -var - Config: TBackpressureConfig; -begin - Config := Pool.WorkQueue.BackpressureConfig; - Config.MaxAttempts := 3; - Config.HighLoadDelay := 200; - Pool.WorkQueue.BackpressureConfig := Config; -end; -``` +`TryEnqueue(Item, TimeoutMS)` attempts an enqueue under the queue lock. If the +buffer is full, it waits on the not-full event and recalculates the remaining +deadline after every wake. `TryDequeue` signals not-full whenever it frees a +slot, and resets not-empty when the last item is removed. ---- +The inverse applies to consumers: enqueue signals not-empty, while the worker +waits without polling. Shutdown explicitly wakes all waiters after setting each +worker's `Terminated` flag. -## Implementation Details +Timeout rules are shared with the Simple pool: -### TryEnqueue — full implementation +- `0`: immediate attempt/check; +- finite value: milliseconds from the start of the call; +- `THREADPOOL_INFINITE`: no deadline. -```pascal -function TThreadSafeQueue.TryEnqueue(AItem: IWorkItem): boolean; -var - Attempts: Integer; -begin - if AItem = nil then Exit(False); +The legacy `TBackpressureConfig` record remains source-compatible. Threshold +and low/medium delay values no longer cause sleeps. New code should express its +deadline directly with `TryQueue`. - Attempts := 0; - while Attempts < FBackpressureConfig.MaxAttempts do - begin - ApplyBackpressure; // adaptive delay based on current LoadFactor +## Lifecycle - FLock.Enter; - try - if FCount < FCapacity then - begin - FItems[FTail] := AItem; - FTail := (FTail + 1) mod FCapacity; - Inc(FCount); - FLastEnqueueTime := Now; - Exit(True); // success - end; - finally - FLock.Leave; - end; +Both pools use the same monotonic state machine: - Inc(Attempts); - if Attempts < FBackpressureConfig.MaxAttempts then - Sleep(10); - end; - - // All attempts exhausted - raise EQueueFullException.Create(Format( - 'Queue is full after %d attempts (Capacity: %d)', - [FBackpressureConfig.MaxAttempts, FCapacity])); -end; +```text +tpsAccepting -> tpsDraining -> tpsStopped ``` -### Worker Execute loop - -```pascal -procedure TProducerConsumerWorkerThread.Execute; -var - Pool: TProducerConsumerThreadPool; - WorkItem: IWorkItem; -begin - Pool := TProducerConsumerThreadPool(FThreadPool); +`Shutdown` performs these steps: - while not Terminated do - begin - if Pool.FWorkQueue.TryDequeue(WorkItem) then - begin - try - WorkItem.Execute; - except - on E: Exception do - begin - Pool.FErrorLock.Enter; - try - Pool.SetLastError(E.Message); - finally - Pool.FErrorLock.Leave; - end; - end; - end; +1. Atomically stop new admission. +2. Wait for submissions that already passed admission to finish enqueueing. +3. Wait for all accepted tasks to finish. +4. Set `Terminated`, wake, join, and free every worker. +5. Publish `tpsStopped` and wake concurrent shutdown callers. - // Decrement counter; signal WaitForAll when last item finishes - Pool.FWorkItemLock.Enter; - try - Dec(Pool.FWorkItemCount); - if Pool.FWorkItemCount = 0 then - Pool.FCompletionEvent.SetEvent; - finally - Pool.FWorkItemLock.Leave; - end; - end - else - Sleep(100); // queue empty — avoid busy-waiting - end; -end; -``` +Queueing after step 1 raises `EThreadPoolShutdown`. `Shutdown` is idempotent, +and invoking it from one of the pool's own workers raises instead of deadlocking +that worker on itself. -### Destructor / cleanup sequence - -```pascal -destructor TProducerConsumerThreadPool.Destroy; -begin - ClearThreads; // sets Terminated, waits for all threads, frees them - FWorkQueue.Free; - FCompletionEvent.Free; - FErrorLock.Free; - FWorkItemLock.Free; - FThreads.Free; - inherited; -end; -``` +## Error containment ---- +Task exceptions remain asynchronous and are stored in `LastError` plus the +bounded `Errors` collection. `OnError` runs on the worker that observed the +failure. The base class catches exceptions raised by the handler so user +diagnostic code cannot terminate a worker or skip completion accounting. -## Thread Safety Summary +## Synchronization summary -| Object | Protects | +| Mechanism | Protects or signals | | --- | --- | -| `FLock` (in TThreadSafeQueue) | Queue head/tail/count during enqueue and dequeue | -| `FWorkItemLock` | `FWorkItemCount` and `FCompletionEvent` | -| `FErrorLock` | `FLastError` writes from worker threads | -| `FCompletionEvent` | Signals `WaitForAll` when all items are done | - ---- - -## Performance Considerations - -- **O(1)** enqueue and dequeue — circular buffer, no shifting -- Backpressure delays add latency on the producer side when the queue is busy; - tune `MaxAttempts` and delay values for your workload -- Workers sleep **100 ms** when the queue is empty — acceptable for batch - workloads, but not suitable for latency-sensitive tasks -- `DEBUG_LOG = True` by default; set it to `False` in production to eliminate - the overhead of timestamped console output - ---- +| Queue critical section | Circular buffer, head, tail, and count | +| Queue not-empty event | Sleeping consumers | +| Queue not-full event | Producers waiting for bounded capacity | +| Work-item critical section | Pending count and completion transition | +| Completion event | `WaitForAll`, including timeout overload | +| Base error lock | `LastError`, `Errors`, and `OnError` | +| Base lifecycle lock/events | Admission, active submitters, draining, stopped | + +## Performance characteristics + +- O(1) queue insertion and removal. +- No worker polling sleeps. +- No artificial delay while the queue still has space. +- Debug logging is disabled by default. +- A bounded queue still serializes its short state updates through one critical + section; batching very small tasks may improve throughput further. ## Limitations -- Fixed queue capacity — no dynamic resizing -- Only the most recent worker exception is stored (`LastError`) -- No task priority, ordering guarantees, or cancellation -- Thread count is fixed after construction — no dynamic scaling -- Debug logging writes to stdout; there is no log-level or handler API +- Queue capacity and worker count are fixed after construction. +- No task priorities, result-bearing futures, or task cancellation yet. +- `OnError` executes on a worker and should remain short and thread-safe. diff --git a/docs/ThreadPool.Simple-API.md b/docs/ThreadPool.Simple-API.md index 56b82fd..b32f559 100644 --- a/docs/ThreadPool.Simple-API.md +++ b/docs/ThreadPool.Simple-API.md @@ -29,6 +29,19 @@ GlobalThreadPool.Queue(@MyProcedure); GlobalThreadPool.WaitForAll; ``` +The timeout overload returns `False` if work remains after the requested number +of milliseconds: + +```pascal +if not GlobalThreadPool.WaitForAll(250) then + WriteLn('Still running'); +``` + +`0` is an immediate check. `THREADPOOL_INFINITE` waits indefinitely. + +`WaitForAll` does not close admission. Coordinate concurrent producers first, +or use `Shutdown` to atomically stop admission before draining. + ### TSimpleThreadPool A manually managed pool for when you need explicit control over thread count or lifetime. @@ -170,7 +183,24 @@ end; Pool.OnError := @Handler.OnTaskError; ``` -### Properties +Exceptions raised by the handler are contained by the pool. They cannot +terminate a worker or prevent task completion accounting. + +## Lifecycle and submission timeouts (v0.8.0) + +`TSimpleThreadPool` is unbounded, so `TryQueue` normally succeeds immediately; +the timeout argument exists so code can use either pool through `IThreadPool`. +After shutdown begins, `Queue` and `TryQueue` raise `EThreadPoolShutdown`. + +```pascal +Pool.TryQueue(@MyProcedure, 0); +Pool.Shutdown; // stop admission, drain accepted work, join workers +``` + +`State` progresses once from `tpsAccepting` to `tpsDraining` to `tpsStopped`. +`Shutdown` is idempotent and destruction calls it automatically. + +## Properties ```pascal property LastError: string; // Raw message of the most recent worker exception @@ -178,16 +208,20 @@ property Errors: TStringArray; // All captured messages (oldest first, c property ErrorCount: Integer; // Number of messages currently in Errors property OnError: TThreadPoolErrorEvent; // Fired (on a worker thread) per failed task property ThreadCount: Integer; // Number of worker threads (read-only) +property State: TThreadPoolState; // Accepting, draining, or stopped ``` -### Methods +## Methods ```pascal procedure Queue(AProcedure: TThreadProcedure); procedure Queue(AMethod: TThreadMethod); procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); -procedure WaitForAll; +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 ``` diff --git a/docs/ThreadPool.Simple-Technical.md b/docs/ThreadPool.Simple-Technical.md index b08058a..c221fe1 100644 --- a/docs/ThreadPool.Simple-Technical.md +++ b/docs/ThreadPool.Simple-Technical.md @@ -19,7 +19,7 @@ graph TD subgraph "Thread Pool" G & S --> |owns| TL[TThreadList] - G & S --> |owns| WQ[TThreadList<br>Work Queue] + G & S --> |owns| WQ[Dynamic circular<br>FIFO queue] G & S --> |owns| CS[TCriticalSection<br>Work Item Count] G & S --> |owns| EV[TEvent<br>Completion Signal] G & S --> |owns| EL[TCriticalSection<br>Error Lock] @@ -59,18 +59,18 @@ A singleton `TSimpleThreadPool` declared in the unit's `var` section. The main pool class. Responsibilities: - Owns and manages a list of `TSimpleWorkerThread` instances (`TThreadList`) -- Maintains a thread-safe work item queue (`TThreadList`) +- Maintains a dynamically growing O(1) circular FIFO queue - Tracks the number of pending work items with a `TCriticalSection` + counter - Signals `WaitForAll` callers via a `TEvent` when the counter reaches zero -- Captures worker exceptions thread-safely via a second `TCriticalSection` +- Captures worker exceptions and protects lifecycle transitions ### TSimpleWorkerThread Each worker thread: - Is created suspended and started explicitly by the pool constructor -- Loops continuously, popping work items from the shared queue -- Calls `Sleep(1)` when the queue is empty to avoid busy-waiting +- Blocks on `FWorkAvailableEvent` while the queue is empty +- Drains available FIFO work after an enqueue wakes it - Terminates cleanly when `Terminated` is set during pool destruction ### TSimpleWorkItem @@ -156,10 +156,12 @@ end; | Mechanism | Purpose | | --- | --- | | `TThreadList` (threads) | Safe iteration and termination of worker threads | -| `TThreadList` (work items) | Safe enqueue / dequeue across threads | +| `TCriticalSection` (FIFO queue) | Safe O(1) enqueue / dequeue across threads | | `TCriticalSection` (work item count) | Atomic increment/decrement of pending counter | | `TEvent` (completion) | Signals `WaitForAll` when counter hits zero | -| `TCriticalSection` (error lock) | Safe write to `FLastError` from worker threads | +| `TEvent` (work available) | Wakes idle workers without polling | +| Lifecycle lock/events | Serialize admission, draining, and stopped state | +| Error lock | Protects `LastError`, `Errors`, and `OnError` | --- @@ -167,28 +169,14 @@ end; ### How worker exceptions are captured -```pascal -// Inside TSimpleWorkerThread.Execute -try - WorkItem.Execute; -except - on E: Exception do - begin - Pool.FErrorLock.Enter; - try - Pool.SetLastError(E.Message); // stores raw message only - Pool.FErrorEvent.SetEvent; - finally - Pool.FErrorLock.Leave; - end; - end; -end; -``` +Workers store the task error through `TThreadPoolBase.SetLastError`. Completion +accounting runs in a separate `finally` block, and exceptions raised by an +`OnError` handler are contained at the callback boundary. ### Key behaviours - `LastError` stores the **raw exception message** — no thread ID prefix -- Only the **most recent** exception is kept; earlier ones are overwritten +- `LastError` keeps the most recent exception and `Errors` keeps the bounded history - The pool **keeps running** after an exception — remaining tasks are processed - Call `ClearLastError` before reusing a pool to reset error state - Exceptions are **not re-raised** on the calling thread; check `LastError` after `WaitForAll` @@ -197,21 +185,32 @@ end; 1. Always check `LastError` after `WaitForAll` 2. Call `ClearLastError` before queuing a new batch if reusing the pool -3. If you need to track all failures, collect them inside the task procedures themselves +3. Use `Errors` when a batch can contain more than one failure --- ## Performance Considerations -- Workers use `Sleep(1)` when idle — low CPU overhead but ~1 ms latency before a newly queued item is picked up +- Idle workers block on an event and are woken immediately by submission +- Queue insertion/removal is O(1); growth only occurs when the ring is full - For very large numbers of tiny tasks, consider batching them into fewer, larger work items to reduce queue overhead - Thread count defaults to `ProcessorCount`; raising it above that can hurt performance due to context-switching --- +## Lifecycle + +The shared base state moves from `tpsAccepting` to `tpsDraining` to +`tpsStopped`. `Shutdown` stops admission, waits for submissions already in +progress, drains every accepted task, wakes and joins the workers, and publishes +the stopped state. It is idempotent. Queueing after draining begins raises +`EThreadPoolShutdown`, and a worker attempting to shut down its own pool is +rejected before it can wait on itself. + +--- + ## Limitations -- Only the most recent worker exception is stored (no error queue) - Exceptions are not propagated to the main thread — must poll `LastError` - Thread count is fixed after construction — no dynamic scaling - No task prioritisation or cancellation diff --git a/docs/release-notes-v0.8.0.md b/docs/release-notes-v0.8.0.md new file mode 100644 index 0000000..f903d47 --- /dev/null +++ b/docs/release-notes-v0.8.0.md @@ -0,0 +1,63 @@ +# ThreadPool for Free Pascal — v0.8.0 + +> 0.8.0 makes threadpool-fp event-driven, deterministic under shutdown and +> failure, and measurably faster under burst and idle workloads. + +## Highlights + +- Both pools now have the same monotonic lifecycle and draining `Shutdown`. +- Idle workers block on events instead of polling with `Sleep`. +- The Simple pool uses a dynamically growing O(1) circular FIFO. +- The bounded pool signals not-empty and not-full transitions and only waits + when its queue is actually full. +- `WaitForAll(TimeoutMS)` and `TryQueue(..., TimeoutMS)` make deadlines explicit. +- Exceptions raised by `OnError` are contained at the callback boundary. +- Existing `Queue(...)` source continues to compile unchanged. +- Debug logging is off by default. + +## Lifecycle contract + +`Shutdown` changes `State` from `tpsAccepting` to `tpsDraining`, prevents new +admission, waits for submissions already in progress, drains all accepted work, +wakes and joins workers, and publishes `tpsStopped`. Repeated calls are safe. +Queueing after shutdown raises `EThreadPoolShutdown`. + +## Timeout contract + +Timeout values are milliseconds: + +- `0` performs an immediate attempt/check; +- finite values bound the call from entry; +- `THREADPOOL_INFINITE` waits without a deadline. + +The Simple pool is unbounded, so `TryQueue` normally returns immediately. The +Producer-Consumer pool returns `False` when no slot becomes available before +the deadline. + +## Verification + +The v0.8.0 suite contains 58 tests covering both legacy behaviour and new +lifecycle, timeout, callback-failure, queue saturation, concurrent +admission/shutdown, and invalid-construction cases. On the development machine +it completes in roughly four seconds with zero heap leaks reported by `heaptrc`. + +The benchmark in [`benchmarks/`](../benchmarks/) was compiled unchanged against +v0.7.0 and v0.8.0 with FPC 3.2.2 `-O3`. Debug logging was disabled in both. +Median of three Windows runs: + +| Workload | v0.7.0 | v0.8.0 | Change | +| --- | ---: | ---: | ---: | +| Simple, 20,000-task burst | 234 ms | 109 ms | 2.1× faster | +| Producer, 20,000-task burst | 3,953 ms | 172 ms | 23.0× faster | +| Simple idle queue-to-start | 7.9 ms | <1 ms | timer-resolution floor | +| Producer idle queue-to-start | 81.3 ms | <1 ms | timer-resolution floor | + +## Upgrade notes + +- Ordinary `Queue`/`WaitForAll` programs need no source changes. +- Code that submitted work after shutdown must now handle + `EThreadPoolShutdown`; the old Simple pool silently discarded that work. +- `TBackpressureConfig` remains available, but threshold and low/medium-delay + fields no longer sleep. Prefer `TryQueue(..., TimeoutMS)` in new code. + +See [CHANGELOG.md](../CHANGELOG.md) for the complete version history. diff --git a/package/lazarus/threadpool_fp.lpk b/package/lazarus/threadpool_fp.lpk index 5b8f839..7e6bd9b 100644 --- a/package/lazarus/threadpool_fp.lpk +++ b/package/lazarus/threadpool_fp.lpk @@ -30,7 +30,7 @@ This library provides two thread pool implementations, each with its own strengt uses ThreadPool.Simple; ``` - Global singleton instance for quick use -- Direct task execution +- Event-driven FIFO scheduling - Automatic thread count management - Best for simple parallel tasks - Lower memory overhead @@ -48,13 +48,14 @@ A thread pool with fixed-size circular buffer (1024 items) and built-in backpres - Configurable capacity (default: 1024 items) - **Backpressure Handling** - - Load-based adaptive delays (10ms to 100ms) - - Automatic retry mechanism (up to 5 attempts) - - Throws EQueueFullException when retries exhausted + - Event-driven not-full signalling + - Timeout-aware TryQueue overloads + - Queue raises EQueueFullException when its compatibility timeout expires -- **Monitoring & Debug** - - Thread-safe error capture with thread IDs - - Detailed debug logging (can be disabled) +- **Lifecycle & Error Safety** + - Deterministic draining shutdown + - Timeout-aware waits + - Debug logging disabled by default Best for: - High-volume task processing with rate control @@ -81,7 +82,7 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. "/> - <Version Minor="7"/> + <Version Minor="8"/> <Files> <Item> <Filename Value="..\..\src\ThreadPool.Simple.pas"/> diff --git a/src/ThreadPool.ProducerConsumer.pas b/src/ThreadPool.ProducerConsumer.pas index 6c5bc24..8cd157d 100644 --- a/src/ThreadPool.ProducerConsumer.pas +++ b/src/ThreadPool.ProducerConsumer.pas @@ -5,10 +5,10 @@ interface uses - Classes, SysUtils, ThreadPool.Types, SyncObjs; + Classes, SysUtils, Math, ThreadPool.Types, SyncObjs; -const - DEBUG_LOG = True; // Set to False to disable logging +var + DEBUG_LOG: Boolean = False; // Opt-in only; disabled by default procedure DebugLog(const Msg: string); @@ -50,16 +50,23 @@ TThreadSafeQueue = class(TObject) FCount: integer; FCapacity: integer; FLock: TCriticalSection; + FNotEmptyEvent: TEvent; + FNotFullEvent: TEvent; FLastEnqueueTime: TDateTime; FBackpressureConfig: TBackpressureConfig; protected function GetLoadFactor: Double; procedure ApplyBackpressure; + function GetDefaultTimeout: Cardinal; public constructor Create(ACapacity: integer); destructor Destroy; override; function TryEnqueue(AItem: IWorkItem): boolean; + function TryEnqueue(AItem: IWorkItem; + ATimeoutMS: Cardinal): boolean; overload; function TryDequeue(out AItem: IWorkItem): boolean; + function WaitForItem(ATimeoutMS: Cardinal): Boolean; + procedure WakeAll; function GetCount: integer; procedure Clear; property LoadFactor: Double read GetLoadFactor; @@ -100,19 +107,33 @@ TProducerConsumerThreadPool = class(TThreadPoolBase) FLocalThreadCount: integer; procedure ClearThreads; - function GetThreadCount: integer; override; - function GetLastError: string; override; - function TryQueueWorkItem(WorkItem: IWorkItem): Boolean; + function TryQueueWorkItem(WorkItem: IWorkItem; + ATimeoutMS: Cardinal): Boolean; + procedure CompleteWorkItem; + function IsCurrentWorkerThread: Boolean; public - constructor Create(AThreadCount: Integer = 0; AQueueSize: Integer = 1024); + constructor Create(AThreadCount: Integer = 0; + AQueueSize: Integer = 1024); reintroduce; destructor Destroy; override; + function GetThreadCount: integer; override; + function GetLastError: string; override; { IThreadPool implementation } procedure Queue(AProcedure: TThreadProcedure); override; procedure Queue(AMethod: TThreadMethod); override; procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: integer); override; procedure Queue(AMethod: TThreadMethodIndex; AIndex: integer); override; - procedure WaitForAll; override; + function TryQueue(AProcedure: TThreadProcedure; + ATimeoutMS: Cardinal): Boolean; overload; override; + function TryQueue(AMethod: TThreadMethod; + ATimeoutMS: Cardinal): Boolean; overload; override; + function TryQueue(AProcedure: TThreadProcedureIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; overload; override; + function TryQueue(AMethod: TThreadMethodIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; overload; override; + procedure WaitForAll; overload; override; + function WaitForAll(ATimeoutMS: Cardinal): Boolean; overload; override; + procedure Shutdown; override; property WorkQueue: TThreadSafeQueue read FWorkQueue; // Added this line property ThreadCount: integer read GetThreadCount; property LastError: string read GetLastError; @@ -142,6 +163,9 @@ constructor TProducerConsumerThreadPool.Create(AThreadCount: Integer = 0; AQueue DebugLog('Creating thread pool with ' + IntToStr(AThreadCount) + ' threads'); inherited Create(AThreadCount); + if AQueueSize <= 0 then + raise EArgumentOutOfRangeException.Create('Queue size must be greater than zero'); + // Set the local thread count to the base thread count FLocalThreadCount := FThreadCount; @@ -167,7 +191,7 @@ constructor TProducerConsumerThreadPool.Create(AThreadCount: Integer = 0; AQueue destructor TProducerConsumerThreadPool.Destroy; begin - ClearThreads; + Shutdown; FWorkQueue.Free; FCompletionEvent.Free; FWorkItemLock.Free; @@ -177,7 +201,7 @@ destructor TProducerConsumerThreadPool.Destroy; { Note: - The retry logic is in one place only: TThreadSafeQueue.TryEnqueue + Bounded wait logic is in one place only: TThreadSafeQueue.TryEnqueue Benefits of using IWorkItem: - Dependency Inversion: We depend on abstractions (interfaces) rather than concrete implementations @@ -186,10 +210,11 @@ destructor TProducerConsumerThreadPool.Destroy; - Testability: Easier to mock work items in unit tests - Interface Segregation: We only need the methods defined in IWorkItem } -function TProducerConsumerThreadPool.TryQueueWorkItem(WorkItem: IWorkItem): Boolean; +function TProducerConsumerThreadPool.TryQueueWorkItem(WorkItem: IWorkItem; + ATimeoutMS: Cardinal): Boolean; begin - Result := False; // Initialize Result to False - + Result := False; + FWorkItemLock.Enter; try Inc(FWorkItemCount); @@ -199,15 +224,10 @@ function TProducerConsumerThreadPool.TryQueueWorkItem(WorkItem: IWorkItem): Bool end; try - // Let TryEnqueue raise its exception directly - FWorkQueue.TryEnqueue(WorkItem); - Result := True; // If we get here, it succeeded - DebugLog(Format('Work item queued (Load: %.1f%%)', - [FWorkQueue.LoadFactor * 100])); - except - on E: Exception do + Result := FWorkQueue.TryEnqueue(WorkItem, ATimeoutMS); + finally + if not Result then begin - DebugLog('TryQueueWorkItem caught exception: ' + E.Message); FWorkItemLock.Enter; try Dec(FWorkItemCount); @@ -216,104 +236,166 @@ function TProducerConsumerThreadPool.TryQueueWorkItem(WorkItem: IWorkItem): Bool finally FWorkItemLock.Leave; end; - raise; // Re-raise the exception end; end; end; +procedure TProducerConsumerThreadPool.CompleteWorkItem; +begin + FWorkItemLock.Enter; + try + Dec(FWorkItemCount); + if FWorkItemCount = 0 then + FCompletionEvent.SetEvent; + finally + FWorkItemLock.Leave; + end; +end; -procedure TProducerConsumerThreadPool.Queue(AProcedure: TThreadProcedure); +function TProducerConsumerThreadPool.TryQueue(AProcedure: TThreadProcedure; + ATimeoutMS: Cardinal): Boolean; var - WorkItem: TProducerConsumerWorkItem; // Change back to class type - WorkItemIntf: IWorkItem; // Add interface variable -begin - WorkItem := TProducerConsumerWorkItem.Create(Self); - WorkItem.FProcedure := AProcedure; - WorkItem.FItemType := witProcedure; - WorkItemIntf := WorkItem; // Assign to interface (increases ref count) - + WorkItem: TProducerConsumerWorkItem; + WorkItemIntf: IWorkItem; +begin + BeginQueue; try - TryQueueWorkItem(WorkItemIntf); - except - on E:Exception do - begin - DebugLog('TProducerConsumerThreadPool.Queue: Exception caught: ' + E.Message); - raise; - end; + WorkItem := TProducerConsumerWorkItem.Create(Self); + WorkItem.FProcedure := AProcedure; + WorkItem.FItemType := witProcedure; + WorkItemIntf := WorkItem; + Result := TryQueueWorkItem(WorkItemIntf, ATimeoutMS); + finally + EndQueue; end; end; -procedure TProducerConsumerThreadPool.Queue(AMethod: TThreadMethod); +function TProducerConsumerThreadPool.TryQueue(AMethod: TThreadMethod; + ATimeoutMS: Cardinal): Boolean; var WorkItem: TProducerConsumerWorkItem; WorkItemIntf: IWorkItem; begin - WorkItem := TProducerConsumerWorkItem.Create(Self); - WorkItem.FMethod := AMethod; - WorkItem.FItemType := witMethod; - WorkItemIntf := WorkItem; - + BeginQueue; try - TryQueueWorkItem(WorkItemIntf); - except - on E: Exception do - begin - DebugLog('TProducerConsumerThreadPool.Queue: Exception caught: ' + E.Message); - raise; - end; + WorkItem := TProducerConsumerWorkItem.Create(Self); + WorkItem.FMethod := AMethod; + WorkItem.FItemType := witMethod; + WorkItemIntf := WorkItem; + Result := TryQueueWorkItem(WorkItemIntf, ATimeoutMS); + finally + EndQueue; end; end; -procedure TProducerConsumerThreadPool.Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); +function TProducerConsumerThreadPool.TryQueue( + AProcedure: TThreadProcedureIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; var WorkItem: TProducerConsumerWorkItem; WorkItemIntf: IWorkItem; begin - WorkItem := TProducerConsumerWorkItem.Create(Self); - WorkItem.FProcedureIndex := AProcedure; - WorkItem.FIndex := AIndex; - WorkItem.FItemType := witProcedureIndex; - WorkItemIntf := WorkItem; - + BeginQueue; try - TryQueueWorkItem(WorkItemIntf); - except - on E: Exception do - begin - DebugLog('TProducerConsumerThreadPool.Queue: Exception caught: ' + E.Message); - raise; - end; + WorkItem := TProducerConsumerWorkItem.Create(Self); + WorkItem.FProcedureIndex := AProcedure; + WorkItem.FIndex := AIndex; + WorkItem.FItemType := witProcedureIndex; + WorkItemIntf := WorkItem; + Result := TryQueueWorkItem(WorkItemIntf, ATimeoutMS); + finally + EndQueue; end; end; -procedure TProducerConsumerThreadPool.Queue(AMethod: TThreadMethodIndex; AIndex: Integer); +function TProducerConsumerThreadPool.TryQueue(AMethod: TThreadMethodIndex; + AIndex: Integer; ATimeoutMS: Cardinal): Boolean; var WorkItem: TProducerConsumerWorkItem; WorkItemIntf: IWorkItem; begin - WorkItem := TProducerConsumerWorkItem.Create(Self); - WorkItem.FMethodIndex := AMethod; - WorkItem.FIndex := AIndex; - WorkItem.FItemType := witMethodIndex; - WorkItemIntf := WorkItem; - + BeginQueue; try - TryQueueWorkItem(WorkItemIntf); - except - on E: Exception do - begin - DebugLog('TProducerConsumerThreadPool.Queue: Exception caught: ' + E.Message); - raise; - end; + WorkItem := TProducerConsumerWorkItem.Create(Self); + WorkItem.FMethodIndex := AMethod; + WorkItem.FIndex := AIndex; + WorkItem.FItemType := witMethodIndex; + WorkItemIntf := WorkItem; + Result := TryQueueWorkItem(WorkItemIntf, ATimeoutMS); + finally + EndQueue; end; end; +procedure TProducerConsumerThreadPool.Queue(AProcedure: TThreadProcedure); +begin + if not TryQueue(AProcedure, FWorkQueue.GetDefaultTimeout) then + raise EQueueFullException.Create('Queue is full (submission timed out)'); +end; + +procedure TProducerConsumerThreadPool.Queue(AMethod: TThreadMethod); +begin + if not TryQueue(AMethod, FWorkQueue.GetDefaultTimeout) then + raise EQueueFullException.Create('Queue is full (submission timed out)'); +end; + +procedure TProducerConsumerThreadPool.Queue(AProcedure: TThreadProcedureIndex; + AIndex: Integer); +begin + if not TryQueue(AProcedure, AIndex, FWorkQueue.GetDefaultTimeout) then + raise EQueueFullException.Create('Queue is full (submission timed out)'); +end; + +procedure TProducerConsumerThreadPool.Queue(AMethod: TThreadMethodIndex; + AIndex: Integer); +begin + if not TryQueue(AMethod, AIndex, FWorkQueue.GetDefaultTimeout) then + raise EQueueFullException.Create('Queue is full (submission timed out)'); +end; procedure TProducerConsumerThreadPool.WaitForAll; begin - DebugLog('Waiting for all work items to complete'); - FCompletionEvent.WaitFor(INFINITE); - DebugLog('All work items completed'); + WaitForAll(THREADPOOL_INFINITE); +end; + +function TProducerConsumerThreadPool.WaitForAll( + ATimeoutMS: Cardinal): Boolean; +begin + Result := FCompletionEvent.WaitFor(ATimeoutMS) = wrSignaled; +end; + +function TProducerConsumerThreadPool.IsCurrentWorkerThread: Boolean; +var + List: TList; + I: Integer; +begin + Result := False; + if not Assigned(FThreads) then + Exit; + List := FThreads.LockList; + try + for I := 0 to List.Count - 1 do + if TThread(List[I]).ThreadID = GetCurrentThreadID then + Exit(True); + finally + FThreads.UnlockList; + end; +end; + +procedure TProducerConsumerThreadPool.Shutdown; +begin + if IsCurrentWorkerThread then + raise EThreadPoolShutdown.Create('Shutdown cannot be called from a pool worker'); + if not BeginShutdown then + Exit; + try + if Assigned(FCompletionEvent) then + WaitForAll; + if Assigned(FThreads) then + ClearThreads; + finally + FinishShutdown; + end; end; procedure TProducerConsumerThreadPool.ClearThreads; @@ -329,6 +411,8 @@ procedure TProducerConsumerThreadPool.ClearThreads; Thread := TThread(List[I]); Thread.Terminate; end; + if Assigned(FWorkQueue) then + FWorkQueue.WakeAll; finally FThreads.UnlockList; end; @@ -396,11 +480,17 @@ function TProducerConsumerWorkItem.GetItemType: integer; constructor TThreadSafeQueue.Create(ACapacity: integer); begin inherited Create; + if ACapacity <= 0 then + raise EArgumentOutOfRangeException.Create('Queue capacity must be greater than zero'); FCapacity := ACapacity; SetLength(FItems, FCapacity); FHead := 0; FTail := 0; FCount := 0; + FNotEmptyEvent := TEvent.Create(nil, True, False, ''); + FNotFullEvent := TEvent.Create(nil, True, True, ''); + { Create the lock after both events so a partial constructor failure can be + destroyed without Clear touching a missing event. } FLock := TCriticalSection.Create; // Initialize default backpressure configuration @@ -415,33 +505,33 @@ constructor TThreadSafeQueue.Create(ACapacity: integer); end; procedure TThreadSafeQueue.ApplyBackpressure; +begin + { Kept for source compatibility. Backpressure is now event-driven: a + producer waits only while the bounded queue is actually full. } +end; + +function TThreadSafeQueue.GetDefaultTimeout: Cardinal; var - CurrentLoad: Double; - WaitTime: Integer; -begin - CurrentLoad := GetLoadFactor; - - // Determine wait time based on load thresholds - if CurrentLoad >= FBackpressureConfig.HighLoadThreshold then - WaitTime := FBackpressureConfig.HighLoadDelay - else if CurrentLoad >= FBackpressureConfig.MediumLoadThreshold then - WaitTime := FBackpressureConfig.MediumLoadDelay - else if CurrentLoad >= FBackpressureConfig.LowLoadThreshold then - WaitTime := FBackpressureConfig.LowLoadDelay + Attempts: Integer; + Total: QWord; +begin + Attempts := FBackpressureConfig.MaxAttempts; + if Attempts <= 1 then + Exit(0); + Total := QWord(Attempts) * QWord(Max(0, FBackpressureConfig.HighLoadDelay)) + + QWord(Attempts - 1) * 10; + if Total > High(Cardinal) - 1 then + Result := High(Cardinal) - 1 else - WaitTime := 0; - - if WaitTime > 0 then - begin - DebugLog(Format('Queue load at %.1f%%, applying backpressure: %dms', - [CurrentLoad * 100, WaitTime])); - Sleep(WaitTime); - end; + Result := Cardinal(Total); end; destructor TThreadSafeQueue.Destroy; begin - Clear; + if Assigned(FLock) then + Clear; + FNotFullEvent.Free; + FNotEmptyEvent.Free; FLock.Free; inherited; end; @@ -459,6 +549,9 @@ function TThreadSafeQueue.TryDequeue(out AItem: IWorkItem): boolean; FItems[FHead] := nil; FHead := (FHead + 1) mod FCapacity; Dec(FCount); + FNotFullEvent.SetEvent; + if FCount = 0 then + FNotEmptyEvent.ResetEvent; Result := True; end; finally @@ -487,11 +580,24 @@ procedure TThreadSafeQueue.Clear; FHead := 0; FTail := 0; FCount := 0; + FNotEmptyEvent.ResetEvent; + FNotFullEvent.SetEvent; finally FLock.Leave; end; end; +function TThreadSafeQueue.WaitForItem(ATimeoutMS: Cardinal): Boolean; +begin + Result := FNotEmptyEvent.WaitFor(ATimeoutMS) = wrSignaled; +end; + +procedure TThreadSafeQueue.WakeAll; +begin + FNotEmptyEvent.SetEvent; + FNotFullEvent.SetEvent; +end; + {$ENDREGION} {$REGION 'TProducerConsumerWorkerThread'} @@ -515,42 +621,24 @@ procedure TProducerConsumerWorkerThread.Execute; while not Terminated do begin - try - if Pool.FWorkQueue.TryDequeue(WorkItem) then - begin - DebugLog('Got work item'); + Pool.FWorkQueue.WaitForItem(INFINITE); + if Terminated then + Break; + + while (not Terminated) and Pool.FWorkQueue.TryDequeue(WorkItem) do + begin + try try WorkItem.Execute; - DebugLog('Work item executed'); except on E: Exception do - begin - DebugLog('Error executing work item: ' + E.Message); - // SetLastError is thread-safe on its own (base class locking), so - // no extra lock is needed. Calling it outside any subclass lock - // also ensures the OnError callback it may fire cannot deadlock. Pool.SetLastError(E.Message); - end; - end; - - Pool.FWorkItemLock.Enter; - try - Dec(Pool.FWorkItemCount); - DebugLog('Work items remaining: ' + IntToStr(Pool.FWorkItemCount)); - if Pool.FWorkItemCount = 0 then - begin - Pool.FCompletionEvent.SetEvent; - DebugLog('All work items completed'); - end; - finally - Pool.FWorkItemLock.Leave; end; - end - else - Sleep(100); // Wait a bit before trying again - except - on E: Exception do - DebugLog('Queue error: ' + E.Message); + finally + WorkItem := nil; + { This must execute even when the task or OnError callback fails. } + Pool.CompleteWorkItem; + end; end; end; DebugLog('Worker thread terminating'); @@ -569,52 +657,55 @@ function TThreadSafeQueue.GetLoadFactor: Double; end; function TThreadSafeQueue.TryEnqueue(AItem: IWorkItem): boolean; -var - Attempts: Integer; begin - DebugLog('TThreadSafeQueue.TryEnqueue: Starting'); + Result := TryEnqueue(AItem, GetDefaultTimeout); +end; - // Early validation +function TThreadSafeQueue.TryEnqueue(AItem: IWorkItem; + ATimeoutMS: Cardinal): boolean; +var + StartedAt, Elapsed: QWord; + Remaining: Cardinal; +begin if AItem = nil then - begin - DebugLog('TThreadSafeQueue.TryEnqueue: Nil item received'); Exit(False); - end; - - Attempts := 0; - - while Attempts < FBackpressureConfig.MaxAttempts do + + StartedAt := GetTickCount64; + repeat begin - ApplyBackpressure; - FLock.Enter; try if FCount < FCapacity then begin - DebugLog('TThreadSafeQueue.TryEnqueue: Adding item to queue'); FItems[FTail] := AItem; FTail := (FTail + 1) mod FCapacity; Inc(FCount); FLastEnqueueTime := Now; - Exit(True); // Success case + FNotEmptyEvent.SetEvent; + if FCount = FCapacity then + FNotFullEvent.ResetEvent; + Exit(True); end; - DebugLog(Format('TThreadSafeQueue.TryEnqueue: Queue full (Count: %d, Capacity: %d)', - [FCount, FCapacity])); finally FLock.Leave; end; - - Inc(Attempts); - if Attempts < FBackpressureConfig.MaxAttempts then - Sleep(10); // Only sleep if we're going to try again + + if ATimeoutMS = 0 then + Exit(False); + if ATimeoutMS = THREADPOOL_INFINITE then + Remaining := INFINITE + else + begin + Elapsed := GetTickCount64 - StartedAt; + if Elapsed >= ATimeoutMS then + Exit(False); + Remaining := ATimeoutMS - Cardinal(Elapsed); + end; + + if FNotFullEvent.WaitFor(Remaining) <> wrSignaled then + Exit(False); end; - - // If we get here, we've exhausted all attempts - DebugLog('TThreadSafeQueue.TryEnqueue: Max attempts reached'); - DebugLog('TThreadSafeQueue.TryEnqueue: Raising an exception: EQueueFullException'); - raise EQueueFullException.Create(Format( - 'Queue is full after %d attempts (Capacity: %d)', - [FBackpressureConfig.MaxAttempts, FCapacity])); + until False; end; end. diff --git a/src/ThreadPool.Simple.pas b/src/ThreadPool.Simple.pas index 6b69886..8722c74 100644 --- a/src/ThreadPool.Simple.pas +++ b/src/ThreadPool.Simple.pas @@ -56,13 +56,21 @@ TSimpleWorkerThread = class(TThread, IWorkerThread) TSimpleThreadPool = class(TThreadPoolBase) private FThreads: TThreadList; - FWorkItems: TThreadList; + FWorkItems: array of TSimpleWorkItem; + FQueueHead: Integer; + FQueueTail: Integer; + FQueueCount: Integer; + FQueueLock: TCriticalSection; FWorkItemLock: TCriticalSection; FWorkItemCount: Integer; FWorkItemEvent: TEvent; - FErrorEvent: TEvent; + FWorkAvailableEvent: TEvent; procedure ClearThreads; procedure ClearWorkItems; + procedure EnqueueWorkItem(AWorkItem: TSimpleWorkItem); + function TryDequeueWorkItem(out AWorkItem: TSimpleWorkItem): Boolean; + procedure CompleteWorkItem; + function IsCurrentWorkerThread: Boolean; public constructor Create(AThreadCount: Integer = 0); override; destructor Destroy; override; @@ -72,7 +80,17 @@ TSimpleThreadPool = class(TThreadPoolBase) procedure Queue(AMethod: TThreadMethod); override; procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); override; procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); override; - procedure WaitForAll; override; + function TryQueue(AProcedure: TThreadProcedure; + ATimeoutMS: Cardinal): Boolean; overload; override; + function TryQueue(AMethod: TThreadMethod; + ATimeoutMS: Cardinal): Boolean; overload; override; + function TryQueue(AProcedure: TThreadProcedureIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; overload; override; + function TryQueue(AMethod: TThreadMethodIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; overload; override; + procedure WaitForAll; overload; override; + function WaitForAll(ATimeoutMS: Cardinal): Boolean; overload; override; + procedure Shutdown; override; function GetThreadCount: Integer; override; function GetLastError: string; override; property ThreadCount: Integer read GetThreadCount; @@ -153,47 +171,32 @@ function TSimpleWorkerThread._Release: Integer; {$IFDEF WINDOWS}stdcall{$ELSE}cd procedure TSimpleWorkerThread.Execute; var Pool: TSimpleThreadPool; - List: TList; WorkItem: TSimpleWorkItem; begin Pool := TSimpleThreadPool(FThreadPool); - + while not Terminated do begin - // Try to get a work item - List := Pool.FWorkItems.LockList; - try - if List.Count > 0 then - begin - WorkItem := TSimpleWorkItem(List[0]); - List.Delete(0); - end - else - WorkItem := nil; - finally - Pool.FWorkItems.UnlockList; - end; + { Workers sleep without polling until submission or shutdown signals them. } + Pool.FWorkAvailableEvent.WaitFor(INFINITE); + if Terminated then + Break; - // Process work item if we got one - if Assigned(WorkItem) then + while (not Terminated) and Pool.TryDequeueWorkItem(WorkItem) do begin try - WorkItem.Execute; - except - on E: Exception do - begin - // Capture error. SetLastError is thread-safe on its own (base class - // locking), and TEvent.SetEvent is thread-safe, so no extra lock is - // needed here. Keeping SetLastError outside any subclass lock also - // ensures the OnError callback it may fire cannot deadlock. - Pool.SetLastError(E.Message); - Pool.FErrorEvent.SetEvent; + try + WorkItem.Execute; + except + on E: Exception do + Pool.SetLastError(E.Message); end; + finally + WorkItem.Free; + { Completion accounting is independent of task and callback failures. } + Pool.CompleteWorkItem; end; - WorkItem.Free; - end - else - Sleep(1); // Prevent busy waiting + end; end; end; @@ -217,28 +220,12 @@ destructor TSimpleWorkItem.Destroy; end; procedure TSimpleWorkItem.Execute; -var - Pool: TSimpleThreadPool; begin - try - // Execute the appropriate work item based on its type - case FItemType of - witProcedure: if Assigned(FProcedure) then FProcedure; - witMethod: if Assigned(FMethod) then FMethod; - witProcedureIndex: if Assigned(FProcedureIndex) then FProcedureIndex(FIndex); - witMethodIndex: if Assigned(FMethodIndex) then FMethodIndex(FIndex); - end; - finally - // Update work item count and signal completion if necessary - Pool := TSimpleThreadPool(FThreadPool); - Pool.FWorkItemLock.Enter; - try - Dec(Pool.FWorkItemCount); - if Pool.FWorkItemCount = 0 then - Pool.FWorkItemEvent.SetEvent; - finally - Pool.FWorkItemLock.Leave; - end; + case FItemType of + witProcedure: if Assigned(FProcedure) then FProcedure; + witMethod: if Assigned(FMethod) then FMethod; + witProcedureIndex: if Assigned(FProcedureIndex) then FProcedureIndex(FIndex); + witMethodIndex: if Assigned(FMethodIndex) then FMethodIndex(FIndex); end; end; @@ -259,16 +246,18 @@ constructor TSimpleThreadPool.Create(AThreadCount: Integer = 0); Thread: TSimpleWorkerThread; begin inherited Create(AThreadCount); - - // Initialize thread-safe collections and synchronization + FThreads := TThreadList.Create; - FWorkItems := TThreadList.Create; + FQueueLock := TCriticalSection.Create; FWorkItemLock := TCriticalSection.Create; - FErrorEvent := TEvent.Create(nil, True, False, ''); + FWorkAvailableEvent := TEvent.Create(nil, True, False, ''); FWorkItemEvent := TEvent.Create(nil, True, True, ''); + SetLength(FWorkItems, 64); + FQueueHead := 0; + FQueueTail := 0; + FQueueCount := 0; FWorkItemCount := 0; - // Create and start worker threads for I := 1 to FThreadCount do begin Thread := TSimpleWorkerThread.Create(Self); @@ -279,49 +268,138 @@ constructor TSimpleThreadPool.Create(AThreadCount: Integer = 0); destructor TSimpleThreadPool.Destroy; begin - // Only drain outstanding work if the pool was fully constructed. If the - // constructor failed partway, FPC still calls this destructor on the - // half-built object, so guard against the sync objects being nil. - if Assigned(FWorkItemEvent) and Assigned(FErrorEvent) then - WaitForAll; // Ensure all tasks complete before destroying - - FShutdown := True; - - if Assigned(FWorkItems) then - ClearWorkItems; - if Assigned(FThreads) then - ClearThreads; - - // Clean up synchronization objects (each may be nil after a partial - // construction, so Free — which is nil-safe — is used throughout). + Shutdown; + ClearWorkItems; + FQueueLock.Free; FWorkItemLock.Free; FThreads.Free; - FWorkItems.Free; + FWorkAvailableEvent.Free; FWorkItemEvent.Free; - FErrorEvent.Free; - inherited Destroy; end; +procedure TSimpleThreadPool.EnqueueWorkItem(AWorkItem: TSimpleWorkItem); +var + NewItems: array of TSimpleWorkItem; + I, NewCapacity: Integer; +begin + NewItems := nil; + FQueueLock.Enter; + try + if FQueueCount = Length(FWorkItems) then + begin + NewCapacity := Length(FWorkItems) * 2; + if NewCapacity = 0 then + NewCapacity := 64; + SetLength(NewItems, NewCapacity); + for I := 0 to FQueueCount - 1 do + NewItems[I] := FWorkItems[(FQueueHead + I) mod Length(FWorkItems)]; + FWorkItems := NewItems; + FQueueHead := 0; + FQueueTail := FQueueCount; + end; + + FWorkItems[FQueueTail] := AWorkItem; + FQueueTail := (FQueueTail + 1) mod Length(FWorkItems); + Inc(FQueueCount); + + FWorkItemLock.Enter; + try + Inc(FWorkItemCount); + FWorkItemEvent.ResetEvent; + finally + FWorkItemLock.Leave; + end; + FWorkAvailableEvent.SetEvent; + finally + FQueueLock.Leave; + end; +end; + +function TSimpleThreadPool.TryDequeueWorkItem( + out AWorkItem: TSimpleWorkItem): Boolean; +begin + FQueueLock.Enter; + try + Result := FQueueCount > 0; + if Result then + begin + AWorkItem := FWorkItems[FQueueHead]; + FWorkItems[FQueueHead] := nil; + FQueueHead := (FQueueHead + 1) mod Length(FWorkItems); + Dec(FQueueCount); + if FQueueCount = 0 then + FWorkAvailableEvent.ResetEvent; + end + else + AWorkItem := nil; + finally + FQueueLock.Leave; + end; +end; + +procedure TSimpleThreadPool.CompleteWorkItem; +begin + FWorkItemLock.Enter; + try + Dec(FWorkItemCount); + if FWorkItemCount = 0 then + FWorkItemEvent.SetEvent; + finally + FWorkItemLock.Leave; + end; +end; + +function TSimpleThreadPool.IsCurrentWorkerThread: Boolean; +var + List: TList; + I: Integer; +begin + Result := False; + if not Assigned(FThreads) then + Exit; + List := FThreads.LockList; + try + for I := 0 to List.Count - 1 do + if TThread(List[I]).ThreadID = GetCurrentThreadID then + Exit(True); + finally + FThreads.UnlockList; + end; +end; + +procedure TSimpleThreadPool.Shutdown; +begin + if IsCurrentWorkerThread then + raise EThreadPoolShutdown.Create('Shutdown cannot be called from a pool worker'); + if not BeginShutdown then + Exit; + try + if Assigned(FWorkItemEvent) then + WaitForAll; + if Assigned(FThreads) then + ClearThreads; + finally + FinishShutdown; + end; +end; + procedure TSimpleThreadPool.ClearThreads; var Thread: TSimpleWorkerThread; List: TList; I: Integer; begin - // Signal all threads to terminate List := FThreads.LockList; try for I := 0 to List.Count - 1 do - begin - Thread := TSimpleWorkerThread(List[I]); - Thread.Terminate; - end; + TSimpleWorkerThread(List[I]).Terminate; + if Assigned(FWorkAvailableEvent) then + FWorkAvailableEvent.SetEvent; finally FThreads.UnlockList; end; - // Wait for all threads to finish and clean up List := FThreads.LockList; try for I := 0 to List.Count - 1 do @@ -338,120 +416,135 @@ procedure TSimpleThreadPool.ClearThreads; procedure TSimpleThreadPool.ClearWorkItems; var - List: TList; - I: Integer; + WorkItem: TSimpleWorkItem; begin - // Clean up any remaining work items - List := FWorkItems.LockList; - try - for I := 0 to List.Count - 1 do - TSimpleWorkItem(List[I]).Free; - List.Clear; - finally - FWorkItems.UnlockList; - end; + if not Assigned(FQueueLock) then + Exit; + while TryDequeueWorkItem(WorkItem) do + WorkItem.Free; + SetLength(FWorkItems, 0); end; -{ Queue overloads for different types of work items } - -procedure TSimpleThreadPool.Queue(AProcedure: TThreadProcedure); +function TSimpleThreadPool.TryQueue(AProcedure: TThreadProcedure; + ATimeoutMS: Cardinal): Boolean; var WorkItem: TSimpleWorkItem; begin - if FShutdown then Exit; // Don't queue if shutting down - - // Update work item count - FWorkItemLock.Enter; + BeginQueue; try - Inc(FWorkItemCount); - if FWorkItemCount > 0 then - FWorkItemEvent.ResetEvent; // Reset completion event + WorkItem := TSimpleWorkItem.Create(Self); + try + WorkItem.FProcedure := AProcedure; + WorkItem.FItemType := witProcedure; + EnqueueWorkItem(WorkItem); + WorkItem := nil; + Result := True; + finally + WorkItem.Free; + end; finally - FWorkItemLock.Leave; + EndQueue; end; - - // Create and queue work item - WorkItem := TSimpleWorkItem.Create(Self); - WorkItem.FProcedure := AProcedure; - WorkItem.FItemType := witProcedure; - - FWorkItems.Add(WorkItem); end; -procedure TSimpleThreadPool.Queue(AMethod: TThreadMethod); +function TSimpleThreadPool.TryQueue(AMethod: TThreadMethod; + ATimeoutMS: Cardinal): Boolean; var WorkItem: TSimpleWorkItem; begin - if FShutdown then Exit; - - FWorkItemLock.Enter; + BeginQueue; try - Inc(FWorkItemCount); - if FWorkItemCount > 0 then - FWorkItemEvent.ResetEvent; + WorkItem := TSimpleWorkItem.Create(Self); + try + WorkItem.FMethod := AMethod; + WorkItem.FItemType := witMethod; + EnqueueWorkItem(WorkItem); + WorkItem := nil; + Result := True; + finally + WorkItem.Free; + end; finally - FWorkItemLock.Leave; + EndQueue; end; - - WorkItem := TSimpleWorkItem.Create(Self); - WorkItem.FMethod := AMethod; - WorkItem.FItemType := witMethod; - - FWorkItems.Add(WorkItem); end; -procedure TSimpleThreadPool.Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); +function TSimpleThreadPool.TryQueue(AProcedure: TThreadProcedureIndex; + AIndex: Integer; ATimeoutMS: Cardinal): Boolean; var WorkItem: TSimpleWorkItem; begin - if FShutdown then Exit; - - FWorkItemLock.Enter; + BeginQueue; try - Inc(FWorkItemCount); - if FWorkItemCount > 0 then - FWorkItemEvent.ResetEvent; + WorkItem := TSimpleWorkItem.Create(Self); + try + WorkItem.FProcedureIndex := AProcedure; + WorkItem.FIndex := AIndex; + WorkItem.FItemType := witProcedureIndex; + EnqueueWorkItem(WorkItem); + WorkItem := nil; + Result := True; + finally + WorkItem.Free; + end; finally - FWorkItemLock.Leave; + EndQueue; end; - - WorkItem := TSimpleWorkItem.Create(Self); - WorkItem.FProcedureIndex := AProcedure; - WorkItem.FIndex := AIndex; - WorkItem.FItemType := witProcedureIndex; - - FWorkItems.Add(WorkItem); end; -procedure TSimpleThreadPool.Queue(AMethod: TThreadMethodIndex; AIndex: Integer); +function TSimpleThreadPool.TryQueue(AMethod: TThreadMethodIndex; + AIndex: Integer; ATimeoutMS: Cardinal): Boolean; var WorkItem: TSimpleWorkItem; begin - if FShutdown then Exit; - - FWorkItemLock.Enter; + BeginQueue; try - Inc(FWorkItemCount); - if FWorkItemCount > 0 then - FWorkItemEvent.ResetEvent; + WorkItem := TSimpleWorkItem.Create(Self); + try + WorkItem.FMethodIndex := AMethod; + WorkItem.FIndex := AIndex; + WorkItem.FItemType := witMethodIndex; + EnqueueWorkItem(WorkItem); + WorkItem := nil; + Result := True; + finally + WorkItem.Free; + end; finally - FWorkItemLock.Leave; + EndQueue; end; - - WorkItem := TSimpleWorkItem.Create(Self); - WorkItem.FMethodIndex := AMethod; - WorkItem.FIndex := AIndex; - WorkItem.FItemType := witMethodIndex; - - FWorkItems.Add(WorkItem); +end; + +procedure TSimpleThreadPool.Queue(AProcedure: TThreadProcedure); +begin + TryQueue(AProcedure, 0); +end; + +procedure TSimpleThreadPool.Queue(AMethod: TThreadMethod); +begin + TryQueue(AMethod, 0); +end; + +procedure TSimpleThreadPool.Queue(AProcedure: TThreadProcedureIndex; + AIndex: Integer); +begin + TryQueue(AProcedure, AIndex, 0); +end; + +procedure TSimpleThreadPool.Queue(AMethod: TThreadMethodIndex; + AIndex: Integer); +begin + TryQueue(AMethod, AIndex, 0); end; procedure TSimpleThreadPool.WaitForAll; begin - FWorkItemEvent.WaitFor(INFINITE); // Wait for all work items to complete - // If there was an error, ensure it's fully captured - if FErrorEvent.WaitFor(100) = wrSignaled then - FErrorEvent.ResetEvent; + WaitForAll(THREADPOOL_INFINITE); +end; + +function TSimpleThreadPool.WaitForAll(ATimeoutMS: Cardinal): Boolean; +begin + Result := FWorkItemEvent.WaitFor(ATimeoutMS) = wrSignaled; end; function TSimpleThreadPool.GetThreadCount: Integer; @@ -467,9 +560,9 @@ function TSimpleThreadPool.GetLastError: string; {$ENDREGION} initialization - GlobalThreadPool := TSimpleThreadPool.Create; // Create global instance + GlobalThreadPool := TSimpleThreadPool.Create; finalization - GlobalThreadPool.Free; // Clean up global instance + GlobalThreadPool.Free; end. diff --git a/src/threadpool.types.pas b/src/threadpool.types.pas index ce576ad..058c5f4 100644 --- a/src/threadpool.types.pas +++ b/src/threadpool.types.pas @@ -14,7 +14,20 @@ interface of this cap. } MAX_STORED_ERRORS = 1000; + { Used by timeout-aware queue and wait operations. } + THREADPOOL_INFINITE = High(Cardinal); + type + EThreadPoolShutdown = class(Exception); + + { A pool accepts work until Shutdown begins, drains every accepted task, and + then becomes permanently stopped. } + TThreadPoolState = ( + tpsAccepting, + tpsDraining, + tpsStopped + ); + { Task Types - Different kinds of work that can be queued } TThreadProcedure = procedure; TThreadMethod = procedure of object; @@ -71,7 +84,17 @@ interface procedure Queue(AMethod: TThreadMethod); overload; procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); overload; procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); overload; - procedure WaitForAll; + function TryQueue(AProcedure: TThreadProcedure; + ATimeoutMS: Cardinal): Boolean; overload; + function TryQueue(AMethod: TThreadMethod; + ATimeoutMS: Cardinal): Boolean; overload; + function TryQueue(AProcedure: TThreadProcedureIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; overload; + function TryQueue(AMethod: TThreadMethodIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; overload; + procedure WaitForAll; overload; + function WaitForAll(ATimeoutMS: Cardinal): Boolean; overload; + procedure Shutdown; procedure ClearLastError; procedure ClearErrors; function GetLastError: string; @@ -80,6 +103,7 @@ interface function GetErrorCount: Integer; function GetOnError: TThreadPoolErrorEvent; procedure SetOnError(AValue: TThreadPoolErrorEvent); + function GetState: TThreadPoolState; property LastError: string read GetLastError; property ThreadCount: Integer read GetThreadCount; { All task error messages captured since the last ClearErrors/ClearLastError, @@ -88,6 +112,7 @@ interface property ErrorCount: Integer read GetErrorCount; { Optional callback fired from a worker thread whenever a task raises. } property OnError: TThreadPoolErrorEvent read GetOnError write SetOnError; + property State: TThreadPoolState read GetState; end; { Base class for thread pool implementations } @@ -102,8 +127,17 @@ TThreadPoolBase = class(TInterfacedObject, IThreadPool) FErrors: TStringList; FErrorsLock: TCriticalSection; FOnError: TThreadPoolErrorEvent; + FLifecycleLock: TCriticalSection; + FNoSubmittersEvent: TEvent; + FStoppedEvent: TEvent; + FState: TThreadPoolState; + FActiveSubmitters: Integer; procedure SetLastError(const AError: string); virtual; + procedure BeginQueue; + procedure EndQueue; + function BeginShutdown: Boolean; + procedure FinishShutdown; public constructor Create(AThreadCount: Integer); virtual; destructor Destroy; override; @@ -113,7 +147,17 @@ TThreadPoolBase = class(TInterfacedObject, IThreadPool) procedure Queue(AMethod: TThreadMethod); virtual; abstract; procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); virtual; abstract; procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); virtual; abstract; + function TryQueue(AProcedure: TThreadProcedure; + ATimeoutMS: Cardinal): Boolean; virtual; abstract; + function TryQueue(AMethod: TThreadMethod; + ATimeoutMS: Cardinal): Boolean; virtual; abstract; + function TryQueue(AProcedure: TThreadProcedureIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; virtual; abstract; + function TryQueue(AMethod: TThreadMethodIndex; AIndex: Integer; + ATimeoutMS: Cardinal): Boolean; virtual; abstract; procedure WaitForAll; virtual; abstract; + function WaitForAll(ATimeoutMS: Cardinal): Boolean; virtual; abstract; + procedure Shutdown; virtual; abstract; procedure ClearLastError; virtual; procedure ClearErrors; virtual; function GetLastError: string; virtual; @@ -122,6 +166,7 @@ TThreadPoolBase = class(TInterfacedObject, IThreadPool) function GetErrorCount: Integer; virtual; function GetOnError: TThreadPoolErrorEvent; virtual; procedure SetOnError(AValue: TThreadPoolErrorEvent); virtual; + function GetState: TThreadPoolState; virtual; { All task error messages captured since the last ClearErrors/ClearLastError, oldest first, capped at MAX_STORED_ERRORS. } @@ -129,6 +174,7 @@ TThreadPoolBase = class(TInterfacedObject, IThreadPool) property ErrorCount: Integer read GetErrorCount; { Optional callback fired from a worker thread whenever a task raises. } property OnError: TThreadPoolErrorEvent read GetOnError write SetOnError; + property State: TThreadPoolState read GetState; end; implementation @@ -143,6 +189,11 @@ constructor TThreadPoolBase.Create(AThreadCount: Integer); FErrors := TStringList.Create; FErrorsLock := TCriticalSection.Create; FOnError := nil; + FLifecycleLock := TCriticalSection.Create; + FNoSubmittersEvent := TEvent.Create(nil, True, True, ''); + FStoppedEvent := TEvent.Create(nil, True, False, ''); + FState := tpsAccepting; + FActiveSubmitters := 0; // Apply thread count safety limits if AThreadCount <= 0 then @@ -155,6 +206,9 @@ constructor TThreadPoolBase.Create(AThreadCount: Integer); destructor TThreadPoolBase.Destroy; begin FShutdown := True; + FStoppedEvent.Free; + FNoSubmittersEvent.Free; + FLifecycleLock.Free; FErrors.Free; FErrorsLock.Free; inherited; @@ -179,7 +233,73 @@ procedure TThreadPoolBase.SetLastError(const AError: string); // Fire the callback outside the lock so a handler cannot deadlock by calling // back into the pool, and so a slow handler does not block other workers. if Assigned(Handler) then - Handler(AError); + begin + { A diagnostic callback must never be able to terminate a worker or skip + its completion accounting. Callback exceptions are deliberately + contained at this boundary. } + try + Handler(AError); + except + { Ignore callback failures. The original task error is already stored. } + end; + end; +end; + +procedure TThreadPoolBase.BeginQueue; +begin + FLifecycleLock.Enter; + try + if FState <> tpsAccepting then + raise EThreadPoolShutdown.Create('Thread pool is shutting down'); + Inc(FActiveSubmitters); + if FActiveSubmitters = 1 then + FNoSubmittersEvent.ResetEvent; + finally + FLifecycleLock.Leave; + end; +end; + +procedure TThreadPoolBase.EndQueue; +begin + FLifecycleLock.Enter; + try + Dec(FActiveSubmitters); + if FActiveSubmitters = 0 then + FNoSubmittersEvent.SetEvent; + finally + FLifecycleLock.Leave; + end; +end; + +function TThreadPoolBase.BeginShutdown: Boolean; +begin + FLifecycleLock.Enter; + try + Result := FState = tpsAccepting; + if Result then + begin + FState := tpsDraining; + FShutdown := True; + end; + finally + FLifecycleLock.Leave; + end; + + if Result then + FNoSubmittersEvent.WaitFor(INFINITE) + else if GetState = tpsDraining then + FStoppedEvent.WaitFor(INFINITE); +end; + +procedure TThreadPoolBase.FinishShutdown; +begin + FLifecycleLock.Enter; + try + FState := tpsStopped; + FStoppedEvent.SetEvent; + finally + FLifecycleLock.Leave; + end; end; procedure TThreadPoolBase.ClearLastError; @@ -217,6 +337,7 @@ function TThreadPoolBase.GetErrors: TStringArray; var I: Integer; begin + Result := nil; FErrorsLock.Enter; try SetLength(Result, FErrors.Count); @@ -257,4 +378,14 @@ procedure TThreadPoolBase.SetOnError(AValue: TThreadPoolErrorEvent); end; end; +function TThreadPoolBase.GetState: TThreadPoolState; +begin + FLifecycleLock.Enter; + try + Result := FState; + finally + FLifecycleLock.Leave; + end; +end; + end. diff --git a/tests/ThreadPool.ProducerConsumer.Tests.pas b/tests/ThreadPool.ProducerConsumer.Tests.pas index 05ff3a2..616b6b9 100644 --- a/tests/ThreadPool.ProducerConsumer.Tests.pas +++ b/tests/ThreadPool.ProducerConsumer.Tests.pas @@ -6,7 +6,7 @@ interface uses Classes, SysUtils, fpcunit, testregistry, ThreadPool.Types, - ThreadPool.ProducerConsumer, syncobjs, DateUtils, Math; + ThreadPool.ProducerConsumer, syncobjs, DateUtils; type { TTestProducerConsumerThreadPool } @@ -17,6 +17,9 @@ TTestProducerConsumerThreadPool = class(TTestCase) FSharedLock: TCriticalSection; FTestResults: TStringList; FOnErrorCount: Integer; + FGateEvent: TEvent; + FAllStartedEvent: TEvent; + FStartedCount: Integer; procedure LogTest(const Msg: string); procedure IncrementCounter; @@ -28,6 +31,9 @@ TTestProducerConsumerThreadPool = class(TTestCase) procedure RaiseTestError; procedure RaiseOtherError; procedure HandlePoolError(const AMessage: string); + procedure HandlePoolErrorAndRaise(const AMessage: string); + procedure GateTask; + procedure ShutdownFromWorker; procedure SleepTask; function MeasureQueueTime(const TaskCount: Integer): Int64; protected @@ -47,12 +53,38 @@ TTestProducerConsumerThreadPool = class(TTestCase) procedure Test11_BackpressureConfig; procedure Test12_LoadFactorCalculation; procedure Test13_BackpressureBehavior; - procedure Test14_AdaptivePerformance; + procedure Test14_ParallelScaling; // Error-collection API (v0.7.0) procedure Test15_ErrorsCollectionCapturesAll; procedure Test16_OnErrorCallbackFires; procedure Test17_ClearErrorsResetsCollection; + + // Lifecycle, timeout, and callback safety (v0.8.0) + procedure Test18_WaitTimeout; + procedure Test19_ShutdownDrainsAndStops; + procedure Test20_QueueAfterShutdownRaises; + procedure Test21_OnErrorExceptionIsContained; + procedure Test22_TryQueueTimeoutWhenFull; + procedure Test23_ConcurrentQueueAndShutdown; + procedure Test24_InvalidQueueSizeRejected; + procedure Test25_WorkerShutdownCannotDeadlock; + end; + + TProducerLifecycleQueueThread = class(TThread) + private + FPool: TProducerConsumerThreadPool; + FStartEvent: TEvent; + FFirstQueuedEvent: TEvent; + FTask: TThreadMethod; + FAccepted: Integer; + FUnexpectedError: string; + public + constructor Create(APool: TProducerConsumerThreadPool; ATask: TThreadMethod; + AStartEvent, AFirstQueuedEvent: TEvent); + procedure Execute; override; + property Accepted: Integer read FAccepted; + property UnexpectedError: string read FUnexpectedError; end; implementation @@ -60,7 +92,42 @@ implementation const TASK_COUNT = 100; PARALLEL_TASKS = 1000; - STRESS_TEST_TASKS = 2048; + +constructor TProducerLifecycleQueueThread.Create( + APool: TProducerConsumerThreadPool; ATask: TThreadMethod; + AStartEvent, AFirstQueuedEvent: TEvent); +begin + inherited Create(True); + FPool := APool; + FTask := ATask; + FStartEvent := AStartEvent; + FFirstQueuedEvent := AFirstQueuedEvent; + FreeOnTerminate := False; +end; + +procedure TProducerLifecycleQueueThread.Execute; +var + I: Integer; +begin + FStartEvent.WaitFor(INFINITE); + try + for I := 1 to 10000 do + begin + try + FPool.Queue(FTask); + Inc(FAccepted); + if FAccepted = 1 then + FFirstQueuedEvent.SetEvent; + except + on E: EThreadPoolShutdown do + Break; + end; + end; + except + on E: Exception do + FUnexpectedError := E.ClassName + ': ' + E.Message; + end; +end; { TTestProducerConsumerThreadPool } @@ -76,6 +143,9 @@ procedure TTestProducerConsumerThreadPool.SetUp; FSharedCounter := 0; FSharedLock := TCriticalSection.Create; FTestResults := TStringList.Create; + FGateEvent := TEvent.Create(nil, True, False, ''); + FAllStartedEvent := TEvent.Create(nil, True, False, ''); + FStartedCount := 0; LogTest('Test setup complete'); end; @@ -83,6 +153,8 @@ procedure TTestProducerConsumerThreadPool.TearDown; begin LogTest('Tearing down test...'); FThreadPool.Free; + FAllStartedEvent.Free; + FGateEvent.Free; FSharedLock.Free; FTestResults.Free; LogTest('Test teardown complete'); @@ -132,7 +204,7 @@ procedure TTestProducerConsumerThreadPool.ProcessMethodWithIndex(AIndex: Integer procedure TTestProducerConsumerThreadPool.LongTask; begin - Sleep(5000); + Sleep(250); end; procedure TTestProducerConsumerThreadPool.SlowTask; @@ -156,6 +228,25 @@ procedure TTestProducerConsumerThreadPool.HandlePoolError(const AMessage: string InterlockedIncrement(FOnErrorCount); end; +procedure TTestProducerConsumerThreadPool.HandlePoolErrorAndRaise( + const AMessage: string); +begin + InterlockedIncrement(FOnErrorCount); + raise Exception.Create('OnError handler failure'); +end; + +procedure TTestProducerConsumerThreadPool.GateTask; +begin + if InterlockedIncrement(FStartedCount) = 4 then + FAllStartedEvent.SetEvent; + FGateEvent.WaitFor(INFINITE); +end; + +procedure TTestProducerConsumerThreadPool.ShutdownFromWorker; +begin + FThreadPool.Shutdown; +end; + procedure TTestProducerConsumerThreadPool.SleepTask; begin Sleep(100); @@ -286,16 +377,7 @@ procedure TTestProducerConsumerThreadPool.Test07_ParallelExecution; LogTest('Test07_ParallelExecution finished'); end; -{ - Test08_QueueFullBehavior - - Previously, Test08 expected an exception. With adaptive backpressure, it - should instead slow down rather than fail. - - Also, Since we now have two different but valid error messages - ("Queue is full" and "Queue is full after maximum attempts"), the assert - in Test08 must accept either message. -} +{ Legacy Queue remains exception-based when its bounded wait expires. } procedure TTestProducerConsumerThreadPool.Test08_QueueFullBehavior; const QUEUE_SIZE = 2; @@ -455,61 +537,41 @@ procedure TTestProducerConsumerThreadPool.Test12_LoadFactorCalculation; procedure TTestProducerConsumerThreadPool.Test13_BackpressureBehavior; var + Pool: TProducerConsumerThreadPool; StartTime: TDateTime; ElapsedMS: Int64; I: Integer; - Config: TBackpressureConfig; begin LogTest('Test13_BackpressureBehavior starting...'); - - // Configure more aggressive backpressure for testing - Config := FThreadPool.WorkQueue.BackpressureConfig; - Config.LowLoadThreshold := 0.3; - Config.LowLoadDelay := 50; // Increased delay - Config.HighLoadThreshold := 0.7; - Config.HighLoadDelay := 100; // Increased delay - FThreadPool.WorkQueue.BackpressureConfig := Config; - - // Measure time with high load - StartTime := Now; - for I := 1 to 500 do // More tasks - FThreadPool.Queue(@SleepTask); - ElapsedMS := MilliSecondsBetween(Now, StartTime); - - // Should have significant delay due to backpressure - AssertTrue('High load should trigger backpressure', - ElapsedMS > Config.HighLoadDelay); - - LogTest(Format('Elapsed time under load: %d ms', [ElapsedMS])); - FThreadPool.WaitForAll; + + { Fill every worker and the single queue slot. Event-driven backpressure + should block only because the queue is actually full, then time out. } + FGateEvent.ResetEvent; + FAllStartedEvent.ResetEvent; + FStartedCount := 0; + Pool := TProducerConsumerThreadPool.Create(4, 1); + try + for I := 1 to 4 do + Pool.Queue(@GateTask); + AssertEquals('All workers should start their gate task', Ord(wrSignaled), + Ord(FAllStartedEvent.WaitFor(2000))); + Pool.Queue(@GateTask); // occupies the only queue slot + + StartTime := Now; + AssertFalse('TryQueue should time out while the queue remains full', + Pool.TryQueue(@GateTask, 50)); + ElapsedMS := MilliSecondsBetween(Now, StartTime); + AssertTrue('Full-queue wait should honour its timeout', ElapsedMS >= 40); + finally + FGateEvent.SetEvent; + Pool.WaitForAll; + Pool.Free; + end; LogTest('Test13_BackpressureBehavior finished'); end; -{ - Test14_AdaptivePerformance evaluates the thread pool's ability to maintain efficient performance - under varying load conditions. The test performs the following steps: - - 1. **Configuration**: It starts by disabling backpressure to ensure that delays introduced by - backpressure do not affect the measurement of task processing times. - - 2. **Low Load Test**: The test queues a single task (`LOW_LOAD_TASKS`) and measures the time - taken to complete it. This establishes a baseline for the thread pool's performance under - minimal load. - - 3. **High Load Test**: It then queues a large number of tasks (`HIGH_LOAD_TASKS`) and measures - the time taken to process all of them. This simulates a high-load scenario to assess how - well the thread pool scales with increased workload. - - 4. **Performance Analysis**: The test calculates the normalized time per task for both low - and high load scenarios. By computing the ratio of normalized low load time to high load - time, it determines the slowdown factor, which indicates how much the performance degrades - under high load. - - The purpose of this test is to ensure that the thread pool can adapt to different levels of - workload without significant performance penalties, thereby validating its scalability and - robustness in handling varying task loads. -} -procedure TTestProducerConsumerThreadPool.Test14_AdaptivePerformance; +{ Verify that independent tasks scale across the fixed worker set. } +procedure TTestProducerConsumerThreadPool.Test14_ParallelScaling; const LOW_LOAD_TASKS = 1; // Single task HIGH_LOAD_TASKS = 32; // Many more tasks @@ -521,18 +583,10 @@ procedure TTestProducerConsumerThreadPool.Test14_AdaptivePerformance; NormalizedLowTime: Double; NormalizedHighTime: Double; Ratio: Double; - Config: TBackpressureConfig; begin - LogTest('Test14_AdaptivePerformance starting...'); + LogTest('Test14_ParallelScaling starting...'); LogTest(Format('Thread count: %d', [FThreadPool.ThreadCount])); - // Disable backpressure for cleaner measurements - Config := FThreadPool.WorkQueue.BackpressureConfig; - Config.LowLoadDelay := 0; - Config.MediumLoadDelay := 0; - Config.HighLoadDelay := 0; - FThreadPool.WorkQueue.BackpressureConfig := Config; - // Measure low load (single task) LogTest('Starting low load test...'); StartTime := Now; @@ -567,7 +621,7 @@ procedure TTestProducerConsumerThreadPool.Test14_AdaptivePerformance; AssertTrue('Low load should be proportionally faster', Ratio > 1.5); - LogTest('Test14_AdaptivePerformance finished'); + LogTest('Test14_ParallelScaling finished'); end; procedure TTestProducerConsumerThreadPool.Test15_ErrorsCollectionCapturesAll; @@ -628,6 +682,145 @@ procedure TTestProducerConsumerThreadPool.Test17_ClearErrorsResetsCollection; LogTest('Test17_ClearErrorsResetsCollection finished'); end; +procedure TTestProducerConsumerThreadPool.Test18_WaitTimeout; +begin + FThreadPool.Queue(@SleepTask); + AssertFalse('Short timeout should report unfinished work', + FThreadPool.WaitForAll(5)); + AssertTrue('Long timeout should observe completion', + FThreadPool.WaitForAll(2000)); +end; + +procedure TTestProducerConsumerThreadPool.Test19_ShutdownDrainsAndStops; +var + I: Integer; +begin + FSharedCounter := 0; + for I := 1 to 200 do + FThreadPool.Queue(@IncrementCounter); + FThreadPool.Shutdown; + AssertEquals('Shutdown must drain every accepted task', 200, FSharedCounter); + AssertEquals('Pool must become stopped', Ord(tpsStopped), + Ord(FThreadPool.State)); + AssertTrue('A stopped drained pool is idle', FThreadPool.WaitForAll(0)); +end; + +procedure TTestProducerConsumerThreadPool.Test20_QueueAfterShutdownRaises; +var + Raised: Boolean; +begin + FThreadPool.Shutdown; + Raised := False; + try + FThreadPool.Queue(@IncrementCounter); + except + on E: EThreadPoolShutdown do + Raised := True; + end; + AssertTrue('Queue after shutdown must raise EThreadPoolShutdown', Raised); +end; + +procedure TTestProducerConsumerThreadPool.Test21_OnErrorExceptionIsContained; +begin + FOnErrorCount := 0; + FThreadPool.OnError := @HandlePoolErrorAndRaise; + try + FThreadPool.Queue(@RaiseTestError); + AssertTrue('Failing task should still complete', + FThreadPool.WaitForAll(2000)); + FThreadPool.Queue(@IncrementCounter); + AssertTrue('Worker pool must remain usable after callback exception', + FThreadPool.WaitForAll(2000)); + AssertEquals('Callback should have run once', 1, FOnErrorCount); + finally + FThreadPool.OnError := nil; + end; +end; + +procedure TTestProducerConsumerThreadPool.Test22_TryQueueTimeoutWhenFull; +var + Pool: TProducerConsumerThreadPool; + I: Integer; +begin + FGateEvent.ResetEvent; + FAllStartedEvent.ResetEvent; + FStartedCount := 0; + Pool := TProducerConsumerThreadPool.Create(4, 1); + try + for I := 1 to 4 do + Pool.Queue(@GateTask); + AssertEquals('All workers should start their gate task', Ord(wrSignaled), + Ord(FAllStartedEvent.WaitFor(2000))); + Pool.Queue(@GateTask); + AssertFalse('A zero-timeout submission should fail immediately when full', + Pool.TryQueue(@GateTask, 0)); + finally + FGateEvent.SetEvent; + Pool.WaitForAll; + Pool.Free; + end; +end; + +procedure TTestProducerConsumerThreadPool.Test23_ConcurrentQueueAndShutdown; +var + Submitter: TProducerLifecycleQueueThread; + StartEvent, FirstQueuedEvent: TEvent; +begin + FSharedCounter := 0; + StartEvent := TEvent.Create(nil, True, False, ''); + FirstQueuedEvent := TEvent.Create(nil, True, False, ''); + Submitter := TProducerLifecycleQueueThread.Create(FThreadPool, + @IncrementCounter, StartEvent, FirstQueuedEvent); + try + Submitter.Start; + StartEvent.SetEvent; + AssertEquals('Submitter should queue at least one task', Ord(wrSignaled), + Ord(FirstQueuedEvent.WaitFor(2000))); + FThreadPool.Shutdown; + Submitter.WaitFor; + AssertEquals('Admission race must not raise an unexpected error', '', + Submitter.UnexpectedError); + AssertEquals('Every accepted task must be drained', Submitter.Accepted, + FSharedCounter); + finally + Submitter.Free; + FirstQueuedEvent.Free; + StartEvent.Free; + end; +end; + +procedure TTestProducerConsumerThreadPool.Test24_InvalidQueueSizeRejected; +var + Pool: TProducerConsumerThreadPool; + Raised: Boolean; +begin + Pool := nil; + Raised := False; + try + try + Pool := TProducerConsumerThreadPool.Create(4, 0); + except + on E: EArgumentOutOfRangeException do + Raised := True; + end; + finally + Pool.Free; + end; + AssertTrue('Zero-sized queues must be rejected', Raised); +end; + +procedure TTestProducerConsumerThreadPool.Test25_WorkerShutdownCannotDeadlock; +begin + FThreadPool.ClearErrors; + FThreadPool.Queue(@ShutdownFromWorker); + AssertTrue('Worker shutdown attempt must finish instead of self-deadlocking', + FThreadPool.WaitForAll(2000)); + AssertEquals('Rejected worker shutdown must leave the pool accepting', + Ord(tpsAccepting), Ord(FThreadPool.State)); + AssertTrue('Rejected worker shutdown should be captured as a task error', + Pos('cannot be called from a pool worker', FThreadPool.LastError) > 0); +end; + initialization RegisterTest(TTestProducerConsumerThreadPool); end. diff --git a/tests/threadpool.simple.tests.pas b/tests/threadpool.simple.tests.pas index f69fb27..83fadd2 100644 --- a/tests/threadpool.simple.tests.pas +++ b/tests/threadpool.simple.tests.pas @@ -36,6 +36,9 @@ TSimpleThreadPoolTests = class(TTestCase) procedure IncrementCounterWithIndex(AIndex: Integer); // Thread-safe OnError handler used by Test23. procedure HandlePoolError(const AMessage: string); + procedure HandlePoolErrorAndRaise(const AMessage: string); + procedure SlowTask; + procedure ShutdownFromWorker; protected procedure SetUp; override; procedure TearDown; override; @@ -77,6 +80,15 @@ TSimpleThreadPoolTests = class(TTestCase) procedure Test24_ClearErrorsResetsCollection; procedure Test25_ErrorCollectionIsCapped; procedure Test26_LastErrorStillWorks; + + // Lifecycle, timeout, and callback safety (v0.8.0) + procedure Test27_WaitTimeout; + procedure Test28_ShutdownDrainsAndStops; + procedure Test29_QueueAfterShutdownRaises; + procedure Test30_OnErrorExceptionIsContained; + procedure Test31_TryQueueCompatibility; + procedure Test32_ConcurrentQueueAndShutdown; + procedure Test33_WorkerShutdownCannotDeadlock; end; var @@ -87,6 +99,7 @@ procedure GlobalIncrementCounter; procedure GlobalIncrementCounterWithIndex(AIndex: Integer); procedure RaiseTestException; procedure RaiseAnotherException; +procedure RaiseFastException; implementation @@ -116,6 +129,11 @@ procedure RaiseAnotherException; raise TTestException.Create('Another exception message'); end; +procedure RaiseFastException; +begin + raise TTestException.Create('Fast exception'); +end; + { TTestObject } constructor TTestObject.Create(ACS: TCriticalSection); @@ -354,6 +372,21 @@ TTestQueueThread = class(TThread) procedure Execute; override; end; + TLifecycleQueueThread = class(TThread) + private + FPool: TSimpleThreadPool; + FStartEvent: TEvent; + FFirstQueuedEvent: TEvent; + FAccepted: Integer; + FUnexpectedError: string; + public + constructor Create(APool: TSimpleThreadPool; AStartEvent, + AFirstQueuedEvent: TEvent); + procedure Execute; override; + property Accepted: Integer read FAccepted; + property UnexpectedError: string read FUnexpectedError; + end; + constructor TTestQueueThread.Create(APool: TSimpleThreadPool; AStartEvent: TEvent); begin inherited Create(True); @@ -371,6 +404,40 @@ procedure TTestQueueThread.Execute; FPool.Queue(@GlobalIncrementCounter); end; +constructor TLifecycleQueueThread.Create(APool: TSimpleThreadPool; + AStartEvent, AFirstQueuedEvent: TEvent); +begin + inherited Create(True); + FPool := APool; + FStartEvent := AStartEvent; + FFirstQueuedEvent := AFirstQueuedEvent; + FreeOnTerminate := False; +end; + +procedure TLifecycleQueueThread.Execute; +var + I: Integer; +begin + FStartEvent.WaitFor(INFINITE); + try + for I := 1 to 10000 do + begin + try + FPool.Queue(@GlobalIncrementCounter); + Inc(FAccepted); + if FAccepted = 1 then + FFirstQueuedEvent.SetEvent; + except + on E: EThreadPoolShutdown do + Break; + end; + end; + except + on E: Exception do + FUnexpectedError := E.ClassName + ': ' + E.Message; + end; +end; + procedure TSimpleThreadPoolTests.Test13_ConcurrentQueueAccess; var Threads: array[1..5] of TTestQueueThread; @@ -570,7 +637,7 @@ procedure TSimpleThreadPoolTests.Test25_ErrorCollectionIsCapped; // ErrorCount must never exceed MAX_STORED_ERRORS regardless of how many fail. FThreadPool.ClearErrors; for I := 1 to MAX_STORED_ERRORS + 250 do - FThreadPool.Queue(@RaiseAnotherException); + FThreadPool.Queue(@RaiseFastException); FThreadPool.WaitForAll; AssertEquals('Error collection should be capped at MAX_STORED_ERRORS', @@ -597,6 +664,127 @@ procedure TSimpleThreadPoolTests.HandlePoolError(const AMessage: string); InterlockedIncrement(FOnErrorCount); end; +procedure TSimpleThreadPoolTests.HandlePoolErrorAndRaise( + const AMessage: string); +begin + InterlockedIncrement(FOnErrorCount); + raise Exception.Create('OnError handler failure'); +end; + +procedure TSimpleThreadPoolTests.SlowTask; +begin + Sleep(100); +end; + +procedure TSimpleThreadPoolTests.ShutdownFromWorker; +begin + FThreadPool.Shutdown; +end; + +procedure TSimpleThreadPoolTests.Test27_WaitTimeout; +begin + FThreadPool.Queue(@SlowTask); + AssertFalse('Short timeout should report unfinished work', + FThreadPool.WaitForAll(5)); + AssertTrue('Long timeout should observe completion', + FThreadPool.WaitForAll(2000)); +end; + +procedure TSimpleThreadPoolTests.Test28_ShutdownDrainsAndStops; +var + I: Integer; +begin + FCounter := 0; + for I := 1 to 200 do + FThreadPool.Queue(@GlobalIncrementCounter); + FThreadPool.Shutdown; + AssertEquals('Shutdown must drain every accepted task', 200, FCounter); + AssertEquals('Pool must become stopped', Ord(tpsStopped), + Ord(FThreadPool.State)); + AssertTrue('A stopped drained pool is idle', FThreadPool.WaitForAll(0)); +end; + +procedure TSimpleThreadPoolTests.Test29_QueueAfterShutdownRaises; +var + Raised: Boolean; +begin + FThreadPool.Shutdown; + Raised := False; + try + FThreadPool.Queue(@GlobalIncrementCounter); + except + on E: EThreadPoolShutdown do + Raised := True; + end; + AssertTrue('Queue after shutdown must raise EThreadPoolShutdown', Raised); +end; + +procedure TSimpleThreadPoolTests.Test30_OnErrorExceptionIsContained; +begin + FOnErrorCount := 0; + FThreadPool.OnError := @HandlePoolErrorAndRaise; + try + FThreadPool.Queue(@RaiseFastException); + AssertTrue('Failing task should still complete', + FThreadPool.WaitForAll(2000)); + FThreadPool.Queue(@GlobalIncrementCounter); + AssertTrue('Worker pool must remain usable after callback exception', + FThreadPool.WaitForAll(2000)); + AssertEquals('Callback should have run once', 1, FOnErrorCount); + finally + FThreadPool.OnError := nil; + end; +end; + +procedure TSimpleThreadPoolTests.Test31_TryQueueCompatibility; +begin + FCounter := 0; + AssertTrue('Unbounded Simple pool should accept TryQueue immediately', + FThreadPool.TryQueue(@GlobalIncrementCounter, 0)); + FThreadPool.WaitForAll; + AssertEquals('TryQueue task should execute', 1, FCounter); +end; + +procedure TSimpleThreadPoolTests.Test32_ConcurrentQueueAndShutdown; +var + Submitter: TLifecycleQueueThread; + StartEvent, FirstQueuedEvent: TEvent; +begin + FCounter := 0; + StartEvent := TEvent.Create(nil, True, False, ''); + FirstQueuedEvent := TEvent.Create(nil, True, False, ''); + Submitter := TLifecycleQueueThread.Create(FThreadPool, StartEvent, + FirstQueuedEvent); + try + Submitter.Start; + StartEvent.SetEvent; + AssertEquals('Submitter should queue at least one task', Ord(wrSignaled), + Ord(FirstQueuedEvent.WaitFor(2000))); + FThreadPool.Shutdown; + Submitter.WaitFor; + AssertEquals('Admission race must not raise an unexpected error', '', + Submitter.UnexpectedError); + AssertEquals('Every accepted task must be drained', Submitter.Accepted, + FCounter); + finally + Submitter.Free; + FirstQueuedEvent.Free; + StartEvent.Free; + end; +end; + +procedure TSimpleThreadPoolTests.Test33_WorkerShutdownCannotDeadlock; +begin + FThreadPool.ClearErrors; + FThreadPool.Queue(@ShutdownFromWorker); + AssertTrue('Worker shutdown attempt must finish instead of self-deadlocking', + FThreadPool.WaitForAll(2000)); + AssertEquals('Rejected worker shutdown must leave the pool accepting', + Ord(tpsAccepting), Ord(FThreadPool.State)); + AssertTrue('Rejected worker shutdown should be captured as a task error', + Pos('cannot be called from a pool worker', FThreadPool.LastError) > 0); +end; + procedure TSimpleThreadPoolTests.IncrementCounter; begin FCS.Enter; From 2cbffca53605fe0c1fb5842e6d290600ac890d13 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah <iwan.kelaiah@gmail.com> Date: Mon, 13 Jul 2026 22:04:06 +1000 Subject: [PATCH 2/3] feat(benchmark): update project file --- benchmarks/ThreadPoolBenchmark.lpi | 68 +++++++++++++++++++++++++++--- 1 file changed, 61 insertions(+), 7 deletions(-) diff --git a/benchmarks/ThreadPoolBenchmark.lpi b/benchmarks/ThreadPoolBenchmark.lpi index 2aaae44..d8d8cb8 100644 --- a/benchmarks/ThreadPoolBenchmark.lpi +++ b/benchmarks/ThreadPoolBenchmark.lpi @@ -8,33 +8,85 @@ <MainUnitHasCreateFormStatements Value="False"/> <MainUnitHasTitleStatement Value="False"/> <MainUnitHasScaledStatement Value="False"/> + <UseDefaultCompilerOptions Value="True"/> </Flags> + <SessionStorage Value="InProjectDir"/> <Title Value="ThreadPoolBenchmark"/> <UseAppBundle Value="False"/> <ResourceType Value="res"/> </General> <BuildModes> <Item Name="Default" Default="True"/> + <Item Name="Debug"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ThreadPoolBenchmark"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <Parsing> + <SyntaxOptions> + <IncludeAssertionCode Value="True"/> + </SyntaxOptions> + </Parsing> + <CodeGeneration> + <Checks> + <IOChecks Value="True"/> + <RangeChecks Value="True"/> + <OverflowChecks Value="True"/> + <StackChecks Value="True"/> + </Checks> + <VerifyObjMethodCallValidity Value="True"/> + </CodeGeneration> + <Linking> + <Debugging> + <DebugInfoType Value="dsDwarf3"/> + <UseHeaptrc Value="True"/> + <TrashVariables Value="True"/> + </Debugging> + </Linking> + </CompilerOptions> + </Item> <Item Name="Release"> <CompilerOptions> <Version Value="11"/> <PathDelim Value="\"/> - <Target><Filename Value="ThreadPoolBenchmark"/></Target> + <Target> + <Filename Value="ThreadPoolBenchmark"/> + </Target> <SearchPaths> <OtherUnitFiles Value="..\src"/> <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> </SearchPaths> <CodeGeneration> <SmartLinkUnit Value="True"/> - <Optimizations><OptimizationLevel Value="3"/></Optimizations> + <Optimizations> + <OptimizationLevel Value="3"/> + </Optimizations> </CodeGeneration> - <Linking><LinkSmart Value="True"/></Linking> + <Linking> + <Debugging> + <GenerateDebugInfo Value="False"/> + <RunWithoutDebug Value="True"/> + <StripSymbols Value="True"/> + </Debugging> + <LinkSmart Value="True"/> + </Linking> </CompilerOptions> </Item> </BuildModes> - <RequiredPackages> - <Item><PackageName Value="FCL"/></Item> - </RequiredPackages> + <PublishOptions> + <Version Value="2"/> + <UseFileFilters Value="True"/> + </PublishOptions> + <RunParams> + <FormatVersion Value="2"/> + </RunParams> <Units> <Unit> <Filename Value="ThreadPoolBenchmark.lpr"/> @@ -45,7 +97,9 @@ <CompilerOptions> <Version Value="11"/> <PathDelim Value="\"/> - <Target><Filename Value="ThreadPoolBenchmark"/></Target> + <Target> + <Filename Value="ThreadPoolBenchmark"/> + </Target> <SearchPaths> <OtherUnitFiles Value="..\src"/> <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> From 8a55c4286e2a62d31587dc3e36040cdfc7ed9988 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah <iwan.kelaiah@gmail.com> Date: Mon, 13 Jul 2026 22:20:06 +1000 Subject: [PATCH 3/3] docs(repo): update documents to latest state of code --- CHANGELOG.md | 2 ++ CONTRIBUTING.md | 6 +++--- README.md | 13 ++++++++++--- docs/ThreadPool.ProducerConsumer-API.md | 8 +++++++- docs/ThreadPool.Simple-API.md | 9 +++++++-- docs/release-notes-v0.8.0.md | 5 +++++ 6 files changed, 34 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c92acc5..615def2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - Worker completion accounting runs independently of task and callback failures - `OnError` exceptions are contained so they cannot terminate a worker or wedge `WaitForAll` +- Callback execution remains synchronous and is not time-limited; blocking tasks + or `OnError` handlers can delay completion and `Shutdown` - Producer backpressure waits only when the queue is full; load-threshold sleeps are removed while `TBackpressureConfig` remains source-compatible - Producer debug logging is disabled by default diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d08bcbe..59eb1ad 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -13,7 +13,7 @@ sizes are welcome — bug reports, documentation fixes, new examples, and code. ## Prerequisites - Free Pascal 3.2.2 or later -- Lazarus 3.6.0 or later (provides `lazbuild`) +- Lazarus 4.0 or later (provides `lazbuild`) - No external dependencies ## Building @@ -40,8 +40,8 @@ lazbuild tests/TestRunner.lpi ./tests/TestRunner -a -p --format=plain ``` -All tests should pass before you submit a change. The full suite can take a -few minutes to run because some tests exercise high task volumes. +All tests should pass before you submit a change. The full suite normally takes +only a few seconds, although timing varies by machine. CI runs the package build, the test suite, and all examples on Linux and Windows for every pull request — please make sure your branch is green. diff --git a/README.md b/README.md index 98f37b9..488a3f1 100644 --- a/README.md +++ b/README.md @@ -137,6 +137,7 @@ backpressure: - `Errors` collection captures **all** failed-task messages (capped, oldest dropped) — *v0.7.0* - Optional `OnError` callback fired per failed task — *v0.7.0* - Exceptions raised by `OnError` are contained and cannot terminate workers + - Callback execution remains synchronous; keep callbacks short and bounded > [!NOTE] > Thread count is determined by `TThread.ProcessorCount` at startup and remains fixed. See [Thread Management](#-thread-management) for details. @@ -234,6 +235,11 @@ unbounded, so its `TryQueue` timeout is accepted for API symmetry but queue capacity cannot time out. After `Shutdown`, all `Queue` and `TryQueue` overloads raise `EThreadPoolShutdown`. +Exception containment does not impose an execution deadline. A task or +`OnError` handler that blocks indefinitely continues to occupy its worker, and +`Shutdown` waits because it drains all accepted work. Use application-level +timeouts or cancellation inside operations that may block. + `WaitForAll` is not an admission barrier for unrelated producer threads. If producers may still submit concurrently, coordinate them first or call `Shutdown`, which closes admission before draining. @@ -341,8 +347,9 @@ Prefer to react the moment a task fails (instead of polling after `WaitForAll`)? Assign an `OnError` callback: ```pascal -// IMPORTANT: OnError is called from a worker thread. Keep the handler short and -// thread-safe; synchronize if it touches the UI or shared state. +// IMPORTANT: OnError is called synchronously from a worker thread. Keep the +// handler short, bounded, and thread-safe; synchronize if it touches the UI or +// shared state. Pool.OnError := @MyHandler.OnTaskError; ``` @@ -531,7 +538,7 @@ All tasks completed successfully! ## ⚙️ Requirements - 💻 Free Pascal 3.2.2 or later -- 📦 Lazarus 3.6.0 or later +- 📦 Lazarus 4.0 or later - 🆓 No external dependencies ## 📚 Documentation diff --git a/docs/ThreadPool.ProducerConsumer-API.md b/docs/ThreadPool.ProducerConsumer-API.md index 181c125..401a04b 100644 --- a/docs/ThreadPool.ProducerConsumer-API.md +++ b/docs/ThreadPool.ProducerConsumer-API.md @@ -168,13 +168,19 @@ end; Or assign `OnError` to react the moment a task fails, instead of polling: ```pascal -// Called from a worker thread — keep it short and thread-safe. +// 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 ```pascal diff --git a/docs/ThreadPool.Simple-API.md b/docs/ThreadPool.Simple-API.md index b32f559..1659801 100644 --- a/docs/ThreadPool.Simple-API.md +++ b/docs/ThreadPool.Simple-API.md @@ -175,8 +175,8 @@ type procedure TMyHandler.OnTaskError(const AMessage: string); begin - // NOTE: called from a worker thread. Keep it short and thread-safe; - // synchronize if you touch the UI or shared state. + // NOTE: called synchronously from a worker thread. Keep it short, bounded, + // and thread-safe; synchronize if you touch the UI or shared state. Log('task failed: ' + AMessage); end; @@ -186,6 +186,11 @@ 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. + ## Lifecycle and submission timeouts (v0.8.0) `TSimpleThreadPool` is unbounded, so `TryQueue` normally succeeds immediately; diff --git a/docs/release-notes-v0.8.0.md b/docs/release-notes-v0.8.0.md index f903d47..91d63a1 100644 --- a/docs/release-notes-v0.8.0.md +++ b/docs/release-notes-v0.8.0.md @@ -22,6 +22,11 @@ admission, waits for submissions already in progress, drains all accepted work, wakes and joins workers, and publishes `tpsStopped`. Repeated calls are safe. Queueing after shutdown raises `EThreadPoolShutdown`. +Task and `OnError` exceptions are contained, but callback execution is +synchronous and has no automatic deadline. A blocking callback can therefore +delay completion and `Shutdown`; operations that may block should implement an +application-level timeout or cancellation mechanism. + ## Timeout contract Timeout values are milliseconds: