From dfdd6df7d18c67cbcd617f85e85ef6cfda280d0f Mon Sep 17 00:00:00 2001 From: Hal <22532613+HalRiveria@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:31:20 -0400 Subject: [PATCH 1/2] Say which runspace job crashed, and report crashes during an abort -JobCrashedHandler only got an error record pointing at the kit's own EndInvoke line, with nothing identifying the job, so a caller couldn't log or retry the input that failed. A worker that threw while an abort waited for it wasn't counted or reported to the handler at all. And a handler that threw was swallowed without a word. Crash handling now goes through one helper, used by both the main loop and the abort drain. It counts the crash and adds it, with its job and error, to a new Crashed list on the return object. It announces the crash, then calls the handler. A handler that declares a second parameter, or has no param() block, also gets the job hashtable. A one-parameter handler is called as before, since one with [CmdletBinding()] would throw on an extra argument. A handler that throws is now reported as a message. The README now says what the handler gets, what counts as a crash (a non-terminating error isn't one), that workers start with default preferences, and how to return an error object from a worker. It also says not to call exit in one. Co-Authored-By: Claude Opus 5.5 --- ParallelKit.Harness.ps1 | 38 +++++++++++++++++++- ParallelKit.ps1 | 78 +++++++++++++++++++++++++++++++++-------- README.md | 15 ++++++-- 3 files changed, 113 insertions(+), 18 deletions(-) diff --git a/ParallelKit.Harness.ps1 b/ParallelKit.Harness.ps1 index e188046..a46725c 100644 --- a/ParallelKit.Harness.ps1 +++ b/ParallelKit.Harness.ps1 @@ -128,6 +128,42 @@ Test-Assert ($out.CrashedCount -eq 1 -and $crashes.Count -eq 1) "a worker that t Test-Assert ($results.Count -eq 2) "the other workers still return their results" Test-Assert (@($script:messages | Where-Object { $_.Level -eq 'Error' -and $_.Message -like '*boom*' }).Count -eq 1) "the crash is reported as an Error message" +# Which job crashed: on the return object, and to a handler that asks for it. +$crashedProp = $out.PSObject.Properties['Crashed'] # read safely: strict mode throws on a missing property +Test-Assert ($null -ne $crashedProp -and @($crashedProp.Value).Count -eq 1 -and $crashedProp.Value[0].Job['Id'] -eq 2 -and $crashedProp.Value[0].Error -is [System.Management.Automation.ErrorRecord]) "the return object lists the crashed job and its error" +$crashWorker = { param([int]$Id) if ($Id -eq 2) { throw "boom" }; return $Id } +$seenJobs = [System.Collections.Generic.List[object]]::new() +$null = Invoke-PkRunspaceJobs -WorkerScript $crashWorker -JobParams $jobs -MaxParallel 2 -JobCrashedHandler { param($e, $job) $seenJobs.Add($job) }.GetNewClosure() +Test-Assert ($seenJobs.Count -eq 1 -and $null -ne $seenJobs[0] -and $seenJobs[0]['Id'] -eq 2) "a -JobCrashedHandler with a second parameter gets the job that crashed" +$seenArgs = @{ Count = -1; Id = $null } +$null = Invoke-PkRunspaceJobs -WorkerScript $crashWorker -JobParams $jobs -MaxParallel 2 -JobCrashedHandler { $seenArgs.Count = $args.Count; $seenArgs.Id = $args[1]['Id'] }.GetNewClosure() +Test-Assert ($seenArgs.Count -eq 2 -and $seenArgs.Id -eq 2) ("a -JobCrashedHandler with no param() gets the error and the job in `$args (count " + $seenArgs.Count + ")") +# One-parameter handlers that would throw on a second argument still work. +$advanced = @{ Calls = 0 } +$null = Invoke-PkRunspaceJobs -WorkerScript $crashWorker -JobParams $jobs -MaxParallel 2 -JobCrashedHandler { [CmdletBinding()] param($e) $advanced.Calls++ }.GetNewClosure() +Test-Assert ($advanced.Calls -eq 1) "a one-parameter [CmdletBinding()] -JobCrashedHandler is still called" +function Receive-HarnessCrash($e, $job) { $script:harnessCrashId = $job['Id'] } +$script:harnessCrashId = $null +$null = Invoke-PkRunspaceJobs -WorkerScript $crashWorker -JobParams $jobs -MaxParallel 2 -JobCrashedHandler ${function:Receive-HarnessCrash} +Test-Assert ($script:harnessCrashId -eq 2) "a function used as -JobCrashedHandler, with function f(`$e, `$job) parameters, gets the job" +# A handler that throws is reported, not swallowed. +Reset-Harness +$null = Invoke-PkRunspaceJobs -WorkerScript $crashWorker -JobParams $jobs -MaxParallel 2 -JobCrashedHandler { param($e) throw "handler bug" } +Test-Assert (@($script:messages | Where-Object { $_.Message -like '*-JobCrashedHandler threw*handler bug*' }).Count -eq 1) "a -JobCrashedHandler that throws is reported as a message" +# Workers that throw while an abort waits for them count as crashes too. +Reset-Harness +$poll = @{ Count = 0 } +$drainCrashes = [System.Collections.Generic.List[object]]::new() +$drainOut = Invoke-PkRunspaceJobs -MaxParallel 2 -JobParams (New-JobList -Jobs @(@{ Id = 1 }, @{ Id = 2 })) -WorkerScript { + param([int]$Id) + for ($i = 0; $i -lt 100; $i++) { if (Test-PkCancelRequested) { break }; [System.Threading.Thread]::Sleep(50) } + [System.Threading.Thread]::Sleep(500) + throw ("drain boom " + $Id) +} -PollCallback { $poll.Count++; if ($poll.Count -eq 3) { Request-PkAbort } }.GetNewClosure() -JobCrashedHandler { param($e, $job) $drainCrashes.Add($job['Id']) }.GetNewClosure() +$drainCrashedProp = $drainOut.PSObject.Properties['Crashed'] +Test-Assert ($drainOut.Aborted -and $drainOut.CrashedCount -eq 2 -and $null -ne $drainCrashedProp -and @($drainCrashedProp.Value).Count -eq 2 -and $drainCrashes.Count -eq 2) ("workers that throw during an abort's drain are counted and reported (counted " + $drainOut.CrashedCount + ", handler calls " + $drainCrashes.Count + ")") +Reset-Harness + $threw = $false try { $null = Invoke-PkRunspaceJobs -WorkerScript { param($Id) } -JobParams (New-JobList -Jobs @(@{ Id = 1 })) -MaxParallel 1 -SharedFunctionNames @('No-SuchFunction') } catch { $threw = ($_.ToString() -like "*No-SuchFunction*") } Test-Assert $threw "an unknown -SharedFunctionNames entry fails up front with the name in the message" @@ -1304,7 +1340,7 @@ Test-Assert ((Get-HarnessShape $shapeOne) -eq 'ExitCode,Stdout,Stderr,StreamTime $shapeDispatch = Invoke-PkProcessDispatch -MaxConcurrent 1 -GroupProvider { return $null } -JobBuilder { param($g) } -JobFinalize { param($j, $p) } Test-Assert ((Get-HarnessShape $shapeDispatch) -eq 'GroupResults,Aborted,ForceKilledCount,JobStarterFailedCount') ("the dispatcher returns an object (" + (Get-HarnessShape $shapeDispatch) + ")") $shapeJobs = Invoke-PkRunspaceJobs -MaxParallel 1 -JobParams @{ Id = 1 } -WorkerScript { param($Id) } -Test-Assert ((Get-HarnessShape $shapeJobs) -eq 'ForceStoppedCount,CrashedCount,Aborted') ("runspace jobs return an object (" + (Get-HarnessShape $shapeJobs) + ")") +Test-Assert ((Get-HarnessShape $shapeJobs) -eq 'ForceStoppedCount,CrashedCount,Aborted,Crashed') ("runspace jobs return an object (" + (Get-HarnessShape $shapeJobs) + ")") $shapeFindings = Test-PkWorkerSafety -WorkerScript { param($Id) $null = [Func[int]]{ 1 } } $shapeFinding = $shapeFindings[0] Test-Assert ((Get-HarnessShape $shapeFinding) -eq 'Rule,Where,Line,Text') ("worker-safety findings are objects (" + (Get-HarnessShape $shapeFinding) + ")") diff --git a/ParallelKit.ps1 b/ParallelKit.ps1 index 5d0b142..c38f51b 100644 --- a/ParallelKit.ps1 +++ b/ParallelKit.ps1 @@ -1730,11 +1730,54 @@ function Get-PkWorkerSafeFunctionNames { 'Register-PkCancellationSource', 'Unregister-PkCancellationSource', 'New-PkBatchResult', 'ConvertTo-PkArgumentString', 'Get-PkForceStopReason', 'Resolve-PkFileSystemPath') } +function Get-PkHandlerParamCount { + # How many parameters a callback declares, or -1 if it has no param() + # block (it takes whatever it's given, in $args). + param([Parameter(Mandatory)][scriptblock]$Callback) + $ast = $Callback.Ast + if ($ast -is [System.Management.Automation.Language.FunctionDefinitionAst]) { + if ($null -ne $ast.Parameters) { return $ast.Parameters.Count } # function f($a, $b) + $ast = $ast.Body + } + if ($null -eq $ast.ParamBlock) { return -1 } + return $ast.ParamBlock.Parameters.Count +} + +function Add-PkCrashedJob { + # A worker threw: count it, record it with its job on the run's outcome, + # announce it, and tell -JobCrashedHandler. Shared by the main loop and + # the abort drain. pk-prefixed parameters: a handler built without + # .GetNewClosure() can see them. + param( + [Parameter(Mandatory)][System.Collections.Specialized.OrderedDictionary]$PkOutcome, + [hashtable]$PkJob, + [Parameter(Mandatory)][System.Management.Automation.ErrorRecord]$PkError, + [scriptblock]$PkHandler, + [string]$PkWhen = "" + ) + $PkOutcome.CrashedCount++ + $PkOutcome.Crashed.Add([pscustomobject][ordered]@{ Job = $PkJob; Error = $PkError }) + Invoke-PkMessageHandler -Level Error -Message ("Error collecting runspace result" + $PkWhen + ": " + $PkError) + if ($null -eq $PkHandler) { return } + try { + # The job only goes to a handler that has room for it: a one-parameter + # handler with [CmdletBinding()] or [Parameter()] would throw on a + # second argument. + $pkParamCount = Get-PkHandlerParamCount -Callback $PkHandler + if ($pkParamCount -eq -1 -or $pkParamCount -ge 2) { $null = & $PkHandler $PkError $PkJob } + else { $null = & $PkHandler $PkError } + } catch { + # Reported, not rethrown: a broken crash handler mustn't take the other workers down. + Invoke-PkMessageHandler -Level Error -Message ("-JobCrashedHandler threw: " + $_) + } +} + function Receive-PkCompletedHandles { param( [System.Collections.Generic.List[hashtable]]$Handles, [scriptblock]$ResultHandler, - [scriptblock]$ErrorHandler + [System.Collections.Specialized.OrderedDictionary]$Outcome, + [scriptblock]$JobCrashedHandler ) $stillActive = [System.Collections.Generic.List[hashtable]]::new() foreach ($h in $Handles) { @@ -1747,7 +1790,7 @@ function Receive-PkCompletedHandles { try { $pkWorkerOutput = $h.PS.EndInvoke($h.Handle) } catch { - if ($null -ne $ErrorHandler) { $null = & $ErrorHandler ("Error collecting runspace result during abort drain: " + $_) } + Add-PkCrashedJob -PkOutcome $Outcome -PkJob $h['Job'] -PkError $_ -PkHandler $JobCrashedHandler -PkWhen " during abort drain" } finally { $h.PS.Dispose() } @@ -1773,7 +1816,9 @@ function Invoke-PkGracefulDrain { [scriptblock]$ResultHandler, [System.Collections.Concurrent.ConcurrentQueue[object]]$MessageQueue, [int]$TimeoutSeconds = 60, - [string]$AbortWaitMessage + [string]$AbortWaitMessage, + [Parameter(Mandatory)][System.Collections.Specialized.OrderedDictionary]$Outcome, + [scriptblock]$JobCrashedHandler ) if ($Handles.Count -eq 0) { @@ -1784,7 +1829,6 @@ function Invoke-PkGracefulDrain { $AbortWaitMessage = "[ABORT] Waiting for current jobs to finish (up to " + $TimeoutSeconds + "s)..." } - $errorHandler = { param($msg) Invoke-PkMessageHandler -Message $msg -Level Error } $localHandles = $Handles $abortDeadline = [System.Diagnostics.Stopwatch]::StartNew() $forceStoppedCount = 0 @@ -1792,14 +1836,14 @@ function Invoke-PkGracefulDrain { if ($AbortWaitMessage -ne "") { Invoke-PkMessageHandler -Message $AbortWaitMessage -Level Warn } while ($localHandles.Count -gt 0 -and -not [PkConsoleCtrl]::IsCloseEvent -and -not [PkConsoleCtrl]::SkipDrain -and $abortDeadline.Elapsed.TotalSeconds -lt $TimeoutSeconds) { - $localHandles = Receive-PkCompletedHandles -Handles $localHandles -ResultHandler $ResultHandler -ErrorHandler $errorHandler + $localHandles = Receive-PkCompletedHandles -Handles $localHandles -ResultHandler $ResultHandler -Outcome $Outcome -JobCrashedHandler $JobCrashedHandler Receive-PkWorkerMessages -Queue $MessageQueue if ($localHandles.Count -gt 0) { [System.Threading.Thread]::Sleep(250) } } # One final pass before declaring anything force-stopped. if ($localHandles.Count -gt 0) { - $localHandles = Receive-PkCompletedHandles -Handles $localHandles -ResultHandler $ResultHandler -ErrorHandler $errorHandler + $localHandles = Receive-PkCompletedHandles -Handles $localHandles -ResultHandler $ResultHandler -Outcome $Outcome -JobCrashedHandler $JobCrashedHandler Receive-PkWorkerMessages -Queue $MessageQueue } @@ -1828,6 +1872,8 @@ function Invoke-PkRunspaceJobs { # functions, so name every function of yours it calls in -SharedFunctionNames. # Each job starts in your current location, in a fresh scope: a folder change # or a variable from an earlier job doesn't carry over. + # Only an error that stops the worker (throw, -ErrorAction Stop) counts as a + # crash; a non-terminating one is printed and the job's output still counts. # The worker code is checked against the worker rules (see the README) # before anything starts, and each job's values before that job starts; # code or values that break them are refused. @@ -1851,7 +1897,11 @@ function Invoke-PkRunspaceJobs { # .PARAMETER PollCallback # { } on the main thread, every loop pass (for progress and the like). # .PARAMETER JobCrashedHandler - # { param($errorRecord) } on the main thread when a worker throws. + # { param($errorRecord, $job) } on the main thread when a worker throws, also + # during an abort's drain. $job is the job hashtable that crashed; leave it out + # of param() if you don't need it. The worker's own error is + # $errorRecord.Exception.InnerException. If the handler throws, that's reported + # as a message and the run carries on. # .PARAMETER DrainTimeoutSeconds # How long an abort waits for running workers before force-stopping them # (default 60). @@ -1861,7 +1911,8 @@ function Invoke-PkRunspaceJobs { # Check the worker code and job values before starting (default on). # -CheckWorkerSafety:$false turns it off. # .OUTPUTS - # An object with ForceStoppedCount, CrashedCount and Aborted. + # An object with ForceStoppedCount, CrashedCount, Aborted and Crashed (one + # @{ Job; Error } entry per worker that threw). # .EXAMPLE # Invoke-PkRunspaceJobs -MaxParallel 4 -JobParams @(@{ Id = 1 }, @{ Id = 2 }) -WorkerScript { param($Id) $Id * 2 } -ResultHandler { param($r) Write-Host $r } param( @@ -1908,7 +1959,7 @@ function Invoke-PkRunspaceJobs { $JobParams = $normalizedJobs } - $outcome = [ordered]@{ ForceStoppedCount = 0; CrashedCount = 0; Aborted = $false } + $outcome = [ordered]@{ ForceStoppedCount = 0; CrashedCount = 0; Aborted = $false; Crashed = [System.Collections.Generic.List[object]]::new() } if ($null -ne $JobParams -and $JobParams.Count -eq 0) { return [pscustomobject]$outcome } if ($CheckWorkerSafety) { @@ -2048,7 +2099,7 @@ function Invoke-PkRunspaceJobs { # A local scope too, so variables a job sets (preference # variables included) don't carry over to the next job. $null = $ps.AddScript($jobScript, $true).AddParameter('PkJobLocation', $jobLocation).AddParameter('PkJobValues', $jobValues) - $handles.Add(@{ PS = $ps; Handle = $ps.BeginInvoke() }) + $handles.Add(@{ PS = $ps; Handle = $ps.BeginInvoke(); Job = $nextJob }) } $null = Test-PkAbortSignal @@ -2071,10 +2122,7 @@ function Invoke-PkRunspaceJobs { } catch { # The worker itself threw - there is no result, so # -ResultHandler is never called for this job. - $outcome.CrashedCount++ - Invoke-PkMessageHandler -Message ("Error collecting runspace result: " + $_) -Level Error - # Swallowed: a broken crash handler must not take the other workers down with it. - if ($null -ne $JobCrashedHandler) { try { $null = & $JobCrashedHandler $_ } catch { } } + Add-PkCrashedJob -PkOutcome $outcome -PkJob $h['Job'] -PkError $_ -PkHandler $JobCrashedHandler } finally { $h.PS.Dispose() } @@ -2090,7 +2138,7 @@ function Invoke-PkRunspaceJobs { } if ($Cts.Token.IsCancellationRequested) { - $drainArgs = @{ Handles = $handles; ResultHandler = $ResultHandler; MessageQueue = $messageQueue; TimeoutSeconds = $DrainTimeoutSeconds } + $drainArgs = @{ Handles = $handles; ResultHandler = $ResultHandler; MessageQueue = $messageQueue; TimeoutSeconds = $DrainTimeoutSeconds; Outcome = $outcome; JobCrashedHandler = $JobCrashedHandler } if ($PSBoundParameters.ContainsKey('AbortWaitMessage')) { $drainArgs.AbortWaitMessage = $AbortWaitMessage } $drainResult = Invoke-PkGracefulDrain @drainArgs $outcome.ForceStoppedCount = $drainResult.ForceStoppedCount diff --git a/README.md b/README.md index 7ba7e2d..8a9fc4f 100644 --- a/README.md +++ b/README.md @@ -184,10 +184,21 @@ Each worker runs in its own runspace: a separate PowerShell session inside the s - Whatever the worker returns is handed to `-ResultHandler` on the main thread, once per object returned. - `-NextJobProvider { ... }` can replace `-JobParams` to produce jobs lazily: it's called on the main thread for the next job hashtable, and returns `$null` when there are no more. - `-PollCallback { ... }` runs on the main thread on every loop pass, for progress display and the like. -- `-JobCrashedHandler { param($err) ... }` is told when a worker throws. That job's result handler isn't called. +- `-JobCrashedHandler { param($err, $job) ... }` is told when a worker throws, including while an abort waits for running workers, with the job hashtable that crashed. Leave `$job` out of `param()` if you don't need it. `$err.Exception.InnerException` is the worker's own error. That job's result handler isn't called, and anything the worker output before throwing is lost. If the handler itself throws, that's reported as a message and the run carries on. +- Only an error that stops the worker counts as a crash: `throw`, a cmdlet with `-ErrorAction Stop`, or any error once the worker sets `$ErrorActionPreference = 'Stop'`. A non-terminating error (a locked file in `Get-FileHash`, a `Write-Error`) is only printed as `Runspace error: ...`, and whatever the worker returned still goes to `-ResultHandler` as if it had worked. Workers start with PowerShell's default preferences and no strict mode, whatever your script uses. To catch every failure and know which input it came from, have the worker return its own error object: + + ```powershell + WorkerScript = { + param($Path) + $ErrorActionPreference = 'Stop' + try { [pscustomobject]@{ Path = $Path; Hash = (Get-FileHash -LiteralPath $Path).Hash; Error = $null } } + catch { [pscustomobject]@{ Path = $Path; Hash = $null; Error = $_.Exception.Message } } + } + ``` +- Don't call `exit` in a worker. It ends that job quietly (not as a crash) and sets your whole script's exit code. - `-MaxParallel` has no default; you choose it. `-DrainTimeoutSeconds` (how long Ctrl+C waits for running workers) defaults to 60. - If `-ResultHandler`, `-PollCallback` or `-NextJobProvider` throws, the run stops: workers still going are told to stop and the exception comes out of the call. -- Returns an object with `ForceStoppedCount`, `CrashedCount` and `Aborted`. +- Returns an object with `ForceStoppedCount`, `CrashedCount`, `Aborted` and `Crashed`, which lists the `Job` and `Error` of each worker that threw. ## The two rules for worker code From d4ad93aa3da640941724935395e6e62a178f1260 Mon Sep 17 00:00:00 2001 From: Hal <22532613+HalRiveria@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:32:10 -0400 Subject: [PATCH 2/2] README: a .NET method exception in a worker isn't a crash either Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 8a9fc4f..a6fb957 100644 --- a/README.md +++ b/README.md @@ -185,7 +185,7 @@ Each worker runs in its own runspace: a separate PowerShell session inside the s - `-NextJobProvider { ... }` can replace `-JobParams` to produce jobs lazily: it's called on the main thread for the next job hashtable, and returns `$null` when there are no more. - `-PollCallback { ... }` runs on the main thread on every loop pass, for progress display and the like. - `-JobCrashedHandler { param($err, $job) ... }` is told when a worker throws, including while an abort waits for running workers, with the job hashtable that crashed. Leave `$job` out of `param()` if you don't need it. `$err.Exception.InnerException` is the worker's own error. That job's result handler isn't called, and anything the worker output before throwing is lost. If the handler itself throws, that's reported as a message and the run carries on. -- Only an error that stops the worker counts as a crash: `throw`, a cmdlet with `-ErrorAction Stop`, or any error once the worker sets `$ErrorActionPreference = 'Stop'`. A non-terminating error (a locked file in `Get-FileHash`, a `Write-Error`) is only printed as `Runspace error: ...`, and whatever the worker returned still goes to `-ResultHandler` as if it had worked. Workers start with PowerShell's default preferences and no strict mode, whatever your script uses. To catch every failure and know which input it came from, have the worker return its own error object: +- Only an error that stops the worker counts as a crash: `throw`, a cmdlet with `-ErrorAction Stop`, or any error once the worker sets `$ErrorActionPreference = 'Stop'`. A non-terminating error (a locked file in `Get-FileHash`, a `Write-Error`), or an exception from a .NET method such as `[System.IO.File]::ReadAllText(...)`, only ends that one statement. It's printed as `Runspace error: ...`, and whatever the worker returned still goes to `-ResultHandler` as if it had worked. Workers start with PowerShell's default preferences and no strict mode, whatever your script uses. To catch every failure and know which input it came from, have the worker return its own error object: ```powershell WorkerScript = {