TryAdd/TryTake for better robustness and responsiveness.

Hierarchical progress

Author: Staffan Gustafsson

module/PSParallel.psd1

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Management.Automation;
 using System.Management.Automation.Runspaces;
@@ -185,8 +186,10 @@ protected override void ProcessRecord()
 		{
 			if(NoProgress)
 			{
-				m_powershellPool.AddInput(ScriptBlock, InputObject);
-				WriteOutputs();
+				while (!m_powershellPool.TryAddInput(ScriptBlock, InputObject))
+				{
+					WriteOutputs();
+				}								
 			}
 			else
 			{
@@ -205,8 +208,10 @@ protected override void EndProcessing()
 					{
 						var pr = m_progressManager.GetCurrentProgressRecord($"Starting processing of {i}", m_powershellPool.ProcessedCount);
 						WriteProgress(pr);
-						m_powershellPool.AddInput(ScriptBlock, i);
-						WriteOutputs();
+						while (!m_powershellPool.TryAddInput(ScriptBlock, i))
+						{
+							WriteOutputs();
+						}												
 					}
 				}
 				while(!m_powershellPool.WaitForAllPowershellCompleted(100))
@@ -241,6 +246,7 @@ protected override void StopProcessing()
 
 		private void WriteOutputs()
 		{
+			Debug.WriteLine("Processing output");
 			if (m_cancelationTokenSource.IsCancellationRequested)
 			{
 				return;

src/PSParallel/InvokeParallelCommand.cs

@@ -6,18 +6,25 @@ namespace PSParallel
 	class PowerShellPoolMember : IDisposable
 	{
 		private readonly PowershellPool m_pool;
+		private readonly int m_index;
 		private readonly PowerShellPoolStreams m_poolStreams;
 		private PowerShell m_powerShell;
 		public PowerShell PowerShell => m_powerShell;
+		public int Index => m_index ;
+
 		private readonly PSDataCollection<PSObject> m_input =new PSDataCollection<PSObject>();
 		private PSDataCollection<PSObject> m_output;
+		private ProgressProjector m_progressProjector;
+		public ProgressProjector ProgressProjector => m_progressProjector;
 
-		public PowerShellPoolMember(PowershellPool pool)
+		public PowerShellPoolMember(PowershellPool pool, int index)
 		{
 			m_pool = pool;
+			m_index = index;
 			m_poolStreams = m_pool.Streams;
-			m_input.Complete();
+			m_input.Complete();			
 			CreatePowerShell();
+			m_progressProjector = new ProgressProjector();			
 		}
 
 		private void PowerShellOnInvocationStateChanged(object sender, PSInvocationStateChangedEventArgs psInvocationStateChangedEventArgs)
@@ -81,10 +88,13 @@ private void UnhookStreamEvents(PSDataStreams streams)
 
 		public void BeginInvoke(ScriptBlock scriptblock, PSObject inputObject)
 		{
-			string command = $"param($_,$PSItem){scriptblock}";
+			m_progressProjector.Start();
+			string command = $"param($_,$PSItem, $PSPArallelIndex,$PSParallelProgressId){scriptblock}";
 			m_powerShell.AddScript(command)
 				.AddParameter("_", inputObject)
-				.AddParameter("PSItem", inputObject);
+				.AddParameter("PSItem", inputObject)
+				.AddParameter("PSParallelIndex", m_index)
+				.AddParameter("PSParallelProgressId", m_index+1000);
 			m_powerShell.BeginInvoke(m_input, m_output);
 		}
 
@@ -115,7 +125,8 @@ private void InformationOnDataAdded(object sender, DataAddedEventArgs dataAddedE
 
 		private void ProgressOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
-			var record = ((PSDataCollection<ProgressRecord>)sender)[dataAddedEventArgs.Index];
+			var record = ((PSDataCollection<ProgressRecord>)sender)[dataAddedEventArgs.Index];			
+			m_progressProjector.ReportProgress(record.PercentComplete);
 			m_poolStreams.Progress.Add(record);
 		}
 

src/PSParallel/PowerShellPoolMember.cs

@@ -1,21 +1,22 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Diagnostics.Contracts;
 using System.Management.Automation;
 using System.Management.Automation.Runspaces;
 using System.Threading;
 
 namespace PSParallel
 {
-	class PowershellPool : IDisposable
+	sealed class PowershellPool : IDisposable
 	{
 		private int m_busyCount;
 		private int m_processedCount;
 		private readonly CancellationToken m_cancellationToken;
 		private readonly RunspacePool m_runspacePool;
 		private readonly List<PowerShellPoolMember> m_poolMembers;
-		private readonly BlockingCollection<PowerShellPoolMember> m_availablePoolMembers = new BlockingCollection<PowerShellPoolMember>(new ConcurrentStack<PowerShellPoolMember> ());
+		private readonly BlockingCollection<PowerShellPoolMember> m_availablePoolMembers = new BlockingCollection<PowerShellPoolMember>(new ConcurrentQueue<PowerShellPoolMember>());
 		public readonly PowerShellPoolStreams Streams = new PowerShellPoolStreams();
 
 		public int ProcessedCount => m_processedCount;
@@ -26,9 +27,9 @@ public PowershellPool(int poolSize, InitialSessionState initialSessionState, Can
 			m_processedCount = 0;
 			m_cancellationToken = cancellationToken;
 
-			for (int i = 0; i < poolSize; i++)
+			for (var i = 0; i < poolSize; i++)
 			{
-				var powerShellPoolMember = new PowerShellPoolMember(this);
+				var powerShellPoolMember = new PowerShellPoolMember(this, i+1);
 				m_poolMembers.Add(powerShellPoolMember);
 				m_availablePoolMembers.Add(powerShellPoolMember);
 			}
@@ -37,16 +38,17 @@ public PowershellPool(int poolSize, InitialSessionState initialSessionState, Can
 			m_runspacePool.SetMaxRunspaces(poolSize);
 		}
 
-		public void AddInput(ScriptBlock scriptblock,PSObject inputObject)
-		{
-			try { 
-				var powerShell = WaitForAvailablePowershell();
-				Interlocked.Increment(ref m_busyCount);
-				powerShell.BeginInvoke(scriptblock, inputObject);
-			}
-			catch(OperationCanceledException)
+		public bool TryAddInput(ScriptBlock scriptblock,PSObject inputObject)
+		{			
+			PowerShellPoolMember poolMember;
+			if(!TryWaitForAvailablePowershell(100, out poolMember))
 			{
+				return false;							
 			}
+
+			Interlocked.Increment(ref m_busyCount);
+			poolMember.BeginInvoke(scriptblock, inputObject);
+			return true;
 		}
 
 		public void Open()
@@ -76,11 +78,19 @@ public bool WaitForAllPowershellCompleted(int timeoutMilliseconds)
 			return false;
 		}
 
-		private PowerShellPoolMember WaitForAvailablePowershell()
-		{
-			var poolmember = m_availablePoolMembers.Take(m_cancellationToken);
-			poolmember.PowerShell.RunspacePool = m_runspacePool;
-			return poolmember;
+		private bool TryWaitForAvailablePowershell(int milliseconds, out PowerShellPoolMember poolMember)
+		{			
+			if(!m_availablePoolMembers.TryTake(out poolMember, milliseconds, m_cancellationToken))
+			{
+				m_cancellationToken.ThrowIfCancellationRequested();
+				Debug.WriteLine($"WaitForAvailablePowershell - TryTake failed");
+				poolMember = null;
+				return false;
+			}
+			
+			poolMember.PowerShell.RunspacePool = m_runspacePool;
+			Debug.WriteLine($"WaitForAvailablePowershell - Busy: {m_busyCount} _processed {m_processedCount}, member = {poolMember.Index}");
+			return true;
 		}
 
 
@@ -95,10 +105,13 @@ public void ReportAvailable(PowerShellPoolMember poolmember)
 		{
 			Interlocked.Decrement(ref m_busyCount);
 			Interlocked.Increment(ref m_processedCount);
-			if(!m_cancellationToken.IsCancellationRequested)
-			{ 
-				m_availablePoolMembers.Add(poolmember);
-			}
+			while (!m_availablePoolMembers.TryAdd(poolmember, 1000, m_cancellationToken))
+			{
+				m_cancellationToken.ThrowIfCancellationRequested();
+				Debug.WriteLine($"WaitForAvailablePowershell - TryAdd failed");
+			}			
+			Debug.WriteLine($"ReportAvailable - Busy: {m_busyCount} _processed {m_processedCount}, member = {poolmember.Index}");	
+			
 		}
 
 		public void ReportStopped(PowerShellPoolMember powerShellPoolMember)

src/PSParallel/PowershellPool.cs

@@ -1,4 +1,5 @@
-using System.Diagnostics;
+using System;
+using System.Diagnostics;
 using System.Management.Automation;
 
 namespace PSParallel
@@ -53,4 +54,37 @@ public ProgressRecord Completed()
 		private int GetPercentComplete(int count) => count*100/TotalCount;
 		public int ActivityId => m_progressRecord.ActivityId;
 	}
+
+
+	class ProgressProjector
+	{
+		private readonly Stopwatch m_stopWatch;
+		private int m_percentComplete;		
+		public ProgressProjector()
+		{
+			m_stopWatch = new Stopwatch();
+			m_percentComplete = -1;
+		}
+
+		public void ReportProgress(int percentComplete)
+		{			
+			m_percentComplete = percentComplete;			
+		}
+
+		public bool IsValid => m_percentComplete > 0 && m_stopWatch.IsRunning;
+		public TimeSpan Elapsed => m_stopWatch.Elapsed;
+		
+		public TimeSpan ProjectedTotalTime => new TimeSpan(Elapsed.Ticks * 100 / m_percentComplete);
+
+		public void Start()
+		{
+			m_stopWatch.Start();
+			m_percentComplete = 0;
+		}
+
+		public void Stop()
+		{
+			m_stopWatch.Stop();
+		}
+	}
 }

src/PSParallel/ProgressManager.cs

@@ -9,7 +9,7 @@
 [assembly: AssemblyConfiguration("")]
 [assembly: AssemblyCompany("")]
 [assembly: AssemblyProduct("PSParallell")]
-[assembly: AssemblyCopyright("Copyright ©  2015")]
+[assembly: AssemblyCopyright("Copyright ©  2016")]
 [assembly: AssemblyTrademark("")]
 [assembly: AssemblyCulture("")]
 
@@ -32,4 +32,4 @@
 // by using the '*' as shown below:
 // [assembly: AssemblyVersion("1.0.*")]
 [assembly: AssemblyVersion("1.0.0.0")]
-[assembly: AssemblyFileVersion("2.1.3.0")]
+[assembly: AssemblyFileVersion("2.2.0.0")]