Cancel and signal to ensure the async processor ends

libs/server/Resp/AsyncProcessor.cs

@@ -29,6 +29,11 @@ internal sealed partial class RespServerSession : ServerSessionBase
         /// </summary>
         SingleWaiterAutoResetEvent asyncWaiter = null;
 
+        /// <summary>
+        /// Cancellation token source for async waiter
+        /// </summary>
+        CancellationTokenSource asyncWaiterCancel = null;
+
         /// <summary>
         /// Semaphore for barrier command to wait for async operations to complete
         /// </summary>
@@ -51,6 +56,7 @@ void NetworkGETPending<TGarnetApi>(ref TGarnetApi storageApi)
 
             if (++asyncStarted == 1) // first async operation on the session, create the IO continuation processor
             {
+                asyncWaiterCancel = new();
                 asyncWaiter = new()
                 {
                     RunContinuationsAsynchronously = true
@@ -131,6 +137,9 @@ async Task AsyncGetProcessor<TGarnetApi>(TGarnetApi storageApi)
                 // Wait for next async operation
                 // We do not need to cancel the wait - it should get garbage collected when the session ends
                 await asyncWaiter.WaitAsync();
+
+                if (asyncWaiterCancel.Token.IsCancellationRequested)
+                    break;
             }
         }
     }

libs/server/Resp/RespServerSession.cs

@@ -173,6 +173,10 @@ public override void Dispose()
 
             subscribeBroker?.RemoveSubscription(this);
 
+            // Cancel the async processor, if any
+            asyncWaiterCancel?.Cancel();
+            asyncWaiter?.Signal();
+
             storageSession.Dispose();
         }