<#
.SYNOPSIS
A parallel ForEach that uses runspaces

.PARAMETER ScriptBlock
ScriptBlock to execute for each InputObject

.PARAMETER ScriptFile
Script file to execute for each InputObject

.PARAMETER InputObject
Object(s) to run script against in parallel

.PARAMETER Throttle
Maximum number of threads to run at one time.  Default: 5

.PARAMETER Timeout
Stop each thread after this many minutes.  Default: 0 (Do not timeout)

WARNING:  This parameter should be used as a failsafe only
Set it for roughly the entire duration you expect for all threads to complete

.PARAMETER SleepTimer
When looping through open threads, wait this many milliseconds before looping again.  Default: 200

.PARAMETER ProgressId
Identity of the progress bar to use for display.  Default: 0

.PARAMETER Activity
String to pass to the progress bar for the -Activity parameter. No progress is displayed unless this is specified.  Default: Blank and do not show progress.

.EXAMPLE
(1..50) | ForEach-Parallel -Throttle 4 -ScriptBlock { $this; Start-Sleep -Seconds (Get-Random -Minimum 0 -Maximum 5) } -Activity "One to fifty"

Send the number 1 through 50 to the scriptblock.  For each, display the number and then sleep for 0 to 5 seconds.  Display a progress bar.  Only execute 4 threads concurrently.

.EXAMPLE
Get-Content .\servers.txt | Foreach-Parallel -Throttle 20 -Timeout 10 -SleepTimer 200 -ScriptFile .\query.ps1 -Verbose

Run query.ps1 against each computer in the servers.txt file.  Run 20 concurrently, timeout a thread if it takes longer than 10 minutes to run, give verbose output.

.FUNCTIONALITY 
PowerShell Language

.NOTES
Credit to Tome Tanasovski for the original ForEach-Parallel, from which this script has been modified.
http://powertoe.wordpress.com/2012/05/03/foreach-parallel/

Added:
 - Progress bar
 - Verbose output
 - Changed $_ to $this within the scriptblock to avoid pipeline contention
#>
[cmdletbinding()]
PARAM (
	[Parameter(Mandatory=$false,position=0,ParameterSetName='ScriptBlock')]
		[scriptblock]$ScriptBlock,

	[Parameter(Mandatory=$false,ParameterSetName='ScriptFile')]
	[ValidateScript({test-path $_ -pathtype leaf})]
		$ScriptFile,

	[Parameter(Mandatory=$true,ValueFromPipeline=$true)]
		[PSObject]$InputObject,

		[int]$Throttle = 5,

		[double]$SleepTimer = 200,

		[double]$Timeout = 0,
		
		[int]$ProgressId = 0,
		
		[string]$Activity = 'Default'
)

BEGIN {
	$stopWatch = [Diagnostics.Stopwatch]::StartNew()
	
	if ($Activity -ne 'Default') { Write-Progress -Id $ProgressId -Activity $Activity -Status "Preparing" -PercentComplete 0 -CurrentOperation "Elapsed time: $($stopWatch.Elapsed.toString().subString(0,8))" }
	
	#Build the scriptblock depending on the parameter used
	switch ($PSCmdlet.ParameterSetName) {
		'ScriptBlock' {$ScriptBlock = $ExecutionContext.InvokeCommand.NewScriptBlock("param(`$this)`r`n" + $Scriptblock.ToString())}
		'ScriptFile' {$scriptblock = [scriptblock]::Create($(get-content $ScriptFile | out-string))}
		Default {Write-Error ("Must provide ScriptBlock or ScriptFile"); Return}
	}
	
	#Define the initial sessionstate, create the runspacepool
	Write-Verbose "Creating runspace pool with $Throttle threads"
	$sessionState = [system.management.automation.runspaces.initialsessionstate]::CreateDefault()
	$pool = [runspacefactory]::CreateRunspacePool(1, $Throttle, $sessionState, $host)
	$pool.open()
	
	#array to hold details on each thread
	$threads = @()

	#If inputObject is bound get a total count and set bound to true
	$bound = $false
	if( $PSBoundParameters.ContainsKey("inputObject") ) {
		$bound = $true
		$totalCount = $inputObject.count
	}
	
	if ($Activity -ne 'Default') { Write-Progress -Id $ProgressId -Activity $Activity -Status "Creating threads" -CurrentOperation "Elapsed time: $($stopWatch.Elapsed.toString().subString(0,8))" }
	
	[int]$ProgressTotal = 0 # Total number of threads created
	[int]$ProgressCount = 0 # Number of threads completed
$runtemplate = @'
	#For each pipeline object, create a new powershell instance, add to runspacepool
	$powershell = [powershell]::Create().AddScript($ScriptBlock).AddArgument($InputObject)
	$powershell.runspacepool=$pool
	$startTime = Get-Date
	
	#add references to inputobject, instance, handle and startTime to threads array
	$threads     += New-Object psobject -Property @{
		Object    = $inputObject;
		instance  = $powershell;
		handle    = $powershell.begininvoke();
		startTime = $startTime
	}
	
	$ProgressTotal++
	Write-Verbose "Added $inputobject to the runspacepool at $startTime"
	if ($Activity -ne 'Default') { Write-Progress -Id $ProgressId -Activity $Activity -Status "$ProgressTotal threads created" -CurrentOperation "Elapsed time: $($stopWatch.Elapsed.toString().subString(0,8))" -PercentComplete ([math]::Min($ProgressTotal,95)) }
'@
}

PROCESS {
	if($bound) {
		$run = $runtemplate -replace 'inputObject', 'object'
		foreach($object in $inputObject) { 
			Invoke-Expression -command $run
		}
	}
	else {
		Invoke-Expression -command $run
	}
}

END {
	$notdone       = $true
	$ProgressTotal = $threads.count;
	$ProgressCount = 0
	if ($Activity -ne 'Default') { Write-Progress -Id $ProgressId -Activity $Activity -Status "$ProgressCount of $ProgressTotal threads completed" -CurrentOperation "Elapsed time: $($stopWatch.Elapsed.toString().subString(0,8))" -PercentComplete (100*$ProgressCount/$ProgressTotal) }
	
	#Loop through threads.
	while ($notdone) {

		$notdone = $false

		for ($i=0; $i -lt $threads.count; $i++) {
			$thread = $threads[$i]
			if ($thread) {

				#If thread is complete, dispose of it.
				if ($thread.handle.iscompleted) {
					Write-verbose "Closing thread for $($thread.Object)"
					$thread.instance.endinvoke($thread.handle)
					$thread.instance.dispose()
					$threads[$i] = $null
					$ProgressCount++
				}

				#Thread exceeded maxruntime timeout threshold
				elseif( $Timeout -ne 0 -and ( (get-date) - $thread.startTime ).totalminutes -gt $Timeout ) {
					Write-Error "Closing thread for $($thread.Object): Thread exceeded $Timeout minute limit" -TargetObject $thread.inputObject
					$thread.instance.dispose()
					$threads[$i] = $null
					$ProgressCount++
				}

				#Thread is running, loop again!
				else {
					$notdone = $true
				}
			}           
		}
		if ($Activity -ne 'Default') {Write-Progress -Id $ProgressId -Activity $Activity -Status "$ProgressCount of $ProgressTotal threads completed" -CurrentOperation "Elapsed time: $($stopWatch.Elapsed.toString().subString(0,8))" -PercentComplete (100*$ProgressCount/$ProgressTotal) }
		
		Start-Sleep -Milliseconds $SleepTimer
	}
	if ($Activity -ne 'Default') { Write-Progress -Id $ProgressId -Activity $Activity -Status "Completed" -CurrentOperation "Elapsed time: $($stopWatch.Elapsed.toString().subString(0,8))" -Completed }
}