From 089eb9b604d9525159e6bf8a4cd185cc53265c4d Mon Sep 17 00:00:00 2001 From: Eggers Jan Date: Thu, 14 Sep 2017 13:37:45 +0200 Subject: [PATCH 1/4] fixed unittest server and client both generated id 2 for differnt package types --- MQTTnet.Core/Client/MqttPacketDispatcher.cs | 22 +++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/MQTTnet.Core/Client/MqttPacketDispatcher.cs b/MQTTnet.Core/Client/MqttPacketDispatcher.cs index 1551d3fa..a825581f 100644 --- a/MQTTnet.Core/Client/MqttPacketDispatcher.cs +++ b/MQTTnet.Core/Client/MqttPacketDispatcher.cs @@ -14,7 +14,7 @@ namespace MQTTnet.Core.Client private readonly object _syncRoot = new object(); private readonly HashSet _receivedPackets = new HashSet(); private readonly ConcurrentDictionary> _packetByResponseType = new ConcurrentDictionary>(); - private readonly ConcurrentDictionary> _packetByIdentifier = new ConcurrentDictionary>(); + private readonly ConcurrentDictionary>> _packetByResponseTypeAndIdentifier = new ConcurrentDictionary>>(); public async Task WaitForPacketAsync(MqttBasePacket request, Type responseType, TimeSpan timeout) { @@ -42,17 +42,21 @@ namespace MQTTnet.Core.Client { if (packet == null) throw new ArgumentNullException(nameof(packet)); + var type = packet.GetType(); var packetDispatched = false; if (packet is IMqttPacketWithIdentifier withIdentifier) { - if (_packetByIdentifier.TryRemove(withIdentifier.PacketIdentifier, out var tcs)) + if (_packetByResponseTypeAndIdentifier.TryGetValue(type, out var byid)) { - tcs.TrySetResult(packet); - packetDispatched = true; + if (byid.TryRemove( withIdentifier.PacketIdentifier, out var tcs)) + { + tcs.TrySetResult( packet ); + packetDispatched = true; + } } } - else if (_packetByResponseType.TryRemove(packet.GetType(), out var tcs)) + else if (_packetByResponseType.TryRemove(type, out var tcs)) { tcs.TrySetResult(packet); packetDispatched = true; @@ -78,7 +82,7 @@ namespace MQTTnet.Core.Client _receivedPackets.Clear(); } - _packetByIdentifier.Clear(); + _packetByResponseTypeAndIdentifier.Clear(); } private TaskCompletionSource AddPacketAwaiter(MqttBasePacket request, Type responseType) @@ -86,7 +90,8 @@ namespace MQTTnet.Core.Client var tcs = new TaskCompletionSource(); if (request is IMqttPacketWithIdentifier withIdent) { - _packetByIdentifier[withIdent.PacketIdentifier] = tcs; + var byId = _packetByResponseTypeAndIdentifier.GetOrAdd(responseType, key => new ConcurrentDictionary>()); + byId[withIdent.PacketIdentifier] = tcs; } else { @@ -100,7 +105,8 @@ namespace MQTTnet.Core.Client { if (request is IMqttPacketWithIdentifier withIdent) { - _packetByIdentifier.TryRemove(withIdent.PacketIdentifier, out var _); + var byId = _packetByResponseTypeAndIdentifier.GetOrAdd(responseType, key => new ConcurrentDictionary>()); + byId.TryRemove(withIdent.PacketIdentifier, out var _); } else { From cf7b7191f8992e971d0c0bb9cdfba715b4f04957 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 14 Sep 2017 21:48:38 +0200 Subject: [PATCH 2/4] if we wait before sending the packets we dont need to store the received packets at all because the awaiter will always be there or is already canceled --- MQTTnet.Core/Client/MqttPacketDispatcher.cs | 41 +-------------------- 1 file changed, 1 insertion(+), 40 deletions(-) diff --git a/MQTTnet.Core/Client/MqttPacketDispatcher.cs b/MQTTnet.Core/Client/MqttPacketDispatcher.cs index a825581f..21c177ca 100644 --- a/MQTTnet.Core/Client/MqttPacketDispatcher.cs +++ b/MQTTnet.Core/Client/MqttPacketDispatcher.cs @@ -1,5 +1,4 @@ using System; -using System.Collections.Generic; using System.Threading.Tasks; using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Exceptions; @@ -11,8 +10,6 @@ namespace MQTTnet.Core.Client { public class MqttPacketDispatcher { - private readonly object _syncRoot = new object(); - private readonly HashSet _receivedPackets = new HashSet(); private readonly ConcurrentDictionary> _packetByResponseType = new ConcurrentDictionary>(); private readonly ConcurrentDictionary>> _packetByResponseTypeAndIdentifier = new ConcurrentDictionary>>(); @@ -21,8 +18,6 @@ namespace MQTTnet.Core.Client if (request == null) throw new ArgumentNullException(nameof(request)); var packetAwaiter = AddPacketAwaiter(request, responseType); - DispatchPendingPackets(); - try { return await packetAwaiter.Task.TimeoutAfter(timeout); @@ -43,8 +38,6 @@ namespace MQTTnet.Core.Client if (packet == null) throw new ArgumentNullException(nameof(packet)); var type = packet.GetType(); - var packetDispatched = false; - if (packet is IMqttPacketWithIdentifier withIdentifier) { if (_packetByResponseTypeAndIdentifier.TryGetValue(type, out var byid)) @@ -52,37 +45,19 @@ namespace MQTTnet.Core.Client if (byid.TryRemove( withIdentifier.PacketIdentifier, out var tcs)) { tcs.TrySetResult( packet ); - packetDispatched = true; } } } else if (_packetByResponseType.TryRemove(type, out var tcs)) { tcs.TrySetResult(packet); - packetDispatched = true; - } - - lock (_syncRoot) - { - if (!packetDispatched) - { - _receivedPackets.Add(packet); - } - else - { - _receivedPackets.Remove(packet); - } } } public void Reset() { - lock (_syncRoot) - { - _receivedPackets.Clear(); - } - _packetByResponseTypeAndIdentifier.Clear(); + _packetByResponseType.Clear(); } private TaskCompletionSource AddPacketAwaiter(MqttBasePacket request, Type responseType) @@ -113,19 +88,5 @@ namespace MQTTnet.Core.Client _packetByResponseType.TryRemove(responseType, out var _); } } - - private void DispatchPendingPackets() - { - List receivedPackets; - lock (_syncRoot) - { - receivedPackets = new List(_receivedPackets); - } - - foreach (var pendingPacket in receivedPackets) - { - Dispatch(pendingPacket); - } - } } } \ No newline at end of file From d1b634511d653f24d0812a3b1259f265f2814e79 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 14 Sep 2017 22:18:54 +0200 Subject: [PATCH 3/4] reduced footprint of client msg queue and numbers can go even higher --- MQTTnet.Core/Server/MqttClientMessageQueue.cs | 73 ++++++------------- .../Server/MqttClientPublishPacketContext.cs | 19 ----- .../PerformanceTest.cs | 7 +- 3 files changed, 30 insertions(+), 69 deletions(-) delete mode 100644 MQTTnet.Core/Server/MqttClientPublishPacketContext.cs diff --git a/MQTTnet.Core/Server/MqttClientMessageQueue.cs b/MQTTnet.Core/Server/MqttClientMessageQueue.cs index 3cbe8823..a8845e05 100644 --- a/MQTTnet.Core/Server/MqttClientMessageQueue.cs +++ b/MQTTnet.Core/Server/MqttClientMessageQueue.cs @@ -6,16 +6,16 @@ using MQTTnet.Core.Adapter; using MQTTnet.Core.Diagnostics; using MQTTnet.Core.Exceptions; using MQTTnet.Core.Packets; +using System.Linq; namespace MQTTnet.Core.Server { public sealed class MqttClientMessageQueue { - private readonly BlockingCollection _pendingPublishPackets = new BlockingCollection(); + private readonly BlockingCollection _pendingPublishPackets = new BlockingCollection(); private readonly MqttServerOptions _options; private CancellationTokenSource _cancellationTokenSource; - private IMqttCommunicationAdapter _adapter; public MqttClientMessageQueue(MqttServerOptions options) { @@ -29,15 +29,14 @@ namespace MQTTnet.Core.Server throw new InvalidOperationException($"{nameof(MqttClientMessageQueue)} already started."); } - _adapter = adapter ?? throw new ArgumentNullException(nameof(adapter)); + if (adapter == null) throw new ArgumentNullException(nameof(adapter)); _cancellationTokenSource = new CancellationTokenSource(); - Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token), _cancellationTokenSource.Token); + Task.Run(() => SendPendingPublishPacketsAsync(_cancellationTokenSource.Token, adapter), _cancellationTokenSource.Token); } public void Stop() { - _adapter = null; _cancellationTokenSource?.Cancel(); _cancellationTokenSource = null; _pendingPublishPackets?.Dispose(); @@ -47,61 +46,37 @@ namespace MQTTnet.Core.Server { if (publishPacket == null) throw new ArgumentNullException(nameof(publishPacket)); - _pendingPublishPackets.Add(new MqttClientPublishPacketContext(publishPacket)); + _pendingPublishPackets.Add(publishPacket); } - private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken) + private async Task SendPendingPublishPacketsAsync(CancellationToken cancellationToken, IMqttCommunicationAdapter adapter) { - foreach (var publishPacket in _pendingPublishPackets.GetConsumingEnumerable(cancellationToken)) + var consumable = _pendingPublishPackets.GetConsumingEnumerable(); + while (!cancellationToken.IsCancellationRequested) { + var packets = consumable.Take(_pendingPublishPackets.Count).ToList(); try { - if (cancellationToken.IsCancellationRequested) - { - return; - } - - if (_adapter == null) - { - continue; - } - - await TrySendPendingPublishPacketAsync(publishPacket).ConfigureAwait(false); + await adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, packets).ConfigureAwait(false); } - catch (Exception e) + catch (MqttCommunicationException exception) { - MqttTrace.Error(nameof(MqttClientMessageQueue), e, "Error while sending pending publish packets."); + MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed."); + foreach (var publishPacket in packets) + { + publishPacket.Dup = true; + _pendingPublishPackets.Add(publishPacket); + } } - } - } - - private async Task TrySendPendingPublishPacketAsync(MqttClientPublishPacketContext publishPacketContext) - { - try - { - if (_adapter == null) + catch (Exception exception) { - return; + MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed."); + foreach (var publishPacket in packets) + { + publishPacket.Dup = true; + _pendingPublishPackets.Add(publishPacket); + } } - - publishPacketContext.PublishPacket.Dup = publishPacketContext.SendTries > 0; - await _adapter.SendPacketsAsync(_options.DefaultCommunicationTimeout, publishPacketContext.PublishPacket).ConfigureAwait(false); - - publishPacketContext.IsSent = true; - } - catch (MqttCommunicationException exception) - { - MqttTrace.Warning(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed."); - _pendingPublishPackets.Add(publishPacketContext); - } - catch (Exception exception) - { - MqttTrace.Error(nameof(MqttClientMessageQueue), exception, "Sending publish packet failed."); - _pendingPublishPackets.Add(publishPacketContext); - } - finally - { - publishPacketContext.SendTries++; } } } diff --git a/MQTTnet.Core/Server/MqttClientPublishPacketContext.cs b/MQTTnet.Core/Server/MqttClientPublishPacketContext.cs deleted file mode 100644 index 98553906..00000000 --- a/MQTTnet.Core/Server/MqttClientPublishPacketContext.cs +++ /dev/null @@ -1,19 +0,0 @@ -using System; -using MQTTnet.Core.Packets; - -namespace MQTTnet.Core.Server -{ - public sealed class MqttClientPublishPacketContext - { - public MqttClientPublishPacketContext(MqttPublishPacket publishPacket) - { - PublishPacket = publishPacket ?? throw new ArgumentNullException(nameof(publishPacket)); - } - - public MqttPublishPacket PublishPacket { get; } - - public int SendTries { get; set; } - - public bool IsSent { get; set; } - } -} diff --git a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs index 8e0f4305..052778ad 100644 --- a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs @@ -18,11 +18,16 @@ namespace MQTTnet.TestApp.NetFramework public static async Task RunAsync() { var server = Task.Run(() => RunServerAsync()); - var client = Task.Run(() => RunClientAsync(300, TimeSpan.FromMilliseconds(10))); + var client = Task.Run(() => RunClientAsync(400, TimeSpan.FromMilliseconds(10))); await Task.WhenAll(server, client).ConfigureAwait(false); } + private static Task RunClientsAsync(int msgChunkSize, TimeSpan interval) + { + return Task.WhenAll(Enumerable.Range(0, 3).Select((i) => Task.Run(() => RunClientAsync(msgChunkSize, interval)))); + } + private static async Task RunClientAsync( int msgChunkSize, TimeSpan interval ) { try From f1ff947697f1a8c92f75c0db70f19e40332f9ebe Mon Sep 17 00:00:00 2001 From: JanEggers Date: Thu, 14 Sep 2017 22:26:28 +0200 Subject: [PATCH 4/4] higher numbers --- Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs index 052778ad..de5cd27e 100644 --- a/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs +++ b/Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs @@ -18,7 +18,7 @@ namespace MQTTnet.TestApp.NetFramework public static async Task RunAsync() { var server = Task.Run(() => RunServerAsync()); - var client = Task.Run(() => RunClientAsync(400, TimeSpan.FromMilliseconds(10))); + var client = Task.Run(() => RunClientAsync(2000, TimeSpan.FromMilliseconds(10))); await Task.WhenAll(server, client).ConfigureAwait(false); }