From 210d57b2548fb70113947534d842986bb48a12c9 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Mon, 19 Nov 2018 21:09:36 -0500 Subject: [PATCH 1/9] IEquatable for speed IEquatable for speed in using EqualityComparer.Default.Equals --- .../ManagedMqttApplicationMessage.cs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs index bba73b0c..e8906e26 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs @@ -2,10 +2,15 @@ namespace MQTTnet.Extensions.ManagedClient { - public class ManagedMqttApplicationMessage + public class ManagedMqttApplicationMessage : IEquatable { public Guid Id { get; set; } = Guid.NewGuid(); public MqttApplicationMessage ApplicationMessage { get; set; } + + public bool Equals(ManagedMqttApplicationMessage other) + { + return Id.Equals(other.Id); + } } } From 9ff39c1fd920eac6d4b714278d68da65ac0e8d85 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Mon, 19 Nov 2018 21:11:47 -0500 Subject: [PATCH 2/9] New functions This is to support the peek-and-publish model for the managed client --- Source/MQTTnet/Internal/BlockingQueue.cs | 32 ++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/Source/MQTTnet/Internal/BlockingQueue.cs b/Source/MQTTnet/Internal/BlockingQueue.cs index d6c28f46..546b6461 100644 --- a/Source/MQTTnet/Internal/BlockingQueue.cs +++ b/Source/MQTTnet/Internal/BlockingQueue.cs @@ -55,6 +55,38 @@ namespace MQTTnet.Internal _gate.WaitOne(); } } + + public TItem PeekAndWait() + { + while (true) + { + lock (_syncRoot) + { + if (_items.Count > 0) + { + return _items.First.Value; + } + + if (_items.Count == 0) + { + _gate.Reset(); + } + } + + _gate.WaitOne(); + } + } + + public void RemoveFirstIfEqual(TItem item) + { + lock (_syncRoot) + { + if (_items.Count > 0 && EqualityComparer.Default.Equals(_items.First.Value, item)) + { + _items.RemoveFirst(); + } + } + } public TItem RemoveFirst() { From b2c31331b67e4eb9d153063152e38b2d95d6bb1a Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Mon, 19 Nov 2018 21:20:50 -0500 Subject: [PATCH 3/9] Peek-and-publish model We had been seeing an issue in which the queue could grow larger than the configured cap. I examined the code and saw that this could happen if _mqttClient.PublishAsync() throws an exception, in which case a message can be re-enqueued without honoring the cap. Furthermore, I saw that it was possible for the DropOldestQueuedMessage strategy to drop messages that were not actually the oldest ones, because when re-enqueueing the messages in the queue are no longer ordered by the original time they entered the queue. It made sense to us to peek at the message when publishing rather than dequeue it, so that when re-enqueueing after an exception 1) the cap is still honored and 2) the order of queued messages isn't altered. It's ok if another thread removes the message that's currently being published from the queue due to the cap, because all we have to do then is check if it's already been removed before removing it ourselves. --- .../ManagedMqttClient.cs | 34 +++++++++++++++++-- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 271e6b93..3568d1f7 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -254,7 +254,15 @@ namespace MQTTnet.Extensions.ManagedClient { while (!cancellationToken.IsCancellationRequested && _mqttClient.IsConnected) { - var message = _messageQueue.Dequeue(); + //Peek at the message without dequeueing in order to prevent the + //possibility of the queue growing beyond the configured cap. + //Previously, messages could be re-enqueued if there was an + //exception, and this re-enqueueing did not honor the cap. + //Furthermore, because re-enqueueing would shuffle the order + //of the messages, the DropOldestQueuedMessage strategy would + //be unable to know which message is actually the oldest and would + //instead drop the first item in the queue. + var message = _messageQueue.PeekAndWait(); if (message == null) { continue; @@ -284,6 +292,16 @@ namespace MQTTnet.Extensions.ManagedClient try { _mqttClient.PublishAsync(message.ApplicationMessage).GetAwaiter().GetResult(); + lock (_messageQueue) //lock to avoid conflict with this.PublishAsync + { + //While publishing this message, this.PublishAsync could have booted this + //message off the queue to make room for another (when using a cap + //with the DropOldestQueuedMessage strategy). If the first item + //in the queue is equal to this message, then it's safe to remove + //it from the queue. If not, that means this.PublishAsync has already + //removed it, in which case we don't want to do anything. + _messageQueue.RemoveFirstIfEqual(message); + } _storageManager?.RemoveAsync(message).GetAwaiter().GetResult(); } catch (MqttCommunicationException exception) @@ -292,9 +310,19 @@ namespace MQTTnet.Extensions.ManagedClient _logger.Warning(exception, $"Publishing application ({message.Id}) message failed."); - if (message.ApplicationMessage.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) + if (message.ApplicationMessage.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce) { - _messageQueue.Enqueue(message); + //If QoS 0, we don't want this message to stay on the queue. + //If QoS 1 or 2, it's possible that, when using a cap, this message + //has been booted off the queue by this.PublishAsync, in which case this + //thread will not continue to try to publish it. While this does + //contradict the expected behavior of QoS 1 and 2, that's also true + //for the usage of a message queue cap, so it's still consistent + //with prior behavior in that way. + lock (_messageQueue) //lock to avoid conflict with this.PublishAsync + { + _messageQueue.RemoveFirstIfEqual(message); + } } } catch (Exception exception) From 2c6c3ac6a4d070f6bbab48e2e9123f596bddad5a Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 21 Nov 2018 20:47:12 -0500 Subject: [PATCH 4/9] Changes to this file no longer needed --- .../ManagedMqttApplicationMessage.cs | 16 ---------------- 1 file changed, 16 deletions(-) delete mode 100644 Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs deleted file mode 100644 index e8906e26..00000000 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs +++ /dev/null @@ -1,16 +0,0 @@ -using System; - -namespace MQTTnet.Extensions.ManagedClient -{ - public class ManagedMqttApplicationMessage : IEquatable - { - public Guid Id { get; set; } = Guid.NewGuid(); - - public MqttApplicationMessage ApplicationMessage { get; set; } - - public bool Equals(ManagedMqttApplicationMessage other) - { - return Id.Equals(other.Id); - } - } -} From baa924121620a9d4b93adb93d876230d0cbabf07 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 21 Nov 2018 21:35:15 -0500 Subject: [PATCH 5/9] Revert "Changes to this file no longer needed" This reverts commit 2c6c3ac6a4d070f6bbab48e2e9123f596bddad5a. --- .../ManagedMqttApplicationMessage.cs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs new file mode 100644 index 00000000..e8906e26 --- /dev/null +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs @@ -0,0 +1,16 @@ +using System; + +namespace MQTTnet.Extensions.ManagedClient +{ + public class ManagedMqttApplicationMessage : IEquatable + { + public Guid Id { get; set; } = Guid.NewGuid(); + + public MqttApplicationMessage ApplicationMessage { get; set; } + + public bool Equals(ManagedMqttApplicationMessage other) + { + return Id.Equals(other.Id); + } + } +} From d1c3d9b453d8879970266dc19450ec0db7d0ab96 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 21 Nov 2018 21:37:01 -0500 Subject: [PATCH 6/9] Removed unnecessary code --- .../ManagedMqttApplicationMessage.cs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs index e8906e26..bba73b0c 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttApplicationMessage.cs @@ -2,15 +2,10 @@ namespace MQTTnet.Extensions.ManagedClient { - public class ManagedMqttApplicationMessage : IEquatable + public class ManagedMqttApplicationMessage { public Guid Id { get; set; } = Guid.NewGuid(); public MqttApplicationMessage ApplicationMessage { get; set; } - - public bool Equals(ManagedMqttApplicationMessage other) - { - return Id.Equals(other.Id); - } } } From 74d59673a8173c051808ba0e24fcc83fc2ce0e23 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 21 Nov 2018 21:38:06 -0500 Subject: [PATCH 7/9] Delegate for comparisons --- Source/MQTTnet/Internal/BlockingQueue.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet/Internal/BlockingQueue.cs b/Source/MQTTnet/Internal/BlockingQueue.cs index 546b6461..2812349a 100644 --- a/Source/MQTTnet/Internal/BlockingQueue.cs +++ b/Source/MQTTnet/Internal/BlockingQueue.cs @@ -77,11 +77,11 @@ namespace MQTTnet.Internal } } - public void RemoveFirstIfEqual(TItem item) + public void RemoveFirstIfEqual(TItem item, Func areEqual) { lock (_syncRoot) { - if (_items.Count > 0 && EqualityComparer.Default.Equals(_items.First.Value, item)) + if (_items.Count > 0 && areEqual(_items.First.Value, item)) { _items.RemoveFirst(); } From 31de6ee186215269ae76300f6a435f295338ddba Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 21 Nov 2018 21:39:42 -0500 Subject: [PATCH 8/9] Delegate comparison function --- .../ManagedMqttClient.cs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index 3568d1f7..da36023e 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -300,7 +300,7 @@ namespace MQTTnet.Extensions.ManagedClient //in the queue is equal to this message, then it's safe to remove //it from the queue. If not, that means this.PublishAsync has already //removed it, in which case we don't want to do anything. - _messageQueue.RemoveFirstIfEqual(message); + _messageQueue.RemoveFirstIfEqual(message, IdsAreEqual); } _storageManager?.RemoveAsync(message).GetAwaiter().GetResult(); } @@ -321,7 +321,7 @@ namespace MQTTnet.Extensions.ManagedClient //with prior behavior in that way. lock (_messageQueue) //lock to avoid conflict with this.PublishAsync { - _messageQueue.RemoveFirstIfEqual(message); + _messageQueue.RemoveFirstIfEqual(message, IdsAreEqual); } } } @@ -335,6 +335,11 @@ namespace MQTTnet.Extensions.ManagedClient ApplicationMessageProcessed?.Invoke(this, new ApplicationMessageProcessedEventArgs(message, transmitException)); } } + + private bool IdsAreEqual(ManagedMqttApplicationMessage message1, ManagedMqttApplicationMessage message2) + { + return message1.Id.Equals(message2.Id); + } private async Task SynchronizeSubscriptionsAsync() { From 2a93fc18e4f7295b6e9dd2e6b97bc5fec167f237 Mon Sep 17 00:00:00 2001 From: Paul Fake Date: Wed, 21 Nov 2018 21:44:26 -0500 Subject: [PATCH 9/9] Made comparison function static --- Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs index da36023e..7647fc7f 100644 --- a/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs +++ b/Source/MQTTnet.Extensions.ManagedClient/ManagedMqttClient.cs @@ -336,7 +336,7 @@ namespace MQTTnet.Extensions.ManagedClient } } - private bool IdsAreEqual(ManagedMqttApplicationMessage message1, ManagedMqttApplicationMessage message2) + private static bool IdsAreEqual(ManagedMqttApplicationMessage message1, ManagedMqttApplicationMessage message2) { return message1.Id.Equals(message2.Id); }