Drain any buffers before returning pooled client

mythz · mythz · commit 8d382270e21c · 2020-11-16T17:29:29.000+08:00
diff --git a/src/ServiceStack.Redis/BufferedReader.cs b/src/ServiceStack.Redis/BufferedReader.cs
@@ -28,6 +28,11 @@ internal BufferedReader(Stream source, int bufferSize)
         {
             _source = source;
             _buffer = new byte[bufferSize];
+            Reset();
+        }
+
+        internal void Reset()
+        {
             _offset = _available = 0;
         }
 
diff --git a/src/ServiceStack.Redis/PooledRedisClientManager.cs b/src/ServiceStack.Redis/PooledRedisClientManager.cs
@@ -250,7 +250,7 @@ private RedisClient GetClient(bool forAsync)
                     if (inActiveClient != null)
                     {
                         WritePoolIndex++;
-                        inActiveClient.Active = true;
+                        inActiveClient.Activate();
 
                         InitClient(inActiveClient);
 
@@ -400,7 +400,7 @@ private RedisClient GetReadOnlyClient(bool forAsync)
                     if (inActiveClient != null)
                     {
                         ReadPoolIndex++;
-                        inActiveClient.Active = true;
+                        inActiveClient.Activate();
 
                         InitClient(inActiveClient);
 
@@ -468,7 +468,7 @@ private int GetInActiveReadClient(out RedisClient inactiveClient)
         {
             var desiredIndex = ReadPoolIndex % readClients.Length;
             //this will loop through all hosts in readClients once even though there are 2 for loops
-            //both loops are used to try to get the prefered host according to the round robin algorithm
+            //both loops are used to try to get the preferred host according to the round robin algorithm
             var readOnlyTotal = RedisResolver.ReadOnlyHostsCount;
             for (int x = 0; x < readOnlyTotal; x++)
             {
@@ -502,7 +502,7 @@ private int GetInActiveReadClient(out RedisClient inactiveClient)
         private RedisClient InitNewClient(RedisClient client)
         {
             client.Id = Interlocked.Increment(ref RedisClientCounter);
-            client.Active = true;
+            client.Activate(newClient:true);
             client.ClientManager = this;
             client.ConnectionFilter = ConnectionFilter;
             if (NamespacePrefix != null)
@@ -543,7 +543,7 @@ public void DisposeClient(RedisNativeClient client)
                     else
                     {
                         client.TrackThread = null;
-                        client.Active = false;
+                        client.Deactivate();
                     }
 
                     Monitor.PulseAll(readClients);
@@ -564,7 +564,7 @@ public void DisposeClient(RedisNativeClient client)
                     else
                     {
                         client.TrackThread = null;
-                        client.Active = false;
+                        client.Deactivate();
                     }
 
                     Monitor.PulseAll(writeClients);
@@ -587,7 +587,7 @@ public void DisposeReadOnlyClient(RedisNativeClient client)
         {
             lock (readClients)
             {
-                client.Active = false;
+                client.Deactivate();
                 Monitor.PulseAll(readClients);
             }
         }
@@ -600,7 +600,7 @@ public void DisposeWriteClient(RedisNativeClient client)
         {
             lock (writeClients)
             {
-                client.Active = false;
+                client.Deactivate();
                 Monitor.PulseAll(writeClients);
             }
         }
diff --git a/src/ServiceStack.Redis/RedisClient.Async.cs b/src/ServiceStack.Redis/RedisClient.Async.cs
@@ -105,7 +105,8 @@ private async ValueTask<T> ExecAsync<T>(Func<IRedisClientAsync, ValueTask<T>> ac
         {
             using (JsConfig.With(new Text.Config { ExcludeTypeInfo = false }))
             {
-                return await action(this).ConfigureAwait(false);
+                var ret = await action(this).ConfigureAwait(false);
+                return ret;
             }
         }
 
@@ -120,11 +121,19 @@ ValueTask<string> IRedisClientAsync.GetValueAsync(string key, CancellationToken
 
         Task<T> ICacheClientAsync.GetAsync<T>(string key, CancellationToken token)
         {
-            return ExecAsync(r =>
-                typeof(T) == typeof(byte[])
-                    ? ((IRedisNativeClientAsync)r).GetAsync(key, token).Await(val => (T)(object)val)
-                    : r.GetValueAsync(key, token).Await(val => JsonSerializer.DeserializeFromString<T>(val))
-            ).AsTask();
+            return ExecAsync(async r => {
+                if (typeof(T) == typeof(byte[]))
+                {
+                    var ret = await ((IRedisNativeClientAsync) r).GetAsync(key, token).ConfigureAwait(false);
+                    return (T) (object) ret;
+                }
+                else
+                {
+                    var val = await r.GetValueAsync(key, token).ConfigureAwait(false);
+                    var ret = JsonSerializer.DeserializeFromString<T>(val);
+                    return ret;
+                }
+            }).AsTask();
         }
 
         async ValueTask<List<string>> IRedisClientAsync.SearchKeysAsync(string pattern, CancellationToken token)
@@ -169,10 +178,10 @@ ValueTask<bool> IRedisClientAsync.AddItemToSortedSetAsync(string setId, string v
             => ((IRedisClientAsync)this).AddItemToSortedSetAsync(setId, value, GetLexicalScore(value), token);
 
         ValueTask<bool> IRedisClientAsync.AddItemToSortedSetAsync(string setId, string value, double score, CancellationToken token)
-            => NativeAsync.ZAddAsync(setId, score, value.ToUtf8Bytes()).IsSuccessAsync();
+            => NativeAsync.ZAddAsync(setId, score, value.ToUtf8Bytes(), token).IsSuccessAsync();
 
         ValueTask<bool> IRedisClientAsync.SetEntryInHashAsync(string hashId, string key, string value, CancellationToken token)
-            => NativeAsync.HSetAsync(hashId, key.ToUtf8Bytes(), value.ToUtf8Bytes()).IsSuccessAsync();
+            => NativeAsync.HSetAsync(hashId, key.ToUtf8Bytes(), value.ToUtf8Bytes(), token).IsSuccessAsync();
 
         ValueTask IRedisClientAsync.SetAllAsync(IDictionary<string, string> map, CancellationToken token)
             => GetSetAllBytes(map, out var keyBytes, out var valBytes) ? NativeAsync.MSetAsync(keyBytes, valBytes, token) : default;
@@ -221,7 +230,7 @@ ValueTask<bool> IRedisClientAsync.ExpireEntryAtAsync(string key, DateTime expire
             : NativeAsync.ExpireAtAsync(key, ConvertToServerDate(expireAt).ToUnixTime(), token);
 
         Task<TimeSpan?> ICacheClientAsync.GetTimeToLiveAsync(string key, CancellationToken token)
-            => NativeAsync.TtlAsync(key, token).Await(ttlSecs => ParseTimeToLiveResult(ttlSecs)).AsTask();
+            => NativeAsync.TtlAsync(key, token).Await(ParseTimeToLiveResult).AsTask();
 
         ValueTask<bool> IRedisClientAsync.PingAsync(CancellationToken token)
             => NativeAsync.PingAsync(token);
@@ -365,7 +374,7 @@ ValueTask IRedisClientAsync.SlowlogResetAsync(CancellationToken token)
             => NativeAsync.SlowlogResetAsync(token);
 
         ValueTask<SlowlogItem[]> IRedisClientAsync.GetSlowlogAsync(int? numberOfRecords, CancellationToken token)
-            => NativeAsync.SlowlogGetAsync(numberOfRecords, token).Await(data => ParseSlowlog(data));
+            => NativeAsync.SlowlogGetAsync(numberOfRecords, token).Await(ParseSlowlog);
 
 
         Task<bool> ICacheClientAsync.SetAsync<T>(string key, T value, CancellationToken token)
diff --git a/src/ServiceStack.Redis/RedisManagerPool.cs b/src/ServiceStack.Redis/RedisManagerPool.cs
@@ -151,7 +151,7 @@ private RedisClient GetClient(bool forAsync)
                     if (inActiveClient != null)
                     {
                         poolIndex++;
-                        inActiveClient.Active = true;
+                        inActiveClient.Activate();
 
                         return !AssertAccessOnlyOnSameThread 
                             ? inActiveClient
@@ -281,7 +281,7 @@ private int GetInActiveClient(out RedisClient inactiveClient)
         private RedisClient InitNewClient(RedisClient client)
         {
             client.Id = Interlocked.Increment(ref RedisClientCounter);
-            client.Active = true;
+            client.Activate(newClient:true);
             client.ClientManager = this;
             client.ConnectionFilter = ConnectionFilter;
 
@@ -303,7 +303,7 @@ public void DisposeClient(RedisNativeClient client)
                     else
                     {
                         client.TrackThread = null;
-                        client.Active = false;
+                        client.Deactivate();
                     }
 
                     Monitor.PulseAll(clients);
@@ -320,7 +320,7 @@ public void DisposeWriteClient(RedisNativeClient client)
         {
             lock (clients)
             {
-                client.Active = false;
+                client.Deactivate();
             }
         }
 
diff --git a/src/ServiceStack.Redis/RedisNativeClient.cs b/src/ServiceStack.Redis/RedisNativeClient.cs
@@ -83,7 +83,7 @@ public DateTime? DeactivatedAt
         internal bool Active
         {
             get => Interlocked.CompareExchange(ref active, 0, 0) == YES;
-            set => Interlocked.Exchange(ref active, value ? YES : NO);
+            private set => Interlocked.Exchange(ref active, value ? YES : NO);
         }
 
         internal IHandleClientDispose ClientManager { get; set; }
diff --git a/src/ServiceStack.Redis/RedisNativeClient_Utils.cs b/src/ServiceStack.Redis/RedisNativeClient_Utils.cs
@@ -557,6 +557,31 @@ private void SendDirectToSocket(ArraySegment<byte> segment)
             }
         }
 
+        /// <summary>
+        /// Called before returning pooled client/socket  
+        /// </summary>
+        internal void Activate(bool newClient=false)
+        {
+            if (!newClient)
+            {
+                //Drain any existing buffers 
+                ResetSendBuffer();
+                bufferedReader.Reset();
+                if (socket?.Available > 0)
+                {
+                    logDebug($"Draining existing socket of {socket.Available} bytes");
+                    var buff = new byte[socket.Available];
+                    socket.Receive(buff, SocketFlags.None);
+                }
+            }
+            Active = true;
+        }
+
+        internal void Deactivate()
+        {
+            Active = false;
+        }
+
         /// <summary>
         /// reset buffer index in send buffer
         /// </summary>

Original file line number	Diff line number	Diff line change
`@@ -28,6 +28,11 @@ internal BufferedReader(Stream source, int bufferSize)`
`28`	`28`	`{`
`29`	`29`	`_source = source;`
`30`	`30`	`_buffer = new byte[bufferSize];`
	`31`	`+ Reset();`
	`32`	`+ }`
	`33`	`+`
	`34`	`+ internal void Reset()`
	`35`	`+ {`
`31`	`36`	`_offset = _available = 0;`
`32`	`37`	`}`
`33`	`38`
Original file line number	Diff line number	Diff line change
`@@ -250,7 +250,7 @@ private RedisClient GetClient(bool forAsync)`
`250`	`250`	`if (inActiveClient != null)`
`251`	`251`	`{`
`252`	`252`	`WritePoolIndex++;`
`253`		`- inActiveClient.Active = true;`
	`253`	`+ inActiveClient.Activate();`
`254`	`254`
`255`	`255`	`InitClient(inActiveClient);`
`256`	`256`
`@@ -400,7 +400,7 @@ private RedisClient GetReadOnlyClient(bool forAsync)`
`400`	`400`	`if (inActiveClient != null)`
`401`	`401`	`{`
`402`	`402`	`ReadPoolIndex++;`
`403`		`- inActiveClient.Active = true;`
	`403`	`+ inActiveClient.Activate();`
`404`	`404`
`405`	`405`	`InitClient(inActiveClient);`
`406`	`406`
`@@ -468,7 +468,7 @@ private int GetInActiveReadClient(out RedisClient inactiveClient)`
`468`	`468`	`{`
`469`	`469`	`var desiredIndex = ReadPoolIndex % readClients.Length;`
`470`	`470`	`//this will loop through all hosts in readClients once even though there are 2 for loops`
`471`		`- //both loops are used to try to get the prefered host according to the round robin algorithm`
	`471`	`+ //both loops are used to try to get the preferred host according to the round robin algorithm`
`472`	`472`	`var readOnlyTotal = RedisResolver.ReadOnlyHostsCount;`
`473`	`473`	`for (int x = 0; x < readOnlyTotal; x++)`
`474`	`474`	`{`
`@@ -502,7 +502,7 @@ private int GetInActiveReadClient(out RedisClient inactiveClient)`
`502`	`502`	`private RedisClient InitNewClient(RedisClient client)`
`503`	`503`	`{`
`504`	`504`	`client.Id = Interlocked.Increment(ref RedisClientCounter);`
`505`		`- client.Active = true;`
	`505`	`+ client.Activate(newClient:true);`
`506`	`506`	`client.ClientManager = this;`
`507`	`507`	`client.ConnectionFilter = ConnectionFilter;`
`508`	`508`	`if (NamespacePrefix != null)`
`@@ -543,7 +543,7 @@ public void DisposeClient(RedisNativeClient client)`
`543`	`543`	`else`
`544`	`544`	`{`
`545`	`545`	`client.TrackThread = null;`
`546`		`- client.Active = false;`
	`546`	`+ client.Deactivate();`
`547`	`547`	`}`
`548`	`548`
`549`	`549`	`Monitor.PulseAll(readClients);`
`@@ -564,7 +564,7 @@ public void DisposeClient(RedisNativeClient client)`
`564`	`564`	`else`
`565`	`565`	`{`
`566`	`566`	`client.TrackThread = null;`
`567`		`- client.Active = false;`
	`567`	`+ client.Deactivate();`
`568`	`568`	`}`
`569`	`569`
`570`	`570`	`Monitor.PulseAll(writeClients);`
`@@ -587,7 +587,7 @@ public void DisposeReadOnlyClient(RedisNativeClient client)`
`587`	`587`	`{`
`588`	`588`	`lock (readClients)`
`589`	`589`	`{`
`590`		`- client.Active = false;`
	`590`	`+ client.Deactivate();`
`591`	`591`	`Monitor.PulseAll(readClients);`
`592`	`592`	`}`
`593`	`593`	`}`
`@@ -600,7 +600,7 @@ public void DisposeWriteClient(RedisNativeClient client)`
`600`	`600`	`{`
`601`	`601`	`lock (writeClients)`
`602`	`602`	`{`
`603`		`- client.Active = false;`
	`603`	`+ client.Deactivate();`
`604`	`604`	`Monitor.PulseAll(writeClients);`
`605`	`605`	`}`
`606`	`606`	`}`
Original file line number	Diff line number	Diff line change
`@@ -151,7 +151,7 @@ private RedisClient GetClient(bool forAsync)`
`151`	`151`	`if (inActiveClient != null)`
`152`	`152`	`{`
`153`	`153`	`poolIndex++;`
`154`		`- inActiveClient.Active = true;`
	`154`	`+ inActiveClient.Activate();`
`155`	`155`
`156`	`156`	`return !AssertAccessOnlyOnSameThread`
`157`	`157`	`? inActiveClient`
`@@ -281,7 +281,7 @@ private int GetInActiveClient(out RedisClient inactiveClient)`
`281`	`281`	`private RedisClient InitNewClient(RedisClient client)`
`282`	`282`	`{`
`283`	`283`	`client.Id = Interlocked.Increment(ref RedisClientCounter);`
`284`		`- client.Active = true;`
	`284`	`+ client.Activate(newClient:true);`
`285`	`285`	`client.ClientManager = this;`
`286`	`286`	`client.ConnectionFilter = ConnectionFilter;`
`287`	`287`
`@@ -303,7 +303,7 @@ public void DisposeClient(RedisNativeClient client)`
`303`	`303`	`else`
`304`	`304`	`{`
`305`	`305`	`client.TrackThread = null;`
`306`		`- client.Active = false;`
	`306`	`+ client.Deactivate();`
`307`	`307`	`}`
`308`	`308`
`309`	`309`	`Monitor.PulseAll(clients);`
`@@ -320,7 +320,7 @@ public void DisposeWriteClient(RedisNativeClient client)`
`320`	`320`	`{`
`321`	`321`	`lock (clients)`
`322`	`322`	`{`
`323`		`- client.Active = false;`
	`323`	`+ client.Deactivate();`
`324`	`324`	`}`
`325`	`325`	`}`
`326`	`326`
Original file line number	Diff line number	Diff line change
`@@ -83,7 +83,7 @@ public DateTime? DeactivatedAt`
`83`	`83`	`internal bool Active`
`84`	`84`	`{`
`85`	`85`	`get => Interlocked.CompareExchange(ref active, 0, 0) == YES;`
`86`		`- set => Interlocked.Exchange(ref active, value ? YES : NO);`
	`86`	`+ private set => Interlocked.Exchange(ref active, value ? YES : NO);`
`87`	`87`	`}`
`88`	`88`
`89`	`89`	`internal IHandleClientDispose ClientManager { get; set; }`