Optimizing memory usage even further.

projects/RabbitMQ.Client/client/api/AsyncDefaultBasicConsumer.cs

@@ -5,8 +5,6 @@
 
 using RabbitMQ.Client.Events;
 
-using TaskExtensions = RabbitMQ.Client.Impl.TaskExtensions;
-
 namespace RabbitMQ.Client
 {
     public class AsyncDefaultBasicConsumer : IBasicConsumer, IAsyncBasicConsumer
@@ -96,7 +94,7 @@ public virtual Task HandleBasicConsumeOk(string consumerTag)
         {
             _consumerTags.Add(consumerTag);
             IsRunning = true;
-            return TaskExtensions.CompletedTask;
+            return Task.CompletedTask;
         }
 
         /// <summary>
@@ -118,7 +116,7 @@ public virtual Task HandleBasicDeliver(string consumerTag,
             ReadOnlyMemory<byte> body)
         {
             // Nothing to do here.
-            return TaskExtensions.CompletedTask;
+            return Task.CompletedTask;
         }
 
         /// <summary>

projects/RabbitMQ.Client/client/api/IBasicPublishBatch.cs

@@ -37,6 +37,8 @@
 //  The Initial Developer of the Original Code is Pivotal Software, Inc.
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
+using System;
+
 namespace RabbitMQ.Client
 {
     public interface IBasicPublishBatch

projects/RabbitMQ.Client/client/exceptions/UnexpectedFrameException.cs

@@ -48,14 +48,10 @@ namespace RabbitMQ.Client.Exceptions
     /// </summary>
     public class UnexpectedFrameException : HardProtocolException
     {
-        internal UnexpectedFrameException(Frame frame)
-            : base("A frame of this type was not expected at this time")
+        internal UnexpectedFrameException(FrameType frameType) : base($"A frame of type {frameType} was not expected at this time")
         {
-            Frame = frame;
         }
 
-        internal  Frame Frame { get; }
-
         public override ushort ReplyCode
         {
             get { return Constants.CommandInvalid; }

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

@@ -45,11 +45,12 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
             string exchange,
             string routingKey,
             IBasicProperties basicProperties,
-            ReadOnlyMemory<byte> body)
+            ReadOnlySpan<byte> body)
         {
-            IMemoryOwner<byte> bodyCopy = MemoryPool<byte>.Shared.Rent(body.Length);
-            body.CopyTo(bodyCopy.Memory);
-            ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, bodyCopy, body.Length));
+            byte[] bodyBytes = ArrayPool<byte>.Shared.Rent(body.Length);
+            Memory<byte> bodyCopy = new Memory<byte>(bodyBytes, 0, body.Length);
+            body.CopyTo(bodyCopy.Span);
+            ScheduleUnlessShuttingDown(new BasicDeliver(consumer, consumerTag, deliveryTag, redelivered, exchange, routingKey, basicProperties, bodyCopy));
         }
 
         public void HandleBasicCancelOk(IBasicConsumer consumer, string consumerTag)

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

@@ -1,6 +1,7 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Runtime.InteropServices;
 using System.Threading.Tasks;
 
 using RabbitMQ.Client.Events;
@@ -15,8 +16,7 @@ sealed class BasicDeliver : Work
         readonly string _exchange;
         readonly string _routingKey;
         readonly IBasicProperties _basicProperties;
-        readonly IMemoryOwner<byte> _body;
-        readonly int _bodyLength;
+        readonly ReadOnlyMemory<byte> _body;
 
         public BasicDeliver(IBasicConsumer consumer, 
             string consumerTag, 
@@ -25,8 +25,7 @@ public BasicDeliver(IBasicConsumer consumer,
             string exchange, 
             string routingKey, 
             IBasicProperties basicProperties,
-            IMemoryOwner<byte> body,
-            int bodyLength) : base(consumer)
+            ReadOnlyMemory<byte> body) : base(consumer)
         {
             _consumerTag = consumerTag;
             _deliveryTag = deliveryTag;
@@ -35,7 +34,6 @@ public BasicDeliver(IBasicConsumer consumer,
             _routingKey = routingKey;
             _basicProperties = basicProperties;
             _body = body;
-            _bodyLength = bodyLength;
         }
 
         protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
@@ -48,7 +46,7 @@ await consumer.HandleBasicDeliver(_consumerTag,
                     _exchange,
                     _routingKey,
                     _basicProperties,
-                    _body.Memory.Slice(0, _bodyLength)).ConfigureAwait(false);
+                    _body).ConfigureAwait(false);
             }
             catch (Exception e)
             {
@@ -61,7 +59,10 @@ await consumer.HandleBasicDeliver(_consumerTag,
             }
             finally
             {
-                _body.Dispose();
+                if (MemoryMarshal.TryGetArray(_body, out ArraySegment<byte> segment))
+                {
+                    ArrayPool<byte>.Shared.Return(segment.Array);
+                }
             }
         }
     }

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

@@ -38,6 +38,8 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
+using System.Buffers;
 using System.Collections.Generic;
 
 using RabbitMQ.Client.Framing.Impl;
@@ -63,7 +65,7 @@ public void Add(string exchange, string routingKey, bool mandatory, IBasicProper
                 _mandatory = mandatory
             };
 
-            _commands.Add(new Command(method, (ContentHeaderBase)bp, body));
+            _commands.Add(new Command(method, (ContentHeaderBase)bp, body, false));
         }
 
         public void Publish()

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

@@ -41,6 +41,7 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Runtime.InteropServices;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing.Impl;
 
@@ -54,31 +55,23 @@ class Command : IDisposable
         // - 4 bytes of frame payload length
         // - 1 byte of payload trailer FrameEnd byte
         private const int EmptyFrameSize = 8;
-        private static readonly byte[] s_emptyByteArray = new byte[0];
-        private readonly IMemoryOwner<byte> _body;
+        private readonly bool _returnBufferOnDispose;
 
         static Command()
         {
             CheckEmptyFrameSize();
         }
 
-        internal Command(MethodBase method) : this(method, null, null, 0)
+        internal Command(MethodBase method) : this(method, null, null, false)
         {
         }
 
-        internal Command(MethodBase method, ContentHeaderBase header, ReadOnlyMemory<byte> body)
+        public Command(MethodBase method, ContentHeaderBase header, ReadOnlyMemory<byte> body, bool returnBufferOnDispose)
         {
             Method = method;
             Header = header;
             Body = body;
-        }
-
-        public Command(MethodBase method, ContentHeaderBase header, IMemoryOwner<byte> body, int bodySize)
-        {
-            Method = method;
-            Header = header;
-            _body = body;
-            Body = _body?.Memory.Slice(0, bodySize) ?? s_emptyByteArray;
+            _returnBufferOnDispose = returnBufferOnDispose;
         }
 
         public ReadOnlyMemory<byte> Body { get; private set; }
@@ -167,9 +160,9 @@ internal static List<OutboundFrame> CalculateFrames(int channelNumber, Connectio
 
         public void Dispose()
         {
-            if (_body is IMemoryOwner<byte>)
+            if(_returnBufferOnDispose && MemoryMarshal.TryGetArray(Body, out ArraySegment<byte> segment))
             {
-                _body.Dispose();
+                ArrayPool<byte>.Shared.Return(segment.Array);
             }
         }
     }

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

@@ -38,6 +38,7 @@
 //  Copyright (c) 2007-2020 VMware, Inc.  All rights reserved.
 //---------------------------------------------------------------------------
 
+using System;
 using System.Buffers;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing.Impl;
@@ -59,7 +60,7 @@ class CommandAssembler
 
         public MethodBase m_method;
         public ContentHeaderBase m_header;
-        public IMemoryOwner<byte> m_body;
+        public Memory<byte> m_body;
         public ProtocolBase m_protocol;
         public int m_remainingBodyBytes;
         private int _offset;
@@ -78,39 +79,40 @@ public Command HandleFrame(InboundFrame f)
                 case AssemblyState.ExpectingMethod:
                     if (!f.IsMethod())
                     {
-                        throw new UnexpectedFrameException(f);
+                        throw new UnexpectedFrameException(f.Type);
                     }
                     m_method = m_protocol.DecodeMethodFrom(f.Payload);
                     m_state = m_method.HasContent ? AssemblyState.ExpectingContentHeader : AssemblyState.Complete;
                     return CompletedCommand();
                 case AssemblyState.ExpectingContentHeader:
                     if (!f.IsHeader())
                     {
-                        throw new UnexpectedFrameException(f);
+                        throw new UnexpectedFrameException(f.Type);
                     }
                     m_header = m_protocol.DecodeContentHeaderFrom(NetworkOrderDeserializer.ReadUInt16(f.Payload));
                     ulong totalBodyBytes = m_header.ReadFrom(f.Payload.Slice(2));
                     if (totalBodyBytes > MaxArrayOfBytesSize)
                     {
-                        throw new UnexpectedFrameException(f);
+                        throw new UnexpectedFrameException(f.Type);
                     }
 
                     m_remainingBodyBytes = (int)totalBodyBytes;
-                    m_body = MemoryPool<byte>.Shared.Rent(m_remainingBodyBytes);
+                    byte[] bodyBytes = ArrayPool<byte>.Shared.Rent(m_remainingBodyBytes);
+                    m_body = new Memory<byte>(bodyBytes, 0, m_remainingBodyBytes);
                     UpdateContentBodyState();
                     return CompletedCommand();
                 case AssemblyState.ExpectingContentBody:
                     if (!f.IsBody())
                     {
-                        throw new UnexpectedFrameException(f);
+                        throw new UnexpectedFrameException(f.Type);
                     }
 
                     if (f.Payload.Length > m_remainingBodyBytes)
                     {
                         throw new MalformedFrameException($"Overlong content body received - {m_remainingBodyBytes} bytes remaining, {f.Payload.Length} bytes received");
                     }
 
-                    f.Payload.CopyTo(m_body.Memory.Slice(_offset));
+                    f.Payload.CopyTo(m_body.Slice(_offset));
                     m_remainingBodyBytes -= f.Payload.Length;
                     _offset += f.Payload.Length;
                     UpdateContentBodyState();
@@ -125,7 +127,7 @@ private Command CompletedCommand()
         {
             if (m_state == AssemblyState.Complete)
             {
-                Command result = new Command(m_method, m_header, m_body, _offset);
+                Command result = new Command(m_method, m_header, m_body, true);
                 Reset();
                 return result;
             }

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

@@ -62,10 +62,11 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
                                        string exchange,
                                        string routingKey,
                                        IBasicProperties basicProperties,
-                                       ReadOnlyMemory<byte> body)
+                                       ReadOnlySpan<byte> body)
         {
-            IMemoryOwner<byte> memoryCopy = MemoryPool<byte>.Shared.Rent(body.Length);
-            body.CopyTo(memoryCopy.Memory);
+            byte[] memoryCopyArray = ArrayPool<byte>.Shared.Rent(body.Length);
+            Memory<byte> memoryCopy = new Memory<byte>(memoryCopyArray, 0, body.Length);
+            body.CopyTo(memoryCopy.Span);
             UnlessShuttingDown(() =>
             {
                 try
@@ -76,7 +77,7 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
                                                 exchange,
                                                 routingKey,
                                                 basicProperties,
-                                                memoryCopy.Memory.Slice(0, body.Length));
+                                                memoryCopy);
                 }
                 catch (Exception e)
                 {
@@ -89,7 +90,7 @@ public void HandleBasicDeliver(IBasicConsumer consumer,
                 }
                 finally
                 {
-                    memoryCopy.Dispose();
+                    ArrayPool<byte>.Shared.Return(memoryCopyArray);
                 }
             });
         }