diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bc952fc..c357c3f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -71,9 +71,4 @@ jobs:
- name: Build all examples
shell: bash
- run: |
- set -e
- for lpi in examples/*/*.lpi; do
- echo "==> Building $lpi"
- lazbuild "$lpi"
- done
+ run: sh ./build-examples.sh Release
diff --git a/.gitignore b/.gitignore
index cec63fc..1527bbe 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,6 +27,9 @@ tests/TestRunner
tests/TestRunner.exe
tests/lib/
+# Consolidated example binary outputs
+/example-bin/
+
# Debug files
*.dbg
*.dcu
@@ -58,4 +61,4 @@ $RECYCLE.BIN/
## Linux
*~
.directory
-.Trash-*
\ No newline at end of file
+.Trash-*
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6a58c3d..f71a449 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,64 @@ 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.9.0] - 2026-07-16
+
+### Added
+
+- `ThreadPool.Tasks`, a shared task coordination unit for both pool
+ implementations
+- `Submit` overloads returning `IThreadPoolTask` handles with pending, running,
+ completed, failed, and cancelled states
+- Timeout-aware `TrySubmit` overloads matching all four existing callback forms
+- Individual task waiting, failure messages, terminal-state observation, and
+ race-safe pending cancellation
+- `EThreadPoolDeadlock` protection when a callback or error handler attempts to
+ wait for its own task
+- `IThreadPoolTaskBatch` for snapshot-based waits, status counts, indexed task
+ access, and batch pending cancellation
+- Chunked `SubmitRange` overloads for indexed procedures and methods, with
+ inclusive bounds, automatic or explicit chunk sizes, and a returned batch
+- v0.8 interface compile sentinel plus task, batch, cancellation-race, bounded
+ admission, lifetime, and range regression tests
+- `TaskCoordination` API tour, plus production-shaped coordinated file-backup
+ and parallel log-analysis examples
+- Root-level PowerShell and POSIX scripts that build every example into
+ `example-bin/`
+- Complete task API documentation
+- Separate benchmark cases for legacy queueing, tracked submission, individual
+ indexed submission, and chunked ranges
+
+### Changed
+
+- Simple pool queue storage now uses `IWorkItem` consistently, allowing shared
+ tracked work items while preserving its dynamically growing O(1) FIFO
+- Both workers execute tracked work through one shared error/state transition
+ boundary
+- Package version advanced to 0.9.0 and includes `ThreadPool.Tasks`
+
+### Performance
+
+- Legacy `Queue` retains its untracked path, so programs that do not request
+ task handles do not allocate task state or completion events
+- Task completion events are created lazily only when unfinished work is waited
+ on
+- Automatic ranges create at most `ThreadCount * 4` queue entries instead of
+ one entry per index
+- Windows/FPC 3.2.2 release checks kept the legacy 20,000-task median within the
+ v0.8.5 10% regression budget; a 200,000-index Simple range was over 60x faster
+ than individual tracked submissions in the orientation run
+
+### Compatibility
+
+- The v0.8 `IThreadPool` interface and GUID are unchanged; new interface-based
+ callers use the separate `IThreadPoolTaskSource` capability
+- Existing `Queue`, `TryQueue`, `WaitForAll`, lifecycle, error, constructor, and
+ `GlobalThreadPool` contracts remain available
+- Cancellation applies only to pending callbacks. It never interrupts running
+ code, and a bounded-queue tombstone may occupy its slot until dequeued
+- Bounded `SubmitRange` calls from the same pool's worker are rejected with
+ `EThreadPoolDeadlock` to prevent queue-starvation deadlocks
+
## [0.8.5] - 2026-07-14
### Added
diff --git a/README.md b/README.md
index 95c7f99..9f0a2c5 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
# ThreadPool for Free Pascal
-[](CHANGELOG.md)
+[](CHANGELOG.md)
[](LICENSE.md)
[](https://www.freepascal.org/)
[](https://www.lazarus-ide.org/)
@@ -19,11 +19,12 @@ producer-consumer workloads that need backpressure.
[Quick start](#quick-start) · [Cheat sheet](docs/CHEATSHEET.md) ·
[API documentation](#documentation) · [Examples](examples/) ·
-[v0.8.5 release notes](docs/release-notes-v0.8.5.md)
+[v0.9.0 release notes](docs/release-notes-v0.9.0.md)
> [!TIP]
-> ✨ **New in v0.8.5:** refreshed project identity, a shorter README, and a new
-> [cheat sheet](docs/CHEATSHEET.md). Runtime behavior is unchanged from v0.8.0.
+> ✨ **New in v0.9.0:** observable task handles, task batches, efficient
+> chunked ranges, and pending-work cancellation. Existing v0.8 queueing code
+> remains source-compatible. See the [task API](docs/ThreadPool.Tasks-API.md).
> [!NOTE]
> This library is designed for simple parallel processing and learning-friendly
@@ -47,6 +48,9 @@ Both implementations provide:
- event-driven workers with no polling sleeps;
- four task forms: procedures, methods, and indexed variants;
- timeout-aware `TryQueue` and `WaitForAll` overloads;
+- observable `Submit`/`TrySubmit` task handles;
+- task batches and chunked `SubmitRange` processing;
+- race-safe cancellation of work that has not started;
- deterministic, draining `Shutdown`;
- captured worker exceptions through `LastError`, `Errors`, and `OnError`; and
- automatic worker-count selection with safety limits.
@@ -130,6 +134,33 @@ end.
The first constructor argument is the worker count; `0` selects
`TThread.ProcessorCount`. The second is queue capacity.
+### Tasks, batches, and ranges
+
+Add `ThreadPool.Tasks` when work needs to be observed or coordinated:
+
+```pascal
+uses
+ ThreadPool.Tasks, ThreadPool.Simple;
+
+var
+ Task: IThreadPoolTask;
+ Batch: IThreadPoolTaskBatch;
+begin
+ Task := GlobalThreadPool.Submit(@DoWork);
+ if Task.WaitFor(250) and (Task.State = ttsFailed) then
+ WriteLn(Task.ErrorMessage);
+
+ Batch := GlobalThreadPool.SubmitRange(@ProcessItem, 0, 999);
+ Batch.WaitFor;
+end;
+```
+
+`Task.Cancel` succeeds only while the task is pending. It never interrupts a
+running callback. A range uses a small number of chunks by default instead of
+creating one queue item per index. See the
+[task API](docs/ThreadPool.Tasks-API.md) for batch counts, timeouts, explicit
+chunk sizes, and bounded-pool rules.
+
## Lifecycle and timeouts
Both pools follow one monotonic lifecycle:
@@ -173,6 +204,9 @@ expires.
> also have no automatic execution deadline; add cancellation or
> application-level timeouts where needed.
+Task handles and batches do not retain their pool. They may be kept after a
+pool is freed, because pool destruction drains accepted work first.
+
## Error handling
Task exceptions are caught so a worker failure does not terminate the pool.
@@ -226,12 +260,30 @@ Requirements:
## Examples
+Build every example in Release mode from the repository root:
+
+```powershell
+.\build-examples.ps1
+```
+
+```sh
+sh ./build-examples.sh
+```
+
+Both scripts discover `examples/*/*.lpi` automatically and place the
+executables in the ignored root-level `example-bin/` directory. Pass `Default`
+to build the default mode, or use `-Rebuild` in PowerShell / `--rebuild` in the
+shell script to force a complete rebuild.
+
| Start with | Demonstrates |
| --- | --- |
| [`Starter`](examples/Starter/) | Smallest compilable program with explanatory comments |
| [`SimpleDemo`](examples/SimpleDemo/) | Procedures, methods, indexes, and the global pool |
| [`ProdConSimpleDemo`](examples/ProdConSimpleDemo/) | Basic bounded-pool ownership and queueing |
| [`SimpleErrorHandlingBasic`](examples/SimpleErrorHandlingBasic/) | Reading captured errors after completion |
+| [`TaskCoordination`](examples/TaskCoordination/) | Task handles, batches, ranges, and cancellation |
+| [`CoordinatedFileBackup`](examples/CoordinatedFileBackup/) | Per-file progress, critical-failure policy, and pending cancellation |
+| [`ParallelLogAnalyzer`](examples/ParallelLogAnalyzer/) | Chunked analysis followed by a parallel reporting phase |
More focused samples cover:
@@ -241,8 +293,11 @@ More focused samples cover:
[`SimpleWordCounter`](examples/SimpleWordCounter/), and
[`ProdConMessageProcessor`](examples/ProdConMessageProcessor/);
- advanced callbacks: [`SimpleErrorHandling`](examples/SimpleErrorHandling/);
-- real I/O: [`ParallelFileHasher`](examples/ParallelFileHasher/) and
- [`ParallelUrlFetcher`](examples/ParallelUrlFetcher/).
+- real I/O: [`ParallelFileHasher`](examples/ParallelFileHasher/),
+ [`ParallelUrlFetcher`](examples/ParallelUrlFetcher/), and
+ [`CoordinatedFileBackup`](examples/CoordinatedFileBackup/);
+- coordinated data processing:
+ [`ParallelLogAnalyzer`](examples/ParallelLogAnalyzer/).
## Documentation
@@ -251,9 +306,10 @@ More focused samples cover:
| [Cheat sheet](docs/CHEATSHEET.md) | Calls and safety rules at a glance |
| [Simple API](docs/ThreadPool.Simple-API.md) | Complete unbounded-pool reference |
| [Producer-Consumer API](docs/ThreadPool.ProducerConsumer-API.md) | Complete bounded-pool reference |
+| [Tasks API](docs/ThreadPool.Tasks-API.md) | Submit, wait, batch, range, and cancellation contracts |
| [Simple technical guide](docs/ThreadPool.Simple-Technical.md) | Internal design and synchronization |
| [Producer-Consumer technical guide](docs/ThreadPool.ProducerConsumer-Technical.md) | Queue and backpressure internals |
-| [v0.8.5 release notes](docs/release-notes-v0.8.5.md) | Current release scope and compatibility |
+| [v0.9.0 release notes](docs/release-notes-v0.9.0.md) | Current release scope and compatibility |
| [Changelog](CHANGELOG.md) | Full version history |
The banner's editable source is
diff --git a/benchmarks/README.md b/benchmarks/README.md
index 6f728c1..0183d12 100644
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -7,11 +7,23 @@ lazbuild --build-mode=Release benchmarks/ThreadPoolBenchmark.lpi
./benchmarks/ThreadPoolBenchmark
```
-The same source compiles against v0.7.0 and v0.8.0. It reports:
+The v0.9.0 benchmark reports:
-- completion time for a 20,000-task burst through each pool;
+- completion time for a 20,000-task legacy `Queue` burst through each pool;
+- completion time for an equivalent tracked `Submit` burst;
+- individual tracked submission versus chunked `SubmitRange` for 200,000
+ indexed calls;
- average queue-to-start latency after workers have been idle.
+For comparisons with v0.7.0 and v0.8.x, use the legacy queue and idle fields;
+the tracked and range APIs did not exist in those releases.
+
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.
+
+Do not enforce absolute millisecond thresholds on shared CI runners. Compare
+five-run medians on the same otherwise-idle machine. The v0.9.0 release budget
+allows at most a 10% regression in the legacy queue medians and expects the
+Simple chunked range case to be at least 5x faster than individual tracked
+indexed submissions.
diff --git a/benchmarks/ThreadPoolBenchmark.lpr b/benchmarks/ThreadPoolBenchmark.lpr
index 653d1e0..a0509f2 100644
--- a/benchmarks/ThreadPoolBenchmark.lpr
+++ b/benchmarks/ThreadPoolBenchmark.lpr
@@ -7,10 +7,11 @@
cthreads,
{$ENDIF}
Classes, SysUtils, SyncObjs,
- ThreadPool.Simple, ThreadPool.ProducerConsumer;
+ ThreadPool.Tasks, ThreadPool.Simple, ThreadPool.ProducerConsumer;
const
BURST_TASKS = 20000;
+ RANGE_ITEMS = 200000;
IDLE_SAMPLES = 10;
IDLE_PAUSE_MS = 120;
@@ -25,6 +26,12 @@ procedure CountTask;
InterlockedIncrement(Counter);
end;
+procedure CountIndexedTask(AIndex: Integer);
+begin
+ if AIndex >= 0 then
+ InterlockedIncrement(Counter);
+end;
+
procedure IdleTask;
begin
LastIdleLatency := GetTickCount64 - SubmittedAt;
@@ -75,6 +82,128 @@ function BenchmarkProducerBurst: QWord;
end;
end;
+function BenchmarkSimpleSubmitBurst: QWord;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+ StartedAt: QWord;
+ I: Integer;
+begin
+ Counter := 0;
+ Pool := TSimpleThreadPool.Create(4);
+ try
+ StartedAt := GetTickCount64;
+ for I := 1 to BURST_TASKS do
+ Task := Pool.Submit(@CountTask);
+ if Task = nil then
+ raise Exception.Create('Simple Submit returned nil');
+ Pool.WaitForAll;
+ Result := GetTickCount64 - StartedAt;
+ if Counter <> BURST_TASKS then
+ raise Exception.CreateFmt('Simple submit burst lost tasks: %d/%d',
+ [Counter, BURST_TASKS]);
+ finally
+ Task := nil;
+ Pool.Free;
+ end;
+end;
+
+function BenchmarkProducerSubmitBurst: QWord;
+var
+ Pool: TProducerConsumerThreadPool;
+ Task: IThreadPoolTask;
+ StartedAt: QWord;
+ I: Integer;
+begin
+ Counter := 0;
+ Pool := TProducerConsumerThreadPool.Create(4, 1024);
+ try
+ StartedAt := GetTickCount64;
+ for I := 1 to BURST_TASKS do
+ Task := Pool.Submit(@CountTask);
+ if Task = nil then
+ raise Exception.Create('Producer Submit returned nil');
+ Pool.WaitForAll;
+ Result := GetTickCount64 - StartedAt;
+ if Counter <> BURST_TASKS then
+ raise Exception.CreateFmt('Producer submit burst lost tasks: %d/%d',
+ [Counter, BURST_TASKS]);
+ finally
+ Task := nil;
+ Pool.Free;
+ end;
+end;
+
+function BenchmarkIndividualIndexedSubmit: QWord;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+ StartedAt: QWord;
+ I: Integer;
+begin
+ Counter := 0;
+ Pool := TSimpleThreadPool.Create(4);
+ try
+ StartedAt := GetTickCount64;
+ for I := 0 to RANGE_ITEMS - 1 do
+ Task := Pool.Submit(@CountIndexedTask, I);
+ if Task = nil then
+ raise Exception.Create('Indexed Submit returned nil');
+ Pool.WaitForAll;
+ Result := GetTickCount64 - StartedAt;
+ if Counter <> RANGE_ITEMS then
+ raise Exception.CreateFmt('Individual range lost indexes: %d/%d',
+ [Counter, RANGE_ITEMS]);
+ finally
+ Task := nil;
+ Pool.Free;
+ end;
+end;
+
+function BenchmarkSimpleRange: QWord;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ StartedAt: QWord;
+begin
+ Counter := 0;
+ Pool := TSimpleThreadPool.Create(4);
+ try
+ StartedAt := GetTickCount64;
+ Batch := Pool.SubmitRange(@CountIndexedTask, 0, RANGE_ITEMS - 1);
+ Batch.WaitFor;
+ Result := GetTickCount64 - StartedAt;
+ if Counter <> RANGE_ITEMS then
+ raise Exception.CreateFmt('Simple range lost indexes: %d/%d',
+ [Counter, RANGE_ITEMS]);
+ finally
+ Batch := nil;
+ Pool.Free;
+ end;
+end;
+
+function BenchmarkProducerRange: QWord;
+var
+ Pool: TProducerConsumerThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ StartedAt: QWord;
+begin
+ Counter := 0;
+ Pool := TProducerConsumerThreadPool.Create(4, 1024);
+ try
+ StartedAt := GetTickCount64;
+ Batch := Pool.SubmitRange(@CountIndexedTask, 0, RANGE_ITEMS - 1);
+ Batch.WaitFor;
+ Result := GetTickCount64 - StartedAt;
+ if Counter <> RANGE_ITEMS then
+ raise Exception.CreateFmt('Producer range lost indexes: %d/%d',
+ [Counter, RANGE_ITEMS]);
+ finally
+ Batch := nil;
+ Pool.Free;
+ end;
+end;
+
function BenchmarkSimpleIdle: Double;
var
Pool: TSimpleThreadPool;
@@ -131,9 +260,15 @@ function BenchmarkProducerIdle: Double;
IdleEvent := TEvent.Create(nil, True, False, '');
try
WriteLn('burst_tasks=', BURST_TASKS);
+ WriteLn('range_items=', RANGE_ITEMS);
WriteLn('idle_samples=', IDLE_SAMPLES);
WriteLn('simple_burst_ms=', BenchmarkSimpleBurst);
WriteLn('producer_burst_ms=', BenchmarkProducerBurst);
+ WriteLn('simple_submit_burst_ms=', BenchmarkSimpleSubmitBurst);
+ WriteLn('producer_submit_burst_ms=', BenchmarkProducerSubmitBurst);
+ WriteLn('individual_indexed_submit_ms=', BenchmarkIndividualIndexedSubmit);
+ WriteLn('simple_range_ms=', BenchmarkSimpleRange);
+ WriteLn('producer_range_ms=', BenchmarkProducerRange);
WriteLn('simple_idle_avg_ms=', FormatFloat('0.00', BenchmarkSimpleIdle));
WriteLn('producer_idle_avg_ms=', FormatFloat('0.00', BenchmarkProducerIdle));
finally
diff --git a/build-examples.ps1 b/build-examples.ps1
new file mode 100644
index 0000000..40702cb
--- /dev/null
+++ b/build-examples.ps1
@@ -0,0 +1,43 @@
+[CmdletBinding()]
+param(
+ [ValidateSet('Default', 'Release')]
+ [string] $BuildMode = 'Release',
+
+ [switch] $Rebuild
+)
+
+Set-StrictMode -Version Latest
+$ErrorActionPreference = 'Stop'
+
+$RepoRoot = $PSScriptRoot
+$ExamplesDirectory = Join-Path $RepoRoot 'examples'
+$OutputDirectory = Join-Path $RepoRoot 'example-bin'
+$LazBuild = Get-Command lazbuild -CommandType Application -ErrorAction Stop
+$Projects = @(
+ Get-ChildItem -LiteralPath $ExamplesDirectory -Directory |
+ ForEach-Object {
+ Get-ChildItem -LiteralPath $_.FullName -Filter '*.lpi' -File
+ } |
+ Sort-Object FullName
+)
+
+if ($Projects.Count -eq 0) {
+ throw "No Lazarus example projects found under '$ExamplesDirectory'."
+}
+
+$null = New-Item -ItemType Directory -Path $OutputDirectory -Force
+$BuildArguments = @("--build-mode=$BuildMode", '--no-write-project')
+if ($Rebuild) {
+ $BuildArguments += '--build-all'
+}
+
+foreach ($Project in $Projects) {
+ $RelativeProject = $Project.FullName.Substring($RepoRoot.Length + 1)
+ Write-Host "==> Building $RelativeProject ($BuildMode)"
+ & $LazBuild.Source @BuildArguments $Project.FullName
+ if ($LASTEXITCODE -ne 0) {
+ throw "lazbuild failed for '$RelativeProject' with exit code $LASTEXITCODE."
+ }
+}
+
+Write-Host "Built $($Projects.Count) examples into '$OutputDirectory'."
diff --git a/build-examples.sh b/build-examples.sh
new file mode 100644
index 0000000..c399530
--- /dev/null
+++ b/build-examples.sh
@@ -0,0 +1,60 @@
+#!/usr/bin/env sh
+
+set -eu
+
+usage() {
+ echo "Usage: sh build-examples.sh [Default|Release] [--rebuild]" >&2
+}
+
+BUILD_MODE="${1:-Release}"
+if [ "$#" -gt 0 ]; then
+ shift
+fi
+
+case "$BUILD_MODE" in
+ Default|Release)
+ ;;
+ *)
+ usage
+ exit 2
+ ;;
+esac
+
+REBUILD=0
+if [ "$#" -gt 0 ]; then
+ if [ "$1" != "--rebuild" ] || [ "$#" -ne 1 ]; then
+ usage
+ exit 2
+ fi
+ REBUILD=1
+fi
+
+if ! command -v lazbuild >/dev/null 2>&1; then
+ echo "lazbuild was not found in PATH. Install Lazarus and try again." >&2
+ exit 127
+fi
+
+SCRIPT_DIRECTORY=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
+EXAMPLES_DIRECTORY="$SCRIPT_DIRECTORY/examples"
+OUTPUT_DIRECTORY="$SCRIPT_DIRECTORY/example-bin"
+
+mkdir -p "$OUTPUT_DIRECTORY"
+set -- "$EXAMPLES_DIRECTORY"/*/*.lpi
+if [ ! -f "$1" ]; then
+ echo "No Lazarus example projects found under '$EXAMPLES_DIRECTORY'." >&2
+ exit 1
+fi
+
+PROJECT_COUNT=0
+for PROJECT in "$@"; do
+ RELATIVE_PROJECT=${PROJECT#"$SCRIPT_DIRECTORY"/}
+ echo "==> Building $RELATIVE_PROJECT ($BUILD_MODE)"
+ if [ "$REBUILD" -eq 1 ]; then
+ lazbuild --build-mode="$BUILD_MODE" --no-write-project --build-all "$PROJECT"
+ else
+ lazbuild --build-mode="$BUILD_MODE" --no-write-project "$PROJECT"
+ fi
+ PROJECT_COUNT=$((PROJECT_COUNT + 1))
+done
+
+echo "Built $PROJECT_COUNT examples into '$OUTPUT_DIRECTORY'."
diff --git a/docs/CHEATSHEET.md b/docs/CHEATSHEET.md
index 8a5ad06..ba33846 100644
--- a/docs/CHEATSHEET.md
+++ b/docs/CHEATSHEET.md
@@ -1,8 +1,10 @@
# ThreadPool for Free Pascal — cheat sheet
-Quick reference for v0.8.5. For full contracts, use the
+Quick reference for v0.9.0. For full contracts, use the
[Simple API](ThreadPool.Simple-API.md) or
-[Producer-Consumer API](ThreadPool.ProducerConsumer-API.md).
+[Producer-Consumer API](ThreadPool.ProducerConsumer-API.md). Task handles,
+batches, ranges, and cancellation are covered by the
+[Tasks API](ThreadPool.Tasks-API.md).
## Choose a pool
@@ -12,6 +14,9 @@ Quick reference for v0.8.5. For full contracts, use the
| A ready-to-use process-wide instance | `GlobalThreadPool` from `ThreadPool.Simple` |
| Bounded memory and explicit queue saturation handling | `ThreadPool.ProducerConsumer` |
| Producer backpressure with a submission deadline | `TProducerConsumerThreadPool.TryQueue` |
+| Observe or cancel one pending task | `Submit` and `IThreadPoolTask` |
+| Coordinate related task handles | `IThreadPoolTaskBatch` |
+| Process an integer range efficiently | `SubmitRange` |
## Import correctly
@@ -23,6 +28,7 @@ uses
{$IFDEF UNIX}
cthreads,
{$ENDIF}
+ ThreadPool.Tasks,
ThreadPool.Simple; // or ThreadPool.ProducerConsumer
```
@@ -59,6 +65,53 @@ procedure ProcessItem(Index: Integer);
procedure TWorker.ProcessItem(Index: Integer);
```
+## Submit observable work
+
+```pascal
+Task := Pool.Submit(@DoWork);
+
+if not Task.WaitFor(250) then
+ WriteLn('Still pending or running')
+else if Task.State = ttsFailed then
+ WriteLn(Task.ErrorMessage);
+```
+
+States are `ttsPending`, `ttsRunning`, `ttsCompleted`, `ttsFailed`, and
+`ttsCancelled`. `WaitFor` returns `True` for any terminal state.
+
+```pascal
+if Task.Cancel then
+ WriteLn('Cancelled before worker start');
+```
+
+Cancellation never interrupts running code. A cancelled entry may remain in
+the queue as a skipped tombstone until a worker reaches it.
+
+## Coordinate a batch
+
+```pascal
+Batch := NewThreadPoolTaskBatch;
+for I := 0 to High(Items) do
+ Batch.Add(Pool.Submit(@ProcessItem, I));
+
+Batch.WaitFor;
+WriteLn(Batch.FinishedCount, '/', Batch.Count);
+```
+
+`Batch.CancelPending` returns the number of pending tasks it cancelled. A wait
+uses the batch entries present when the call starts and one overall timeout.
+
+## Submit a chunked range
+
+```pascal
+Batch := Pool.SubmitRange(@ProcessItem, 0, High(Items));
+Batch.WaitFor;
+```
+
+Bounds are inclusive. The default chunk size `0` creates at most four chunks
+per worker. Pass a positive size for explicit chunks; smaller chunks improve
+load balance and cancellation granularity but add queue overhead.
+
## Handle bounded-queue saturation
Prefer `TryQueue` when using the Producer-Consumer pool:
@@ -153,5 +206,5 @@ On Windows, run `tests/TestRunner.exe` in the final command.
- [README](../README.md)
- [Examples](../examples/)
-- [v0.8.5 release notes](release-notes-v0.8.5.md)
+- [v0.9.0 release notes](release-notes-v0.9.0.md)
- [Changelog](../CHANGELOG.md)
diff --git a/docs/ThreadPool.ProducerConsumer-API.md b/docs/ThreadPool.ProducerConsumer-API.md
index 401a04b..c314b71 100644
--- a/docs/ThreadPool.ProducerConsumer-API.md
+++ b/docs/ThreadPool.ProducerConsumer-API.md
@@ -20,6 +20,10 @@ For simpler use cases see `ThreadPool.Simple`.
>
> See the [official FPC documentation on `cthreads`](https://www.freepascal.org/docs-html/rtl/cthreads/index.html).
+v0.9 observable `Submit` tasks, batches, ranges, and pending cancellation are
+documented in the [ThreadPool.Tasks API](ThreadPool.Tasks-API.md). Bounded
+`TrySubmit` uses the same deadline rules as `TryQueue`.
+
---
## Constructor
@@ -391,6 +395,7 @@ end;
- Fixed queue capacity — no dynamic resizing
- `LastError` holds only the most recent worker exception (use `Errors` to collect all; capped at `MAX_STORED_ERRORS`)
-- No task priority or cancellation support
+- No task priorities or forced cancellation of running callbacks; v0.9 can
+ cancel pending submitted work
- No dynamic thread scaling after construction
- Not suitable for real-time or UI-thread work
diff --git a/docs/ThreadPool.ProducerConsumer-Technical.md b/docs/ThreadPool.ProducerConsumer-Technical.md
index b575e2e..4e09487 100644
--- a/docs/ThreadPool.ProducerConsumer-Technical.md
+++ b/docs/ThreadPool.ProducerConsumer-Technical.md
@@ -106,5 +106,6 @@ diagnostic code cannot terminate a worker or skip completion accounting.
## Limitations
- Queue capacity and worker count are fixed after construction.
-- No task priorities, result-bearing futures, or task cancellation yet.
+- No task priorities, result-bearing futures, or forced cancellation of
+ running callbacks. v0.9 can cancel pending submitted work.
- `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 1659801..b7d641f 100644
--- a/docs/ThreadPool.Simple-API.md
+++ b/docs/ThreadPool.Simple-API.md
@@ -12,6 +12,10 @@
>
> See the [official FPC documentation on `cthreads`](https://www.freepascal.org/docs-html/rtl/cthreads/index.html).
+v0.9 observable `Submit` tasks, batches, ranges, and pending cancellation are
+documented in the [ThreadPool.Tasks API](ThreadPool.Tasks-API.md). The queueing
+surface below remains unchanged.
+
## Thread Pool Types
### GlobalThreadPool
diff --git a/docs/ThreadPool.Simple-Technical.md b/docs/ThreadPool.Simple-Technical.md
index c221fe1..2f151c4 100644
--- a/docs/ThreadPool.Simple-Technical.md
+++ b/docs/ThreadPool.Simple-Technical.md
@@ -213,4 +213,5 @@ rejected before it can wait on itself.
- 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
+- No task prioritisation or forced cancellation of running callbacks; v0.9 can
+ cancel pending submitted work
diff --git a/docs/ThreadPool.Tasks-API.md b/docs/ThreadPool.Tasks-API.md
new file mode 100644
index 0000000..df0813f
--- /dev/null
+++ b/docs/ThreadPool.Tasks-API.md
@@ -0,0 +1,238 @@
+# ThreadPool.Tasks API
+
+`ThreadPool.Tasks` adds observable task handles, batches, chunked ranges, and
+pending-work cancellation to both thread pools. It requires Free Pascal 3.2.2
+or later and uses only FCL units.
+
+On Unix-like systems, `cthreads` must remain the first unit in the program's
+`uses` clause.
+
+```pascal
+uses
+ {$IFDEF UNIX}
+ cthreads,
+ {$ENDIF}
+ ThreadPool.Tasks, ThreadPool.Simple;
+```
+
+The existing v0.8 `Queue` API remains available. Use `Submit` when one piece of
+work needs a handle; use `Queue` for the smallest fire-and-forget path.
+
+## Individual tasks
+
+Both `TSimpleThreadPool` and `TProducerConsumerThreadPool` provide four
+`Submit` overloads matching the existing callback forms:
+
+```pascal
+function Submit(AProcedure: TThreadProcedure): IThreadPoolTask;
+function Submit(AMethod: TThreadMethod): IThreadPoolTask;
+function Submit(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer): IThreadPoolTask;
+function Submit(AMethod: TThreadMethodIndex;
+ AIndex: Integer): IThreadPoolTask;
+```
+
+The returned interface manages its own lifetime. It does not own or retain the
+pool and remains readable after the pool has been shut down and freed.
+
+```pascal
+Task := Pool.Submit(@DoWork);
+
+if not Task.WaitFor(250) then
+ WriteLn('Task has not finished')
+else
+ case Task.State of
+ ttsCompleted: WriteLn('Done');
+ ttsFailed: WriteLn('Failed: ', Task.ErrorMessage);
+ ttsCancelled: WriteLn('Cancelled');
+ end;
+```
+
+### Task states
+
+| State | Meaning |
+| --- | --- |
+| `ttsPending` | Accepted but no worker has claimed it |
+| `ttsRunning` | A worker won the start transition |
+| `ttsCompleted` | Callback returned normally |
+| `ttsFailed` | Callback raised an exception |
+| `ttsCancelled` | Cancellation won before worker start |
+
+`IsFinished` is true for completed, failed, and cancelled tasks.
+
+### Waiting
+
+```pascal
+Task.WaitFor; // no deadline
+Finished := Task.WaitFor(100); // milliseconds
+Finished := Task.WaitFor(0); // immediate check
+Finished := Task.WaitFor(THREADPOOL_INFINITE);
+```
+
+A `True` result means that the task reached any terminal state. Waiting does
+not re-raise a worker exception; inspect `State` and `ErrorMessage`.
+
+Multiple threads may observe or wait for the same task. Completion events are
+created lazily only when unfinished work is actually waited on.
+
+A callback cannot wait for its own task handle. That would prevent its terminal
+transition, so the library raises `EThreadPoolDeadlock` instead. The same guard
+also protects an `OnError` handler that tries to wait for the failing task.
+
+### Failure reporting
+
+A failed submitted task reports its message in two places:
+
+- the task's `ErrorMessage`; and
+- the existing pool-level `LastError`, `Errors`, and `OnError` API.
+
+`OnError` still runs synchronously on the worker. A task becomes terminal after
+the error callback returns, preserving the existing completion behaviour.
+
+## Timeout-aware submission
+
+Each callback form also has a `TrySubmit` overload:
+
+```pascal
+if not Pool.TrySubmit(@DoWork, 50, Task) then
+ WriteLn('Queue remained full for 50 ms');
+```
+
+For indexed callbacks, the order is callback, index, timeout, output task:
+
+```pascal
+Accepted := Pool.TrySubmit(@ProcessItem, I, 50, Task);
+```
+
+On the bounded Producer-Consumer pool, `False` means that admission reached its
+deadline, and the output task is `nil`. `Submit` uses the same bounded
+compatibility timeout as `Queue` and raises `EQueueFullException` on expiry.
+
+The Simple pool is unbounded, so `TrySubmit` normally returns `True` and its
+timeout exists for API symmetry. Both pools raise `EThreadPoolShutdown` if
+shutdown has begun.
+
+## Cancelling pending work
+
+```pascal
+if Task.Cancel then
+ WriteLn('The callback will not run')
+else
+ WriteLn('The task had already started or finished');
+```
+
+`Cancel` is a single-winner state transition:
+
+```text
+pending -> running -> completed or failed
+ |
+ +-----> cancelled
+```
+
+- It returns `True` only for `pending -> cancelled`.
+- It is safe to call repeatedly.
+- It never terminates a worker or interrupts a running callback.
+- Cancellation is not an error and does not fire `OnError`.
+
+A cancelled item remains as a tombstone until a worker reaches its queue
+position. Its handle is terminal immediately and its callback is skipped. In a
+bounded queue, the tombstone can occupy its slot briefly; v0.9 deliberately
+avoids an expensive queue-wide removal scan.
+
+If a submitted object method cannot be cancelled because it is running, keep
+the callback target alive until the task finishes. A successful cancellation
+guarantees that the callback will not be invoked.
+
+## Batches
+
+Create a batch and add handles from either pool:
+
+```pascal
+Batch := NewThreadPoolTaskBatch;
+for I := 0 to High(Items) do
+ Batch.Add(Pool.Submit(@ProcessItem, I));
+
+if not Batch.WaitFor(1000) then
+ WriteLn('Some tasks remain');
+```
+
+`IThreadPoolTaskBatch` provides:
+
+| Member | Meaning |
+| --- | --- |
+| `Add(Task)` | Add one non-nil task handle |
+| `WaitFor` | Wait indefinitely for the current snapshot |
+| `WaitFor(TimeoutMS)` | Wait using one overall deadline |
+| `CancelPending` | Attempt every pending cancellation; return wins |
+| `Count` | Number of entries |
+| `FinishedCount` | Completed, failed, or cancelled entries |
+| `FailedCount` | Failed entries |
+| `CancelledCount` | Cancelled entries |
+| `Tasks[Index]` | Read an individual handle |
+
+An empty batch is already finished. `WaitFor` snapshots the entries at call
+entry; tasks added concurrently are included in a later wait. Adding the same
+handle twice creates two entries. Count properties are thread-safe snapshots,
+but tasks can change state immediately after a count is read.
+
+## Chunked ranges
+
+`SubmitRange` submits an inclusive integer range and returns a populated batch:
+
+```pascal
+RangeTasks := Pool.SubmitRange(@ProcessItem, 0, High(Items));
+RangeTasks.WaitFor;
+```
+
+Procedure and object-method indexed callbacks are supported. Bounds follow
+Pascal's `for I := First to Last` convention:
+
+- `First > Last` returns an empty batch;
+- `AChunkSize = 0` selects automatic chunking;
+- `AChunkSize > 0` selects that many indexes per queued task; and
+- a negative chunk size raises `EArgumentOutOfRangeException`.
+
+Automatic mode creates at most `ThreadCount * 4` chunks. Range arithmetic uses
+64-bit intermediates, so valid `Integer` bounds do not overflow while chunks
+are calculated.
+
+One batch entry represents one chunk. Cancelling a pending entry skips the
+whole chunk; a running chunk finishes. Smaller explicit chunks trade queue
+overhead for finer cancellation and load balancing.
+
+If an index callback raises, that chunk stops and becomes failed while other
+chunks continue. Catch exceptions inside the callback when every index must be
+attempted.
+
+The bounded pool rejects `SubmitRange` from one of its own workers with
+`EThreadPoolDeadlock`. Blocking a fixed worker set while it tries to feed its
+own full queue can otherwise starve the pool. Submit bounded ranges from a
+coordinating thread. The coordinating call may wait for queue space until all
+of its small set of chunks has been accepted.
+
+## Capability interface
+
+New interface-oriented code can depend on `IThreadPoolTaskSource`. It describes
+the common `Submit`, `TrySubmit`, and `SubmitRange` surface implemented by both
+pools.
+
+The v0.8 `IThreadPool` interface and GUID are unchanged. This is intentional:
+third-party `IThreadPool` implementations continue to compile without adding
+v0.9 methods.
+
+## Examples
+
+See [`examples/TaskCoordination`](../examples/TaskCoordination/) for individual
+waiting, batches, ranges, state inspection, and best-effort pending
+cancellation in one small program.
+
+Two production-shaped examples show how those pieces fit into complete
+workflows:
+
+- [`CoordinatedFileBackup`](../examples/CoordinatedFileBackup/) submits one
+ observable task per file, reports batch progress, reacts to a critical
+ failure, cancels only work that has not started, and waits before releasing
+ callback-owned state.
+- [`ParallelLogAnalyzer`](../examples/ParallelLogAnalyzer/) analyzes 100,000
+ records with automatic range chunking, uses the range batch as a phase
+ barrier, then coordinates independent report sections as a second batch.
diff --git a/docs/release-notes-v0.9.0.md b/docs/release-notes-v0.9.0.md
new file mode 100644
index 0000000..e7471e2
--- /dev/null
+++ b/docs/release-notes-v0.9.0.md
@@ -0,0 +1,99 @@
+# ThreadPool for Free Pascal v0.9.0
+
+v0.9.0 adds observable work and coordination while keeping the v0.8 queueing
+surface source-compatible.
+
+## Highlights
+
+- `Submit` returns an `IThreadPoolTask` that can be observed and waited.
+- `TrySubmit` combines a task handle with bounded admission deadlines.
+- `Cancel` prevents a pending callback from starting without terminating a
+ worker thread.
+- `IThreadPoolTaskBatch` coordinates related handles and cancels their pending
+ work as a group.
+- `SubmitRange` divides inclusive integer ranges into a small number of chunks.
+- Both Simple and Producer-Consumer pools implement the same task semantics.
+
+## Quick start
+
+```pascal
+uses
+ ThreadPool.Tasks, ThreadPool.Simple;
+
+var
+ Task: IThreadPoolTask;
+ Batch: IThreadPoolTaskBatch;
+begin
+ Task := GlobalThreadPool.Submit(@DoWork);
+ Task.WaitFor;
+
+ Batch := GlobalThreadPool.SubmitRange(@ProcessItem, 0, 999);
+ Batch.WaitFor;
+end;
+```
+
+See the [task API](ThreadPool.Tasks-API.md) and
+[`TaskCoordination`](../examples/TaskCoordination/) example for the complete
+contract.
+
+For real workflow patterns, see
+[`CoordinatedFileBackup`](../examples/CoordinatedFileBackup/) for observation,
+failure policy, and pending cancellation, and
+[`ParallelLogAnalyzer`](../examples/ParallelLogAnalyzer/) for efficient range
+chunking and phase coordination.
+
+## Cancellation contract
+
+Cancellation is intentionally narrow and safe. `Task.Cancel` succeeds only
+while the task is pending. Once a worker transitions it to running,
+cancellation returns `False` and the callback finishes normally.
+
+Waiting for the current callback's own task would deadlock its terminal
+transition, so it is rejected with `EThreadPoolDeadlock`.
+
+Cancelled entries are skipped when dequeued. A cancelled handle becomes
+terminal immediately, although the queue entry remains as a tombstone until a
+worker reaches it. This keeps handles independent from pool lifetime and avoids
+queue-wide removal scans.
+
+## Range contract
+
+Range bounds are inclusive. Automatic chunking creates at most four chunks per
+worker; a positive explicit chunk size selects the number of indexes per task.
+A range returns a batch whose entries represent chunks.
+
+One index failure stops its chunk and marks that task failed. Other chunks
+continue. A pending chunk cancellation skips all of its indexes, while a
+running chunk finishes.
+
+The bounded pool rejects nested `SubmitRange` from one of its own workers.
+Blocking a fixed worker set while it submits to its own full queue can
+otherwise deadlock through starvation.
+
+## Compatibility
+
+The existing `IThreadPool` interface and GUID are unchanged. The new
+`IThreadPoolTaskSource` capability keeps third-party v0.8 implementations from
+having to add methods.
+
+Existing programs can continue using:
+
+- all four `Queue` and `TryQueue` forms;
+- pool-wide `WaitForAll` and draining `Shutdown`;
+- `LastError`, `Errors`, and `OnError`;
+- the existing constructors and managed `GlobalThreadPool`.
+
+The test suite includes an independent v0.8-style `IThreadPool`
+implementation as a compile sentinel.
+
+## Performance and verification
+
+Legacy `Queue` remains untracked. New task state is allocated only for
+`Submit`, and its completion event is lazy.
+
+The release benchmark now separates legacy queue, tracked submit, and range
+paths. On the Windows/FPC 3.2.2 orientation run, the legacy 20,000-task median
+remained within the 10% v0.8.5 budget, and chunking a 200,000-index Simple range
+was over 60x faster than submitting each index as a tracked task.
+
+The v0.9 suite contains 81 tests and reports no leaks under heap tracing.
diff --git a/docs/v0.9.0-plan.md b/docs/v0.9.0-plan.md
new file mode 100644
index 0000000..01ff9e6
--- /dev/null
+++ b/docs/v0.9.0-plan.md
@@ -0,0 +1,411 @@
+# v0.9.0 development plan
+
+Status: implemented on `feat/v0.9.0-task-coordination`; Windows verification
+complete, Linux CI verification pending
+
+## Exit criterion
+
+> Users can submit work, observe and wait for individual tasks, coordinate
+> batches, parallelize ranges efficiently, and cancel pending work without
+> breaking existing v0.8.x programs.
+
+This criterion is complete only when it holds for both `ThreadPool.Simple` and
+`ThreadPool.ProducerConsumer` on Free Pascal 3.2.2 or later, on Windows and
+Linux.
+
+## Design principles
+
+1. Preserve the v0.8 API and behaviour. Existing `Queue`, `TryQueue`,
+ `WaitForAll`, `Shutdown`, error handling, constructors, and
+ `GlobalThreadPool` ownership remain valid.
+2. Add one task state machine shared by both pool implementations. Queue
+ storage and backpressure may differ; task semantics must not.
+3. Keep ownership explicit. A task handle owns task state, not the pool. A
+ handle may safely outlive its pool, and retaining a handle must not keep a
+ pool alive.
+4. Cancel queued callbacks, never worker threads. Free Pascal has no safe
+ general-purpose forced thread termination.
+5. Keep the legacy fast path fast. `Queue` callers that do not request a handle
+ should not pay for a task handle or completion event.
+6. Prefer small FCL-only types, ordinary interfaces, and the four callback
+ forms already familiar to users. Do not require generics, anonymous
+ functions, or external packages.
+
+## Proposed public API
+
+Add a `ThreadPool.Tasks` unit containing the new public contracts. Both
+concrete pools expose the same overloads and implement a new capability
+interface. Do **not** add methods to the existing `IThreadPool` interface:
+doing that would break third-party classes that currently implement it.
+
+The names below are the proposed contract. They should be compiled in a small
+API spike before implementation is spread across both pools.
+
+```pascal
+type
+ TThreadPoolTaskState = (
+ ttsPending,
+ ttsRunning,
+ ttsCompleted,
+ ttsFailed,
+ ttsCancelled
+ );
+
+ IThreadPoolTask = interface
+ ['{new-guid}']
+ procedure WaitFor; overload;
+ function WaitFor(ATimeoutMS: Cardinal): Boolean; overload;
+ function Cancel: Boolean;
+ function GetState: TThreadPoolTaskState;
+ function GetErrorMessage: string;
+ function GetIsFinished: Boolean;
+ property State: TThreadPoolTaskState read GetState;
+ property ErrorMessage: string read GetErrorMessage;
+ property IsFinished: Boolean read GetIsFinished;
+ end;
+
+ IThreadPoolTaskBatch = interface
+ ['{new-guid}']
+ procedure Add(const ATask: IThreadPoolTask);
+ procedure WaitFor; overload;
+ function WaitFor(ATimeoutMS: Cardinal): Boolean; overload;
+ function CancelPending: Integer;
+ function GetCount: Integer;
+ function GetFinishedCount: Integer;
+ function GetFailedCount: Integer;
+ function GetCancelledCount: Integer;
+ function GetTask(AIndex: Integer): IThreadPoolTask;
+ property Count: Integer read GetCount;
+ property FinishedCount: Integer read GetFinishedCount;
+ property FailedCount: Integer read GetFailedCount;
+ property CancelledCount: Integer read GetCancelledCount;
+ property Tasks[AIndex: Integer]: IThreadPoolTask read GetTask; default;
+ end;
+
+function NewThreadPoolTaskBatch: IThreadPoolTaskBatch;
+```
+
+Each pool gets the existing four callback forms for these operations:
+
+```pascal
+function Submit(...): IThreadPoolTask; overload;
+function TrySubmit(...; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+function SubmitRange(...; AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
+```
+
+`IThreadPoolTaskSource` should describe that shared surface for callers that
+need an interface. `TSimpleThreadPool`, `TProducerConsumerThreadPool`, and
+`GlobalThreadPool` also expose the methods directly, which is the simplest
+route for new Free Pascal users.
+
+### Typical use
+
+```pascal
+var
+ Task: IThreadPoolTask;
+begin
+ Task := GlobalThreadPool.Submit(@DoWork);
+ if not Task.WaitFor(250) then
+ WriteLn('Still running or queued')
+ else if Task.State = ttsFailed then
+ WriteLn(Task.ErrorMessage);
+end;
+```
+
+```pascal
+var
+ Batch: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ Batch := NewThreadPoolTaskBatch;
+ for I := 0 to 99 do
+ Batch.Add(GlobalThreadPool.Submit(@ProcessItem, I));
+
+ Batch.WaitFor;
+end;
+```
+
+```pascal
+var
+ RangeTasks: IThreadPoolTaskBatch;
+begin
+ RangeTasks := GlobalThreadPool.SubmitRange(
+ @ProcessItem, 0, High(Items));
+ RangeTasks.WaitFor;
+end;
+```
+
+## Required semantics
+
+### Submission and waiting
+
+- `Submit` returns a non-nil handle for accepted work. The task may already be
+ running or finished when `Submit` returns.
+- On the bounded pool, `Submit` uses the same compatibility timeout and
+ `EQueueFullException` behaviour as `Queue`.
+- `TrySubmit` returns `False` only when bounded admission reaches its deadline;
+ in that case its output task is `nil`. Shutdown still raises
+ `EThreadPoolShutdown`, matching `TryQueue`.
+- `WaitFor(0)` is an immediate check. Finite timeouts are measured from call
+ entry. `THREADPOOL_INFINITE` means no deadline, exactly as in v0.8.
+- A successful wait means the task reached any terminal state. Callers inspect
+ `State` to distinguish completion, failure, and cancellation.
+- A task failure continues to update `LastError`, `Errors`, and `OnError` as it
+ does in v0.8, and also stores the message in `Task.ErrorMessage`.
+- Multiple threads may observe or wait for the same task.
+
+### Task state machine
+
+```text
+ callback returned
+ +------------------> completed
+ |
+pending ----------> running
+ | |
+ | Cancel wins +------------------> failed
+ | callback raised
+ v
+cancelled
+```
+
+- Exactly one of the worker's start transition and the caller's cancel
+ transition wins while a task is pending.
+- `Cancel` returns `True` only for `pending -> cancelled`. It returns `False`
+ for running or terminal tasks and is safe to call repeatedly.
+- Cancellation never interrupts a running callback and never fires `OnError`.
+- A cancelled queue entry may remain as a cheap tombstone until a worker pops
+ it. Its callback is skipped, its task handle is terminal immediately, and
+ the pool's accepted-work accounting is decremented exactly once when the
+ tombstone is drained. This avoids an unsafe pool reference inside handles.
+- In the bounded pool, a cancelled tombstone may occupy its queue slot until it
+ reaches a worker. This trade-off must be documented and benchmarked; eager
+ arbitrary queue removal is not required for v0.9.0.
+
+### Batch behaviour
+
+- A batch is an independent, thread-safe collection of task handles. It does
+ not own or retain a pool.
+- `Add(nil)` raises an argument exception. Adding the same handle twice is
+ permitted and represents two entries; document this explicitly.
+- `WaitFor` takes a snapshot of the current entries at call entry. Tasks added
+ later belong to a later wait. This removes an otherwise ambiguous race
+ between producers and waiters.
+- Timed waits use one overall deadline, not a fresh timeout for every entry.
+- An empty batch is already finished.
+- `CancelPending` attempts cancellation for every entry and returns the number
+ of transitions it won. Running work is left alone.
+- Count properties are snapshots and need not describe one globally atomic
+ instant while tasks are changing state.
+
+### Range behaviour
+
+- Bounds are inclusive, matching Pascal's `for I := First to Last` syntax.
+- `First > Last` returns an empty, already-finished batch.
+- `AChunkSize > 0` is an explicit number of indexes per queued task.
+- `AChunkSize = 0` chooses approximately
+ `ceil(ItemCount / (ThreadCount * 4))`, giving at most four tasks per worker.
+ Use 64-bit intermediate arithmetic so extreme `Integer` bounds do not
+ overflow.
+- One batch entry represents one chunk, not one index. This is the source of
+ the performance improvement over a loop of individual submissions.
+- Cancelling a pending chunk skips every index in that chunk. A running chunk
+ finishes; callers can select a smaller explicit chunk size when they value
+ cancellation granularity over throughput.
+- If one index raises, that chunk stops and becomes failed. Other chunks keep
+ running. A callback that must attempt every index should catch its own
+ per-index errors.
+- Initial v0.9 range submission is not nested-parallelism support. A bounded
+ pool worker must not block while submitting more work to its own pool; detect
+ and reject that case rather than permit a starvation deadlock.
+
+## Internal architecture
+
+### Units and responsibilities
+
+| Unit | Responsibility |
+| --- | --- |
+| `ThreadPool.Types` | Keep v0.8 callback types, lifecycle, errors, and `IThreadPool` stable |
+| `ThreadPool.Tasks` | New task/batch contracts, state control, lazy wait event, and shared tracked work items |
+| `ThreadPool.Simple` | Unbounded FIFO admission/dequeue and Simple worker ownership |
+| `ThreadPool.ProducerConsumer` | Bounded FIFO admission/dequeue and backpressure |
+
+Keep task state transitions in one implementation in `ThreadPool.Tasks`.
+Workers in both pools should use the same small internal tracked-work-item
+interface:
+
+1. Attempt `pending -> running`.
+2. Skip execution if cancellation already won.
+3. Execute the existing callback wrapper.
+4. Publish `completed` or capture the error and publish `failed`.
+5. Run the pool's accepted-work completion accounting once in `finally`.
+
+The task control object should use a `TCriticalSection` for state, error text,
+and lazy event creation. Create its manual-reset `TEvent` only when a caller
+actually waits on unfinished work. This keeps `Submit` cheaper when handles are
+used only for observation and avoids creating an operating-system event for
+every task.
+
+Do not make `IThreadPoolTask` retain a pool or expose internal work items.
+Do not use `TThread.Terminate` as task cancellation; workers use it only during
+the existing draining pool shutdown.
+
+### Compatibility boundary
+
+- Leave `IThreadPool` and its GUID exactly unchanged.
+- Do not reorder existing enum members or change existing exception classes.
+- Do not rename units, classes, public fields, methods, or constructor defaults.
+- Keep `Queue` and `TryQueue` on the untracked path. New `Submit` calls use the
+ tracked path.
+- Keep `WaitForAll` defined over all accepted pool work, including cancelled
+ tombstones not yet drained.
+- Compile representative v0.8.0 and v0.8.5 client source unchanged in CI.
+
+## Test plan
+
+Tests belong in a dedicated `ThreadPool.Tasks.Tests` unit for shared contracts,
+with pool-specific admission and lifecycle cases in the two existing suites.
+Prefer events and gates over timing sleeps.
+
+### Task contract
+
+- all success, failure, and cancellation transitions;
+- wait before, during, and after completion;
+- zero, finite, and infinite wait timeouts;
+- multiple concurrent waiters and observers;
+- cancel/start race repeated enough to show exactly one winner;
+- cancel is idempotent and cannot cancel running work;
+- a cancelled callback is never invoked;
+- task failure message agrees with the pool error collection;
+- `OnError` failure cannot strand a task in `running`;
+- task handle remains readable after its pool is freed;
+- callback target lifetime examples remain safe.
+
+### Batch contract
+
+- empty and single-task batches;
+- mixed completed, failed, cancelled, pending, and running entries;
+- one overall timeout across many entries;
+- snapshot-at-wait behaviour while another thread adds tasks;
+- concurrent waiters and count readers;
+- `CancelPending` count and repeated calls;
+- invalid index and nil-add validation.
+
+### Range contract
+
+- empty, one-item, negative-bound, and ordinary inclusive ranges;
+- every index runs exactly once;
+- procedure and method callback overloads;
+- automatic and explicit chunk sizes;
+- no more than `ThreadCount * 4` automatic chunks;
+- chunk failure does not stop other chunks;
+- pending chunk cancellation skips its indexes;
+- bounded queue with a capacity smaller than the chunk count;
+- shutdown racing with range admission has no lost accounting or deadlock;
+- 64-bit bound calculations avoid overflow.
+
+### Compatibility and robustness
+
+- retain all 58 v0.8.5 tests unchanged;
+- add source-compatibility fixtures that use only v0.8 APIs;
+- run the full suite with heap tracing and require zero leaks;
+- build package, tests, benchmark, and all examples on Windows and Linux;
+- add stress cases for submit/cancel, submit/shutdown, and wait/cancel races.
+
+## Performance plan
+
+The current Windows/FPC 3.2.2 `-O3` baseline measured on 2026-07-14 is:
+
+| Benchmark | Baseline |
+| --- | ---: |
+| Simple 20,000-task `Queue` burst | 109 ms |
+| Producer-Consumer 20,000-task `Queue` burst | 156 ms |
+| Simple idle queue-to-start average | below 1 ms |
+| Producer-Consumer idle queue-to-start average | below 1 ms |
+
+These are single-run orientation numbers; release decisions use the median of
+at least five otherwise-idle runs on identical hardware.
+
+Extend the benchmark with separate `Queue`, `Submit`, and `SubmitRange` cases.
+Use these release budgets:
+
+- legacy `Queue` burst median: no more than 10% slower than v0.8.5;
+- idle queue-to-start latency: remain below the timer's 1 ms resolution;
+- `Submit` overhead: report it separately and keep it bounded; it must not alter
+ the legacy result;
+- a large trivial `SubmitRange` workload: at least 5x faster than submitting
+ every index separately, with at most `ThreadCount * 4` queued tasks in
+ automatic mode;
+- cancellation and batch operations: O(1) per task, with no queue-wide scan on
+ the normal submit/dequeue path.
+
+Do not enforce absolute millisecond limits in shared CI runners. Store the
+benchmark format and comparison procedure so maintainers can make a local
+median comparison before release.
+
+## Delivery slices
+
+### 1. Contract and compatibility harness
+
+- Add the new unit and compile-only API spike.
+- Add unchanged v0.8 client fixtures.
+- Freeze state, timeout, error, lifetime, and cancellation semantics in tests.
+
+### 2. Individual task handles
+
+- Implement the shared task control and lazy wait event.
+- Add `Submit`/`TrySubmit` to Simple, then Producer-Consumer parity.
+- Add race, failure, lifecycle, and handle-after-pool-free tests.
+- Benchmark legacy `Queue` before moving on.
+
+### 3. Batches
+
+- Implement the snapshot-based task collection.
+- Add wait, count, and cancel coordination tests.
+- Add a small batch example using both pools.
+
+### 4. Chunked ranges
+
+- Implement procedure and method range work items.
+- Add overflow-safe automatic chunking and explicit chunk size.
+- Add correctness, failure, cancellation, bounded-queue, and performance tests.
+
+### 5. Hardening and release candidate
+
+- Run race stress tests and heap tracing on Windows and Linux.
+- Compare five-run release benchmark medians with v0.8.5.
+- Update package contents/version, README, cheat sheet, API/technical docs,
+ changelog, release notes, and examples.
+- Build a clean checkout and verify every documented command.
+
+## Release checklist
+
+- [x] Both pools implement identical task and batch semantics.
+- [x] Individual tasks can be observed, timed-waited, and indefinitely waited.
+- [x] Pending cancellation has a tested single-winner race with worker start.
+- [x] Batch waiting, counts, and pending cancellation are documented and tested.
+- [x] Automatic range submission uses at most four chunks per worker and runs
+ each included index exactly once.
+- [x] All v0.8.5 tests and compatibility fixtures pass unchanged.
+- [ ] Package, tests, benchmark, and all examples build on Windows and Linux
+ (Windows complete; Linux requires CI).
+- [x] Heap tracing reports no leaks.
+- [x] Legacy performance stays inside the agreed budget.
+- [x] New-developer examples cover task, batch, range, failure, and cancellation.
+- [x] The v0.9.0 changelog calls out precise cancellation and lifetime limits.
+
+## Explicit non-goals for v0.9.0
+
+- forcibly stopping running callbacks;
+- cooperative cancellation tokens inside callbacks;
+- return values, futures, or generic typed tasks;
+- task priorities or work stealing;
+- dynamic worker counts;
+- dependency graphs or continuations;
+- nested parallel ranges on the same bounded pool;
+- immediate arbitrary removal of cancelled bounded-queue entries.
+
+These can be considered after the task contract has shipped and gained real
+usage. Keeping them out of v0.9.0 makes the exit criterion achievable without
+turning a small Free Pascal library into a general scheduling framework.
diff --git a/examples/CoordinatedFileBackup/CoordinatedFileBackup.lpi b/examples/CoordinatedFileBackup/CoordinatedFileBackup.lpi
new file mode 100644
index 0000000..61fc3da
--- /dev/null
+++ b/examples/CoordinatedFileBackup/CoordinatedFileBackup.lpi
@@ -0,0 +1,64 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ -
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/examples/CoordinatedFileBackup/CoordinatedFileBackup.lpr b/examples/CoordinatedFileBackup/CoordinatedFileBackup.lpr
new file mode 100644
index 0000000..a73255c
--- /dev/null
+++ b/examples/CoordinatedFileBackup/CoordinatedFileBackup.lpr
@@ -0,0 +1,275 @@
+program CoordinatedFileBackup;
+
+{$mode objfpc}{$H+}{$J-}
+
+uses
+ {$IFDEF UNIX}
+ cthreads,
+ {$ENDIF}
+ Classes, SysUtils, ThreadPool.Tasks, ThreadPool.ProducerConsumer;
+
+const
+ FILE_COUNT = 18;
+ WORKER_COUNT = 4;
+ QUEUE_SIZE = 32;
+ MISSING_FILE_INDEX = 0;
+ COPY_BUFFER_SIZE = 4096;
+ BLOCKS_PER_FILE = 16;
+
+type
+ TStringArray = array of string;
+ TChecksumArray = array of QWord;
+
+ { One job object owns the data used by every callback. It must outlive the
+ pool and every task handle that refers to one of its methods. }
+ TBackupJob = class
+ private
+ FRootDirectory: string;
+ FSourceDirectory: string;
+ FTargetDirectory: string;
+ FFileNames: TStringArray;
+ FChecksums: TChecksumArray;
+ function SourcePath(AIndex: Integer): string;
+ function TargetPath(AIndex: Integer): string;
+ function ChecksumFile(const AFileName: string): QWord;
+ public
+ constructor Create;
+ procedure CreateDemoFiles;
+ procedure CopyAndVerify(AIndex: Integer);
+ procedure Cleanup;
+ property FileNames: TStringArray read FFileNames;
+ property Checksums: TChecksumArray read FChecksums;
+ end;
+
+function TaskStateName(AState: TThreadPoolTaskState): string;
+begin
+ Result := 'unknown';
+ case AState of
+ ttsPending: Result := 'pending';
+ ttsRunning: Result := 'running';
+ ttsCompleted: Result := 'completed';
+ ttsFailed: Result := 'failed';
+ ttsCancelled: Result := 'cancelled';
+ end;
+end;
+
+constructor TBackupJob.Create;
+var
+ I: Integer;
+begin
+ inherited Create;
+ FRootDirectory := IncludeTrailingPathDelimiter(GetTempDir(False)) +
+ 'threadpool-fp-backup-' + IntToStr(GetProcessID) + '-' +
+ IntToStr(Int64(GetTickCount64));
+ FSourceDirectory := IncludeTrailingPathDelimiter(FRootDirectory) + 'source';
+ FTargetDirectory := IncludeTrailingPathDelimiter(FRootDirectory) + 'target';
+ SetLength(FFileNames, FILE_COUNT);
+ SetLength(FChecksums, FILE_COUNT);
+ FFileNames[MISSING_FILE_INDEX] := 'required-settings.ini';
+ for I := 1 to High(FFileNames) do
+ FFileNames[I] := Format('customer-document-%.2d.dat', [I]);
+end;
+
+function TBackupJob.SourcePath(AIndex: Integer): string;
+begin
+ Result := IncludeTrailingPathDelimiter(FSourceDirectory) +
+ FFileNames[AIndex];
+end;
+
+function TBackupJob.TargetPath(AIndex: Integer): string;
+begin
+ Result := IncludeTrailingPathDelimiter(FTargetDirectory) +
+ FFileNames[AIndex];
+end;
+
+procedure TBackupJob.Cleanup;
+var
+ I: Integer;
+begin
+ for I := 0 to High(FFileNames) do
+ begin
+ DeleteFile(SourcePath(I));
+ DeleteFile(TargetPath(I));
+ end;
+ RemoveDir(FSourceDirectory);
+ RemoveDir(FTargetDirectory);
+ RemoveDir(FRootDirectory);
+end;
+
+procedure TBackupJob.CreateDemoFiles;
+var
+ Buffer: array[0..COPY_BUFFER_SIZE - 1] of Byte;
+ OutputFile: TFileStream;
+ I, J, Block: Integer;
+begin
+ Cleanup;
+ ForceDirectories(FSourceDirectory);
+ ForceDirectories(FTargetDirectory);
+
+ { Index zero is intentionally absent. A backup policy can treat a missing
+ required settings file as fatal and cancel files still waiting in line. }
+ for I := 1 to High(FFileNames) do
+ begin
+ for J := 0 to High(Buffer) do
+ Buffer[J] := Byte((I * 31 + J) mod 256);
+
+ OutputFile := TFileStream.Create(SourcePath(I), fmCreate);
+ try
+ for Block := 1 to BLOCKS_PER_FILE do
+ OutputFile.WriteBuffer(Buffer, SizeOf(Buffer));
+ finally
+ OutputFile.Free;
+ end;
+ end;
+end;
+
+function TBackupJob.ChecksumFile(const AFileName: string): QWord;
+var
+ Buffer: array[0..COPY_BUFFER_SIZE - 1] of Byte;
+ InputFile: TFileStream;
+ BytesRead, I: Integer;
+begin
+ Result := 0;
+ for I := 0 to High(Buffer) do
+ Buffer[I] := 0;
+ InputFile := TFileStream.Create(AFileName, fmOpenRead or fmShareDenyNone);
+ try
+ repeat
+ BytesRead := InputFile.Read(Buffer, SizeOf(Buffer));
+ for I := 0 to BytesRead - 1 do
+ Inc(Result, Buffer[I]);
+ until BytesRead = 0;
+ finally
+ InputFile.Free;
+ end;
+end;
+
+procedure TBackupJob.CopyAndVerify(AIndex: Integer);
+var
+ Buffer: array[0..COPY_BUFFER_SIZE - 1] of Byte;
+ InputFile, OutputFile: TFileStream;
+ BytesRead, I: Integer;
+ SourceChecksum, TargetChecksum: QWord;
+begin
+ if not FileExists(SourcePath(AIndex)) then
+ raise Exception.CreateFmt('Required source is missing: %s',
+ [FFileNames[AIndex]]);
+
+ SourceChecksum := 0;
+ for I := 0 to High(Buffer) do
+ Buffer[I] := 0;
+ InputFile := TFileStream.Create(SourcePath(AIndex),
+ fmOpenRead or fmShareDenyNone);
+ try
+ OutputFile := TFileStream.Create(TargetPath(AIndex), fmCreate);
+ try
+ repeat
+ BytesRead := InputFile.Read(Buffer, SizeOf(Buffer));
+ if BytesRead > 0 then
+ begin
+ OutputFile.WriteBuffer(Buffer, BytesRead);
+ for I := 0 to BytesRead - 1 do
+ Inc(SourceChecksum, Buffer[I]);
+
+ { This models a rate-limited remote backup target. It also keeps
+ enough work pending to make cancellation visible in the demo. }
+ Sleep(2);
+ end;
+ until BytesRead = 0;
+ finally
+ OutputFile.Free;
+ end;
+ finally
+ InputFile.Free;
+ end;
+
+ TargetChecksum := ChecksumFile(TargetPath(AIndex));
+ if TargetChecksum <> SourceChecksum then
+ raise Exception.CreateFmt('Verification failed: %s', [FFileNames[AIndex]]);
+
+ { Each callback owns a distinct array slot, so no lock is needed. }
+ FChecksums[AIndex] := TargetChecksum;
+end;
+
+function FindFailedTask(const ABatch: IThreadPoolTaskBatch): Integer;
+var
+ I: Integer;
+begin
+ Result := -1;
+ for I := 0 to ABatch.Count - 1 do
+ if ABatch[I].State = ttsFailed then
+ Exit(I);
+end;
+
+procedure PrintProgress(const ABatch: IThreadPoolTaskBatch);
+begin
+ WriteLn('Progress: ', ABatch.FinishedCount, '/', ABatch.Count,
+ ' finished (failed=', ABatch.FailedCount,
+ ', cancelled=', ABatch.CancelledCount, ')');
+end;
+
+var
+ Pool: TProducerConsumerThreadPool;
+ Job: TBackupJob;
+ Batch: IThreadPoolTaskBatch;
+ Task: IThreadPoolTask;
+ FailedIndex, CancelledCount, I: Integer;
+begin
+ Pool := TProducerConsumerThreadPool.Create(WORKER_COUNT, QUEUE_SIZE);
+ Job := TBackupJob.Create;
+ try
+ Job.CreateDemoFiles;
+ Batch := NewThreadPoolTaskBatch;
+
+ WriteLn('Starting a coordinated backup of ', FILE_COUNT, ' files...');
+ for I := 0 to FILE_COUNT - 1 do
+ begin
+ Task := Pool.Submit(@Job.CopyAndVerify, I);
+ Batch.Add(Task);
+ end;
+
+ { Observe the batch without blocking forever. A real service could update
+ a UI, answer a health endpoint, or enforce a deadline in this loop. }
+ repeat
+ FailedIndex := FindFailedTask(Batch);
+ if FailedIndex >= 0 then
+ begin
+ WriteLn('Critical backup error: ', Batch[FailedIndex].ErrorMessage);
+ CancelledCount := Batch.CancelPending;
+ WriteLn('Policy response: cancelled ', CancelledCount,
+ ' file(s) that had not started.');
+ Break;
+ end;
+
+ if Batch.WaitFor(10) then
+ Break;
+ PrintProgress(Batch);
+ until False;
+
+ Batch.WaitFor;
+
+ { Cancelled queue entries are harmless tombstones. WaitForAll lets the
+ bounded pool consume them before the callback owner is released. }
+ Pool.WaitForAll;
+ PrintProgress(Batch);
+
+ WriteLn;
+ WriteLn('Per-file outcome:');
+ for I := 0 to Batch.Count - 1 do
+ begin
+ Write(' ', Job.FileNames[I], ': ', TaskStateName(Batch[I].State));
+ case Batch[I].State of
+ ttsCompleted:
+ Write(' (checksum ', Job.Checksums[I], ')');
+ ttsFailed:
+ Write(' (', Batch[I].ErrorMessage, ')');
+ end;
+ WriteLn;
+ end;
+ finally
+ Pool.WaitForAll;
+ Pool.Free;
+ Job.Cleanup;
+ Job.Free;
+ end;
+end.
diff --git a/examples/CoordinatedFileBackup/README.md b/examples/CoordinatedFileBackup/README.md
new file mode 100644
index 0000000..2022356
--- /dev/null
+++ b/examples/CoordinatedFileBackup/README.md
@@ -0,0 +1,22 @@
+# Coordinated file backup
+
+This self-contained example models a backup to a rate-limited remote target.
+It creates demo files, submits one observable task per file to a bounded pool,
+and groups the handles in a batch.
+
+The required settings file is intentionally missing. The coordinator observes
+that task's failure and applies an all-or-nothing policy: work that is still
+pending is cancelled, while copies already running finish safely. It then
+waits for the pool before releasing the object used by the callbacks.
+
+The exact completed/cancelled counts may vary with scheduling. The important
+invariants are that one file fails, pending tasks do not start after successful
+cancellation, and running tasks are never interrupted.
+
+Build with Lazarus or run:
+
+```text
+lazbuild CoordinatedFileBackup.lpi
+```
+
+The generated source and target directories are removed after the report.
diff --git a/examples/ParallelFileHasher/ParallelFileHasher.lpi b/examples/ParallelFileHasher/ParallelFileHasher.lpi
index dcb18fc..79965c9 100644
--- a/examples/ParallelFileHasher/ParallelFileHasher.lpi
+++ b/examples/ParallelFileHasher/ParallelFileHasher.lpi
@@ -21,7 +21,7 @@
-
+
@@ -57,7 +57,7 @@
-
+
@@ -98,7 +98,7 @@
-
+
diff --git a/examples/ParallelLogAnalyzer/ParallelLogAnalyzer.lpi b/examples/ParallelLogAnalyzer/ParallelLogAnalyzer.lpi
new file mode 100644
index 0000000..6d528ae
--- /dev/null
+++ b/examples/ParallelLogAnalyzer/ParallelLogAnalyzer.lpi
@@ -0,0 +1,64 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ -
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/examples/ParallelLogAnalyzer/ParallelLogAnalyzer.lpr b/examples/ParallelLogAnalyzer/ParallelLogAnalyzer.lpr
new file mode 100644
index 0000000..fe14479
--- /dev/null
+++ b/examples/ParallelLogAnalyzer/ParallelLogAnalyzer.lpr
@@ -0,0 +1,208 @@
+program ParallelLogAnalyzer;
+
+{$mode objfpc}{$H+}{$J-}
+
+uses
+ {$IFDEF UNIX}
+ cthreads,
+ {$ENDIF}
+ Classes, SysUtils, ThreadPool.Tasks, ThreadPool.Simple;
+
+const
+ ENTRY_COUNT = 100000;
+ ENDPOINT_COUNT = 4;
+ INVALID_ENTRY_INDEX = 12345;
+ ENDPOINT_NAMES: array[0..ENDPOINT_COUNT - 1] of string = (
+ '/catalog', '/checkout', '/account', '/health');
+
+type
+ TAccessLogEntry = record
+ EndpointIndex: Integer;
+ StatusCode: Integer;
+ LatencyMS: Integer;
+ BytesSent: Integer;
+ end;
+
+ TEntryAnalysis = record
+ IsValid: Boolean;
+ IsError: Boolean;
+ IsSlow: Boolean;
+ end;
+
+ TEndpointSummary = record
+ RequestCount: Integer;
+ ErrorCount: Integer;
+ SlowCount: Integer;
+ BytesSent: Int64;
+ end;
+
+ TAccessLogEntryArray = array of TAccessLogEntry;
+ TEntryAnalysisArray = array of TEntryAnalysis;
+ TEndpointSummaryArray = array of TEndpointSummary;
+
+ TLogAnalysisJob = class
+ private
+ FEntries: TAccessLogEntryArray;
+ FAnalysis: TEntryAnalysisArray;
+ FSummaries: TEndpointSummaryArray;
+ FInvalidCount: Integer;
+ public
+ constructor Create(AEntryCount: Integer);
+ procedure AnalyzeEntry(AIndex: Integer);
+ procedure SummarizeEndpoint(AEndpointIndex: Integer);
+ procedure CountInvalidEntries;
+ procedure PrintReport;
+ end;
+
+constructor TLogAnalysisJob.Create(AEntryCount: Integer);
+var
+ I: Integer;
+begin
+ inherited Create;
+ SetLength(FEntries, AEntryCount);
+ SetLength(FAnalysis, AEntryCount);
+ SetLength(FSummaries, ENDPOINT_COUNT);
+
+ { Generate deterministic parsed log records so the example is fast,
+ dependency-free, and produces the same report on every platform. }
+ for I := 0 to High(FEntries) do
+ begin
+ FEntries[I].EndpointIndex := I mod ENDPOINT_COUNT;
+ FEntries[I].LatencyMS := 20 + ((I * 37) mod 900);
+ FEntries[I].BytesSent := 300 + ((I * 53) mod 20000);
+
+ if (I mod 97) = 0 then
+ FEntries[I].StatusCode := 500
+ else if (I mod 31) = 0 then
+ FEntries[I].StatusCode := 404
+ else
+ FEntries[I].StatusCode := 200;
+ end;
+
+ { A malformed line in a real access log should be reported without losing
+ the other records that happen to share its range chunk. }
+ FEntries[INVALID_ENTRY_INDEX].StatusCode := 0;
+end;
+
+procedure TLogAnalysisJob.AnalyzeEntry(AIndex: Integer);
+var
+ Entry: TAccessLogEntry;
+begin
+ Entry := FEntries[AIndex];
+ FAnalysis[AIndex].IsValid :=
+ (Entry.EndpointIndex >= 0) and
+ (Entry.EndpointIndex < ENDPOINT_COUNT) and
+ (Entry.StatusCode >= 100) and (Entry.StatusCode <= 599) and
+ (Entry.LatencyMS >= 0) and (Entry.BytesSent >= 0);
+
+ if FAnalysis[AIndex].IsValid then
+ begin
+ FAnalysis[AIndex].IsError := Entry.StatusCode >= 400;
+ FAnalysis[AIndex].IsSlow := Entry.LatencyMS >= 500;
+ end;
+end;
+
+procedure TLogAnalysisJob.SummarizeEndpoint(AEndpointIndex: Integer);
+var
+ I: Integer;
+ Summary: TEndpointSummary;
+begin
+ Summary.RequestCount := 0;
+ Summary.ErrorCount := 0;
+ Summary.SlowCount := 0;
+ Summary.BytesSent := 0;
+ for I := 0 to High(FEntries) do
+ if FAnalysis[I].IsValid and
+ (FEntries[I].EndpointIndex = AEndpointIndex) then
+ begin
+ Inc(Summary.RequestCount);
+ if FAnalysis[I].IsError then
+ Inc(Summary.ErrorCount);
+ if FAnalysis[I].IsSlow then
+ Inc(Summary.SlowCount);
+ Inc(Summary.BytesSent, FEntries[I].BytesSent);
+ end;
+
+ { Each report task writes one distinct slot. All analysis slots are now
+ read-only because the range batch below is used as a phase barrier. }
+ FSummaries[AEndpointIndex] := Summary;
+end;
+
+procedure TLogAnalysisJob.CountInvalidEntries;
+var
+ I: Integer;
+begin
+ FInvalidCount := 0;
+ for I := 0 to High(FAnalysis) do
+ if not FAnalysis[I].IsValid then
+ Inc(FInvalidCount);
+end;
+
+procedure TLogAnalysisJob.PrintReport;
+var
+ I: Integer;
+begin
+ WriteLn;
+ WriteLn('Access-log report');
+ WriteLn('-----------------');
+ for I := 0 to ENDPOINT_COUNT - 1 do
+ WriteLn(Format('%-10s requests=%6d errors=%4d slow=%5d bytes=%d',
+ [ENDPOINT_NAMES[I], FSummaries[I].RequestCount,
+ FSummaries[I].ErrorCount, FSummaries[I].SlowCount,
+ FSummaries[I].BytesSent]));
+ WriteLn('Malformed records: ', FInvalidCount);
+end;
+
+var
+ Pool: TSimpleThreadPool;
+ Job: TLogAnalysisJob;
+ AnalysisBatch, ReportBatch: IThreadPoolTaskBatch;
+ ReportTasks: array[0..ENDPOINT_COUNT - 1] of IThreadPoolTask;
+ InvalidCountTask: IThreadPoolTask;
+ StartedAt: QWord;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ Job := TLogAnalysisJob.Create(ENTRY_COUNT);
+ try
+ { Phase 1: automatic chunking keeps queue traffic low while preserving a
+ task handle for each chunk. Range bounds are inclusive. }
+ StartedAt := GetTickCount64;
+ AnalysisBatch := Pool.SubmitRange(@Job.AnalyzeEntry, 0,
+ ENTRY_COUNT - 1);
+ WriteLn('Analyzing ', ENTRY_COUNT, ' records in ', AnalysisBatch.Count,
+ ' chunks...');
+ while not AnalysisBatch.WaitFor(10) do
+ WriteLn('Analysis progress: ', AnalysisBatch.FinishedCount, '/',
+ AnalysisBatch.Count, ' chunks');
+
+ if AnalysisBatch.FailedCount > 0 then
+ raise Exception.CreateFmt('Analysis failed in %d chunk(s)',
+ [AnalysisBatch.FailedCount]);
+ WriteLn('Analysis phase complete in ', GetTickCount64 - StartedAt, ' ms.');
+
+ { Phase 2 starts only after phase 1. Independent report sections and the
+ validation count can then scan the immutable analysis in parallel. }
+ ReportBatch := NewThreadPoolTaskBatch;
+ for I := 0 to ENDPOINT_COUNT - 1 do
+ begin
+ ReportTasks[I] := Pool.Submit(@Job.SummarizeEndpoint, I);
+ ReportBatch.Add(ReportTasks[I]);
+ end;
+ InvalidCountTask := Pool.Submit(@Job.CountInvalidEntries);
+ ReportBatch.Add(InvalidCountTask);
+
+ ReportBatch.WaitFor;
+ if ReportBatch.FailedCount > 0 then
+ raise Exception.CreateFmt('Report failed in %d section(s)',
+ [ReportBatch.FailedCount]);
+
+ WriteLn('Report sections coordinated: ', ReportBatch.FinishedCount, '/',
+ ReportBatch.Count);
+ Job.PrintReport;
+ finally
+ Pool.WaitForAll;
+ Pool.Free;
+ Job.Free;
+ end;
+end.
diff --git a/examples/ParallelLogAnalyzer/README.md b/examples/ParallelLogAnalyzer/README.md
new file mode 100644
index 0000000..6810bf5
--- /dev/null
+++ b/examples/ParallelLogAnalyzer/README.md
@@ -0,0 +1,20 @@
+# Parallel log analyzer
+
+This example models a two-phase access-log reporting pipeline over 100,000
+deterministic records:
+
+1. `SubmitRange` validates and classifies entries using a small number of
+ automatically sized chunks.
+2. The completed range batch acts as a phase barrier. Independent endpoint
+ summaries and malformed-record counting are then submitted as one batch.
+
+Callbacks write separate result slots, and the second phase reads only data
+made immutable by the barrier, so the workflow needs no application-level
+locks. One malformed record is retained as a validation result instead of
+raising from its callback and discarding the rest of that chunk.
+
+Build with Lazarus or run:
+
+```text
+lazbuild ParallelLogAnalyzer.lpi
+```
diff --git a/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi b/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi
index bbb0e89..90d8f40 100644
--- a/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi
+++ b/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi
@@ -21,7 +21,7 @@
-
+
@@ -57,7 +57,7 @@
-
+
@@ -103,7 +103,7 @@
-
+
diff --git a/examples/ProdConMessageProcessor/ProdConMessageProcessor.lpi b/examples/ProdConMessageProcessor/ProdConMessageProcessor.lpi
index c9ae83f..a539d5b 100644
--- a/examples/ProdConMessageProcessor/ProdConMessageProcessor.lpi
+++ b/examples/ProdConMessageProcessor/ProdConMessageProcessor.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/ProdConSimpleDemo/ProdConSimpleDemo.lpi b/examples/ProdConSimpleDemo/ProdConSimpleDemo.lpi
index a0146dd..e8eb2c5 100644
--- a/examples/ProdConSimpleDemo/ProdConSimpleDemo.lpi
+++ b/examples/ProdConSimpleDemo/ProdConSimpleDemo.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/ProdConSquareNumbers/ProdConSquareNumbers.lpi b/examples/ProdConSquareNumbers/ProdConSquareNumbers.lpi
index 708e892..8621614 100644
--- a/examples/ProdConSquareNumbers/ProdConSquareNumbers.lpi
+++ b/examples/ProdConSquareNumbers/ProdConSquareNumbers.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/SimpleDemo/SimpleDemo.lpi b/examples/SimpleDemo/SimpleDemo.lpi
index e208d83..42e1b8d 100644
--- a/examples/SimpleDemo/SimpleDemo.lpi
+++ b/examples/SimpleDemo/SimpleDemo.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/SimpleErrorHandling/SimpleErrorHandling.lpi b/examples/SimpleErrorHandling/SimpleErrorHandling.lpi
index ae4980b..05967ee 100644
--- a/examples/SimpleErrorHandling/SimpleErrorHandling.lpi
+++ b/examples/SimpleErrorHandling/SimpleErrorHandling.lpi
@@ -21,7 +21,7 @@
-
+
@@ -57,7 +57,7 @@
-
+
@@ -98,7 +98,7 @@
-
+
diff --git a/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi b/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi
index 93187f3..3a3b650 100644
--- a/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi
+++ b/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi
@@ -21,7 +21,7 @@
-
+
@@ -57,7 +57,7 @@
-
+
@@ -98,7 +98,7 @@
-
+
diff --git a/examples/SimpleSquareNumbers/SimpleSquareNumbers.lpi b/examples/SimpleSquareNumbers/SimpleSquareNumbers.lpi
index 25746f4..773583f 100644
--- a/examples/SimpleSquareNumbers/SimpleSquareNumbers.lpi
+++ b/examples/SimpleSquareNumbers/SimpleSquareNumbers.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/SimpleThreadpoolDemo/SimpleThreadpoolDemo.lpi b/examples/SimpleThreadpoolDemo/SimpleThreadpoolDemo.lpi
index 9372804..4258c0a 100644
--- a/examples/SimpleThreadpoolDemo/SimpleThreadpoolDemo.lpi
+++ b/examples/SimpleThreadpoolDemo/SimpleThreadpoolDemo.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/SimpleWordCounter/SimpleWordCounter.lpi b/examples/SimpleWordCounter/SimpleWordCounter.lpi
index 1c59692..bcc3e73 100644
--- a/examples/SimpleWordCounter/SimpleWordCounter.lpi
+++ b/examples/SimpleWordCounter/SimpleWordCounter.lpi
@@ -21,7 +21,7 @@
-
+
@@ -56,7 +56,7 @@
-
+
@@ -97,7 +97,7 @@
-
+
diff --git a/examples/Starter/Starter.lpi b/examples/Starter/Starter.lpi
index dc1425c..87f7044 100644
--- a/examples/Starter/Starter.lpi
+++ b/examples/Starter/Starter.lpi
@@ -22,7 +22,7 @@
-
+
@@ -60,7 +60,7 @@
-
+
@@ -104,7 +104,7 @@
-
+
diff --git a/examples/TaskCoordination/TaskCoordination.lpi b/examples/TaskCoordination/TaskCoordination.lpi
new file mode 100644
index 0000000..4c62ca7
--- /dev/null
+++ b/examples/TaskCoordination/TaskCoordination.lpi
@@ -0,0 +1,64 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ -
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/examples/TaskCoordination/TaskCoordination.lpr b/examples/TaskCoordination/TaskCoordination.lpr
new file mode 100644
index 0000000..aa3ec2b
--- /dev/null
+++ b/examples/TaskCoordination/TaskCoordination.lpr
@@ -0,0 +1,86 @@
+program TaskCoordination;
+
+{$mode objfpc}{$H+}{$J-}
+
+uses
+ {$IFDEF UNIX}
+ cthreads,
+ {$ENDIF}
+ Classes, SysUtils, ThreadPool.Tasks, ThreadPool.Simple;
+
+const
+ ITEM_COUNT = 20;
+
+var
+ Results: array[0..ITEM_COUNT - 1] of Integer;
+
+procedure DoOneTask;
+begin
+ Sleep(10);
+end;
+
+procedure FailTask;
+begin
+ raise Exception.Create('demonstration failure');
+end;
+
+procedure ProcessItem(AIndex: Integer);
+begin
+ Results[AIndex] := AIndex * AIndex;
+end;
+
+function TaskStateName(AState: TThreadPoolTaskState): string;
+begin
+ Result := 'unknown';
+ case AState of
+ ttsPending: Result := 'pending';
+ ttsRunning: Result := 'running';
+ ttsCompleted: Result := 'completed';
+ ttsFailed: Result := 'failed';
+ ttsCancelled: Result := 'cancelled';
+ end;
+end;
+
+var
+ Task: IThreadPoolTask;
+ Batch, RangeTasks: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ { Submit returns a handle for observing and waiting for one task. }
+ Task := GlobalThreadPool.Submit(@DoOneTask);
+ if Task.WaitFor(1000) then
+ WriteLn('Individual task: ', TaskStateName(Task.State));
+
+ Task := GlobalThreadPool.Submit(@FailTask);
+ Task.WaitFor;
+ if Task.State = ttsFailed then
+ WriteLn('Failed task: ', Task.ErrorMessage);
+
+ { A batch coordinates any collection of task handles. }
+ Batch := NewThreadPoolTaskBatch;
+ for I := 1 to 3 do
+ Batch.Add(GlobalThreadPool.Submit(@DoOneTask));
+ Batch.WaitFor;
+ WriteLn('Batch finished: ', Batch.FinishedCount, '/', Batch.Count);
+
+ { SubmitRange queues chunks rather than one task per index. Bounds are
+ inclusive, and automatic chunking is selected by the default zero size. }
+ for I := 0 to High(Results) do
+ Results[I] := 0;
+ RangeTasks := GlobalThreadPool.SubmitRange(
+ @ProcessItem, 0, High(Results));
+ RangeTasks.WaitFor;
+ WriteLn('Range chunks: ', RangeTasks.Count);
+ WriteLn('Last square: ', Results[High(Results)]);
+
+ { Cancel succeeds only while a task is still pending. It never terminates a
+ callback that has started running. }
+ Task := GlobalThreadPool.Submit(@DoOneTask);
+ if Task.Cancel then
+ WriteLn('Optional task cancelled before it started')
+ else
+ begin
+ Task.WaitFor;
+ WriteLn('Optional task had already started: ', TaskStateName(Task.State));
+ end;
+end.
diff --git a/package/lazarus/threadpool_fp.lpk b/package/lazarus/threadpool_fp.lpk
index 1117a3e..d22d26c 100644
--- a/package/lazarus/threadpool_fp.lpk
+++ b/package/lazarus/threadpool_fp.lpk
@@ -23,8 +23,9 @@ A lightweight, dependency-free thread pool library with two event-driven impleme
- ThreadPool.Simple: an unbounded FIFO with a managed global instance
- ThreadPool.ProducerConsumer: a bounded FIFO with timeout-aware backpressure
+- ThreadPool.Tasks: observable tasks, batches, ranges, and pending cancellation
-Both pools provide draining shutdown, timeout-aware waits, four task forms, and captured worker errors. See README.md and docs/CHEATSHEET.md for usage."/>
+Both pools provide draining shutdown, timeout-aware waits, four task forms, observable submission, and captured worker errors. See README.md and docs/CHEATSHEET.md for usage."/>
-
+
-
@@ -56,6 +57,10 @@ SOFTWARE. "/>
+ -
+
+
+
-
diff --git a/package/lazarus/threadpool_fp.pas b/package/lazarus/threadpool_fp.pas
index 47a4d2c..20c9f28 100644
--- a/package/lazarus/threadpool_fp.pas
+++ b/package/lazarus/threadpool_fp.pas
@@ -8,7 +8,8 @@
interface
uses
- ThreadPool.Simple, ThreadPool.Types, ThreadPool.ProducerConsumer;
+ ThreadPool.Simple, ThreadPool.Types, ThreadPool.Tasks,
+ ThreadPool.ProducerConsumer;
implementation
diff --git a/src/ThreadPool.ProducerConsumer.pas b/src/ThreadPool.ProducerConsumer.pas
index 8cd157d..9fd1d81 100644
--- a/src/ThreadPool.ProducerConsumer.pas
+++ b/src/ThreadPool.ProducerConsumer.pas
@@ -5,7 +5,7 @@
interface
uses
- Classes, SysUtils, Math, ThreadPool.Types, SyncObjs;
+ Classes, SysUtils, Math, ThreadPool.Types, ThreadPool.Tasks, SyncObjs;
var
DEBUG_LOG: Boolean = False; // Opt-in only; disabled by default
@@ -97,7 +97,7 @@ TProducerConsumerWorkItem = class(TInterfacedObject, IWorkItem)
{$REGION 'Public API: TProducerConsumerThreadPool'}
{ Producer-consumer thread pool implementation }
- TProducerConsumerThreadPool = class(TThreadPoolBase)
+ TProducerConsumerThreadPool = class(TThreadPoolBase, IThreadPoolTaskSource)
private
FThreads: TThreadList;
FWorkQueue: TThreadSafeQueue; // Use our custom thread-safe queue
@@ -109,6 +109,7 @@ TProducerConsumerThreadPool = class(TThreadPoolBase)
procedure ClearThreads;
function TryQueueWorkItem(WorkItem: IWorkItem;
ATimeoutMS: Cardinal): Boolean;
+ function SubmitRangeWorkItem(const AWorkItem: IWorkItem): Boolean;
procedure CompleteWorkItem;
function IsCurrentWorkerThread: Boolean;
public
@@ -131,6 +132,26 @@ TProducerConsumerThreadPool = class(TThreadPoolBase)
ATimeoutMS: Cardinal): Boolean; overload; override;
function TryQueue(AMethod: TThreadMethodIndex; AIndex: Integer;
ATimeoutMS: Cardinal): Boolean; overload; override;
+ function Submit(AProcedure: TThreadProcedure): IThreadPoolTask; overload;
+ function Submit(AMethod: TThreadMethod): IThreadPoolTask; overload;
+ function Submit(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer): IThreadPoolTask; overload;
+ function Submit(AMethod: TThreadMethodIndex;
+ AIndex: Integer): IThreadPoolTask; overload;
+ function TrySubmit(AProcedure: TThreadProcedure; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AMethod: TThreadMethod; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AProcedure: TThreadProcedureIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AMethod: TThreadMethodIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean; overload;
+ function SubmitRange(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
+ function SubmitRange(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
procedure WaitForAll; overload; override;
function WaitForAll(ATimeoutMS: Cardinal): Boolean; overload; override;
procedure Shutdown; override;
@@ -252,6 +273,171 @@ procedure TProducerConsumerThreadPool.CompleteWorkItem;
end;
end;
+function TProducerConsumerThreadPool.SubmitRangeWorkItem(
+ const AWorkItem: IWorkItem): Boolean;
+begin
+ Result := TryQueueWorkItem(AWorkItem, THREADPOOL_INFINITE);
+end;
+
+function TProducerConsumerThreadPool.Submit(
+ AProcedure: TThreadProcedure): IThreadPoolTask;
+begin
+ if not TrySubmit(AProcedure, FWorkQueue.GetDefaultTimeout, Result) then
+ raise EQueueFullException.Create('Queue is full (submission timed out)');
+end;
+
+function TProducerConsumerThreadPool.Submit(
+ AMethod: TThreadMethod): IThreadPoolTask;
+begin
+ if not TrySubmit(AMethod, FWorkQueue.GetDefaultTimeout, Result) then
+ raise EQueueFullException.Create('Queue is full (submission timed out)');
+end;
+
+function TProducerConsumerThreadPool.Submit(
+ AProcedure: TThreadProcedureIndex; AIndex: Integer): IThreadPoolTask;
+begin
+ if not TrySubmit(AProcedure, AIndex, FWorkQueue.GetDefaultTimeout,
+ Result) then
+ raise EQueueFullException.Create('Queue is full (submission timed out)');
+end;
+
+function TProducerConsumerThreadPool.Submit(
+ AMethod: TThreadMethodIndex; AIndex: Integer): IThreadPoolTask;
+begin
+ if not TrySubmit(AMethod, AIndex, FWorkQueue.GetDefaultTimeout,
+ Result) then
+ raise EQueueFullException.Create('Queue is full (submission timed out)');
+end;
+
+function TProducerConsumerThreadPool.TrySubmit(
+ AProcedure: TThreadProcedure; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AProcedure, ATask);
+ try
+ BeginQueue;
+ try
+ Result := TryQueueWorkItem(WorkItem, ATimeoutMS);
+ finally
+ EndQueue;
+ end;
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TProducerConsumerThreadPool.TrySubmit(AMethod: TThreadMethod;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AMethod, ATask);
+ try
+ BeginQueue;
+ try
+ Result := TryQueueWorkItem(WorkItem, ATimeoutMS);
+ finally
+ EndQueue;
+ end;
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TProducerConsumerThreadPool.TrySubmit(
+ AProcedure: TThreadProcedureIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AProcedure, AIndex, ATask);
+ try
+ BeginQueue;
+ try
+ Result := TryQueueWorkItem(WorkItem, ATimeoutMS);
+ finally
+ EndQueue;
+ end;
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TProducerConsumerThreadPool.TrySubmit(
+ AMethod: TThreadMethodIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AMethod, AIndex, ATask);
+ try
+ BeginQueue;
+ try
+ Result := TryQueueWorkItem(WorkItem, ATimeoutMS);
+ finally
+ EndQueue;
+ end;
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TProducerConsumerThreadPool.SubmitRange(
+ AProcedure: TThreadProcedureIndex; AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer): IThreadPoolTaskBatch;
+begin
+ if GetThreadPoolRangeChunkSize(AFirstIndex, ALastIndex,
+ FThreadCount, AChunkSize) = 0 then
+ Exit(NewThreadPoolTaskBatch);
+ if IsCurrentWorkerThread then
+ raise EThreadPoolDeadlock.Create(
+ 'SubmitRange cannot be called from a worker of the same bounded pool');
+ BeginQueue;
+ try
+ Result := NewThreadPoolRangeBatch(AProcedure, AFirstIndex,
+ ALastIndex, FThreadCount, AChunkSize, @SubmitRangeWorkItem);
+ finally
+ EndQueue;
+ end;
+end;
+
+function TProducerConsumerThreadPool.SubmitRange(
+ AMethod: TThreadMethodIndex; AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer): IThreadPoolTaskBatch;
+begin
+ if GetThreadPoolRangeChunkSize(AFirstIndex, ALastIndex,
+ FThreadCount, AChunkSize) = 0 then
+ Exit(NewThreadPoolTaskBatch);
+ if IsCurrentWorkerThread then
+ raise EThreadPoolDeadlock.Create(
+ 'SubmitRange cannot be called from a worker of the same bounded pool');
+ BeginQueue;
+ try
+ Result := NewThreadPoolRangeBatch(AMethod, AFirstIndex,
+ ALastIndex, FThreadCount, AChunkSize, @SubmitRangeWorkItem);
+ finally
+ EndQueue;
+ end;
+end;
+
function TProducerConsumerThreadPool.TryQueue(AProcedure: TThreadProcedure;
ATimeoutMS: Cardinal): Boolean;
var
@@ -628,12 +814,7 @@ procedure TProducerConsumerWorkerThread.Execute;
while (not Terminated) and Pool.FWorkQueue.TryDequeue(WorkItem) do
begin
try
- try
- WorkItem.Execute;
- except
- on E: Exception do
- Pool.SetLastError(E.Message);
- end;
+ ExecuteThreadPoolWorkItem(WorkItem, @Pool.SetLastError);
finally
WorkItem := nil;
{ This must execute even when the task or OnError callback fails. }
diff --git a/src/ThreadPool.Simple.pas b/src/ThreadPool.Simple.pas
index 8722c74..8f1dba8 100644
--- a/src/ThreadPool.Simple.pas
+++ b/src/ThreadPool.Simple.pas
@@ -5,7 +5,7 @@
interface
uses
- Classes, SysUtils, SyncObjs, ThreadPool.Types;
+ Classes, SysUtils, SyncObjs, ThreadPool.Types, ThreadPool.Tasks;
type
{$REGION 'Internal: Work Item'}
@@ -53,10 +53,10 @@ TSimpleWorkerThread = class(TThread, IWorkerThread)
{$REGION 'Public API: TSimpleThreadPool'}
{ Simple thread pool implementation }
- TSimpleThreadPool = class(TThreadPoolBase)
+ TSimpleThreadPool = class(TThreadPoolBase, IThreadPoolTaskSource)
private
FThreads: TThreadList;
- FWorkItems: array of TSimpleWorkItem;
+ FWorkItems: array of IWorkItem;
FQueueHead: Integer;
FQueueTail: Integer;
FQueueCount: Integer;
@@ -67,8 +67,10 @@ TSimpleThreadPool = class(TThreadPoolBase)
FWorkAvailableEvent: TEvent;
procedure ClearThreads;
procedure ClearWorkItems;
- procedure EnqueueWorkItem(AWorkItem: TSimpleWorkItem);
- function TryDequeueWorkItem(out AWorkItem: TSimpleWorkItem): Boolean;
+ procedure EnqueueWorkItem(const AWorkItem: IWorkItem);
+ function TryDequeueWorkItem(out AWorkItem: IWorkItem): Boolean;
+ function TrySubmitWorkItem(const AWorkItem: IWorkItem): Boolean;
+ function SubmitRangeWorkItem(const AWorkItem: IWorkItem): Boolean;
procedure CompleteWorkItem;
function IsCurrentWorkerThread: Boolean;
public
@@ -88,6 +90,26 @@ TSimpleThreadPool = class(TThreadPoolBase)
ATimeoutMS: Cardinal): Boolean; overload; override;
function TryQueue(AMethod: TThreadMethodIndex; AIndex: Integer;
ATimeoutMS: Cardinal): Boolean; overload; override;
+ function Submit(AProcedure: TThreadProcedure): IThreadPoolTask; overload;
+ function Submit(AMethod: TThreadMethod): IThreadPoolTask; overload;
+ function Submit(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer): IThreadPoolTask; overload;
+ function Submit(AMethod: TThreadMethodIndex;
+ AIndex: Integer): IThreadPoolTask; overload;
+ function TrySubmit(AProcedure: TThreadProcedure; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AMethod: TThreadMethod; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AProcedure: TThreadProcedureIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AMethod: TThreadMethodIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean; overload;
+ function SubmitRange(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
+ function SubmitRange(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
procedure WaitForAll; overload; override;
function WaitForAll(ATimeoutMS: Cardinal): Boolean; overload; override;
procedure Shutdown; override;
@@ -171,7 +193,7 @@ function TSimpleWorkerThread._Release: Integer; {$IFDEF WINDOWS}stdcall{$ELSE}cd
procedure TSimpleWorkerThread.Execute;
var
Pool: TSimpleThreadPool;
- WorkItem: TSimpleWorkItem;
+ WorkItem: IWorkItem;
begin
Pool := TSimpleThreadPool(FThreadPool);
@@ -185,14 +207,9 @@ procedure TSimpleWorkerThread.Execute;
while (not Terminated) and Pool.TryDequeueWorkItem(WorkItem) do
begin
try
- try
- WorkItem.Execute;
- except
- on E: Exception do
- Pool.SetLastError(E.Message);
- end;
+ ExecuteThreadPoolWorkItem(WorkItem, @Pool.SetLastError);
finally
- WorkItem.Free;
+ WorkItem := nil;
{ Completion accounting is independent of task and callback failures. }
Pool.CompleteWorkItem;
end;
@@ -278,9 +295,9 @@ destructor TSimpleThreadPool.Destroy;
inherited Destroy;
end;
-procedure TSimpleThreadPool.EnqueueWorkItem(AWorkItem: TSimpleWorkItem);
+procedure TSimpleThreadPool.EnqueueWorkItem(const AWorkItem: IWorkItem);
var
- NewItems: array of TSimpleWorkItem;
+ NewItems: array of IWorkItem;
I, NewCapacity: Integer;
begin
NewItems := nil;
@@ -317,7 +334,7 @@ procedure TSimpleThreadPool.EnqueueWorkItem(AWorkItem: TSimpleWorkItem);
end;
function TSimpleThreadPool.TryDequeueWorkItem(
- out AWorkItem: TSimpleWorkItem): Boolean;
+ out AWorkItem: IWorkItem): Boolean;
begin
FQueueLock.Enter;
try
@@ -416,32 +433,178 @@ procedure TSimpleThreadPool.ClearThreads;
procedure TSimpleThreadPool.ClearWorkItems;
var
- WorkItem: TSimpleWorkItem;
+ WorkItem: IWorkItem;
begin
if not Assigned(FQueueLock) then
Exit;
while TryDequeueWorkItem(WorkItem) do
- WorkItem.Free;
+ WorkItem := nil;
SetLength(FWorkItems, 0);
end;
+function TSimpleThreadPool.TrySubmitWorkItem(
+ const AWorkItem: IWorkItem): Boolean;
+begin
+ BeginQueue;
+ try
+ EnqueueWorkItem(AWorkItem);
+ Result := True;
+ finally
+ EndQueue;
+ end;
+end;
+
+function TSimpleThreadPool.SubmitRangeWorkItem(
+ const AWorkItem: IWorkItem): Boolean;
+begin
+ EnqueueWorkItem(AWorkItem);
+ Result := True;
+end;
+
+function TSimpleThreadPool.Submit(
+ AProcedure: TThreadProcedure): IThreadPoolTask;
+begin
+ if not TrySubmit(AProcedure, 0, Result) then
+ Result := nil;
+end;
+
+function TSimpleThreadPool.Submit(
+ AMethod: TThreadMethod): IThreadPoolTask;
+begin
+ if not TrySubmit(AMethod, 0, Result) then
+ Result := nil;
+end;
+
+function TSimpleThreadPool.Submit(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer): IThreadPoolTask;
+begin
+ if not TrySubmit(AProcedure, AIndex, 0, Result) then
+ Result := nil;
+end;
+
+function TSimpleThreadPool.Submit(AMethod: TThreadMethodIndex;
+ AIndex: Integer): IThreadPoolTask;
+begin
+ if not TrySubmit(AMethod, AIndex, 0, Result) then
+ Result := nil;
+end;
+
+function TSimpleThreadPool.TrySubmit(AProcedure: TThreadProcedure;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AProcedure, ATask);
+ try
+ Result := TrySubmitWorkItem(WorkItem);
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TSimpleThreadPool.TrySubmit(AMethod: TThreadMethod;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AMethod, ATask);
+ try
+ Result := TrySubmitWorkItem(WorkItem);
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TSimpleThreadPool.TrySubmit(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AProcedure, AIndex, ATask);
+ try
+ Result := TrySubmitWorkItem(WorkItem);
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TSimpleThreadPool.TrySubmit(AMethod: TThreadMethodIndex;
+ AIndex: Integer; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean;
+var
+ WorkItem: IWorkItem;
+begin
+ ATask := nil;
+ WorkItem := NewTrackedWorkItem(AMethod, AIndex, ATask);
+ try
+ Result := TrySubmitWorkItem(WorkItem);
+ except
+ ATask := nil;
+ raise;
+ end;
+ if not Result then
+ ATask := nil;
+end;
+
+function TSimpleThreadPool.SubmitRange(
+ AProcedure: TThreadProcedureIndex; AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer): IThreadPoolTaskBatch;
+begin
+ if GetThreadPoolRangeChunkSize(AFirstIndex, ALastIndex,
+ FThreadCount, AChunkSize) = 0 then
+ Exit(NewThreadPoolTaskBatch);
+ BeginQueue;
+ try
+ Result := NewThreadPoolRangeBatch(AProcedure, AFirstIndex,
+ ALastIndex, FThreadCount, AChunkSize, @SubmitRangeWorkItem);
+ finally
+ EndQueue;
+ end;
+end;
+
+function TSimpleThreadPool.SubmitRange(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer): IThreadPoolTaskBatch;
+begin
+ if GetThreadPoolRangeChunkSize(AFirstIndex, ALastIndex,
+ FThreadCount, AChunkSize) = 0 then
+ Exit(NewThreadPoolTaskBatch);
+ BeginQueue;
+ try
+ Result := NewThreadPoolRangeBatch(AMethod, AFirstIndex,
+ ALastIndex, FThreadCount, AChunkSize, @SubmitRangeWorkItem);
+ finally
+ EndQueue;
+ end;
+end;
+
function TSimpleThreadPool.TryQueue(AProcedure: TThreadProcedure;
ATimeoutMS: Cardinal): Boolean;
var
- WorkItem: TSimpleWorkItem;
+ WorkItemObject: TSimpleWorkItem;
+ WorkItem: IWorkItem;
begin
BeginQueue;
try
- WorkItem := TSimpleWorkItem.Create(Self);
- try
- WorkItem.FProcedure := AProcedure;
- WorkItem.FItemType := witProcedure;
- EnqueueWorkItem(WorkItem);
- WorkItem := nil;
- Result := True;
- finally
- WorkItem.Free;
- end;
+ WorkItemObject := TSimpleWorkItem.Create(Self);
+ WorkItemObject.FProcedure := AProcedure;
+ WorkItemObject.FItemType := witProcedure;
+ WorkItem := WorkItemObject;
+ EnqueueWorkItem(WorkItem);
+ Result := True;
finally
EndQueue;
end;
@@ -450,20 +613,17 @@ function TSimpleThreadPool.TryQueue(AProcedure: TThreadProcedure;
function TSimpleThreadPool.TryQueue(AMethod: TThreadMethod;
ATimeoutMS: Cardinal): Boolean;
var
- WorkItem: TSimpleWorkItem;
+ WorkItemObject: TSimpleWorkItem;
+ WorkItem: IWorkItem;
begin
BeginQueue;
try
- WorkItem := TSimpleWorkItem.Create(Self);
- try
- WorkItem.FMethod := AMethod;
- WorkItem.FItemType := witMethod;
- EnqueueWorkItem(WorkItem);
- WorkItem := nil;
- Result := True;
- finally
- WorkItem.Free;
- end;
+ WorkItemObject := TSimpleWorkItem.Create(Self);
+ WorkItemObject.FMethod := AMethod;
+ WorkItemObject.FItemType := witMethod;
+ WorkItem := WorkItemObject;
+ EnqueueWorkItem(WorkItem);
+ Result := True;
finally
EndQueue;
end;
@@ -472,21 +632,18 @@ function TSimpleThreadPool.TryQueue(AMethod: TThreadMethod;
function TSimpleThreadPool.TryQueue(AProcedure: TThreadProcedureIndex;
AIndex: Integer; ATimeoutMS: Cardinal): Boolean;
var
- WorkItem: TSimpleWorkItem;
+ WorkItemObject: TSimpleWorkItem;
+ WorkItem: IWorkItem;
begin
BeginQueue;
try
- WorkItem := TSimpleWorkItem.Create(Self);
- try
- WorkItem.FProcedureIndex := AProcedure;
- WorkItem.FIndex := AIndex;
- WorkItem.FItemType := witProcedureIndex;
- EnqueueWorkItem(WorkItem);
- WorkItem := nil;
- Result := True;
- finally
- WorkItem.Free;
- end;
+ WorkItemObject := TSimpleWorkItem.Create(Self);
+ WorkItemObject.FProcedureIndex := AProcedure;
+ WorkItemObject.FIndex := AIndex;
+ WorkItemObject.FItemType := witProcedureIndex;
+ WorkItem := WorkItemObject;
+ EnqueueWorkItem(WorkItem);
+ Result := True;
finally
EndQueue;
end;
@@ -495,21 +652,18 @@ function TSimpleThreadPool.TryQueue(AProcedure: TThreadProcedureIndex;
function TSimpleThreadPool.TryQueue(AMethod: TThreadMethodIndex;
AIndex: Integer; ATimeoutMS: Cardinal): Boolean;
var
- WorkItem: TSimpleWorkItem;
+ WorkItemObject: TSimpleWorkItem;
+ WorkItem: IWorkItem;
begin
BeginQueue;
try
- WorkItem := TSimpleWorkItem.Create(Self);
- try
- WorkItem.FMethodIndex := AMethod;
- WorkItem.FIndex := AIndex;
- WorkItem.FItemType := witMethodIndex;
- EnqueueWorkItem(WorkItem);
- WorkItem := nil;
- Result := True;
- finally
- WorkItem.Free;
- end;
+ WorkItemObject := TSimpleWorkItem.Create(Self);
+ WorkItemObject.FMethodIndex := AMethod;
+ WorkItemObject.FIndex := AIndex;
+ WorkItemObject.FItemType := witMethodIndex;
+ WorkItem := WorkItemObject;
+ EnqueueWorkItem(WorkItem);
+ Result := True;
finally
EndQueue;
end;
diff --git a/src/ThreadPool.Tasks.pas b/src/ThreadPool.Tasks.pas
new file mode 100644
index 0000000..a13fad4
--- /dev/null
+++ b/src/ThreadPool.Tasks.pas
@@ -0,0 +1,838 @@
+unit ThreadPool.Tasks;
+
+{$mode objfpc}{$H+}{$J-}
+
+interface
+
+uses
+ Classes, SysUtils, SyncObjs, ThreadPool.Types;
+
+type
+ EThreadPoolDeadlock = class(Exception);
+ EThreadPoolTaskSubmission = class(Exception);
+
+ { Observable lifecycle of one submitted task. }
+ TThreadPoolTaskState = (
+ ttsPending,
+ ttsRunning,
+ ttsCompleted,
+ ttsFailed,
+ ttsCancelled
+ );
+
+ IThreadPoolTask = interface
+ ['{D5E86E1A-9585-4CC3-90AD-76C1B34CD9F3}']
+ procedure WaitFor; overload;
+ function WaitFor(ATimeoutMS: Cardinal): Boolean; overload;
+ function Cancel: Boolean;
+ function GetState: TThreadPoolTaskState;
+ function GetErrorMessage: string;
+ function GetIsFinished: Boolean;
+ property State: TThreadPoolTaskState read GetState;
+ property ErrorMessage: string read GetErrorMessage;
+ property IsFinished: Boolean read GetIsFinished;
+ end;
+
+ TThreadPoolTaskArray = array of IThreadPoolTask;
+
+ { Internal admission callback used by the shared range builder. }
+ TThreadPoolWorkItemSubmit = function(
+ const AWorkItem: IWorkItem): Boolean of object;
+
+ { A thread-safe collection of task handles. WaitFor operates on a snapshot
+ taken when the call begins, so producers may continue adding tasks without
+ changing an in-progress wait. }
+ IThreadPoolTaskBatch = interface
+ ['{B593B216-E9A4-4242-9A6B-F3CB6B4AFBE8}']
+ procedure Add(const ATask: IThreadPoolTask);
+ procedure WaitFor; overload;
+ function WaitFor(ATimeoutMS: Cardinal): Boolean; overload;
+ function CancelPending: Integer;
+ function GetCount: Integer;
+ function GetFinishedCount: Integer;
+ function GetFailedCount: Integer;
+ function GetCancelledCount: Integer;
+ function GetTask(AIndex: Integer): IThreadPoolTask;
+ property Count: Integer read GetCount;
+ property FinishedCount: Integer read GetFinishedCount;
+ property FailedCount: Integer read GetFailedCount;
+ property CancelledCount: Integer read GetCancelledCount;
+ property Tasks[AIndex: Integer]: IThreadPoolTask read GetTask; default;
+ end;
+
+ { Optional v0.9 capability interface. IThreadPool remains unchanged so
+ existing third-party implementations retain source compatibility. }
+ IThreadPoolTaskSource = interface
+ ['{2898DB24-5D5D-4CF3-9A1B-BDA89D4127A8}']
+ function Submit(AProcedure: TThreadProcedure): IThreadPoolTask; overload;
+ function Submit(AMethod: TThreadMethod): IThreadPoolTask; overload;
+ function Submit(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer): IThreadPoolTask; overload;
+ function Submit(AMethod: TThreadMethodIndex;
+ AIndex: Integer): IThreadPoolTask; overload;
+ function TrySubmit(AProcedure: TThreadProcedure; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AMethod: TThreadMethod; ATimeoutMS: Cardinal;
+ out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AProcedure: TThreadProcedureIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean; overload;
+ function TrySubmit(AMethod: TThreadMethodIndex; AIndex: Integer;
+ ATimeoutMS: Cardinal; out ATask: IThreadPoolTask): Boolean; overload;
+ function SubmitRange(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
+ function SubmitRange(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex: Integer;
+ AChunkSize: Integer = 0): IThreadPoolTaskBatch; overload;
+ end;
+
+ { Internal worker bridge. It is separate from IWorkItem so the v0.8
+ interface and its GUID do not change. }
+ IThreadPoolTrackedWorkItem = interface
+ ['{9B013788-41B1-4C03-95E1-5DA8A9E80904}']
+ function TryStart: Boolean;
+ procedure MarkCompleted;
+ procedure MarkFailed(const AMessage: string);
+ end;
+
+function NewThreadPoolTaskBatch: IThreadPoolTaskBatch;
+
+{ Internal factories shared by the two pool implementations. }
+function NewTrackedWorkItem(AProcedure: TThreadProcedure;
+ out ATask: IThreadPoolTask): IWorkItem; overload;
+function NewTrackedWorkItem(AMethod: TThreadMethod;
+ out ATask: IThreadPoolTask): IWorkItem; overload;
+function NewTrackedWorkItem(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer; out ATask: IThreadPoolTask): IWorkItem; overload;
+function NewTrackedWorkItem(AMethod: TThreadMethodIndex;
+ AIndex: Integer; out ATask: IThreadPoolTask): IWorkItem; overload;
+function NewTrackedRangeWorkItem(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex: Integer;
+ out ATask: IThreadPoolTask): IWorkItem; overload;
+function NewTrackedRangeWorkItem(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex: Integer;
+ out ATask: IThreadPoolTask): IWorkItem; overload;
+function GetThreadPoolRangeChunkSize(AFirstIndex, ALastIndex,
+ AThreadCount, ARequestedChunkSize: Integer): Int64;
+procedure ExecuteThreadPoolWorkItem(const AWorkItem: IWorkItem;
+ AOnError: TThreadPoolErrorEvent);
+function NewThreadPoolRangeBatch(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex, AThreadCount, AChunkSize: Integer;
+ ASubmit: TThreadPoolWorkItemSubmit): IThreadPoolTaskBatch; overload;
+function NewThreadPoolRangeBatch(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex, AThreadCount, AChunkSize: Integer;
+ ASubmit: TThreadPoolWorkItemSubmit): IThreadPoolTaskBatch; overload;
+
+implementation
+
+type
+ TThreadPoolTaskControl = class(TInterfacedObject, IThreadPoolTask)
+ private
+ FLock: TCriticalSection;
+ FCompletionEvent: TEvent;
+ FState: TThreadPoolTaskState;
+ FErrorMessage: string;
+ FRunningThreadID: TThreadID;
+ procedure SignalCompletion;
+ class function IsTerminal(AState: TThreadPoolTaskState): Boolean; static;
+ public
+ constructor Create;
+ destructor Destroy; override;
+ procedure WaitFor; overload;
+ function WaitFor(ATimeoutMS: Cardinal): Boolean; overload;
+ function Cancel: Boolean;
+ function GetState: TThreadPoolTaskState;
+ function GetErrorMessage: string;
+ function GetIsFinished: Boolean;
+ function TryStart: Boolean;
+ procedure MarkCompleted;
+ procedure MarkFailed(const AMessage: string);
+ end;
+
+ TTrackedWorkItem = class(TInterfacedObject, IWorkItem,
+ IThreadPoolTrackedWorkItem)
+ private
+ FProcedure: TThreadProcedure;
+ FMethod: TThreadMethod;
+ FProcedureIndex: TThreadProcedureIndex;
+ FMethodIndex: TThreadMethodIndex;
+ FIndex: Integer;
+ FFirstIndex: Integer;
+ FLastIndex: Integer;
+ FItemType: TWorkItemType;
+ FIsRange: Boolean;
+ FTask: IThreadPoolTask;
+ FControl: TThreadPoolTaskControl;
+ public
+ constructor Create(AControl: TThreadPoolTaskControl;
+ const ATask: IThreadPoolTask);
+ procedure Execute;
+ function GetItemType: Integer;
+ function TryStart: Boolean;
+ procedure MarkCompleted;
+ procedure MarkFailed(const AMessage: string);
+ end;
+
+ TThreadPoolTaskBatch = class(TInterfacedObject, IThreadPoolTaskBatch)
+ private
+ FLock: TCriticalSection;
+ FTasks: TThreadPoolTaskArray;
+ FCount: Integer;
+ function Snapshot: TThreadPoolTaskArray;
+ public
+ constructor Create;
+ destructor Destroy; override;
+ procedure Add(const ATask: IThreadPoolTask);
+ procedure WaitFor; overload;
+ function WaitFor(ATimeoutMS: Cardinal): Boolean; overload;
+ function CancelPending: Integer;
+ function GetCount: Integer;
+ function GetFinishedCount: Integer;
+ function GetFailedCount: Integer;
+ function GetCancelledCount: Integer;
+ function GetTask(AIndex: Integer): IThreadPoolTask;
+ end;
+
+{ TThreadPoolTaskControl }
+
+constructor TThreadPoolTaskControl.Create;
+begin
+ inherited Create;
+ FLock := TCriticalSection.Create;
+ FCompletionEvent := nil;
+ FState := ttsPending;
+ FErrorMessage := '';
+ FRunningThreadID := 0;
+end;
+
+destructor TThreadPoolTaskControl.Destroy;
+begin
+ FCompletionEvent.Free;
+ FLock.Free;
+ inherited;
+end;
+
+class function TThreadPoolTaskControl.IsTerminal(
+ AState: TThreadPoolTaskState): Boolean;
+begin
+ Result := AState in [ttsCompleted, ttsFailed, ttsCancelled];
+end;
+
+procedure TThreadPoolTaskControl.SignalCompletion;
+begin
+ { FLock must be held so a waiter cannot miss the transition while lazily
+ creating the event. }
+ if Assigned(FCompletionEvent) then
+ FCompletionEvent.SetEvent;
+end;
+
+procedure TThreadPoolTaskControl.WaitFor;
+begin
+ WaitFor(THREADPOOL_INFINITE);
+end;
+
+function TThreadPoolTaskControl.WaitFor(ATimeoutMS: Cardinal): Boolean;
+var
+ CompletionEvent: TEvent;
+begin
+ CompletionEvent := nil;
+ FLock.Enter;
+ try
+ if IsTerminal(FState) then
+ Exit(True);
+ if (FState = ttsRunning) and
+ (FRunningThreadID = GetCurrentThreadID) then
+ raise EThreadPoolDeadlock.Create(
+ 'A task cannot wait for its own completion');
+ if not Assigned(FCompletionEvent) then
+ FCompletionEvent := TEvent.Create(nil, True, False, '');
+ CompletionEvent := FCompletionEvent;
+ finally
+ FLock.Leave;
+ end;
+
+ Result := CompletionEvent.WaitFor(ATimeoutMS) = wrSignaled;
+end;
+
+function TThreadPoolTaskControl.Cancel: Boolean;
+begin
+ FLock.Enter;
+ try
+ Result := FState = ttsPending;
+ if Result then
+ begin
+ FState := ttsCancelled;
+ SignalCompletion;
+ end;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function TThreadPoolTaskControl.GetState: TThreadPoolTaskState;
+begin
+ FLock.Enter;
+ try
+ Result := FState;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function TThreadPoolTaskControl.GetErrorMessage: string;
+begin
+ FLock.Enter;
+ try
+ Result := FErrorMessage;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function TThreadPoolTaskControl.GetIsFinished: Boolean;
+begin
+ FLock.Enter;
+ try
+ Result := IsTerminal(FState);
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function TThreadPoolTaskControl.TryStart: Boolean;
+begin
+ FLock.Enter;
+ try
+ Result := FState = ttsPending;
+ if Result then
+ begin
+ FState := ttsRunning;
+ FRunningThreadID := GetCurrentThreadID;
+ end;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+procedure TThreadPoolTaskControl.MarkCompleted;
+begin
+ FLock.Enter;
+ try
+ if FState = ttsRunning then
+ begin
+ FState := ttsCompleted;
+ FRunningThreadID := 0;
+ SignalCompletion;
+ end;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+procedure TThreadPoolTaskControl.MarkFailed(const AMessage: string);
+begin
+ FLock.Enter;
+ try
+ if FState = ttsRunning then
+ begin
+ FErrorMessage := AMessage;
+ FState := ttsFailed;
+ FRunningThreadID := 0;
+ SignalCompletion;
+ end;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+{ TTrackedWorkItem }
+
+constructor TTrackedWorkItem.Create(AControl: TThreadPoolTaskControl;
+ const ATask: IThreadPoolTask);
+begin
+ inherited Create;
+ FControl := AControl;
+ FTask := ATask;
+ FItemType := witProcedure;
+ FIndex := 0;
+ FFirstIndex := 0;
+ FLastIndex := -1;
+ FIsRange := False;
+end;
+
+procedure TTrackedWorkItem.Execute;
+var
+ CurrentIndex: Integer;
+begin
+ if FIsRange then
+ begin
+ CurrentIndex := FFirstIndex;
+ while True do
+ begin
+ case FItemType of
+ witProcedureIndex:
+ if Assigned(FProcedureIndex) then
+ FProcedureIndex(CurrentIndex);
+ witMethodIndex:
+ if Assigned(FMethodIndex) then
+ FMethodIndex(CurrentIndex);
+ end;
+ if CurrentIndex = FLastIndex then
+ Break;
+ Inc(CurrentIndex);
+ end;
+ Exit;
+ 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;
+
+function TTrackedWorkItem.GetItemType: Integer;
+begin
+ Result := Ord(FItemType);
+end;
+
+function TTrackedWorkItem.TryStart: Boolean;
+begin
+ Result := FControl.TryStart;
+end;
+
+procedure TTrackedWorkItem.MarkCompleted;
+begin
+ FControl.MarkCompleted;
+end;
+
+procedure TTrackedWorkItem.MarkFailed(const AMessage: string);
+begin
+ FControl.MarkFailed(AMessage);
+end;
+
+{ TThreadPoolTaskBatch }
+
+constructor TThreadPoolTaskBatch.Create;
+begin
+ inherited Create;
+ FLock := TCriticalSection.Create;
+ SetLength(FTasks, 8);
+ FCount := 0;
+end;
+
+destructor TThreadPoolTaskBatch.Destroy;
+begin
+ SetLength(FTasks, 0);
+ FLock.Free;
+ inherited;
+end;
+
+procedure TThreadPoolTaskBatch.Add(const ATask: IThreadPoolTask);
+var
+ NewCapacity: Integer;
+begin
+ if ATask = nil then
+ raise EArgumentException.Create('Task must not be nil');
+
+ FLock.Enter;
+ try
+ if FCount = Length(FTasks) then
+ begin
+ NewCapacity := Length(FTasks) * 2;
+ if NewCapacity = 0 then
+ NewCapacity := 8;
+ SetLength(FTasks, NewCapacity);
+ end;
+ FTasks[FCount] := ATask;
+ Inc(FCount);
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function TThreadPoolTaskBatch.Snapshot: TThreadPoolTaskArray;
+var
+ I: Integer;
+begin
+ Result := nil;
+ FLock.Enter;
+ try
+ SetLength(Result, FCount);
+ for I := 0 to FCount - 1 do
+ Result[I] := FTasks[I];
+ finally
+ FLock.Leave;
+ end;
+end;
+
+procedure TThreadPoolTaskBatch.WaitFor;
+begin
+ WaitFor(THREADPOOL_INFINITE);
+end;
+
+function TThreadPoolTaskBatch.WaitFor(ATimeoutMS: Cardinal): Boolean;
+var
+ TasksSnapshot: TThreadPoolTaskArray;
+ StartedAt, Elapsed: QWord;
+ Remaining: Cardinal;
+ I: Integer;
+begin
+ TasksSnapshot := Snapshot;
+ StartedAt := GetTickCount64;
+ for I := 0 to High(TasksSnapshot) do
+ begin
+ if ATimeoutMS = THREADPOOL_INFINITE then
+ Remaining := THREADPOOL_INFINITE
+ else
+ begin
+ Elapsed := GetTickCount64 - StartedAt;
+ if Elapsed >= ATimeoutMS then
+ Remaining := 0
+ else
+ Remaining := ATimeoutMS - Cardinal(Elapsed);
+ end;
+ if not TasksSnapshot[I].WaitFor(Remaining) then
+ Exit(False);
+ end;
+ Result := True;
+end;
+
+function TThreadPoolTaskBatch.CancelPending: Integer;
+var
+ TasksSnapshot: TThreadPoolTaskArray;
+ I: Integer;
+begin
+ Result := 0;
+ TasksSnapshot := Snapshot;
+ for I := 0 to High(TasksSnapshot) do
+ if TasksSnapshot[I].Cancel then
+ Inc(Result);
+end;
+
+function TThreadPoolTaskBatch.GetCount: Integer;
+begin
+ FLock.Enter;
+ try
+ Result := FCount;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function TThreadPoolTaskBatch.GetFinishedCount: Integer;
+var
+ TasksSnapshot: TThreadPoolTaskArray;
+ I: Integer;
+begin
+ Result := 0;
+ TasksSnapshot := Snapshot;
+ for I := 0 to High(TasksSnapshot) do
+ if TasksSnapshot[I].IsFinished then
+ Inc(Result);
+end;
+
+function TThreadPoolTaskBatch.GetFailedCount: Integer;
+var
+ TasksSnapshot: TThreadPoolTaskArray;
+ I: Integer;
+begin
+ Result := 0;
+ TasksSnapshot := Snapshot;
+ for I := 0 to High(TasksSnapshot) do
+ if TasksSnapshot[I].State = ttsFailed then
+ Inc(Result);
+end;
+
+function TThreadPoolTaskBatch.GetCancelledCount: Integer;
+var
+ TasksSnapshot: TThreadPoolTaskArray;
+ I: Integer;
+begin
+ Result := 0;
+ TasksSnapshot := Snapshot;
+ for I := 0 to High(TasksSnapshot) do
+ if TasksSnapshot[I].State = ttsCancelled then
+ Inc(Result);
+end;
+
+function TThreadPoolTaskBatch.GetTask(AIndex: Integer): IThreadPoolTask;
+begin
+ FLock.Enter;
+ try
+ if (AIndex < 0) or (AIndex >= FCount) then
+ raise EArgumentOutOfRangeException.CreateFmt(
+ 'Task index %d is outside the batch', [AIndex]);
+ Result := FTasks[AIndex];
+ finally
+ FLock.Leave;
+ end;
+end;
+
+function NewThreadPoolTaskBatch: IThreadPoolTaskBatch;
+begin
+ Result := TThreadPoolTaskBatch.Create;
+end;
+
+function NewTaskControl(out ATask: IThreadPoolTask): TThreadPoolTaskControl;
+begin
+ Result := TThreadPoolTaskControl.Create;
+ ATask := Result;
+end;
+
+function NewTrackedWorkItem(AProcedure: TThreadProcedure;
+ out ATask: IThreadPoolTask): IWorkItem;
+var
+ Control: TThreadPoolTaskControl;
+ WorkItem: TTrackedWorkItem;
+begin
+ Control := NewTaskControl(ATask);
+ WorkItem := TTrackedWorkItem.Create(Control, ATask);
+ WorkItem.FProcedure := AProcedure;
+ WorkItem.FItemType := witProcedure;
+ Result := WorkItem;
+end;
+
+function NewTrackedWorkItem(AMethod: TThreadMethod;
+ out ATask: IThreadPoolTask): IWorkItem;
+var
+ Control: TThreadPoolTaskControl;
+ WorkItem: TTrackedWorkItem;
+begin
+ Control := NewTaskControl(ATask);
+ WorkItem := TTrackedWorkItem.Create(Control, ATask);
+ WorkItem.FMethod := AMethod;
+ WorkItem.FItemType := witMethod;
+ Result := WorkItem;
+end;
+
+function NewTrackedWorkItem(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer; out ATask: IThreadPoolTask): IWorkItem;
+var
+ Control: TThreadPoolTaskControl;
+ WorkItem: TTrackedWorkItem;
+begin
+ Control := NewTaskControl(ATask);
+ WorkItem := TTrackedWorkItem.Create(Control, ATask);
+ WorkItem.FProcedureIndex := AProcedure;
+ WorkItem.FIndex := AIndex;
+ WorkItem.FItemType := witProcedureIndex;
+ Result := WorkItem;
+end;
+
+function NewTrackedWorkItem(AMethod: TThreadMethodIndex;
+ AIndex: Integer; out ATask: IThreadPoolTask): IWorkItem;
+var
+ Control: TThreadPoolTaskControl;
+ WorkItem: TTrackedWorkItem;
+begin
+ Control := NewTaskControl(ATask);
+ WorkItem := TTrackedWorkItem.Create(Control, ATask);
+ WorkItem.FMethodIndex := AMethod;
+ WorkItem.FIndex := AIndex;
+ WorkItem.FItemType := witMethodIndex;
+ Result := WorkItem;
+end;
+
+function NewTrackedRangeWorkItem(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex: Integer;
+ out ATask: IThreadPoolTask): IWorkItem;
+var
+ Control: TThreadPoolTaskControl;
+ WorkItem: TTrackedWorkItem;
+begin
+ Control := NewTaskControl(ATask);
+ WorkItem := TTrackedWorkItem.Create(Control, ATask);
+ WorkItem.FProcedureIndex := AProcedure;
+ WorkItem.FFirstIndex := AFirstIndex;
+ WorkItem.FLastIndex := ALastIndex;
+ WorkItem.FItemType := witProcedureIndex;
+ WorkItem.FIsRange := True;
+ Result := WorkItem;
+end;
+
+function NewTrackedRangeWorkItem(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex: Integer;
+ out ATask: IThreadPoolTask): IWorkItem;
+var
+ Control: TThreadPoolTaskControl;
+ WorkItem: TTrackedWorkItem;
+begin
+ Control := NewTaskControl(ATask);
+ WorkItem := TTrackedWorkItem.Create(Control, ATask);
+ WorkItem.FMethodIndex := AMethod;
+ WorkItem.FFirstIndex := AFirstIndex;
+ WorkItem.FLastIndex := ALastIndex;
+ WorkItem.FItemType := witMethodIndex;
+ WorkItem.FIsRange := True;
+ Result := WorkItem;
+end;
+
+function GetThreadPoolRangeChunkSize(AFirstIndex, ALastIndex,
+ AThreadCount, ARequestedChunkSize: Integer): Int64;
+var
+ ItemCount, TargetTaskCount: Int64;
+begin
+ if ARequestedChunkSize < 0 then
+ raise EArgumentOutOfRangeException.Create(
+ 'Range chunk size must be zero or greater');
+ if AFirstIndex > ALastIndex then
+ Exit(0);
+ if ARequestedChunkSize > 0 then
+ Exit(ARequestedChunkSize);
+
+ ItemCount := Int64(ALastIndex) - Int64(AFirstIndex) + 1;
+ TargetTaskCount := Int64(AThreadCount) * 4;
+ if TargetTaskCount < 1 then
+ TargetTaskCount := 1;
+ Result := (ItemCount + TargetTaskCount - 1) div TargetTaskCount;
+ if Result < 1 then
+ Result := 1;
+end;
+
+procedure ExecuteThreadPoolWorkItem(const AWorkItem: IWorkItem;
+ AOnError: TThreadPoolErrorEvent);
+var
+ TrackedWorkItem: IThreadPoolTrackedWorkItem;
+ ErrorMessage: string;
+begin
+ TrackedWorkItem := nil;
+ if Supports(AWorkItem, IThreadPoolTrackedWorkItem, TrackedWorkItem) then
+ begin
+ if not TrackedWorkItem.TryStart then
+ Exit;
+ try
+ AWorkItem.Execute;
+ TrackedWorkItem.MarkCompleted;
+ except
+ on E: Exception do
+ begin
+ ErrorMessage := E.Message;
+ try
+ if Assigned(AOnError) then
+ AOnError(ErrorMessage);
+ finally
+ TrackedWorkItem.MarkFailed(ErrorMessage);
+ end;
+ end;
+ end;
+ Exit;
+ end;
+
+ try
+ AWorkItem.Execute;
+ except
+ on E: Exception do
+ if Assigned(AOnError) then
+ AOnError(E.Message);
+ end;
+end;
+
+function NewThreadPoolRangeBatch(AProcedure: TThreadProcedureIndex;
+ AFirstIndex, ALastIndex, AThreadCount, AChunkSize: Integer;
+ ASubmit: TThreadPoolWorkItemSubmit): IThreadPoolTaskBatch;
+var
+ Batch: IThreadPoolTaskBatch;
+ Task: IThreadPoolTask;
+ WorkItem: IWorkItem;
+ EffectiveChunkSize, ChunkFirst, ChunkLast: Int64;
+begin
+ if not Assigned(ASubmit) then
+ raise EArgumentException.Create('Range submit callback must not be nil');
+ Batch := NewThreadPoolTaskBatch;
+ EffectiveChunkSize := GetThreadPoolRangeChunkSize(AFirstIndex,
+ ALastIndex, AThreadCount, AChunkSize);
+ if EffectiveChunkSize = 0 then
+ Exit(Batch);
+
+ try
+ ChunkFirst := AFirstIndex;
+ while ChunkFirst <= Int64(ALastIndex) do
+ begin
+ ChunkLast := ChunkFirst + EffectiveChunkSize - 1;
+ if ChunkLast > Int64(ALastIndex) then
+ ChunkLast := ALastIndex;
+ WorkItem := NewTrackedRangeWorkItem(AProcedure,
+ Integer(ChunkFirst), Integer(ChunkLast), Task);
+ if not ASubmit(WorkItem) then
+ begin
+ Task.Cancel;
+ raise EThreadPoolTaskSubmission.Create(
+ 'Range chunk was not accepted');
+ end;
+ try
+ Batch.Add(Task);
+ except
+ Task.Cancel;
+ raise;
+ end;
+ WorkItem := nil;
+ Task := nil;
+ ChunkFirst := ChunkLast + 1;
+ end;
+ except
+ Batch.CancelPending;
+ raise;
+ end;
+ Result := Batch;
+end;
+
+function NewThreadPoolRangeBatch(AMethod: TThreadMethodIndex;
+ AFirstIndex, ALastIndex, AThreadCount, AChunkSize: Integer;
+ ASubmit: TThreadPoolWorkItemSubmit): IThreadPoolTaskBatch;
+var
+ Batch: IThreadPoolTaskBatch;
+ Task: IThreadPoolTask;
+ WorkItem: IWorkItem;
+ EffectiveChunkSize, ChunkFirst, ChunkLast: Int64;
+begin
+ if not Assigned(ASubmit) then
+ raise EArgumentException.Create('Range submit callback must not be nil');
+ Batch := NewThreadPoolTaskBatch;
+ EffectiveChunkSize := GetThreadPoolRangeChunkSize(AFirstIndex,
+ ALastIndex, AThreadCount, AChunkSize);
+ if EffectiveChunkSize = 0 then
+ Exit(Batch);
+
+ try
+ ChunkFirst := AFirstIndex;
+ while ChunkFirst <= Int64(ALastIndex) do
+ begin
+ ChunkLast := ChunkFirst + EffectiveChunkSize - 1;
+ if ChunkLast > Int64(ALastIndex) then
+ ChunkLast := ALastIndex;
+ WorkItem := NewTrackedRangeWorkItem(AMethod,
+ Integer(ChunkFirst), Integer(ChunkLast), Task);
+ if not ASubmit(WorkItem) then
+ begin
+ Task.Cancel;
+ raise EThreadPoolTaskSubmission.Create(
+ 'Range chunk was not accepted');
+ end;
+ try
+ Batch.Add(Task);
+ except
+ Task.Cancel;
+ raise;
+ end;
+ WorkItem := nil;
+ Task := nil;
+ ChunkFirst := ChunkLast + 1;
+ end;
+ except
+ Batch.CancelPending;
+ raise;
+ end;
+ Result := Batch;
+end;
+
+end.
diff --git a/tests/TestRunner.lpr b/tests/TestRunner.lpr
index 377faa4..35b11c0 100644
--- a/tests/TestRunner.lpr
+++ b/tests/TestRunner.lpr
@@ -8,7 +8,9 @@
{$ENDIF}
Classes, consoletestrunner,
ThreadPool.Simple.Tests,
- ThreadPool.ProducerConsumer.Tests;
+ ThreadPool.ProducerConsumer.Tests,
+ ThreadPool.Tasks.Tests,
+ V08.Compatibility.Tests;
type
diff --git a/tests/ThreadPool.Tasks.Tests.pas b/tests/ThreadPool.Tasks.Tests.pas
new file mode 100644
index 0000000..7c3c573
--- /dev/null
+++ b/tests/ThreadPool.Tasks.Tests.pas
@@ -0,0 +1,701 @@
+unit ThreadPool.Tasks.Tests;
+
+{$mode objfpc}{$H+}{$J-}
+
+interface
+
+uses
+ Classes, SysUtils, SyncObjs, fpcunit, testregistry,
+ ThreadPool.Types, ThreadPool.Tasks, ThreadPool.Simple,
+ ThreadPool.ProducerConsumer;
+
+type
+ TTaskWaitThread = class(TThread)
+ private
+ FTask: IThreadPoolTask;
+ FWaitResult: Boolean;
+ protected
+ procedure Execute; override;
+ public
+ constructor Create(const ATask: IThreadPoolTask);
+ property WaitResult: Boolean read FWaitResult;
+ end;
+
+ TThreadPoolTasksTests = class(TTestCase)
+ private
+ FLock: TCriticalSection;
+ FGateEvent: TEvent;
+ FAllStartedEvent: TEvent;
+ FExpectedStarts: Integer;
+ FStartedCount: Integer;
+ FCounter: Integer;
+ FCoverage: array of Integer;
+ FRangeFirst: Integer;
+ FRangeFailIndex: Integer;
+ FNestedPool: TProducerConsumerThreadPool;
+ FNestedRejected: Integer;
+ FSelfTask: IThreadPoolTask;
+ FSelfWaitRejected: Integer;
+ procedure CountTask;
+ procedure GateTask;
+ procedure RaiseTask;
+ procedure ProcessIndex(AIndex: Integer);
+ procedure SubmitNestedRange;
+ procedure WaitOnOwnTask;
+ procedure PrepareGate(AExpectedStarts: Integer);
+ procedure PrepareCoverage(AFirstIndex, ALastIndex: Integer);
+ protected
+ procedure SetUp; override;
+ procedure TearDown; override;
+ published
+ procedure Test01_TaskCapabilityInterfaces;
+ procedure Test02_SimpleTaskCompletes;
+ procedure Test03_ProducerTaskCompletes;
+ procedure Test04_TaskWaitTimeoutAndMultipleWaiters;
+ procedure Test05_TaskFailureIsObservable;
+ procedure Test06_SimplePendingCancellation;
+ procedure Test07_ProducerPendingCancellation;
+ procedure Test08_RunningTaskCannotBeCancelled;
+ procedure Test09_TaskHandleOutlivesPool;
+ procedure Test10_ProducerTrySubmitTimeoutReturnsNil;
+ procedure Test11_BatchWaitAndCounts;
+ procedure Test12_BatchCancelsPendingTasks;
+ procedure Test13_SimpleAutomaticRangeRunsExactlyOnce;
+ procedure Test14_ProducerExplicitRangeChunks;
+ procedure Test15_EmptyAndNegativeRangeBounds;
+ procedure Test16_RangeFailureIsIsolated;
+ procedure Test17_RangePendingCancellation;
+ procedure Test18_BoundedNestedRangeIsRejected;
+ procedure Test19_InvalidRangeChunkSize;
+ procedure Test20_SubmitCancelRaceHasOneTerminalOutcome;
+ procedure Test21_SubmitAfterShutdownIsRejected;
+ procedure Test22_TaskCannotWaitOnItself;
+ end;
+
+implementation
+
+{ TTaskWaitThread }
+
+constructor TTaskWaitThread.Create(const ATask: IThreadPoolTask);
+begin
+ inherited Create(True);
+ FreeOnTerminate := False;
+ FTask := ATask;
+ FWaitResult := False;
+end;
+
+procedure TTaskWaitThread.Execute;
+begin
+ FWaitResult := FTask.WaitFor(2000);
+end;
+
+{ TThreadPoolTasksTests }
+
+procedure TThreadPoolTasksTests.SetUp;
+begin
+ FLock := TCriticalSection.Create;
+ FGateEvent := TEvent.Create(nil, True, False, '');
+ FAllStartedEvent := TEvent.Create(nil, True, False, '');
+ FExpectedStarts := 0;
+ FStartedCount := 0;
+ FCounter := 0;
+ FRangeFirst := 0;
+ FRangeFailIndex := High(Integer);
+ FNestedPool := nil;
+ FNestedRejected := 0;
+ FSelfTask := nil;
+ FSelfWaitRejected := 0;
+end;
+
+procedure TThreadPoolTasksTests.TearDown;
+begin
+ FGateEvent.SetEvent;
+ FAllStartedEvent.Free;
+ FGateEvent.Free;
+ FLock.Free;
+ SetLength(FCoverage, 0);
+end;
+
+procedure TThreadPoolTasksTests.PrepareGate(AExpectedStarts: Integer);
+begin
+ FLock.Enter;
+ try
+ FExpectedStarts := AExpectedStarts;
+ FStartedCount := 0;
+ FAllStartedEvent.ResetEvent;
+ FGateEvent.ResetEvent;
+ finally
+ FLock.Leave;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.PrepareCoverage(AFirstIndex,
+ ALastIndex: Integer);
+begin
+ FRangeFirst := AFirstIndex;
+ if AFirstIndex > ALastIndex then
+ SetLength(FCoverage, 0)
+ else
+ SetLength(FCoverage, Int64(ALastIndex) - Int64(AFirstIndex) + 1);
+ if Length(FCoverage) > 0 then
+ FillChar(FCoverage[0], Length(FCoverage) * SizeOf(Integer), 0);
+ FCounter := 0;
+end;
+
+procedure TThreadPoolTasksTests.CountTask;
+begin
+ FLock.Enter;
+ try
+ Inc(FCounter);
+ finally
+ FLock.Leave;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.GateTask;
+begin
+ FLock.Enter;
+ try
+ Inc(FStartedCount);
+ if FStartedCount >= FExpectedStarts then
+ FAllStartedEvent.SetEvent;
+ finally
+ FLock.Leave;
+ end;
+ FGateEvent.WaitFor(INFINITE);
+end;
+
+procedure TThreadPoolTasksTests.RaiseTask;
+begin
+ raise Exception.Create('tracked task failure');
+end;
+
+procedure TThreadPoolTasksTests.ProcessIndex(AIndex: Integer);
+var
+ CoverageIndex: Integer;
+begin
+ if AIndex = FRangeFailIndex then
+ raise Exception.CreateFmt('range failure at %d', [AIndex]);
+ CoverageIndex := AIndex - FRangeFirst;
+ FLock.Enter;
+ try
+ if (CoverageIndex >= 0) and (CoverageIndex < Length(FCoverage)) then
+ Inc(FCoverage[CoverageIndex]);
+ Inc(FCounter);
+ finally
+ FLock.Leave;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.SubmitNestedRange;
+var
+ Batch: IThreadPoolTaskBatch;
+begin
+ try
+ Batch := FNestedPool.SubmitRange(@ProcessIndex, 0, 3);
+ Batch.WaitFor;
+ except
+ on E: EThreadPoolDeadlock do
+ InterlockedIncrement(FNestedRejected);
+ end;
+end;
+
+procedure TThreadPoolTasksTests.WaitOnOwnTask;
+begin
+ FGateEvent.WaitFor(INFINITE);
+ try
+ FSelfTask.WaitFor;
+ except
+ on E: EThreadPoolDeadlock do
+ InterlockedIncrement(FSelfWaitRejected);
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test01_TaskCapabilityInterfaces;
+var
+ SimpleSource, ProducerSource: IThreadPoolTaskSource;
+begin
+ SimpleSource := TSimpleThreadPool.Create(4);
+ ProducerSource := TProducerConsumerThreadPool.Create(4, 8);
+ AssertTrue('Simple pool should expose task capabilities',
+ SimpleSource <> nil);
+ AssertTrue('Producer pool should expose task capabilities',
+ ProducerSource <> nil);
+end;
+
+procedure TThreadPoolTasksTests.Test02_SimpleTaskCompletes;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ try
+ Task := Pool.Submit(@CountTask);
+ AssertTrue('Task should finish', Task.WaitFor(2000));
+ AssertEquals(Ord(ttsCompleted), Ord(Task.State));
+ AssertEquals(1, FCounter);
+ AssertEquals('', Task.ErrorMessage);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test03_ProducerTaskCompletes;
+var
+ Pool: TProducerConsumerThreadPool;
+ Task: IThreadPoolTask;
+begin
+ Pool := TProducerConsumerThreadPool.Create(4, 8);
+ try
+ Task := Pool.Submit(@CountTask);
+ Task.WaitFor;
+ AssertEquals(Ord(ttsCompleted), Ord(Task.State));
+ AssertEquals(1, FCounter);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test04_TaskWaitTimeoutAndMultipleWaiters;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+ Waiter1, Waiter2: TTaskWaitThread;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ Waiter1 := nil;
+ Waiter2 := nil;
+ PrepareGate(1);
+ try
+ Task := Pool.Submit(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ AssertFalse('Immediate wait should time out', Task.WaitFor(0));
+ Waiter1 := TTaskWaitThread.Create(Task);
+ Waiter2 := TTaskWaitThread.Create(Task);
+ Waiter1.Start;
+ Waiter2.Start;
+ FGateEvent.SetEvent;
+ Waiter1.WaitFor;
+ Waiter2.WaitFor;
+ AssertTrue(Waiter1.WaitResult);
+ AssertTrue(Waiter2.WaitResult);
+ AssertEquals(Ord(ttsCompleted), Ord(Task.State));
+ finally
+ FGateEvent.SetEvent;
+ Waiter1.Free;
+ Waiter2.Free;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test05_TaskFailureIsObservable;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ try
+ Task := Pool.Submit(@RaiseTask);
+ Task.WaitFor;
+ AssertEquals(Ord(ttsFailed), Ord(Task.State));
+ AssertEquals('tracked task failure', Task.ErrorMessage);
+ AssertEquals('tracked task failure', Pool.LastError);
+ AssertEquals(1, Pool.ErrorCount);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test06_SimplePendingCancellation;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ PrepareGate(Pool.ThreadCount);
+ try
+ for I := 1 to Pool.ThreadCount do
+ Pool.Queue(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ Task := Pool.Submit(@CountTask);
+ AssertTrue('Pending task should be cancelled', Task.Cancel);
+ AssertFalse('Cancellation should be idempotent', Task.Cancel);
+ AssertTrue(Task.WaitFor(0));
+ AssertEquals(Ord(ttsCancelled), Ord(Task.State));
+ FGateEvent.SetEvent;
+ Pool.WaitForAll;
+ AssertEquals('Cancelled callback must not run', 0, FCounter);
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test07_ProducerPendingCancellation;
+var
+ Pool: TProducerConsumerThreadPool;
+ Task: IThreadPoolTask;
+ I: Integer;
+begin
+ Pool := TProducerConsumerThreadPool.Create(4, 4);
+ PrepareGate(Pool.ThreadCount);
+ try
+ for I := 1 to Pool.ThreadCount do
+ Pool.Queue(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ Task := Pool.Submit(@CountTask);
+ AssertTrue(Task.Cancel);
+ AssertEquals(Ord(ttsCancelled), Ord(Task.State));
+ FGateEvent.SetEvent;
+ Pool.WaitForAll;
+ AssertEquals(0, FCounter);
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test08_RunningTaskCannotBeCancelled;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ PrepareGate(1);
+ try
+ Task := Pool.Submit(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ AssertEquals(Ord(ttsRunning), Ord(Task.State));
+ AssertFalse(Task.Cancel);
+ FGateEvent.SetEvent;
+ Task.WaitFor;
+ AssertEquals(Ord(ttsCompleted), Ord(Task.State));
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test09_TaskHandleOutlivesPool;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ Task := Pool.Submit(@CountTask);
+ Pool.Free;
+ AssertTrue(Task.WaitFor(0));
+ AssertEquals(Ord(ttsCompleted), Ord(Task.State));
+ AssertEquals(1, FCounter);
+end;
+
+procedure TThreadPoolTasksTests.Test10_ProducerTrySubmitTimeoutReturnsNil;
+var
+ Pool: TProducerConsumerThreadPool;
+ PendingTask, RejectedTask: IThreadPoolTask;
+ I: Integer;
+begin
+ Pool := TProducerConsumerThreadPool.Create(4, 1);
+ PrepareGate(Pool.ThreadCount);
+ try
+ for I := 1 to Pool.ThreadCount do
+ Pool.Queue(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ PendingTask := Pool.Submit(@CountTask);
+ AssertFalse(Pool.TrySubmit(@CountTask, 0, RejectedTask));
+ AssertTrue('Rejected task handle must be nil', RejectedTask = nil);
+ FGateEvent.SetEvent;
+ PendingTask.WaitFor;
+ AssertEquals(1, FCounter);
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test11_BatchWaitAndCounts;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ Batch := NewThreadPoolTaskBatch;
+ try
+ AssertTrue('Empty batch is complete', Batch.WaitFor(0));
+ Batch.Add(Pool.Submit(@CountTask));
+ Batch.Add(Pool.Submit(@RaiseTask));
+ AssertTrue(Batch.WaitFor(2000));
+ AssertEquals(2, Batch.Count);
+ AssertEquals(2, Batch.FinishedCount);
+ AssertEquals(1, Batch.FailedCount);
+ AssertEquals(0, Batch.CancelledCount);
+ AssertEquals(1, FCounter);
+ AssertEquals(Ord(ttsCompleted), Ord(Batch[0].State));
+ AssertEquals(Ord(ttsFailed), Ord(Batch[1].State));
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test12_BatchCancelsPendingTasks;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ Batch := NewThreadPoolTaskBatch;
+ PrepareGate(Pool.ThreadCount);
+ try
+ for I := 1 to Pool.ThreadCount do
+ Pool.Queue(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ Batch.Add(Pool.Submit(@CountTask));
+ Batch.Add(Pool.Submit(@CountTask));
+ AssertEquals(2, Batch.CancelPending);
+ AssertEquals(0, Batch.CancelPending);
+ AssertTrue(Batch.WaitFor(0));
+ AssertEquals(2, Batch.CancelledCount);
+ FGateEvent.SetEvent;
+ Pool.WaitForAll;
+ AssertEquals(0, FCounter);
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test13_SimpleAutomaticRangeRunsExactlyOnce;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ PrepareCoverage(0, 100);
+ try
+ Batch := Pool.SubmitRange(@ProcessIndex, 0, 100);
+ AssertTrue(Batch.Count <= Pool.ThreadCount * 4);
+ Batch.WaitFor;
+ AssertEquals(101, FCounter);
+ for I := 0 to 100 do
+ AssertEquals('Index should run exactly once', 1, FCoverage[I]);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test14_ProducerExplicitRangeChunks;
+var
+ Pool: TProducerConsumerThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ Pool := TProducerConsumerThreadPool.Create(4, 2);
+ PrepareCoverage(-20, 19);
+ try
+ Batch := Pool.SubmitRange(@ProcessIndex, -20, 19, 7);
+ AssertEquals('40 indexes in chunks of 7', 6, Batch.Count);
+ Batch.WaitFor;
+ AssertEquals(40, FCounter);
+ for I := 0 to 39 do
+ AssertEquals(1, FCoverage[I]);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test15_EmptyAndNegativeRangeBounds;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ try
+ Batch := Pool.SubmitRange(@ProcessIndex, 5, 4);
+ AssertEquals(0, Batch.Count);
+ AssertTrue(Batch.WaitFor(0));
+
+ PrepareCoverage(-3, 3);
+ Batch := Pool.SubmitRange(@ProcessIndex, -3, 3, 2);
+ Batch.WaitFor;
+ AssertEquals(7, FCounter);
+ for I := 0 to 6 do
+ AssertEquals(1, FCoverage[I]);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test16_RangeFailureIsIsolated;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ PrepareCoverage(0, 31);
+ FRangeFailIndex := 5;
+ try
+ Batch := Pool.SubmitRange(@ProcessIndex, 0, 31, 4);
+ Batch.WaitFor;
+ AssertEquals(1, Batch.FailedCount);
+ AssertEquals(1, Pool.ErrorCount);
+ AssertEquals(Ord(ttsFailed), Ord(Batch[1].State));
+ AssertTrue('Other chunks should continue', FCounter > 20);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test17_RangePendingCancellation;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ PrepareCoverage(0, 31);
+ PrepareGate(Pool.ThreadCount);
+ try
+ for I := 1 to Pool.ThreadCount do
+ Pool.Queue(@GateTask);
+ AssertEquals(Ord(wrSignaled), Ord(FAllStartedEvent.WaitFor(2000)));
+ Batch := Pool.SubmitRange(@ProcessIndex, 0, 31, 1);
+ AssertEquals(32, Batch.CancelPending);
+ AssertTrue(Batch.WaitFor(0));
+ FGateEvent.SetEvent;
+ Pool.WaitForAll;
+ AssertEquals(0, FCounter);
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test18_BoundedNestedRangeIsRejected;
+var
+ Task: IThreadPoolTask;
+begin
+ FNestedPool := TProducerConsumerThreadPool.Create(4, 4);
+ PrepareCoverage(0, 3);
+ try
+ Task := FNestedPool.Submit(@SubmitNestedRange);
+ Task.WaitFor;
+ AssertEquals(1, FNestedRejected);
+ AssertEquals(Ord(ttsCompleted), Ord(Task.State));
+ finally
+ FNestedPool.Free;
+ FNestedPool := nil;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test19_InvalidRangeChunkSize;
+var
+ Pool: TSimpleThreadPool;
+ Batch: IThreadPoolTaskBatch;
+ RaisedExpected: Boolean;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ RaisedExpected := False;
+ try
+ try
+ Batch := Pool.SubmitRange(@ProcessIndex, 0, 10, -1);
+ AssertTrue(Batch <> nil);
+ except
+ on E: EArgumentOutOfRangeException do
+ RaisedExpected := True;
+ end;
+ AssertTrue('Negative chunk size should be rejected', RaisedExpected);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test20_SubmitCancelRaceHasOneTerminalOutcome;
+const
+ TASK_COUNT = 5000;
+var
+ Pool: TSimpleThreadPool;
+ Tasks: TThreadPoolTaskArray;
+ CompletedCount, CancelledCount: Integer;
+ I: Integer;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ Tasks := nil;
+ SetLength(Tasks, TASK_COUNT);
+ CompletedCount := 0;
+ CancelledCount := 0;
+ try
+ for I := 0 to TASK_COUNT - 1 do
+ begin
+ Tasks[I] := Pool.Submit(@CountTask);
+ Tasks[I].Cancel;
+ end;
+ Pool.WaitForAll;
+
+ for I := 0 to TASK_COUNT - 1 do
+ begin
+ case Tasks[I].State of
+ ttsCompleted: Inc(CompletedCount);
+ ttsCancelled: Inc(CancelledCount);
+ else
+ Fail('Every raced task must have one terminal outcome');
+ end;
+ AssertFalse('Terminal task cannot be cancelled again', Tasks[I].Cancel);
+ end;
+ AssertEquals(TASK_COUNT, CompletedCount + CancelledCount);
+ AssertEquals(CompletedCount, FCounter);
+ finally
+ Pool.Free;
+ SetLength(Tasks, 0);
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test21_SubmitAfterShutdownIsRejected;
+var
+ Pool: TSimpleThreadPool;
+ Task: IThreadPoolTask;
+ RaisedExpected: Boolean;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ RaisedExpected := False;
+ try
+ Pool.Shutdown;
+ try
+ Task := Pool.Submit(@CountTask);
+ AssertTrue(Task <> nil);
+ except
+ on E: EThreadPoolShutdown do
+ RaisedExpected := True;
+ end;
+ AssertTrue('Submit after shutdown should be rejected', RaisedExpected);
+ finally
+ Pool.Free;
+ end;
+end;
+
+procedure TThreadPoolTasksTests.Test22_TaskCannotWaitOnItself;
+var
+ Pool: TSimpleThreadPool;
+begin
+ Pool := TSimpleThreadPool.Create(4);
+ FGateEvent.ResetEvent;
+ try
+ FSelfTask := Pool.Submit(@WaitOnOwnTask);
+ FGateEvent.SetEvent;
+ AssertTrue(FSelfTask.WaitFor(2000));
+ AssertEquals(1, FSelfWaitRejected);
+ AssertEquals(Ord(ttsCompleted), Ord(FSelfTask.State));
+ finally
+ FGateEvent.SetEvent;
+ Pool.Free;
+ FSelfTask := nil;
+ end;
+end;
+
+initialization
+ RegisterTest(TThreadPoolTasksTests);
+
+end.
diff --git a/tests/V08.Compatibility.Tests.pas b/tests/V08.Compatibility.Tests.pas
new file mode 100644
index 0000000..997711f
--- /dev/null
+++ b/tests/V08.Compatibility.Tests.pas
@@ -0,0 +1,176 @@
+unit V08.Compatibility.Tests;
+
+{$mode objfpc}{$H+}{$J-}
+
+interface
+
+uses
+ SysUtils, fpcunit, testregistry, ThreadPool.Types;
+
+type
+ { Compile sentinel for the exact v0.8 IThreadPool surface. If methods are
+ added to that interface, this unchanged third-party implementation stops
+ compiling and exposes the source-compatibility break. }
+ TV08ThirdPartyPool = class(TInterfacedObject, IThreadPool)
+ private
+ FOnError: TThreadPoolErrorEvent;
+ public
+ procedure Queue(AProcedure: TThreadProcedure); overload;
+ procedure Queue(AMethod: TThreadMethod); overload;
+ procedure Queue(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer); overload;
+ procedure Queue(AMethod: TThreadMethodIndex;
+ AIndex: Integer); overload;
+ 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;
+ function GetThreadCount: Integer;
+ function GetErrors: TStringArray;
+ function GetErrorCount: Integer;
+ function GetOnError: TThreadPoolErrorEvent;
+ procedure SetOnError(AValue: TThreadPoolErrorEvent);
+ function GetState: TThreadPoolState;
+ end;
+
+ TV08CompatibilityTests = class(TTestCase)
+ published
+ procedure Test01_ThirdPartyInterfaceImplementationStillCompiles;
+ end;
+
+implementation
+
+procedure TV08ThirdPartyPool.Queue(AProcedure: TThreadProcedure);
+begin
+ if Assigned(AProcedure) then
+ AProcedure;
+end;
+
+procedure TV08ThirdPartyPool.Queue(AMethod: TThreadMethod);
+begin
+ if Assigned(AMethod) then
+ AMethod;
+end;
+
+procedure TV08ThirdPartyPool.Queue(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer);
+begin
+ if Assigned(AProcedure) then
+ AProcedure(AIndex);
+end;
+
+procedure TV08ThirdPartyPool.Queue(AMethod: TThreadMethodIndex;
+ AIndex: Integer);
+begin
+ if Assigned(AMethod) then
+ AMethod(AIndex);
+end;
+
+function TV08ThirdPartyPool.TryQueue(AProcedure: TThreadProcedure;
+ ATimeoutMS: Cardinal): Boolean;
+begin
+ Queue(AProcedure);
+ Result := True;
+end;
+
+function TV08ThirdPartyPool.TryQueue(AMethod: TThreadMethod;
+ ATimeoutMS: Cardinal): Boolean;
+begin
+ Queue(AMethod);
+ Result := True;
+end;
+
+function TV08ThirdPartyPool.TryQueue(AProcedure: TThreadProcedureIndex;
+ AIndex: Integer; ATimeoutMS: Cardinal): Boolean;
+begin
+ Queue(AProcedure, AIndex);
+ Result := True;
+end;
+
+function TV08ThirdPartyPool.TryQueue(AMethod: TThreadMethodIndex;
+ AIndex: Integer; ATimeoutMS: Cardinal): Boolean;
+begin
+ Queue(AMethod, AIndex);
+ Result := True;
+end;
+
+procedure TV08ThirdPartyPool.WaitForAll;
+begin
+end;
+
+function TV08ThirdPartyPool.WaitForAll(ATimeoutMS: Cardinal): Boolean;
+begin
+ Result := True;
+end;
+
+procedure TV08ThirdPartyPool.Shutdown;
+begin
+end;
+
+procedure TV08ThirdPartyPool.ClearLastError;
+begin
+end;
+
+procedure TV08ThirdPartyPool.ClearErrors;
+begin
+end;
+
+function TV08ThirdPartyPool.GetLastError: string;
+begin
+ Result := '';
+end;
+
+function TV08ThirdPartyPool.GetThreadCount: Integer;
+begin
+ Result := 0;
+end;
+
+function TV08ThirdPartyPool.GetErrors: TStringArray;
+begin
+ Result := nil;
+end;
+
+function TV08ThirdPartyPool.GetErrorCount: Integer;
+begin
+ Result := 0;
+end;
+
+function TV08ThirdPartyPool.GetOnError: TThreadPoolErrorEvent;
+begin
+ Result := FOnError;
+end;
+
+procedure TV08ThirdPartyPool.SetOnError(AValue: TThreadPoolErrorEvent);
+begin
+ FOnError := AValue;
+end;
+
+function TV08ThirdPartyPool.GetState: TThreadPoolState;
+begin
+ Result := tpsAccepting;
+end;
+
+procedure TV08CompatibilityTests.
+ Test01_ThirdPartyInterfaceImplementationStillCompiles;
+var
+ Pool: IThreadPool;
+begin
+ Pool := TV08ThirdPartyPool.Create;
+ AssertEquals(Ord(tpsAccepting), Ord(Pool.State));
+ AssertTrue(Pool.WaitForAll(0));
+end;
+
+initialization
+ RegisterTest(TV08CompatibilityTests);
+
+end.