diff --git a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs b/Source/MQTTnet.AspnetCore/ReaderExtensions.cs index d3b00ab6..c7d289fd 100644 --- a/Source/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/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, - in ReadOnlySequence input, - out MqttPacket packet, - out SequencePosition consumed, + public static bool TryDecode(this MqttPacketFormatterAdapter formatter, + in ReadOnlySequence input, + out MqttPacket packet, + 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 int headerLength, out var bodyLength)) + if (!TryReadBodyLength(ref copy, out var 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 receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, new ArraySegment(buffer, 0, buffer.Length), buffer.Length + 2); - + var bodySegment = GetArraySegment(ref bodySlice); + + 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 GetMemory(this in ReadOnlySequence input) + static ArraySegment GetArraySegment(ref ReadOnlySequence 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(array); } static bool TryReadBodyLength(ref ReadOnlySequence 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 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 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 valueSpan, int index) + { + throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", valueSpan.Slice(1, index).ToArray())})."); + } } } diff --git a/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index 9cac34d0..899b655f 100644 --- a/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Source/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -1,6 +1,7 @@ - + + true Exe Full net6.0 diff --git a/Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs b/Source/MQTTnet.Benchmarks/MemoryCopyBenchmark.cs new file mode 100644 index 00000000..39fa4f38 --- /dev/null +++ b/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); + } + + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs b/Source/MQTTnet.Benchmarks/ReaderExtensionsBenchmark.cs new file mode 100644 index 00000000..1d50b904 --- /dev/null +++ b/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 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(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 GetMemory(in ReadOnlySequence input) + { + if (input.IsSingleSegment) + { + return input.First; + } + + // Should be rare + return input.ToArray(); + } + + static bool TryReadBodyLength(ref ReadOnlySequence 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; + } + } + + } +} diff --git a/Source/MQTTnet/Formatter/MqttBufferReader.cs b/Source/MQTTnet/Formatter/MqttBufferReader.cs index 51c8225f..4b3ef3c4 100644 --- a/Source/MQTTnet/Formatter/MqttBufferReader.cs +++ b/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); - Position += length; + Memory.Copy(_buffer, _position, result, 0, 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(_buffer.AsSpan().Slice(Position)); - value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value); + var value = MemoryMarshal.Read(_buffer.AsSpan(_position)); + if (BitConverter.IsLittleEndian) + { + value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value); + } #else - var byte0 = _buffer[_offset + Position]; - var byte1 = _buffer[_offset + Position + 1]; - var byte2 = _buffer[_offset + Position + 2]; - var byte3 = _buffer[_offset + Position + 3]; + var byte0 = _buffer[_position]; + var byte1 = _buffer[_position + 1]; + var byte2 = _buffer[_position + 2]; + 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); - - Position += bufferLength; + Memory.Copy(_buffer, _position, buffer, 0, 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(_buffer.AsSpan().Slice(Position)); - value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value); + var value = MemoryMarshal.Read(_buffer.AsSpan(_position)); + if (BitConverter.IsLittleEndian) + { + value = System.Buffers.Binary.BinaryPrimitives.ReverseEndianness(value); + } #else - var msb = _buffer[_offset + Position]; - var lsb = _buffer[_offset + Position + 1]; + var msb = _buffer[_position]; + 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 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 = 0; - _maxPosition = length; + _position = offset; + _maxPosition = offset + 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"); } } } diff --git a/Source/MQTTnet/Formatter/MqttBufferWriter.cs b/Source/MQTTnet/Formatter/MqttBufferWriter.cs index 7e740630..8e572d22 100644 --- a/Source/MQTTnet/Formatter/MqttBufferWriter.cs +++ b/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); } } diff --git a/Source/MQTTnet/Formatter/MqttPacketBuffer.cs b/Source/MQTTnet/Formatter/MqttPacketBuffer.cs index 8bdd51c8..7a3afea1 100644 --- a/Source/MQTTnet/Formatter/MqttPacketBuffer.cs +++ b/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); - Array.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count); + Memory.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.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); - Array.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count); + Memory.Copy(Packet.Array, Packet.Offset, buffer, 0, Packet.Count); + Memory.Copy(Payload.Array, Payload.Offset, buffer, Packet.Count, Payload.Count); return new ArraySegment(buffer); } diff --git a/Source/MQTTnet/Internal/Memory.cs b/Source/MQTTnet/Internal/Memory.cs new file mode 100644 index 00000000..0d53f6aa --- /dev/null +++ b/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 + } + } +} diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index c8bf7f0a..6153cce0 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -42,6 +42,7 @@ MIT true 1591 + true