< Summary

Information
Class: IceRpc.Internal.IceRpcProtocolConnection
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Internal/IceRpcProtocolConnection.cs
Tag: 2300_35243572715
Line coverage
91%
Covered lines: 863
Uncovered lines: 81
Coverable lines: 944
Total lines: 1616
Line coverage: 91.4%
Branch coverage
87%
Covered branches: 190
Total branches: 216
Branch coverage: 87.9%
Method coverage
100%
Covered methods: 34
Fully covered methods: 21
Total methods: 34
Method coverage: 100%
Full method coverage: 61.7%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_IsServer()100%11100%
.ctor(...)50%22100%
ConnectAsync(...)50%2281.81%
PerformConnectAsync()91.66%121288.73%
DisposeAsync()100%88100%
PerformDisposeAsync()92.85%1414100%
InvokeAsync(...)75%161270.83%
PerformInvokeAsync()78.12%343288.37%
OnGoAway()100%66100%
DecodeHeader()100%22100%
EncodeHeader()100%11100%
ShutdownAsync(...)100%1010100%
PerformShutdownAsync()81.25%201674.6%
DecodeFieldDictionary(...)87.5%8888.67%
AcceptRequestsAsync()75%9877.5%
CheckPeerHeaderSize(...)100%22100%
DecrementDispatchInvocationCount()100%88100%
DecrementStreamInputOutputCount()100%88100%
DispatchRequestAsync()96.66%303094.54%
PerformDispatchRequestAsync()83.33%66100%
DecodeHeader()100%11100%
EncodeHeader()100%22100%
EncodeFieldDictionary(...)100%22100%
IncrementDispatchInvocationCount()100%44100%
IncrementStreamInputOutputCount(...)100%22100%
ReadGoAwayAsync()100%2286.36%
ReceiveControlFrameHeaderAsync()100%44100%
ReceiveSettingsFrameBody()100%22100%
RefuseNewInvocations(...)100%22100%
ScheduleInactivityCheck()100%11100%
SendControlFrameAsync(...)100%11100%
EncodeFrame()100%11100%
SendRequestPayloadContinuation(...)100%11100%
PerformSendRequestPayloadContinuationAsync()50%2286.95%

File(s)

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

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Transports;
 4using System.Buffers;
 5using System.Collections.Immutable;
 6using System.Diagnostics;
 7using System.IO.Pipelines;
 8using System.Security.Authentication;
 9using ZeroC.Slice.Codec;
 10
 11namespace IceRpc.Internal;
 12
 13internal sealed class IceRpcProtocolConnection : IProtocolConnection
 14{
 15    private const int MaxGoAwayFrameBodySize = 16;
 16    private const int MaxSettingsFrameBodySize = 1024;
 17
 150418    private bool IsServer => _transportConnectionInformation is not null;
 19
 20    private Task? _acceptRequestsTask;
 21
 22    private Task? _connectTask;
 23    private IConnectionContext? _connectionContext; // non-null once the connection is established
 24    private IMultiplexedStream? _controlStream;
 25
 26    // The number of outstanding dispatches and invocations.
 27    // DisposeAsync waits until this count reaches 0 (using _dispatchesAndInvocationsCompleted) before disposing the
 28    // underlying transport connection. So when this count is greater than 0, we know _transportConnection and other
 29    // fields are not disposed.
 30    // _dispatchInvocationCount is also used for the inactivity check: the connection remains active while
 31    // _dispatchInvocationCount > 0 or _streamInputOutputCount > 0.
 32    private int _dispatchInvocationCount;
 33
 34    private readonly SemaphoreSlim? _dispatchSemaphore;
 35
 36    private readonly IDispatcher? _dispatcher;
 41537    private readonly TaskCompletionSource _dispatchesAndInvocationsCompleted =
 41538        new(TaskCreationOptions.RunContinuationsAsynchronously);
 39
 40    private Task? _disposeTask;
 41
 42    // This cancellation token source is canceled when the connection is disposed.
 41543    private readonly CancellationTokenSource _disposedCts = new();
 44
 45    // Canceled when we receive the GoAway frame from the peer.
 41546    private readonly CancellationTokenSource _goAwayCts = new();
 47
 48    // The GoAway frame received from the peer. Read it only after _goAwayCts is canceled.
 49    private IceRpcGoAway _goAwayFrame;
 50
 51    // The number of bytes we need to encode a size up to _maxPeerHeaderSize. It's 2 for DefaultMaxIceRpcHeaderSize.
 41552    private int _headerSizeLength = 2;
 53
 54    private readonly TimeSpan _inactivityTimeout;
 55    private readonly Timer _inactivityTimeoutTimer;
 56    private string? _invocationRefusedMessage;
 57
 58    // The ID of the last bidirectional stream accepted by this connection. It's null as long as no bidirectional stream
 59    // was accepted.
 60    private ulong? _lastRemoteBidirectionalStreamId;
 61
 62    // The ID of the last unidirectional stream accepted by this connection. It's null as long as no unidirectional
 63    // stream (other than _remoteControlStream) was accepted.
 64    private ulong? _lastRemoteUnidirectionalStreamId;
 65
 66    private readonly int _maxLocalHeaderSize;
 41567    private int _maxPeerHeaderSize = ConnectionOptions.DefaultMaxIceRpcHeaderSize;
 68
 41569    private readonly Lock _mutex = new();
 70
 71    private Task? _readGoAwayTask;
 72
 73    // A connection refuses invocations when it's disposed, shut down, shutting down or merely "shutdown requested".
 74    private bool _refuseInvocations;
 75
 76    private IMultiplexedStream? _remoteControlStream;
 77
 78    private readonly CancellationTokenSource _shutdownOrGoAwayCts;
 79
 80    // The thread that completes this TCS can run the continuations, and as a result its result must be set without
 81    // holding a lock on _mutex.
 41582    private readonly TaskCompletionSource _shutdownRequestedTcs = new();
 83
 84    private Task? _shutdownTask;
 85
 86    // Keeps track of the number of stream Input and Output that are not completed yet.
 87    // It's not the same as the _dispatchInvocationCount: a dispatch or invocation can be completed while the
 88    // application is still reading an incoming frame payload that corresponds to a stream input.
 89    // ShutdownAsync waits for both _streamInputOutputCount and _dispatchInvocationCount to reach 0, while DisposeAsync
 90    // only waits for _dispatchInvocationCount to reach 0.
 91    private int _streamInputOutputCount;
 92
 93    // The streams are completed when _shutdownTask is not null and _streamInputOutputCount is 0.
 41594    private readonly TaskCompletionSource _streamsCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
 95
 96    private readonly ITaskExceptionObserver? _taskExceptionObserver;
 97
 98    private readonly IMultiplexedConnection _transportConnection;
 99
 100    // Only set for server connections.
 101    private readonly TransportConnectionInformation? _transportConnectionInformation;
 102
 103    public Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> ConnectAsync(
 104        CancellationToken cancellationToken)
 408105    {
 106        Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> result;
 107
 108        lock (_mutex)
 408109        {
 408110            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 111
 406112            if (_connectTask is not null)
 0113            {
 0114                throw new InvalidOperationException("Cannot call connect more than once.");
 115            }
 116
 406117            result = PerformConnectAsync();
 406118            _connectTask = result;
 406119        }
 406120        return result;
 121
 122        async Task<(TransportConnectionInformation ConnectionInformation, Task ShutdownRequested)> PerformConnectAsync()
 406123        {
 124            // Make sure we execute the function without holding the connection mutex lock.
 406125            await Task.Yield();
 126
 127            // _disposedCts is not disposed at this point because DisposeAsync waits for the completion of _connectTask.
 406128            using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(
 406129                cancellationToken,
 406130                _disposedCts.Token);
 131
 132            TransportConnectionInformation transportConnectionInformation;
 133
 134            try
 406135            {
 136                // If the transport connection information is null, we need to connect the transport connection. It's
 137                // null for client connections. The transport connection of a server connection is established by
 138                // Server.
 406139                transportConnectionInformation = _transportConnectionInformation ??
 406140                    await _transportConnection.ConnectAsync(connectCts.Token).ConfigureAwait(false);
 141
 381142                _controlStream = await _transportConnection.CreateStreamAsync(
 381143                    false,
 381144                    connectCts.Token).ConfigureAwait(false);
 145
 372146                var settings = new IceRpcSettings(
 372147                    _maxLocalHeaderSize == ConnectionOptions.DefaultMaxIceRpcHeaderSize ?
 372148                        ImmutableDictionary<IceRpcSettingKey, ulong>.Empty :
 372149                        new Dictionary<IceRpcSettingKey, ulong>
 372150                        {
 372151                            [IceRpcSettingKey.MaxHeaderSize] = (ulong)_maxLocalHeaderSize
 372152                        });
 153
 154                try
 372155                {
 372156                    await SendControlFrameAsync(
 372157                        IceRpcControlFrameType.Settings,
 372158                        settings.Encode,
 372159                        connectCts.Token).ConfigureAwait(false);
 368160                }
 4161                catch
 4162                {
 163                    // If we fail to send the Settings frame, we are in an abortive closure and we close Output to allow
 164                    // the peer to continue if it's waiting for us. This could happen when the cancellation token is
 165                    // canceled.
 4166                    _controlStream!.Output.CompleteOutput(success: false);
 4167                    throw;
 168                }
 169
 170                // Wait for the remote control stream to be accepted and read the protocol Settings frame
 368171                _remoteControlStream = await _transportConnection.AcceptStreamAsync(
 368172                    connectCts.Token).ConfigureAwait(false);
 173
 355174                await ReceiveControlFrameHeaderAsync(
 355175                    IceRpcControlFrameType.Settings,
 355176                    connectCts.Token).ConfigureAwait(false);
 177
 349178                await ReceiveSettingsFrameBody(connectCts.Token).ConfigureAwait(false);
 344179            }
 25180            catch (OperationCanceledException)
 25181            {
 25182                cancellationToken.ThrowIfCancellationRequested();
 183
 7184                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 7185                throw new IceRpcException(
 7186                    IceRpcError.OperationAborted,
 7187                    "The connection establishment was aborted because the connection was disposed.");
 188            }
 7189            catch (InvalidDataException exception)
 7190            {
 7191                throw new IceRpcException(
 7192                    IceRpcError.ConnectionAborted,
 7193                    "The connection establishment was aborted by an icerpc protocol error.",
 7194                    exception);
 195            }
 1196            catch (AuthenticationException)
 1197            {
 1198                throw;
 199            }
 29200            catch (IceRpcException)
 29201            {
 29202                throw;
 203            }
 0204            catch (Exception exception)
 0205            {
 0206                Debug.Fail($"ConnectAsync failed with an unexpected exception: {exception}");
 0207                throw;
 208            }
 209
 210            // This needs to be set before starting the accept requests task below.
 344211            _connectionContext = new ConnectionContext(this, transportConnectionInformation);
 212
 213            // We assign _readGoAwayTask and _acceptRequestsTask with _mutex locked to make sure this assignment
 214            // occurs before the start of DisposeAsync. Once _disposeTask is not null, _readGoAwayTask etc are
 215            // immutable.
 216            lock (_mutex)
 344217            {
 344218                if (_disposeTask is not null)
 0219                {
 0220                    throw new IceRpcException(
 0221                        IceRpcError.OperationAborted,
 0222                        "The connection establishment was aborted because the connection was disposed.");
 223                }
 224
 225                // Read the go away frame from the control stream.
 344226                _readGoAwayTask = ReadGoAwayAsync(_disposedCts.Token);
 227
 228                // Start a task that accepts requests (the "accept requests loop")
 344229                _acceptRequestsTask = AcceptRequestsAsync(_shutdownOrGoAwayCts.Token);
 344230            }
 231
 232            // The _acceptRequestsTask waits for this PerformConnectAsync completion before reading anything. As soon as
 233            // it receives a request, it will cancel this inactivity check.
 344234            ScheduleInactivityCheck();
 235
 344236            return (transportConnectionInformation, _shutdownRequestedTcs.Task);
 344237        }
 406238    }
 239
 240    public ValueTask DisposeAsync()
 441241    {
 242        lock (_mutex)
 441243        {
 441244            if (_disposeTask is null)
 415245            {
 415246                RefuseNewInvocations("The connection was disposed.");
 247
 415248                if (_streamInputOutputCount == 0)
 402249                {
 250                    // That's only for consistency. _streamsCompleted.Task matters only to ShutdownAsync.
 402251                    _streamsCompleted.TrySetResult();
 402252                }
 415253                if (_dispatchInvocationCount == 0)
 406254                {
 406255                    _dispatchesAndInvocationsCompleted.TrySetResult();
 406256                }
 257
 415258                _shutdownTask ??= Task.CompletedTask;
 415259                _disposeTask = PerformDisposeAsync();
 415260            }
 441261        }
 441262        return new(_disposeTask);
 263
 264        async Task PerformDisposeAsync()
 415265        {
 266            // Make sure we execute the code below without holding the mutex lock.
 415267            await Task.Yield();
 268
 415269            _disposedCts.Cancel();
 270
 271            // We don't lock _mutex since once _disposeTask is not null, _connectTask etc are immutable.
 272
 415273            if (_connectTask is not null)
 406274            {
 275                // We wait for _dispatchesAndInvocationsCompleted (since dispatches and invocations are somewhat under
 276                // our control), but not for _streamsCompleted, since we can't make the application complete the
 277                // incoming payload pipe readers.
 278                try
 406279                {
 406280                    await Task.WhenAll(
 406281                        _connectTask,
 406282                        _acceptRequestsTask ?? Task.CompletedTask,
 406283                        _readGoAwayTask ?? Task.CompletedTask,
 406284                        _shutdownTask,
 406285                        _dispatchesAndInvocationsCompleted.Task).ConfigureAwait(false);
 80286                }
 326287                catch
 326288                {
 289                    // Expected if any of these tasks failed or was canceled. Each task takes care of handling
 290                    // unexpected exceptions so there's no need to handle them here.
 326291                }
 406292            }
 293
 294            // If the application is still reading some incoming payload, the disposal of the transport connection can
 295            // abort this reading.
 415296            await _transportConnection.DisposeAsync().ConfigureAwait(false);
 297
 298            // It's safe to complete the output since write operations have been completed by the transport connection
 299            // disposal.
 415300            _controlStream?.Output.Complete();
 301
 302            // It's safe to complete the input since read operations have been completed by the transport connection
 303            // disposal.
 415304            _remoteControlStream?.Input.Complete();
 305
 415306            _dispatchSemaphore?.Dispose();
 415307            _disposedCts.Dispose();
 415308            _goAwayCts.Dispose();
 415309            _shutdownOrGoAwayCts.Dispose();
 310
 415311            await _inactivityTimeoutTimer.DisposeAsync().ConfigureAwait(false);
 415312        }
 441313    }
 314
 315    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken = default)
 1450316    {
 1450317        if (request.Protocol != Protocol.IceRpc)
 1318        {
 1319            throw new InvalidOperationException(
 1320                $"Cannot send {request.Protocol} request on {Protocol.IceRpc} connection.");
 321        }
 322
 323        lock (_mutex)
 1449324        {
 1449325            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 326
 1447327            if (_refuseInvocations)
 6328            {
 6329                throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage);
 330            }
 1441331            if (_connectTask is null)
 0332            {
 0333                throw new InvalidOperationException("Cannot invoke on a connection before connecting it.");
 334            }
 1441335            if (!IsServer && !_connectTask.IsCompletedSuccessfully)
 0336            {
 0337                throw new InvalidOperationException(
 0338                    "Cannot invoke on a client connection that is not fully established.");
 339            }
 340            // It's possible but rare to invoke on a server connection that is still connecting.
 341
 1441342            if (request.ServiceAddress.Fragment.Length > 0)
 0343            {
 0344                throw new NotSupportedException("The icerpc protocol does not support fragments.");
 345            }
 346
 1441347            IncrementDispatchInvocationCount();
 1441348        }
 349
 1441350        return PerformInvokeAsync();
 351
 352        async Task<IncomingResponse> PerformInvokeAsync()
 1441353        {
 354            // Since _dispatchInvocationCount > 0, _disposedCts is not disposed.
 1441355            using var invocationCts = CancellationTokenSource.CreateLinkedTokenSource(
 1441356                cancellationToken,
 1441357                _disposedCts.Token);
 358
 1441359            PipeReader? streamInput = null;
 360
 361            // This try/catch block cleans up streamInput (when not null) and decrements the dispatch-invocation count.
 362            try
 1441363            {
 364                // Create the stream.
 365                IMultiplexedStream stream;
 366                try
 1441367                {
 368                    // We want to cancel CreateStreamAsync as soon as the connection is being shutdown or received a
 369                    // GoAway frame.
 1441370                    using CancellationTokenRegistration _ = _shutdownOrGoAwayCts.Token.UnsafeRegister(
 6371                        cts => ((CancellationTokenSource)cts!).Cancel(),
 1441372                        invocationCts);
 373
 1441374                    stream = await _transportConnection.CreateStreamAsync(
 1441375                        bidirectional: !request.IsOneway,
 1441376                        invocationCts.Token).ConfigureAwait(false);
 377
 1432378                    streamInput = stream.IsBidirectional ? stream.Input : null;
 1432379                }
 8380                catch (OperationCanceledException)
 8381                {
 8382                    cancellationToken.ThrowIfCancellationRequested();
 383
 384                    // Connection was shut down or disposed and we did not read the payload at all.
 6385                    throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage);
 386                }
 1387                catch (IceRpcException exception)
 1388                {
 1389                    RefuseNewInvocations("The connection was lost.");
 1390                    throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage, exception);
 391                }
 0392                catch (Exception exception)
 0393                {
 0394                    Debug.Fail($"CreateStreamAsync failed with an unexpected exception: {exception}");
 0395                    RefuseNewInvocations("The connection was lost.");
 0396                    throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage, exception);
 397                }
 398
 1432399                using CancellationTokenRegistration tokenRegistration = _goAwayCts.Token.UnsafeRegister(
 1432400                    OnGoAway,
 1432401                    invocationCts);
 402
 403                PipeWriter payloadWriter;
 404
 405                try
 1432406                {
 407                    lock (_mutex)
 1432408                    {
 1432409                        if (_refuseInvocations)
 0410                        {
 411                            // Both stream.Output and stream.Output are completed by catch blocks below.
 0412                            throw new IceRpcException(IceRpcError.InvocationRefused, _invocationRefusedMessage);
 413                        }
 414
 1432415                        IncrementStreamInputOutputCount(stream.IsBidirectional);
 416
 417                        // Decorate the stream to decrement the input/output count on Complete.
 1432418                        stream = new MultiplexedStreamDecorator(stream, DecrementStreamInputOutputCount);
 1432419                        streamInput = stream.IsBidirectional ? stream.Input : null;
 1432420                    }
 421
 1432422                    EncodeHeader(stream.Output);
 1431423                    payloadWriter = request.GetPayloadWriter(stream.Output);
 1431424                }
 1425                catch
 1426                {
 1427                    stream.Output.CompleteOutput(success: false);
 1428                    throw;
 429                }
 430
 431                // From now on, we only use payloadWriter to write and we make sure to complete it.
 432
 1431433                bool hasContinuation = request.PayloadContinuation is not null;
 434                FlushResult flushResult;
 435
 436                try
 1431437                {
 1431438                    flushResult = await payloadWriter.CopyFromAsync(
 1431439                        request.Payload,
 1431440                        stream.WritesClosed,
 1431441                        endStream: !hasContinuation,
 1431442                        invocationCts.Token).ConfigureAwait(false);
 1423443                }
 8444                catch
 8445                {
 8446                    payloadWriter.CompleteOutput(success: false);
 8447                    request.PayloadContinuation?.Complete();
 8448                    throw;
 449                }
 450                finally
 1431451                {
 1431452                    request.Payload.Complete();
 1431453                }
 454
 1423455                if (flushResult.IsCompleted || flushResult.IsCanceled || !hasContinuation)
 1411456                {
 457                    // The remote reader doesn't want more data, or the copying was canceled, or there is no
 458                    // continuation: we're done.
 1411459                    payloadWriter.CompleteOutput(!flushResult.IsCanceled);
 1411460                    request.PayloadContinuation?.Complete();
 1411461                }
 462                else
 12463                {
 464                    // Sends the payload continuation in a background thread.
 12465                    SendRequestPayloadContinuation(
 12466                        request,
 12467                        payloadWriter,
 12468                        stream.WritesClosed,
 12469                        OnGoAway,
 12470                        invocationCts.Token);
 12471                }
 472
 1423473                if (request.IsOneway)
 1010474                {
 1010475                    return new IncomingResponse(request, _connectionContext!);
 476                }
 477
 413478                Debug.Assert(streamInput is not null);
 479
 480                try
 413481                {
 413482                    ReadResult readResult = await streamInput.ReadSliceSegmentAsync(
 413483                        _maxLocalHeaderSize,
 413484                        invocationCts.Token).ConfigureAwait(false);
 485
 486                    // Nothing cancels the stream input pipe reader.
 390487                    Debug.Assert(!readResult.IsCanceled);
 488
 390489                    if (readResult.Buffer.IsEmpty)
 0490                    {
 0491                        throw new IceRpcException(
 0492                            IceRpcError.IceRpcError,
 0493                            "Received an icerpc response with an empty header.");
 494                    }
 495
 390496                    (StatusCode statusCode, string? errorMessage, IDictionary<ResponseFieldKey, ReadOnlySequence<byte>> 
 390497                        DecodeHeader(readResult.Buffer);
 389498                    stream.Input.AdvanceTo(readResult.Buffer.End);
 499
 389500                    if (statusCode == StatusCode.TruncatedPayload && invocationCts.Token.IsCancellationRequested)
 0501                    {
 502                        // Canceling the sending of the payload continuation triggers the completion of the stream
 503                        // output. This may lead to a TruncatedPayload if the dispatch is currently reading the payload
 504                        // continuation. In such cases, we prioritize throwing an OperationCanceledException.
 0505                        fieldsPipeReader?.Complete();
 0506                        invocationCts.Token.ThrowIfCancellationRequested();
 0507                    }
 508
 389509                    var response = new IncomingResponse(
 389510                        request,
 389511                        _connectionContext!,
 389512                        statusCode,
 389513                        errorMessage,
 389514                        fields,
 389515                        fieldsPipeReader)
 389516                    {
 389517                        Payload = streamInput
 389518                    };
 519
 389520                    streamInput = null; // response now owns the stream input
 389521                    return response;
 522                }
 2523                catch (InvalidDataException exception)
 2524                {
 2525                    throw new IceRpcException(
 2526                        IceRpcError.IceRpcError,
 2527                        "Received an icerpc response with an invalid header.",
 2528                        exception);
 529                }
 530
 531                void OnGoAway(object? cts)
 13532                {
 13533                    if (!stream.IsStarted ||
 13534                        stream.Id >=
 13535                            (stream.IsBidirectional ?
 13536                                _goAwayFrame.BidirectionalStreamId :
 13537                                _goAwayFrame.UnidirectionalStreamId))
 4538                    {
 539                        // The request wasn't received by the peer so it's safe to cancel the invocation.
 4540                        ((CancellationTokenSource)cts!).Cancel();
 4541                    }
 13542                }
 543            }
 16544            catch (OperationCanceledException exception) when (exception.CancellationToken == invocationCts.Token)
 13545            {
 13546                cancellationToken.ThrowIfCancellationRequested();
 547
 6548                if (_disposedCts.IsCancellationRequested)
 3549                {
 550                    // DisposeAsync aborted the request.
 3551                    throw new IceRpcException(IceRpcError.OperationAborted);
 552                }
 553                else
 3554                {
 3555                    Debug.Assert(_goAwayCts.IsCancellationRequested);
 3556                    throw new IceRpcException(IceRpcError.InvocationCanceled, "The connection is shutting down.");
 557                }
 558            }
 559            finally
 1441560            {
 1441561                streamInput?.Complete();
 1441562                DecrementDispatchInvocationCount();
 1441563            }
 564
 565            static (StatusCode StatusCode, string? ErrorMessage, IDictionary<ResponseFieldKey, ReadOnlySequence<byte>>, 
 566                ReadOnlySequence<byte> buffer)
 390567            {
 390568                var decoder = new SliceDecoder(buffer);
 569
 390570                StatusCode statusCode = decoder.DecodeStatusCode();
 390571                string? errorMessage = statusCode == StatusCode.Ok ? null : decoder.DecodeString();
 572
 390573                (IDictionary<ResponseFieldKey, ReadOnlySequence<byte>> fields, PipeReader? pipeReader) =
 390574                    DecodeFieldDictionary(
 390575                        ref decoder,
 393576                        (ref SliceDecoder decoder) => decoder.DecodeResponseFieldKey());
 577
 389578                return (statusCode, errorMessage, fields, pipeReader);
 389579            }
 580
 581            void EncodeHeader(PipeWriter streamOutput)
 1432582            {
 1432583                var encoder = new SliceEncoder(streamOutput);
 584
 585                // Write the IceRpc request header.
 1432586                Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(_headerSizeLength);
 587
 588                // We use UnflushedBytes because EncodeFieldDictionary can write directly to streamOutput.
 1432589                long headerStartPos = streamOutput.UnflushedBytes;
 590
 1432591                var header = new IceRpcRequestHeader(request.ServiceAddress.Path, request.Operation);
 592
 1432593                header.Encode(ref encoder);
 594
 1432595                EncodeFieldDictionary(
 1432596                    request.Fields,
 14597                    (ref SliceEncoder encoder, RequestFieldKey key) => encoder.EncodeRequestFieldKey(key),
 1432598                    ref encoder,
 1432599                    streamOutput);
 600
 601                // We're done with the header encoding, write the header size.
 1432602                int headerSize = (int)(streamOutput.UnflushedBytes - headerStartPos);
 1432603                CheckPeerHeaderSize(headerSize);
 1431604                SliceEncoder.EncodeVarUInt62((uint)headerSize, sizePlaceholder);
 1431605            }
 1399606        }
 1441607    }
 608
 609    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 111610    {
 611        lock (_mutex)
 111612        {
 111613            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 614
 109615            if (_shutdownTask is not null)
 2616            {
 2617                throw new InvalidOperationException("Cannot call ShutdownAsync more than once.");
 618            }
 107619            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 3620            {
 3621                throw new InvalidOperationException("Cannot shut down a protocol connection before it's connected.");
 622            }
 623
 104624            RefuseNewInvocations("The connection was shut down.");
 625
 104626            if (_streamInputOutputCount == 0)
 82627            {
 82628                _streamsCompleted.TrySetResult();
 82629            }
 104630            if (_dispatchInvocationCount == 0)
 80631            {
 80632                _dispatchesAndInvocationsCompleted.TrySetResult();
 80633            }
 634
 104635            _shutdownTask = PerformShutdownAsync();
 104636        }
 104637        return _shutdownTask;
 638
 639        async Task PerformShutdownAsync()
 104640        {
 104641            await Task.Yield(); // exit mutex lock
 642
 104643            _shutdownOrGoAwayCts.Cancel();
 644
 645            try
 104646            {
 104647                Debug.Assert(_acceptRequestsTask is not null);
 104648                Debug.Assert(_controlStream is not null);
 104649                Debug.Assert(_readGoAwayTask is not null);
 104650                Debug.Assert(_remoteControlStream is not null);
 651
 104652                await _acceptRequestsTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 653
 89654                using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedCts.Token);
 655
 656                // Once shutdownTask is not null, _lastRemoteBidirectionalStreamId and _lastRemoteUnidirectionalStreamId
 657                // are immutable.
 658
 659                // When this peer is the server endpoint, the first accepted bidirectional stream ID is 0. When this
 660                // peer is the client endpoint, the first accepted bidirectional stream ID is 1.
 89661                IceRpcGoAway goAwayFrame = new(
 89662                    _lastRemoteBidirectionalStreamId is ulong value ? value + 4 : (IsServer ? 0ul : 1ul),
 89663                    (_lastRemoteUnidirectionalStreamId ?? _remoteControlStream.Id) + 4);
 664
 665                try
 89666                {
 89667                    _ = await SendControlFrameAsync(
 89668                        IceRpcControlFrameType.GoAway,
 89669                        goAwayFrame.Encode,
 89670                        cts.Token).ConfigureAwait(false);
 671
 672                    // Wait for the peer to send back a GoAway frame. The task should already be completed if the
 673                    // shutdown was initiated by the peer.
 87674                    await _readGoAwayTask.WaitAsync(cts.Token).ConfigureAwait(false);
 675
 676                    // Wait for all streams (other than the control streams) to have their Input and Output completed.
 77677                    await _streamsCompleted.Task.WaitAsync(cts.Token).ConfigureAwait(false);
 678
 679                    // Close the control stream to notify the peer that on our side, all the streams completed and that
 680                    // it can close the transport connection whenever it likes.
 76681                    _controlStream.Output.CompleteOutput(success: true);
 76682                }
 13683                catch
 13684                {
 685                    // If we fail to send the GoAway frame or some other failure occur (such as
 686                    // OperationCanceledException) we are in an abortive closure and we close Output to allow
 687                    // the peer to continue if it's waiting for us.
 13688                    _controlStream.Output.CompleteOutput(success: false);
 13689                    throw;
 690                }
 691
 692                // Wait for the peer notification that on its side all the streams are completed. It's important to wait
 693                // for this notification before closing the connection. In particular with QUIC where closing the
 694                // connection before all the streams are processed could lead to a stream failure.
 695                try
 76696                {
 697                    // Wait for the _remoteControlStream Input completion.
 76698                    ReadResult readResult = await _remoteControlStream.Input.ReadAsync(cts.Token).ConfigureAwait(false);
 699
 74700                    Debug.Assert(!readResult.IsCanceled);
 701
 74702                    if (!readResult.IsCompleted || !readResult.Buffer.IsEmpty)
 0703                    {
 0704                        throw new IceRpcException(
 0705                            IceRpcError.IceRpcError,
 0706                            "Received bytes on the control stream after receiving the GoAway frame.");
 707                    }
 708
 709                    // We can now safely close the connection.
 74710                    await _transportConnection.CloseAsync(MultiplexedConnectionCloseError.NoError, cts.Token)
 74711                        .ConfigureAwait(false);
 73712                }
 2713                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.ConnectionClosedByPeer)
 0714                {
 715                    // Expected if the peer closed the connection first.
 0716                }
 717
 718                // We wait for the completion of the dispatches that we created (and, secondarily, invocations).
 73719                await _dispatchesAndInvocationsCompleted.Task.WaitAsync(cts.Token).ConfigureAwait(false);
 73720            }
 9721            catch (OperationCanceledException)
 9722            {
 9723                cancellationToken.ThrowIfCancellationRequested();
 724
 3725                Debug.Assert(_disposedCts.Token.IsCancellationRequested);
 3726                throw new IceRpcException(
 3727                    IceRpcError.OperationAborted,
 3728                    "The connection shutdown was aborted because the connection was disposed.");
 729            }
 0730            catch (InvalidDataException exception)
 0731            {
 0732                throw new IceRpcException(
 0733                    IceRpcError.IceRpcError,
 0734                    "The connection shutdown was aborted by an icerpc protocol error.",
 0735                    exception);
 736            }
 22737            catch (IceRpcException)
 22738            {
 22739                throw;
 740            }
 0741            catch (Exception exception)
 0742            {
 0743                Debug.Fail($"ShutdownAsync failed with an unexpected exception: {exception}");
 0744                throw;
 745            }
 73746        }
 104747    }
 748
 415749    internal IceRpcProtocolConnection(
 415750        IMultiplexedConnection transportConnection,
 415751        TransportConnectionInformation? transportConnectionInformation,
 415752        ConnectionOptions options,
 415753        ITaskExceptionObserver? taskExceptionObserver)
 415754    {
 415755        _shutdownOrGoAwayCts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token, _goAwayCts.Token);
 756
 415757        _taskExceptionObserver = taskExceptionObserver;
 758
 415759        _transportConnection = transportConnection;
 415760        _dispatcher = options.Dispatcher;
 415761        _maxLocalHeaderSize = options.MaxIceRpcHeaderSize;
 415762        _transportConnectionInformation = transportConnectionInformation;
 763
 415764        if (options.MaxDispatches > 0)
 415765        {
 415766            _dispatchSemaphore = new SemaphoreSlim(
 415767                initialCount: options.MaxDispatches,
 415768                maxCount: options.MaxDispatches);
 415769        }
 770
 415771        _inactivityTimeout = options.InactivityTimeout;
 415772        _inactivityTimeoutTimer = new Timer(_ =>
 5773        {
 5774            bool requestShutdown = false;
 415775
 415776            lock (_mutex)
 5777            {
 5778                if (_shutdownTask is null && _dispatchInvocationCount == 0 && _streamInputOutputCount == 0)
 5779                {
 5780                    requestShutdown = true;
 5781                    RefuseNewInvocations(
 5782                        $"The connection was shut down because it was inactive for over {_inactivityTimeout.TotalSeconds
 5783                }
 5784            }
 415785
 5786            if (requestShutdown)
 5787            {
 415788                // TrySetResult must be called outside the mutex lock
 5789                _shutdownRequestedTcs.TrySetResult();
 5790            }
 420791        });
 415792    }
 793
 794    private static (IDictionary<TKey, ReadOnlySequence<byte>>, PipeReader?) DecodeFieldDictionary<TKey>(
 795        ref SliceDecoder decoder,
 796        DecodeFunc<TKey> decodeKeyFunc) where TKey : struct
 1810797    {
 1810798        int count = decoder.DecodeSize();
 799
 800        IDictionary<TKey, ReadOnlySequence<byte>> fields;
 801        PipeReader? pipeReader;
 1810802        if (count == 0)
 1790803        {
 1790804            fields = ImmutableDictionary<TKey, ReadOnlySequence<byte>>.Empty;
 1790805            pipeReader = null;
 1790806            decoder.CheckEndOfBuffer();
 1790807        }
 808        else
 20809        {
 810            // We don't use the normal collection allocation check here because SizeOf<ReadOnlySequence<byte>> is quite
 811            // large (24).
 812            // For example, say we decode a fields dictionary with a single field with an empty value. It's encoded
 813            // using 1 byte (dictionary size) + 1 byte (key) + 1 byte (value size) = 3 bytes. The decoder's default max
 814            // allocation size is 3 * 8 = 24. If we simply call IncreaseCollectionAllocation(1, 4 + 24), we'll exceed
 815            // the default collection allocation limit. (sizeof TKey is currently 4 but could/should increase to 8).
 816
 817            // Each field consumes at least 2 bytes: 1 for the key and one for the value size.
 20818            if ((long)count * 2 > decoder.Remaining)
 2819            {
 2820                throw new InvalidDataException("Too many fields.");
 821            }
 822
 18823            fields = new Dictionary<TKey, ReadOnlySequence<byte>>(count);
 18824            var pipe = new Pipe();
 18825            decoder.CopyTo(pipe.Writer);
 18826            pipe.Writer.Complete();
 827
 828            try
 18829            {
 18830                _ = pipe.Reader.TryRead(out ReadResult readResult);
 18831                var fieldsDecoder = new SliceDecoder(readResult.Buffer);
 832
 72833                for (int i = 0; i < count; ++i)
 19834                {
 835                    // Decode the field key.
 19836                    TKey key = decodeKeyFunc(ref fieldsDecoder);
 837
 838                    // Decode and check the field value size.
 839                    int valueSize;
 840                    try
 19841                    {
 19842                        valueSize = checked((int)fieldsDecoder.DecodeVarUInt62());
 19843                    }
 0844                    catch (OverflowException exception)
 0845                    {
 0846                        throw new InvalidDataException("The field size can't be larger than int.MaxValue.", exception);
 847                    }
 848
 19849                    if (valueSize > fieldsDecoder.Remaining)
 0850                    {
 0851                        throw new InvalidDataException(
 0852                            $"The value of field '{key}' extends beyond the end of the buffer.");
 853                    }
 854
 855                    // Create a ROS reference to the field value by slicing the fields pipe reader ROS.
 19856                    ReadOnlySequence<byte> value = readResult.Buffer.Slice(fieldsDecoder.Consumed, valueSize);
 857                    try
 19858                    {
 19859                        fields.Add(key, value);
 18860                    }
 1861                    catch (ArgumentException exception)
 1862                    {
 1863                        throw new InvalidDataException(
 1864                            $"Received icerpc header with duplicate field key '{key}'.", exception);
 865                    }
 866
 867                    // Skip the field value to prepare the decoder to read the next field value.
 18868                    fieldsDecoder.Skip(valueSize);
 18869                }
 17870                fieldsDecoder.CheckEndOfBuffer();
 871
 17872                pipe.Reader.AdvanceTo(readResult.Buffer.Start); // complete read without consuming anything
 873
 17874                pipeReader = pipe.Reader;
 17875            }
 1876            catch
 1877            {
 1878                pipe.Reader.Complete();
 1879                throw;
 880            }
 17881        }
 882
 883        // The caller is responsible for completing the pipe reader.
 1807884        return (fields, pipeReader);
 1807885    }
 886
 887    private async Task AcceptRequestsAsync(CancellationToken cancellationToken)
 344888    {
 344889        await Task.Yield(); // exit mutex lock
 890
 891        // Wait for _connectTask (which spawned the task running this method) to complete. This way, we won't dispatch
 892        // any request until _connectTask has completed successfully, and indirectly we won't make any invocation until
 893        // _connectTask has completed successfully. The creation of the _acceptRequestsTask is the last action taken by
 894        // _connectTask and as a result this await can't fail.
 344895        await _connectTask!.ConfigureAwait(false);
 896
 897        try
 344898        {
 899            // We check the cancellation token for each iteration because we want to exit the accept requests loop as
 900            // soon as ShutdownAsync/GoAway requests this cancellation, even when more streams can be accepted without
 901            // waiting.
 1777902            while (!cancellationToken.IsCancellationRequested)
 1777903            {
 904                // When _dispatcher is null, the multiplexed connection MaxUnidirectionalStreams and
 905                // MaxBidirectionalStreams options are configured to not accept any request-stream from the peer. As a
 906                // result, when _dispatcher is null, this call will block indefinitely until the cancellation token is
 907                // canceled by ShutdownAsync, GoAway or DisposeAsync.
 1777908                IMultiplexedStream stream = await _transportConnection.AcceptStreamAsync(cancellationToken)
 1777909                    .ConfigureAwait(false);
 910
 911                lock (_mutex)
 1433912                {
 913                    // We don't want to increment _dispatchInvocationCount/_streamInputOutputCount when the connection
 914                    // is shutting down or being disposed.
 1433915                    if (_shutdownTask is not null)
 0916                    {
 917                        // Note that cancellationToken may not be canceled yet at this point.
 0918                        throw new OperationCanceledException();
 919                    }
 920
 921                    // The logic in IncrementStreamInputOutputCount requires that we increment the dispatch-invocation
 922                    // count first.
 1433923                    IncrementDispatchInvocationCount();
 1433924                    IncrementStreamInputOutputCount(stream.IsBidirectional);
 925
 926                    // Decorate the stream to decrement the stream input/output count on Complete.
 1433927                    stream = new MultiplexedStreamDecorator(stream, DecrementStreamInputOutputCount);
 928
 929                    // The multiplexed connection guarantees that the IDs of accepted streams of a given type have ever
 930                    // increasing values.
 931
 1433932                    if (stream.IsBidirectional)
 424933                    {
 424934                        _lastRemoteBidirectionalStreamId = stream.Id;
 424935                    }
 936                    else
 1009937                    {
 1009938                        _lastRemoteUnidirectionalStreamId = stream.Id;
 1009939                    }
 1433940                }
 941
 942                // Start a task to read the stream and dispatch the request. We pass CancellationToken.None to Task.Run
 943                // because DispatchRequestAsync must clean-up the stream and the dispatch-invocation count.
 2866944                _ = Task.Run(() => DispatchRequestAsync(stream), CancellationToken.None);
 1433945            }
 0946        }
 231947        catch (OperationCanceledException)
 231948        {
 949            // Expected, the associated cancellation token source was canceled.
 231950        }
 113951        catch (IceRpcException)
 113952        {
 113953            RefuseNewInvocations("The connection was lost");
 113954            _ = _shutdownRequestedTcs.TrySetResult();
 113955            throw;
 956        }
 0957        catch (Exception exception)
 0958        {
 0959            Debug.Fail($"The accept stream task failed with an unexpected exception: {exception}");
 0960            RefuseNewInvocations("The connection was lost");
 0961            _ = _shutdownRequestedTcs.TrySetResult();
 0962            throw;
 963        }
 231964    }
 965
 966    private void CheckPeerHeaderSize(int headerSize)
 1833967    {
 1833968        if (headerSize > _maxPeerHeaderSize)
 2969        {
 2970            throw new IceRpcException(
 2971                IceRpcError.LimitExceeded,
 2972                $"The header size ({headerSize}) for an icerpc request or response is greater than the peer's max header
 973        }
 1831974    }
 975
 976    private void DecrementDispatchInvocationCount()
 2886977    {
 978        lock (_mutex)
 2886979        {
 2886980            if (--_dispatchInvocationCount == 0)
 803981            {
 803982                if (_shutdownTask is not null)
 30983                {
 30984                    _dispatchesAndInvocationsCompleted.TrySetResult();
 30985                }
 773986                else if (!_refuseInvocations && _streamInputOutputCount == 0)
 656987                {
 656988                    ScheduleInactivityCheck();
 656989                }
 803990            }
 2886991        }
 2886992    }
 993
 994    /// <summary>Decrements the stream input/output count.</summary>
 995    private void DecrementStreamInputOutputCount()
 3706996    {
 997        lock (_mutex)
 3706998        {
 3706999            if (--_streamInputOutputCount == 0)
 7951000            {
 7951001                if (_shutdownTask is not null)
 281002                {
 281003                    _streamsCompleted.TrySetResult();
 281004                }
 7671005                else if (!_refuseInvocations && _dispatchInvocationCount == 0)
 1071006                {
 1007                    // We enable the inactivity check in order to complete _shutdownRequestedTcs when inactive for too
 1008                    // long. _refuseInvocations is true when the connection is either about to be "shutdown requested",
 1009                    // or shut down / disposed. We don't need to complete _shutdownRequestedTcs in any of these
 1010                    // situations.
 1071011                    ScheduleInactivityCheck();
 1071012                }
 7951013            }
 37061014        }
 37061015    }
 1016
 1017    private async Task DispatchRequestAsync(IMultiplexedStream stream)
 14331018    {
 1019        // _disposedCts is not disposed since we own a dispatch count.
 14331020        CancellationToken cancellationToken = stream.IsBidirectional ?
 14331021            stream.WritesClosed.AsCancellationToken(_disposedCts.Token) :
 14331022            _disposedCts.Token;
 1023
 14331024        PipeReader? fieldsPipeReader = null;
 1025        IDictionary<RequestFieldKey, ReadOnlySequence<byte>> fields;
 1026        IceRpcRequestHeader header;
 1027
 14331028        PipeReader? streamInput = stream.Input;
 14331029        PipeWriter? streamOutput = stream.IsBidirectional ? stream.Output : null;
 14331030        bool success = false;
 1031
 1032        try
 14331033        {
 1034            try
 14331035            {
 14331036                ReadResult readResult = await streamInput.ReadSliceSegmentAsync(
 14331037                    _maxLocalHeaderSize,
 14331038                    cancellationToken).ConfigureAwait(false);
 1039
 14231040                if (readResult.Buffer.IsEmpty)
 01041                {
 01042                    throw new IceRpcException(IceRpcError.IceRpcError, "Received icerpc request with empty header.");
 1043                }
 1044
 14231045                (header, fields, fieldsPipeReader) = DecodeHeader(readResult.Buffer);
 14181046                streamInput.AdvanceTo(readResult.Buffer.End);
 14181047            }
 91048            catch (InvalidDataException exception)
 91049            {
 91050                var rpcException = new IceRpcException(
 91051                    IceRpcError.IceRpcError,
 91052                    "Received invalid icerpc request header.",
 91053                    exception);
 1054
 91055                if (_taskExceptionObserver is null)
 11056                {
 11057                    throw rpcException;
 1058                }
 1059                else
 81060                {
 81061                    _taskExceptionObserver.DispatchRefused(
 81062                        _connectionContext!.TransportConnectionInformation,
 81063                        rpcException);
 81064                    return; // success remains false
 1065                }
 1066            }
 61067            catch (Exception exception) when (_taskExceptionObserver is not null)
 31068            {
 31069                _taskExceptionObserver.DispatchRefused(_connectionContext!.TransportConnectionInformation, exception);
 31070                return; // success remains false
 1071            }
 1072
 14181073            using var request = new IncomingRequest(Protocol.IceRpc, _connectionContext!)
 14181074            {
 14181075                Fields = fields,
 14181076                IsOneway = !stream.IsBidirectional,
 14181077                Operation = header.Operation,
 14181078                Path = header.Path,
 14181079                Payload = streamInput
 14181080            };
 1081
 14181082            streamInput = null; // the request now owns streamInput
 1083
 1084            try
 14181085            {
 14181086                OutgoingResponse response = await PerformDispatchRequestAsync(request, cancellationToken)
 14181087                    .ConfigureAwait(false);
 1088
 14101089                if (!request.IsOneway)
 4021090                {
 4021091                    Debug.Assert(streamOutput is not null);
 4021092                    EncodeHeader(response);
 1093
 4001094                    PipeWriter payloadWriter = response.GetPayloadWriter(streamOutput);
 1095
 1096                    // Remains false if a copy throws or is canceled.
 4001097                    bool payloadWriterSuccess = false;
 1098
 1099                    try
 4001100                    {
 1101                        // We don't use cancellationToken here because it's canceled shortly afterwards by the
 1102                        // completion of writesClosed. This works around https://github.com/dotnet/runtime/issues/82704
 1103                        // where the stream would otherwise be aborted after the successful write. It's also fine to
 1104                        // just use _disposedCts.Token: if writes are closed because the peer is not longer interested
 1105                        // in the response, the write operations will raise an IceRpcException(StreamAborted) which is
 1106                        // ignored.
 4001107                        bool hasContinuation = response.PayloadContinuation is not null;
 1108
 4001109                        FlushResult flushResult = await payloadWriter.CopyFromAsync(
 4001110                            response.Payload,
 4001111                            stream.WritesClosed,
 4001112                            endStream: !hasContinuation,
 4001113                            _disposedCts.Token).ConfigureAwait(false);
 1114
 3961115                        if (!flushResult.IsCompleted && !flushResult.IsCanceled && hasContinuation)
 21116                        {
 21117                            flushResult = await payloadWriter.CopyFromAsync(
 21118                                response.PayloadContinuation!,
 21119                                stream.WritesClosed,
 21120                                endStream: true,
 21121                                _disposedCts.Token).ConfigureAwait(false);
 11122                        }
 1123
 3951124                        payloadWriterSuccess = !flushResult.IsCanceled;
 3951125                    }
 1126                    finally
 4001127                    {
 4001128                        payloadWriter.CompleteOutput(payloadWriterSuccess);
 4001129                        response.Payload.Complete();
 4001130                        response.PayloadContinuation?.Complete();
 4001131                    }
 3951132                }
 14031133            }
 151134            catch (Exception exception) when (_taskExceptionObserver is not null)
 71135            {
 71136                _taskExceptionObserver.DispatchFailed(
 71137                    request,
 71138                    _connectionContext!.TransportConnectionInformation,
 71139                    exception);
 71140                return; // success remains false
 1141            }
 14031142            success = true;
 14031143        }
 11144        catch (IceRpcException)
 11145        {
 1146            // Expected, with for example:
 1147            //  - IceRpcError.ConnectionAborted when the peer aborts the connection
 1148            //  - IceRpcError.IceRpcError when the request header is invalid
 1149            //  - IceRpcError.TruncatedData when the request header is truncated
 11150        }
 111151        catch (OperationCanceledException exception) when (
 111152            exception.CancellationToken == cancellationToken ||
 111153            exception.CancellationToken == _disposedCts.Token)
 111154        {
 1155            // Expected if the dispatch is canceled by the peer or the connection is disposed.
 111156        }
 01157        catch (Exception exception)
 01158        {
 1159            // This exception is unexpected when running the IceRPC test suite. A test that expects this exception must
 1160            // install a task exception observer.
 01161            Debug.Fail($"icerpc dispatch failed with an unexpected exception: {exception}");
 1162
 1163            // Generate unobserved task exception (UTE). If this exception is expected (e.g. an expected payload read
 1164            // exception) and the application wants to avoid this UTE, it must configure a non-null logger to install
 1165            // a task exception observer.
 01166            throw;
 1167        }
 1168        finally
 14331169        {
 14331170            if (!success)
 301171            {
 1172                // We always need to complete streamOutput when an exception is thrown. For example, we received an
 1173                // invalid request header that we could not decode.
 301174                streamOutput?.CompleteOutput(success: false);
 301175                streamInput?.Complete();
 301176            }
 14331177            fieldsPipeReader?.Complete();
 1178
 14331179            DecrementDispatchInvocationCount();
 14331180        }
 1181
 1182        async Task<OutgoingResponse> PerformDispatchRequestAsync(
 1183            IncomingRequest request,
 1184            CancellationToken cancellationToken)
 14181185        {
 14181186            Debug.Assert(_dispatcher is not null);
 1187
 1188            OutgoingResponse response;
 1189
 1190            try
 14181191            {
 14181192                if (_dispatchSemaphore is SemaphoreSlim dispatchSemaphore)
 14181193                {
 14181194                    await dispatchSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
 14181195                }
 1196
 1197                try
 14181198                {
 14181199                    response = await _dispatcher.DispatchAsync(request, cancellationToken).ConfigureAwait(false);
 13961200                }
 1201                finally
 14181202                {
 14181203                    _dispatchSemaphore?.Release();
 14181204                }
 1205
 13961206                if (response != request.Response)
 11207                {
 11208                    throw new InvalidOperationException(
 11209                        "The dispatcher did not return the last response created for this request.");
 1210                }
 13951211            }
 91212            catch (OperationCanceledException exception) when (cancellationToken == exception.CancellationToken)
 81213            {
 81214                throw;
 1215            }
 151216            catch (Exception exception)
 151217            {
 151218                var dispatchException = DispatchException.FromException(exception);
 151219                Debug.Assert(!dispatchException.ConvertToInternalError);
 151220                response = new OutgoingResponse(
 151221                    request,
 151222                    dispatchException.StatusCode,
 151223                    dispatchException.ErrorMessage);
 151224            }
 1225
 14101226            return response;
 14101227        }
 1228
 1229        static (IceRpcRequestHeader, IDictionary<RequestFieldKey, ReadOnlySequence<byte>>, PipeReader?) DecodeHeader(
 1230            ReadOnlySequence<byte> buffer)
 14231231        {
 14231232            var decoder = new SliceDecoder(buffer);
 14231233            var header = new IceRpcRequestHeader(ref decoder);
 1234
 1235            // Ensure that the encoded path is a valid service address path.
 1236            try
 14231237            {
 14231238                ServiceAddress.CheckPath(header.Path);
 14201239            }
 31240            catch (FormatException exception)
 31241            {
 31242                throw new InvalidDataException(exception.Message, exception);
 1243            }
 1244
 14201245            (IDictionary<RequestFieldKey, ReadOnlySequence<byte>> fields, PipeReader? pipeReader) =
 14201246                DecodeFieldDictionary(
 14201247                    ref decoder,
 14361248                    (ref SliceDecoder decoder) => decoder.DecodeRequestFieldKey());
 1249
 14181250            return (header, fields, pipeReader);
 14181251        }
 1252
 1253        void EncodeHeader(OutgoingResponse response)
 4021254        {
 4021255            var encoder = new SliceEncoder(streamOutput);
 1256
 1257            // Write the IceRpc response header.
 4021258            Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(_headerSizeLength);
 1259
 1260            // We use UnflushedBytes because EncodeFieldDictionary can write directly to streamOutput.
 4021261            long headerStartPos = streamOutput.UnflushedBytes;
 1262
 4021263            encoder.EncodeStatusCode(response.StatusCode);
 4021264            if (response.StatusCode > StatusCode.Ok)
 341265            {
 341266                encoder.EncodeString(response.ErrorMessage!);
 341267            }
 1268
 4021269            EncodeFieldDictionary(
 4021270                response.Fields,
 51271                (ref SliceEncoder encoder, ResponseFieldKey key) => encoder.EncodeResponseFieldKey(key),
 4021272                ref encoder,
 4021273                streamOutput);
 1274
 1275            // We're done with the header encoding, write the header size.
 4011276            int headerSize = (int)(streamOutput.UnflushedBytes - headerStartPos);
 4011277            CheckPeerHeaderSize(headerSize);
 4001278            SliceEncoder.EncodeVarUInt62((uint)headerSize, sizePlaceholder);
 4001279        }
 14331280    }
 1281
 1282    /// <summary>Encodes the fields dictionary at the end of a request or response header.</summary>
 1283    /// <remarks>This method can write bytes directly to <paramref name="output"/> without going through
 1284    /// <paramref name="encoder"/>.</remarks>
 1285    private void EncodeFieldDictionary<TKey>(
 1286        IDictionary<TKey, OutgoingFieldValue> fields,
 1287        EncodeAction<TKey> encodeKeyAction,
 1288        ref SliceEncoder encoder,
 1289        PipeWriter output) where TKey : struct =>
 18341290        encoder.EncodeDictionary(
 18341291            fields,
 18341292            encodeKeyAction,
 18341293            (ref SliceEncoder encoder, OutgoingFieldValue value) =>
 191294                {
 191295                    if (value.WriteAction is Action<IBufferWriter<byte>> writeAction)
 91296                    {
 91297                        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(_headerSizeLength);
 91298                        long startPos = output.UnflushedBytes;
 91299                        writeAction(output);
 81300                        SliceEncoder.EncodeVarUInt62((ulong)(output.UnflushedBytes - startPos), sizePlaceholder);
 81301                    }
 18341302                    else
 101303                    {
 101304                        encoder.EncodeSize(checked((int)value.ByteSequence.Length));
 101305                        encoder.WriteByteSequence(value.ByteSequence);
 101306                    }
 18521307                });
 1308
 1309    /// <summary>Increments the dispatch-invocation count.</summary>
 1310    /// <remarks>This method must be called with _mutex locked.</remarks>
 1311    private void IncrementDispatchInvocationCount()
 28861312    {
 28861313        if (_dispatchInvocationCount++ == 0 && _streamInputOutputCount == 0)
 8031314        {
 1315            // Cancel inactivity check.
 8031316            _inactivityTimeoutTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
 8031317        }
 28861318    }
 1319
 1320    /// <summary>Increments the stream input/output count.</summary>
 1321    /// <remarks>This method must be called with _mutex locked.</remarks>
 1322    private void IncrementStreamInputOutputCount(bool bidirectional)
 28651323    {
 28651324        Debug.Assert(_dispatchInvocationCount > 0);
 28651325        _streamInputOutputCount += bidirectional ? 2 : 1;
 28651326    }
 1327
 1328    private async Task ReadGoAwayAsync(CancellationToken cancellationToken)
 3441329    {
 3441330        await Task.Yield(); // exit mutex lock
 1331
 1332        // Wait for _connectTask (which spawned the task running this method) to complete. This await can't fail.
 1333        // This guarantees this method won't request a shutdown until after _connectTask completed successfully.
 3441334        await _connectTask!.ConfigureAwait(false);
 1335
 3441336        PipeReader remoteInput = _remoteControlStream!.Input!;
 1337
 1338        try
 3441339        {
 1340            // Wait to receive the GoAway frame.
 3441341            await ReceiveControlFrameHeaderAsync(IceRpcControlFrameType.GoAway, cancellationToken)
 3441342                .ConfigureAwait(false);
 1343
 901344            ReadResult readResult = await remoteInput.ReadSliceSegmentAsync(
 901345                MaxGoAwayFrameBodySize,
 901346                cancellationToken).ConfigureAwait(false);
 1347
 1348            // We don't call CancelPendingRead on remoteInput
 861349            Debug.Assert(!readResult.IsCanceled);
 1350
 1351            try
 861352            {
 861353                _goAwayFrame =
 1721354                    readResult.Buffer.DecodeSliceBuffer((ref SliceDecoder decoder) => new IceRpcGoAway(ref decoder));
 841355            }
 1356            finally
 861357            {
 861358                remoteInput.AdvanceTo(readResult.Buffer.End);
 861359            }
 1360
 841361            RefuseNewInvocations("The connection was shut down because it received a GoAway frame from the peer.");
 841362            _goAwayCts.Cancel();
 841363            _ = _shutdownRequestedTcs.TrySetResult();
 841364        }
 1341365        catch (OperationCanceledException)
 1341366        {
 1367            // The connection is disposed and we let this exception cancel the task.
 1341368            throw;
 1369        }
 1161370        catch (IceRpcException)
 1161371        {
 1161372            RefuseNewInvocations("The connection was lost");
 1161373            _ = _shutdownRequestedTcs.TrySetResult();
 1374
 1375            // We let the task complete with this expected exception.
 1161376            throw;
 1377        }
 101378        catch (InvalidDataException exception)
 101379        {
 101380            RefuseNewInvocations("The connection was lost");
 101381            _ = _shutdownRequestedTcs.TrySetResult();
 1382
 1383            // "expected" in the sense it should not trigger a Debug.Fail.
 101384            throw new IceRpcException(
 101385                IceRpcError.IceRpcError,
 101386                "The ReadGoAway task was aborted by an icerpc protocol error.",
 101387                exception);
 1388        }
 01389        catch (Exception exception)
 01390        {
 01391            Debug.Fail($"The read go away task failed with an unexpected exception: {exception}");
 01392            RefuseNewInvocations("The connection was lost");
 01393            _ = _shutdownRequestedTcs.TrySetResult();
 01394            throw;
 1395        }
 841396    }
 1397
 1398    private async ValueTask ReceiveControlFrameHeaderAsync(
 1399        IceRpcControlFrameType expectedFrameType,
 1400        CancellationToken cancellationToken)
 6991401    {
 6991402        ReadResult readResult = await _remoteControlStream!.Input.ReadAsync(cancellationToken).ConfigureAwait(false);
 1403
 1404        // We don't call CancelPendingRead on _remoteControlStream.Input.
 4451405        Debug.Assert(!readResult.IsCanceled);
 1406
 4451407        if (readResult.Buffer.IsEmpty)
 21408        {
 21409            throw new InvalidDataException(
 21410                "Failed to read the frame type because no more data is available from the control stream.");
 1411        }
 1412
 4431413        var frameType = (IceRpcControlFrameType)readResult.Buffer.FirstSpan[0];
 4431414        if (frameType != expectedFrameType)
 41415        {
 41416            throw new InvalidDataException($"Received frame type {frameType} but expected {expectedFrameType}.");
 1417        }
 4391418        _remoteControlStream!.Input.AdvanceTo(readResult.Buffer.GetPosition(1));
 4391419    }
 1420
 1421    private async ValueTask ReceiveSettingsFrameBody(CancellationToken cancellationToken)
 3491422    {
 1423        // We are still in the single-threaded initialization at this point.
 1424
 3491425        PipeReader input = _remoteControlStream!.Input;
 3491426        ReadResult readResult = await input.ReadSliceSegmentAsync(
 3491427            MaxSettingsFrameBodySize,
 3491428            cancellationToken).ConfigureAwait(false);
 1429
 1430        // We don't call CancelPendingRead on _remoteControlStream.Input
 3481431        Debug.Assert(!readResult.IsCanceled);
 1432
 1433        try
 3481434        {
 3481435            IceRpcSettings settings =
 6961436                readResult.Buffer.DecodeSliceBuffer((ref SliceDecoder decoder) => new IceRpcSettings(ref decoder));
 1437
 3451438            if (settings.Value.TryGetValue(IceRpcSettingKey.MaxHeaderSize, out ulong value))
 31439            {
 1440                // a varuint62 always fits in a long
 1441                try
 31442                {
 31443                    _maxPeerHeaderSize = ConnectionOptions.IceRpcCheckMaxHeaderSize((long)value);
 21444                }
 11445                catch (ArgumentOutOfRangeException exception)
 11446                {
 11447                    throw new InvalidDataException("Received invalid maximum header size setting.", exception);
 1448                }
 21449                _headerSizeLength = SliceEncoder.GetVarUInt62EncodedSize(value);
 21450            }
 1451            // all other settings are unknown and ignored
 3441452        }
 1453        finally
 3481454        {
 3481455            input.AdvanceTo(readResult.Buffer.End);
 3481456        }
 3441457    }
 1458
 1459    private void RefuseNewInvocations(string message)
 8481460    {
 1461        lock (_mutex)
 8481462        {
 8481463            _refuseInvocations = true;
 8481464            _invocationRefusedMessage ??= message;
 8481465        }
 8481466    }
 1467
 1468    // The inactivity check executes once in _inactivityTimeout. By then either:
 1469    // - the connection is no longer inactive (and the inactivity check is canceled or being canceled)
 1470    // - the connection is still inactive and we request shutdown
 1471    private void ScheduleInactivityCheck() =>
 11071472        _inactivityTimeoutTimer.Change(_inactivityTimeout, Timeout.InfiniteTimeSpan);
 1473
 1474    private ValueTask<FlushResult> SendControlFrameAsync(
 1475        IceRpcControlFrameType frameType,
 1476        EncodeAction encodeAction,
 1477        CancellationToken cancellationToken)
 4611478    {
 4611479        PipeWriter output = _controlStream!.Output;
 1480
 4611481        EncodeFrame(output);
 1482
 4611483        return output.FlushAsync(cancellationToken); // Flush
 1484
 1485        void EncodeFrame(IBufferWriter<byte> buffer)
 4611486        {
 4611487            var encoder = new SliceEncoder(buffer);
 4611488            encoder.EncodeIceRpcControlFrameType(frameType);
 4611489            Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(_headerSizeLength);
 4611490            int startPos = encoder.EncodedByteCount; // does not include the size
 4611491            encodeAction.Invoke(ref encoder);
 4611492            int frameSize = encoder.EncodedByteCount - startPos;
 4611493            SliceEncoder.EncodeVarUInt62((uint)frameSize, sizePlaceholder);
 4611494        }
 4611495    }
 1496
 1497    /// <summary>Sends the payload continuation of an outgoing request in the background.</summary>
 1498    /// <remarks>We send the payload continuation on a separate thread with Task.Run: this ensures that the synchronous
 1499    /// activity that could result from reading or writing the payload continuation doesn't delay in any way the
 1500    /// caller. </remarks>
 1501    /// <param name="request">The outgoing request.</param>
 1502    /// <param name="payloadWriter">The payload writer.</param>
 1503    /// <param name="writesClosed">A task that completes when we can no longer write to payloadWriter.</param>
 1504    /// <param name="onGoAway">An action to execute with a CTS when we receive the GoAway frame from the peer.</param>
 1505    /// <param name="cancellationToken">The cancellation token of the invocation; the associated CTS is disposed when
 1506    /// the invocation completes.</param>
 1507    private void SendRequestPayloadContinuation(
 1508        OutgoingRequest request,
 1509        PipeWriter payloadWriter,
 1510        Task writesClosed,
 1511        Action<object?> onGoAway,
 1512        CancellationToken cancellationToken)
 121513    {
 121514        Debug.Assert(request.PayloadContinuation is not null);
 1515
 1516        // First "detach" the continuation.
 121517        PipeReader payloadContinuation = request.PayloadContinuation;
 121518        request.PayloadContinuation = null;
 1519
 1520        lock (_mutex)
 121521        {
 121522            Debug.Assert(_dispatchInvocationCount > 0); // as a result, can't be disposed.
 1523
 1524            // Give the task its own dispatch-invocation count. This ensures the transport connection won't be disposed
 1525            // while the continuation is being sent.
 121526            IncrementDispatchInvocationCount();
 121527        }
 1528
 1529        // This background task owns payloadContinuation, payloadWriter and 1 dispatch-invocation count, and must clean
 1530        // them up. Hence CancellationToken.None.
 121531        _ = Task.Run(PerformSendRequestPayloadContinuationAsync, CancellationToken.None);
 1532
 1533        async Task PerformSendRequestPayloadContinuationAsync()
 121534        {
 121535            bool success = false;
 1536
 1537            try
 121538            {
 1539                // Since _dispatchInvocationCount > 0, _disposedCts is not disposed.
 121540                using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedCts.Token);
 1541
 1542                // This token registration is needed for one-way requests and is redundant for two-way requests.
 1543                // We want GoAway to cancel the sending of one-way requests that have not been received by the peer,
 1544                // especially when these requests have payload continuations.
 121545                using CancellationTokenRegistration tokenRegistration = _goAwayCts.Token.UnsafeRegister(onGoAway, cts);
 1546
 1547                try
 121548                {
 1549                    // The cancellation of the InvokeAsync's cancellationToken cancels cts only until InvokeAsync's
 1550                    // PerformInvokeAsync completes. Afterwards, the cancellation of InvokeAsync's cancellationToken has
 1551                    // no effect on cts, so it doesn't cancel the copying of payloadContinuation.
 121552                    FlushResult flushResult = await payloadWriter.CopyFromAsync(
 121553                        payloadContinuation,
 121554                        writesClosed,
 121555                        endStream: true,
 121556                        cts.Token).ConfigureAwait(false);
 1557
 51558                    success = !flushResult.IsCanceled;
 51559                }
 31560                catch (OperationCanceledException exception) when (exception.CancellationToken == cts.Token)
 21561                {
 1562                    // Process/translate this exception primarily for the benefit of _taskExceptionObserver.
 1563
 1564                    // Can be because cancellationToken was canceled by DisposeAsync or GoAway; that's fine.
 21565                    cancellationToken.ThrowIfCancellationRequested();
 1566
 11567                    if (_disposedCts.IsCancellationRequested)
 01568                    {
 1569                        // DisposeAsync aborted the request.
 01570                        throw new IceRpcException(IceRpcError.OperationAborted);
 1571                    }
 1572                    else
 11573                    {
 1574                        // When _goAwayCts is canceled and onGoAway cancels its argument:
 1575                        // - if PerformInvokeAsync is no longer running (typical for a one-way request), we get here
 1576                        // - if PerformInvokeAsync is still running, we may get here or cancellationToken gets canceled
 1577                        // first.
 11578                        Debug.Assert(_goAwayCts.IsCancellationRequested);
 11579                        throw new IceRpcException(IceRpcError.InvocationCanceled, "The connection is shutting down.");
 1580                    }
 1581                }
 51582            }
 71583            catch (Exception exception) when (_taskExceptionObserver is not null)
 51584            {
 51585                _taskExceptionObserver.RequestPayloadContinuationFailed(
 51586                    request,
 51587                    _connectionContext!.TransportConnectionInformation,
 51588                    exception);
 51589            }
 11590            catch (OperationCanceledException exception) when (exception.CancellationToken == cancellationToken)
 11591            {
 1592                // Expected.
 11593            }
 11594            catch (IceRpcException)
 11595            {
 1596                // Expected, with for example IceRpcError.ConnectionAborted when the peer aborts the connection.
 11597            }
 01598            catch (Exception exception)
 01599            {
 1600                // This exception is unexpected when running the IceRPC test suite. A test that expects such an
 1601                // exception must install a task exception observer.
 01602                Debug.Fail($"Failed to send payload continuation of request {request}: {exception}");
 1603
 1604                // If Debug is not enabled and there is no task exception observer, we rethrow to generate an
 1605                // Unobserved Task Exception.
 01606                throw;
 1607            }
 1608            finally
 121609            {
 121610                payloadWriter.CompleteOutput(success);
 121611                payloadContinuation.Complete();
 121612                DecrementDispatchInvocationCount();
 121613            }
 121614        }
 121615    }
 1616}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
GitHub | reportgenerator.io

Methods/Properties

get_IsServer()
.ctor(IceRpc.Transports.IMultiplexedConnection,IceRpc.Transports.TransportConnectionInformation,IceRpc.ConnectionOptions,IceRpc.Internal.ITaskExceptionObserver)
ConnectAsync(System.Threading.CancellationToken)
PerformConnectAsync()
DisposeAsync()
PerformDisposeAsync()
InvokeAsync(IceRpc.OutgoingRequest,System.Threading.CancellationToken)
PerformInvokeAsync()
OnGoAway()
DecodeHeader()
EncodeHeader()
ShutdownAsync(System.Threading.CancellationToken)
PerformShutdownAsync()
DecodeFieldDictionary(ZeroC.Slice.Codec.SliceDecoder&,ZeroC.Slice.Codec.DecodeFunc`1<TKey>)
AcceptRequestsAsync()
CheckPeerHeaderSize(System.Int32)
DecrementDispatchInvocationCount()
DecrementStreamInputOutputCount()
DispatchRequestAsync()
PerformDispatchRequestAsync()
DecodeHeader()
EncodeHeader()
EncodeFieldDictionary(System.Collections.Generic.IDictionary`2<TKey,IceRpc.OutgoingFieldValue>,ZeroC.Slice.Codec.EncodeAction`1<TKey>,ZeroC.Slice.Codec.SliceEncoder&,System.IO.Pipelines.PipeWriter)
IncrementDispatchInvocationCount()
IncrementStreamInputOutputCount(System.Boolean)
ReadGoAwayAsync()
ReceiveControlFrameHeaderAsync()
ReceiveSettingsFrameBody()
RefuseNewInvocations(System.String)
ScheduleInactivityCheck()
SendControlFrameAsync(IceRpc.Internal.IceRpcControlFrameType,ZeroC.Slice.Codec.EncodeAction,System.Threading.CancellationToken)
EncodeFrame()
SendRequestPayloadContinuation(IceRpc.OutgoingRequest,System.IO.Pipelines.PipeWriter,System.Threading.Tasks.Task,System.Action`1<System.Object>,System.Threading.CancellationToken)
PerformSendRequestPayloadContinuationAsync()