Merge pull request #218 from caleblloyd/f_reaper

Add Connection Pool Reaper & ConnectionIdleTimeout.

Author: bgrainger

.ci/test.ps1

@@ -24,6 +24,11 @@ dotnet test tests\SideBySide\SideBySide.csproj -c Release
 if ($LASTEXITCODE -ne 0){
     exit $LASTEXITCODE;
 }
+echo "Executing Debug Only tests"
+dotnet test tests\SideBySide\SideBySide.csproj -c Debug --filter "FullyQualifiedName~SideBySide.DebugOnlyTests"
+if ($LASTEXITCODE -ne 0){
+    exit $LASTEXITCODE;
+}
 
 echo "Executing tests with Compression, No SSL"
 Copy-Item -Force .ci\config\config.compression.json tests\SideBySide\config.json

.travis.yml

@@ -18,6 +18,7 @@ script:
 - dotnet test tests/MySqlConnector.Tests/MySqlConnector.Tests.csproj -c Release -f netcoreapp1.1.1
 - dotnet build tests/SideBySide/SideBySide.csproj -c Release -f netcoreapp1.1.1
 - echo 'Executing tests with No Compression, No SSL' && ./.ci/use-config.sh config.json 172.17.0.1 3307 && time dotnet test tests/SideBySide/SideBySide.csproj -c Release -f netcoreapp1.1.1
+- echo 'Executing Debug Only tests' && time dotnet test tests/SideBySide/SideBySide.csproj -c Debug -f netcoreapp1.1.1 --filter "FullyQualifiedName~SideBySide.DebugOnlyTests"
 - echo 'Executing tests with Compression, No SSL' && ./.ci/use-config.sh config.compression.json 172.17.0.1 3307 && time dotnet test tests/SideBySide/SideBySide.csproj -c Release -f netcoreapp1.1.1
 - echo 'Executing tests with No Compression, SSL' && ./.ci/use-config.sh config.ssl.json 172.17.0.1 3307 && time dotnet test tests/SideBySide/SideBySide.csproj -c Release -f netcoreapp1.1.1
 - echo 'Executing tests with Compression, SSL' && ./.ci/use-config.sh config.compression+ssl.json 172.17.0.1 3307 && time dotnet test tests/SideBySide/SideBySide.csproj -c Release -f netcoreapp1.1.1

docs/content/connection-options.md

@@ -106,18 +106,28 @@ Connection pooling is enabled by default.  These options are used to configure i
   <tr>
     <td>Connection Lifetime, ConnectionLifeTime</td>
     <td>0</td>
-    <td>When a connection is returned to the pool, its creation time is compared with the current time, and the connection is destroyed if that time span (in seconds) exceeds the value specified by Connection Lifetime. This is useful in clustered configurations to force load balancing between a running server and a server just brought online. A value of zero (0) causes pooled connections to have the maximum connection timeout.</td>
+    <td>When a connection is returned to the pool, its creation time is compared with the current time, and the connection is destroyed if that time span (in seconds) exceeds the value specified by Connection Lifetime. This is useful in clustered configurations to force load balancing between a running server and a server just brought online. A value of zero (0) means pooled connections will never incur a ConnectionLifeTime timeout.</td>
   </tr>
   <tr>
     <td>Connection Reset, ConnectionReset	</td>
     <td>false</td>
     <td>If true, the connection state is reset when it is retrieved from the pool. The default value of false avoids making an additional server round trip when obtaining a connection, but the connection state is not reset.</td>
   </tr>
+  <tr>
+    <td>Connection Idle Timeout, ConnectionIdleTimeout</td>
+    <td>180</td>
+    <td>The amount of time in seconds that a connection can remain idle in the pool. Any connection that is idle for longer is subject to being closed by a background task that runs every minute, unless there are only MinimumPoolSize connections left in the pool. A value of zero (0) means pooled connections will never incur a ConnectionIdleTimeout.</td>
+  </tr>
   <tr>
     <td>Maximum Pool Size, Max Pool Size, MaximumPoolsize, maxpoolsize</td>
     <td>100</td>
     <td>The maximum number of connections allowed in the pool.</td>
   </tr>
+  <tr>
+    <td>Minimum Pool Size, Min Pool Size, MinimumPoolSize, minpoolsize</td>
+    <td>0</td>
+    <td>The minimum number of connections to leave in the pool if ConnectionIdleTimeout is reached.</td>
+  </tr>
 </table>
 
 Other Options

src/MySqlConnector/MySqlClient/ConnectionPool.cs

@@ -22,8 +22,17 @@ public async Task<MySqlSession> GetSessionAsync(IOBehavior ioBehavior, Cancellat
 
 			try
 			{
-				// check for a pooled session
-				if (m_sessions.TryDequeue(out var session))
+				// check for a waiting session
+				MySqlSession session = null;
+				lock (m_sessions)
+				{
+					if (m_sessions.Count > 0)
+					{
+						session = m_sessions.First.Value;
+						m_sessions.RemoveFirst();
+					}
+				}
+				if (session != null)
 				{
 					if (session.PoolGeneration != m_generation || !await session.TryPingAsync(ioBehavior, cancellationToken).ConfigureAwait(false))
 					{
@@ -42,6 +51,7 @@ public async Task<MySqlSession> GetSessionAsync(IOBehavior ioBehavior, Cancellat
 					}
 				}
 
+				// create a new session
 				session = new MySqlSession(this, m_generation);
 				await session.ConnectAsync(m_connectionSettings, ioBehavior, cancellationToken).ConfigureAwait(false);
 				return session;
@@ -73,7 +83,8 @@ public void Return(MySqlSession session)
 			try
 			{
 				if (SessionIsHealthy(session))
-					m_sessions.Enqueue(session);
+					lock (m_sessions)
+						m_sessions.AddFirst(session);
 				else
 					session.DisposeAsync(IOBehavior.Synchronous, CancellationToken.None).ConfigureAwait(false);
 			}
@@ -87,45 +98,85 @@ public async Task ClearAsync(IOBehavior ioBehavior, CancellationToken cancellati
 		{
 			// increment the generation of the connection pool
 			Interlocked.Increment(ref m_generation);
+			await CleanPoolAsync(ioBehavior, session => session.PoolGeneration != m_generation, false, cancellationToken).ConfigureAwait(false);
+		}
 
-			var waitTimeout = TimeSpan.FromMilliseconds(10);
-			while (true)
+		public async Task ReapAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
+		{
+			if (m_connectionSettings.ConnectionIdleTimeout == 0)
+				return;
+			await CleanPoolAsync(ioBehavior, session => (DateTime.UtcNow - session.LastReturnedUtc).TotalSeconds >= m_connectionSettings.ConnectionIdleTimeout, true, cancellationToken).ConfigureAwait(false);
+		}
+
+		private async Task CleanPoolAsync(IOBehavior ioBehavior, Func<MySqlSession, bool> shouldCleanFn, bool respectMinPoolSize, CancellationToken cancellationToken)
+		{
+			// synchronize access to this method as only one clean routine should be run at a time
+			if (ioBehavior == IOBehavior.Asynchronous)
+				await m_cleanSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+			else
+				m_cleanSemaphore.Wait(cancellationToken);
+
+			try
 			{
-				// try to get an open slot; if this fails, connection pool is full and sessions will be disposed when returned to pool
-				if (ioBehavior == IOBehavior.Asynchronous)
-				{
-					if (!await m_sessionSemaphore.WaitAsync(waitTimeout, cancellationToken).ConfigureAwait(false))
-						return;
-				}
-				else
+				var waitTimeout = TimeSpan.FromMilliseconds(10);
+				while (true)
 				{
-					if (!m_sessionSemaphore.Wait(waitTimeout, cancellationToken))
-						return;
-				}
+					// if respectMinPoolSize is true, return if (leased sessions + waiting sessions <= minPoolSize)
+					if (respectMinPoolSize)
+						lock (m_sessions)
+							if (m_connectionSettings.MaximumPoolSize - m_sessionSemaphore.CurrentCount + m_sessions.Count <= m_connectionSettings.MinimumPoolSize)
+								return;
+
+					// try to get an open slot; if this fails, connection pool is full and sessions will be disposed when returned to pool
+					if (ioBehavior == IOBehavior.Asynchronous)
+					{
+						if (!await m_sessionSemaphore.WaitAsync(waitTimeout, cancellationToken).ConfigureAwait(false))
+							return;
+					}
+					else
+					{
+						if (!m_sessionSemaphore.Wait(waitTimeout, cancellationToken))
+							return;
+					}
 
-				try
-				{
-					if (m_sessions.TryDequeue(out var session))
+					try
 					{
-						if (session.PoolGeneration != m_generation)
+						// check for a waiting session
+						MySqlSession session = null;
+						lock (m_sessions)
+						{
+							if (m_sessions.Count > 0)
+							{
+								session = m_sessions.Last.Value;
+								m_sessions.RemoveLast();
+							}
+						}
+						if (session == null)
+							return;
+
+						if (shouldCleanFn(session))
 						{
-							// session generation does not match pool generation; dispose of it and continue iterating
+							// session should be cleaned; dispose it and keep iterating
 							await session.DisposeAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
-							continue;
 						}
 						else
 						{
-							// session generation matches pool generation; put it back in the queue and stop iterating
-							m_sessions.Enqueue(session);
+							// session should not be cleaned; put it back in the queue and stop iterating
+							lock (m_sessions)
+								m_sessions.AddLast(session);
+							return;
 						}
 					}
-					return;
-				}
-				finally
-				{
-					m_sessionSemaphore.Release();
+					finally
+					{
+						m_sessionSemaphore.Release();
+					}
 				}
 			}
+			finally
+			{
+				m_cleanSemaphore.Release();
+			}
 		}
 
 		public static ConnectionPool GetPool(ConnectionSettings cs)
@@ -144,25 +195,51 @@ public static ConnectionPool GetPool(ConnectionSettings cs)
 
 		public static async Task ClearPoolsAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
 		{
-			var pools = new List<ConnectionPool>(s_pools.Values);
-
-			foreach (var pool in pools)
+			foreach (var pool in s_pools.Values)
 				await pool.ClearAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
 		}
 
+		public static async Task ReapPoolsAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)
+		{
+			foreach (var pool in s_pools.Values)
+				await pool.ReapAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
+		}
+
 		private ConnectionPool(ConnectionSettings cs)
 		{
 			m_connectionSettings = cs;
 			m_generation = 0;
+			m_cleanSemaphore = new SemaphoreSlim(1);
 			m_sessionSemaphore = new SemaphoreSlim(cs.MaximumPoolSize);
-			m_sessions = new ConcurrentQueue<MySqlSession>();
+			m_sessions = new LinkedList<MySqlSession>();
 		}
 
 		static readonly ConcurrentDictionary<string, ConnectionPool> s_pools = new ConcurrentDictionary<string, ConnectionPool>();
+#if DEBUG
+		static readonly TimeSpan ReaperInterval = TimeSpan.FromSeconds(1);
+#else
+		static readonly TimeSpan ReaperInterval = TimeSpan.FromMinutes(1);
+#endif
+		static readonly Task Reaper = Task.Run(async () => {
+			while (true)
+			{
+				var task = Task.Delay(ReaperInterval);
+				try
+				{
+					await ReapPoolsAsync(IOBehavior.Asynchronous, new CancellationTokenSource(ReaperInterval).Token).ConfigureAwait(false);
+				}
+				catch
+				{
+					// do nothing; we'll try to reap again
+				}
+				await task.ConfigureAwait(false);
+			}
+		});
 
 		int m_generation;
+		readonly SemaphoreSlim m_cleanSemaphore;
 		readonly SemaphoreSlim m_sessionSemaphore;
-		readonly ConcurrentQueue<MySqlSession> m_sessions;
+		readonly LinkedList<MySqlSession> m_sessions;
 		readonly ConnectionSettings m_connectionSettings;
 	}
 }

src/MySqlConnector/MySqlClient/MySqlConnectionStringBuilder.cs

@@ -85,6 +85,12 @@ public bool ConnectionReset
 			set => MySqlConnectionStringOption.ConnectionReset.SetValue(this, value);
 		}
 
+		public uint ConnectionIdleTimeout
+		{
+			get => MySqlConnectionStringOption.ConnectionIdleTimeout.GetValue(this);
+			set => MySqlConnectionStringOption.ConnectionIdleTimeout.SetValue(this, value);
+		}
+
 		public uint MinimumPoolSize
 		{
 			get => MySqlConnectionStringOption.MinimumPoolSize.GetValue(this);
@@ -231,6 +237,7 @@ internal abstract class MySqlConnectionStringOption
 		public static readonly MySqlConnectionStringOption<bool> Pooling;
 		public static readonly MySqlConnectionStringOption<uint> ConnectionLifeTime;
 		public static readonly MySqlConnectionStringOption<bool> ConnectionReset;
+		public static readonly MySqlConnectionStringOption<uint> ConnectionIdleTimeout;
 		public static readonly MySqlConnectionStringOption<uint> MinimumPoolSize;
 		public static readonly MySqlConnectionStringOption<uint> MaximumPoolSize;
 
@@ -321,6 +328,10 @@ static MySqlConnectionStringOption()
 				keys: new[] { "Connection Reset", "ConnectionReset" },
 				defaultValue: true));
 
+			AddOption(ConnectionIdleTimeout = new MySqlConnectionStringOption<uint>(
+				keys: new[] { "Connection Idle Timeout", "ConnectionIdleTimeout" },
+				defaultValue: 180));
+
 			AddOption(MinimumPoolSize = new MySqlConnectionStringOption<uint>(
 				keys: new[] { "Minimum Pool Size", "Min Pool Size", "MinimumPoolSize", "minpoolsize" },
 				defaultValue: 0));

src/MySqlConnector/MySqlConnector.csproj

@@ -36,6 +36,10 @@
     <PackageReference Include="System.Threading.Tasks.Extensions" Version="4.3.0" />
   </ItemGroup>
 
+  <PropertyGroup Condition=" '$(Configuration)' == 'Debug' ">
+    <DefineConstants>DEBUG</DefineConstants>
+  </PropertyGroup>
+
   <ItemGroup>
     <Compile Update="MySqlClient\MySqlCommand.cs">
       <SubType>Component</SubType>

src/MySqlConnector/Serialization/ConnectionSettings.cs

@@ -38,6 +38,7 @@ public ConnectionSettings(MySqlConnectionStringBuilder csb)
 			Pooling = csb.Pooling;
 			ConnectionLifeTime = (int)csb.ConnectionLifeTime;
 			ConnectionReset = csb.ConnectionReset;
+			ConnectionIdleTimeout = (int)csb.ConnectionIdleTimeout;
 			if (csb.MinimumPoolSize > csb.MaximumPoolSize)
 				throw new MySqlException("MaximumPoolSize must be greater than or equal to MinimumPoolSize");
 			MinimumPoolSize = (int)csb.MinimumPoolSize;
@@ -80,6 +81,7 @@ private ConnectionSettings(ConnectionSettings other, bool? useCompression)
 			Pooling = other.Pooling;
 			ConnectionLifeTime = other.ConnectionLifeTime;
 			ConnectionReset = other.ConnectionReset;
+			ConnectionIdleTimeout = other.ConnectionIdleTimeout;
 			MinimumPoolSize = other.MinimumPoolSize;
 			MaximumPoolSize = other.MaximumPoolSize;
 
@@ -116,6 +118,7 @@ private ConnectionSettings(ConnectionSettings other, bool? useCompression)
 		internal readonly bool Pooling;
 		internal readonly int ConnectionLifeTime;
 		internal readonly bool ConnectionReset;
+		internal readonly int ConnectionIdleTimeout;
 		internal readonly int MinimumPoolSize;
 		internal readonly int MaximumPoolSize;
 
@@ -134,4 +137,3 @@ private ConnectionSettings(ConnectionSettings other, bool? useCompression)
 	}
 
 }
-

src/MySqlConnector/Serialization/MySqlSession.cs

@@ -34,9 +34,14 @@ public MySqlSession(ConnectionPool pool, int poolGeneration)
 		public DateTime CreatedUtc { get; }
 		public ConnectionPool Pool { get; }
 		public int PoolGeneration { get; }
+		public DateTime LastReturnedUtc { get; private set; }
 		public string DatabaseOverride { get; set; }
 
-		public void ReturnToPool() => Pool?.Return(this);
+		public void ReturnToPool()
+		{
+			LastReturnedUtc = DateTime.UtcNow;
+			Pool?.Return(this);
+		}
 
 		public bool IsConnected => m_state == State.Connected;
 

tests/MySqlConnector.Tests/MySqlConnectionStringBuilderTests.cs

@@ -25,6 +25,7 @@ public void Defaults()
 			Assert.Equal("", csb.Database);
 #if !BASELINE
 			Assert.Equal(false, csb.BufferResultSets);
+			Assert.Equal(180u, csb.ConnectionIdleTimeout);
 			Assert.Equal(false, csb.ForceSynchronous);
 #endif
 			Assert.Equal(0u, csb.Keepalive);
@@ -64,6 +65,7 @@ public void ParseConnectionString()
 				                   "ConnectionReset=false;" +
 				                   "Convert Zero Datetime=true;" +
 #if !BASELINE
+				                   "connectionidletimeout=30;" +
 				                   "bufferresultsets=true;" +
 				                   "forcesynchronous=true;" +
 #endif
@@ -91,6 +93,7 @@ public void ParseConnectionString()
 			Assert.Equal("schema_name", csb.Database);
 #if !BASELINE
 			Assert.Equal(true, csb.BufferResultSets);
+			Assert.Equal(30u, csb.ConnectionIdleTimeout);
 			Assert.Equal(true, csb.ForceSynchronous);
 #endif
 			Assert.Equal(90u, csb.Keepalive);