An easy to use concurrency manager for PowerShell

PSJobManager.md

PowerShell Job Manager

An easy to use concurrency manager for PowerShell

Source Code

class JobManager
{
    [Collections.Generic.Dictionary[String, Guid]] $Jobs;
    [Int32] $MaxConcurrentJobs;
    [Int32] $SleepMs;

    JobManager([Int32] $maxConcurrentJobs, [Int32] $sleepMs)
    {
        $this.Jobs = [Collections.Generic.Dictionary[String, Guid]]::new();
        $this.MaxConcurrentJobs = $maxConcurrentJobs;
        $this.SleepMs = $sleepMs;
    }

    [Management.Automation.Job] GetJob([String] $id)
    {
        if (!$this.Jobs.Keys.Contains($id))
        {
            return $null;
        }
        
        return (Get-Job `
            -InstanceId $this.Jobs[$id] `
            -ErrorAction "SilentlyContinue");
    }

    [Management.Automation.Job[]] GetAllJobs()
    {
        return (Get-Job `
            -InstanceId ($this.Jobs.Values | Select-Object -ExpandProperty "Guid") `
            -ErrorAction "SilentlyContinue");
    }

    [Int32] GetActiveJobCount()
    {
        return ($this.GetAllJobs() `
            | Where-Object {
                $_.State -ne "Completed" `
                -and $_.State -ne "Stopped" `
                -and $_.State -ne "Failed"
            }).Count;
    }

    [Void] StartJob([String] $id, [Hashtable] $context, [Object] $init, [Bool] $wait)
    {
        if ($wait)
        {
            while ($this.GetActiveJobCount() -ge $this.MaxConcurrentJobs)
            {
                (Start-Sleep -Milliseconds $this.SleepMs);
            }
        }
        
        $this.Jobs.Add($id, (Start-Job `
            -InitializationScript $init `
            -ScriptBlock { Execute -Context $using:context }).InstanceId);
    }

    [Void] StopJob([String] $id)
    {
        if (!$this.Jobs.Keys.Contains($id))
        {
            return;
        }
        
        (Stop-Job `
            -InstanceId $this.Jobs[$id] `
            -ErrorAction "SilentlyContinue");
    }

    [Void] StopAllJobs()
    {
        (Stop-Job `
            -InstanceId ($this.Jobs.Values | Select-Object -ExpandProperty "Guid") `
            -ErrorAction "SilentlyContinue");
    }

    [Void] AwaitJobs()
    {
        while ($this.GetActiveJobCount() -gt 0)
        {
            (Start-Sleep -Milliseconds $this.SleepMs);
        }
    }

    [Void] RemoveJob([String] $id)
    {
        if (!$this.Jobs.Keys.Contains($id))
        {
            return;
        }

        (Remove-Job `
            -InstanceId $this.Jobs[$id] `
            -ErrorAction "SilentlyContinue" `
            -Force);

        $this.Jobs.Remove($id);
    }

    [Void] RemoveAllJobs()
    {
        (Remove-Job `
            -InstanceId ($this.Jobs.Values | Select-Object -ExpandProperty "Guid") `
            -ErrorAction "SilentlyContinue" `
            -Force);

        $this.Jobs.Clear();
    }
}

Sample Usage

$instanceId = [Guid]::NewGuid().ToString();
$outputDirectory = "$pwd\$instanceId";

if (!(Test-Path -Path $outputDirectory))
{
    (New-Item -Path $outputDirectory -ItemType "Directory" -Force | Out-Null);
}

$urls = [Collections.Generic.Dictionary[String, String]]::new();
$urls.Add("DuckDuckGo", "https://duckduckgo.com");
$urls.Add("Google", "https://google.com");
$urls.Add("Bing", "https://bing.com");
$jobManager = [JobManager]::new(10, 100);

$init = {
    function Execute([Hashtable] $Context)
    {
        $response = (Invoke-WebRequest -Uri $Context["Url"] -ErrorAction "SilentlyContinue");
        ($response | Export-Clixml -Path $Context["OutputPath"]);
    }
};

(Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] Starting jobs...");

foreach ($key in $urls.Keys)
{
    $jobManager.StartJob(
        $key,
        @{
            OutputPath = "$outputDirectory\$key.xml";
            Url = $urls[$key];
        },
        $init,
        $false
    );

    (Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] `t<$key> Sending request to `"$($urls[$key])`"...");
}

(Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] Awaiting jobs...");

$jobManager.AwaitJobs();

(Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] Aggregating job outputs...");

foreach ($key in $jobManager.Jobs.Keys)
{
    if (!(Test-Path -Path "$outputDirectory\$key.xml"))
    {
        continue;
    }

    $output = (Import-Clixml -Path "$outputDirectory\$key.xml");
    (Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] `t<$key> Received response status code $($output.StatusCode)!");

    (Remove-Item -Path "$outputDirectory\$key.xml" -Force | Out-Null);
}

(Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] Cleaning up...");
(Remove-Item -Path $outputDirectory -Force | Out-Null);

$jobManager.RemoveAllJobs();

(Write-Host "[$(Get-Date -Format("yyyy-MM-dd HH:mm:ss"))] Finished!");