Refactor code

.github/workflows/ReleaseNotes.md

@@ -1,5 +1,7 @@
 * [Core] Fixed a memory leak in _AsyncSignal_ implementation (#1586, thanks to @mario-zelger).
 * [Core] Fixed an issue with bounds handling in _MqttBufferReader_ (#1593).
+* [Core] Performance improvements and reduced memory usage due to improved buffer copies (#1599, thanks to @xljiulang).
 * [Server] Fix not properly reset statistics (#1587, thanks to @damikun).
 * [Server] Now using an empty string as the sender client ID for injected application messages (#1583, thanks to @xljiulang).
-* [Server] Improved memory usage for ASP.NET connections (#1582, thanks to @xljiulang).
+* [Server] Improved memory usage for ASP.NET connections (#1582, thanks to @xljiulang).
+* [Server] Improved memory usage and performance for ASP.NET integration (#1596, thanks to @xljiulang).

Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs

@@ -33,7 +33,7 @@ namespace MQTTnet.Benchmarks
         [Benchmark]
         public void Memory_Copy()
         {
-            MQTTnet.Internal.Memory.Copy(source, 0, target, 0, Length);
+            MQTTnet.Internal.MqttMemoryHelper.Copy(source, 0, target, 0, Length);
         }
 
     }

Source/MQTTnet.Tests/Formatter/MqttBufferReader_Tests.cs

@@ -2,6 +2,9 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+using System;
+using System.Collections.Generic;
+using System.IO;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
@@ -67,6 +70,172 @@ namespace MQTTnet.Tests.Formatter
             Assert.AreEqual(5, remainingData.Length);
         }
 
+        [TestMethod]
+        public void Read_Various_Positions_and_Offsets()
+        {
+            const int NumBufferElements = 1000;
+            const int NumTestSequences = 10000;
+
+            const string TestString = "The quick brown fox jumps over the lazy dog.";
+
+            var rndSeed = Environment.TickCount;
+            // rndSeed = ...; // assign random seed here if there is an error so that it can be repeated
+            var rnd = new Random(rndSeed);
+            // rndSeed is printed when an assert fails
+
+            var bufferElements = new List<ElementReference>();
+
+            byte[] elementBuffer;
+
+            // Fill a buffer with strings or numbers in big endian format
+            using (var ms = new MemoryStream())
+            {
+                using (var bw = new BinaryWriter(ms))
+                {
+                    var bufferOffset = 0;
+
+                    for (var i = 0; i < NumBufferElements; i++)
+                    {
+                        var nextElementType = (ElementReference.BufferElementType)rnd.Next(ElementReference.NumBufferElementTypes);
+
+                        byte[] elementBytes = null;
+                        uint elementNumberValue = 0; // all element number values fit into uint
+                        string elementStringValue = null;
+                        var elementSize = 0;
+
+                        var alreadyBigEndian = false;
+                        switch (nextElementType)
+                        {
+                            case ElementReference.BufferElementType.Byte:
+                                var byteValue = (byte)i;
+                                elementNumberValue = byteValue;
+                                elementSize = sizeof(byte);
+                                elementBytes = new byte[1] { byteValue };
+                                alreadyBigEndian = true; // nothing to swap
+                                break;
+                            case ElementReference.BufferElementType.TwoByteInt:
+                                var ushortValue = (ushort)i;
+                                elementNumberValue = ushortValue;
+                                elementSize = sizeof(ushort);
+                                elementBytes = BitConverter.GetBytes(ushortValue);
+                                break;
+                            case ElementReference.BufferElementType.FourByteInt:
+                                var uintValue = (uint)i;
+                                elementNumberValue = uintValue;
+                                elementSize = sizeof(uint);
+                                elementBytes = BitConverter.GetBytes(uintValue);
+                                break;
+                            case ElementReference.BufferElementType.VariableSizeInt:
+                            {
+                                elementNumberValue = (uint)i;
+                                var writer = new MqttBufferWriter(4, 4);
+                                writer.WriteVariableByteInteger(elementNumberValue);
+                                elementSize = writer.Length;
+                                elementBytes = new byte[elementSize];
+                                var buffer = writer.GetBuffer();
+                                Array.Copy(buffer, elementBytes, elementSize);
+                                alreadyBigEndian = true; // nothing to swap
+                            }
+                                break;
+                            case ElementReference.BufferElementType.String:
+                            {
+                                var stringLen = rnd.Next(TestString.Length);
+                                elementStringValue = TestString.Substring(0, stringLen); // could be empty
+                                var writer = new MqttBufferWriter(stringLen + 1, stringLen + 1);
+                                writer.WriteString(elementStringValue);
+                                elementSize = writer.Length;
+                                elementBytes = new byte[elementSize];
+                                var buffer = writer.GetBuffer();
+                                Array.Copy(buffer, elementBytes, elementSize);
+                                alreadyBigEndian = true; // nothing to swap
+                            }
+                                break;
+                        }
+
+                        if (BitConverter.IsLittleEndian && !alreadyBigEndian)
+                        {
+                            Array.Reverse(elementBytes);
+                        }
+
+                        bw.Write(elementBytes);
+                        bufferElements.Add(new ElementReference(nextElementType, elementSize, elementNumberValue, elementStringValue, bufferOffset));
+                        bufferOffset += elementSize;
+                    }
+
+                    elementBuffer = ms.ToArray();
+                }
+            }
+
+            for (var i = 0; i < NumTestSequences; i++)
+            {
+                var firstElementIndex = rnd.Next(NumBufferElements - 1);
+                // segmentElementIndex will be < NumBufferElements - 1, which means there is at least one byte left for segment element count
+                var elementCount = rnd.Next(NumBufferElements - firstElementIndex) + 1;
+                var lastElementIndex = firstElementIndex + elementCount - 1;
+                // Use element index to get buffer offsets
+                var lastElement = bufferElements[lastElementIndex];
+                var segmentStartPosition = bufferElements[firstElementIndex].BufferOffset;
+                var segmentEndPosition = lastElement.BufferOffset + lastElement.Size;
+                var segmentLength = segmentEndPosition - segmentStartPosition;
+
+                var reader = new MqttBufferReader();
+                reader.SetBuffer(elementBuffer, segmentStartPosition, segmentLength);
+
+                // read all elements in the buffer segment; values should be as expected              
+                for (var n = 0; n < elementCount; n++)
+                {
+                    var element = bufferElements[firstElementIndex + n];
+                    uint elementNumberValue = 0;
+                    string elementStringValue = null;
+
+                    var expectedPosition = element.BufferOffset - segmentStartPosition;
+                    Assert.AreEqual(expectedPosition, reader.Position);
+
+                    switch (element.Type)
+                    {
+                        case ElementReference.BufferElementType.Byte:
+                        {
+                            elementNumberValue = reader.ReadByte();
+                        }
+                            break;
+                        case ElementReference.BufferElementType.TwoByteInt:
+                        {
+                            elementNumberValue = reader.ReadTwoByteInteger();
+                        }
+                            break;
+                        case ElementReference.BufferElementType.FourByteInt:
+                        {
+                            elementNumberValue = reader.ReadFourByteInteger();
+                        }
+                            break;
+                        case ElementReference.BufferElementType.VariableSizeInt:
+                        {
+                            elementNumberValue = reader.ReadVariableByteInteger();
+                        }
+                            break;
+                        case ElementReference.BufferElementType.String:
+                        {
+                            elementStringValue = reader.ReadString();
+                        }
+                            break;
+                    }
+
+                    if (elementStringValue != null)
+                    {
+                        Assert.AreEqual(elementStringValue, element.StringValue, $"'{elementStringValue}' not equal '{element.StringValue}' with random seed {rndSeed}.");
+                    }
+                    else
+                    {
+                        Assert.AreEqual(elementNumberValue, element.NumberValue, $"{elementNumberValue} not equal {element.NumberValue} with random seed {rndSeed}.");
+                    }
+                }
+
+                // confirm end of stream
+                Assert.AreEqual(0, reader.BytesLeft, "BytesLeft not zero as expected with random seed " + rndSeed);
+                Assert.IsTrue(reader.EndOfStream, "Not end of stream as expected with random seed " + rndSeed);
+            }
+        }
+
         [TestMethod]
         public void Report_Correct_Length_For_Full_Buffer()
         {
@@ -106,5 +275,36 @@ namespace MQTTnet.Tests.Formatter
             Assert.IsFalse(reader.EndOfStream);
             Assert.AreEqual(5, reader.BytesLeft);
         }
+
+
+        // Helper class to build up a reference to elements of various types in a buffer
+        class ElementReference
+        {
+            public enum BufferElementType
+            {
+                Byte,
+                TwoByteInt,
+                FourByteInt,
+                VariableSizeInt,
+                String
+            }
+
+            public const int NumBufferElementTypes = 5;
+
+            public ElementReference(BufferElementType type, int size, uint numberValue, string stringValue, int bufferOffset)
+            {
+                Type = type;
+                Size = size;
+                NumberValue = numberValue;
+                StringValue = stringValue;
+                BufferOffset = bufferOffset;
+            }
+
+            public int BufferOffset { get; }
+            public uint NumberValue { get; }
+            public int Size { get; }
+            public string StringValue { get; }
+            public BufferElementType Type { get; }
+        }
     }
 }

Source/MQTTnet/Formatter/MqttBufferReader.cs

@@ -2,21 +2,23 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using MQTTnet.Exceptions;
-using MQTTnet.Internal;
 using System;
 using System.Runtime.CompilerServices;
-using System.Runtime.InteropServices;
 using System.Text;
+using MQTTnet.Exceptions;
+using MQTTnet.Internal;
+#if NETCOREAPP3_0_OR_GREATER
+using System.Buffers.Binary;
+#endif
 
 namespace MQTTnet.Formatter
 {
     public sealed class MqttBufferReader
     {
         byte[] _buffer = EmptyBuffer.Array;
+        int _maxPosition;
         int _offset;
         int _position;
-        int _maxPosition;
 
         public int BytesLeft => _maxPosition - _position;
 
@@ -36,7 +38,7 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(length);
 
             var result = new byte[length];
-            Memory.Copy(_buffer, _position, result, 0, length);
+            MqttMemoryHelper.Copy(_buffer, _position, result, 0, length);
             _position += length;
 
             return result;
@@ -53,11 +55,7 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(4);
 
 #if NETCOREAPP3_0_OR_GREATER
-            var value = MemoryMarshal.Read<uint>(_buffer.AsSpan(_position));
+            var value = BinaryPrimitives.ReadUInt32BigEndian(_buffer.AsSpan(_position));
-            if (BitConverter.IsLittleEndian)
-            {
-                value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
-            }
 #else
             var byte0 = _buffer[_position];
             var byte1 = _buffer[_position + 1];
@@ -80,7 +78,7 @@ namespace MQTTnet.Formatter
             }
 
             var buffer = new byte[bufferLength];
-            Memory.Copy(_buffer, _position, buffer, 0, bufferLength);
+            MqttMemoryHelper.Copy(_buffer, _position, buffer, 0, bufferLength);
             _position += bufferLength;
 
             return buffer;
@@ -102,7 +100,7 @@ namespace MQTTnet.Formatter
             var result = Encoding.UTF8.GetString(_buffer.AsSpan(_position, length));
 #else
             var result = Encoding.UTF8.GetString(_buffer, _position, length);
-#endif            
+#endif
 
             _position += length;
             return result;
@@ -113,11 +111,7 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(2);
 
 #if NETCOREAPP3_0_OR_GREATER
-            var value = MemoryMarshal.Read<ushort>(_buffer.AsSpan(_position));
+            var value = BinaryPrimitives.ReadUInt16BigEndian(_buffer.AsSpan(_position));
-            if (BitConverter.IsLittleEndian)
-            {
-                value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
-            }
 #else
             var msb = _buffer[_position];
             var lsb = _buffer[_position + 1];

Source/MQTTnet/Formatter/MqttBufferWriter.cs

@@ -126,7 +126,7 @@ namespace MQTTnet.Formatter
 
             EnsureAdditionalCapacity(count);
 
-            Memory.Copy(buffer, offset, _buffer, _position, count);
+            MqttMemoryHelper.Copy(buffer, offset, _buffer, _position, count);
             IncreasePosition(count);
         }
 
@@ -150,7 +150,7 @@ namespace MQTTnet.Formatter
                 _buffer[_position] = (byte)(valueLength >> 8);
                 _buffer[_position + 1] = (byte)valueLength;
 
-                Memory.Copy(value, 0, _buffer, _position + 2, valueLength);
+                MqttMemoryHelper.Copy(value, 0, _buffer, _position + 2, valueLength);
                 IncreasePosition(valueLength + 2);
             }
         }

Source/MQTTnet/Formatter/MqttPacketBuffer.cs

@@ -43,8 +43,8 @@ namespace MQTTnet.Formatter
             }
 
             var buffer = new byte[Length];
-            Memory.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
+            MqttMemoryHelper.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
-            Memory.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
+            MqttMemoryHelper.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
 
             return buffer;
         }
@@ -57,8 +57,8 @@ namespace MQTTnet.Formatter
             }
 
             var buffer = new byte[Length];
-            Memory.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
+            MqttMemoryHelper.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
-            Memory.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
+            MqttMemoryHelper.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
 
             return new ArraySegment<byte>(buffer);
         }

Source/MQTTnet/Internal/MqttMemoryHelper.cs

@@ -3,7 +3,7 @@ using System.Runtime.CompilerServices;
 
 namespace MQTTnet.Internal
 {
-    public static class Memory
+    public static class MqttMemoryHelper
     {
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public static void Copy(byte[] source, int sourceIndex, byte[] destination, int destinationIndex, int length)
@@ -13,11 +13,11 @@ namespace MQTTnet.Internal
 #elif NET461_OR_GREATER || NETSTANDARD1_3_OR_GREATER
             unsafe
             {
-                fixed (byte* pSoure = &source[sourceIndex])
+                fixed (byte* sourceHandle = &source[sourceIndex])
                 {
-                    fixed (byte* pDestination = &destination[destinationIndex])
+                    fixed (byte* destinationHandle = &destination[destinationIndex])
                     {
-                        System.Buffer.MemoryCopy(pSoure, pDestination, length, length);
+                        System.Buffer.MemoryCopy(sourceHandle, destinationHandle, length, length);
                     }
                 }
             }