From 55b36edbe987a812965ebf006ce32788c712116a Mon Sep 17 00:00:00 2001 From: Malcolm Daigle Date: Mon, 13 Jul 2026 18:09:16 -0700 Subject: [PATCH] Refactor ChannelDbConnectionPool to a pump-based, fast-path-free model Remove the optimistic fast path so every request queues on the idle/creation channel, giving strict FIFO ordering to parked waiters. Connection creation is decoupled from the requesting caller and runs on background "pump" tasks that grow the pool as demand and capacity allow. Creation results (connection, error, or bare wake) travel over the channel via a new CreateOutcome payload; a background creation error is delivered to a waiter and rethrown there. - Add CreateOutcome struct (connection | error | bare wake) as the channel payload. - Widen IdleConnectionChannel to carry CreateOutcome; Count tracks connection- bearing outcomes only. - Add _waiterCount / _pendingCreates accounting and TryPumpCreate / LaunchCreate. - Rewrite GetInternalConnection to register demand, pump, then block on the channel. - Background creates use the pool's CreationTimeout (not the caller's budget); a caller's ConnectTimeout governs only its channel wait. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ConnectionPool/ChannelDbConnectionPool.cs | 270 +++++++++++++----- .../SqlClient/ConnectionPool/CreateOutcome.cs | 73 +++++ .../ConnectionPool/IdleConnectionChannel.cs | 50 ++-- .../ChannelDbConnectionPoolTest.cs | 190 ++++++++++-- .../IdleConnectionChannelTest.cs | 98 +++---- 5 files changed, 513 insertions(+), 168 deletions(-) create mode 100644 src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/ConnectionPool/CreateOutcome.cs 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); } }