diff --git a/.github/workflows/ReleaseNotes.md b/.github/workflows/ReleaseNotes.md index 687239f2..b2cc9759 100644 --- a/.github/workflows/ReleaseNotes.md +++ b/.github/workflows/ReleaseNotes.md @@ -1 +1,2 @@ +* [Core] Fixed several dead locks by implementing a new async signal (#1552) * [Server] Fixed duplicated invocation of the event _ClientAcknowledgedPublishPacketAsync_ for QoS level 2 (#1550) \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs b/Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs new file mode 100644 index 00000000..df63fd40 --- /dev/null +++ b/Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs @@ -0,0 +1,114 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using MQTTnet.Internal; + +namespace MQTTnet.Tests.Internal +{ + [TestClass] + public sealed class AsyncSignal_Tests + { + [TestMethod] + [ExpectedException(typeof(TaskCanceledException))] + public async Task Cancel_If_No_Signal() + { + var asyncSignal = new AsyncSignal(); + + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2))) + { + await asyncSignal.WaitAsync(timeout.Token); + + Assert.Fail("There is no signal. So we must fail here!"); + } + } + + [TestMethod] + [ExpectedException(typeof(ObjectDisposedException))] + public async Task Dispose_Properly() + { + var asyncSignal = new AsyncSignal(); + + // The timeout will not be reached but another task will kill the async signal. + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(999))) + { + _ = Task.Run( + async () => + { + await Task.Delay(2000, CancellationToken.None); + asyncSignal.Dispose(); + }, + CancellationToken.None); + + await asyncSignal.WaitAsync(timeout.Token); + + Assert.Fail("There is no signal. So we must fail here!"); + } + } + + [TestMethod] + public async Task Loop_Signal() + { + var asyncSignal = new AsyncSignal(); + + for (var i = 0; i < 10; i++) + { + asyncSignal.Set(); + + // WaitAsync should return directly because the signal is available. + await asyncSignal.WaitAsync(); + } + } + + [TestMethod] + public async Task Signal() + { + var asyncSignal = new AsyncSignal(); + + // The timeout will not be reached but another task will kill the async signal. + using (var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(3))) + { + var stopwatch = Stopwatch.StartNew(); + + _ = Task.Run( + async () => + { + await Task.Delay(1000, CancellationToken.None); + asyncSignal.Set(); + }, + CancellationToken.None); + + await asyncSignal.WaitAsync(timeout.Token); + + stopwatch.Stop(); + + Assert.IsTrue(stopwatch.ElapsedMilliseconds > 900); + } + } + + [TestMethod] + [ExpectedException(typeof(InvalidOperationException))] + public async Task Fail_For_Two_Waiters() + { + var asyncSignal = new AsyncSignal(); + + // This thread will wait properly because it is the first waiter. + _ = Task.Run( + async () => + { + await asyncSignal.WaitAsync(); + }, + CancellationToken.None); + + await Task.Delay(1000); + + // Now the current thread must fail because there is already a waiter. + await asyncSignal.WaitAsync(); + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs b/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs index 814d6232..b09159fd 100644 --- a/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs +++ b/Source/MQTTnet.Tests/Internal/MqttPacketBus_Tests.cs @@ -15,7 +15,7 @@ namespace MQTTnet.Tests.Internal public sealed class MqttPacketBus_Tests { [TestMethod] - [ExpectedException(typeof(OperationCanceledException))] + [ExpectedException(typeof(TaskCanceledException))] public async Task Wait_With_Empty_Bus() { var bus = new MqttPacketBus(); diff --git a/Source/MQTTnet.Tests/Server/Load_Tests.cs b/Source/MQTTnet.Tests/Server/Load_Tests.cs index 7cba9c12..7c446ee0 100644 --- a/Source/MQTTnet.Tests/Server/Load_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Load_Tests.cs @@ -55,7 +55,7 @@ namespace MQTTnet.Tests.Server }); } - SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(60)); + SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(90)); Assert.AreEqual(100000, receivedMessages); } diff --git a/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs b/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs index b514a399..70e9bffd 100644 --- a/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Subscription_TopicHash_Tests.cs @@ -2,10 +2,8 @@ using System; using System.Collections.Generic; using System.Linq; using System.Text; -using System.Threading; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; -using MQTTnet.Client; using MQTTnet.Protocol; using MQTTnet.Server; using MQTTnet.Tests.Mockups; @@ -13,11 +11,11 @@ using MQTTnet.Tests.Mockups; namespace MQTTnet.Tests.Server { [TestClass] - public class Subscription_TopicHash_Tests + public sealed class SubscriptionTopicHashTests { enum TopicHashSelector { NoWildcard, SingleWildcard, MultiWildcard }; - MQTTnet.Server.MqttSession _clientSession; + MqttSession _clientSession; [TestMethod] public void Match_Hash_Test_SingleWildCard() @@ -26,13 +24,9 @@ namespace MQTTnet.Tests.Server var l1 = "topic1"; var l2 = "+"; var l3 = "prop1"; - var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3); + var topic = $"{l0}/{l1}/{l2}/{l3}"; - - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); @@ -65,13 +59,9 @@ namespace MQTTnet.Tests.Server var l1 = "topic1"; var l2 = "sub1"; var l3 = "#"; - var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3); + var topic = $"{l0}/{l1}/{l2}/{l3}"; - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); @@ -104,13 +94,9 @@ namespace MQTTnet.Tests.Server var l1 = "topic1"; var l2 = "sub1"; var l3 = "prop1"; - var topic = string.Format("{0}/{1}/{2}/{3}", l0, l1, l2, l3); + var topic = $"{l0}/{l1}/{l2}/{l3}"; - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsFalse(topicHasWildcard, "Wildcard detected when not wildcard present"); @@ -144,14 +130,14 @@ namespace MQTTnet.Tests.Server public void Match_Hash_Test_LongTopic_MultiWildcard() { var sb = new StringBuilder(); - const int NumLevels = 8; - var levelNames = new string[NumLevels]; - for (var i = 0; i < NumLevels; ++i) + const int numLevels = 8; + var levelNames = new string[numLevels]; + for (var i = 0; i < numLevels; ++i) { if (i > 0) sb.Append("/"); string levelName; - if (i == NumLevels - 1) + if (i == numLevels - 1) { // last one is # levelName = "#"; @@ -167,11 +153,7 @@ namespace MQTTnet.Tests.Server // UInt64 is limited to 8 levels - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); @@ -214,14 +196,14 @@ namespace MQTTnet.Tests.Server public void Match_Hash_Test_LongTopic_SingleWildCard() { var sb = new StringBuilder(); - const int NumLevels = 8; - var levelNames = new string[NumLevels]; - for (var i = 0; i < NumLevels; ++i) + const int numLevels = 8; + var levelNames = new string[numLevels]; + for (var i = 0; i < numLevels; ++i) { if (i > 0) sb.Append("/"); string levelName; - if (i == NumLevels - 1) + if (i == numLevels - 1) { // last one is + levelName = "+"; @@ -235,11 +217,7 @@ namespace MQTTnet.Tests.Server } var topic = sb.ToString(); - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); @@ -284,9 +262,9 @@ namespace MQTTnet.Tests.Server public void Match_Hash_Test_LongTopic_NoWildCard() { var sb = new StringBuilder(); - const int NumLevels = 9; - var levelNames = new string[NumLevels]; - for(var i=0; i < NumLevels; ++i) + const int numLevels = 9; + var levelNames = new string[numLevels]; + for(var i=0; i < numLevels; ++i) { if (i > 0) sb.Append("/"); @@ -296,11 +274,7 @@ namespace MQTTnet.Tests.Server } var topic = sb.ToString(); - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsFalse(topicHasWildcard, "Wildcard detected when not present"); @@ -332,11 +306,7 @@ namespace MQTTnet.Tests.Server { var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/+"; - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); } @@ -349,11 +319,7 @@ namespace MQTTnet.Tests.Server { var topic = "asdfasdf/asdfasdf/asdfasdf/asdfasdf/asdfas/dfaf/assfdgsdfgdf/#"; - UInt64 topicHash; - UInt64 topicHashMask; - bool topicHasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out topicHashMask, out topicHasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var topicHashMask, out var topicHasWildcard); Assert.IsTrue(topicHasWildcard, "Wildcard not detected"); } @@ -422,14 +388,10 @@ namespace MQTTnet.Tests.Server [TestMethod] public void Check_Hash_Bucket_Depth() { - Dictionary> topicsByPublisher; - Dictionary> singleWildcardTopicsByPublisher; - Dictionary> multiWildcardTopicsByPublisher; + const int numPublishers = 5000; + const int numTopicsPerPublisher = 10; - const int NumPublishers = 5000; - const int NumTopicsPerPublisher = 10; - - TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher); + TopicGenerator.Generate(numPublishers, numTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher); // There will be many 'similar' topics ending with, i.e. "sensor100", "sensor101", ... // Hash bucket depths should remain low. @@ -444,10 +406,7 @@ namespace MQTTnet.Tests.Server var topics = t.Value; foreach (var topic in topics) { - UInt64 topicHash; - UInt64 hashMask; - bool hasWildcard; - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out hashMask, out hasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard); bucketDepths.TryGetValue(topicHash, out var currentValue); ++currentValue; @@ -489,16 +448,12 @@ namespace MQTTnet.Tests.Server void CheckTopicHash(string topic, ulong expectedHash, ulong expectedHashMask) { - ulong topicHash; - ulong hashMask; - bool hasWildcard; - - MQTTnet.Server.MqttSubscription.CalculateTopicHash(topic, out topicHash, out hashMask, out hasWildcard); + MqttSubscription.CalculateTopicHash(topic, out var topicHash, out var hashMask, out var hasWildcard); Console.WriteLine(); Console.WriteLine("Topic: " + topic); - Console.WriteLine(string.Format("Hash: {0:X8}", topicHash)); - Console.WriteLine(string.Format("Hash Mask: {0:X8}", hashMask)); + Console.WriteLine($"Hash: {topicHash:X8}"); + Console.WriteLine($"Hash Mask: {hashMask:X8}"); Assert.AreEqual(expectedHash, topicHash, "Topic hash not as expected. Has the hash function changed?"); Assert.AreEqual(expectedHashMask, hashMask, "Topic hash mask not as expected"); @@ -507,27 +462,23 @@ namespace MQTTnet.Tests.Server async Task> PrepareTopicHashSubscriptions(TopicHashSelector selector) { - Dictionary> topicsByPublisher; - Dictionary> singleWildcardTopicsByPublisher; - Dictionary> multiWildcardTopicsByPublisher; + const int numPublishers = 1; + const int numTopicsPerPublisher = 10000; - const int NumPublishers = 1; - const int NumTopicsPerPublisher = 10000; - - TopicGenerator.Generate(NumPublishers, NumTopicsPerPublisher, out topicsByPublisher, out singleWildcardTopicsByPublisher, out multiWildcardTopicsByPublisher); + TopicGenerator.Generate(numPublishers, numTopicsPerPublisher, out var topicsByPublisher, out var singleWildcardTopicsByPublisher, out var multiWildcardTopicsByPublisher); var topics = topicsByPublisher.FirstOrDefault().Value; var singleWildcardTopics = singleWildcardTopicsByPublisher.FirstOrDefault().Value; var multiWildcardTopics = multiWildcardTopicsByPublisher.FirstOrDefault().Value; - const string ClientId = "Client1"; - var logger = new Mockups.TestLogger(); - var serverOptions = new MQTTnet.Server.MqttServerOptions(); - var eventContainer = new MQTTnet.Server.MqttServerEventContainer(); + const string clientId = "Client1"; + var logger = new TestLogger(); + var serverOptions = new MqttServerOptions(); + var eventContainer = new MqttServerEventContainer(); var retainedMessagesManager = new MqttRetainedMessagesManager(eventContainer, logger); var sessionManager = new MqttClientSessionsManager(serverOptions, retainedMessagesManager, eventContainer, logger); - _clientSession = new MQTTnet.Server.MqttSession( - ClientId, + _clientSession = new MqttSession( + clientId, false, new Dictionary(), serverOptions, @@ -553,16 +504,18 @@ namespace MQTTnet.Tests.Server foreach (var t in topicsToSubscribe) { var subPacket = new Packets.MqttSubscribePacket(); - var filter = new Packets.MqttTopicFilter(); - filter.Topic = t; + var filter = new Packets.MqttTopicFilter + { + Topic = t + }; subPacket.TopicFilters.Add(filter); - await _clientSession.SubscriptionsManager.Subscribe(subPacket, default(CancellationToken)); + await _clientSession.SubscriptionsManager.Subscribe(subPacket, default); } return topics; } - int CheckTopicSubscriptions(MQTTnet.Server.MqttSession clientSession, List topicsToFind, string subject) + int CheckTopicSubscriptions(MqttSession clientSession, List topicsToFind, string subject) { var matchCount = 0; @@ -577,12 +530,9 @@ namespace MQTTnet.Tests.Server { var topicToFind = topicsToFind[countDown]; - UInt64 topicHash; - UInt64 hashMask; - bool hasWildcard; - MQTTnet.Server.MqttSubscription.CalculateTopicHash((string)topicToFind, out topicHash, out hashMask, out hasWildcard); + MqttSubscription.CalculateTopicHash(topicToFind, out var topicHash, out var hashMask, out var hasWildcard); - var result = clientSession.SubscriptionsManager.CheckSubscriptions((string)topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient"); + var result = clientSession.SubscriptionsManager.CheckSubscriptions(topicToFind, topicHash, MqttQualityOfServiceLevel.AtMostOnce, "OtherClient"); if (result.IsSubscribed) { ++matchCount; diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 7a9db45d..4b030150 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -13,7 +13,6 @@ using MQTTnet.Channel; using MQTTnet.Diagnostics; using MQTTnet.Exceptions; using MQTTnet.Formatter; -using MQTTnet.Implementations; using MQTTnet.Internal; using MQTTnet.Packets; @@ -296,8 +295,26 @@ namespace MQTTnet.Adapter while (totalBytesRead < buffer.Length) { - var bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); + // Check two times for cancellation because the call to _ReadAsync_ might block for some time. + if (cancellationToken.IsCancellationRequested) + { + return ReadFixedHeaderResult.Cancelled; + } + int bytesRead; + try + { + bytesRead = await _channel.ReadAsync(buffer, totalBytesRead, buffer.Length - totalBytesRead, cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return ReadFixedHeaderResult.Cancelled; + } + catch (SocketException) + { + return ReadFixedHeaderResult.ConnectionClosed; + } + if (cancellationToken.IsCancellationRequested) { return ReadFixedHeaderResult.Cancelled; diff --git a/Source/MQTTnet/Client/MqttClient.cs b/Source/MQTTnet/Client/MqttClient.cs index 92004fe8..48c32bf4 100644 --- a/Source/MQTTnet/Client/MqttClient.cs +++ b/Source/MQTTnet/Client/MqttClient.cs @@ -827,6 +827,11 @@ namespace MQTTnet.Client return; } + if (exception is AggregateException aggregateException) + { + exception = aggregateException.GetBaseException(); + } + if (exception is OperationCanceledException) { } diff --git a/Source/MQTTnet/Internal/AsyncQueue.cs b/Source/MQTTnet/Internal/AsyncQueue.cs index 5605c0f8..c54f5467 100644 --- a/Source/MQTTnet/Internal/AsyncQueue.cs +++ b/Source/MQTTnet/Internal/AsyncQueue.cs @@ -11,64 +11,39 @@ namespace MQTTnet.Internal { public sealed class AsyncQueue : IDisposable { + readonly AsyncSignal _signal = new AsyncSignal(); readonly object _syncRoot = new object(); - SemaphoreSlim _semaphore = new SemaphoreSlim(0); + ConcurrentQueue _queue = new ConcurrentQueue(); + bool _isDisposed; + public int Count => _queue.Count; + public void Clear() + { + Interlocked.Exchange(ref _queue, new ConcurrentQueue()); + } + + public void Dispose() + { + lock (_syncRoot) + { + _signal?.Dispose(); + + _isDisposed = true; + } + } + public void Enqueue(TItem item) { lock (_syncRoot) { _queue.Enqueue(item); - _semaphore?.Release(); + _signal.Set(); } } - public async Task> TryDequeueAsync(CancellationToken cancellationToken) - { - while (!cancellationToken.IsCancellationRequested) - { - try - { - Task task; - lock (_syncRoot) - { - if (_semaphore == null) - { - return new AsyncQueueDequeueResult(false, default); - } - - task = _semaphore.WaitAsync(cancellationToken); - } - - await task.ConfigureAwait(false); - - if (cancellationToken.IsCancellationRequested) - { - return new AsyncQueueDequeueResult(false, default); - } - - if (_queue.TryDequeue(out var item)) - { - return new AsyncQueueDequeueResult(true, item); - } - } - catch (ArgumentNullException) - { - // The semaphore throws this internally sometimes. - return new AsyncQueueDequeueResult(false, default); - } - catch (OperationCanceledException) - { - return new AsyncQueueDequeueResult(false, default); - } - } - - return new AsyncQueueDequeueResult(false, default); - } - public AsyncQueueDequeueResult TryDequeue() { if (_queue.TryDequeue(out var item)) @@ -79,18 +54,42 @@ namespace MQTTnet.Internal return new AsyncQueueDequeueResult(false, default); } - public void Clear() + public async Task> TryDequeueAsync(CancellationToken cancellationToken) { - Interlocked.Exchange(ref _queue, new ConcurrentQueue()); - } - - public void Dispose() - { - lock (_syncRoot) + while (!cancellationToken.IsCancellationRequested) { - _semaphore?.Dispose(); - _semaphore = null; + try + { + Task task; + lock (_syncRoot) + { + if (_isDisposed) + { + return new AsyncQueueDequeueResult(false, default); + } + + task = _signal.WaitAsync(cancellationToken); + } + + await task.ConfigureAwait(false); + + if (cancellationToken.IsCancellationRequested) + { + return new AsyncQueueDequeueResult(false, default); + } + + if (_queue.TryDequeue(out var item)) + { + return new AsyncQueueDequeueResult(true, item); + } + } + catch (OperationCanceledException) + { + return new AsyncQueueDequeueResult(false, default); + } } + + return new AsyncQueueDequeueResult(false, default); } } -} +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/AsyncSignal.cs b/Source/MQTTnet/Internal/AsyncSignal.cs new file mode 100644 index 00000000..cde8532c --- /dev/null +++ b/Source/MQTTnet/Internal/AsyncSignal.cs @@ -0,0 +1,113 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace MQTTnet.Internal +{ + public sealed class AsyncSignal : IDisposable + { + readonly object _syncRoot = new object(); + + bool _isDisposed; + bool _isSignaled; + AsyncSignalWaiter _waiter; + + public void Dispose() + { + lock (_syncRoot) + { + _waiter?.Dispose(); + _waiter = null; + + _isDisposed = true; + } + } + + public void Set() + { + lock (_syncRoot) + { + _isSignaled = true; + + // If there is already a waiting task let it run. + _waiter?.Approve(); + _waiter = null; + } + } + + public Task WaitAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + + lock (_syncRoot) + { + ThrowIfDisposed(); + + if (_isSignaled) + { + return CompletedTask.Instance; + } + + if (_waiter != null) + { + if (!_waiter.Task.IsCompleted) + { + throw new InvalidOperationException("Only one waiting task is permitted per async signal."); + } + + _waiter.Dispose(); + } + + _waiter = new AsyncSignalWaiter(cancellationToken); + return _waiter.Task; + } + } + + void ThrowIfDisposed() + { + if (_isDisposed) + { + throw new ObjectDisposedException(nameof(AsyncSignal)); + } + } + + sealed class AsyncSignalWaiter : IDisposable + { + readonly AsyncTaskCompletionSource _promise = new AsyncTaskCompletionSource(); + + // ReSharper disable once FieldCanBeMadeReadOnly.Local + CancellationTokenRegistration _cancellationTokenRegistration; + + public AsyncSignalWaiter(CancellationToken cancellationToken) + { + if (cancellationToken.CanBeCanceled) + { + _cancellationTokenRegistration = cancellationToken.Register(Cancel); + } + } + + public Task Task => _promise.Task; + + public void Approve() + { + _promise.TrySetResult(true); + } + + public void Dispose() + { + _cancellationTokenRegistration.Dispose(); + + _promise.TrySetException(new ObjectDisposedException(nameof(AsyncSignalWaiter))); + } + + void Cancel() + { + _promise.TrySetCanceled(); + } + } + } +} \ No newline at end of file diff --git a/Source/MQTTnet/Internal/MqttPacketBus.cs b/Source/MQTTnet/Internal/MqttPacketBus.cs index af02cbed..002e3d58 100644 --- a/Source/MQTTnet/Internal/MqttPacketBus.cs +++ b/Source/MQTTnet/Internal/MqttPacketBus.cs @@ -20,7 +20,7 @@ namespace MQTTnet.Internal new LinkedList() }; - readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1); + readonly AsyncSignal _signal = new AsyncSignal(); readonly object _syncRoot = new object(); int _activePartition = (int)MqttPacketBusPartition.Health; @@ -82,7 +82,16 @@ namespace MQTTnet.Internal // No partition contains data so that we have to wait and put // the worker back to the thread pool. - await _semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await _signal.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (ObjectDisposedException) + { + // The cancelled token should "hide" the disposal of the signal. + cancellationToken.ThrowIfCancellationRequested(); + throw; + } } cancellationToken.ThrowIfCancellationRequested(); @@ -92,7 +101,7 @@ namespace MQTTnet.Internal public void Dispose() { - _semaphore?.Dispose(); + _signal.Dispose(); } public void DropFirstItem(MqttPacketBusPartition partition) @@ -118,11 +127,7 @@ namespace MQTTnet.Internal lock (_syncRoot) { _partitions[(int)partition].AddLast(item); - - if (_semaphore.CurrentCount == 0) - { - _semaphore.Release(); - } + _signal.Set(); } }