Invoke-Parallel.ps1
|
function Invoke-Parallel { <# .Synopsis Invokes PowerShell in parallel .Description Invokes PowerShell in parallel and in the same proess space. This maximizes performance for parallel execution in PowerShell, by removing the drag associated with background processes. .Example 1..10 | Invoke-Parallel -Command { (Get-Date).ToShortTimeString() + "." + (Get-Date).Millisecond } -SliceSize 1 -MaxRunning 10 .Example 1..10 | Invoke-Parallel -Command { foreach ($a in $args) { "$a - " + ((Get-Date).ToShortTimeString() + "." + (Get-Date).Millisecond) } } .Notes Invoke-Parallel is an alternative to using PowerShell workflows to run PowerShell in parallel. Because PowerShell workflows involve process isolation, each workflow takes up a lot of overhead for a small operation. This overhead is considerably increased when the workflow is expected to return information, since this information has to be deserialized and brought into the process via interprocess communication (IPC). Doing this takes up a lot of time and requires the movement of a lot of memory, which also slows down workflows. Invoke-Parallel uses PowerShell RunspacePools to run code in parallel within the current process, which saves a lot of overhead. .Link Update-SQL #> [OutputType([PSObject])] param( # The list of input objects [Parameter(Mandatory=$true,ValueFromPipeline=$true,Position=1)] [PSObject[]] $InputObject, # The command [Parameter(Mandatory=$true,Position=0)] [ScriptBlock] $Command, # The size of each slice of data. By default, this is the square root of the number of items [Parameter(Position=3)] [Uint32] $SliceSize, # The maximum number of running items. By default, this is the square root of the slice size. [Parameter(Position=4)] [Uint32] $MaxRunning ) begin { #region Declare the type if (-not ('StartAutomating.ParallelInvoker' -as [Type])) { $t = Add-Type -TypeDefinition @" using System; using System.Collections; using System.Collections.Generic; using System.Management.Automation; using System.Management.Automation.Runspaces; namespace StartAutomating { public class ParallelInvoker { public ParallelInvoker() { combinedOutput.DataAdded += new EventHandler<DataAddedEventArgs>(outputCollection_DataAdded); } void outputCollection_DataAdded(object sender, DataAddedEventArgs e) { PSDataCollection<PSObject> collection = sender as PSDataCollection<PSObject>; if (collection != null) { PSObject lastOutput = collection[e.Index]; pendingOutput.Enqueue(lastOutput); } else { // Progress record PSDataCollection<ProgressRecord> progressCollection = sender as PSDataCollection<ProgressRecord>; ProgressRecord lastProgress = progressCollection[e.Index]; pendingOutput.Enqueue(new PSObject(lastProgress)); } } PSDataCollection<PSObject> combinedOutput = new PSDataCollection<PSObject>(); Queue<PSObject> pendingOutput = new Queue<PSObject>(); public IEnumerator<PSObject> InvokeParallel(ScriptBlock sb, PSObject[]psObject, uint maxRunning = 0, uint sliceSize = 0) { combinedOutput.Clear(); pendingOutput.Clear(); List<PowerShell> runningJobs = new List<PowerShell>(); if (sliceSize == 0) { sliceSize = (uint)System.Math.Sqrt(psObject.Length); } if (maxRunning == 0) { maxRunning = (uint)System.Math.Sqrt(sliceSize); } if (maxRunning == 0) { maxRunning = 1; } RunspacePool runspacePool = RunspaceFactory.CreateRunspacePool(1, (int)maxRunning); runspacePool.Open(); uint index = 0; uint innerIndex = 0; List<PSObject> slice; string ssb = sb.ToString(); for(; index<psObject.Length;index+=sliceSize) { slice = new List<PSObject>(); for(innerIndex = index; innerIndex < psObject.Length && innerIndex < (index + sliceSize); innerIndex++) { slice.Add(psObject[innerIndex]); } PowerShell psCmd = PowerShell.Create(); psCmd.AddScript(ssb); psCmd.AddParameters(slice); psCmd.Commands.Commands[0].MergeMyResults(PipelineResultTypes.All, PipelineResultTypes.Output); psCmd.RunspacePool = runspacePool; psCmd.Streams.Progress.DataAdded += new EventHandler<DataAddedEventArgs>(outputCollection_DataAdded); IAsyncResult invocation = psCmd.BeginInvoke<Object, PSObject>(null, combinedOutput); runningJobs.Add(psCmd); } bool incompleteJobs = false; List<PowerShell> jobsToRemove = new List<PowerShell>(); do { incompleteJobs = false; while (pendingOutput.Count > 0) { yield return pendingOutput.Dequeue(); } foreach (PowerShell rj in runningJobs) { if (rj.InvocationStateInfo.State != PSInvocationState.Completed && rj.InvocationStateInfo.State != PSInvocationState.Failed && rj.InvocationStateInfo.State != PSInvocationState.Stopped && rj.InvocationStateInfo.State != PSInvocationState.Disconnected) { incompleteJobs = true; } else { if (rj.InvocationStateInfo.State == PSInvocationState.Failed && rj.InvocationStateInfo.Reason != null) { yield return new PSObject(new ErrorRecord(rj.InvocationStateInfo.Reason, " ", ErrorCategory.WriteError, rj)); } jobsToRemove.Add(rj); } } foreach (PowerShell ps in jobsToRemove) { runningJobs.Remove(ps); ps.Dispose(); } jobsToRemove.Clear(); } while (incompleteJobs); while (pendingOutput.Count > 0) { yield return pendingOutput.Dequeue(); } runspacePool.Close(); runspacePool.Dispose(); combinedOutput.Clear(); combinedOutput.Dispose(); } } } "@ -PassThru $null = $t } $accumulatededInput = New-Object Collections.Generic.List[PSObject] #endregion Declare the type } process { #region Accumulate Input $null = $accumulatededInput.AddRange($InputObject) #endregion Accumulate Input } end { #region Run in parallel foreach ($_ in ((New-OBject StartAutomating.ParallelInvoker).InvokeParallel($Command, $accumulatededInput.ToArray(), $MaxRunning, $SliceSize))) { if ($_ -is [Management.Automation.ProgressRecord]) { $progressParameters = @{ Id = $_.ActivityId ParentId = $_.ParentActivityId PercentComplete = $_.PercentComplete Activity = $_.Activity Status = $_.StatusDescription SecondsRemaining = $_.SecondsRemaining Completed = if ($_.RecordType -eq 'Processing') { $false } else { $true } } if ($progressParameters.SecondsRemaining -eq -1) { $progressParameters.Remove('SecondsRemaining') } if ($progressParameters.Id -eq -1) { $progressParameters.Remove('Id') } if ($progressParameters.ParentId -eq -1) { $progressParameters.Remove('ParentId') } if (-not $progressParameters.Completed) { $progressParameters.Remove('Completed') } Write-Progress @progressParameters } else { $_ } } #endregion Run in parallel } } |