From 10740f12c4e98bb22bb0a053050ae0e3f2e4afe6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Warin?= Date: Tue, 15 May 2018 15:28:14 +0200 Subject: [PATCH 1/6] Replace the Async coordination primitives --- .../Internal/AsyncAutoResetEvent.cs | 101 ++++++++- .../MQTTnet.NetStandard/Internal/AsyncLock.cs | 25 ++- .../Server/MqttClientPendingMessagesQueue.cs | 1 - .../Server/MqttClientSessionsManager.cs | 44 +--- .../Server/MqttRetainedMessagesManager.cs | 56 +++-- .../AsyncAutoResentEventTests.cs | 192 +++++++++++++++++- Tests/MQTTnet.Core.Tests/AsyncLockTests.cs | 9 +- 7 files changed, 325 insertions(+), 103 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs b/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs index ad20646c..0a304e65 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs @@ -1,26 +1,109 @@ using System; +using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; namespace MQTTnet.Internal { - public sealed class AsyncAutoResetEvent : IDisposable + // Inspired from Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/11/building-async-coordination-primitives-part-2-asyncautoresetevent/) and Chris Gillum (https://stackoverflow.com/a/43012490) + public class AsyncAutoResetEvent { - private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(0, 1); + private readonly LinkedList> waiters = new LinkedList>(); - public Task WaitOneAsync(CancellationToken cancellationToken) + private bool isSignaled; + + public AsyncAutoResetEvent(bool signaled = false) { - return _semaphore.WaitAsync(cancellationToken); + this.isSignaled = signaled; + } + + public Task WaitOneAsync() + { + return this.WaitOneAsync(CancellationToken.None); + } + + public Task WaitOneAsync(TimeSpan timeout) + { + return this.WaitOneAsync(timeout, CancellationToken.None); + } + + public Task WaitOneAsync(CancellationToken cancellationToken) + { + return this.WaitOneAsync(Timeout.InfiniteTimeSpan, cancellationToken); + } + + public async Task WaitOneAsync(TimeSpan timeout, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + + TaskCompletionSource tcs; + + lock (this.waiters) + { + if (this.isSignaled) + { + this.isSignaled = false; + return true; + } + else if (timeout == TimeSpan.Zero) + { + return this.isSignaled; + } + else + { + tcs = new TaskCompletionSource(); + this.waiters.AddLast(tcs); + } + } + + Task winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)); + if (winner == tcs.Task) + { + // The task was signaled. + return true; + } + else + { + // We timed-out; remove our reference to the task. + // This is an O(n) operation since waiters is a LinkedList. + lock (this.waiters) + { + bool removed = this.waiters.Remove(tcs); + if (winner.Status == TaskStatus.Canceled) + { + throw new OperationCanceledException(cancellationToken); + } + else + { + throw new TimeoutException(); + } + } + } } public void Set() { - _semaphore.Release(); - } + TaskCompletionSource toRelease = null; - public void Dispose() - { - _semaphore?.Dispose(); + lock (this.waiters) + { + if (this.waiters.Count > 0) + { + // Signal the first task in the waiters list. + toRelease = this.waiters.First.Value; + this.waiters.RemoveFirst(); + } + else if (!this.isSignaled) + { + // No tasks are pending + this.isSignaled = true; + } + } + + if (toRelease != null) + { + toRelease.SetResult(true); + } } } } diff --git a/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs b/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs index 145e385f..d4029493 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs @@ -4,23 +4,32 @@ using System.Threading.Tasks; namespace MQTTnet.Internal { - public sealed class AsyncLock : IDisposable + // From Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/12/building-async-coordination-primitives-part-6-asynclock/) + public sealed class AsyncLock { - private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); + private readonly SemaphoreSlim m_semaphore = new SemaphoreSlim(1, 1); + private readonly Task m_releaser; - public Task EnterAsync(CancellationToken cancellationToken) + public AsyncLock() { - return _semaphore.WaitAsync(cancellationToken); + m_releaser = Task.FromResult((IDisposable)new Releaser(this)); } - public void Exit() + public Task LockAsync(CancellationToken cancellationToken) { - _semaphore.Release(); + var wait = m_semaphore.WaitAsync(cancellationToken); + return wait.IsCompleted ? + m_releaser : + wait.ContinueWith((_, state) => (IDisposable)state, + m_releaser.Result, cancellationToken, + TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } - public void Dispose() + private sealed class Releaser : IDisposable { - _semaphore?.Dispose(); + private readonly AsyncLock m_toRelease; + internal Releaser(AsyncLock toRelease) { m_toRelease = toRelease; } + public void Dispose() { m_toRelease.m_semaphore.Release(); } } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs index fc74d4c5..f08e18e3 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientPendingMessagesQueue.cs @@ -154,7 +154,6 @@ namespace MQTTnet.Server public void Dispose() { - _queueAutoResetEvent?.Dispose(); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index e50b8c24..e47f3f32 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -119,8 +119,7 @@ namespace MQTTnet.Server public async Task StopAsync() { - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { foreach (var session in _sessions) { @@ -129,16 +128,11 @@ namespace MQTTnet.Server _sessions.Clear(); } - finally - { - _sessionsLock.Exit(); - } } public async Task> GetConnectedClientsAsync() { - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient { @@ -149,10 +143,6 @@ namespace MQTTnet.Server PendingApplicationMessages = s.Value.PendingMessagesQueue.Count }).ToList(); } - finally - { - _sessionsLock.Exit(); - } } public void StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) @@ -165,8 +155,7 @@ namespace MQTTnet.Server if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { if (!_sessions.TryGetValue(clientId, out var session)) { @@ -175,10 +164,6 @@ namespace MQTTnet.Server await session.SubscribeAsync(topicFilters).ConfigureAwait(false); } - finally - { - _sessionsLock.Exit(); - } } public async Task UnsubscribeAsync(string clientId, IList topicFilters) @@ -186,8 +171,7 @@ namespace MQTTnet.Server if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { if (!_sessions.TryGetValue(clientId, out var session)) { @@ -196,15 +180,10 @@ namespace MQTTnet.Server await session.UnsubscribeAsync(topicFilters).ConfigureAwait(false); } - finally - { - _sessionsLock.Exit(); - } } public void Dispose() { - _sessionsLock?.Dispose(); } private MqttConnectReturnCode ValidateConnection(MqttConnectPacket connectPacket) @@ -226,8 +205,8 @@ namespace MQTTnet.Server private async Task PrepareClientSessionAsync(MqttConnectPacket connectPacket) { - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + { var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession); if (isSessionPresent) @@ -261,10 +240,6 @@ namespace MQTTnet.Server return new GetOrCreateClientSessionResult { IsExistingSession = isExistingSession, Session = clientSession }; } - finally - { - _sessionsLock.Exit(); - } } private async Task DispatchApplicationMessageAsync(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) @@ -294,18 +269,13 @@ namespace MQTTnet.Server _logger.Error(exception, "Error while processing application message"); } - await _sessionsLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { foreach (var clientSession in _sessions.Values) { await clientSession.EnqueueApplicationMessageAsync(applicationMessage).ConfigureAwait(false); } } - finally - { - _sessionsLock.Exit(); - } } private MqttApplicationMessageInterceptorContext InterceptApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs index be0a5b7b..e00ff444 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs @@ -29,24 +29,22 @@ namespace MQTTnet.Server return; } - await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { - var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false); - - _messages.Clear(); - foreach (var retainedMessage in retainedMessages) + try { - _messages[retainedMessage.Topic] = retainedMessage; + var retainedMessages = await _options.Storage.LoadRetainedMessagesAsync().ConfigureAwait(false); + + _messages.Clear(); + foreach (var retainedMessage in retainedMessages) + { + _messages[retainedMessage.Topic] = retainedMessage; + } + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while loading retained messages."); } - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while loading retained messages."); - } - finally - { - _messagesLock.Exit(); } } @@ -54,18 +52,16 @@ namespace MQTTnet.Server { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { - await HandleMessageInternalAsync(clientId, applicationMessage); - } - catch (Exception exception) - { - _logger.Error(exception, "Unhandled exception while handling retained messages."); - } - finally - { - _messagesLock.Exit(); + try + { + await HandleMessageInternalAsync(clientId, applicationMessage); + } + catch (Exception exception) + { + _logger.Error(exception, "Unhandled exception while handling retained messages."); + } } } @@ -73,8 +69,7 @@ namespace MQTTnet.Server { var retainedMessages = new List(); - await _messagesLock.EnterAsync(CancellationToken.None).ConfigureAwait(false); - try + using (var releaser = await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { foreach (var retainedMessage in _messages.Values) { @@ -90,17 +85,12 @@ namespace MQTTnet.Server } } } - finally - { - _messagesLock.Exit(); - } return retainedMessages; } public void Dispose() { - _messagesLock?.Dispose(); } private async Task HandleMessageInternalAsync(string clientId, MqttApplicationMessage applicationMessage) diff --git a/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs b/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs index 77275eb9..879f9256 100644 --- a/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs +++ b/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs @@ -1,34 +1,210 @@ -using System.Threading; -using System.Threading.Tasks; -using Microsoft.VisualStudio.TestTools.UnitTesting; +using Microsoft.VisualStudio.TestTools.UnitTesting; using MQTTnet.Internal; +using System; +using System.Threading; +using System.Threading.Tasks; namespace MQTTnet.Core.Tests { [TestClass] + // Inspired from the vs-threading tests (https://github.com/Microsoft/vs-threading/blob/master/src/Microsoft.VisualStudio.Threading.Tests/AsyncAutoResetEventTests.cs) public class AsyncAutoResetEventTests { + private AsyncAutoResetEvent evt; + + public AsyncAutoResetEventTests() + { + this.evt = new AsyncAutoResetEvent(); + } + + [TestMethod] + public async Task SingleThreadedPulse() + { + for (int i = 0; i < 5; i++) + { + var t = this.evt.WaitOneAsync(); + Assert.IsFalse(t.IsCompleted); + this.evt.Set(); + await t; + Assert.IsTrue(t.IsCompleted); + } + } + + [TestMethod] + public async Task MultipleSetOnlySignalsOnce() + { + this.evt.Set(); + this.evt.Set(); + await this.evt.WaitOneAsync(); + var t = this.evt.WaitOneAsync(); + Assert.IsFalse(t.IsCompleted); + await Task.Delay(500); + Assert.IsFalse(t.IsCompleted); + this.evt.Set(); + await t; + Assert.IsTrue(t.IsCompleted); + } + + [TestMethod] + public async Task OrderPreservingQueue() + { + var waiters = new Task[5]; + for (int i = 0; i < waiters.Length; i++) + { + waiters[i] = this.evt.WaitOneAsync(); + } + + for (int i = 0; i < waiters.Length; i++) + { + this.evt.Set(); + await waiters[i]; + } + } + + /// + /// Verifies that inlining continuations do not have to complete execution before Set() returns. + /// + [TestMethod] + public async Task SetReturnsBeforeInlinedContinuations() + { + var setReturned = new ManualResetEventSlim(); + var inlinedContinuation = this.evt.WaitOneAsync() + .ContinueWith(delegate + { + // Arrange to synchronously block the continuation until Set() has returned, + // which would deadlock if Set does not return until inlined continuations complete. + Assert.IsTrue(setReturned.Wait(500)); + }); + await Task.Delay(100); + this.evt.Set(); + setReturned.Set(); + Assert.IsTrue(inlinedContinuation.Wait(500)); + } + + [TestMethod] + public void WaitAsync_WithCancellationToken() + { + var cts = new CancellationTokenSource(); + Task waitTask = this.evt.WaitOneAsync(cts.Token); + Assert.IsFalse(waitTask.IsCompleted); + + // Cancel the request and ensure that it propagates to the task. + cts.Cancel(); + try + { + waitTask.GetAwaiter().GetResult(); + Assert.IsTrue(false, "Task was expected to transition to a canceled state."); + } + catch (System.OperationCanceledException ex) + { + Assert.AreEqual(cts.Token, ex.CancellationToken); + } + + // Now set the event and verify that a future waiter gets the signal immediately. + this.evt.Set(); + waitTask = this.evt.WaitOneAsync(); + Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status); + } + + [TestMethod] + public void WaitAsync_WithCancellationToken_Precanceled() + { + // We construct our own pre-canceled token so that we can do + // a meaningful identity check later. + var tokenSource = new CancellationTokenSource(); + tokenSource.Cancel(); + var token = tokenSource.Token; + + // Verify that a pre-set signal is not reset by a canceled wait request. + this.evt.Set(); + try + { + this.evt.WaitOneAsync(token).GetAwaiter().GetResult(); + Assert.IsTrue(false, "Task was expected to transition to a canceled state."); + } + catch (OperationCanceledException ex) + { + Assert.AreEqual(token, ex.CancellationToken); + } + + // Verify that the signal was not acquired. + Task waitTask = this.evt.WaitOneAsync(); + Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status); + } + + [TestMethod] + public async Task WaitAsync_WithTimeout() + { + Task waitTask = this.evt.WaitOneAsync(TimeSpan.FromMilliseconds(500)); + Assert.IsFalse(waitTask.IsCompleted); + + // Cancel the request and ensure that it propagates to the task. + await Task.Delay(1000); + try + { + waitTask.GetAwaiter().GetResult(); + Assert.IsTrue(false, "Task was expected to transition to a timeout state."); + } + catch (System.TimeoutException) + { + Assert.IsTrue(true); + } + + // Now set the event and verify that a future waiter gets the signal immediately. + this.evt.Set(); + waitTask = this.evt.WaitOneAsync(TimeSpan.FromMilliseconds(500)); + Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status); + } + + [TestMethod] + public void WaitAsync_Canceled_DoesNotInlineContinuations() + { + var cts = new CancellationTokenSource(); + var task = this.evt.WaitOneAsync(cts.Token); + + var completingActionFinished = new ManualResetEventSlim(); + var continuation = task.ContinueWith( + _ => Assert.IsTrue(completingActionFinished.Wait(500)), + CancellationToken.None, + TaskContinuationOptions.None, + TaskScheduler.Default); + + cts.Cancel(); + completingActionFinished.Set(); + + // Rethrow the exception if it turned out it deadlocked. + continuation.GetAwaiter().GetResult(); + } + [TestMethod] public async Task AsyncAutoResetEvent() { var aare = new AsyncAutoResetEvent(); - var increment = 0; var globalI = 0; #pragma warning disable 4014 Task.Run(async () => #pragma warning restore 4014 { await aare.WaitOneAsync(CancellationToken.None); - globalI += increment; + globalI += 1; + }); + +#pragma warning disable 4014 + Task.Run(async () => +#pragma warning restore 4014 + { + await aare.WaitOneAsync(CancellationToken.None); + globalI += 2; }); await Task.Delay(500); - increment = 1; + aare.Set(); + await Task.Delay(500); aare.Set(); await Task.Delay(100); - Assert.AreEqual(1, globalI); + Assert.AreEqual(3, globalI); } } -} +} \ No newline at end of file diff --git a/Tests/MQTTnet.Core.Tests/AsyncLockTests.cs b/Tests/MQTTnet.Core.Tests/AsyncLockTests.cs index 56c70500..43f2486c 100644 --- a/Tests/MQTTnet.Core.Tests/AsyncLockTests.cs +++ b/Tests/MQTTnet.Core.Tests/AsyncLockTests.cs @@ -21,22 +21,17 @@ namespace MQTTnet.Core.Tests #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed threads[i] = Task.Run(async () => { - await @lock.EnterAsync(CancellationToken.None); - try + using (var releaser = await @lock.LockAsync(CancellationToken.None)) { var localI = globalI; await Task.Delay(10); // Increase the chance for wrong data. localI++; globalI = localI; } - finally - { - @lock.Exit(); - } }); #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed } - + Task.WaitAll(threads); Assert.AreEqual(ThreadsCount, globalI); } From 5287bf5cac2f832b39baa8580b736ebddd8d17d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Warin?= Date: Tue, 15 May 2018 16:54:37 +0200 Subject: [PATCH 2/6] Codacy/PR Quality Review --- .../MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs | 9 ++++++--- Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs | 7 ++++++- Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs | 6 +++--- 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs b/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs index 0a304e65..a8fa8529 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs @@ -12,7 +12,10 @@ namespace MQTTnet.Internal private bool isSignaled; - public AsyncAutoResetEvent(bool signaled = false) + public AsyncAutoResetEvent() : this(false) + { } + + public AsyncAutoResetEvent(bool signaled) { this.isSignaled = signaled; } @@ -56,7 +59,7 @@ namespace MQTTnet.Internal } } - Task winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)); + Task winner = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)).ConfigureAwait(false); if (winner == tcs.Task) { // The task was signaled. @@ -68,7 +71,7 @@ namespace MQTTnet.Internal // This is an O(n) operation since waiters is a LinkedList. lock (this.waiters) { - bool removed = this.waiters.Remove(tcs); + this.waiters.Remove(tcs); if (winner.Status == TaskStatus.Canceled) { throw new OperationCanceledException(cancellationToken); diff --git a/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs b/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs index d4029493..5e60a6a9 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs @@ -5,7 +5,7 @@ using System.Threading.Tasks; namespace MQTTnet.Internal { // From Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/12/building-async-coordination-primitives-part-6-asynclock/) - public sealed class AsyncLock + public sealed class AsyncLock : IDisposable { private readonly SemaphoreSlim m_semaphore = new SemaphoreSlim(1, 1); private readonly Task m_releaser; @@ -25,6 +25,11 @@ namespace MQTTnet.Internal TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } + public void Dispose() + { + this.m_semaphore?.Dispose(); + } + private sealed class Releaser : IDisposable { private readonly AsyncLock m_toRelease; diff --git a/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs b/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs index 879f9256..a4909b4a 100644 --- a/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs +++ b/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs @@ -10,7 +10,7 @@ namespace MQTTnet.Core.Tests // Inspired from the vs-threading tests (https://github.com/Microsoft/vs-threading/blob/master/src/Microsoft.VisualStudio.Threading.Tests/AsyncAutoResetEventTests.cs) public class AsyncAutoResetEventTests { - private AsyncAutoResetEvent evt; + private readonly AsyncAutoResetEvent evt; public AsyncAutoResetEventTests() { @@ -57,7 +57,7 @@ namespace MQTTnet.Core.Tests for (int i = 0; i < waiters.Length; i++) { this.evt.Set(); - await waiters[i]; + await waiters[i].ConfigureAwait(false); } } @@ -139,7 +139,7 @@ namespace MQTTnet.Core.Tests Assert.IsFalse(waitTask.IsCompleted); // Cancel the request and ensure that it propagates to the task. - await Task.Delay(1000); + await Task.Delay(1000).ConfigureAwait(false); try { waitTask.GetAwaiter().GetResult(); From ba66c108fa78dd019b3fc2777807a23e14d8573c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Warin?= Date: Tue, 15 May 2018 17:39:45 +0200 Subject: [PATCH 3/6] Fixes and clean the new topic matcher from israellot --- .../Server/MqttTopicFilterComparer.cs | 128 ++++++++++++++---- .../MQTTnet.Benchmarks.csproj | 1 + Tests/MQTTnet.Benchmarks/Program.cs | 4 + .../TopicFilterComparerBenchmark.cs | 91 +++++++++++++ 4 files changed, 195 insertions(+), 29 deletions(-) create mode 100644 Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs index 00e0bd65..c02be734 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttTopicFilterComparer.cs @@ -4,46 +4,116 @@ namespace MQTTnet.Server { public static class MqttTopicFilterComparer { - private static readonly char[] TopicLevelSeparator = { '/' }; + private const char LEVEL_SEPARATOR = '/'; + private const char WILDCARD_MULTI_LEVEL = '#'; + private const char WILDCARD_SINGLE_LEVEL = '+'; public static bool IsMatch(string topic, string filter) { - if (topic == null) throw new ArgumentNullException(nameof(topic)); - if (filter == null) throw new ArgumentNullException(nameof(filter)); + if (string.IsNullOrEmpty(topic)) throw new ArgumentNullException(nameof(topic)); + if (string.IsNullOrEmpty(filter)) throw new ArgumentNullException(nameof(filter)); - if (string.Equals(topic, filter, StringComparison.Ordinal)) + int spos = 0; + int slen = filter.Length; + int tpos = 0; + int tlen = topic.Length; + + while (spos < slen && tpos < tlen) { - return true; - } - - var fragmentsTopic = topic.Split(TopicLevelSeparator, StringSplitOptions.None); - var fragmentsFilter = filter.Split(TopicLevelSeparator, StringSplitOptions.None); - - // # > In either case it MUST be the last character specified in the Topic Filter [MQTT-4.7.1-2]. - for (var i = 0; i < fragmentsFilter.Length; i++) - { - if (fragmentsFilter[i] == "+") + if (filter[spos] == topic[tpos]) { - continue; + if (tpos == tlen - 1) + { + /* Check for e.g. foo matching foo/# */ + if (spos == slen - 3 + && filter[spos + 1] == LEVEL_SEPARATOR + && filter[spos + 2] == WILDCARD_MULTI_LEVEL) + { + return true; + } + } + spos++; + tpos++; + if (spos == slen && tpos == tlen) + { + return true; + } + else if (tpos == tlen && spos == slen - 1 && filter[spos] == WILDCARD_SINGLE_LEVEL) + { + if (spos > 0 && filter[spos - 1] != LEVEL_SEPARATOR) + { + // Invalid filter string + return false; + } + spos++; + return true; + } } - - if (fragmentsFilter[i] == "#") + else { - return true; - } - - if (i >= fragmentsTopic.Length) - { - return false; - } - - if (!string.Equals(fragmentsFilter[i], fragmentsTopic[i], StringComparison.Ordinal)) - { - return false; + if (filter[spos] == WILDCARD_SINGLE_LEVEL) + { + /* Check for bad "+foo" or "a/+foo" subscription */ + if (spos > 0 && filter[spos - 1] != LEVEL_SEPARATOR) + { + // Invalid filter string + return false; + } + /* Check for bad "foo+" or "foo+/a" subscription */ + if (spos < slen - 1 && filter[spos + 1] != LEVEL_SEPARATOR) + { + // Invalid filter string + return false; + } + spos++; + while (tpos < tlen && topic[tpos] != LEVEL_SEPARATOR) + { + tpos++; + } + if (tpos == tlen && spos == slen) + { + return true; + } + } + else if (filter[spos] == WILDCARD_MULTI_LEVEL) + { + if (spos > 0 && filter[spos - 1] != LEVEL_SEPARATOR) + { + // Invalid filter string + return false; + } + if (spos + 1 != slen) + { + // Invalid filter string + return false; + } + else + { + return true; + } + } + else + { + /* Check for e.g. foo/bar matching foo/+/# */ + if (spos > 0 + && spos + 2 == slen + && tpos == tlen + && filter[spos - 1] == WILDCARD_SINGLE_LEVEL + && filter[spos] == LEVEL_SEPARATOR + && filter[spos + 1] == WILDCARD_MULTI_LEVEL) + { + return true; + } + return false; + } } } + if (tpos < tlen || spos < slen) + { + return false; + } - return fragmentsTopic.Length == fragmentsFilter.Length; + return false; } } } diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index e068fa9d..8d3a7c89 100644 --- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -147,6 +147,7 @@ + diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs index 363cf8f2..a65de421 100644 --- a/Tests/MQTTnet.Benchmarks/Program.cs +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -12,6 +12,7 @@ namespace MQTTnet.Benchmarks Console.WriteLine("1 = MessageProcessingBenchmark"); Console.WriteLine("2 = SerializerBenchmark"); Console.WriteLine("3 = LoggerBenchmark"); + Console.WriteLine("4 = TopicFilterComparerBenchmark"); var pressedKey = Console.ReadKey(true); switch (pressedKey.KeyChar) @@ -25,6 +26,9 @@ namespace MQTTnet.Benchmarks case '3': BenchmarkRunner.Run(); break; + case '4': + BenchmarkRunner.Run(); + break; } Console.ReadLine(); diff --git a/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs b/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs new file mode 100644 index 00000000..6d9a0321 --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/TopicFilterComparerBenchmark.cs @@ -0,0 +1,91 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Attributes.Exporters; +using BenchmarkDotNet.Attributes.Jobs; +using MQTTnet.Server; +using System; + +namespace MQTTnet.Benchmarks +{ + [ClrJob] + [RPlotExporter] + [MemoryDiagnoser] + public class TopicFilterComparerBenchmark + { + private static readonly char[] TopicLevelSeparator = { '/' }; + + [GlobalSetup] + public void Setup() + { + } + + [Benchmark] + public void MqttTopicFilterComparer_10000_StringSplitMethod() + { + for (var i = 0; i < 10000; i++) + { + LegacyMethodByStringSplit("sport/tennis/player1", "sport/#"); + LegacyMethodByStringSplit("sport/tennis/player1/ranking", "sport/#/ranking"); + LegacyMethodByStringSplit("sport/tennis/player1/score/wimbledon", "sport/+/player1/#"); + LegacyMethodByStringSplit("sport/tennis/player1", "sport/tennis/+"); + LegacyMethodByStringSplit("/finance", "+/+"); + LegacyMethodByStringSplit("/finance", "/+"); + LegacyMethodByStringSplit("/finance", "+"); + } + } + + [Benchmark] + public void MqttTopicFilterComparer_10000_LoopMethod() + { + for (var i = 0; i < 10000; i++) + { + MqttTopicFilterComparer.IsMatch("sport/tennis/player1", "sport/#"); + MqttTopicFilterComparer.IsMatch("sport/tennis/player1/ranking", "sport/#/ranking"); + MqttTopicFilterComparer.IsMatch("sport/tennis/player1/score/wimbledon", "sport/+/player1/#"); + MqttTopicFilterComparer.IsMatch("sport/tennis/player1", "sport/tennis/+"); + MqttTopicFilterComparer.IsMatch("/finance", "+/+"); + MqttTopicFilterComparer.IsMatch("/finance", "/+"); + MqttTopicFilterComparer.IsMatch("/finance", "+"); + } + } + + private static bool LegacyMethodByStringSplit(string topic, string filter) + { + if (topic == null) throw new ArgumentNullException(nameof(topic)); + if (filter == null) throw new ArgumentNullException(nameof(filter)); + + if (string.Equals(topic, filter, StringComparison.Ordinal)) + { + return true; + } + + var fragmentsTopic = topic.Split(TopicLevelSeparator, StringSplitOptions.None); + var fragmentsFilter = filter.Split(TopicLevelSeparator, StringSplitOptions.None); + + // # > In either case it MUST be the last character specified in the Topic Filter [MQTT-4.7.1-2]. + for (var i = 0; i < fragmentsFilter.Length; i++) + { + if (fragmentsFilter[i] == "+") + { + continue; + } + + if (fragmentsFilter[i] == "#") + { + return true; + } + + if (i >= fragmentsTopic.Length) + { + return false; + } + + if (!string.Equals(fragmentsFilter[i], fragmentsTopic[i], StringComparison.Ordinal)) + { + return false; + } + } + + return fragmentsTopic.Length == fragmentsFilter.Length; + } + } +} From c1352d50c1c9a4d615de12b11010ea7552d2a345 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Warin?= Date: Wed, 16 May 2018 10:00:29 +0200 Subject: [PATCH 4/6] Review code style to match with the existing code --- .../Internal/AsyncAutoResetEvent.cs | 44 ++++++++-------- .../MQTTnet.NetStandard/Internal/AsyncLock.cs | 28 +++++++---- .../Server/MqttClientSessionsManager.cs | 13 +++-- .../Server/MqttRetainedMessagesManager.cs | 6 +-- .../AsyncAutoResentEventTests.cs | 50 +++++++++---------- 5 files changed, 72 insertions(+), 69 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs b/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs index a8fa8529..f20733d8 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/AsyncAutoResetEvent.cs @@ -8,31 +8,30 @@ namespace MQTTnet.Internal // Inspired from Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/11/building-async-coordination-primitives-part-2-asyncautoresetevent/) and Chris Gillum (https://stackoverflow.com/a/43012490) public class AsyncAutoResetEvent { - private readonly LinkedList> waiters = new LinkedList>(); - - private bool isSignaled; + private readonly LinkedList> _waiters = new LinkedList>(); + private bool _isSignaled; public AsyncAutoResetEvent() : this(false) { } public AsyncAutoResetEvent(bool signaled) { - this.isSignaled = signaled; + _isSignaled = signaled; } public Task WaitOneAsync() { - return this.WaitOneAsync(CancellationToken.None); + return WaitOneAsync(CancellationToken.None); } public Task WaitOneAsync(TimeSpan timeout) { - return this.WaitOneAsync(timeout, CancellationToken.None); + return WaitOneAsync(timeout, CancellationToken.None); } public Task WaitOneAsync(CancellationToken cancellationToken) { - return this.WaitOneAsync(Timeout.InfiniteTimeSpan, cancellationToken); + return WaitOneAsync(Timeout.InfiniteTimeSpan, cancellationToken); } public async Task WaitOneAsync(TimeSpan timeout, CancellationToken cancellationToken) @@ -41,21 +40,21 @@ namespace MQTTnet.Internal TaskCompletionSource tcs; - lock (this.waiters) + lock (_waiters) { - if (this.isSignaled) + if (_isSignaled) { - this.isSignaled = false; + _isSignaled = false; return true; } else if (timeout == TimeSpan.Zero) { - return this.isSignaled; + return _isSignaled; } else { tcs = new TaskCompletionSource(); - this.waiters.AddLast(tcs); + _waiters.AddLast(tcs); } } @@ -69,9 +68,9 @@ namespace MQTTnet.Internal { // We timed-out; remove our reference to the task. // This is an O(n) operation since waiters is a LinkedList. - lock (this.waiters) + lock (_waiters) { - this.waiters.Remove(tcs); + _waiters.Remove(tcs); if (winner.Status == TaskStatus.Canceled) { throw new OperationCanceledException(cancellationToken); @@ -88,25 +87,22 @@ namespace MQTTnet.Internal { TaskCompletionSource toRelease = null; - lock (this.waiters) + lock (_waiters) { - if (this.waiters.Count > 0) + if (_waiters.Count > 0) { // Signal the first task in the waiters list. - toRelease = this.waiters.First.Value; - this.waiters.RemoveFirst(); + toRelease = _waiters.First.Value; + _waiters.RemoveFirst(); } - else if (!this.isSignaled) + else if (!_isSignaled) { // No tasks are pending - this.isSignaled = true; + _isSignaled = true; } } - if (toRelease != null) - { - toRelease.SetResult(true); - } + toRelease?.SetResult(true); } } } diff --git a/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs b/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs index 5e60a6a9..1de1b1a9 100644 --- a/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs +++ b/Frameworks/MQTTnet.NetStandard/Internal/AsyncLock.cs @@ -7,34 +7,42 @@ namespace MQTTnet.Internal // From Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/12/building-async-coordination-primitives-part-6-asynclock/) public sealed class AsyncLock : IDisposable { - private readonly SemaphoreSlim m_semaphore = new SemaphoreSlim(1, 1); - private readonly Task m_releaser; + private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1); + private readonly Task _releaser; public AsyncLock() { - m_releaser = Task.FromResult((IDisposable)new Releaser(this)); + _releaser = Task.FromResult((IDisposable)new Releaser(this)); } public Task LockAsync(CancellationToken cancellationToken) { - var wait = m_semaphore.WaitAsync(cancellationToken); + Task wait = _semaphore.WaitAsync(cancellationToken); return wait.IsCompleted ? - m_releaser : + _releaser : wait.ContinueWith((_, state) => (IDisposable)state, - m_releaser.Result, cancellationToken, + _releaser.Result, cancellationToken, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } public void Dispose() { - this.m_semaphore?.Dispose(); + _semaphore?.Dispose(); } private sealed class Releaser : IDisposable { - private readonly AsyncLock m_toRelease; - internal Releaser(AsyncLock toRelease) { m_toRelease = toRelease; } - public void Dispose() { m_toRelease.m_semaphore.Release(); } + private readonly AsyncLock _toRelease; + + internal Releaser(AsyncLock toRelease) + { + _toRelease = toRelease; + } + + public void Dispose() + { + _toRelease._semaphore.Release(); + } } } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index e47f3f32..9281a971 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -119,7 +119,7 @@ namespace MQTTnet.Server public async Task StopAsync() { - using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { foreach (var session in _sessions) { @@ -132,7 +132,7 @@ namespace MQTTnet.Server public async Task> GetConnectedClientsAsync() { - using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { return _sessions.Where(s => s.Value.IsConnected).Select(s => new ConnectedMqttClient { @@ -155,7 +155,7 @@ namespace MQTTnet.Server if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { if (!_sessions.TryGetValue(clientId, out var session)) { @@ -171,7 +171,7 @@ namespace MQTTnet.Server if (clientId == null) throw new ArgumentNullException(nameof(clientId)); if (topicFilters == null) throw new ArgumentNullException(nameof(topicFilters)); - using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { if (!_sessions.TryGetValue(clientId, out var session)) { @@ -205,8 +205,7 @@ namespace MQTTnet.Server private async Task PrepareClientSessionAsync(MqttConnectPacket connectPacket) { - using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) - + using (await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { var isSessionPresent = _sessions.TryGetValue(connectPacket.ClientId, out var clientSession); if (isSessionPresent) @@ -269,7 +268,7 @@ namespace MQTTnet.Server _logger.Error(exception, "Error while processing application message"); } - using (var releaser = await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _sessionsLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { foreach (var clientSession in _sessions.Values) { diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs index e00ff444..89765b70 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs @@ -29,7 +29,7 @@ namespace MQTTnet.Server return; } - using (var releaser = await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { try { @@ -52,7 +52,7 @@ namespace MQTTnet.Server { if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); - using (var releaser = await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { try { @@ -69,7 +69,7 @@ namespace MQTTnet.Server { var retainedMessages = new List(); - using (var releaser = await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) + using (await _messagesLock.LockAsync(CancellationToken.None).ConfigureAwait(false)) { foreach (var retainedMessage in _messages.Values) { diff --git a/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs b/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs index a4909b4a..7087ff04 100644 --- a/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs +++ b/Tests/MQTTnet.Core.Tests/AsyncAutoResentEventTests.cs @@ -10,11 +10,11 @@ namespace MQTTnet.Core.Tests // Inspired from the vs-threading tests (https://github.com/Microsoft/vs-threading/blob/master/src/Microsoft.VisualStudio.Threading.Tests/AsyncAutoResetEventTests.cs) public class AsyncAutoResetEventTests { - private readonly AsyncAutoResetEvent evt; + private readonly AsyncAutoResetEvent _aare; public AsyncAutoResetEventTests() { - this.evt = new AsyncAutoResetEvent(); + _aare = new AsyncAutoResetEvent(); } [TestMethod] @@ -22,9 +22,9 @@ namespace MQTTnet.Core.Tests { for (int i = 0; i < 5; i++) { - var t = this.evt.WaitOneAsync(); + var t = _aare.WaitOneAsync(); Assert.IsFalse(t.IsCompleted); - this.evt.Set(); + _aare.Set(); await t; Assert.IsTrue(t.IsCompleted); } @@ -33,14 +33,14 @@ namespace MQTTnet.Core.Tests [TestMethod] public async Task MultipleSetOnlySignalsOnce() { - this.evt.Set(); - this.evt.Set(); - await this.evt.WaitOneAsync(); - var t = this.evt.WaitOneAsync(); + _aare.Set(); + _aare.Set(); + await _aare.WaitOneAsync(); + var t = _aare.WaitOneAsync(); Assert.IsFalse(t.IsCompleted); await Task.Delay(500); Assert.IsFalse(t.IsCompleted); - this.evt.Set(); + _aare.Set(); await t; Assert.IsTrue(t.IsCompleted); } @@ -51,12 +51,12 @@ namespace MQTTnet.Core.Tests var waiters = new Task[5]; for (int i = 0; i < waiters.Length; i++) { - waiters[i] = this.evt.WaitOneAsync(); + waiters[i] = _aare.WaitOneAsync(); } for (int i = 0; i < waiters.Length; i++) { - this.evt.Set(); + _aare.Set(); await waiters[i].ConfigureAwait(false); } } @@ -68,7 +68,7 @@ namespace MQTTnet.Core.Tests public async Task SetReturnsBeforeInlinedContinuations() { var setReturned = new ManualResetEventSlim(); - var inlinedContinuation = this.evt.WaitOneAsync() + var inlinedContinuation = _aare.WaitOneAsync() .ContinueWith(delegate { // Arrange to synchronously block the continuation until Set() has returned, @@ -76,7 +76,7 @@ namespace MQTTnet.Core.Tests Assert.IsTrue(setReturned.Wait(500)); }); await Task.Delay(100); - this.evt.Set(); + _aare.Set(); setReturned.Set(); Assert.IsTrue(inlinedContinuation.Wait(500)); } @@ -85,7 +85,7 @@ namespace MQTTnet.Core.Tests public void WaitAsync_WithCancellationToken() { var cts = new CancellationTokenSource(); - Task waitTask = this.evt.WaitOneAsync(cts.Token); + Task waitTask = _aare.WaitOneAsync(cts.Token); Assert.IsFalse(waitTask.IsCompleted); // Cancel the request and ensure that it propagates to the task. @@ -95,14 +95,14 @@ namespace MQTTnet.Core.Tests waitTask.GetAwaiter().GetResult(); Assert.IsTrue(false, "Task was expected to transition to a canceled state."); } - catch (System.OperationCanceledException ex) + catch (OperationCanceledException ex) { Assert.AreEqual(cts.Token, ex.CancellationToken); } // Now set the event and verify that a future waiter gets the signal immediately. - this.evt.Set(); - waitTask = this.evt.WaitOneAsync(); + _aare.Set(); + waitTask = _aare.WaitOneAsync(); Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status); } @@ -116,10 +116,10 @@ namespace MQTTnet.Core.Tests var token = tokenSource.Token; // Verify that a pre-set signal is not reset by a canceled wait request. - this.evt.Set(); + _aare.Set(); try { - this.evt.WaitOneAsync(token).GetAwaiter().GetResult(); + _aare.WaitOneAsync(token).GetAwaiter().GetResult(); Assert.IsTrue(false, "Task was expected to transition to a canceled state."); } catch (OperationCanceledException ex) @@ -128,14 +128,14 @@ namespace MQTTnet.Core.Tests } // Verify that the signal was not acquired. - Task waitTask = this.evt.WaitOneAsync(); + Task waitTask = _aare.WaitOneAsync(); Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status); } [TestMethod] public async Task WaitAsync_WithTimeout() { - Task waitTask = this.evt.WaitOneAsync(TimeSpan.FromMilliseconds(500)); + Task waitTask = _aare.WaitOneAsync(TimeSpan.FromMilliseconds(500)); Assert.IsFalse(waitTask.IsCompleted); // Cancel the request and ensure that it propagates to the task. @@ -145,14 +145,14 @@ namespace MQTTnet.Core.Tests waitTask.GetAwaiter().GetResult(); Assert.IsTrue(false, "Task was expected to transition to a timeout state."); } - catch (System.TimeoutException) + catch (TimeoutException) { Assert.IsTrue(true); } // Now set the event and verify that a future waiter gets the signal immediately. - this.evt.Set(); - waitTask = this.evt.WaitOneAsync(TimeSpan.FromMilliseconds(500)); + _aare.Set(); + waitTask = _aare.WaitOneAsync(TimeSpan.FromMilliseconds(500)); Assert.AreEqual(TaskStatus.RanToCompletion, waitTask.Status); } @@ -160,7 +160,7 @@ namespace MQTTnet.Core.Tests public void WaitAsync_Canceled_DoesNotInlineContinuations() { var cts = new CancellationTokenSource(); - var task = this.evt.WaitOneAsync(cts.Token); + var task = _aare.WaitOneAsync(cts.Token); var completingActionFinished = new ManualResetEventSlim(); var continuation = task.ContinueWith( From 932aa1fd39115dc1a605509d1b732ca33b2f839a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Warin?= Date: Wed, 16 May 2018 11:46:09 +0200 Subject: [PATCH 5/6] Fixes chkr1011's refactoing and pass all tests --- .../MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs | 6 ++++-- .../Server/MqttRetainedMessagesManager.cs | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index b9ab0136..62a10ee1 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -136,9 +136,9 @@ namespace MQTTnet.Server return Task.FromResult((IList)result); } - public void StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) + public Task StartDispatchApplicationMessage(MqttClientSession senderClientSession, MqttApplicationMessage applicationMessage) { - Task.Run(() => DispatchApplicationMessageAsync(senderClientSession, applicationMessage)); + return DispatchApplicationMessageAsync(senderClientSession, applicationMessage); } public Task SubscribeAsync(string clientId, IList topicFilters) @@ -237,6 +237,7 @@ namespace MQTTnet.Server { try { + _logger.Verbose($"DispatchApplicationMessageAsync"); var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); if (interceptorContext.CloseConnection) { @@ -257,6 +258,7 @@ namespace MQTTnet.Server foreach (var clientSession in _sessions.Values) { + _logger.Verbose($"DispatchApplicationMessageAsync for {clientSession.ClientId} "); clientSession.EnqueueApplicationMessage(applicationMessage); } } diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs index f2ab9448..3ab53cbf 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttRetainedMessagesManager.cs @@ -82,7 +82,7 @@ namespace MQTTnet.Server { var saveIsRequired = false; - if (applicationMessage.Payload?.Length > 0) + if (applicationMessage.Payload?.Length == 0) { saveIsRequired = _messages.TryRemove(applicationMessage.Topic, out _); _logger.Info("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic); From 5e0bc17341fdc9c3625c2ad31cedf0a0c12dd8b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Warin?= Date: Wed, 16 May 2018 20:47:27 +0200 Subject: [PATCH 6/6] Removing the debug log messages --- .../MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs index 62a10ee1..e798a513 100644 --- a/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs +++ b/Frameworks/MQTTnet.NetStandard/Server/MqttClientSessionsManager.cs @@ -237,7 +237,6 @@ namespace MQTTnet.Server { try { - _logger.Verbose($"DispatchApplicationMessageAsync"); var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); if (interceptorContext.CloseConnection) { @@ -258,7 +257,6 @@ namespace MQTTnet.Server foreach (var clientSession in _sessions.Values) { - _logger.Verbose($"DispatchApplicationMessageAsync for {clientSession.ClientId} "); clientSession.EnqueueApplicationMessage(applicationMessage); } }