diff --git a/ParallelKit.Harness.ps1 b/ParallelKit.Harness.ps1 index a46725c..7f00eb8 100644 --- a/ParallelKit.Harness.ps1 +++ b/ParallelKit.Harness.ps1 @@ -1981,6 +1981,48 @@ try { } Write-Host "" +# ============================================================================== +Write-Host "=== Y: output that's been dropped is freed during the run ===" -ForegroundColor Cyan +# 30 programs printing ~150 KB each. A caller who drops each output once it's +# handled shouldn't have the kit keep it alive until the run ends. Weak +# references show what a full GC can free while the last job is finishing. +# A few of the most recent jobs can still be referenced from the dispatch +# loop (5-7 at -MaxConcurrent 4, however long the run); before the fix all were. +Reset-Harness +$chatty = @(1..30 | ForEach-Object { @{ FilePath = $cmdExe; Arguments = ('/c for /l %i in (1,1,1500) do @echo ' + ('x' * 100)) } }) +$batchWeak = [System.Collections.Generic.List[System.WeakReference]]::new() +$batchSeen = @{ Alive = -1; Largest = 0 } +$null = Invoke-PkProcessBatch -Commands $chatty -MaxConcurrent 4 -OnComplete { + param($command, $result) + if ($result.Stdout.Length -gt $batchSeen.Largest) { $batchSeen.Largest = $result.Stdout.Length } + $batchWeak.Add([System.WeakReference]::new($result.Stdout)) + $result.Stdout = $null + $result.Stderr = $null + if ($batchWeak.Count -eq 30) { + [GC]::Collect(); [GC]::WaitForPendingFinalizers(); [GC]::Collect() + $batchSeen.Alive = @($batchWeak | Select-Object -First 29 | Where-Object { $_.IsAlive }).Count + } +}.GetNewClosure() +Test-Assert ($batchSeen.Largest -gt 100000) ("precondition: each program printed a lot (" + $batchSeen.Largest + " chars)") +Test-Assert ($batchSeen.Alive -ge 0 -and $batchSeen.Alive -le 15) ("batch: output dropped in -OnComplete can be freed before the batch ends (" + $batchSeen.Alive + " of 29 still held)") + +$feed = @{ Given = $false } +$dispatchWeak = [System.Collections.Generic.List[System.WeakReference]]::new() +$dispatchSeen = @{ Alive = -1 } +$null = Invoke-PkProcessDispatch -MaxConcurrent 4 ` + -GroupProvider { if ($feed.Given) { return $null }; $feed.Given = $true; return 'one big group' }.GetNewClosure() ` + -JobBuilder { param($g) foreach ($c in $chatty) { $c.Clone() } }.GetNewClosure() ` + -JobFinalize { + param($job, $pending) + $dispatchWeak.Add([System.WeakReference]::new((Complete-PkProcess -Job $job).Stdout)) + if ($dispatchWeak.Count -eq 30) { + [GC]::Collect(); [GC]::WaitForPendingFinalizers(); [GC]::Collect() + $dispatchSeen.Alive = @($dispatchWeak | Select-Object -First 29 | Where-Object { $_.IsAlive }).Count + } + }.GetNewClosure() +Test-Assert ($dispatchSeen.Alive -ge 0 -and $dispatchSeen.Alive -le 15) ("dispatcher: a finished job's output can be freed before its group ends (" + $dispatchSeen.Alive + " of 29 still held)") +Write-Host "" + Set-PkMessageHandler -Handler $null Reset-PkAbortState diff --git a/ParallelKit.ps1 b/ParallelKit.ps1 index c38f51b..f6912f6 100644 --- a/ParallelKit.ps1 +++ b/ParallelKit.ps1 @@ -1207,6 +1207,41 @@ function Invoke-PkProcessDispatch { } } +function Add-PkDispatchGroupJobs { + # Builds one group's jobs, tags them with the group and queues them. + # Returns how many were queued. Its own function, not inline in the + # dispatch loop, so the group's job list dies with this call. Kept in the + # loop's frame it stayed alive for the whole run, and with it every + # finished job's output, even output the caller had already dropped. + # Nulling the variables in the loop didn't help once the kit had run many + # times in a session (most likely PowerShell's compiled code keeping its + # own references); a separate call did. + param( + [Parameter(Mandatory)][scriptblock]$JobBuilder, + $GroupInfo, + [Parameter(Mandatory)][object]$GroupKey, + [Parameter(Mandatory)][System.Collections.Generic.Queue[hashtable]]$Pending + ) + $rawJobs = & $JobBuilder $GroupInfo + # Assigned inside each branch, NOT `$jobs = if (...) {...} else {...}`: + # captured through an if/else, a single hashtable gets + # enumerated into its key/value pairs. + if ($null -eq $rawJobs) { $rawJobList = @() } else { $rawJobList = @($rawJobs) } + # Keep only real job hashtables: a builder that lets a stray + # value escape (a HashSet.Add's True, say) must not crash the run. + $jobs = [System.Collections.Generic.List[hashtable]]::new() + foreach ($rawJob in $rawJobList) { if ($rawJob -is [hashtable]) { $jobs.Add($rawJob) } } + if ($jobs.Count -ne $rawJobList.Count) { + Write-PkMessage ("ProcessDispatch: -JobBuilder returned " + ($rawJobList.Count - $jobs.Count) + " value(s) that are not job hashtables - ignored. Each job must be a hashtable.") -Level Warn + } + foreach ($j in $jobs) { + $j['GroupKey'] = $GroupKey + if (-not $j.ContainsKey('Group')) { $j['Group'] = $GroupInfo } # leave a job's own 'Group' field alone + $Pending.Enqueue($j) + } + return $jobs.Count +} + function Invoke-PkProcessDispatchCore { # The dispatch loop itself. Call Invoke-PkProcessDispatch, not this. param( @@ -1263,24 +1298,8 @@ function Invoke-PkProcessDispatchCore { # share one job count. Jobs carry this key as .GroupKey and the # group value itself as .Group. $groupKey = [object]::new() - $rawJobs = & $JobBuilder $groupInfo - # Assigned inside each branch, NOT `$jobs = if (...) {...} else {...}`: - # captured through an if/else, a single hashtable gets - # enumerated into its key/value pairs. - if ($null -eq $rawJobs) { $rawJobList = @() } else { $rawJobList = @($rawJobs) } - # Keep only real job hashtables: a builder that lets a stray - # value escape (a HashSet.Add's True, say) must not crash the run. - $jobs = [System.Collections.Generic.List[hashtable]]::new() - foreach ($rawJob in $rawJobList) { if ($rawJob -is [hashtable]) { $jobs.Add($rawJob) } } - if ($jobs.Count -ne $rawJobList.Count) { - Write-PkMessage ("ProcessDispatch: -JobBuilder returned " + ($rawJobList.Count - $jobs.Count) + " value(s) that are not job hashtables - ignored. Each job must be a hashtable.") -Level Warn - } - foreach ($j in $jobs) { - $j['GroupKey'] = $groupKey - if (-not $j.ContainsKey('Group')) { $j['Group'] = $groupInfo } # leave a job's own 'Group' field alone - $pending.Enqueue($j) - } - $state = @{ GroupInfo = $groupInfo; Outstanding = $jobs.Count; Completed = $false } + $jobCount = Add-PkDispatchGroupJobs -JobBuilder $JobBuilder -GroupInfo $groupInfo -GroupKey $groupKey -Pending $pending + $state = @{ GroupInfo = $groupInfo; Outstanding = $jobCount; Completed = $false } $groupStates[$groupKey] = $state $activeGroupCount++ if ($state.Outstanding -le 0) { @@ -1727,7 +1746,7 @@ function Get-PkWorkerSafeFunctionNames { return @('Write-PkMessage', 'Test-PkAbortSignal', 'Test-PkCancelRequested', 'Initialize-PkThreadPool', 'Stop-PkProcessTree', 'Stop-PkJobProcesses', 'Resolve-PkEncoding', 'Start-PkProcess', 'Wait-PkProcess', 'Complete-PkProcess', 'Invoke-PkProcessWithTimeout', 'Invoke-PkProcessDispatch', 'Invoke-PkProcessDispatchCore', 'Invoke-PkProcessBatch', - 'Register-PkCancellationSource', 'Unregister-PkCancellationSource', 'New-PkBatchResult', 'ConvertTo-PkArgumentString', 'Get-PkForceStopReason', 'Resolve-PkFileSystemPath') + 'Register-PkCancellationSource', 'Unregister-PkCancellationSource', 'New-PkBatchResult', 'ConvertTo-PkArgumentString', 'Get-PkForceStopReason', 'Resolve-PkFileSystemPath', 'Add-PkDispatchGroupJobs') } function Get-PkHandlerParamCount { diff --git a/README.md b/README.md index a6fb957..f1b6efd 100644 --- a/README.md +++ b/README.md @@ -322,7 +322,7 @@ Other pieces: `Request-PkAbort` triggers the same abort from code. `Test-PkAbort - 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 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. ## Testing