many code improvements

santisq · santisq · commit 91e7fc27fcbc · 2025-03-18T17:37:13.000-03:00
diff --git a/src/PSParallelPipeline/Commands/InvokeParallelCommand.cs b/src/PSParallelPipeline/Commands/InvokeParallelCommand.cs
@@ -90,7 +90,7 @@ protected override void ProcessRecord()
         }
         catch (OperationCanceledException exception)
         {
-            _worker.Wait();
+            _worker.WaitForCompletion();
             exception.WriteTimeoutError(this);
         }
     }
@@ -107,7 +107,7 @@ protected override void EndProcessing()
                 ProcessOutput(data);
             }
 
-            _worker.Wait();
+            _worker.WaitForCompletion();
         }
         catch (Exception _) when (_ is PipelineStoppedException or FlowControlException)
         {
@@ -116,7 +116,7 @@ protected override void EndProcessing()
         }
         catch (OperationCanceledException exception)
         {
-            _worker.Wait();
+            _worker.WaitForCompletion();
             exception.WriteTimeoutError(this);
         }
     }
@@ -125,33 +125,33 @@ private void ProcessOutput(PSOutputData data)
     {
         switch (data.Type)
         {
-            case Type.Success:
+            case OutputType.Success:
                 WriteObject(data.Output);
                 break;
 
-            case Type.Error:
+            case OutputType.Error:
                 WriteError((ErrorRecord)data.Output);
                 break;
 
-            case Type.Debug:
+            case OutputType.Debug:
                 DebugRecord debug = (DebugRecord)data.Output;
                 WriteDebug(debug.Message);
                 break;
 
-            case Type.Information:
+            case OutputType.Information:
                 WriteInformation((InformationRecord)data.Output);
                 break;
 
-            case Type.Progress:
+            case OutputType.Progress:
                 WriteProgress((ProgressRecord)data.Output);
                 break;
 
-            case Type.Verbose:
+            case OutputType.Verbose:
                 VerboseRecord verbose = (VerboseRecord)data.Output;
                 WriteVerbose(verbose.Message);
                 break;
 
-            case Type.Warning:
+            case OutputType.Warning:
                 WarningRecord warning = (WarningRecord)data.Output;
                 WriteWarning(warning.Message);
                 break;
@@ -161,7 +161,7 @@ private void ProcessOutput(PSOutputData data)
     private void CancelAndWait()
     {
         _cts.Cancel();
-        _worker?.Wait();
+        _worker?.WaitForCompletion();
     }
 
     protected override void StopProcessing() => CancelAndWait();
diff --git a/src/PSParallelPipeline/OutputType.cs b/src/PSParallelPipeline/OutputType.cs
@@ -0,0 +1,12 @@
+namespace PSParallelPipeline;
+
+internal enum OutputType
+{
+    Success,
+    Error,
+    Debug,
+    Information,
+    Progress,
+    Verbose,
+    Warning
+}
diff --git a/src/PSParallelPipeline/PSOutputData.cs b/src/PSParallelPipeline/PSOutputData.cs
@@ -1,36 +1,25 @@
 namespace PSParallelPipeline;
 
-internal enum Type
-{
-    Success,
-    Error,
-    Debug,
-    Information,
-    Progress,
-    Verbose,
-    Warning
-}
-
-internal record struct PSOutputData(Type Type, object Output)
+internal record struct PSOutputData(OutputType Type, object Output)
 {
     internal static PSOutputData WriteObject(object sendToPipeline) =>
-        new(Type.Success, sendToPipeline);
+        new(OutputType.Success, sendToPipeline);
 
     internal static PSOutputData WriteError(object error) =>
-        new(Type.Error, error);
+        new(OutputType.Error, error);
 
     internal static PSOutputData WriteDebug(object debug) =>
-        new(Type.Debug, debug);
+        new(OutputType.Debug, debug);
 
     internal static PSOutputData WriteInformation(object information) =>
-        new(Type.Information, information);
+        new(OutputType.Information, information);
 
     internal static PSOutputData WriteProgress(object progress) =>
-        new(Type.Progress, progress);
+        new(OutputType.Progress, progress);
 
     internal static PSOutputData WriteVerbose(object verbose) =>
-        new(Type.Verbose, verbose);
+        new(OutputType.Verbose, verbose);
 
     internal static PSOutputData WriteWarning(object warning) =>
-        new(Type.Warning, warning);
+        new(OutputType.Warning, warning);
 }
diff --git a/src/PSParallelPipeline/PSTask.cs b/src/PSParallelPipeline/PSTask.cs
@@ -9,44 +9,66 @@ namespace PSParallelPipeline;
 
 internal sealed class PSTask
 {
+    private const string SetVariableCommand = "Set-Variable";
+
+    private const string DollarUnderbar = "_";
+
+    private const string StopParsingOp = "--%";
+
     private readonly PowerShell _powershell;
 
     private readonly PSDataStreams _internalStreams;
 
-    private readonly RunspacePool _pool;
+    private Runspace? _runspace;
 
-    private PSOutputStreams OutputStreams { get => _pool.Streams; }
+    private readonly PSOutputStreams _outputStreams;
 
-    private Runspace Runspace
-    {
-        get => _powershell.Runspace;
-        set => _powershell.Runspace = value;
-    }
+    private readonly CancellationToken _token;
+
+    private readonly RunspacePool _pool;
 
     private PSTask(RunspacePool pool)
     {
         _powershell = PowerShell.Create();
         _internalStreams = _powershell.Streams;
+        _outputStreams = pool.Streams;
+        _token = pool.Token;
         _pool = pool;
     }
 
-    static internal async Task<PSTask> CreateAsync(
+    static internal PSTask Create(
         object? input,
         RunspacePool runspacePool,
         TaskSettings settings)
     {
         PSTask ps = new(runspacePool);
         SetStreams(ps._internalStreams, runspacePool.Streams);
-        ps.Runspace = await runspacePool
-            .GetRunspaceAsync()
-            .ConfigureAwait(false);
 
         return ps
             .AddInput(input)
             .AddScript(settings.Script)
             .AddUsingStatements(settings.UsingStatements);
     }
 
+    internal async Task InvokeAsync()
+    {
+        try
+        {
+            using CancellationTokenRegistration _ = _token.Register(Cancel);
+            _runspace = await _pool.GetRunspaceAsync().ConfigureAwait(false);
+            _powershell.Runspace = _runspace;
+            await InvokePowerShellAsync(_powershell, _outputStreams.Success).ConfigureAwait(false);
+        }
+        catch (Exception exception)
+        {
+            _outputStreams.AddOutput(exception.CreateProcessingTaskError(this));
+        }
+        finally
+        {
+            CompleteTask();
+        }
+    }
+
     private static void SetStreams(
         PSDataStreams streams,
         PSOutputStreams outputStreams)
@@ -71,8 +93,8 @@ private PSTask AddInput(object? inputObject)
         if (inputObject is not null)
         {
             _powershell
-                .AddCommand("Set-Variable", useLocalScope: true)
-                .AddArgument("_")
+                .AddCommand(SetVariableCommand, useLocalScope: true)
+                .AddArgument(DollarUnderbar)
                 .AddArgument(inputObject);
         }
 
@@ -89,33 +111,27 @@ private PSTask AddUsingStatements(Dictionary<string, object?> usingParams)
     {
         if (usingParams.Count > 0)
         {
-            _powershell.AddParameter("--%", usingParams);
+            _powershell.AddParameter(StopParsingOp, usingParams);
         }
 
         return this;
     }
 
-    internal async Task InvokeAsync()
+    private void CompleteTask()
     {
-        try
-        {
-            using CancellationTokenRegistration _ = _pool.RegisterCancellation(Cancel);
-            await InvokePowerShellAsync(_powershell, OutputStreams.Success).ConfigureAwait(false);
-        }
-        catch (Exception exception)
-        {
-            OutputStreams.AddOutput(exception.CreateProcessingTaskError(this));
-        }
-        finally
+        _powershell.Dispose();
+        if (!_token.IsCancellationRequested && _runspace is not null)
         {
-            _powershell.Dispose();
-            _pool.PushRunspace(Runspace);
+            _pool.PushRunspace(_runspace);
+            return;
         }
+
+        _runspace?.Dispose();
     }
 
     private void Cancel()
     {
         _powershell.Dispose();
-        Runspace.Dispose();
+        _runspace?.Dispose();
     }
 }
diff --git a/src/PSParallelPipeline/RunspacePool.cs b/src/PSParallelPipeline/RunspacePool.cs
@@ -8,71 +8,59 @@ namespace PSParallelPipeline;
 
 internal sealed class RunspacePool : IDisposable
 {
-    private readonly PoolSettings _settings;
-
-    private readonly CancellationToken _token;
+    private readonly SemaphoreSlim _semaphore;
 
-    private InitialSessionState InitialSessionState { get => _settings.InitialSessionState; }
+    private readonly PoolSettings _settings;
 
     private readonly ConcurrentQueue<Runspace> _pool = [];
 
-    private readonly ConcurrentDictionary<Guid, Runspace> _created;
-
     private bool UseNewRunspace { get => _settings.UseNewRunspace; }
 
-    private readonly SemaphoreSlim _semaphore;
+    internal int MaxRunspaces { get => _settings.MaxRunspaces; }
 
-    internal PSOutputStreams Streams { get; }
+    internal CancellationToken Token { get; }
 
-    internal int MaxRunspaces { get => _settings.MaxRunspaces; }
+    internal PSOutputStreams Streams { get; }
 
     internal RunspacePool(
         PoolSettings settings,
         PSOutputStreams streams,
         CancellationToken token)
     {
-        _settings = settings;
         Streams = streams;
-        _token = token;
+        Token = token;
+        _settings = settings;
         _semaphore = new SemaphoreSlim(MaxRunspaces, MaxRunspaces);
-        _created = new ConcurrentDictionary<Guid, Runspace>(
-            Environment.ProcessorCount,
-            MaxRunspaces);
     }
 
     internal void PushRunspace(Runspace runspace)
     {
-        if (_token.IsCancellationRequested)
+        if (Token.IsCancellationRequested)
         {
             return;
         }
 
         if (UseNewRunspace)
         {
             runspace.Dispose();
-            _created.TryRemove(runspace.InstanceId, out _);
             runspace = CreateRunspace();
         }
 
         _pool.Enqueue(runspace);
         _semaphore.Release();
     }
 
-    internal CancellationTokenRegistration RegisterCancellation(Action callback) =>
-        _token.Register(callback);
-
     private Runspace CreateRunspace()
     {
-        Runspace rs = RunspaceFactory.CreateRunspace(InitialSessionState);
-        _created[rs.InstanceId] = rs;
+        Runspace rs = RunspaceFactory.CreateRunspace(_settings.InitialSessionState);
         rs.Open();
         return rs;
     }
 
     internal async Task<Runspace> GetRunspaceAsync()
     {
         await _semaphore
-            .WaitAsync(_token)
+            .WaitAsync(Token)
             .ConfigureAwait(false);
 
         if (_pool.TryDequeue(out Runspace runspace))
@@ -85,7 +73,7 @@ await _semaphore
 
     public void Dispose()
     {
-        foreach (Runspace runspace in _created.Values)
+        foreach (Runspace runspace in _pool)
         {
             runspace.Dispose();
         }
diff --git a/src/PSParallelPipeline/Worker.cs b/src/PSParallelPipeline/Worker.cs
@@ -33,7 +33,7 @@ internal Worker(
         _pool = new RunspacePool(poolSettings, _streams, _token);
     }
 
-    internal void Wait() => _worker?.ConfigureAwait(false).GetAwaiter().GetResult();
+    internal void WaitForCompletion() => _worker?.GetAwaiter().GetResult();
 
     internal void Enqueue(object? input) => _input.Add(input, _token);
 
@@ -58,17 +58,12 @@ private async Task Start()
                     await ProcessAnyAsync(tasks).ConfigureAwait(false);
                 }
 
-                PSTask task = await PSTask
-                    .CreateAsync(
-                        input: input,
-                        runspacePool: _pool,
-                        settings: _taskSettings)
-                    .ConfigureAwait(false);
-
-                tasks.Add(task.InvokeAsync());
+                tasks.Add(PSTask
+                    .Create(input, _pool, _taskSettings)
+                    .InvokeAsync());
             }
         }
-        catch
+        catch (OperationCanceledException)
         { }
         finally
         {
@@ -87,8 +82,7 @@ private static async Task ProcessAnyAsync(List<Task> tasks)
             .ConfigureAwait(false);
 
         tasks.Remove(task);
-        await task
-            .ConfigureAwait(false);
+        await task.ConfigureAwait(false);
     }
 
     public void Dispose()

Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ protected override void ProcessRecord()`
`90`	`90`	`}`
`91`	`91`	`catch (OperationCanceledException exception)`
`92`	`92`	`{`
`93`		`- _worker.Wait();`
	`93`	`+ _worker.WaitForCompletion();`
`94`	`94`	`exception.WriteTimeoutError(this);`
`95`	`95`	`}`
`96`	`96`	`}`
`@@ -107,7 +107,7 @@ protected override void EndProcessing()`
`107`	`107`	`ProcessOutput(data);`
`108`	`108`	`}`
`109`	`109`
`110`		`- _worker.Wait();`
	`110`	`+ _worker.WaitForCompletion();`
`111`	`111`	`}`
`112`	`112`	`catch (Exception _) when (_ is PipelineStoppedException or FlowControlException)`
`113`	`113`	`{`
`@@ -116,7 +116,7 @@ protected override void EndProcessing()`
`116`	`116`	`}`
`117`	`117`	`catch (OperationCanceledException exception)`
`118`	`118`	`{`
`119`		`- _worker.Wait();`
	`119`	`+ _worker.WaitForCompletion();`
`120`	`120`	`exception.WriteTimeoutError(this);`
`121`	`121`	`}`
`122`	`122`	`}`
`@@ -125,33 +125,33 @@ private void ProcessOutput(PSOutputData data)`
`125`	`125`	`{`
`126`	`126`	`switch (data.Type)`
`127`	`127`	`{`
`128`		`- case Type.Success:`
	`128`	`+ case OutputType.Success:`
`129`	`129`	`WriteObject(data.Output);`
`130`	`130`	`break;`
`131`	`131`
`132`		`- case Type.Error:`
	`132`	`+ case OutputType.Error:`
`133`	`133`	`WriteError((ErrorRecord)data.Output);`
`134`	`134`	`break;`
`135`	`135`
`136`		`- case Type.Debug:`
	`136`	`+ case OutputType.Debug:`
`137`	`137`	`DebugRecord debug = (DebugRecord)data.Output;`
`138`	`138`	`WriteDebug(debug.Message);`
`139`	`139`	`break;`
`140`	`140`
`141`		`- case Type.Information:`
	`141`	`+ case OutputType.Information:`
`142`	`142`	`WriteInformation((InformationRecord)data.Output);`
`143`	`143`	`break;`
`144`	`144`
`145`		`- case Type.Progress:`
	`145`	`+ case OutputType.Progress:`
`146`	`146`	`WriteProgress((ProgressRecord)data.Output);`
`147`	`147`	`break;`
`148`	`148`
`149`		`- case Type.Verbose:`
	`149`	`+ case OutputType.Verbose:`
`150`	`150`	`VerboseRecord verbose = (VerboseRecord)data.Output;`
`151`	`151`	`WriteVerbose(verbose.Message);`
`152`	`152`	`break;`
`153`	`153`
`154`		`- case Type.Warning:`
	`154`	`+ case OutputType.Warning:`
`155`	`155`	`WarningRecord warning = (WarningRecord)data.Output;`
`156`	`156`	`WriteWarning(warning.Message);`
`157`	`157`	`break;`
`@@ -161,7 +161,7 @@ private void ProcessOutput(PSOutputData data)`
`161`	`161`	`private void CancelAndWait()`
`162`	`162`	`{`
`163`	`163`	`_cts.Cancel();`
`164`		`- _worker?.Wait();`
	`164`	`+ _worker?.WaitForCompletion();`
`165`	`165`	`}`
`166`	`166`
`167`	`167`	`protected override void StopProcessing() => CancelAndWait();`
Original file line number	Diff line number	Diff line change
`@@ -8,71 +8,59 @@ namespace PSParallelPipeline;`
`8`	`8`
`9`	`9`	`internal sealed class RunspacePool : IDisposable`
`10`	`10`	`{`
`11`		`- private readonly PoolSettings _settings;`
`12`		`-`
`13`		`- private readonly CancellationToken _token;`
	`11`	`+ private readonly SemaphoreSlim _semaphore;`
`14`	`12`
`15`		`- private InitialSessionState InitialSessionState { get => _settings.InitialSessionState; }`
	`13`	`+ private readonly PoolSettings _settings;`
`16`	`14`
`17`	`15`	`private readonly ConcurrentQueue<Runspace> _pool = [];`
`18`	`16`
`19`		`- private readonly ConcurrentDictionary<Guid, Runspace> _created;`
`20`		`-`
`21`	`17`	`private bool UseNewRunspace { get => _settings.UseNewRunspace; }`
`22`	`18`
`23`		`- private readonly SemaphoreSlim _semaphore;`
	`19`	`+ internal int MaxRunspaces { get => _settings.MaxRunspaces; }`
`24`	`20`
`25`		`- internal PSOutputStreams Streams { get; }`
	`21`	`+ internal CancellationToken Token { get; }`
`26`	`22`
`27`		`- internal int MaxRunspaces { get => _settings.MaxRunspaces; }`
	`23`	`+ internal PSOutputStreams Streams { get; }`
`28`	`24`
`29`	`25`	`internal RunspacePool(`
`30`	`26`	`PoolSettings settings,`
`31`	`27`	`PSOutputStreams streams,`
`32`	`28`	`CancellationToken token)`
`33`	`29`	`{`
`34`		`- _settings = settings;`
`35`	`30`	`Streams = streams;`
`36`		`- _token = token;`
	`31`	`+ Token = token;`
	`32`	`+ _settings = settings;`
`37`	`33`	`_semaphore = new SemaphoreSlim(MaxRunspaces, MaxRunspaces);`
`38`		`- _created = new ConcurrentDictionary<Guid, Runspace>(`
`39`		`- Environment.ProcessorCount,`
`40`		`- MaxRunspaces);`
`41`	`34`	`}`
`42`	`35`
`43`	`36`	`internal void PushRunspace(Runspace runspace)`
`44`	`37`	`{`
`45`		`- if (_token.IsCancellationRequested)`
	`38`	`+ if (Token.IsCancellationRequested)`
`46`	`39`	`{`
`47`	`40`	`return;`
`48`	`41`	`}`
`49`	`42`
`50`	`43`	`if (UseNewRunspace)`
`51`	`44`	`{`
`52`	`45`	`runspace.Dispose();`
`53`		`- _created.TryRemove(runspace.InstanceId, out _);`
`54`	`46`	`runspace = CreateRunspace();`
`55`	`47`	`}`
`56`	`48`
`57`	`49`	`_pool.Enqueue(runspace);`
`58`	`50`	`_semaphore.Release();`
`59`	`51`	`}`
`60`	`52`
`61`		`- internal CancellationTokenRegistration RegisterCancellation(Action callback) =>`
`62`		`- _token.Register(callback);`
`63`		`-`
`64`	`53`	`private Runspace CreateRunspace()`
`65`	`54`	`{`
`66`		`- Runspace rs = RunspaceFactory.CreateRunspace(InitialSessionState);`
`67`		`- _created[rs.InstanceId] = rs;`
	`55`	`+ Runspace rs = RunspaceFactory.CreateRunspace(_settings.InitialSessionState);`
`68`	`56`	`rs.Open();`
`69`	`57`	`return rs;`
`70`	`58`	`}`
`71`	`59`
`72`	`60`	`internal async Task<Runspace> GetRunspaceAsync()`
`73`	`61`	`{`
`74`	`62`	`await _semaphore`
`75`		`- .WaitAsync(_token)`
	`63`	`+ .WaitAsync(Token)`
`76`	`64`	`.ConfigureAwait(false);`
`77`	`65`
`78`	`66`	`if (_pool.TryDequeue(out Runspace runspace))`
`@@ -85,7 +73,7 @@ await _semaphore`
`85`	`73`
`86`	`74`	`public void Dispose()`
`87`	`75`	`{`
`88`		`- foreach (Runspace runspace in _created.Values)`
	`76`	`+ foreach (Runspace runspace in _pool)`
`89`	`77`	`{`
`90`	`78`	`runspace.Dispose();`
`91`	`79`	`}`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ internal Worker(`
`33`	`33`	`_pool = new RunspacePool(poolSettings, _streams, _token);`
`34`	`34`	`}`
`35`	`35`
`36`		`- internal void Wait() => _worker?.ConfigureAwait(false).GetAwaiter().GetResult();`
	`36`	`+ internal void WaitForCompletion() => _worker?.GetAwaiter().GetResult();`
`37`	`37`
`38`	`38`	`internal void Enqueue(object? input) => _input.Add(input, _token);`
`39`	`39`
`@@ -58,17 +58,12 @@ private async Task Start()`
`58`	`58`	`await ProcessAnyAsync(tasks).ConfigureAwait(false);`
`59`	`59`	`}`
`60`	`60`
`61`		`- PSTask task = await PSTask`
`62`		`- .CreateAsync(`
`63`		`- input: input,`
`64`		`- runspacePool: _pool,`
`65`		`- settings: _taskSettings)`
`66`		`- .ConfigureAwait(false);`
`67`		`-`
`68`		`- tasks.Add(task.InvokeAsync());`
	`61`	`+ tasks.Add(PSTask`
	`62`	`+ .Create(input, _pool, _taskSettings)`
	`63`	`+ .InvokeAsync());`
`69`	`64`	`}`
`70`	`65`	`}`
`71`		`- catch`
	`66`	`+ catch (OperationCanceledException)`
`72`	`67`	`{ }`
`73`	`68`	`finally`
`74`	`69`	`{`
`@@ -87,8 +82,7 @@ private static async Task ProcessAnyAsync(List<Task> tasks)`
`87`	`82`	`.ConfigureAwait(false);`
`88`	`83`
`89`	`84`	`tasks.Remove(task);`
`90`		`- await task`
`91`		`- .ConfigureAwait(false);`
	`85`	`+ await task.ConfigureAwait(false);`
`92`	`86`	`}`
`93`	`87`
`94`	`88`	`public void Dispose()`