From b77798f8853e9fdf2f0c5e7591224a8100b9be22 Mon Sep 17 00:00:00 2001 From: Jan Eggers Date: Mon, 25 Jun 2018 15:31:49 +0200 Subject: [PATCH] dont use thread static buffers they are overlapping on server due to async io --- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 6 +++- Source/MQTTnet/Serializer/MqttPacketReader.cs | 35 +++++-------------- .../MQTTnet.Benchmarks/SerializerBenchmark.cs | 4 ++- .../MqttPacketReaderTests.cs | 4 ++- .../MqttPacketSerializerTests.cs | 9 +++-- 5 files changed, 27 insertions(+), 31 deletions(-) diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 61912ef0..0059f201 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -23,6 +23,10 @@ namespace MQTTnet.Adapter private readonly IMqttNetChildLogger _logger; private readonly IMqttChannel _channel; + private readonly byte[] _fixedHeaderBuffer = new byte[2]; + + private readonly byte[] _singleByteBuffer = new byte[1]; + private bool _isDisposed; public MqttChannelAdapter(IMqttChannel channel, IMqttPacketSerializer serializer, IMqttNetChildLogger logger) @@ -163,7 +167,7 @@ namespace MQTTnet.Adapter private async Task ReceiveAsync(IMqttChannel channel, CancellationToken cancellationToken) { - var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, cancellationToken).ConfigureAwait(false); + var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, _fixedHeaderBuffer, _singleByteBuffer, cancellationToken).ConfigureAwait(false); try { diff --git a/Source/MQTTnet/Serializer/MqttPacketReader.cs b/Source/MQTTnet/Serializer/MqttPacketReader.cs index 826747cf..32a92855 100644 --- a/Source/MQTTnet/Serializer/MqttPacketReader.cs +++ b/Source/MQTTnet/Serializer/MqttPacketReader.cs @@ -9,19 +9,13 @@ 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) + public static async Task ReadFixedHeaderAsync(IMqttChannel channel, byte[] fixedHeaderBuffer, byte[] singleByteBuffer, 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 = InitializeFixedHeaderBuffer(); + var buffer = fixedHeaderBuffer; var totalBytesRead = 0; while (totalBytesRead < buffer.Length) @@ -41,12 +35,12 @@ namespace MQTTnet.Serializer { return new MqttFixedHeader(buffer[0], 0); } - - var bodyLength = ReadBodyLength(channel, buffer[1], cancellationToken); + + var bodyLength = ReadBodyLength(channel, buffer[1], singleByteBuffer, cancellationToken); return new MqttFixedHeader(buffer[0], bodyLength); } - private static int ReadBodyLength(IMqttChannel channel, byte initialEncodedByte, CancellationToken cancellationToken) + private static int ReadBodyLength(IMqttChannel channel, byte initialEncodedByte, byte[] singleByteBuffer, CancellationToken cancellationToken) { // 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 = 128; @@ -61,7 +55,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 = ReadByte(channel, cancellationToken); + encodedByte = ReadByte(channel, singleByteBuffer, cancellationToken); value += (byte)(encodedByte & 127) * multiplier; if (multiplier > 128 * 128 * 128) @@ -75,27 +69,16 @@ namespace MQTTnet.Serializer return value; } - private static byte ReadByte(IMqttChannel channel, CancellationToken cancellationToken) + private static byte ReadByte(IMqttChannel channel, byte[] singleByteBuffer, CancellationToken cancellationToken) { - var buffer = InitializeSingleByteBuffer(); - var readCount = channel.ReadAsync(buffer, 0, 1, cancellationToken).GetAwaiter().GetResult(); + var readCount = channel.ReadAsync(singleByteBuffer, 0, 1, cancellationToken).GetAwaiter().GetResult(); if (readCount <= 0) { cancellationToken.ThrowIfCancellationRequested(); ExceptionHelper.ThrowGracefulSocketClose(); } - return buffer[0]; - } - - private static byte[] InitializeFixedHeaderBuffer() - { - return _fixedHeaderBuffer ?? (_fixedHeaderBuffer = new byte[2]); - } - - private static byte[] InitializeSingleByteBuffer() - { - return _singleByteBuffer ?? (_singleByteBuffer = new byte[1]); + return singleByteBuffer[0]; } } } diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 28de60ff..8beb7b49 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -47,12 +47,14 @@ namespace MQTTnet.Benchmarks public void Deserialize_10000_Messages() { var channel = new BenchmarkMqttChannel(_serializedPacket); + var fixedHeader = new byte[2]; + var singleByteBuffer = new byte[1]; for (var i = 0; i < 10000; i++) { channel.Reset(); - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult(); + var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); var receivedPacket = new ReceivedMqttPacket( header.Flags, diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs index f983622b..39aeeff2 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketReaderTests.cs @@ -14,7 +14,9 @@ namespace MQTTnet.Core.Tests [ExpectedException(typeof(MqttCommunicationClosedGracefullyException))] public void MqttPacketReader_EmptyStream() { - MqttPacketReader.ReadFixedHeaderAsync(new TestMqttChannel(new MemoryStream()), CancellationToken.None).GetAwaiter().GetResult(); + var fixedHeader = new byte[2]; + var singleByteBuffer = new byte[1]; + MqttPacketReader.ReadFixedHeaderAsync(new TestMqttChannel(new MemoryStream()), fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); } } } diff --git a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs index 61b69eb5..d8d23e29 100644 --- a/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs +++ b/Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs @@ -476,7 +476,9 @@ namespace MQTTnet.Core.Tests using (var headerStream = new MemoryStream(Join(buffer1))) { var channel = new TestMqttChannel(headerStream); - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult(); + var fixedHeader = new byte[2]; + var singleByteBuffer = new byte[1]; + var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) { @@ -498,7 +500,10 @@ namespace MQTTnet.Core.Tests using (var headerStream = new MemoryStream(Join(buffer1))) { var channel = new TestMqttChannel(headerStream); - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult(); + var fixedHeader = new byte[2]; + var singleByteBuffer = new byte[1]; + + var header = MqttPacketReader.ReadFixedHeaderAsync(channel, fixedHeader, singleByteBuffer, CancellationToken.None).GetAwaiter().GetResult(); using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.RemainingLength)) {