Fix #2664 Backlog: move to full synchronous behavior since we're on a dedicated thread (#2667)

Resolving #2664.

This went through a few iterations early on, but in the current form the issue is that we can await hop to a thread pool thread and have our wait blocking there - not awesome. We want to avoid that entirely and do a full sync path here.

Author: NickCraver

docs/ReleaseNotes.md

@@ -8,9 +8,10 @@ Current package versions:
 
 ## Unreleased
 
-- Add new `LoggingTunnel` API; see https://stackexchange.github.io/StackExchange.Redis/Logging ([#2660 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2660))
 - **Potentiallty Breaking**: Fix `CheckTrustedIssuer` certificate validation for broken chain scenarios ([#2665 by NickCraver](https://github.com/StackExchange/StackExchange.Redis/pull/2665))
   - Users inadvertently trusting a remote cert with a broken chain could not be failing custom validation before this change. This is only in play if you are using `ConfigurationOptions.TrustIssuer` at all.
+- Add new `LoggingTunnel` API; see https://stackexchange.github.io/StackExchange.Redis/Logging ([#2660 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2660))
+- Fix [#2664](https://github.com/StackExchange/StackExchange.Redis/issues/2664): Move ProcessBacklog to fully sync to prevent thread pool hopping and blocking on awaits ([#2667 by NickCraver](https://github.com/StackExchange/StackExchange.Redis/pull/2667))
 
 ## 2.7.27
 

src/StackExchange.Redis/PhysicalBridge.cs

@@ -904,7 +904,7 @@ private void StartBacklogProcessor()
                 // to unblock the thread-pool when there could be sync-over-async callers. Note that in reality,
                 // the initial "enough" of the back-log processor is typically sync, which means that the thread
                 // we start is actually useful, despite thinking "but that will just go async and back to the pool"
-                var thread = new Thread(s => ((PhysicalBridge)s!).ProcessBacklogAsync().RedisFireAndForget())
+                var thread = new Thread(s => ((PhysicalBridge)s!).ProcessBacklog())
                 {
                     IsBackground = true,                  // don't keep process alive (also: act like the thread-pool used to)
                     Name = "StackExchange.Redis Backlog", // help anyone looking at thread-dumps
@@ -985,7 +985,7 @@ internal enum BacklogStatus : byte
         /// Process the backlog(s) in play if any.
         /// This means flushing commands to an available/active connection (if any) or spinning until timeout if not.
         /// </summary>
-        private async Task ProcessBacklogAsync()
+        private void ProcessBacklog()
         {
             _backlogStatus = BacklogStatus.Starting;
             try
@@ -997,7 +997,7 @@ private async Task ProcessBacklogAsync()
                         // TODO: vNext handoff this backlog to another primary ("can handle everything") connection
                         // and remove any per-server commands. This means we need to track a bit of whether something
                         // was server-endpoint-specific in PrepareToPushMessageToBridge (was the server ref null or not)
-                        await ProcessBridgeBacklogAsync().ForAwait();
+                        ProcessBridgeBacklog();
                     }
 
                     // The cost of starting a new thread is high, and we can bounce in and out of the backlog a lot.
@@ -1070,7 +1070,7 @@ private async Task ProcessBacklogAsync()
         /// </summary>
         private readonly AutoResetEvent _backlogAutoReset = new AutoResetEvent(false);
 
-        private async Task ProcessBridgeBacklogAsync()
+        private void ProcessBridgeBacklog()
         {
             // Importantly: don't assume we have a physical connection here
             // We are very likely to hit a state where it's not re-established or even referenced here
@@ -1097,10 +1097,10 @@ private async Task ProcessBridgeBacklogAsync()
 
                     // try and get the lock; if unsuccessful, retry
 #if NETCOREAPP
-                    gotLock = await _singleWriterMutex.WaitAsync(TimeoutMilliseconds).ForAwait();
+                    gotLock = _singleWriterMutex.Wait(TimeoutMilliseconds);
                     if (gotLock) break; // got the lock; now go do something with it
 #else
-                    token = await _singleWriterMutex.TryWaitAsync().ForAwait();
+                    token = _singleWriterMutex.TryWait();
                     if (token.Success) break; // got the lock; now go do something with it
 #endif
                 }
@@ -1132,7 +1132,9 @@ private async Task ProcessBridgeBacklogAsync()
                         if (result == WriteResult.Success)
                         {
                             _backlogStatus = BacklogStatus.Flushing;
-                            result = await physical.FlushAsync(false).ForAwait();
+#pragma warning disable CS0618 // Type or member is obsolete
+                            result = physical.FlushSync(false, TimeoutMilliseconds);
+#pragma warning restore CS0618 // Type or member is obsolete
                         }
 
                         _backlogStatus = BacklogStatus.MarkingInactive;