< Summary

Information
Class: IceRpc.Internal.IceProtocolConnection
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Internal/IceProtocolConnection.cs
Tag: 1986_28452893481
Line coverage
89%
Covered lines: 853
Uncovered lines: 98
Coverable lines: 951
Total lines: 1597
Line coverage: 89.6%
Branch coverage
83%
Covered branches: 209
Total branches: 250
Branch coverage: 83.6%
Method coverage
100%
Covered methods: 37
Fully covered methods: 18
Total methods: 37
Method coverage: 100%
Full method coverage: 48.6%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
get_IsServer()100%11100%
.ctor(...)62.5%88100%
ConnectAsync(...)50%2281.81%
PerformConnectAsync()83.33%211879.41%
EncodeValidateConnectionFrame()100%11100%
DecodeValidateConnectionFrame()100%11100%
DisposeAsync()100%66100%
PerformDisposeAsync()87.5%88100%
InvokeAsync(...)75%8888.23%
PerformInvokeAsync()80.76%262694.84%
ShutdownAsync(...)87.5%8888.88%
PerformShutdownAsync()100%8889.18%
EncodeCloseConnectionFrame()100%11100%
SendHeartbeat()75%44100%
SendValidateConnectionFrameAsync()100%3237.5%
EncodeValidateConnectionFrame()100%11100%
DecodeRequestIdAndHeader(...)75%131281.25%
DecodeResponseHeader(...)77.27%232285.71%
EncodeRequestHeader(...)83.33%6692.1%
EncodeResponseHeader(...)100%2020100%
ReadFullPayloadAsync()50%4475%
AcquireWriteLockAsync()100%22100%
CreateFrameReaderAsync()100%1173.33%
DecrementDispatchInvocationCount()100%66100%
DispatchRequestAsync()94.44%181896.29%
IncrementDispatchInvocationCount()100%22100%
ScheduleInactivityCheck()100%11100%
ReadFramesAsync()75%262075.58%
AbortTwowayInvocations()100%22100%
ReadFailed()100%11100%
ReadReplyAsync()80%101093.33%
ReadRequestAsync()75%202092.85%
<ReadRequestAsync()100%1170.83%
RefuseNewInvocations(...)100%22100%
SendControlFrameAsync()100%11100%
WriteFailed(...)100%11100%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Internal/IceProtocolConnection.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Ice.Codec;
 4using IceRpc.Transports;
 5using IceRpc.Transports.Internal;
 6using System.Buffers;
 7using System.Collections.Immutable;
 8using System.Diagnostics;
 9using System.IO.Pipelines;
 10using System.Security.Authentication;
 11
 12namespace IceRpc.Internal;
 13
 14/// <summary>Implements <see cref="IProtocolConnection" /> for the ice protocol.</summary>
 15internal sealed class IceProtocolConnection : IProtocolConnection
 16{
 117    private static readonly IDictionary<RequestFieldKey, ReadOnlySequence<byte>> _idempotentFields =
 118        new Dictionary<RequestFieldKey, ReadOnlySequence<byte>>
 119        {
 120            [RequestFieldKey.Idempotent] = default
 121        }.ToImmutableDictionary();
 22
 21623    private bool IsServer => _transportConnectionInformation is not null;
 24
 25    private IConnectionContext? _connectionContext; // non-null once the connection is established
 26    private Task? _connectTask;
 27    private readonly IDispatcher _dispatcher;
 28
 29    // The number of outstanding dispatches and invocations.
 30    private int _dispatchInvocationCount;
 31
 32    // We don't want the continuation to run from the dispatch or invocation thread.
 22733    private readonly TaskCompletionSource _dispatchesAndInvocationsCompleted =
 22734        new(TaskCreationOptions.RunContinuationsAsynchronously);
 35
 36    private readonly SemaphoreSlim? _dispatchSemaphore;
 37
 38    // This cancellation token source is canceled when the connection is disposed.
 22739    private readonly CancellationTokenSource _disposedCts = new();
 40
 41    private Task? _disposeTask;
 42    private readonly IDuplexConnection _duplexConnection;
 43    private readonly DuplexConnectionReader _duplexConnectionReader;
 44    private readonly IceDuplexConnectionWriter _duplexConnectionWriter;
 22745    private bool _heartbeatEnabled = true;
 22746    private Task _heartbeatTask = Task.CompletedTask;
 47    private readonly TimeSpan _inactivityTimeout;
 48    private readonly Timer _inactivityTimeoutTimer;
 49    private string? _invocationRefusedMessage;
 50    private int _lastRequestId;
 51    private readonly int _maxFrameSize;
 22752    private readonly Lock _mutex = new();
 53    private readonly PipeOptions _pipeOptions;
 54    private Task? _readFramesTask;
 55
 56    // A connection refuses invocations when it's disposed, shut down, shutting down or merely "shutdown requested".
 57    private bool _refuseInvocations;
 58
 59    // Does ShutdownAsync send a close connection frame?
 22760    private bool _sendCloseConnectionFrame = true;
 61
 62    private Task? _shutdownTask;
 63
 64    // The thread that completes this TCS can run the continuations, and as a result its result must be set without
 65    // holding a lock on _mutex.
 22766    private readonly TaskCompletionSource _shutdownRequestedTcs = new();
 67
 68    // Only set for server connections.
 69    private readonly TransportConnectionInformation? _transportConnectionInformation;
 70
 71    private readonly CancellationTokenSource _twowayDispatchesCts;
 22772    private readonly Dictionary<int, TaskCompletionSource<PipeReader>> _twowayInvocations = new();
 73
 74    private Exception? _writeException; // protected by _writeSemaphore
 22775    private readonly SemaphoreSlim _writeSemaphore = new(1, 1);
 76
 77    public Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> ConnectAsync(
 78        CancellationToken cancellationToken)
 22679    {
 80        Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> result;
 81        lock (_mutex)
 22682        {
 22683            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 84
 22485            if (_connectTask is not null)
 086            {
 087                throw new InvalidOperationException("Cannot call connect more than once.");
 88            }
 89
 22490            result = PerformConnectAsync();
 22491            _connectTask = result;
 22492        }
 22493        return result;
 94
 95        async Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> PerformConnectAsync()
 22496        {
 97            // Make sure we execute the function without holding the connection mutex lock.
 22498            await Task.Yield();
 99
 100            // _disposedCts is not disposed at this point because DisposeAsync waits for the completion of _connectTask.
 224101            using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(
 224102                cancellationToken,
 224103                _disposedCts.Token);
 104
 105            TransportConnectionInformation transportConnectionInformation;
 106
 107            try
 224108            {
 109                // If the transport connection information is null, we need to connect the transport connection. It's
 110                // null for client connections. The transport connection of a server connection is established by
 111                // Server.
 224112                transportConnectionInformation = _transportConnectionInformation ??
 224113                    await _duplexConnection.ConnectAsync(connectCts.Token).ConfigureAwait(false);
 114
 216115                if (IsServer)
 105116                {
 117                    // Send ValidateConnection frame.
 105118                    await SendControlFrameAsync(EncodeValidateConnectionFrame, connectCts.Token).ConfigureAwait(false);
 119
 120                    // The SendControlFrameAsync is a "write" that schedules a heartbeat when the idle timeout is not
 121                    // infinite. So no need to call ScheduleHeartbeat.
 103122                }
 123                else
 111124                {
 111125                    ReadOnlySequence<byte> buffer = await _duplexConnectionReader.ReadAtLeastAsync(
 111126                        IceDefinitions.PrologueSize,
 111127                        connectCts.Token).ConfigureAwait(false);
 128
 101129                    (IcePrologue validateConnectionFrame, long consumed) = DecodeValidateConnectionFrame(buffer);
 101130                    _duplexConnectionReader.AdvanceTo(buffer.GetPosition(consumed), buffer.End);
 131
 101132                    IceDefinitions.CheckPrologue(validateConnectionFrame);
 100133                    if (validateConnectionFrame.FrameSize != IceDefinitions.PrologueSize)
 0134                    {
 0135                        throw new InvalidDataException(
 0136                            $"Received ice frame with only '{validateConnectionFrame.FrameSize}' bytes.");
 137                    }
 100138                    if (validateConnectionFrame.FrameType != IceFrameType.ValidateConnection)
 0139                    {
 0140                        throw new InvalidDataException(
 0141                            $"Expected '{nameof(IceFrameType.ValidateConnection)}' frame but received frame type '{valid
 142                    }
 143
 144                    // The client connection is now connected, so we schedule the first heartbeat.
 100145                    if (_duplexConnection is IceDuplexConnectionDecorator decorator)
 100146                    {
 100147                        decorator.ScheduleHeartbeat();
 100148                    }
 100149                }
 203150            }
 11151            catch (OperationCanceledException)
 11152            {
 11153                cancellationToken.ThrowIfCancellationRequested();
 154
 5155                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 5156                throw new IceRpcException(
 5157                    IceRpcError.OperationAborted,
 5158                    "The connection establishment was aborted because the connection was disposed.");
 159            }
 1160            catch (InvalidDataException exception)
 1161            {
 1162                throw new IceRpcException(
 1163                    IceRpcError.ConnectionAborted,
 1164                    "The connection was aborted by an ice protocol error.",
 1165                    exception);
 166            }
 1167            catch (AuthenticationException)
 1168            {
 1169                throw;
 170            }
 8171            catch (IceRpcException)
 8172            {
 8173                throw;
 174            }
 0175            catch (Exception exception)
 0176            {
 0177                Debug.Fail($"ConnectAsync failed with an unexpected exception: {exception}");
 0178                throw;
 179            }
 180
 181            // We assign _readFramesTask with _mutex locked to make sure this assignment occurs before the start of
 182            // DisposeAsync. Once _disposeTask is not null, _readFramesTask is immutable.
 183            lock (_mutex)
 203184            {
 203185                if (_disposeTask is not null)
 0186                {
 0187                    throw new IceRpcException(
 0188                        IceRpcError.OperationAborted,
 0189                        "The connection establishment was aborted because the connection was disposed.");
 190                }
 191
 192                // This needs to be set before starting the read frames task below.
 203193                _connectionContext = new ConnectionContext(this, transportConnectionInformation);
 194
 203195                _readFramesTask = ReadFramesAsync(_disposedCts.Token);
 203196            }
 197
 198            // The _readFramesTask waits for this PerformConnectAsync completion before reading anything. As soon as
 199            // it receives a request, it will cancel this inactivity check.
 203200            ScheduleInactivityCheck();
 201
 203202            return (transportConnectionInformation, _shutdownRequestedTcs.Task);
 203
 204            static void EncodeValidateConnectionFrame(IBufferWriter<byte> writer)
 105205            {
 105206                var encoder = new IceEncoder(writer);
 105207                IceDefinitions.ValidateConnectionFrame.Encode(ref encoder);
 105208            }
 209
 210            static (IcePrologue, long) DecodeValidateConnectionFrame(ReadOnlySequence<byte> buffer)
 101211            {
 101212                var decoder = new IceDecoder(buffer);
 101213                return (new IcePrologue(ref decoder), decoder.Consumed);
 101214            }
 203215        }
 224216    }
 217
 218    public ValueTask DisposeAsync()
 248219    {
 220        lock (_mutex)
 248221        {
 248222            if (_disposeTask is null)
 227223            {
 227224                RefuseNewInvocations("The connection was disposed.");
 225
 227226                _shutdownTask ??= Task.CompletedTask;
 227227                if (_dispatchInvocationCount == 0)
 218228                {
 218229                    _dispatchesAndInvocationsCompleted.TrySetResult();
 218230                }
 231
 227232                _heartbeatEnabled = false; // makes _heartbeatTask immutable
 233
 227234                _disposeTask = PerformDisposeAsync();
 227235            }
 248236        }
 248237        return new(_disposeTask);
 238
 239        async Task PerformDisposeAsync()
 227240        {
 241            // Make sure we execute the code below without holding the mutex lock.
 227242            await Task.Yield();
 243
 227244            _disposedCts.Cancel();
 245
 246            // We don't lock _mutex since once _disposeTask is not null, _connectTask etc are immutable.
 247
 227248            if (_connectTask is not null)
 224249            {
 250                // Wait for all writes to complete. This can't take forever since all writes are canceled by
 251                // _disposedCts.Token.
 224252                await _writeSemaphore.WaitAsync().ConfigureAwait(false);
 253
 254                try
 224255                {
 224256                    await Task.WhenAll(
 224257                        _connectTask,
 224258                        _readFramesTask ?? Task.CompletedTask,
 224259                        _heartbeatTask,
 224260                        _dispatchesAndInvocationsCompleted.Task,
 224261                        _shutdownTask).ConfigureAwait(false);
 160262                }
 64263                catch
 64264                {
 265                    // Expected if any of these tasks failed or was canceled. Each task takes care of handling
 266                    // unexpected exceptions so there's no need to handle them here.
 64267                }
 224268            }
 269
 227270            _duplexConnection.Dispose();
 271
 272            // It's safe to dispose the reader/writer since no more threads are sending/receiving data.
 227273            _duplexConnectionReader.Dispose();
 227274            _duplexConnectionWriter.Dispose();
 275
 227276            _disposedCts.Dispose();
 227277            _twowayDispatchesCts.Dispose();
 278
 227279            _dispatchSemaphore?.Dispose();
 227280            _writeSemaphore.Dispose();
 227281            await _inactivityTimeoutTimer.DisposeAsync().ConfigureAwait(false);
 227282        }
 248283    }
 284
 285    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken = default)
 1399286    {
 1399287        if (request.Protocol != Protocol.Ice)
 1288        {
 1289            throw new InvalidOperationException(
 1290                $"Cannot send {request.Protocol} request on {Protocol.Ice} connection.");
 291        }
 292
 293        lock (_mutex)
 1398294        {
 1398295            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 296
 1397297            if (_refuseInvocations)
 1298            {
 1299                throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage);
 300            }
 1396301            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 0302            {
 0303                throw new InvalidOperationException("Cannot invoke on a connection that is not fully established.");
 304            }
 305
 1396306            IncrementDispatchInvocationCount();
 1396307        }
 308
 1396309        return PerformInvokeAsync();
 310
 311        async Task<IncomingResponse> PerformInvokeAsync()
 1396312        {
 313            // Since _dispatchInvocationCount > 0, _disposedCts is not disposed.
 1396314            using var invocationCts =
 1396315                CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token, cancellationToken);
 316
 1396317            PipeReader? frameReader = null;
 1396318            bool responseCreated = false;
 1396319            TaskCompletionSource<PipeReader>? responseCompletionSource = null;
 1396320            int requestId = 0;
 321
 322            try
 1396323            {
 324                // Read the full payload. This can take some time so this needs to be done before acquiring the write
 325                // semaphore.
 1396326                ReadOnlySequence<byte> payloadBuffer = await ReadFullPayloadAsync(request.Payload, invocationCts.Token)
 1396327                    .ConfigureAwait(false);
 328
 329                try
 1396330                {
 331                    // Wait for the writing of other frames to complete.
 1396332                    using SemaphoreLock _ = await AcquireWriteLockAsync(invocationCts.Token).ConfigureAwait(false);
 333
 334                    // Assign the request ID for two-way invocations and keep track of the invocation for receiving the
 335                    // response. The request ID is only assigned once the write semaphore is acquired. We don't want a
 336                    // canceled request to allocate a request ID that won't be used.
 337                    lock (_mutex)
 1396338                    {
 1396339                        if (_refuseInvocations)
 0340                        {
 341                            // It's InvocationCanceled and not InvocationRefused because we've read the payload.
 0342                            throw new IceRpcException(IceRpcError.InvocationCanceled, _invocationRefusedMessage);
 343                        }
 344
 1396345                        if (!request.IsOneway)
 390346                        {
 347                            // wrap around back to 1 if we reach int.MaxValue. 0 means one-way.
 390348                            _lastRequestId = _lastRequestId == int.MaxValue ? 1 : _lastRequestId + 1;
 390349                            requestId = _lastRequestId;
 350
 351                            // RunContinuationsAsynchronously because we don't want the "read frames loop" to run the
 352                            // continuation.
 390353                            responseCompletionSource = new(TaskCreationOptions.RunContinuationsAsynchronously);
 390354                            _twowayInvocations[requestId] = responseCompletionSource;
 390355                        }
 1396356                    }
 357
 1396358                    int payloadSize = checked((int)payloadBuffer.Length);
 359
 360                    try
 1396361                    {
 1396362                        EncodeRequestHeader(_duplexConnectionWriter, request, requestId, payloadSize);
 363
 364                        // We write to the duplex connection with _disposedCts.Token instead of invocationCts.Token.
 365                        // Canceling this write operation is fatal to the connection.
 1396366                        await _duplexConnectionWriter.WriteAsync(payloadBuffer, _disposedCts.Token)
 1396367                            .ConfigureAwait(false);
 1395368                    }
 1369                    catch (Exception exception)
 1370                    {
 1371                        WriteFailed(exception);
 1372                        throw;
 373                    }
 1395374                }
 1375                catch (IceRpcException exception) when (exception.IceRpcError != IceRpcError.InvocationCanceled)
 1376                {
 377                    // Since we could not send the request, the server cannot dispatch it and it's safe to retry.
 378                    // This includes the situation where await AcquireWriteLockAsync throws because a previous write
 379                    // failed.
 1380                    throw new IceRpcException(
 1381                        IceRpcError.InvocationCanceled,
 1382                        "Failed to send ice request.",
 1383                        exception);
 384                }
 385                finally
 1396386                {
 387                    // We've read the payload (see ReadFullPayloadAsync) and we are now done with it.
 1396388                    request.Payload.Complete();
 1396389                }
 390
 1395391                if (request.IsOneway)
 1006392                {
 393                    // We're done, there's no response for one-way requests.
 1006394                    return new IncomingResponse(request, _connectionContext!);
 395                }
 396
 397                // Wait to receive the response.
 389398                Debug.Assert(responseCompletionSource is not null);
 389399                frameReader = await responseCompletionSource.Task.WaitAsync(invocationCts.Token).ConfigureAwait(false);
 400
 369401                if (!frameReader.TryRead(out ReadResult readResult))
 0402                {
 0403                    throw new InvalidDataException($"Received empty response frame for request with id '{requestId}'.");
 404                }
 405
 369406                Debug.Assert(readResult.IsCompleted);
 407
 369408                (StatusCode statusCode, string? errorMessage, SequencePosition consumed) =
 369409                    DecodeResponseHeader(readResult.Buffer, requestId);
 410
 369411                frameReader.AdvanceTo(consumed);
 412
 369413                var response = new IncomingResponse(
 369414                    request,
 369415                    _connectionContext!,
 369416                    statusCode,
 369417                    errorMessage)
 369418                {
 369419                    Payload = frameReader
 369420                };
 421
 369422                responseCreated = true; // the response now owns frameReader
 369423                return response;
 424            }
 9425            catch (OperationCanceledException)
 9426            {
 9427                cancellationToken.ThrowIfCancellationRequested();
 428
 3429                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 3430                throw new IceRpcException(
 3431                    IceRpcError.OperationAborted,
 3432                    "The invocation was aborted because the connection was disposed.");
 433            }
 434            finally
 1396435            {
 436                // If responseCompletionSource is not completed, we want to complete it to prevent another method from
 437                // setting an unobservable exception in it. And if it's already completed with an exception, we observe
 438                // this exception.
 1396439                if (responseCompletionSource is not null &&
 1396440                    !responseCompletionSource.TrySetResult(InvalidPipeReader.Instance))
 380441                {
 442                    try
 380443                    {
 444                        // Retrieve (or re-retrieve) the response PipeReader. The cleanup at the end of this finally
 445                        // completes it unless a response was created, in which case the response owns it.
 380446                        frameReader = await responseCompletionSource.Task.ConfigureAwait(false);
 369447                    }
 11448                    catch
 11449                    {
 450                        // observe exception, if any
 11451                    }
 380452                }
 453
 454                lock (_mutex)
 1396455                {
 456                    // Unregister the two-way invocation if registered.
 1396457                    if (requestId > 0 && !_refuseInvocations)
 369458                    {
 369459                        _twowayInvocations.Remove(requestId);
 369460                    }
 461
 1396462                    DecrementDispatchInvocationCount();
 1396463                }
 464
 1396465                if (!responseCreated)
 1027466                {
 1027467                    frameReader?.Complete();
 1027468                }
 469                // else the response owns the PipeReader
 1396470            }
 0471        }
 2771472    }
 473
 474    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 69475    {
 476        lock (_mutex)
 69477        {
 69478            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 479
 67480            if (_shutdownTask is not null)
 0481            {
 0482                throw new InvalidOperationException("Cannot call ShutdownAsync more than once.");
 483            }
 67484            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 3485            {
 3486                throw new InvalidOperationException("Cannot shut down a protocol connection before it's connected.");
 487            }
 488
 64489            RefuseNewInvocations("The connection was shut down.");
 490
 64491            if (_dispatchInvocationCount == 0)
 52492            {
 52493                _dispatchesAndInvocationsCompleted.TrySetResult();
 52494            }
 64495            _shutdownTask = PerformShutdownAsync(_sendCloseConnectionFrame);
 64496        }
 497
 64498        return _shutdownTask;
 499
 500        async Task PerformShutdownAsync(bool sendCloseConnectionFrame)
 64501        {
 64502            await Task.Yield(); // exit mutex lock
 503
 504            try
 64505            {
 64506                Debug.Assert(_readFramesTask is not null);
 507
 508                // Since DisposeAsync waits for the _shutdownTask completion, _disposedCts is not disposed at this
 509                // point.
 64510                using var shutdownCts = CancellationTokenSource.CreateLinkedTokenSource(
 64511                    cancellationToken,
 64512                    _disposedCts.Token);
 513
 514                // Wait for dispatches and invocations to complete.
 64515                await _dispatchesAndInvocationsCompleted.Task.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 516
 517                // Stops sending heartbeats. We can't do earlier: while we're waiting for dispatches and invocations to
 518                // complete, we need to keep sending heartbeats otherwise the peer could see the connection as idle and
 519                // abort it.
 520                lock (_mutex)
 60521                {
 60522                    _heartbeatEnabled = false; // makes _heartbeatTask immutable
 60523                }
 524
 525                // Wait for the last send heartbeat to complete before sending the CloseConnection frame or disposing
 526                // the duplex connection. _heartbeatTask is immutable once _shutdownTask set. _heartbeatTask can be
 527                // canceled by DisposeAsync.
 60528                await _heartbeatTask.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 529
 60530                if (sendCloseConnectionFrame)
 27531                {
 532                    // Send CloseConnection frame.
 27533                    await SendControlFrameAsync(EncodeCloseConnectionFrame, shutdownCts.Token).ConfigureAwait(false);
 534
 535                    // Wait for the peer to abort the connection as an acknowledgment for this CloseConnection frame.
 536                    // The peer can also send us a CloseConnection frame if it started shutting down at the same time.
 537                    // We can't just return and dispose the duplex connection since the peer can still be reading frames
 538                    // (including the CloseConnection frame) and we don't want to abort this reading.
 25539                    await _readFramesTask.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 25540                }
 541                else
 33542                {
 543                    // _readFramesTask should be already completed or nearly completed.
 33544                    await _readFramesTask.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 545
 546                    // _readFramesTask succeeded means the peer is waiting for us to abort the duplex connection;
 547                    // we oblige.
 23548                    _duplexConnection.Dispose();
 23549                }
 48550            }
 7551            catch (OperationCanceledException)
 7552            {
 7553                cancellationToken.ThrowIfCancellationRequested();
 554
 2555                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 2556                throw new IceRpcException(
 2557                    IceRpcError.OperationAborted,
 2558                    "The connection shutdown was aborted because the connection was disposed.");
 559            }
 9560            catch (IceRpcException)
 9561            {
 9562                throw;
 563            }
 0564            catch (Exception exception)
 0565            {
 0566                Debug.Fail($"ShutdownAsync failed with an unexpected exception: {exception}");
 0567                throw;
 568            }
 569
 570            static void EncodeCloseConnectionFrame(IBufferWriter<byte> writer)
 26571            {
 26572                var encoder = new IceEncoder(writer);
 26573                IceDefinitions.CloseConnectionFrame.Encode(ref encoder);
 26574            }
 48575        }
 64576    }
 577
 227578    internal IceProtocolConnection(
 227579        IDuplexConnection duplexConnection,
 227580        TransportConnectionInformation? transportConnectionInformation,
 227581        ConnectionOptions options)
 227582    {
 227583        _twowayDispatchesCts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 584
 585        // With ice, we always listen for incoming frames (responses) so we need a dispatcher for incoming requests even
 586        // if we don't expect any. This dispatcher throws an ice ObjectNotExistException back to the client, which makes
 587        // more sense than throwing an UnknownException.
 227588        _dispatcher = options.Dispatcher ?? NotFoundDispatcher.Instance;
 589
 227590        _maxFrameSize = options.MaxIceFrameSize;
 227591        _transportConnectionInformation = transportConnectionInformation;
 592
 227593        if (options.MaxDispatches > 0)
 227594        {
 227595            _dispatchSemaphore = new SemaphoreSlim(
 227596                initialCount: options.MaxDispatches,
 227597                maxCount: options.MaxDispatches);
 227598        }
 599
 227600        _inactivityTimeout = options.InactivityTimeout;
 601
 602        // The readerScheduler doesn't matter (we don't call pipe.Reader.ReadAsync on the resulting pipe), and the
 603        // writerScheduler doesn't matter (pipe.Writer.FlushAsync never blocks).
 227604        _pipeOptions = new PipeOptions(
 227605            pool: options.Pool,
 227606            minimumSegmentSize: options.MinSegmentSize,
 227607            pauseWriterThreshold: 0,
 227608            useSynchronizationContext: false);
 609
 227610        if (options.IceIdleTimeout != Timeout.InfiniteTimeSpan)
 227611        {
 227612            duplexConnection = new IceDuplexConnectionDecorator(
 227613                duplexConnection,
 227614                readIdleTimeout: options.EnableIceIdleCheck ? options.IceIdleTimeout : Timeout.InfiniteTimeSpan,
 227615                writeIdleTimeout: options.IceIdleTimeout,
 227616                SendHeartbeat);
 227617        }
 618
 227619        _duplexConnection = duplexConnection;
 227620        _duplexConnectionReader = new DuplexConnectionReader(_duplexConnection, options.Pool, options.MinSegmentSize);
 227621        _duplexConnectionWriter =
 227622            new IceDuplexConnectionWriter(_duplexConnection, options.Pool, options.MinSegmentSize);
 623
 227624        _inactivityTimeoutTimer = new Timer(_ =>
 5625        {
 5626            bool requestShutdown = false;
 227627
 227628            lock (_mutex)
 5629            {
 5630                if (_dispatchInvocationCount == 0 && _shutdownTask is null)
 5631                {
 5632                    requestShutdown = true;
 5633                    RefuseNewInvocations(
 5634                        $"The connection was shut down because it was inactive for over {_inactivityTimeout.TotalSeconds
 5635                }
 5636            }
 227637
 5638            if (requestShutdown)
 5639            {
 227640                // TrySetResult must be called outside the mutex lock.
 5641                _shutdownRequestedTcs.TrySetResult();
 5642            }
 232643        });
 644
 645        void SendHeartbeat()
 13646        {
 647            lock (_mutex)
 13648            {
 13649                if (_heartbeatTask.IsCompletedSuccessfully && _heartbeatEnabled)
 13650                {
 13651                    _heartbeatTask = SendValidateConnectionFrameAsync(_disposedCts.Token);
 13652                }
 13653            }
 654
 655            async Task SendValidateConnectionFrameAsync(CancellationToken cancellationToken)
 13656            {
 657                // Make sure we execute the function without holding the connection mutex lock.
 13658                await Task.Yield();
 659
 660                try
 13661                {
 13662                    await SendControlFrameAsync(EncodeValidateConnectionFrame, cancellationToken).ConfigureAwait(false);
 13663                }
 0664                catch (OperationCanceledException)
 0665                {
 666                    // Canceled by DisposeAsync
 0667                    throw;
 668                }
 0669                catch (IceRpcException)
 0670                {
 671                    // Expected, typically the peer aborted the connection.
 0672                    throw;
 673                }
 0674                catch (Exception exception)
 0675                {
 0676                    Debug.Fail($"The heartbeat task completed due to an unhandled exception: {exception}");
 0677                    throw;
 678                }
 679
 680                static void EncodeValidateConnectionFrame(IBufferWriter<byte> writer)
 13681                {
 13682                    var encoder = new IceEncoder(writer);
 13683                    IceDefinitions.ValidateConnectionFrame.Encode(ref encoder);
 13684                }
 13685            }
 13686        }
 227687    }
 688
 689    private static (int RequestId, IceRequestHeader Header, PipeReader? ContextReader, int Consumed) DecodeRequestIdAndH
 690        ReadOnlySequence<byte> buffer)
 1393691    {
 1393692        var decoder = new IceDecoder(buffer);
 693
 1393694        int requestId = decoder.DecodeInt();
 695
 1393696        var requestHeader = new IceRequestHeader(ref decoder);
 1393697        requestHeader.Facet.CheckFacetCount();
 698
 1393699        Pipe? contextPipe = null;
 1393700        long pos = decoder.Consumed;
 1393701        int count = decoder.DecodeSize();
 1393702        if (count > 0)
 7703        {
 28704            for (int i = 0; i < count; ++i)
 7705            {
 7706                decoder.Skip(decoder.DecodeSize()); // Skip the key
 7707                decoder.Skip(decoder.DecodeSize()); // Skip the value
 7708            }
 7709            contextPipe = new Pipe();
 7710            contextPipe.Writer.Write(buffer.Slice(pos, decoder.Consumed - pos));
 7711            contextPipe.Writer.Complete();
 7712        }
 713
 1393714        var encapsulationHeader = new EncapsulationHeader(ref decoder);
 715
 1393716        if (encapsulationHeader.PayloadEncodingMajor != 1 ||
 1393717            encapsulationHeader.PayloadEncodingMinor != 1)
 0718        {
 0719            throw new InvalidDataException(
 0720                $"Unsupported payload encoding '{encapsulationHeader.PayloadEncodingMajor}.{encapsulationHeader.PayloadE
 721        }
 722
 1393723        int payloadSize = encapsulationHeader.EncapsulationSize - 6;
 1393724        if (payloadSize != (buffer.Length - decoder.Consumed))
 0725        {
 0726            throw new InvalidDataException(
 0727                $"Request payload size mismatch: expected {payloadSize} bytes, read {buffer.Length - decoder.Consumed} b
 728        }
 729
 1393730        return (requestId, requestHeader, contextPipe?.Reader, (int)decoder.Consumed);
 1393731    }
 732
 733    private static (StatusCode StatusCode, string? ErrorMessage, SequencePosition Consumed) DecodeResponseHeader(
 734        ReadOnlySequence<byte> buffer,
 735        int requestId)
 369736    {
 369737        var replyStatus = (ReplyStatus)buffer.FirstSpan[0];
 738
 369739        if (replyStatus <= ReplyStatus.UserException)
 333740        {
 741            const int headerSize = 7; // reply status byte + encapsulation header
 742
 743            // read and check encapsulation header (6 bytes long)
 744
 333745            if (buffer.Length < headerSize)
 0746            {
 0747                throw new InvalidDataException($"Received invalid frame header for request with id '{requestId}'.");
 748            }
 749
 333750            EncapsulationHeader encapsulationHeader =
 666751                buffer.Slice(1, 6).DecodeIceBuffer((ref IceDecoder decoder) => new EncapsulationHeader(ref decoder));
 752
 753            // Sanity check
 333754            int payloadSize = encapsulationHeader.EncapsulationSize - 6;
 333755            if (payloadSize != buffer.Length - headerSize)
 0756            {
 0757                throw new InvalidDataException(
 0758                    $"Response payload size/frame size mismatch: payload size is {payloadSize} bytes but frame has {buff
 759            }
 760
 333761            SequencePosition consumed = buffer.GetPosition(headerSize);
 762
 333763            return replyStatus == ReplyStatus.Ok ? (StatusCode.Ok, null, consumed) :
 333764                // Set the error message to the empty string, because null is not allowed for status code > Ok.
 333765                (StatusCode.ApplicationError, "", consumed);
 766        }
 767        else
 36768        {
 769            // An ice system exception.
 770
 36771            StatusCode statusCode = replyStatus switch
 36772            {
 14773                ReplyStatus.ObjectNotExist => StatusCode.NotFound,
 0774                ReplyStatus.FacetNotExist => StatusCode.NotFound,
 2775                ReplyStatus.OperationNotExist => StatusCode.NotImplemented,
 4776                ReplyStatus.InvalidData => StatusCode.InvalidData,
 1777                ReplyStatus.Unauthorized => StatusCode.Unauthorized,
 1778                ReplyStatus.NotSupported => StatusCode.NotSupported,
 14779                _ => StatusCode.InternalError
 36780            };
 781
 36782            var decoder = new IceDecoder(buffer.Slice(1));
 783
 784            string message;
 36785            switch (replyStatus)
 786            {
 787                case ReplyStatus.FacetNotExist:
 788                case ReplyStatus.ObjectNotExist:
 789                case ReplyStatus.OperationNotExist:
 790
 16791                    var requestFailed = new RequestFailedExceptionData(ref decoder);
 792
 16793                    string target = requestFailed.Facet.Count > 0 ?
 16794                        $"{requestFailed.Identity.ToPath()}#{requestFailed.Facet.ToFragment()}" : requestFailed.Identity
 795
 16796                    message =
 16797                        $"The dispatch failed with status code {statusCode} while dispatching '{requestFailed.Operation}
 16798                    break;
 799                default:
 20800                    message = decoder.DecodeString();
 20801                    break;
 802            }
 36803            decoder.CheckEndOfBuffer();
 36804            return (statusCode, message, buffer.End);
 805        }
 369806    }
 807
 808    private static void EncodeRequestHeader(
 809        IceDuplexConnectionWriter output,
 810        OutgoingRequest request,
 811        int requestId,
 812        int payloadSize)
 1396813    {
 1396814        var encoder = new IceEncoder(output);
 815
 816        // Write the request header.
 1396817        encoder.WriteByteSpan(IceDefinitions.FramePrologue);
 1396818        encoder.EncodeIceFrameType(IceFrameType.Request);
 1396819        encoder.EncodeByte(0); // compression status
 820
 1396821        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(4);
 822
 1396823        encoder.EncodeInt(requestId);
 824
 1396825        byte encodingMajor = 1;
 1396826        byte encodingMinor = 1;
 827
 828        // Request header.
 1396829        var requestHeader = new IceRequestHeader(
 1396830            IceIdentity.Parse(request.ServiceAddress.Path),
 1396831            request.ServiceAddress.Fragment.ToFacet(),
 1396832            request.Operation,
 1396833            request.Fields.ContainsKey(RequestFieldKey.Idempotent) ? OperationMode.Idempotent : OperationMode.Normal);
 1396834        requestHeader.Encode(ref encoder);
 1396835        int directWriteSize = 0;
 1396836        if (request.Fields.TryGetValue(RequestFieldKey.Context, out OutgoingFieldValue requestField))
 7837        {
 7838            if (requestField.WriteAction is Action<IBufferWriter<byte>> writeAction)
 7839            {
 840                // This writes directly to the underlying output; we measure how many bytes are written.
 7841                long start = output.UnflushedBytes;
 7842                writeAction(output);
 7843                directWriteSize = (int)(output.UnflushedBytes - start);
 7844            }
 845            else
 0846            {
 0847                encoder.WriteByteSequence(requestField.ByteSequence);
 0848            }
 7849        }
 850        else
 1389851        {
 1389852            encoder.EncodeSize(0);
 1389853        }
 854
 855        // We ignore all other fields. They can't be sent over ice.
 856
 1396857        new EncapsulationHeader(
 1396858            encapsulationSize: payloadSize + 6,
 1396859            encodingMajor,
 1396860            encodingMinor).Encode(ref encoder);
 861
 1396862        int frameSize = checked(encoder.EncodedByteCount + directWriteSize + payloadSize);
 1396863        IceEncoder.EncodeInt(frameSize, sizePlaceholder);
 1396864    }
 865
 866    private static void EncodeResponseHeader(
 867        IBufferWriter<byte> writer,
 868        OutgoingResponse response,
 869        IncomingRequest request,
 870        int requestId,
 871        int payloadSize)
 1380872    {
 1380873        var encoder = new IceEncoder(writer);
 874
 875        // Write the response header.
 876
 1380877        encoder.WriteByteSpan(IceDefinitions.FramePrologue);
 1380878        encoder.EncodeIceFrameType(IceFrameType.Reply);
 1380879        encoder.EncodeByte(0); // compression status
 1380880        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(4);
 881
 1380882        encoder.EncodeInt(requestId);
 883
 1380884        if (response.StatusCode > StatusCode.ApplicationError ||
 1380885            (response.StatusCode == StatusCode.ApplicationError && payloadSize == 0))
 38886        {
 887            // system exception
 38888            switch (response.StatusCode)
 889            {
 890                case StatusCode.NotFound:
 891                case StatusCode.NotImplemented:
 18892                    encoder.EncodeReplyStatus(response.StatusCode == StatusCode.NotFound ?
 18893                        ReplyStatus.ObjectNotExist : ReplyStatus.OperationNotExist);
 894
 18895                    new RequestFailedExceptionData(
 18896                        IceIdentity.Parse(request.Path),
 18897                        request.Fragment.ToFacet(),
 18898                        request.Operation).Encode(ref encoder);
 18899                    break;
 900                case StatusCode.InternalError:
 8901                    encoder.EncodeReplyStatus(ReplyStatus.UnknownException);
 8902                    encoder.EncodeString(response.ErrorMessage!);
 8903                    break;
 904                case StatusCode.InvalidData:
 4905                    encoder.EncodeReplyStatus(ReplyStatus.InvalidData);
 4906                    encoder.EncodeString(response.ErrorMessage!);
 4907                    break;
 908                case StatusCode.Unauthorized:
 1909                    encoder.EncodeReplyStatus(ReplyStatus.Unauthorized);
 1910                    encoder.EncodeString(response.ErrorMessage!);
 1911                    break;
 912                case StatusCode.NotSupported:
 1913                    encoder.EncodeReplyStatus(ReplyStatus.NotSupported);
 1914                    encoder.EncodeString(response.ErrorMessage!);
 1915                    break;
 916                default:
 6917                    encoder.EncodeReplyStatus(ReplyStatus.UnknownException);
 6918                    encoder.EncodeString(
 6919                        $"{response.ErrorMessage} {{ Original StatusCode = {response.StatusCode} }}");
 6920                    break;
 921            }
 38922        }
 923        else
 1342924        {
 1342925            encoder.EncodeReplyStatus((ReplyStatus)response.StatusCode);
 926
 927            // When IceRPC receives a response, it ignores the response encoding. So this "1.1" is only relevant to
 928            // a ZeroC Ice client that decodes the response. The only Slice encoding such a client can possibly use
 929            // to decode the response payload is 1.1 or 1.0, and we don't care about interop with 1.0.
 1342930            var encapsulationHeader = new EncapsulationHeader(
 1342931                encapsulationSize: payloadSize + 6,
 1342932                payloadEncodingMajor: 1,
 1342933                payloadEncodingMinor: 1);
 1342934            encapsulationHeader.Encode(ref encoder);
 1342935        }
 936
 1380937        int frameSize = encoder.EncodedByteCount + payloadSize;
 1380938        IceEncoder.EncodeInt(frameSize, sizePlaceholder);
 1380939    }
 940
 941    /// <summary>Reads the full Ice payload from the given pipe reader.</summary>
 942    private static async ValueTask<ReadOnlySequence<byte>> ReadFullPayloadAsync(
 943        PipeReader payload,
 944        CancellationToken cancellationToken)
 2743945    {
 946        // We use ReadAtLeastAsync instead of ReadAsync to bypass the PauseWriterThreshold when the payload is
 947        // backed by a Pipe.
 2743948        ReadResult readResult = await payload.ReadAtLeastAsync(int.MaxValue, cancellationToken).ConfigureAwait(false);
 949
 2740950        if (readResult.IsCanceled)
 0951        {
 0952            throw new InvalidOperationException("Unexpected call to CancelPendingRead on ice payload.");
 953        }
 954
 2740955        return readResult.IsCompleted ? readResult.Buffer :
 2740956            throw new ArgumentException("The payload size is greater than int.MaxValue.", nameof(payload));
 2740957    }
 958
 959    /// <summary>Acquires exclusive access to _duplexConnectionWriter.</summary>
 960    /// <returns>A <see cref="SemaphoreLock" /> that releases the acquired semaphore in its Dispose method.</returns>
 961    private async ValueTask<SemaphoreLock> AcquireWriteLockAsync(CancellationToken cancellationToken)
 2921962    {
 2921963        SemaphoreLock semaphoreLock = await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false);
 964
 965        // _writeException is protected by _writeSemaphore
 2921966        if (_writeException is not null)
 1967        {
 1968            semaphoreLock.Dispose();
 969
 1970            throw new IceRpcException(
 1971                IceRpcError.ConnectionAborted,
 1972                "The connection was aborted because a previous write operation failed.",
 1973                _writeException);
 974        }
 975
 2920976        return semaphoreLock;
 2920977    }
 978
 979    /// <summary>Creates a pipe reader to simplify the reading of a request or response frame. The frame is read fully
 980    /// and buffered into an internal pipe.</summary>
 981    private async ValueTask<PipeReader> CreateFrameReaderAsync(int size, CancellationToken cancellationToken)
 2767982    {
 2767983        var pipe = new Pipe(_pipeOptions);
 984
 985        try
 2767986        {
 2767987            await _duplexConnectionReader.FillBufferWriterAsync(pipe.Writer, size, cancellationToken)
 2767988                .ConfigureAwait(false);
 2767989        }
 0990        catch
 0991        {
 0992            pipe.Reader.Complete();
 0993            throw;
 994        }
 995        finally
 2767996        {
 2767997            pipe.Writer.Complete();
 2767998        }
 999
 27671000        return pipe.Reader;
 27671001    }
 1002
 1003    private void DecrementDispatchInvocationCount()
 27871004    {
 1005        lock (_mutex)
 27871006        {
 27871007            if (--_dispatchInvocationCount == 0)
 12411008            {
 12411009                if (_shutdownTask is not null)
 191010                {
 191011                    _dispatchesAndInvocationsCompleted.TrySetResult();
 191012                }
 1013                // We enable the inactivity check in order to complete ShutdownRequested when inactive for too long.
 1014                // _refuseInvocations is true when the connection is either about to be "shutdown requested", or shut
 1015                // down / disposed. We don't need to complete ShutdownRequested in any of these situations.
 12221016                else if (!_refuseInvocations)
 12071017                {
 12071018                    ScheduleInactivityCheck();
 12071019                }
 12411020            }
 27871021        }
 27871022    }
 1023
 1024    /// <summary>Dispatches an incoming request. This method executes in a task spawn from the read frames loop.
 1025    /// </summary>
 1026    private async Task DispatchRequestAsync(IncomingRequest request, int requestId, PipeReader? contextReader)
 13911027    {
 13911028        CancellationToken cancellationToken = request.IsOneway ? _disposedCts.Token : _twowayDispatchesCts.Token;
 1029
 1030        OutgoingResponse? response;
 1031        try
 13911032        {
 1033            // The dispatcher can complete the incoming request payload to release its memory as soon as possible.
 1034            try
 13911035            {
 1036                // _dispatcher.DispatchAsync may very well ignore the cancellation token and we don't want to keep
 1037                // dispatching when the cancellation token is canceled.
 13911038                cancellationToken.ThrowIfCancellationRequested();
 1039
 13911040                response = await _dispatcher.DispatchAsync(request, cancellationToken).ConfigureAwait(false);
 13721041            }
 1042            finally
 13911043            {
 13911044                _dispatchSemaphore?.Release();
 13911045            }
 1046
 13721047            if (response != request.Response)
 11048            {
 11049                throw new InvalidOperationException(
 11050                    "The dispatcher did not return the last response created for this request.");
 1051            }
 13711052        }
 201053        catch when (request.IsOneway)
 01054        {
 1055            // ignored since we're not returning anything
 01056            response = null;
 01057        }
 101058        catch (OperationCanceledException exception) when (exception.CancellationToken == cancellationToken)
 91059        {
 1060            // expected when the connection is disposed or the request is canceled by the peer's shutdown
 91061            response = null;
 91062        }
 111063        catch (Exception exception)
 111064        {
 111065            if (exception is not DispatchException dispatchException)
 71066            {
 71067                StatusCode statusCode = exception is InvalidDataException ?
 71068                    StatusCode.InvalidData : StatusCode.InternalError;
 71069                dispatchException = new DispatchException(statusCode, innerException: exception);
 71070            }
 111071            response = dispatchException.ToOutgoingResponse(request);
 111072        }
 1073        finally
 13911074        {
 13911075            request.Payload.Complete();
 13911076            contextReader?.Complete();
 1077
 1078            // The field values are now invalid - they point to potentially recycled and reused memory. We
 1079            // replace Fields by an empty dictionary to prevent accidental access to this reused memory.
 13911080            request.Fields = ImmutableDictionary<RequestFieldKey, ReadOnlySequence<byte>>.Empty;
 13911081        }
 1082
 1083        try
 13911084        {
 13911085            if (response is not null)
 13821086            {
 1087                // Read the full response payload. This can take some time so this needs to be done before acquiring
 1088                // the write semaphore.
 13821089                ReadOnlySequence<byte> payload = ReadOnlySequence<byte>.Empty;
 1090
 13821091                if (response.StatusCode <= StatusCode.ApplicationError)
 13471092                {
 1093                    try
 13471094                    {
 13471095                        payload = await ReadFullPayloadAsync(response.Payload, cancellationToken)
 13471096                            .ConfigureAwait(false);
 13441097                    }
 21098                    catch (OperationCanceledException exception) when (exception.CancellationToken == cancellationToken)
 21099                    {
 21100                        throw;
 1101                    }
 11102                    catch (Exception exception)
 11103                    {
 1104                        // We "encode" the exception in the error message.
 1105
 11106                        response = new OutgoingResponse(
 11107                            request,
 11108                            StatusCode.InternalError,
 11109                            "The dispatch failed to read the response payload.",
 11110                            exception);
 11111                    }
 13451112                }
 1113                // else payload remains empty because the payload of a dispatch exception (if any) cannot be sent
 1114                // over ice.
 1115
 13801116                int payloadSize = checked((int)payload.Length);
 1117
 1118                // Wait for writing of other frames to complete.
 13801119                using SemaphoreLock _ = await AcquireWriteLockAsync(cancellationToken).ConfigureAwait(false);
 1120                try
 13801121                {
 13801122                    EncodeResponseHeader(_duplexConnectionWriter, response, request, requestId, payloadSize);
 1123
 1124                    // We write to the duplex connection with _disposedCts.Token instead of cancellationToken.
 1125                    // Canceling this write operation is fatal to the connection.
 13801126                    await _duplexConnectionWriter.WriteAsync(payload, _disposedCts.Token).ConfigureAwait(false);
 13791127                }
 11128                catch (Exception exception)
 11129                {
 11130                    WriteFailed(exception);
 11131                    throw;
 1132                }
 13791133            }
 13881134        }
 31135        catch (OperationCanceledException exception) when (
 31136            exception.CancellationToken == _disposedCts.Token ||
 31137            exception.CancellationToken == cancellationToken)
 31138        {
 1139            // expected when the connection is disposed or the request is canceled by the peer's shutdown
 31140        }
 1141        finally
 13911142        {
 13911143            DecrementDispatchInvocationCount();
 13911144        }
 13911145    }
 1146
 1147    /// <summary>Increments the dispatch-invocation count.</summary>
 1148    /// <remarks>This method must be called with _mutex locked.</remarks>
 1149    private void IncrementDispatchInvocationCount()
 27871150    {
 27871151        if (_dispatchInvocationCount++ == 0)
 12411152        {
 1153            // Cancel inactivity check.
 12411154            _inactivityTimeoutTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
 12411155        }
 27871156    }
 1157
 1158    private void ScheduleInactivityCheck() =>
 14101159        _inactivityTimeoutTimer.Change(_inactivityTimeout, Timeout.InfiniteTimeSpan);
 1160
 1161    /// <summary>Reads incoming frames and returns successfully when a CloseConnection frame is received or when the
 1162    /// connection is aborted during ShutdownAsync or canceled by DisposeAsync.</summary>
 1163    private async Task ReadFramesAsync(CancellationToken cancellationToken)
 2031164    {
 2031165        await Task.Yield(); // exit mutex lock
 1166
 1167        // Wait for _connectTask (which spawned the task running this method) to complete. This way, we won't dispatch
 1168        // any request until _connectTask has completed successfully, and indirectly we won't make any invocation until
 1169        // _connectTask has completed successfully. The creation of the _readFramesTask is the last action taken by
 1170        // _connectTask and as a result this await can't fail.
 2031171        await _connectTask!.ConfigureAwait(false);
 1172
 1173        try
 2031174        {
 29831175            while (!cancellationToken.IsCancellationRequested)
 29811176            {
 29811177                ReadOnlySequence<byte> buffer = await _duplexConnectionReader.ReadAtLeastAsync(
 29811178                    IceDefinitions.PrologueSize,
 29811179                    cancellationToken).ConfigureAwait(false);
 1180
 1181                // First decode and check the prologue.
 1182
 28081183                ReadOnlySequence<byte> prologueBuffer = buffer.Slice(0, IceDefinitions.PrologueSize);
 1184
 28081185                IcePrologue prologue =
 56161186                    prologueBuffer.DecodeIceBuffer((ref IceDecoder decoder) => new IcePrologue(ref decoder));
 1187
 28081188                _duplexConnectionReader.AdvanceTo(prologueBuffer.End);
 1189
 28081190                IceDefinitions.CheckPrologue(prologue);
 28071191                if (prologue.FrameSize > _maxFrameSize)
 11192                {
 11193                    throw new InvalidDataException(
 11194                        $"Received frame with size ({prologue.FrameSize}) greater than max frame size.");
 1195                }
 28061196                if (prologue.FrameSize < IceDefinitions.PrologueSize)
 11197                {
 11198                    throw new InvalidDataException(
 11199                        $"Received frame with size ({prologue.FrameSize}) smaller than the prologue size.");
 1200                }
 1201
 28051202                if (prologue.CompressionStatus == 2)
 01203                {
 1204                    // The exception handler calls ReadFailed.
 01205                    throw new IceRpcException(
 01206                        IceRpcError.ConnectionAborted,
 01207                        "The connection was aborted because it received a compressed ice frame, and IceRPC does not supp
 1208                }
 1209
 1210                // Then process the frame based on its type.
 28051211                switch (prologue.FrameType)
 1212                {
 1213                    case IceFrameType.CloseConnection:
 251214                    {
 251215                        if (prologue.FrameSize != IceDefinitions.PrologueSize)
 01216                        {
 01217                            throw new InvalidDataException(
 01218                                $"Received {nameof(IceFrameType.CloseConnection)} frame with unexpected data.");
 1219                        }
 1220
 1221                        lock (_mutex)
 251222                        {
 251223                            RefuseNewInvocations(
 251224                                "The connection was shut down because it received a CloseConnection frame from the peer.
 1225
 1226                            // By exiting the "read frames loop" below, we are refusing new dispatches as well.
 1227
 1228                            // Only one side sends the CloseConnection frame.
 251229                            _sendCloseConnectionFrame = false;
 251230                        }
 1231
 1232                        // Even though we're in the "read frames loop", it's ok to cancel CTS and a "synchronous" TCS
 1233                        // below. We won't be reading anything else so it's ok to run continuations synchronously.
 1234
 1235                        // Abort two-way invocations that are waiting for a response (it will never come).
 1236                        // We use InvocationCanceled (not ConnectionAborted) because the ice protocol guarantees the
 1237                        // peer has sent responses for all two-way requests it accepted before sending CloseConnection.
 1238                        // These pending two-way requests were never processed by the peer, so it's safe for a retry
 1239                        // interceptor to retry them unconditionally.
 251240                        AbortTwowayInvocations(
 251241                            IceRpcError.InvocationCanceled,
 251242                            "The invocation was canceled by the shutdown of the peer.");
 1243
 1244                        // Cancel two-way dispatches since the peer is not interested in the responses. This does not
 1245                        // cancel ongoing writes to _duplexConnection: we don't send incomplete/invalid data.
 251246                        _twowayDispatchesCts.Cancel();
 1247
 1248                        // We keep sending heartbeats. If the shutdown request / shutdown is not fulfilled quickly, they
 1249                        // tell the peer we're still alive and maybe stuck waiting for invocations and dispatches to
 1250                        // complete.
 1251
 1252                        // We request a shutdown that will dispose _duplexConnection once all invocations and dispatches
 1253                        // have completed.
 251254                        _shutdownRequestedTcs.TrySetResult();
 251255                        return;
 1256                    }
 1257
 1258                    case IceFrameType.Request:
 13931259                        await ReadRequestAsync(prologue.FrameSize, cancellationToken).ConfigureAwait(false);
 13931260                        break;
 1261
 1262                    case IceFrameType.RequestBatch:
 1263                        // The exception handler calls ReadFailed.
 01264                        throw new IceRpcException(
 01265                            IceRpcError.ConnectionAborted,
 01266                            "The connection was aborted because it received a batch request, and IceRPC does not support
 1267
 1268                    case IceFrameType.Reply:
 13741269                        await ReadReplyAsync(prologue.FrameSize, cancellationToken).ConfigureAwait(false);
 13741270                        break;
 1271
 1272                    case IceFrameType.ValidateConnection:
 131273                    {
 131274                        if (prologue.FrameSize != IceDefinitions.PrologueSize)
 01275                        {
 01276                            throw new InvalidDataException(
 01277                                $"Received {nameof(IceFrameType.ValidateConnection)} frame with unexpected data.");
 1278                        }
 131279                        break;
 1280                    }
 1281
 1282                    default:
 01283                    {
 01284                        throw new InvalidDataException(
 01285                            $"Received Ice frame with unknown frame type '{prologue.FrameType}'.");
 1286                    }
 1287                }
 27801288            } // while
 21289        }
 701290        catch (OperationCanceledException)
 701291        {
 1292            // canceled by DisposeAsync, no need to throw anything
 701293        }
 1031294        catch (IceRpcException exception) when (
 1031295            exception.IceRpcError == IceRpcError.ConnectionAborted &&
 1031296            _dispatchesAndInvocationsCompleted.Task.IsCompleted)
 691297        {
 1298            // The peer acknowledged receipt of the CloseConnection frame by aborting the duplex connection. Return.
 1299            // See ShutdownAsync.
 691300        }
 341301        catch (IceRpcException exception)
 341302        {
 341303            ReadFailed(exception);
 341304            throw;
 1305        }
 31306        catch (InvalidDataException exception)
 31307        {
 31308            ReadFailed(exception);
 31309            throw new IceRpcException(
 31310                IceRpcError.ConnectionAborted,
 31311                "The connection was aborted by an ice protocol error.",
 31312                exception);
 1313        }
 01314        catch (Exception exception)
 01315        {
 01316            Debug.Fail($"The read frames task completed due to an unhandled exception: {exception}");
 01317            ReadFailed(exception);
 01318            throw;
 1319        }
 1320
 1321        // Aborts all pending two-way invocations. Must be called outside the mutex lock after setting
 1322        // _refuseInvocations to true.
 1323        void AbortTwowayInvocations(IceRpcError error, string message, Exception? exception = null)
 621324        {
 621325            Debug.Assert(_refuseInvocations);
 1326
 1327            // _twowayInvocations is immutable once _refuseInvocations is true.
 2101328            foreach (TaskCompletionSource<PipeReader> responseCompletionSource in _twowayInvocations.Values)
 121329            {
 1330                // _twowayInvocations can hold completed completion sources.
 121331                _ = responseCompletionSource.TrySetException(new IceRpcException(error, message, exception));
 121332            }
 621333        }
 1334
 1335        // Takes appropriate action after a read failure.
 1336        void ReadFailed(Exception exception)
 371337        {
 1338            // We also prevent new one-way invocations even though they don't need to read the connection.
 371339            RefuseNewInvocations("The connection was lost because a read operation failed.");
 1340
 1341            // It's ok to cancel CTS and a "synchronous" TCS below. We won't be reading anything else so it's ok to run
 1342            // continuations synchronously.
 1343
 371344            AbortTwowayInvocations(
 371345                IceRpcError.ConnectionAborted,
 371346                "The invocation was aborted because the connection was lost.",
 371347                exception);
 1348
 1349            // ReadFailed is called when the connection is dead or the peer sent us a non-supported frame (e.g. a
 1350            // batch request). We don't need to allow outstanding two-way dispatches to complete in these situations, so
 1351            // we cancel them to speed-up the shutdown.
 371352            _twowayDispatchesCts.Cancel();
 1353
 1354            lock (_mutex)
 371355            {
 1356                // Don't send a close connection frame since we can't wait for the peer's acknowledgment.
 371357                _sendCloseConnectionFrame = false;
 371358            }
 1359
 371360            _ = _shutdownRequestedTcs.TrySetResult();
 371361        }
 1661362    }
 1363
 1364    /// <summary>Reads a reply (incoming response) and completes the invocation response completion source with this
 1365    /// response. This method executes "synchronously" in the read frames loop.</summary>
 1366    private async Task ReadReplyAsync(int replyFrameSize, CancellationToken cancellationToken)
 13741367    {
 1368        // Read the remainder of the frame immediately into frameReader.
 13741369        PipeReader replyFrameReader = await CreateFrameReaderAsync(
 13741370            replyFrameSize - IceDefinitions.PrologueSize,
 13741371            cancellationToken).ConfigureAwait(false);
 1372
 13741373        bool completeFrameReader = true;
 1374
 1375        try
 13741376        {
 1377            // Read and decode request ID
 13741378            if (!replyFrameReader.TryRead(out ReadResult readResult) || readResult.Buffer.Length < 4)
 01379            {
 01380                throw new InvalidDataException("Received a response with an invalid request ID.");
 1381            }
 1382
 13741383            ReadOnlySequence<byte> requestIdBuffer = readResult.Buffer.Slice(0, 4);
 27481384            int requestId = requestIdBuffer.DecodeIceBuffer((ref IceDecoder decoder) => decoder.DecodeInt());
 13741385            replyFrameReader.AdvanceTo(requestIdBuffer.End);
 1386
 1387            lock (_mutex)
 13741388            {
 13741389                if (_twowayInvocations.TryGetValue(
 13741390                    requestId,
 13741391                    out TaskCompletionSource<PipeReader>? responseCompletionSource))
 3691392                {
 1393                    // continuation runs asynchronously
 3691394                    if (responseCompletionSource.TrySetResult(replyFrameReader))
 3691395                    {
 3691396                        completeFrameReader = false;
 3691397                    }
 1398                    // else this invocation just completed and is about to remove itself from _twowayInvocations,
 1399                    // or _twowayInvocations is immutable and contains entries for completed invocations.
 3691400                }
 1401                // else the request ID carried by the response is bogus or corresponds to a request that was previously
 1402                // discarded (for example, because its deadline expired).
 13741403            }
 13741404        }
 1405        finally
 13741406        {
 13741407            if (completeFrameReader)
 10051408            {
 10051409                replyFrameReader.Complete();
 10051410            }
 13741411        }
 13741412    }
 1413
 1414    /// <summary>Reads and then dispatches an incoming request in a separate dispatch task. This method executes
 1415    /// "synchronously" in the read frames loop.</summary>
 1416    private async Task ReadRequestAsync(int requestFrameSize, CancellationToken cancellationToken)
 13931417    {
 1418        // Read the request frame.
 13931419        PipeReader requestFrameReader = await CreateFrameReaderAsync(
 13931420            requestFrameSize - IceDefinitions.PrologueSize,
 13931421            cancellationToken).ConfigureAwait(false);
 1422
 1423        // Decode its header.
 1424        int requestId;
 1425        IceRequestHeader requestHeader;
 13931426        PipeReader? contextReader = null;
 1427        IDictionary<RequestFieldKey, ReadOnlySequence<byte>>? fields;
 13931428        Task? dispatchTask = null;
 1429
 1430        try
 13931431        {
 13931432            if (!requestFrameReader.TryRead(out ReadResult readResult))
 01433            {
 01434                throw new InvalidDataException("Received an invalid request frame.");
 1435            }
 1436
 13931437            Debug.Assert(readResult.IsCompleted);
 1438
 13931439            (requestId, requestHeader, contextReader, int consumed) = DecodeRequestIdAndHeader(readResult.Buffer);
 13931440            requestFrameReader.AdvanceTo(readResult.Buffer.GetPosition(consumed));
 1441
 13931442            if (contextReader is null)
 13861443            {
 13861444                fields = requestHeader.OperationMode == OperationMode.Normal ?
 13861445                    ImmutableDictionary<RequestFieldKey, ReadOnlySequence<byte>>.Empty : _idempotentFields;
 13861446            }
 1447            else
 71448            {
 71449                contextReader.TryRead(out ReadResult result);
 71450                Debug.Assert(result.Buffer.Length > 0 && result.IsCompleted);
 71451                fields = new Dictionary<RequestFieldKey, ReadOnlySequence<byte>>()
 71452                {
 71453                    [RequestFieldKey.Context] = result.Buffer
 71454                };
 1455
 71456                if (requestHeader.OperationMode != OperationMode.Normal)
 01457                {
 1458                    // OperationMode can be Idempotent or Nonmutating.
 01459                    fields[RequestFieldKey.Idempotent] = default;
 01460                }
 71461            }
 1462
 13931463            bool releaseDispatchSemaphore = false;
 13931464            if (_dispatchSemaphore is SemaphoreSlim dispatchSemaphore)
 13931465            {
 1466                // This prevents us from receiving any new frames if we're already dispatching the maximum number
 1467                // of requests. We need to do this in the "accept from network loop" to apply back pressure to the
 1468                // caller.
 1469                try
 13931470                {
 13931471                    await dispatchSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
 13921472                    releaseDispatchSemaphore = true;
 13921473                }
 11474                catch (OperationCanceledException)
 11475                {
 1476                    // and return below
 11477                }
 13931478            }
 1479
 1480            lock (_mutex)
 13931481            {
 13931482                if (_shutdownTask is not null)
 21483                {
 1484                    // The connection is (being) disposed or the connection is shutting down and received a request.
 1485                    // We simply discard it. For a graceful shutdown, the two-way invocation in the peer will throw
 1486                    // IceRpcException(InvocationCanceled). We also discard one-way requests: if we accepted them, they
 1487                    // could delay our shutdown and make it time out.
 21488                    if (releaseDispatchSemaphore)
 11489                    {
 11490                        _dispatchSemaphore!.Release();
 11491                    }
 21492                    return;
 1493                }
 1494
 13911495                IncrementDispatchInvocationCount();
 13911496            }
 1497
 1498            // The scheduling of the task can't be canceled since we want to make sure DispatchRequestAsync will
 1499            // cleanup (decrement _dispatchCount etc.) if DisposeAsync is called. dispatchTask takes ownership of the
 1500            // requestFrameReader and contextReader.
 13911501            dispatchTask = Task.Run(
 13911502                async () =>
 13911503                {
 13911504                    using var request = new IncomingRequest(Protocol.Ice, _connectionContext!)
 13911505                    {
 13911506                        Fields = fields,
 13911507                        Fragment = requestHeader.Facet.ToFragment(),
 13911508                        IsOneway = requestId == 0,
 13911509                        Operation = requestHeader.Operation,
 13911510                        Path = requestHeader.Identity.ToPath(),
 13911511                        Payload = requestFrameReader,
 13911512                    };
 13911513
 13911514                    try
 13911515                    {
 13911516                        await DispatchRequestAsync(
 13911517                            request,
 13911518                            requestId,
 13911519                            contextReader).ConfigureAwait(false);
 13911520                    }
 01521                    catch (IceRpcException)
 01522                    {
 13911523                        // expected when the peer aborts the connection.
 01524                    }
 01525                    catch (Exception exception)
 01526                    {
 13911527                        // With ice, a dispatch cannot throw an exception that comes from the application code:
 13911528                        // any exception thrown when reading the response payload is converted into a DispatchException
 13911529                        // response, and the response header has no fields to encode.
 01530                        Debug.Fail($"ice dispatch {request} failed with an unexpected exception: {exception}");
 01531                        throw;
 13911532                    }
 13911533                },
 13911534                CancellationToken.None);
 13911535        }
 1536        finally
 13931537        {
 13931538            if (dispatchTask is null)
 21539            {
 21540                requestFrameReader.Complete();
 21541                contextReader?.Complete();
 21542            }
 13931543        }
 13931544    }
 1545
 1546    private void RefuseNewInvocations(string message)
 3631547    {
 1548        lock (_mutex)
 3631549        {
 3631550            _refuseInvocations = true;
 3631551            _invocationRefusedMessage ??= message;
 3631552        }
 3631553    }
 1554
 1555    /// <summary>Sends a control frame. It takes care of acquiring and releasing the write lock and calls
 1556    /// <see cref="WriteFailed" /> if a failure occurs while writing to _duplexConnectionWriter.</summary>
 1557    /// <param name="encode">Encodes the control frame.</param>
 1558    /// <param name="cancellationToken">The cancellation token.</param>
 1559    /// <remarks>If the cancellation token is canceled while writing to the duplex connection, the connection is
 1560    /// aborted.</remarks>
 1561    private async ValueTask SendControlFrameAsync(
 1562        Action<IBufferWriter<byte>> encode,
 1563        CancellationToken cancellationToken)
 1451564    {
 1451565        using SemaphoreLock _ = await AcquireWriteLockAsync(cancellationToken).ConfigureAwait(false);
 1566
 1567        try
 1441568        {
 1441569            encode(_duplexConnectionWriter);
 1441570            await _duplexConnectionWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
 1411571        }
 31572        catch (Exception exception)
 31573        {
 31574            WriteFailed(exception);
 31575            throw;
 1576        }
 1411577    }
 1578
 1579    /// <summary>Takes appropriate action after a write failure.</summary>
 1580    /// <remarks>Must be called outside the mutex lock but after acquiring _writeSemaphore.</remarks>
 1581    private void WriteFailed(Exception exception)
 51582    {
 51583        Debug.Assert(_writeException is null);
 51584        _writeException = exception; // protected by _writeSemaphore
 1585
 1586        // We can't send new invocations without writing to the connection.
 51587        RefuseNewInvocations("The connection was lost because a write operation failed.");
 1588
 1589        // We can't send responses so these dispatches can be canceled.
 51590        _twowayDispatchesCts.Cancel();
 1591
 1592        // We don't change _sendClosedConnectionFrame. If the _readFrameTask is still running, we want ShutdownAsync
 1593        // to send CloseConnection - and fail.
 1594
 51595        _ = _shutdownRequestedTcs.TrySetResult();
 51596    }
 1597}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

.cctor()
get_IsServer()
.ctor(IceRpc.Transports.IDuplexConnection,IceRpc.Transports.TransportConnectionInformation,IceRpc.ConnectionOptions)
ConnectAsync(System.Threading.CancellationToken)
PerformConnectAsync()
EncodeValidateConnectionFrame()
DecodeValidateConnectionFrame()
DisposeAsync()
PerformDisposeAsync()
InvokeAsync(IceRpc.OutgoingRequest,System.Threading.CancellationToken)
PerformInvokeAsync()
ShutdownAsync(System.Threading.CancellationToken)
PerformShutdownAsync()
EncodeCloseConnectionFrame()
SendHeartbeat()
SendValidateConnectionFrameAsync()
EncodeValidateConnectionFrame()
DecodeRequestIdAndHeader(System.Buffers.ReadOnlySequence`1<System.Byte>)
DecodeResponseHeader(System.Buffers.ReadOnlySequence`1<System.Byte>,System.Int32)
EncodeRequestHeader(IceRpc.Internal.IceDuplexConnectionWriter,IceRpc.OutgoingRequest,System.Int32,System.Int32)
EncodeResponseHeader(System.Buffers.IBufferWriter`1<System.Byte>,IceRpc.OutgoingResponse,IceRpc.IncomingRequest,System.Int32,System.Int32)
ReadFullPayloadAsync()
AcquireWriteLockAsync()
CreateFrameReaderAsync()
DecrementDispatchInvocationCount()
DispatchRequestAsync()
IncrementDispatchInvocationCount()
ScheduleInactivityCheck()
ReadFramesAsync()
AbortTwowayInvocations()
ReadFailed()
ReadReplyAsync()
ReadRequestAsync()
<ReadRequestAsync()
RefuseNewInvocations(System.String)
SendControlFrameAsync()
WriteFailed(System.Exception)