From 56e9e15ac006876cac3ff4d97607a0f4e0349b92 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Sun, 17 Jun 2018 13:47:58 +0200 Subject: [PATCH] Refactor Serializer memory usage. --- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 2 + .../Serializer/IMqttPacketSerializer.cs | 2 + .../Serializer/MqttPacketSerializer.cs | 73 ++++++++++++------- Source/MQTTnet/Serializer/MqttPacketWriter.cs | 21 ++++++ .../MQTTnet.Benchmarks/SerializerBenchmark.cs | 55 +++++++++++--- 5 files changed, 117 insertions(+), 36 deletions(-) diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index b21f4b30..cfbf743d 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -95,6 +95,8 @@ namespace MQTTnet.Adapter var packetData = PacketSerializer.Serialize(packet); await _channel.WriteAsync(packetData.Array, packetData.Offset, packetData.Count, cancellationToken).ConfigureAwait(false); + + PacketSerializer.FreeBuffer(); } catch (Exception exception) { diff --git a/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs b/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs index a81071bd..e3066f9f 100644 --- a/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs +++ b/Source/MQTTnet/Serializer/IMqttPacketSerializer.cs @@ -11,5 +11,7 @@ namespace MQTTnet.Serializer ArraySegment Serialize(MqttBasePacket mqttPacket); MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket); + + void FreeBuffer(); } } \ No newline at end of file diff --git a/Source/MQTTnet/Serializer/MqttPacketSerializer.cs b/Source/MQTTnet/Serializer/MqttPacketSerializer.cs index 811acf70..aafaa90f 100644 --- a/Source/MQTTnet/Serializer/MqttPacketSerializer.cs +++ b/Source/MQTTnet/Serializer/MqttPacketSerializer.cs @@ -11,15 +11,19 @@ namespace MQTTnet.Serializer { private const int FixedHeaderSize = 1; + [ThreadStatic] + private static MqttPacketWriter _packetWriter; + public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311; public ArraySegment Serialize(MqttBasePacket packet) { if (packet == null) throw new ArgumentNullException(nameof(packet)); - var packetWriter = new MqttPacketWriter(); + var packetWriter = InitializePacketWriter(); // Leave enough head space for max header size (fixed + 4 variable remaining length = 5 bytes) + packetWriter.Reset(); packetWriter.Seek(5); var fixedHeader = SerializePacket(packet, packetWriter); @@ -39,28 +43,6 @@ namespace MQTTnet.Serializer return new ArraySegment(buffer, headerOffset, packetWriter.Length - headerOffset); } - private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter packetWriter) - { - switch (packet) - { - case MqttConnectPacket connectPacket: return Serialize(connectPacket, packetWriter); - case MqttConnAckPacket connAckPacket: return Serialize(connAckPacket, packetWriter); - case MqttDisconnectPacket _: return SerializeEmptyPacket(MqttControlPacketType.Disconnect); - case MqttPingReqPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingReq); - case MqttPingRespPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingResp); - case MqttPublishPacket publishPacket: return Serialize(publishPacket, packetWriter); - case MqttPubAckPacket pubAckPacket: return Serialize(pubAckPacket, packetWriter); - case MqttPubRecPacket pubRecPacket: return Serialize(pubRecPacket, packetWriter); - case MqttPubRelPacket pubRelPacket: return Serialize(pubRelPacket, packetWriter); - case MqttPubCompPacket pubCompPacket: return Serialize(pubCompPacket, packetWriter); - case MqttSubscribePacket subscribePacket: return Serialize(subscribePacket, packetWriter); - case MqttSubAckPacket subAckPacket: return Serialize(subAckPacket, packetWriter); - case MqttUnsubscribePacket unsubscribePacket: return Serialize(unsubscribePacket, packetWriter); - case MqttUnsubAckPacket unsubAckPacket: return Serialize(unsubAckPacket, packetWriter); - default: throw new MqttProtocolViolationException("Packet type invalid."); - } - } - public MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket) { if (receivedMqttPacket == null) throw new ArgumentNullException(nameof(receivedMqttPacket)); @@ -92,6 +74,43 @@ namespace MQTTnet.Serializer } } + public void FreeBuffer() + { + InitializePacketWriter().FreeBuffer(); + } + + private byte SerializePacket(MqttBasePacket packet, MqttPacketWriter packetWriter) + { + switch (packet) + { + case MqttConnectPacket connectPacket: return Serialize(connectPacket, packetWriter); + case MqttConnAckPacket connAckPacket: return Serialize(connAckPacket, packetWriter); + case MqttDisconnectPacket _: return SerializeEmptyPacket(MqttControlPacketType.Disconnect); + case MqttPingReqPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingReq); + case MqttPingRespPacket _: return SerializeEmptyPacket(MqttControlPacketType.PingResp); + case MqttPublishPacket publishPacket: return Serialize(publishPacket, packetWriter); + case MqttPubAckPacket pubAckPacket: return Serialize(pubAckPacket, packetWriter); + case MqttPubRecPacket pubRecPacket: return Serialize(pubRecPacket, packetWriter); + case MqttPubRelPacket pubRelPacket: return Serialize(pubRelPacket, packetWriter); + case MqttPubCompPacket pubCompPacket: return Serialize(pubCompPacket, packetWriter); + case MqttSubscribePacket subscribePacket: return Serialize(subscribePacket, packetWriter); + case MqttSubAckPacket subAckPacket: return Serialize(subAckPacket, packetWriter); + case MqttUnsubscribePacket unsubscribePacket: return Serialize(unsubscribePacket, packetWriter); + case MqttUnsubAckPacket unsubAckPacket: return Serialize(unsubAckPacket, packetWriter); + default: throw new MqttProtocolViolationException("Packet type invalid."); + } + } + + private static MqttPacketWriter InitializePacketWriter() + { + if (_packetWriter == null) + { + _packetWriter = new MqttPacketWriter(); + } + + return _packetWriter; + } + private static MqttBasePacket DeserializeUnsubAck(MqttPacketBodyReader body) { ThrowIfBodyIsEmpty(body); @@ -310,7 +329,7 @@ namespace MQTTnet.Serializer var packet = new MqttConnAckPacket(); var acknowledgeFlags = body.ReadByte(); - + if (ProtocolVersion == MqttProtocolVersion.V311) { packet.IsSessionPresent = (acknowledgeFlags & 0x1) > 0; @@ -372,7 +391,7 @@ namespace MQTTnet.Serializer connectFlags |= 0x20; } } - + if (packet.Password != null && packet.Username == null) { throw new MqttProtocolViolationException("If the User Name Flag is set to 0, the Password Flag MUST be set to 0 [MQTT-3.1.2-22]."); @@ -387,7 +406,7 @@ namespace MQTTnet.Serializer { connectFlags |= 0x80; } - + packetWriter.Write(connectFlags); packetWriter.Write(packet.KeepAlivePeriod); packetWriter.WriteWithLengthPrefix(packet.ClientId); @@ -424,7 +443,7 @@ namespace MQTTnet.Serializer { connectAcknowledgeFlags |= 0x1; } - + packetWriter.Write(connectAcknowledgeFlags); } else diff --git a/Source/MQTTnet/Serializer/MqttPacketWriter.cs b/Source/MQTTnet/Serializer/MqttPacketWriter.cs index c0c49fcb..a2c6f5bf 100644 --- a/Source/MQTTnet/Serializer/MqttPacketWriter.cs +++ b/Source/MQTTnet/Serializer/MqttPacketWriter.cs @@ -93,6 +93,11 @@ namespace MQTTnet.Serializer IncreasePostition(count); } + public void Reset() + { + Length = 0; + } + public void Seek(int offset) { EnsureCapacity(offset); @@ -104,6 +109,22 @@ namespace MQTTnet.Serializer return _buffer; } + public void FreeBuffer() + { + // This method frees the used memory by shrinking the buffer. This is required because the buffer + // is used across several messages. In general this is not a big issue because subsequent Ping packages + // have the same size but a very big publish package with 100 MB of payload will increase the buffer + // a lot and the size will never reduced. So this method tries to find a size which can be held in + // memory for a long time without causing troubles. + + if (_buffer.Length < 4096) + { + return; + } + + Array.Resize(ref _buffer, 4096); + } + private void EnsureAdditionalCapacity(int additionalCapacity) { var freeSpace = _buffer.Length - _position; diff --git a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs index 5696c6a0..2e9f22c3 100644 --- a/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs @@ -6,8 +6,9 @@ using BenchmarkDotNet.Attributes.Jobs; using BenchmarkDotNet.Attributes.Exporters; using System; using System.Threading; -using System.IO; +using System.Threading.Tasks; using MQTTnet.Adapter; +using MQTTnet.Channel; namespace MQTTnet.Benchmarks { @@ -38,6 +39,7 @@ namespace MQTTnet.Benchmarks for (var i = 0; i < 10000; i++) { _serializer.Serialize(_packet); + _serializer.FreeBuffer(); } } @@ -46,16 +48,51 @@ namespace MQTTnet.Benchmarks { for (var i = 0; i < 10000; i++) { - using (var stream = new MemoryStream()) - { - stream.Write(_serializedPacket.Array, _serializedPacket.Offset, _serializedPacket.Count); - stream.Position = 0; + var channel = new BenchmarkMqttChannel(_serializedPacket); - var channel = new TestMqttChannel(stream); + var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult(); + _serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength))); + } + } - var header = MqttPacketReader.ReadFixedHeaderAsync(channel, CancellationToken.None).GetAwaiter().GetResult(); - _serializer.Deserialize(new ReceivedMqttPacket(header.Flags, new MqttPacketBodyReader(_serializedPacket.Array, _serializedPacket.Count - header.RemainingLength))); - } + private class BenchmarkMqttChannel : IMqttChannel + { + private readonly ArraySegment _buffer; + private int _position; + + public BenchmarkMqttChannel(ArraySegment buffer) + { + _buffer = buffer; + _position = _buffer.Offset; + } + + public string Endpoint { get; } + + public Task ConnectAsync(CancellationToken cancellationToken) + { + throw new NotImplementedException(); + } + + public Task DisconnectAsync() + { + throw new NotImplementedException(); + } + + public Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + Array.Copy(_buffer.Array, _position, buffer, offset, count); + _position += count; + + return Task.FromResult(count); + } + + public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + throw new NotImplementedException(); + } + + public void Dispose() + { } } }