Minimal System.IO.Pipelines integration to prepare for full-async work.

Author: stebet

projects/RabbitMQ.Client/RabbitMQ.Client.csproj

@@ -63,6 +63,7 @@
     <PackageReference Include="MinVer" Version="4.3.0" PrivateAssets="all" />
     <PackageReference Include="System.Memory" Version="4.5.5" />
     <PackageReference Include="System.Threading.Channels" Version="7.0.0" />
+    <PackageReference Include="System.IO.Pipelines" Version="6.0.3" />
   </ItemGroup>
 
 </Project>

projects/RabbitMQ.Client/client/impl/Connection.Receive.cs

@@ -43,11 +43,11 @@ internal sealed partial class Connection
         private readonly IFrameHandler _frameHandler;
         private readonly Task _mainLoopTask;
 
-        private void MainLoop()
+        private async Task MainLoop()
         {
             try
             {
-                ReceiveLoop();
+                await ReceiveLoop();
             }
             catch (EndOfStreamException eose)
             {
@@ -56,7 +56,7 @@ private void MainLoop()
             }
             catch (HardProtocolException hpe)
             {
-                HardProtocolExceptionHandler(hpe);
+                await HardProtocolExceptionHandler(hpe);
             }
             catch (Exception ex)
             {
@@ -66,11 +66,11 @@ private void MainLoop()
             FinishClose();
         }
 
-        private void ReceiveLoop()
+        private async Task ReceiveLoop()
         {
             while (!_closed)
             {
-                InboundFrame frame = _frameHandler.ReadFrame();
+                InboundFrame frame = await _frameHandler.ReadFrame();
                 NotifyHeartbeatListener();
 
                 bool shallReturn = true;
@@ -139,7 +139,7 @@ private void HandleMainLoopException(ShutdownEventArgs reason)
             LogCloseError($"Unexpected connection closure: {reason}", new Exception(reason.ToString()));
         }
 
-        private void HardProtocolExceptionHandler(HardProtocolException hpe)
+        private async Task HardProtocolExceptionHandler(HardProtocolException hpe)
         {
             if (SetCloseReason(hpe.ShutdownReason))
             {
@@ -151,7 +151,7 @@ private void HardProtocolExceptionHandler(HardProtocolException hpe)
                     _session0.Transmit(in cmd);
                     if (hpe.CanShutdownCleanly)
                     {
-                        ClosingLoop();
+                        await ClosingLoop();
                     }
                 }
                 catch (IOException ioe)
@@ -168,13 +168,13 @@ private void HardProtocolExceptionHandler(HardProtocolException hpe)
         ///<remarks>
         /// Loop only used while quiescing. Use only to cleanly close connection
         ///</remarks>
-        private void ClosingLoop()
+        private async Task ClosingLoop()
         {
             try
             {
                 _frameHandler.ReadTimeout = TimeSpan.Zero;
                 // Wait for response/socket closure or timeout
-                ReceiveLoop();
+                await ReceiveLoop();
             }
             catch (ObjectDisposedException ode)
             {

projects/RabbitMQ.Client/client/impl/Connection.cs

@@ -79,7 +79,7 @@ public Connection(ConnectionConfig config, IFrameHandler frameHandler)
                 ["connection_name"] = ClientProvidedName
             };
 
-            _mainLoopTask = Task.Factory.StartNew(MainLoop, TaskCreationOptions.LongRunning);
+            _mainLoopTask = Task.Run(MainLoop);
             try
             {
                 Open();

projects/RabbitMQ.Client/client/impl/Frame.cs

@@ -31,10 +31,13 @@
 
 using System;
 using System.Buffers;
+using System.Diagnostics;
 using System.IO;
+using System.IO.Pipelines;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
 using System.Runtime.ExceptionServices;
+using System.Threading.Tasks;
 
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing.Impl;
@@ -130,14 +133,14 @@ internal static class Heartbeat
 
             /// <summary>
             /// Compiler trick to directly refer to static data in the assembly, see here: https://github.com/dotnet/roslyn/pull/24621
-            /// </summary>
-            private static ReadOnlySpan<byte> Payload => new byte[]
-            {
-                Constants.FrameHeartbeat,
-                0, 0, // channel
-                0, 0, 0, 0, // payload length
-                Constants.FrameEnd
-            };
+            /// A heartbeat frame has the following layout:
+            /// +--------------------+------------------+-----------------+--------------------------+
+            /// | Frame Type(1 byte) | Channel(2 bytes) | Length(4 bytes) | End Frame Marker(1 byte) |
+            /// +--------------------+------------------+-----------------+--------------------------+
+            /// | 0x08               | 0x0000           | 0x00000000      | 0xCE                     |
+            /// +--------------------+------------------+-----------------+--------------------------+
+            ///</summary>
+            private static ReadOnlySpan<byte> Payload => new byte[] { Constants.FrameHeartbeat, 0, 0, 0, 0, 0, 0, Constants.FrameEnd };
 
             public static Memory<byte> GetHeartbeatFrame()
             {
@@ -201,7 +204,7 @@ private static int GetBodyFrameCount(int maxPayloadBytes, int length)
         }
     }
 
-    internal readonly ref struct InboundFrame
+    internal readonly struct InboundFrame
     {
         public readonly FrameType Type;
         public readonly int Channel;
@@ -216,22 +219,24 @@ private InboundFrame(FrameType type, int channel, ReadOnlyMemory<byte> payload,
             _rentedArray = rentedArray;
         }
 
-        private static void ProcessProtocolHeader(Stream reader, ReadOnlySpan<byte> frameHeader)
+        private static void ProcessProtocolHeader(ReadOnlySequence<byte> buffer)
         {
             try
             {
-                if (frameHeader[0] != 'M' || frameHeader[1] != 'Q' || frameHeader[2] != 'P')
+                if (buffer.Length < 8)
                 {
-                    throw new MalformedFrameException("Invalid AMQP protocol header from server");
+                    throw new EndOfStreamException();
                 }
 
-                int serverMinor = reader.ReadByte();
-                if (serverMinor == -1)
+                Span<byte> tempSpan = stackalloc byte[8];
+                buffer.Slice(0, 8).CopyTo(tempSpan);
+
+                if (tempSpan[1] != 'M' || tempSpan[2] != 'Q' || tempSpan[3] != 'P')
                 {
-                    throw new EndOfStreamException();
+                    throw new MalformedFrameException("Invalid AMQP protocol header from server");
                 }
 
-                throw new PacketNotRecognizedException(frameHeader[3], frameHeader[4], frameHeader[5], serverMinor);
+                throw new PacketNotRecognizedException(tempSpan[4], tempSpan[5], tempSpan[6], tempSpan[7]);
             }
             catch (EndOfStreamException)
             {
@@ -246,38 +251,59 @@ private static void ProcessProtocolHeader(Stream reader, ReadOnlySpan<byte> fram
             }
         }
 
-        internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, uint maxMessageSize)
+        internal static async ValueTask<InboundFrame> ReadFromPipe(PipeReader reader, uint maxMessageSize)
         {
-            try
+            // Try a synchronous read first, then go async
+            if (!reader.TryRead(out ReadResult result))
             {
-                ReadFromStream(reader, frameHeaderBuffer, frameHeaderBuffer.Length);
+                result = await reader.ReadAsync().ConfigureAwait(false);
             }
-            catch (IOException ioe)
+
+            ReadOnlySequence<byte> buffer = result.Buffer;
+
+            if (result.IsCompleted || buffer.Length == 0)
             {
-                // If it's a WSAETIMEDOUT SocketException, unwrap it.
-                // This might happen when the limit of half-open connections is
-                // reached.
-                if (ioe?.InnerException is SocketException exception && exception.SocketErrorCode == SocketError.TimedOut)
-                {
-                    ExceptionDispatchInfo.Capture(exception).Throw();
-                }
-                else
+                throw new EndOfStreamException("Pipe is completed.");
+            }
+
+            byte firstByte = buffer.First.Span[0];
+            if (firstByte == 'A')
+            {
+                ProcessProtocolHeader(buffer);
+            }
+
+            InboundFrame frame;
+
+            while (!TryReadFrame(ref buffer, maxMessageSize, out frame))
+            {
+                reader.AdvanceTo(buffer.Start, buffer.End);
+                
+                // No need to try a synchronous read since we have an incomplete frame anyway, so we'll always need to go async
+                result = await reader.ReadAsync().ConfigureAwait(false);
+
+                if (result.IsCompleted || buffer.Length == 0)
                 {
-                    throw;
+                    throw new EndOfStreamException("Pipe is completed.");
                 }
+
+                buffer = result.Buffer;
             }
 
-            byte firstByte = frameHeaderBuffer[0];
-            if (firstByte == 'A')
+            reader.AdvanceTo(buffer.Start);
+            return frame;
+        }
+
+        internal static bool TryReadFrame(ref ReadOnlySequence<byte> buffer, uint maxMessageSize, out InboundFrame frame)
+        {
+            if(buffer.Length < 7)
             {
-                // Probably an AMQP protocol header, otherwise meaningless
-                ProcessProtocolHeader(reader, frameHeaderBuffer.AsSpan(1, 6));
+                frame = default;
+                return false;
             }
 
-            FrameType type = (FrameType)firstByte;
-            var frameHeaderSpan = new ReadOnlySpan<byte>(frameHeaderBuffer, 1, 6);
-            int channel = NetworkOrderDeserializer.ReadUInt16(frameHeaderSpan);
-            int payloadSize = NetworkOrderDeserializer.ReadInt32(frameHeaderSpan.Slice(2, 4));
+            FrameType type = (FrameType)buffer.First.Span[0];
+            int channel = NetworkOrderDeserializer.ReadUInt16(buffer.Slice(1));
+            int payloadSize = NetworkOrderDeserializer.ReadInt32(buffer.Slice(3));
             if ((maxMessageSize > 0) && (payloadSize > maxMessageSize))
             {
                 string msg = $"Frame payload size '{payloadSize}' exceeds maximum of '{maxMessageSize}' bytes";
@@ -287,46 +313,29 @@ internal static InboundFrame ReadFrom(Stream reader, byte[] frameHeaderBuffer, u
             const int EndMarkerLength = 1;
             // Is returned by InboundFrame.ReturnPayload in Connection.MainLoopIteration
             int readSize = payloadSize + EndMarkerLength;
-            byte[] payloadBytes = ArrayPool<byte>.Shared.Rent(readSize);
-            try
-            {
-                ReadFromStream(reader, payloadBytes, readSize);
-            }
-            catch (Exception)
-            {
-                // Early EOF.
-                ArrayPool<byte>.Shared.Return(payloadBytes);
-                throw new MalformedFrameException($"Short frame - expected to read {readSize} bytes");
-            }
 
-            if (payloadBytes[payloadSize] != Constants.FrameEnd)
+            if ((buffer.Length - 7) < readSize)
             {
-                ArrayPool<byte>.Shared.Return(payloadBytes);
-                throw new MalformedFrameException($"Bad frame end marker: {payloadBytes[payloadSize]}");
+                frame = default;
+                return false;
             }
-
-            RabbitMqClientEventSource.Log.DataReceived(payloadSize + Framing.BaseFrameSize);
-            return new InboundFrame(type, channel, new Memory<byte>(payloadBytes, 0, payloadSize), payloadBytes);
-        }
-
-        [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        private static void ReadFromStream(Stream reader, byte[] buffer, int toRead)
-        {
-            int bytesRead = 0;
-            do
+            else
             {
-                int read = reader.Read(buffer, bytesRead, toRead - bytesRead);
-                if (read == 0)
+                byte[] payloadBytes = ArrayPool<byte>.Shared.Rent(readSize);
+                ReadOnlySequence<byte> framePayload = buffer.Slice(7, readSize);
+                framePayload.CopyTo(payloadBytes);
+
+                if (payloadBytes[payloadSize] != Constants.FrameEnd)
                 {
-                    ThrowEndOfStream();
+                    ArrayPool<byte>.Shared.Return(payloadBytes);
+                    throw new MalformedFrameException($"Bad frame end marker: {payloadBytes[payloadSize]}");
                 }
 
-                bytesRead += read;
-            } while (bytesRead != toRead);
-
-            static void ThrowEndOfStream()
-            {
-                throw new EndOfStreamException("Reached the end of the stream. Possible authentication failure.");
+                RabbitMqClientEventSource.Log.DataReceived(payloadSize + Framing.BaseFrameSize);
+                frame = new InboundFrame(type, channel, new Memory<byte>(payloadBytes, 0, payloadSize), payloadBytes);
+                // Advance the buffer
+                buffer = buffer.Slice(7 + readSize);
+                return true;
             }
         }
 

projects/RabbitMQ.Client/client/impl/IFrameHandler.cs

@@ -31,6 +31,7 @@
 
 using System;
 using System.Net;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -57,7 +58,7 @@ internal interface IFrameHandler
         ///<summary>Read a frame from the underlying
         ///transport. Returns null if the read operation timed out
         ///(see Timeout property).</summary>
-        InboundFrame ReadFrame();
+        ValueTask<InboundFrame> ReadFrame();
 
         void SendHeader();
 

projects/RabbitMQ.Client/client/impl/Session.cs

@@ -29,6 +29,8 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System.Diagnostics;
+
 using RabbitMQ.Client.Framing.Impl;
 
 namespace RabbitMQ.Client.Impl