diff --git a/MQTTnet.Core/Client/MqttPacketDispatcher.cs b/MQTTnet.Core/Client/MqttPacketDispatcher.cs index 1551d3fa..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,18 +10,14 @@ 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> _packetByIdentifier = new ConcurrentDictionary>(); + private readonly ConcurrentDictionary>> _packetByResponseTypeAndIdentifier = new ConcurrentDictionary>>(); public async Task WaitForPacketAsync(MqttBasePacket request, Type responseType, TimeSpan timeout) { if (request == null) throw new ArgumentNullException(nameof(request)); var packetAwaiter = AddPacketAwaiter(request, responseType); - DispatchPendingPackets(); - try { return await packetAwaiter.Task.TimeoutAfter(timeout); @@ -42,43 +37,27 @@ namespace MQTTnet.Core.Client { if (packet == null) throw new ArgumentNullException(nameof(packet)); - var packetDispatched = false; - + var type = packet.GetType(); 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 ); + } } } - else if (_packetByResponseType.TryRemove(packet.GetType(), out var tcs)) + 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(); - } - - _packetByIdentifier.Clear(); + _packetByResponseTypeAndIdentifier.Clear(); + _packetByResponseType.Clear(); } private TaskCompletionSource AddPacketAwaiter(MqttBasePacket request, Type responseType) @@ -86,7 +65,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,26 +80,13 @@ 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 { _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 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..de5cd27e 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(2000, 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