diff --git a/ParallelKit.Harness.ps1 b/ParallelKit.Harness.ps1 index 075c745..ed56e9c 100644 --- a/ParallelKit.Harness.ps1 +++ b/ParallelKit.Harness.ps1 @@ -2357,6 +2357,168 @@ Test-Assert ($workerWait.Count -eq 1 -and $workerWait[0] -eq 2) "a batch inside Reset-Harness Write-Host "" +# ============================================================================== +Write-Host "=== AC: the dispatcher's per-job time limit ===" -ForegroundColor Cyan + +function Invoke-LimitGroup { + # One group of jobs through the dispatcher. Logs each finalized job. + param([hashtable[]]$Jobs, [hashtable]$Extra = @{}) + $log = [System.Collections.Generic.List[object]]::new() + $feed = @{ Given = $false } + $sw = [System.Diagnostics.Stopwatch]::StartNew() + $dispatchParams = @{ + MaxConcurrent = 6; AbortTimeoutSeconds = 30 + GroupProvider = { if ($feed.Given) { return $null }; $feed.Given = $true; return 'g' }.GetNewClosure() + JobBuilder = { param($g) return $Jobs }.GetNewClosure() + JobFinalize = { + param($job, $q) + $done = $null + if ($job['Proc'] -is [System.Diagnostics.Process]) { $done = Complete-PkProcess -Job $job } + $exit = $null + if ($null -ne $done) { $exit = $done.ExitCode } + $log.Add([pscustomobject]@{ Name = $job['Name']; TimedOut = $job['TimedOut']; ExitCode = $exit; At = $sw.Elapsed.TotalSeconds }) + if ($job['Retry'] -gt 0) { $job['Retry']--; $q.Enqueue($job) } + }.GetNewClosure() + } + foreach ($k in $Extra.Keys) { $dispatchParams[$k] = $Extra[$k] } + $result = $null + $problem = $null + try { $result = Invoke-PkProcessDispatch @dispatchParams } catch { $problem = [string]$_ } + return @{ Log = $log; Result = $result; Problem = $problem; Secs = $sw.Elapsed.TotalSeconds } +} +function Get-LimitEntry { param($Run, [string]$Name) return ,@($Run.Log | Where-Object { $_.Name -eq $Name }) } +function Get-LimitPingCount { param([int]$N) return @(Get-CimInstance Win32_Process -Filter "Name = 'PING.EXE'" | Where-Object { $_.CommandLine -like ("*-n " + $N + " 127.0.0.1*") }).Count } + +# -TimeoutSeconds, and a job's own key overriding it in both directions. +$limitRun = Invoke-LimitGroup -Extra @{ TimeoutSeconds = 2 } -Jobs @( + @{ Name = 'slow'; FilePath = $cmdExe; Arguments = ('/c ""' + $pingExe + '" -n 37 127.0.0.1"') }, + @{ Name = 'fast'; FilePath = $cmdExe; Arguments = '/c exit 0' }, + @{ Name = 'optout'; FilePath = $pingExe; Arguments = '-n 4 127.0.0.1'; TimeoutSeconds = 0 }, + @{ Name = 'shorter'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1'; TimeoutSeconds = 0.5 }) +$slow = Get-LimitEntry $limitRun 'slow'; $fast = Get-LimitEntry $limitRun 'fast'; $optout = Get-LimitEntry $limitRun 'optout'; $shorter = Get-LimitEntry $limitRun 'shorter' +Test-Assert ($null -eq $limitRun.Problem -and $slow.Count -eq 1 -and $slow[0].TimedOut -eq $true -and $slow[0].ExitCode -ne 0 -and $slow[0].At -ge 2 -and $slow[0].At -lt 8) ("-TimeoutSeconds kills an overrunning job, and -JobFinalize sees TimedOut (problem: " + $limitRun.Problem + ")") +Test-Assert ((Get-LimitPingCount 37) -eq 0) "and what it started is killed with it" +Test-Assert ($fast.Count -eq 1 -and $fast[0].TimedOut -eq $false -and $fast[0].ExitCode -eq 0) "a job that finishes in time has TimedOut = `$false" +Test-Assert ($optout.Count -eq 1 -and $optout[0].TimedOut -eq $false -and $optout[0].ExitCode -eq 0 -and $optout[0].At -ge 2.5) "a job's TimeoutSeconds = 0 opts out of -TimeoutSeconds" +Test-Assert ($shorter.Count -eq 1 -and $shorter[0].TimedOut -eq $true -and $shorter[0].At -lt 1.9) "a job's own TimeoutSeconds beats -TimeoutSeconds, and can be fractional" + +# A job's key with no -TimeoutSeconds; a job without the key isn't limited. +$limitRun = Invoke-LimitGroup -Jobs @( + @{ Name = 'limited'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1'; TimeoutSeconds = 1 }, + @{ Name = 'free'; FilePath = $pingExe; Arguments = '-n 3 127.0.0.1' }) +$limited = Get-LimitEntry $limitRun 'limited'; $free = Get-LimitEntry $limitRun 'free' +Test-Assert ($limited.Count -eq 1 -and $limited[0].TimedOut -eq $true -and $limited[0].At -lt 5) ("a job's TimeoutSeconds works without -TimeoutSeconds (problem: " + $limitRun.Problem + ")") +Test-Assert ($free.Count -eq 1 -and $free[0].TimedOut -eq $false -and $free[0].ExitCode -eq 0) "and a job without the key runs to its end" + +# Still in force while an abort waits for running jobs. +$limitCts = [System.Threading.CancellationTokenSource]::new() +$limitCts.CancelAfter(500) +$limitCleaned = [System.Collections.Generic.List[object]]::new() +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'drain'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1' }) -Extra @{ TimeoutSeconds = 2; CancelToken = $limitCts.Token; JobAbortCleanup = { param($job) $limitCleaned.Add($job['Name']) }.GetNewClosure() } +$drain = Get-LimitEntry $limitRun 'drain' +Test-Assert ($null -ne $limitRun.Result -and $limitRun.Result.Aborted -and $limitRun.Result.ForceKilledCount -eq 0 -and $drain.Count -eq 1 -and $drain[0].TimedOut -eq $true -and $limitCleaned.Count -eq 0 -and $limitRun.Secs -lt 10) ("a limit fires during an abort's wait, and the job is finalized (" + [int]$limitRun.Secs + "s, problem: " + $limitRun.Problem + ")") +$limitCts.Dispose() + +# A starter that sets no Stopwatch is still timed. +$noWatchStarter = { param($job) $s = Start-PkProcess -FilePath $job['FilePath'] -Arguments $job['Arguments']; $job['Proc'] = $s.Proc; $job['StdoutTask'] = $s.StdoutTask; $job['StderrTask'] = $s.StderrTask; return $job } +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'nowatch'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1' }) -Extra @{ TimeoutSeconds = 1; JobStarter = $noWatchStarter } +$noWatch = Get-LimitEntry $limitRun 'nowatch' +Test-Assert ($noWatch.Count -eq 1 -and $noWatch[0].TimedOut -eq $true -and $noWatch[0].At -lt 5) ("a starter with no Stopwatch is still timed (problem: " + $limitRun.Problem + ")") + +# A job with no real process is left to its own -JobPollCompleted. +$customClock = [System.Diagnostics.Stopwatch]::StartNew() +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'custom' }) -Extra @{ + TimeoutSeconds = 1 + JobStarter = { param($job) $job['Proc'] = 'not a process'; return $job } + JobPollCompleted = { param($job) return ($customClock.Elapsed.TotalSeconds -ge 1.8) }.GetNewClosure() +} +$custom = Get-LimitEntry $limitRun 'custom' +Test-Assert ($null -eq $limitRun.Problem -and $null -ne $limitRun.Result -and $custom.Count -eq 1 -and $custom[0].TimedOut -eq $false -and $custom[0].At -ge 1.7) ("a job whose .Proc isn't a process isn't timed (problem: " + $limitRun.Problem + ")") + +# A program that can't be stopped gets one try and one warning, not a taskkill +# on every pass. A Process that was never started can't be stopped or read. +Reset-Harness +$stuckClock = [System.Diagnostics.Stopwatch]::StartNew() +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'stuck' }) -Extra @{ + TimeoutSeconds = 0.2 + JobStarter = { param($job) $job['Proc'] = [System.Diagnostics.Process]::new(); return $job } + JobPollCompleted = { param($job) return ($stuckClock.Elapsed.TotalSeconds -ge 1) }.GetNewClosure() + JobFinalize = { param($job, $q) } +} +$stuckWarnings = @($script:messages | Where-Object { $_.Level -eq 'Warn' -and $_.Message -like "*couldn't be stopped*" }).Count +Test-Assert ($null -eq $limitRun.Problem -and $null -ne $limitRun.Result -and $stuckWarnings -eq 1) ("a program that can't be stopped gets one warning and the run carries on (warnings " + $stuckWarnings + ", problem: " + $limitRun.Problem + ")") +Reset-Harness + +# Fractional, and under 0.5 s: not rounded to 0, which would mean no limit. +$limitRun = Invoke-LimitGroup -Extra @{ TimeoutSeconds = 0.4 } -Jobs @(@{ Name = 'fraction'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1' }) +$fraction = Get-LimitEntry $limitRun 'fraction' +Test-Assert ($fraction.Count -eq 1 -and $fraction[0].TimedOut -eq $true -and $fraction[0].At -lt 5) "a fractional -TimeoutSeconds isn't rounded away" +$fractionBatch = Invoke-PkProcessBatch -TimeoutSeconds 0.4 -Commands @(@{ FilePath = $pingExe; Arguments = '-n 30 127.0.0.1' }) +Test-Assert ($fractionBatch.Results[0].TimedOut -eq $true) "and neither is the batch's" + +# On the main thread too, a plain callback reads the caller's $TimeoutSeconds: +# reading the kit's 0 here would finish the job at once. +$TimeoutSeconds = 1.5 +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'plain'; FilePath = $cmdExe; Arguments = '/c exit 0' }) -Extra @{ + TimeoutSeconds = 30 + JobPollCompleted = { param($job) $job['Proc'].HasExited -and $job['Stopwatch'].Elapsed.TotalSeconds -ge $TimeoutSeconds } +} +Remove-Variable -Name TimeoutSeconds +$plain = Get-LimitEntry $limitRun 'plain' +Test-Assert ($plain.Count -eq 1 -and $plain[0].At -ge 1.4 -and $plain[0].At -lt 10) ("a plain -JobPollCompleted reads the caller's `$TimeoutSeconds, not the dispatcher's (finished at " + $(if ($plain.Count) { [Math]::Round($plain[0].At, 1) }) + " s)") + +# A bad key fails that job's start, before anything runs. +Reset-Harness +$limitRun = Invoke-LimitGroup -Jobs @( + @{ Name = 'word'; FilePath = $pingExe; Arguments = '-n 33 127.0.0.1'; TimeoutSeconds = 'soon' }, + @{ Name = 'minus'; FilePath = $pingExe; Arguments = '-n 33 127.0.0.1'; TimeoutSeconds = -1 }, + @{ Name = 'nan'; FilePath = $pingExe; Arguments = '-n 33 127.0.0.1'; TimeoutSeconds = 'NaN' }) +$badKeyMessages = @($script:messages | Where-Object { $_.Level -eq 'Error' -and $_.Message -like "*TimeoutSeconds must be*" }).Count +Test-Assert ($null -ne $limitRun.Result -and $limitRun.Result.JobStarterFailedCount -eq 3 -and $limitRun.Log.Count -eq 0 -and $badKeyMessages -eq 3) ("a bad TimeoutSeconds key is a failed start with a clear message (messages " + $badKeyMessages + ", problem: " + $limitRun.Problem + ")") +Reset-Harness + +# A retry of a timed-out job gets a fresh clock. +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'retry'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1'; Retry = 1 }) -Extra @{ TimeoutSeconds = 1 } +$retried = Get-LimitEntry $limitRun 'retry' +Test-Assert ($retried.Count -eq 2 -and $retried[0].TimedOut -and $retried[1].TimedOut -and ($retried[1].At - $retried[0].At) -ge 0.8) ("a retried job gets a fresh clock (" + $retried.Count + " runs, problem: " + $limitRun.Problem + ")") + +$limitRun = Invoke-LimitGroup -Jobs @(@{ Name = 'x'; FilePath = $cmdExe; Arguments = '/c exit 0' }) -Extra @{ TimeoutSeconds = -1 } +Test-Assert ($null -ne $limitRun.Problem -and $limitRun.Problem -like '*-TimeoutSeconds must be*') "a negative -TimeoutSeconds is refused by the dispatcher" + +# The batch: a command's own key, and a bad one still gets a result. +$keyBatch = Invoke-PkProcessBatch -MaxConcurrent 4 -TimeoutSeconds 1 -Commands @( + @{ Name = 'own'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1'; TimeoutSeconds = 0.5 }, + @{ Name = 'optout'; FilePath = $pingExe; Arguments = '-n 3 127.0.0.1'; TimeoutSeconds = 0 }, + @{ Name = 'inherit'; FilePath = $pingExe; Arguments = '-n 30 127.0.0.1' }, + @{ Name = 'bad'; FilePath = $pingExe; Arguments = '-n 33 127.0.0.1'; TimeoutSeconds = 'soon' }) +$keyResult = @{} +foreach ($r in $keyBatch.Results) { $keyResult[$r.Command['Name']] = $r } +Test-Assert ($keyBatch.Results.Count -eq 4 -and $keyResult['own'].TimedOut -and $keyResult['inherit'].TimedOut) ("batch: a command's TimeoutSeconds key and -TimeoutSeconds both fire (" + $keyBatch.Results.Count + " results)") +Test-Assert ($keyResult.ContainsKey('optout') -and -not $keyResult['optout'].TimedOut -and $keyResult['optout'].ExitCode -eq 0) "batch: a command's TimeoutSeconds = 0 opts out" +Test-Assert ($keyResult.ContainsKey('bad') -and $keyResult['bad'].StartError -like '*TimeoutSeconds must be*' -and (Get-LimitPingCount 33) -eq 0) "batch: a bad TimeoutSeconds key is that command's StartError, and it doesn't run" +Reset-Harness + +# Inside a worker, where the helpers are plain copies and a plain callback sees +# the kit's frames: it must read the worker's own $TimeoutSeconds. +$limitWorker = [System.Collections.Generic.List[object]]::new() +$limitOut = Invoke-PkRunspaceJobs -MaxParallel 1 -JobParams (New-JobList -Jobs @(@{ PingExe = $pingExe })) -ResultHandler { param($r) $limitWorker.Add($r) }.GetNewClosure() -WorkerScript { + param([string]$PingExe) + $TimeoutSeconds = 99 + $seen = @{ Value = $null } + $feed = @{ Given = $false } + $flags = [System.Collections.Generic.List[object]]::new() + $null = Invoke-PkProcessDispatch -MaxConcurrent 1 -TimeoutSeconds 1 ` + -GroupProvider { if ($feed.Given) { return $null }; $feed.Given = $true; return 'g' }.GetNewClosure() ` + -JobBuilder { param($g) return @(@{ FilePath = $PingExe; Arguments = '-n 30 127.0.0.1' }) }.GetNewClosure() ` + -JobPollCompleted { param($job) $seen.Value = $TimeoutSeconds; $job['Proc'].HasExited } ` + -JobFinalize { param($job, $q) $null = Complete-PkProcess -Job $job; $flags.Add($job['TimedOut']) }.GetNewClosure() + return @{ Flags = ($flags -join ','); Seen = $seen.Value } +} +Test-Assert ($limitOut.CrashedCount -eq 0 -and $limitWorker.Count -eq 1 -and $limitWorker[0]['Flags'] -eq 'True') "a time limit works inside a worker" +Test-Assert ($limitWorker.Count -eq 1 -and $limitWorker[0]['Seen'] -eq 99) "and a plain callback there reads the worker's `$TimeoutSeconds, not the dispatcher's" +Reset-Harness +Write-Host "" + Set-PkMessageHandler -Handler $null Reset-PkAbortState diff --git a/ParallelKit.ps1 b/ParallelKit.ps1 index 2c60236..48ef15a 100644 --- a/ParallelKit.ps1 +++ b/ParallelKit.ps1 @@ -953,6 +953,41 @@ function Stop-PkJobProcesses { try { $null = Stop-PkProcessTree -Process @($procs) } catch { } # best effort; each job is still waited on and disposed after } +function Get-PkJobTimeLimit { + # A dispatcher job's time limit in seconds: its own TimeoutSeconds key if it + # has one, else the default. 0 = no limit. Throws on a bad key. + param([hashtable]$Job, [double]$DefaultTimeoutSeconds = 0) + $value = $Job['TimeoutSeconds'] + if ($null -eq $value) { return $DefaultTimeoutSeconds } + $limit = 0.0 + $parsed = [double]::TryParse([string]$value, [System.Globalization.NumberStyles]::Float, [System.Globalization.CultureInfo]::InvariantCulture, [ref]$limit) + if (-not $parsed -or [double]::IsNaN($limit) -or [double]::IsInfinity($limit) -or $limit -lt 0) { + throw ("TimeoutSeconds must be a number of seconds, 0 (no limit) or more (got '" + $value + "').") + } + return $limit +} + +function Stop-PkTimedOutJob { + # Kills a dispatcher job's program tree once it passes the job's TimeLimit and + # sets TimedOut; the poll then sees it exited and it is finalized like any + # other. Only a .Proc that is a real Process is timed: anything else is left + # to the job's own -JobPollCompleted. + param([hashtable]$Job) + $proc = $Job['Proc'] + $clock = $Job['Stopwatch'] + if ($proc -isnot [System.Diagnostics.Process] -or $clock -isnot [System.Diagnostics.Stopwatch]) { return } + if ($proc.HasExited -or $clock.Elapsed.TotalSeconds -lt $Job['TimeLimit']) { return } + # Only flagged if this call did the killing: a program that exited on its + # own right at the deadline didn't time out. One try only: this runs every + # pass, and a program that can't be killed would start a taskkill each time. + $Job['TimeLimit'] = 0 + if (Stop-PkProcessTree -Process $proc) { + $Job['TimedOut'] = $true + } elseif (-not $proc.HasExited) { + Write-PkMessage ("ProcessDispatch: a job passed its time limit, but its program (process " + $proc.Id + ") couldn't be stopped. It holds its slot until it exits.") -Level Warn + } +} + function Wait-PkProcess { # .SYNOPSIS # Waits for a program to exit, with a timeout that can't overrun. @@ -1151,9 +1186,10 @@ function Invoke-PkProcessDispatch { # # The dispatcher keeps its own state on each job hashtable. Don't use these keys # for anything else: GroupKey, SlotReleased, Proc, StdoutTask, StderrTask, - # Stopwatch. It also sets Group to the job's group, unless the job already has a - # Group of its own. The default starter reads FilePath, Arguments, - # WorkingDirectory and OutputEncoding from the job. + # Stopwatch, TimeLimit, TimedOut. It also sets Group to the job's group, unless + # the job already has a Group of its own. The default starter reads FilePath, + # Arguments, WorkingDirectory and OutputEncoding from the job. A job's + # TimeoutSeconds key overrides -TimeoutSeconds. # .PARAMETER GroupProvider # { } returning the next group, or $null when there are no more. # .PARAMETER JobBuilder @@ -1163,7 +1199,7 @@ function Invoke-PkProcessDispatch { # Start-PkProcess on the job's FilePath and Arguments. # .PARAMETER JobPollCompleted # { param($job) } returning $true once the job is finished. Default: - # $job['Proc'].HasExited. Also the place for a per-job time limit. + # $job['Proc'].HasExited. # .PARAMETER JobFinalize # { param($job, $pendingQueue) } called once per finished job; collect its # result with Complete-PkProcess. Enqueue new job hashtables onto $pendingQueue @@ -1191,6 +1227,11 @@ function Invoke-PkProcessDispatch { # .PARAMETER GroupReleaseCheck # { param($group) } returning $true to release a zero-job group's queue-ahead # slot; until then it holds the slot. Runs on every loop pass, so keep it cheap. + # .PARAMETER TimeoutSeconds + # Kill a job's program, and anything it started, still running after this long + # (default 0, no limit). A job's own TimeoutSeconds key overrides it, and 0 there + # opts the job out. The killed job still goes through -JobFinalize, with + # $job['TimedOut'] = $true. A job whose .Proc isn't a process isn't timed. # .OUTPUTS # An object with GroupResults, Aborted, ForceKilledCount and # JobStarterFailedCount. Each group result has GroupInfo (the group), Completed @@ -1212,7 +1253,8 @@ function Invoke-PkProcessDispatch { [System.Threading.CancellationToken]$CancelToken, [int]$AbortTimeoutSeconds = 60, [int]$MaxQueueAhead = 3, - [scriptblock]$GroupReleaseCheck = $null + [scriptblock]$GroupReleaseCheck = $null, + [double]$TimeoutSeconds = 0 ) # pk-prefixed, parameters included: inside a worker runspace the kit isn't # a module, so a callback without .GetNewClosure() sees this frame's @@ -1230,7 +1272,8 @@ function Invoke-PkProcessDispatch { $pkAbortTimeoutSeconds = $AbortTimeoutSeconds $pkMaxQueueAhead = $MaxQueueAhead $pkGroupReleaseCheck = $GroupReleaseCheck - Remove-Variable -Name GroupProvider, JobBuilder, JobStarter, JobPollCompleted, JobFinalize, JobAbortCleanup, GroupCompleteCallback, MaxConcurrent, AbortPollCallback, CancelToken, AbortTimeoutSeconds, MaxQueueAhead, GroupReleaseCheck -Scope Local + $pkTimeoutSeconds = $TimeoutSeconds + Remove-Variable -Name GroupProvider, JobBuilder, JobStarter, JobPollCompleted, JobFinalize, JobAbortCleanup, GroupCompleteCallback, MaxConcurrent, AbortPollCallback, CancelToken, AbortTimeoutSeconds, MaxQueueAhead, GroupReleaseCheck, TimeoutSeconds -Scope Local # 0 or less would never start a job (or pull a group) and spin forever. if ($pkMaxConcurrent -isnot [System.Threading.SemaphoreSlim]) { @@ -1241,6 +1284,8 @@ function Invoke-PkProcessDispatch { $pkMaxConcurrent = $pkMaxConcurrentInt } if ($pkMaxQueueAhead -lt 1) { throw "Invoke-PkProcessDispatch: -MaxQueueAhead must be at least 1." } + # Invoke-PkProcessBatch has the same check, so its error names the batch. + if ($pkTimeoutSeconds -lt 0) { throw "Invoke-PkProcessDispatch: -TimeoutSeconds must be 0 (no limit) or more." } # Two pool threads per running process for its output reads - make sure # they exist up front. For a shared budget, the slots free right now @@ -1295,7 +1340,7 @@ function Invoke-PkProcessDispatch { -JobPollCompleted $pkJobPollCompleted -JobFinalize $pkJobFinalize -JobAbortCleanup $pkJobAbortCleanup ` -GroupCompleteCallback $pkGroupCompleteCallback -MaxConcurrent $pkMaxConcurrent -AbortPollCallback $pkAbortPollCallback ` -CancelToken $pkCancelToken -AbortTimeoutSeconds $pkAbortTimeoutSeconds -MaxQueueAhead $pkMaxQueueAhead ` - -GroupReleaseCheck $pkGroupReleaseCheck -RequireProc $pkRequireProc + -GroupReleaseCheck $pkGroupReleaseCheck -RequireProc $pkRequireProc -TimeoutSeconds $pkTimeoutSeconds } finally { if ($null -ne $pkOwnCts) { Unregister-PkCancellationSource -Cts $pkOwnCts @@ -1363,7 +1408,8 @@ function Invoke-PkProcessDispatchCore { [int]$AbortTimeoutSeconds, [int]$MaxQueueAhead, [scriptblock]$GroupReleaseCheck, - [bool]$RequireProc = $false + [bool]$RequireProc = $false, + [double]$TimeoutSeconds = 0 ) # pk-prefixed, parameters included: inside a worker runspace the kit isn't # a module, so a callback without .GetNewClosure() sees this frame's @@ -1382,7 +1428,8 @@ function Invoke-PkProcessDispatchCore { $pkMaxQueueAhead = $MaxQueueAhead $pkGroupReleaseCheck = $GroupReleaseCheck $pkRequireProc = $RequireProc - Remove-Variable -Name GroupProvider, JobBuilder, JobStarter, JobPollCompleted, JobFinalize, JobAbortCleanup, GroupCompleteCallback, MaxConcurrent, AbortPollCallback, CancelToken, AbortTimeoutSeconds, MaxQueueAhead, GroupReleaseCheck, RequireProc -Scope Local + $pkTimeoutSeconds = $TimeoutSeconds + Remove-Variable -Name GroupProvider, JobBuilder, JobStarter, JobPollCompleted, JobFinalize, JobAbortCleanup, GroupCompleteCallback, MaxConcurrent, AbortPollCallback, CancelToken, AbortTimeoutSeconds, MaxQueueAhead, GroupReleaseCheck, RequireProc, TimeoutSeconds -Scope Local $pkIsSharedBudget = ($pkMaxConcurrent -is [System.Threading.SemaphoreSlim]) @@ -1457,6 +1504,8 @@ function Invoke-PkProcessDispatchCore { $pkJob['Proc'] = $null; $pkJob['StdoutTask'] = $null; $pkJob['StderrTask'] = $null; $pkJob['Stopwatch'] = $null } $pkJob['SlotReleased'] = $false + # Before starting, so a bad key never leaves a program running. + $pkJob['TimeLimit'] = Get-PkJobTimeLimit -Job $pkJob -DefaultTimeoutSeconds $pkTimeoutSeconds $pkStartedRaw = & $pkJobStarter $pkJob foreach ($pkItem in @($pkStartedRaw)) { if ($pkItem -isnot [hashtable] -or [object]::ReferenceEquals($pkItem, $pkJob)) { continue } @@ -1467,6 +1516,8 @@ function Invoke-PkProcessDispatchCore { if ($pkRequireProc -and $null -eq $pkJob['Proc']) { throw "JobStarter did not set the job's .Proc (set .Proc on the job and return it, or supply -JobPollCompleted)." } + $pkJob['TimedOut'] = $false + if ($pkJob['TimeLimit'] -gt 0 -and $null -eq $pkJob['Stopwatch']) { $pkJob['Stopwatch'] = [System.Diagnostics.Stopwatch]::StartNew() } $pkRunning.Add($pkJob) } catch { # Release BEFORE reporting: if reporting throws, the slot @@ -1496,6 +1547,7 @@ function Invoke-PkProcessDispatchCore { # and the mandatory final pass below - keep all three in sync. $pkStillRunning = [System.Collections.Generic.List[hashtable]]::new() foreach ($pkJob in $pkRunning) { + if ($pkJob['TimeLimit'] -gt 0) { Stop-PkTimedOutJob -Job $pkJob } if (& $pkJobPollCompleted $pkJob) { $pkPendingBefore = $pkPending.Count $null = & $pkJobFinalize $pkJob $pkPending @@ -1570,6 +1622,7 @@ function Invoke-PkProcessDispatchCore { # main poll pass above and the final pass below) $pkStillWaiting = [System.Collections.Generic.List[hashtable]]::new() foreach ($pkJob in $pkRunning) { + if ($pkJob['TimeLimit'] -gt 0) { Stop-PkTimedOutJob -Job $pkJob } if (& $pkJobPollCompleted $pkJob) { $pkPendingBefore = $pkPending.Count $null = & $pkJobFinalize $pkJob $pkPending @@ -1619,7 +1672,8 @@ function Invoke-PkProcessDispatchCore { # A window-close can exit the loop above before a single poll, and # a job can finish in the last 250ms before the deadline. One more # pass so a finished job is never thrown away as "force-stopped". - # (keep in sync with the two blocks above) + # (keep in sync with the two blocks above, except the time limit: + # whatever is still running is force-stopped together just below) if ($pkRunning.Count -gt 0) { $pkStillWaiting = [System.Collections.Generic.List[hashtable]]::new() foreach ($pkJob in $pkRunning) { @@ -1748,8 +1802,9 @@ function Invoke-PkProcessBatch { # each one's exit code and output. # .DESCRIPTION # Each command is a hashtable: @{ FilePath = ...; Arguments = ... } plus, - # optionally, WorkingDirectory, OutputEncoding (overrides -OutputEncoding) and - # any keys of your own, which come back on the result's Command. Inside + # optionally, WorkingDirectory, OutputEncoding (overrides -OutputEncoding), + # TimeoutSeconds (overrides -TimeoutSeconds) and any keys of your own, which + # come back on the result's Command. Inside # Invoke-PkRun, Ctrl+C lets running programs finish (up to -AbortTimeoutSeconds) # and starts no new ones; a second Ctrl+C kills them straight away. # .PARAMETER Commands @@ -1783,7 +1838,7 @@ function Invoke-PkProcessBatch { [int]$MaxConcurrent = 4, [scriptblock]$OnComplete = $null, [int]$AbortTimeoutSeconds = 60, - [int]$TimeoutSeconds = 0, + [double]$TimeoutSeconds = 0, $OutputEncoding = $null ) # pk-prefixed, parameters included: inside a worker runspace the kit isn't @@ -1799,6 +1854,7 @@ function Invoke-PkProcessBatch { # Checked once here, so a bad name fails the call instead of every command. $null = Resolve-PkEncoding -Encoding $pkOutputEncoding + # Invoke-PkProcessDispatch has the same check; this one names the batch. if ($pkTimeoutSeconds -lt 0) { throw "Invoke-PkProcessBatch: -TimeoutSeconds must be 0 (no limit) or more." } $pkResults = [System.Collections.Generic.List[object]]::new() @@ -1807,6 +1863,7 @@ function Invoke-PkProcessBatch { # .GetNewClosure() block only sees global ones), so take these along. $pkNewBatchResult = ${function:New-PkBatchResult} $pkStartProcessCore = ${function:Start-PkProcessCore} + $pkGetJobTimeLimit = ${function:Get-PkJobTimeLimit} $pkCallerState = $PSCmdlet.SessionState $pkGroupProvider = { @@ -1822,7 +1879,16 @@ function Invoke-PkProcessBatch { if ($null -eq $pkC) { continue } $pkEncoding = $pkOutputEncoding if ($pkC.ContainsKey('OutputEncoding')) { $pkEncoding = $pkC['OutputEncoding'] } - $pkJobs.Add(@{ FilePath = $pkC['FilePath']; Arguments = [string]$pkC['Arguments']; WorkingDirectory = [string]$pkC['WorkingDirectory']; OutputEncoding = $pkEncoding; Command = $pkC }) + # Here, not left to the dispatcher: a job it refuses never reaches the + # starter below, so the command would get no result. + try { + $null = & $pkGetJobTimeLimit -Job $pkC + } catch { + $pkResults.Add((& $pkNewBatchResult -Command $pkC -StartError ([string]$_))) + Write-PkMessage ("Invoke-PkProcessBatch: a command couldn't start: " + $_) -Level Error + continue + } + $pkJobs.Add(@{ FilePath = $pkC['FilePath']; Arguments = [string]$pkC['Arguments']; WorkingDirectory = [string]$pkC['WorkingDirectory']; OutputEncoding = $pkEncoding; TimeoutSeconds = $pkC['TimeoutSeconds']; Command = $pkC }) } return $pkJobs }.GetNewClosure() @@ -1836,7 +1902,7 @@ function Invoke-PkProcessBatch { throw } $pkJob['Proc'] = $pkStarted.Proc; $pkJob['StdoutTask'] = $pkStarted.StdoutTask; $pkJob['StderrTask'] = $pkStarted.StderrTask - $pkJob['Stopwatch'] = $pkStarted.Stopwatch; $pkJob['TimedOut'] = $false + $pkJob['Stopwatch'] = $pkStarted.Stopwatch return $pkJob }.GetNewClosure() @@ -1856,20 +1922,7 @@ function Invoke-PkProcessBatch { $pkDispatchParams = @{ GroupProvider = $pkGroupProvider; JobBuilder = $pkJobBuilder; JobStarter = $pkJobStarter JobFinalize = $pkJobFinalize; JobAbortCleanup = $pkJobAbortCleanup; MaxConcurrent = $pkMaxConcurrent - AbortTimeoutSeconds = $pkAbortTimeoutSeconds; MaxQueueAhead = 1 - } - if ($pkTimeoutSeconds -gt 0) { - # Kill an overrunning program; the next poll sees it exited and it - # goes through $pkJobFinalize like any other, flagged TimedOut. - $pkDispatchParams['JobPollCompleted'] = { - param($pkJob) - if (-not $pkJob['Proc'].HasExited -and $pkJob['Stopwatch'].Elapsed.TotalSeconds -ge $pkTimeoutSeconds) { - # Only flagged if this call did the killing: a program that - # exited on its own right at the deadline didn't time out. - if (Stop-PkProcessTree -Process $pkJob['Proc']) { $pkJob['TimedOut'] = $true } - } - return $pkJob['Proc'].HasExited - }.GetNewClosure() + AbortTimeoutSeconds = $pkAbortTimeoutSeconds; MaxQueueAhead = 1; TimeoutSeconds = $pkTimeoutSeconds } $pkDispatch = Invoke-PkProcessDispatch @pkDispatchParams @@ -1885,7 +1938,7 @@ function Get-PkWorkerSafeFunctionNames { return @('Write-PkMessage', 'Test-PkAbortSignal', 'Test-PkCancelRequested', 'Initialize-PkThreadPool', 'Stop-PkProcessTree', 'Stop-PkJobProcesses', 'Resolve-PkEncoding', 'Start-PkProcess', 'Wait-PkProcess', 'Wait-PkRunningJobs', 'Complete-PkProcess', 'Invoke-PkProcessWithTimeout', 'Invoke-PkProcessDispatch', 'Invoke-PkProcessDispatchCore', 'Invoke-PkProcessBatch', - 'Register-PkCancellationSource', 'Unregister-PkCancellationSource', 'New-PkBatchResult', 'ConvertTo-PkArgumentString', 'Get-PkForceStopReason', 'Resolve-PkFileSystemPath', 'Add-PkDispatchGroupJobs', 'Start-PkProcessCore') + 'Register-PkCancellationSource', 'Unregister-PkCancellationSource', 'New-PkBatchResult', 'ConvertTo-PkArgumentString', 'Get-PkForceStopReason', 'Resolve-PkFileSystemPath', 'Add-PkDispatchGroupJobs', 'Start-PkProcessCore', 'Get-PkJobTimeLimit', 'Stop-PkTimedOutJob') } function Get-PkHandlerParamCount { diff --git a/README.md b/README.md index d06c8db..d120c5a 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,7 @@ Programs run in your current location (`Get-Location`) unless a command says oth `-OnComplete` runs as each command finishes, but only for commands that actually ran. Check `Results` for the ones that couldn't start or were aborted. Defaults: `-MaxConcurrent` 4, `-AbortTimeoutSeconds` 60 (how long Ctrl+C waits for running programs), `-TimeoutSeconds` 0 (no limit). -`-TimeoutSeconds 600` kills any program still running after 10 minutes. Its result has `TimedOut = $true`, still goes through `-OnComplete`, and the rest of the batch carries on. The default, 0, means no limit. Any programs it started are killed with it. +`-TimeoutSeconds 600` kills any program still running after 10 minutes. Its result has `TimedOut = $true`, still goes through `-OnComplete`, and the rest of the batch carries on. The default, 0, means no limit. Any programs it started are killed with it. It can be fractional. A command's own `TimeoutSeconds` key overrides this for that command; 0 opts it out of the limit. A bad value there gives that command a `StartError`. Output is decoded in the console's code page unless you say otherwise, which is right for Windows' own tools. Many cross-platform tools write UTF-8 instead, and their non-ASCII output (accented or Japanese file names, say) comes back garbled. Pass `-OutputEncoding 'utf-8'` for the whole batch, or put an `OutputEncoding` key on one command. It also takes a code page number (`65001`, `437`) or a `[System.Text.Encoding]`, and the same option exists on `Invoke-PkProcessWithTimeout` and `Start-PkProcess` (and as a job key for the dispatcher's default starter). @@ -240,13 +240,14 @@ All of these callbacks run on whichever thread called the dispatcher: the main t - `-JobFinalize` is called once per finished job. Collect its exit code and output there with `Complete-PkProcess -Job $job`; the dispatcher disposes the process right after. - Groups are pulled lazily, at most `-MaxQueueAhead` (default 3) ahead. - A group can be any value: a hashtable, a path, a name. Each job gets its group as `$job.Group`. -- The dispatcher keeps its state on the job hashtables. Don't use these names for your own job fields: `GroupKey`, `SlotReleased`, `Proc`, `StdoutTask`, `StderrTask`, `Stopwatch`. (`Group` is yours if you set it yourself.) +- The dispatcher keeps its state on the job hashtables. Don't use these names for your own job fields: `GroupKey`, `SlotReleased`, `Proc`, `StdoutTask`, `StderrTask`, `Stopwatch`, `TimeLimit`, `TimedOut`. (`Group` is yours if you set it yourself.) +- `-TimeoutSeconds 600` kills a job's program, and anything it started, if it's still running after 10 minutes. The default, 0, means no limit, and it can be fractional. A job's own `TimeoutSeconds` key overrides it; 0 opts the job out, and a bad value is a failed start. The killed job still goes through `-JobFinalize`, where `$job['TimedOut']` is `$true`, and other jobs carry on. The limit stays in force while an abort waits for running jobs. A retry gets a fresh clock. A job whose `.Proc` isn't a process isn't timed: `-JobPollCompleted` is the place for its limit. - To retry a job or add follow-up work, enqueue job hashtables onto `$pending` inside `-JobFinalize`. They join the same group, and the group isn't complete until they've all finished. A retry can start within milliseconds of the failure, so cap the number of retries, and don't count on the gap for something like a locked file to clear. - Supply `-JobStarter { param($job) ... }` to start jobs yourself. It must set `$job.Proc` and should return the job. A starter that is just `Start-PkProcess ...` works too: its result is folded into your job. Throw to report a job that couldn't start. - Whatever your callbacks output is thrown away, apart from the return values listed here. They can't corrupt the dispatcher's result by accident. - `-MaxConcurrent` can be a `SemaphoreSlim` shared by several dispatchers that must stay under one combined limit. It works inside workers, so two runspace jobs can each run a dispatcher against one shared budget. - `-GroupReleaseCheck { param($group) }` makes a group with zero jobs keep its queue-ahead slot until the check returns true. Use it to stop one dispatcher racing ahead of another that it's paired with. The dispatcher doesn't return until every held group has been released, unless the run is cancelled or aborted. It runs on every loop pass, so keep it cheap. -- `-JobPollCompleted { param($job) }` replaces the default "has the process exited" check. Use it for a custom starter whose job isn't a plain process, or to give each job a time limit (see "Limits"). +- `-JobPollCompleted { param($job) }` replaces the default "has the process exited" check. Use it for a custom starter whose job isn't a plain process. - `-AbortPollCallback { param($runningJobs) }` runs on every loop pass, with the jobs currently running. That's about every 50 ms, and can be many times more often when programs finish quickly, so keep it cheap. - `-CancelToken` stops the dispatcher when cancelled. An abort stops it either way, without cancelling your token. - An exception from one of your callbacks stops the dispatcher: running processes are killed and the exception is rethrown. @@ -316,7 +317,6 @@ Other pieces: `Request-PkAbort` triggers the same abort from code. `Test-PkAbort - After changing `ParallelKit.ps1` itself, start a new PowerShell window. A session keeps the kit's compiled parts until it closes; the kit tells you if it finds an older copy loaded. - One run at a time per process. The abort state is shared. A nested `Invoke-PkRun` just runs inside the outer one. - A worker stuck inside a long .NET call can't be interrupted. Force-stop stops waiting for it and returns. The worker is NOT killed: once that call returns it can carry on to the end of its script. Before any step that can't be undone (deleting a source file, say), have the worker check `Test-PkCancelRequested` and skip the step if it's true. -- `Invoke-PkProcessDispatch` has no per-program time limit (`Invoke-PkProcessBatch` has `-TimeoutSeconds`): a program that hangs holds its slot until it exits or the run is aborted. You can add one through `-JobPollCompleted`, using the stopwatch the default starter puts on each job: `{ param($job) if ($job['Stopwatch'].Elapsed.TotalMinutes -gt 10) { $null = Stop-PkProcessTree -Process $job['Proc'] }; $job['Proc'].HasExited }`. The killed job then goes through `-JobFinalize` like any other, with a non-zero exit code. - Programs run hidden, and there's no way to send them input. A program that stops to ask a question will sit there until it's killed. - A program's output is collected in memory and handed over when it exits; there's no live line-by-line output. A program that prints gigabytes will use that much memory. A batch keeps every result, output included, until it returns, which adds up over thousands of chatty programs. If you only need the output in `-OnComplete`, drop it there with `$result.Stdout = $null; $result.Stderr = $null` and it's freed as the batch goes.