Invoke-Async – Asynchronous parallel processing for any workload in PowerShell

Update (Mar 16, 2018):

Please check-out my most recent blog post related to this on how to use this to invoke your own PowerShell function in parallel:

https://sqljana.wordpress.com/2018/03/16/powershell-sql-server-run-in-parallel-collect-sql-results-with-print-output-from-across-your-sql-farm-fast/

Why asynchronous processing?

If it takes a man 10 hours to do something, technically, it would take just one hour if 10 men did it. Right? Not quite, but it is somewhat true in the analogous computing world if resources are hardly maxed out when 10 threads are at work. It is more true when the jobs happen to be waiting for another computer to respond with an answer. For example, if you wanted to find the free disk space on a 100 hosts, with “asynchronous processing”, you could be simultaneously querying 10 hosts or more at a time.

Traditional asynchronous processing in PowerShell:

Traditionally, if you wanted to process things in parallel using PowerShell, one would use the Start-Job, Stop-Job, Receive-Job, Remove-Job, Get-Job etc. (i.e., the *-Job set of commands)

Get-Command -Name '*-Job'

CommandType     Name
-----------     ----
Cmdlet          Get-Job
Cmdlet          Receive-Job
Cmdlet          Remove-Job
Cmdlet          Resume-Job
Cmdlet          Start-Job
Cmdlet          Stop-Job
Cmdlet          Suspend-Job
Cmdlet          Wait-Job

Additionally, you would have to come-up with your own mechanism to manage starting, processing, stopping and aggregating the results. There are numerous variations on the internet. I just did a quick search and came up with these

• Scaling and Queuing PowerShell Background Jobs
• Multithreading Revisited – Using Jobs
• Can Powershell Run Commands in Parallel?
• Weekend Scripter: Max Out PowerShell in a Little Bit of Time (Using Runspaces)

The last link uses Runspaces instead of *-Job functions. It is written by  Jason Walker a Microsoft employee whose presentation on asynchronous processing I had the pleasure of attending. Be sure to check out Part 1 of the article.

Why I use Invoke-Aysnc?

Although it is not the best way to parallelize (see performance comparison between different methods), I started using the below function from TechNet gallery and I love it

• Invoke-Async – Allows you to run any cmdlet/function/scriptblock asynchronously

The primary reasons I like it are

1. Can parallelize anything. It simply does not matter what operation it is.
2. Allows one to specify the number of threads
3. Aggregates all the results
4. Allows specification of parameters other than the list one needs to iterate through
5. It is ingeniously simple and elegant. I was able to easily make code changes
6. The author (jrich) is very responsive to questions/comments

Once you download, it is easy to follow the examples and get going. I will only point out the things that are not obvious.

Each launched job gets its own new session

What you are asking it to run asynchronously will get its own PowerShell session. So, if you are doing this from the ISE, everything that you have dot-sourced or imported (modules) in the ISE will not be available in the asynchronous job unless you also dot-source or import explicitly in the script block you provide it to run. Sometimes, it may not be worthwhile to do this. Please evaluate before you start using it.

Changes required to the script

I made two changes to the script (changes marked with “Jana”)

1. Bug fix to make it work when only 1 item is specified in the collection to process (the author might have already fixed this)
2. Abort after [x] errors. I needed this additional parameter because, it did not make sense to process 285 items if the first 15 failed for example, as it is highly likely that there is some issue and all 300 will fail.

#Source: http://gallery.technet.microsoft.com/scriptcenter/Invoke-Async-Allows-you-to-83b0c9f0#content


function Invoke-Async{
param(
#The data group to process, such as server names.
[parameter(Mandatory=$true,ValueFromPipeLine=$true)]
[object[]]$Set,
#The parameter name that the set belongs to, such as Computername.
[parameter(Mandatory=$true)]
[string] $SetParam,
#The Cmdlet for Function you'd like to process with.
[parameter(Mandatory=$true, ParameterSetName='cmdlet')]
[string]$Cmdlet,
#The ScriptBlock you'd like to process with
[parameter(Mandatory=$true, ParameterSetName='ScriptBlock')]
[scriptblock]$ScriptBlock,
#any aditional parameters to be forwarded to the cmdlet/function/scriptblock
[hashtable]$Params,
#number of jobs to spin up, default being 10.
[int]$ThreadCount=10,
#return performance data
[switch]$Measure,
#return abort threshold (if non-negative/non-zero, bails after this many errors have occured)
[int]$AbortAfterErrorCount=-1

)
Begin
{
    [int] $ErrorCounter = 0                                                                              #20141031 Jana - Added to track errors and bail
    [int] $AllowedErrorCount = if ($AbortAfterErrorCount -le 0) {9999999} else {$AbortAfterErrorCount}   #20141031 Jana - Added to track errors and bail
    $Threads = @()
    $Length = $JobsLeft = $Set.Length

    $Count = 0
    if($Length -lt $ThreadCount){$ThreadCount=$Length}
    $timer = @(1..$ThreadCount  | ForEach-Object{$null})
    $Jobs = @(1..$ThreadCount  | ForEach-Object{$null})

    If($PSCmdlet.ParameterSetName -eq 'cmdlet')
    {
        $CmdType = (Get-Command $Cmdlet).CommandType
        if($CmdType -eq 'Alias')
        {
            $CmdType = (Get-Command (Get-Command $Cmdlet).ResolvedCommandName).CommandType
        }

        If($CmdType -eq 'Function')
        {
            $ScriptBlock = (Get-Item Function:\$Cmdlet).ScriptBlock
            1..$ThreadCount | ForEach-Object{ $Threads += [powershell]::Create().AddScript($ScriptBlock)}
        }
        ElseIf($CmdType -eq "Cmdlet")
        {
            1..$ThreadCount  | ForEach-Object{ $Threads += [powershell]::Create().AddCommand($Cmdlet)}
        }
    }
    Else
    {
        1..$ThreadCount | ForEach-Object{ $Threads += [powershell]::Create().AddScript($ScriptBlock)}
    }

    If($Params){$Threads | ForEach-Object{$_.AddParameters($Params) | Out-Null}}

}
Process
{
    while($JobsLeft)
    {
        #20140929 Jana - Bug fix - Changed "-lt" to "-le" because it does not execute if if there is only 1 item in the set total to begin with!
        #for($idx = 0; $idx -lt ($ThreadCount-1) ; $idx++)
        for($idx = 0; $idx -le ($ThreadCount-1) ; $idx++)
        {

            $SetParamObj = $Threads[$idx].Commands.Commands[0].Parameters| Where-Object {$_.Name -eq $SetParam}

            #NOTE: Only hits this block after atleast one item has been kicked off..so during very first pass, skips this.
            If ($Jobs[$idx] -ne $null)
            {
                If($Jobs[$idx].IsCompleted) #job ran ok, clear it out
                {
                    $result = $null
                    if($threads[$idx].InvocationStateInfo.State -eq "Failed")
                    {
                        $result  = $Threads[$idx].InvocationStateInfo.Reason

                        #Will write out the hashtable values in the error instead of "Set Item: System.Collections.Hashtable Exception: ...."
                        $OutError = "Set Item: $($($SetParamObj.Value)| Out-String )"
                        Write-Error "$OutError Exception: $result"

                        #Write-Error "Set Item: $($SetParamObj.Value) Exception: $result"

                        #This was the original code by the original author (always leave this commented)
                        #Write-Error "Set Item: $($SetParamObj) Exception: $result"

                        #20141031 Jana - Added to track errors and bail
                        $ErrorCounter++
                        if ($ErrorCounter -ge $AllowedErrorCount)
                        {
                            break;
                        }
                    }
                    else
                    {
                        $result = $Threads[$idx].EndInvoke($Jobs[$idx])
                    }
                    $ts = (New-TimeSpan -Start $timer[$idx] -End (Get-Date))
                    if($Measure)
                    {
                        new-object psobject -Property @{
                            TimeSpan = $ts
                            Output = $result
                            SetItem = $SetParamObj.Value
                            }
                    }
                    else
                    {
                        $result
                    }
                    $Jobs[$idx] = $null
                    $JobsLeft-- #one less left

                    write-verbose "Completed: $($SetParamObj.Value) in $ts"
                    #write-verbose "Completed: $SetParamObj in $ts"
                    write-progress -Activity "Processing Set" -Status "$JobsLeft jobs left" -PercentComplete (($length-$jobsleft)/$length*100)
                }
            }

            If(($Count -lt $Length) -and ($Jobs[$idx] -eq $null)) #add job if there is more to process
            {
                write-verbose "starting: $($Set[$Count])"
                $timer[$idx] = get-date
                $Threads[$idx].Commands.Commands[0].Parameters.Remove($SetParamObj) | Out-Null #check for success?
                $Threads[$idx].AddParameter($SetParam,$Set[$Count]) | Out-Null
                $Jobs[$idx] = $Threads[$idx].BeginInvoke()
                $Count++
            }

        }

        #20141031 Jana - Added to track errors and bail
        if ($ErrorCounter -ge $AllowedErrorCount)
        {
            break;
        }
    }
}
End
{
    $Threads | ForEach-Object{$_.runspace.close();$_.Dispose()}
}
}

Original source for above script:

http://gallery.technet.microsoft.com/scriptcenter/Invoke-Async-Allows-you-to-83b0c9f0#content

Again, I want to give the author “jrich” his due respect and appreciation for the good work.