reaper implementation

Author: caleblloyd

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,47 @@ 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>();
+		static readonly TimeSpan ReaperInterval = TimeSpan.FromMinutes(1);
+		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/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;