From fcf276dd1a66a08ec4805d6784646cad34659c93 Mon Sep 17 00:00:00 2001 From: Christian <6939810+chkr1011@users.noreply.github.com> Date: Thu, 21 Apr 2022 20:48:52 +0200 Subject: [PATCH] Improve Server packet management. --- .../Internal/MqttPacketBus_Tests.cs | 39 +++++++++++++++-- .../MqttPacketSerializationHelper.cs | 1 - Source/MQTTnet/Internal/MqttPacketBus.cs | 39 ++++++++++++++--- Source/MQTTnet/Server/Internal/MqttClient.cs | 20 ++++----- .../Internal/MqttClientSessionsManager.cs | 6 +-- Source/MQTTnet/Server/Internal/MqttSession.cs | 43 ++++++++----------- Source/MQTTnet/Server/MqttServer.cs | 2 +- .../Server/Status/MqttSessionStatus.cs | 4 +- 8 files changed, 101 insertions(+), 53 deletions(-) diff --git a/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs b/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs index b701a26d..814d6232 100644 --- a/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs +++ b/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs @@ -26,6 +26,37 @@ namespace MQTTnet.Tests.Internal } } + [TestMethod] + public async Task Fill_From_Different_Task() + { + const int MessageCount = 500; + + var delayRandom = new Random(); + + var bus = new MqttPacketBus(); + + _ = Task.Run( + () => + { + for (int i = 0; i < MessageCount; i++) + { + bus.EnqueueItem(new MqttPacketBusItem(MqttPingReqPacket.Instance), MqttPacketBusPartition.Health); + + Thread.Sleep(delayRandom.Next(0, 10)); + } + }); + + for (var i = 0; i < MessageCount; i++) + { + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10))) + { + await bus.DequeueItemAsync(timeout.Token); + } + } + + Assert.AreEqual(0, bus.TotalItemsCount); + } + [TestMethod] public void Alternate_Priorities() { @@ -43,7 +74,7 @@ namespace MQTTnet.Tests.Internal bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); bus.EnqueueItem(new MqttPacketBusItem(new MqttPingRespPacket()), MqttPacketBusPartition.Health); - Assert.AreEqual(9, bus.ItemsCount); + Assert.AreEqual(9, bus.TotalItemsCount); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPublishPacket)); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); @@ -55,7 +86,7 @@ namespace MQTTnet.Tests.Internal Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttSubAckPacket)); Assert.IsInstanceOfType(bus.DequeueItemAsync(CancellationToken.None).Result.Packet, typeof(MqttPingRespPacket)); - Assert.AreEqual(0, bus.ItemsCount); + Assert.AreEqual(0, bus.TotalItemsCount); } [TestMethod] @@ -67,7 +98,7 @@ namespace MQTTnet.Tests.Internal bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); bus.EnqueueItem(new MqttPacketBusItem(new MqttPublishPacket()), MqttPacketBusPartition.Data); - Assert.AreEqual(3, bus.ItemsCount); + Assert.AreEqual(3, bus.TotalItemsCount); var exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Control); Assert.AreEqual(0, exportedPackets.Count); @@ -78,7 +109,7 @@ namespace MQTTnet.Tests.Internal exportedPackets = bus.ExportPackets(MqttPacketBusPartition.Data); Assert.AreEqual(3, exportedPackets.Count); - Assert.AreEqual(3, bus.ItemsCount); + Assert.AreEqual(3, bus.TotalItemsCount); } [TestMethod] diff --git a/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs b/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs index 4d98e8c2..530b185d 100644 --- a/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs +++ b/Source/MQTTnet.Tests/MqttPacketSerializationHelper.cs @@ -3,7 +3,6 @@ using System.Threading; using MQTTnet.Adapter; using MQTTnet.Diagnostics; using MQTTnet.Formatter; -using MQTTnet.Internal; using MQTTnet.Packets; using MQTTnet.Tests.Mockups; diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs index e5919c15..af02cbed 100644 --- a/Source/MQTTnet/Internal/MqttPacketBus.cs +++ b/Source/MQTTnet/Internal/MqttPacketBus.cs @@ -20,12 +20,12 @@ namespace MQTTnet.Internal new LinkedList() }; - readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0); + readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1); readonly object _syncRoot = new object(); int _activePartition = (int)MqttPacketBusPartition.Health; - public int ItemsCount + public int TotalItemsCount { get { @@ -55,12 +55,26 @@ namespace MQTTnet.Internal { for (var i = 0; i < 3; i++) { + // Iterate through the partitions in order to ensure processing of health packets + // even if lots of data packets are enqueued. + + // Partition | Messages (left = oldest). + // DATA | [#]######################### + // CONTROL | [#]############# + // HEALTH | [#]#### + + // In this sample the 3 oldest messages from the partitions are processed in a row. + // Then the next 3 from all 3 partitions. + MoveActivePartition(); - if (_partitions[_activePartition].Count > 0) + var activePartition = _partitions[_activePartition]; + + if (activePartition.First != null) { - var item = _partitions[_activePartition].First; - _partitions[_activePartition].RemoveFirst(); + var item = activePartition.First; + activePartition.RemoveFirst(); + return item.Value; } } @@ -104,9 +118,12 @@ namespace MQTTnet.Internal lock (_syncRoot) { _partitions[(int)partition].AddLast(item); - } - _semaphore.Release(); + if (_semaphore.CurrentCount == 0) + { + _semaphore.Release(); + } + } } public List ExportPackets(MqttPacketBusPartition partition) @@ -117,6 +134,14 @@ namespace MQTTnet.Internal } } + public int ItemsCount(MqttPacketBusPartition partition) + { + lock (_syncRoot) + { + return _partitions[(int)partition].Count; + } + } + public int PartitionItemsCount(MqttPacketBusPartition partition) { lock (_syncRoot) diff --git a/Source/MQTTnet/Server/Internal/MqttClient.cs b/Source/MQTTnet/Server/Internal/MqttClient.cs index 718b4850..6265539c 100644 --- a/Source/MQTTnet/Server/Internal/MqttClient.cs +++ b/Source/MQTTnet/Server/Internal/MqttClient.cs @@ -189,7 +189,6 @@ namespace MQTTnet.Server while (!cancellationToken.IsCancellationRequested) { var packet = await ChannelAdapter.ReceivePacketAsync(cancellationToken).ConfigureAwait(false); - if (packet == null) { return; @@ -251,8 +250,7 @@ namespace MQTTnet.Server else if (packet is MqttPingReqPacket) { // See: The Server MUST send a PINGRESP packet in response to a PINGREQ packet [MQTT-3.12.4-1]. - //await SendPacketAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false); - Session.EnqueuePacket(new MqttPacketBusItem(MqttPingRespPacket.Instance)); + Session.EnqueueHealthPacket(new MqttPacketBusItem(MqttPingRespPacket.Instance)); } else if (packet is MqttPingRespPacket) { @@ -342,7 +340,7 @@ namespace MQTTnet.Server if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { publishPacket.Dup = true; - Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); + Session.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); } } @@ -366,13 +364,13 @@ namespace MQTTnet.Server void HandleIncomingPubRecPacket(MqttPubRecPacket pubRecPacket) { var pubRelPacket = _packetFactories.PubRel.Create(pubRecPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueuePacket(new MqttPacketBusItem(pubRelPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubRelPacket)); } void HandleIncomingPubRelPacket(MqttPubRelPacket pubRelPacket) { var pubCompPacket = _packetFactories.PubComp.Create(pubRelPacket, MqttApplicationMessageReceivedReasonCode.Success); - Session.EnqueuePacket(new MqttPacketBusItem(pubCompPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubCompPacket)); } async Task HandleIncomingSubscribePacket(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) @@ -381,7 +379,7 @@ namespace MQTTnet.Server var subAckPacket = _packetFactories.SubAck.Create(subscribePacket, subscribeResult); - Session.EnqueuePacket(new MqttPacketBusItem(subAckPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(subAckPacket)); if (subscribeResult.CloseConnection) { @@ -394,7 +392,7 @@ namespace MQTTnet.Server foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) { var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); - Session.EnqueuePacket(new MqttPacketBusItem(publishPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(publishPacket)); } } } @@ -405,7 +403,7 @@ namespace MQTTnet.Server var unsubAckPacket = _packetFactories.UnsubAck.Create(unsubscribePacket, unsubscribeResult); - Session.EnqueuePacket(new MqttPacketBusItem(unsubAckPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(unsubAckPacket)); if (unsubscribeResult.CloseConnection) { @@ -470,13 +468,13 @@ namespace MQTTnet.Server case MqttQualityOfServiceLevel.AtLeastOnce: { var pubAckPacket = _packetFactories.PubAck.Create(publishPacket, interceptingPublishEventArgs); - Session.EnqueuePacket(new MqttPacketBusItem(pubAckPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubAckPacket)); break; } case MqttQualityOfServiceLevel.ExactlyOnce: { var pubRecPacket = _packetFactories.PubRec.Create(publishPacket, interceptingPublishEventArgs); - Session.EnqueuePacket(new MqttPacketBusItem(pubRecPacket)); + Session.EnqueueControlPacket(new MqttPacketBusItem(pubRecPacket)); break; } default: diff --git a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs index 8517888a..7d2e47ec 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientSessionsManager.cs @@ -181,7 +181,7 @@ namespace MQTTnet.Server newPublishPacket.Retain = false; } - session.EnqueuePacket(new MqttPacketBusItem(newPublishPacket)); + session.EnqueueDataPacket(new MqttPacketBusItem(newPublishPacket)); deliveryCount++; _logger.Verbose("Client '{0}': Queued PUBLISH packet with topic '{1}'.", session.Id, applicationMessage.Topic); @@ -229,7 +229,7 @@ namespace MQTTnet.Server } } - public Task> GetClientStatusAsync() + public Task> GetClientStatusesAsync() { var result = new List(); @@ -424,7 +424,7 @@ namespace MQTTnet.Server foreach (var retainedApplicationMessage in subscribeResult.RetainedMessages) { var publishPacket = _packetFactories.Publish.Create(retainedApplicationMessage.ApplicationMessage); - clientSession.EnqueuePacket(new MqttPacketBusItem(publishPacket)); + clientSession.EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); } } } diff --git a/Source/MQTTnet/Server/Internal/MqttSession.cs b/Source/MQTTnet/Server/Internal/MqttSession.cs index 9bb83374..7053a547 100644 --- a/Source/MQTTnet/Server/Internal/MqttSession.cs +++ b/Source/MQTTnet/Server/Internal/MqttSession.cs @@ -24,7 +24,7 @@ namespace MQTTnet.Server readonly Dictionary _unacknowledgedPublishPackets = new Dictionary(); - // Bookkeeping to know if this is a subscribing client; lazy intialize later. + // Bookkeeping to know if this is a subscribing client; lazy initialize later. HashSet _subscribedTopics; public MqttSession( @@ -100,14 +100,9 @@ namespace MQTTnet.Server SubscriptionsManager.Dispose(); } - public void EnqueuePacket(MqttPacketBusItem packetBusItem) + public void EnqueueDataPacket(MqttPacketBusItem packetBusItem) { - if (packetBusItem == null) - { - throw new ArgumentNullException(nameof(packetBusItem)); - } - - if (_packetBus.ItemsCount >= _serverOptions.MaxPendingMessagesPerClient) + if (_packetBus.ItemsCount(MqttPacketBusPartition.Data) >= _serverOptions.MaxPendingMessagesPerClient) { if (_serverOptions.PendingMessagesOverflowStrategy == MqttPendingMessagesOverflowStrategy.DropNewMessage) { @@ -122,25 +117,25 @@ namespace MQTTnet.Server } } - if (packetBusItem.Packet is MqttPublishPacket publishPacket) + var publishPacket = (MqttPublishPacket)packetBusItem.Packet; + if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) { - if (publishPacket.QualityOfServiceLevel > MqttQualityOfServiceLevel.AtMostOnce) - { - _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket; - } + _unacknowledgedPublishPackets[publishPacket.PacketIdentifier] = publishPacket; + } - _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data); - } - else if (packetBusItem.Packet is MqttPingReqPacket || packetBusItem.Packet is MqttPingRespPacket) - { - _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health); - } - else - { - _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control); - } + _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Data); } + public void EnqueueControlPacket(MqttPacketBusItem packetBusItem) + { + _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Control); + } + + public void EnqueueHealthPacket(MqttPacketBusItem packetBusItem) + { + _packetBus.EnqueueItem(packetBusItem, MqttPacketBusPartition.Health); + } + public void Recover() { // TODO: Keep the bus and only insert pending items again. @@ -149,7 +144,7 @@ namespace MQTTnet.Server foreach (var publishPacket in _unacknowledgedPublishPackets.Values.ToList()) { - EnqueuePacket(new MqttPacketBusItem(publishPacket)); + EnqueueDataPacket(new MqttPacketBusItem(publishPacket)); } } diff --git a/Source/MQTTnet/Server/MqttServer.cs b/Source/MQTTnet/Server/MqttServer.cs index 4084e8f5..fa3f49f2 100644 --- a/Source/MQTTnet/Server/MqttServer.cs +++ b/Source/MQTTnet/Server/MqttServer.cs @@ -182,7 +182,7 @@ namespace MQTTnet.Server { ThrowIfNotStarted(); - return _clientSessionsManager.GetClientStatusAsync(); + return _clientSessionsManager.GetClientStatusesAsync(); } public Task> GetRetainedMessagesAsync() diff --git a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs index 7dfd641e..e9319395 100644 --- a/Source/MQTTnet/Server/Status/MqttSessionStatus.cs +++ b/Source/MQTTnet/Server/Status/MqttSessionStatus.cs @@ -33,7 +33,7 @@ namespace MQTTnet.Server if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); var publishPacketFactory = new MqttPublishPacketFactory(); - _session.EnqueuePacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage))); + _session.EnqueueDataPacket(new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage))); return PlatformAbstractionLayer.CompletedTask; } @@ -44,7 +44,7 @@ namespace MQTTnet.Server var publishPacketFactory = new MqttPublishPacketFactory(); var packetBusItem = new MqttPacketBusItem(publishPacketFactory.Create(applicationMessage)); - _session.EnqueuePacket(packetBusItem); + _session.EnqueueDataPacket(packetBusItem); return packetBusItem.WaitForDeliveryAsync(); }