Merge branch 'master' of https://github.com/dotnet/MQTTnet

Source/MQTTnet.AspnetCore/ReaderExtensions.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 System;
-using System.Buffers;
 using MQTTnet.Adapter;
 using MQTTnet.Exceptions;
 using MQTTnet.Formatter;
 using MQTTnet.Packets;
+using System;
+using System.Buffers;
+using System.Linq;
+using System.Runtime.InteropServices;
 
 namespace MQTTnet.AspNetCore
 {
     public static class ReaderExtensions
     {
-        public static bool TryDecode(this MqttPacketFormatterAdapter formatter, 
+        public static bool TryDecode(this MqttPacketFormatterAdapter formatter,
-            in ReadOnlySequence<byte> input, 
+            in ReadOnlySequence<byte> input,
-            out MqttPacket packet, 
+            out MqttPacket packet,
-            out SequencePosition consumed, 
+            out SequencePosition consumed,
             out SequencePosition observed,
             out int bytesRead)
         {
@@ -33,22 +35,22 @@ namespace MQTTnet.AspNetCore
                 return false;
             }
 
-            var fixedHeader = copy.First.Span[0];
+            if (!TryReadBodyLength(ref copy, out var headerLength, out var bodyLength))
-            if (!TryReadBodyLength(ref copy, out int headerLength, out var bodyLength))
             {
                 return false;
             }
 
+            var fixedHeader = copy.First.Span[0];
+            copy = copy.Slice(headerLength);
             if (copy.Length < bodyLength)
             {
                 return false;
             }
 
             var bodySlice = copy.Slice(0, bodyLength);
-            var buffer = bodySlice.GetMemory().ToArray();
+            var bodySegment = GetArraySegment(ref bodySlice); 
-            
+    
-            var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, new ArraySegment<byte>(buffer, 0, buffer.Length), buffer.Length + 2);
+            var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, bodySegment, headerLength + bodyLength);
-
             if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
             {
                 formatter.DetectProtocolVersion(receivedMqttPacket);
@@ -61,18 +63,41 @@ namespace MQTTnet.AspNetCore
             return true;
         }
 
-        static ReadOnlyMemory<byte> GetMemory(this in ReadOnlySequence<byte> input)
+        static ArraySegment<byte> GetArraySegment(ref ReadOnlySequence<byte> input)
         {
-            if (input.IsSingleSegment)
+            if (input.IsSingleSegment && MemoryMarshal.TryGetArray(input.First, out var segment))
             {
-                return input.First;
+                return segment;
             }
 
             // Should be rare
-            return input.ToArray();
+            var array = input.ToArray();
+            return new ArraySegment<byte>(array);
         }
 
         static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int headerLength, out int bodyLength)
+        {
+            var valueSequence = input.Slice(0, Math.Min(5, input.Length));
+            if (valueSequence.IsSingleSegment)
+            {
+                var valueSpan =
+#if NET5_0_OR_GREATER
+                    valueSequence.FirstSpan;
+#else
+                    valueSequence.First.Span;
+#endif
+                return TryReadBodyLength(valueSpan, out headerLength, out bodyLength);
+            }
+            else
+            {
+                Span<byte> valueSpan = stackalloc byte[8];
+                valueSequence.CopyTo(valueSpan);
+                valueSpan = valueSpan.Slice(0, (int)valueSequence.Length);
+                return TryReadBodyLength(valueSpan, out headerLength, out bodyLength);
+            }
+        }
+
+        static bool TryReadBodyLength(ReadOnlySpan<byte> span, out int headerLength, out int bodyLength)
         {
             // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
             var multiplier = 1;
@@ -81,33 +106,35 @@ namespace MQTTnet.AspNetCore
             var index = 1;
             headerLength = 0;
             bodyLength = 0;
-            
-            var temp = input.Slice(0, Math.Min(5, input.Length)).GetMemory().Span;
 
             do
             {
-                if (index == temp.Length)
+                if (index == span.Length)
                 {
                     return false;
                 }
 
-                encodedByte = temp[index];
+                encodedByte = span[index];
                 index++;
 
                 value += (byte)(encodedByte & 127) * multiplier;
                 if (multiplier > 128 * 128 * 128)
                 {
-                    throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.Slice(1, index).ToArray())}).");
+                    ThrowProtocolViolationException(span, index);
                 }
 
                 multiplier *= 128;
             } while ((encodedByte & 128) != 0);
 
-            input = input.Slice(index);
-
             headerLength = index;
             bodyLength = value;
             return true;
         }
+
+
+        static void ThrowProtocolViolationException(ReadOnlySpan<byte> valueSpan, int index)
+        {
+            throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", valueSpan.Slice(1, index).ToArray())}).");
+        }
     }
 }

Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj

@@ -1,6 +1,7 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
+        <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
         <OutputType>Exe</OutputType>
         <DebugType>Full</DebugType>
         <TargetFrameworks>net6.0</TargetFrameworks>

Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs

@@ -0,0 +1,40 @@
+using BenchmarkDotNet.Attributes;
+using BenchmarkDotNet.Jobs;
+using System;
+
+
+namespace MQTTnet.Benchmarks
+{
+    [SimpleJob(RuntimeMoniker.Net60)]
+    [RPlotExporter, RankColumn]
+    [MemoryDiagnoser]
+    public class MemoryCopyBenchmark
+    {
+        const int max_length = 1024 * 8;
+        private byte[] source;
+        private byte[] target;
+
+        [Params(64 - 1, 128 - 1, 256 - 1, 512 - 1, 1024 - 1, 2048 - 1, 5096 - 1)]
+        public int Length { get; set; }
+
+        [GlobalSetup]
+        public void Setup()
+        {
+            source = new byte[max_length];
+            target = new byte[max_length];
+        }
+
+        [Benchmark(Baseline = true)]
+        public void Array_Copy()
+        {
+            Array.Copy(source, 0, target, 0, Length);
+        }
+
+        [Benchmark]
+        public void Memory_Copy()
+        {
+            MQTTnet.Internal.Memory.Copy(source, 0, target, 0, Length);
+        }
+
+    }
+}

Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs

@@ -0,0 +1,241 @@
+using BenchmarkDotNet.Attributes;
+using BenchmarkDotNet.Jobs;
+using MQTTnet.Adapter;
+using MQTTnet.AspNetCore;
+using MQTTnet.Exceptions;
+using MQTTnet.Formatter;
+using MQTTnet.Packets;
+using System;
+using System.Buffers;
+using System.IO;
+using System.IO.Pipelines;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Benchmarks
+{
+    [SimpleJob(RuntimeMoniker.Net60)]
+    [RPlotExporter, RankColumn]
+    [MemoryDiagnoser]
+    public class ReaderExtensionsBenchmark
+    {
+        private MqttPacketFormatterAdapter mqttPacketFormatter;
+        private MemoryStream stream;
+
+        [GlobalSetup]
+        public void GlobalSetup()
+        {
+            mqttPacketFormatter = new MqttPacketFormatterAdapter(MqttProtocolVersion.V311, new MqttBufferWriter(4096, 65535));
+            var mqttMessage = new MqttApplicationMessageBuilder()
+            .WithTopic("topic")
+            .WithPayload(new byte[10 * 1024])
+            .Build();
+
+            var packetFactories = new MqttPacketFactories();
+            var packet = packetFactories.Publish.Create(mqttMessage);
+
+            var buffer = mqttPacketFormatter.Encode(packet);
+            stream = new MemoryStream();
+            stream.Write(buffer.Packet);
+            stream.Write(buffer.Payload);
+            mqttPacketFormatter.Cleanup();
+        }
+
+        [Benchmark(Baseline = true)]
+        public async Task Before()
+        {
+            stream.Position = 0;
+            var input = PipeReader.Create(stream);
+
+            while (true)
+            {
+                ReadResult readResult;
+                var readTask = input.ReadAsync();
+                if (readTask.IsCompleted)
+                {
+                    readResult = readTask.Result;
+                }
+                else
+                {
+                    readResult = await readTask.ConfigureAwait(false);
+                }
+
+                var buffer = readResult.Buffer;
+
+                var consumed = buffer.Start;
+                var observed = buffer.Start;
+
+                try
+                {
+                    if (!buffer.IsEmpty)
+                    {
+                        if (ReaderExtensions_Before.TryDecode(mqttPacketFormatter, buffer, out var packet, out consumed, out observed, out var received))
+                        {
+                            break;
+                        }
+                    }
+                    else if (readResult.IsCompleted)
+                    {
+                        throw new MqttCommunicationException("Connection Aborted");
+                    }
+                }
+                finally
+                {
+                    // The buffer was sliced up to where it was consumed, so we can just advance to the start.
+                    // We mark examined as buffer.End so that if we didn't receive a full frame, we'll wait for more data
+                    // before yielding the read again.
+                    input.AdvanceTo(consumed, observed);
+                }
+            }
+        }
+
+        [Benchmark]
+        public async Task After()
+        {
+            stream.Position = 0;
+            var input = PipeReader.Create(stream);
+
+            while (true)
+            {
+                ReadResult readResult;
+                var readTask = input.ReadAsync();
+                if (readTask.IsCompleted)
+                {
+                    readResult = readTask.Result;
+                }
+                else
+                {
+                    readResult = await readTask.ConfigureAwait(false);
+                }
+
+                var buffer = readResult.Buffer;
+
+                var consumed = buffer.Start;
+                var observed = buffer.Start;
+
+                try
+                {
+                    if (!buffer.IsEmpty)
+                    {
+                        if (ReaderExtensions.TryDecode(mqttPacketFormatter, buffer, out var packet, out consumed, out observed, out var received))
+                        {
+                            break;
+                        }
+                    }
+                    else if (readResult.IsCompleted)
+                    {
+                        throw new MqttCommunicationException("Connection Aborted");
+                    }
+                }
+                finally
+                {
+                    // The buffer was sliced up to where it was consumed, so we can just advance to the start.
+                    // We mark examined as buffer.End so that if we didn't receive a full frame, we'll wait for more data
+                    // before yielding the read again.
+                    input.AdvanceTo(consumed, observed);
+                }
+            }
+        }
+
+        public static class ReaderExtensions_Before
+        {
+            public static bool TryDecode(MqttPacketFormatterAdapter formatter,
+                in ReadOnlySequence<byte> input,
+                out MqttPacket packet,
+                out SequencePosition consumed,
+                out SequencePosition observed,
+                out int bytesRead)
+            {
+                if (formatter == null) throw new ArgumentNullException(nameof(formatter));
+
+                packet = null;
+                consumed = input.Start;
+                observed = input.End;
+                bytesRead = 0;
+                var copy = input;
+
+                if (copy.Length < 2)
+                {
+                    return false;
+                }
+
+                var fixedHeader = copy.First.Span[0];
+                if (!TryReadBodyLength(ref copy, out int headerLength, out var bodyLength))
+                {
+                    return false;
+                }
+
+                if (copy.Length < bodyLength)
+                {
+                    return false;
+                }
+
+                var bodySlice = copy.Slice(0, bodyLength);
+                var buffer = GetMemory(bodySlice).ToArray();
+
+                var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, new ArraySegment<byte>(buffer, 0, buffer.Length), buffer.Length + 2);
+
+                if (formatter.ProtocolVersion == MqttProtocolVersion.Unknown)
+                {
+                    formatter.DetectProtocolVersion(receivedMqttPacket);
+                }
+
+                packet = formatter.Decode(receivedMqttPacket);
+                consumed = bodySlice.End;
+                observed = bodySlice.End;
+                bytesRead = headerLength + bodyLength;
+                return true;
+            }
+
+            static ReadOnlyMemory<byte> GetMemory(in ReadOnlySequence<byte> input)
+            {
+                if (input.IsSingleSegment)
+                {
+                    return input.First;
+                }
+
+                // Should be rare
+                return input.ToArray();
+            }
+
+            static bool TryReadBodyLength(ref ReadOnlySequence<byte> input, out int headerLength, out int bodyLength)
+            {
+                // Alorithm taken from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html.
+                var multiplier = 1;
+                var value = 0;
+                byte encodedByte;
+                var index = 1;
+                headerLength = 0;
+                bodyLength = 0;
+
+                var temp = GetMemory(input.Slice(0, Math.Min(5, input.Length))).Span;
+
+                do
+                {
+                    if (index == temp.Length)
+                    {
+                        return false;
+                    }
+
+                    encodedByte = temp[index];
+                    index++;
+
+                    value += (byte)(encodedByte & 127) * multiplier;
+                    if (multiplier > 128 * 128 * 128)
+                    {
+                        throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", temp.Slice(1, index).ToArray())}).");
+                    }
+
+                    multiplier *= 128;
+                } while ((encodedByte & 128) != 0);
+
+                input = input.Slice(index);
+
+                headerLength = index;
+                bodyLength = value;
+                return true;
+            }
+        }
+
+    }
+}

Source/MQTTnet/Formatter/MqttBufferReader.cs

@@ -2,25 +2,27 @@
 // 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.Runtime.CompilerServices;
-using System.Text;
 using MQTTnet.Exceptions;
 using MQTTnet.Internal;
+using System;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+using System.Text;
 
 namespace MQTTnet.Formatter
 {
     public sealed class MqttBufferReader
     {
         byte[] _buffer = EmptyBuffer.Array;
-        int _maxPosition;
         int _offset;
+        int _position;
+        int _maxPosition;
 
-        public int BytesLeft => _maxPosition - Position;
+        public int BytesLeft => _maxPosition - _position;
 
         public bool EndOfStream => BytesLeft == 0;
 
-        public int Position { get; private set; }
+        public int Position => _position - _offset;
 
         public byte[] ReadBinaryData()
         {
@@ -34,8 +36,8 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(length);
 
             var result = new byte[length];
-            Array.Copy(_buffer, Position + _offset, result, 0, length);
+            Memory.Copy(_buffer, _position, result, 0, length);
-            Position += length;
+            _position += length;
 
             return result;
         }
@@ -43,8 +45,7 @@ namespace MQTTnet.Formatter
         public byte ReadByte()
         {
             ValidateReceiveBuffer(1);
-
+            return _buffer[_position++];
-            return _buffer[Position++];
         }
 
         public uint ReadFourByteInteger()
@@ -52,34 +53,35 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(4);
 
 #if NETCOREAPP3_0_OR_GREATER
-            var value = System.Runtime.InteropServices.MemoryMarshal.Read<uint>(_buffer.AsSpan().Slice(Position));
+            var value = MemoryMarshal.Read<uint>(_buffer.AsSpan(_position));
-            value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
+            if (BitConverter.IsLittleEndian)
+            {
+                value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
+            }
 #else
-            var byte0 = _buffer[_offset + Position];
+            var byte0 = _buffer[_position];
-            var byte1 = _buffer[_offset + Position + 1];
+            var byte1 = _buffer[_position + 1];
-            var byte2 = _buffer[_offset + Position + 2];
+            var byte2 = _buffer[_position + 2];
-            var byte3 = _buffer[_offset + Position + 3];
+            var byte3 = _buffer[_position + 3];
 
             var value = (uint)((byte0 << 24) | (byte1 << 16) | (byte2 << 8) | byte3);
 #endif
 
-            Position += 4;
+            _position += 4;
             return value;
         }
 
         public byte[] ReadRemainingData()
         {
-            var bufferLength = _maxPosition - Position - _offset;
+            var bufferLength = BytesLeft;
-
             if (bufferLength == 0)
             {
                 return EmptyBuffer.Array;
             }
 
             var buffer = new byte[bufferLength];
-            Array.Copy(_buffer, Position + _offset, buffer, 0, bufferLength);
+            Memory.Copy(_buffer, _position, buffer, 0, bufferLength);
-
+            _position += bufferLength;
-            Position += bufferLength;
 
             return buffer;
         }
@@ -97,12 +99,12 @@ namespace MQTTnet.Formatter
 
 #if NETCOREAPP3_0_OR_GREATER
             // AsSpan() version is slightly faster. Not much but at least a little bit.
-            var result = Encoding.UTF8.GetString(_buffer.AsSpan(_offset + Position, length));
+            var result = Encoding.UTF8.GetString(_buffer.AsSpan(_position, length));
 #else
-            var result = Encoding.UTF8.GetString(_buffer, _offset + Position, length);
+            var result = Encoding.UTF8.GetString(_buffer, _position, length);
 #endif            
 
-            Position += length;
+            _position += length;
             return result;
         }
 
@@ -111,16 +113,19 @@ namespace MQTTnet.Formatter
             ValidateReceiveBuffer(2);
 
 #if NETCOREAPP3_0_OR_GREATER
-            var value = System.Runtime.InteropServices.MemoryMarshal.Read<ushort>(_buffer.AsSpan().Slice(Position));
+            var value = MemoryMarshal.Read<ushort>(_buffer.AsSpan(_position));
-            value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
+            if (BitConverter.IsLittleEndian)
+            {
+                value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value);
+            }
 #else
-            var msb = _buffer[_offset + Position];
+            var msb = _buffer[_position];
-            var lsb = _buffer[_offset + Position + 1];
+            var lsb = _buffer[_position + 1];
 
             var value = (ushort)((msb << 8) | lsb);
 #endif
-            
+
-            Position += 2;
+            _position += 2;
             return value;
         }
 
@@ -148,26 +153,29 @@ namespace MQTTnet.Formatter
 
         public void Seek(int position)
         {
-            Position = position;
+            _position = _offset + position;
+        }
+
+        public void SetBuffer(ArraySegment<byte> buffer)
+        {
+            SetBuffer(buffer.Array, buffer.Offset, buffer.Count);
         }
 
         public void SetBuffer(byte[] buffer, int offset, int length)
         {
             _buffer = buffer ?? throw new ArgumentNullException(nameof(buffer));
             _offset = offset;
-
+            _position = offset;
-            Position = 0;
+            _maxPosition = offset + length;
-            _maxPosition = length;
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         void ValidateReceiveBuffer(int length)
         {
-            var newPosition = Position + _offset + length;
+            var newPosition = _position + length;
-
             if (_maxPosition < newPosition)
             {
-                throw new MqttProtocolViolationException($"Expected at least {Position + length} bytes but there are only {_maxPosition} bytes");
+                throw new MqttProtocolViolationException($"Expected at least {newPosition} bytes but there are only {_maxPosition} bytes");
             }
         }
     }

Source/MQTTnet/Formatter/MqttBufferWriter.cs

@@ -6,6 +6,7 @@ using System;
 using System.Runtime.CompilerServices;
 using System.Text;
 using MQTTnet.Exceptions;
+using MQTTnet.Internal;
 using MQTTnet.Protocol;
 
 namespace MQTTnet.Formatter
@@ -125,7 +126,7 @@ namespace MQTTnet.Formatter
 
             EnsureAdditionalCapacity(count);
 
-            Array.Copy(buffer, offset, _buffer, _position, count);
+            Memory.Copy(buffer, offset, _buffer, _position, count);
             IncreasePosition(count);
         }
 
@@ -149,7 +150,7 @@ namespace MQTTnet.Formatter
                 _buffer[_position] = (byte)(valueLength >> 8);
                 _buffer[_position + 1] = (byte)valueLength;
 
-                Array.Copy(value, 0, _buffer, _position + 2, valueLength);
+                Memory.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];
-            Array.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
+            Memory.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
-            Array.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
+            Memory.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
 
             return buffer;
         }
@@ -57,8 +57,8 @@ namespace MQTTnet.Formatter
             }
 
             var buffer = new byte[Length];
-            Array.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
+            Memory.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count);
-            Array.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
+            Memory.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count);
 
             return new ArraySegment<byte>(buffer);
         }

Source/MQTTnet/Internal/Memory.cs

@@ -0,0 +1,29 @@
+using System;
+using System.Runtime.CompilerServices;
+
+namespace MQTTnet.Internal
+{
+    public static class Memory
+    {
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public static void Copy(byte[] source, int sourceIndex, byte[] destination, int destinationIndex, int length)
+        {
+#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1
+            source.AsSpan(sourceIndex, length).CopyTo(destination.AsSpan(destinationIndex, length));
+#elif NET461_OR_GREATER || NETSTANDARD1_3_OR_GREATER
+            unsafe
+            {
+                fixed (byte* pSoure = &source[sourceIndex])
+                {
+                    fixed (byte* pDestination = &destination[destinationIndex])
+                    {
+                        System.Buffer.MemoryCopy(pSoure, pDestination, length, length);
+                    }
+                }
+            }
+#else
+            Array.Copy(source, sourceIndex, destination, destinationIndex, length);
+#endif
+        }
+    }
+}

Source/MQTTnet/MQTTnet.csproj

@@ -42,6 +42,7 @@
         <PackageLicenseExpression>MIT</PackageLicenseExpression>
         <DisableImplicitAspNetCoreAnalyzers>true</DisableImplicitAspNetCoreAnalyzers>
         <NoWarn>1591</NoWarn>
+        <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
     </PropertyGroup>
 
     <PropertyGroup Condition="'$(TargetFramework)'=='uap10.0'">