Merge pull request #152 from cnblogs/implement-reset-async

Implement ResetAsync

Author: cnblogs-dudu

Enyim.Caching/Memcached/MemcachedNode.cs

@@ -278,6 +278,7 @@ private class InternalPoolImpl : IDisposable
             private MemcachedNode ownerNode;
             private readonly EndPoint _endPoint;
             private readonly TimeSpan queueTimeout;
+            private readonly TimeSpan _receiveTimeout;
             private SemaphoreSlim _semaphore;
 
             private readonly object initLock = new Object();
@@ -293,11 +294,14 @@ internal InternalPoolImpl(
                     throw new InvalidOperationException("maxItems must be larger than minItems", null);
                 if (config.QueueTimeout < TimeSpan.Zero)
                     throw new InvalidOperationException("queueTimeout must be >= TimeSpan.Zero", null);
+                if (config.ReceiveTimeout < TimeSpan.Zero)
+                    throw new InvalidOperationException("ReceiveTimeout must be >= TimeSpan.Zero", null);
 
                 this.ownerNode = ownerNode;
                 this.isAlive = true;
                 _endPoint = ownerNode.EndPoint;
                 this.queueTimeout = config.QueueTimeout;
+                _receiveTimeout = config.ReceiveTimeout;
 
                 this.minItems = config.MinPoolSize;
                 this.maxItems = config.MaxPoolSize;
@@ -356,7 +360,7 @@ internal async Task InitPoolAsync()
                             {
                                 _freeItems.Push(await CreateSocketAsync());
                             }
-                            catch(Exception ex)
+                            catch (Exception ex)
                             {
                                 _logger.LogError(ex, $"Failed to put {nameof(PooledSocket)} {i} in Pool");
                             }
@@ -561,9 +565,22 @@ public async Task<IPooledSocketResult> AcquireAsync()
 
                     try
                     {
-                        retval.Reset();
+                        var resetTask = retval.ResetAsync();
 
-                        message = "Socket was reset. " + retval.InstanceId;
+                        if (await Task.WhenAny(resetTask, Task.Delay(_receiveTimeout)) == resetTask)
+                        {
+                            await resetTask;
+                        }
+                        else
+                        {
+                            message = "Timeout to reset an acquired socket. InstanceId " + retval.InstanceId;
+                            _logger.LogError(message);
+                            MarkAsDead();
+                            result.Fail(message);
+                            return result;
+                        }
+
+                        message = "Socket was reset. InstanceId " + retval.InstanceId;
                         if (_isDebugEnabled) _logger.LogDebug(message);
 
                         result.Pass(message);
@@ -796,7 +813,7 @@ protected internal virtual async Task<PooledSocket> CreateSocketAsync()
             }
             catch (Exception ex)
             {
-                var endPointStr =  _endPoint.ToString().Replace("Unspecified/", string.Empty);
+                var endPointStr = _endPoint.ToString().Replace("Unspecified/", string.Empty);
                 _logger.LogError(ex, $"Failed to {nameof(CreateSocketAsync)} to {endPointStr}");
                 throw;
             }
@@ -874,7 +891,7 @@ protected virtual async Task<IPooledSocketResult> ExecuteOperationAsync(IOperati
                     _logger.LogDebug("pooledSocket.WriteAsync...");
 
                     var writeSocketTask = pooledSocket.WriteAsync(b);
-                    if(await Task.WhenAny(writeSocketTask, Task.Delay(_config.ConnectionTimeout)) != writeSocketTask)
+                    if (await Task.WhenAny(writeSocketTask, Task.Delay(_config.ConnectionTimeout)) != writeSocketTask)
                     {
                         result.Fail("Timeout to pooledSocket.WriteAsync");
                         return result;

Enyim.Caching/Memcached/PooledSocket.cs

@@ -158,8 +158,7 @@ public int Available
 
         public void Reset()
         {
-            // discard any buffered data
-            _inputStream.Flush();
+            // _inputStream.Flush();
 
             int available = _socket.Available;
 
@@ -168,18 +167,39 @@ public void Reset()
                 if (_logger.IsEnabled(LogLevel.Warning))
                     _logger.LogWarning(
                         "Socket bound to {0} has {1} unread data! This is probably a bug in the code. InstanceID was {2}.",
-                        _socket.RemoteEndPoint, available, this.InstanceId);
+                        _socket.RemoteEndPoint, available, InstanceId);
 
                 byte[] data = new byte[available];
 
-                this.Read(data, 0, available);
+                Read(data, 0, available);
+            }
 
+            if (_logger.IsEnabled(LogLevel.Debug))
+                _logger.LogDebug("Socket {0} was reset", InstanceId);
+        }
+
+        public async Task ResetAsync()
+        {
+            // await _inputStream.FlushAsync();
+
+            int available = _socket.Available;
+
+            if (available > 0)
+            {
                 if (_logger.IsEnabled(LogLevel.Warning))
-                    _logger.LogWarning(Encoding.ASCII.GetString(data));
+                {
+                    _logger.LogWarning(
+                        "Socket bound to {0} has {1} unread data! This is probably a bug in the code. InstanceID was {2}.",
+                        _socket.RemoteEndPoint, available, InstanceId);
+                }
+
+                byte[] data = new byte[available];
+
+                await ReadAsync(data, 0, available);
             }
 
             if (_logger.IsEnabled(LogLevel.Debug))
-                _logger.LogDebug("Socket {0} was reset", this.InstanceId);
+                _logger.LogDebug("Socket {0} was reset", InstanceId);
         }
 
         /// <summary>