Reducing thread usage for connections, merging the pipeline read loop into the MainLoop in the Connection, getting rid of those channels while at it.

Author: Stefán J. Sigurðarson

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

@@ -33,6 +33,7 @@
 using System.Buffers;
 using System.Collections.Generic;
 using System.IO;
+using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
 using System.Reflection;
@@ -342,10 +343,29 @@ public async ValueTask ClosingLoop()
             {
                 _frameHandler.ReadTimeout = TimeSpan.Zero;
                 // Wait for response/socket closure or timeout
+                bool allowSync = true;
                 while (!_closed)
                 {
-                    await _frameHandler.FrameReader.WaitToReadAsync().ConfigureAwait(false);
-                    MainLoopIteration();
+                    // Let's read some bytes
+                    if (!(allowSync && _frameHandler.PipeReader.TryRead(out ReadResult readResult)))
+                    {
+                        readResult = await _frameHandler.PipeReader.ReadAsync().ConfigureAwait(false);
+                    }
+
+                    int handled = 0;
+                    ReadOnlySequence<byte> buffer = readResult.Buffer;
+                    if (!buffer.IsEmpty)
+                    {
+                        handled = MainLoopIteration(ref buffer);
+                    }
+
+                    allowSync = handled != 0;
+                    _frameHandler.PipeReader.AdvanceTo(buffer.Start, buffer.End);
+
+                    if (handled == 0 && readResult.IsCompleted)
+                    {
+                        throw new EndOfStreamException();
+                    }
                 }
             }
             catch (ObjectDisposedException ode)
@@ -484,17 +504,31 @@ public async Task MainLoop()
                 bool shutdownCleanly = false;
                 try
                 {
+                    bool allowSync = true;
                     while (_running)
                     {
-                        ValueTask<bool> waitToRead = _frameHandler.FrameReader.WaitToReadAsync();
-                        if (!waitToRead.IsCompleted)
-                        {
-                            await waitToRead.ConfigureAwait(false);
-                        }
-
                         try
                         {
-                            MainLoopIteration();
+                            // Let's read some bytes
+                            if (!(allowSync && _frameHandler.PipeReader.TryRead(out ReadResult readResult)))
+                            {
+                                readResult = await _frameHandler.PipeReader.ReadAsync().ConfigureAwait(false);
+                            }
+
+                            int handled = 0;
+                            ReadOnlySequence<byte> buffer = readResult.Buffer;
+                            if (!buffer.IsEmpty)
+                            {
+                                handled = MainLoopIteration(ref buffer);
+                            }
+
+                            allowSync = handled != 0;
+                            _frameHandler.PipeReader.AdvanceTo(buffer.Start, buffer.End);
+
+                            if (handled == 0 && readResult.IsCompleted)
+                            {
+                                throw new EndOfStreamException();
+                            }
                         }
                         catch (SoftProtocolException spe)
                         {
@@ -550,9 +584,10 @@ public async Task MainLoop()
             }
         }
 
-        public void MainLoopIteration()
+        public int MainLoopIteration(ref ReadOnlySequence<byte> buffer)
         {
-            while (_frameHandler.FrameReader.TryRead(out InboundFrame frame))
+            int handled = 0;
+            while (InboundFrame.TryReadFrame(ref buffer, out InboundFrame frame))
             {
                 NotifyHeartbeatListener();
 
@@ -597,7 +632,11 @@ public void MainLoopIteration()
                 {
                     frame.ReturnPayload();
                 }
+
+                handled++;
             }
+
+            return handled;
         }
 
         private void NotifyHeartbeatListener()

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

@@ -30,8 +30,8 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.IO.Pipelines;
 using System.Net;
-using System.Threading.Channels;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -53,7 +53,7 @@ internal interface IFrameHandler
         ///<summary>Socket write timeout. System.Threading.Timeout.InfiniteTimeSpan signals "infinity".</summary>
         TimeSpan WriteTimeout { set; }
 
-        ChannelReader<InboundFrame> FrameReader { get; }
+        PipeReader PipeReader { get; }
 
         void Close();
 

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

@@ -30,7 +30,6 @@
 //---------------------------------------------------------------------------
 
 using System;
-using System.Buffers;
 using System.IO;
 using System.IO.Pipelines;
 using System.Net;
@@ -71,12 +70,10 @@ internal class SocketFrameHandler : IFrameHandler
         private readonly ITcpClient _socket;
         private readonly ChannelWriter<OutgoingFrame> _channelWriter;
         private readonly ChannelReader<OutgoingFrame> _channelReader;
-        private readonly ChannelWriter<InboundFrame> _frameWriter;
-        public ChannelReader<InboundFrame> FrameReader { get; }
-        private readonly PipeReader _pipeReader;
+        public PipeReader PipeReader { get; private set; }
+
         private readonly PipeWriter _pipeWriter;
         private readonly Task _writerTask;
-        private readonly Task _readerTask;
         private readonly object _semaphore = new object();
         private readonly byte[] _frameHeaderBuffer;
         private readonly IMeasuredDuplexPipe _pipe;
@@ -94,18 +91,8 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint, Func<AddressFamily, ITcpClie
                     SingleWriter = false
                 });
 
-            var frameChannel = Channel.CreateUnbounded<InboundFrame>(
-                new UnboundedChannelOptions
-                {
-                    AllowSynchronousContinuations = true,
-                    SingleReader = true,
-                    SingleWriter = true
-                });
-
             _channelReader = channel.Reader;
             _channelWriter = channel.Writer;
-            _frameWriter = frameChannel.Writer;
-            FrameReader = frameChannel.Reader;
 
             // Resolve the hostname to know if it's even possible to even try IPv6
             IPAddress[] adds = Dns.GetHostAddresses(endpoint.HostName);
@@ -162,12 +149,11 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint, Func<AddressFamily, ITcpClie
                 _pipe = SocketConnection.Create(_socket.Client);
             }
 
-            _pipeReader = _pipe.Input;
+            PipeReader = _pipe.Input;
             _pipeWriter = _pipe.Output;
 
             WriteTimeout = writeTimeout;
             _writerTask = Task.Run(WriteLoop, CancellationToken.None);
-            _readerTask = Task.Run(ReadLoop, CancellationToken.None);
         }
         public AmqpTcpEndpoint Endpoint { get; set; }
 
@@ -226,7 +212,6 @@ public void Close()
                 if (!_closed)
                 {
                     _channelWriter.TryComplete();
-                    _frameWriter.TryComplete();
                     try { _writerTask.GetAwaiter().GetResult(); } catch { }
 
                     if (_pipe != null)
@@ -246,7 +231,7 @@ public void Close()
 
         public void SendHeader()
         {
-            Span<byte> headerBytes = stackalloc byte[8];
+            Span<byte> headerBytes = _pipeWriter.GetSpan(8);
             headerBytes[0] = (byte)'A';
             headerBytes[1] = (byte)'M';
             headerBytes[2] = (byte)'Q';
@@ -266,64 +251,18 @@ public void SendHeader()
                 headerBytes[7] = (byte)Endpoint.Protocol.MinorVersion;
             }
 
-            Memory<byte> pipeMemory = _pipeWriter.GetMemory(8);
-            headerBytes.CopyTo(pipeMemory.Span);
-            _pipeWriter.Advance(headerBytes.Length);
-            _pipeWriter.FlushAsync().AsTask().GetAwaiter().GetResult();
-        }
-
-        public void Write(OutgoingFrame frame)
-        {
-            _channelWriter.TryWrite(frame);
-        }
-
-        private async Task ReadLoop()
-        {
-            bool allowSync = true;
-            try
+            _pipeWriter.Advance(8);
+            if (!_pipeWriter.FlushAsync().AsTask().Wait(_writeableStateTimeout))
             {
-                while (true)
-                {
-                    // Let's read some bytes
-                    if (!(allowSync && _pipeReader.TryRead(out ReadResult readResult)))
-                    {
-                        readResult = await _pipeReader.ReadAsync().ConfigureAwait(false);
-                    }
-
-                    int handled = 0;
-                    ReadOnlySequence<byte> buffer = readResult.Buffer;
-                    if (!buffer.IsEmpty)
-                    {
-                        handled = ProcessBuffer(ref buffer);
-                    }
-
-                    allowSync = handled != 0;
-                    _pipeReader.AdvanceTo(buffer.Start, buffer.End);
-
-                    if (handled == 0 && readResult.IsCompleted)
-                    {
-                        break;
-                    }
-                }
+                var timeout = new TimeoutException();
+                _pipeWriter.Complete(timeout);
+                _channelWriter.Complete(timeout);
             }
-            catch (Exception e)
-            {
-                _frameWriter.TryComplete(e);
-            }
-
-            _frameWriter.TryComplete(new EndOfStreamException());
         }
 
-        private int ProcessBuffer(ref ReadOnlySequence<byte> buffer)
+        public void Write(OutgoingFrame frame)
         {
-            int handled = 0;
-            while (InboundFrame.TryReadFrame(ref buffer, out InboundFrame frame))
-            {
-                _frameWriter.TryWrite(frame);
-                handled++;
-            }
-
-            return handled;
+            _channelWriter.TryWrite(frame);
         }
 
         private async Task WriteLoop()