Implement pipelining

docs/content/connection-options.md

@@ -1,5 +1,5 @@
 ---
-lastmod: 2021-02-06
+lastmod: 2021-11-25
 date: 2016-10-16
 title: Connection Options
 customtitle: MySQL Connection String for C# .NET Core Programs
@@ -390,6 +390,11 @@ These are the other options that MySqlConnector supports. They are set to sensib
     <td>false</td>
     <td>When set to <code>false</code> or no (strongly recommended), security-sensitive information, such as the password, is not returned as part of the connection string if the connection is open or has ever been in an open state. Resetting the connection string resets all connection string values, including the password. Recognized values are true, false, yes, and no.</td>
   </tr>
+  <tr id="Pipelining">
+    <td>Pipelining</td>
+    <td>true</td>
+    <td>When set to <code>true</code>, queries will be "pipelined" (when possible) by sending multiple packets to the server before waiting for a response. This improves performance (by reducing latency) but is not compatible with some servers (most notably Amazon Aurora RDS). Set to <code>false</code> to disable this behavior.</td>
+  </tr>
   <tr id="ServerRedirectionMode">
     <td>Server Redirection Mode, ServerRedirectionMode</td>
     <td>Disabled</td>

docs/content/troubleshooting/aurora-freeze.md

@@ -0,0 +1,21 @@
+---
+date: 2021-11-25
+title: Aurora Freeze
+customtitle: "MySqlConnection.Open Freezes with Amazon Aurora"
+weight: 5
+menu:
+  main:
+    parent: troubleshooting
+---
+
+# MySqlConnection.Open Freezes with Amazon Aurora
+
+When using Amazon Aurora RDS, you may experience a hang when calling `MySqlConnection.Open()`. The immediate prior log statement will be:
+
+```
+[TRACE]  ServerSession  Session0.0 ServerVersion=5.7.12 supports reset connection and pipelining; sending pipelined reset connection request
+```
+
+The cause of this problem is Amazon Aurora not correctly supporting pipelining in the MySQL protocol. This is known to be a problem with 2.x versions of Aurora (that implement MySQL 5.7), but not with 3.x versions (that implement MySQL 8.0).
+
+To work around it, add `Pipelining = False;` to your connection string to disable the pipelining feature.

src/MySqlConnector/Core/ConnectionPool.cs

@@ -62,7 +62,7 @@ public async ValueTask<ServerSession> GetSessionAsync(MySqlConnection connection
 				else
 				{
 					if (ConnectionSettings.ConnectionReset || session.DatabaseOverride is not null)
-						reuseSession = await session.TryResetConnectionAsync(ConnectionSettings, connection, false, ioBehavior, cancellationToken).ConfigureAwait(false);
+						reuseSession = await session.TryResetConnectionAsync(ConnectionSettings, connection, ioBehavior, cancellationToken).ConfigureAwait(false);
 					else
 						reuseSession = true;
 				}

src/MySqlConnector/Core/ConnectionSettings.cs

@@ -140,6 +140,7 @@ public ConnectionSettings(MySqlConnectionStringBuilder csb)
 		Keepalive = csb.Keepalive;
 		NoBackslashEscapes = csb.NoBackslashEscapes;
 		PersistSecurityInfo = csb.PersistSecurityInfo;
+		Pipelining = csb.ContainsKey("Pipelining") ? csb.Pipelining : default(bool?);
 		ServerRedirectionMode = csb.ServerRedirectionMode;
 		ServerRsaPublicKeyFile = csb.ServerRsaPublicKeyFile;
 		ServerSPN = csb.ServerSPN;
@@ -233,6 +234,7 @@ private static MySqlGuidFormat GetEffectiveGuidFormat(MySqlGuidFormat guidFormat
 	public uint Keepalive { get; }
 	public bool NoBackslashEscapes { get; }
 	public bool PersistSecurityInfo { get; }
+	public bool? Pipelining { get; }
 	public MySqlServerRedirectionMode ServerRedirectionMode { get; }
 	public string ServerRsaPublicKeyFile { get; }
 	public string ServerSPN { get; }
@@ -316,6 +318,7 @@ private ConnectionSettings(ConnectionSettings other, string host, int port, stri
 		Keepalive = other.Keepalive;
 		NoBackslashEscapes = other.NoBackslashEscapes;
 		PersistSecurityInfo = other.PersistSecurityInfo;
+		Pipelining = other.Pipelining;
 		ServerRedirectionMode = other.ServerRedirectionMode;
 		ServerRsaPublicKeyFile = other.ServerRsaPublicKeyFile;
 		ServerSPN = other.ServerSPN;

src/MySqlConnector/Core/ServerSession.cs

@@ -468,9 +468,36 @@ public async Task DisposeAsync(IOBehavior ioBehavior, CancellationToken cancella
 				m_characterSet = ServerVersion.Version >= ServerVersions.SupportsUtf8Mb4 ? CharacterSet.Utf8Mb4GeneralCaseInsensitive : CharacterSet.Utf8GeneralCaseInsensitive;
 				m_setNamesPayload = ServerVersion.Version >= ServerVersions.SupportsUtf8Mb4 ? s_setNamesUtf8mb4Payload : s_setNamesUtf8Payload;
 
-				Log.Debug("Session{0} made connection; ServerVersion={1}; ConnectionId={2}; Compression={3}; Attributes={4}; DeprecateEof={5}; Ssl={6}; SessionTrack={7}",
+				// disable pipelining for RDS MySQL 5.7 (assuming Aurora); otherwise take it from the connection string or default to true
+				if (!cs.Pipelining.HasValue && ServerVersion.Version.Major == 5 && ServerVersion.Version.Minor == 7 && HostName.EndsWith(".rds.amazonaws.com", StringComparison.OrdinalIgnoreCase))
+				{
+					m_logArguments[1] = HostName;
+					Log.Debug("Session{0} auto-detected Aurora 5.7 at '{1}'; disabling pipelining", m_logArguments);
+					m_supportsPipelining = false;
+				}
+				else
+				{
+					// pipelining is not currently compatible with compression
+					m_supportsPipelining = !cs.UseCompression && (cs.Pipelining ?? true);
+
+					// for pipelining, concatenate reset connection and SET NAMES query into one buffer
+					if (m_supportsPipelining)
+					{
+						m_pipelinedResetConnectionBytes = new byte[m_setNamesPayload.Span.Length + 9];
+
+						// first packet: reset connection
+						m_pipelinedResetConnectionBytes[0] = 1;
+						m_pipelinedResetConnectionBytes[4] = (byte) CommandKind.ResetConnection;
+
+						// second packet: SET NAMES query
+						m_pipelinedResetConnectionBytes[5] = (byte) m_setNamesPayload.Span.Length;
+						m_setNamesPayload.Span.CopyTo(m_pipelinedResetConnectionBytes.AsSpan().Slice(9));
+					}
+				}
+
+				Log.Debug("Session{0} made connection; ServerVersion={1}; ConnectionId={2}; Compression={3}; Attributes={4}; DeprecateEof={5}; Ssl={6}; SessionTrack={7}; Pipelining={8}",
 					m_logArguments[0], ServerVersion.OriginalString, ConnectionId,
-					m_useCompression, m_supportsConnectionAttributes, m_supportsDeprecateEof, serverSupportsSsl, m_supportsSessionTrack);
+					m_useCompression, m_supportsConnectionAttributes, m_supportsDeprecateEof, serverSupportsSsl, m_supportsSessionTrack, m_supportsPipelining);
 
 				if (cs.SslMode != MySqlSslMode.None && (cs.SslMode != MySqlSslMode.Preferred || serverSupportsSsl))
 				{
@@ -555,11 +582,10 @@ public async Task DisposeAsync(IOBehavior ioBehavior, CancellationToken cancella
 		return statusInfo;
 	}
 
-	public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, MySqlConnection connection, bool returnToPool, IOBehavior ioBehavior, CancellationToken cancellationToken)
+	public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, MySqlConnection connection, IOBehavior ioBehavior, CancellationToken cancellationToken)
 	{
 		VerifyState(State.Connected);
 
-		var success = false;
 		try
 		{
 			// clear all prepared statements; resetting the connection will clear them on the server
@@ -569,6 +595,26 @@ public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, MySqlConn
 			if (DatabaseOverride is null && (ServerVersion.Version.CompareTo(ServerVersions.SupportsResetConnection) >= 0 || ServerVersion.MariaDbVersion?.CompareTo(ServerVersions.MariaDbSupportsResetConnection) >= 0))
 			{
 				m_logArguments[1] = ServerVersion.OriginalString;
+
+				if (m_supportsPipelining)
+				{
+					Log.Trace("Session{0} ServerVersion={1} supports reset connection and pipelining; sending pipelined reset connection request", m_logArguments);
+
+					// send both packets at once
+					await m_payloadHandler!.ByteHandler.WriteBytesAsync(m_pipelinedResetConnectionBytes!, ioBehavior).ConfigureAwait(false);
+
+					// read two OK replies
+					m_payloadHandler.SetNextSequenceNumber(1);
+					payload = await ReceiveReplyAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
+					OkPayload.Create(payload.Span, SupportsDeprecateEof, SupportsSessionTrack);
+
+					m_payloadHandler.SetNextSequenceNumber(1);
+					payload = await ReceiveReplyAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
+					OkPayload.Create(payload.Span, SupportsDeprecateEof, SupportsSessionTrack);
+
+					return true;
+				}
+
 				Log.Trace("Session{0} ServerVersion={1} supports reset connection; sending reset connection request", m_logArguments);
 				await SendAsync(ResetConnectionPayload.Instance, ioBehavior, cancellationToken).ConfigureAwait(false);
 				payload = await ReceiveReplyAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
@@ -606,7 +652,7 @@ public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, MySqlConn
 			payload = await ReceiveReplyAsync(ioBehavior, cancellationToken).ConfigureAwait(false);
 			OkPayload.Create(payload.Span, SupportsDeprecateEof, SupportsSessionTrack);
 
-			success = true;
+			return true;
 		}
 		catch (IOException ex)
 		{
@@ -625,12 +671,7 @@ public async Task<bool> TryResetConnectionAsync(ConnectionSettings cs, MySqlConn
 			Log.Trace(ex, "Session{0} ignoring SocketException in TryResetConnectionAsync", m_logArguments);
 		}
 
-		if (returnToPool && Pool is not null)
-		{
-			await Pool.ReturnAsync(ioBehavior, this).ConfigureAwait(false);
-		}
-
-		return success;
+		return false;
 	}
 
 	private async Task<PayloadData> SwitchAuthenticationAsync(ConnectionSettings cs, string password, PayloadData payload, IOBehavior ioBehavior, CancellationToken cancellationToken)
@@ -1888,7 +1929,9 @@ protected override void OnStatementBegin(int index)
 	bool m_supportsConnectionAttributes;
 	bool m_supportsDeprecateEof;
 	bool m_supportsSessionTrack;
+	bool m_supportsPipelining;
 	CharacterSet m_characterSet;
 	PayloadData m_setNamesPayload;
+	byte[]? m_pipelinedResetConnectionBytes;
 	Dictionary<string, PreparedStatements>? m_preparedStatements;
 }

src/MySqlConnector/MySqlConnectionStringBuilder.cs

@@ -678,8 +678,21 @@ public bool PersistSecurityInfo
 	}
 
 	/// <summary>
-	/// Whether to use server redirection.
+	/// Enables query pipelining.
 	/// </summary>
+	[Category("Other")]
+	[DefaultValue(true)]
+	[Description("Enables query pipelining.")]
+	[DisplayName("Pipelining")]
+	public bool Pipelining
+	{
+		get => MySqlConnectionStringOption.Pipelining.GetValue(this);
+		set => MySqlConnectionStringOption.Pipelining.SetValue(this, value);
+	}
+
+	/// <summary>
+		/// Whether to use server redirection.
+		/// </summary>
 	[Category("Connection")]
 	[DefaultValue(MySqlServerRedirectionMode.Disabled)]
 	[Description("Whether to use server redirection.")]
@@ -911,6 +924,7 @@ internal abstract class MySqlConnectionStringOption
 	public static readonly MySqlConnectionStringValueOption<bool> NoBackslashEscapes;
 	public static readonly MySqlConnectionStringValueOption<bool> OldGuids;
 	public static readonly MySqlConnectionStringValueOption<bool> PersistSecurityInfo;
+	public static readonly MySqlConnectionStringValueOption<bool> Pipelining;
 	public static readonly MySqlConnectionStringValueOption<MySqlServerRedirectionMode> ServerRedirectionMode;
 	public static readonly MySqlConnectionStringReferenceOption<string> ServerRsaPublicKeyFile;
 	public static readonly MySqlConnectionStringReferenceOption<string> ServerSPN;
@@ -1178,6 +1192,10 @@ static MySqlConnectionStringOption()
 			keys: new[] { "Persist Security Info", "PersistSecurityInfo" },
 			defaultValue: false));
 
+		AddOption(Pipelining = new(
+			keys: new[] { "Pipelining" },
+			defaultValue: true));
+
 		AddOption(ServerRedirectionMode = new(
 			keys: new[] { "Server Redirection Mode", "ServerRedirectionMode" },
 			defaultValue: MySqlServerRedirectionMode.Disabled));

src/MySqlConnector/Protocol/Serialization/CompressedPayloadHandler.cs

@@ -21,11 +21,11 @@ public void Dispose()
 		Utility.Dispose(ref m_uncompressedStream);
 	}
 
-	public void StartNewConversation()
-	{
-		m_compressedSequenceNumber = 0;
-		m_uncompressedSequenceNumber = 0;
-	}
+	public void StartNewConversation() =>
+		m_compressedSequenceNumber = m_uncompressedSequenceNumber = 0;
+
+	public void SetNextSequenceNumber(int sequenceNumber) =>
+		throw new NotSupportedException();
 
 	public IByteHandler ByteHandler
 	{

src/MySqlConnector/Protocol/Serialization/IPayloadHandler.cs

@@ -8,6 +8,13 @@ internal interface IPayloadHandler : IDisposable
 	/// </summary>
 	void StartNewConversation();
 
+	/// <summary>
+	/// Forces the next sequence number to be the specified value.
+	/// </summary>
+	/// <param name="sequenceNumber">The next sequence number.</param>
+	/// <remarks>This should only be used in advanced scenarios.</remarks>
+	void SetNextSequenceNumber(int sequenceNumber);
+
 	/// <summary>
 	/// Gets or sets the underlying <see cref="IByteHandler"/> that data is read from and written to.
 	/// </summary>

src/MySqlConnector/Protocol/Serialization/StandardPayloadHandler.cs

@@ -16,10 +16,11 @@ public void Dispose()
 		Utility.Dispose(ref m_byteHandler);
 	}
 
-	public void StartNewConversation()
-	{
+	public void StartNewConversation() =>
 		m_sequenceNumber = 0;
-	}
+
+	public void SetNextSequenceNumber(int sequenceNumber) =>
+		m_sequenceNumber = (byte) sequenceNumber;
 
 	public IByteHandler ByteHandler
 	{

tests/MySqlConnector.Tests/MySqlConnectionStringBuilderTests.cs

@@ -71,6 +71,9 @@ public void Defaults()
 #endif
 		Assert.False(csb.OldGuids);
 		Assert.False(csb.PersistSecurityInfo);
+#if !BASELINE
+		Assert.True(csb.Pipelining);
+#endif
 		Assert.True(csb.Pooling);
 		Assert.Equal(3306u, csb.Port);
 		Assert.Equal("", csb.Server);
@@ -135,6 +138,7 @@ public void ParseConnectionString()
 				"load balance=random;" +
 				"guidformat=timeswapbinary16;" +
 				"nobackslashescapes=true;" +
+				"pipelining=false;" +
 				"server redirection mode=required;" +
 				"server spn=mariadb/host.example.com@EXAMPLE.COM;" +
 				"use xa transactions=false;" +
@@ -197,6 +201,7 @@ public void ParseConnectionString()
 		Assert.Equal(MySqlLoadBalance.Random, csb.LoadBalance);
 		Assert.Equal(MySqlGuidFormat.TimeSwapBinary16, csb.GuidFormat);
 		Assert.True(csb.NoBackslashEscapes);
+		Assert.False(csb.Pipelining);
 		Assert.Equal(MySqlServerRedirectionMode.Required, csb.ServerRedirectionMode);
 		Assert.Equal("mariadb/host.example.com@EXAMPLE.COM", csb.ServerSPN);
 		Assert.False(csb.UseXaTransactions);