PSTask.cs

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.Management.Automation.Host;
using System.Management.Automation.Language;
using System.Management.Automation.Remoting.Internal;
using System.Management.Automation.Runspaces;
using System.Management.Automation.Security;
using System.Threading;

using Dbg = System.Management.Automation.Diagnostics;

namespace System.Management.Automation.PSTasks
{
    #region PSTask

    /// <summary>
    /// Class to encapsulate synchronous running scripts in parallel.
    /// </summary>
    internal sealed class PSTask : PSTaskBase
    {
        #region Members

        private readonly PSTaskDataStreamWriter _dataStreamWriter;

        #endregion

        #region Constructor

        /// <summary>
        /// Initializes a new instance of the <see cref="PSTask"/> class.
        /// </summary>
        /// <param name="scriptBlock">Script block to run in task.</param>
        /// <param name="usingValuesMap">Using values passed into script block.</param>
        /// <param name="dollarUnderbar">Dollar underbar variable value.</param>
        /// <param name="currentLocationPath">Current working directory.</param>
        /// <param name="dataStreamWriter">Cmdlet data stream writer.</param>
        public PSTask(
            ScriptBlock scriptBlock,
            Dictionary<string, object> usingValuesMap,
            object dollarUnderbar,
            string currentLocationPath,
            PSTaskDataStreamWriter dataStreamWriter)
            : base(
                scriptBlock,
                usingValuesMap,
                dollarUnderbar,
                currentLocationPath)
        {
            _dataStreamWriter = dataStreamWriter;
        }

        #endregion

        #region Overrides

        /// <summary>
        /// Initialize PowerShell object.
        /// </summary>
        protected override void InitializePowershell()
        {
            // Writer data stream handlers
            _output.DataAdded += (sender, args) => HandleOutputData();
            _powershell.Streams.Error.DataAdded += (sender, args) => HandleErrorData();
            _powershell.Streams.Warning.DataAdded += (sender, args) => HandleWarningData();
            _powershell.Streams.Verbose.DataAdded += (sender, args) => HandleVerboseData();
            _powershell.Streams.Debug.DataAdded += (sender, args) => HandleDebugData();
            _powershell.Streams.Progress.DataAdded += (sender, args) => HandleProgressData();
            _powershell.Streams.Information.DataAdded += (sender, args) => HandleInformationData();

            // State change handler
            _powershell.InvocationStateChanged += (sender, args) => HandleStateChanged(args);
        }

        #endregion

        #region Writer data stream handlers

        private void HandleOutputData()
        {
            foreach (var item in _output.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Output, item));
            }
        }

        private void HandleErrorData()
        {
            foreach (var item in _powershell.Streams.Error.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Error, item));
            }
        }

        private void HandleWarningData()
        {
            foreach (var item in _powershell.Streams.Warning.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Warning, item.Message));
            }
        }

        private void HandleVerboseData()
        {
            foreach (var item in _powershell.Streams.Verbose.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Verbose, item.Message));
            }
        }

        private void HandleDebugData()
        {
            foreach (var item in _powershell.Streams.Debug.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Debug, item.Message));
            }
        }

        private void HandleInformationData()
        {
            foreach (var item in _powershell.Streams.Information.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Information, item));
            }
        }

        private void HandleProgressData()
        {
            foreach (var item in _powershell.Streams.Progress.ReadAll())
            {
                _dataStreamWriter.Add(
                    new PSStreamObject(PSStreamObjectType.Progress, item));
            }
        }

        #endregion

        #region Event handlers

        private void HandleStateChanged(PSInvocationStateChangedEventArgs stateChangeInfo)
        {
            if (_dataStreamWriter != null)
            {
                // Treat any terminating exception as a non-terminating error record
                var newStateInfo = stateChangeInfo.InvocationStateInfo;
                if (newStateInfo.Reason != null)
                {
                    var errorRecord = new ErrorRecord(
                        newStateInfo.Reason,
                        "PSTaskException",
                        ErrorCategory.InvalidOperation,
                        this);

                    _dataStreamWriter.Add(
                        new PSStreamObject(PSStreamObjectType.Error, errorRecord));
                }
            }

            RaiseStateChangedEvent(stateChangeInfo);
        }

        #endregion
    }

    /// <summary>
    /// Class to encapsulate asynchronous running scripts in parallel as jobs.
    /// </summary>
    internal sealed class PSJobTask : PSTaskBase
    {
        #region Members

        private readonly Job _job;

        #endregion

        #region Constructor

        /// <summary>
        /// Initializes a new instance of the <see cref="PSJobTask"/> class.
        /// </summary>
        /// <param name="scriptBlock">Script block to run.</param>
        /// <param name="usingValuesMap">Using variable values passed to script block.</param>
        /// <param name="dollarUnderbar">Dollar underbar variable value for script block.</param>
        /// <param name="currentLocationPath">Current working directory.</param>
        /// <param name="job">Job object associated with task.</param>
        public PSJobTask(
            ScriptBlock scriptBlock,
            Dictionary<string, object> usingValuesMap,
            object dollarUnderbar,
            string currentLocationPath,
            Job job) : base(
                scriptBlock,
                usingValuesMap,
                dollarUnderbar,
                currentLocationPath)
        {
            _job = job;
        }

        #endregion

        #region Overrides

        /// <summary>
        /// Initialize PowerShell object.
        /// </summary>
        protected override void InitializePowershell()
        {
            // Job data stream handlers
            _output.DataAdded += (sender, args) => HandleJobOutputData();
            _powershell.Streams.Error.DataAdded += (sender, args) => HandleJobErrorData();
            _powershell.Streams.Warning.DataAdded += (sender, args) => HandleJobWarningData();
            _powershell.Streams.Verbose.DataAdded += (sender, args) => HandleJobVerboseData();
            _powershell.Streams.Debug.DataAdded += (sender, args) => HandleJobDebugData();
            _powershell.Streams.Information.DataAdded += (sender, args) => HandleJobInformationData();

            // State change handler
            _powershell.InvocationStateChanged += (sender, args) => HandleStateChanged(args);
        }

        #endregion

        #region Job data stream handlers

        private void HandleJobOutputData()
        {
            foreach (var item in _output.ReadAll())
            {
                _job.Output.Add(item);
                _job.Results.Add(
                    new PSStreamObject(PSStreamObjectType.Output, item));
            }
        }

        private void HandleJobErrorData()
        {
            foreach (var item in _powershell.Streams.Error.ReadAll())
            {
                _job.Error.Add(item);
                _job.Results.Add(
                    new PSStreamObject(PSStreamObjectType.Error, item));
            }
        }

        private void HandleJobWarningData()
        {
            foreach (var item in _powershell.Streams.Warning.ReadAll())
            {
                _job.Warning.Add(item);
                _job.Results.Add(
                    new PSStreamObject(PSStreamObjectType.Warning, item.Message));
            }
        }

        private void HandleJobVerboseData()
        {
            foreach (var item in _powershell.Streams.Verbose.ReadAll())
            {
                _job.Verbose.Add(item);
                _job.Results.Add(
                    new PSStreamObject(PSStreamObjectType.Verbose, item.Message));
            }
        }

        private void HandleJobDebugData()
        {
            foreach (var item in _powershell.Streams.Debug.ReadAll())
            {
                _job.Debug.Add(item);
                _job.Results.Add(
                    new PSStreamObject(PSStreamObjectType.Debug, item.Message));
            }
        }

        private void HandleJobInformationData()
        {
            foreach (var item in _powershell.Streams.Information.ReadAll())
            {
                _job.Information.Add(item);
                _job.Results.Add(
                    new PSStreamObject(PSStreamObjectType.Information, item));
            }
        }

        #endregion

        #region Event handlers

        private void HandleStateChanged(PSInvocationStateChangedEventArgs stateChangeInfo)
        {
            RaiseStateChangedEvent(stateChangeInfo);
        }

        #endregion

        #region Properties

        /// <summary>
        /// Gets Debugger.
        /// </summary>
        public Debugger Debugger
        {
            get => _powershell.Runspace.Debugger;
        }

        #endregion
    }

    /// <summary>
    /// Base class to encapsulate running a PowerShell script concurrently in a cmdlet or job context.
    /// </summary>
    internal abstract class PSTaskBase : IDisposable
    {
        #region Members

        private readonly ScriptBlock _scriptBlockToRun;
        private readonly Dictionary<string, object> _usingValuesMap;
        private readonly object _dollarUnderbar;
        private readonly int _id;
        private readonly string _currentLocationPath;
        private Runspace _runspace;
        protected PowerShell _powershell;
        protected PSDataCollection<PSObject> _output;

        public const string RunspaceName = "PSTask";

        private static int s_taskId;

        #endregion

        #region Events

        /// <summary>
        /// Event that fires when the task running state changes.
        /// </summary>
        public event EventHandler<PSInvocationStateChangedEventArgs> StateChanged;

        internal void RaiseStateChangedEvent(PSInvocationStateChangedEventArgs args)
        {
            StateChanged.SafeInvoke(this, args);
        }

        #endregion

        #region Properties

        /// <summary>
        /// Gets current running state of the task.
        /// </summary>
        public PSInvocationState State
        {
            get
            {
                PowerShell ps = _powershell;
                if (ps != null)
                {
                    return ps.InvocationStateInfo.State;
                }

                return PSInvocationState.NotStarted;
            }
        }

        /// <summary>
        /// Gets Task Id.
        /// </summary>
        public int Id { get => _id; }

        /// <summary>
        /// Gets Task Runspace.
        /// </summary>
        public Runspace Runspace { get => _runspace; }

        #endregion

        #region Constructor

        private PSTaskBase()
        {
            _id = Interlocked.Increment(ref s_taskId);
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="PSTaskBase"/> class.
        /// </summary>
        /// <param name="scriptBlock">Script block to run.</param>
        /// <param name="usingValuesMap">Using variable values passed to script block.</param>
        /// <param name="dollarUnderbar">Dollar underbar variable value.</param>
        /// <param name="currentLocationPath">Current working directory.</param>
        protected PSTaskBase(
            ScriptBlock scriptBlock,
            Dictionary<string, object> usingValuesMap,
            object dollarUnderbar,
            string currentLocationPath) : this()
        {
            _scriptBlockToRun = scriptBlock;
            _usingValuesMap = usingValuesMap;
            _dollarUnderbar = dollarUnderbar;
            _currentLocationPath = currentLocationPath;
        }

        #endregion

        #region Abstract Methods

        /// <summary>
        /// Initialize PowerShell object.
        /// </summary>
        protected abstract void InitializePowershell();

        #endregion

        #region IDisposable

        /// <summary>
        /// Dispose PSTaskBase instance.
        /// </summary>
        public void Dispose()
        {
            _powershell.Dispose();
            _output.Dispose();
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Start task.
        /// </summary>
        /// <param name="runspace">Runspace used to run task.</param>
        public void Start(Runspace runspace)
        {
            if (_powershell != null)
            {
                Dbg.Assert(false, "A PSTask can be started only once.");
                return;
            }

            Dbg.Assert(runspace != null, "Task runspace cannot be null.");
            _runspace = runspace;

            // If available, set current working directory on the runspace.
            // Temporarily set the newly created runspace as the thread default runspace for any needed module loading.
            if (_currentLocationPath != null)
            {
                var oldDefaultRunspace = Runspace.DefaultRunspace;
                try
                {
                    Runspace.DefaultRunspace = runspace;
                    var context = new CmdletProviderContext(runspace.ExecutionContext)
                    {
                        // _currentLocationPath denotes the current path as-is, and should not be attempted expanded.
                        SuppressWildcardExpansion = true
                    };
                    runspace.ExecutionContext.SessionState.Internal.SetLocation(_currentLocationPath, context);
                }
                catch (DriveNotFoundException)
                {
                    // Allow task to run if current drive is not available.
                }
                finally
                {
                    Runspace.DefaultRunspace = oldDefaultRunspace;
                }
            }

            // Create the PowerShell command pipeline for the provided script block
            // The script will run on the provided Runspace in a new thread by default
            _powershell = PowerShell.Create(runspace);

            // Initialize PowerShell object data streams and event handlers
            _output = new PSDataCollection<PSObject>();
            InitializePowershell();

            // Start the script running in a new thread
            _powershell.AddScript(_scriptBlockToRun.ToString());
            _powershell.Commands.Commands[0].DollarUnderbar = _dollarUnderbar;
            if (_usingValuesMap != null && _usingValuesMap.Count > 0)
            {
                _powershell.AddParameter(Parser.VERBATIM_ARGUMENT, _usingValuesMap);
            }

            _powershell.BeginInvoke<object, PSObject>(input: null, output: _output);
        }

        /// <summary>
        /// Signals the running task to stop.
        /// </summary>
        public void SignalStop() => _powershell?.BeginStop(null, null);

        #endregion
    }

    #endregion

    #region PSTaskDataStreamWriter

    /// <summary>
    /// Class that handles writing task data stream objects to a cmdlet.
    /// </summary>
    internal sealed class PSTaskDataStreamWriter : IDisposable
    {
        #region Members

        private readonly PSCmdlet _cmdlet;
        private readonly PSDataCollection<PSStreamObject> _dataStream;
        private readonly int _cmdletThreadId;

        #endregion

        #region Properties

        /// <summary>
        /// Gets wait-able handle that signals when new data has been added to
        /// the data stream collection.
        /// </summary>
        /// <returns>Data added wait handle.</returns>
        internal WaitHandle DataAddedWaitHandle
        {
            get => _dataStream.WaitHandle;
        }

        #endregion

        #region Constructor

        private PSTaskDataStreamWriter() { }

        /// <summary>
        /// Initializes a new instance of the <see cref="PSTaskDataStreamWriter"/> class.
        /// </summary>
        /// <param name="psCmdlet">Parent cmdlet.</param>
        public PSTaskDataStreamWriter(PSCmdlet psCmdlet)
        {
            _cmdlet = psCmdlet;
            _cmdletThreadId = Environment.CurrentManagedThreadId;
            _dataStream = new PSDataCollection<PSStreamObject>();
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Add data stream object to the writer.
        /// </summary>
        /// <param name="streamObject">Data stream object to write.</param>
        public void Add(PSStreamObject streamObject)
        {
            _dataStream.Add(streamObject);
        }

        /// <summary>
        /// Write all objects in data stream collection to the cmdlet data stream.
        /// </summary>
        public void WriteImmediate()
        {
            CheckCmdletThread();

            foreach (var item in _dataStream.ReadAll())
            {
                item.WriteStreamObject(cmdlet: _cmdlet, overrideInquire: true);
            }
        }

        /// <summary>
        /// Waits for data stream objects to be added to the collection, and writes them
        /// to the cmdlet data stream.
        /// This method returns only after the writer has been closed.
        /// </summary>
        public void WaitAndWrite()
        {
            CheckCmdletThread();

            while (true)
            {
                _dataStream.WaitHandle.WaitOne();
                WriteImmediate();

                if (!_dataStream.IsOpen)
                {
                    WriteImmediate();
                    break;
                }
            }
        }

        /// <summary>
        /// Closes the stream writer.
        /// </summary>
        public void Close()
        {
            _dataStream.Complete();
        }

        #endregion

        #region Private Methods

        private void CheckCmdletThread()
        {
            if (Environment.CurrentManagedThreadId != _cmdletThreadId)
            {
                throw new PSInvalidOperationException(InternalCommandStrings.PSTaskStreamWriterWrongThread);
            }
        }

        #endregion

        #region IDisposable

        /// <summary>
        /// Dispose the stream writer.
        /// </summary>
        public void Dispose()
        {
            _dataStream.Dispose();
        }

        #endregion
    }

    #endregion

    #region PSTaskPool

    /// <summary>
    /// Pool for running PSTasks, with limit of total number of running tasks at a time.
    /// </summary>
    internal sealed class PSTaskPool : IDisposable
    {
        #region Members

        private readonly ManualResetEvent _addAvailable;
        private readonly int _sizeLimit;
        private readonly ManualResetEvent _stopAll;
        private readonly object _syncObject;
        private readonly Dictionary<int, PSTaskBase> _taskPool;
        private readonly ConcurrentQueue<Runspace> _runspacePool;
        private readonly ConcurrentDictionary<int, Runspace> _activeRunspaces;
        private readonly WaitHandle[] _waitHandles;
        private readonly bool _useRunspacePool;
        private bool _isOpen;
        private bool _stopping;
        private int _createdRunspaceCount;

        private const int AddAvailable = 0;
        private const int Stop = 1;

        #endregion

        #region Constructor

        private PSTaskPool() { }

        /// <summary>
        /// Initializes a new instance of the <see cref="PSTaskPool"/> class.
        /// </summary>
        /// <param name="size">Total number of allowed running objects in pool at one time.</param>
        /// <param name="useNewRunspace">When true, a new runspace object is created for the task instead of reusing one from the pool.</param>
        public PSTaskPool(
            int size,
            bool useNewRunspace)
        {
            _sizeLimit = size;
            _useRunspacePool = !useNewRunspace;
            _isOpen = true;
            _syncObject = new object();
            _addAvailable = new ManualResetEvent(true);
            _stopAll = new ManualResetEvent(false);
            _waitHandles = new WaitHandle[]
            {
                _addAvailable,      // index 0
                _stopAll,           // index 1
            };
            _taskPool = new Dictionary<int, PSTaskBase>(size);
            _activeRunspaces = new ConcurrentDictionary<int, Runspace>();
            if (_useRunspacePool)
            {
                _runspacePool = new ConcurrentQueue<Runspace>();
            }
        }

        #endregion

        #region Events

        /// <summary>
        /// Event that fires when pool is closed and drained of all tasks.
        /// </summary>
        public event EventHandler<EventArgs> PoolComplete;

        #endregion

        #region Properties

        /// <summary>
        /// Gets a value indicating whether a pool is currently open for accepting tasks.
        /// </summary>
        public bool IsOpen
        {
            get => _isOpen;
        }

        /// <summary>
        /// Gets a value of the count of total runspaces allocated.
        /// </summary>
        public int AllocatedRunspaceCount
        {
            get => _createdRunspaceCount;
        }

        #endregion

        #region IDisposable

        /// <summary>
        /// Dispose task pool.
        /// </summary>
        public void Dispose()
        {
            _addAvailable.Dispose();
            _stopAll.Dispose();

            DisposeRunspaces();
        }

        /// <summary>
        /// Dispose runspaces.
        /// </summary>
        internal void DisposeRunspaces()
        {
            foreach (var item in _activeRunspaces)
            {
                item.Value.Dispose();
            }

            _activeRunspaces.Clear();
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Method to add a task to the pool.
        /// If the pool is full, then this method blocks until space is available.
        /// This method is not multi-thread safe and assumes only one thread waits and adds tasks.
        /// </summary>
        /// <param name="task">Task to be added to pool.</param>
        /// <returns>True when task is successfully added.</returns>
        public bool Add(PSTaskBase task)
        {
            if (!_isOpen)
            {
                return false;
            }

            // Block until either space is available, or a stop is commanded
            var index = WaitHandle.WaitAny(_waitHandles);

            switch (index)
            {
                case AddAvailable:
                    var runspace = GetRunspace(task.Id);
                    task.StateChanged += HandleTaskStateChangedDelegate;
                    lock (_syncObject)
                    {
                        if (!_isOpen)
                        {
                            return false;
                        }

                        _taskPool.Add(task.Id, task);
                        if (_taskPool.Count == _sizeLimit)
                        {
                            _addAvailable.Reset();
                        }

                        task.Start(runspace);
                    }

                    return true;

                case Stop:
                    return false;

                default:
                    return false;
            }
        }

        /// <summary>
        /// Add child job task to task pool.
        /// </summary>
        /// <param name="childJob">Child job to be added to pool.</param>
        /// <returns>True when child job is successfully added.</returns>
        public bool Add(PSTaskChildJob childJob)
        {
            return Add(childJob.Task);
        }

        /// <summary>
        /// Signals all running tasks to stop and closes pool for any new tasks.
        /// </summary>
        public void StopAll()
        {
            _stopping = true;

            // Accept no more input
            Close();
            _stopAll.Set();

            // Stop all running tasks
            PSTaskBase[] tasksToStop;
            lock (_syncObject)
            {
                tasksToStop = new PSTaskBase[_taskPool.Values.Count];
                _taskPool.Values.CopyTo(tasksToStop, 0);
            }

            foreach (var task in tasksToStop)
            {
                task.Dispose();
            }

            // Dispose all active runspaces
            DisposeRunspaces();
            _stopping = false;
        }

        /// <summary>
        /// Closes the pool and prevents any new tasks from being added.
        /// </summary>
        public void Close()
        {
            _isOpen = false;
            CheckForComplete();
        }

        #endregion

        #region Private Methods

        private void HandleTaskStateChangedDelegate(object sender, PSInvocationStateChangedEventArgs args) => HandleTaskStateChanged(sender, args);

        private void HandleTaskStateChanged(object sender, PSInvocationStateChangedEventArgs args)
        {
            var task = sender as PSTaskBase;
            Dbg.Assert(task != null, "State changed sender must always be PSTaskBase");
            var stateInfo = args.InvocationStateInfo;
            switch (stateInfo.State)
            {
                // Look for completed state and remove
                case PSInvocationState.Completed:
                case PSInvocationState.Stopped:
                case PSInvocationState.Failed:
                    ReturnRunspace(task);
                    lock (_syncObject)
                    {
                        _taskPool.Remove(task.Id);
                        if (_taskPool.Count == (_sizeLimit - 1))
                        {
                            _addAvailable.Set();
                        }
                    }

                    task.StateChanged -= HandleTaskStateChangedDelegate;
                    if (!_stopping || stateInfo.State != PSInvocationState.Stopped)
                    {
                        // StopAll disposes tasks.
                        task.Dispose();
                    }

                    CheckForComplete();
                    break;
            }
        }

        private void CheckForComplete()
        {
            bool isTaskPoolComplete;
            lock (_syncObject)
            {
                isTaskPoolComplete = !_isOpen && _taskPool.Count == 0;
            }

            if (isTaskPoolComplete)
            {
                try
                {
                    PoolComplete.SafeInvoke(
                        this,
                        new EventArgs());
                }
                catch
                {
                    Dbg.Assert(false, "Exceptions should not be thrown on event thread");
                }
            }
        }

        private Runspace GetRunspace(int taskId)
        {
            var runspaceName = string.Create(CultureInfo.InvariantCulture, $"{PSTask.RunspaceName}:{taskId}");

            if (_useRunspacePool && _runspacePool.TryDequeue(out Runspace runspace))
            {
                if (runspace.RunspaceStateInfo.State == RunspaceState.Opened &&
                    runspace.RunspaceAvailability == RunspaceAvailability.Available)
                {
                    try
                    {
                        runspace.ResetRunspaceState();
                        runspace.Name = runspaceName;
                        return runspace;
                    }
                    catch
                    {
                        // If the runspace cannot be reset for any reason, remove it.
                    }
                }

                RemoveActiveRunspace(runspace);
            }

            // Create and initialize a new Runspace
            var iss = InitialSessionState.CreateDefault2();
            switch (SystemPolicy.GetSystemLockdownPolicy())
            {
                case SystemEnforcementMode.Enforce:
                    iss.LanguageMode = PSLanguageMode.ConstrainedLanguage;
                    break;

                case SystemEnforcementMode.Audit:
                    // In audit mode, CL restrictions are not enforced and instead audit
                    // log entries are created.
                    iss.LanguageMode = PSLanguageMode.ConstrainedLanguage;
                    break;

                case SystemEnforcementMode.None:
                    iss.LanguageMode = PSLanguageMode.FullLanguage;
                    break;
            }

            runspace = RunspaceFactory.CreateRunspace(iss);
            runspace.Name = runspaceName;
            _activeRunspaces.TryAdd(runspace.Id, runspace);
            runspace.Open();
            _createdRunspaceCount++;

            return runspace;
        }

        private void ReturnRunspace(PSTaskBase task)
        {
            var runspace = task.Runspace;
            Dbg.Assert(runspace != null, "Task runspace cannot be null.");
            if (_useRunspacePool &&
                runspace.RunspaceStateInfo.State == RunspaceState.Opened &&
                runspace.RunspaceAvailability == RunspaceAvailability.Available)
            {
                _runspacePool.Enqueue(runspace);
                return;
            }

            RemoveActiveRunspace(runspace);
        }

        private void RemoveActiveRunspace(Runspace runspace)
        {
            runspace.Dispose();
            _activeRunspaces.TryRemove(runspace.Id, out Runspace _);
        }

        #endregion
    }

    #endregion

    #region PSTaskJobs

    /// <summary>
    /// Job for running ForEach-Object parallel task child jobs asynchronously.
    /// </summary>
    public sealed class PSTaskJob : Job
    {
        #region Members

        private readonly PSTaskPool _taskPool;