diff --git a/.vscode/launch.json b/.vscode/launch.json index 4fdc927..81b33ff 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -18,7 +18,7 @@ ], "cwd": "${workspaceFolder}", "stopAtEntry": false, - "console": "externalTerminal", + "console": "integratedTerminal", }, { "name": "PowerShell Launch Current File", @@ -27,13 +27,6 @@ "script": "${file}", "cwd": "${workspaceFolder}" }, - { - "name": ".NET FullCLR Attach", - "type": "clr", - "request": "attach", - "processId": "${command:pickProcess}", - "justMyCode": true, - }, { "name": ".NET CoreCLR Attach", "type": "coreclr", diff --git a/README.md b/README.md index c88cb78..29f29d4 100644 --- a/README.md +++ b/README.md @@ -72,19 +72,24 @@ $message = 'world!' ## `-Variables`, `-Functions`, `-ModuleNames`, and `-ModulePaths` Parameters -- [`-Variables` Parameter](./docs/en-US/Invoke-Parallel.md#-variables): Pass variables directly to parallel runspaces. +- [`-Variables` Parameter](./docs/en-US/Invoke-Parallel.md#-variables): Pass variables directly using hashtables, multiple dictionaries, or wildcard patterns to automatically match caller variables. ```powershell + # Using a hashtable or dictionary 'hello ' | Invoke-Parallel { $_ + $msg } -Variables @{ msg = 'world!' } - # hello world! + + # Using wildcards to import matching caller variables + $appUser = 'John Doe' + $appRole = 'Admin' + 0..5 | Invoke-Parallel { "Processing $_ $appUser ($appRole)" } -Variables 'app*' ``` -- [`-Functions` Parameter](./docs/en-US/Invoke-Parallel.md#-functions): Use local functions in parallel scopes without redefining them. +- [`-Functions` Parameter](./docs/en-US/Invoke-Parallel.md#-functions): Use local functions in parallel scopes by name or wildcard pattern without redefining them. ```powershell - function Get-Message {param($MyParam) $MyParam + 'world!' } - 'hello ' | Invoke-Parallel { Get-Message $_ } -Functions Get-Message - # hello world! + function Get-Greeting { param($s) "Hello $s" } + function Get-Farewell { param($s) "Goodbye $s" } + 0..5 | Invoke-Parallel { Get-Greeting $_; Get-Farewell$_ } -Functions Get-* ``` - [`-ModuleNames` Parameter](./docs/en-US/Invoke-Parallel.md#-modulenames): Import system-installed modules into parallel runspaces by name, using modules discoverable via `$env:PSModulePath`. diff --git a/docs/en-US/Invoke-Parallel.md b/docs/en-US/Invoke-Parallel.md index c3f9bc5..b281475 100644 --- a/docs/en-US/Invoke-Parallel.md +++ b/docs/en-US/Invoke-Parallel.md @@ -19,7 +19,7 @@ Invoke-Parallel [-InputObject ] [-ThrottleLimit ] [-TimeoutSeconds ] - [-Variables ] + [-Variables ] [-Functions ] [-ModuleNames ] [-ModulePaths ] @@ -67,7 +67,29 @@ This example demonstrates the [`-Variables` parameter](#-variables), which passe the parallel scope using a hashtable. The key `message` in the hashtable defines the variable name available within the script block, serving as an alternative to the `$using:` scope modifier. -### Example 3: Adding to a thread-safe collection with `$using:` +### Example 3: Import variables using wildcards + +```powershell +$appUser = 'John Doe' +$appRole = 'Admin' + +0..5 | Invoke-Parallel { "Processing $_ for $appUser ($appRole)" } -Variables app* +``` + +This example uses wildcards with `-Variables` to dynamically import all caller variables starting with `app` (in this case `$appUser` and `$appRole`) into the parallel scope. + +### Example 4: Passing multiple dictionaries to `-Variables` + +```powershell +$dict1 = @{ Server = 'localhost' } +$dict2 = @{ Port = 8080 } + +0..5 | Invoke-Parallel { "Connecting to $Server:$Port" } -Variables $dict1, $dict2 +``` + +This example demonstrates passing an array of dictionaries to `-Variables`. Key/value pairs from both dictionaries are merged into the parallel scope. + +### Example 5: Adding to a thread-safe collection with `$using:` ```powershell $dict = [System.Collections.Concurrent.ConcurrentDictionary[int, object]]::new() @@ -78,7 +100,7 @@ $dict[$PID] This example uses a thread-safe dictionary to store process objects by ID, leveraging the `$using:` modifier for variable access. -### Example 4: Adding to a thread-safe collection with `-Variables` +### Example 6: Adding to a thread-safe collection with `-Variables` ```powershell $dict = [System.Collections.Concurrent.ConcurrentDictionary[int, object]]::new() @@ -88,18 +110,18 @@ $dict[$PID] Similar to Example 3, this demonstrates the same functionality using `-Variables` instead of `$using:`. -### Example 5: Using the `-Functions` parameter +### Example 7: Using the `-Functions` parameter with wildcards ```powershell -function Greet { param($s) "$s hey there!" } +function Get-Greeting { param($s) "Hello $s" } +function Get-Farewell { param($s) "Goodbye $s" } -0..10 | Invoke-Parallel { Greet $_ } -Functions Greet +0..5 | Invoke-Parallel { Get-Greeting $_; Get-Farewell $_ } -Functions Get-* ``` -This example imports a local function `Greet` into the parallel scope using [`-Functions` parameter](#-functions), -allowing its use within the script block. +This example imports all functions matching the wildcard pattern `Get-*` from the local session into the parallel scope using the [`-Functions` parameter](#-functions). -### Example 6: Setting a timeout with `-TimeoutSeconds` +### Example 8: Setting a timeout with `-TimeoutSeconds` ```powershell 0..10 | Invoke-Parallel { Start-Sleep 1 } -TimeoutSeconds 3 @@ -108,7 +130,7 @@ allowing its use within the script block. This example limits execution to 3 seconds, stopping all running script blocks and ignoring unprocessed input once the timeout is reached. -### Example 7: Creating new runspaces with `-UseNewRunspace` +### Example 9: Creating new runspaces with `-UseNewRunspace` ```powershell 0..3 | Invoke-Parallel { [runspace]::DefaultRunspace.InstanceId } -ThrottleLimit 2 @@ -133,7 +155,7 @@ timeout is reached. This example contrasts default runspace reuse with the `-UseNewRunspace` switch, showing unique runspace IDs for each invocation in the latter case. -### Example 8: Using the `-ModuleNames` parameter +### Example 10: Using the `-ModuleNames` parameter ```powershell Import-Csv users.csv | Invoke-Parallel { Get-ADUser $_.UserPrincipalName } -ModuleNames ActiveDirectory @@ -142,7 +164,7 @@ Import-Csv users.csv | Invoke-Parallel { Get-ADUser $_.UserPrincipalName } -Modu This example imports the `ActiveDirectory` module into the parallel scope using `-ModuleNames`, enabling the `Get-ADUser` cmdlet within the script block. -### Example 9: Using the `-ModulePaths` parameter +### Example 11: Using the `-ModulePaths` parameter ```powershell $moduleDir = Join-Path $PSScriptRoot "CustomModule" @@ -160,10 +182,14 @@ function to be used in the parallel script block. ### -Functions -Specifies an array of function names from the local session to include in the runspaces’ +Specifies an array of function names or wildcard patterns from the local session to include in the runspaces’ [Initial Session State](https://learn.microsoft.com/en-us/dotnet/api/system.management.automation.runspaces.initialsessionstate). This enables their use within the parallel script block. +> [!NOTE] +> +> If a specified name or wildcard pattern fails to match any function in the caller's scope, a terminating error is thrown. + > [!TIP] > > This parameter is the recommended way to make local functions available in the parallel scope. @@ -179,7 +205,7 @@ Required: False Position: Named Default value: None Accept pipeline input: False -Accept wildcard characters: False +Accept wildcard characters: True ``` ### -InputObject @@ -270,8 +296,17 @@ Accept wildcard characters: False ### -Variables -Provides a hashtable of variables to make available in the parallel scope. Keys define the variable names within the -script block. +Provides dictionaries (`IDictionary`), variable names, or wildcard patterns to make matching caller variables available in the parallel scope. + +You can supply arguments in multiple forms: + +- __Dictionaries__: Pass one or more dictionaries (e.g., hashtables). Keys specify variable names in the parallel scope. Duplicate keys across multiple dictionaries are ignored. +- __Variable Names or Wildcard Patterns__: Pass strings containing variable names or wildcard expressions (e.g., `*`, `foo*`, `[ab]*`). __Non-built-in__ variables in the caller scope matching the pattern will be imported automatically. + +> [!NOTE] +> +> - When resolving variables by string/wildcard pattern, variables holding a `ScriptBlock` value are silently skipped. +> - If a string or wildcard pattern does not match any variable in the caller's scope, a terminating error is thrown. > [!TIP] > @@ -287,7 +322,7 @@ Required: False Position: Named Default value: None Accept pipeline input: False -Accept wildcard characters: False +Accept wildcard characters: True ``` ### -ModuleNames diff --git a/module/PSParallelPipeline.psd1 b/module/PSParallelPipeline.psd1 index 43db821..4ccbecf 100644 --- a/module/PSParallelPipeline.psd1 +++ b/module/PSParallelPipeline.psd1 @@ -11,7 +11,7 @@ RootModule = 'bin/netstandard2.0/PSParallelPipeline.dll' # Version number of this module. - ModuleVersion = '1.2.5' + ModuleVersion = '1.3.0' # Supported PSEditions # CompatiblePSEditions = @() diff --git a/src/PSParallelPipeline/Commands/InvokeParallelCommand.cs b/src/PSParallelPipeline/Commands/InvokeParallelCommand.cs index cb39d28..6e57221 100644 --- a/src/PSParallelPipeline/Commands/InvokeParallelCommand.cs +++ b/src/PSParallelPipeline/Commands/InvokeParallelCommand.cs @@ -33,12 +33,15 @@ public sealed class InvokeParallelCommand : PSCmdlet, IDisposable [Parameter] [ValidateNotNullOrEmpty] + [VariableTransformation] + [SupportsWildcards] [Alias("vars")] - public Hashtable? Variables { get; set; } + public IDictionary[]? Variables { get; set; } [Parameter] [ValidateNotNullOrEmpty] [ArgumentCompleter(typeof(CommandCompleter))] + [SupportsWildcards] [Alias("funcs")] public string[]? Functions { get; set; } @@ -83,11 +86,7 @@ protected override void BeginProcessing() protected override void ProcessRecord() { - if (_worker is null) - { - return; - } - + if (_worker is null) return; InputObject.ThrowIfInputObjectIsScriptBlock(this); try diff --git a/src/PSParallelPipeline/ExceptionExtensions.cs b/src/PSParallelPipeline/ExceptionExtensions.cs new file mode 100644 index 0000000..9be06a9 --- /dev/null +++ b/src/PSParallelPipeline/ExceptionExtensions.cs @@ -0,0 +1,122 @@ +using System; +using System.IO; +using System.Management.Automation; +using Microsoft.PowerShell.Commands; + +namespace PSParallelPipeline; + +internal static class ExceptionExtensions +{ + private const string ScriptBlockNotSupported = + "Passed-in script block variables are not supported, and can result in undefined behavior."; + + private static readonly PSArgumentException PassedInVariableCannotBeScriptBlock = new(ScriptBlockNotSupported); + + private static readonly PSArgumentException UsingVariableCannotBeScriptBlock = new( + $"A $using: variable cannot be a script block. {ScriptBlockNotSupported}"); + + private static readonly PSArgumentException InputObjectCannotBeScriptBlock = new( + $"Piped input object cannot be a script block. {ScriptBlockNotSupported}"); + + extension(Exception exception) + { + internal void WriteTimeoutError(PSCmdlet cmdlet) => + cmdlet.WriteError(new ErrorRecord( + new TimeoutException("Timeout has been reached.", exception), + "TimeOutReached", + ErrorCategory.OperationTimeout, + cmdlet)); + + internal PSOutputData CreateProcessingTaskError() => + PSOutputData.CreateError(new ErrorRecord( + exception, "ProcessingTask", ErrorCategory.NotSpecified, null)); + } + + extension(object? value) + { + internal bool IsNotScriptBlock() => + value is not ScriptBlock and not PSObject { BaseObject: ScriptBlock }; + + internal void ThrowIfInputObjectIsScriptBlock(PSCmdlet cmdlet) + { + if (value.IsNotScriptBlock()) return; + + ErrorRecord error = new( + InputObjectCannotBeScriptBlock, + nameof(InputObjectCannotBeScriptBlock), + ErrorCategory.InvalidType, value); + + cmdlet.ThrowTerminatingError(error); + } + } + + extension(PSCmdlet cmdlet) + { + internal void ThrowIfVariableIsScriptBlock(object? value) + { + if (value.IsNotScriptBlock()) return; + + ErrorRecord error = new( + PassedInVariableCannotBeScriptBlock, + nameof(PassedInVariableCannotBeScriptBlock), + ErrorCategory.InvalidType, value); + + cmdlet.ThrowTerminatingError(error); + } + + internal void ThrowIfUsingValueIsScriptBlock(object? value) + { + if (value.IsNotScriptBlock()) return; + + ErrorRecord error = new( + UsingVariableCannotBeScriptBlock, + nameof(UsingVariableCannotBeScriptBlock), + ErrorCategory.InvalidType, value); + + cmdlet.ThrowTerminatingError(error); + } + + internal void ThrowFunctionNotFoundError(string function) + { + Exception ex = new CommandNotFoundException( + $"Could not find any function matching the name or pattern '{function}'."); + ErrorRecord error = new(ex, "FunctionNotFound", ErrorCategory.ObjectNotFound, function); + cmdlet.ThrowTerminatingError(error); + } + } + + extension(ProviderInfo provider) + { + internal void ThrowIfInvalidProvider(string path, PSCmdlet cmdlet) + { + if (provider.ImplementingType == typeof(FileSystemProvider)) return; + + ErrorRecord error = new( + new NotSupportedException( + $"The resolved path '{path}' is not a FileSystem path but '{provider.Name}'."), + "NotFileSystemPath", + ErrorCategory.InvalidArgument, + path); + + cmdlet.ThrowTerminatingError(error); + } + } + + extension(string path) + { + internal void ThrowIfNotDirectory(PSCmdlet cmdlet) + { + if (Directory.Exists(path)) return; + + ErrorRecord error = new( + new ArgumentException( + $"The specified path '{path}' does not exist or is not a directory. " + + "The path must be a valid directory containing one or more PowerShell modules."), + "NotDirectoryPath", + ErrorCategory.InvalidArgument, + path); + + cmdlet.ThrowTerminatingError(error); + } + } +} diff --git a/src/PSParallelPipeline/ExceptionHelper.cs b/src/PSParallelPipeline/ExceptionHelper.cs deleted file mode 100644 index e4fe057..0000000 --- a/src/PSParallelPipeline/ExceptionHelper.cs +++ /dev/null @@ -1,134 +0,0 @@ -using System; -using System.IO; -using System.Management.Automation; -using Microsoft.PowerShell.Commands; - -namespace PSParallelPipeline; - -internal static class ExceptionHelper -{ - private const string NotSupported = - "Passed-in script block variables are not supported, and can result in undefined behavior."; - - internal static void WriteTimeoutError(this Exception exception, PSCmdlet cmdlet) => - cmdlet.WriteError(new ErrorRecord( - new TimeoutException("Timeout has been reached.", exception), - "TimeOutReached", - ErrorCategory.OperationTimeout, - cmdlet)); - - internal static PSOutputData CreateProcessingTaskError(this Exception exception, object context) => - PSOutputData.CreateError(new ErrorRecord( - exception, "ProcessingTask", ErrorCategory.NotSpecified, context)); - - internal static void ThrowFunctionNotFoundError( - this CommandNotFoundException exception, - Cmdlet cmdlet, - string function) => - cmdlet.ThrowTerminatingError(new ErrorRecord( - exception, "FunctionNotFound", ErrorCategory.ObjectNotFound, function)); - - private static bool ValueIsNotScriptBlock(object? value) => - value is not ScriptBlock and not PSObject { BaseObject: ScriptBlock }; - - internal static CommandInfo ThrowIfFunctionNotFoundError( - this CommandInfo? command, - string function) - { - if (command is not null) - { - return command; - } - - throw new CommandNotFoundException( - $"The function with name '{function}' could not be found."); - } - - internal static void ThrowIfVariableIsScriptBlock(this PSCmdlet cmdlet, object? value) - { - if (ValueIsNotScriptBlock(value)) - { - return; - } - - cmdlet.ThrowTerminatingError(new ErrorRecord( - new PSArgumentException(NotSupported), - "PassedInVariableCannotBeScriptBlock", - ErrorCategory.InvalidType, - value)); - } - - internal static void ThrowIfInputObjectIsScriptBlock(this object? value, PSCmdlet cmdlet) - { - if (ValueIsNotScriptBlock(value)) - { - return; - } - - cmdlet.ThrowTerminatingError(new ErrorRecord( - new PSArgumentException( - string.Concat( - "Piped input object cannot be a script block. ", - NotSupported)), - "InputObjectCannotBeScriptBlock", - ErrorCategory.InvalidType, - value)); - } - - internal static void ThrowIfUsingValueIsScriptBlock(this PSCmdlet cmdlet, object? value) - { - if (ValueIsNotScriptBlock(value)) - { - return; - } - - cmdlet.ThrowTerminatingError(new ErrorRecord( - new PSArgumentException( - string.Concat( - "A $using: variable cannot be a script block. ", - NotSupported)), - "UsingVariableCannotBeScriptBlock", - ErrorCategory.InvalidType, - value)); - } - - internal static void ThrowIfInvalidProvider( - this ProviderInfo provider, - string path, - PSCmdlet cmdlet) - { - if (provider.ImplementingType == typeof(FileSystemProvider)) - { - return; - } - - ErrorRecord error = new( - new NotSupportedException( - $"The resolved path '{path}' is not a FileSystem path but '{provider.Name}'."), - "NotFileSystemPath", - ErrorCategory.InvalidArgument, - path); - - cmdlet.ThrowTerminatingError(error); - } - - internal static void ThrowIfNotDirectory( - this string path, - PSCmdlet cmdlet) - { - if (Directory.Exists(path)) - { - return; - } - - ErrorRecord error = new( - new ArgumentException( - $"The specified path '{path}' does not exist or is not a directory. " + - "The path must be a valid directory containing one or more PowerShell modules."), - "NotDirectoryPath", - ErrorCategory.InvalidArgument, - path); - - cmdlet.ThrowTerminatingError(error); - } -} diff --git a/src/PSParallelPipeline/Extensions.cs b/src/PSParallelPipeline/Extensions.cs deleted file mode 100644 index e0e326d..0000000 --- a/src/PSParallelPipeline/Extensions.cs +++ /dev/null @@ -1,236 +0,0 @@ -using System; -using System.Collections; -using System.Collections.Generic; -using System.Linq; -using System.Management.Automation; -using System.Management.Automation.Language; -using System.Management.Automation.Runspaces; -using System.Runtime.CompilerServices; -using System.Text; -using System.Threading.Tasks; - -namespace PSParallelPipeline; - -internal static class Extensions -{ - internal static InitialSessionState AddFunctions( - this InitialSessionState initialSessionState, - string[]? functionsToAdd, - PSCmdlet cmdlet) - { - if (functionsToAdd is not null) - { - foreach (string function in functionsToAdd) - { - try - { - CommandInfo commandInfo = cmdlet - .InvokeCommand - .GetCommand(function, CommandTypes.Function) - .ThrowIfFunctionNotFoundError(function); - - initialSessionState.Commands.Add( - new SessionStateFunctionEntry( - name: function, - definition: commandInfo.Definition)); - } - catch (CommandNotFoundException exception) - { - exception.ThrowFunctionNotFoundError(cmdlet, function); - } - } - } - - return initialSessionState; - } - - internal static InitialSessionState AddVariables( - this InitialSessionState initialSessionState, - Hashtable? variables, - PSCmdlet cmdlet) - { - if (variables is not null) - { - foreach (DictionaryEntry pair in variables) - { - cmdlet.ThrowIfVariableIsScriptBlock(pair.Value); - initialSessionState.Variables.Add(new SessionStateVariableEntry( - name: LanguagePrimitives.ConvertTo(pair.Key), - value: pair.Value, - description: null)); - } - } - - return initialSessionState; - } - - internal static InitialSessionState ImportModules( - this InitialSessionState initialSessionState, - string[]? modulesToImport) - { - if (modulesToImport is not null) - { - initialSessionState.ImportPSModule(modulesToImport); - } - - return initialSessionState; - } - - internal static InitialSessionState ImportModulesFromPath( - this InitialSessionState initialSessionState, - string[]? modulePaths, - PSCmdlet cmdlet) - { - - if (modulePaths is not null) - { - foreach (string path in modulePaths) - { - string resolved = cmdlet.ResolvePath(path); - initialSessionState.ImportPSModulesFromPath(resolved); - } - } - - return initialSessionState; - } - - private static string ResolvePath(this PSCmdlet cmdlet, string path) - { - string resolved = cmdlet.SessionState.Path.GetUnresolvedProviderPathFromPSPath( - path: path, - provider: out ProviderInfo provider, - drive: out _); - - provider.ThrowIfInvalidProvider(path, cmdlet); - resolved.ThrowIfNotDirectory(cmdlet); - return resolved.TrimEnd('\\', '/'); - } - - internal static Dictionary GetUsingParameters( - this ScriptBlock script, - PSCmdlet cmdlet) - { - Dictionary usingParams = []; - IEnumerable usingExpressionAsts = script.Ast - .FindAll(a => a is UsingExpressionAst, true) - .Cast(); - - foreach (UsingExpressionAst usingStatement in usingExpressionAsts) - { - VariableExpressionAst backingVariableAst = UsingExpressionAst - .ExtractUsingVariable(usingStatement); - - string varPath = backingVariableAst.VariablePath.UserPath; - - string varText = usingStatement.ToString(); - if (usingStatement.SubExpression is VariableExpressionAst) - { - varText = varText.ToLowerInvariant(); - } - - string key = Convert.ToBase64String(Encoding.Unicode.GetBytes(varText)); - object? value = cmdlet.GetVariableValue(varPath); - cmdlet.ThrowIfUsingValueIsScriptBlock(value); - - if (usingParams.ContainsKey(key)) - { - continue; - } - - if (usingStatement.SubExpression is MemberExpressionAst or IndexExpressionAst) - { - value = ExtractUsingExpressionValue(value, usingStatement.SubExpression); - cmdlet.ThrowIfUsingValueIsScriptBlock(value); - } - - usingParams.Add(key, value); - } - - return usingParams; - } - - private static object? ExtractUsingExpressionValue( - object? value, - ExpressionAst ast) - { - VariableExpressionAst usingVariable = (VariableExpressionAst)ast - .Find(a => a is VariableExpressionAst, false); - - ExpressionAst lookupAst = new ConstantExpressionAst(ast.Extent, value); - Ast? currentAst = usingVariable; - - while ((currentAst = currentAst?.Parent) is not null) - { - switch (currentAst) - { - case IndexExpressionAst indexAst: - lookupAst = new IndexExpressionAst( - extent: indexAst.Extent, - target: lookupAst, - index: (ExpressionAst)indexAst.Index.Copy()); - currentAst = indexAst; - break; - - case MemberExpressionAst memberAst: - lookupAst = new MemberExpressionAst( - extent: memberAst.Extent, - expression: lookupAst, - member: (ExpressionAst)memberAst.Member.Copy(), - memberAst.Static); - currentAst = memberAst; - break; - - default: - goto CreateAst; - } - } - - CreateAst: - ScriptBlockAst extractionAst = new( - extent: ast.Extent, - paramBlock: null, - statements: new StatementBlockAst( - extent: ast.Extent, - statements: [ - new PipelineAst( - extent: ast.Extent, - pipelineElements: [ - new CommandExpressionAst( - extent: ast.Extent, - expression: lookupAst, - redirections: null) - ]) - ], - traps: null), - isFilter: false); - - return extractionAst - .GetScriptBlock() - .InvokeReturnAsIs(); - } - - internal static Task InvokePowerShellAsync( - this PowerShell powerShell, - PSDataCollection output) - => Task.Factory.FromAsync( - powerShell.BeginInvoke(null, output), - powerShell.EndInvoke); - - internal static ConfiguredTaskAwaitable NoContext(this Task task) => task.ConfigureAwait(false); - - internal static ConfiguredTaskAwaitable NoContext(this Task task) => task.ConfigureAwait(false); - - public static IEnumerable DistinctBy( - this IEnumerable source, - Func keySelector) - { - HashSet seenKeys = []; - foreach (TSource element in source) - { - if (seenKeys.Add(keySelector(element))) - { - yield return element; - } - } - } -} diff --git a/src/PSParallelPipeline/MiscExtensions.cs b/src/PSParallelPipeline/MiscExtensions.cs new file mode 100644 index 0000000..44feedc --- /dev/null +++ b/src/PSParallelPipeline/MiscExtensions.cs @@ -0,0 +1,298 @@ +using System; +using System.Collections; +using System.Collections.Generic; +using System.Linq; +using System.Management.Automation; +using System.Management.Automation.Language; +using System.Management.Automation.Runspaces; +using System.Runtime.CompilerServices; +using System.Text; +using System.Threading.Tasks; + +namespace PSParallelPipeline; + +internal static class MiscExtensions +{ + extension(InitialSessionState initialSessionState) + { + internal InitialSessionState AddFunctions( + string[]? functionsToAdd, + PSCmdlet cmdlet) + { + if (functionsToAdd is not null) + { + foreach (string function in functionsToAdd) + { + IEnumerable commands = cmdlet + .InvokeCommand + .GetCommands( + name: function, + commandTypes: CommandTypes.Function, + nameIsPattern: WildcardPattern.ContainsWildcardCharacters(function)); + + bool addedOne = false; + foreach (CommandInfo command in commands) + { + addedOne = true; + initialSessionState.Commands.Add( + new SessionStateFunctionEntry( + name: command.Name, + definition: command.Definition)); + } + + if (!addedOne) + cmdlet.ThrowFunctionNotFoundError(function); + } + } + + return initialSessionState; + } + + internal InitialSessionState AddVariables( + IDictionary[]? variables, + PSCmdlet cmdlet) + { + if (variables is not null) + { + foreach (IDictionary dict in variables) + { + foreach (DictionaryEntry pair in dict) + { + cmdlet.ThrowIfVariableIsScriptBlock(pair.Value); + initialSessionState.Variables.Add(new SessionStateVariableEntry( + name: LanguagePrimitives.ConvertTo(pair.Key), + value: pair.Value, + description: null)); + } + } + } + + return initialSessionState; + } + + internal InitialSessionState ImportModules( + string[]? modulesToImport) + { + if (modulesToImport is not null) + initialSessionState.ImportPSModule(modulesToImport); + + return initialSessionState; + } + + internal InitialSessionState ImportModulesFromPath( + string[]? modulePaths, + PSCmdlet cmdlet) + { + + if (modulePaths is not null) + { + foreach (string path in modulePaths) + { + string resolved = cmdlet.ResolvePath(path); + initialSessionState.ImportPSModulesFromPath(resolved); + } + } + + return initialSessionState; + } + } + + extension(PSCmdlet cmdlet) + { + private string ResolvePath(string path) + { + string resolved = cmdlet.SessionState.Path.GetUnresolvedProviderPathFromPSPath( + path: path, + provider: out ProviderInfo provider, + drive: out _); + + provider.ThrowIfInvalidProvider(path, cmdlet); + resolved.ThrowIfNotDirectory(cmdlet); + return resolved.TrimEnd('\\', '/'); + } + } + + extension(ScriptBlock script) + { + private static object? ExtractUsingExpressionValue(object? value, ExpressionAst ast) + { + VariableExpressionAst usingVariable = (VariableExpressionAst)ast + .Find(a => a is VariableExpressionAst, false); + + ExpressionAst lookupAst = new ConstantExpressionAst(ast.Extent, value); + Ast? currentAst = usingVariable; + + while ((currentAst = currentAst?.Parent) is not null) + { + switch (currentAst) + { + case IndexExpressionAst indexAst: + lookupAst = new IndexExpressionAst( + extent: indexAst.Extent, + target: lookupAst, + index: (ExpressionAst)indexAst.Index.Copy()); + currentAst = indexAst; + break; + + case MemberExpressionAst memberAst: + lookupAst = new MemberExpressionAst( + extent: memberAst.Extent, + expression: lookupAst, + member: (ExpressionAst)memberAst.Member.Copy(), + memberAst.Static); + currentAst = memberAst; + break; + + default: + goto CreateAst; + } + } + + CreateAst: + ScriptBlockAst extractionAst = new( + extent: ast.Extent, + paramBlock: null, + statements: new StatementBlockAst( + extent: ast.Extent, + statements: [ + new PipelineAst( + extent: ast.Extent, + pipelineElements: [ + new CommandExpressionAst( + extent: ast.Extent, + expression: lookupAst, + redirections: null) + ]) + ], + traps: null), + isFilter: false); + + return extractionAst + .GetScriptBlock() + .InvokeReturnAsIs(); + } + + internal Dictionary GetUsingParameters(PSCmdlet cmdlet) + { + Dictionary usingParams = []; + IEnumerable usingExpressionAsts = script.Ast + .FindAll(a => a is UsingExpressionAst, true) + .Cast(); + + foreach (UsingExpressionAst usingStatement in usingExpressionAsts) + { + VariableExpressionAst backingVariableAst = UsingExpressionAst + .ExtractUsingVariable(usingStatement); + + string varPath = backingVariableAst.VariablePath.UserPath; + + string varText = usingStatement.ToString(); + if (usingStatement.SubExpression is VariableExpressionAst) + { + varText = varText.ToLowerInvariant(); + } + + string key = Convert.ToBase64String(Encoding.Unicode.GetBytes(varText)); + object? value = cmdlet.GetVariableValue(varPath); + cmdlet.ThrowIfUsingValueIsScriptBlock(value); + + if (usingParams.ContainsKey(key)) + { + continue; + } + + if (usingStatement.SubExpression is MemberExpressionAst or IndexExpressionAst) + { + value = ExtractUsingExpressionValue(value, usingStatement.SubExpression); + cmdlet.ThrowIfUsingValueIsScriptBlock(value); + } + + usingParams.Add(key, value); + } + + return usingParams; + } + } + + + + extension(PowerShell powershell) + { + internal Task InvokeAsync(PSDataCollection output) + => Task.Factory.FromAsync( + powershell.BeginInvoke(null, output), + powershell.EndInvoke); + + internal PowerShell AddInput(object? inputObject) + { + const string SetVariableCommand = "Set-Variable"; + const string DollarUnderbar = "_"; + + if (inputObject is not null) + powershell + .AddCommand(SetVariableCommand, useLocalScope: true) + .AddArgument(DollarUnderbar) + .AddArgument(inputObject); + + return powershell; + } + + internal PowerShell AddScript(TaskSettings settings) + { + powershell.AddScript(settings.Script, useLocalScope: true); + return powershell; + } + + internal PowerShell AddUsingStatements(TaskSettings settings) + { + const string StopParsingOp = "--%"; + + if (settings.UsingStatements.Count > 0) + powershell.AddParameter(StopParsingOp, settings.UsingStatements); + + return powershell; + } + + internal PowerShell WithStreams(PSOutputStreams outputStreams) + { + PSDataStreams streams = powershell.Streams; + streams.Error = outputStreams.Error; + streams.Debug = outputStreams.Debug; + streams.Information = outputStreams.Information; + streams.Progress = outputStreams.Progress; + streams.Verbose = outputStreams.Verbose; + streams.Warning = outputStreams.Warning; + return powershell; + } + } + + extension(Task task) + { + internal ConfiguredTaskAwaitable NoContext() => task.ConfigureAwait(false); + } + + extension(Task task) + { + internal ConfiguredTaskAwaitable NoContext() => task.ConfigureAwait(false); + } + + extension(IEnumerable source) + { + public IEnumerable DistinctBy(Func keySelector) + { + HashSet seenKeys = []; + foreach (TSource element in source) + { + if (seenKeys.Add(keySelector(element))) + yield return element; + } + } + } + + extension(string x) + { + internal bool Matches(string y) => + x.Equals(y, StringComparison.OrdinalIgnoreCase) + || WildcardPattern.Get(x, WildcardOptions.IgnoreCase).IsMatch(y); + } +} diff --git a/src/PSParallelPipeline/PSTask.cs b/src/PSParallelPipeline/PSTask.cs index 73828c1..bcf5b43 100644 --- a/src/PSParallelPipeline/PSTask.cs +++ b/src/PSParallelPipeline/PSTask.cs @@ -1,135 +1,36 @@ -using System; using System.Threading; using System.Threading.Tasks; -using System.Collections.Generic; using System.Management.Automation; using System.Management.Automation.Runspaces; namespace PSParallelPipeline; -internal sealed class PSTask +internal static class PSTask { - private const string SetVariableCommand = "Set-Variable"; - - private const string DollarUnderbar = "_"; - - private const string StopParsingOp = "--%"; - - private bool _canceled; - - private readonly PowerShell _powershell; - - private readonly PSDataStreams _internalStreams; - - private Runspace? _runspace; - - private readonly PSOutputStreams _outputStreams; - - 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; - } - - internal static PSTask Create( + internal static async Task InvokeAsync( object? input, - RunspacePool runspacePool, - TaskSettings settings) - { - PSTask ps = new(runspacePool); - SetStreams(ps._internalStreams, runspacePool.Streams); - - return ps + Runspace runspace, + PSOutputStreams streams, + TaskSettings settings, + CancellationToken token) + { + using PowerShell powershell = PowerShell + .Create() + .WithStreams(streams) .AddInput(input) - .AddScript(settings.Script) - .AddUsingStatements(settings.UsingStatements); - } - - internal async Task InvokeAsync() - { - try - { - using CancellationTokenRegistration _ = _token.Register(Cancel); - _runspace = await _pool.GetRunspaceAsync().NoContext(); - _powershell.Runspace = _runspace; - await _powershell.InvokePowerShellAsync(_outputStreams.Success).NoContext(); - } - catch (Exception exception) - { - _outputStreams.AddError(exception.CreateProcessingTaskError(this)); - } - finally - { - CompleteTask(); - } - } + .AddScript(settings) + .AddUsingStatements(settings); - private static void SetStreams( - PSDataStreams streams, - PSOutputStreams outputStreams) - { - streams.Error = outputStreams.Error; - streams.Debug = outputStreams.Debug; - streams.Information = outputStreams.Information; - streams.Progress = outputStreams.Progress; - streams.Verbose = outputStreams.Verbose; - streams.Warning = outputStreams.Warning; - } + powershell.Runspace = runspace; - private PSTask AddInput(object? inputObject) - { - if (inputObject is not null) + using CancellationTokenRegistration _ = token.Register(() => { - _powershell - .AddCommand(SetVariableCommand, useLocalScope: true) - .AddArgument(DollarUnderbar) - .AddArgument(inputObject); - } - - return this; - } + powershell.BeginStop(null, null); + runspace.Dispose(); + }); - private PSTask AddScript(string script) - { - _powershell.AddScript(script, useLocalScope: true); - return this; - } - - private PSTask AddUsingStatements(Dictionary usingParams) - { - if (usingParams.Count > 0) - { - _powershell.AddParameter(StopParsingOp, usingParams); - } - - return this; - } - - private void CompleteTask() - { - _powershell.Dispose(); - if (_canceled) - { - _runspace?.Dispose(); - return; - } - - if (_runspace is not null) - { - _pool.PushRunspace(_runspace); - } - } - - internal void Cancel() - { - _powershell.BeginStop(null, null); - _canceled = true; + await powershell + .InvokeAsync(streams.Success) + .NoContext(); } } diff --git a/src/PSParallelPipeline/RunspacePool.cs b/src/PSParallelPipeline/RunspacePool.cs index 6c0c028..1fdea1a 100644 --- a/src/PSParallelPipeline/RunspacePool.cs +++ b/src/PSParallelPipeline/RunspacePool.cs @@ -10,32 +10,36 @@ internal sealed class RunspacePool : IDisposable { private readonly SemaphoreSlim _semaphore; - private readonly PoolSettings _settings; + private readonly InitialSessionState _initialSessionState; private readonly ConcurrentQueue _pool = []; - private bool UseNewRunspace { get => _settings.UseNewRunspace; } + private readonly bool _useNewRunspace; - internal int MaxRunspaces { get => _settings.MaxRunspaces; } + private readonly int _maxRunspaces; - internal CancellationToken Token { get; } + private readonly CancellationToken _token; - internal PSOutputStreams Streams { get; } + private readonly PSOutputStreams _streams; internal RunspacePool( PoolSettings settings, PSOutputStreams streams, CancellationToken token) { - Streams = streams; - Token = token; - _settings = settings; - _semaphore = new SemaphoreSlim(MaxRunspaces, MaxRunspaces); + _streams = streams; + _token = token; + _initialSessionState = settings.InitialSessionState; + _useNewRunspace = settings.UseNewRunspace; + _maxRunspaces = settings.MaxRunspaces; + _semaphore = new SemaphoreSlim(_maxRunspaces, _maxRunspaces); } - internal void PushRunspace(Runspace runspace) + private void PushRunspace(Runspace? runspace) { - if (UseNewRunspace) + if (runspace is null) return; + + if (_useNewRunspace) { runspace.Dispose(); _semaphore.Release(); @@ -48,27 +52,47 @@ internal void PushRunspace(Runspace runspace) private Runspace CreateRunspace() { - Runspace rs = RunspaceFactory.CreateRunspace(_settings.InitialSessionState); + Runspace rs = RunspaceFactory.CreateRunspace(_initialSessionState); rs.Open(); return rs; } private Task CreateRunspaceAsync() => - Task.Run(CreateRunspace, cancellationToken: Token); + Task.Run(CreateRunspace, cancellationToken: _token); - internal async Task GetRunspaceAsync() + private async Task GetRunspaceAsync() { - await _semaphore.WaitAsync(Token).NoContext(); - if (_pool.TryDequeue(out Runspace runspace)) return runspace; + await _semaphore.WaitAsync(_token).NoContext(); + if (_pool.TryDequeue(out Runspace runspace)) + return runspace; + return await CreateRunspaceAsync().NoContext(); } + internal async Task InvokePowerShellAsync(object? input, TaskSettings settings) + { + Runspace? runspace = null; + + try + { + runspace = await GetRunspaceAsync().NoContext(); + await PSTask.InvokeAsync(input, runspace, _streams, settings, _token); + } + catch (Exception exception) + { + _streams.AddError(exception.CreateProcessingTaskError()); + } + finally + { + PushRunspace(runspace); + } + } + + public void Dispose() { foreach (Runspace runspace in _pool) - { runspace.Dispose(); - } _semaphore.Dispose(); GC.SuppressFinalize(this); diff --git a/src/PSParallelPipeline/VariableTransformation.cs b/src/PSParallelPipeline/VariableTransformation.cs new file mode 100644 index 0000000..958c0f9 --- /dev/null +++ b/src/PSParallelPipeline/VariableTransformation.cs @@ -0,0 +1,62 @@ +using System; +using System.Collections; +using System.Collections.Generic; +using System.Linq; +using System.Management.Automation; + +namespace PSParallelPipeline; + +public sealed class VariableTransformation : ArgumentTransformationAttribute +{ + private static readonly HashSet s_defaultVars = [.. GetDefaultVariables()]; + + public override object Transform(EngineIntrinsics engineIntrinsics, object inputData) + { + if (inputData is PSObject pso) inputData = pso.BaseObject; + if (inputData is IDictionary) return inputData; + + PSVariable[] vars = [.. engineIntrinsics.InvokeProvider.ChildItem + .Get("variable:", true) + .Select(pso => pso.BaseObject) + .Cast() + .Where(var => var.Value.IsNotScriptBlock() && !s_defaultVars.Contains(var.Name))]; + + Hashtable parallelVars = []; + foreach (object? input in LanguagePrimitives.ConvertTo(inputData)) + { + if (input is null) + throw new ArgumentNullException(); + + if (LanguagePrimitives.TryConvertTo(input, out Hashtable hash)) + { + foreach (DictionaryEntry entry in hash) + parallelVars[entry.Key] = entry.Value; + + continue; + } + + bool shouldThrow = true; + string inputAsString = LanguagePrimitives.ConvertTo(input); + foreach (PSVariable var in vars) + { + if (inputAsString.Matches(var.Name)) + { + shouldThrow = false; + parallelVars[var.Name] = var.Value; + } + } + + if (shouldThrow) + throw new ItemNotFoundException( + $"Could not find any variable matching the name or pattern '{inputAsString}'."); + } + + return parallelVars; + } + + private static IEnumerable GetDefaultVariables() + { + using PowerShell ps = PowerShell.Create().AddCommand("Get-Variable"); + return ps.Invoke().Select(e => e.Name); + } +} diff --git a/src/PSParallelPipeline/Worker.cs b/src/PSParallelPipeline/Worker.cs index d6f16a8..46af244 100644 --- a/src/PSParallelPipeline/Worker.cs +++ b/src/PSParallelPipeline/Worker.cs @@ -10,27 +10,24 @@ internal sealed class Worker { private readonly Task _worker; + private readonly PoolSettings _poolSettings; + private readonly TaskSettings _taskSettings; private readonly BlockingCollection _input = []; private readonly BlockingCollection _output = []; - private readonly RunspacePool _pool; - private readonly CancellationToken _token; - private readonly PSOutputStreams _streams; - internal Worker( PoolSettings poolSettings, TaskSettings taskSettings, CancellationToken token) { _token = token; + _poolSettings = poolSettings; _taskSettings = taskSettings; - _streams = new PSOutputStreams(_output); - _pool = new RunspacePool(poolSettings, _streams, _token); _worker = Task.Run(Start, cancellationToken: _token); } @@ -46,22 +43,24 @@ internal Worker( private async Task Start() { - List tasks = new(_pool.MaxRunspaces); + int max = _poolSettings.MaxRunspaces; + using PSOutputStreams streams = new(_output); + using RunspacePool pool = new(_poolSettings, streams, _token); + List tasks = new(max); try { + Task task; foreach (object? input in _input.GetConsumingEnumerable(_token)) { - if (tasks.Count == tasks.Capacity) + if (tasks.Count == max) { - Task task = await Task.WhenAny(tasks).NoContext(); + task = await Task.WhenAny(tasks).NoContext(); tasks.Remove(task); await task.NoContext(); } - tasks.Add(PSTask - .Create(input, _pool, _taskSettings) - .InvokeAsync()); + tasks.Add(pool.InvokePowerShellAsync(input, _taskSettings)); } } catch (OperationCanceledException) @@ -69,9 +68,7 @@ private async Task Start() finally { if (tasks.Count > 0) - { await Task.WhenAll(tasks).NoContext(); - } _output.CompleteAdding(); } @@ -79,9 +76,7 @@ private async Task Start() public void Dispose() { - _pool.Dispose(); _input.Dispose(); - _streams.Dispose(); _output.Dispose(); GC.SuppressFinalize(this); } diff --git a/tests/PSParallelPipeline.tests.ps1 b/tests/PSParallelPipeline.tests.ps1 index 9ca1ec7..4247b9c 100644 --- a/tests/PSParallelPipeline.tests.ps1 +++ b/tests/PSParallelPipeline.tests.ps1 @@ -103,6 +103,10 @@ Describe PSParallelPipeline { } Context 'Variables Parameter' { + BeforeAll { + $foo, $bar, $baz = 1..3 + } + It 'Makes variables available in the parallel scope' { $items = 0..10 | Invoke-Parallel { $message -f $_ } -Variables @{ message = 'Hello world from {0:D2}' @@ -111,6 +115,43 @@ Describe PSParallelPipeline { $shouldBe = 0..10 | ForEach-Object { 'Hello world from {0:D2}' -f $_ } $items | Should -BeExactly $shouldBe } + + It 'Supports input strings' { + $null | Invoke-Parallel { $foo, $bar, $baz } -Variables foo, bar, baz | + Should -BeExactly 1, 2, 3 + } + + It 'Supports wildcard strings' { + $null | Invoke-Parallel { $foo, $bar, $baz } -Variables foo, b* | + Should -BeExactly 1, 2, 3 + + $null | Invoke-Parallel { $foo, $bar, $baz } -Variables * | + Should -BeExactly 1, 2, 3 + } + + It 'Supports multiple hashtables' { + $null | Invoke-Parallel { $foo, $bar, $baz } -Variables foo, @{ bar = 2 }, @{ baz = 3 } | + Should -BeExactly 1, 2, 3 + } + + It 'Strips PSObject wrapper' { + $vars = Write-Output foo, bar, baz + $null | Invoke-Parallel { $foo, $bar, $baz } -Variables $vars | + Should -BeExactly 1, 2, 3 + } + + It 'Should throw if a variable could not be found' { + { $null | Invoke-Parallel { } -Variable xyz } | + Should -Throw -ExceptionType ([ParameterBindingException]) + + { $null | Invoke-Parallel { } -Variable xyz* } | + Should -Throw -ExceptionType ([ParameterBindingException]) + } + + It 'Should throw if null value' { + { $null | Invoke-Parallel { } -Variable foo, $null } | + Should -Throw -ExceptionType ([ParameterBindingException]) + } } Context 'Functions Parameter' { @@ -124,6 +165,16 @@ Describe PSParallelPipeline { { Invoke-Parallel -Functions Test-NotExist { } } | Should -Throw -ExceptionType ([CommandNotFoundException]) } + + It 'Supports wildcard strings' { + 0..10 | Invoke-Parallel { Test-Function $_ } -Functions Test-F* | + Sort-Object | + Should -BeExactly @(0..10 | ForEach-Object { Test-Function $_ }) + + 0..10 | Invoke-Parallel { Test-Function $_ } -Functions * | + Sort-Object | + Should -BeExactly @(0..10 | ForEach-Object { Test-Function $_ }) + } } Context 'ThrottleLimit Parameter' { @@ -238,12 +289,15 @@ Describe PSParallelPipeline { $shouldBe = 0..10 | ForEach-Object { 'Hello world from {0:D2}' -f $_ } $items | Should -BeExactly $shouldBe + + $foo = 1 + $null | Invoke-Parallel { $using:foo; $using:foo } | Should -BeExactly 1, 1 } It 'Allows indexing on $using: passed-in variables' { $arr = 0..10; $hash = @{ foo = 'bar' } - 1 | Invoke-Parallel { $using:arr[-1] } | Should -BeExactly 10 - 1 | Invoke-Parallel { $using:hash['FOO'] } | Should -BeExactly 'bar' + 1 | Invoke-Parallel { $using:arr[0]; $using:arr[-1] } | Should -BeExactly 0, 10 + 1 | Invoke-Parallel { $using:hash['Foo']; $using:hash['FOO'] } | Should -BeExactly bar, bar } It 'Allows member access on $using: passed-in variables' { diff --git a/tools/requiredModules.psd1 b/tools/requiredModules.psd1 index ea1ae51..fb04cda 100644 --- a/tools/requiredModules.psd1 +++ b/tools/requiredModules.psd1 @@ -1,6 +1,5 @@ @{ - InvokeBuild = '5.12.2' + InvokeBuild = '5.14.23' platyPS = '0.14.2' - PSScriptAnalyzer = '1.24.0' - Pester = '5.7.1' + Pester = '6.0.1' }