1565 high cpu load from asyncsignal (#1566)

Author: chkr1011

Source/MQTTnet.Tests/Internal/AsyncSignal_Tests.cs

@@ -52,12 +52,26 @@ public async Task Dispose_Properly()
  }
 
  [TestMethod]
- public async Task Loop_Signal()
+ public async Task Reset_Signal()
  {
  var asyncSignal = new AsyncSignal();
 
+ // WaitAsync should fail because no signal is available.
  for (var i = 0; i < 10; i++)
  {
+ try
+ {
+ using (var timeout = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)))
+ {
+ await asyncSignal.WaitAsync(timeout.Token);
+ }
+ 
+ Assert.Fail("This must fail because the signal is not yet set.");
+ }
+ catch (OperationCanceledException)
+ {
+ }
+ 
  asyncSignal.Set();
 
  // WaitAsync should return directly because the signal is available.

Source/MQTTnet.Tests/Server/Load_Tests.cs

@@ -55,7 +55,7 @@ await client.PublishAsync(message)
  });
  }
 
- SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(90));
+ SpinWait.SpinUntil(() => receivedMessages == 100000, TimeSpan.FromSeconds(120));
 
  Assert.AreEqual(100000, receivedMessages);
  }

Source/MQTTnet.Tests/Server/Session_Tests.cs

@@ -150,7 +150,7 @@ void OnReceive()
  // Try to connect 50 clients at the same time.
  var clients = await Task.WhenAll(Enumerable.Range(0, 50).Select(i => ConnectAndSubscribe(testEnvironment, options, OnReceive)));
 
- await Task.Delay(5000);
+ await Task.Delay(TimeSpan.FromSeconds(10));
 
  var connectedClients = clients.Where(c => c?.TryPingAsync().GetAwaiter().GetResult() == true).ToList();
 
@@ -162,7 +162,7 @@ void OnReceive()
  var sendClient = await testEnvironment.ConnectClient(option2);
  await sendClient.PublishStringAsync("aaa", "1");
 
- await Task.Delay(3000);
+ await Task.Delay(TimeSpan.FromSeconds(5));
 
  Assert.AreEqual(true, hasReceive);
  }

Source/MQTTnet/Internal/AsyncQueue.cs

@@ -29,7 +29,7 @@ public void Dispose()
  {
  lock (_syncRoot)
  {
- _signal?.Dispose();
+ _signal.Dispose();
 
  _isDisposed = true;
  }
@@ -48,10 +48,10 @@ public AsyncQueueDequeueResult<TItem> TryDequeue()
  {
  if (_queue.TryDequeue(out var item))
  {
- return new AsyncQueueDequeueResult<TItem>(true, item);
+ return AsyncQueueDequeueResult<TItem>.Success(item);
  }
 
- return new AsyncQueueDequeueResult<TItem>(false, default);
+ return AsyncQueueDequeueResult<TItem>.NonSuccess;
  }
 
  public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationToken cancellationToken)
@@ -60,36 +60,42 @@ public async Task<AsyncQueueDequeueResult<TItem>> TryDequeueAsync(CancellationTo
  {
  try
  {
- Task task;
+ Task task = null;
  lock (_syncRoot)
  {
  if (_isDisposed)
  {
- return new AsyncQueueDequeueResult<TItem>(false, default);
+ return AsyncQueueDequeueResult<TItem>.NonSuccess;
  }
 
- task = _signal.WaitAsync(cancellationToken);
+ if (_queue.IsEmpty)
+ {
+ task = _signal.WaitAsync(cancellationToken);
+ }
  }
 
- await task.ConfigureAwait(false);
-
+ if (task != null)
+ {
+ await task.ConfigureAwait(false); 
+ }
+ 
  if (cancellationToken.IsCancellationRequested)
  {
- return new AsyncQueueDequeueResult<TItem>(false, default);
+ return AsyncQueueDequeueResult<TItem>.NonSuccess;
  }
 
  if (_queue.TryDequeue(out var item))
  {
- return new AsyncQueueDequeueResult<TItem>(true, item);
+ return AsyncQueueDequeueResult<TItem>.Success(item);
  }
  }
  catch (OperationCanceledException)
  {
- return new AsyncQueueDequeueResult<TItem>(false, default);
+ return AsyncQueueDequeueResult<TItem>.NonSuccess;
  }
  }
 
- return new AsyncQueueDequeueResult<TItem>(false, default);
+ return AsyncQueueDequeueResult<TItem>.NonSuccess;
  }
  }
 }

Source/MQTTnet/Internal/AsyncQueueDequeueResult.cs

@@ -4,8 +4,10 @@
 
 namespace MQTTnet.Internal
 {
- public class AsyncQueueDequeueResult<TItem>
+ public sealed class AsyncQueueDequeueResult<TItem>
  {
+ public static readonly AsyncQueueDequeueResult<TItem> NonSuccess = new AsyncQueueDequeueResult<TItem>(false, default);
+
  public AsyncQueueDequeueResult(bool isSuccess, TItem item)
  {
  IsSuccess = isSuccess;
@@ -15,5 +17,10 @@ public AsyncQueueDequeueResult(bool isSuccess, TItem item)
  public bool IsSuccess { get; }
 
  public TItem Item { get; }
+
+ public static AsyncQueueDequeueResult<TItem> Success(TItem item)
+ {
+ return new AsyncQueueDequeueResult<TItem>(true, item);
+ }
  }
 }

Source/MQTTnet/Internal/AsyncSignal.cs

@@ -33,9 +33,17 @@ public void Set()
  {
  _isSignaled = true;
 
+ Cleanup();
+ 
  // If there is already a waiting task let it run.
- _waiter?.Approve();
- _waiter = null;
+ if (_waiter != null)
+ {
+ _waiter.Approve();
+ _waiter = null;
+
+ // Since we already got a waiter the signal must be reset right now!
+ _isSignaled = false;
+ }
  }
  }
 
@@ -47,8 +55,11 @@ public Task WaitAsync(CancellationToken cancellationToken = default)
  {
  ThrowIfDisposed();
 
+ Cleanup();
+
  if (_isSignaled)
  {
+ _isSignaled = false;
  return CompletedTask.Instance;
  }
 
@@ -58,7 +69,7 @@ public Task WaitAsync(CancellationToken cancellationToken = default)
  {
  throw new InvalidOperationException("Only one waiting task is permitted per async signal.");
  }
- 
+
  _waiter.Dispose();
  }
 
@@ -67,6 +78,15 @@ public Task WaitAsync(CancellationToken cancellationToken = default)
  }
  }
 
+ void Cleanup()
+ {
+ // Cleanup if the previous waiter was cancelled.
+ if (_waiter != null && _waiter.Task.IsCanceled)
+ {
+ _waiter = null;
+ }
+ }
+
  void ThrowIfDisposed()
  {
  if (_isDisposed)