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..a6fb957 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`), 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 = { + 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