Runspaces/Runspaces.ps1
|
class Runspaces { $Caller = $null #-------------------------------------- # Private properties #-------------------------------------- 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 #-------------------------------------- # Public properties #-------------------------------------- [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 #-------------------------------------- # Constructor #-------------------------------------- Runspaces( [string]$ScriptFile) { $this._scriptFile = $ScriptFile } #-------------------------------------- # Private methods #-------------------------------------- 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()# $false ) if( -not $more ) { $_this.endJobs() } else { Write-Host "^" -NoNewline } }) } #Write-Host "starting DoEvents timer ..." $this._timer.start() $this._ticks = 0 } hidden [void] endJobs() { if($this._timer) { $this._timer.stop() } $this._ticks = 0 $this._running = $false #if (-not $this.Quiet) { # Write-Progress -Id $this._progressId -Activity "Running Query" -Status "Starting threads" -Completed #} $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() { #No max queue specified? Estimate one. 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 they want to import variables or modules, create a clean runspace, get loaded items, use those to exclude items if ($this.ImportVariables -or $this.ImportModules -or $this.ImportFunctions) { $StandardUserEnv = [powershell]::Create().addscript({ #Get modules, snapins, functions in this clean runspace $Modules = Get-Module | Select-Object -ExpandProperty Name # NOTE: -snapins removed from support in core 6 & 7 #$Snapins = $null #if( $PSVersionTable.PSVersion.Major -eq 5 ) { $Snapins = Get-PSSnapin | Select-Object -ExpandProperty Name #} $Functions = Get-ChildItem function:\ | Select-Object -ExpandProperty Name #Get variables in this clean runspace #Called last to get vars like $? into session $Variables = Get-Variable | Select-Object -ExpandProperty Name #Return a hashtable where we can access each. @{ Variables = $Variables Modules = $Modules Snapins = $Snapins Functions = $Functions } },$true).invoke()[0] if ($this.ImportVariables) { #Exclude common parameters, bound parameters, and automatic variables 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 ", ")" # we don't use 'Get-Variable -Exclude', because it uses regexps. # One of the veriables that we pass is '$?'. # There could be other variables with such problems. # Scope 2 required if we move to a real module $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) { $this._userModules = @( Get-Module | Where-Object {$StandardUserEnv.Modules -notcontains $_.Name -and (Test-Path $_.Path -ErrorAction SilentlyContinue)} | Select-Object -ExpandProperty Path ) #if( $PSVersionTable.PSVersion.Major -eq 5 ) { $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() { $this._scriptBlock = [scriptblock]::Create( $(Get-Content $this._scriptFile | out-string) ) #TR "`$ScriptBlock: $($this._scriptBlock | Out-String)" TR "Creating runspace pool and session states" #If specified, add variables and modules/snapins to session state $sessionstate = [System.Management.Automation.Runspaces.InitialSessionState]::CreateDefault() 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( $PSVersionTable.PSVersion.Major -eq 5 ) { 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)) } } #Create runspace pool $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() #New-Object System.Collections.ArrayList <# #If inputObject is bound get a total count and set bound to true $bound = $PSBoundParameters.keys -contains "InputObject" if(-not $bound) { [System.Collections.ArrayList]$allObjects = @() } #> $this._allObjects = [System.Collections.ArrayList]::new() #Set up log file if specified 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 } #write initial log entry $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 #counts for progress $this._totalCount = 0 $this._completedCount = 0 $this._startedCount = 0 $this.statusUpdate($true) } hidden [void] addRunspaceJobs($pack) { [void]$this._allObjects.add( $pack ) #counts for progress $this._totalCount += $this._allObjects.count foreach($object in $this._allObjects) { #region add scripts to runspace pool #Create the powershell instance, set verbose if needed, supply the scriptblock and parameters $powershell = [powershell]::Create() <# if ($VerbosePreference -eq 'Continue') { [void]$powerShell.AddScript({$VerbosePreference = 'Continue'}) } #> [void]$powershell.AddScript($this._scriptBlock).AddArgument($object) if ($this.Parameter) { [void]$powershell.AddArgument($this.Parameter) } <# # $Using support from Boe Prox if ($UsingVariableData) { Foreach($UsingVariable in $UsingVariableData) { Write-Verbose "Adding $($UsingVariable.Name) with value: $($UsingVariable.Value)" [void]$PowerShell.AddArgument($UsingVariable.Value) } } #> #Add the runspace into the powershell instance $powershell.RunspacePool = $this._runspacepool #Create a temporary collection for each runspace $temp = "" | Select-Object PowerShell, StartTime, object, Runspace $temp.PowerShell = $powershell $temp.StartTime = Get-Date $temp.object = $object #Save the handle output when calling BeginInvoke() that will be used later to end the runspace $temp.Runspace = $powershell.BeginInvoke() $this._startedCount++ $this._commandFile += "$($temp.object.ts_RuleSet) " if( $this._beginCallback ) { $this._beginCallback.Invoke($this, $temp.object.ts_RuleSet) } #Add the temp tracking info to $runspaces collection TR ( "Adding {0} to collection at {1}" -f $temp.object, $temp.starttime.tostring() ) $this._runspaces.Add($temp) | Out-Null #loop through existing runspaces one time $this.getRunspaceData()# $false ) #If we have more running than max queue (used to control timeout accuracy) #Script scope resolves odd PowerShell 2 issue $firstRun = $true while ($this._runspaces.count -ge $this.MaxQueue) { #give verbose output if($firstRun) { TR "$($this._runspaces.count) items running - exceeded $($this.MaxQueue) limit." } $firstRun = $false #run get-runspace data and sleep for a short while $this.getRunspaceData()# $false ) Start-Sleep -Milliseconds $this.SleepTimer } #endregion add scripts to runspace pool } $this._allObjects.Clear() } hidden [void] beginJobs($pack) { if( -not $this._running) { $this._running = $true if( $this._startCallback) { $this._startCallback.Invoke($this) } } #Use Try/Finally to catch Ctrl+C and clean up. try { $this.addRunspaceJobs($pack) TR ( "Finish processing the remaining runspace jobs: {0}" -f ( @($this._runspaces | Where-Object {$Null -ne $_.Runspace}).Count) ) $this.waitForJobs() } finally { <# #Close the runspace pool, unless we specified no close on timeout and something timed out if ( ($this._timedOutTasks -eq $false) -or ( ($this._timedOutTasks -eq $true) -and ($this.NoCloseOnTimeout -eq $false) ) ) { Write-Verbose "Closing the runspace pool" $this._runspacepool.close() } #collect garbage [gc]::Collect() #> } #$this._running = $false #if( $this._stopCallback) { $this._stopCallback.Invoke($this) } } 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]$ruleName = "Saving $($this._startedCount) filters..." #"Executing rule: $($args[0])" #write-host ("Creating shortcut for {0} with path {1} in folder {2}..." -f $_.Key, $_.Value[0], $_.Value[1]) [string]$messageText = $this._messageText -f $this._commandFile #$($this._startedCount) if($completed) { $messageText = "Ready" } #[string]$counter = "$($this._completedCount)" #"$($args[1])" [int]$cnt = $($($this._completedCount)*10 + $this._ticks) [string]$counter = "$cnt" #"$($args[1])" #[string]$maxcount = "$($this._totalCount)" #"$($args[2])"\ [int]$maxcnt = $($($this._totalCount)*10) [string]$maxcount = "$maxcnt" #"$($args[2])"\ if( $this._completedCount -gt $this._totalCount ) { [int]$maxcnt = $($($this._completedCount)*10) #$maxcount = "$($this._completedCount)" $maxcount = "$maxcnt" } if($cnt -ge $maxcnt) { $counter = "$maxcnt" } [string]$message = "$messageText,$counter,$maxcount" if($this._statusCallback) { $this._statusCallback.Invoke($this, $message) } } } hidden [boolean] getRunspaceData() { # [boolean]$wait = $false) { #[cmdletbinding()] #param( [switch]$Wait ) #loop through runspaces #if $wait is specified, keep looping until all complete $more = $false # Do { #set more to false for tracking completion $more = $false #Progress bar if we have inputobject count (bound parameter) $this.statusUpdate($false) #run through each runspace. $this._commandFile = "" foreach($runspace in $this._runspaces) { #get the duration - inaccurate $currentdate = Get-Date $runtime = $currentdate - $runspace.startTime $runMin = [math]::Round( $runtime.totalminutes ,2 ) #set up log object $log = "" | Select-Object Date, Action, Runtime, Status, Details $log.Action = "Removing:'$($runspace.object)'" $log.Date = $currentdate $log.Runtime = "$runMin minutes" #If runspace completed, end invoke, dispose, recycle, counter++ If ($runspace.Runspace.isCompleted) { $this._completedCount++ #check if there were errors if($runspace.powershell.Streams.Error.Count -gt 0) { #set the logging info and move the file to completed $log.status = "CompletedWithErrors" Write-Verbose ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] foreach($ErrorRecord in $runspace.powershell.Streams.Error) { Write-Error -ErrorRecord $ErrorRecord } } else { #add logging details and cleanup $log.status = "Completed" Write-Verbose ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] } #everything is logged, clean up the runspace $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) } } #If runtime exceeds max, dispose the runspace elseIf ( $this.RunspaceTimeout -ne 0 -and $runtime.totalseconds -gt $this.RunspaceTimeout) { $this._completedCount++ $this._timedOutTasks = $true #add logging details and cleanup $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)" #Depending on how it hangs, we could still get stuck here as dispose calls a synchronous method on the powershell instance if (!$this.NoCloseOnTimeout) { $runspace.powershell.dispose() } $runspace.Runspace = $null $runspace.powershell = $null if( $this._errorCallback ) { $this._errorCallback.Invoke($this, $runspace.object.ts_RuleSet) } } #If runspace isn't null set more to true elseIf ($null -ne $runspace.Runspace ) { $log = $null $more = $true if( $null -ne $runspace.object.ts_RuleSet) { $this._commandFile += "$($runspace.object.ts_RuleSet) "} } #log the results if a log file was indicated if($this.LogFile -and $log) { ($log | ConvertTo-Csv -Delimiter ";" -NoTypeInformation)[1] | out-file $this.LogFile -append } } #Clean out unused runspace jobs $temphash = $this._runspaces.clone() $temphash | Where-Object { $Null -eq $_.runspace } | ForEach-Object { $this._runspaces.remove($_) } #sleep for a bit if we will loop again # if($wait){ Start-Sleep -milliseconds $this.SleepTimer } #Loop again only if -wait parameter and there are more runspaces to process # } while ($more -and $wait) return $more #End of runspace function } 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#.Clone() <# Write-Host "<<<<<<<<<<<<<<<<<<< updateFilter >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" Write-Host ">>>>>BEFORE: filterNodes:" Write-Host "-------------------------" Write-Host ($this._filterNodes | FT | Out-String) #> [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) <# Write-Host ">>>>> termNode---------------" Write-Host ($termNode | FT | Out-String) Write-Host "-----------------------------" #> if ($operation -eq "add") { #if( $null -ne $term) { $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 { } <# Write-Host ">>>>>AFTER: filterNodes:" Write-Host "------------------------" Write-Host ($this._filterNodes | FT | Out-String) Write-Host ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" #> } #-------------------------------------- # Public methods #-------------------------------------- [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.updateFilter($pack) $this.beginJobs($pack) return $null #$this._filterNodes } [void] Clear() { $this.endJobs() #if( $this._stopCallback) { $this._stopCallback.Invoke($this) } } [void] Close (){ $this.endJobs() #Close the runspace pool, unless we specified no close on timeout and something timed out if ( ($this._timedOutTasks -eq $false) -or ( ($this._timedOutTasks -eq $true) -and ($this.NoCloseOnTimeout -eq $false) ) ) { Write-Verbose "Closing the runspace pool" $this._runspacepool.close() } #collect garbage [gc]::Collect() # if( $this._closeCallback) { $this._closeCallback.Invoke($this) } } } |