< Summary

Information
Class: IceRpc.Internal.IceProtocolConnection
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Internal/IceProtocolConnection.cs
Tag: 2300_35243572715
Line coverage
89%
Covered lines: 863
Uncovered lines: 98
Coverable lines: 961
Total lines: 1614
Line coverage: 89.8%
Branch coverage
83%
Covered branches: 208
Total branches: 248
Branch coverage: 83.8%
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()84.61%262694.28%
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(...)79.16%252488.88%
EncodeRequestHeader(...)83.33%6692.1%
EncodeResponseHeader(...)100%2020100%
ReadFullPayloadAsync()50%4475%
AcquireWriteLockAsync()100%22100%
CreateFrameReaderAsync()100%1173.33%
DecrementDispatchInvocationCount()100%66100%
DispatchRequestAsync()92.85%141496.25%
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
 21823    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.
 22933    private readonly TaskCompletionSource _dispatchesAndInvocationsCompleted =
 22934        new(TaskCreationOptions.RunContinuationsAsynchronously);
 35
 36    private readonly SemaphoreSlim? _dispatchSemaphore;
 37
 38    // This cancellation token source is canceled when the connection is disposed.
 22939    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;
 22945    private bool _heartbeatEnabled = true;
 22946    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;
 22952    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?
 22960    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.
 22966    private readonly TaskCompletionSource _shutdownRequestedTcs = new();
 67
 68    // Only set for server connections.
 69    private readonly TransportConnectionInformation? _transportConnectionInformation;
 70
 71    private readonly CancellationTokenSource _twowayDispatchesCts;
 22972    private readonly Dictionary<int, TaskCompletionSource<PipeReader>> _twowayInvocations = new();
 73
 74    private Exception? _writeException; // protected by _writeSemaphore
 22975    private readonly SemaphoreSlim _writeSemaphore = new(1, 1);
 76
 77    public Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> ConnectAsync(
 78        CancellationToken cancellationToken)
 22879    {
 80        Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> result;
 81        lock (_mutex)
 22882        {
 22883            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 84
 22685            if (_connectTask is not null)
 086            {
 087                throw new InvalidOperationException("Cannot call connect more than once.");
 88            }
 89
 22690            result = PerformConnectAsync();
 22691            _connectTask = result;
 22692        }
 22693        return result;
 94
 95        async Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> PerformConnectAsync()
 22696        {
 97            // Make sure we execute the function without holding the connection mutex lock.
 22698            await Task.Yield();
 99
 100            // _disposedCts is not disposed at this point because DisposeAsync waits for the completion of _connectTask.
 226101            using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(
 226102                cancellationToken,
 226103                _disposedCts.Token);
 104
 105            TransportConnectionInformation transportConnectionInformation;
 106
 107            try
 226108            {
 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.
 226112                transportConnectionInformation = _transportConnectionInformation ??
 226113                    await _duplexConnection.ConnectAsync(connectCts.Token).ConfigureAwait(false);
 114
 218115                if (IsServer)
 106116                {
 117                    // Send ValidateConnection frame.
 106118                    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.
 104122                }
 123                else
 112124                {
 112125                    ReadOnlySequence<byte> buffer = await _duplexConnectionReader.ReadAtLeastAsync(
 112126                        IceDefinitions.PrologueSize,
 112127                        connectCts.Token).ConfigureAwait(false);
 128
 102129                    (IcePrologue validateConnectionFrame, long consumed) = DecodeValidateConnectionFrame(buffer);
 102130                    _duplexConnectionReader.AdvanceTo(buffer.GetPosition(consumed), buffer.End);
 131
 102132                    IceDefinitions.CheckPrologue(validateConnectionFrame);
 101133                    if (validateConnectionFrame.FrameSize != IceDefinitions.PrologueSize)
 0134                    {
 0135                        throw new InvalidDataException(
 0136                            $"Received ice frame with only '{validateConnectionFrame.FrameSize}' bytes.");
 137                    }
 101138                    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.
 101145                    if (_duplexConnection is IceDuplexConnectionDecorator decorator)
 101146                    {
 101147                        decorator.ScheduleHeartbeat();
 101148                    }
 101149                }
 205150            }
 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)
 205184            {
 205185                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.
 205193                _connectionContext = new ConnectionContext(this, transportConnectionInformation);
 194
 205195                _readFramesTask = ReadFramesAsync(_disposedCts.Token);
 205196            }
 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.
 205200            ScheduleInactivityCheck();
 201
 205202            return (transportConnectionInformation, _shutdownRequestedTcs.Task);
 203
 204            static void EncodeValidateConnectionFrame(IBufferWriter<byte> writer)
 106205            {
 106206                var encoder = new IceEncoder(writer);
 106207                IceDefinitions.ValidateConnectionFrame.Encode(ref encoder);
 106208            }
 209
 210            static (IcePrologue, long) DecodeValidateConnectionFrame(ReadOnlySequence<byte> buffer)
 102211            {
 102212                var decoder = new IceDecoder(buffer);
 102213                return (new IcePrologue(ref decoder), decoder.Consumed);
 102214            }
 205215        }
 226216    }
 217
 218    public ValueTask DisposeAsync()
 250219    {
 220        lock (_mutex)
 250221        {
 250222            if (_disposeTask is null)
 229223            {
 229224                RefuseNewInvocations("The connection was disposed.");
 225
 229226                _shutdownTask ??= Task.CompletedTask;
 229227                if (_dispatchInvocationCount == 0)
 221228                {
 221229                    _dispatchesAndInvocationsCompleted.TrySetResult();
 221230                }
 231
 229232                _heartbeatEnabled = false; // makes _heartbeatTask immutable
 233
 229234                _disposeTask = PerformDisposeAsync();
 229235            }
 250236        }
 250237        return new(_disposeTask);
 238
 239        async Task PerformDisposeAsync()
 229240        {
 241            // Make sure we execute the code below without holding the mutex lock.
 229242            await Task.Yield();
 243
 229244            _disposedCts.Cancel();
 245
 246            // We don't lock _mutex since once _disposeTask is not null, _connectTask etc are immutable.
 247
 229248            if (_connectTask is not null)
 226249            {
 250                // Wait for all writes to complete. This can't take forever since all writes are canceled by
 251                // _disposedCts.Token.
 226252                await _writeSemaphore.WaitAsync().ConfigureAwait(false);
 253
 254                try
 226255                {
 226256                    await Task.WhenAll(
 226257                        _connectTask,
 226258                        _readFramesTask ?? Task.CompletedTask,
 226259                        _heartbeatTask,
 226260                        _dispatchesAndInvocationsCompleted.Task,
 226261                        _shutdownTask).ConfigureAwait(false);
 130262                }
 96263                catch
 96264                {
 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.
 96267                }
 226268            }
 269
 229270            _duplexConnection.Dispose();
 271
 272            // It's safe to dispose the reader/writer since no more threads are sending/receiving data.
 229273            _duplexConnectionReader.Dispose();
 229274            _duplexConnectionWriter.Dispose();
 275
 229276            _disposedCts.Dispose();
 229277            _twowayDispatchesCts.Dispose();
 278
 229279            _dispatchSemaphore?.Dispose();
 229280            _writeSemaphore.Dispose();
 229281            await _inactivityTimeoutTimer.DisposeAsync().ConfigureAwait(false);
 229282        }
 250283    }
 284
 285    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken = default)
 1400286    {
 1400287        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)
 1399294        {
 1399295            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 296
 1398297            if (_refuseInvocations)
 1298            {
 1299                throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage);
 300            }
 1397301            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 0302            {
 0303                throw new InvalidOperationException("Cannot invoke on a connection that is not fully established.");
 304            }
 305
 1397306            IncrementDispatchInvocationCount();
 1397307        }
 308
 1397309        return PerformInvokeAsync();
 310
 311        async Task<IncomingResponse> PerformInvokeAsync()
 1397312        {
 313            // Since _dispatchInvocationCount > 0, _disposedCts is not disposed.
 1397314            using var invocationCts =
 1397315                CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token, cancellationToken);
 316
 1397317            PipeReader? frameReader = null;
 1397318            bool responseCreated = false;
 1397319            TaskCompletionSource<PipeReader>? responseCompletionSource = null;
 1397320            int requestId = 0;
 321
 322            try
 1397323            {
 324                // Read the full payload. This can take some time so this needs to be done before acquiring the write
 325                // semaphore.
 1397326                ReadOnlySequence<byte> payloadBuffer = await ReadFullPayloadAsync(request.Payload, invocationCts.Token)
 1397327                    .ConfigureAwait(false);
 328
 329                try
 1397330                {
 331                    // Wait for the writing of other frames to complete.
 1397332                    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)
 1397338                    {
 1397339                        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
 1397345                        if (!request.IsOneway)
 391346                        {
 347                            // wrap around back to 1 if we reach int.MaxValue. 0 means one-way.
 391348                            _lastRequestId = _lastRequestId == int.MaxValue ? 1 : _lastRequestId + 1;
 391349                            requestId = _lastRequestId;
 350
 351                            // RunContinuationsAsynchronously because we don't want the "read frames loop" to run the
 352                            // continuation.
 391353                            responseCompletionSource = new(TaskCreationOptions.RunContinuationsAsynchronously);
 391354                            _twowayInvocations[requestId] = responseCompletionSource;
 391355                        }
 1397356                    }
 357
 1397358                    int payloadSize = checked((int)payloadBuffer.Length);
 359
 360                    try
 1397361                    {
 1397362                        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.
 1397366                        await _duplexConnectionWriter.WriteAsync(payloadBuffer, _disposedCts.Token)
 1397367                            .ConfigureAwait(false);
 1396368                    }
 1369                    catch (Exception exception)
 1370                    {
 1371                        WriteFailed(exception);
 1372                        throw;
 373                    }
 1396374                }
 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
 1397386                {
 387                    // We've read the payload (see ReadFullPayloadAsync) and we are now done with it.
 1397388                    request.Payload.Complete();
 1397389                }
 390
 1396391                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.
 390398                Debug.Assert(responseCompletionSource is not null);
 390399                frameReader = await responseCompletionSource.Task.WaitAsync(invocationCts.Token).ConfigureAwait(false);
 400
 401                StatusCode statusCode;
 402                string? errorMessage;
 403                SequencePosition consumed;
 404                try
 370405                {
 370406                    if (!frameReader.TryRead(out ReadResult readResult))
 0407                    {
 0408                        throw new InvalidDataException(
 0409                            $"Received empty response frame for request with ID '{requestId}'.");
 410                    }
 411
 370412                    Debug.Assert(readResult.IsCompleted);
 413
 370414                    (statusCode, errorMessage, consumed) = DecodeResponseHeader(readResult.Buffer, requestId);
 369415                }
 1416                catch (InvalidDataException exception)
 1417                {
 1418                    throw new IceRpcException(
 1419                        IceRpcError.IceRpcError,
 1420                        "Received an ice response with an invalid header.",
 1421                        exception);
 422                }
 423
 369424                frameReader.AdvanceTo(consumed);
 425
 369426                var response = new IncomingResponse(
 369427                    request,
 369428                    _connectionContext!,
 369429                    statusCode,
 369430                    errorMessage)
 369431                {
 369432                    Payload = frameReader
 369433                };
 434
 369435                responseCreated = true; // the response now owns frameReader
 369436                return response;
 437            }
 9438            catch (OperationCanceledException)
 9439            {
 9440                cancellationToken.ThrowIfCancellationRequested();
 441
 3442                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 3443                throw new IceRpcException(
 3444                    IceRpcError.OperationAborted,
 3445                    "The invocation was aborted because the connection was disposed.");
 446            }
 447            finally
 1397448            {
 449                // If responseCompletionSource is not completed, we want to complete it to prevent another method from
 450                // setting an unobservable exception in it. And if it's already completed with an exception, we observe
 451                // this exception.
 1397452                if (responseCompletionSource is not null &&
 1397453                    !responseCompletionSource.TrySetResult(InvalidPipeReader.Instance))
 381454                {
 455                    try
 381456                    {
 457                        // Retrieve (or re-retrieve) the response PipeReader. The cleanup at the end of this finally
 458                        // completes it unless a response was created, in which case the response owns it.
 381459                        frameReader = await responseCompletionSource.Task.ConfigureAwait(false);
 370460                    }
 11461                    catch
 11462                    {
 463                        // observe exception, if any
 11464                    }
 381465                }
 466
 467                lock (_mutex)
 1397468                {
 469                    // Unregister the two-way invocation if registered.
 1397470                    if (requestId > 0 && !_refuseInvocations)
 369471                    {
 369472                        _twowayInvocations.Remove(requestId);
 369473                    }
 474
 1397475                    DecrementDispatchInvocationCount();
 1397476                }
 477
 1397478                if (!responseCreated)
 1028479                {
 1028480                    frameReader?.Complete();
 1028481                }
 482                // else the response owns the PipeReader
 1397483            }
 0484        }
 2772485    }
 486
 487    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 71488    {
 489        lock (_mutex)
 71490        {
 71491            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 492
 69493            if (_shutdownTask is not null)
 0494            {
 0495                throw new InvalidOperationException("Cannot call ShutdownAsync more than once.");
 496            }
 69497            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 3498            {
 3499                throw new InvalidOperationException("Cannot shut down a protocol connection before it's connected.");
 500            }
 501
 66502            RefuseNewInvocations("The connection was shut down.");
 503
 66504            if (_dispatchInvocationCount == 0)
 54505            {
 54506                _dispatchesAndInvocationsCompleted.TrySetResult();
 54507            }
 66508            _shutdownTask = PerformShutdownAsync(_sendCloseConnectionFrame);
 66509        }
 510
 66511        return _shutdownTask;
 512
 513        async Task PerformShutdownAsync(bool sendCloseConnectionFrame)
 66514        {
 66515            await Task.Yield(); // exit mutex lock
 516
 517            try
 66518            {
 66519                Debug.Assert(_readFramesTask is not null);
 520
 521                // Since DisposeAsync waits for the _shutdownTask completion, _disposedCts is not disposed at this
 522                // point.
 66523                using var shutdownCts = CancellationTokenSource.CreateLinkedTokenSource(
 66524                    cancellationToken,
 66525                    _disposedCts.Token);
 526
 527                // Wait for dispatches and invocations to complete.
 66528                await _dispatchesAndInvocationsCompleted.Task.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 529
 530                // Stops sending heartbeats. We can't do earlier: while we're waiting for dispatches and invocations to
 531                // complete, we need to keep sending heartbeats otherwise the peer could see the connection as idle and
 532                // abort it.
 533                lock (_mutex)
 62534                {
 62535                    _heartbeatEnabled = false; // makes _heartbeatTask immutable
 62536                }
 537
 538                // Wait for the last send heartbeat to complete before sending the CloseConnection frame or disposing
 539                // the duplex connection. _heartbeatTask is immutable once _shutdownTask set. _heartbeatTask can be
 540                // canceled by DisposeAsync.
 62541                await _heartbeatTask.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 542
 62543                if (sendCloseConnectionFrame)
 29544                {
 545                    // Send CloseConnection frame.
 29546                    await SendControlFrameAsync(EncodeCloseConnectionFrame, shutdownCts.Token).ConfigureAwait(false);
 547
 548                    // Wait for the peer to abort the connection as an acknowledgment for this CloseConnection frame.
 549                    // The peer can also send us a CloseConnection frame if it started shutting down at the same time.
 550                    // We can't just return and dispose the duplex connection since the peer can still be reading frames
 551                    // (including the CloseConnection frame) and we don't want to abort this reading.
 25552                    await _readFramesTask.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 24553                }
 554                else
 33555                {
 556                    // _readFramesTask should be already completed or nearly completed.
 33557                    await _readFramesTask.WaitAsync(shutdownCts.Token).ConfigureAwait(false);
 558
 559                    // _readFramesTask succeeded means the peer is waiting for us to abort the duplex connection;
 560                    // we oblige.
 23561                    _duplexConnection.Dispose();
 23562                }
 47563            }
 7564            catch (OperationCanceledException)
 7565            {
 7566                cancellationToken.ThrowIfCancellationRequested();
 567
 3568                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 3569                throw new IceRpcException(
 3570                    IceRpcError.OperationAborted,
 3571                    "The connection shutdown was aborted because the connection was disposed.");
 572            }
 12573            catch (IceRpcException)
 12574            {
 12575                throw;
 576            }
 0577            catch (Exception exception)
 0578            {
 0579                Debug.Fail($"ShutdownAsync failed with an unexpected exception: {exception}");
 0580                throw;
 581            }
 582
 583            static void EncodeCloseConnectionFrame(IBufferWriter<byte> writer)
 27584            {
 27585                var encoder = new IceEncoder(writer);
 27586                IceDefinitions.CloseConnectionFrame.Encode(ref encoder);
 27587            }
 47588        }
 66589    }
 590
 229591    internal IceProtocolConnection(
 229592        IDuplexConnection duplexConnection,
 229593        TransportConnectionInformation? transportConnectionInformation,
 229594        ConnectionOptions options)
 229595    {
 229596        _twowayDispatchesCts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 597
 598        // With ice, we always listen for incoming frames (responses) so we need a dispatcher for incoming requests even
 599        // if we don't expect any. This dispatcher throws an ice ObjectNotExistException back to the client, which makes
 600        // more sense than throwing an UnknownException.
 229601        _dispatcher = options.Dispatcher ?? NotFoundDispatcher.Instance;
 602
 229603        _maxFrameSize = options.MaxIceFrameSize;
 229604        _transportConnectionInformation = transportConnectionInformation;
 605
 229606        if (options.MaxDispatches > 0)
 229607        {
 229608            _dispatchSemaphore = new SemaphoreSlim(
 229609                initialCount: options.MaxDispatches,
 229610                maxCount: options.MaxDispatches);
 229611        }
 612
 229613        _inactivityTimeout = options.InactivityTimeout;
 614
 615        // The readerScheduler doesn't matter (we don't call pipe.Reader.ReadAsync on the resulting pipe), and the
 616        // writerScheduler doesn't matter (pipe.Writer.FlushAsync never blocks).
 229617        _pipeOptions = new PipeOptions(
 229618            pool: options.Pool,
 229619            minimumSegmentSize: options.MinSegmentSize,
 229620            pauseWriterThreshold: 0,
 229621            useSynchronizationContext: false);
 622
 229623        if (options.IceIdleTimeout != Timeout.InfiniteTimeSpan)
 229624        {
 229625            duplexConnection = new IceDuplexConnectionDecorator(
 229626                duplexConnection,
 229627                readIdleTimeout: options.EnableIceIdleCheck ? options.IceIdleTimeout : Timeout.InfiniteTimeSpan,
 229628                writeIdleTimeout: options.IceIdleTimeout,
 229629                SendHeartbeat);
 229630        }
 631
 229632        _duplexConnection = duplexConnection;
 229633        _duplexConnectionReader = new DuplexConnectionReader(_duplexConnection, options.Pool, options.MinSegmentSize);
 229634        _duplexConnectionWriter =
 229635            new IceDuplexConnectionWriter(_duplexConnection, options.Pool, options.MinSegmentSize);
 636
 229637        _inactivityTimeoutTimer = new Timer(_ =>
 5638        {
 5639            bool requestShutdown = false;
 229640
 229641            lock (_mutex)
 5642            {
 5643                if (_dispatchInvocationCount == 0 && _shutdownTask is null)
 5644                {
 5645                    requestShutdown = true;
 5646                    RefuseNewInvocations(
 5647                        $"The connection was shut down because it was inactive for over {_inactivityTimeout.TotalSeconds
 5648                }
 5649            }
 229650
 5651            if (requestShutdown)
 5652            {
 229653                // TrySetResult must be called outside the mutex lock.
 5654                _shutdownRequestedTcs.TrySetResult();
 5655            }
 234656        });
 657
 658        void SendHeartbeat()
 14659        {
 660            lock (_mutex)
 14661            {
 14662                if (_heartbeatTask.IsCompletedSuccessfully && _heartbeatEnabled)
 14663                {
 14664                    _heartbeatTask = SendValidateConnectionFrameAsync(_disposedCts.Token);
 14665                }
 14666            }
 667
 668            async Task SendValidateConnectionFrameAsync(CancellationToken cancellationToken)
 14669            {
 670                // Make sure we execute the function without holding the connection mutex lock.
 14671                await Task.Yield();
 672
 673                try
 14674                {
 14675                    await SendControlFrameAsync(EncodeValidateConnectionFrame, cancellationToken).ConfigureAwait(false);
 14676                }
 0677                catch (OperationCanceledException)
 0678                {
 679                    // Canceled by DisposeAsync
 0680                    throw;
 681                }
 0682                catch (IceRpcException)
 0683                {
 684                    // Expected, typically the peer aborted the connection.
 0685                    throw;
 686                }
 0687                catch (Exception exception)
 0688                {
 0689                    Debug.Fail($"The heartbeat task completed due to an unhandled exception: {exception}");
 0690                    throw;
 691                }
 692
 693                static void EncodeValidateConnectionFrame(IBufferWriter<byte> writer)
 14694                {
 14695                    var encoder = new IceEncoder(writer);
 14696                    IceDefinitions.ValidateConnectionFrame.Encode(ref encoder);
 14697                }
 14698            }
 14699        }
 229700    }
 701
 702    private static (int RequestId, IceRequestHeader Header, PipeReader? ContextReader, int Consumed) DecodeRequestIdAndH
 703        ReadOnlySequence<byte> buffer)
 1394704    {
 1394705        var decoder = new IceDecoder(buffer);
 706
 1394707        int requestId = decoder.DecodeInt();
 708
 1394709        var requestHeader = new IceRequestHeader(ref decoder);
 1394710        requestHeader.Facet.CheckFacetCount();
 711
 1394712        Pipe? contextPipe = null;
 1394713        long pos = decoder.Consumed;
 1394714        int count = decoder.DecodeSize();
 1394715        if (count > 0)
 7716        {
 28717            for (int i = 0; i < count; ++i)
 7718            {
 7719                decoder.Skip(decoder.DecodeSize()); // Skip the key
 7720                decoder.Skip(decoder.DecodeSize()); // Skip the value
 7721            }
 7722            contextPipe = new Pipe();
 7723            contextPipe.Writer.Write(buffer.Slice(pos, decoder.Consumed - pos));
 7724            contextPipe.Writer.Complete();
 7725        }
 726
 1394727        var encapsulationHeader = new EncapsulationHeader(ref decoder);
 728
 1394729        if (encapsulationHeader.PayloadEncodingMajor != 1 ||
 1394730            encapsulationHeader.PayloadEncodingMinor != 1)
 0731        {
 0732            throw new InvalidDataException(
 0733                $"Unsupported payload encoding '{encapsulationHeader.PayloadEncodingMajor}.{encapsulationHeader.PayloadE
 734        }
 735
 1394736        int payloadSize = encapsulationHeader.EncapsulationSize - 6;
 1394737        if (payloadSize != (buffer.Length - decoder.Consumed))
 0738        {
 0739            throw new InvalidDataException(
 0740                $"Request payload size mismatch: expected {payloadSize} bytes, read {buffer.Length - decoder.Consumed} b
 741        }
 742
 1394743        return (requestId, requestHeader, contextPipe?.Reader, (int)decoder.Consumed);
 1394744    }
 745
 746    private static (StatusCode StatusCode, string? ErrorMessage, SequencePosition Consumed) DecodeResponseHeader(
 747        ReadOnlySequence<byte> buffer,
 748        int requestId)
 370749    {
 370750        if (buffer.IsEmpty)
 0751        {
 0752            throw new InvalidDataException($"Received empty response header for request with ID '{requestId}'.");
 753        }
 754
 370755        var replyStatus = (ReplyStatus)buffer.FirstSpan[0];
 756
 370757        if (replyStatus <= ReplyStatus.UserException)
 334758        {
 759            const int headerSize = 7; // reply status byte + encapsulation header
 760
 761            // read and check encapsulation header (6 bytes long)
 762
 334763            if (buffer.Length < headerSize)
 0764            {
 0765                throw new InvalidDataException($"Received invalid frame header for request with ID '{requestId}'.");
 766            }
 767
 334768            EncapsulationHeader encapsulationHeader =
 668769                buffer.Slice(1, 6).DecodeIceBuffer((ref IceDecoder decoder) => new EncapsulationHeader(ref decoder));
 770
 771            // Sanity check
 334772            int payloadSize = encapsulationHeader.EncapsulationSize - 6;
 334773            if (payloadSize != buffer.Length - headerSize)
 1774            {
 1775                throw new InvalidDataException(
 1776                    $"Response payload size/frame size mismatch: payload size is {payloadSize} bytes but frame has {buff
 777            }
 778
 333779            SequencePosition consumed = buffer.GetPosition(headerSize);
 780
 333781            return replyStatus == ReplyStatus.Ok ? (StatusCode.Ok, null, consumed) :
 333782                // Set the error message to the empty string, because null is not allowed for status code > Ok.
 333783                (StatusCode.ApplicationError, "", consumed);
 784        }
 785        else
 36786        {
 787            // An ice system exception.
 788
 36789            StatusCode statusCode = replyStatus switch
 36790            {
 14791                ReplyStatus.ObjectNotExist => StatusCode.NotFound,
 0792                ReplyStatus.FacetNotExist => StatusCode.NotFound,
 2793                ReplyStatus.OperationNotExist => StatusCode.NotImplemented,
 3794                ReplyStatus.InvalidData => StatusCode.InvalidData,
 1795                ReplyStatus.Unauthorized => StatusCode.Unauthorized,
 2796                ReplyStatus.NotSupported => StatusCode.NotSupported,
 14797                _ => StatusCode.InternalError
 36798            };
 799
 36800            var decoder = new IceDecoder(buffer.Slice(1));
 801
 802            string message;
 36803            switch (replyStatus)
 804            {
 805                case ReplyStatus.FacetNotExist:
 806                case ReplyStatus.ObjectNotExist:
 807                case ReplyStatus.OperationNotExist:
 808
 16809                    var requestFailed = new RequestFailedExceptionData(ref decoder);
 810
 16811                    string target = requestFailed.Facet.Count > 0 ?
 16812                        $"{requestFailed.Identity.ToPath()}#{requestFailed.Facet.ToFragment()}" : requestFailed.Identity
 813
 16814                    message =
 16815                        $"The dispatch failed with status code {statusCode} while dispatching '{requestFailed.Operation}
 16816                    break;
 817                default:
 20818                    message = decoder.DecodeString();
 20819                    break;
 820            }
 36821            decoder.CheckEndOfBuffer();
 36822            return (statusCode, message, buffer.End);
 823        }
 369824    }
 825
 826    private static void EncodeRequestHeader(
 827        IceDuplexConnectionWriter output,
 828        OutgoingRequest request,
 829        int requestId,
 830        int payloadSize)
 1397831    {
 1397832        var encoder = new IceEncoder(output);
 833
 834        // Write the request header.
 1397835        encoder.WriteByteSpan(IceDefinitions.FramePrologue);
 1397836        encoder.EncodeIceFrameType(IceFrameType.Request);
 1397837        encoder.EncodeByte(0); // compression status
 838
 1397839        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(4);
 840
 1397841        encoder.EncodeInt(requestId);
 842
 1397843        byte encodingMajor = 1;
 1397844        byte encodingMinor = 1;
 845
 846        // Request header.
 1397847        var requestHeader = new IceRequestHeader(
 1397848            IceIdentity.Parse(request.ServiceAddress.Path),
 1397849            request.ServiceAddress.Fragment.ToFacet(),
 1397850            request.Operation,
 1397851            request.Fields.ContainsKey(RequestFieldKey.Idempotent) ? OperationMode.Idempotent : OperationMode.Normal);
 1397852        requestHeader.Encode(ref encoder);
 1397853        int directWriteSize = 0;
 1397854        if (request.Fields.TryGetValue(RequestFieldKey.Context, out OutgoingFieldValue requestField))
 7855        {
 7856            if (requestField.WriteAction is Action<IBufferWriter<byte>> writeAction)
 7857            {
 858                // This writes directly to the underlying output; we measure how many bytes are written.
 7859                long start = output.UnflushedBytes;
 7860                writeAction(output);
 7861                directWriteSize = (int)(output.UnflushedBytes - start);
 7862            }
 863            else
 0864            {
 0865                encoder.WriteByteSequence(requestField.ByteSequence);
 0866            }
 7867        }
 868        else
 1390869        {
 1390870            encoder.EncodeSize(0);
 1390871        }
 872
 873        // We ignore all other fields. They can't be sent over ice.
 874
 1397875        new EncapsulationHeader(
 1397876            encapsulationSize: payloadSize + 6,
 1397877            encodingMajor,
 1397878            encodingMinor).Encode(ref encoder);
 879
 1397880        int frameSize = checked(encoder.EncodedByteCount + directWriteSize + payloadSize);
 1397881        IceEncoder.EncodeInt(frameSize, sizePlaceholder);
 1397882    }
 883
 884    private static void EncodeResponseHeader(
 885        IBufferWriter<byte> writer,
 886        OutgoingResponse response,
 887        IncomingRequest request,
 888        int requestId,
 889        int payloadSize)
 1381890    {
 1381891        var encoder = new IceEncoder(writer);
 892
 893        // Write the response header.
 894
 1381895        encoder.WriteByteSpan(IceDefinitions.FramePrologue);
 1381896        encoder.EncodeIceFrameType(IceFrameType.Reply);
 1381897        encoder.EncodeByte(0); // compression status
 1381898        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(4);
 899
 1381900        encoder.EncodeInt(requestId);
 901
 1381902        if (response.StatusCode > StatusCode.ApplicationError ||
 1381903            (response.StatusCode == StatusCode.ApplicationError && payloadSize == 0))
 38904        {
 905            // system exception
 38906            switch (response.StatusCode)
 907            {
 908                case StatusCode.NotFound:
 909                case StatusCode.NotImplemented:
 18910                    encoder.EncodeReplyStatus(response.StatusCode == StatusCode.NotFound ?
 18911                        ReplyStatus.ObjectNotExist : ReplyStatus.OperationNotExist);
 912
 18913                    new RequestFailedExceptionData(
 18914                        IceIdentity.Parse(request.Path),
 18915                        request.Fragment.ToFacet(),
 18916                        request.Operation).Encode(ref encoder);
 18917                    break;
 918                case StatusCode.InternalError:
 7919                    encoder.EncodeReplyStatus(ReplyStatus.UnknownException);
 7920                    encoder.EncodeString(response.ErrorMessage!);
 7921                    break;
 922                case StatusCode.InvalidData:
 3923                    encoder.EncodeReplyStatus(ReplyStatus.InvalidData);
 3924                    encoder.EncodeString(response.ErrorMessage!);
 3925                    break;
 926                case StatusCode.Unauthorized:
 1927                    encoder.EncodeReplyStatus(ReplyStatus.Unauthorized);
 1928                    encoder.EncodeString(response.ErrorMessage!);
 1929                    break;
 930                case StatusCode.NotSupported:
 2931                    encoder.EncodeReplyStatus(ReplyStatus.NotSupported);
 2932                    encoder.EncodeString(response.ErrorMessage!);
 2933                    break;
 934                default:
 7935                    encoder.EncodeReplyStatus(ReplyStatus.UnknownException);
 7936                    encoder.EncodeString(
 7937                        $"{response.ErrorMessage} {{ Original StatusCode = {response.StatusCode} }}");
 7938                    break;
 939            }
 38940        }
 941        else
 1343942        {
 1343943            encoder.EncodeReplyStatus((ReplyStatus)response.StatusCode);
 944
 945            // When IceRPC receives a response, it ignores the response encoding. So this "1.1" is only relevant to
 946            // a ZeroC Ice client that decodes the response. The only Slice encoding such a client can possibly use
 947            // to decode the response payload is 1.1 or 1.0, and we don't care about interop with 1.0.
 1343948            var encapsulationHeader = new EncapsulationHeader(
 1343949                encapsulationSize: payloadSize + 6,
 1343950                payloadEncodingMajor: 1,
 1343951                payloadEncodingMinor: 1);
 1343952            encapsulationHeader.Encode(ref encoder);
 1343953        }
 954
 1381955        int frameSize = encoder.EncodedByteCount + payloadSize;
 1381956        IceEncoder.EncodeInt(frameSize, sizePlaceholder);
 1381957    }
 958
 959    /// <summary>Reads the full Ice payload from the given pipe reader.</summary>
 960    private static async ValueTask<ReadOnlySequence<byte>> ReadFullPayloadAsync(
 961        PipeReader payload,
 962        CancellationToken cancellationToken)
 2745963    {
 964        // We use ReadAtLeastAsync instead of ReadAsync to bypass the PauseWriterThreshold when the payload is
 965        // backed by a Pipe.
 2745966        ReadResult readResult = await payload.ReadAtLeastAsync(int.MaxValue, cancellationToken).ConfigureAwait(false);
 967
 2742968        if (readResult.IsCanceled)
 0969        {
 0970            throw new InvalidOperationException("Unexpected call to CancelPendingRead on ice payload.");
 971        }
 972
 2742973        return readResult.IsCompleted ? readResult.Buffer :
 2742974            throw new ArgumentException("The payload size is greater than int.MaxValue.", nameof(payload));
 2742975    }
 976
 977    /// <summary>Acquires exclusive access to _duplexConnectionWriter.</summary>
 978    /// <returns>A <see cref="SemaphoreLock" /> that releases the acquired semaphore in its Dispose method.</returns>
 979    private async ValueTask<SemaphoreLock> AcquireWriteLockAsync(CancellationToken cancellationToken)
 2927980    {
 2927981        SemaphoreLock semaphoreLock = await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false);
 982
 983        // _writeException is protected by _writeSemaphore
 2926984        if (_writeException is not null)
 1985        {
 1986            semaphoreLock.Dispose();
 987
 1988            throw new IceRpcException(
 1989                IceRpcError.ConnectionAborted,
 1990                "The connection was aborted because a previous write operation failed.",
 1991                _writeException);
 992        }
 993
 2925994        return semaphoreLock;
 2925995    }
 996
 997    /// <summary>Creates a pipe reader to simplify the reading of a request or response frame. The frame is read fully
 998    /// and buffered into an internal pipe.</summary>
 999    private async ValueTask<PipeReader> CreateFrameReaderAsync(int size, CancellationToken cancellationToken)
 27711000    {
 27711001        var pipe = new Pipe(_pipeOptions);
 1002
 1003        try
 27711004        {
 27711005            await _duplexConnectionReader.FillBufferWriterAsync(pipe.Writer, size, cancellationToken)
 27711006                .ConfigureAwait(false);
 27711007        }
 01008        catch
 01009        {
 01010            pipe.Reader.Complete();
 01011            throw;
 1012        }
 1013        finally
 27711014        {
 27711015            pipe.Writer.Complete();
 27711016        }
 1017
 27711018        return pipe.Reader;
 27711019    }
 1020
 1021    private void DecrementDispatchInvocationCount()
 27891022    {
 1023        lock (_mutex)
 27891024        {
 27891025            if (--_dispatchInvocationCount == 0)
 12431026            {
 12431027                if (_shutdownTask is not null)
 181028                {
 181029                    _dispatchesAndInvocationsCompleted.TrySetResult();
 181030                }
 1031                // We enable the inactivity check in order to complete ShutdownRequested when inactive for too long.
 1032                // _refuseInvocations is true when the connection is either about to be "shutdown requested", or shut
 1033                // down / disposed. We don't need to complete ShutdownRequested in any of these situations.
 12251034                else if (!_refuseInvocations)
 12081035                {
 12081036                    ScheduleInactivityCheck();
 12081037                }
 12431038            }
 27891039        }
 27891040    }
 1041
 1042    /// <summary>Dispatches an incoming request. This method executes in a task spawn from the read frames loop.
 1043    /// </summary>
 1044    private async Task DispatchRequestAsync(IncomingRequest request, int requestId, PipeReader? contextReader)
 13921045    {
 13921046        CancellationToken cancellationToken = request.IsOneway ? _disposedCts.Token : _twowayDispatchesCts.Token;
 1047
 1048        OutgoingResponse? response;
 1049        try
 13921050        {
 1051            // The dispatcher can complete the incoming request payload to release its memory as soon as possible.
 1052            try
 13921053            {
 1054                // _dispatcher.DispatchAsync may very well ignore the cancellation token and we don't want to keep
 1055                // dispatching when the cancellation token is canceled.
 13921056                cancellationToken.ThrowIfCancellationRequested();
 1057
 13921058                response = await _dispatcher.DispatchAsync(request, cancellationToken).ConfigureAwait(false);
 13731059            }
 1060            finally
 13921061            {
 13921062                _dispatchSemaphore?.Release();
 13921063            }
 1064
 13731065            if (response != request.Response)
 11066            {
 11067                throw new InvalidOperationException(
 11068                    "The dispatcher did not return the last response created for this request.");
 1069            }
 13721070        }
 201071        catch when (request.IsOneway)
 01072        {
 1073            // ignored since we're not returning anything
 01074            response = null;
 01075        }
 101076        catch (OperationCanceledException exception) when (exception.CancellationToken == cancellationToken)
 91077        {
 1078            // expected when the connection is disposed or the request is canceled by the peer's shutdown
 91079            response = null;
 91080        }
 111081        catch (Exception exception)
 111082        {
 111083            var dispatchException = DispatchException.FromException(exception);
 111084            Debug.Assert(!dispatchException.ConvertToInternalError);
 111085            response = new OutgoingResponse(
 111086                request,
 111087                dispatchException.StatusCode,
 111088                dispatchException.ErrorMessage);
 111089        }
 1090        finally
 13921091        {
 13921092            request.Payload.Complete();
 13921093            contextReader?.Complete();
 1094
 1095            // The field values are now invalid - they point to potentially recycled and reused memory. We
 1096            // replace Fields by an empty dictionary to prevent accidental access to this reused memory.
 13921097            request.Fields = ImmutableDictionary<RequestFieldKey, ReadOnlySequence<byte>>.Empty;
 13921098        }
 1099
 1100        try
 13921101        {
 13921102            if (response is not null)
 13831103            {
 1104                // Read the full response payload. This can take some time so this needs to be done before acquiring
 1105                // the write semaphore.
 13831106                ReadOnlySequence<byte> payload = ReadOnlySequence<byte>.Empty;
 1107
 13831108                if (response.StatusCode <= StatusCode.ApplicationError)
 13481109                {
 1110                    try
 13481111                    {
 13481112                        payload = await ReadFullPayloadAsync(response.Payload, cancellationToken)
 13481113                            .ConfigureAwait(false);
 13451114                    }
 21115                    catch (OperationCanceledException exception) when (exception.CancellationToken == cancellationToken)
 21116                    {
 21117                        throw;
 1118                    }
 11119                    catch (Exception exception)
 11120                    {
 1121                        // We "encode" the exception in the error message.
 1122
 11123                        response = new OutgoingResponse(
 11124                            request,
 11125                            StatusCode.InternalError,
 11126                            "The dispatch failed to read the response payload.",
 11127                            exception);
 11128                    }
 13461129                }
 1130                // else payload remains empty because the payload of a dispatch exception (if any) cannot be sent
 1131                // over ice.
 1132
 13811133                int payloadSize = checked((int)payload.Length);
 1134
 1135                // Wait for writing of other frames to complete.
 13811136                using SemaphoreLock _ = await AcquireWriteLockAsync(cancellationToken).ConfigureAwait(false);
 1137                try
 13811138                {
 13811139                    EncodeResponseHeader(_duplexConnectionWriter, response, request, requestId, payloadSize);
 1140
 1141                    // We write to the duplex connection with _disposedCts.Token instead of cancellationToken.
 1142                    // Canceling this write operation is fatal to the connection.
 13811143                    await _duplexConnectionWriter.WriteAsync(payload, _disposedCts.Token).ConfigureAwait(false);
 13801144                }
 11145                catch (Exception exception)
 11146                {
 11147                    WriteFailed(exception);
 11148                    throw;
 1149                }
 13801150            }
 13891151        }
 31152        catch (OperationCanceledException exception) when (
 31153            exception.CancellationToken == _disposedCts.Token ||
 31154            exception.CancellationToken == cancellationToken)
 31155        {
 1156            // expected when the connection is disposed or the request is canceled by the peer's shutdown
 31157        }
 1158        finally
 13921159        {
 13921160            DecrementDispatchInvocationCount();
 13921161        }
 13921162    }
 1163
 1164    /// <summary>Increments the dispatch-invocation count.</summary>
 1165    /// <remarks>This method must be called with _mutex locked.</remarks>
 1166    private void IncrementDispatchInvocationCount()
 27891167    {
 27891168        if (_dispatchInvocationCount++ == 0)
 12431169        {
 1170            // Cancel inactivity check.
 12431171            _inactivityTimeoutTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
 12431172        }
 27891173    }
 1174
 1175    private void ScheduleInactivityCheck() =>
 14131176        _inactivityTimeoutTimer.Change(_inactivityTimeout, Timeout.InfiniteTimeSpan);
 1177
 1178    /// <summary>Reads incoming frames and returns successfully when a CloseConnection frame is received or when the
 1179    /// connection is aborted during ShutdownAsync or canceled by DisposeAsync.</summary>
 1180    private async Task ReadFramesAsync(CancellationToken cancellationToken)
 2051181    {
 2051182        await Task.Yield(); // exit mutex lock
 1183
 1184        // Wait for _connectTask (which spawned the task running this method) to complete. This way, we won't dispatch
 1185        // any request until _connectTask has completed successfully, and indirectly we won't make any invocation until
 1186        // _connectTask has completed successfully. The creation of the _readFramesTask is the last action taken by
 1187        // _connectTask and as a result this await can't fail.
 2051188        await _connectTask!.ConfigureAwait(false);
 1189
 1190        try
 2051191        {
 29901192            while (!cancellationToken.IsCancellationRequested)
 29891193            {
 29891194                ReadOnlySequence<byte> buffer = await _duplexConnectionReader.ReadAtLeastAsync(
 29891195                    IceDefinitions.PrologueSize,
 29891196                    cancellationToken).ConfigureAwait(false);
 1197
 1198                // First decode and check the prologue.
 1199
 28131200                ReadOnlySequence<byte> prologueBuffer = buffer.Slice(0, IceDefinitions.PrologueSize);
 1201
 28131202                IcePrologue prologue =
 56261203                    prologueBuffer.DecodeIceBuffer((ref IceDecoder decoder) => new IcePrologue(ref decoder));
 1204
 28131205                _duplexConnectionReader.AdvanceTo(prologueBuffer.End);
 1206
 28131207                IceDefinitions.CheckPrologue(prologue);
 28121208                if (prologue.FrameSize > _maxFrameSize)
 11209                {
 11210                    throw new InvalidDataException(
 11211                        $"Received frame with size ({prologue.FrameSize}) greater than {nameof(ConnectionOptions.MaxIceF
 1212                }
 28111213                if (prologue.FrameSize < IceDefinitions.PrologueSize)
 11214                {
 11215                    throw new InvalidDataException(
 11216                        $"Received frame with size ({prologue.FrameSize}) smaller than the prologue size.");
 1217                }
 1218
 28101219                if (prologue.CompressionStatus == 2)
 01220                {
 1221                    // The exception handler calls ReadFailed.
 01222                    throw new IceRpcException(
 01223                        IceRpcError.ConnectionAborted,
 01224                        "The connection was aborted because it received a compressed ice frame, and IceRPC does not supp
 1225                }
 1226
 1227                // Then process the frame based on its type.
 28101228                switch (prologue.FrameType)
 1229                {
 1230                    case IceFrameType.CloseConnection:
 251231                    {
 251232                        if (prologue.FrameSize != IceDefinitions.PrologueSize)
 01233                        {
 01234                            throw new InvalidDataException(
 01235                                $"Received {nameof(IceFrameType.CloseConnection)} frame with unexpected data.");
 1236                        }
 1237
 1238                        lock (_mutex)
 251239                        {
 251240                            RefuseNewInvocations(
 251241                                "The connection was shut down because it received a CloseConnection frame from the peer.
 1242
 1243                            // By exiting the "read frames loop" below, we are refusing new dispatches as well.
 1244
 1245                            // Only one side sends the CloseConnection frame.
 251246                            _sendCloseConnectionFrame = false;
 251247                        }
 1248
 1249                        // Even though we're in the "read frames loop", it's ok to cancel CTS and a "synchronous" TCS
 1250                        // below. We won't be reading anything else so it's ok to run continuations synchronously.
 1251
 1252                        // Abort two-way invocations that are waiting for a response (it will never come).
 1253                        // We use InvocationCanceled (not ConnectionAborted) because the ice protocol guarantees the
 1254                        // peer has sent responses for all two-way requests it accepted before sending CloseConnection.
 1255                        // These pending two-way requests were never processed by the peer, so it's safe for a retry
 1256                        // interceptor to retry them unconditionally.
 251257                        AbortTwowayInvocations(
 251258                            IceRpcError.InvocationCanceled,
 251259                            "The invocation was canceled by the shutdown of the peer.");
 1260
 1261                        // Cancel two-way dispatches since the peer is not interested in the responses. This does not
 1262                        // cancel ongoing writes to _duplexConnection: we don't send incomplete/invalid data.
 251263                        _twowayDispatchesCts.Cancel();
 1264
 1265                        // We keep sending heartbeats. If the shutdown request / shutdown is not fulfilled quickly, they
 1266                        // tell the peer we're still alive and maybe stuck waiting for invocations and dispatches to
 1267                        // complete.
 1268
 1269                        // We request a shutdown that will dispose _duplexConnection once all invocations and dispatches
 1270                        // have completed.
 251271                        _shutdownRequestedTcs.TrySetResult();
 251272                        return;
 1273                    }
 1274
 1275                    case IceFrameType.Request:
 13941276                        await ReadRequestAsync(prologue.FrameSize, cancellationToken).ConfigureAwait(false);
 13941277                        break;
 1278
 1279                    case IceFrameType.RequestBatch:
 1280                        // The exception handler calls ReadFailed.
 01281                        throw new IceRpcException(
 01282                            IceRpcError.ConnectionAborted,
 01283                            "The connection was aborted because it received a batch request, and IceRPC does not support
 1284
 1285                    case IceFrameType.Reply:
 13771286                        await ReadReplyAsync(prologue.FrameSize, cancellationToken).ConfigureAwait(false);
 13771287                        break;
 1288
 1289                    case IceFrameType.ValidateConnection:
 141290                    {
 141291                        if (prologue.FrameSize != IceDefinitions.PrologueSize)
 01292                        {
 01293                            throw new InvalidDataException(
 01294                                $"Received {nameof(IceFrameType.ValidateConnection)} frame with unexpected data.");
 1295                        }
 141296                        break;
 1297                    }
 1298
 1299                    default:
 01300                    {
 01301                        throw new InvalidDataException(
 01302                            $"Received Ice frame with unknown frame type '{prologue.FrameType}'.");
 1303                    }
 1304                }
 27851305            } // while
 11306        }
 731307        catch (OperationCanceledException)
 731308        {
 1309            // canceled by DisposeAsync, no need to throw anything
 731310        }
 1031311        catch (IceRpcException exception) when (
 1031312            exception.IceRpcError == IceRpcError.ConnectionAborted &&
 1031313            _dispatchesAndInvocationsCompleted.Task.IsCompleted)
 401314        {
 1315            // The peer acknowledged receipt of the CloseConnection frame by aborting the duplex connection. Return.
 1316            // See ShutdownAsync.
 401317        }
 631318        catch (IceRpcException exception)
 631319        {
 631320            ReadFailed(exception);
 631321            throw;
 1322        }
 31323        catch (InvalidDataException exception)
 31324        {
 31325            ReadFailed(exception);
 31326            throw new IceRpcException(
 31327                IceRpcError.ConnectionAborted,
 31328                "The connection was aborted by an ice protocol error.",
 31329                exception);
 1330        }
 01331        catch (Exception exception)
 01332        {
 01333            Debug.Fail($"The read frames task completed due to an unhandled exception: {exception}");
 01334            ReadFailed(exception);
 01335            throw;
 1336        }
 1337
 1338        // Aborts all pending two-way invocations. Must be called outside the mutex lock after setting
 1339        // _refuseInvocations to true.
 1340        void AbortTwowayInvocations(IceRpcError error, string message, Exception? exception = null)
 911341        {
 911342            Debug.Assert(_refuseInvocations);
 1343
 1344            // _twowayInvocations is immutable once _refuseInvocations is true.
 2991345            foreach (TaskCompletionSource<PipeReader> responseCompletionSource in _twowayInvocations.Values)
 131346            {
 1347                // _twowayInvocations can hold completed completion sources.
 131348                _ = responseCompletionSource.TrySetException(new IceRpcException(error, message, exception));
 131349            }
 911350        }
 1351
 1352        // Takes appropriate action after a read failure.
 1353        void ReadFailed(Exception exception)
 661354        {
 1355            // We also prevent new one-way invocations even though they don't need to read the connection.
 661356            RefuseNewInvocations("The connection was lost because a read operation failed.");
 1357
 1358            // It's ok to cancel CTS and a "synchronous" TCS below. We won't be reading anything else so it's ok to run
 1359            // continuations synchronously.
 1360
 661361            AbortTwowayInvocations(
 661362                IceRpcError.ConnectionAborted,
 661363                "The invocation was aborted because the connection was lost.",
 661364                exception);
 1365
 1366            // ReadFailed is called when the connection is dead or the peer sent us a non-supported frame (e.g. a
 1367            // batch request). We don't need to allow outstanding two-way dispatches to complete in these situations, so
 1368            // we cancel them to speed-up the shutdown.
 661369            _twowayDispatchesCts.Cancel();
 1370
 1371            lock (_mutex)
 661372            {
 1373                // Don't send a close connection frame since we can't wait for the peer's acknowledgment.
 661374                _sendCloseConnectionFrame = false;
 661375            }
 1376
 661377            _ = _shutdownRequestedTcs.TrySetResult();
 661378        }
 1391379    }
 1380
 1381    /// <summary>Reads a reply (incoming response) and completes the invocation response completion source with this
 1382    /// response. This method executes "synchronously" in the read frames loop.</summary>
 1383    private async Task ReadReplyAsync(int replyFrameSize, CancellationToken cancellationToken)
 13771384    {
 1385        // Read the remainder of the frame immediately into frameReader.
 13771386        PipeReader replyFrameReader = await CreateFrameReaderAsync(
 13771387            replyFrameSize - IceDefinitions.PrologueSize,
 13771388            cancellationToken).ConfigureAwait(false);
 1389
 13771390        bool completeFrameReader = true;
 1391
 1392        try
 13771393        {
 1394            // Read and decode request ID
 13771395            if (!replyFrameReader.TryRead(out ReadResult readResult) || readResult.Buffer.Length < 4)
 01396            {
 01397                throw new InvalidDataException("Received a response with an invalid request ID.");
 1398            }
 1399
 13771400            ReadOnlySequence<byte> requestIdBuffer = readResult.Buffer.Slice(0, 4);
 27541401            int requestId = requestIdBuffer.DecodeIceBuffer((ref IceDecoder decoder) => decoder.DecodeInt());
 13771402            replyFrameReader.AdvanceTo(requestIdBuffer.End);
 1403
 1404            lock (_mutex)
 13771405            {
 13771406                if (_twowayInvocations.TryGetValue(
 13771407                    requestId,
 13771408                    out TaskCompletionSource<PipeReader>? responseCompletionSource))
 3701409                {
 1410                    // continuation runs asynchronously
 3701411                    if (responseCompletionSource.TrySetResult(replyFrameReader))
 3701412                    {
 3701413                        completeFrameReader = false;
 3701414                    }
 1415                    // else this invocation just completed and is about to remove itself from _twowayInvocations,
 1416                    // or _twowayInvocations is immutable and contains entries for completed invocations.
 3701417                }
 1418                // else the request ID carried by the response is bogus or corresponds to a request that was previously
 1419                // discarded (for example, because its deadline expired).
 13771420            }
 13771421        }
 1422        finally
 13771423        {
 13771424            if (completeFrameReader)
 10071425            {
 10071426                replyFrameReader.Complete();
 10071427            }
 13771428        }
 13771429    }
 1430
 1431    /// <summary>Reads and then dispatches an incoming request in a separate dispatch task. This method executes
 1432    /// "synchronously" in the read frames loop.</summary>
 1433    private async Task ReadRequestAsync(int requestFrameSize, CancellationToken cancellationToken)
 13941434    {
 1435        // Read the request frame.
 13941436        PipeReader requestFrameReader = await CreateFrameReaderAsync(
 13941437            requestFrameSize - IceDefinitions.PrologueSize,
 13941438            cancellationToken).ConfigureAwait(false);
 1439
 1440        // Decode its header.
 1441        int requestId;
 1442        IceRequestHeader requestHeader;
 13941443        PipeReader? contextReader = null;
 1444        IDictionary<RequestFieldKey, ReadOnlySequence<byte>>? fields;
 13941445        Task? dispatchTask = null;
 1446
 1447        try
 13941448        {
 13941449            if (!requestFrameReader.TryRead(out ReadResult readResult))
 01450            {
 01451                throw new InvalidDataException("Received an invalid request frame.");
 1452            }
 1453
 13941454            Debug.Assert(readResult.IsCompleted);
 1455
 13941456            (requestId, requestHeader, contextReader, int consumed) = DecodeRequestIdAndHeader(readResult.Buffer);
 13941457            requestFrameReader.AdvanceTo(readResult.Buffer.GetPosition(consumed));
 1458
 13941459            if (contextReader is null)
 13871460            {
 13871461                fields = requestHeader.OperationMode == OperationMode.Normal ?
 13871462                    ImmutableDictionary<RequestFieldKey, ReadOnlySequence<byte>>.Empty : _idempotentFields;
 13871463            }
 1464            else
 71465            {
 71466                contextReader.TryRead(out ReadResult result);
 71467                Debug.Assert(result.Buffer.Length > 0 && result.IsCompleted);
 71468                fields = new Dictionary<RequestFieldKey, ReadOnlySequence<byte>>()
 71469                {
 71470                    [RequestFieldKey.Context] = result.Buffer
 71471                };
 1472
 71473                if (requestHeader.OperationMode != OperationMode.Normal)
 01474                {
 1475                    // OperationMode can be Idempotent or Nonmutating.
 01476                    fields[RequestFieldKey.Idempotent] = default;
 01477                }
 71478            }
 1479
 13941480            bool releaseDispatchSemaphore = false;
 13941481            if (_dispatchSemaphore is SemaphoreSlim dispatchSemaphore)
 13941482            {
 1483                // This prevents us from receiving any new frames if we're already dispatching the maximum number
 1484                // of requests. We need to do this in the "accept from network loop" to apply back pressure to the
 1485                // caller.
 1486                try
 13941487                {
 13941488                    await dispatchSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
 13931489                    releaseDispatchSemaphore = true;
 13931490                }
 11491                catch (OperationCanceledException)
 11492                {
 1493                    // and return below
 11494                }
 13941495            }
 1496
 1497            lock (_mutex)
 13941498            {
 13941499                if (_shutdownTask is not null)
 21500                {
 1501                    // The connection is (being) disposed or the connection is shutting down and received a request.
 1502                    // We simply discard it. For a graceful shutdown, the two-way invocation in the peer will throw
 1503                    // IceRpcException(InvocationCanceled). We also discard one-way requests: if we accepted them, they
 1504                    // could delay our shutdown and make it time out.
 21505                    if (releaseDispatchSemaphore)
 11506                    {
 11507                        _dispatchSemaphore!.Release();
 11508                    }
 21509                    return;
 1510                }
 1511
 13921512                IncrementDispatchInvocationCount();
 13921513            }
 1514
 1515            // The scheduling of the task can't be canceled since we want to make sure DispatchRequestAsync will
 1516            // cleanup (decrement _dispatchCount etc.) if DisposeAsync is called. dispatchTask takes ownership of the
 1517            // requestFrameReader and contextReader.
 13921518            dispatchTask = Task.Run(
 13921519                async () =>
 13921520                {
 13921521                    using var request = new IncomingRequest(Protocol.Ice, _connectionContext!)
 13921522                    {
 13921523                        Fields = fields,
 13921524                        Fragment = requestHeader.Facet.ToFragment(),
 13921525                        IsOneway = requestId == 0,
 13921526                        Operation = requestHeader.Operation,
 13921527                        Path = requestHeader.Identity.ToPath(),
 13921528                        Payload = requestFrameReader,
 13921529                    };
 13921530
 13921531                    try
 13921532                    {
 13921533                        await DispatchRequestAsync(
 13921534                            request,
 13921535                            requestId,
 13921536                            contextReader).ConfigureAwait(false);
 13921537                    }
 01538                    catch (IceRpcException)
 01539                    {
 13921540                        // expected when the peer aborts the connection.
 01541                    }
 01542                    catch (Exception exception)
 01543                    {
 13921544                        // With ice, a dispatch cannot throw an exception that comes from the application code:
 13921545                        // any exception thrown when reading the response payload is converted into a DispatchException
 13921546                        // response, and the response header has no fields to encode.
 01547                        Debug.Fail($"ice dispatch {request} failed with an unexpected exception: {exception}");
 01548                        throw;
 13921549                    }
 13921550                },
 13921551                CancellationToken.None);
 13921552        }
 1553        finally
 13941554        {
 13941555            if (dispatchTask is null)
 21556            {
 21557                requestFrameReader.Complete();
 21558                contextReader?.Complete();
 21559            }
 13941560        }
 13941561    }
 1562
 1563    private void RefuseNewInvocations(string message)
 3971564    {
 1565        lock (_mutex)
 3971566        {
 3971567            _refuseInvocations = true;
 3971568            _invocationRefusedMessage ??= message;
 3971569        }
 3971570    }
 1571
 1572    /// <summary>Sends a control frame. It takes care of acquiring and releasing the write lock and calls
 1573    /// <see cref="WriteFailed" /> if a failure occurs while writing to _duplexConnectionWriter.</summary>
 1574    /// <param name="encode">Encodes the control frame.</param>
 1575    /// <param name="cancellationToken">The cancellation token.</param>
 1576    /// <remarks>If the cancellation token is canceled while writing to the duplex connection, the connection is
 1577    /// aborted.</remarks>
 1578    private async ValueTask SendControlFrameAsync(
 1579        Action<IBufferWriter<byte>> encode,
 1580        CancellationToken cancellationToken)
 1491581    {
 1491582        using SemaphoreLock _ = await AcquireWriteLockAsync(cancellationToken).ConfigureAwait(false);
 1583
 1584        try
 1471585        {
 1471586            encode(_duplexConnectionWriter);
 1471587            await _duplexConnectionWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
 1431588        }
 41589        catch (Exception exception)
 41590        {
 41591            WriteFailed(exception);
 41592            throw;
 1593        }
 1431594    }
 1595
 1596    /// <summary>Takes appropriate action after a write failure.</summary>
 1597    /// <remarks>Must be called outside the mutex lock but after acquiring _writeSemaphore.</remarks>
 1598    private void WriteFailed(Exception exception)
 61599    {
 61600        Debug.Assert(_writeException is null);
 61601        _writeException = exception; // protected by _writeSemaphore
 1602
 1603        // We can't send new invocations without writing to the connection.
 61604        RefuseNewInvocations("The connection was lost because a write operation failed.");
 1605
 1606        // We can't send responses so these dispatches can be canceled.
 61607        _twowayDispatchesCts.Cancel();
 1608
 1609        // We don't change _sendClosedConnectionFrame. If the _readFrameTask is still running, we want ShutdownAsync
 1610        // to send CloseConnection - and fail.
 1611
 61612        _ = _shutdownRequestedTcs.TrySetResult();
 61613    }
 1614}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
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)