diff --git a/CHANGELOG.md b/CHANGELOG.md index 9001cc0c4ec..c255e9233af 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,9 @@ Changelog Unreleased ---------- +- Improve pipeline for native commands. + Start native process in Prepare() instead of Complete(). + `ping | grep` doesn't block anymore. - Added -Top and -Bottom parameters to Sort-Object v6.0.0-alpha.12 - 2016-11-03 diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 9195cb793f9..937134ad567 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -17,6 +17,8 @@ using System.Runtime.Serialization; using System.Globalization; using System.Diagnostics.CodeAnalysis; +using System.Collections.Concurrent; +using System.Collections.Generic; #if CORECLR // Use stubs for SerializableAttribute and ISerializable related types. @@ -306,6 +308,8 @@ internal override void Prepare(IDictionary psDefaultParameterValues) { this.NativeParameterBinderController.BindParameters(arguments); } + + InitNativeProcess(); } /// @@ -315,9 +319,10 @@ internal override void ProcessRecord() { while (Read()) { - // Accumulate everything from the pipe and execute at the end. _inputWriter.Add(Command.CurrentPipelineObject); } + + ConsumeAvailableNativeProcessOutput(blocking: false); } /// @@ -330,17 +335,29 @@ internal override void ProcessRecord() /// private ProcessInputWriter _inputWriter = null; - /// - /// This is used for reading input form the process - /// - private ProcessOutputReader _outputReader = null; - /// /// Is true if this command is to be run "standalone" - that is, with /// no redirection. /// private bool _runStandAlone; + /// + /// Indicate whether we need to consider redirecting the output/error of the current native command. + /// Usually a windows program which is the last command in a pipeline can be executed as 'background' -- we don't need to capture its output/error streams. + /// + private bool _isRunningInBackground; + + /// + /// This output queue helps us keep the output and error (if redirected) order correct. + /// We could do a blocking read in the Complete block instead, + /// but then we would not be able to restore the order reasonable. + /// + private BlockingCollection _nativeProcessOutputQueue; + + private bool _scrapeHostOutput; + + private Host.Coordinates _startPosition; + /// /// object used for synchronization between StopProcessing thread and /// Pipeline thread. @@ -356,12 +373,8 @@ internal override void ProcessRecord() /// /// The native command could not be run /// - internal override void Complete() + private void InitNativeProcess() { - // Indicate whether we need to consider redirecting the output/error of the current native command. - // Usually a windows program which is the last command in a pipeline can be executed as 'background' -- we don't need to capture its output/error streams. - bool background; - // Figure out if we're going to run this process "standalone" i.e. without // redirecting anything. This is a bit tricky as we always run redirected so // we have to see if the redirection is actually being done at the topmost level or not. @@ -384,12 +397,10 @@ internal override void Complete() throw new PipelineStoppedException(); } - // If a problem occurred in running the program, this exception will - // be set and should be rethrown at the end of the try/catch block... - Exception exceptionToRethrow = null; - Host.Coordinates startPosition = new Host.Coordinates(); - bool scrapeHostOutput = false; + _startPosition = new Host.Coordinates(); + _scrapeHostOutput = false; + Exception exceptionToRethrow = null; try { // If this process is being run standalone, tell the host, which might want @@ -405,15 +416,15 @@ internal override void Complete() { if (this.Command.Context.EngineHostInterface.UI.IsTranscribing) { - scrapeHostOutput = true; - startPosition = this.Command.Context.EngineHostInterface.UI.RawUI.CursorPosition; - startPosition.X = 0; + _scrapeHostOutput = true; + _startPosition = this.Command.Context.EngineHostInterface.UI.RawUI.CursorPosition; + _startPosition.X = 0; } } catch (Host.HostException) { // The host doesn't support scraping via its RawUI interface - scrapeHostOutput = false; + _scrapeHostOutput = false; } } @@ -499,14 +510,14 @@ internal override void Complete() // Something like // ls | notepad | sort.exe // should block until the notepad process is terminated. - background = false; + _isRunningInBackground = false; } else { - background = true; + _isRunningInBackground = true; if (startInfo.UseShellExecute == false) { - background = IsWindowsApplication(_nativeProcess.StartInfo.FileName); + _isRunningInBackground = IsWindowsApplication(_nativeProcess.StartInfo.FileName); } } @@ -528,90 +539,181 @@ internal override void Complete() } } } - - if (background == false) - { - //if output is redirected, start reading output of process. - if (startInfo.RedirectStandardOutput || startInfo.RedirectStandardError) - { - lock (_sync) - { - if (!_stopped) - { - _outputReader = new ProcessOutputReader(_nativeProcess, Path, redirectOutput, redirectError); - _outputReader.Start(); - } - } - if (_outputReader != null) - { - ProcessOutputHelper(); - } - } - } } catch (Exception) { StopProcessing(); throw; } - finally + + if (_isRunningInBackground == false) + { + InitOutputQueue(); + } + } + catch (Win32Exception e) + { + exceptionToRethrow = e; + + } // try + catch (PipelineStoppedException) + { + // If we're stopping the process, just rethrow this exception... + throw; + } + catch (Exception e) + { + CommandProcessorBase.CheckForSevereException(e); + + exceptionToRethrow = e; + } + + // An exception was thrown while attempting to run the program + // so wrap and rethrow it here... + if (exceptionToRethrow != null) + { + // It's a system exception so wrap it in one of ours and re-throw. + string message = StringUtil.Format(ParserStrings.ProgramFailedToExecute, + this.NativeCommandName, exceptionToRethrow.Message, + this.Command.MyInvocation.PositionMessage); + ApplicationFailedException appFailedException = new ApplicationFailedException(message, exceptionToRethrow); + + // There is no need to set this exception here since this exception will eventually be caught by pipeline processor. + // this.commandRuntime.PipelineProcessor.ExecutionFailed = true; + + throw appFailedException; + } + } + + private void InitOutputQueue() + { + //if output is redirected, start reading output of process in queue. + if (_nativeProcess.StartInfo.RedirectStandardOutput || _nativeProcess.StartInfo.RedirectStandardError) + { + lock (_sync) + { + if (!_stopped) + { + _nativeProcessOutputQueue = new BlockingCollection(); + // we don't assign the handler to anything, because it's used only for objects marshaling + new ProcessOutputHandler(_nativeProcess, _nativeProcessOutputQueue); + } + } + } + } + + private ProcessOutputObject DequeueProcessOutput(bool blocking) + { + if (blocking) + { + // If adding was completed and collection is empty (IsCompleted == true) + // there is no need to do a blocking Take(), we should just return. + if (!_nativeProcessOutputQueue.IsCompleted) { - if (background == false) + try + { + // If adding is not complete we need a try {} catch {} + // to mitigate a concurrent call to CompleteAdding(). + return _nativeProcessOutputQueue.Take(); + } + catch (InvalidOperationException) { - //Wait for process to exit - _nativeProcess.WaitForExit(); + // It's a normal situation: another thread can mark collection as CompleteAdding + // in a concurrent way and we will rise an exception in Take(). + // Although it's a normal situation it's not the most common path + // and will be executed only on the race condtion case. + } + } - //Wait for input writer to finish. - _inputWriter.Done(); + return null; + } + else + { + ProcessOutputObject record = null; + _nativeProcessOutputQueue.TryTake(out record); + return record; + } + } - //Wait for outputReader to finish - if (_outputReader != null) + /// + /// Read the output from the native process and send it down the line. + /// + private void ConsumeAvailableNativeProcessOutput(bool blocking) + { + if (_isRunningInBackground == false) + { + if (_nativeProcess.StartInfo.RedirectStandardOutput || _nativeProcess.StartInfo.RedirectStandardError) + { + ProcessOutputObject record; + while ((record = DequeueProcessOutput(blocking)) != null) + { + if (this.Command.Context.CurrentPipelineStopping) { - _outputReader.Done(); + this.StopProcessing(); + return; } - // Capture screen output if we are transcribing - if (this.Command.Context.EngineHostInterface.UI.IsTranscribing && - scrapeHostOutput) - { - Host.Coordinates endPosition = this.Command.Context.EngineHostInterface.UI.RawUI.CursorPosition; - endPosition.X = this.Command.Context.EngineHostInterface.UI.RawUI.BufferSize.Width - 1; + ProcessOutputRecord(record); + } + } + } + } - // If the end position is before the start position, then capture the entire buffer. - if (endPosition.Y < startPosition.Y) - { - startPosition.Y = 0; - } + internal override void Complete() + { + Exception exceptionToRethrow = null; + try + { + if (_isRunningInBackground == false) + { + //Wait for input writer to finish. + _inputWriter.Done(); - Host.BufferCell[,] bufferContents = this.Command.Context.EngineHostInterface.UI.RawUI.GetBufferContents( - new Host.Rectangle(startPosition, endPosition)); + // read all the available output in the blocking way + ConsumeAvailableNativeProcessOutput(blocking: true); + _nativeProcess.WaitForExit(); - StringBuilder lineContents = new StringBuilder(); - StringBuilder bufferText = new StringBuilder(); + // Capture screen output if we are transcribing + if (this.Command.Context.EngineHostInterface.UI.IsTranscribing && + _scrapeHostOutput) + { + Host.Coordinates endPosition = this.Command.Context.EngineHostInterface.UI.RawUI.CursorPosition; + endPosition.X = this.Command.Context.EngineHostInterface.UI.RawUI.BufferSize.Width - 1; - for (int row = 0; row < bufferContents.GetLength(0); row++) - { - if (row > 0) - { - bufferText.Append(Environment.NewLine); - } + // If the end position is before the start position, then capture the entire buffer. + if (endPosition.Y < _startPosition.Y) + { + _startPosition.Y = 0; + } - lineContents.Clear(); - for (int column = 0; column < bufferContents.GetLength(1); column++) - { - lineContents.Append(bufferContents[row, column].Character); - } + Host.BufferCell[,] bufferContents = this.Command.Context.EngineHostInterface.UI.RawUI.GetBufferContents( + new Host.Rectangle(_startPosition, endPosition)); + + StringBuilder lineContents = new StringBuilder(); + StringBuilder bufferText = new StringBuilder(); + + for (int row = 0; row < bufferContents.GetLength(0); row++) + { + if (row > 0) + { + bufferText.Append(Environment.NewLine); + } - bufferText.Append(lineContents.ToString().TrimEnd(Utils.Separators.SpaceOrTab)); + lineContents.Clear(); + for (int column = 0; column < bufferContents.GetLength(1); column++) + { + lineContents.Append(bufferContents[row, column].Character); } - this.Command.Context.InternalHost.UI.TranscribeResult(bufferText.ToString()); + bufferText.Append(lineContents.ToString().TrimEnd(Utils.Separators.SpaceOrTab)); } - this.Command.Context.SetVariable(SpecialVariables.LastExitCodeVarPath, _nativeProcess.ExitCode); - if (_nativeProcess.ExitCode != 0) - this.commandRuntime.PipelineProcessor.ExecutionFailed = true; + this.Command.Context.InternalHost.UI.TranscribeResult(bufferText.ToString()); } + + this.Command.Context.SetVariable(SpecialVariables.LastExitCodeVarPath, _nativeProcess.ExitCode); + if (_nativeProcess.ExitCode != 0) + this.commandRuntime.PipelineProcessor.ExecutionFailed = true; } } catch (Win32Exception e) @@ -631,7 +733,7 @@ internal override void Complete() } finally { - if (!redirectOutput) + if (!_nativeProcess.StartInfo.RedirectStandardOutput) { this.Command.Context.EngineHostInterface.NotifyEndApplication(); } @@ -893,12 +995,6 @@ internal void StopProcessing() //Stop input writer _inputWriter.Stop(); - //stop output writer - if (_outputReader != null) - { - _outputReader.Stop(); - } - KillProcess(_nativeProcess); } } @@ -924,84 +1020,67 @@ private void CleanUp() } } - /// - /// This method process the output - /// - private void ProcessOutputHelper() + private void ProcessOutputRecord(ProcessOutputObject outputValue) { - Dbg.Assert(_outputReader != null, "this should be called only when output has been created"); - - object value = _outputReader.Read(); - while (value != AutomationNull.Value) + Dbg.Assert(outputValue != null, "only object of type ProcessOutputObject expected"); + + if (outputValue.Stream == MinishellStream.Error) { - ProcessOutputObject outputValue = value as ProcessOutputObject; - Dbg.Assert(outputValue != null, "only object of type ProcessOutputObject expected"); - - if (outputValue.Stream == MinishellStream.Error) - { - ErrorRecord record = outputValue.Data as ErrorRecord; - Dbg.Assert(record != null, "ProcessReader should ensure that data is ErrorRecord"); - record.SetInvocationInfo(this.Command.MyInvocation); - this.commandRuntime._WriteErrorSkipAllowCheck(record, isNativeError: true); - } - else if (outputValue.Stream == MinishellStream.Output) - { - this.commandRuntime._WriteObjectSkipAllowCheck(outputValue.Data); - } - else if (outputValue.Stream == MinishellStream.Debug) - { - string temp = outputValue.Data as string; - Dbg.Assert(temp != null, "ProcessReader should ensure that data is string"); - this.Command.PSHostInternal.UI.WriteDebugLine(temp); - } - else if (outputValue.Stream == MinishellStream.Verbose) - { - string temp = outputValue.Data as string; - Dbg.Assert(temp != null, "ProcessReader should ensure that data is string"); - this.Command.PSHostInternal.UI.WriteVerboseLine(temp); - } - else if (outputValue.Stream == MinishellStream.Warning) - { - string temp = outputValue.Data as string; - Dbg.Assert(temp != null, "ProcessReader should ensure that data is string"); - this.Command.PSHostInternal.UI.WriteWarningLine(temp); - } - else if (outputValue.Stream == MinishellStream.Progress) + ErrorRecord record = outputValue.Data as ErrorRecord; + Dbg.Assert(record != null, "ProcessReader should ensure that data is ErrorRecord"); + record.SetInvocationInfo(this.Command.MyInvocation); + this.commandRuntime._WriteErrorSkipAllowCheck(record, isNativeError: true); + } + else if (outputValue.Stream == MinishellStream.Output) + { + this.commandRuntime._WriteObjectSkipAllowCheck(outputValue.Data); + } + else if (outputValue.Stream == MinishellStream.Debug) + { + string temp = outputValue.Data as string; + Dbg.Assert(temp != null, "ProcessReader should ensure that data is string"); + this.Command.PSHostInternal.UI.WriteDebugLine(temp); + } + else if (outputValue.Stream == MinishellStream.Verbose) + { + string temp = outputValue.Data as string; + Dbg.Assert(temp != null, "ProcessReader should ensure that data is string"); + this.Command.PSHostInternal.UI.WriteVerboseLine(temp); + } + else if (outputValue.Stream == MinishellStream.Warning) + { + string temp = outputValue.Data as string; + Dbg.Assert(temp != null, "ProcessReader should ensure that data is string"); + this.Command.PSHostInternal.UI.WriteWarningLine(temp); + } + else if (outputValue.Stream == MinishellStream.Progress) + { + PSObject temp = outputValue.Data as PSObject; + if (temp != null) { - PSObject temp = outputValue.Data as PSObject; - if (temp != null) + long sourceId = 0; + PSMemberInfo info = temp.Properties["SourceId"]; + if (info != null) { - long sourceId = 0; - PSMemberInfo info = temp.Properties["SourceId"]; - if (info != null) - { - sourceId = (long)info.Value; - } - info = temp.Properties["Record"]; - ProgressRecord rec = null; - if (info != null) - { - rec = info.Value as ProgressRecord; - } - if (rec != null) - { - this.Command.PSHostInternal.UI.WriteProgress(sourceId, rec); - } + sourceId = (long)info.Value; + } + info = temp.Properties["Record"]; + ProgressRecord rec = null; + if (info != null) + { + rec = info.Value as ProgressRecord; + } + if (rec != null) + { + this.Command.PSHostInternal.UI.WriteProgress(sourceId, rec); } } - else if (outputValue.Stream == MinishellStream.Information) - { - InformationRecord record = outputValue.Data as InformationRecord; - Dbg.Assert(record != null, "ProcessReader should ensure that data is InformationRecord"); - this.commandRuntime.WriteInformation(record); - } - - if (this.Command.Context.CurrentPipelineStopping) - { - this.StopProcessing(); - break; - } - value = _outputReader.Read(); + } + else if (outputValue.Stream == MinishellStream.Information) + { + InformationRecord record = outputValue.Data as InformationRecord; + Dbg.Assert(record != null, "ProcessReader should ensure that data is InformationRecord"); + this.commandRuntime.WriteInformation(record); } } @@ -1107,7 +1186,7 @@ private bool IsDownstreamOutDefault(Pipe downstreamPipe) /// private void CalculateIORedirection(out bool redirectOutput, out bool redirectError, out bool redirectInput) { - redirectInput = true; + redirectInput = this.Command.MyInvocation.PipelinePosition > 0; redirectOutput = true; redirectError = true; @@ -1163,9 +1242,6 @@ private void CalculateIORedirection(out bool redirectOutput, out bool redirectEr redirectError = true; } - if (_inputWriter.Count == 0 && (!this.Command.MyInvocation.ExpectingInput)) - redirectInput = false; - // Remoting server consideration. // Currently, the WinRM is using std io pipes to communicate with PowerShell server. // To protect these std io pipes from access from user command, we have replaced the original std io pipes with null pipes. @@ -1344,676 +1420,447 @@ private bool IsMiniShell() internal static bool IsServerSide { get; set; } } - /// - /// Helper class to handle writing input to a process. - /// - internal class ProcessInputWriter + internal class ProcessOutputHandler { - #region constructor - - private InternalCommand _command; - /// - /// Creates an instance of ProcessInputWriter - /// - internal ProcessInputWriter(InternalCommand command) - { - Dbg.Assert(command != null, "Caller should validate the parameter"); - _command = command; - } + internal const string XmlCliTag = "#< CLIXML"; - #endregion constructor + private int _refCount; + private BlockingCollection _queue; + private bool _isFirstOutput; + private bool _isFirstError; + private bool _isXmlCliOutput; + private bool _isXmlCliError; + private string _processFileName; - /// - /// Input is collected in this list - /// - private ArrayList _inputList = new ArrayList(); - /// - /// Add an object to write to process - /// - /// - internal void Add(object input) + public ProcessOutputHandler(Process process, BlockingCollection queue) { - _inputList.Add(input); - } + Debug.Assert(process.StartInfo.RedirectStandardOutput || process.StartInfo.RedirectStandardError, "Caller should redirect at least one stream"); + _refCount = 0; + _processFileName = process.StartInfo.FileName; + _queue = queue; - /// - /// Count of object in inputlist - /// - internal int Count - { - get + // we incrementing refCount on the same thread and before running any processing + // so it's safe to do it without Interlocked. + if (process.StartInfo.RedirectStandardOutput) { _refCount++; } + if (process.StartInfo.RedirectStandardError) { _refCount++; } + + // once we have _refCount, we can start processing + if (process.StartInfo.RedirectStandardOutput) { - return _inputList.Count; + _isFirstOutput = true; + _isXmlCliOutput = false; + process.OutputDataReceived += new DataReceivedEventHandler(OutputHandler); + process.BeginOutputReadLine(); } - } - - /// - /// Stream to which input is written - /// - private StreamWriter _streamWriter; - - /// - /// Format of input. - /// - private NativeCommandIOFormat _inputFormat; - - /// - /// Thread which writes the input - /// - private Thread _inputThread; - - /// - /// Start writing input to process - /// - /// - /// process to which input is written - /// - /// - /// - internal void Start(Process process, NativeCommandIOFormat inputFormat) - { - Dbg.Assert(process != null, "caller should validate the parameter"); - - //Get the encoding for writing to native command. Note we get the Encoding - //from the current scope so a script or function can use a different encoding - //than global value. - Encoding pipeEncoding = _command.Context.GetVariableValue(SpecialVariables.OutputEncodingVarPath) as System.Text.Encoding ?? - Encoding.ASCII; - - _streamWriter = new StreamWriter(process.StandardInput.BaseStream, - pipeEncoding); - _inputFormat = inputFormat; - if (inputFormat == NativeCommandIOFormat.Text) + if (process.StartInfo.RedirectStandardError) { - ConvertToString(); + _isFirstError = true; + _isXmlCliError = false; + process.ErrorDataReceived += new DataReceivedEventHandler(ErrorHandler); + process.BeginErrorReadLine(); } - _inputThread = new Thread(new ThreadStart(this.WriterThreadProc)); - _inputThread.Start(); - } - - private bool _stopping = false; - /// - /// Stop writing input to process - /// - internal void Stop() - { - _stopping = true; } - /// - /// This method wait for writer thread to finish. - /// - internal void Done() + private void decrementRefCount() { - if (_inputThread != null) + Debug.Assert(_refCount > 0, "RefCount should always be positive, when we are trying to decrement it"); + if (Interlocked.Decrement(ref _refCount) == 0) { - _inputThread.Join(); + _queue.CompleteAdding(); } } - /// - /// Thread procedure for writing data to the child process... - /// - private void WriterThreadProc() + private void OutputHandler(object sender, DataReceivedEventArgs outputReceived) { - try + if (outputReceived.Data != null) { - if (_inputFormat == NativeCommandIOFormat.Text) + if (_isFirstOutput) + { + _isFirstOutput = false; + if (string.Equals(outputReceived.Data, XmlCliTag, StringComparison.Ordinal)) + { + _isXmlCliOutput = true; + return; + } + } + + if (_isXmlCliOutput) { - WriteTextInput(); + foreach (var record in DeserializeCliXmlObject(outputReceived.Data, true)) + { + _queue.Add(record); + } } else { - WriteXmlInput(); + _queue.Add(new ProcessOutputObject(outputReceived.Data, MinishellStream.Output)); } } - catch (System.IO.IOException) + else { + decrementRefCount(); } } - private void WriteTextInput() + private void ErrorHandler(object sender, DataReceivedEventArgs errorReceived) { - try + if (errorReceived.Data != null) { - foreach (object o in _inputList) + if (string.Equals(errorReceived.Data, XmlCliTag, StringComparison.Ordinal)) { - if (_stopping) return; - - string line = PSObject.ToStringParser(_command.Context, o); - _streamWriter.Write(line); + _isXmlCliError = true; + return; } - } - finally + + if (_isXmlCliError) + { + foreach (var record in DeserializeCliXmlObject(errorReceived.Data, false)) + { + _queue.Add(record); + } + } + else + { + ErrorRecord errorRecord; + if (_isFirstError) + { + _isFirstError = false; + // Produce a regular error record for the first line of the output + errorRecord = new ErrorRecord(new RemoteException(errorReceived.Data), "NativeCommandError", ErrorCategory.NotSpecified, errorReceived.Data); + } + else + { + // Wrap the rest of the output in ErrorRecords with the "NativeCommandErrorMessage" error ID + errorRecord = new ErrorRecord(new RemoteException(errorReceived.Data), "NativeCommandErrorMessage", ErrorCategory.NotSpecified, null); + } + + _queue.Add(new ProcessOutputObject(errorRecord, MinishellStream.Error)); + } + } + else { - _streamWriter.Dispose(); + decrementRefCount(); } } - private void WriteXmlInput() + private List DeserializeCliXmlObject(string xml, bool isOutput) { + var result = new List(); try { - //Write header - _streamWriter.WriteLine("#< CLIXML"); - - // When (if) switching to XmlTextWriter.Create remember the OmitXmlDeclaration difference - XmlWriter writer = XmlWriter.Create(_streamWriter); - Serializer ser = new Serializer(writer); - foreach (object o in _inputList) + using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml))) { - if (_stopping) return; - ser.Serialize(o); + XmlReader xmlReader = XmlReader.Create(streamReader, InternalDeserializer.XmlReaderSettingsForCliXml); + Deserializer des = new Deserializer(xmlReader); + while (!des.Done()) + { + string streamName; + object obj = des.Deserialize(out streamName); + + //Decide the stream to which data belongs + MinishellStream stream = MinishellStream.Unknown; + if (streamName != null) + { + stream = StringToMinishellStreamConverter.ToMinishellStream(streamName); + } + if (stream == MinishellStream.Unknown) + { + stream = isOutput ? MinishellStream.Output : MinishellStream.Error; + } + + //Null is allowed only in output stream + if (stream != MinishellStream.Output && obj == null) + { + continue; + } + + if (stream == MinishellStream.Error) + { + if (obj is PSObject) + { + obj = ErrorRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); + } + else + { + string errorMessage = null; + try + { + errorMessage = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); + } + catch (PSInvalidCastException) + { + continue; + } + obj = new ErrorRecord(new RemoteException(errorMessage), + "NativeCommandError", ErrorCategory.NotSpecified, errorMessage); + } + } + else if (stream == MinishellStream.Information) + { + if (obj is PSObject) + { + obj = InformationRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); + } + else + { + string messageData = null; + try + { + messageData = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); + } + catch (PSInvalidCastException) + { + continue; + } + + obj = new InformationRecord(messageData, null); + } + } + else if (stream == MinishellStream.Debug || + stream == MinishellStream.Verbose || + stream == MinishellStream.Warning) + { + //Convert to string + try + { + obj = LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); + } + catch (PSInvalidCastException) + { + continue; + } + } + result.Add(new ProcessOutputObject(obj, stream)); + } } - ser.Done(); } - finally + catch (XmlException originalException) { - _streamWriter.Dispose(); - } - } + string template = NativeCP.CliXmlError; + string message = string.Format( + null, + template, + isOutput ? MinishellStream.Output : MinishellStream.Error, + _processFileName, + originalException.Message); + XmlException newException = new XmlException( + message, + originalException); - /// - /// Formats the input objects using out-string. Output of out-string - /// is given as input to native command processor. - /// This method is to be called from the pipeline thread and not from the - /// thread which writes input in to process. - /// - private void ConvertToString() - { - Dbg.Assert(_inputFormat == NativeCommandIOFormat.Text, "InputFormat should be Text"); + ErrorRecord error = new ErrorRecord( + newException, + "ProcessStreamReader_CliXmlError", + ErrorCategory.SyntaxError, + _processFileName); + result.Add(new ProcessOutputObject(error, MinishellStream.Error)); + } - PipelineProcessor p = new PipelineProcessor(); - p.Add(_command.Context.CreateCommand("out-string", false)); - object[] result = (object[])p.SynchronousExecuteEnumerate(_inputList.ToArray()); - _inputList = new ArrayList(result); + return result; } } /// - /// This helper class reads the output from error and output streams of - /// process. + /// Helper class to handle writing input to a process. /// - internal class ProcessOutputReader + internal class ProcessInputWriter { - #region constructor - - /// - /// Process whose output is to be read. - /// - private Process _process; - - /// - /// Path of the process application - /// - private string _processPath; - - private bool _redirectOutput; - - private bool _redirectError; + #region constructor + private InternalCommand _command; /// - /// Process whose output is read + /// Creates an instance of ProcessInputWriter /// - internal ProcessOutputReader(Process process, string processPath, bool redirectOutput, bool redirectError) + internal ProcessInputWriter(InternalCommand command) { - Dbg.Assert(process != null, "caller should validate the parameter"); - Dbg.Assert(processPath != null, "caller should validate the parameter"); - Dbg.Assert(redirectOutput || redirectError, "Either redirectOutput or redirectError must be true"); - _process = process; - _processPath = processPath; - _redirectOutput = redirectOutput; - _redirectError = redirectError; + Dbg.Assert(command != null, "Caller should validate the parameter"); + _command = command; } #endregion constructor - /// - /// Reader for output stream of process - /// - private ProcessStreamReader _outputReader; - /// - /// Reader for error stream of process - /// - private ProcessStreamReader _errorReader; - /// - /// Synchronized object queue in which object read form output and - /// error streams are deposited - /// - private ObjectStream _processOutput; + private SteppablePipeline _pipeline; + private Serializer _xmlSerializer; /// - /// Start reading the output/error. Note all the work is done asynchronously. + /// Add an object to write to process /// - internal void Start() + /// + internal void Add(object input) { - _processOutput = new ObjectStream(128); + if (_stopping || _streamWriter == null) + { + // if _streamWriter is already null, then we already called Dispose() + // so we should just discard the input. + return; + } - // Start async reading of error and output - // readercount variable is used by multiple threads to close "processOutput" ObjectStream. - // Without properly initializing the readercount, the ObjectStream might get - // closed early. readerCount is protected here by using the lock. - lock (_readerLock) + if (_inputFormat == NativeCommandIOFormat.Text) { - if (_redirectOutput) - { - _readerCount++; - _outputReader = new ProcessStreamReader(_process.StandardOutput, _processPath, true, _processOutput.ObjectWriter, this); - _outputReader.Start(); - } - if (_redirectError) - { - _readerCount++; - _errorReader = new ProcessStreamReader(_process.StandardError, _processPath, false, _processOutput.ObjectWriter, this); - _errorReader.Start(); - } + AddTextInput(input); + } + else // Xml + { + AddXmlInput(input); } } - /// - /// Stops reading from streams. This is called from NativeCommandProcessor's StopProcessing - /// method. Note return of this method doesn't mean reading has stopped and all threads are - /// done. - /// Use Done to ensure that all reading threads are finished. - /// - internal void Stop() + private void AddTextInput(object input) { - if (_processOutput != null) - { - try - { - //Close the reader for the stream. - _processOutput.ObjectReader.Close(); - } - catch (Exception e) // ignore non-severe exceptions - { - CommandProcessorBase.CheckForSevereException(e); - } + AddTextInputFromFormattedArray(_pipeline.Process(input)); + } + private void AddTextInputFromFormattedArray(Array formattedObjects) + { + foreach (var item in formattedObjects) + { + string line = PSObject.ToStringParser(_command.Context, item); + // if process is already finished and we are trying to write something to it, + // we will get IOException try { - _processOutput.Close(); + _streamWriter.WriteLine(line); } - catch (Exception e) // ignore non-severe exceptions + catch (IOException) { - CommandProcessorBase.CheckForSevereException(e); + // we are assuming that process is already finished + // we should just stop processing at this point + this.Dispose(); + // stop foreach execution + break; } } } - /// - /// This method returns when all output reader threads have returned - /// - internal void Done() - { - if (_outputReader != null) - { - _outputReader.Done(); - } - if (_errorReader != null) - { - _errorReader.Done(); - } - } - - /// - /// Return one object which was read from the process. - /// - /// - /// AutomationNull.Value if no more objects. - /// object of type ProcessOutputObject otherwise - /// - internal object Read() - { - return _processOutput.ObjectReader.Read(); - } - - /// - /// object used for synchronizing ReaderDone call between two readers - /// - private object _readerLock = new object(); - - /// - /// Count of readers - this is set by Start. If both output and error - /// are redirected, it will be 2. If only one is redirected, it'll be 1. - /// - private int _readerCount; - - /// - /// This method is called by output or error reader when they are - /// done reading. When it is called two times, we close the writer. - /// - /// - internal void ReaderDone(bool isOutput) + private void AddXmlInput(object input) { - int temp; - lock (_readerLock) + try { - temp = --_readerCount; + _xmlSerializer.Serialize(input); } - if (temp == 0) + catch (IOException) { - _processOutput.ObjectWriter.Close(); + // we are assuming that process is already finished + // we should just stop processing at this point + this.Dispose(); } } - } - - /// - /// This class reads the string from output or error streams of process - /// and processes them appropriately. - /// - /// - /// This class is not thread safe. It is assumed that NativeCommandProcessor - /// class will synchronize access to this class between different threads. - /// - internal class ProcessStreamReader - { - #region constructor - - /// - /// Stream from which data is read. - /// - private StreamReader _streamReader; /// - /// Flag which tells if streamReader is for stdout or stderr stream of process - /// - private bool _isOutput; - - /// - /// Writer to which data read from stream are written - /// - private PipelineWriter _writer; - - /// - /// Path to the process. This is used for setting the name of the thread. + /// Stream to which input is written /// - private string _processPath; + private StreamWriter _streamWriter; /// - /// ProcessReader which owns this stream reader + /// Format of input. /// - private ProcessOutputReader _processOutputReader; + private NativeCommandIOFormat _inputFormat; /// - /// Creates an instance of ProcessStreamReader + /// Start writing input to process /// - /// - /// Stream from which data is read - /// - /// - /// Path to the process. This is used for setting the name of the thread. - /// - /// - /// if true stream is output stream of process - /// else stream is error stream. - /// - /// - /// Processed data is written to it + /// + /// process to which input is written /// - /// - /// ProcessOutputReader which owns this stream reader + /// /// - internal ProcessStreamReader(StreamReader streamReader, string processPath, bool isOutput, - PipelineWriter writer, ProcessOutputReader processOutputReader) + internal void Start(Process process, NativeCommandIOFormat inputFormat) { - Dbg.Assert(streamReader != null, "Caller should validate the parameter"); - Dbg.Assert(processPath != null, "Caller should validate the parameter"); - Dbg.Assert(writer != null, "Caller should validate the parameter"); - Dbg.Assert(processOutputReader != null, "Caller should validate the parameter"); - - _streamReader = streamReader; - _processPath = processPath; - _isOutput = isOutput; - _writer = writer; - _processOutputReader = processOutputReader; - } + Dbg.Assert(process != null, "caller should validate the paramter"); - #endregion constructor + //Get the encoding for writing to native command. Note we get the Encoding + //from the current scope so a script or function can use a different encoding + //than global value. + Encoding pipeEncoding = _command.Context.GetVariableValue(SpecialVariables.OutputEncodingVarPath) as System.Text.Encoding ?? + Encoding.ASCII; - /// - /// Thread on which reading happens - /// - private Thread _thread = null; + _streamWriter = new StreamWriter(process.StandardInput.BaseStream, pipeEncoding); + _streamWriter.AutoFlush = true; - /// - /// Launches a new thread to start reading. - /// - internal void Start() - { - _thread = new Thread(new ThreadStart(ReaderStartProc)); - if (_isOutput) + _inputFormat = inputFormat; + + if (_inputFormat == NativeCommandIOFormat.Xml) { - _thread.Name = string.Format(CultureInfo.InvariantCulture, "{0} :Output Reader", _processPath); + _streamWriter.WriteLine(ProcessOutputHandler.XmlCliTag); + _xmlSerializer = new Serializer(XmlWriter.Create(_streamWriter)); } - else + else // Text { - _thread.Name = string.Format(CultureInfo.InvariantCulture, "{0} :Error Reader", _processPath); + _pipeline = ScriptBlock.Create("Out-String -Stream").GetSteppablePipeline(); + _pipeline.Begin(true); } - _thread.Start(); } - /// - /// This method returns when reader thread has returned. - /// - internal void Done() - { - if (_thread != null) - { - _thread.Join(); - } - } + bool _stopping = false; /// - /// Thread proc for reading + /// Stop writing input to process /// - private void ReaderStartProc() + internal void Stop() { - try - { - ReaderStartProcHelper(); - } - catch (Exception ex) - { - CommandProcessorBase.CheckForSevereException(ex); - } - finally - { - _processOutputReader.ReaderDone(_isOutput); - } + _stopping = true; } - private void ReaderStartProcHelper() + internal void Dispose() { - //read the first line to detect the format. - //for xml, first line is #< CLIXML - string line = _streamReader.ReadLine(); - if (line == null) - { - // nothing to do - } - else if (line.Equals("#< CLIXML", StringComparison.Ordinal) == false) + // we allow call Dispose() multiply times. + // For example one time from ProcessRecord() code path, + // when we detect that process already finished + // and once from Done() code path. + // Even though Dispose() could be called multiple times, + // the calls are on the same thread, so there is no race condition + if (_pipeline != null) { - ReadText(line); + _pipeline.Dispose(); + _pipeline = null; } - else + + if (_xmlSerializer != null) { - ReadXml(); + _xmlSerializer = null; } - } - private void ReadText(string line) - { - if (_isOutput) + // streamWriter can be null if we didn't call Start method + if (_streamWriter != null) { - while (line != null) + try { - AddObjectToWriter(line, MinishellStream.Output); - line = _streamReader.ReadLine(); + _streamWriter.Dispose(); } - } - else - { - // - // Produce a regular error record for the first line of the output - // - ErrorRecord errorRecord = new ErrorRecord(new RemoteException(line), - "NativeCommandError", ErrorCategory.NotSpecified, line); - AddObjectToWriter(errorRecord, MinishellStream.Error); - - // - // Wrap the rest of the output in ErrorRecords with the "NativeCommandErrorMessage" error ID - // - while ((line = _streamReader.ReadLine()) != null) + catch (IOException) { - AddObjectToWriter( - new ErrorRecord( - new RemoteException(line), - "NativeCommandErrorMessage", - ErrorCategory.NotSpecified, - null), - MinishellStream.Error); + // on unix, if process is already finished attempt to dispose it will + // lead to "Broken pipe" exception. + // we are ignoring it here } + _streamWriter = null; } } - private void ReadXml() + internal void Done() { - try + if (_inputFormat == NativeCommandIOFormat.Xml) { - XmlReader xmlReader = XmlReader.Create(_streamReader, InternalDeserializer.XmlReaderSettingsForCliXml); - Deserializer des = new Deserializer(xmlReader); - while (!des.Done()) + if (_xmlSerializer != null) { - string streamName; - object obj = des.Deserialize(out streamName); - - //Decide the stream to which data belongs - MinishellStream stream = MinishellStream.Unknown; - if (streamName != null) - { - stream = StringToMinishellStreamConverter.ToMinishellStream(streamName); - } - if (stream == MinishellStream.Unknown) - { - stream = _isOutput ? MinishellStream.Output : MinishellStream.Error; - } - - //Null is allowed only in output stream - if (stream != MinishellStream.Output && obj == null) - { - continue; - } - - if (stream == MinishellStream.Error) - { - if (obj is PSObject) - { - obj = ErrorRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); - } - else - { - string errorMessage = null; - try - { - errorMessage = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); - } - catch (PSInvalidCastException) - { - continue; - } - obj = new ErrorRecord(new RemoteException(errorMessage), - "NativeCommandError", ErrorCategory.NotSpecified, errorMessage); - } - } - else if (stream == MinishellStream.Information) - { - if (obj is PSObject) - { - obj = InformationRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); - } - else - { - string messageData = null; - try - { - messageData = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); - } - catch (PSInvalidCastException) - { - continue; - } - - obj = new InformationRecord(messageData, null); - } - } - else if (stream == MinishellStream.Debug || - stream == MinishellStream.Verbose || - stream == MinishellStream.Warning) - { - //Convert to string - try - { - obj = LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); - } - catch (PSInvalidCastException) - { - continue; - } - } - AddObjectToWriter(obj, stream); + _xmlSerializer.Done(); } } - catch (XmlException originalException) - { - string template = NativeCP.CliXmlError; - string message = string.Format( - null, - template, - _isOutput ? MinishellStream.Output : MinishellStream.Error, - _processPath, - originalException.Message); - XmlException newException = new XmlException( - message, - originalException); - - ErrorRecord error = new ErrorRecord( - newException, - "ProcessStreamReader_CliXmlError", - ErrorCategory.SyntaxError, - _processPath); - AddObjectToWriter(error, MinishellStream.Error); - } - } - - /// - /// Adds one object to writer - /// - private void AddObjectToWriter(object data, MinishellStream stream) - { - try + else // Text { - ProcessOutputObject dataObject = new ProcessOutputObject(data, stream); - //writer is shared between Error and Output reader. - lock (_writer) + // if _pipeline == null, we already called Dispose(), + // for example, because downstream process finished + if (_pipeline != null) { - _writer.Write(dataObject); + var finalResults = _pipeline.End(); + AddTextInputFromFormattedArray(finalResults); } } - catch (PipelineClosedException) - { - // The output queue may have been closed asynchronously - ; - } - catch (System.ObjectDisposedException) - { - // The output queue may have been disposed asynchronously when StopProcessing is called... - ; - } + + Dispose(); } } - + #if !CORECLR // There is no GUI application on OneCore, so powershell on OneCore should always have a console attached. /// diff --git a/test/powershell/Host/ConsoleHost.Tests.ps1 b/test/powershell/Host/ConsoleHost.Tests.ps1 index 1ab7fed1daa..fba8a3b61b9 100644 --- a/test/powershell/Host/ConsoleHost.Tests.ps1 +++ b/test/powershell/Host/ConsoleHost.Tests.ps1 @@ -1,5 +1,52 @@ using namespace System.Diagnostics +# Minishell (Singleshell) is a powershell concept. +# Its primary use-case is when somebody executes a scriptblock in the new powershell process. +# The objects are automatically marshelled to the child process and +# back to the parent session, so users can avoid custom +# serialization to pass objects between two processes. + +Describe 'minishell for native executables' -Tag 'CI' { + + BeforeAll { + $powershell = Join-Path -Path $PsHome -ChildPath "powershell" + } + + Context 'Streams from minishell' { + + It 'gets a hashtable object from minishell' { + $output = & $powershell -noprofile { @{'a' = 'b'} } + ($output | measure).Count | Should Be 1 + ($output.GetType().Name) | Should Be 'Hashtable' + $output['a'] | Should Be 'b' + } + + It 'gets the error stream from minishell' { + $output = & $powershell -noprofile { Write-Error 'foo' } 2>&1 + ($output | measure).Count | Should Be 1 + ($output.GetType().Name) | Should Be 'ErrorRecord' + $output.FullyQualifiedErrorId | Should Be 'Microsoft.PowerShell.Commands.WriteErrorException' + } + + It 'gets the information stream from minishell' { + $output = & $powershell -noprofile { Write-Information 'foo' } 6>&1 + ($output.GetType().Name) | Should Be 'InformationRecord' + $output | Should Be 'foo' + } + } + + Context 'Streams to minishell' { + It "passes input into minishell" { + $a = 1,2,3 + $val = $a | & $powershell -noprofile -command { $input } + $val.Count | Should Be 3 + $val[0] | Should Be 1 + $val[1] | Should Be 2 + $val[2] | Should Be 3 + } + } +} + Describe "ConsoleHost unit tests" -tags "Feature" { BeforeAll { @@ -53,15 +100,6 @@ Describe "ConsoleHost unit tests" -tags "Feature" { } } - It "Verify Simple Interop Scenario Child Single Shell" { - $a = 1,2,3 - $val = $a | & $powershell -noprofile -command { $input } - $val.Count | Should Be 3 - $val[0] | Should Be 1 - $val[1] | Should Be 2 - $val[2] | Should Be 3 - } - It "Verify Validate Dollar Error Populated should throw exception" { $origEA = $ErrorActionPreference $ErrorActionPreference = "Stop" diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 index 7a34facb85d..286c230c5ca 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 @@ -1,3 +1,24 @@ +Describe 'native commands lifecycle' -tags 'Feature' { + + BeforeAll { + $powershell = Join-Path -Path $PsHome -ChildPath "powershell" + } + + It "native | ps | native doesn't block" { + $first = $true + & $powershell -command '1..10 | % {Start-Sleep -mill 100; $_}' | %{$_} | & $powershell -command '$input' | % { + if ($first) + { + $first = $false + $firstTime = [datetime]::Now + } + $lastTime = [datetime]::Now + } + + $lastTime - $firstTime | Should BeGreaterThan ([timespan]::new(0, 0, 0, 0, 100)) # 100 milliseconds + } +} + Describe "Native Command Processor" -tags "Feature" { BeforeAll { diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 index 7c191802ef6..523a7439e51 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 @@ -51,3 +51,20 @@ Describe "Native streams behavior with PowerShell" -Tags 'CI' { } } } + +Describe 'piping powershell objects to finished native executable' -Tags 'CI' { + # Find where test/powershell is so we can find the echoargs command relative to it + $powershellTestDir = $PSScriptRoot + while ($powershellTestDir -notmatch 'test[\\/]powershell$') { + $powershellTestDir = Split-Path $powershellTestDir + } + $echoArgs = Join-Path (Split-Path $powershellTestDir) tools/EchoArgs/bin/echoargs + + It 'doesn''t throw any exceptions, when we are piping to the closed executable' { + 1..3 | % { + Start-Sleep -Milliseconds 100 + # yeild some multi-line formatted object + @{'a' = 'b'} + } | & $echoArgs | Should Be $null + } +}