Avoid synchronous I/O in BufferedReadStream (#11379)

Fixes #

### Context


### Changes Made


### Testing


### Notes

Author: SimaTian

src/Shared/BufferedReadStream.cs

@@ -3,6 +3,12 @@
 
 using System;
 using System.IO;
+using System.IO.Pipes;
+using System.Threading;
+
+#if NET451_OR_GREATER || NETCOREAPP
+using System.Threading.Tasks;
+#endif
 
 #nullable disable
 
@@ -11,14 +17,14 @@ namespace Microsoft.Build.BackEnd
     internal class BufferedReadStream : Stream
     {
         private const int BUFFER_SIZE = 1024;
-        private Stream _innerStream;
+        private NamedPipeServerStream _innerStream;
         private byte[] _buffer;
 
         // The number of bytes in the buffer that have been read from the underlying stream but not read by consumers of this stream
         private int _currentlyBufferedByteCount;
         private int _currentIndexInBuffer;
 
-        public BufferedReadStream(Stream innerStream)
+        public BufferedReadStream(NamedPipeServerStream innerStream)
         {
             _innerStream = innerStream;
             _buffer = new byte[BUFFER_SIZE];
@@ -120,6 +126,62 @@ public override int Read(byte[] buffer, int offset, int count)
             }
         }
 
+#if NET451_OR_GREATER || NETCOREAPP
+        public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            if (count > BUFFER_SIZE)
+            {
+                // Trying to read more data than the buffer can hold
+                int alreadyCopied = CopyToBuffer(buffer, offset);
+
+#pragma warning disable CA1835 // Prefer the 'Memory'-based overloads for 'ReadAsync' and 'WriteAsync'
+                int innerReadCount = await _innerStream.ReadAsync(buffer, offset + alreadyCopied, count - alreadyCopied, cancellationToken);
+#pragma warning restore CA1835 // Prefer the 'Memory'-based overloads for 'ReadAsync' and 'WriteAsync'
+                return innerReadCount + alreadyCopied;
+            }
+            else if (count <= _currentlyBufferedByteCount)
+            {
+                // Enough data buffered to satisfy read request
+                Array.Copy(_buffer, _currentIndexInBuffer, buffer, offset, count);
+                _currentIndexInBuffer += count;
+                _currentlyBufferedByteCount -= count;
+                return count;
+            }
+            else
+            {
+                // Need to read more data
+                int alreadyCopied = CopyToBuffer(buffer, offset);
+
+#pragma warning disable CA1835 // Prefer the 'Memory'-based overloads for 'ReadAsync' and 'WriteAsync'
+                int innerReadCount = await _innerStream.ReadAsync(_buffer, 0, BUFFER_SIZE, cancellationToken);
+#pragma warning restore CA1835 // Prefer the 'Memory'-based overloads for 'ReadAsync' and 'WriteAsync'
+                _currentIndexInBuffer = 0;
+                _currentlyBufferedByteCount = innerReadCount;
+
+                int remainingCopyCount = alreadyCopied + innerReadCount >= count ? count - alreadyCopied : innerReadCount;
+                Array.Copy(_buffer, 0, buffer, offset + alreadyCopied, remainingCopyCount);
+                _currentIndexInBuffer += remainingCopyCount;
+                _currentlyBufferedByteCount -= remainingCopyCount;
+
+                return alreadyCopied + remainingCopyCount;
+            }
+
+            int CopyToBuffer(byte[] buffer, int offset)
+            {
+                int alreadyCopied = 0;
+                if (_currentlyBufferedByteCount > 0)
+                {
+                    Array.Copy(_buffer, _currentIndexInBuffer, buffer, offset, _currentlyBufferedByteCount);
+                    alreadyCopied = _currentlyBufferedByteCount;
+                    _currentIndexInBuffer = 0;
+                    _currentlyBufferedByteCount = 0;
+                }
+
+                return alreadyCopied;
+            }
+        }
+#endif
+
         public override long Seek(long offset, SeekOrigin origin)
         {
             throw new NotSupportedException();

src/Shared/NodeEndpointOutOfProcBase.cs

@@ -15,12 +15,11 @@
 using Microsoft.Build.Shared;
 #if FEATURE_SECURITY_PERMISSIONS || FEATURE_PIPE_SECURITY
 using System.Security.AccessControl;
-
 #endif
 #if FEATURE_PIPE_SECURITY && FEATURE_NAMED_PIPE_SECURITY_CONSTRUCTOR
 using System.Security.Principal;
 #endif
-#if !FEATURE_APM
+#if NET451_OR_GREATER || NETCOREAPP
 using System.Threading.Tasks;
 #endif
 
@@ -511,7 +510,7 @@ private void PacketPumpProc()
             }
         }
 
-        private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
+        private void RunReadLoop(BufferedReadStream localReadPipe, NamedPipeServerStream localWritePipe,
             ConcurrentQueue<INodePacket> localPacketQueue, AutoResetEvent localPacketAvailable, AutoResetEvent localTerminatePacketPump)
         {
             // Ordering of the wait handles is important.  The first signalled wait handle in the array
@@ -520,25 +519,30 @@ private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
             // spammed to the endpoint and it never gets an opportunity to shutdown.
             CommunicationsUtilities.Trace("Entering read loop.");
             byte[] headerByte = new byte[5];
-#if FEATURE_APM
-            IAsyncResult result = localReadPipe.BeginRead(headerByte, 0, headerByte.Length, null, null);
-#else
+#if NET451_OR_GREATER
+            Task<int> readTask = localReadPipe.ReadAsync(headerByte, 0, headerByte.Length, CancellationToken.None);
+#elif NETCOREAPP
             Task<int> readTask = CommunicationsUtilities.ReadAsync(localReadPipe, headerByte, headerByte.Length);
+#else
+            IAsyncResult result = localReadPipe.BeginRead(headerByte, 0, headerByte.Length, null, null);
 #endif
 
-            bool exitLoop = false;
-            do
+            // Ordering is important.  We want packetAvailable to supercede terminate otherwise we will not properly wait for all
+            // packets to be sent by other threads which are shutting down, such as the logging thread.
+            WaitHandle[] handles = new WaitHandle[]
             {
-                // Ordering is important.  We want packetAvailable to supercede terminate otherwise we will not properly wait for all
-                // packets to be sent by other threads which are shutting down, such as the logging thread.
-                WaitHandle[] handles = new WaitHandle[] {
-#if FEATURE_APM
-                    result.AsyncWaitHandle,
+#if NET451_OR_GREATER || NETCOREAPP
+                ((IAsyncResult)readTask).AsyncWaitHandle,
 #else
-                    ((IAsyncResult)readTask).AsyncWaitHandle,
+                result.AsyncWaitHandle,
 #endif
-                    localPacketAvailable, localTerminatePacketPump };
+                localPacketAvailable,
+                localTerminatePacketPump,
+            };
 
+            bool exitLoop = false;
+            do
+            {
                 int waitId = WaitHandle.WaitAny(handles);
                 switch (waitId)
                 {
@@ -547,10 +551,10 @@ private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
                             int bytesRead = 0;
                             try
                             {
-#if FEATURE_APM
-                                bytesRead = localReadPipe.EndRead(result);
-#else
+#if NET451_OR_GREATER || NETCOREAPP
                                 bytesRead = readTask.Result;
+#else
+                                bytesRead = localReadPipe.EndRead(result);
 #endif
                             }
                             catch (Exception e)
@@ -591,7 +595,7 @@ private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
                                 break;
                             }
 
-                            NodePacketType packetType = (NodePacketType)Enum.ToObject(typeof(NodePacketType), headerByte[0]);
+                            NodePacketType packetType = (NodePacketType)headerByte[0];
 
                             try
                             {
@@ -607,10 +611,18 @@ private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
                                 break;
                             }
 
-#if FEATURE_APM
+#if NET451_OR_GREATER
+                            readTask = localReadPipe.ReadAsync(headerByte, 0, headerByte.Length, CancellationToken.None);
+#elif NETCOREAPP
+                            readTask = CommunicationsUtilities.ReadAsync(localReadPipe, headerByte, headerByte.Length);
+#else
                             result = localReadPipe.BeginRead(headerByte, 0, headerByte.Length, null, null);
+#endif
+
+#if NET451_OR_GREATER || NETCOREAPP
+                            handles[0] = ((IAsyncResult)readTask).AsyncWaitHandle;
 #else
-                            readTask = CommunicationsUtilities.ReadAsync(localReadPipe, headerByte, headerByte.Length);
+                            handles[0] = result.AsyncWaitHandle;
 #endif
                         }
 
@@ -673,8 +685,8 @@ private void RunReadLoop(Stream localReadPipe, Stream localWritePipe,
             while (!exitLoop);
         }
 
-        #endregion
+#endregion
 
-        #endregion
+#endregion
     }
 }