Sane progress

Author: powercode

module/PSParallel.psd1

@@ -286,7 +286,7 @@ public override void EndProcessing()
 					_progressManager.TotalCount = _input.Count;
 					foreach (var i in _input)
 					{
-						var processed = Pool.ProcessedCount + Pool.GetPartiallyProcessedCount();
+						var processed = Pool.GetEstimatedProgressCount();
 						_progressManager.UpdateCurrentProgressRecord($"Starting processing of {i}", processed);
 						WriteProgress(_progressManager.ProgressRecord);
 						while (!Pool.TryAddInput(ScriptBlock, i))
@@ -298,7 +298,7 @@ public override void EndProcessing()
 					while (!Pool.WaitForAllPowershellCompleted(100))
 					{
 
-						_progressManager.UpdateCurrentProgressRecord("All work queued. Waiting for remaining work to complete.", Pool.ProcessedCount);
+						_progressManager.UpdateCurrentProgressRecord("All work queued. Waiting for remaining work to complete.", Pool.GetEstimatedProgressCount());
 						WriteProgress(_progressManager.ProgressRecord);
 
 						if (Stopping)
@@ -322,11 +322,9 @@ public override void WriteProgress(Collection<ProgressRecord> progress)
 					p.ParentActivityId = _progressManager.ActivityId;
 					WriteProgress(p);
 				}
-				_progressManager.UpdateCurrentProgressRecord(Pool.ProcessedCount + Pool.GetPartiallyProcessedCount());
+				_progressManager.UpdateCurrentProgressRecord(Pool.GetEstimatedProgressCount());
 				WriteProgress(_progressManager.ProgressRecord);
 			}
 		}
-
-
 	}
 }

src/PSParallel/InvokeParallelCommand.cs

@@ -15,8 +15,12 @@ class PowerShellPoolMember : IDisposable
 		private readonly PSDataCollection<PSObject> m_input =new PSDataCollection<PSObject>();
 		private PSDataCollection<PSObject> m_output;
 		private int m_percentComplete;
-		public int PercentComplete => m_percentComplete;
-		
+		public int PercentComplete
+		{
+			get { return m_percentComplete; }
+			set { m_percentComplete = value; }
+		}
+
 
 		public PowerShellPoolMember(PowershellPool pool, int index)
 		{

src/PSParallel/PowerShellPoolMember.cs

@@ -3,29 +3,27 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.Contracts;
-using System.Linq;
 using System.Management.Automation;
 using System.Management.Automation.Runspaces;
 using System.Threading;
 
+
 namespace PSParallel
 {
 	sealed class PowershellPool : IDisposable
 	{
+		private readonly object _countLock = new object();
 		private int _busyCount;
-		private int _processedCount;
 		private readonly CancellationToken _cancellationToken;
 		private readonly RunspacePool _runspacePool;
 		private readonly List<PowerShellPoolMember> _poolMembers;
 		private readonly BlockingCollection<PowerShellPoolMember> _availablePoolMembers = new BlockingCollection<PowerShellPoolMember>(new ConcurrentQueue<PowerShellPoolMember>());
 		public readonly PowerShellPoolStreams Streams = new PowerShellPoolStreams();
-
-		public int ProcessedCount => _processedCount;		
+		private int _processedCount;
 
 		public PowershellPool(int poolSize, InitialSessionState initialSessionState, CancellationToken cancellationToken)
 		{
 			_poolMembers= new List<PowerShellPoolMember>(poolSize);
-			_processedCount = 0;
 			_cancellationToken = cancellationToken;
 
 			for (var i = 0; i < poolSize; i++)
@@ -39,7 +37,7 @@ public PowershellPool(int poolSize, InitialSessionState initialSessionState, Can
 			_runspacePool.SetMaxRunspaces(poolSize);
 		}
 
-		public int GetPartiallyProcessedCount()
+		private int GetPartiallyProcessedCount()
 		{
 			var totalPercentComplete = 0;
 			var count = _poolMembers.Count;
@@ -55,16 +53,24 @@ public int GetPartiallyProcessedCount()
 					percentComplete = 100;
 				}
 				totalPercentComplete += percentComplete;
-			}			
-			return totalPercentComplete / 100;
+			}
+			var partiallyProcessedCount = totalPercentComplete / 100;
+			return partiallyProcessedCount;
+		}
+
+		public int GetEstimatedProgressCount()
+		{
+			lock(_countLock) {
+				return _processedCount + GetPartiallyProcessedCount();
+			}
 		}
 
 		public bool TryAddInput(ScriptBlock scriptblock,PSObject inputObject)
 		{			
 			PowerShellPoolMember poolMember;
 			if(!TryWaitForAvailablePowershell(100, out poolMember))
 			{
-				return false;							
+				return false;
 			}
 
 			Interlocked.Increment(ref _busyCount);
@@ -110,7 +116,6 @@ private bool TryWaitForAvailablePowershell(int milliseconds, out PowerShellPoolM
 			}
 
 			poolMember.PowerShell.RunspacePool = _runspacePool;
-			Debug.WriteLine($"WaitForAvailablePowershell - Busy: {_busyCount} _processed {_processedCount}, member = {poolMember.Index}");
 			return true;
 		}
 
@@ -125,13 +130,18 @@ public void Dispose()
 		public void ReportAvailable(PowerShellPoolMember poolmember)
 		{
 			Interlocked.Decrement(ref _busyCount);
-			Interlocked.Increment(ref _processedCount);
+			lock (_countLock)
+			{
+				_processedCount++;
+				poolmember.PercentComplete = 0;
+			}
+			
+			poolmember.PercentComplete = 0;
 			while (!_availablePoolMembers.TryAdd(poolmember, 1000, _cancellationToken))
 			{
 				_cancellationToken.ThrowIfCancellationRequested();
 				Debug.WriteLine("WaitForAvailablePowershell - TryAdd failed");
 			}
-			Debug.WriteLine($"ReportAvailable - Busy: {_busyCount} _processed {_processedCount}, member = {poolmember.Index}");
 		}
 
 		public void ReportStopped(PowerShellPoolMember powerShellPoolMember)

src/PSParallel/PowershellPool.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Diagnostics;
 using System.Management.Automation;
-
 namespace PSParallel
 {
 	class ProgressManager
@@ -33,7 +32,7 @@ public void UpdateCurrentProgressRecord(int count)
 					_progressRecord.PercentComplete = percentComplete;
 					_progressRecord.SecondsRemaining = GetSecondsRemaining(count);
 				}				
-			}			
+			}
 		}
 
 		public void UpdateCurrentProgressRecord(string currentOperation, int count)
@@ -54,8 +53,18 @@ public ProgressRecord Completed()
 			return _progressRecord;
 		}
 
-		private int GetSecondsRemaining(int count) => count == 0 ? -1 : (int) ((TotalCount - count)*_stopwatch.ElapsedMilliseconds/1000/count);
-		private int GetPercentComplete(int count) => count*100/TotalCount;
+		private int GetSecondsRemaining(int count)
+		{
+			var secondsRemaining = count == 0 ? -1 : (int) ((TotalCount - count)*_stopwatch.ElapsedMilliseconds/1000/count);
+			return secondsRemaining;
+		}
+
+		private int GetPercentComplete(int count)
+		{
+			var percentComplete = count*100/TotalCount;
+			return percentComplete;
+		}
+			
 		public int ActivityId => _progressRecord.ActivityId;
 	}
 

src/PSParallel/ProgressManager.cs

@@ -32,4 +32,4 @@
 // by using the '*' as shown below:
 // [assembly: AssemblyVersion("1.0.*")]
 [assembly: AssemblyVersion("1.0.0.0")]
-[assembly: AssemblyFileVersion("2.2.0.0")]
+[assembly: AssemblyFileVersion("2.2.2.0")]