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/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; + } + } + + } +}