From ca0980a5dacec194ec15af24ea67de78948f6df6 Mon Sep 17 00:00:00 2001 From: Paul Higinbotham Date: Mon, 16 Dec 2019 15:56:18 -0800 Subject: [PATCH 1/3] Fix key exchange hang for outofproc transports by using separate processing threads for session and command protocol messages. --- .../fanin/OutOfProcTransportManager.cs | 90 +++++++++++++++++-- 1 file changed, 81 insertions(+), 9 deletions(-) diff --git a/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs b/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs index c23b875d5b3..af985f33d63 100644 --- a/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs +++ b/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs @@ -9,6 +9,7 @@ * elevation to support local machine remoting). */ +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Globalization; @@ -467,6 +468,8 @@ internal abstract class OutOfProcessClientSessionTransportManagerBase : BaseClie private OutOfProcessUtils.DataProcessingDelegates _dataProcessingCallbacks; private Dictionary _cmdTransportManagers; private Timer _closeTimeOutTimer; + private readonly BlockingCollection _sessionMessageQueue; + private readonly BlockingCollection _commandMessageQueue; protected OutOfProcessTextWriter stdInWriter; protected PowerShellTraceSource _tracer; @@ -507,6 +510,20 @@ internal OutOfProcessClientSessionTransportManagerBase( // timers initialization _closeTimeOutTimer = new Timer(OnCloseTimeOutTimerElapsed, null, Timeout.Infinite, Timeout.Infinite); + // Session message processing + _sessionMessageQueue = new BlockingCollection(); + var sessionThread = new Thread(ProcessMessageProc); + sessionThread.Name = "SessionMessageProcessing"; + sessionThread.IsBackground = true; + sessionThread.Start(_sessionMessageQueue); + + // Command message processing + _commandMessageQueue = new BlockingCollection(); + var commandThread = new Thread(ProcessMessageProc); + commandThread.Name = "CommandMessageProcessing"; + commandThread.IsBackground = true; + commandThread.Start(_commandMessageQueue); + _tracer = PowerShellTraceSourceFactory.GetTraceSource(); } @@ -601,7 +618,7 @@ internal override BaseClientCommandTransportManager CreateClientCommandTransport } /// - /// Kills the server process and disposes other resources. + /// Terminates the server process and disposes other resources. /// /// internal override void Dispose(bool isDisposing) @@ -611,6 +628,8 @@ internal override void Dispose(bool isDisposing) { _cmdTransportManagers.Clear(); _closeTimeOutTimer.Dispose(); + _sessionMessageQueue.Dispose(); + _commandMessageQueue.Dispose(); } } @@ -663,29 +682,82 @@ private void OnCloseSessionCompleted() { // stop timer _closeTimeOutTimer.Change(Timeout.Infinite, Timeout.Infinite); + + // Stop protocol message processing threads. + _sessionMessageQueue.CompleteAdding(); + _commandMessageQueue.CompleteAdding(); + RaiseCloseCompleted(); CleanupConnection(); } protected abstract void CleanupConnection(); + private void ProcessMessageProc(object state) + { + var messageQueue = state as BlockingCollection; + + try + { + while (true) + { + var data = messageQueue.Take(); + try + { + OutOfProcessUtils.ProcessData(data, _dataProcessingCallbacks); + } + catch (Exception exception) + { + PSRemotingTransportException psrte = + new PSRemotingTransportException(PSRemotingErrorId.IPCErrorProcessingServerData, + RemotingErrorIdStrings.IPCErrorProcessingServerData, + exception.Message); + RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, TransportMethodEnum.ReceiveShellOutputEx)); + } + } + } + catch (InvalidOperationException) + { + // Normal session message processing thread end. + } + } + + private const string GUIDTAG = "PSGuid='"; + private const int GUID_STR_LEN = 36; // GUID string: 32 digits plus 4 dashes + private Guid GetMessageGuid(string data) + { + // Scan data packet for a GUID. + var iTag = data.IndexOf(GUIDTAG, StringComparison.OrdinalIgnoreCase); + if (iTag > -1) + { + try + { + var psGuidString = data.Substring(iTag + GUIDTAG.Length, GUID_STR_LEN); + return new Guid(psGuidString); + } + catch { } + } + + return Guid.Empty; + } + #endregion #region Event Handlers protected void HandleOutputDataReceived(string data) { - try + // Route protocol message based on whether it is a session or command message. + // Session messages have empty Guid values. + if (Guid.Equals(GetMessageGuid(data), Guid.Empty)) { - OutOfProcessUtils.ProcessData(data, _dataProcessingCallbacks); + // Session message + _sessionMessageQueue.Add(data); } - catch (Exception exception) + else { - PSRemotingTransportException psrte = - new PSRemotingTransportException(PSRemotingErrorId.IPCErrorProcessingServerData, - RemotingErrorIdStrings.IPCErrorProcessingServerData, - exception.Message); - RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, TransportMethodEnum.ReceiveShellOutputEx)); + // Command message + _commandMessageQueue.Add(data); } } From 55ab0c176e0b230c5186fe8a7f12cb4df89e63d3 Mon Sep 17 00:00:00 2001 From: Paul Higinbotham Date: Tue, 17 Dec 2019 09:13:17 -0800 Subject: [PATCH 2/3] Address CodeFactor issues --- .../remoting/fanin/OutOfProcTransportManager.cs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs b/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs index af985f33d63..448c277834f 100644 --- a/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs +++ b/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs @@ -464,12 +464,12 @@ internal abstract class OutOfProcessClientSessionTransportManagerBase : BaseClie { #region Data + private readonly BlockingCollection _sessionMessageQueue; + private readonly BlockingCollection _commandMessageQueue; private PrioritySendDataCollection.OnDataAvailableCallback _onDataAvailableToSendCallback; private OutOfProcessUtils.DataProcessingDelegates _dataProcessingCallbacks; private Dictionary _cmdTransportManagers; private Timer _closeTimeOutTimer; - private readonly BlockingCollection _sessionMessageQueue; - private readonly BlockingCollection _commandMessageQueue; protected OutOfProcessTextWriter stdInWriter; protected PowerShellTraceSource _tracer; @@ -709,7 +709,8 @@ private void ProcessMessageProc(object state) catch (Exception exception) { PSRemotingTransportException psrte = - new PSRemotingTransportException(PSRemotingErrorId.IPCErrorProcessingServerData, + new PSRemotingTransportException( + PSRemotingErrorId.IPCErrorProcessingServerData, RemotingErrorIdStrings.IPCErrorProcessingServerData, exception.Message); RaiseErrorHandler(new TransportErrorOccuredEventArgs(psrte, TransportMethodEnum.ReceiveShellOutputEx)); @@ -724,6 +725,7 @@ private void ProcessMessageProc(object state) private const string GUIDTAG = "PSGuid='"; private const int GUID_STR_LEN = 36; // GUID string: 32 digits plus 4 dashes + private Guid GetMessageGuid(string data) { // Scan data packet for a GUID. @@ -735,7 +737,9 @@ private Guid GetMessageGuid(string data) var psGuidString = data.Substring(iTag + GUIDTAG.Length, GUID_STR_LEN); return new Guid(psGuidString); } - catch { } + catch + { + } } return Guid.Empty; From 4863655d409dc50908c5258bc41878deb55adb68 Mon Sep 17 00:00:00 2001 From: Paul Higinbotham Date: Tue, 17 Dec 2019 09:37:41 -0800 Subject: [PATCH 3/3] Add clarifying comment per CR. --- .../engine/remoting/fanin/OutOfProcTransportManager.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs b/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs index 448c277834f..c94f5a74617 100644 --- a/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs +++ b/src/System.Management.Automation/engine/remoting/fanin/OutOfProcTransportManager.cs @@ -728,7 +728,7 @@ private void ProcessMessageProc(object state) private Guid GetMessageGuid(string data) { - // Scan data packet for a GUID. + // Perform quick scan for data packet for a GUID, ignoring any errors. var iTag = data.IndexOf(GUIDTAG, StringComparison.OrdinalIgnoreCase); if (iTag > -1) { @@ -739,6 +739,8 @@ private Guid GetMessageGuid(string data) } catch { + // Ignore any malformed packet errors here and return an empty Guid. + // Packet errors will be reported later during message processing. } }