#region Fields and properties

static readonly ILogger Logger = NpgsqlLoggingConfiguration.ConnectionLogger;

readonly int _max;

readonly int _min;

readonly TimeSpan _connectionLifetime;

volatile int _numConnectors;

volatile int _idleCount;

/// <summary>

/// Tracks all connectors currently managed by this pool, whether idle or busy.

/// Only updated rarely - when physical connections are opened/closed - but is read in perf-sensitive contexts.

/// </summary>

private protected readonly NpgsqlConnector?[] Connectors;

readonly MultiHostConnectorPool? _parentPool;

/// <summary>

/// Reader side for the idle connector channel. Contains nulls in order to release waiting attempts after

/// a connector has been physically closed/broken.

/// </summary>

readonly ChannelReader<NpgsqlConnector?> _idleConnectorReader;

internal ChannelWriter<NpgsqlConnector?> IdleConnectorWriter { get; }

/// <summary>

/// Incremented every time this pool is cleared via <see cref="NpgsqlConnection.ClearPool"/> or

/// <see cref="NpgsqlConnection.ClearAllPools"/>. Allows us to identify connections which were

/// created before the clear.

/// </summary>

volatile int _clearCounter;

static readonly TimerCallback PruningTimerCallback = PruneIdleConnectors;

readonly Timer _pruningTimer;

readonly TimeSpan _pruningSamplingInterval;

readonly int _pruningSampleSize;

readonly int[] _pruningSamples;

readonly int _pruningMedianIndex;

volatile bool _pruningTimerEnabled;

int _pruningSampleIndex;

static readonly SingleThreadSynchronizationContext SingleThreadSynchronizationContext = new("NpgsqlRemainingAsyncSendWorker");

#endregion

internal sealed override (int Total, int Idle, int Busy) Statistics

{

get

{

var numConnectors = _numConnectors;

var idleCount = _idleCount;

return (numConnectors, idleCount, numConnectors - idleCount);

}

}

internal sealed override bool OwnsConnectors => true;

internal ConnectorPool(NpgsqlConnectionStringBuilder settings, string connString, MultiHostConnectorPool? parentPool = null)

: base(settings, connString)

{

if (settings.MaxPoolSize < settings.MinPoolSize)

throw new ArgumentException($"Connection can't have 'Max Pool Size' {settings.MaxPoolSize} under 'Min Pool Size' {settings.MinPoolSize}");

_parentPool = parentPool;

// We enforce Max Pool Size, so no need to to create a bounded channel (which is less efficient)

// On the consuming side, we have the multiplexing write loop but also non-multiplexing Rents

// On the producing side, we have connections being released back into the pool (both multiplexing and not)

var idleChannel = Channel.CreateUnbounded<NpgsqlConnector?>();

_idleConnectorReader = idleChannel.Reader;

IdleConnectorWriter = idleChannel.Writer;

_max = settings.MaxPoolSize;

_min = settings.MinPoolSize;

if (settings.ConnectionPruningInterval == 0)

throw new ArgumentException("ConnectionPruningInterval can't be 0.");

var connectionIdleLifetime = TimeSpan.FromSeconds(settings.ConnectionIdleLifetime);

var pruningSamplingInterval = TimeSpan.FromSeconds(settings.ConnectionPruningInterval);

if (connectionIdleLifetime < pruningSamplingInterval)

throw new ArgumentException($"Connection can't have ConnectionIdleLifetime {connectionIdleLifetime} under ConnectionPruningInterval {_pruningSamplingInterval}");

_pruningTimer = new Timer(PruningTimerCallback, this, Timeout.Infinite, Timeout.Infinite);

_pruningSampleSize = DivideRoundingUp(settings.ConnectionIdleLifetime, settings.ConnectionPruningInterval);

_pruningMedianIndex = DivideRoundingUp(_pruningSampleSize, 2) - 1; // - 1 to go from length to index

_pruningSamplingInterval = pruningSamplingInterval;

_pruningSamples = new int[_pruningSampleSize];

_pruningTimerEnabled = false;

_connectionLifetime = TimeSpan.FromSeconds(settings.ConnectionLifetime);

Connectors = new NpgsqlConnector[_max];

}

internal sealed override ValueTask<NpgsqlConnector> Get(

NpgsqlConnection conn, NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

return TryGetIdleConnector(out var connector)

? new ValueTask<NpgsqlConnector>(connector)

: RentAsync(conn, timeout, async, cancellationToken);

async ValueTask<NpgsqlConnector> RentAsync(

NpgsqlConnection conn, NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

// First, try to open a new physical connector. This will fail if we're at max capacity.

var connector = await OpenNewConnector(conn, timeout, async, cancellationToken);

if (connector != null)

return connector;

// We're at max capacity. 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.

using var linkedSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);

var finalToken = linkedSource.Token;

linkedSource.CancelAfter(timeout.CheckAndGetTimeLeft());

while (true)

{

try

{

if (async)

{

connector = await _idleConnectorReader.ReadAsync(finalToken);

if (CheckIdleConnector(connector))

return connector;

}

else

{

// Channels don't have a sync API. To avoid sync-over-async issues, we use a special single-

// thread synchronization context which ensures that callbacks are executed on a dedicated

// thread.

// Note that AsTask isn't safe here for getting the result, since it still causes some continuation code

// to get executed on the TP (which can cause deadlocks).

using (SingleThreadSynchronizationContext.Enter())

using (var mre = new ManualResetEventSlim())

{

_idleConnectorReader.WaitToReadAsync(finalToken).GetAwaiter().OnCompleted(() => mre.Set());

mre.Wait(finalToken);

}

}

}

catch (OperationCanceledException)

{

cancellationToken.ThrowIfCancellationRequested();

Debug.Assert(finalToken.IsCancellationRequested);

throw new NpgsqlException(

$"The connection pool has been exhausted, either raise 'Max Pool Size' (currently {_max}) " +

$"or 'Timeout' (currently {Settings.Timeout} seconds) in your connection string.",

new TimeoutException());

}

catch (ChannelClosedException)

{

throw new NpgsqlException("The connection pool has been shut down.");

}

// If we're here, our waiting attempt on the idle connector channel was released with a null

// (or bad connector), or we're in sync mode. Check again if a new idle connector has appeared since we last checked.

if (TryGetIdleConnector(out connector))

return connector;

// We might have closed a connector in the meantime and no longer be at max capacity

// so try to open a new connector and if that fails, loop again.

connector = await OpenNewConnector(conn, timeout, async, cancellationToken);

if (connector != null)

return connector;

}

}

}

[MethodImpl(MethodImplOptions.AggressiveInlining)]

internal sealed override bool TryGetIdleConnector([NotNullWhen(true)] out NpgsqlConnector? connector)

{

while (_idleConnectorReader.TryRead(out var nullableConnector))

{

if (CheckIdleConnector(nullableConnector))

{

connector = nullableConnector;

return true;

}

}

connector = null;

return false;