Increasing buffer size for both send/receive for better throughput.

Removing use og SunchronizedList and replacing with just two longs to keep track of outstanding confirms.

Author: Stefán J. Sigurðarson

projects/client/RabbitMQ.Client/src/client/api/ConnectionFactoryBase.cs

@@ -59,7 +59,9 @@ public static ITcpClient DefaultSocketFactory(AddressFamily addressFamily)
         {
             var socket = new Socket(addressFamily, SocketType.Stream, ProtocolType.Tcp)
             {
-                NoDelay = true
+                NoDelay = true,
+                ReceiveBufferSize = 65536,
+                SendBufferSize = 65536
             };
             return new TcpClientAdapter(socket);
         }

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

@@ -71,7 +71,7 @@ interface IFrameHandler
 
         void SendHeader();
 
-        void WriteFrame(OutboundFrame frame);
+        void WriteFrame(OutboundFrame frame, bool flush = true);
 
         void WriteFrameSet(IList<OutboundFrame> frames);
     }

projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs

@@ -42,6 +42,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
+using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
 
@@ -69,12 +70,15 @@ abstract class ModelBase : IFullModel, IRecoverable
 
         private readonly object _shutdownLock = new object();
         private readonly object _rpcLock = new object();
+        private readonly object _confirmLock = new object();
 
-        private readonly SynchronizedList<ulong> _unconfirmedSet = new SynchronizedList<ulong>();
+        private ulong _maxDeliveryId;
+        private ulong _deliveredItems = 0;
 
         private EventHandler<ShutdownEventArgs> _modelShutdown;
 
         private bool _onlyAcksReceived = true;
+        private ulong _nextPublishSeqNo;
 
         public IConsumerDispatcher ConsumerDispatcher { get; private set; }
 
@@ -99,7 +103,7 @@ public ModelBase(ISession session, ConsumerWorkService workService)
         protected void Initialise(ISession session)
         {
             CloseReason = null;
-            NextPublishSeqNo = 0;
+            _nextPublishSeqNo = 0;
             Session = session;
             Session.CommandReceived = HandleCommand;
             Session.SessionShutdown += OnSessionShutdown;
@@ -176,7 +180,9 @@ public bool IsOpen
             get { return CloseReason == null; }
         }
 
-        public ulong NextPublishSeqNo { get; private set; }
+
+
+        public ulong NextPublishSeqNo { get => _nextPublishSeqNo; }
 
         public ISession Session { get; private set; }
 
@@ -490,8 +496,10 @@ public virtual void OnModelShutdown(ShutdownEventArgs reason)
                     }
                 }
             }
-            lock (_unconfirmedSet.SyncRoot)
-                Monitor.Pulse(_unconfirmedSet.SyncRoot);
+            lock (_confirmLock)
+            {
+                Monitor.Pulse(_confirmLock);
+            }
 
             _flowControlBlock.Set();
         }
@@ -1069,20 +1077,12 @@ public abstract void BasicNack(ulong deliveryTag,
 
         internal void AllocatatePublishSeqNos(int count)
         {
-            int c = 0;
-            lock (_unconfirmedSet.SyncRoot)
+
+            lock (_confirmLock)
             {
-                while (c < count)
+                if (_nextPublishSeqNo > 0)
                 {
-                    if (NextPublishSeqNo > 0)
-                    {
-                        if (!_unconfirmedSet.Contains(NextPublishSeqNo))
-                        {
-                            _unconfirmedSet.Add(NextPublishSeqNo);
-                        }
-                        NextPublishSeqNo++;
-                    }
-                    c++;
+                    _nextPublishSeqNo = InterlockedEx.Add(ref _nextPublishSeqNo, (ulong)count);
                 }
             }
         }
@@ -1102,17 +1102,14 @@ public void BasicPublish(string exchange,
             {
                 basicProperties = CreateBasicProperties();
             }
-            if (NextPublishSeqNo > 0)
+            if (_nextPublishSeqNo > 0)
             {
-                lock (_unconfirmedSet.SyncRoot)
+                lock (_confirmLock)
                 {
-                    if (!_unconfirmedSet.Contains(NextPublishSeqNo))
-                    {
-                        _unconfirmedSet.Add(NextPublishSeqNo);
-                    }
-                    NextPublishSeqNo++;
+                    _nextPublishSeqNo = InterlockedEx.Increment(ref _nextPublishSeqNo);
                 }
             }
+
             _Private_BasicPublish(exchange,
                 routingKey,
                 mandatory,
@@ -1170,8 +1167,9 @@ public void ConfirmSelect()
         {
             if (NextPublishSeqNo == 0UL)
             {
-                NextPublishSeqNo = 1;
+                _nextPublishSeqNo = 1;
             }
+
             _Private_ConfirmSelect(false);
         }
 
@@ -1331,7 +1329,7 @@ public bool WaitForConfirms(TimeSpan timeout, out bool timedOut)
             }
             bool isWaitInfinite = timeout.TotalMilliseconds == Timeout.Infinite;
             Stopwatch stopwatch = Stopwatch.StartNew();
-            lock (_unconfirmedSet.SyncRoot)
+            lock (_confirmLock)
             {
                 while (true)
                 {
@@ -1340,7 +1338,7 @@ public bool WaitForConfirms(TimeSpan timeout, out bool timedOut)
                         throw new AlreadyClosedException(CloseReason);
                     }
 
-                    if (_unconfirmedSet.Count == 0)
+                    if (_deliveredItems == _nextPublishSeqNo - 1)
                     {
                         bool aux = _onlyAcksReceived;
                         _onlyAcksReceived = true;
@@ -1349,13 +1347,12 @@ public bool WaitForConfirms(TimeSpan timeout, out bool timedOut)
                     }
                     if (isWaitInfinite)
                     {
-                        Monitor.Wait(_unconfirmedSet.SyncRoot);
+                        Monitor.Wait(_confirmLock);
                     }
                     else
                     {
                         TimeSpan elapsed = stopwatch.Elapsed;
-                        if (elapsed > timeout || !Monitor.Wait(
-                            _unconfirmedSet.SyncRoot, timeout - elapsed))
+                        if (elapsed > timeout || !Monitor.Wait(_confirmLock, timeout - elapsed))
                         {
                             timedOut = true;
                             return _onlyAcksReceived;
@@ -1411,32 +1408,27 @@ internal void SendCommands(IList<Command> commands)
 
         protected virtual void handleAckNack(ulong deliveryTag, bool multiple, bool isNack)
         {
-            lock (_unconfirmedSet.SyncRoot)
+            lock (_confirmLock)
             {
-                if (multiple)
-                {
-                    for (ulong i = _unconfirmedSet[0]; i <= deliveryTag; i++)
-                    {
-                        // removes potential duplicates
-                        while (_unconfirmedSet.Remove(i))
-                        {
-                        }
-                    }
-                }
-                else
+                _deliveredItems = InterlockedEx.Increment(ref _deliveredItems);
+
+                if (multiple && _maxDeliveryId < deliveryTag)
                 {
-                    while (_unconfirmedSet.Remove(deliveryTag))
-                    {
-                    }
+                    _maxDeliveryId = deliveryTag;
                 }
+
+                _deliveredItems = Math.Max(_maxDeliveryId, _deliveredItems);
                 _onlyAcksReceived = _onlyAcksReceived && !isNack;
-                if (_unconfirmedSet.Count == 0)
+                if (_deliveredItems == _nextPublishSeqNo - 1)
                 {
-                    Monitor.Pulse(_unconfirmedSet.SyncRoot);
+                    Monitor.Pulse(_confirmLock);
                 }
             }
+
         }
 
+
+
         private QueueDeclareOk QueueDeclare(string queue, bool passive, bool durable, bool exclusive,
             bool autoDelete, IDictionary<string, object> arguments)
         {
@@ -1478,5 +1470,26 @@ public class QueueDeclareRpcContinuation : SimpleBlockingRpcContinuation
         {
             public QueueDeclareOk m_result;
         }
+
+        public static class InterlockedEx
+        {
+            public static ulong Increment(ref ulong location)
+            {
+                long incrementedSigned = Interlocked.Increment(ref Unsafe.As<ulong, long>(ref location));
+                return Unsafe.As<long, ulong>(ref incrementedSigned);
+            }
+
+            public static ulong Decrement(ref ulong location)
+            {
+                long decrementedSigned = Interlocked.Decrement(ref Unsafe.As<ulong, long>(ref location));
+                return Unsafe.As<long, ulong>(ref decrementedSigned);
+            }
+
+            public static ulong Add(ref ulong location, ulong value)
+            {
+                long addSigned = Interlocked.Add(ref Unsafe.As<ulong, long>(ref location), Unsafe.As<ulong, long>(ref value));
+                return Unsafe.As<long, ulong>(ref addSigned);
+            }
+        }
     }
 }

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

@@ -121,7 +121,7 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
                 }
             }
             _reader = new BufferedStream(netstream, _socket.Client.ReceiveBufferSize);
-            _writer = netstream;
+            _writer = new BufferedStream(netstream, _socket.Client.SendBufferSize);
 
             WriteTimeout = writeTimeout;
         }
@@ -222,10 +222,10 @@ public void SendHeader()
                 headerBytes[7] = (byte)Endpoint.Protocol.MinorVersion;
             }
 
-            Write(new ArraySegment<byte>(headerBytes));
+            Write(new ArraySegment<byte>(headerBytes), true);
         }
 
-        public void WriteFrame(OutboundFrame frame)
+        public void WriteFrame(OutboundFrame frame, bool flush = true)
         {
             int bufferSize = frame.GetMinimumBufferSize();
             using (IMemoryOwner<byte> memory = MemoryPool<byte>.Shared.Rent(bufferSize))
@@ -235,11 +235,10 @@ public void WriteFrame(OutboundFrame frame)
                 _socket.Client.Poll(_writeableStateTimeoutMicroSeconds, SelectMode.SelectWrite);
                 if (MemoryMarshal.TryGetArray(slice.Slice(0, frame.ByteCount), out ArraySegment<byte> segment))
                 {
-                    Write(segment);
+                    Write(segment, flush);
                     return;
                 }
 
-
                 throw new InvalidOperationException("Unable to get array segment from memory.");
             }
         }
@@ -248,7 +247,11 @@ public void WriteFrameSet(IList<OutboundFrame> frames)
         {
             for (int i = 0; i < frames.Count; i++)
             {
-                WriteFrame(frames[i]);
+                WriteFrame(frames[i], false);
+            }
+            lock (_streamLock)
+            {
+                _writer.Flush();
             }
             /*
             int bufferSize = 0;
@@ -282,11 +285,15 @@ public void WriteFrameSet(IList<OutboundFrame> frames)
             */
         }
 
-        private void Write(ArraySegment<byte> bufferSegment)
+        private void Write(ArraySegment<byte> bufferSegment, bool flush)
         {
             lock (_streamLock)
             {
                 _writer.Write(bufferSegment.Array, bufferSegment.Offset, bufferSegment.Count);
+                if (flush)
+                {
+                    _writer.Flush();
+                }
             }
         }