diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index cfbf743d..94e0f332 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -157,10 +157,6 @@ namespace MQTTnet.Adapter private async Task ReceiveAsync(IMqttChannel channel, CancellationToken cancellationToken) { var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, cancellationToken).ConfigureAwait(false); - if (fixedHeader == null) - { - return null; - } try { diff --git a/Source/MQTTnet/Serializer/Extensions.cs b/Source/MQTTnet/Serializer/Extensions.cs index 1a14de4f..00ad7cfb 100644 --- a/Source/MQTTnet/Serializer/Extensions.cs +++ b/Source/MQTTnet/Serializer/Extensions.cs @@ -12,7 +12,10 @@ namespace MQTTnet.Serializer } var buffer = new byte[source.Count]; - Buffer.BlockCopy(source.Array, source.Offset, buffer, 0, buffer.Length); + if (buffer.Length > 0) + { + Array.Copy(source.Array, source.Offset, buffer, 0, buffer.Length); + } return buffer; } diff --git a/Source/MQTTnet/Serializer/MqttFixedHeader.cs b/Source/MQTTnet/Serializer/MqttFixedHeader.cs index d87f63d4..a8c2015b 100644 --- a/Source/MQTTnet/Serializer/MqttFixedHeader.cs +++ b/Source/MQTTnet/Serializer/MqttFixedHeader.cs @@ -1,6 +1,6 @@ namespace MQTTnet.Serializer { - public class MqttFixedHeader + public struct MqttFixedHeader { public MqttFixedHeader(byte flags, int remainingLength) { @@ -10,6 +10,6 @@ public byte Flags { get; } - public int RemainingLength { get; set; } + public int RemainingLength { get; } } } diff --git a/Source/MQTTnet/Serializer/MqttPacketReader.cs b/Source/MQTTnet/Serializer/MqttPacketReader.cs index d50fae5d..5bbed528 100644 --- a/Source/MQTTnet/Serializer/MqttPacketReader.cs +++ b/Source/MQTTnet/Serializer/MqttPacketReader.cs @@ -1,4 +1,5 @@ -using System.Threading; +using System; +using System.Threading; using System.Threading.Tasks; using MQTTnet.Channel; using MQTTnet.Exceptions; @@ -8,23 +9,29 @@ namespace MQTTnet.Serializer { public static class MqttPacketReader { + [ThreadStatic] + private static byte[] _fixedHeaderBuffer; + + [ThreadStatic] + private static byte[] _singleByteBuffer; + public static async Task ReadFixedHeaderAsync(IMqttChannel channel, CancellationToken cancellationToken) { // The MQTT fixed header contains 1 byte of flags and at least 1 byte for the remaining data length. // So in all cases at least 2 bytes must be read for a complete MQTT packet. // async/await is used here because the next packet is received in a couple of minutes so the performance // impact is acceptable according to a useless waiting thread. - var buffer = new byte[2]; + var buffer = InitializeFixedHeaderBuffer(); var totalBytesRead = 0; while (totalBytesRead < buffer.Length) { - var bytesRead = await channel.ReadAsync(buffer, 0, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); + var bytesRead = await channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); if (bytesRead <= 0) { if (cancellationToken.IsCancellationRequested) { - return null; + throw new TaskCanceledException(); } ExceptionHelper.ThrowGracefulSocketClose(); @@ -56,7 +63,7 @@ namespace MQTTnet.Serializer // is too big for reading 1 byte in a row. We expect that the remaining data was sent // directly after the initial bytes. If the client disconnects just in this moment we // will get an exception anyway. - encodedByte = ReadByteAsync(channel, cancellationToken).GetAwaiter().GetResult(); + encodedByte = ReadByte(channel, cancellationToken); value += (byte)(encodedByte & 127) * multiplier; if (multiplier > 128 * 128 * 128) @@ -70,10 +77,10 @@ namespace MQTTnet.Serializer return value; } - private static async Task ReadByteAsync(IMqttChannel channel, CancellationToken cancellationToken) + private static byte ReadByte(IMqttChannel channel, CancellationToken cancellationToken) { - var buffer = new byte[1]; - var readCount = await channel.ReadAsync(buffer, 0, 1, cancellationToken).ConfigureAwait(false); + var buffer = InitializeSingleByteBuffer(); + var readCount = channel.ReadAsync(buffer, 0, 1, cancellationToken).GetAwaiter().GetResult(); if (readCount <= 0) { ExceptionHelper.ThrowGracefulSocketClose(); @@ -81,5 +88,15 @@ namespace MQTTnet.Serializer return buffer[0]; } + + private static byte[] InitializeFixedHeaderBuffer() + { + return _fixedHeaderBuffer ?? (_fixedHeaderBuffer = new byte[2]); + } + + private static byte[] InitializeSingleByteBuffer() + { + return _singleByteBuffer ?? (_singleByteBuffer = new byte[1]); + } } } diff --git a/Source/MQTTnet/Serializer/MqttPacketSerializer.cs b/Source/MQTTnet/Serializer/MqttPacketSerializer.cs index aafaa90f..159ba895 100644 --- a/Source/MQTTnet/Serializer/MqttPacketSerializer.cs +++ b/Source/MQTTnet/Serializer/MqttPacketSerializer.cs @@ -199,19 +199,18 @@ namespace MQTTnet.Serializer private static MqttBasePacket DeserializePublish(ReceivedMqttPacket receivedMqttPacket) { - var body = receivedMqttPacket.Body; - ThrowIfBodyIsEmpty(body); + ThrowIfBodyIsEmpty(receivedMqttPacket.Body); var retain = (receivedMqttPacket.FixedHeader & 0x1) > 0; var qualityOfServiceLevel = (MqttQualityOfServiceLevel)(receivedMqttPacket.FixedHeader >> 1 & 0x3); var dup = (receivedMqttPacket.FixedHeader & 0x3) > 0; - var topic = body.ReadStringWithLengthPrefix(); + var topic = receivedMqttPacket.Body.ReadStringWithLengthPrefix(); ushort? packetIdentifier = null; if (qualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - packetIdentifier = body.ReadUInt16(); + packetIdentifier = receivedMqttPacket.Body.ReadUInt16(); } var packet = new MqttPublishPacket @@ -219,7 +218,7 @@ namespace MQTTnet.Serializer PacketIdentifier = packetIdentifier, Retain = retain, Topic = topic, - Payload = body.ReadRemainingData().ToArray(), + Payload = receivedMqttPacket.Body.ReadRemainingData().ToArray(), QualityOfServiceLevel = qualityOfServiceLevel, Dup = dup }; diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs index 53b71fd6..f6ff61a3 100644 --- a/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingBenchmark.cs @@ -9,6 +9,7 @@ namespace MQTTnet.Benchmarks { [ClrJob] [RPlotExporter, RankColumn] + [MemoryDiagnoser] public class MessageProcessingBenchmark { private IMqttServer _mqttServer; diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 2e9f22c3..28de60ff 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -46,12 +46,19 @@ namespace MQTTnet.Benchmarks [Benchmark] public void Deserialize_10000_Messages() { + var channel = new BenchmarkMqttChannel(_serializedPacket); + for (var i = 0; i < 10000; i++) { - var channel = new BenchmarkMqttChannel(_serializedPacket); + channel.Reset(); var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult(); - _serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength))); + + var receivedPacket = new ReceivedMqttPacket( + header.Flags, + new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength)); + + _serializer.Deserialize(receivedPacket); } } @@ -68,6 +75,11 @@ namespace MQTTnet.Benchmarks public string Endpoint { get; } + public void Reset() + { + _position = _buffer.Offset; + } + public Task ConnectAsync(CancellationToken cancellationToken) { throw new NotImplementedException();