Don't serialize returning connections to pool in background.

This was a bottleneck in highly concurrent applications that could generate unacceptably high latencies when waiting for a connection to be reset and returned to the pool.

Signed-off-by: Bradley Grainger <bgrainger@gmail.com>

Author: bgrainger

src/MySqlConnector/Core/BackgroundConnectionResetHelper.cs

@@ -11,12 +11,12 @@ internal static class BackgroundConnectionResetHelper
 	{
 		public static void AddSession(ServerSession session, MySqlConnection? owningConnection)
 		{
-			var resetTask = session.TryResetConnectionAsync(session.Pool!.ConnectionSettings, IOBehavior.Asynchronous, default);
+			var resetTask = session.TryResetConnectionAsync(session.Pool!.ConnectionSettings, owningConnection, true, IOBehavior.Asynchronous, default);
 			lock (s_lock)
-				s_sessions.Add(new SessionResetTask(session, resetTask, owningConnection));
+				s_resetTasks.Add(resetTask);
 
 			if (Log.IsDebugEnabled())
-				Log.Debug("Started Session{0} reset in background; waiting SessionCount: {1}.", session.Id, s_sessions.Count);
+				Log.Debug("Started Session{0} reset in background; waiting TaskCount: {1}.", session.Id, s_resetTasks.Count);
 
 			// release only if it is likely to succeed
 			if (s_semaphore.CurrentCount == 0)
@@ -69,7 +69,6 @@ public static async Task ReturnSessionsAsync()
 			Log.Info("Started BackgroundConnectionResetHelper worker.");
 
 			List<Task<bool>> localTasks = new();
-			List<SessionResetTask> localSessions = new();
 
 			// keep running until stopped
 			while (!s_cancellationTokenSource.IsCancellationRequested)
@@ -85,36 +84,18 @@ public static async Task ReturnSessionsAsync()
 					{
 						lock (s_lock)
 						{
-							if (s_sessions.Count == 0)
-							{
-								if (localTasks.Count == 0)
-									break;
-							}
-							else
-							{
-								foreach (var session in s_sessions)
-								{
-									localSessions.Add(session);
-									localTasks.Add(session.ResetTask);
-								}
-								s_sessions.Clear();
-							}
+							localTasks.AddRange(s_resetTasks);
+							s_resetTasks.Clear();
 						}
 
+						if (localTasks.Count == 0)
+							break;
+
 						if (Log.IsDebugEnabled())
-							Log.Debug("Found SessionCount {0} session(s) to return.", localSessions.Count);
+							Log.Debug("Found TaskCount {0} task(s) to process.", localTasks.Count);
 
-						while (localTasks.Count != 0)
-						{
-							var completedTask = await Task.WhenAny(localTasks).ConfigureAwait(false);
-							var index = localTasks.IndexOf(completedTask);
-							var session = localSessions[index].Session;
-							var connection = localSessions[index].OwningConnection;
-							localSessions.RemoveAt(index);
-							localTasks.RemoveAt(index);
-							await session.Pool!.ReturnAsync(IOBehavior.Asynchronous, session).ConfigureAwait(false);
-							GC.KeepAlive(connection);
-						}
+						await Task.WhenAll(localTasks);
+						localTasks.Clear();
 					}
 				}
 				catch (Exception ex) when (!(ex is OperationCanceledException oce && oce.CancellationToken == s_cancellationTokenSource.Token))
@@ -124,25 +105,11 @@ public static async Task ReturnSessionsAsync()
 			}
 		}
 
-		internal readonly struct SessionResetTask
-		{
-			public SessionResetTask(ServerSession session, Task<bool> resetTask, MySqlConnection? owningConnection)
-			{
-				Session = session;
-				ResetTask = resetTask;
-				OwningConnection = owningConnection;
-			}
-
-			public ServerSession Session { get; }
-			public Task<bool> ResetTask { get; }
-			public MySqlConnection? OwningConnection { get; }
-		}
-
 		static readonly IMySqlConnectorLogger Log = MySqlConnectorLogManager.CreateLogger(nameof(BackgroundConnectionResetHelper));
 		static readonly object s_lock = new();
 		static readonly SemaphoreSlim s_semaphore = new(1, 1);
 		static readonly CancellationTokenSource s_cancellationTokenSource = new();
-		static readonly List<SessionResetTask> s_sessions = new();
+		static readonly List<Task<bool>> s_resetTasks = new();
 		static Task? s_workerTask;
 	}
 }

src/MySqlConnector/Core/ConnectionPool.cs

@@ -68,7 +68,7 @@ public async ValueTask<ServerSession> GetSessionAsync(MySqlConnection connection
 					{
 						if ((ConnectionSettings.ConnectionReset && ConnectionSettings.DeferConnectionReset) || session.DatabaseOverride is not null)
 						{
-							reuseSession = await session.TryResetConnectionAsync(ConnectionSettings, ioBehavior, cancellationToken).ConfigureAwait(false);
+							reuseSession = await session.TryResetConnectionAsync(ConnectionSettings, null, false, ioBehavior, cancellationToken).ConfigureAwait(false);
 						}
 						else if ((unchecked((uint) Environment.TickCount) - session.LastReturnedTicks) >= ConnectionSettings.ConnectionIdlePingTime)
 						{
@@ -158,7 +158,11 @@ private int GetSessionHealth(ServerSession session)
 			return 0;
 		}
 
-		public async Task ReturnAsync(IOBehavior ioBehavior, ServerSession session)
+#if NET45 || NET461 || NET471 || NETSTANDARD1_3 || NETSTANDARD2_0
+		public async ValueTask<int> ReturnAsync(IOBehavior ioBehavior, ServerSession session)
+#else
+		public async ValueTask ReturnAsync(IOBehavior ioBehavior, ServerSession session)
+#endif
 		{
 			if (Log.IsDebugEnabled())
 				Log.Debug("Pool{0} receiving Session{1} back", m_logArguments[0], session.Id);
@@ -188,6 +192,10 @@ public async Task ReturnAsync(IOBehavior ioBehavior, ServerSession session)
 			{
 				m_sessionSemaphore.Release();
 			}
+
+#if NET45 || NET461 || NET471 || NETSTANDARD1_3 || NETSTANDARD2_0
+			return default;
+#endif
 		}
 
 		public async Task ClearAsync(IOBehavior ioBehavior, CancellationToken cancellationToken)

src/MySqlConnector/Core/ServerSession.cs

@@ -68,7 +68,11 @@ public ServerSession(ConnectionPool? pool, int poolGeneration, int id)
 		public bool SupportsSessionTrack => m_supportsSessionTrack;
 		public bool ProcAccessDenied { get; set; }
 
-		public Task ReturnToPoolAsync(IOBehavior ioBehavior, MySqlConnection? owningConnection)
+#if NET45 || NET461 || NET471 || NETSTANDARD1_3 || NETSTANDARD2_0
+		public ValueTask<int> ReturnToPoolAsync(IOBehavior ioBehavior, MySqlConnection? owningConnection)
+#else
+		public ValueTask ReturnToPoolAsync(IOBehavior ioBehavior, MySqlConnection? owningConnection)
+#endif
 		{
 			if (Log.IsDebugEnabled())
 			{
@@ -77,11 +81,11 @@ public Task ReturnToPoolAsync(IOBehavior ioBehavior, MySqlConnection? owningConn
 			}
 			LastReturnedTicks = unchecked((uint) Environment.TickCount);
 			if (Pool is null)
-				return Utility.CompletedTask;
+				return default;
 			if (!Pool.ConnectionSettings.ConnectionReset || Pool.ConnectionSettings.DeferConnectionReset)
 				return Pool.ReturnAsync(ioBehavior, this);
 			BackgroundConnectionResetHelper.AddSession(this, owningConnection);
-			return Utility.CompletedTask;
+			return default;
 		}
 
 		public bool IsConnected
@@ -529,10 +533,11 @@ public async Task DisposeAsync(IOBehavior ioBehavior, CancellationToken cancella
 			return statusInfo;
 		}
 
-		public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, IOBehavior ioBehavior, CancellationToken cancellationToken)
+		public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, MySqlConnection? owningConnection, bool returnToPool, IOBehavior ioBehavior, CancellationToken cancellationToken)
 		{
 			VerifyState(State.Connected);
 
+			var success = false;
 			try
 			{
 				// clear all prepared statements; resetting the connection will clear them on the server
@@ -578,7 +583,7 @@ public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, IOBehavio
 				payload = await ReceiveReplyAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
 				OkPayload.Create(payload.Span, SupportsDeprecateEof, SupportsSessionTrack);
 
-				return true;
+				success = true;
 			}
 			catch (IOException ex)
 			{
@@ -597,7 +602,15 @@ public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, IOBehavio
 				Log.Debug(ex, "Session{0} ignoring SocketException in TryResetConnectionAsync", m_logArguments);
 			}
 
-			return false;
+			if (returnToPool && Pool is not null)
+			{
+				await Pool.ReturnAsync(ioBehavior, this).ConfigureAwait(false);
+
+				// make sure the MySqlConnection is kept alive until the session is returned to the pool; this prevents it from potentially being detected as "leaked"
+				GC.KeepAlive(owningConnection);
+			}
+
+			return success;
 		}
 
 		private async Task<PayloadData> SwitchAuthenticationAsync(ConnectionSettings cs, PayloadData payload, IOBehavior ioBehavior, CancellationToken cancellationToken)