Manginging runspaces w/o using RunspacePool.

Improving progress handling.

Author: Staffan Gustafsson

src/PSParallel/InvokeParallelCommand.cs

@@ -77,8 +77,9 @@ private static IEnumerable<PSVariable> GetVariables(SessionState sessionState)
 			try
 			{
 				string[] noTouchVariables = { "null", "true", "false", "Error" };
-				var variables = sessionState.InvokeProvider.Item.Get("Variable:");
+				var variables = sessionState.InvokeProvider.Item.Get("Variable:");				
 				var psVariables = (IEnumerable<PSVariable>)variables[0].BaseObject;
+				
 				return psVariables.Where(p => !noTouchVariables.Contains(p.Name));
 			}
 			catch (DriveNotFoundException)
@@ -102,11 +103,22 @@ private static void CaptureVariables(SessionState sessionState, InitialSessionSt
 			foreach (var variable in variables)
 			{
 				var existing = initialSessionState.Variables[variable.Name].FirstOrDefault();
-				if (existing != null && (existing.Options & (ScopedItemOptions.Constant | ScopedItemOptions.ReadOnly)) != ScopedItemOptions.None)
+				if (existing != null)
 				{
-					continue;
+					if ((existing.Options & (ScopedItemOptions.Constant | ScopedItemOptions.ReadOnly)) != ScopedItemOptions.None)
+					{
+						continue;
+					}
+					else
+					{
+						initialSessionState.Variables.Remove(existing.Name, existing.GetType());
+						initialSessionState.Variables.Add(new SessionStateVariableEntry(variable.Name, variable.Value, variable.Description, variable.Options, variable.Attributes));
+					}
+				}
+				else
+				{
+					initialSessionState.Variables.Add(new SessionStateVariableEntry(variable.Name, variable.Value, variable.Description, variable.Options, variable.Attributes));
 				}
-				initialSessionState.Variables.Add(new SessionStateVariableEntry(variable.Name, variable.Value, variable.Description, variable.Options, variable.Attributes));
 			}
 		}
 
@@ -134,7 +146,7 @@ InitialSessionState GetSessionState()
 				}
 				return InitialSessionState;
 			}
-			return GetSessionState(SessionState);
+			return GetSessionState(base.SessionState);
 		}
 
 
@@ -143,8 +155,7 @@ protected override void BeginProcessing()
 		{
 			ValidateParameters();
 			var iss = GetSessionState();
-			PowershellPool = new PowershellPool(ThrottleLimit, iss, _cancelationTokenSource.Token);
-			PowershellPool.Open();
+			PowershellPool = new PowershellPool(ThrottleLimit, iss, _cancelationTokenSource.Token);			
 			_worker = NoProgress ? (WorkerBase) new NoProgressWorker(this) : new ProgressWorker(this);
 		}
 
@@ -268,6 +279,7 @@ class ProgressWorker : WorkerBase
 		{
 			readonly ProgressManager _progressManager;
 			private readonly List<PSObject> _input;
+			private int _lastEstimate = -1;
 			public ProgressWorker(InvokeParallelCommand cmdlet) : base(cmdlet)
 			{
 				_progressManager = new ProgressManager(cmdlet.ProgressId, cmdlet.ProgressActivity, $"Processing with {cmdlet.ThrottleLimit} workers", cmdlet.ParentProgressId);
@@ -282,25 +294,30 @@ public override void ProcessRecord(PSObject inputObject)
 			public override void EndProcessing()
 			{
 				try
-				{
+				{					
 					_progressManager.TotalCount = _input.Count;
+					var lastPercentComplete = -1;
 					foreach (var i in _input)
 					{
 						var processed = Pool.GetEstimatedProgressCount();
-						_progressManager.UpdateCurrentProgressRecord($"Starting processing of {i}", processed);
-						WriteProgress(_progressManager.ProgressRecord);
+						_lastEstimate = processed;
+						_progressManager.SetCurrentOperation($"Starting processing of {i}");
+						_progressManager.UpdateCurrentProgressRecord(processed);
+						var pr = _progressManager.ProgressRecord;
+						if (lastPercentComplete != pr.PercentComplete)
+						{
+							WriteProgress(pr);
+						}
+						
 						while (!Pool.TryAddInput(ScriptBlock, i))
 						{
 							WriteOutputs();
 						}
 					}
-
+					_progressManager.SetCurrentOperation("All work queued. Waiting for remaining work to complete.");										
 					while (!Pool.WaitForAllPowershellCompleted(100))
 					{
-
-						_progressManager.UpdateCurrentProgressRecord("All work queued. Waiting for remaining work to complete.", Pool.GetEstimatedProgressCount());
-						WriteProgress(_progressManager.ProgressRecord);
-
+						WriteProgressIfUpdated();
 						if (Stopping)
 						{
 							return;
@@ -311,6 +328,7 @@ public override void EndProcessing()
 				}
 				finally
 				{
+					_progressManager.UpdateCurrentProgressRecord(Pool.GetEstimatedProgressCount());
 					WriteProgress(_progressManager.Completed());
 				}
 			}
@@ -319,11 +337,27 @@ public override void WriteProgress(Collection<ProgressRecord> progress)
 			{
 				foreach (var p in progress)
 				{
-					p.ParentActivityId = _progressManager.ActivityId;
+					if (p.ActivityId != _progressManager.ActivityId)
+					{
+						p.ParentActivityId = _progressManager.ActivityId;
+					}
 					WriteProgress(p);
 				}
-				_progressManager.UpdateCurrentProgressRecord(Pool.GetEstimatedProgressCount());
-				WriteProgress(_progressManager.ProgressRecord);
+				if (progress.Count > 0)
+				{
+					WriteProgressIfUpdated();
+				}
+			}
+
+			private void WriteProgressIfUpdated()
+			{
+				var estimatedCompletedCount = Pool.GetEstimatedProgressCount();
+				if (_lastEstimate != estimatedCompletedCount)
+				{
+					_lastEstimate = estimatedCompletedCount;
+					_progressManager.UpdateCurrentProgressRecord(estimatedCompletedCount);
+					WriteProgress(_progressManager.ProgressRecord);
+				}
 			}
 		}
 	}

src/PSParallel/PowerShellPoolMember.cs

@@ -1,33 +1,37 @@
 using System;
 using System.Management.Automation;
+using System.Management.Automation.Runspaces;
 
 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 int m_percentComplete;
+		private readonly PowershellPool _pool;
+		private readonly int _index;
+		private readonly PowerShellPoolStreams _poolStreams;
+		private readonly Runspace _runspace;		
+		private PowerShell _powerShell;
+		public PowerShell PowerShell => _powerShell;
+		public int Index => _index ;
+
+		private readonly PSDataCollection<PSObject> _input =new PSDataCollection<PSObject>();
+		private PSDataCollection<PSObject> _output;
+		private int _percentComplete;
 		public int PercentComplete
 		{
-			get { return m_percentComplete; }
-			set { m_percentComplete = value; }
+			get { return _percentComplete; }
+			set { _percentComplete = value; }
 		}
 
 
-		public PowerShellPoolMember(PowershellPool pool, int index)
+		public PowerShellPoolMember(PowershellPool pool, int index, Runspace runspace)
 		{
-			m_pool = pool;
-			m_index = index;
-			m_poolStreams = m_pool.Streams;
-			m_input.Complete();			
+			_pool = pool;
+			_index = index;
+			_runspace = runspace;
+			_runspace.Open();
+			_poolStreams = _pool.Streams;
+			_input.Complete();			
 			CreatePowerShell();			
 		}
 
@@ -37,34 +41,35 @@ private void PowerShellOnInvocationStateChanged(object sender, PSInvocationState
 			{
 				case PSInvocationState.Stopped:
 					ReleasePowerShell();
-					m_pool.ReportStopped(this);
+					_pool.ReportStopped(this);
 					break;
 				case PSInvocationState.Completed:
 				case PSInvocationState.Failed:
 					ReleasePowerShell();
 					CreatePowerShell();
-					m_pool.ReportAvailable(this);
+					_pool.ReportAvailable(this);
 					break;
 			}
 		}
 
 		private void CreatePowerShell()
 		{
 			var powerShell = PowerShell.Create();
+			powerShell.Runspace = _runspace;
 			HookStreamEvents(powerShell.Streams);
 			powerShell.InvocationStateChanged += PowerShellOnInvocationStateChanged;
-			m_powerShell = powerShell;
-			m_output = new PSDataCollection<PSObject>();
-			m_output.DataAdded += OutputOnDataAdded;
+			_powerShell = powerShell;
+			_output = new PSDataCollection<PSObject>();
+			_output.DataAdded += OutputOnDataAdded;
 		}
 
 		private void ReleasePowerShell()
 		{
-			UnhookStreamEvents(m_powerShell.Streams);
-			m_powerShell.InvocationStateChanged -= PowerShellOnInvocationStateChanged;
-			m_output.DataAdded -= OutputOnDataAdded;
-			m_powerShell.Dispose();
-			m_powerShell = null;
+			UnhookStreamEvents(_powerShell.Streams);
+			_powerShell.InvocationStateChanged -= PowerShellOnInvocationStateChanged;
+			_output.DataAdded -= OutputOnDataAdded;
+			_powerShell.Dispose();
+			_powerShell = null;
 		}
 
 
@@ -92,90 +97,96 @@ private void UnhookStreamEvents(PSDataStreams streams)
 
 		public void BeginInvoke(ScriptBlock scriptblock, PSObject inputObject)
 		{
-			m_percentComplete = 0;
+			_percentComplete = 0;
 			string command = $"param($_,$PSItem, $PSPArallelIndex,$PSParallelProgressId){scriptblock}";
-			m_powerShell.AddScript(command)
+			_powerShell.AddScript(command)
 				.AddParameter("_", inputObject)
 				.AddParameter("PSItem", inputObject)
-				.AddParameter("PSParallelIndex", m_index)
-				.AddParameter("PSParallelProgressId", m_index+1000);
-			m_powerShell.BeginInvoke(m_input, m_output);
+				.AddParameter("PSParallelIndex", _index)
+				.AddParameter("PSParallelProgressId", _index+1000);
+			_powerShell.BeginInvoke(_input, _output);
+		}
+
+		internal void Reset()
+		{			
+			_runspace.ResetRunspaceState();
 		}
 
 		public void Dispose()
 		{
-			var ps = m_powerShell;
+			var ps = _powerShell;
 			if (ps != null)
 			{
 				UnhookStreamEvents(ps.Streams);
 				ps.Dispose();
 			}
-			m_output.Dispose();
-			m_input.Dispose();
+			_output.Dispose();
+			_input.Dispose();
+			_runspace.Dispose();
 		}
 
 		private void OutputOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var item = ((PSDataCollection<PSObject>)sender)[dataAddedEventArgs.Index];
-			m_poolStreams.Output.Add(item);
+			_poolStreams.Output.Add(item);
 		}
 
 
 		private void InformationOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var ir = ((PSDataCollection<InformationRecord>)sender)[dataAddedEventArgs.Index];
-			m_poolStreams.Information.Add(ir);
+			_poolStreams.Information.Add(ir);
 		}
 
 		private void ProgressOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var record = ((PSDataCollection<ProgressRecord>)sender)[dataAddedEventArgs.Index];			
-			m_percentComplete = record.PercentComplete;
-			m_poolStreams.AddProgress(record, m_index);
+			_percentComplete = record.PercentComplete;
+			_poolStreams.AddProgress(record, _index);
 		}
 
 		private void ErrorOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var record = ((PSDataCollection<ErrorRecord>)sender)[dataAddedEventArgs.Index];
-			m_poolStreams.Error.Add(record);
+			_poolStreams.Error.Add(record);
 		}
 
 		private void DebugOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var record = ((PSDataCollection<DebugRecord>)sender)[dataAddedEventArgs.Index];
-			m_poolStreams.Debug.Add(record);
+			_poolStreams.Debug.Add(record);
 		}
 
 		private void WarningOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var record = ((PSDataCollection<WarningRecord>)sender)[dataAddedEventArgs.Index];
-			m_poolStreams.Warning.Add(record);
+			_poolStreams.Warning.Add(record);
 		}
 
 		private void VerboseOnDataAdded(object sender, DataAddedEventArgs dataAddedEventArgs)
 		{
 			var record = ((PSDataCollection<VerboseRecord>)sender)[dataAddedEventArgs.Index];
-			m_poolStreams.Verbose.Add(record);
+			_poolStreams.Verbose.Add(record);
 		}
 
 		public void Stop()
 		{
-			if(m_powerShell.InvocationStateInfo.State != PSInvocationState.Stopped) 
+			if(_powerShell.InvocationStateInfo.State != PSInvocationState.Stopped) 
 			{ 
-				UnhookStreamEvents(m_powerShell.Streams);
-				m_powerShell.BeginStop(OnStopped, null);
+				UnhookStreamEvents(_powerShell.Streams);
+				_powerShell.BeginStop(OnStopped, null);
 			}
 		}
 
 		private void OnStopped(IAsyncResult ar)
 		{
-			var ps = m_powerShell;
+			var ps = _powerShell;
 			if (ps == null)
 			{
 				return;
 			}
 			ps.EndStop(ar);
-			m_powerShell = null;
+			_powerShell = null;
 		}
 	}
 }