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

Performance optimizations

Browse Source
This commit is contained in:
Christian Kratky
2017-11-25 23:10:48 +01:00
parent 24ab285066
commit 611fabcd01
11 changed files with 105 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs
+8 -5
View File
@@ -60,8 +60,11 @@ namespace MQTTnet.Adapter
_logger.Trace<MqttChannelAdapter>("TX >>> {0} [Timeout={1}]", packet, timeout);
var writeBuffer = PacketSerializer.Serialize(packet);
await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
var chunks = PacketSerializer.Serialize(packet);
foreach (var chunk in chunks)
{
await _channel.SendStream.WriteAsync(chunk.Array, chunk.Offset, chunk.Count, cancellationToken).ConfigureAwait(false);
}
}
if (timeout > TimeSpan.Zero)
@@ -114,7 +117,7 @@ namespace MQTTnet.Adapter
{
receivedMqttPacket?.Dispose();
}
});
}).ConfigureAwait(false);
return packet;
}
@@ -140,8 +143,8 @@ namespace MQTTnet.Adapter
var readBytesCount = await stream.ReadAsync(body, offset, body.Length - offset, cancellationToken).ConfigureAwait(false);
offset += readBytesCount;
} while (offset < header.BodyLength);
return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length));
return new ReceivedMqttPacket(header, new MemoryStream(body, 0, body.Length, false, true));
}
private static async Task ExecuteAndWrapExceptionAsync(Func<Task> action)
Frameworks/MQTTnet.NetStandard/Serializer/IMqttPacketSerializer.cs
+4 -2
View File
@@ -1,4 +1,6 @@
using MQTTnet.Adapter;
using System;
using System.Collections.Generic;
using MQTTnet.Adapter;
using MQTTnet.Packets;
namespace MQTTnet.Serializer
@@ -7,7 +9,7 @@ namespace MQTTnet.Serializer
{
MqttProtocolVersion ProtocolVersion { get; set; }
byte[] Serialize(MqttBasePacket mqttPacket);
IEnumerable<ArraySegment<byte>> Serialize(MqttBasePacket mqttPacket);
MqttBasePacket Deserialize(ReceivedMqttPacket receivedMqttPacket);
}
Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketReader.cs
+16 -6
View File
@@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;
@@ -30,7 +29,7 @@ namespace MQTTnet.Serializer
{
return null;
}
var fixedHeader = (byte)buffer;
var controlPacketType = (MqttControlPacketType)(fixedHeader >> 4);
var bodyLength = ReadBodyLengthFromSource(stream, cancellationToken);
@@ -57,12 +56,22 @@ namespace MQTTnet.Serializer
public string ReadStringWithLengthPrefix()
{
var buffer = ReadWithLengthPrefix();
if (buffer.Length == 0)
{
return string.Empty;
}
return Encoding.UTF8.GetString(buffer, 0, buffer.Length);
}
public byte[] ReadWithLengthPrefix()
{
var length = ReadUInt16();
if (length == 0)
{
return new byte[0];
}
return ReadBytes(length);
}
@@ -78,7 +87,7 @@ namespace MQTTnet.Serializer
var value = 0;
byte encodedByte;
var readBytes = new List<int>();
////var readBytes = new List<int>();
do
{
if (cancellationToken.IsCancellationRequested)
@@ -87,19 +96,20 @@ namespace MQTTnet.Serializer
}
var buffer = stream.ReadByte();
readBytes.Add(buffer);
if (buffer == -1)
{
throw new MqttCommunicationException("Connection closed while reading remaining length data.");
}
////readBytes.Add(buffer);
encodedByte = (byte)buffer;
value += (byte)(encodedByte & 127) * multiplier;
multiplier *= 128;
if (multiplier > 128 * 128 * 128)
{
throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
//throw new MqttProtocolViolationException($"Remaining length is invalid (Data={string.Join(",", readBytes)}).");
throw new MqttProtocolViolationException("Remaining length is invalid.");
}
} while ((encodedByte & 128) != 0);
Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketSerializer.cs
+21 -17
View File
@@ -1,12 +1,12 @@
using System;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using MQTTnet.Adapter;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;
namespace MQTTnet.Serializer
{
@@ -17,25 +17,29 @@ namespace MQTTnet.Serializer
public MqttProtocolVersion ProtocolVersion { get; set; } = MqttProtocolVersion.V311;
public byte[] Serialize(MqttBasePacket packet)
public IEnumerable<ArraySegment<byte>> Serialize(MqttBasePacket packet)
{
if (packet == null) throw new ArgumentNullException(nameof(packet));
using (var stream = new MemoryStream())
using (var stream = new MemoryStream(128))
using (var writer = new MqttPacketWriter(stream))
{
var fixedHeader = SerializePacket(packet, writer);
var headerBuffer = new List<byte> { fixedHeader };
MqttPacketWriter.WriteRemainingLength((int)stream.Length, headerBuffer);
var remainingLength = (int)stream.Length;
writer.Write(fixedHeader);
MqttPacketWriter.WriteRemainingLength(remainingLength, writer);
var headerLength = (int)stream.Length - remainingLength;
var header = headerBuffer.ToArray();
var body = stream.ToArray();
var buffer = new byte[header.Length + body.Length];
Buffer.BlockCopy(header, 0, buffer, 0, header.Length);
Buffer.BlockCopy(body, 0, buffer, header.Length, body.Length);
return buffer;
#if NET461 || NET452 || NETSTANDARD2_0
var buffer = stream.GetBuffer();
#else
var buffer = stream.ToArray();
#endif
return new List<ArraySegment<byte>>
{
new ArraySegment<byte>(buffer, remainingLength, headerLength),
new ArraySegment<byte>(buffer, 0, remainingLength)
};
}
}
Frameworks/MQTTnet.NetStandard/Serializer/MqttPacketWriter.cs
+3 -4
View File
@@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using MQTTnet.Protocol;
@@ -56,11 +55,11 @@ namespace MQTTnet.Serializer
Write(value);
}
public static void WriteRemainingLength(int length, List<byte> target)
public static void WriteRemainingLength(int length, BinaryWriter target)
{
if (length == 0)
{
target.Add(0);
target.Write((byte)0);
return;
}
@@ -75,7 +74,7 @@ namespace MQTTnet.Serializer
encodedByte = encodedByte | 128;
}
target.Add((byte)encodedByte);
target.Write((byte)encodedByte);
} while (x > 0);
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSession.cs
+13 -25
View File
@@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet.Adapter;
@@ -14,8 +13,6 @@ namespace MQTTnet.Server
{
public sealed class MqttClientSession
{
private readonly HashSet<ushort> _unacknowledgedPublishPackets = new HashSet<ushort>();
private readonly MqttClientSubscriptionsManager _subscriptionsManager;
private readonly MqttClientSessionsManager _sessionsManager;
private readonly MqttClientPendingMessagesQueue _pendingMessagesQueue;
@@ -99,21 +96,22 @@ namespace MQTTnet.Server
if (willMessage != null)
{
_willMessage = null; //clear willmessage so it is send just once
await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage);
await _sessionsManager.DispatchApplicationMessageAsync(this, willMessage).ConfigureAwait(false);
}
}
}
public void EnqueuePublishPacket(MqttPublishPacket publishPacket)
public void EnqueueApplicationMessage(MqttApplicationMessage applicationMessage)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
var result = _subscriptionsManager.CheckSubscriptions(publishPacket);
var result = _subscriptionsManager.CheckSubscriptions(applicationMessage);
if (!result.IsSubscribed)
{
return;
}
var publishPacket = applicationMessage.ToPublishPacket();
publishPacket.QualityOfServiceLevel = result.QualityOfServiceLevel;
_pendingMessagesQueue.Enqueue(publishPacket);
}
@@ -134,12 +132,12 @@ namespace MQTTnet.Server
catch (MqttCommunicationException exception)
{
_logger.Warning<MqttClientSession>(exception, "Client '{0}': Communication exception while processing client packets.", ClientId);
await StopAsync();
await StopAsync().ConfigureAwait(false);
}
catch (Exception exception)
{
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId);
await StopAsync();
await StopAsync().ConfigureAwait(false);
}
}
@@ -200,7 +198,7 @@ namespace MQTTnet.Server
if (subscribeResult.CloseConnection)
{
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttDisconnectPacket()).ConfigureAwait(false);
await StopAsync();
await StopAsync().ConfigureAwait(false);
}
}
@@ -209,7 +207,7 @@ namespace MQTTnet.Server
var retainedMessages = await _sessionsManager.GetRetainedMessagesAsync(subscribePacket).ConfigureAwait(false);
foreach (var publishPacket in retainedMessages)
{
EnqueuePublishPacket(publishPacket.ToPublishPacket());
EnqueueApplicationMessage(publishPacket);
}
}
@@ -221,27 +219,22 @@ namespace MQTTnet.Server
{
case MqttQualityOfServiceLevel.AtMostOnce:
{
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
return;
}
case MqttQualityOfServiceLevel.AtLeastOnce:
{
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier });
new MqttPubAckPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);
return;
}
case MqttQualityOfServiceLevel.ExactlyOnce:
{
// QoS 2 is implement as method "B" [4.3.3 QoS 2: Exactly once delivery]
lock (_unacknowledgedPublishPackets)
{
_unacknowledgedPublishPackets.Add(publishPacket.PacketIdentifier);
}
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage);
await _sessionsManager.DispatchApplicationMessageAsync(this, applicationMessage).ConfigureAwait(false);
await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken,
new MqttPubRecPacket { PacketIdentifier = publishPacket.PacketIdentifier }).ConfigureAwait(false);
@@ -255,11 +248,6 @@ namespace MQTTnet.Server
private Task HandleIncomingPubRelPacketAsync(IMqttChannelAdapter adapter, MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
lock (_unacknowledgedPublishPackets)
{
_unacknowledgedPublishPackets.Remove(pubRelPacket.PacketIdentifier);
}
return adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttPubCompPacket { PacketIdentifier = pubRelPacket.PacketIdentifier });
}
}
Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs
+4 -5
View File
@@ -6,7 +6,6 @@ using System.Threading.Tasks;
using MQTTnet.Adapter;
using MQTTnet.Diagnostics;
using MQTTnet.Exceptions;
using MQTTnet.Internal;
using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Serializer;
@@ -59,7 +58,7 @@ namespace MQTTnet.Server
return;
}
var clientSession = await GetOrCreateClientSessionAsync(connectPacket);
var clientSession = await GetOrCreateClientSessionAsync(connectPacket).ConfigureAwait(false);
await clientAdapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, cancellationToken, new MqttConnAckPacket
{
@@ -105,7 +104,7 @@ namespace MQTTnet.Server
{
foreach (var session in _sessions)
{
await session.Value.StopAsync();
await session.Value.StopAsync().ConfigureAwait(false);
}
_sessions.Clear();
@@ -162,7 +161,7 @@ namespace MQTTnet.Server
{
foreach (var clientSession in _sessions.Values.ToList())
{
clientSession.EnqueuePublishPacket(applicationMessage.ToPublishPacket());
clientSession.EnqueueApplicationMessage(applicationMessage);
}
}
}
@@ -200,7 +199,7 @@ namespace MQTTnet.Server
if (connectPacket.CleanSession)
{
_sessions.Remove(connectPacket.ClientId);
await clientSession.StopAsync();
await clientSession.StopAsync().ConfigureAwait(false);
clientSession = null;
_logger.Trace<MqttClientSessionsManager>("Stopped existing session of client '{0}'.", connectPacket.ClientId);
Frameworks/MQTTnet.NetStandard/Server/MqttClientSubscriptionsManager.cs
+5 -5
View File
@@ -64,23 +64,23 @@ namespace MQTTnet.Server
return unsubscribePacket.CreateResponse<MqttUnsubAckPacket>();
}
public CheckSubscriptionsResult CheckSubscriptions(MqttPublishPacket publishPacket)
public CheckSubscriptionsResult CheckSubscriptions(MqttApplicationMessage applicationMessage)
{
if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket));
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage));
lock (_subscriptions)
{
foreach (var subscription in _subscriptions)
{
if (!MqttTopicFilterComparer.IsMatch(publishPacket.Topic, subscription.Key))
if (!MqttTopicFilterComparer.IsMatch(applicationMessage.Topic, subscription.Key))
{
continue;
}
var effectiveQos = subscription.Value;
if (publishPacket.QualityOfServiceLevel < effectiveQos)
if (applicationMessage.QualityOfServiceLevel < effectiveQos)
{
effectiveQos = publishPacket.QualityOfServiceLevel;
effectiveQos = applicationMessage.QualityOfServiceLevel;
}
return new CheckSubscriptionsResult
Tests/MQTTnet.Core.Tests/MqttPacketSerializerTests.cs
+18 -6
View File
@@ -1,4 +1,5 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;
@@ -392,9 +393,9 @@ namespace MQTTnet.Core.Tests
private static void SerializeAndCompare(MqttBasePacket packet, string expectedBase64Value, MqttProtocolVersion protocolVersion = MqttProtocolVersion.V311)
{
var serializer = new MqttPacketSerializer { ProtocolVersion = protocolVersion };
var buffer = serializer.Serialize(packet);
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer));
var chunks = serializer.Serialize(packet);
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(chunks)));
}
private static void DeserializeAndCompare(MqttBasePacket packet, string expectedBase64Value)
@@ -403,18 +404,29 @@ namespace MQTTnet.Core.Tests
var buffer1 = serializer.Serialize(packet);
using (var headerStream = new MemoryStream(buffer1))
using (var headerStream = new MemoryStream(Join(buffer1)))
{
var header = MqttPacketReader.ReadHeaderFromSource(headerStream, CancellationToken.None);
using (var bodyStream = new MemoryStream(buffer1, (int)headerStream.Position, header.BodyLength))
using (var bodyStream = new MemoryStream(Join(buffer1), (int)headerStream.Position, header.BodyLength))
{
var deserializedPacket = serializer.Deserialize(new ReceivedMqttPacket(header, bodyStream));
var buffer2 = serializer.Serialize(deserializedPacket);
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(buffer2));
Assert.AreEqual(expectedBase64Value, Convert.ToBase64String(Join(buffer2)));
}
}
}
private static byte[] Join(IEnumerable<ArraySegment<byte>> chunks)
{
var buffer = new MemoryStream();
foreach (var chunk in chunks)
{
buffer.Write(chunk.Array, chunk.Offset, chunk.Count);
}
return buffer.ToArray();
}
}
}
Tests/MQTTnet.Core.Tests/MqttSubscriptionsManagerTests.cs
+3 -3
View File
@@ -18,7 +18,7 @@ namespace MQTTnet.Core.Tests
sm.Subscribe(sp, "");
var pp = new MqttPublishPacket
var pp = new MqttApplicationMessage
{
Topic = "A/B/C",
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
@@ -37,7 +37,7 @@ namespace MQTTnet.Core.Tests
sm.Subscribe(sp, "");
var pp = new MqttPublishPacket
var pp = new MqttApplicationMessage
{
Topic = "A/B/X",
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
@@ -56,7 +56,7 @@ namespace MQTTnet.Core.Tests
sm.Subscribe(sp, "");
var pp = new MqttPublishPacket
var pp = new MqttApplicationMessage
{
Topic = "A/B/C",
QualityOfServiceLevel = MqttQualityOfServiceLevel.AtMostOnce
Tests/MQTTnet.TestApp.NetCore/PerformanceTest.cs
+10 -10
View File
@@ -171,16 +171,16 @@ namespace MQTTnet.TestApp.NetCore
var msgs = 0;
var stopwatch = Stopwatch.StartNew();
mqttServer.ApplicationMessageReceived += (sender, args) =>
{
msgs++;
if (stopwatch.ElapsedMilliseconds > 1000)
{
Console.WriteLine($"received {msgs}");
msgs = 0;
stopwatch.Restart();
}
};
////mqttServer.ApplicationMessageReceived += (sender, args) =>
////{
//// msgs++;
//// if (stopwatch.ElapsedMilliseconds > 1000)
//// {
//// Console.WriteLine($"received {msgs}");
//// msgs = 0;
//// stopwatch.Restart();
//// }
////};
await mqttServer.StartAsync(new MqttServerOptions());
Console.WriteLine("Press any key to exit.");