Add scaffolding for pool throttling (#2667)

* Add basic rate limiters

* Add license headers. Clean up usings.

* Add docs.

* Instantiate pooling data source via builder. Set rate limiter based on connection string options.

* Compilation fixes. Remove auth params for now.

* Use delegate to better capture callback signature.

* Use state type to remove closed params. Clean up docs.

* Implement review suggestions.

* Implement dispose.

* Dispose next rate limiter.

* Dispose rate limiter on pool shutdown.

* Address review nits.

src/Microsoft.Data.SqlClient/netcore/src/Microsoft.Data.SqlClient.csproj

@@ -954,6 +954,11 @@
     <Compile Include="Microsoft\Data\SqlClientX\IO\TdsWriter.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\IO\TdsWriteStream.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\PoolingDataSource.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\AsyncFlagFunc.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\BlockingPeriodRateLimiter.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\ConcurrencyRateLimiter.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\RateLimiterBase.cs" />
+    <Compile Include="Microsoft\Data\SqlClientX\RateLimiters\PassthroughRateLimiter.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\SqlConnectionX.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\SqlConnector.cs" />
     <Compile Include="Microsoft\Data\SqlClientX\SqlDataSource.cs" />
@@ -982,7 +987,7 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="Microsoft.Data.SqlClient.SNI.runtime" Version="$(MicrosoftDataSqlClientSNIRuntimeVersion)" />
-    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="$(MicrosoftExtensionsCachingMemoryVersion)"  />
+    <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="$(MicrosoftExtensionsCachingMemoryVersion)" />
     <!-- Enable the project reference for debugging purposes. -->
     <!-- <ProjectReference Include="$(SqlServerSourceCode)\Microsoft.SqlServer.Server.csproj" /> -->
     <PackageReference Include="Microsoft.SqlServer.Server" Version="$(MicrosoftSqlServerServerVersion)" />

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/PoolingDataSource.cs

@@ -13,6 +13,7 @@
 using System.Threading.Tasks;
 using Microsoft.Data.ProviderBase;
 using Microsoft.Data.SqlClient;
+using Microsoft.Data.SqlClientX.RateLimiters;
 
 namespace Microsoft.Data.SqlClientX
 {
@@ -21,22 +22,29 @@ namespace Microsoft.Data.SqlClientX
     /// </summary>
     internal sealed class PoolingDataSource : SqlDataSource
     {
-        private DbConnectionPoolGroupOptions _connectionPoolGroupOptions;
+        private readonly DbConnectionPoolGroupOptions _connectionPoolGroupOptions;
+        private RateLimiterBase _connectionRateLimiter;
 
         internal int MinPoolSize => _connectionPoolGroupOptions.MinPoolSize;
         internal int MaxPoolSize => _connectionPoolGroupOptions.MaxPoolSize;
 
+        internal int ObjectID => _objectID;
+
+        private static int _objectTypeCount; // EventSource counter
+        private readonly int _objectID = Interlocked.Increment(ref _objectTypeCount);
+
         /// <summary>
         /// Initializes a new PoolingDataSource.
         /// </summary>
         //TODO: support auth contexts and provider info
-        PoolingDataSource(SqlConnectionStringBuilder connectionStringBuilder,
+        internal PoolingDataSource(SqlConnectionStringBuilder connectionStringBuilder,
             SqlCredential credential,
-            RemoteCertificateValidationCallback userCertificateValidationCallback,
-            Action<X509CertificateCollection> clientCertificatesCallback,
-            DbConnectionPoolGroupOptions options) : base(connectionStringBuilder, credential, userCertificateValidationCallback, clientCertificatesCallback)
+            DbConnectionPoolGroupOptions options,
+            RateLimiterBase connectionRateLimiter)
+            : base(connectionStringBuilder, credential)
         {
             _connectionPoolGroupOptions = options;
+            _connectionRateLimiter = connectionRateLimiter;
             //TODO: other construction
         }
 
@@ -46,10 +54,32 @@ internal override ValueTask<SqlConnector> GetInternalConnection(SqlConnectionX o
             throw new NotImplementedException();
         }
 
+        internal readonly struct OpenInternalConnectionState
+        {
+            readonly SqlConnectionX _owningConnection;
+            readonly TimeSpan _timeout;
+
+            internal OpenInternalConnectionState(SqlConnectionX owningConnection, TimeSpan timeout)
+            {
+                _owningConnection = owningConnection;
+                _timeout = timeout;
+            }
+        }
+
         /// <inheritdoc/>
         internal override ValueTask<SqlConnector> OpenNewInternalConnection(SqlConnectionX owningConnection, TimeSpan timeout, bool async, CancellationToken cancellationToken)
         {
-            throw new NotImplementedException();
+            return _connectionRateLimiter.Execute<OpenInternalConnectionState, SqlConnector>(
+                RateLimitedOpen,
+                new OpenInternalConnectionState(owningConnection, timeout),
+                async,
+                cancellationToken
+            );
+
+            ValueTask<SqlConnector> RateLimitedOpen(OpenInternalConnectionState state, bool async, CancellationToken cancellationToken)
+            {
+                throw new NotImplementedException();
+            }
         }
 
         /// <inheritdoc/>
@@ -75,7 +105,16 @@ internal void WarmUp()
         {
             throw new NotImplementedException();
         }
+
+        internal void Shutdown()
+        {
+            SqlClientEventSource.Log.TryPoolerTraceEvent("<prov.DbConnectionPool.Shutdown|RES|INFO|CPOOL> {0}", ObjectID);
+            if (_connectionRateLimiter != null) {
+                _connectionRateLimiter.Dispose();
+                _connectionRateLimiter = null;
+            }
+        }
     }
 }
 
-#endif
+#endif

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/AsyncFlagFunc.cs

@@ -0,0 +1,20 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Threading;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A function that operates asynchronously based on a flag. If isAsync is true, the function operates asynchronously.
+    /// If isAsync is false, the function operates synchronously.
+    /// </summary>
+    /// <typeparam name="TState">The type accepted by the callback as input.</typeparam>
+    /// <typeparam name="TResult">The type returned by the callback.</typeparam>
+    /// <param name="state">An instance of State to be passed to the callback.</param>
+    /// <param name="isAsync">Indicates whether the function should operate asynchronously.</param>
+    /// <param name="cancellationToken">Allows cancellation of the operation.</param>
+    /// <returns>Returns the result of the callback.</returns>
+    internal delegate TResult AsyncFlagFunc<in TState, out TResult>(TState state, bool isAsync, CancellationToken cancellationToken);
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/BlockingPeriodRateLimiter.cs

@@ -0,0 +1,34 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+#nullable enable
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A rate limiter that enforces a backoff (blocking) period upon error. 
+    /// Each subsequent error increases the blocking duration, up to a maximum, until a success occurs.
+    /// </summary>
+    internal sealed class BlockingPeriodRateLimiter : RateLimiterBase
+    {
+        /// <inheritdoc/>
+        internal override ValueTask<TResult> Execute<State, TResult>(
+            AsyncFlagFunc<State, ValueTask<TResult>> callback, 
+            State state, 
+            bool isAsync, 
+            CancellationToken cancellationToken = default)
+        {
+            throw new NotImplementedException();
+        }
+
+        public override void Dispose()
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/ConcurrencyRateLimiter.cs

@@ -0,0 +1,67 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A rate limiter that enforces a concurrency limit. 
+    /// When the limit is reached, new requests must wait until a spot is freed upon completion of an existing request.
+    /// </summary>
+    internal class ConcurrencyRateLimiter : RateLimiterBase
+    {
+        private readonly SemaphoreSlim _concurrencyLimitSemaphore;
+
+        /// <summary>
+        /// Initializes a new ConcurrencyRateLimiter with the specified concurrency limit.
+        /// </summary>
+        /// <param name="concurrencyLimit">The maximum number of concurrent requests.</param>
+        internal ConcurrencyRateLimiter(int concurrencyLimit)
+        {
+            _concurrencyLimitSemaphore = new SemaphoreSlim(concurrencyLimit);
+        }
+
+        /// <inheritdoc/>
+        internal sealed override async ValueTask<TResult> Execute<State, TResult>(AsyncFlagFunc<State, ValueTask<TResult>> callback, State state, bool isAsync, CancellationToken cancellationToken = default)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            //TODO: in the future, we can enforce order
+            if (isAsync)
+            {
+                await _concurrencyLimitSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+            }
+            else
+            {
+                _concurrencyLimitSemaphore.Wait(cancellationToken);
+            }
+
+            try
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                if (Next != null)
+                {
+                    return await Next.Execute(callback, state, isAsync, cancellationToken).ConfigureAwait(false);
+                }
+                else
+                {
+                    return await callback(state, isAsync, cancellationToken).ConfigureAwait(false);
+                }
+            }
+            finally
+            {
+                _concurrencyLimitSemaphore.Release();
+            }
+        }
+
+        public override void Dispose()
+        {
+            _concurrencyLimitSemaphore.Dispose();
+            Next.Dispose();
+        }
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/PassthroughRateLimiter.cs

@@ -0,0 +1,39 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// A no-op rate limiter that simply executes the callback or passes through to the next rate limiter.
+    /// </summary>
+    internal sealed class PassthroughRateLimiter : RateLimiterBase
+    {
+        //TODO: no state, add static instance
+
+        /// <inheritdoc/>
+        internal override ValueTask<TResult> Execute<State, TResult>(
+            AsyncFlagFunc<State, ValueTask<TResult>> callback,
+            State state,
+            bool isAsync,
+            CancellationToken cancellationToken = default)
+        {
+            if (Next != null)
+            {
+                return Next.Execute(callback, state, isAsync, cancellationToken);
+            }
+            else
+            {
+                return callback(state, isAsync, cancellationToken);
+            }
+        }
+
+        public override void Dispose()
+        {
+            Next.Dispose();
+        }
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/RateLimiters/RateLimiterBase.cs

@@ -0,0 +1,40 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Data.SqlClientX.RateLimiters
+{
+    /// <summary>
+    /// An interface for rate limiters that execute arbitraty code. Intended to be small and self contained and chained together to achieve more complex behavior.
+    /// </summary>
+    internal abstract class RateLimiterBase : IDisposable
+    {
+
+        /// <summary>
+        /// The next rate limiter that should be executed within the context of this rate limiter.
+        /// </summary>
+        protected readonly RateLimiterBase Next;
+
+        /// <summary>
+        /// Execute the provided callback within the context of the rate limit, or pass the responsibility along to the next rate limiter.
+        /// </summary>
+        /// <typeparam name="State">The type accepted by the callback as input.</typeparam>
+        /// <typeparam name="TResult">The type of the result returned by the callback.</typeparam>
+        /// <param name="callback">The callback function to execute.</param>
+        /// <param name="state">An instance of State to be passed to the callback.</param>
+        /// <param name="isAsync">Whether this method should run asynchronously.</param>
+        /// <param name="cancellationToken">Cancels outstanding requests.</param>
+        /// <returns>Returns the result of the callback or the next rate limiter.</returns>
+        internal abstract ValueTask<TResult> Execute<State, TResult>(
+            AsyncFlagFunc<State, ValueTask<TResult>> callback,
+            State state,
+            bool isAsync,
+            CancellationToken cancellationToken = default);
+
+        public abstract void Dispose();
+    }
+}

src/Microsoft.Data.SqlClient/netcore/src/Microsoft/Data/SqlClientX/SqlDataSource.cs

@@ -22,10 +22,6 @@ internal abstract class SqlDataSource : DbDataSource
     {
         private readonly SqlConnectionStringBuilder _connectionStringBuilder;
 
-        private readonly RemoteCertificateValidationCallback _userCertificateValidationCallback;
-
-        private readonly Action<X509CertificateCollection> _clientCertificatesCallback;
-
         internal SqlCredential Credential { get; }
 
         //TODO: return SqlConnection after it is updated to wrap SqlConnectionX 
@@ -40,14 +36,10 @@ protected override SqlConnectionX CreateDbConnection()
 
         internal SqlDataSource(
             SqlConnectionStringBuilder connectionStringBuilder,
-            SqlCredential credential,
-            RemoteCertificateValidationCallback userCertificateValidationCallback,
-            Action<X509CertificateCollection> clientCertificatesCallback)
+            SqlCredential credential)
         {
             _connectionStringBuilder = connectionStringBuilder;
             Credential = credential;
-            _userCertificateValidationCallback = userCertificateValidationCallback;
-            _clientCertificatesCallback = clientCertificatesCallback;
         }
 
         /// <inheritdoc />