< Summary

Information
Class: IceRpc.ConnectionCache
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ConnectionCache.cs
Tag: 1986_28452893481
Line coverage
71%
Covered lines: 267
Uncovered lines: 107
Coverable lines: 374
Total lines: 683
Line coverage: 71.3%
Branch coverage
67%
Covered branches: 59
Total branches: 88
Branch coverage: 67%
Method coverage
90%
Covered methods: 18
Fully covered methods: 4
Total methods: 20
Method coverage: 90%
Full method coverage: 20%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
.ctor()100%210%
DisposeAsync()100%66100%
PerformDisposeAsync()100%2278.57%
InvokeAsync(...)37.5%14854.16%
PerformInvokeAsync()66.66%12643.75%
ShutdownAsync(...)75%4485.71%
PerformShutdownAsync()50%7440%
CreateConnectTask()50%10866.66%
DisposePendingConnectionAsync()75%5461.11%
ShutdownWhenRequestedAsync()100%11100%
GetActiveConnectionAsync()64.28%191470.9%
RemoveFromActiveAsync(...)50%6685.71%
ShutdownAndDisposeConnectionAsync()75%4482.35%
TryGetActiveConnection(...)80%101084.61%
get_Current()75%44100%
get_Count()100%210%
get_CurrentIndex()100%11100%
MoveNext()83.33%6686.66%
.ctor(...)50%2272.72%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ConnectionCache.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Features;
 4using IceRpc.Transports;
 5using Microsoft.Extensions.Logging;
 6using Microsoft.Extensions.Logging.Abstractions;
 7using System.Diagnostics;
 8using System.Diagnostics.CodeAnalysis;
 9using System.Runtime.ExceptionServices;
 10
 11namespace IceRpc;
 12
 13/// <summary>Represents an invoker that routes outgoing requests to connections it manages.</summary>
 14/// <remarks><para>The connection cache routes requests based on the request's <see cref="IServerAddressFeature" />
 15/// feature or the server addresses of the request's target service.</para>
 16/// <para>The connection cache keeps at most one active connection per server address.</para></remarks>
 17public sealed class ConnectionCache : IInvoker, IAsyncDisposable
 18{
 19    // Connected connections.
 1220    private readonly Dictionary<ServerAddress, IProtocolConnection> _activeConnections =
 1221        new(ServerAddressComparer.OptionalTransport);
 22
 23    private readonly IClientProtocolConnectionFactory _connectionFactory;
 24
 25    private readonly TimeSpan _connectTimeout;
 26
 27    // A detached connection is a protocol connection that is connecting, shutting down or being disposed. Both
 28    // ShutdownAsync and DisposeAsync wait for detached connections to reach 0 using _detachedConnectionsTcs. Such a
 29    // connection is "detached" because it's not in _activeConnections.
 30    private int _detachedConnectionCount;
 31
 1232    private readonly TaskCompletionSource _detachedConnectionsTcs =
 1233        new(TaskCreationOptions.RunContinuationsAsynchronously);
 34
 35    // A cancellation token source that is canceled when DisposeAsync is called.
 1236    private readonly CancellationTokenSource _disposedCts = new();
 37
 38    private Task? _disposeTask;
 39
 1240    private readonly Lock _mutex = new();
 41
 42    // New connections in the process of connecting.
 1243    private readonly Dictionary<ServerAddress, (IProtocolConnection Connection, Task ConnectTask)> _pendingConnections =
 1244        new(ServerAddressComparer.OptionalTransport);
 45
 46    private readonly bool _preferExistingConnection;
 47
 48    private Task? _shutdownTask;
 49
 50    private readonly TimeSpan _shutdownTimeout;
 51
 52    /// <summary>Constructs a connection cache.</summary>
 53    /// <param name="options">The connection cache options.</param>
 54    /// <param name="duplexClientTransport">The duplex client transport. <see langword="null" /> is equivalent to <see
 55    /// cref="IDuplexClientTransport.Default" />.</param>
 56    /// <param name="multiplexedClientTransport">The multiplexed client transport. <see langword="null" /> is equivalent
 57    /// to <see cref="IMultiplexedClientTransport.Default" />.</param>
 58    /// <param name="logger">The logger. <see langword="null" /> is equivalent to <see cref="NullLogger.Instance"
 59    /// />.</param>
 1260    public ConnectionCache(
 1261        ConnectionCacheOptions options,
 1262        IDuplexClientTransport? duplexClientTransport = null,
 1263        IMultiplexedClientTransport? multiplexedClientTransport = null,
 1264        ILogger? logger = null)
 1265    {
 1266        _connectionFactory = new ClientProtocolConnectionFactory(
 1267            options.ConnectionOptions,
 1268            options.ConnectTimeout,
 1269            options.ClientAuthenticationOptions,
 1270            duplexClientTransport,
 1271            multiplexedClientTransport,
 1272            logger);
 73
 1274        _connectTimeout = options.ConnectTimeout;
 1275        _shutdownTimeout = options.ShutdownTimeout;
 76
 1277        _preferExistingConnection = options.PreferExistingConnection;
 1278    }
 79
 80    /// <summary>Constructs a connection cache using the default options.</summary>
 81    public ConnectionCache()
 082        : this(new ConnectionCacheOptions())
 083    {
 084    }
 85
 86    /// <summary>Releases all resources allocated by the cache. The cache disposes all the connections it
 87    /// created.</summary>
 88    /// <returns>A value task that completes when the disposal of all connections created by this cache has completed.
 89    /// This includes connections that were active when this method is called and connections whose disposal was
 90    /// initiated prior to this call.</returns>
 91    /// <remarks>The disposal of an underlying connection of the cache  aborts invocations, cancels dispatches and
 92    /// disposes the underlying transport connection without waiting for the peer. To wait for invocations and
 93    /// dispatches to complete, call <see cref="ShutdownAsync" /> first. If the configured dispatcher does not complete
 94    /// promptly when its cancellation token is canceled, the disposal can hang.</remarks>
 95    public ValueTask DisposeAsync()
 1396    {
 97        lock (_mutex)
 1398        {
 1399            if (_disposeTask is null)
 12100            {
 12101                _shutdownTask ??= Task.CompletedTask;
 12102                if (_detachedConnectionCount == 0)
 11103                {
 11104                    _ = _detachedConnectionsTcs.TrySetResult();
 11105                }
 106
 12107                _disposeTask = PerformDisposeAsync();
 12108            }
 13109            return new(_disposeTask);
 110        }
 111
 112        async Task PerformDisposeAsync()
 12113        {
 12114            await Task.Yield(); // exit mutex lock
 115
 12116            _disposedCts.Cancel();
 117
 118            // Wait for shutdown before disposing connections.
 119            try
 12120            {
 12121                await _shutdownTask.ConfigureAwait(false);
 12122            }
 0123            catch
 0124            {
 125                // Ignore exceptions.
 0126            }
 127
 128            // Since a pending connection is "detached", it's disposed via the connectTask, not directly by this method.
 12129            await Task.WhenAll(
 8130                _activeConnections.Values.Select(connection => connection.DisposeAsync().AsTask())
 12131                    .Append(_detachedConnectionsTcs.Task)).ConfigureAwait(false);
 132
 12133            _disposedCts.Dispose();
 12134        }
 13135    }
 136
 137    /// <summary>Sends an outgoing request and returns the corresponding incoming response.</summary>
 138    /// <param name="request">The outgoing request being sent.</param>
 139    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 140    /// <returns>The corresponding <see cref="IncomingResponse" />.</returns>
 141    /// <exception cref="InvalidOperationException">Thrown if no <see cref="IServerAddressFeature" /> feature is set and
 142    /// the request's service address has no server addresses.</exception>
 143    /// <exception cref="IceRpcException">Thrown with <see cref="IceRpcError.InvocationRefused" /> if the connection
 144    /// cache is shutdown, or with <see cref="IceRpcError.NoConnection" /> if the request's
 145    /// <see cref="IServerAddressFeature" /> feature has no server addresses.</exception>
 146    /// <exception cref="ObjectDisposedException">Thrown if this connection cache is disposed.</exception>
 147    /// <remarks><para>If the request <see cref="IServerAddressFeature" /> feature is not set, the cache sets it from
 148    /// the server addresses of the target service.</para>
 149    /// <para>It then looks for an active connection. The <see cref="ConnectionCacheOptions.PreferExistingConnection" />
 150    /// property influences how the cache selects this active connection. If no active connection can be found, the
 151    /// cache creates a new connection to one of the server addresses from the <see cref="IServerAddressFeature" />
 152    /// feature.</para>
 153    /// <para>If the connection establishment to <see cref="IServerAddressFeature.ServerAddress" /> fails, <see
 154    /// cref="IServerAddressFeature.ServerAddress" /> is appended at the end of <see
 155    /// cref="IServerAddressFeature.AltServerAddresses" /> and the first address from <see
 156    /// cref="IServerAddressFeature.AltServerAddresses" /> replaces <see cref="IServerAddressFeature.ServerAddress" />.
 157    /// The cache tries again to find or establish a connection to <see cref="IServerAddressFeature.ServerAddress" />.
 158    /// If unsuccessful, the cache repeats this process until success or until it tried all the addresses. If all the
 159    /// attempts fail, this method throws the exception from the last attempt.</para></remarks>
 160    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken)
 94161    {
 94162        if (request.Features.Get<IServerAddressFeature>() is IServerAddressFeature serverAddressFeature)
 0163        {
 0164            if (serverAddressFeature.ServerAddress is null)
 0165            {
 0166                throw new IceRpcException(
 0167                    IceRpcError.NoConnection,
 0168                    $"Could not invoke '{request.Operation}' on '{request.ServiceAddress}': tried all server addresses w
 169            }
 0170        }
 171        else
 94172        {
 94173            if (request.ServiceAddress.ServerAddress is null)
 0174            {
 0175                throw new InvalidOperationException("Cannot send a request to a service without a server address.");
 176            }
 177
 94178            serverAddressFeature = new ServerAddressFeature(request.ServiceAddress);
 94179            request.Features = request.Features.With(serverAddressFeature);
 94180        }
 181
 182        lock (_mutex)
 94183        {
 94184            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 185
 94186            if (_shutdownTask is not null)
 0187            {
 0188                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 189            }
 94190        }
 191
 94192        return PerformInvokeAsync();
 193
 194        async Task<IncomingResponse> PerformInvokeAsync()
 94195        {
 94196            Debug.Assert(serverAddressFeature.ServerAddress is not null);
 197
 198            // When InvokeAsync (or ConnectAsync) throws an IceRpcException(InvocationRefused) we retry unless the
 199            // cache is being shutdown.
 94200            while (true)
 94201            {
 94202                IProtocolConnection? connection = null;
 94203                if (_preferExistingConnection)
 92204                {
 92205                    _ = TryGetActiveConnection(serverAddressFeature, out connection);
 92206                }
 94207                connection ??= await GetActiveConnectionAsync(serverAddressFeature, cancellationToken)
 94208                    .ConfigureAwait(false);
 209
 210                try
 94211                {
 94212                    return await connection.InvokeAsync(request, cancellationToken).ConfigureAwait(false);
 213                }
 0214                catch (ObjectDisposedException)
 0215                {
 216                    // This can occasionally happen if we find a connection that was just closed and then automatically
 217                    // disposed by this connection cache.
 0218                }
 0219                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.InvocationRefused)
 0220                {
 221                    // The connection is refusing new invocations.
 0222                }
 0223                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.OperationAborted)
 0224                {
 225                    lock (_mutex)
 0226                    {
 0227                        if (_disposeTask is null)
 0228                        {
 0229                            throw new IceRpcException(
 0230                                IceRpcError.ConnectionAborted,
 0231                                "The underlying connection was disposed while the invocation was in progress.");
 232                        }
 233                        else
 0234                        {
 0235                            throw;
 236                        }
 237                    }
 238                }
 239
 240                // Make sure connection is no longer in _activeConnection before we retry.
 0241                _ = RemoveFromActiveAsync(serverAddressFeature.ServerAddress.Value, connection);
 0242            }
 94243        }
 94244    }
 245
 246    /// <summary>Gracefully shuts down all connections created by this cache.</summary>
 247    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 248    /// <returns>A task that completes successfully once the shutdown of all connections created by this cache has
 249    /// completed. This includes connections that were active when this method is called and connections whose shutdown
 250    /// was initiated prior to this call.</returns>
 251    /// <exception cref="InvalidOperationException">Thrown if this method is called more than once.</exception>
 252    /// <exception cref="ObjectDisposedException">Thrown if the connection cache is disposed.</exception>
 253    /// <remarks><para>The returned task can also complete with one of the following exceptions:</para>
 254    /// <list type="bullet">
 255    /// <item><description><see cref="IceRpcException" /> with error <see cref="IceRpcError.OperationAborted" /> if the
 256    /// connection cache is disposed while being shut down.</description></item>
 257    /// <item><description><see cref="OperationCanceledException" /> if cancellation was requested through the
 258    /// cancellation token.</description></item>
 259    /// <item><description><see cref="TimeoutException" /> if the shutdown timed out.</description></item>
 260    /// </list>
 261    /// </remarks>
 262    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 11263    {
 264        lock (_mutex)
 11265        {
 11266            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 267
 11268            if (_shutdownTask is not null)
 0269            {
 0270                throw new InvalidOperationException("The connection cache is already shut down or shutting down.");
 271            }
 272
 11273            if (_detachedConnectionCount == 0)
 7274            {
 7275                _detachedConnectionsTcs.SetResult();
 7276            }
 277
 11278            _shutdownTask = PerformShutdownAsync();
 11279        }
 280
 11281        return _shutdownTask;
 282
 283        async Task PerformShutdownAsync()
 11284        {
 11285            await Task.Yield(); // exit mutex lock
 286
 11287            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedCts.Token);
 11288            cts.CancelAfter(_shutdownTimeout);
 289
 290            try
 11291            {
 292                // Since a pending connection is "detached", it's shutdown and disposed via the connectTask, not
 293                // directly by this method.
 294                try
 11295                {
 11296                    await Task.WhenAll(
 8297                        _activeConnections.Values.Select(connection => connection.ShutdownAsync(cts.Token))
 11298                            .Append(_detachedConnectionsTcs.Task.WaitAsync(cts.Token))).ConfigureAwait(false);
 11299                }
 0300                catch (OperationCanceledException)
 0301                {
 0302                    throw;
 303                }
 0304                catch
 0305                {
 306                    // Ignore other connection shutdown failures.
 307
 308                    // Throw OperationCanceledException if this WhenAll exception is hiding an OCE.
 0309                    cts.Token.ThrowIfCancellationRequested();
 0310                }
 11311            }
 0312            catch (OperationCanceledException)
 0313            {
 0314                cancellationToken.ThrowIfCancellationRequested();
 315
 0316                if (_disposedCts.IsCancellationRequested)
 0317                {
 0318                    throw new IceRpcException(
 0319                        IceRpcError.OperationAborted,
 0320                        "The shutdown was aborted because the connection cache was disposed.");
 321                }
 322                else
 0323                {
 0324                    throw new TimeoutException(
 0325                        $"The connection cache shut down timed out after {_shutdownTimeout.TotalSeconds} s.");
 326                }
 327            }
 11328        }
 11329    }
 330
 331    private async Task CreateConnectTask(IProtocolConnection connection, ServerAddress serverAddress)
 20332    {
 20333        await Task.Yield(); // exit mutex lock
 334
 335        // This task "owns" a detachedConnectionCount and as a result _disposedCts can't be disposed.
 20336        using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 20337        cts.CancelAfter(_connectTimeout);
 338
 339        Task shutdownRequested;
 20340        Task? connectTask = null;
 341
 342        try
 20343        {
 344            try
 20345            {
 20346                (_, shutdownRequested) = await connection.ConnectAsync(cts.Token).ConfigureAwait(false);
 17347            }
 0348            catch (OperationCanceledException)
 0349            {
 0350                if (_disposedCts.IsCancellationRequested)
 0351                {
 0352                    throw new IceRpcException(
 0353                        IceRpcError.OperationAborted,
 0354                        "The connection establishment was aborted because the connection cache was disposed.");
 355                }
 356                else
 0357                {
 0358                    throw new TimeoutException(
 0359                        $"The connection establishment timed out after {_connectTimeout.TotalSeconds} s.");
 360                }
 361            }
 17362        }
 3363        catch
 3364        {
 365            lock (_mutex)
 3366            {
 367                // connectTask is executing this method and about to throw.
 3368                connectTask = _pendingConnections[serverAddress].ConnectTask;
 3369                _pendingConnections.Remove(serverAddress);
 3370            }
 371
 3372            _ = DisposePendingConnectionAsync(connection, connectTask);
 3373            throw;
 374        }
 375
 376        lock (_mutex)
 17377        {
 17378            if (_shutdownTask is null)
 17379            {
 380                // the connection is now "attached" in _activeConnections
 17381                _activeConnections.Add(serverAddress, connection);
 17382                _detachedConnectionCount--;
 17383            }
 384            else
 0385            {
 0386                connectTask = _pendingConnections[serverAddress].ConnectTask;
 0387            }
 17388            bool removed = _pendingConnections.Remove(serverAddress);
 17389            Debug.Assert(removed);
 17390        }
 391
 17392        if (connectTask is null)
 17393        {
 17394            _ = ShutdownWhenRequestedAsync(connection, serverAddress, shutdownRequested);
 17395        }
 396        else
 0397        {
 398            // As soon as this method completes successfully, we shut down then dispose the connection.
 0399            _ = DisposePendingConnectionAsync(connection, connectTask);
 0400        }
 401
 402        async Task DisposePendingConnectionAsync(IProtocolConnection connection, Task connectTask)
 3403        {
 404            try
 3405            {
 3406                await connectTask.ConfigureAwait(false);
 407
 408                // Since we own a detachedConnectionCount, _disposedCts is not disposed.
 0409                using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 0410                cts.CancelAfter(_shutdownTimeout);
 0411                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 0412            }
 3413            catch
 3414            {
 415                // Observe and ignore exceptions.
 3416            }
 417
 3418            await connection.DisposeAsync().ConfigureAwait(false);
 419
 420            lock (_mutex)
 3421            {
 3422                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 0423                {
 0424                    _detachedConnectionsTcs.SetResult();
 0425                }
 3426            }
 3427        }
 428
 429        async Task ShutdownWhenRequestedAsync(
 430            IProtocolConnection connection,
 431            ServerAddress serverAddress,
 432            Task shutdownRequested)
 17433        {
 17434            await shutdownRequested.ConfigureAwait(false);
 13435            await RemoveFromActiveAsync(serverAddress, connection).ConfigureAwait(false);
 13436        }
 17437    }
 438
 439    /// <summary>Gets an active connection, by creating and connecting (if necessary) a new protocol connection.
 440    /// </summary>
 441    /// <param name="serverAddressFeature">The server address feature.</param>
 442    /// <param name="cancellationToken">The cancellation token of the invocation calling this method.</param>
 443    private async Task<IProtocolConnection> GetActiveConnectionAsync(
 444        IServerAddressFeature serverAddressFeature,
 445        CancellationToken cancellationToken)
 17446    {
 17447        Debug.Assert(serverAddressFeature.ServerAddress is not null);
 17448        Exception? connectionException = null;
 449        (IProtocolConnection Connection, Task ConnectTask) pendingConnectionValue;
 17450        var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 20451        while (enumerator.MoveNext())
 20452        {
 20453            ServerAddress serverAddress = enumerator.Current;
 20454            if (enumerator.CurrentIndex > 0)
 3455            {
 456                // Rotate the server addresses before each new connection attempt after the initial attempt
 3457                serverAddressFeature.RotateAddresses();
 3458            }
 459
 460            try
 20461            {
 462                lock (_mutex)
 20463                {
 20464                    if (_disposeTask is not null)
 0465                    {
 0466                        throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 467                    }
 20468                    else if (_shutdownTask is not null)
 0469                    {
 0470                        throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache is shut down.");
 471                    }
 472
 20473                    if (_activeConnections.TryGetValue(serverAddress, out IProtocolConnection? connection))
 0474                    {
 0475                        return connection;
 476                    }
 477
 20478                    if (!_pendingConnections.TryGetValue(serverAddress, out pendingConnectionValue))
 20479                    {
 20480                        connection = _connectionFactory.CreateConnection(serverAddress);
 20481                        _detachedConnectionCount++;
 20482                        pendingConnectionValue = (connection, CreateConnectTask(connection, serverAddress));
 20483                        _pendingConnections.Add(serverAddress, pendingConnectionValue);
 20484                    }
 20485                }
 486                // ConnectTask itself takes care of scheduling its exception observation when it fails.
 20487                await pendingConnectionValue.ConnectTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 17488                return pendingConnectionValue.Connection;
 489            }
 0490            catch (TimeoutException exception)
 0491            {
 0492                connectionException = exception;
 0493            }
 3494            catch (IceRpcException exception) when (exception.IceRpcError is
 3495                IceRpcError.ConnectionAborted or
 3496                IceRpcError.ConnectionRefused or
 3497                IceRpcError.ServerBusy or
 3498                IceRpcError.ServerUnreachable)
 3499            {
 500                // keep going unless the connection cache was disposed or shut down
 3501                connectionException = exception;
 502                lock (_mutex)
 3503                {
 3504                    if (_shutdownTask is not null)
 0505                    {
 0506                        throw;
 507                    }
 3508                }
 3509            }
 3510        }
 511
 0512        Debug.Assert(connectionException is not null);
 0513        ExceptionDispatchInfo.Throw(connectionException);
 0514        Debug.Assert(false);
 0515        throw connectionException;
 17516    }
 517
 518    /// <summary>Removes the connection from _activeConnections, and when successful, shuts down and disposes this
 519    /// connection.</summary>
 520    /// <param name="serverAddress">The server address key in _activeConnections.</param>
 521    /// <param name="connection">The connection to shutdown and dispose after removal.</param>
 522    private Task RemoveFromActiveAsync(ServerAddress serverAddress, IProtocolConnection connection)
 13523    {
 524        lock (_mutex)
 13525        {
 526            // Check identity before removing: a concurrent reconnect may have already replaced this connection
 527            // with a new one at the same server address. Removing without the identity check would evict the
 528            // healthy replacement and leak it outside the cache.
 13529            if (_shutdownTask is null &&
 13530                _activeConnections.TryGetValue(serverAddress, out IProtocolConnection? existing) &&
 13531                existing == connection)
 9532            {
 9533                _activeConnections.Remove(serverAddress);
 534                // it's now our connection.
 9535                _detachedConnectionCount++;
 9536            }
 537            else
 4538            {
 539                // Another task owns this connection
 4540                return Task.CompletedTask;
 541            }
 9542        }
 543
 9544        return ShutdownAndDisposeConnectionAsync();
 545
 546        async Task ShutdownAndDisposeConnectionAsync()
 9547        {
 548            // _disposedCts is not disposed since we own a detachedConnectionCount
 9549            using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 9550            cts.CancelAfter(_shutdownTimeout);
 551
 552            try
 9553            {
 9554                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 9555            }
 0556            catch
 0557            {
 558                // Ignore connection shutdown failures
 0559            }
 560
 9561            await connection.DisposeAsync().ConfigureAwait(false);
 562
 563            lock (_mutex)
 9564            {
 9565                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 5566                {
 5567                    _detachedConnectionsTcs.SetResult();
 5568                }
 9569            }
 9570        }
 13571    }
 572
 573    /// <summary>Tries to get an existing connection matching one of the addresses of the server address feature.
 574    /// </summary>
 575    /// <param name="serverAddressFeature">The server address feature.</param>
 576    /// <param name="connection">When this method returns <see langword="true" />, this argument contains an active
 577    /// connection; otherwise, it is set to <see langword="null" />.</param>
 578    /// <returns><see langword="true" /> when an active connection matching any of the addresses of the server address
 579    /// feature is found; otherwise, <see langword="false"/>.</returns>
 580    private bool TryGetActiveConnection(
 581        IServerAddressFeature serverAddressFeature,
 582        [NotNullWhen(true)] out IProtocolConnection? connection)
 92583    {
 584        lock (_mutex)
 92585        {
 92586            connection = null;
 92587            if (_disposeTask is not null)
 0588            {
 0589                throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 590            }
 591
 92592            if (_shutdownTask is not null)
 0593            {
 0594                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 595            }
 596
 92597            var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 112598            while (enumerator.MoveNext())
 97599            {
 97600                ServerAddress serverAddress = enumerator.Current;
 97601                if (_activeConnections.TryGetValue(serverAddress, out connection))
 77602                {
 77603                    if (enumerator.CurrentIndex > 0)
 1604                    {
 605                        // This altServerAddress becomes the main server address, and the existing main
 606                        // server address becomes the first alt server address.
 1607                        serverAddressFeature.AltServerAddresses = serverAddressFeature.AltServerAddresses
 1608                            .RemoveAt(enumerator.CurrentIndex - 1)
 1609                            .Insert(0, serverAddressFeature.ServerAddress!.Value);
 1610                        serverAddressFeature.ServerAddress = serverAddress;
 1611                    }
 77612                    return true;
 613                }
 20614            }
 15615            return false;
 616        }
 92617    }
 618
 619    /// <summary>A helper struct that implements an enumerator that allows iterating the addresses of an
 620    /// <see cref="IServerAddressFeature" /> without allocations.</summary>
 621    private struct ServerAddressEnumerator
 622    {
 623        internal readonly ServerAddress Current
 624        {
 625            get
 117626            {
 117627                Debug.Assert(CurrentIndex >= 0 && CurrentIndex <= _altServerAddresses.Count);
 117628                if (CurrentIndex == 0)
 109629                {
 109630                    Debug.Assert(_mainServerAddress is not null);
 109631                    return _mainServerAddress.Value;
 632                }
 633                else
 8634                {
 8635                    return _altServerAddresses[CurrentIndex - 1];
 636                }
 117637            }
 638        }
 639
 0640        internal int Count { get; }
 641
 955642        internal int CurrentIndex { get; private set; } = -1;
 643
 644        private readonly ServerAddress? _mainServerAddress;
 645        private readonly IList<ServerAddress> _altServerAddresses;
 646
 647        internal bool MoveNext()
 132648        {
 132649            if (CurrentIndex == -1)
 109650            {
 109651                if (_mainServerAddress is not null)
 109652                {
 109653                    CurrentIndex++;
 109654                    return true;
 655                }
 656                else
 0657                {
 0658                    return false;
 659                }
 660            }
 23661            else if (CurrentIndex < _altServerAddresses.Count)
 8662            {
 8663                CurrentIndex++;
 8664                return true;
 665            }
 15666            return false;
 132667        }
 668
 669        internal ServerAddressEnumerator(IServerAddressFeature serverAddressFeature)
 109670        {
 109671            _mainServerAddress = serverAddressFeature.ServerAddress;
 109672            _altServerAddresses = serverAddressFeature.AltServerAddresses;
 109673            if (_mainServerAddress is null)
 0674            {
 0675                Count = 0;
 0676            }
 677            else
 109678            {
 109679                Count = _altServerAddresses.Count + 1;
 109680            }
 109681        }
 682    }
 683}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

.ctor(IceRpc.ConnectionCacheOptions,IceRpc.Transports.IDuplexClientTransport,IceRpc.Transports.IMultiplexedClientTransport,Microsoft.Extensions.Logging.ILogger)
.ctor()
DisposeAsync()
PerformDisposeAsync()
InvokeAsync(IceRpc.OutgoingRequest,System.Threading.CancellationToken)
PerformInvokeAsync()
ShutdownAsync(System.Threading.CancellationToken)
PerformShutdownAsync()
CreateConnectTask()
DisposePendingConnectionAsync()
ShutdownWhenRequestedAsync()
GetActiveConnectionAsync()
RemoveFromActiveAsync(IceRpc.ServerAddress,IceRpc.IProtocolConnection)
ShutdownAndDisposeConnectionAsync()
TryGetActiveConnection(IceRpc.Features.IServerAddressFeature,IceRpc.IProtocolConnection&)
get_Current()
get_Count()
get_CurrentIndex()
MoveNext()
.ctor(IceRpc.Features.IServerAddressFeature)