Improved GetClientAsync so it uses async locks instead of thread locks.

Author: Pete Ness

src/ServiceStack.Redis/PooledRedisClientManager.Async.cs

@@ -25,7 +25,7 @@ ValueTask<ICacheClientAsync> IRedisClientsManagerAsync.GetCacheClientAsync(Cance
             => new RedisClientManagerCacheClient(this).AsValueTaskResult<ICacheClientAsync>();
 
         ValueTask<IRedisClientAsync> IRedisClientsManagerAsync.GetClientAsync(CancellationToken token)
-            => GetClient(true).AsValueTaskResult<IRedisClientAsync>();
+            => GetClientAsync();
 
         ValueTask<ICacheClientAsync> IRedisClientsManagerAsync.GetReadOnlyCacheClientAsync(CancellationToken token)
             => new RedisClientManagerCacheClient(this) { ReadOnly = true }.AsValueTaskResult<ICacheClientAsync>();

src/ServiceStack.Redis/PooledRedisClientManager.cs

@@ -15,6 +15,8 @@
 using System.IO;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
+using Cleary.AsyncExtensions;
 using ServiceStack.Caching;
 using ServiceStack.Logging;
 using ServiceStack.Text;
@@ -34,6 +36,9 @@ public partial class PooledRedisClientManager
         private const string PoolTimeoutError =
             "Redis Timeout expired. The timeout period elapsed prior to obtaining a connection from the pool. This may have occurred because all pooled connections were in use.";
 
+        private AsyncMonitor readMonitor;
+        private AsyncMonitor writeMonitor;
+
         protected readonly int PoolSizeMultiplier = 20;
         public int RecheckPoolAfterMs = 100;
         public int? PoolTimeout { get; set; }
@@ -42,7 +47,7 @@ public partial class PooledRedisClientManager
         public int? SocketReceiveTimeout { get; set; }
         public int? IdleTimeOutSecs { get; set; }
         public bool AssertAccessOnlyOnSameThread { get; set; }
-
+        
         /// <summary>
         /// Gets or sets object key prefix.
         /// </summary>
@@ -211,15 +216,128 @@ protected virtual void OnStart()
         {
             this.Start();
         }
+        
+        private async Task<bool> waitForWriter(int msTimeout)
+        {
+           writeMonitor = new AsyncMonitor();
+           var cts = new CancellationTokenSource();
+           cts.CancelAfter(msTimeout);
+           await writeMonitor.WaitAsync(cts.Token);
+           if (cts.IsCancellationRequested) return false;
+           cts.Cancel();
+           return true;
+        }
 
         /// <summary>
         /// Returns a Read/Write client (The default) using the hosts defined in ReadWriteHosts
         /// </summary>
         /// <returns></returns>
         public IRedisClient GetClient() => GetClient(false);
 
+        private async ValueTask<IRedisClientAsync> GetClientAsync()
+        {
+            try
+            {
+                var poolTimedOut = false;
+                var inactivePoolIndex = -1;
+
+                do
+                {
+                   RedisClient inActiveClient;
+                   lock (writeClients)
+                   {
+                      AssertValidReadWritePool();
+
+                      // If it's -1, then we want to try again after a delay of some kind. So if it's NOT negative one, process it...
+                      if ((inactivePoolIndex = GetInActiveWriteClient(out inActiveClient)) != -1)
+                      {
+                         //inActiveClient != null only for Valid InActive Clients
+                         if (inActiveClient != null)
+                         {
+                            WritePoolIndex++;
+                            inActiveClient.Activate();
+
+                            InitClient(inActiveClient);
+
+                            return inActiveClient;
+                         }
+                         else
+                         {
+                            // Still need to be in lock for this!
+                            break;
+                         }
+                      }
+                   }
+
+                   // Didn't get one, so let's wait a bit for a new one.
+                   if (PoolTimeout.HasValue)
+                   {
+                      if (!await waitForWriter(PoolTimeout.Value))
+                      {
+                         poolTimedOut = true;
+                         break;
+                      }
+                   }
+                   else
+                   {
+                      await waitForWriter(RecheckPoolAfterMs);
+                   }
+                } while (true); // Just keep repeating until we get a
+
+                if (poolTimedOut)
+                    throw new TimeoutException(PoolTimeoutError);
+
+                //Reaches here when there's no Valid InActive Clients
+                try
+                {
+                    //inactivePoolIndex = index of reservedSlot || index of invalid client
+                    var existingClient = writeClients[inactivePoolIndex];
+                    if (existingClient != null && existingClient != reservedSlot && existingClient.HadExceptions)
+                    {
+                        RedisState.DeactivateClient(existingClient);
+                    }
+
+                    var newClient = InitNewClient(RedisResolver.CreateMasterClient(inactivePoolIndex));
+
+                    //Put all blocking I/O or potential Exceptions before lock
+                    lock (writeClients)
+                    {
+                        //If existingClient at inactivePoolIndex changed (failover) return new client outside of pool
+                        if (writeClients[inactivePoolIndex] != existingClient)
+                        {
+                            if (Log.IsDebugEnabled)
+                                Log.Debug("writeClients[inactivePoolIndex] != existingClient: {0}".Fmt(writeClients[inactivePoolIndex]));
+
+                            return newClient; //return client outside of pool
+                        }
+
+                        WritePoolIndex++;
+                        writeClients[inactivePoolIndex] = newClient;
+
+                        return (!AssertAccessOnlyOnSameThread)
+                            ? newClient
+                            : newClient.LimitAccessToThread(Thread.CurrentThread.ManagedThreadId, Environment.StackTrace);
+                    }
+                }
+                catch
+                {
+                    //Revert free-slot for any I/O exceptions that can throw (before lock)
+                    lock (writeClients)
+                    {
+                        writeClients[inactivePoolIndex] = null; //free slot
+                    }
+                    throw;
+                }
+            }
+            finally
+            {
+                RedisState.DisposeExpiredClients();
+            }
+        }
+
         private RedisClient GetClient(bool forAsync)
         {
+           if (forAsync) throw new Exception("Call GetClientAsync instead");
             try
             {
                 var poolTimedOut = false;
@@ -547,6 +665,7 @@ public void DisposeClient(RedisNativeClient client)
                     }
 
                     Monitor.PulseAll(readClients);
+                    readMonitor?.PulseAll();
                     return;
                 }
             }
@@ -568,15 +687,23 @@ public void DisposeClient(RedisNativeClient client)
                     }
 
                     Monitor.PulseAll(writeClients);
+                    writeMonitor?.PulseAll();
                     return;
                 }
             }
 
             //Client not found in any pool, pulse both pools.
             lock (readClients)
-                Monitor.PulseAll(readClients);
+            {
+               Monitor.PulseAll(readClients);
+               readMonitor?.PulseAll();
+            }
+
             lock (writeClients)
-                Monitor.PulseAll(writeClients);
+            {
+               Monitor.PulseAll(writeClients);
+               writeMonitor?.PulseAll();
+            }
         }
 
         /// <summary>
@@ -589,6 +716,7 @@ public void DisposeReadOnlyClient(RedisNativeClient client)
             {
                 client.Deactivate();
                 Monitor.PulseAll(readClients);
+                readMonitor?.PulseAll();
             }
         }
 
@@ -602,6 +730,7 @@ public void DisposeWriteClient(RedisNativeClient client)
             {
                 client.Deactivate();
                 Monitor.PulseAll(writeClients);
+                writeMonitor?.PulseAll();
             }
         }
 

src/ServiceStack.Redis/ServiceStack.Redis.Core.csproj

@@ -11,6 +11,7 @@
     <PackageTags>Redis;NoSQL;Client;Distributed;Cache;PubSub;Messaging;Transactions</PackageTags>
   </PropertyGroup>
   <ItemGroup>
+    <PackageReference Include="Cleary.AsyncExtensions" Version="4.9.3" />
     <PackageReference Include="ServiceStack.Common.Core" Version="$(Version)" />
   </ItemGroup>
   <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.0' ">