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

.ci/versions.json

@@ -1,4 +1,4 @@
 {
-  "erlang": "25.2",
-  "rabbitmq": "3.11.6"
+  "erlang": "25.2.3",
+  "rabbitmq": "3.11.9"
 }

projects/Benchmarks/Benchmarks.csproj

@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="BenchmarkDotNet" Version="0.13.4" />
+    <PackageReference Include="BenchmarkDotNet" Version="0.13.5" />
     <PackageReference Include="Ductus.FluentDocker" Version="2.10.59" />
   </ItemGroup>
 

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="7.0.0" />
   </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().ConfigureAwait(false);
             }
             catch (EndOfStreamException eose)
             {
@@ -56,7 +56,7 @@ private void MainLoop()
             }
             catch (HardProtocolException hpe)
             {
-                HardProtocolExceptionHandler(hpe);
+                await HardProtocolExceptionHandler(hpe).ConfigureAwait(false);
             }
             catch (Exception ex)
             {
@@ -66,57 +66,70 @@ private void MainLoop()
             FinishClose();
         }
 
-        private void ReceiveLoop()
+        private async Task ReceiveLoop()
         {
             while (!_closed)
             {
-                InboundFrame frame = _frameHandler.ReadFrame();
+                while (_frameHandler.TryReadFrame(out InboundFrame frame))
+                {
+                    NotifyHeartbeatListener();
+                    ProcessFrame(frame);
+                }
+
+                // Done reading frames synchronously, go async
+                InboundFrame asyncFrame = await _frameHandler.ReadFrameAsync()
+                   .ConfigureAwait(false);
                 NotifyHeartbeatListener();
+                ProcessFrame(asyncFrame);
+            }
+        }
 
-                bool shallReturn = true;
-                if (frame.Channel == 0)
+        private void ProcessFrame(InboundFrame frame)
+        {
+            bool shallReturnPayload = true;
+            if (frame.Channel == 0)
+            {
+                if (frame.Type == FrameType.FrameHeartbeat)
                 {
-                    if (frame.Type == FrameType.FrameHeartbeat)
-                    {
-                        // Ignore it: we've already just reset the heartbeat
-                    }
-                    else
-                    {
-                        // 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);
-                    }
+                    // Ignore it: we've already recently reset the heartbeat
                 }
                 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);
-                    }
+                    // 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.
+                    shallReturnPayload = _session0.HandleFrame(in frame);
                 }
-
-                if (shallReturn)
+            }
+            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)
                 {
-                    frame.ReturnPayload();
+                    // 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.)
+                    ISession session = _sessionManager.Lookup(frame.Channel);
+                    shallReturnPayload = session.HandleFrame(in frame);
                 }
             }
+
+            if (shallReturnPayload)
+            {
+                frame.ReturnPayload();
+            }
         }
 
         ///<remarks>
@@ -139,7 +152,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 +164,8 @@ private void HardProtocolExceptionHandler(HardProtocolException hpe)
                     _session0.Transmit(in cmd);
                     if (hpe.CanShutdownCleanly)
                     {
-                        ClosingLoop();
+                        await ClosingLoop()
+                           .ConfigureAwait(false);
                     }
                 }
                 catch (IOException ioe)
@@ -168,13 +182,14 @@ 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()
+                   .ConfigureAwait(false);
             }
             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();