Runspaces/Runspaces.ps1
|
class Runspaces { $Caller = $null static [string] $JobModuleName = 'SRxCore' static [boolean] $FailOnErrors = $true static [int] $MaxReportedErrors = 5 hidden $_jobScriptBlock = $null hidden $_host = $Host hidden [System.Windows.Forms.Timer]$_timer = $null hidden [System.Management.Automation.ScriptBlock]$_closeCallback = $null hidden [System.Management.Automation.ScriptBlock]$_beginCallback = $null hidden [System.Management.Automation.ScriptBlock]$_endCallback = $null hidden [System.Management.Automation.ScriptBlock]$_errorCallback = $null hidden [System.Management.Automation.ScriptBlock]$_startCallback = $null hidden [System.Management.Automation.ScriptBlock]$_statusCallback = $null hidden [System.Management.Automation.ScriptBlock]$_stopCallback = $null hidden [string]$_messageText = "Saving {0} filters..." hidden $_progressId = $null hidden $_running = $false hidden [int]$_startedCount = 0 hidden [int]$_totalCount = 0 hidden [int]$_completedCount = 0 hidden [string]$_commandFile = "" hidden [int]$_ticks = 0 hidden $_runspacepool = $null hidden $_runspaces = $null hidden [System.Collections.ArrayList]$_allObjects = $null hidden [boolean]$_timedOutTasks = $false hidden [string]$_scriptFile = $null hidden $_scriptBlock = $null hidden [System.Collections.ArrayList] $_filterNodes = $null hidden $_userModules = $null hidden $_userSnapins = $null hidden $_userVariables = $null hidden $_userFunctions = $null [PSObject]$Parameter = $null [boolean]$ImportVariables = $false [boolean]$ImportModules = $false [boolean]$ImportFunctions = $false [int]$Throttle = 20 [int]$SleepTimer = 200 [int]$RunspaceTimeout = 0 [boolean]$NoCloseOnTimeout = $false [int]$MaxQueue = $null [boolean]$AppendLog = $false [string]$LogFile = $null [boolean]$Quiet = $false [string[]]$ModuleFilter = @() [boolean]$SilenceImportWarnings = $true Runspaces( [string]$ScriptFile) { $this._scriptFile = $ScriptFile } hidden [void] waitForJobs() { if( -not $this._timer) { $this._timer = [System.Windows.Forms.Timer]::new() $this._timer.Interval = 5000 $this._timer.Tag = $this $this._timer.add_tick( { param($source, $e) $_this = $source.Tag $_this._ticks++ $more = $_this.getRunspaceData() if( -not $more ) { $_this.endJobs() } else { Write-Host "^" -NoNewline } }) } $this._timer.start() $this._ticks = 0 } hidden [void] endJobs() { if($this._timer) { $this._timer.stop() } $this._ticks = 0 $this._running = $false $this.statusUpdate($true) $this._totalCount = 0 $this._completedCount = 0 $this._startedCount = 0 $this._allObjects.Clear() if( $this._stopCallback) { $this._stopCallback.Invoke($this) } } hidden [void] load() { if( -not $this.MaxQueue ) { if($this.RunspaceTimeout -ne 0){ $this.MaxQueue = $this.Throttle } else{ $this.MaxQueue = $this.Throttle * 3 } } $this._progressId = Get-Random TR "Throttle: '$($this.Throttle)' SleepTimer '$($this.SleepTimer)' runSpaceTimeout '$($this.RunspaceTimeout)' maxQueue '$($this.MaxQueue)' logFile '$($this.LogFile)'" if ($this.ImportVariables -or $this.ImportModules -or $this.ImportFunctions) { $StandardUserEnv = [powershell]::Create().addscript({ $Modules = Get-Module | Select-Object -ExpandProperty Name $Snapins = Get-PSSnapin | Select-Object -ExpandProperty Name $Functions = Get-ChildItem function:\ | Select-Object -ExpandProperty Name $Variables = Get-Variable | Select-Object -ExpandProperty Name @{ Variables = $Variables Modules = $Modules Snapins = $Snapins Functions = $Functions } },$true).invoke()[0] if ($this.ImportVariables) { Function _temp {[cmdletbinding(SupportsShouldProcess=$True)] param() } $VariablesToExclude = @( (Get-Command _temp | Select-Object -ExpandProperty parameters).Keys + $PSBoundParameters.Keys + $StandardUserEnv.Variables ) Write-Verbose "Excluding variables $( ($VariablesToExclude | Sort-Object ) -join ", ")" $this._userVariables = @( Get-Variable | Where-Object { -not ($VariablesToExclude -contains $_.Name) } ) Write-Verbose "Found variables to import: $( ($this._userVariables | Select-Object -expandproperty Name | Sort-Object ) -join ", " | Out-String).`n" } if ($this.ImportModules) { $mods = @( Get-Module | Where-Object {$StandardUserEnv.Modules -notcontains $_.Name -and (Test-Path $_.Path -ErrorAction SilentlyContinue)} ) $filter = @($this.ModuleFilter | Where-Object { -not [string]::IsNullOrWhiteSpace([string]$_) }) if ($filter.Count -gt 0) { $keep = New-Object System.Collections.ArrayList foreach ($m in $mods) { $hit = $false foreach ($f in $filter) { if ([string]$m.Name -like [string]$f) { $hit = $true; break } } if ($hit) { [void]$keep.Add($m) } else { TR " > Module skipped (ModuleFilter): $($m.Name)" } } $mods = @($keep) } $this._userModules = @( $mods | Select-Object -ExpandProperty Path ) $this._userSnapins = @( Get-PSSnapin | Select-Object -ExpandProperty Name | Where-Object {$StandardUserEnv.Snapins -notcontains $_ } ) } if($this.ImportFunctions) { $this._userFunctions = @( Get-ChildItem function:\ | Where-Object { $StandardUserEnv.Functions -notcontains $_.Name } ) } } } hidden [void] init() { $scriptText = $(Get-Content $this._scriptFile | out-string) $this._scriptBlock = [scriptblock]::Create( $scriptText ) $this._jobScriptBlock = [scriptblock]::Create( [Runspaces]::JobWrapperText($scriptText) ) TR "Creating runspace pool and session states" $sessionstate = [System.Management.Automation.Runspaces.InitialSessionState]::CreateDefault() if ($this.SilenceImportWarnings) { $sessionstate.Variables.Add((New-Object -TypeName System.Management.Automation.Runspaces.SessionStateVariableEntry -ArgumentList 'WarningPreference', ([System.Management.Automation.ActionPreference]::SilentlyContinue), $null)) $sessionstate.Variables.Add((New-Object -TypeName System.Management.Automation.Runspaces.SessionStateVariableEntry -ArgumentList 'VerbosePreference', ([System.Management.Automation.ActionPreference]::SilentlyContinue), $null)) } if($this.ImportVariables -and $this._userVariables.count -gt 0) { foreach($Variable in $this._userVariables) { $sessionstate.Variables.Add((New-Object -TypeName System.Management.Automation.Runspaces.SessionStateVariableEntry -ArgumentList $Variable.Name, $Variable.Value, $null) ) } } if ($this.ImportModules) { if($this._userModules.count -gt 0) { foreach($ModulePath in $this._userModules) { TR " > ModulePath = $ModulePath" $sessionstate.ImportPSModule($ModulePath) } } if($this._userSnapins.count -gt 0) { foreach($PSSnapin in $this._userSnapins) { [void]$sessionstate.ImportPSSnapIn($PSSnapin, [ref]$null) } } } if($this.ImportFunctions -and $this._userFunctions.count -gt 0) { foreach ($FunctionDef in $this._userFunctions) { $sessionstate.Commands.Add((New-Object System.Management.Automation.Runspaces.SessionStateFunctionEntry -ArgumentList $FunctionDef.Name,$FunctionDef.ScriptBlock)) } } $this._runspacepool = [runspacefactory]::CreateRunspacePool(1, $this.Throttle, $sessionstate, $this._host) $this._runspacepool.Open() TR "Creating empty collection to hold runspace jobs" $this._runspaces = [System.Collections.ArrayList]::new() $this._allObjects = [System.Collections.ArrayList]::new() if( $this.LogFile -and (-not (Test-Path $this.LogFile) -or $this.AppendLog -eq $false)){ New-Item -ItemType file -Path $this.LogFile -Force | Out-Null ("" | Select-Object -Property Date, Action, Runtime, Status, Details | ConvertTo-Csv -NoTypeInformation -Delimiter ";")[0] | Out-File $this.LogFile } $log = "" | Select-Object -Property Date, Action, Runtime, Status, Details $log.Date = Get-Date $log.Action = "Batch processing started" $log.Runtime = $null $log.Status = "Started" $log.Details = $null if($this.LogFile) { ($log | convertto-csv -Delimiter ";" -NoTypeInformation)[1] | Out-File $this.LogFile -Append } $this._timedOutTasks = $false $this._totalCount = 0 $this._completedCount = 0 $this._startedCount = 0 $this.statusUpdate($true) } hidden [void] addRunspaceJobs($pack) { [void]$this._allObjects.add( $pack ) $this._totalCount += $this._allObjects.count foreach($object in $this._allObjects) { $powershell = [powershell]::Create() $isJobTask = $false try { $isJobTask = -not [string]::IsNullOrEmpty([string]$object.ts_JobToken) } catch { $isJobTask = $false } if ($isJobTask -and $null -ne $this._jobScriptBlock) { [void]$powershell.AddScript($this._jobScriptBlock).AddArgument($object) } else { [void]$powershell.AddScript($this._scriptBlock).AddArgument($object) } if ($this.Parameter) { [void]$powershell.AddArgument($this.Parameter) } $powershell.RunspacePool = $this._runspacepool $temp = "" | Select-Object PowerShell, StartTime, object, Runspace $temp.PowerShell = $powershell $temp.StartTime = Get-Date $temp.object = $object $temp.Runspace = $powershell.BeginInvoke() $this._startedCount++ $this._commandFile += "$($temp.object.ts_RuleSet) " if( $this._beginCallback ) { $this._beginCallback.Invoke($this, $temp.object.ts_RuleSet) } TR ( "Adding {0} to collection at {1}" -f $temp.object, $temp.starttime.tostring() ) $this._runspaces.Add($temp) | Out-Null $this.getRunspaceData() $firstRun = $true while ($this._runspaces.count -ge $this.MaxQueue) { if($firstRun) { TR "$($this._runspaces.count) items running - exceeded $($this.MaxQueue) limit." } $firstRun = $false $this.getRunspaceData() Start-Sleep -Milliseconds $this.SleepTimer } } $this._allObjects.Clear() } hidden [void] beginJobs($pack) { if( -not $this._running) { $this._running = $true if( $this._startCallback) { $this._startCallback.Invoke($this) } } try { $this.addRunspaceJobs($pack) TR ( "Finish processing the remaining runspace jobs: {0}" -f ( @($this._runspaces | Where-Object {$Null -ne $_.Runspace}).Count) ) $this.waitForJobs() } finally { } } hidden [void] statusUpdate( [boolean]$completed) { if (-not $this.Quiet) { if( -not $completed) { Write-Progress -Id $this._progressId -Activity "Running Query" -Status "Starting threads"` -CurrentOperation "$($this._startedCount) threads defined - $($this._totalCount) input objects - $($this._completedCount) input objects processed"` -PercentComplete $( Try { $($this._completedCount) / $($this._totalCount) * 100 } Catch {0} ) } else { Write-Progress -Id $this._progressId -Activity "Running Query" -Status "Starting threads" -Completed } } if( $this._statusCallback) { [string]$messageText = $this._messageText -f $this._commandFile if($completed) { $messageText = "Ready" } [int]$cnt = $($($this._completedCount)*10 + $this._ticks) [string]$counter = "$cnt" [int]$maxcnt = $($($this._totalCount)*10) [string]$maxcount = "$maxcnt" if( $this._completedCount -gt $this._totalCount ) { [int]$maxcnt = $($($this._completedCount)*10) $maxcount = "$maxcnt" } if($cnt -ge $maxcnt) { $counter = "$maxcnt" } [string]$message = "$messageText,$counter,$maxcount" if($this._statusCallback) { $this._statusCallback.Invoke($this, $message) } } } hidden [boolean] getRunspaceData() { $more = $false $more = $false $this.statusUpdate($false) $this._commandFile = "" foreach($runspace in $this._runspaces) { $currentdate = Get-Date $runtime = $currentdate - $runspace.startTime $runMin = [math]::Round( $runtime.totalminutes ,2 ) $log = "" | Select-Object Date, Action, Runtime, Status, Details $log.Action = "Removing:'$($runspace.object)'" $log.Date = $currentdate $log.Runtime = "$runMin minutes" If ($runspace.Runspace.isCompleted) { $this._completedCount++ if($runspace.powershell.Streams.Error.Count -gt 0) { $log.status = "CompletedWithErrors" Write-Verbose ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] foreach($ErrorRecord in $runspace.powershell.Streams.Error) { Write-Error -ErrorRecord $ErrorRecord } } else { $log.status = "Completed" Write-Verbose ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] } $runspace.powershell.EndInvoke($runspace.Runspace) $runspace.powershell.dispose() $runspace.Runspace = $null $runspace.powershell = $null if( $this._endCallback ) { $this._endCallback.Invoke($this, $runspace.object.ts_RuleSet, $runspace.object) } } elseIf ( $this.RunspaceTimeout -ne 0 -and $runtime.totalseconds -gt $this.RunspaceTimeout) { $this._completedCount++ $this._timedOutTasks = $true $log.status = "TimedOut" Write-Verbose ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] Write-Error "Runspace timed out at $($runtime.totalseconds) seconds for the object:`n$($runspace.object | out-string)" if (!$this.NoCloseOnTimeout) { $runspace.powershell.dispose() } $runspace.Runspace = $null $runspace.powershell = $null if( $this._errorCallback ) { $this._errorCallback.Invoke($this, $runspace.object.ts_RuleSet, $runspace.object) } } elseIf ($null -ne $runspace.Runspace ) { $log = $null $more = $true if( $null -ne $runspace.object.ts_RuleSet) { $this._commandFile += "$($runspace.object.ts_RuleSet) "} } if($this.LogFile -and $log) { ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] | out-file $this.LogFile -append } } $temphash = $this._runspaces.clone() $temphash | Where-Object { $Null -eq $_.runspace } | ForEach-Object { $this._runspaces.remove($_) } return $more } hidden [void] removeFilter( $filterNode) { $selection = ($($this._filterNodes) | Where-Object { -not (($($_.xPath) -eq $($filterNode.xPath)) -and (($_.Scope) -eq $($filterNode.Scope) -and (($_.optype) -eq $($filterNode.optype)))) }) $this._filterNodes = [System.Collections.ArrayList]::new() if( $selection ) { $this._filterNodes += $selection } } hidden [void] addFilter( $filterNode) { $this.removeFilter( $filterNode) $this._filterNodes += $filterNode } hidden [void] updateFilter($pack) { $this._filterNodes = $pack.filterNodes [System.Collections.Stack]$nodesStack = $pack.nodesStack.Clone() while( $($nodesStack.Count) -gt 1 ) { $nodesStack.Pop() } $termNode = $nodesStack.Peek() [string]$operation = $($termNode.Operation) [int]$optype = $($termNode.optype) if ($operation -eq "add") { $details = @{ xPath = $termNode.xPath Scope = $termNode.Scope optype = $optype Filter = $termNode.Attributes.Clone() } $object = New-Object PSObject -Property $details $this.addFilter($object) } elseif($operation -eq "remove") { $this.removeFilter($termNode) } else { } } static [string] JobWrapperText([string]$scriptText) { $head = @' param($__pack) $__ok = $true $__c = $null $__errBase = $Error.Count $__errs = New-Object System.Collections.ArrayList # Phase 7 IT5: errors that reached the task's error stream try { # IT3.1 r3: never run against Explorer's OWN $global:SRxEnv. ImportVariables hands the # pool a reference to it; its script methods belong to Explorer's runspace, so every # call from this thread is marshalled into Explorer's busy UI runspace and breaks both # sides ("'Test-Path' / 'New-Object' is not recognized"). The ruleset script builds a # runspace-local one (ExecuteRulesetJob.ps1 -> Initialize-SRxEnv), as a child process does. Remove-Variable -Name SRxEnv -Scope Global -Force -ErrorAction Ignore # IT3.1 r4: that Initialize-SRxEnv must not touch the console (title / buffer size): # $Host of a pool runspace is Explorer's own console. -NoHostUI by default in THIS runspace. if ($null -eq $global:PSDefaultParameterValues) { $global:PSDefaultParameterValues = @{} } $global:PSDefaultParameterValues['Initialize-SRxEnv:NoHostUI'] = $true # the SRxCore that exports the CURRENT Job functions (not any other loaded copy) $__mod = $null $__cmd = Get-Command Connect-SRxJobsChild -ErrorAction Ignore if ($null -ne $__cmd) { $__mod = $__cmd.Module } if ($null -eq $__mod) { $__mod = Get-Module -Name '__MODULE__' -ErrorAction Ignore | Sort-Object Version -Descending | Select-Object -First 1 } if ($null -ne $__mod) { $__c = & $__mod { [SRxBasePipeChildAPI]::new() } } else { $__c = New-Object -TypeName 'SRxBasePipeChildAPI' } # older class copies have no NoSync: then no sync worker is started anyway in a runspace if ($null -ne $__c.PSObject.Properties['NoSync']) { $__c.NoSync = $true } if (-not $__c.Pull([string]$__pack.ts_JobToken, [string]$__pack.ts_JobPipe)) { throw ('Job token not pulled: ' + $__c.LastError) } $global:SRxJobChild = $__c $global:__onRuleCallback = $__c.RuleCallback() $global:__onRuleCompletedCallback = $__c.RuleCompletedCallback() [void]$__c.Hello() # legacy pipe-Shell / pipe-Explorer client calls of the script -> this Job (same as Job windows) if (Get-Command Install-SRxJobChannels -ErrorAction Ignore) { try { Install-SRxJobChannels } catch { [void]$__c.Post('log', ('Install-SRxJobChannels: ' + $_.Exception.Message)) } } $__task = $__c.Command if ($null -eq $__task) { $__task = $__pack } # IT3.1 r5/r6: the task's own SRxEnv is built HERE (the ruleset script's # Initialize-SRxEnv then finds it); its changes are diffed and sent to # Explorer through the Job ('env' before 'complete'). The SRxEnv object is # passed explicitly (a class method may see another runspace's globals). $__syncEnv = ($null -ne $__c.PSObject.Methods['CaptureEnvBaseline']) -and [bool]$__c.SyncEnv if ($__syncEnv -and $null -eq $global:SRxEnv) { if (Get-Command Initialize-SRxEnv -ErrorAction Ignore) { $__sid = $null try { $__sid = $__task.ts_SourceID } catch { $__sid = $null } try { Initialize-SRxEnv -SourceID $__sid } # -NoHostUI via PSDefaultParameterValues (r4) catch { [void]$__c.Post('problem', ('Initialize-SRxEnv failed: ' + $_.Exception.Message)) } } else { [void]$__c.Post('problem', 'Initialize-SRxEnv is not available in the runspace - SRxEnv changes cannot be reported') } } if ($__syncEnv) { if ($null -eq $global:SRxEnv) { [void]$__c.Post('problem', 'No SRxEnv in the runspace after Initialize-SRxEnv') } else { $__c.CaptureEnvBaseline($global:SRxEnv) } } $__errBase = $Error.Count # errors of the ruleset script only $__inner = { '@ $tail = @' } # Phase 7 IT5: only UNHANDLED errors (the error stream) count - $Error also holds # handled ones (try / catch, -ErrorAction SilentlyContinue), e.g. a connection probe & $__inner $__task 2>&1 | ForEach-Object { if ($_ -is [System.Management.Automation.ErrorRecord]) { [void]$__errs.Add($_) } } } catch { $__ok = $false if ($null -ne $__c -and $__c.Ready) { [void]$__c.Post('problem', ('Task failed: ' + $_.Exception.Message)) } else { throw } } finally { if ($null -ne $__c -and $__c.Ready) { $__new = $__errs.Count if ($__new -gt 0) { $__n = [Math]::Min($__new, __MAXERR__) for ($__i = 0; $__i -lt $__n; $__i++) { [void]$__c.Post('problem', ('Error: ' + [string]$__errs[$__i])) } if ($__new -gt $__n) { [void]$__c.Post('problem', ('... and {0} more error(s)' -f ($__new - $__n))) } if (__FAILONERR__) { $__ok = $false } } # IT3.1 r6: 'env' from THIS runspace's SRxEnv, before 'complete' (also on failure) if ($null -ne $__c.PSObject.Methods['PublishEnvChanges'] -and [bool]$__c.SyncEnv) { try { [void]$__c.PublishEnvChanges($global:SRxEnv) } catch { [void]$__c.Post('problem', ('SRxEnv report failed: ' + $_.Exception.Message)) } } [void]$__c.Complete($__ok) } $global:SRxJobChild = $null $global:__onRuleCallback = $null $global:__onRuleCompletedCallback = $null } '@ $h = $head.Replace('__MODULE__', [Runspaces]::JobModuleName) $t = $tail.Replace('__MAXERR__', [string][Runspaces]::MaxReportedErrors) if ([Runspaces]::FailOnErrors) { $t = $t.Replace('__FAILONERR__', '$true') } else { $t = $t.Replace('__FAILONERR__', '$false') } return ($h + "`n" + $scriptText + "`n" + $t) } [void] MessageText($formattedString) { $this._messageText = $formattedString } [void] OnStart( [System.Management.Automation.ScriptBlock]$onStartCallback ) { $this._startCallback = $onStartCallback } [void] OnStatus( [System.Management.Automation.ScriptBlock]$onStatusCallback ) { $this._statusCallback = $onStatusCallback } [void] OnBegin( [System.Management.Automation.ScriptBlock]$onBeginCallback ) { $this._beginCallback = $onBeginCallback } [void] OnEnd( [System.Management.Automation.ScriptBlock]$onEndCallback ) { $this._endCallback = $onEndCallback } [void] OnError( [System.Management.Automation.ScriptBlock]$onErrorCallback ) { $this._errorCallback = $onErrorCallback } [void] OnStop( [System.Management.Automation.ScriptBlock]$onStopCallback ) { $this._stopCallback = $onStopCallback } [void] OnClose( [System.Management.Automation.ScriptBlock]$onCloseCallback ) { $this._closeCallback = $onCloseCallback } [void] Open (){ $this.load() $this.init() } [boolean]Running() { return $this._running } [System.Collections.ArrayList] AddPack($pack) { $this.beginJobs($pack) return $null } [void] Clear() { $this.endJobs() } [void] Close (){ $this.endJobs() if ( ($this._timedOutTasks -eq $false) -or ( ($this._timedOutTasks -eq $true) -and ($this.NoCloseOnTimeout -eq $false) ) ) { Write-Verbose "Closing the runspace pool" $this._runspacepool.close() } [gc]::Collect() if( $this._closeCallback) { $this._closeCallback.Invoke($this) } } } |