From a05fffd9dce1974be8db7c912745bbc480cb42b6 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Mon, 10 Oct 2016 15:21:56 -0700 Subject: [PATCH 01/20] Add a minishell clixml tests --- .../NativeExecution/NativeMinishell.Tests.ps1 | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 new file mode 100644 index 00000000000..44db92b6eef --- /dev/null +++ b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 @@ -0,0 +1,18 @@ +# Minishell is a powershell concept. +# It's primare use-case is when somebody executes a scriptblock in the new powershell process. +# The objects are automatically marshelled 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" + } + + It 'gets a hashtable object from minishell' { + $output = & powershell { @{'a' = 'b'} } + ($output | measure).Count | Should Be 1 + ($output.GetType().Name) | Should Be 'Hashtable' + $output['a'] | Should Be 'b' + } +} From b93042670f15aeb1b473ab420f4421ddc3be5b78 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Mon, 10 Oct 2016 18:57:39 -0700 Subject: [PATCH 02/20] Improve pipeline for native commands - Start native process in Prepare() instead of Complete() in NativeCommandProcessor. - Remove unneeded input, output, error threads. Replaced by simpler primitives. --- .../engine/NativeCommandProcessor.cs | 1212 ++++++----------- 1 file changed, 442 insertions(+), 770 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 9195cb793f9..b1e3ac36331 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. @@ -140,6 +142,8 @@ internal ProcessOutputObject(object data, MinishellStream stream) /// internal class NativeCommandProcessor : CommandProcessorBase { + private static string XmlCliTag = "#< CLIXML"; + #region ctor/native command properties /// @@ -306,6 +310,8 @@ internal override void Prepare(IDictionary psDefaultParameterValues) { this.NativeParameterBinderController.BindParameters(arguments); } + + InitNativeProcess(); } /// @@ -315,7 +321,6 @@ internal override void ProcessRecord() { while (Read()) { - // Accumulate everything from the pipe and execute at the end. _inputWriter.Add(Command.CurrentPipelineObject); } } @@ -330,17 +335,39 @@ 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 _background; + + private ProcessStartInfo _startInfo; + + /// + /// 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 ConcurrentQueue _nativeProcessOutputQueue; + + private bool _scrapeHostOutput; + + private bool _redirectInput; + + private Host.Coordinates _startPosition; + + /// + /// 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... + /// + private Exception _exceptionToRethrow = null; + /// /// object used for synchronization between StopProcessing thread and /// Pipeline thread. @@ -356,12 +383,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. @@ -369,26 +392,22 @@ internal override void Complete() //Calculate if input and output are redirected. bool redirectOutput; bool redirectError; - bool redirectInput; - CalculateIORedirection(out redirectOutput, out redirectError, out redirectInput); + CalculateIORedirection(out redirectOutput, out redirectError, out _redirectInput); // Find out if it's the only command in the pipeline. bool soloCommand = this.Command.MyInvocation.PipelineLength == 1; // Get the start info for the process. - ProcessStartInfo startInfo = GetProcessStartInfo(redirectOutput, redirectError, redirectInput, soloCommand); + _startInfo = GetProcessStartInfo(redirectOutput, redirectError, _redirectInput, soloCommand); if (this.Command.Context.CurrentPipelineStopping) { 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; try { @@ -405,15 +424,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; } } @@ -433,7 +452,7 @@ internal override void Complete() try { _nativeProcess = new Process(); - _nativeProcess.StartInfo = startInfo; + _nativeProcess.StartInfo = _startInfo; _nativeProcess.Start(); } catch (Win32Exception) @@ -445,7 +464,7 @@ internal override void Complete() // See if there is a file association for this command. If so // then we'll use that. If there's no file association, then // try shell execute... - string executable = FindExecutable(startInfo.FileName); + string executable = FindExecutable(_startInfo.FileName); bool notDone = true; if (!String.IsNullOrEmpty(executable)) { @@ -455,10 +474,10 @@ internal override void Complete() ConsoleVisibility.AllocateHiddenConsole(); } - string oldArguments = startInfo.Arguments; - string oldFileName = startInfo.FileName; - startInfo.Arguments = "\"" + startInfo.FileName + "\" " + startInfo.Arguments; - startInfo.FileName = executable; + string oldArguments = _startInfo.Arguments; + string oldFileName = _startInfo.FileName; + _startInfo.Arguments = "\"" + _startInfo.FileName + "\" " + _startInfo.Arguments; + _startInfo.FileName = executable; try { _nativeProcess.Start(); @@ -467,8 +486,8 @@ internal override void Complete() catch (Win32Exception) { // Restore the old filename and arguments to try shell execute last... - startInfo.Arguments = oldArguments; - startInfo.FileName = oldFileName; + _startInfo.Arguments = oldArguments; + _startInfo.FileName = oldFileName; } } // We got here because there was either no executable found for this @@ -476,12 +495,12 @@ internal override void Complete() // we will try launching one last time using ShellExecute... if (notDone) { - if (soloCommand && startInfo.UseShellExecute == false) + if (soloCommand && _startInfo.UseShellExecute == false) { - startInfo.UseShellExecute = true; - startInfo.RedirectStandardInput = false; - startInfo.RedirectStandardOutput = false; - startInfo.RedirectStandardError = false; + _startInfo.UseShellExecute = true; + _startInfo.RedirectStandardInput = false; + _startInfo.RedirectStandardOutput = false; + _startInfo.RedirectStandardError = false; _nativeProcess.Start(); } else @@ -499,21 +518,21 @@ internal override void Complete() // Something like // ls | notepad | sort.exe // should block until the notepad process is terminated. - background = false; + _background = false; } else { - background = true; - if (startInfo.UseShellExecute == false) + _background = true; + if (_startInfo.UseShellExecute == false) { - background = IsWindowsApplication(_nativeProcess.StartInfo.FileName); + _background = IsWindowsApplication(_nativeProcess.StartInfo.FileName); } } try { //If input is redirected, start input to process. - if (startInfo.RedirectStandardInput) + if (_startInfo.RedirectStandardInput) { NativeCommandIOFormat inputFormat = NativeCommandIOFormat.Text; if (_isMiniShell) @@ -528,95 +547,337 @@ 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 (_background == 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 List DeserializeCliXmlObject(string xml, bool isOutput) + { + var result = new List(); + try + { + using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml))) { - if (background == false) + XmlReader xmlReader = XmlReader.Create(streamReader, InternalDeserializer.XmlReaderSettingsForCliXml); + Deserializer des = new Deserializer(xmlReader); + while (!des.Done()) { - //Wait for process to exit - _nativeProcess.WaitForExit(); + string streamName; + object obj = des.Deserialize(out streamName); - //Wait for input writer to finish. - _inputWriter.Done(); + //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; + } - //Wait for outputReader to finish - if (_outputReader != null) + //Null is allowed only in output stream + if (stream != MinishellStream.Output && obj == null) { - _outputReader.Done(); + continue; } - // Capture screen output if we are transcribing - if (this.Command.Context.EngineHostInterface.UI.IsTranscribing && - scrapeHostOutput) + 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) { - Host.Coordinates endPosition = this.Command.Context.EngineHostInterface.UI.RawUI.CursorPosition; - endPosition.X = this.Command.Context.EngineHostInterface.UI.RawUI.BufferSize.Width - 1; + 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; + } - // If the end position is before the start position, then capture the entire buffer. - if (endPosition.Y < startPosition.Y) + 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) { - startPosition.Y = 0; + continue; } + } + result.Add(new ProcessOutputObject(obj, stream)); + } + } + } + catch (XmlException originalException) + { + string template = NativeCP.CliXmlError; + string message = string.Format( + null, + template, + isOutput ? MinishellStream.Output : MinishellStream.Error, + Path, + originalException.Message); + XmlException newException = new XmlException( + message, + originalException); - Host.BufferCell[,] bufferContents = this.Command.Context.EngineHostInterface.UI.RawUI.GetBufferContents( - new Host.Rectangle(startPosition, endPosition)); + ErrorRecord error = new ErrorRecord( + newException, + "ProcessStreamReader_CliXmlError", + ErrorCategory.SyntaxError, + Path); + result.Add(new ProcessOutputObject(error, MinishellStream.Error)); + } + + return result; + } - StringBuilder lineContents = new StringBuilder(); - StringBuilder bufferText = new StringBuilder(); + private void InitOutputQueue() + { + //if output is redirected, start reading output of process in queue. + if (_startInfo.RedirectStandardOutput || _startInfo.RedirectStandardError) + { + lock (_sync) + { + if (!_stopped) + { + _nativeProcessOutputQueue = new ConcurrentQueue(); - for (int row = 0; row < bufferContents.GetLength(0); row++) + if (_startInfo.RedirectStandardOutput) + { + bool isFirstOutput = true; + bool isXmlCliOutput = false; + _nativeProcess.OutputDataReceived += new DataReceivedEventHandler((sender, e) => { - if (row > 0) + if (e.Data != null) { - bufferText.Append(Environment.NewLine); + if (isFirstOutput) + { + isFirstOutput = false; + if (e.Data == XmlCliTag) + { + isXmlCliOutput = true; + return; + } + } + + if (isXmlCliOutput) + { + foreach (var record in DeserializeCliXmlObject(e.Data, true)) + { + _nativeProcessOutputQueue.Enqueue(record); + } + } + else + { + _nativeProcessOutputQueue.Enqueue(new ProcessOutputObject(e.Data, MinishellStream.Output)); + } } + }); + _nativeProcess.BeginOutputReadLine(); + } - lineContents.Clear(); - for (int column = 0; column < bufferContents.GetLength(1); column++) + if (_startInfo.RedirectStandardError) + { + bool isFirstError = true; + bool isXmlCliError = false; + _nativeProcess.ErrorDataReceived += new DataReceivedEventHandler((sender, e) => + { + if (e.Data != null) { - lineContents.Append(bufferContents[row, column].Character); + if (e.Data == XmlCliTag) + { + isXmlCliError = true; + return; + } + + if (isXmlCliError) + { + foreach (var record in DeserializeCliXmlObject(e.Data, false)) + { + _nativeProcessOutputQueue.Enqueue(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(e.Data), "NativeCommandError", ErrorCategory.NotSpecified, e.Data); + } + else + { + // Wrap the rest of the output in ErrorRecords with the "NativeCommandErrorMessage" error ID + errorRecord = new ErrorRecord(new RemoteException(e.Data), "NativeCommandErrorMessage", ErrorCategory.NotSpecified, null); + } + + _nativeProcessOutputQueue.Enqueue(new ProcessOutputObject(errorRecord, MinishellStream.Error)); + } } + }); + _nativeProcess.BeginErrorReadLine(); + } + } + } + } + } + + /// + /// Read the output from the native process and send it down the line + /// + private void ConsumeAvailableNativeProcessOutput() + { + if (_background == false) + { + if (_startInfo.RedirectStandardOutput || _startInfo.RedirectStandardError) + { + ProcessOutputObject record; + while (_nativeProcessOutputQueue.TryDequeue(out record)) + { + ProcessOutputRecord(record); + } + } + } + } + + internal override void Complete() + { + try + { + if (_background == false) + { + //Wait for input writer to finish. + _inputWriter.Done(); + + //Wait for process to exit + _nativeProcess.WaitForExit(); + + ConsumeAvailableNativeProcessOutput(); + + // 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; + + // If the end position is before the start position, then capture the entire buffer. + if (endPosition.Y < _startPosition.Y) + { + _startPosition.Y = 0; + } + + 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) { - exceptionToRethrow = e; + _exceptionToRethrow = e; } // try catch (PipelineStoppedException) { @@ -627,11 +888,11 @@ internal override void Complete() { CommandProcessorBase.CheckForSevereException(e); - exceptionToRethrow = e; + _exceptionToRethrow = e; } finally { - if (!redirectOutput) + if (!_startInfo.RedirectStandardOutput) { this.Command.Context.EngineHostInterface.NotifyEndApplication(); } @@ -641,13 +902,13 @@ internal override void Complete() // An exception was thrown while attempting to run the program // so wrap and rethrow it here... - if (exceptionToRethrow != null) + 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.NativeCommandName, _exceptionToRethrow.Message, this.Command.MyInvocation.PositionMessage); - ApplicationFailedException appFailedException = new ApplicationFailedException(message, exceptionToRethrow); + 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; @@ -893,12 +1154,6 @@ internal void StopProcessing() //Stop input writer _inputWriter.Stop(); - //stop output writer - if (_outputReader != null) - { - _outputReader.Stop(); - } - KillProcess(_nativeProcess); } } @@ -924,84 +1179,74 @@ 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"); + Dbg.Assert(outputValue != null, "only object of type ProcessOutputObject expected"); - object value = _outputReader.Read(); - while (value != AutomationNull.Value) + + 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); - } + } + 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(); + if (this.Command.Context.CurrentPipelineStopping) + { + this.StopProcessing(); + return; } } @@ -1107,7 +1352,7 @@ private bool IsDownstreamOutDefault(Pipe downstreamPipe) /// private void CalculateIORedirection(out bool redirectOutput, out bool redirectError, out bool redirectInput) { - redirectInput = true; + redirectInput = this.Command.MyInvocation.PipelineLength > 0; redirectOutput = true; redirectError = true; @@ -1163,9 +1408,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. @@ -1359,32 +1601,30 @@ internal ProcessInputWriter(InternalCommand command) { Dbg.Assert(command != null, "Caller should validate the parameter"); _command = command; + + _pipeline = ScriptBlock.Create("Out-String -Stream").GetSteppablePipeline(); + _pipeline.Begin(true); } #endregion constructor - /// - /// Input is collected in this list - /// - private ArrayList _inputList = new ArrayList(); + private SteppablePipeline _pipeline; + /// /// Add an object to write to process /// /// internal void Add(object input) { - _inputList.Add(input); - } - - /// - /// Count of object in inputlist - /// - internal int Count - { - get + Array formattedObjects = _pipeline.Process(input); + if (_stopping) return; + foreach (var item in formattedObjects) { - return _inputList.Count; + string line = PSObject.ToStringParser(_command.Context, item); + _streamWriter.WriteLine(line); } + + _streamWriter.Flush(); } /// @@ -1397,11 +1637,6 @@ internal int Count /// private NativeCommandIOFormat _inputFormat; - /// - /// Thread which writes the input - /// - private Thread _inputThread; - /// /// Start writing input to process /// @@ -1412,7 +1647,7 @@ internal int Count /// internal void Start(Process process, NativeCommandIOFormat inputFormat) { - Dbg.Assert(process != null, "caller should validate the parameter"); + Dbg.Assert(process != null, "caller should validate the paramter"); //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 @@ -1423,16 +1658,10 @@ internal void Start(Process process, NativeCommandIOFormat inputFormat) _streamWriter = new StreamWriter(process.StandardInput.BaseStream, pipeEncoding); _inputFormat = inputFormat; - - if (inputFormat == NativeCommandIOFormat.Text) - { - ConvertToString(); - } - _inputThread = new Thread(new ThreadStart(this.WriterThreadProc)); - _inputThread.Start(); } - private bool _stopping = false; + bool _stopping = false; + /// /// Stop writing input to process /// @@ -1446,574 +1675,17 @@ internal void Stop() /// internal void Done() { - if (_inputThread != null) - { - _inputThread.Join(); - } - } + _pipeline.End(); + _pipeline.Dispose(); - /// - /// Thread procedure for writing data to the child process... - /// - private void WriterThreadProc() - { - try - { - if (_inputFormat == NativeCommandIOFormat.Text) - { - WriteTextInput(); - } - else - { - WriteXmlInput(); - } - } - catch (System.IO.IOException) + // streamWriter is present, only if we call Start method + if (_streamWriter != null) { - } - } - - private void WriteTextInput() - { - try - { - foreach (object o in _inputList) - { - if (_stopping) return; - - string line = PSObject.ToStringParser(_command.Context, o); - _streamWriter.Write(line); - } - } - finally - { - _streamWriter.Dispose(); - } - } - - private void WriteXmlInput() - { - 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) - { - if (_stopping) return; - ser.Serialize(o); - } - ser.Done(); - } - finally - { - _streamWriter.Dispose(); - } - } - - /// - /// 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"); - - PipelineProcessor p = new PipelineProcessor(); - p.Add(_command.Context.CreateCommand("out-string", false)); - object[] result = (object[])p.SynchronousExecuteEnumerate(_inputList.ToArray()); - _inputList = new ArrayList(result); - } - } - - /// - /// This helper class reads the output from error and output streams of - /// process. - /// - internal class ProcessOutputReader - { - #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; - - /// - /// Process whose output is read - /// - internal ProcessOutputReader(Process process, string processPath, bool redirectOutput, bool redirectError) - { - 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; - } - - #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; - - /// - /// Start reading the output/error. Note all the work is done asynchronously. - /// - internal void Start() - { - _processOutput = new ObjectStream(128); - - // 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 (_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(); - } - } - } - - /// - /// 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() - { - if (_processOutput != null) - { - try - { - //Close the reader for the stream. - _processOutput.ObjectReader.Close(); - } - catch (Exception e) // ignore non-severe exceptions - { - CommandProcessorBase.CheckForSevereException(e); - } - - try - { - _processOutput.Close(); - } - catch (Exception e) // ignore non-severe exceptions - { - CommandProcessorBase.CheckForSevereException(e); - } - } - } - - /// - /// 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) - { - int temp; - lock (_readerLock) - { - temp = --_readerCount; - } - if (temp == 0) - { - _processOutput.ObjectWriter.Close(); - } - } - } - - /// - /// 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. - /// - private string _processPath; - - /// - /// ProcessReader which owns this stream reader - /// - private ProcessOutputReader _processOutputReader; - - /// - /// Creates an instance of ProcessStreamReader - /// - /// - /// 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 - /// - /// - /// ProcessOutputReader which owns this stream reader - /// - internal ProcessStreamReader(StreamReader streamReader, string processPath, bool isOutput, - PipelineWriter writer, ProcessOutputReader processOutputReader) - { - 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; - } - - #endregion constructor - - /// - /// Thread on which reading happens - /// - private Thread _thread = null; - - /// - /// Launches a new thread to start reading. - /// - internal void Start() - { - _thread = new Thread(new ThreadStart(ReaderStartProc)); - if (_isOutput) - { - _thread.Name = string.Format(CultureInfo.InvariantCulture, "{0} :Output Reader", _processPath); - } - else - { - _thread.Name = string.Format(CultureInfo.InvariantCulture, "{0} :Error Reader", _processPath); - } - _thread.Start(); - } - - /// - /// This method returns when reader thread has returned. - /// - internal void Done() - { - if (_thread != null) - { - _thread.Join(); - } - } - - /// - /// Thread proc for reading - /// - private void ReaderStartProc() - { - try - { - ReaderStartProcHelper(); - } - catch (Exception ex) - { - CommandProcessorBase.CheckForSevereException(ex); - } - finally - { - _processOutputReader.ReaderDone(_isOutput); - } - } - - private void ReaderStartProcHelper() - { - //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) - { - ReadText(line); - } - else - { - ReadXml(); - } - } - - private void ReadText(string line) - { - if (_isOutput) - { - while (line != null) - { - AddObjectToWriter(line, MinishellStream.Output); - line = _streamReader.ReadLine(); - } - } - 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) - { - AddObjectToWriter( - new ErrorRecord( - new RemoteException(line), - "NativeCommandErrorMessage", - ErrorCategory.NotSpecified, - null), - MinishellStream.Error); - } - } - } - - private void ReadXml() - { - try - { - 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; - } - } - AddObjectToWriter(obj, stream); - } - } - 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 - { - ProcessOutputObject dataObject = new ProcessOutputObject(data, stream); - //writer is shared between Error and Output reader. - lock (_writer) - { - _writer.Write(dataObject); - } - } - 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... - ; + _streamWriter.Dispose(); } } } - + #if !CORECLR // There is no GUI application on OneCore, so powershell on OneCore should always have a console attached. /// From 6acc1e06291d6ac3ed82d0c9dbe556ffbad160e4 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Tue, 11 Oct 2016 14:58:39 -0700 Subject: [PATCH 03/20] Add error stream and information stream tests for native pipes --- .../NativeExecution/NativeMinishell.Tests.ps1 | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 index 44db92b6eef..6cb14d9e781 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 @@ -10,9 +10,22 @@ Describe 'minishell for native executables' -Tag 'CI' { } It 'gets a hashtable object from minishell' { - $output = & powershell { @{'a' = 'b'} } + $output = & $powershell { @{'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 { 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 { Write-Information 'foo' } 6>&1 + ($output.GetType().Name) | Should Be 'InformationRecord' + $output | Should Be 'foo' + } } From 747f5ff164828401e40516cf78acbc206d1c28d2 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Wed, 12 Oct 2016 16:33:23 -0700 Subject: [PATCH 04/20] Make native pipeline consume available output in the process block --- .../engine/NativeCommandProcessor.cs | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index b1e3ac36331..72e8394cb50 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -323,6 +323,8 @@ internal override void ProcessRecord() { _inputWriter.Add(Command.CurrentPipelineObject); } + + ConsumeAvailableNativeProcessOutput(); } /// @@ -801,10 +803,14 @@ private void InitOutputQueue() } /// - /// Read the output from the native process and send it down the line + /// Read the output from the native process and send it down the line. /// - private void ConsumeAvailableNativeProcessOutput() + /// + /// True if there was any new input available, otherwise false. + /// + private bool ConsumeAvailableNativeProcessOutput() { + bool isNewData = false; if (_background == false) { if (_startInfo.RedirectStandardOutput || _startInfo.RedirectStandardError) @@ -813,9 +819,11 @@ private void ConsumeAvailableNativeProcessOutput() while (_nativeProcessOutputQueue.TryDequeue(out record)) { ProcessOutputRecord(record); + isNewData = true; } } } + return isNewData; } internal override void Complete() @@ -827,9 +835,16 @@ internal override void Complete() //Wait for input writer to finish. _inputWriter.Done(); - //Wait for process to exit - _nativeProcess.WaitForExit(); + //Wait for the process to exit and consume available output + while (!_nativeProcess.HasExited) + { + // TODO: Currently the main pipeline Thread is just spinning. + // It would be much better to park it until new input is available. + ConsumeAvailableNativeProcessOutput(); + } + // read all the available output one more time + _nativeProcess.WaitForExit(); ConsumeAvailableNativeProcessOutput(); // Capture screen output if we are transcribing From 584aec7c92e11e71d704fcae9f09fd134d3b12fa Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Wed, 12 Oct 2016 16:33:38 -0700 Subject: [PATCH 05/20] Add motivation example for the native pipeline changes --- .../NativeExecution/NativeMinishell.Tests.ps1 | 49 +++++++++++++------ 1 file changed, 34 insertions(+), 15 deletions(-) diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 index 6cb14d9e781..eee6bc8844f 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 @@ -9,23 +9,42 @@ Describe 'minishell for native executables' -Tag 'CI' { $powershell = Join-Path -Path $PsHome -ChildPath "powershell" } - It 'gets a hashtable object from minishell' { - $output = & $powershell { @{'a' = 'b'} } - ($output | measure).Count | Should Be 1 - ($output.GetType().Name) | Should Be 'Hashtable' - $output['a'] | Should Be 'b' - } + Context 'Streams' { + + It 'gets a hashtable object from minishell' { + $output = & $powershell { @{'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 { 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 error stream from minishell' { + $output = & $powershell { 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 { Write-Information 'foo' } 6>&1 + ($output.GetType().Name) | Should Be 'InformationRecord' + $output | Should Be 'foo' + } } - It 'gets the information stream from minishell' { - $output = & $powershell { Write-Information 'foo' } 6>&1 - ($output.GetType().Name) | Should Be 'InformationRecord' - $output | Should Be 'foo' + Context 'native commands lifecycle' { + It "native | ps | native doesn't block" { + $first = $true + & $powershell -command '1..5 | % {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 + } } } From df6662aa2e33151a994aeca187e5383f1fcd1bc3 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Wed, 12 Oct 2016 19:12:54 -0700 Subject: [PATCH 06/20] Fix flushing mechanics for input writer in native pipe --- .../engine/NativeCommandProcessor.cs | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 72e8394cb50..8d1d891a9c4 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1631,15 +1631,19 @@ internal ProcessInputWriter(InternalCommand command) /// internal void Add(object input) { + if (_stopping || _streamWriter == null) + { + // if _streamWriter is already null, then we already called Done() + // so we should just discard the input. + return; + } + Array formattedObjects = _pipeline.Process(input); - if (_stopping) return; foreach (var item in formattedObjects) { string line = PSObject.ToStringParser(_command.Context, item); _streamWriter.WriteLine(line); } - - _streamWriter.Flush(); } /// @@ -1670,8 +1674,9 @@ internal void Start(Process process, NativeCommandIOFormat inputFormat) Encoding pipeEncoding = _command.Context.GetVariableValue(SpecialVariables.OutputEncodingVarPath) as System.Text.Encoding ?? Encoding.ASCII; - _streamWriter = new StreamWriter(process.StandardInput.BaseStream, - pipeEncoding); + _streamWriter = new StreamWriter(process.StandardInput.BaseStream, pipeEncoding); + _streamWriter.AutoFlush = true; + _inputFormat = inputFormat; } @@ -1696,7 +1701,8 @@ internal void Done() // streamWriter is present, only if we call Start method if (_streamWriter != null) { - _streamWriter.Dispose(); + _streamWriter.Close(); + _streamWriter = null; } } } From 666bf887d7cbd8a2eb905534f2ce5fc84209764d Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Wed, 12 Oct 2016 19:15:41 -0700 Subject: [PATCH 07/20] Use StreamWriter.Dispose() instead of Close() --- .../engine/NativeCommandProcessor.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 8d1d891a9c4..9673c415a46 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1701,7 +1701,7 @@ internal void Done() // streamWriter is present, only if we call Start method if (_streamWriter != null) { - _streamWriter.Close(); + _streamWriter.Dispose(); _streamWriter = null; } } From 9658e6268218b695ecfbd7acb6a18be26137d289 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 21 Oct 2016 17:07:11 -0700 Subject: [PATCH 08/20] Add a test for broken linux pipe in native pipeline --- .../NativeExecution/NativeStreams.Tests.ps1 | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 index 7c191802ef6..ecac247d822 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 @@ -51,3 +51,16 @@ 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; $_ } | & $echoArgs | Should Be $null + } +} From a6fc5387c2b73c7b7aff6a96d573b3dfae013899 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 21 Oct 2016 18:45:29 -0700 Subject: [PATCH 09/20] Add a code to handle finished process correctly on Unix --- .../engine/NativeCommandProcessor.cs | 38 +++++++++++++++++-- .../NativeExecution/NativeStreams.Tests.ps1 | 6 ++- 2 files changed, 39 insertions(+), 5 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 9673c415a46..6015f8191a1 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1642,7 +1642,20 @@ internal void Add(object input) foreach (var item in formattedObjects) { string line = PSObject.ToStringParser(_command.Context, item); - _streamWriter.WriteLine(line); + // if process is already finished and we are trying to write something to it, + // we will get IOException + try + { + _streamWriter.WriteLine(line); + } + catch (IOException) + { + // we are assuming that process is already finished + // we should just stop processing at this point + this.Done(); + // stop foreach execution + break; + } } } @@ -1695,13 +1708,30 @@ internal void Stop() /// internal void Done() { - _pipeline.End(); - _pipeline.Dispose(); + // we allow call Done() multiply times. + // For example one time from Process() code path, + // when we detect that process already finished + // and once from End() code path. + if (_pipeline != null) + { + _pipeline.End(); + _pipeline.Dispose(); + _pipeline = null; + } // streamWriter is present, only if we call Start method if (_streamWriter != null) { - _streamWriter.Dispose(); + try + { + _streamWriter.Dispose(); + } + catch (IOException) + { + // on unix, if process is already finished attempt to dispose it will + // lead to "Broken pipe" exception. + // we are ignoring it here + } _streamWriter = null; } } diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 index ecac247d822..523a7439e51 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeStreams.Tests.ps1 @@ -61,6 +61,10 @@ Describe 'piping powershell objects to finished native executable' -Tags 'CI' { $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; $_ } | & $echoArgs | Should Be $null + 1..3 | % { + Start-Sleep -Milliseconds 100 + # yeild some multi-line formatted object + @{'a' = 'b'} + } | & $echoArgs | Should Be $null } } From 89782f92d2ce20143d77d5465c06dad23cf64b00 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Wed, 26 Oct 2016 19:06:48 -0700 Subject: [PATCH 10/20] Clean-up unnesesary fields from NativeCommandProcessor --- .../engine/NativeCommandProcessor.cs | 116 ++++++++---------- 1 file changed, 51 insertions(+), 65 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 6015f8191a1..fd9a62efa52 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -142,7 +142,7 @@ internal ProcessOutputObject(object data, MinishellStream stream) /// internal class NativeCommandProcessor : CommandProcessorBase { - private static string XmlCliTag = "#< CLIXML"; + private const string XmlCliTag = "#< CLIXML"; #region ctor/native command properties @@ -347,9 +347,7 @@ internal override void ProcessRecord() /// 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 _background; - - private ProcessStartInfo _startInfo; + private bool _isRunningInBackground; /// /// This output queue helps us keep the output and error (if redirected) order correct. @@ -360,16 +358,8 @@ internal override void ProcessRecord() private bool _scrapeHostOutput; - private bool _redirectInput; - private Host.Coordinates _startPosition; - /// - /// 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... - /// - private Exception _exceptionToRethrow = null; - /// /// object used for synchronization between StopProcessing thread and /// Pipeline thread. @@ -394,14 +384,15 @@ private void InitNativeProcess() //Calculate if input and output are redirected. bool redirectOutput; bool redirectError; + bool redirectInput; - CalculateIORedirection(out redirectOutput, out redirectError, out _redirectInput); + CalculateIORedirection(out redirectOutput, out redirectError, out redirectInput); // Find out if it's the only command in the pipeline. bool soloCommand = this.Command.MyInvocation.PipelineLength == 1; // Get the start info for the process. - _startInfo = GetProcessStartInfo(redirectOutput, redirectError, _redirectInput, soloCommand); + ProcessStartInfo startInfo = GetProcessStartInfo(redirectOutput, redirectError, redirectInput, soloCommand); if (this.Command.Context.CurrentPipelineStopping) { @@ -411,6 +402,7 @@ private void InitNativeProcess() _startPosition = new Host.Coordinates(); _scrapeHostOutput = false; + Exception exceptionToRethrow = null; try { // If this process is being run standalone, tell the host, which might want @@ -454,7 +446,7 @@ private void InitNativeProcess() try { _nativeProcess = new Process(); - _nativeProcess.StartInfo = _startInfo; + _nativeProcess.StartInfo = startInfo; _nativeProcess.Start(); } catch (Win32Exception) @@ -466,7 +458,7 @@ private void InitNativeProcess() // See if there is a file association for this command. If so // then we'll use that. If there's no file association, then // try shell execute... - string executable = FindExecutable(_startInfo.FileName); + string executable = FindExecutable(startInfo.FileName); bool notDone = true; if (!String.IsNullOrEmpty(executable)) { @@ -476,10 +468,10 @@ private void InitNativeProcess() ConsoleVisibility.AllocateHiddenConsole(); } - string oldArguments = _startInfo.Arguments; - string oldFileName = _startInfo.FileName; - _startInfo.Arguments = "\"" + _startInfo.FileName + "\" " + _startInfo.Arguments; - _startInfo.FileName = executable; + string oldArguments = startInfo.Arguments; + string oldFileName = startInfo.FileName; + startInfo.Arguments = "\"" + startInfo.FileName + "\" " + startInfo.Arguments; + startInfo.FileName = executable; try { _nativeProcess.Start(); @@ -488,8 +480,8 @@ private void InitNativeProcess() catch (Win32Exception) { // Restore the old filename and arguments to try shell execute last... - _startInfo.Arguments = oldArguments; - _startInfo.FileName = oldFileName; + startInfo.Arguments = oldArguments; + startInfo.FileName = oldFileName; } } // We got here because there was either no executable found for this @@ -497,12 +489,12 @@ private void InitNativeProcess() // we will try launching one last time using ShellExecute... if (notDone) { - if (soloCommand && _startInfo.UseShellExecute == false) + if (soloCommand && startInfo.UseShellExecute == false) { - _startInfo.UseShellExecute = true; - _startInfo.RedirectStandardInput = false; - _startInfo.RedirectStandardOutput = false; - _startInfo.RedirectStandardError = false; + startInfo.UseShellExecute = true; + startInfo.RedirectStandardInput = false; + startInfo.RedirectStandardOutput = false; + startInfo.RedirectStandardError = false; _nativeProcess.Start(); } else @@ -520,21 +512,21 @@ private void InitNativeProcess() // Something like // ls | notepad | sort.exe // should block until the notepad process is terminated. - _background = false; + _isRunningInBackground = false; } else { - _background = true; - if (_startInfo.UseShellExecute == false) + _isRunningInBackground = true; + if (startInfo.UseShellExecute == false) { - _background = IsWindowsApplication(_nativeProcess.StartInfo.FileName); + _isRunningInBackground = IsWindowsApplication(_nativeProcess.StartInfo.FileName); } } try { //If input is redirected, start input to process. - if (_startInfo.RedirectStandardInput) + if (startInfo.RedirectStandardInput) { NativeCommandIOFormat inputFormat = NativeCommandIOFormat.Text; if (_isMiniShell) @@ -556,14 +548,14 @@ private void InitNativeProcess() throw; } - if (_background == false) + if (_isRunningInBackground == false) { InitOutputQueue(); } } catch (Win32Exception e) { - _exceptionToRethrow = e; + exceptionToRethrow = e; } // try catch (PipelineStoppedException) @@ -575,18 +567,18 @@ private void InitNativeProcess() { CommandProcessorBase.CheckForSevereException(e); - _exceptionToRethrow = e; + exceptionToRethrow = e; } // An exception was thrown while attempting to run the program // so wrap and rethrow it here... - if (_exceptionToRethrow != null) + 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.NativeCommandName, exceptionToRethrow.Message, this.Command.MyInvocation.PositionMessage); - ApplicationFailedException appFailedException = new ApplicationFailedException(message, _exceptionToRethrow); + 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; @@ -713,7 +705,7 @@ private List DeserializeCliXmlObject(string xml, bool isOut private void InitOutputQueue() { //if output is redirected, start reading output of process in queue. - if (_startInfo.RedirectStandardOutput || _startInfo.RedirectStandardError) + if (_nativeProcess.StartInfo.RedirectStandardOutput || _nativeProcess.StartInfo.RedirectStandardError) { lock (_sync) { @@ -721,7 +713,7 @@ private void InitOutputQueue() { _nativeProcessOutputQueue = new ConcurrentQueue(); - if (_startInfo.RedirectStandardOutput) + if (_nativeProcess.StartInfo.RedirectStandardOutput) { bool isFirstOutput = true; bool isXmlCliOutput = false; @@ -755,7 +747,7 @@ private void InitOutputQueue() _nativeProcess.BeginOutputReadLine(); } - if (_startInfo.RedirectStandardError) + if (_nativeProcess.StartInfo.RedirectStandardError) { bool isFirstError = true; bool isXmlCliError = false; @@ -805,32 +797,33 @@ private void InitOutputQueue() /// /// Read the output from the native process and send it down the line. /// - /// - /// True if there was any new input available, otherwise false. - /// - private bool ConsumeAvailableNativeProcessOutput() + private void ConsumeAvailableNativeProcessOutput() { - bool isNewData = false; - if (_background == false) + if (_isRunningInBackground == false) { - if (_startInfo.RedirectStandardOutput || _startInfo.RedirectStandardError) + if (_nativeProcess.StartInfo.RedirectStandardOutput || _nativeProcess.StartInfo.RedirectStandardError) { ProcessOutputObject record; while (_nativeProcessOutputQueue.TryDequeue(out record)) { + if (this.Command.Context.CurrentPipelineStopping) + { + this.StopProcessing(); + return; + } + ProcessOutputRecord(record); - isNewData = true; } } } - return isNewData; } internal override void Complete() { + Exception exceptionToRethrow = null; try { - if (_background == false) + if (_isRunningInBackground == false) { //Wait for input writer to finish. _inputWriter.Done(); @@ -892,7 +885,7 @@ internal override void Complete() } catch (Win32Exception e) { - _exceptionToRethrow = e; + exceptionToRethrow = e; } // try catch (PipelineStoppedException) { @@ -903,11 +896,11 @@ internal override void Complete() { CommandProcessorBase.CheckForSevereException(e); - _exceptionToRethrow = e; + exceptionToRethrow = e; } finally { - if (!_startInfo.RedirectStandardOutput) + if (!_nativeProcess.StartInfo.RedirectStandardOutput) { this.Command.Context.EngineHostInterface.NotifyEndApplication(); } @@ -917,13 +910,13 @@ internal override void Complete() // An exception was thrown while attempting to run the program // so wrap and rethrow it here... - if (_exceptionToRethrow != null) + 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.NativeCommandName, exceptionToRethrow.Message, this.Command.MyInvocation.PositionMessage); - ApplicationFailedException appFailedException = new ApplicationFailedException(message, _exceptionToRethrow); + 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; @@ -1197,7 +1190,6 @@ private void CleanUp() private void ProcessOutputRecord(ProcessOutputObject outputValue) { Dbg.Assert(outputValue != null, "only object of type ProcessOutputObject expected"); - if (outputValue.Stream == MinishellStream.Error) { @@ -1257,12 +1249,6 @@ private void ProcessOutputRecord(ProcessOutputObject outputValue) Dbg.Assert(record != null, "ProcessReader should ensure that data is InformationRecord"); this.commandRuntime.WriteInformation(record); } - - if (this.Command.Context.CurrentPipelineStopping) - { - this.StopProcessing(); - return; - } } /// @@ -1367,7 +1353,7 @@ private bool IsDownstreamOutDefault(Pipe downstreamPipe) /// private void CalculateIORedirection(out bool redirectOutput, out bool redirectError, out bool redirectInput) { - redirectInput = this.Command.MyInvocation.PipelineLength > 0; + redirectInput = this.Command.MyInvocation.PipelinePosition > 0; redirectOutput = true; redirectError = true; From f0a4a7d8213a5479fe6ac79ed616639228780c8f Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Mon, 7 Nov 2016 18:38:25 -0800 Subject: [PATCH 11/20] Replace spin-lock by a proper BlockingCollection in NativeCommandProcessor --- .../engine/NativeCommandProcessor.cs | 413 +++++++++++------- .../NativeExecution/NativeMinishell.Tests.ps1 | 2 +- 2 files changed, 251 insertions(+), 164 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index fd9a62efa52..41d1a31fe85 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -324,7 +324,7 @@ internal override void ProcessRecord() _inputWriter.Add(Command.CurrentPipelineObject); } - ConsumeAvailableNativeProcessOutput(); + ConsumeAvailableNativeProcessOutput(blocking: false); } /// @@ -354,7 +354,7 @@ internal override void ProcessRecord() /// We could do a blocking read in the Complete block instead, /// but then we would not be able to restore the order reasonable. /// - private ConcurrentQueue _nativeProcessOutputQueue; + private BlockingCollection _nativeProcessOutputQueue; private bool _scrapeHostOutput; @@ -587,119 +587,242 @@ private void InitNativeProcess() } } - private List DeserializeCliXmlObject(string xml, bool isOutput) + private class ProcessOutputHandler { - var result = new List(); - try + private int _refCount; + private BlockingCollection _queue; + private bool _isFirstOutput; + private bool _isFirstError; + private bool _isXmlCliOutput; + private bool _isXmlCliError; + private string _path; + + public ProcessOutputHandler(Process process, BlockingCollection queue) + { + Debug.Assert(process.StartInfo.RedirectStandardOutput || process.StartInfo.RedirectStandardError, "Caller should redirect at least one stream"); + _refCount = 0; + _path = process.StartInfo.FileName; + _queue = queue; + + // 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) + { + _isFirstOutput = true; + _isXmlCliOutput = false; + process.OutputDataReceived += new DataReceivedEventHandler(OutputHandler); + process.BeginOutputReadLine(); + } + + if (process.StartInfo.RedirectStandardError) + { + _isFirstError = true; + _isXmlCliError = false; + process.ErrorDataReceived += new DataReceivedEventHandler(ErrorHandler); + process.BeginErrorReadLine(); + } + } + + private void decrementRefCount() { - using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml))) + Debug.Assert(_refCount > 0, "RefCount should always be positive, when we are trying to decrement it"); + if (Interlocked.Decrement(ref _refCount) == 0) { - XmlReader xmlReader = XmlReader.Create(streamReader, InternalDeserializer.XmlReaderSettingsForCliXml); - Deserializer des = new Deserializer(xmlReader); - while (!des.Done()) - { - string streamName; - object obj = des.Deserialize(out streamName); + _queue.CompleteAdding(); + } + } - //Decide the stream to which data belongs - MinishellStream stream = MinishellStream.Unknown; - if (streamName != null) + private void OutputHandler(object sender, DataReceivedEventArgs e) + { + if (e.Data != null) + { + if (_isFirstOutput) + { + _isFirstOutput = false; + if (string.Equals(e.Data, XmlCliTag, StringComparison.Ordinal)) { - stream = StringToMinishellStreamConverter.ToMinishellStream(streamName); + _isXmlCliOutput = true; + return; } - if (stream == MinishellStream.Unknown) + } + + if (_isXmlCliOutput) + { + foreach (var record in DeserializeCliXmlObject(e.Data, true)) { - stream = isOutput ? MinishellStream.Output : MinishellStream.Error; + _queue.Add(record); } + } + else + { + _queue.Add(new ProcessOutputObject(e.Data, MinishellStream.Output)); + } + } + else + { + decrementRefCount(); + } + } - //Null is allowed only in output stream - if (stream != MinishellStream.Output && obj == null) + private void ErrorHandler(object sender, DataReceivedEventArgs e) + { + if (e.Data != null) + { + if (string.Equals(e.Data, XmlCliTag, StringComparison.Ordinal)) + { + _isXmlCliError = true; + return; + } + + if (_isXmlCliError) + { + foreach (var record in DeserializeCliXmlObject(e.Data, false)) { - continue; + _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(e.Data), "NativeCommandError", ErrorCategory.NotSpecified, e.Data); + } + else + { + // Wrap the rest of the output in ErrorRecords with the "NativeCommandErrorMessage" error ID + errorRecord = new ErrorRecord(new RemoteException(e.Data), "NativeCommandErrorMessage", ErrorCategory.NotSpecified, null); } - if (stream == MinishellStream.Error) + _queue.Add(new ProcessOutputObject(errorRecord, MinishellStream.Error)); + } + } + else + { + decrementRefCount(); + } + } + + private List DeserializeCliXmlObject(string xml, bool isOutput) + { + var result = new List(); + try + { + using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml))) + { + XmlReader xmlReader = XmlReader.Create(streamReader, InternalDeserializer.XmlReaderSettingsForCliXml); + Deserializer des = new Deserializer(xmlReader); + while (!des.Done()) { - if (obj is PSObject) + string streamName; + object obj = des.Deserialize(out streamName); + + //Decide the stream to which data belongs + MinishellStream stream = MinishellStream.Unknown; + if (streamName != null) { - obj = ErrorRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); + stream = StringToMinishellStreamConverter.ToMinishellStream(streamName); } - else + if (stream == MinishellStream.Unknown) { - string errorMessage = null; - try + 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) { - errorMessage = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); + obj = ErrorRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); } - catch (PSInvalidCastException) + else { - continue; + 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); } - obj = new ErrorRecord(new RemoteException(errorMessage), - "NativeCommandError", ErrorCategory.NotSpecified, errorMessage); } - } - else if (stream == MinishellStream.Information) - { - if (obj is PSObject) + else if (stream == MinishellStream.Information) { - obj = InformationRecord.FromPSObjectForRemoting(PSObject.AsPSObject(obj)); + 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 + else if (stream == MinishellStream.Debug || + stream == MinishellStream.Verbose || + stream == MinishellStream.Warning) { - string messageData = null; + //Convert to string try { - messageData = (string)LanguagePrimitives.ConvertTo(obj, typeof(string), CultureInfo.InvariantCulture); + obj = 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)); } - result.Add(new ProcessOutputObject(obj, stream)); } } - } - catch (XmlException originalException) - { - string template = NativeCP.CliXmlError; - string message = string.Format( - null, - template, - isOutput ? MinishellStream.Output : MinishellStream.Error, - Path, - originalException.Message); - XmlException newException = new XmlException( - message, - originalException); + catch (XmlException originalException) + { + string template = NativeCP.CliXmlError; + string message = string.Format( + null, + template, + isOutput ? MinishellStream.Output : MinishellStream.Error, + _path, + originalException.Message); + XmlException newException = new XmlException( + message, + originalException); + + ErrorRecord error = new ErrorRecord( + newException, + "ProcessStreamReader_CliXmlError", + ErrorCategory.SyntaxError, + _path); + result.Add(new ProcessOutputObject(error, MinishellStream.Error)); + } - ErrorRecord error = new ErrorRecord( - newException, - "ProcessStreamReader_CliXmlError", - ErrorCategory.SyntaxError, - Path); - result.Add(new ProcessOutputObject(error, MinishellStream.Error)); + return result; } - - return result; } private void InitOutputQueue() @@ -711,100 +834,64 @@ private void InitOutputQueue() { if (!_stopped) { - _nativeProcessOutputQueue = new ConcurrentQueue(); - - if (_nativeProcess.StartInfo.RedirectStandardOutput) - { - bool isFirstOutput = true; - bool isXmlCliOutput = false; - _nativeProcess.OutputDataReceived += new DataReceivedEventHandler((sender, e) => - { - if (e.Data != null) - { - if (isFirstOutput) - { - isFirstOutput = false; - if (e.Data == XmlCliTag) - { - isXmlCliOutput = true; - return; - } - } - - if (isXmlCliOutput) - { - foreach (var record in DeserializeCliXmlObject(e.Data, true)) - { - _nativeProcessOutputQueue.Enqueue(record); - } - } - else - { - _nativeProcessOutputQueue.Enqueue(new ProcessOutputObject(e.Data, MinishellStream.Output)); - } - } - }); - _nativeProcess.BeginOutputReadLine(); - } - - if (_nativeProcess.StartInfo.RedirectStandardError) - { - bool isFirstError = true; - bool isXmlCliError = false; - _nativeProcess.ErrorDataReceived += new DataReceivedEventHandler((sender, e) => - { - if (e.Data != null) - { - if (e.Data == XmlCliTag) - { - isXmlCliError = true; - return; - } + _nativeProcessOutputQueue = new BlockingCollection(); + // we don't assign the handler to anything, because it's used only for objects marshaling + new ProcessOutputHandler(_nativeProcess, _nativeProcessOutputQueue); + } + } + } + } - if (isXmlCliError) - { - foreach (var record in DeserializeCliXmlObject(e.Data, false)) - { - _nativeProcessOutputQueue.Enqueue(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(e.Data), "NativeCommandError", ErrorCategory.NotSpecified, e.Data); - } - else - { - // Wrap the rest of the output in ErrorRecords with the "NativeCommandErrorMessage" error ID - errorRecord = new ErrorRecord(new RemoteException(e.Data), "NativeCommandErrorMessage", ErrorCategory.NotSpecified, null); - } - - _nativeProcessOutputQueue.Enqueue(new ProcessOutputObject(errorRecord, MinishellStream.Error)); - } - } - }); - _nativeProcess.BeginErrorReadLine(); - } + private ProcessOutputObject DequeueProcessOutput(bool blocking) + { + if (blocking) + { + // if adding was completed, there is no need to do a blocking Take(), + // if collection is empty + if (_nativeProcessOutputQueue.IsAddingCompleted) + { + if (_nativeProcessOutputQueue.Count > 0) + { + return _nativeProcessOutputQueue.Take(); } } + else + { + try + { + return _nativeProcessOutputQueue.Take(); + } + catch (InvalidOperationException) + { + // 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. + } + } + + // collection is empty or exception been raised + return null; + } + else + { + ProcessOutputObject record = null; + _nativeProcessOutputQueue.TryTake(out record); + return record; } } /// /// Read the output from the native process and send it down the line. /// - private void ConsumeAvailableNativeProcessOutput() + private void ConsumeAvailableNativeProcessOutput(bool blocking) { if (_isRunningInBackground == false) { if (_nativeProcess.StartInfo.RedirectStandardOutput || _nativeProcess.StartInfo.RedirectStandardError) { ProcessOutputObject record; - while (_nativeProcessOutputQueue.TryDequeue(out record)) + while ((record = DequeueProcessOutput(blocking)) != null) { if (this.Command.Context.CurrentPipelineStopping) { @@ -831,14 +918,12 @@ internal override void Complete() //Wait for the process to exit and consume available output while (!_nativeProcess.HasExited) { - // TODO: Currently the main pipeline Thread is just spinning. - // It would be much better to park it until new input is available. - ConsumeAvailableNativeProcessOutput(); + ConsumeAvailableNativeProcessOutput(blocking: true); } // read all the available output one more time _nativeProcess.WaitForExit(); - ConsumeAvailableNativeProcessOutput(); + ConsumeAvailableNativeProcessOutput(blocking: true); // Capture screen output if we are transcribing if (this.Command.Context.EngineHostInterface.UI.IsTranscribing && @@ -1695,9 +1780,11 @@ internal void Stop() internal void Done() { // we allow call Done() multiply times. - // For example one time from Process() code path, + // For example one time from ProcessRecord() code path, // when we detect that process already finished - // and once from End() code path. + // and once from Complete() code path. + // Even though Done() could be called multiple times, + // the calls are on the same thread, so there is no race condition. if (_pipeline != null) { _pipeline.End(); diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 index eee6bc8844f..db008565377 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 @@ -1,5 +1,5 @@ # Minishell is a powershell concept. -# It's primare use-case is when somebody executes a scriptblock in the new powershell process. +# Its primare use-case is when somebody executes a scriptblock in the new powershell process. # The objects are automatically marshelled back to the parent session, so users can avoid custom # serialization to pass objects between two processes. From c0bda7a55a2d795318ca8c0f2aa053badc9b4993 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Mon, 7 Nov 2016 19:01:32 -0800 Subject: [PATCH 12/20] Add native pipelines entry to changelog.md --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) 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 From 616c15556f44bd20d01303d92225c103aef805e8 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 11 Nov 2016 11:53:40 -0800 Subject: [PATCH 13/20] Re-shuffle minishell tests --- test/powershell/Host/ConsoleHost.Tests.ps1 | 56 ++++++++++++++++--- .../NativeCommandProcessor.Tests.ps1 | 21 +++++++ .../NativeExecution/NativeMinishell.Tests.ps1 | 50 ----------------- 3 files changed, 68 insertions(+), 59 deletions(-) delete mode 100644 test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 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..d87bdc99a5d 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 'CI' { + + BeforeAll { + $powershell = Join-Path -Path $PsHome -ChildPath "powershell" + } + + It "native | ps | native doesn't block" { + $first = $true + & $powershell -command '1..5 | % {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/NativeMinishell.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 deleted file mode 100644 index db008565377..00000000000 --- a/test/powershell/Language/Scripting/NativeExecution/NativeMinishell.Tests.ps1 +++ /dev/null @@ -1,50 +0,0 @@ -# Minishell is a powershell concept. -# Its primare use-case is when somebody executes a scriptblock in the new powershell process. -# The objects are automatically marshelled 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' { - - It 'gets a hashtable object from minishell' { - $output = & $powershell { @{'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 { 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 { Write-Information 'foo' } 6>&1 - ($output.GetType().Name) | Should Be 'InformationRecord' - $output | Should Be 'foo' - } - } - - Context 'native commands lifecycle' { - It "native | ps | native doesn't block" { - $first = $true - & $powershell -command '1..5 | % {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 - } - } -} From e038360087064334c89741665a9ae02adedcb62e Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 11 Nov 2016 13:20:47 -0800 Subject: [PATCH 14/20] Prettify NativeCommmandProcessor.cs - Move ProcessOutputHandler outside of NativeCommmandProcessor - Remove redundent ConsumeAvailableNativeProcessOutput() blocking call --- .../engine/NativeCommandProcessor.cs | 501 +++++++++--------- 1 file changed, 250 insertions(+), 251 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 41d1a31fe85..0180825e7d5 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -142,8 +142,6 @@ internal ProcessOutputObject(object data, MinishellStream stream) /// internal class NativeCommandProcessor : CommandProcessorBase { - private const string XmlCliTag = "#< CLIXML"; - #region ctor/native command properties /// @@ -587,244 +585,6 @@ private void InitNativeProcess() } } - private class ProcessOutputHandler - { - private int _refCount; - private BlockingCollection _queue; - private bool _isFirstOutput; - private bool _isFirstError; - private bool _isXmlCliOutput; - private bool _isXmlCliError; - private string _path; - - public ProcessOutputHandler(Process process, BlockingCollection queue) - { - Debug.Assert(process.StartInfo.RedirectStandardOutput || process.StartInfo.RedirectStandardError, "Caller should redirect at least one stream"); - _refCount = 0; - _path = process.StartInfo.FileName; - _queue = queue; - - // 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) - { - _isFirstOutput = true; - _isXmlCliOutput = false; - process.OutputDataReceived += new DataReceivedEventHandler(OutputHandler); - process.BeginOutputReadLine(); - } - - if (process.StartInfo.RedirectStandardError) - { - _isFirstError = true; - _isXmlCliError = false; - process.ErrorDataReceived += new DataReceivedEventHandler(ErrorHandler); - process.BeginErrorReadLine(); - } - } - - private void decrementRefCount() - { - Debug.Assert(_refCount > 0, "RefCount should always be positive, when we are trying to decrement it"); - if (Interlocked.Decrement(ref _refCount) == 0) - { - _queue.CompleteAdding(); - } - } - - private void OutputHandler(object sender, DataReceivedEventArgs e) - { - if (e.Data != null) - { - if (_isFirstOutput) - { - _isFirstOutput = false; - if (string.Equals(e.Data, XmlCliTag, StringComparison.Ordinal)) - { - _isXmlCliOutput = true; - return; - } - } - - if (_isXmlCliOutput) - { - foreach (var record in DeserializeCliXmlObject(e.Data, true)) - { - _queue.Add(record); - } - } - else - { - _queue.Add(new ProcessOutputObject(e.Data, MinishellStream.Output)); - } - } - else - { - decrementRefCount(); - } - } - - private void ErrorHandler(object sender, DataReceivedEventArgs e) - { - if (e.Data != null) - { - if (string.Equals(e.Data, XmlCliTag, StringComparison.Ordinal)) - { - _isXmlCliError = true; - return; - } - - if (_isXmlCliError) - { - foreach (var record in DeserializeCliXmlObject(e.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(e.Data), "NativeCommandError", ErrorCategory.NotSpecified, e.Data); - } - else - { - // Wrap the rest of the output in ErrorRecords with the "NativeCommandErrorMessage" error ID - errorRecord = new ErrorRecord(new RemoteException(e.Data), "NativeCommandErrorMessage", ErrorCategory.NotSpecified, null); - } - - _queue.Add(new ProcessOutputObject(errorRecord, MinishellStream.Error)); - } - } - else - { - decrementRefCount(); - } - } - - private List DeserializeCliXmlObject(string xml, bool isOutput) - { - var result = new List(); - try - { - using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml))) - { - 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)); - } - } - } - catch (XmlException originalException) - { - string template = NativeCP.CliXmlError; - string message = string.Format( - null, - template, - isOutput ? MinishellStream.Output : MinishellStream.Error, - _path, - originalException.Message); - XmlException newException = new XmlException( - message, - originalException); - - ErrorRecord error = new ErrorRecord( - newException, - "ProcessStreamReader_CliXmlError", - ErrorCategory.SyntaxError, - _path); - result.Add(new ProcessOutputObject(error, MinishellStream.Error)); - } - - return result; - } - } - private void InitOutputQueue() { //if output is redirected, start reading output of process in queue. @@ -846,12 +606,15 @@ private ProcessOutputObject DequeueProcessOutput(bool blocking) { if (blocking) { - // if adding was completed, there is no need to do a blocking Take(), - // if collection is empty + // If adding was completed and collection is empty, + // there is no need to do a blocking Take(). if (_nativeProcessOutputQueue.IsAddingCompleted) { if (_nativeProcessOutputQueue.Count > 0) { + // This is a common codepath and although it has a duplicated code, + // we are keeping it outside of try {} catch {} to improve the perf + // for the common case. return _nativeProcessOutputQueue.Take(); } } @@ -859,6 +622,8 @@ private ProcessOutputObject DequeueProcessOutput(bool blocking) { try { + // If adding is not complete we need a try {} catch {} + // to mitigate a concurrent call to CompleteAdding(). return _nativeProcessOutputQueue.Take(); } catch (InvalidOperationException) @@ -870,7 +635,7 @@ private ProcessOutputObject DequeueProcessOutput(bool blocking) } } - // collection is empty or exception been raised + // collection is empty return null; } else @@ -915,15 +680,9 @@ internal override void Complete() //Wait for input writer to finish. _inputWriter.Done(); - //Wait for the process to exit and consume available output - while (!_nativeProcess.HasExited) - { - ConsumeAvailableNativeProcessOutput(blocking: true); - } - - // read all the available output one more time - _nativeProcess.WaitForExit(); + // read all the available output in the blocking way ConsumeAvailableNativeProcessOutput(blocking: true); + _nativeProcess.WaitForExit(); // Capture screen output if we are transcribing if (this.Command.Context.EngineHostInterface.UI.IsTranscribing && @@ -1672,6 +1431,246 @@ private bool IsMiniShell() internal static bool IsServerSide { get; set; } } + internal class ProcessOutputHandler + { + private const string XmlCliTag = "#< CLIXML"; + + private int _refCount; + private BlockingCollection _queue; + private bool _isFirstOutput; + private bool _isFirstError; + private bool _isXmlCliOutput; + private bool _isXmlCliError; + private string _processFileName; + + public ProcessOutputHandler(Process process, BlockingCollection queue) + { + Debug.Assert(process.StartInfo.RedirectStandardOutput || process.StartInfo.RedirectStandardError, "Caller should redirect at least one stream"); + _refCount = 0; + _processFileName = process.StartInfo.FileName; + _queue = queue; + + // 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) + { + _isFirstOutput = true; + _isXmlCliOutput = false; + process.OutputDataReceived += new DataReceivedEventHandler(OutputHandler); + process.BeginOutputReadLine(); + } + + if (process.StartInfo.RedirectStandardError) + { + _isFirstError = true; + _isXmlCliError = false; + process.ErrorDataReceived += new DataReceivedEventHandler(ErrorHandler); + process.BeginErrorReadLine(); + } + } + + private void decrementRefCount() + { + Debug.Assert(_refCount > 0, "RefCount should always be positive, when we are trying to decrement it"); + if (Interlocked.Decrement(ref _refCount) == 0) + { + _queue.CompleteAdding(); + } + } + + private void OutputHandler(object sender, DataReceivedEventArgs outputReceived) + { + if (outputReceived.Data != null) + { + if (_isFirstOutput) + { + _isFirstOutput = false; + if (string.Equals(outputReceived.Data, XmlCliTag, StringComparison.Ordinal)) + { + _isXmlCliOutput = true; + return; + } + } + + if (_isXmlCliOutput) + { + foreach (var record in DeserializeCliXmlObject(outputReceived.Data, true)) + { + _queue.Add(record); + } + } + else + { + _queue.Add(new ProcessOutputObject(outputReceived.Data, MinishellStream.Output)); + } + } + else + { + decrementRefCount(); + } + } + + private void ErrorHandler(object sender, DataReceivedEventArgs errorReceived) + { + if (errorReceived.Data != null) + { + if (string.Equals(errorReceived.Data, XmlCliTag, StringComparison.Ordinal)) + { + _isXmlCliError = true; + return; + } + + 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 + { + decrementRefCount(); + } + } + + private List DeserializeCliXmlObject(string xml, bool isOutput) + { + var result = new List(); + try + { + using (var streamReader = new MemoryStream(Encoding.UTF8.GetBytes(xml))) + { + 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)); + } + } + } + catch (XmlException originalException) + { + string template = NativeCP.CliXmlError; + string message = string.Format( + null, + template, + isOutput ? MinishellStream.Output : MinishellStream.Error, + _processFileName, + originalException.Message); + XmlException newException = new XmlException( + message, + originalException); + + ErrorRecord error = new ErrorRecord( + newException, + "ProcessStreamReader_CliXmlError", + ErrorCategory.SyntaxError, + _processFileName); + result.Add(new ProcessOutputObject(error, MinishellStream.Error)); + } + + return result; + } + } + /// /// Helper class to handle writing input to a process. /// From 0ce50a82a76d57419e7594383c72f12441f8d61f Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 11 Nov 2016 13:43:50 -0800 Subject: [PATCH 15/20] Resurect Xml serialization logic inside ProcessInputWriter --- .../engine/NativeCommandProcessor.cs | 49 +++++++++++++++++-- 1 file changed, 45 insertions(+), 4 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 0180825e7d5..0924661bb55 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1433,7 +1433,7 @@ private bool IsMiniShell() internal class ProcessOutputHandler { - private const string XmlCliTag = "#< CLIXML"; + internal const string XmlCliTag = "#< CLIXML"; private int _refCount; private BlockingCollection _queue; @@ -1686,14 +1686,12 @@ internal ProcessInputWriter(InternalCommand command) { Dbg.Assert(command != null, "Caller should validate the parameter"); _command = command; - - _pipeline = ScriptBlock.Create("Out-String -Stream").GetSteppablePipeline(); - _pipeline.Begin(true); } #endregion constructor private SteppablePipeline _pipeline; + private Serializer _xmlSerializer; /// /// Add an object to write to process @@ -1708,6 +1706,18 @@ internal void Add(object input) return; } + if (_inputFormat == NativeCommandIOFormat.Text) + { + AddTextInput(input); + } + else // Xml + { + AddXmlInput(input); + } + } + + private void AddTextInput(object input) + { Array formattedObjects = _pipeline.Process(input); foreach (var item in formattedObjects) { @@ -1729,6 +1739,20 @@ internal void Add(object input) } } + private void AddXmlInput(object input) + { + try + { + _xmlSerializer.Serialize(input); + } + catch (IOException) + { + // we are assuming that process is already finished + // we should just stop processing at this point + this.Done(); + } + } + /// /// Stream to which input is written /// @@ -1761,6 +1785,17 @@ internal void Start(Process process, NativeCommandIOFormat inputFormat) _streamWriter.AutoFlush = true; _inputFormat = inputFormat; + + if (_inputFormat == NativeCommandIOFormat.Xml) + { + _streamWriter.WriteLine(ProcessOutputHandler.XmlCliTag); + _xmlSerializer = new Serializer(XmlWriter.Create(_streamWriter)); + } + else // Text + { + _pipeline = ScriptBlock.Create("Out-String -Stream").GetSteppablePipeline(); + _pipeline.Begin(true); + } } bool _stopping = false; @@ -1791,6 +1826,12 @@ internal void Done() _pipeline = null; } + if (_xmlSerializer != null) + { + _xmlSerializer.Done(); + _xmlSerializer = null; + } + // streamWriter is present, only if we call Start method if (_streamWriter != null) { From 7914c75504e5c5974cd5b0f9a182acfbc14aa042 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 11 Nov 2016 14:23:29 -0800 Subject: [PATCH 16/20] Fix bug in native pipe, where we loose End() output --- .../engine/NativeCommandProcessor.cs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 0924661bb55..73c14574284 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1718,7 +1718,11 @@ internal void Add(object input) private void AddTextInput(object input) { - Array formattedObjects = _pipeline.Process(input); + AddTextInputFromFormattedArray(_pipeline.Process(input)); + } + + private void AddTextInputFromFormattedArray(Array formattedObjects) + { foreach (var item in formattedObjects) { string line = PSObject.ToStringParser(_command.Context, item); @@ -1821,7 +1825,7 @@ internal void Done() // the calls are on the same thread, so there is no race condition. if (_pipeline != null) { - _pipeline.End(); + AddTextInputFromFormattedArray(_pipeline.End()); _pipeline.Dispose(); _pipeline = null; } From 1508526a3fb3135073f8dfe1817a70f5b94d3928 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 11 Nov 2016 15:46:58 -0800 Subject: [PATCH 17/20] Fix recursive Done() problem in steppable pipeline finalization --- .../engine/NativeCommandProcessor.cs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 73c14574284..690cb829c26 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1825,9 +1825,14 @@ internal void Done() // the calls are on the same thread, so there is no race condition. if (_pipeline != null) { - AddTextInputFromFormattedArray(_pipeline.End()); + var finalResults = _pipeline.End(); _pipeline.Dispose(); _pipeline = null; + // AddTextInputFromFormattedArray can recursively call Done(), + // if the downstream process already exited. + // to Prevent it, we first finalize the pipeline and set it to null, + // then calling the result processing for the last time + AddTextInputFromFormattedArray(finalResults); } if (_xmlSerializer != null) From 32837f06ee105674f078c2e80b5b4d0dfaf87382 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Fri, 11 Nov 2016 16:42:23 -0800 Subject: [PATCH 18/20] Trying to make 'native command lifecycle' test unflaky Also move it to the Feature category --- .../NativeExecution/NativeCommandProcessor.Tests.ps1 | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 b/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 index d87bdc99a5d..286c230c5ca 100644 --- a/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 +++ b/test/powershell/Language/Scripting/NativeExecution/NativeCommandProcessor.Tests.ps1 @@ -1,4 +1,4 @@ -Describe 'native commands lifecycle' -tags 'CI' { +Describe 'native commands lifecycle' -tags 'Feature' { BeforeAll { $powershell = Join-Path -Path $PsHome -ChildPath "powershell" @@ -6,7 +6,7 @@ Describe 'native commands lifecycle' -tags 'CI' { It "native | ps | native doesn't block" { $first = $true - & $powershell -command '1..5 | % {Start-Sleep -mill 100; $_}' | %{$_} | & $powershell -command '$input' | % { + & $powershell -command '1..10 | % {Start-Sleep -mill 100; $_}' | %{$_} | & $powershell -command '$input' | % { if ($first) { $first = $false From 825352faa3653a8b1271aecbfded9e2851730641 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Tue, 15 Nov 2016 15:14:09 -0800 Subject: [PATCH 19/20] Separate Done() and Dispose() in ProcessInputWriter NativeCommandProcessor.cs --- .../engine/NativeCommandProcessor.cs | 51 ++++++++++++------- 1 file changed, 32 insertions(+), 19 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index 690cb829c26..dd1d36e20bd 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -1701,7 +1701,7 @@ internal void Add(object input) { if (_stopping || _streamWriter == null) { - // if _streamWriter is already null, then we already called Done() + // if _streamWriter is already null, then we already called Dispose() // so we should just discard the input. return; } @@ -1736,7 +1736,7 @@ private void AddTextInputFromFormattedArray(Array formattedObjects) { // we are assuming that process is already finished // we should just stop processing at this point - this.Done(); + this.Dispose(); // stop foreach execution break; } @@ -1753,7 +1753,7 @@ private void AddXmlInput(object input) { // we are assuming that process is already finished // we should just stop processing at this point - this.Done(); + this.Dispose(); } } @@ -1812,36 +1812,26 @@ internal void Stop() _stopping = true; } - /// - /// This method wait for writer thread to finish. - /// - internal void Done() + internal void Dispose() { - // we allow call Done() multiply times. + // we allow call Dispose() multiply times. // For example one time from ProcessRecord() code path, // when we detect that process already finished - // and once from Complete() code path. - // Even though Done() could be called multiple times, - // the calls are on the same thread, so there is no race condition. + // 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) { - var finalResults = _pipeline.End(); _pipeline.Dispose(); _pipeline = null; - // AddTextInputFromFormattedArray can recursively call Done(), - // if the downstream process already exited. - // to Prevent it, we first finalize the pipeline and set it to null, - // then calling the result processing for the last time - AddTextInputFromFormattedArray(finalResults); } if (_xmlSerializer != null) { - _xmlSerializer.Done(); _xmlSerializer = null; } - // streamWriter is present, only if we call Start method + // streamWriter can be null if we didn't call Start method if (_streamWriter != null) { try @@ -1857,6 +1847,29 @@ internal void Done() _streamWriter = null; } } + + internal void Done() + { + if (_inputFormat == NativeCommandIOFormat.Xml) + { + if (_xmlSerializer != null) + { + _xmlSerializer.Done(); + } + } + else // Text + { + // if _pipeline == null, we already called Dispose(), + // for example, because downstream process finished + if (_pipeline != null) + { + var finalResults = _pipeline.End(); + AddTextInputFromFormattedArray(finalResults); + } + } + + Dispose(); + } } #if !CORECLR // There is no GUI application on OneCore, so powershell on OneCore should always have a console attached. From 4868cee9cf92315f85b830acfda33625fdd236f3 Mon Sep 17 00:00:00 2001 From: Sergei Vorobev Date: Tue, 15 Nov 2016 17:29:05 -0800 Subject: [PATCH 20/20] Remove over-optimization in NativeCommandProcessor.DequeueProcessOutput --- .../engine/NativeCommandProcessor.cs | 17 +++-------------- 1 file changed, 3 insertions(+), 14 deletions(-) diff --git a/src/System.Management.Automation/engine/NativeCommandProcessor.cs b/src/System.Management.Automation/engine/NativeCommandProcessor.cs index dd1d36e20bd..937134ad567 100644 --- a/src/System.Management.Automation/engine/NativeCommandProcessor.cs +++ b/src/System.Management.Automation/engine/NativeCommandProcessor.cs @@ -606,19 +606,9 @@ private ProcessOutputObject DequeueProcessOutput(bool blocking) { if (blocking) { - // If adding was completed and collection is empty, - // there is no need to do a blocking Take(). - if (_nativeProcessOutputQueue.IsAddingCompleted) - { - if (_nativeProcessOutputQueue.Count > 0) - { - // This is a common codepath and although it has a duplicated code, - // we are keeping it outside of try {} catch {} to improve the perf - // for the common case. - return _nativeProcessOutputQueue.Take(); - } - } - else + // 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) { try { @@ -635,7 +625,6 @@ private ProcessOutputObject DequeueProcessOutput(bool blocking) } } - // collection is empty return null; } else