From 8f1ec27ab7a82706ec1de6f8de90cc0f56b14ba2 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah Date: Sat, 13 Jun 2026 11:03:36 +1000 Subject: [PATCH 1/5] feat(repo): collect all task errors + OnError callback (v0.7.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a richer error-handling API to both thread pools. LastError is unchanged and fully backward compatible. Core (implemented once in TThreadPoolBase, inherited by both pools): - Errors: TStringArray — every captured task error message, oldest first - ErrorCount: Integer - OnError: TThreadPoolErrorEvent — optional callback fired on a worker thread per failed task; invoked outside the lock to avoid deadlocks - ClearErrors — clears the collection and LastError - MAX_STORED_ERRORS (1000) caps the collection; oldest dropped beyond it - Removed the now-redundant per-pool FErrorLock; the base class serializes error capture Tests (43 total, +8): - ThreadPool.Simple.Tests: collection-captures-all, OnError fires per failure, ClearErrors resets, cap enforced, LastError back-compat - ThreadPool.ProducerConsumer.Tests: matching coverage on the second pool Examples: - SimpleErrorHandlingBasic — the easy way: poll Errors/ErrorCount/LastError after WaitForAll; no callback, class, or locking - SimpleErrorHandling — advanced: OnError with a thread-safe handler Docs / housekeeping: - README, both API docs, CHANGELOG, docs/release-notes-v0.7.0.md updated; corrected an inaccurate "error messages include thread IDs" claim (raw messages) - Bump version to 0.7.0 (README badge, package .lpk) - Planned/In Progress now points to the 0.8.0 performance milestone - .gitignore: ignore lazbuild's local packagefiles.xml cache --- .claude/settings.local.json | 12 +- .gitignore | 3 + CHANGELOG.md | 20 +++ README.md | 68 +++++++-- docs/ThreadPool.ProducerConsumer-API.md | 32 +++- docs/ThreadPool.Simple-API.md | 52 ++++++- docs/release-notes-v0.7.0.md | 81 ++++++++++ .../SimpleErrorHandling.lpi | 122 +++++++++++++++ .../SimpleErrorHandling.lpr | 127 ++++++++++++++++ .../SimpleErrorHandlingBasic.lpi | 122 +++++++++++++++ .../SimpleErrorHandlingBasic.lpr | 77 ++++++++++ package/lazarus/threadpool_fp.lpk | 2 +- src/ThreadPool.ProducerConsumer.pas | 13 +- src/ThreadPool.Simple.pas | 17 +-- src/threadpool.types.pas | 142 +++++++++++++++++- tests/ThreadPool.ProducerConsumer.Tests.pas | 81 +++++++++- tests/threadpool.simple.tests.pas | 96 +++++++++++- 17 files changed, 1018 insertions(+), 49 deletions(-) create mode 100644 docs/release-notes-v0.7.0.md create mode 100644 examples/SimpleErrorHandling/SimpleErrorHandling.lpi create mode 100644 examples/SimpleErrorHandling/SimpleErrorHandling.lpr create mode 100644 examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi create mode 100644 examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpr diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 12853d2..64aa0a0 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -9,7 +9,17 @@ "Bash(grep '\\\\.lpi$')", "Bash(lazbuild tests/TestRunner.lpi)", "Bash(git remote *)", - "Bash(git status *)" + "Bash(git status *)", + "Bash(git checkout *)", + "Bash(git pull *)", + "Bash(lazbuild package/lazarus/threadpool_fp.lpk)", + "Bash(./tests/TestRunner.exe -a -p --format=plain)", + "Bash(fpc -Fu../src -Futests threadpooltests.pas)", + "Read(//c/Users/iwank/Documents/github/**)", + "Bash(lazbuild examples/SimpleErrorHandling/SimpleErrorHandling.lpi)", + "Bash(./examples/SimpleErrorHandling/SimpleErrorHandling.exe)", + "Bash(lazbuild examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi)", + "Bash(./examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.exe)" ] } } diff --git a/.gitignore b/.gitignore index 98f276a..2d30508 100644 --- a/.gitignore +++ b/.gitignore @@ -36,6 +36,9 @@ tests/lib/ .vscode/ .idea/ +# lazbuild-generated local package-link cache +packagefiles.xml + # OS specific files ## Windows Thumbs.db diff --git a/CHANGELOG.md b/CHANGELOG.md index efc09dd..09f9168 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,26 @@ 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.7.0] - 2026-06-12 + +### Added + +- Error-collection API on both pools (implemented once in `TThreadPoolBase`, inherited by `TSimpleThreadPool` and `TProducerConsumerThreadPool`): + - `Errors: TStringArray` — every captured task error message, oldest first + - `ErrorCount: Integer` + - `ClearErrors` — clears the collection and `LastError` + - `OnError: TThreadPoolErrorEvent` — optional callback fired (on a worker thread) each time a task raises +- `MAX_STORED_ERRORS` constant (1000) caps the collection; oldest messages are dropped beyond it so a high volume of failing tasks cannot exhaust memory +- 8 new unit tests covering the error-collection API across both pools (collection captures all, `OnError` fires per failure, `ClearErrors` resets, cap enforced, `LastError` back-compat) +- New examples: `examples/SimpleErrorHandlingBasic` (the easy way — poll `Errors`/`ErrorCount`/`LastError` after `WaitForAll`, no callback or locking) and `examples/SimpleErrorHandling` (advanced — adds the `OnError` callback with a thread-safe handler) + +### Changed + +- `LastError` is unchanged and remains backward-compatible (still the most recent error) +- Removed the now-redundant per-pool `FErrorLock`; error capture is serialized by the base class, which also fires `OnError` outside its lock to avoid deadlocks +- README/API docs: documented the new `Errors`/`ErrorCount`/`OnError`/`ClearErrors` API and corrected the `LastError`-overwrite caveats; corrected an inaccurate "error messages include thread IDs" claim (messages are raw) +- README "Planned/In Progress": next milestone is a performance & robustness pass (0.8.0) + ## [0.6.5] - 2026-06-11 ### Added diff --git a/README.md b/README.md index 478997a..56d7549 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # 🚀 ThreadPool for Free Pascal -[![Version](https://img.shields.io/badge/version-0.6.5-8B5CF6.svg)](CHANGELOG.md) +[![Version](https://img.shields.io/badge/version-0.7.0-8B5CF6.svg)](CHANGELOG.md) [![License: MIT](https://img.shields.io/badge/License-MIT-1E3A8A.svg)](https://opensource.org/licenses/MIT) [![Free Pascal](https://img.shields.io/badge/Free%20Pascal-3.2.2+-3B82F6.svg)](https://www.freepascal.org/) [![Lazarus](https://img.shields.io/badge/Lazarus-4.0+-60A5FA.svg)](https://www.lazarus-ide.org/) @@ -124,9 +124,10 @@ A thread pool with fixed-size circular buffer (1024 items) and built-in backpres - Protected error handling - **Error Management** - - Thread-specific error capture - - Error messages with thread IDs - - Continuous operation after exceptions + - Worker exceptions are caught automatically; the pool keeps running + - `LastError` for the most recent failure + - `Errors` collection captures **all** failed-task messages (capped, oldest dropped) — *v0.7.0* + - Optional `OnError` callback fired per failed task — *v0.7.0* > [!NOTE] > Thread count is determined by `TThread.ProcessorCount` at startup and remains fixed. See [Thread Management](#-thread-management) for details. @@ -273,6 +274,38 @@ begin end. ``` +### Capturing all task errors (v0.7.0) + +`LastError` only holds the **most recent** failure. To inspect **every** failed +task, use the `Errors` collection (oldest first, capped at `MAX_STORED_ERRORS = 1000`). +This works the same on both pools: + +```pascal +var + Msg: string; +begin + Pool.ClearErrors; + for i := 0 to 9 do + Pool.Queue(@RiskyProc, i); + Pool.WaitForAll; + + WriteLn(Pool.ErrorCount, ' task(s) failed:'); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + + Pool.ClearErrors; // resets the collection and LastError +end; +``` + +Prefer to react the moment a task fails (instead of polling after `WaitForAll`)? +Assign an `OnError` callback: + +```pascal +// IMPORTANT: OnError is called from a worker thread. Keep the handler short and +// thread-safe; synchronize if it touches the UI or shared state. +Pool.OnError := @MyHandler.OnTaskError; +``` + ### Tips > [!NOTE] @@ -323,7 +356,7 @@ All four `Queue` overloads share the same pattern — pick the one that fits you | Indexed method | `Queue(@MyObj.MyMethod, i)` | Loop parallelism + object state | Parallel array transform on an object | > [!NOTE] -> `LastError` is **overwritten** (not appended) each time a task raises an exception. If you queue multiple tasks, only the last error is stored. Check `LastError` immediately after `WaitForAll` and call `ClearLastError` before reusing the pool. +> `LastError` holds only the **most recent** exception. To see **every** failed task, use the `Errors` collection (added in v0.7.0) — see [Capturing all task errors](#capturing-all-task-errors-v070). Call `ClearErrors` before reusing the pool. ## 📚 Examples @@ -357,6 +390,15 @@ All four `Queue` overloads share the same pattern — pick the one that fits you - Queue full handling - Performance comparison +5. 🛡️ **Error Handling — Basic** (`examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpr`) + - **Start here** for error handling — the easy way + - Just queue, `WaitForAll`, then read `Errors`/`ErrorCount`/`LastError` + - No callback, no custom class, no locking required + +6. 🛡️ **Error Handling — Advanced** (`examples/SimpleErrorHandling/SimpleErrorHandling.lpr`) + - Adds the `OnError` callback to react *while* tasks run + - Shows the thread-safe handler pattern (needed only because the handler keeps shared state) + ### Producer-Consumer Examples 5. 🎓 **Simple Demo** (`examples/ProdConSimpleDemo/ProdConSimpleDemo.lpr`) @@ -516,9 +558,11 @@ for i := 0 to 99 do GlobalThreadPool.WaitForAll; ``` -### 3. Only the last error is kept +### 3. `LastError` only keeps the most recent error -`LastError` is overwritten on every exception — not appended. If multiple tasks fail, you only see the last one. +`LastError` is overwritten on every exception. If multiple tasks fail and you +only check `LastError`, you see just the last one. Use the `Errors` collection +to capture all of them — see [Capturing all task errors](#capturing-all-task-errors-v070) below. ```pascal // Queue several tasks that might fail @@ -526,10 +570,12 @@ for i := 0 to 9 do Pool.Queue(@RiskyProc, i); Pool.WaitForAll; -// Only the LAST exception is in LastError +// Only the LAST exception is in LastError... if Pool.LastError <> '' then - WriteLn('At least one task failed: ', Pool.LastError); -Pool.ClearLastError; + WriteLn('Most recent failure: ', Pool.LastError); +// ...but Errors has them all: +WriteLn(Pool.ErrorCount, ' task(s) failed in total'); +Pool.ClearErrors; ``` ### 4. Freeing the global pool manually @@ -547,7 +593,7 @@ GlobalThreadPool.WaitForAll; ## 🚧 Planned/In Progress -- Richer error handling — collect all task errors (not just the last) and an optional `OnError` callback (planned for 0.7.0) +- Performance & robustness pass — event-driven idle workers (remove the Simple pool's poll loop), shutdown review, stress/soak tests (planned for 0.8.0) - Support for `procedure Queue(AMethod: TProc; AArgs: array of Const);` - More comprehensive tests - More examples diff --git a/docs/ThreadPool.ProducerConsumer-API.md b/docs/ThreadPool.ProducerConsumer-API.md index 05e7265..6ce86ac 100644 --- a/docs/ThreadPool.ProducerConsumer-API.md +++ b/docs/ThreadPool.ProducerConsumer-API.md @@ -129,12 +129,36 @@ Pool.WaitForAll; if Pool.LastError <> '' then begin // LastError holds the raw message of the most recent worker exception. - // Earlier failures are overwritten — only the last one is available. WriteLn('Task error: ', Pool.LastError); Pool.ClearLastError; // reset before reuse end; ``` +To inspect **every** task failure (not just the last), use the `Errors` +collection added in v0.7.0 — oldest first, capped at `MAX_STORED_ERRORS = 1000`: + +```pascal +var + Msg: string; +begin + Pool.ClearErrors; + // ... queue tasks ... + Pool.WaitForAll; + + WriteLn(Pool.ErrorCount, ' task(s) failed:'); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + Pool.ClearErrors; // resets the collection and LastError +end; +``` + +Or assign `OnError` to react the moment a task fails, instead of polling: + +```pascal +// Called from a worker thread — keep it short and thread-safe. +Pool.OnError := @Handler.OnTaskError; +``` + ### Full pattern ```pascal @@ -167,10 +191,14 @@ end; ```pascal property ThreadCount: Integer; // read-only; number of worker threads property LastError: string; // read-only; last worker exception message +property Errors: TStringArray; // read-only; all captured messages (capped) +property ErrorCount: Integer; // read-only; count of messages in Errors +property OnError: TThreadPoolErrorEvent; // fired (on a worker thread) per failed task property WorkQueue: TThreadSafeQueue; // access to queue for monitoring/config procedure WaitForAll; procedure ClearLastError; +procedure ClearErrors; // clears both Errors and LastError ``` --- @@ -324,7 +352,7 @@ end; ## Limitations - Fixed queue capacity — no dynamic resizing -- Only the most recent worker exception is stored in `LastError` +- `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 dynamic thread scaling after construction - Not suitable for real-time or UI-thread work diff --git a/docs/ThreadPool.Simple-API.md b/docs/ThreadPool.Simple-API.md index 3fd5604..56b82fd 100644 --- a/docs/ThreadPool.Simple-API.md +++ b/docs/ThreadPool.Simple-API.md @@ -118,7 +118,6 @@ begin if Pool.LastError <> '' then begin // LastError holds the raw exception message of the most recent failure. - // If multiple tasks fail, only the last exception is stored. WriteLn('Error: ', Pool.LastError); Pool.ClearLastError; // Reset before reusing the pool end; @@ -128,11 +127,57 @@ begin end; ``` +### Collecting all errors (v0.7.0) + +`LastError` only holds the most recent failure. To inspect **every** task error, +use the `Errors` collection (oldest first, capped at `MAX_STORED_ERRORS = 1000`): + +```pascal +var + Msg: string; +begin + Pool.ClearErrors; + for i := 0 to N - 1 do + Pool.Queue(@RiskyProcedure); + Pool.WaitForAll; + + WriteLn(Pool.ErrorCount, ' task(s) failed:'); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + + Pool.ClearErrors; // resets the collection and LastError +end; +``` + +### Reacting to errors as they happen (v0.7.0) + +Assign `OnError` to be notified the moment a task fails, instead of polling after +`WaitForAll`: + +```pascal +type + TMyHandler = class + procedure OnTaskError(const AMessage: string); + end; + +procedure TMyHandler.OnTaskError(const AMessage: string); +begin + // NOTE: called from a worker thread. Keep it short and thread-safe; + // synchronize if you touch the UI or shared state. + Log('task failed: ' + AMessage); +end; + +Pool.OnError := @Handler.OnTaskError; +``` + ### Properties ```pascal -property LastError: string; // Raw message of the most recent worker exception -property ThreadCount: Integer; // Number of worker threads (read-only) +property LastError: string; // Raw message of the most recent worker exception +property Errors: TStringArray; // All captured messages (oldest first, capped) +property ErrorCount: Integer; // Number of messages currently in Errors +property OnError: TThreadPoolErrorEvent; // Fired (on a worker thread) per failed task +property ThreadCount: Integer; // Number of worker threads (read-only) ``` ### Methods @@ -144,6 +189,7 @@ procedure Queue(AProcedure: TThreadProcedureIndex; AIndex: Integer); procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); procedure WaitForAll; procedure ClearLastError; +procedure ClearErrors; // clears both Errors and LastError ``` --- diff --git a/docs/release-notes-v0.7.0.md b/docs/release-notes-v0.7.0.md new file mode 100644 index 0000000..f52cc6e --- /dev/null +++ b/docs/release-notes-v0.7.0.md @@ -0,0 +1,81 @@ +# ThreadPool for Free Pascal — v0.7.0 + +This release adds **richer error handling**. Both thread pools can now report +*every* task that failed — not just the most recent one — and notify you the +moment a task raises. Existing code keeps working unchanged: `LastError` behaves +exactly as before. + +## What's new + +### Collect every task error + +Previously, `LastError` held only the most recent failure — if several tasks +failed, earlier errors were lost. The new `Errors` collection keeps them all +(oldest first): + +```pascal +var + Msg: string; +begin + Pool.ClearErrors; + for i := 0 to 9 do + Pool.Queue(@RiskyProc, i); + Pool.WaitForAll; + + WriteLn(Pool.ErrorCount, ' task(s) failed:'); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + + Pool.ClearErrors; // resets the collection and LastError +end; +``` + +The collection is capped at `MAX_STORED_ERRORS` (1000); beyond that the oldest +messages are dropped, so a flood of failing tasks can't exhaust memory. + +### React to failures as they happen + +Assign an `OnError` callback to be notified immediately, instead of polling after +`WaitForAll`: + +```pascal +// Called from a worker thread — keep it short and thread-safe; synchronize if +// it touches the UI or shared state. +Pool.OnError := @MyHandler.OnTaskError; +``` + +### New API (both pools) + +| Member | Description | +| --- | --- | +| `Errors: TStringArray` | All captured task error messages, oldest first | +| `ErrorCount: Integer` | Number of messages currently held | +| `OnError: TThreadPoolErrorEvent` | Fired (on a worker thread) per failed task | +| `ClearErrors` | Clears both `Errors` and `LastError` | + +The API lives in `TThreadPoolBase`, so `TSimpleThreadPool` and +`TProducerConsumerThreadPool` both get it. + +## Under the hood + +- Removed the now-redundant per-pool error lock; error capture is serialized by + the base class, which fires `OnError` outside its lock to avoid deadlocks. +- Documentation corrected: task error messages are raw exception messages (an + earlier doc note incorrectly claimed they included thread IDs). + +## Upgrade notes + +- **No breaking changes.** `LastError` and `ClearLastError` behave exactly as + before; the new members are purely additive. +- `OnError` handlers run on a worker thread — synchronize any access to shared or + UI state. + +## Verification + +- Test suite: **43 tests, 0 errors, 0 failures, 0 unfreed memory blocks** + (8 new tests cover the error-collection API on both pools). +- Package and all 8 examples build cleanly. + +## Full changelog + +See [CHANGELOG.md](../CHANGELOG.md) for the complete version history. diff --git a/examples/SimpleErrorHandling/SimpleErrorHandling.lpi b/examples/SimpleErrorHandling/SimpleErrorHandling.lpi new file mode 100644 index 0000000..ae4980b --- /dev/null +++ b/examples/SimpleErrorHandling/SimpleErrorHandling.lpi @@ -0,0 +1,122 @@ + + + + + + + + + + + + + + <UseAppBundle Value="False"/> + <ResourceType Value="res"/> + </General> + <BuildModes> + <Item Name="Default" Default="True"/> + <Item Name="Debug"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="SimpleErrorHandling"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <Parsing> + <SyntaxOptions> + <IncludeAssertionCode Value="True"/> + </SyntaxOptions> + </Parsing> + <CodeGeneration> + <Checks> + <IOChecks Value="True"/> + <RangeChecks Value="True"/> + <OverflowChecks Value="True"/> + <StackChecks Value="True"/> + </Checks> + <VerifyObjMethodCallValidity Value="True"/> + </CodeGeneration> + <Linking> + <Debugging> + <DebugInfoType Value="dsDwarf3"/> + <UseHeaptrc Value="True"/> + <TrashVariables Value="True"/> + <UseExternalDbgSyms Value="True"/> + </Debugging> + </Linking> + </CompilerOptions> + </Item> + <Item Name="Release"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="SimpleErrorHandling"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <CodeGeneration> + <SmartLinkUnit Value="True"/> + <Optimizations> + <OptimizationLevel Value="3"/> + </Optimizations> + </CodeGeneration> + <Linking> + <Debugging> + <GenerateDebugInfo Value="False"/> + <RunWithoutDebug Value="True"/> + </Debugging> + <LinkSmart Value="True"/> + </Linking> + </CompilerOptions> + </Item> + </BuildModes> + <PublishOptions> + <Version Value="2"/> + <UseFileFilters Value="True"/> + </PublishOptions> + <RunParams> + <FormatVersion Value="2"/> + </RunParams> + <Units> + <Unit> + <Filename Value="SimpleErrorHandling.lpr"/> + <IsPartOfProject Value="True"/> + </Unit> + </Units> + </ProjectOptions> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="SimpleErrorHandling"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + </CompilerOptions> + <Debugging> + <Exceptions> + <Item> + <Name Value="EAbort"/> + </Item> + <Item> + <Name Value="ECodetoolError"/> + </Item> + <Item> + <Name Value="EFOpenError"/> + </Item> + </Exceptions> + </Debugging> +</CONFIG> diff --git a/examples/SimpleErrorHandling/SimpleErrorHandling.lpr b/examples/SimpleErrorHandling/SimpleErrorHandling.lpr new file mode 100644 index 0000000..104ca10 --- /dev/null +++ b/examples/SimpleErrorHandling/SimpleErrorHandling.lpr @@ -0,0 +1,127 @@ +program SimpleErrorHandling; + +{ Error handling with ThreadPool (v0.7.0) + ======================================== + Demonstrates the richer error-handling API added in v0.7.0: + + * LastError — the most recent task error (unchanged from earlier versions) + * Errors — EVERY failed task's message, not just the last + * ErrorCount — how many tasks have failed + * OnError — a callback fired the moment a task fails + * ClearErrors — reset the collection (and LastError) before reusing the pool + + HOW TO COMPILE AND RUN + ---------------------- + Option A — Free Pascal compiler: + fpc -Fu../../src SimpleErrorHandling.lpr && ./SimpleErrorHandling + + Option B — Lazarus IDE / lazbuild: + lazbuild SimpleErrorHandling.lpi && ./SimpleErrorHandling + + Works the same with TProducerConsumerThreadPool — the error API lives on the + shared base class. } + +{$mode objfpc}{$H+}{$J-} + +uses + {$IFDEF UNIX} + cthreads, // MUST be first: enables threading support on Unix/Linux + {$ENDIF} + Classes, SysUtils, SyncObjs, ThreadPool.Simple; + +type + { Collects OnError notifications. The handler runs on a worker thread, so it + must be thread-safe — here a critical section guards the counter. } + TErrorWatcher = class + private + FLock: TCriticalSection; + FCount: Integer; + public + constructor Create; + destructor Destroy; override; + procedure HandleError(const AMessage: string); // assigned to Pool.OnError + property Count: Integer read FCount; + end; + +constructor TErrorWatcher.Create; +begin + inherited Create; + FLock := TCriticalSection.Create; + FCount := 0; +end; + +destructor TErrorWatcher.Destroy; +begin + FLock.Free; + inherited Destroy; +end; + +procedure TErrorWatcher.HandleError(const AMessage: string); +begin + // Called from a worker thread — keep it short and synchronized. + FLock.Enter; + try + Inc(FCount); + WriteLn(Format(' [OnError] task failed: %s', [AMessage])); + finally + FLock.Leave; + end; +end; + +{ Some tasks succeed, some fail. The index decides which. } +procedure MaybeFail(index: Integer); +begin + if (index mod 3) = 0 then + raise Exception.CreateFmt('task %d hit a problem', [index]); + // Successful tasks simply do nothing here. +end; + +var + Pool: TSimpleThreadPool; + Watcher: TErrorWatcher; + Msg: string; + i: Integer; +begin + Pool := TSimpleThreadPool.Create(4); + Watcher := TErrorWatcher.Create; + try + WriteLn('=== ThreadPool error handling (v0.7.0) ==='); + WriteLn; + + // 1) React to failures as they happen via OnError. + Pool.OnError := @Watcher.HandleError; + + // 2) Queue 10 tasks; indices divisible by 3 (0,3,6,9) will fail. + WriteLn('Queueing 10 tasks (indices divisible by 3 will fail)...'); + Pool.ClearErrors; // start from a clean slate + for i := 0 to 9 do + Pool.Queue(@MaybeFail, i); + + Pool.WaitForAll; + WriteLn; + + // 3) After WaitForAll, inspect the collected errors. + WriteLn(Format('ErrorCount: %d task(s) failed', [Pool.ErrorCount])); + WriteLn(Format('OnError fired %d time(s)', [Watcher.Count])); + WriteLn(Format('LastError (most recent only): "%s"', [Pool.LastError])); + WriteLn; + + WriteLn('All failures (from Pool.Errors):'); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + WriteLn; + + // 4) Reset and reuse the pool. + Pool.ClearErrors; + WriteLn(Format('After ClearErrors: ErrorCount = %d, LastError = "%s"', + [Pool.ErrorCount, Pool.LastError])); + + finally + Watcher.Free; + Pool.Free; + end; + + WriteLn; + WriteLn('Press enter to quit ...'); + ReadLn; +end. diff --git a/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi b/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi new file mode 100644 index 0000000..93187f3 --- /dev/null +++ b/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi @@ -0,0 +1,122 @@ +<?xml version="1.0" encoding="UTF-8"?> +<CONFIG> + <ProjectOptions> + <Version Value="12"/> + <PathDelim Value="\"/> + <General> + <Flags> + <MainUnitHasCreateFormStatements Value="False"/> + <MainUnitHasTitleStatement Value="False"/> + <MainUnitHasScaledStatement Value="False"/> + </Flags> + <SessionStorage Value="InProjectDir"/> + <Title Value="SimpleErrorHandlingBasic"/> + <UseAppBundle Value="False"/> + <ResourceType Value="res"/> + </General> + <BuildModes> + <Item Name="Default" Default="True"/> + <Item Name="Debug"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="SimpleErrorHandlingBasic"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <Parsing> + <SyntaxOptions> + <IncludeAssertionCode Value="True"/> + </SyntaxOptions> + </Parsing> + <CodeGeneration> + <Checks> + <IOChecks Value="True"/> + <RangeChecks Value="True"/> + <OverflowChecks Value="True"/> + <StackChecks Value="True"/> + </Checks> + <VerifyObjMethodCallValidity Value="True"/> + </CodeGeneration> + <Linking> + <Debugging> + <DebugInfoType Value="dsDwarf3"/> + <UseHeaptrc Value="True"/> + <TrashVariables Value="True"/> + <UseExternalDbgSyms Value="True"/> + </Debugging> + </Linking> + </CompilerOptions> + </Item> + <Item Name="Release"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="SimpleErrorHandlingBasic"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <CodeGeneration> + <SmartLinkUnit Value="True"/> + <Optimizations> + <OptimizationLevel Value="3"/> + </Optimizations> + </CodeGeneration> + <Linking> + <Debugging> + <GenerateDebugInfo Value="False"/> + <RunWithoutDebug Value="True"/> + </Debugging> + <LinkSmart Value="True"/> + </Linking> + </CompilerOptions> + </Item> + </BuildModes> + <PublishOptions> + <Version Value="2"/> + <UseFileFilters Value="True"/> + </PublishOptions> + <RunParams> + <FormatVersion Value="2"/> + </RunParams> + <Units> + <Unit> + <Filename Value="SimpleErrorHandlingBasic.lpr"/> + <IsPartOfProject Value="True"/> + </Unit> + </Units> + </ProjectOptions> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="SimpleErrorHandlingBasic"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + </CompilerOptions> + <Debugging> + <Exceptions> + <Item> + <Name Value="EAbort"/> + </Item> + <Item> + <Name Value="ECodetoolError"/> + </Item> + <Item> + <Name Value="EFOpenError"/> + </Item> + </Exceptions> + </Debugging> +</CONFIG> diff --git a/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpr b/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpr new file mode 100644 index 0000000..ad1fd3d --- /dev/null +++ b/examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpr @@ -0,0 +1,77 @@ +program SimpleErrorHandlingBasic; + +{ Error handling the EASY way (v0.7.0) + ===================================== + The simplest possible way to handle task errors: queue your work, call + WaitForAll, then read the results. No callback, no custom class, no locking. + + * LastError — the most recent task error (a single string) + * ErrorCount — how many tasks failed + * Errors — the message of EVERY failed task + * ClearErrors — reset before reusing the pool + + Reading these AFTER WaitForAll is completely safe — the pool handles all the + threading for you. You only need locks/callbacks if you want to react to + failures *while tasks are still running* (see the SimpleErrorHandling example + for that more advanced pattern). + + HOW TO COMPILE AND RUN + ---------------------- + Option A — Free Pascal compiler: + fpc -Fu../../src SimpleErrorHandlingBasic.lpr && ./SimpleErrorHandlingBasic + + Option B — Lazarus IDE / lazbuild: + lazbuild SimpleErrorHandlingBasic.lpi && ./SimpleErrorHandlingBasic } + +{$mode objfpc}{$H+}{$J-} + +uses + {$IFDEF UNIX} + cthreads, // MUST be first: enables threading support on Unix/Linux + {$ENDIF} + Classes, SysUtils, ThreadPool.Simple; + +{ A task that fails for some inputs and succeeds for others. } +procedure ProcessItem(index: Integer); +begin + if (index mod 3) = 0 then + raise Exception.CreateFmt('could not process item %d', [index]); + // Successful items just do their work here (nothing, for the demo). +end; + +var + Msg: string; + i: Integer; +begin + WriteLn('=== Easy error handling (v0.7.0) ==='); + WriteLn; + + // 1) Queue some work. Items where index is divisible by 3 will fail. + WriteLn('Queueing 10 tasks (items divisible by 3 will fail)...'); + for i := 0 to 9 do + GlobalThreadPool.Queue(@ProcessItem, i); + + // 2) Wait for everything to finish. + GlobalThreadPool.WaitForAll; + + // 3) Now it is safe to read the results — no locking needed. + WriteLn; + if GlobalThreadPool.ErrorCount = 0 then + WriteLn('All tasks succeeded!') + else + begin + WriteLn(GlobalThreadPool.ErrorCount, ' task(s) failed:'); + for Msg in GlobalThreadPool.Errors do + WriteLn(' - ', Msg); + + WriteLn; + WriteLn('Most recent error (LastError): ', GlobalThreadPool.LastError); + end; + + // 4) Clear before reusing the global pool elsewhere in your program. + GlobalThreadPool.ClearErrors; + + WriteLn; + WriteLn('Press enter to quit ...'); + ReadLn; +end. diff --git a/package/lazarus/threadpool_fp.lpk b/package/lazarus/threadpool_fp.lpk index 5c31584..5b8f839 100644 --- a/package/lazarus/threadpool_fp.lpk +++ b/package/lazarus/threadpool_fp.lpk @@ -81,7 +81,7 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. "/> - <Version Minor="6" Release="5"/> + <Version Minor="7"/> <Files> <Item> <Filename Value="..\..\src\ThreadPool.Simple.pas"/> diff --git a/src/ThreadPool.ProducerConsumer.pas b/src/ThreadPool.ProducerConsumer.pas index 029177a..6c5bc24 100644 --- a/src/ThreadPool.ProducerConsumer.pas +++ b/src/ThreadPool.ProducerConsumer.pas @@ -95,7 +95,6 @@ TProducerConsumerThreadPool = class(TThreadPoolBase) FThreads: TThreadList; FWorkQueue: TThreadSafeQueue; // Use our custom thread-safe queue FCompletionEvent: TEvent; - FErrorLock: TCriticalSection; FWorkItemCount: integer; FWorkItemLock: TCriticalSection; FLocalThreadCount: integer; @@ -151,7 +150,6 @@ constructor TProducerConsumerThreadPool.Create(AThreadCount: Integer = 0; AQueue FThreads := TThreadList.Create; FWorkQueue := TThreadSafeQueue.Create(AQueueSize); FCompletionEvent := TEvent.Create(nil, True, True, ''); - FErrorLock := TCriticalSection.Create; FWorkItemLock := TCriticalSection.Create; FWorkItemCount := 0; FLastError := ''; @@ -172,7 +170,6 @@ destructor TProducerConsumerThreadPool.Destroy; ClearThreads; FWorkQueue.Free; FCompletionEvent.Free; - FErrorLock.Free; FWorkItemLock.Free; FThreads.Free; inherited; @@ -529,12 +526,10 @@ procedure TProducerConsumerWorkerThread.Execute; on E: Exception do begin DebugLog('Error executing work item: ' + E.Message); - Pool.FErrorLock.Enter; - try - Pool.SetLastError(E.Message); - finally - Pool.FErrorLock.Leave; - end; + // SetLastError is thread-safe on its own (base class locking), so + // no extra lock is needed. Calling it outside any subclass lock + // also ensures the OnError callback it may fire cannot deadlock. + Pool.SetLastError(E.Message); end; end; diff --git a/src/ThreadPool.Simple.pas b/src/ThreadPool.Simple.pas index 5781d51..6b69886 100644 --- a/src/ThreadPool.Simple.pas +++ b/src/ThreadPool.Simple.pas @@ -60,7 +60,6 @@ TSimpleThreadPool = class(TThreadPoolBase) FWorkItemLock: TCriticalSection; FWorkItemCount: Integer; FWorkItemEvent: TEvent; - FErrorLock: TCriticalSection; FErrorEvent: TEvent; procedure ClearThreads; procedure ClearWorkItems; @@ -183,14 +182,12 @@ procedure TSimpleWorkerThread.Execute; except on E: Exception do begin - // Capture error - Pool.FErrorLock.Enter; - try - Pool.SetLastError(E.Message); - Pool.FErrorEvent.SetEvent; - finally - Pool.FErrorLock.Leave; - end; + // Capture error. SetLastError is thread-safe on its own (base class + // locking), and TEvent.SetEvent is thread-safe, so no extra lock is + // needed here. Keeping SetLastError outside any subclass lock also + // ensures the OnError callback it may fire cannot deadlock. + Pool.SetLastError(E.Message); + Pool.FErrorEvent.SetEvent; end; end; WorkItem.Free; @@ -267,7 +264,6 @@ constructor TSimpleThreadPool.Create(AThreadCount: Integer = 0); FThreads := TThreadList.Create; FWorkItems := TThreadList.Create; FWorkItemLock := TCriticalSection.Create; - FErrorLock := TCriticalSection.Create; FErrorEvent := TEvent.Create(nil, True, False, ''); FWorkItemEvent := TEvent.Create(nil, True, True, ''); FWorkItemCount := 0; @@ -303,7 +299,6 @@ destructor TSimpleThreadPool.Destroy; FWorkItems.Free; FWorkItemEvent.Free; FErrorEvent.Free; - FErrorLock.Free; inherited Destroy; end; diff --git a/src/threadpool.types.pas b/src/threadpool.types.pas index 4ab77aa..ce576ad 100644 --- a/src/threadpool.types.pas +++ b/src/threadpool.types.pas @@ -5,7 +5,14 @@ interface uses - Classes, SysUtils, Math; + Classes, SysUtils, SyncObjs, Math; + +const + { Upper bound on how many task error messages a pool retains. When exceeded, + the oldest messages are dropped so a high volume of failing tasks cannot + exhaust memory. LastError always reflects the most recent error regardless + of this cap. } + MAX_STORED_ERRORS = 1000; type { Task Types - Different kinds of work that can be queued } @@ -13,6 +20,14 @@ interface TThreadMethod = procedure of object; TThreadProcedureIndex = procedure(index: Integer); TThreadMethodIndex = procedure(index: Integer) of object; + + { Fired (from a worker thread) each time a queued task raises an exception. + AMessage is the exception's Message. Handlers must be thread-safe and + should not block, since they run on the worker thread that caught the error. } + TThreadPoolErrorEvent = procedure(const AMessage: string) of object; + + { Snapshot of captured error messages, oldest first. } + TStringArray = array of string; { Work item types } TWorkItemType = ( @@ -58,10 +73,21 @@ interface procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); overload; procedure WaitForAll; procedure ClearLastError; + procedure ClearErrors; function GetLastError: string; function GetThreadCount: Integer; + function GetErrors: TStringArray; + function GetErrorCount: Integer; + function GetOnError: TThreadPoolErrorEvent; + procedure SetOnError(AValue: TThreadPoolErrorEvent); property LastError: string read GetLastError; property ThreadCount: Integer read GetThreadCount; + { All task error messages captured since the last ClearErrors/ClearLastError, + oldest first, capped at MAX_STORED_ERRORS. } + property Errors: TStringArray read GetErrors; + property ErrorCount: Integer read GetErrorCount; + { Optional callback fired from a worker thread whenever a task raises. } + property OnError: TThreadPoolErrorEvent read GetOnError write SetOnError; end; { Base class for thread pool implementations } @@ -70,12 +96,18 @@ TThreadPoolBase = class(TInterfacedObject, IThreadPool) FLastError: string; FThreadCount: Integer; FShutdown: Boolean; - + { Collected error messages and the lock guarding them plus FLastError and + FOnError. This lock is internal to the base class and independent of any + lock a subclass holds around its own SetLastError call. } + FErrors: TStringList; + FErrorsLock: TCriticalSection; + FOnError: TThreadPoolErrorEvent; + procedure SetLastError(const AError: string); virtual; public constructor Create(AThreadCount: Integer); virtual; destructor Destroy; override; - + { IThreadPool implementation } procedure Queue(AProcedure: TThreadProcedure); virtual; abstract; procedure Queue(AMethod: TThreadMethod); virtual; abstract; @@ -83,8 +115,20 @@ TThreadPoolBase = class(TInterfacedObject, IThreadPool) procedure Queue(AMethod: TThreadMethodIndex; AIndex: Integer); virtual; abstract; procedure WaitForAll; virtual; abstract; procedure ClearLastError; virtual; + procedure ClearErrors; virtual; function GetLastError: string; virtual; function GetThreadCount: Integer; virtual; + function GetErrors: TStringArray; virtual; + function GetErrorCount: Integer; virtual; + function GetOnError: TThreadPoolErrorEvent; virtual; + procedure SetOnError(AValue: TThreadPoolErrorEvent); virtual; + + { All task error messages captured since the last ClearErrors/ClearLastError, + oldest first, capped at MAX_STORED_ERRORS. } + property Errors: TStringArray read GetErrors; + property ErrorCount: Integer read GetErrorCount; + { Optional callback fired from a worker thread whenever a task raises. } + property OnError: TThreadPoolErrorEvent read GetOnError write SetOnError; end; implementation @@ -96,7 +140,10 @@ constructor TThreadPoolBase.Create(AThreadCount: Integer); inherited Create; FShutdown := False; FLastError := ''; - + FErrors := TStringList.Create; + FErrorsLock := TCriticalSection.Create; + FOnError := nil; + // Apply thread count safety limits if AThreadCount <= 0 then AThreadCount := TThread.ProcessorCount @@ -108,22 +155,57 @@ constructor TThreadPoolBase.Create(AThreadCount: Integer); destructor TThreadPoolBase.Destroy; begin FShutdown := True; + FErrors.Free; + FErrorsLock.Free; inherited; end; procedure TThreadPoolBase.SetLastError(const AError: string); +var + Handler: TThreadPoolErrorEvent; begin - FLastError := AError; + FErrorsLock.Enter; + try + FLastError := AError; + FErrors.Add(AError); + // Bound memory: drop oldest entries beyond the cap. + while FErrors.Count > MAX_STORED_ERRORS do + FErrors.Delete(0); + Handler := FOnError; + finally + FErrorsLock.Leave; + end; + + // Fire the callback outside the lock so a handler cannot deadlock by calling + // back into the pool, and so a slow handler does not block other workers. + if Assigned(Handler) then + Handler(AError); end; procedure TThreadPoolBase.ClearLastError; begin - FLastError := ''; + FErrorsLock.Enter; + try + FLastError := ''; + FErrors.Clear; + finally + FErrorsLock.Leave; + end; +end; + +procedure TThreadPoolBase.ClearErrors; +begin + ClearLastError; end; function TThreadPoolBase.GetLastError: string; begin - Result := FLastError; + FErrorsLock.Enter; + try + Result := FLastError; + finally + FErrorsLock.Leave; + end; end; function TThreadPoolBase.GetThreadCount: Integer; @@ -131,4 +213,48 @@ function TThreadPoolBase.GetThreadCount: Integer; Result := FThreadCount; end; -end. +function TThreadPoolBase.GetErrors: TStringArray; +var + I: Integer; +begin + FErrorsLock.Enter; + try + SetLength(Result, FErrors.Count); + for I := 0 to FErrors.Count - 1 do + Result[I] := FErrors[I]; + finally + FErrorsLock.Leave; + end; +end; + +function TThreadPoolBase.GetErrorCount: Integer; +begin + FErrorsLock.Enter; + try + Result := FErrors.Count; + finally + FErrorsLock.Leave; + end; +end; + +function TThreadPoolBase.GetOnError: TThreadPoolErrorEvent; +begin + FErrorsLock.Enter; + try + Result := FOnError; + finally + FErrorsLock.Leave; + end; +end; + +procedure TThreadPoolBase.SetOnError(AValue: TThreadPoolErrorEvent); +begin + FErrorsLock.Enter; + try + FOnError := AValue; + finally + FErrorsLock.Leave; + end; +end; + +end. diff --git a/tests/ThreadPool.ProducerConsumer.Tests.pas b/tests/ThreadPool.ProducerConsumer.Tests.pas index 4f32044..05ff3a2 100644 --- a/tests/ThreadPool.ProducerConsumer.Tests.pas +++ b/tests/ThreadPool.ProducerConsumer.Tests.pas @@ -16,7 +16,8 @@ TTestProducerConsumerThreadPool = class(TTestCase) FSharedCounter: Integer; FSharedLock: TCriticalSection; FTestResults: TStringList; - + FOnErrorCount: Integer; + procedure LogTest(const Msg: string); procedure IncrementCounter; procedure AddToResults; @@ -25,6 +26,8 @@ TTestProducerConsumerThreadPool = class(TTestCase) procedure SlowTask; procedure LongTask; procedure RaiseTestError; + procedure RaiseOtherError; + procedure HandlePoolError(const AMessage: string); procedure SleepTask; function MeasureQueueTime(const TaskCount: Integer): Int64; protected @@ -45,6 +48,11 @@ TTestProducerConsumerThreadPool = class(TTestCase) procedure Test12_LoadFactorCalculation; procedure Test13_BackpressureBehavior; procedure Test14_AdaptivePerformance; + + // Error-collection API (v0.7.0) + procedure Test15_ErrorsCollectionCapturesAll; + procedure Test16_OnErrorCallbackFires; + procedure Test17_ClearErrorsResetsCollection; end; implementation @@ -137,6 +145,17 @@ procedure TTestProducerConsumerThreadPool.RaiseTestError; raise Exception.Create('Test error'); end; +procedure TTestProducerConsumerThreadPool.RaiseOtherError; +begin + raise Exception.Create('Other error'); +end; + +procedure TTestProducerConsumerThreadPool.HandlePoolError(const AMessage: string); +begin + // Fired from a worker thread — increment atomically. + InterlockedIncrement(FOnErrorCount); +end; + procedure TTestProducerConsumerThreadPool.SleepTask; begin Sleep(100); @@ -547,10 +566,68 @@ procedure TTestProducerConsumerThreadPool.Test14_AdaptivePerformance; LogTest(Format('Performance ratio: %.2fx faster under low load', [Ratio])); AssertTrue('Low load should be proportionally faster', Ratio > 1.5); - + LogTest('Test14_AdaptivePerformance finished'); end; +procedure TTestProducerConsumerThreadPool.Test15_ErrorsCollectionCapturesAll; +var + Errors: TStringArray; +begin + LogTest('Test15_ErrorsCollectionCapturesAll starting...'); + // Two distinct failing tasks: the collection should keep BOTH, where the old + // single-slot LastError kept only the most recent. + FThreadPool.ClearErrors; + FThreadPool.Queue(@RaiseTestError); + FThreadPool.Queue(@RaiseOtherError); + FThreadPool.WaitForAll; + + AssertEquals('Both task errors should be collected', 2, FThreadPool.ErrorCount); + + Errors := FThreadPool.Errors; + AssertEquals('Errors array length should match ErrorCount', 2, Length(Errors)); + // Order of completion is not guaranteed, so assert presence, not position. + AssertTrue('Collection should contain the first error message', + (Errors[0] = 'Test error') or (Errors[1] = 'Test error')); + AssertTrue('Collection should contain the second error message', + (Errors[0] = 'Other error') or (Errors[1] = 'Other error')); + // LastError still works and holds one of the two (back-compat). + AssertTrue('LastError should still hold one of the errors', + (FThreadPool.LastError = 'Test error') or (FThreadPool.LastError = 'Other error')); + LogTest('Test15_ErrorsCollectionCapturesAll finished'); +end; + +procedure TTestProducerConsumerThreadPool.Test16_OnErrorCallbackFires; +begin + LogTest('Test16_OnErrorCallbackFires starting...'); + FOnErrorCount := 0; + FThreadPool.ClearErrors; + FThreadPool.OnError := @HandlePoolError; + try + FThreadPool.Queue(@RaiseTestError); + FThreadPool.Queue(@RaiseOtherError); + FThreadPool.WaitForAll; + + AssertEquals('OnError should fire once per failed task', 2, FOnErrorCount); + finally + FThreadPool.OnError := nil; // avoid dangling callback after this test + end; + LogTest('Test16_OnErrorCallbackFires finished'); +end; + +procedure TTestProducerConsumerThreadPool.Test17_ClearErrorsResetsCollection; +begin + LogTest('Test17_ClearErrorsResetsCollection starting...'); + FThreadPool.Queue(@RaiseTestError); + FThreadPool.WaitForAll; + AssertTrue('Precondition: an error was captured', FThreadPool.ErrorCount > 0); + + FThreadPool.ClearErrors; + AssertEquals('ClearErrors should empty the collection', 0, FThreadPool.ErrorCount); + AssertEquals('ClearErrors should also clear LastError', '', FThreadPool.LastError); + LogTest('Test17_ClearErrorsResetsCollection finished'); +end; + initialization RegisterTest(TTestProducerConsumerThreadPool); end. diff --git a/tests/threadpool.simple.tests.pas b/tests/threadpool.simple.tests.pas index 4b5be15..f69fb27 100644 --- a/tests/threadpool.simple.tests.pas +++ b/tests/threadpool.simple.tests.pas @@ -30,9 +30,12 @@ TSimpleThreadPoolTests = class(TTestCase) FTestObject: TTestObject; FCounter: Integer; FCS: TCriticalSection; - + FOnErrorCount: Integer; + procedure IncrementCounter; procedure IncrementCounterWithIndex(AIndex: Integer); + // Thread-safe OnError handler used by Test23. + procedure HandlePoolError(const AMessage: string); protected procedure SetUp; override; procedure TearDown; override; @@ -67,6 +70,13 @@ TSimpleThreadPoolTests = class(TTestCase) procedure Test19_ExceptionMessage; procedure Test20_MultipleExceptions; procedure Test21_ExceptionAfterClear; + + // Error-collection API (v0.7.0) + procedure Test22_ErrorsCollectionCapturesAll; + procedure Test23_OnErrorCallbackFires; + procedure Test24_ClearErrorsResetsCollection; + procedure Test25_ErrorCollectionIsCapped; + procedure Test26_LastErrorStillWorks; end; var @@ -503,6 +513,90 @@ procedure TSimpleThreadPoolTests.Test21_ExceptionAfterClear; AssertEquals('Should work normally after clearing error', 1, FCounter); end; +procedure TSimpleThreadPoolTests.Test22_ErrorsCollectionCapturesAll; +var + Errors: TStringArray; +begin + // Two distinct failing tasks: the collection should keep BOTH, where the old + // single-slot LastError kept only the most recent. + FThreadPool.ClearErrors; + FThreadPool.Queue(@RaiseTestException); + FThreadPool.Queue(@RaiseAnotherException); + FThreadPool.WaitForAll; + + AssertEquals('Both task errors should be collected', 2, FThreadPool.ErrorCount); + + Errors := FThreadPool.Errors; + AssertEquals('Errors array length should match ErrorCount', 2, Length(Errors)); + // Order of completion is not guaranteed, so assert presence, not position. + AssertTrue('Collection should contain the first error message', + (Errors[0] = 'Test exception message') or (Errors[1] = 'Test exception message')); + AssertTrue('Collection should contain the second error message', + (Errors[0] = 'Another exception message') or (Errors[1] = 'Another exception message')); +end; + +procedure TSimpleThreadPoolTests.Test23_OnErrorCallbackFires; +begin + FOnErrorCount := 0; + FThreadPool.ClearErrors; + FThreadPool.OnError := @HandlePoolError; + try + FThreadPool.Queue(@RaiseTestException); + FThreadPool.Queue(@RaiseAnotherException); + FThreadPool.WaitForAll; + + AssertEquals('OnError should fire once per failed task', 2, FOnErrorCount); + finally + FThreadPool.OnError := nil; // avoid dangling callback after this test + end; +end; + +procedure TSimpleThreadPoolTests.Test24_ClearErrorsResetsCollection; +begin + FThreadPool.Queue(@RaiseTestException); + FThreadPool.WaitForAll; + AssertTrue('Precondition: an error was captured', FThreadPool.ErrorCount > 0); + + FThreadPool.ClearErrors; + AssertEquals('ClearErrors should empty the collection', 0, FThreadPool.ErrorCount); + AssertEquals('ClearErrors should also clear LastError', '', FThreadPool.LastError); +end; + +procedure TSimpleThreadPoolTests.Test25_ErrorCollectionIsCapped; +var + I: Integer; +begin + // Queue more failing tasks than the cap so the oldest entries are dropped; + // ErrorCount must never exceed MAX_STORED_ERRORS regardless of how many fail. + FThreadPool.ClearErrors; + for I := 1 to MAX_STORED_ERRORS + 250 do + FThreadPool.Queue(@RaiseAnotherException); + FThreadPool.WaitForAll; + + AssertEquals('Error collection should be capped at MAX_STORED_ERRORS', + MAX_STORED_ERRORS, FThreadPool.ErrorCount); +end; + +procedure TSimpleThreadPoolTests.Test26_LastErrorStillWorks; +begin + // Back-compat: LastError keeps reflecting the most recent error even with the + // new collection in place. + FThreadPool.ClearErrors; + FThreadPool.Queue(@RaiseTestException); + FThreadPool.WaitForAll; + + AssertEquals('LastError should hold the most recent error message', + 'Test exception message', FThreadPool.LastError); + AssertTrue('Errors collection should also contain it', + FThreadPool.ErrorCount >= 1); +end; + +procedure TSimpleThreadPoolTests.HandlePoolError(const AMessage: string); +begin + // Fired from a worker thread — increment atomically. + InterlockedIncrement(FOnErrorCount); +end; + procedure TSimpleThreadPoolTests.IncrementCounter; begin FCS.Enter; From 11d5690f409022cfecccf9e3747c2453d587cc77 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah <iwan.kelaiah@gmail.com> Date: Sat, 13 Jun 2026 11:49:46 +1000 Subject: [PATCH 2/5] chore(repo): stop tracking .claude/settings.local.json This is per-developer Claude Code config that was accidentally committed. Add it to .gitignore and remove it from tracking (local copy is kept). --- .claude/settings.local.json | 25 ------------------------- .gitignore | 3 +++ 2 files changed, 3 insertions(+), 25 deletions(-) delete mode 100644 .claude/settings.local.json diff --git a/.claude/settings.local.json b/.claude/settings.local.json deleted file mode 100644 index 64aa0a0..0000000 --- a/.claude/settings.local.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "permissions": { - "allow": [ - "Bash(wc -l /c/Users/iwank/Documents/github/threadpool-fp/src/*.pas)", - "Bash(grep -iE '\\\\.\\(o|ppu|a|compiled|bak|lps|res\\)$|/lib/|/backup/')", - "Bash(grep -iE 'threadpool.*\\\\.pas$|\\\\.lpk$|\\\\.lpi$|\\\\.lpr$')", - "Bash(grep -iE '\\\\.\\(o|ppu|bak\\)$|/lib/|/backup/')", - "Bash(fpc -iV)", - "Bash(grep '\\\\.lpi$')", - "Bash(lazbuild tests/TestRunner.lpi)", - "Bash(git remote *)", - "Bash(git status *)", - "Bash(git checkout *)", - "Bash(git pull *)", - "Bash(lazbuild package/lazarus/threadpool_fp.lpk)", - "Bash(./tests/TestRunner.exe -a -p --format=plain)", - "Bash(fpc -Fu../src -Futests threadpooltests.pas)", - "Read(//c/Users/iwank/Documents/github/**)", - "Bash(lazbuild examples/SimpleErrorHandling/SimpleErrorHandling.lpi)", - "Bash(./examples/SimpleErrorHandling/SimpleErrorHandling.exe)", - "Bash(lazbuild examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.lpi)", - "Bash(./examples/SimpleErrorHandlingBasic/SimpleErrorHandlingBasic.exe)" - ] - } -} diff --git a/.gitignore b/.gitignore index 2d30508..cec63fc 100644 --- a/.gitignore +++ b/.gitignore @@ -36,6 +36,9 @@ tests/lib/ .vscode/ .idea/ +# Claude Code local settings (per-developer, not shared) +.claude/settings.local.json + # lazbuild-generated local package-link cache packagefiles.xml From eed6a1cd75a73366a4a3cafa0aad503c20cf7150 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah <iwan.kelaiah@gmail.com> Date: Sat, 11 Jul 2026 09:41:53 +1000 Subject: [PATCH 3/5] feat(examples): add parallel file and URL demos Add Lazarus/FPC projects demonstrating v0.7.0 error handling for parallel file hashing and URL fetching. Move v0.5.0 release notes into docs. --- .../release-notes-v0.5.0.md | 0 .../ParallelFileHasher/ParallelFileHasher.lpi | 122 +++++++++++ .../ParallelFileHasher/ParallelFileHasher.lpr | 192 ++++++++++++++++++ .../ParallelUrlFetcher/ParallelUrlFetcher.lpi | 127 ++++++++++++ .../ParallelUrlFetcher/ParallelUrlFetcher.lpr | 167 +++++++++++++++ 5 files changed, 608 insertions(+) rename release-notes-v0.5.0.md => docs/release-notes-v0.5.0.md (100%) create mode 100644 examples/ParallelFileHasher/ParallelFileHasher.lpi create mode 100644 examples/ParallelFileHasher/ParallelFileHasher.lpr create mode 100644 examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi create mode 100644 examples/ParallelUrlFetcher/ParallelUrlFetcher.lpr diff --git a/release-notes-v0.5.0.md b/docs/release-notes-v0.5.0.md similarity index 100% rename from release-notes-v0.5.0.md rename to docs/release-notes-v0.5.0.md diff --git a/examples/ParallelFileHasher/ParallelFileHasher.lpi b/examples/ParallelFileHasher/ParallelFileHasher.lpi new file mode 100644 index 0000000..dcb18fc --- /dev/null +++ b/examples/ParallelFileHasher/ParallelFileHasher.lpi @@ -0,0 +1,122 @@ +<?xml version="1.0" encoding="UTF-8"?> +<CONFIG> + <ProjectOptions> + <Version Value="12"/> + <PathDelim Value="\"/> + <General> + <Flags> + <MainUnitHasCreateFormStatements Value="False"/> + <MainUnitHasTitleStatement Value="False"/> + <MainUnitHasScaledStatement Value="False"/> + </Flags> + <SessionStorage Value="InProjectDir"/> + <Title Value="ParallelFileHasher"/> + <UseAppBundle Value="False"/> + <ResourceType Value="res"/> + </General> + <BuildModes> + <Item Name="Default" Default="True"/> + <Item Name="Debug"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ParallelFileHasher"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <Parsing> + <SyntaxOptions> + <IncludeAssertionCode Value="True"/> + </SyntaxOptions> + </Parsing> + <CodeGeneration> + <Checks> + <IOChecks Value="True"/> + <RangeChecks Value="True"/> + <OverflowChecks Value="True"/> + <StackChecks Value="True"/> + </Checks> + <VerifyObjMethodCallValidity Value="True"/> + </CodeGeneration> + <Linking> + <Debugging> + <DebugInfoType Value="dsDwarf3"/> + <UseHeaptrc Value="True"/> + <TrashVariables Value="True"/> + <UseExternalDbgSyms Value="True"/> + </Debugging> + </Linking> + </CompilerOptions> + </Item> + <Item Name="Release"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ParallelFileHasher"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <CodeGeneration> + <SmartLinkUnit Value="True"/> + <Optimizations> + <OptimizationLevel Value="3"/> + </Optimizations> + </CodeGeneration> + <Linking> + <Debugging> + <GenerateDebugInfo Value="False"/> + <RunWithoutDebug Value="True"/> + </Debugging> + <LinkSmart Value="True"/> + </Linking> + </CompilerOptions> + </Item> + </BuildModes> + <PublishOptions> + <Version Value="2"/> + <UseFileFilters Value="True"/> + </PublishOptions> + <RunParams> + <FormatVersion Value="2"/> + </RunParams> + <Units> + <Unit> + <Filename Value="ParallelFileHasher.lpr"/> + <IsPartOfProject Value="True"/> + </Unit> + </Units> + </ProjectOptions> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ParallelFileHasher"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + </CompilerOptions> + <Debugging> + <Exceptions> + <Item> + <Name Value="EAbort"/> + </Item> + <Item> + <Name Value="ECodetoolError"/> + </Item> + <Item> + <Name Value="EFOpenError"/> + </Item> + </Exceptions> + </Debugging> +</CONFIG> diff --git a/examples/ParallelFileHasher/ParallelFileHasher.lpr b/examples/ParallelFileHasher/ParallelFileHasher.lpr new file mode 100644 index 0000000..c144a4e --- /dev/null +++ b/examples/ParallelFileHasher/ParallelFileHasher.lpr @@ -0,0 +1,192 @@ +program ParallelFileHasher; + +{ Parallel file hashing with natural error handling (v0.7.0) + ========================================================= + A realistic use of a thread pool: compute the MD5 of many files at once. + Hashing is CPU- and I/O-bound, so spreading it across worker threads is a + genuine win — unlike toy demos that just Sleep(). + + The error handling here is NOT simulated. A worker simply opens each file and + hashes it. If a file is missing, locked, or unreadable, TFileStream.Create + raises EFOpenError on the worker thread. The pool catches that real exception + and records it, so afterwards you can see exactly which files failed and why: + + * OnError — log each failure the moment it happens (worker thread) + * ErrorCount — how many files could not be hashed + * Errors — the message for every failed file + * LastError — the most recent failure + + To make the demo self-contained it creates a few sample files, then queues + those PLUS one path that does not exist — producing a real, not faked, error. + + HOW TO COMPILE AND RUN + ---------------------- + Option A — Free Pascal compiler: + fpc -Fu../../src ParallelFileHasher.lpr && ./ParallelFileHasher + + Option B — Lazarus IDE / lazbuild: + lazbuild ParallelFileHasher.lpi && ./ParallelFileHasher } + +{$mode objfpc}{$H+}{$J-} + +uses + {$IFDEF UNIX} + cthreads, // MUST be first: enables threading support on Unix/Linux + {$ENDIF} + Classes, SysUtils, SyncObjs, md5, ThreadPool.Simple; + +const + SAMPLE_COUNT = 4; + +var + { Work is addressed by index: a task receives an index into these parallel + arrays. Each task writes only its OWN slot, so the result array needs no + lock (the array is never resized while tasks run). } + Files: array of string; + Hashes: array of string; + +{ Logs failures as they happen. Runs on a worker thread, so the WriteLn is + guarded by a critical section to keep lines from interleaving. } +type + TFailureLog = class + private + FLock: TCriticalSection; + public + constructor Create; + destructor Destroy; override; + procedure OnTaskError(const AMessage: string); // assigned to Pool.OnError + end; + +constructor TFailureLog.Create; +begin + inherited Create; + FLock := TCriticalSection.Create; +end; + +destructor TFailureLog.Destroy; +begin + FLock.Free; + inherited Destroy; +end; + +procedure TFailureLog.OnTaskError(const AMessage: string); +begin + FLock.Enter; + try + WriteLn(' [OnError] ', AMessage); + finally + FLock.Leave; + end; +end; + +{ The actual work: hash one file, streaming it in chunks so large files do not + have to fit in memory. Any I/O problem raises here and the pool captures it. } +procedure HashFile(index: Integer); +var + Stream: TFileStream; + Ctx: TMD5Context; + Digest: TMD5Digest; + Buffer: array[0..65535] of Byte; + BytesRead: Integer; +begin + // A missing or locked file raises EFOpenError right here — a natural failure. + Stream := TFileStream.Create(Files[index], fmOpenRead or fmShareDenyWrite); + try + MD5Init(Ctx); + repeat + BytesRead := Stream.Read(Buffer, SizeOf(Buffer)); + if BytesRead > 0 then + MD5Update(Ctx, Buffer, BytesRead); + until BytesRead = 0; + MD5Final(Ctx, Digest); + Hashes[index] := MD5Print(Digest); + finally + Stream.Free; + end; +end; + +{ Create a handful of real files so the demo needs no external data. } +procedure CreateSampleFiles; +var + i, j: Integer; + F: TextFile; +begin + for i := 0 to SAMPLE_COUNT - 1 do + begin + AssignFile(F, Files[i]); + Rewrite(F); + try + for j := 0 to i do + WriteLn(F, 'Sample line ', j, ' for ', Files[i]); + finally + CloseFile(F); + end; + end; +end; + +procedure DeleteSampleFiles; +var + i: Integer; +begin + for i := 0 to SAMPLE_COUNT - 1 do + DeleteFile(Files[i]); +end; + +var + Pool: TSimpleThreadPool; + Log: TFailureLog; + i: Integer; + Msg: string; +begin + WriteLn('=== Parallel file hasher (v0.7.0 error handling) ==='); + WriteLn; + + // Build the work list: SAMPLE_COUNT real files, plus one that does not exist. + SetLength(Files, SAMPLE_COUNT + 1); + for i := 0 to SAMPLE_COUNT - 1 do + Files[i] := Format('hash_sample_%d.txt', [i]); + Files[SAMPLE_COUNT] := 'this_file_does_not_exist.txt'; // will fail naturally + SetLength(Hashes, Length(Files)); + + CreateSampleFiles; + + Pool := TSimpleThreadPool.Create(4); + Log := TFailureLog.Create; + try + Pool.OnError := @Log.OnTaskError; + + WriteLn('Hashing ', Length(Files), ' files across ', Pool.ThreadCount, + ' worker threads...'); + for i := 0 to High(Files) do + Pool.Queue(@HashFile, i); + + Pool.WaitForAll; + WriteLn; + + // Results: a file with an empty hash is one that failed. + WriteLn('Results:'); + for i := 0 to High(Files) do + if Hashes[i] <> '' then + WriteLn(Format(' OK %s %s', [Hashes[i], Files[i]])) + else + WriteLn(Format(' FAIL %-32s %s', ['(not hashed)', Files[i]])); + WriteLn; + + // Aggregate view from the pool's error API. + WriteLn(Format('%d file(s) failed:', [Pool.ErrorCount])); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + if Pool.ErrorCount > 0 then + WriteLn('LastError: ', Pool.LastError); + + Pool.ClearErrors; // ready for reuse + finally + Log.Free; + Pool.Free; + DeleteSampleFiles; + end; + + WriteLn; + WriteLn('Press enter to quit ...'); + ReadLn; +end. diff --git a/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi b/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi new file mode 100644 index 0000000..bbb0e89 --- /dev/null +++ b/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpi @@ -0,0 +1,127 @@ +<?xml version="1.0" encoding="UTF-8"?> +<CONFIG> + <ProjectOptions> + <Version Value="12"/> + <PathDelim Value="\"/> + <General> + <Flags> + <MainUnitHasCreateFormStatements Value="False"/> + <MainUnitHasTitleStatement Value="False"/> + <MainUnitHasScaledStatement Value="False"/> + </Flags> + <SessionStorage Value="InProjectDir"/> + <Title Value="ParallelUrlFetcher"/> + <UseAppBundle Value="False"/> + <ResourceType Value="res"/> + </General> + <BuildModes> + <Item Name="Default" Default="True"/> + <Item Name="Debug"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ParallelUrlFetcher"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <Parsing> + <SyntaxOptions> + <IncludeAssertionCode Value="True"/> + </SyntaxOptions> + </Parsing> + <CodeGeneration> + <Checks> + <IOChecks Value="True"/> + <RangeChecks Value="True"/> + <OverflowChecks Value="True"/> + <StackChecks Value="True"/> + </Checks> + <VerifyObjMethodCallValidity Value="True"/> + </CodeGeneration> + <Linking> + <Debugging> + <DebugInfoType Value="dsDwarf3"/> + <UseHeaptrc Value="True"/> + <TrashVariables Value="True"/> + <UseExternalDbgSyms Value="True"/> + </Debugging> + </Linking> + </CompilerOptions> + </Item> + <Item Name="Release"> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ParallelUrlFetcher"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + <CodeGeneration> + <SmartLinkUnit Value="True"/> + <Optimizations> + <OptimizationLevel Value="3"/> + </Optimizations> + </CodeGeneration> + <Linking> + <Debugging> + <GenerateDebugInfo Value="False"/> + <RunWithoutDebug Value="True"/> + </Debugging> + <LinkSmart Value="True"/> + </Linking> + </CompilerOptions> + </Item> + </BuildModes> + <PublishOptions> + <Version Value="2"/> + <UseFileFilters Value="True"/> + </PublishOptions> + <RunParams> + <FormatVersion Value="2"/> + </RunParams> + <RequiredPackages> + <Item> + <PackageName Value="FCL"/> + </Item> + </RequiredPackages> + <Units> + <Unit> + <Filename Value="ParallelUrlFetcher.lpr"/> + <IsPartOfProject Value="True"/> + </Unit> + </Units> + </ProjectOptions> + <CompilerOptions> + <Version Value="11"/> + <PathDelim Value="\"/> + <Target> + <Filename Value="ParallelUrlFetcher"/> + </Target> + <SearchPaths> + <IncludeFiles Value="$(ProjOutDir)"/> + <OtherUnitFiles Value="..\..\src"/> + <UnitOutputDirectory Value="lib\$(TargetCPU)-$(TargetOS)"/> + </SearchPaths> + </CompilerOptions> + <Debugging> + <Exceptions> + <Item> + <Name Value="EAbort"/> + </Item> + <Item> + <Name Value="ECodetoolError"/> + </Item> + <Item> + <Name Value="EFOpenError"/> + </Item> + </Exceptions> + </Debugging> +</CONFIG> diff --git a/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpr b/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpr new file mode 100644 index 0000000..a33de19 --- /dev/null +++ b/examples/ParallelUrlFetcher/ParallelUrlFetcher.lpr @@ -0,0 +1,167 @@ +program ParallelUrlFetcher; + +{ Parallel URL fetching with natural error handling (v0.7.0) + ========================================================= + Fetching many URLs is the classic case for a thread pool: each request spends + most of its time waiting on the network, so running them concurrently is a + large, real speed-up. + + The failures here are real network conditions, not simulated ones: + * a host that does not resolve raises ESocketError (DNS/connection failure) + * a 4xx/5xx response is detected from the status code and reported as an error + Both are caught by the pool, so afterwards you can see which URLs failed: + + * OnError — log each failure as it happens (worker thread) + * ErrorCount / Errors — the full list of failures after WaitForAll + * LastError — the most recent failure + + REQUIREMENTS + ------------ + * Internet access (the demo contacts real hosts). + * HTTPS support needs the OpenSSL libraries available at runtime; the + opensslsockets unit below wires fphttpclient up to them. + + IMPORTANT: each task uses its OWN TFPHTTPClient. A single client instance is + not safe to share across threads. + + HOW TO COMPILE AND RUN + ---------------------- + Option A — Free Pascal compiler: + fpc -Fu../../src ParallelUrlFetcher.lpr && ./ParallelUrlFetcher + + Option B — Lazarus IDE / lazbuild: + lazbuild ParallelUrlFetcher.lpi && ./ParallelUrlFetcher } + +{$mode objfpc}{$H+}{$J-} + +uses + {$IFDEF UNIX} + cthreads, // MUST be first: enables threading support on Unix/Linux + {$ENDIF} + Classes, SysUtils, SyncObjs, fphttpclient, opensslsockets, + ThreadPool.ProducerConsumer; + +var + { Indexed work: a task receives an index into these parallel arrays and writes + only its own slot, so the result array needs no lock. } + Urls: array of string; + Results: array of string; + +{ Thread-safe logger for OnError notifications (runs on worker threads). } +type + TFetchLog = class + private + FLock: TCriticalSection; + public + constructor Create; + destructor Destroy; override; + procedure OnTaskError(const AMessage: string); // assigned to Pool.OnError + end; + +constructor TFetchLog.Create; +begin + inherited Create; + FLock := TCriticalSection.Create; +end; + +destructor TFetchLog.Destroy; +begin + FLock.Free; + inherited Destroy; +end; + +procedure TFetchLog.OnTaskError(const AMessage: string); +begin + FLock.Enter; + try + WriteLn(' [OnError] ', AMessage); + finally + FLock.Leave; + end; +end; + +{ The work: GET one URL. Connection/DNS problems raise inside Get; a non-2xx + status is a real failure we surface ourselves by raising. Either way the pool + records it against this task. } +procedure FetchUrl(index: Integer); +var + Client: TFPHTTPClient; + Body: string; +begin + Client := TFPHTTPClient.Create(nil); // one client per task — never shared + try + Client.AllowRedirect := True; + Client.ConnectTimeout := 10000; // ms + Client.IOTimeout := 10000; // ms + + // A bad host or refused connection raises ESocketError here. + Body := Client.Get(Urls[index]); + + // fphttpclient does not treat 4xx/5xx as exceptions, so we decide here that + // anything outside 2xx is a failure for this program. + if (Client.ResponseStatusCode < 200) or (Client.ResponseStatusCode >= 300) then + raise Exception.CreateFmt('HTTP %d for %s', + [Client.ResponseStatusCode, Urls[index]]); + + Results[index] := Format('OK (%d) - %d bytes', + [Client.ResponseStatusCode, Length(Body)]); + finally + Client.Free; + end; +end; + +var + Pool: TProducerConsumerThreadPool; + Log: TFetchLog; + i: Integer; + Msg: string; +begin + WriteLn('=== Parallel URL fetcher (v0.7.0 error handling) ==='); + WriteLn; + + // A mix of good and bad URLs so both success and the two failure modes show. + Urls := [ + 'https://example.com', + 'https://www.iana.org/help/example-domains', + 'https://httpbin.org/status/404', // real 4xx response + 'https://no-such-host.invalid/' // real DNS failure + ]; + SetLength(Results, Length(Urls)); + + Pool := TProducerConsumerThreadPool.Create; + Log := TFetchLog.Create; + try + Pool.OnError := @Log.OnTaskError; + + WriteLn('Fetching ', Length(Urls), ' URLs across ', Pool.ThreadCount, + ' worker threads...'); + for i := 0 to High(Urls) do + Pool.Queue(@FetchUrl, i); + + Pool.WaitForAll; + WriteLn; + + WriteLn('Results:'); + for i := 0 to High(Urls) do + if Results[i] <> '' then + WriteLn(Format(' %-40s %s', [Urls[i], Results[i]])) + else + WriteLn(Format(' %-40s FAILED', [Urls[i]])); + WriteLn; + + WriteLn(Format('%d request(s) failed:', [Pool.ErrorCount])); + for Msg in Pool.Errors do + WriteLn(' - ', Msg); + if Pool.ErrorCount > 0 then + WriteLn('LastError: ', Pool.LastError); + + Pool.ClearErrors; + finally + Log.Free; + Pool.Free; + end; + + WriteLn; + WriteLn('Press enter to quit ...'); + ReadLn; +end. From e2adc7364037b615043c20b7f4837989e4f3d531 Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah <iwan.kelaiah@gmail.com> Date: Sat, 11 Jul 2026 13:33:26 +1000 Subject: [PATCH 4/5] docs(0.7.0): update related docs --- CHANGELOG.md | 8 ++++++-- docs/release-notes-v0.7.0.md | 12 +++++++++++- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 09f9168..167369a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ 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.7.0] - 2026-06-12 +## [0.7.0] - 2026-07-11 ### Added @@ -15,7 +15,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - `OnError: TThreadPoolErrorEvent` — optional callback fired (on a worker thread) each time a task raises - `MAX_STORED_ERRORS` constant (1000) caps the collection; oldest messages are dropped beyond it so a high volume of failing tasks cannot exhaust memory - 8 new unit tests covering the error-collection API across both pools (collection captures all, `OnError` fires per failure, `ClearErrors` resets, cap enforced, `LastError` back-compat) -- New examples: `examples/SimpleErrorHandlingBasic` (the easy way — poll `Errors`/`ErrorCount`/`LastError` after `WaitForAll`, no callback or locking) and `examples/SimpleErrorHandling` (advanced — adds the `OnError` callback with a thread-safe handler) +- New examples: + - `examples/SimpleErrorHandlingBasic` — the easy way: poll `Errors`/`ErrorCount`/`LastError` after `WaitForAll`, no callback or locking + - `examples/SimpleErrorHandling` — advanced error handling with the `OnError` callback and a thread-safe handler + - `examples/ParallelFileHasher` — hashes multiple files concurrently and demonstrates real file I/O failures + - `examples/ParallelUrlFetcher` — fetches multiple URLs concurrently and demonstrates real HTTP/DNS failures ### Changed diff --git a/docs/release-notes-v0.7.0.md b/docs/release-notes-v0.7.0.md index f52cc6e..2f7a4ec 100644 --- a/docs/release-notes-v0.7.0.md +++ b/docs/release-notes-v0.7.0.md @@ -70,11 +70,21 @@ The API lives in `TThreadPoolBase`, so `TSimpleThreadPool` and - `OnError` handlers run on a worker thread — synchronize any access to shared or UI state. +## Examples + +- `SimpleErrorHandlingBasic` — the easy way: inspect `Errors`, `ErrorCount`, and + `LastError` after `WaitForAll`. +- `SimpleErrorHandling` — advanced handling with an `OnError` callback. +- `ParallelFileHasher` — hashes multiple files concurrently and reports real + file I/O failures. +- `ParallelUrlFetcher` — fetches multiple URLs concurrently and reports real + HTTP/DNS failures. + ## Verification - Test suite: **43 tests, 0 errors, 0 failures, 0 unfreed memory blocks** (8 new tests cover the error-collection API on both pools). -- Package and all 8 examples build cleanly. +- Package and all 10 top-level examples build cleanly. ## Full changelog From 8066f54a1058bf36bb00ed481193098f6b3b250f Mon Sep 17 00:00:00 2001 From: Iwan Kelaiah <iwan.kelaiah@gmail.com> Date: Sat, 11 Jul 2026 15:01:40 +1000 Subject: [PATCH 5/5] docs(0.7.0): update README.md --- README.md | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 56d7549..f58023a 100644 --- a/README.md +++ b/README.md @@ -399,25 +399,35 @@ All four `Queue` overloads share the same pattern — pick the one that fits you - Adds the `OnError` callback to react *while* tasks run - Shows the thread-safe handler pattern (needed only because the handler keeps shared state) +7. 📁 **Parallel File Hasher** (`examples/ParallelFileHasher/ParallelFileHasher.lpr`) + - Hashes multiple files concurrently with `ThreadPool.Simple` + - Demonstrates real file I/O failures captured through `Errors`/`OnError` + - Creates and cleans up its own sample files + ### Producer-Consumer Examples -5. 🎓 **Simple Demo** (`examples/ProdConSimpleDemo/ProdConSimpleDemo.lpr`) +1. 🎓 **Simple Demo** (`examples/ProdConSimpleDemo/ProdConSimpleDemo.lpr`) - Basic usage with ProducerConsumerThreadPool - Demonstrates procedures and methods - Shows proper object lifetime -6. 🔢 **Square Numbers** (`examples/ProdConSquareNumbers/ProdConSquareNumbers.lpr`) +2. 🔢 **Square Numbers** (`examples/ProdConSquareNumbers/ProdConSquareNumbers.lpr`) - High volume task processing - Queue full handling - Backpressure demonstration - Performance monitoring -7. 📝 **Message Processor** (`examples/ProdConMessageProcessor/ProdConMessageProcessor.lpr`) +3. 📝 **Message Processor** (`examples/ProdConMessageProcessor/ProdConMessageProcessor.lpr`) - Queue-based task processing - Thread-safe message handling - Graceful shutdown - Error handling patterns +4. 🌐 **Parallel URL Fetcher** (`examples/ParallelUrlFetcher/ParallelUrlFetcher.lpr`) + - Fetches multiple URLs concurrently with `ThreadPool.ProducerConsumer` + - Demonstrates real HTTP/DNS failures captured through `Errors`/`OnError` + - Uses one HTTP client per task so clients are not shared across threads + ## 🛠️ Installation