diff --git a/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/ChannelDbConnectionPool.cs b/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/ChannelDbConnectionPool.cs
index f5d758ebb7..322fc568c2 100644
--- a/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/ChannelDbConnectionPool.cs
+++ b/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/ChannelDbConnectionPool.cs
@@ -6,6 +6,7 @@
using System.Collections.ObjectModel;
using System.Data.Common;
using System.Runtime.CompilerServices;
+using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
@@ -71,11 +72,35 @@ internal sealed class ChannelDbConnectionPool : IDbConnectionPool, IDisposable
private readonly ConnectionPoolSlots _connectionSlots;
///
- /// The idle connection channel. Contains nulls in order to release waiting attempts after
- /// a connection has been physically closed/broken. Also tracks the count of non-null idle connections.
+ /// The idle/creation channel. Every connection request queues here (there is no optimistic
+ /// fast path around it), which gives strict FIFO ordering to parked waiters. It carries
+ /// values: a connection (idle or freshly pumped), a captured
+ /// creation error to rethrow on a waiter, or a bare wake ("capacity changed, re-evaluate").
+ /// Also tracks the count of connection-bearing outcomes (idle connections).
///
private readonly IdleConnectionChannel _idleChannel;
+ ///
+ /// The number of requests currently parked waiting for an outcome on .
+ /// Incremented immediately before a request pumps and parks, and decremented once its wait
+ /// completes. The pump reads this to decide whether there is unmet demand worth creating a
+ /// connection for, and slot-freeing events read it to decide whether a bare wake is needed.
+ ///
+ /// Ordering hinge: a waiter increments this before it evaluates pool capacity and
+ /// parks. Combined with the fact that a slot-freeing event frees its slot before it reads
+ /// this counter, that guarantees no lost wake-up: either the waiter observes the freed slot
+ /// and creates a connection itself, or the freeing event observes the waiter and wakes it.
+ /// Updated via .
+ ///
+ private int _waiterCount;
+
+ ///
+ /// The number of background create ("pump") tasks currently in flight. Used by the pump's
+ /// demand/capacity gate so it launches at most one create per unit of unmet demand and never
+ /// over-commits past . Updated via .
+ ///
+ private int _pendingCreates;
+
///
/// The current generation of the pool. Incremented atomically on each call.
/// Connections stamped with a generation that does not match are considered stale and are destroyed
@@ -212,11 +237,11 @@ public void Clear()
// Any connections from a previous generation that are returned to the pool
// after we start draining will fail the _clearCounter comparison and will be closed.
int numToDrain = IdleCount;
- while (numToDrain > 0 && _idleChannel.TryRead(out DbConnectionInternal? connection))
+ while (numToDrain > 0 && _idleChannel.TryRead(out CreateOutcome outcome))
{
- if (connection is not null)
+ if (outcome.Connection is not null)
{
- RemoveConnection(connection);
+ RemoveConnection(outcome.Connection);
numToDrain--;
}
}
@@ -282,7 +307,7 @@ public void ReturnInternalConnection(DbConnectionInternal connection, DbConnecti
}
else
{
- if (!_idleChannel.TryWrite(connection))
+ if (!_idleChannel.TryWrite(new CreateOutcome(connection)))
{
// The channel has been completed (pool is shutting down). Race window
// between the State check above and TryWrite: destroy instead of pooling.
@@ -348,11 +373,12 @@ public void Shutdown()
// Clear() may short-circuit if another caller is already draining. Because the
// channel is now completed, no new items can be enqueued, so it is safe to do a
// final unbounded drain to mop up anything Clear() may have skipped.
- while (_idleChannel.TryRead(out DbConnectionInternal? connection))
+ while (_idleChannel.TryRead(out CreateOutcome outcome))
{
+ DbConnectionInternal? connection = outcome.Connection;
if (connection is null)
{
- // null sentinels are wake-up signals only; nothing to destroy.
+ // Bare-wake and error outcomes carry no connection; nothing to destroy.
continue;
}
@@ -543,9 +569,9 @@ public bool TryGetConnection(
},
cleanupCallback: (newConnection) =>
{
- // If we fail to open a connection, we need to write a null to the idle channel to
- // wake up any waiters
- _idleChannel?.TryWrite(null);
+ // Creation failed or produced no slot. Error propagation and re-driving of
+ // remaining waiters are handled by the pump task (LaunchCreate); here we only
+ // dispose whatever partial connection may have been produced.
newConnection?.Dispose();
});
@@ -611,10 +637,18 @@ private void RemoveConnection(DbConnectionInternal connection)
{
_connectionSlots.TryRemove(connection);
- // Removing a connection from the pool opens a free slot.
- // Write a null to the idle connection channel to wake up a waiter, who can now open a new
- // connection. Statement order is important since we have synchronous completions on the channel.
- _idleChannel.TryWrite(null);
+ // Removing a connection frees a slot. If a request is currently parked waiting for a
+ // connection, wake one so it can re-pump and create a replacement using its own owning
+ // connection. The bare-wake outcome carries neither a connection nor an error.
+ //
+ // The wake is skipped when no request is waiting: there is no one to notify, and the
+ // freed slot will be observed by the next request's own pump. This also prevents stale
+ // wakes from accumulating in the channel. See _waiterCount for the ordering argument
+ // that guarantees a genuinely-parked waiter is never missed.
+ if (Volatile.Read(ref _waiterCount) > 0)
+ {
+ _idleChannel.TryWrite(default);
+ }
connection.Dispose();
@@ -623,38 +657,124 @@ private void RemoveConnection(DbConnectionInternal connection)
}
///
- /// Tries to read a connection from the idle connection channel.
+ /// The pump: signals demand and, when there is unmet demand and free capacity, launches
+ /// background create tasks to grow the pool. Called by a request that is about to park (so
+ /// its demand is reflected in first) and whenever a create task
+ /// completes and frees its accounting.
+ ///
+ /// The gate is a cheap synchronous check, so most calls are near-free. Each launched task
+ /// reserves a slot and opens one physical connection, so multiple waiters are served in
+ /// parallel (bounded by and, once rate limiting lands, by the
+ /// limiter inside the create task).
///
- /// A connection from the idle channel, or null if the channel is empty.
- private DbConnectionInternal? GetIdleConnection()
+ /// A live requesting connection whose creation options seed the
+ /// physical open. Every pump is triggered by a request that is actively waiting, so this is
+ /// always a real, in-use connection.
+ private void TryPumpCreate(DbConnection owningConnection)
{
- // The channel may contain nulls. Read until we find a non-null connection or exhaust the channel.
- while (_idleChannel.TryRead(out DbConnectionInternal? connection))
+ while (true)
{
- if (connection is null)
+ int waiters = Volatile.Read(ref _waiterCount);
+ int pending = Volatile.Read(ref _pendingCreates);
+
+ // Demand already covered by in-flight creates plus connections sitting in the
+ // channel? Then nothing to do.
+ if (pending + _idleChannel.Count >= waiters)
{
- continue;
+ return;
}
- if (!IsLiveConnection(connection))
+ // Any capacity to create? ReservationCount counts slots already held by live
+ // connections; adding pending creates (each of which will reserve a slot) gives the
+ // projected occupancy. This is advisory - ConnectionPoolSlots.Add re-checks
+ // atomically - but it prevents a busy loop of no-slot creates when the pool is full.
+ if (_connectionSlots.ReservationCount + pending >= MaxPoolSize)
{
- RemoveConnection(connection);
- continue;
+ return;
}
- return connection;
+ // Reserve a create optimistically, then launch it. If we lose a race the create
+ // task will find no slot and no-op, which is harmless.
+ Interlocked.Increment(ref _pendingCreates);
+ LaunchCreate(owningConnection);
}
+ }
- return null;
+ ///
+ /// Runs a single background connection create and publishes the result to the channel for
+ /// whichever waiter is at the FIFO head:
+ ///
+ /// - success: the connection is written as a connection outcome;
+ /// - failure: the captured error is written as an error outcome so a waiter
+ /// rethrows it (one failure is consumed by one waiter);
+ /// - no slot available: nothing is written (capacity is unchanged).
+ ///
+ ///
+ private void LaunchCreate(DbConnection owningConnection)
+ {
+ _ = Task.Run(() =>
+ {
+ try
+ {
+ // Creation is decoupled from any specific caller, so it uses the pool's
+ // configured CreationTimeout for the physical open rather than a caller's
+ // remaining budget. A caller's own ConnectTimeout governs only how long it waits
+ // on the channel. CreationTimeout of 0 maps to an infinite timer (TimeSpan.Zero
+ // has zero ticks, which TimeoutTimer treats as infinite).
+ TimeoutTimer timeout = TimeoutTimer.StartNew(
+ TimeSpan.FromMilliseconds(PoolGroupOptions.CreationTimeout));
+
+ DbConnectionInternal? connection =
+ OpenNewInternalConnection(owningConnection, CancellationToken.None, timeout);
+
+ if (connection is not null && !_idleChannel.TryWrite(new CreateOutcome(connection)))
+ {
+ // The channel was completed (pool shutting down) between creation and
+ // publish. Destroy the orphan rather than leaking it.
+ RemoveConnection(connection);
+ }
+ }
+ catch (Exception ex)
+ {
+ // Deliver the error to a waiter so the originating request surface sees a real
+ // failure instead of only timing out. The linger guard avoids leaving a stale
+ // error in the channel for an unrelated future caller when nobody is waiting.
+ if (Volatile.Read(ref _waiterCount) > 0)
+ {
+ _idleChannel.TryWrite(new CreateOutcome(ExceptionDispatchInfo.Capture(ex)));
+ }
+ else
+ {
+ SqlClientEventSource.Log.TryPoolerTraceEvent(
+ " {0}, create failed with no waiter to receive it: {1}", Id, ex);
+ }
+ }
+ finally
+ {
+ Interlocked.Decrement(ref _pendingCreates);
+
+ // Re-drive remaining waiters. A failed create frees its slot and a no-slot
+ // create may run once capacity opens, so nudge a parked waiter to re-evaluate.
+ // Guarded so stale wakes don't accumulate when nobody is waiting.
+ if (Volatile.Read(ref _waiterCount) > 0)
+ {
+ _idleChannel.TryWrite(default);
+ }
+ }
+ });
}
///
- /// Gets an internal connection from the pool, either by retrieving an idle connection or opening a new one.
+ /// Gets an internal connection from the pool. Every request queues on the idle/creation
+ /// channel - there is no optimistic fast path around it - which gives strict FIFO ordering
+ /// to parked waiters. A request signals demand via the pump, which creates connections on
+ /// background tasks as capacity allows, then blocks on the channel until an outcome arrives.
///
- /// The DbConnection that will own this internal connection
+ /// The DbConnection that will own this internal connection.
/// A boolean indicating whether the operation should be asynchronous.
- /// The overall timeout budget for this connection request. Time spent waiting
- /// in the pool is deducted from the budget available for physical connection creation.
+ /// The overall timeout budget for this request. It bounds only how long
+ /// the request waits on the channel; physical connection creation uses the pool's
+ /// CreationTimeout instead.
/// Returns a DbConnectionInternal that is retrieved from the pool.
///
/// Thrown when an OperationCanceledException is caught, indicating that the timeout period
@@ -662,7 +782,7 @@ private void RemoveConnection(DbConnectionInternal connection)
///
///
/// Thrown when a ChannelClosedException is caught, indicating that the connection pool
- /// has been shut down.
+ /// has been shut down. Also rethrows a background creation error delivered to this waiter.
///
private async Task GetInternalConnection(
DbConnection owningConnection,
@@ -676,46 +796,58 @@ private async Task GetInternalConnection(
using CancellationTokenSource cancellationTokenSource = timeout.CreateCancellationTokenSource();
CancellationToken cancellationToken = cancellationTokenSource.Token;
- // Continue looping until we create or retrieve a connection
+ // Continue looping until we retrieve a live connection.
do
{
- try
- {
- // Optimistically try to get an idle connection from the channel
- // Doesn't wait if the channel is empty, just returns null.
- connection ??= GetIdleConnection();
-
+ CreateOutcome outcome;
- // If we didn't find an idle connection, try to open a new one.
- connection ??= OpenNewInternalConnection(
- owningConnection,
- cancellationToken,
- timeout);
-
- // If we're at max capacity and couldn't open a connection. Block on the idle channel with a
- // timeout. Note that Channels guarantee fair FIFO behavior to callers of ReadAsync
- // (first-come, first-served), which is crucial to us.
- if (async)
+ // Fast check for an immediately-available outcome. This is FIFO-safe: a buffered
+ // item only exists when no reader is parked (the channel hands writes directly to
+ // parked readers), so this cannot barge ahead of a waiting request.
+ if (!_idleChannel.TryRead(out outcome))
+ {
+ // Nothing buffered. Register demand *before* pumping and parking so the pump and
+ // any concurrent slot-freeing event both observe this waiter (see _waiterCount).
+ Interlocked.Increment(ref _waiterCount);
+ try
{
- connection ??= await _idleChannel.ReadAsync(cancellationToken).ConfigureAwait(false);
+ TryPumpCreate(owningConnection);
+
+ // Block until an outcome arrives. Channels guarantee FIFO delivery to parked
+ // ReadAsync callers, which is what preserves fair ordering here.
+ if (async)
+ {
+ outcome = await _idleChannel.ReadAsync(cancellationToken).ConfigureAwait(false);
+ }
+ else
+ {
+ outcome = ReadChannelSyncOverAsync(cancellationToken);
+ }
}
- else
+ catch (OperationCanceledException)
{
- connection ??= ReadChannelSyncOverAsync(cancellationToken);
+ throw ADP.PooledOpenTimeout();
+ }
+ catch (ChannelClosedException)
+ {
+ throw new InvalidOperationException(StringsHelper.GetString(Strings.SQL_ConnectionPoolShutDown));
+ }
+ finally
+ {
+ Interlocked.Decrement(ref _waiterCount);
}
}
- catch (OperationCanceledException)
- {
- throw ADP.PooledOpenTimeout();
- }
- catch (ChannelClosedException)
- {
- throw new InvalidOperationException(StringsHelper.GetString(Strings.SQL_ConnectionPoolShutDown));
- }
+
+ // A background create failed; propagate its error to this waiter.
+ outcome.Error?.Throw();
+
+ connection = outcome.Connection;
if (connection is not null && !IsLiveConnection(connection))
{
- // If the connection is not live, we need to remove it from the pool and try again.
+ // Stale or dead connection: remove it (which frees its slot and re-drives
+ // waiters) and loop to try again. A bare-wake outcome (no connection, no error)
+ // also lands here as null and simply loops.
RemoveConnection(connection);
connection = null;
}
@@ -727,13 +859,13 @@ private async Task GetInternalConnection(
}
///
- /// Performs a blocking synchronous read from the idle connection channel.
+ /// Performs a blocking synchronous read from the idle/creation channel.
///
/// Cancels the read operation.
- /// The connection read from the channel.
- private DbConnectionInternal? ReadChannelSyncOverAsync(CancellationToken cancellationToken)
+ /// The outcome read from the channel.
+ private CreateOutcome ReadChannelSyncOverAsync(CancellationToken cancellationToken)
{
- // If there are no connections in the channel, then ReadAsync will block until one is available.
+ // If there are no outcomes in the channel, then ReadAsync will block until one is available.
// Channels doesn't offer a sync API, so running ReadAsync synchronously on this thread may spawn
// additional new async work items in the managed thread pool if there are no items available in the
// channel. We need to ensure that we don't block all available managed threads with these child
@@ -743,7 +875,7 @@ private async Task GetInternalConnection(
_syncOverAsyncSemaphore.Wait(cancellationToken);
try
{
- ConfiguredValueTaskAwaitable.ConfiguredValueTaskAwaiter awaiter =
+ ConfiguredValueTaskAwaitable.ConfiguredValueTaskAwaiter awaiter =
_idleChannel.ReadAsync(cancellationToken).ConfigureAwait(false).GetAwaiter();
using ManualResetEventSlim mres = new ManualResetEventSlim(false, 0);
@@ -827,14 +959,14 @@ internal void PruneConnections(int count)
while (count > 0
&& IsRunning
&& _connectionSlots.ReservationCount > MinPoolSize
- && _idleChannel.TryRead(out var connection))
+ && _idleChannel.TryRead(out var outcome))
{
- if (connection is null)
+ if (outcome.Connection is null)
{
continue;
}
- RemoveConnection(connection);
+ RemoveConnection(outcome.Connection);
count--;
}
}
diff --git a/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/CreateOutcome.cs b/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/CreateOutcome.cs
new file mode 100644
index 0000000000..bf59c36b8e
--- /dev/null
+++ b/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/CreateOutcome.cs
@@ -0,0 +1,73 @@
+// 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.Runtime.ExceptionServices;
+using Microsoft.Data.ProviderBase;
+
+#nullable enable
+
+namespace Microsoft.Data.SqlClient.ConnectionPool
+{
+ ///
+ /// The payload carried by the pool's idle/creation channel. A single value can represent one
+ /// of three things that a waiting request may observe:
+ ///
+ ///
+ /// -
+ /// A connection ( is non-null): either an idle
+ /// connection that was returned to the pool, or a freshly created connection published by a
+ /// background create ("pump") task. The waiter takes it.
+ ///
+ /// -
+ /// An error ( is non-null): a background create task failed.
+ /// The FIFO-head waiter rethrows the captured exception. One failure is consumed by exactly one
+ /// waiter, so demand and error delivery stay balanced.
+ ///
+ /// -
+ /// A bare wake (both and null —
+ /// a default value): a "capacity changed,
+ /// re-evaluate" signal written when a slot is freed but no connection or error is available to
+ /// hand out (e.g. a dead connection was removed, or a create completed without producing one).
+ /// A woken waiter loops and re-pumps using its own owning connection.
+ ///
+ ///
+ ///
+ internal readonly struct CreateOutcome
+ {
+ ///
+ /// The connection to hand to a waiter, or when this outcome carries
+ /// an error or is a bare wake.
+ ///
+ internal DbConnectionInternal? Connection { get; }
+
+ ///
+ /// The captured creation error to rethrow on a waiter, or when this
+ /// outcome carries a connection or is a bare wake.
+ ///
+ internal ExceptionDispatchInfo? Error { get; }
+
+ ///
+ /// Creates an outcome that hands a connection to a waiter.
+ ///
+ internal CreateOutcome(DbConnectionInternal connection)
+ {
+ Connection = connection;
+ Error = null;
+ }
+
+ ///
+ /// Creates an outcome that rethrows a captured creation error on a waiter.
+ ///
+ internal CreateOutcome(ExceptionDispatchInfo error)
+ {
+ Connection = null;
+ Error = error;
+ }
+
+ ///
+ /// when this outcome carries a connection.
+ ///
+ internal bool HasConnection => Connection is not null;
+ }
+}
diff --git a/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/IdleConnectionChannel.cs b/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/IdleConnectionChannel.cs
index 348ad33d9e..5abc9c1eae 100644
--- a/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/IdleConnectionChannel.cs
+++ b/src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/IdleConnectionChannel.cs
@@ -4,27 +4,27 @@
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Channels;
-using Microsoft.Data.ProviderBase;
#nullable enable
namespace Microsoft.Data.SqlClient.ConnectionPool
{
///
- /// Wraps an unbounded of idle connections and tracks the number of
- /// non-null connections it contains. Unbounded channels do not support
- /// , so this class maintains the count via
- /// operations on every read and write of a non-null value.
+ /// Wraps an unbounded of values and tracks
+ /// the number of connection-bearing outcomes it currently holds (i.e. idle connections).
+ /// Unbounded channels do not support , so this class
+ /// maintains the count via operations on every read and write of a
+ /// connection-bearing outcome. Bare-wake and error outcomes do not affect the count.
///
internal sealed class IdleConnectionChannel
{
- private readonly ChannelReader _reader;
- private readonly ChannelWriter _writer;
+ private readonly ChannelReader _reader;
+ private readonly ChannelWriter _writer;
private volatile int _count;
internal IdleConnectionChannel()
{
- var channel = Channel.CreateUnbounded();
+ var channel = Channel.CreateUnbounded();
_reader = channel.Reader;
_writer = channel.Writer;
}
@@ -40,20 +40,20 @@ internal IdleConnectionChannel()
internal bool Complete() => _writer.TryComplete();
///
- /// The number of non-null connections currently in the channel.
+ /// The number of connection-bearing outcomes (idle connections) currently in the channel.
///
internal int Count => _count;
///
- /// Writes a connection (or null wake-up signal) to the channel.
- /// Increments the idle count when is not null.
+ /// Writes an outcome to the channel. Increments the idle count when the outcome carries a
+ /// connection.
///
/// if the value was written; otherwise .
- internal bool TryWrite(DbConnectionInternal? connection)
+ internal bool TryWrite(CreateOutcome outcome)
{
- if (_writer.TryWrite(connection))
+ if (_writer.TryWrite(outcome))
{
- if (connection is not null)
+ if (outcome.HasConnection)
{
Interlocked.Increment(ref _count);
}
@@ -64,14 +64,14 @@ internal bool TryWrite(DbConnectionInternal? connection)
}
///
- /// Tries to read a value from the channel without blocking.
- /// Decrements the idle count when a non-null connection is read.
+ /// Tries to read an outcome from the channel without blocking.
+ /// Decrements the idle count when a connection-bearing outcome is read.
///
- internal bool TryRead(out DbConnectionInternal? connection)
+ internal bool TryRead(out CreateOutcome outcome)
{
- if (_reader.TryRead(out connection))
+ if (_reader.TryRead(out outcome))
{
- if (connection is not null)
+ if (outcome.HasConnection)
{
Interlocked.Decrement(ref _count);
}
@@ -83,19 +83,19 @@ internal bool TryRead(out DbConnectionInternal? connection)
}
///
- /// Asynchronously reads a value from the channel.
- /// Decrements the idle count when a non-null connection is read.
+ /// Asynchronously reads an outcome from the channel.
+ /// Decrements the idle count when a connection-bearing outcome is read.
///
- internal async ValueTask ReadAsync(CancellationToken cancellationToken)
+ internal async ValueTask ReadAsync(CancellationToken cancellationToken)
{
- var connection = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
+ var outcome = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
- if (connection is not null)
+ if (outcome.HasConnection)
{
Interlocked.Decrement(ref _count);
}
- return connection;
+ return outcome;
}
}
}
diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/ChannelDbConnectionPoolTest.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/ChannelDbConnectionPoolTest.cs
index 914e63019a..3b984d9f6e 100644
--- a/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/ChannelDbConnectionPoolTest.cs
+++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/ChannelDbConnectionPoolTest.cs
@@ -1156,6 +1156,119 @@ private static void BackdateReturnedTime(DbConnectionInternal connection, TimeSp
#endregion
+ #region Pump behavior
+
+ [Fact]
+ public async Task Pump_ConcurrentAsyncRequests_CreateConnectionsInParallelUpToMax()
+ {
+ // Empty pool, many simultaneous async requests. The pump should create one connection
+ // per unmet waiter (in parallel on background tasks), up to MaxPoolSize.
+ const int count = 10;
+ var pool = ConstructPool(SuccessfulConnectionFactory);
+
+ var tasks = new Task[count];
+ for (int i = 0; i < count; i++)
+ {
+ var tcs = new TaskCompletionSource();
+ pool.TryGetConnection(
+ new SqlConnection(),
+ tcs,
+ TimeoutTimer.StartNew(TimeSpan.FromSeconds(15)),
+ out _);
+ tasks[i] = tcs.Task;
+ }
+
+ var connections = await Task.WhenAll(tasks);
+
+ foreach (var connection in connections)
+ {
+ Assert.NotNull(connection);
+ }
+ // Count distinct physical connections created.
+ Assert.Equal(count, pool.Count);
+ }
+
+ [Fact]
+ public void Pump_CreationError_IsDeliveredToCaller()
+ {
+ // A background create failure must surface as a real exception on a waiting caller, not
+ // just a timeout. This matters when the pool blocking period is off (e.g. Azure), where
+ // each failed open should reach a caller. Verified on the synchronous path.
+ var pool = ConstructPool(new ThrowingSqlConnectionFactory());
+
+ var ex = Assert.Throws(() =>
+ pool.TryGetConnection(
+ new SqlConnection(),
+ taskCompletionSource: null,
+ TimeoutTimer.StartNew(TimeSpan.FromSeconds(15)),
+ out _));
+
+ Assert.Equal(CustomPumpException.Sentinel, ex.Message);
+ }
+
+ [Fact]
+ public async Task Pump_CreationErrorAsync_IsDeliveredToCaller()
+ {
+ var pool = ConstructPool(new ThrowingSqlConnectionFactory());
+ var tcs = new TaskCompletionSource();
+
+ pool.TryGetConnection(
+ new SqlConnection(),
+ tcs,
+ TimeoutTimer.StartNew(TimeSpan.FromSeconds(15)),
+ out _);
+
+ var ex = await Assert.ThrowsAsync(() => tcs.Task);
+ Assert.Equal(CustomPumpException.Sentinel, ex.Message);
+ }
+
+ [Fact]
+ public async Task Pump_MaxPoolDoomedReturn_UnblocksWaiter()
+ {
+ // A request parked at max capacity must be unblocked when a slot frees up because a
+ // dead connection was removed. This exercises the bare-wake -> re-pump -> create path.
+ var poolGroupOptions = new DbConnectionPoolGroupOptions(
+ poolByIdentity: false,
+ minPoolSize: 0,
+ maxPoolSize: 1,
+ creationTimeout: 15_000,
+ loadBalanceTimeout: 0,
+ hasTransactionAffinity: true,
+ idleTimeout: 0);
+ var pool = ConstructPool(SuccessfulConnectionFactory, poolGroupOptions: poolGroupOptions);
+
+ // Caller A takes the pool's only connection.
+ var ownerA = new SqlConnection();
+ pool.TryGetConnection(
+ ownerA,
+ taskCompletionSource: null,
+ TimeoutTimer.StartNew(TimeSpan.FromSeconds(15)),
+ out DbConnectionInternal? connectionA);
+ Assert.NotNull(connectionA);
+
+ // Caller B requests while the pool is at max with nothing idle, so it parks.
+ var tcsB = new TaskCompletionSource();
+ pool.TryGetConnection(
+ new SqlConnection(),
+ tcsB,
+ TimeoutTimer.StartNew(TimeSpan.FromSeconds(15)),
+ out _);
+
+ // Give B a moment to park on the channel; it cannot complete yet.
+ await Task.Delay(100);
+ Assert.False(tcsB.Task.IsCompleted);
+
+ // Doom A's connection (Clear bumps the generation) and return it. The return fails the
+ // liveness check, so the connection is removed, freeing the slot and waking B.
+ pool.Clear();
+ pool.ReturnInternalConnection(connectionA!, ownerA);
+
+ var connectionB = await tcsB.Task.WaitAsync(TimeSpan.FromSeconds(10));
+ Assert.NotNull(connectionB);
+ }
+
+ #endregion
+
#region Test classes
internal class SuccessfulSqlConnectionFactory : SqlConnectionFactory
{
@@ -1188,6 +1301,27 @@ protected override DbConnectionInternal CreateConnection(
}
}
+ internal sealed class CustomPumpException : Exception
+ {
+ internal const string Sentinel = "pump-create-failure-sentinel";
+
+ internal CustomPumpException() : base(Sentinel) { }
+ }
+
+ internal class ThrowingSqlConnectionFactory : SqlConnectionFactory
+ {
+ protected override DbConnectionInternal CreateConnection(
+ SqlConnectionOptions options,
+ ConnectionPoolKey poolKey,
+ DbConnectionPoolGroupProviderInfo poolGroupProviderInfo,
+ IDbConnectionPool pool,
+ DbConnection owningConnection,
+ TimeoutTimer timeout)
+ {
+ throw new CustomPumpException();
+ }
+ }
+
internal class StubDbConnectionInternal : DbConnectionInternal
{
#region Not Implemented Members
@@ -1444,39 +1578,43 @@ public async Task ConcurrentCallers_ShouldTimeoutIndependently()
}
///
- /// Verifies that the the pool hands to the
- /// connection factory reports a reduced remaining-time budget once the
- /// timer's clock has advanced. This guarantees the factory observes the
- /// actual remaining budget at the moment of the call rather than a
- /// fresh, full timeout.
+ /// Verifies that connection creation is decoupled from the requesting caller's timeout
+ /// budget. In the pump model a background create task opens the physical connection using a
+ /// fresh timer built from the pool's configured CreationTimeout, not the caller's
+ /// remaining budget. The caller's own timer only bounds how long it waits on the channel.
///
- ///
- /// Drives elapsed time deterministically with a
- /// so the test does not depend on real
- /// wall-clock waits or thread sleeps.
- ///
[Fact]
- public void GetConnection_TimeoutTimerReflectsPoolWaitTime()
+ public void GetConnection_CreatesConnectionWithFreshCreationTimeout()
{
- // Arrange: a capturing factory and a fake-time-backed timer with a
- // 30-second budget anchored at virtual time t = 0.
+ // Arrange: a capturing factory and a pool whose CreationTimeout is a distinctive,
+ // comfortably large value so we can tell it apart from the caller's budget.
+ const int creationTimeoutMs = 30_000;
var factory = new SuccessfulSqlConnectionFactory();
- var pool = ConstructPool(factory);
- var owner = new SqlConnection("Timeout=30");
- var fakeTime = new FakeTimeProvider();
- TimeoutTimer timer = TimeoutTimer.StartNew(TimeSpan.FromSeconds(30), fakeTime);
+ var poolGroupOptions = new DbConnectionPoolGroupOptions(
+ poolByIdentity: false,
+ minPoolSize: 0,
+ maxPoolSize: 50,
+ creationTimeout: creationTimeoutMs,
+ loadBalanceTimeout: 0,
+ hasTransactionAffinity: true,
+ idleTimeout: 0);
+ var pool = ConstructPool(factory, poolGroupOptions: poolGroupOptions);
- // Act: advance virtual time by 5 seconds before invoking the pool,
- // simulating budget that was consumed elsewhere (e.g., waiting on a
- // pool slot) before the factory was called.
- fakeTime.Advance(TimeSpan.FromSeconds(5));
- pool.TryGetConnection(owner, taskCompletionSource: null, timer, out DbConnectionInternal? connection);
+ // The caller's timer carries a different, much smaller budget than CreationTimeout.
+ var owner = new SqlConnection("Timeout=5");
+ TimeoutTimer callerTimer = TimeoutTimer.StartNew(TimeSpan.FromSeconds(5));
+
+ // Act
+ pool.TryGetConnection(owner, taskCompletionSource: null, callerTimer, out DbConnectionInternal? connection);
- // Assert: factory received the same timer, and it reports the
- // reduced 25-second remaining budget.
+ // Assert: the factory did NOT receive the caller's timer; it received a fresh timer
+ // whose budget matches the pool's CreationTimeout.
Assert.NotNull(connection);
- Assert.Same(timer, factory.CapturedTimeout);
- Assert.Equal(25_000, factory.CapturedTimeout!.MillisecondsRemainingInt);
+ Assert.NotNull(factory.CapturedTimeout);
+ Assert.NotSame(callerTimer, factory.CapturedTimeout);
+ // Only a few milliseconds can elapse between the pump starting the timer and the factory
+ // reading it, so the remaining budget stays very close to the configured CreationTimeout.
+ Assert.InRange(factory.CapturedTimeout!.MillisecondsRemainingInt, creationTimeoutMs - 5_000, creationTimeoutMs);
}
#endregion
diff --git a/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/IdleConnectionChannelTest.cs b/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/IdleConnectionChannelTest.cs
index dacc266753..3edae59e81 100644
--- a/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/IdleConnectionChannelTest.cs
+++ b/src/Microsoft.Data.SqlClient/tests/UnitTests/ConnectionPool/IdleConnectionChannelTest.cs
@@ -19,23 +19,25 @@ namespace Microsoft.Data.SqlClient.UnitTests.ConnectionPool
{
public class IdleConnectionChannelTest
{
+ private static CreateOutcome Conn() => new CreateOutcome(new StubDbConnectionInternal());
+
#region TryWrite
[Fact]
- public void TryWrite_NonNullConnection_IncrementsCount()
+ public void TryWrite_ConnectionOutcome_IncrementsCount()
{
var channel = new IdleConnectionChannel();
- Assert.True(channel.TryWrite(new StubDbConnectionInternal()));
+ Assert.True(channel.TryWrite(Conn()));
Assert.Equal(1, channel.Count);
}
[Fact]
- public void TryWrite_NullConnection_DoesNotIncrementCount()
+ public void TryWrite_BareWake_DoesNotIncrementCount()
{
var channel = new IdleConnectionChannel();
- Assert.True(channel.TryWrite(null));
+ Assert.True(channel.TryWrite(default));
Assert.Equal(0, channel.Count);
}
@@ -44,10 +46,10 @@ public void TryWrite_MultipleConnections_TracksCountCorrectly()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(null);
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
+ channel.TryWrite(Conn());
+ channel.TryWrite(default);
+ channel.TryWrite(Conn());
Assert.Equal(3, channel.Count);
}
@@ -57,33 +59,33 @@ public void TryWrite_MultipleConnections_TracksCountCorrectly()
#region TryRead
[Fact]
- public void TryRead_NonNullConnection_DecrementsCount()
+ public void TryRead_ConnectionOutcome_DecrementsCount()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
Assert.Equal(1, channel.Count);
- Assert.True(channel.TryRead(out var connection));
- Assert.NotNull(connection);
+ Assert.True(channel.TryRead(out var outcome));
+ Assert.NotNull(outcome.Connection);
Assert.Equal(0, channel.Count);
}
[Fact]
- public void TryRead_NullConnection_DoesNotDecrementCount()
+ public void TryRead_BareWake_DoesNotDecrementCount()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(null);
+ channel.TryWrite(Conn());
+ channel.TryWrite(default);
Assert.Equal(1, channel.Count);
- // Read the non-null connection first (FIFO)
+ // Read the connection outcome first (FIFO)
Assert.True(channel.TryRead(out var first));
- Assert.NotNull(first);
+ Assert.NotNull(first.Connection);
Assert.Equal(0, channel.Count);
- // Read the null
+ // Read the bare wake
Assert.True(channel.TryRead(out var second));
- Assert.Null(second);
+ Assert.Null(second.Connection);
Assert.Equal(0, channel.Count);
}
@@ -92,8 +94,8 @@ public void TryRead_EmptyChannel_ReturnsFalse()
{
var channel = new IdleConnectionChannel();
- Assert.False(channel.TryRead(out var connection));
- Assert.Null(connection);
+ Assert.False(channel.TryRead(out var outcome));
+ Assert.Null(outcome.Connection);
Assert.Equal(0, channel.Count);
}
@@ -102,34 +104,34 @@ public void TryRead_EmptyChannel_ReturnsFalse()
#region ReadAsync
[Fact]
- public async Task ReadAsync_NonNullConnection_DecrementsCount()
+ public async Task ReadAsync_ConnectionOutcome_DecrementsCount()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
Assert.Equal(1, channel.Count);
- var connection = await channel.ReadAsync(CancellationToken.None);
+ var outcome = await channel.ReadAsync(CancellationToken.None);
- Assert.NotNull(connection);
+ Assert.NotNull(outcome.Connection);
Assert.Equal(0, channel.Count);
}
[Fact]
- public async Task ReadAsync_NullConnection_DoesNotDecrementCount()
+ public async Task ReadAsync_BareWake_DoesNotDecrementCount()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(null);
+ channel.TryWrite(Conn());
+ channel.TryWrite(default);
Assert.Equal(1, channel.Count);
- // First read returns the non-null connection (FIFO)
+ // First read returns the connection outcome (FIFO)
var first = await channel.ReadAsync(CancellationToken.None);
- Assert.NotNull(first);
+ Assert.NotNull(first.Connection);
Assert.Equal(0, channel.Count);
- // Second read returns null
+ // Second read returns the bare wake
var second = await channel.ReadAsync(CancellationToken.None);
- Assert.Null(second);
+ Assert.Null(second.Connection);
Assert.Equal(0, channel.Count);
}
@@ -142,10 +144,10 @@ public async Task ReadAsync_WaitsForWrite()
var readTask = channel.ReadAsync(CancellationToken.None);
Assert.False(readTask.IsCompleted);
- channel.TryWrite(expected);
+ channel.TryWrite(new CreateOutcome(expected));
- var connection = await readTask;
- Assert.Same(expected, connection);
+ var outcome = await readTask;
+ Assert.Same(expected, outcome.Connection);
Assert.Equal(0, channel.Count);
}
@@ -188,8 +190,8 @@ public void TryWrite_AfterComplete_ReturnsFalseAndDoesNotIncrementCount()
var channel = new IdleConnectionChannel();
channel.Complete();
- Assert.False(channel.TryWrite(new StubDbConnectionInternal()));
- Assert.False(channel.TryWrite(null));
+ Assert.False(channel.TryWrite(Conn()));
+ Assert.False(channel.TryWrite(default));
Assert.Equal(0, channel.Count);
}
@@ -197,17 +199,17 @@ public void TryWrite_AfterComplete_ReturnsFalseAndDoesNotIncrementCount()
public void TryRead_AfterComplete_DrainsBufferedItems()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
+ channel.TryWrite(Conn());
Assert.Equal(2, channel.Count);
channel.Complete();
// Completion only stops new writes; already-buffered items remain readable.
Assert.True(channel.TryRead(out var first));
- Assert.NotNull(first);
+ Assert.NotNull(first.Connection);
Assert.True(channel.TryRead(out var second));
- Assert.NotNull(second);
+ Assert.NotNull(second.Connection);
Assert.Equal(0, channel.Count);
// Once drained, further reads return false.
@@ -218,12 +220,12 @@ public void TryRead_AfterComplete_DrainsBufferedItems()
public async Task ReadAsync_AfterCompleteAndDrain_ThrowsChannelClosedException()
{
var channel = new IdleConnectionChannel();
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
channel.Complete();
// Buffered item is still readable.
var buffered = await channel.ReadAsync(CancellationToken.None);
- Assert.NotNull(buffered);
+ Assert.NotNull(buffered.Connection);
// After the channel is drained, ReadAsync faults with ChannelClosedException.
await Assert.ThrowsAsync(
@@ -255,9 +257,9 @@ public void WriteAndReadSequence_CountStaysConsistent()
var channel = new IdleConnectionChannel();
// Write 3
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(new StubDbConnectionInternal());
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
+ channel.TryWrite(Conn());
+ channel.TryWrite(Conn());
Assert.Equal(3, channel.Count);
// Read 2
@@ -266,7 +268,7 @@ public void WriteAndReadSequence_CountStaysConsistent()
Assert.Equal(1, channel.Count);
// Write 1 more
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
Assert.Equal(2, channel.Count);
// Read remaining 2
@@ -296,7 +298,7 @@ async Task WriteAndRead()
for (int i = 0; i < iterations; i++)
{
- channel.TryWrite(new StubDbConnectionInternal());
+ channel.TryWrite(Conn());
await channel.ReadAsync(CancellationToken.None);
}
}