From 39edd22a9d482b17d5ab1c7b55e4c93b223ee326 Mon Sep 17 00:00:00 2001 From: Christian Kratky Date: Mon, 4 Jun 2018 22:19:07 +0200 Subject: [PATCH] Refactor thread management for incoming packets. --- Build/MQTTnet.nuspec | 2 +- .../Adapter/MqttChannelAdapter.cs | 3 +- .../Client/IMqttClientOptions.cs | 3 +- .../MQTTnet.NetStandard/Client/MqttClient.cs | 77 +++++-------------- .../Client/MqttClientOptions.cs | 3 +- .../Client/MqttClientOptionsBuilder.cs | 7 -- .../Client/MqttPacketDispatcher.cs | 9 ++- ...eceivedApplicationMessageProcessingMode.cs | 8 -- 8 files changed, 31 insertions(+), 81 deletions(-) delete mode 100644 Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs diff --git a/Build/MQTTnet.nuspec b/Build/MQTTnet.nuspec index 0f0049c1..fc9e1a6f 100644 --- a/Build/MQTTnet.nuspec +++ b/Build/MQTTnet.nuspec @@ -15,7 +15,7 @@ * [Core] Added several packet validations. * [Core] Log messages now contain the complete source path including parent components. * [Core] The adapter now has an _Endpoint_ definition as string containing remote IP and port. -* [Client] Received messages are now processed in the worker thread by default. Added a new setting for switching back to dedicated threads. +* [Client] Received messages are now processed completely in the worker thread without creating new Tasks. * [Client] Fixed wrong calculation for sending keep alive packets (thanks to @cstichlberger) * [ManagedClient] Fixed a loading issue of stored application messages (thanks to @JTrotta). * [Server] Added support for other WebSocket sub protocol formats like mqttv-3.1.1 (thanks to @israellot). diff --git a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs index 92c8f881..941feeb0 100644 --- a/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs +++ b/Frameworks/MQTTnet.NetStandard/Adapter/MqttChannelAdapter.cs @@ -135,9 +135,10 @@ namespace MQTTnet.Adapter { var fixedHeader = await MqttPacketReader.ReadFixedHeaderAsync(channel, cancellationToken).ConfigureAwait(false); - ReadingPacketStarted?.Invoke(this, EventArgs.Empty); try { + ReadingPacketStarted?.Invoke(this, EventArgs.Empty); + var bodyLength = await MqttPacketReader.ReadBodyLengthAsync(channel, cancellationToken).ConfigureAwait(false); if (bodyLength == 0) { diff --git a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs index aa873af0..176f5e43 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/IMqttClientOptions.cs @@ -14,8 +14,7 @@ namespace MQTTnet.Client TimeSpan CommunicationTimeout { get; } TimeSpan KeepAlivePeriod { get; } TimeSpan? KeepAliveSendInterval { get; } - MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; } - + MqttApplicationMessage WillMessage { get; } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs index f7267d2a..eeb3d333 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClient.cs @@ -19,7 +19,7 @@ namespace MQTTnet.Client private readonly Stopwatch _sendTracker = new Stopwatch(); private readonly SemaphoreSlim _disconnectLock = new SemaphoreSlim(1, 1); private readonly MqttPacketDispatcher _packetDispatcher = new MqttPacketDispatcher(); - + private readonly IMqttClientAdapterFactory _adapterFactory; private readonly IMqttNetChildLogger _logger; @@ -390,25 +390,7 @@ namespace MQTTnet.Client while (!cancellationToken.IsCancellationRequested) { var packet = await _adapter.ReceivePacketAsync(TimeSpan.Zero, cancellationToken).ConfigureAwait(false); - - if (cancellationToken.IsCancellationRequested) - { - return; - } - - if (packet == null) - { - continue; - } - - if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.SingleThread) - { - await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); - } - else if (_options.ReceivedApplicationMessageProcessingMode == MqttReceivedApplicationMessageProcessingMode.DedicatedThread) - { - StartProcessReceivedPacketAsync(packet, cancellationToken); - } + await ProcessReceivedPacketAsync(packet, cancellationToken).ConfigureAwait(false); } } catch (Exception exception) @@ -434,40 +416,30 @@ namespace MQTTnet.Client } } - private async Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) + private Task ProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) { - try + if (packet is MqttPublishPacket publishPacket) { - if (packet is MqttPublishPacket publishPacket) - { - await ProcessReceivedPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); - return; - } - - if (packet is MqttPingReqPacket) - { - await SendAsync(new MqttPingRespPacket(), cancellationToken).ConfigureAwait(false); - return; - } - - if (packet is MqttDisconnectPacket) - { - await DisconnectAsync().ConfigureAwait(false); - return; - } - - if (packet is MqttPubRelPacket pubRelPacket) - { - await ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken).ConfigureAwait(false); - return; - } - - _packetDispatcher.Dispatch(packet); + return ProcessReceivedPublishPacketAsync(publishPacket, cancellationToken); } - catch (Exception exception) + + if (packet is MqttPingReqPacket) { - _logger.Error(exception, "Unhandled exception while processing received packet."); + return SendAsync(new MqttPingRespPacket(), cancellationToken); } + + if (packet is MqttDisconnectPacket) + { + return DisconnectAsync(); + } + + if (packet is MqttPubRelPacket pubRelPacket) + { + return ProcessReceivedPubRelPacket(pubRelPacket, cancellationToken); + } + + _packetDispatcher.Dispatch(packet); + return Task.FromResult(0); } private Task ProcessReceivedPublishPacketAsync(MqttPublishPacket publishPacket, CancellationToken cancellationToken) @@ -518,13 +490,6 @@ namespace MQTTnet.Client cancellationToken); } - private void StartProcessReceivedPacketAsync(MqttBasePacket packet, CancellationToken cancellationToken) - { -#pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed - Task.Run(() => ProcessReceivedPacketAsync(packet, cancellationToken), cancellationToken); -#pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed - } - private void FireApplicationMessageReceivedEvent(MqttPublishPacket publishPacket) { try diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs index ae0517a6..7d3ad5d8 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptions.cs @@ -14,8 +14,7 @@ namespace MQTTnet.Client public TimeSpan CommunicationTimeout { get; set; } = TimeSpan.FromSeconds(10); public TimeSpan KeepAlivePeriod { get; set; } = TimeSpan.FromSeconds(15); public TimeSpan? KeepAliveSendInterval { get; set; } - public MqttReceivedApplicationMessageProcessingMode ReceivedApplicationMessageProcessingMode { get; set; } = MqttReceivedApplicationMessageProcessingMode.SingleThread; - + public MqttApplicationMessage WillMessage { get; set; } } } diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs index 2eefb813..985cda7c 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttClientOptionsBuilder.cs @@ -114,13 +114,6 @@ namespace MQTTnet.Client return this; } - public MqttClientOptionsBuilder WithReceivedApplicationMessageProcessingMode( - MqttReceivedApplicationMessageProcessingMode mode) - { - _options.ReceivedApplicationMessageProcessingMode = mode; - return this; - } - public IMqttClientOptions Build() { if (_tlsOptions != null) diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs index 1a1021c4..417b3510 100644 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs +++ b/Frameworks/MQTTnet.NetStandard/Client/MqttPacketDispatcher.cs @@ -13,7 +13,7 @@ namespace MQTTnet.Client { foreach (var awaiter in _awaiters) { - awaiter.Value.SetException(exception); + Task.Run(() => awaiter.Value.SetException(exception)); // Task.Run fixes a dead lock. Without this the client only receives one message. } } @@ -30,9 +30,10 @@ namespace MQTTnet.Client var type = packet.GetType(); var key = new Tuple(identifier, type); - if (_awaiters.TryRemove(key, out var tcs)) + if (_awaiters.TryRemove(key, out var awaiter)) { - tcs.TrySetResult(packet); + awaiter.SetResult(packet); + Task.Run(() => awaiter.SetResult(packet)); // Task.Run fixes a dead lock. Without this the client only receives one message. return; } @@ -70,7 +71,7 @@ namespace MQTTnet.Client } var key = new Tuple(identifier ?? 0, typeof(TResponsePacket)); - _awaiters.TryRemove(key, out var _); + _awaiters.TryRemove(key, out _); } } } \ No newline at end of file diff --git a/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs b/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs deleted file mode 100644 index 651abe5b..00000000 --- a/Frameworks/MQTTnet.NetStandard/Client/MqttReceivedApplicationMessageProcessingMode.cs +++ /dev/null @@ -1,8 +0,0 @@ -namespace MQTTnet.Client -{ - public enum MqttReceivedApplicationMessageProcessingMode - { - SingleThread, - DedicatedThread - } -}