YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-15 00:15:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Serializer memory usage.

Browse Source
This commit is contained in:
Christian Kratky
2018-06-17 13:47:58 +02:00
parent beb54acc2f
commit 56e9e15ac0
5 changed files with 117 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Adapter/MqttChannelAdapter.cs
+2
View File
@@ -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)
{
Source/MQTTnet/Serializer/IMqttPacketSerializer.cs
+2
View File
@@ -11,5 +11,7 @@ namespace MQTTnet.Serializer
ArraySegment<byte> Serialize(MqttBasePacket mqttPacket);
MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket);
void FreeBuffer();
}
}
Source/MQTTnet/Serializer/MqttPacketSerializer.cs
+46 -27
View File
@@ -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<byte> 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<byte>(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
Source/MQTTnet/Serializer/MqttPacketWriter.cs
+21
View File
@@ -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;
Tests/MQTTnet.Benchmarks/SerializerBenchmark.cs
+46 -9
View File
@@ -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<byte> _buffer;
private int _position;
public BenchmarkMqttChannel(ArraySegment<byte> 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<int> 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()
{
}
}
}