[REVERT] 11546 refactor common pipe code (#11648)

Author: YuliiaKovalova

src/Build.UnitTests/BackEnd/NodeEndpointInProc_Tests.cs

@@ -100,7 +100,7 @@ public void UnregisterPacketHandler(NodePacketType packetType)
                 throw new NotImplementedException();
             }
 
-            public INodePacket DeserializePacket(NodePacketType packetType, ITranslator translator)
+            public void DeserializeAndRoutePacket(int nodeId, NodePacketType packetType, ITranslator translator)
             {
                 throw new NotImplementedException();
             }

src/Build/BackEnd/Client/MSBuildClient.cs

@@ -7,6 +7,7 @@
 using System.Diagnostics;
 using System.Globalization;
 using System.IO;
+using System.IO.Pipes;
 using System.Threading;
 using Microsoft.Build.BackEnd;
 using Microsoft.Build.BackEnd.Client;
@@ -74,9 +75,19 @@ public sealed class MSBuildClient
         private readonly string _pipeName;
 
         /// <summary>
-        /// The named pipe client for client-server communication.
+        /// The named pipe stream for client-server communication.
         /// </summary>
-        private NodePipeClient _pipeClient = null!;
+        private NamedPipeClientStream _nodeStream = null!;
+
+        /// <summary>
+        /// A way to cache a byte array when writing out packets
+        /// </summary>
+        private readonly MemoryStream _packetMemoryStream;
+
+        /// <summary>
+        /// A binary writer to help write into <see cref="_packetMemoryStream"/>
+        /// </summary>
+        private readonly BinaryWriter _binaryWriter;
 
         /// <summary>
         /// Used to estimate the size of the build with an ETW trace.
@@ -119,14 +130,26 @@ public MSBuildClient(
             // Client <-> Server communication stream
             _handshake = GetHandshake();
             _pipeName = OutOfProcServerNode.GetPipeName(_handshake);
+            _packetMemoryStream = new MemoryStream();
+            _binaryWriter = new BinaryWriter(_packetMemoryStream);
 
             CreateNodePipeStream();
         }
 
         private void CreateNodePipeStream()
         {
-            _pipeClient = new NodePipeClient(_pipeName, _handshake);
-            _packetPump = new MSBuildClientPacketPump(_pipeClient);
+#pragma warning disable SA1111, SA1009 // Closing parenthesis should be on line of last parameter
+            _nodeStream = new NamedPipeClientStream(
+                serverName: ".",
+                _pipeName,
+                PipeDirection.InOut,
+                PipeOptions.Asynchronous
+#if FEATURE_PIPEOPTIONS_CURRENTUSERONLY
+                | PipeOptions.CurrentUserOnly
+#endif
+            );
+#pragma warning restore SA1111, SA1009 // Closing parenthesis should be on line of last parameter
+            _packetPump = new MSBuildClientPacketPump(_nodeStream);
         }
 
         /// <summary>
@@ -400,7 +423,7 @@ private bool TrySendPacket(Func<INodePacket> packetResolver)
             try
             {
                 packet = packetResolver();
-                _pipeClient.WritePacket(packet);
+                WritePacket(_nodeStream, packet);
                 CommunicationsUtilities.Trace("Command packet of type '{0}' sent...", packet.Type);
             }
             catch (Exception ex)
@@ -598,7 +621,7 @@ private bool TryConnectToServer(int timeoutMilliseconds)
                 tryAgain = false;
                 try
                 {
-                    _pipeClient.ConnectToServer(Math.Max(1, timeoutMilliseconds - (int)sw.ElapsedMilliseconds));
+                    NodeProviderOutOfProcBase.ConnectToPipeStream(_nodeStream, _pipeName, _handshake, Math.Max(1, timeoutMilliseconds - (int)sw.ElapsedMilliseconds));
                 }
                 catch (Exception ex)
                 {
@@ -621,5 +644,30 @@ private bool TryConnectToServer(int timeoutMilliseconds)
 
             return true;
         }
+
+        private void WritePacket(Stream nodeStream, INodePacket packet)
+        {
+            MemoryStream memoryStream = _packetMemoryStream;
+            memoryStream.SetLength(0);
+
+            ITranslator writeTranslator = BinaryTranslator.GetWriteTranslator(memoryStream);
+
+            // Write header
+            memoryStream.WriteByte((byte)packet.Type);
+
+            // Pad for packet length
+            _binaryWriter.Write(0);
+
+            // Reset the position in the write buffer.
+            packet.Translate(writeTranslator);
+
+            int packetStreamLength = (int)memoryStream.Position;
+
+            // Now write in the actual packet length
+            memoryStream.Position = 1;
+            _binaryWriter.Write(packetStreamLength - 5);
+
+            nodeStream.Write(memoryStream.GetBuffer(), 0, packetStreamLength);
+        }
     }
 }

src/Build/BackEnd/Client/MSBuildClientPacketPump.cs

@@ -2,9 +2,15 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using System;
+using System.Buffers.Binary;
 using System.Collections.Concurrent;
+using System.IO;
 using System.Threading;
+
+#if NET
 using System.Threading.Tasks;
+#endif
+
 using Microsoft.Build.Internal;
 using Microsoft.Build.Shared;
 
@@ -42,15 +48,25 @@ internal sealed class MSBuildClientPacketPump : INodePacketHandler, INodePacketF
         /// </summary>
         private readonly NodePacketFactory _packetFactory;
 
+        /// <summary>
+        /// The memory stream for a read buffer.
+        /// </summary>
+        private readonly MemoryStream _readBufferMemoryStream;
+
         /// <summary>
         /// The thread which runs the asynchronous packet pump
         /// </summary>
         private Thread? _packetPumpThread;
 
         /// <summary>
-        /// The pipe client from where to read packets.
+        /// The stream from where to read packets.
         /// </summary>
-        private readonly NodePipeClient _pipeClient;
+        private readonly Stream _stream;
+
+        /// <summary>
+        /// The binary translator for reading packets.
+        /// </summary>
+        private readonly ITranslator _binaryReadTranslator;
 
         /// <summary>
         /// True if this side is gracefully disconnecting.
@@ -59,19 +75,21 @@ internal sealed class MSBuildClientPacketPump : INodePacketHandler, INodePacketF
         /// </summary>
         private bool _isServerDisconnecting;
 
-        public MSBuildClientPacketPump(NodePipeClient pipeClient)
+        public MSBuildClientPacketPump(Stream stream)
         {
-            ErrorUtilities.VerifyThrowArgumentNull(pipeClient);
+            ErrorUtilities.VerifyThrowArgumentNull(stream);
 
-            _pipeClient = pipeClient;
-            _pipeClient.RegisterPacketFactory(this);
+            _stream = stream;
             _isServerDisconnecting = false;
             _packetFactory = new NodePacketFactory();
 
             ReceivedPacketsQueue = new ConcurrentQueue<INodePacket>();
             PacketReceivedEvent = new AutoResetEvent(false);
             PacketPumpCompleted = new ManualResetEvent(false);
             _packetPumpShutdownEvent = new ManualResetEvent(false);
+
+            _readBufferMemoryStream = new MemoryStream();
+            _binaryReadTranslator = BinaryTranslator.GetReadTranslator(_readBufferMemoryStream, InterningBinaryReader.CreateSharedBuffer());
         }
 
         #region INodePacketFactory Members
@@ -97,13 +115,14 @@ public void UnregisterPacketHandler(NodePacketType packetType)
         }
 
         /// <summary>
-        /// Deserializes a packet.
+        /// Deserializes and routes a packer to the appropriate handler.
         /// </summary>
+        /// <param name="nodeId">The node from which the packet was received.</param>
         /// <param name="packetType">The packet type.</param>
         /// <param name="translator">The translator to use as a source for packet data.</param>
-        public INodePacket DeserializePacket(NodePacketType packetType, ITranslator translator)
+        public void DeserializeAndRoutePacket(int nodeId, NodePacketType packetType, ITranslator translator)
         {
-            return _packetFactory.DeserializePacket(packetType, translator);
+            _packetFactory.DeserializeAndRoutePacket(nodeId, packetType, translator);
         }
 
         /// <summary>
@@ -165,16 +184,21 @@ public void Stop()
         /// </remarks>
         private void PacketPumpProc()
         {
-            RunReadLoop(_pipeClient, _packetPumpShutdownEvent);
+            RunReadLoop(_stream, _packetPumpShutdownEvent);
         }
 
-        private void RunReadLoop(NodePipeClient pipeClient, ManualResetEvent localPacketPumpShutdownEvent)
+        private void RunReadLoop(Stream localStream, ManualResetEvent localPacketPumpShutdownEvent)
         {
             CommunicationsUtilities.Trace("Entering read loop.");
 
             try
             {
-                Task<INodePacket> readTask = pipeClient.ReadPacketAsync();
+                byte[] headerByte = new byte[5];
+#if FEATURE_APM
+                IAsyncResult result = localStream.BeginRead(headerByte, 0, headerByte.Length, null, null);
+#else
+                Task<int> readTask = CommunicationsUtilities.ReadAsync(localStream, headerByte, headerByte.Length).AsTask();
+#endif
 
                 bool continueReading = true;
                 do
@@ -186,7 +210,11 @@ private void RunReadLoop(NodePipeClient pipeClient, ManualResetEvent localPacket
                     WaitHandle[] handles =
                     [
                         localPacketPumpShutdownEvent,
+#if FEATURE_APM
+                        result.AsyncWaitHandle
+#else
                         ((IAsyncResult)readTask).AsyncWaitHandle
+#endif
                     ];
                     int waitId = WaitHandle.WaitAny(handles);
                     switch (waitId)
@@ -198,27 +226,80 @@ private void RunReadLoop(NodePipeClient pipeClient, ManualResetEvent localPacket
                             break;
 
                         case 1:
-                            INodePacket packet = readTask.GetAwaiter().GetResult();
-
-                            if (packet.Type == NodePacketType.NodeShutdown)
                             {
-                                if (!_isServerDisconnecting)
+                                // Client recieved a packet header. Read the rest of it.
+                                int headerBytesRead = 0;
+#if FEATURE_APM
+                                headerBytesRead = localStream.EndRead(result);
+#else
+                                headerBytesRead = readTask.Result;
+#endif
+
+                                if ((headerBytesRead != headerByte.Length) && !localPacketPumpShutdownEvent.WaitOne(0))
                                 {
-                                    ErrorUtilities.ThrowInternalError("Server disconnected abruptly.");
+                                    // Incomplete read. Abort.
+                                    if (headerBytesRead == 0)
+                                    {
+                                        if (_isServerDisconnecting)
+                                        {
+                                            continueReading = false;
+                                            break;
+                                        }
+
+                                        ErrorUtilities.ThrowInternalError("Server disconnected abruptly");
+                                    }
+                                    else
+                                    {
+                                        ErrorUtilities.ThrowInternalError("Incomplete header read.  {0} of {1} bytes read", headerBytesRead, headerByte.Length);
+                                    }
                                 }
 
-                                continueReading = false;
-                                break;
-                            }
+                                NodePacketType packetType = (NodePacketType)Enum.ToObject(typeof(NodePacketType), headerByte[0]);
 
-                            _packetFactory.RoutePacket(0, packet);
+                                int packetLength = BinaryPrimitives.ReadInt32LittleEndian(new Span<byte>(headerByte, 1, 4));
+                                int packetBytesRead = 0;
 
-                            continueReading = packet.Type != NodePacketType.ServerNodeBuildResult;
-                            if (continueReading)
-                            {
-                                readTask = pipeClient.ReadPacketAsync();
-                            }
+                                _readBufferMemoryStream.Position = 0;
+                                _readBufferMemoryStream.SetLength(packetLength);
+                                byte[] packetData = _readBufferMemoryStream.GetBuffer();
 
+                                while (packetBytesRead < packetLength)
+                                {
+                                    int bytesRead = localStream.Read(packetData, packetBytesRead, packetLength - packetBytesRead);
+                                    if (bytesRead == 0)
+                                    {
+                                        // Incomplete read.  Abort.
+                                        ErrorUtilities.ThrowInternalError("Incomplete packet read. {0} of {1} bytes read", packetBytesRead, packetLength);
+                                    }
+
+                                    packetBytesRead += bytesRead;
+                                }
+
+                                try
+                                {
+                                    _packetFactory.DeserializeAndRoutePacket(0, packetType, _binaryReadTranslator);
+                                }
+                                catch
+                                {
+                                    // Error while deserializing or handling packet. Logging additional info.
+                                    CommunicationsUtilities.Trace("Packet factory failed to receive package. Exception while deserializing packet {0}.", packetType);
+                                    throw;
+                                }
+
+                                if (packetType == NodePacketType.ServerNodeBuildResult)
+                                {
+                                    continueReading = false;
+                                }
+                                else
+                                {
+                                    // Start reading the next package header.
+#if FEATURE_APM
+                                    result = localStream.BeginRead(headerByte, 0, headerByte.Length, null, null);
+#else
+                                    readTask = CommunicationsUtilities.ReadAsync(localStream, headerByte, headerByte.Length).AsTask();
+#endif
+                                }
+                            }
                             break;
 
                         default:

src/Build/BackEnd/Components/Communications/NodeManager.cs

@@ -240,13 +240,19 @@ public void UnregisterPacketHandler(NodePacketType packetType)
         }
 
         /// <summary>
-        /// Takes a serializer and deserializes the packet.
+        /// Takes a serializer, deserializes the packet and routes it to the appropriate handler.
         /// </summary>
+        /// <param name="nodeId">The node from which the packet was received.</param>
         /// <param name="packetType">The packet type.</param>
         /// <param name="translator">The translator containing the data from which the packet should be reconstructed.</param>
-        public INodePacket DeserializePacket(NodePacketType packetType, ITranslator translator)
+        public void DeserializeAndRoutePacket(int nodeId, NodePacketType packetType, ITranslator translator)
         {
-            return _packetFactory.DeserializePacket(packetType, translator);
+            if (packetType == NodePacketType.NodeShutdown)
+            {
+                RemoveNodeFromMapping(nodeId);
+            }
+
+            _packetFactory.DeserializeAndRoutePacket(nodeId, packetType, translator);
         }
 
         /// <summary>

src/Build/BackEnd/Components/Communications/NodeProviderInProc.cs

@@ -286,11 +286,10 @@ public void UnregisterPacketHandler(NodePacketType packetType)
         /// <summary>
         /// Deserializes and routes a packet.  Not used in the in-proc node.
         /// </summary>
-        public INodePacket DeserializePacket(NodePacketType packetType, ITranslator translator)
+        public void DeserializeAndRoutePacket(int nodeId, NodePacketType packetType, ITranslator translator)
         {
             // Not used
             ErrorUtilities.ThrowInternalErrorUnreachable();
-            return null;
         }
 
         /// <summary>