Async backlog (#1612)

* alternative implementation along the lines of #1574:

- make the backlog processor async
- check for work *before* trying to get the lock

* release notes

Author: mgravell

docs/ReleaseNotes.md

@@ -1,6 +1,6 @@
 # Release Notes
 
-## 2.2.0
+## 2.2.2
 
 - add .NET 5 target
 - fix mutex race condition (#1585 via arsnyder16)
@@ -13,6 +13,7 @@
 - fix `Int64` parse scenario (#1568 via arsnyder16)
 - force replication check during failover (via joroda)
 - documentation tweaks (multiple)
+- fix backlog contention issue (#1612, see also #1574 via devbv)
 
 ## 2.1.58
 

src/StackExchange.Redis/PhysicalBridge.cs

@@ -1,14 +1,11 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
-using System.IO;
-using System.IO.Pipelines;
 using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
-using Pipelines.Sockets.Unofficial;
 using Pipelines.Sockets.Unofficial.Threading;
 using static Pipelines.Sockets.Unofficial.Threading.MutexSlim;
 using static StackExchange.Redis.ConnectionMultiplexer;
@@ -56,9 +53,7 @@ public PhysicalBridge(ServerEndPoint serverEndPoint, ConnectionType type, int ti
             Name = Format.ToString(serverEndPoint.EndPoint) + "/" + ConnectionType.ToString();
             TimeoutMilliseconds = timeoutMilliseconds;
             _singleWriterMutex = new MutexSlim(timeoutMilliseconds: timeoutMilliseconds);
-            _weakRefThis = new WeakReference(this);
         }
-        private readonly WeakReference _weakRefThis;
 
         private readonly int TimeoutMilliseconds;
 
@@ -771,22 +766,15 @@ private bool PushToBacklog(Message message, bool onlyIfExists)
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private void StartBacklogProcessor()
         {
-            var sched = Multiplexer.SocketManager?.Scheduler ?? PipeScheduler.ThreadPool;
 #if DEBUG
             _backlogProcessorRequestedTime = Environment.TickCount;
 #endif
-            sched.Schedule(s_ProcessBacklog, _weakRefThis);
+            Task.Run(ProcessBacklogAsync);
         }
 #if DEBUG
         private volatile int _backlogProcessorRequestedTime;
 #endif
 
-        private static readonly Action<object> s_ProcessBacklog = s =>
-        {
-            var wr = (WeakReference)s;
-            if (wr.Target is PhysicalBridge bridge) bridge.ProcessBacklog();
-        };
-
         private void CheckBacklogForTimeouts() // check the head of the backlog queue, consuming anything that looks dead
         {
             lock (_backlog)
@@ -810,6 +798,7 @@ private void CheckBacklogForTimeouts() // check the head of the backlog queue, c
         internal enum BacklogStatus : byte
         {
             Inactive,
+            Starting,
             Started,
             CheckingForWork,
             CheckingForTimeout,
@@ -823,7 +812,7 @@ internal enum BacklogStatus : byte
             Faulted,
         }
         private volatile BacklogStatus _backlogStatus;
-        private void ProcessBacklog()
+        private async Task ProcessBacklogAsync()
         {
             LockToken token = default;
             try
@@ -833,12 +822,19 @@ private void ProcessBacklog()
                 var msToStartWorker = unchecked(tryToAcquireTime - _backlogProcessorRequestedTime);
                 int failureCount = 0;
 #endif
-                while(true)
+                _backlogStatus = BacklogStatus.Starting;
+                while (true)
                 {
-                    // try and get the lock; if unsuccessful, check for termination
-                    token = _singleWriterMutex.TryWait();
-                    if (token) break; // got the lock
-                    lock (_backlog) { if (_backlog.Count == 0) return; }
+                    // check whether the backlog is empty *before* even trying to get the lock
+                    lock (_backlog)
+                    {
+                        if (_backlog.Count == 0) return; // nothing to do
+                    }
+
+                    // try and get the lock; if unsuccessful, retry
+                    token = await _singleWriterMutex.TryWaitAsync().ConfigureAwait(false);
+                    if (token.Success) break; // got the lock; now go do something with it
+
 #if DEBUG
                     failureCount++;
 #endif
@@ -887,9 +883,7 @@ private void ProcessBacklog()
                             if (result == WriteResult.Success)
                             {
                                 _backlogStatus = BacklogStatus.Flushing;
-#pragma warning disable CS0618
-                                result = physical.FlushSync(false, timeout);
-#pragma warning restore CS0618
+                                result = await physical.FlushAsync(false).ConfigureAwait(false);
                             }
 
                             _backlogStatus = BacklogStatus.MarkingInactive;