From cd1f079f5b78b98edd01175a50b6c3db89c715e5 Mon Sep 17 00:00:00 2001 From: Hal <22532613+HalRiveria@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:41:39 -0500 Subject: [PATCH 1/2] Wake the dispatch and runspace loops as soon as a job finishes The dispatch loop slept 50 ms on every pass, even one where a job had just finished, and noticed the free slot only after another sleep. That capped each slot at about 8 programs a second. The runspace loop's 100 ms sleep capped short workers at about 9 jobs a second per slot. The dispatch loop now goes straight round after a pass where a job finished, and otherwise waits on the exit handles of programs still running plus the cancel token, capped at 50 ms. It waits only on programs that haven't exited, so a -JobPollCompleted that keeps saying "not done" can't spin a core. It falls back to a sleep for a job with no real process or more than 63 running. The runspace loop does the same with its workers' wait handles and a 100 ms cap. Co-Authored-By: Claude Opus 5.5 --- ParallelKit.Harness.ps1 | 76 +++++++++++++++++++++++++++++++++++++++++ ParallelKit.ps1 | 65 +++++++++++++++++++++++++++++++++-- README.md | 4 +-- 3 files changed, 140 insertions(+), 5 deletions(-) diff --git a/ParallelKit.Harness.ps1 b/ParallelKit.Harness.ps1 index c9fbd90..8214413 100644 --- a/ParallelKit.Harness.ps1 +++ b/ParallelKit.Harness.ps1 @@ -2259,6 +2259,82 @@ Test-Assert ($softRun.Aborted -and $softRun.CrashedCount -eq 0 -and $softOut.Cou Reset-Harness Write-Host "" +# ============================================================================== +Write-Host "=== AB: a finished job's slot is refilled without waiting out a sleep ===" -ForegroundColor Cyan + +# Timing, so up to three tries: a busy machine can only make a run slower, and +# polling can't get under its floor however idle the machine is. +# Two 50 ms sleeps per program at -MaxConcurrent 1 can't average under 100 ms. +$bestPerProgram = [double]::MaxValue +for ($try = 0; $try -lt 3 -and $bestPerProgram -ge 90; $try++) { + $shortCommands = foreach ($i in 1..30) { @{ FilePath = $cmdExe; Arguments = '/d /c exit 0' } } + $sw = [System.Diagnostics.Stopwatch]::StartNew() + $shortBatch = Invoke-PkProcessBatch -Commands $shortCommands -MaxConcurrent 1 + $sw.Stop() + $bestPerProgram = [Math]::Min($bestPerProgram, $sw.Elapsed.TotalMilliseconds / 30) +} +Test-Assert ($shortBatch.Results.Count -eq 30) "every short program has a result" +Test-Assert ($bestPerProgram -lt 90) ("short programs aren't held back by the poll (best average " + [int]$bestPerProgram + " ms each)") + +# The same floor for runspace jobs: one 100 ms sleep per 20 ms worker. +$bestPerJob = [double]::MaxValue +for ($try = 0; $try -lt 3 -and $bestPerJob -ge 70; $try++) { + $sw = [System.Diagnostics.Stopwatch]::StartNew() + $null = Invoke-PkRunspaceJobs -MaxParallel 1 -JobParams (New-JobList -Jobs @(foreach ($i in 1..20) { @{ Id = $i } })) -WorkerScript { + param([int]$Id) + $spin = [System.Diagnostics.Stopwatch]::StartNew() + while ($spin.ElapsedMilliseconds -lt 20) { } + } -ResultHandler { param($r) } + $sw.Stop() + $bestPerJob = [Math]::Min($bestPerJob, $sw.Elapsed.TotalMilliseconds / 20) +} +Test-Assert ($bestPerJob -lt 70) ("short workers aren't held back by the poll (best average " + [int]$bestPerJob + " ms each)") + +# A -JobPollCompleted that keeps saying "not done" after a process exits must +# not leave that process's signalled handle in the wait, or the loop spins: +# first while the second program still runs, then with both exited. +$spinState = @{ Given = $false; Polls = 0 } +$spinRun = Invoke-PkProcessDispatch -MaxConcurrent 2 ` + -GroupProvider { if ($spinState.Given) { return $null }; $spinState.Given = $true; return 'g' }.GetNewClosure() ` + -JobBuilder { param($g) @{ FilePath = $cmdExe; Arguments = '/d /c exit 0' }, @{ FilePath = $pingExe; Arguments = '-n 2 127.0.0.1' } }.GetNewClosure() ` + -JobPollCompleted { param($job) $spinState.Polls++; $job['Proc'].HasExited -and $job['Stopwatch'].Elapsed.TotalMilliseconds -gt 2000 }.GetNewClosure() ` + -JobFinalize { param($job, $q) $null = Complete-PkProcess -Job $job } +Test-Assert ($spinRun.GroupResults.Count -eq 1 -and $spinState.Polls -lt 500) ("a lagging -JobPollCompleted doesn't spin the loop (" + $spinState.Polls + " polls in 2 s)") + +# More running programs than one wait can hold. +$wideCommands = foreach ($i in 1..70) { @{ FilePath = $pingExe; Arguments = '-n 2 127.0.0.1' } } +$wide = Invoke-PkProcessBatch -Commands $wideCommands -MaxConcurrent 70 +$wideOk = @($wide.Results | Where-Object { $_.ExitCode -eq 0 }).Count +Test-Assert ($wideOk -eq 70) "70 programs at once all finish ($wideOk ok)" +$wideRun = Invoke-PkRunspaceJobs -MaxParallel 70 -JobParams (New-JobList -Jobs @(foreach ($i in 1..70) { @{ Id = $i } })) -WorkerScript { + param([int]$Id) + [System.Threading.Thread]::Sleep(500) + return $Id +} -ResultHandler { param($r) } +Test-Assert ($wideRun.Aborted -eq $false -and $wideRun.CrashedCount -eq 0) "70 runspace jobs at once all finish" + +# A starter whose .Proc isn't a Process still gets waited on and finished. +$fakeState = @{ Given = $false; Finished = 0; Started = [System.Diagnostics.Stopwatch]::StartNew() } +$fakeRun = Invoke-PkProcessDispatch -MaxConcurrent 2 ` + -GroupProvider { if ($fakeState.Given) { return $null }; $fakeState.Given = $true; return 'g' }.GetNewClosure() ` + -JobBuilder { param($g) @{ Fake = 1 }, @{ FilePath = $pingExe; Arguments = '-n 2 127.0.0.1' } }.GetNewClosure() ` + -JobStarter { param($job) if ($job['Fake']) { $job['Proc'] = 'not a process' } else { $s = Start-PkProcess -FilePath $job['FilePath'] -Arguments $job['Arguments']; foreach ($k in $s.Keys) { $job[$k] = $s[$k] } }; return $job } ` + -JobPollCompleted { param($job) if ($job['Fake']) { $fakeState.Started.ElapsedMilliseconds -gt 300 } else { $job['Proc'].HasExited } }.GetNewClosure() ` + -JobFinalize { param($job, $q) $fakeState.Finished++; if (-not $job['Fake']) { $null = Complete-PkProcess -Job $job } }.GetNewClosure() ` + -JobAbortCleanup { param($job) } +Test-Assert ($fakeState.Finished -eq 2 -and $fakeRun.JobStarterFailedCount -eq 0) "a job whose .Proc isn't a process doesn't stop the wait ($($fakeState.Finished) finished)" + +# The same loop inside a worker, where the wait helper is a plain copy. +$workerWait = [System.Collections.Generic.List[object]]::new() +$null = Invoke-PkRunspaceJobs -MaxParallel 1 -JobParams (New-JobList -Jobs @(@{ CmdExe = $cmdExe })) -WorkerScript { + param([string]$CmdExe) + $inner = Invoke-PkProcessBatch -Commands @(@{ FilePath = $CmdExe; Arguments = '/d /c exit 0' }, @{ FilePath = $CmdExe; Arguments = '/d /c ping -n 2 127.0.0.1 >nul' }) -MaxConcurrent 2 + return @($inner.Results | Where-Object { $_.ExitCode -eq 0 }).Count +} -ResultHandler { param($r) $workerWait.Add($r) }.GetNewClosure() +Test-Assert ($workerWait.Count -eq 1 -and $workerWait[0] -eq 2) "a batch inside a worker waits and finishes (results: $($workerWait -join ','))" +Reset-Harness +Write-Host "" + Set-PkMessageHandler -Handler $null Reset-PkAbortState diff --git a/ParallelKit.ps1 b/ParallelKit.ps1 index 0b8b58c..c98adcc 100644 --- a/ParallelKit.ps1 +++ b/ParallelKit.ps1 @@ -982,6 +982,49 @@ function Wait-PkProcess { } } +function Wait-PkRunningJobs { + # The dispatch loop's pause: until a running program exits, the token is + # cancelled, or TimeoutMs passes. Only programs that haven't exited are waited + # on: a -JobPollCompleted can keep saying "not done" after the exit, and the + # signalled handle would make every wait return at once and spin a core. + # A plain sleep when a job has no real process, or more run than WaitAny can + # hold (64 handles, one of them the token). + param( + [System.Collections.Generic.List[hashtable]]$Jobs, + [System.Threading.CancellationToken]$CancelToken, + [int]$TimeoutMs + ) + $waits = [System.Collections.Generic.List[System.Threading.WaitHandle]]::new() + $origHandles = [System.Collections.Generic.List[object]]::new() + $canWait = $true + try { + foreach ($job in $Jobs) { + $proc = $job['Proc'] + if ($proc -isnot [System.Diagnostics.Process] -or $waits.Count -ge 63) { $canWait = $false; break } + if ($proc.HasExited) { continue } + $procHandle = $proc.Handle + $waitHandle = [System.Threading.ManualResetEvent]::new($false) + $origHandles.Add($waitHandle.SafeWaitHandle) + $waitHandle.SafeWaitHandle = [Microsoft.Win32.SafeHandles.SafeWaitHandle]::new($procHandle, $false) + $waits.Add($waitHandle) + } + if ($canWait -and $waits.Count -gt 0) { + $waits.Add($CancelToken.WaitHandle) + $null = [System.Threading.WaitHandle]::WaitAny($waits.ToArray(), $TimeoutMs) + return + } + } catch { + # A process that can't be read or waited on (a starter's own, say) is + # the job's -JobPollCompleted to deal with; here it just means sleeping. + } finally { + for ($i = 0; $i -lt $origHandles.Count; $i++) { + $waits[$i].SafeWaitHandle = $origHandles[$i] + $waits[$i].Dispose() + } + } + [System.Threading.Thread]::Sleep($TimeoutMs) +} + function Complete-PkProcess { # .SYNOPSIS # Collects the exit code and output of a program that has exited. @@ -1492,9 +1535,13 @@ function Invoke-PkProcessDispatchCore { $pkStillRunning.Add($pkJob) } } + # A job that finished freed a slot: go straight round and refill it. + # The wait's 50 ms cap is how soon an abort flag or a shared budget's + # free slot is noticed, neither of which wakes it. + $pkAnyFinished = ($pkStillRunning.Count -lt $pkRunning.Count) $pkRunning = $pkStillRunning - [System.Threading.Thread]::Sleep(50) + if (-not $pkAnyFinished) { Wait-PkRunningJobs -Jobs $pkRunning -CancelToken $pkCancelToken -TimeoutMs 50 } } } catch { # Best-effort teardown on the way out; the original error is rethrown. @@ -1831,7 +1878,7 @@ function Invoke-PkProcessBatch { function Get-PkWorkerSafeFunctionNames { # Kit functions every worker runspace gets automatically. return @('Write-PkMessage', 'Test-PkAbortSignal', 'Test-PkCancelRequested', 'Initialize-PkThreadPool', 'Stop-PkProcessTree', 'Stop-PkJobProcesses', 'Resolve-PkEncoding', - 'Start-PkProcess', 'Wait-PkProcess', 'Complete-PkProcess', 'Invoke-PkProcessWithTimeout', + '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') } @@ -2252,7 +2299,19 @@ function Invoke-PkRunspaceJobs { } } $handles = $stillPending - if ($handles.Count -gt 0 -and ($scanExhausted -or $handles.Count -ge $MaxParallel)) { [System.Threading.Thread]::Sleep(100) } + if ($handles.Count -gt 0 -and ($scanExhausted -or $handles.Count -ge $MaxParallel)) { + # Until a worker finishes or the token is cancelled. The 100 ms cap + # keeps worker messages, -PollCallback and the abort flag moving. + # WaitAny holds 64 handles at most, one of them the token. + if ($handles.Count -le 63) { + $workerWaits = [System.Collections.Generic.List[System.Threading.WaitHandle]]::new() + foreach ($h in $handles) { $workerWaits.Add($h.Handle.AsyncWaitHandle) } + $workerWaits.Add($Cts.Token.WaitHandle) + $null = [System.Threading.WaitHandle]::WaitAny($workerWaits.ToArray(), 100) + } else { + [System.Threading.Thread]::Sleep(100) + } + } } if ($Cts.Token.IsCancellationRequested) { diff --git a/README.md b/README.md index 41a952a..3b8fd64 100644 --- a/README.md +++ b/README.md @@ -181,7 +181,7 @@ Each worker runs in its own runspace: a separate PowerShell session inside the s - Each job starts in your current location (the real folder, if you're on a `New-PSDrive` drive), in a fresh scope. A folder change or a variable one job makes, preference variables like `$ErrorActionPreference` included, doesn't carry over to the next job. Relative paths in .NET calls such as `[System.IO.File]::ReadAllText('x.txt')` still go by the process's working directory, as they do outside workers, so give .NET full paths. - 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. +- `-PollCallback { ... }` runs on the main thread on every loop pass, for progress display and the like. That's at least every 0.1 s, and can be many times more often when jobs finish quickly, so keep it cheap. - `-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: @@ -247,7 +247,7 @@ All of these callbacks run on whichever thread called the dispatcher: the main t - `-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. - `-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"). -- `-AbortPollCallback { param($runningJobs) }` runs on every loop pass, with the jobs currently running. +- `-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. - Returns an object with `GroupResults`, `Aborted`, `ForceKilledCount` and `JobStarterFailedCount`. `GroupResults` holds one object with `GroupInfo`, `Completed` and `Aborted` per group, where `GroupInfo` is the group value. From 145d1dcd6494771b774a446e1bb1fe17c15e81b3 Mon Sep 17 00:00:00 2001 From: Hal <22532613+HalRiveria@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:54:02 -0500 Subject: [PATCH 2/2] Keep shared-budget dispatchers taking turns; wait on up to 62 handles Refilling a freed slot at once let a dispatcher on a shared SemaphoreSlim keep every slot until its queue ran out, so a sibling got none. A dispatcher on a shared budget still pauses after a finished job. WaitAny holds at most 63 handles on an STA thread, so waiting on 63 plus the token threw under PowerShell 7. Both loops now wait on up to 62 plus the token, and on those even when more are running, rather than falling back to a sleep. Co-Authored-By: Claude Opus 5.5 --- ParallelKit.Harness.ps1 | 22 +++++++++++++++++++++ ParallelKit.ps1 | 43 +++++++++++++++++++++++------------------ README.md | 6 +++--- 3 files changed, 49 insertions(+), 22 deletions(-) diff --git a/ParallelKit.Harness.ps1 b/ParallelKit.Harness.ps1 index 8214413..075c745 100644 --- a/ParallelKit.Harness.ps1 +++ b/ParallelKit.Harness.ps1 @@ -2324,6 +2324,28 @@ $fakeRun = Invoke-PkProcessDispatch -MaxConcurrent 2 ` -JobAbortCleanup { param($job) } Test-Assert ($fakeState.Finished -eq 2 -and $fakeRun.JobStarterFailedCount -eq 0) "a job whose .Proc isn't a process doesn't stop the wait ($($fakeState.Finished) finished)" +# Dispatchers sharing a budget take turns: one with a queue must not retake each +# slot it frees before the other has seen it free. A starts first, so every run +# of A's would come before B's first if A held on to the slot. +$turnBudget = [System.Threading.SemaphoreSlim]::new(1, 1) +$turnLog = [System.Collections.Concurrent.ConcurrentQueue[string]]::new() +$turnJobs = New-JobList -Jobs @( + @{ Tag = 'A'; DelayMs = 0; Budget = $turnBudget; Log = $turnLog; Exe = $cmdExe }, + @{ Tag = 'B'; DelayMs = 300; Budget = $turnBudget; Log = $turnLog; Exe = $cmdExe }) +$null = Invoke-PkRunspaceJobs -MaxParallel 2 -JobParams $turnJobs -ResultHandler { param($r) } -WorkerScript { + param([string]$Tag, [int]$DelayMs, [System.Threading.SemaphoreSlim]$Budget, $Log, [string]$Exe) + [System.Threading.Thread]::Sleep($DelayMs) + $feed = @{ Given = $false } + $null = Invoke-PkProcessDispatch -MaxConcurrent $Budget ` + -GroupProvider { if ($feed.Given) { return $null }; $feed.Given = $true; return 'g' }.GetNewClosure() ` + -JobBuilder { param($g) foreach ($i in 1..30) { @{ FilePath = $Exe; Arguments = '/d /c ping -n 1 127.0.0.1 >nul' } } }.GetNewClosure() ` + -JobFinalize { param($job, $q) $null = Complete-PkProcess -Job $job; $Log.Enqueue($Tag) }.GetNewClosure() +} +$turnOrder = -join $turnLog.ToArray() +$turnBudget.Dispose() +$turnFirstB = $turnOrder.IndexOf('B') +Test-Assert ($turnOrder.Length -eq 60 -and $turnFirstB -ge 0 -and $turnFirstB -lt 15) ("dispatchers sharing a budget take turns ($turnFirstB of A's ran before B's first: $turnOrder)") + # The same loop inside a worker, where the wait helper is a plain copy. $workerWait = [System.Collections.Generic.List[object]]::new() $null = Invoke-PkRunspaceJobs -MaxParallel 1 -JobParams (New-JobList -Jobs @(@{ CmdExe = $cmdExe })) -WorkerScript { diff --git a/ParallelKit.ps1 b/ParallelKit.ps1 index c98adcc..2c60236 100644 --- a/ParallelKit.ps1 +++ b/ParallelKit.ps1 @@ -971,6 +971,7 @@ function Wait-PkProcess { [Parameter(Mandatory)] [int]$TimeoutMs ) + # Same handle wrapping as Wait-PkRunningJobs. $waitHandle = [System.Threading.ManualResetEvent]::new($false) $origHandle = $waitHandle.SafeWaitHandle $waitHandle.SafeWaitHandle = [Microsoft.Win32.SafeHandles.SafeWaitHandle]::new($Process.Handle, $false) @@ -987,8 +988,9 @@ function Wait-PkRunningJobs { # cancelled, or TimeoutMs passes. Only programs that haven't exited are waited # on: a -JobPollCompleted can keep saying "not done" after the exit, and the # signalled handle would make every wait return at once and spin a core. - # A plain sleep when a job has no real process, or more run than WaitAny can - # hold (64 handles, one of them the token). + # A job with no real process, or past the 63 handles WaitAny can hold on an STA + # thread (one of them the token), is only noticed when the time is up. With + # nothing to wait on, a plain sleep. Same handle wrapping as Wait-PkProcess. param( [System.Collections.Generic.List[hashtable]]$Jobs, [System.Threading.CancellationToken]$CancelToken, @@ -996,19 +998,19 @@ function Wait-PkRunningJobs { ) $waits = [System.Collections.Generic.List[System.Threading.WaitHandle]]::new() $origHandles = [System.Collections.Generic.List[object]]::new() - $canWait = $true try { foreach ($job in $Jobs) { + if ($waits.Count -ge 62) { break } $proc = $job['Proc'] - if ($proc -isnot [System.Diagnostics.Process] -or $waits.Count -ge 63) { $canWait = $false; break } - if ($proc.HasExited) { continue } + if ($proc -isnot [System.Diagnostics.Process] -or $proc.HasExited) { continue } $procHandle = $proc.Handle $waitHandle = [System.Threading.ManualResetEvent]::new($false) - $origHandles.Add($waitHandle.SafeWaitHandle) + $origHandle = $waitHandle.SafeWaitHandle $waitHandle.SafeWaitHandle = [Microsoft.Win32.SafeHandles.SafeWaitHandle]::new($procHandle, $false) $waits.Add($waitHandle) + $origHandles.Add($origHandle) } - if ($canWait -and $waits.Count -gt 0) { + if ($waits.Count -gt 0) { $waits.Add($CancelToken.WaitHandle) $null = [System.Threading.WaitHandle]::WaitAny($waits.ToArray(), $TimeoutMs) return @@ -1188,7 +1190,7 @@ function Invoke-PkProcessDispatch { # How many groups to pull ahead of the ones running (default 3). # .PARAMETER GroupReleaseCheck # { param($group) } returning $true to release a zero-job group's queue-ahead - # slot; until then it holds the slot. + # slot; until then it holds the slot. Runs on every loop pass, so keep it cheap. # .OUTPUTS # An object with GroupResults, Aborted, ForceKilledCount and # JobStarterFailedCount. Each group result has GroupInfo (the group), Completed @@ -1536,12 +1538,15 @@ function Invoke-PkProcessDispatchCore { } } # A job that finished freed a slot: go straight round and refill it. - # The wait's 50 ms cap is how soon an abort flag or a shared budget's - # free slot is noticed, neither of which wakes it. + # Not on a shared budget, where a sibling dispatcher only looks for a + # free slot every 50 ms: refilling at once took every slot until this + # dispatcher's queue ran out, and the sibling got none. + # The wait's 50 ms cap is how soon an abort flag or a sibling's released + # slot is noticed, neither of which wakes it. $pkAnyFinished = ($pkStillRunning.Count -lt $pkRunning.Count) $pkRunning = $pkStillRunning - if (-not $pkAnyFinished) { Wait-PkRunningJobs -Jobs $pkRunning -CancelToken $pkCancelToken -TimeoutMs 50 } + if ($pkIsSharedBudget -or -not $pkAnyFinished) { Wait-PkRunningJobs -Jobs $pkRunning -CancelToken $pkCancelToken -TimeoutMs 50 } } } catch { # Best-effort teardown on the way out; the original error is rethrown. @@ -2302,15 +2307,15 @@ function Invoke-PkRunspaceJobs { if ($handles.Count -gt 0 -and ($scanExhausted -or $handles.Count -ge $MaxParallel)) { # Until a worker finishes or the token is cancelled. The 100 ms cap # keeps worker messages, -PollCallback and the abort flag moving. - # WaitAny holds 64 handles at most, one of them the token. - if ($handles.Count -le 63) { - $workerWaits = [System.Collections.Generic.List[System.Threading.WaitHandle]]::new() - foreach ($h in $handles) { $workerWaits.Add($h.Handle.AsyncWaitHandle) } - $workerWaits.Add($Cts.Token.WaitHandle) - $null = [System.Threading.WaitHandle]::WaitAny($workerWaits.ToArray(), 100) - } else { - [System.Threading.Thread]::Sleep(100) + # WaitAny holds 63 handles at most on an STA thread, one of them the + # token; a worker past that is only noticed when the time is up. + $workerWaits = [System.Collections.Generic.List[System.Threading.WaitHandle]]::new() + foreach ($h in $handles) { + if ($workerWaits.Count -ge 62) { break } + $workerWaits.Add($h.Handle.AsyncWaitHandle) } + $workerWaits.Add($Cts.Token.WaitHandle) + $null = [System.Threading.WaitHandle]::WaitAny($workerWaits.ToArray(), 100) } } diff --git a/README.md b/README.md index 3b8fd64..d06c8db 100644 --- a/README.md +++ b/README.md @@ -241,11 +241,11 @@ All of these callbacks run on whichever thread called the dispatcher: the main t - 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.) -- 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. +- 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. +- `-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"). - `-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. @@ -306,7 +306,7 @@ Other pieces: `Request-PkAbort` triggers the same abort from code. `Test-PkAbort - **`-MaxConcurrent` for programs.** For programs that use one CPU core each, start at `[Environment]::ProcessorCount`. Many encoders (image, video, compression) already use several cores per file, so fewer copies at once is often as fast and uses less memory. Programs that mostly wait on disk or network can go higher. Try a few values on real work; the best number depends on the tool. - **`-MaxParallel` for runspace jobs.** Same idea: the core count for CPU-heavy PowerShell work. Each runspace costs some memory and startup time, so very high values rarely help. -- **Several dispatchers at once** (for example, one per runspace job) should share one `SemaphoreSlim` as `-MaxConcurrent`, so their combined total stays under the limit you picked. +- **Several dispatchers at once** (for example, one per runspace job) should share one `SemaphoreSlim` as `-MaxConcurrent`, so their combined total stays under the limit you picked. They take turns: each pauses up to 50 ms after a program finishes so the others can get the free slot, which costs very short programs some speed compared with a dispatcher of their own. ## Limits