Added an option to specify max idle time for a pooled socket. (#185)

Add an option to specify max idle time for a pooled socket

Co-authored-by: Roman Polunin <romanspo@gmail.com>
Co-authored-by: dudu <dudu@cnblogs.com>

Author: romanpolunin

src/Enyim.Caching/Configuration/ISocketPoolConfiguration.cs

@@ -68,6 +68,16 @@ TimeSpan DeadTimeout
             set;
         }
 
+        /// <summary>
+        /// Gets or sets a value that specified the amount of time after which a pooled socket will be discarded. Discarding will only happen at the time when the socket is picked up from the socket pool, e.g. it depends on usage.
+        /// </summary>
+        /// <returns>The value of the connection idle timeout.</returns>
+        TimeSpan ConnectionIdleTimeout
+        {
+            get;
+            set;
+        }
+
         TimeSpan InitPoolTimeout { get; set; }
 
         INodeFailurePolicyFactory FailurePolicyFactory { get; set; }

src/Enyim.Caching/Configuration/MemcachedClientConfiguration.cs

@@ -81,6 +81,9 @@ public MemcachedClientConfiguration(
                 SocketPool.QueueTimeout = options.SocketPool.QueueTimeout;
                 _logger.LogInformation($"{nameof(SocketPool.QueueTimeout)}: {SocketPool.QueueTimeout}");
 
+                SocketPool.ConnectionIdleTimeout = options.SocketPool.ConnectionIdleTimeout;
+                _logger.LogInformation($"{nameof(SocketPool.ConnectionIdleTimeout)}: {SocketPool.ConnectionIdleTimeout}");
+
                 SocketPool.InitPoolTimeout = options.SocketPool.InitPoolTimeout;
 
                 SocketPool.FailurePolicyFactory = options.SocketPool.FailurePolicyFactory;

src/Enyim.Caching/Configuration/MemcachedClientOptions.cs

@@ -72,6 +72,7 @@ public class SocketPoolOptions
         public TimeSpan ReceiveTimeout { get; set; } = new TimeSpan(0, 0, 10);
         public TimeSpan DeadTimeout { get; set; } = new TimeSpan(0, 0, 10);
         public TimeSpan QueueTimeout { get; set; } = new TimeSpan(0, 0, 0, 0, 100);
+        public TimeSpan ConnectionIdleTimeout { get; set; } = TimeSpan.Zero;
         public TimeSpan InitPoolTimeout { get; set; } = new TimeSpan(0, 1, 0);
         public INodeFailurePolicyFactory FailurePolicyFactory { get; set; } = new ThrottlingFailurePolicyFactory(5, TimeSpan.FromMilliseconds(2000));
 
@@ -93,6 +94,7 @@ public void CheckTimeout()
             CheckTimeout(nameof(ReceiveTimeout), ReceiveTimeout);
             CheckTimeout(nameof(DeadTimeout), DeadTimeout);
             CheckTimeout(nameof(QueueTimeout), QueueTimeout);
+            CheckTimeout(nameof(ConnectionIdleTimeout), ConnectionIdleTimeout);
         }
 
         private void CheckTimeout(string paramName, TimeSpan value)

src/Enyim.Caching/Configuration/SocketPoolConfiguration.cs

@@ -15,6 +15,7 @@ public class SocketPoolConfiguration : ISocketPoolConfiguration
         private TimeSpan _receiveTimeout = new TimeSpan(0, 0, 10);
         private TimeSpan _deadTimeout = new TimeSpan(0, 0, 10);
         private TimeSpan _queueTimeout = new TimeSpan(0, 0, 0, 0, 100);
+        private TimeSpan _connectionIdleTimeout = TimeSpan.Zero;
         private TimeSpan _initPoolTimeout = new TimeSpan(0, 1, 0);
         private INodeFailurePolicyFactory _failurePolicyFactory = new ThrottlingFailurePolicyFactory(5, TimeSpan.FromMilliseconds(2000));
 
@@ -95,6 +96,18 @@ TimeSpan ISocketPoolConfiguration.DeadTimeout
             }
         }
 
+        TimeSpan ISocketPoolConfiguration.ConnectionIdleTimeout
+        {
+            get { return _connectionIdleTimeout; }
+            set
+            {
+                if (value < TimeSpan.Zero)
+                    throw new ArgumentOutOfRangeException("value", "value must be positive");
+
+                _connectionIdleTimeout = value;
+            }
+        }
+
         INodeFailurePolicyFactory ISocketPoolConfiguration.FailurePolicyFactory
         {
             get { return _failurePolicyFactory; }

src/Enyim.Caching/Memcached/MemcachedNode.cs

@@ -3,7 +3,9 @@
 using Enyim.Caching.Memcached.Results;
 using Enyim.Caching.Memcached.Results.Extensions;
 using Enyim.Collections;
+
 using Microsoft.Extensions.Logging;
+
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
@@ -25,7 +27,7 @@ namespace Enyim.Caching.Memcached
     public class MemcachedNode : IMemcachedNode
     {
         private readonly ILogger _logger;
-        private static readonly object SyncRoot = new Object();
+        private static readonly object SyncRoot = new();
         private bool _isDisposed;
         private readonly EndPoint _endPoint;
         private readonly ISocketPoolConfiguration _config;
@@ -46,8 +48,8 @@ public MemcachedNode(
             EndPointString = endpoint?.ToString().Replace("Unspecified/", string.Empty);
             _config = socketPoolConfig;
 
-            if (socketPoolConfig.ConnectionTimeout.TotalMilliseconds >= Int32.MaxValue)
-                throw new InvalidOperationException("ConnectionTimeout must be < Int32.MaxValue");
+            if (socketPoolConfig.ConnectionTimeout.TotalMilliseconds >= int.MaxValue)
+                throw new InvalidOperationException("ConnectionTimeout must be < int.MaxValue");
 
             if (socketPoolConfig.InitPoolTimeout.TotalSeconds < 1)
             {
@@ -281,6 +283,7 @@ private class InternalPoolImpl : IDisposable
             private readonly EndPoint _endPoint;
             private readonly TimeSpan _queueTimeout;
             private readonly TimeSpan _receiveTimeout;
+            private readonly TimeSpan _connectionIdleTimeout;
             private SemaphoreSlim _semaphore;
 
             private readonly object initLock = new Object();
@@ -298,12 +301,15 @@ internal InternalPoolImpl(
                     throw new InvalidOperationException("queueTimeout must be >= TimeSpan.Zero", null);
                 if (config.ReceiveTimeout < TimeSpan.Zero)
                     throw new InvalidOperationException("ReceiveTimeout must be >= TimeSpan.Zero", null);
+                if (config.ConnectionIdleTimeout < TimeSpan.Zero)
+                    throw new InvalidOperationException("ConnectionIdleTimeout must be >= TimeSpan.Zero", null);
 
                 _ownerNode = ownerNode;
                 _isAlive = true;
                 _endPoint = ownerNode.EndPoint;
                 _queueTimeout = config.QueueTimeout;
                 _receiveTimeout = config.ReceiveTimeout;
+                _connectionIdleTimeout = config.ConnectionIdleTimeout;
 
                 _minItems = config.MinPoolSize;
                 _maxItems = config.MaxPoolSize;
@@ -344,7 +350,7 @@ internal void InitPool()
                 }
                 catch (Exception e)
                 {
-                    _logger.LogError("Could not init pool.", new EventId(0), e);
+                    _logger.LogError(e, "Could not init pool.");
 
                     MarkAsDead();
                 }
@@ -379,7 +385,7 @@ internal async Task InitPoolAsync()
                 }
                 catch (Exception e)
                 {
-                    _logger.LogError("Could not init pool.", new EventId(0), e);
+                    _logger.LogError(e, "Could not init pool.");
 
                     MarkAsDead();
                 }
@@ -418,10 +424,9 @@ public DateTime MarkedAsDeadUtc
             public IPooledSocketResult Acquire()
             {
                 var result = new PooledSocketResult();
-                var message = string.Empty;
-
                 if (_isDebugEnabled) _logger.LogDebug($"Acquiring stream from pool on node '{_endPoint}'");
 
+                string message;
                 if (!_isAlive || _isDisposed)
                 {
                     message = "Pool is dead or disposed, returning null. " + _endPoint;
@@ -432,8 +437,6 @@ public IPooledSocketResult Acquire()
                     return result;
                 }
 
-                PooledSocket retval = null;
-
                 if (!_semaphore.Wait(_queueTimeout))
                 {
                     message = "Pool is full, timeouting. " + _endPoint;
@@ -456,20 +459,23 @@ public IPooledSocketResult Acquire()
                     return result;
                 }
 
+
+                PooledSocket socket;
                 // do we have free items?
-                if (_freeItems.TryPop(out retval))
+                if (TryPopPooledSocket(out socket))
                 {
                     #region [ get it from the pool         ]
 
                     try
                     {
-                        retval.Reset();
+                        socket.Reset();
 
-                        message = "Socket was reset. " + retval.InstanceId;
+                        message = "Socket was reset. " + socket.InstanceId;
                         if (_isDebugEnabled) _logger.LogDebug(message);
 
                         result.Pass(message);
-                        result.Value = retval;
+                        socket.UpdateLastUsed();
+                        result.Value = socket;
                         return result;
                     }
                     catch (Exception e)
@@ -495,9 +501,9 @@ public IPooledSocketResult Acquire()
                 {
                     // okay, create the new item
                     var startTime = DateTime.Now;
-                    retval = CreateSocket();
+                    socket = CreateSocket();
                     _logger.LogInformation("MemcachedAcquire-CreateSocket: {0}ms", (DateTime.Now - startTime).TotalMilliseconds);
-                    result.Value = retval;
+                    result.Value = socket;
                     result.Pass();
                 }
                 catch (Exception e)
@@ -542,7 +548,7 @@ public async Task<IPooledSocketResult> AcquireAsync()
                     return result;
                 }
 
-                PooledSocket retval = null;
+                PooledSocket socket = null;
 
                 if (!await _semaphore.WaitAsync(_queueTimeout))
                 {
@@ -566,13 +572,13 @@ public async Task<IPooledSocketResult> AcquireAsync()
                 }
 
                 // do we have free items?
-                if (_freeItems.TryPop(out retval))
+                if (TryPopPooledSocket(out socket))
                 {
                     #region [ get it from the pool         ]
 
                     try
                     {
-                        var resetTask = retval.ResetAsync();
+                        var resetTask = socket.ResetAsync();
 
                         if (await Task.WhenAny(resetTask, Task.Delay(_receiveTimeout)) == resetTask)
                         {
@@ -581,19 +587,20 @@ public async Task<IPooledSocketResult> AcquireAsync()
                         else
                         {
                             _semaphore.Release();
-                            retval.IsAlive = false;
+                            socket.IsAlive = false;
 
-                            message = "Timeout to reset an acquired socket. InstanceId " + retval.InstanceId;
+                            message = "Timeout to reset an acquired socket. InstanceId " + socket.InstanceId;
                             _logger.LogError(message);
                             result.Fail(message);
                             return result;
                         }
 
-                        message = "Socket was reset. InstanceId " + retval.InstanceId;
+                        message = "Socket was reset. InstanceId " + socket.InstanceId;
                         if (_isDebugEnabled) _logger.LogDebug(message);
 
                         result.Pass(message);
-                        result.Value = retval;
+                        socket.UpdateLastUsed();
+                        result.Value = socket;
                         return result;
                     }
                     catch (Exception e)
@@ -619,9 +626,9 @@ public async Task<IPooledSocketResult> AcquireAsync()
                 {
                     // okay, create the new item
                     var startTime = DateTime.Now;
-                    retval = await CreateSocketAsync();
+                    socket = await CreateSocketAsync();
                     _logger.LogInformation("MemcachedAcquire-CreateSocket: {0}ms", (DateTime.Now - startTime).TotalMilliseconds);
-                    result.Value = retval;
+                    result.Value = socket;
                     result.Pass();
                 }
                 catch (Exception e)
@@ -737,6 +744,34 @@ private void ReleaseSocket(PooledSocket socket)
                 }
             }
 
+            private bool TryPopPooledSocket(out PooledSocket pooledSocket)
+            {
+                if (_freeItems.TryPop(out var socket))
+                {
+                    if (_connectionIdleTimeout > TimeSpan.Zero &&
+                        socket.LastUsed < DateTime.UtcNow.Subtract(_connectionIdleTimeout))
+                    {
+                        try
+                        {
+                            _logger.LogInformation("Connection idle timeout {idleTimeout} reached.", _connectionIdleTimeout);
+                            socket.Destroy();
+                        }
+                        catch (Exception ex)
+                        {
+                            _logger.LogError(ex, $"Failed to destroy {nameof(PooledSocket)}");
+                        }
+
+                        pooledSocket = null;
+                        return false;
+                    }
+
+                    pooledSocket = socket;
+                    return true;
+                }
+
+                pooledSocket = null;
+                return false;
+            }
 
             ~InternalPoolImpl()
             {
@@ -758,12 +793,10 @@ public void Dispose()
                     _isAlive = false;
                     _isDisposed = true;
 
-                    PooledSocket ps;
-
-                    while (_freeItems.TryPop(out ps))
+                    while (_freeItems.TryPop(out var socket))
                     {
-                        try { ps.Destroy(); }
-                        catch { }
+                        try { socket.Destroy(); }
+                        catch (Exception ex) { _logger.LogError(ex, $"failed to destroy {nameof(PooledSocket)}"); }
                     }
 
                     _ownerNode = null;

src/Enyim.Caching/Memcached/PooledSocket.cs

@@ -176,6 +176,13 @@ public int Available
             get { return _socket.Available; }
         }
 
+        public DateTime LastUsed { get; set; } = DateTime.UtcNow;
+
+        public void UpdateLastUsed()
+        {
+            LastUsed = DateTime.UtcNow;
+        }
+
         public void Reset()
         {
             // _inputStream.Flush();

test/Enyim.Caching.Tests/MemcachedClientCasTests.cs

@@ -6,33 +6,31 @@
 
 namespace Enyim.Caching.Tests
 {
-
     public class MemcachedClientCasTests : MemcachedClientTestsBase
-	{
-
-		[Fact]
-		public void When_Storing_Item_With_Valid_Cas_Result_Is_Successful()
-		{
-			var key = GetUniqueKey("cas");
-			var value = GetRandomString();
-			var storeResult = Store(StoreMode.Add, key, value);
-			StoreAssertPass(storeResult);
-
-			var casResult = _client.ExecuteCas(StoreMode.Set, key, value, storeResult.Cas);
-			StoreAssertPass(casResult);
-		}
-
-		[Fact]
-		public void When_Storing_Item_With_Invalid_Cas_Result_Is_Not_Successful()
-		{
-			var key = GetUniqueKey("cas");
-			var value = GetRandomString();
-			var storeResult = Store(StoreMode.Add, key, value);
-			StoreAssertPass(storeResult);
-
-			var casResult = _client.ExecuteCas(StoreMode.Set, key, value, storeResult.Cas - 1);
-			StoreAssertFail(casResult);
-		}
+    {
+        [Fact]
+        public void When_Storing_Item_With_Valid_Cas_Result_Is_Successful()
+        {
+            var key = GetUniqueKey("cas");
+            var value = GetRandomString();
+            var storeResult = Store(StoreMode.Add, key, value);
+            StoreAssertPass(storeResult);
+
+            var casResult = _client.ExecuteCas(StoreMode.Set, key, value, storeResult.Cas);
+            StoreAssertPass(casResult);
+        }
+
+        [Fact]
+        public void When_Storing_Item_With_Invalid_Cas_Result_Is_Not_Successful()
+        {
+            var key = GetUniqueKey("cas");
+            var value = GetRandomString();
+            var storeResult = Store(StoreMode.Add, key, value);
+            StoreAssertPass(storeResult);
+
+            var casResult = _client.ExecuteCas(StoreMode.Set, key, value, storeResult.Cas - 1);
+            StoreAssertFail(casResult);
+        }
 
 
         [Fact]

test/Enyim.Caching.Tests/MemcachedClientTestsBase.cs

@@ -26,7 +26,7 @@ public MemcachedClientTestsBase(Action<MemcachedClientOptions> onAddEnyimMemcach
                 onAddEnyimMemcached?.Invoke(options);
             });
 
-            services.AddLogging(builder => builder.SetMinimumLevel(LogLevel.Information).AddConsole());
+            services.AddLogging(builder => builder.SetMinimumLevel(LogLevel.Warning).AddConsole());
             IServiceProvider serviceProvider = services.BuildServiceProvider();
             _client = serviceProvider.GetService<IMemcachedClient>() as MemcachedClient;
         }