Simplifying code, reducing blocking and threads by trying to inline writes and resulting to backlog writes if that fails.

Author: Stefán J. Sigurðarson

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

@@ -593,38 +593,37 @@ public int MainLoopIteration(ref ReadOnlySequence<byte> buffer)
 
                 bool shallReturn = true;
                 // We have received an actual frame.
-                if (frame.Type == FrameType.FrameHeartbeat)
+                if (frame.Type != FrameType.FrameHeartbeat)
                 {
-                    // Ignore it: we've already just reset the heartbeat
-                }
-                else if (frame.Channel == 0)
-                {
-                    // In theory, we could get non-connection.close-ok
-                    // frames here while we're quiescing (m_closeReason !=
-                    // null). In practice, there's a limited number of
-                    // things the server can ask of us on channel 0 -
-                    // essentially, just connection.close. That, combined
-                    // with the restrictions on pipelining, mean that
-                    // we're OK here to handle channel 0 traffic in a
-                    // quiescing situation, even though technically we
-                    // should be ignoring everything except
-                    // connection.close-ok.
-                    shallReturn = _session0.HandleFrame(in frame);
-                }
-                else
-                {
-                    // If we're still m_running, but have a m_closeReason,
-                    // then we must be quiescing, which means any inbound
-                    // frames for non-zero channels (and any inbound
-                    // commands on channel zero that aren't
-                    // Connection.CloseOk) must be discarded.
-                    if (_closeReason is null)
+                    if (frame.Channel == 0)
                     {
-                        // No close reason, not quiescing the
-                        // connection. Handle the frame. (Of course, the
-                        // Session itself may be quiescing this particular
-                        // channel, but that's none of our concern.)
-                        shallReturn = _sessionManager.Lookup(frame.Channel).HandleFrame(in frame);
+                        // In theory, we could get non-connection.close-ok
+                        // frames here while we're quiescing (m_closeReason !=
+                        // null). In practice, there's a limited number of
+                        // things the server can ask of us on channel 0 -
+                        // essentially, just connection.close. That, combined
+                        // with the restrictions on pipelining, mean that
+                        // we're OK here to handle channel 0 traffic in a
+                        // quiescing situation, even though technically we
+                        // should be ignoring everything except
+                        // connection.close-ok.
+                        shallReturn = _session0.HandleFrame(in frame);
+                    }
+                    else
+                    {
+                        // If we're still m_running, but have a m_closeReason,
+                        // then we must be quiescing, which means any inbound
+                        // frames for non-zero channels (and any inbound
+                        // commands on channel zero that aren't
+                        // Connection.CloseOk) must be discarded.
+                        if (_closeReason is null)
+                        {
+                            // No close reason, not quiescing the
+                            // connection. Handle the frame. (Of course, the
+                            // Session itself may be quiescing this particular
+                            // channel, but that's none of our concern.)
+                            shallReturn = _sessionManager.Lookup(frame.Channel).HandleFrame(in frame);
+                        }
                     }
                 }
 

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

@@ -45,8 +45,8 @@ internal static class Framing
          * +------------+---------+----------------+---------+------------------+
          * | 1 byte     | 2 bytes | 4 bytes        | x bytes | 1 byte           |
          * +------------+---------+----------------+---------+------------------+ */
-        private const int BaseFrameSize = 1 + 2 + 4 + 1;
-        private const int StartPayload = 7;
+        internal const int BaseFrameSize = 1 + 2 + 4 + 1;
+        internal const int StartPayload = 7;
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private static int WriteBaseFrame(Span<byte> span, FrameType type, ushort channel, int payloadLength)
@@ -152,90 +152,104 @@ private InboundFrame(FrameType type, int channel, ReadOnlyMemory<byte> payload,
             _rentedArray = rentedArray;
         }
 
-        private static void ProcessProtocolHeader(ReadOnlySpan<byte> protocolError)
-        {
-            byte b1 = protocolError[0];
-            byte b2 = protocolError[1];
-            byte b3 = protocolError[2];
-            if (b1 != 'M' || b2 != 'Q' || b3 != 'P')
-            {
-                throw new MalformedFrameException("Invalid AMQP protocol header from server");
-            }
-
-            int transportHigh = protocolError[3];
-            int transportLow = protocolError[4];
-            int serverMajor = protocolError[5];
-            int serverMinor = protocolError[6];
-            throw new PacketNotRecognizedException(transportHigh, transportLow, serverMajor, serverMinor);
-        }
-
         internal static bool TryReadFrame(ref ReadOnlySequence<byte> buffer, out InboundFrame frame)
         {
-            // We'll always need to read at least 8 bytes (type (1) + channel (2) + payloadSize (4) + end marker (1)) or (8 bytes of protocol error, see ProcessProtocolHeader).
-            if (buffer.Length < 8)
+            // We'll always need to read at least 8 bytes (the Framing.BaseFrameSize) or 8 bytes of protocol error (see ProcessProtocolHeader).
+            if (buffer.Length < Framing.BaseFrameSize)
             {
                 frame = default;
                 return false;
             }
 
+            // Let's first check if we have a protocol header.
             if (buffer.First.Span[0] == 'A')
             {
-                // Probably an AMQP protocol header, otherwise meaningless
-                if (buffer.First.Length >= 8)
-                {
-                    ProcessProtocolHeader(buffer.First.Span.Slice(1, 7));
-                }
-                else
-                {
-                    Span<byte> protocolError = stackalloc byte[7];
-                    buffer.Slice(1, 7).CopyTo(protocolError);
-                    ProcessProtocolHeader(protocolError);
-                }
+                ProcessProtocolHeader(buffer);
             }
 
             int type = buffer.First.Span[0];
-            int channel;
-            int payloadSize;
+            ParseFrameHeader(buffer, out int channel, out int payloadSize);
 
-            if (buffer.First.Length >= 7)
-            {
-                channel = NetworkOrderDeserializer.ReadUInt16(buffer.First.Span.Slice(1, 2));
-                payloadSize = NetworkOrderDeserializer.ReadInt32(buffer.First.Span.Slice(3, 4)); // FIXME - throw exn on unreasonable value
-            }
-            else
-            {
-                Span<byte> headerBytes = stackalloc byte[6];
-                buffer.Slice(1, 6).CopyTo(headerBytes);
-                channel = NetworkOrderDeserializer.ReadUInt16(headerBytes.Slice(0, 2));
-                payloadSize = NetworkOrderDeserializer.ReadInt32(headerBytes.Slice(2, 4)); // FIXME - throw exn on unreasonable value
-            }
+            // We'll need to read the payloadSize + FrameEndMarker (1 byte)
+            int readSize = payloadSize + 1;
 
-            const int EndMarkerLength = 1;
-            int readSize = payloadSize + EndMarkerLength;
-
-            // Do we have enough bytes to read an entire frame (type + channel + payloadSize + payload + end marker)
-            if (buffer.Length < (7 + readSize))
+            // Do we have enough bytes to read the rest of the frame
+            if (buffer.Length < (Framing.StartPayload + readSize))
             {
                 frame = default;
                 return false;
             }
 
-            // Is returned by InboundFrame.ReturnPayload in Connection.MainLoopIteration
+            // The rented array is returned by InboundFrame.ReturnPayload in Connection.MainLoopIteration
             byte[] payloadBytes = ArrayPool<byte>.Shared.Rent(readSize);
             Memory<byte> payloadMemory = payloadBytes.AsMemory(0, readSize);
             ReadOnlySequence<byte> payloadSlice = buffer.Slice(7, readSize);
             payloadSlice.CopyTo(payloadMemory.Span);
+
+            // Let's validate that the frame contains a valid Frame End Marker
             if (payloadBytes[payloadSize] != Constants.FrameEnd)
             {
                 ArrayPool<byte>.Shared.Return(payloadBytes);
                 throw new MalformedFrameException($"Bad frame end marker: {payloadBytes[payloadSize]}");
             }
 
-            buffer = buffer.Slice(payloadSlice.End);
+            // We have a frame!
             frame = new InboundFrame((FrameType)type, channel, payloadMemory.Slice(0, payloadSize), payloadBytes);
+
+            // Update the buffer
+            buffer = buffer.Slice(payloadSlice.End);
             return true;
         }
 
+        private static void ProcessProtocolHeader(ReadOnlySequence<byte> buffer)
+        {
+            if (buffer.First.Length >= 8)
+            {
+                EvaluateProtocolError(buffer.First.Span.Slice(1, 7));
+            }
+            else
+            {
+                Span<byte> tempBuffer = stackalloc byte[7];
+                buffer.Slice(1, 7).CopyTo(tempBuffer);
+                EvaluateProtocolError(tempBuffer);
+            }
+
+            void EvaluateProtocolError(ReadOnlySpan<byte> protocolError)
+            {
+                byte b1 = protocolError[0];
+                byte b2 = protocolError[1];
+                byte b3 = protocolError[2];
+                if (b1 != 'M' || b2 != 'Q' || b3 != 'P')
+                {
+                    throw new MalformedFrameException("Invalid AMQP protocol header from server");
+                }
+
+                int transportHigh = protocolError[3];
+                int transportLow = protocolError[4];
+                int serverMajor = protocolError[5];
+                int serverMinor = protocolError[6];
+                throw new PacketNotRecognizedException(transportHigh, transportLow, serverMajor, serverMinor);
+            }
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+
+        private static void ParseFrameHeader(ReadOnlySequence<byte> buffer, out int channel, out int payloadSize)
+        {
+            if (buffer.First.Length >= 7)
+            {
+                channel = NetworkOrderDeserializer.ReadUInt16(buffer.First.Span.Slice(1, 2));
+                payloadSize = NetworkOrderDeserializer.ReadInt32(buffer.First.Span.Slice(3, 4)); // FIXME - throw exn on unreasonable value
+            }
+            else
+            {
+                Span<byte> headerBytes = stackalloc byte[6];
+                buffer.Slice(1, 6).CopyTo(headerBytes);
+                channel = NetworkOrderDeserializer.ReadUInt16(headerBytes.Slice(0, 2));
+                payloadSize = NetworkOrderDeserializer.ReadInt32(headerBytes.Slice(2, 4)); // FIXME - throw exn on unreasonable value
+            }
+        }
+
         public byte[] TakeoverPayload()
         {
             return _rentedArray;