Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add scaffolding for pool throttling #2667

Merged
merged 14 commits into from
Jul 15, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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" />
Expand Down Expand Up @@ -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)" />
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand All @@ -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
}

Expand All @@ -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/>
Expand All @@ -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
Original file line number Diff line number Diff line change
@@ -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);
}
Original file line number Diff line number Diff line change
@@ -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();
}
}
}
Original file line number Diff line number Diff line change
@@ -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();
}
}
}
Original file line number Diff line number Diff line change
@@ -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();
}
}
}
Original file line number Diff line number Diff line change
@@ -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();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -40,14 +36,10 @@ protected override SqlConnectionX CreateDbConnection()

internal SqlDataSource(
SqlConnectionStringBuilder connectionStringBuilder,
SqlCredential credential,
RemoteCertificateValidationCallback userCertificateValidationCallback,
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

removing these for now until we fully implement additional auth pathways

Action<X509CertificateCollection> clientCertificatesCallback)
SqlCredential credential)
{
_connectionStringBuilder = connectionStringBuilder;
Credential = credential;
_userCertificateValidationCallback = userCertificateValidationCallback;
_clientCertificatesCallback = clientCertificatesCallback;
}

/// <inheritdoc />
Expand Down
Loading