< Summary

Information
Class: IceRpc.Transports.Slic.Internal.SlicConnection
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Transports/Slic/Internal/SlicConnection.cs
Tag: 1986_28452893481
Line coverage
90%
Covered lines: 1007
Uncovered lines: 111
Coverable lines: 1118
Total lines: 1789
Line coverage: 90%
Branch coverage
90%
Covered branches: 315
Total branches: 348
Branch coverage: 90.5%
Method coverage
97%
Covered methods: 48
Fully covered methods: 25
Total methods: 49
Method coverage: 97.9%
Full method coverage: 51%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_IsServer()100%11100%
get_MinSegmentSize()100%11100%
get_PeerInitialStreamWindowSize()100%11100%
get_PeerMaxStreamFrameSize()100%11100%
get_Pool()100%11100%
get_InitialStreamWindowSize()100%11100%
get_PauseWriterThreshold()100%11100%
get_StreamWindowUpdateThreshold()100%11100%
.ctor(...)100%44100%
AcceptStreamAsync()100%66100%
ConnectAsync(...)75%4484.61%
PerformConnectAsync()100%262696.19%
DecodeInitialize()75%4489.47%
DecodeInitializeAckOrVersion()66.66%6690%
ReadFrameAsync()87.5%8885%
CloseAsync()92.85%141495.45%
CreateStreamAsync()100%141497.61%
DisposeAsync()100%22100%
PerformDisposeAsync()93.75%161690.69%
SendPingAsync()100%1144.44%
SendReadPing()50%22100%
SendWritePing()0%620%
FillBufferWriterAsync(...)100%11100%
ReleaseStream(...)100%1010100%
ThrowIfClosed()100%22100%
WriteConnectionFrameAsync()100%22100%
WriteStreamFrame(...)50%22100%
WriteStreamFrameAsync()83.33%6678.57%
WriteStreamDataFrameAsync()94.44%363693.18%
EncodeStreamFrameHeader()100%22100%
AddStream(...)100%66100%
DecodeParameters(...)76.92%362675.4%
DecodeParamValue()100%1166.66%
EncodeParameters()83.33%66100%
EncodeParameter()100%11100%
IsUnknownStream(...)100%1212100%
ReadFrameAsync(...)95.65%232394.87%
ReadCloseFrameAsync()100%1313100%
ReadPingFrameAndWritePongFrameAsync()100%22100%
WritePongFrameAsync()100%1180.95%
ReadPongFrameAsync()66.66%6686.66%
ReadStreamWindowUpdateFrameAsync()100%22100%
ReadFrameBodyAsync()100%44100%
ReadFrameHeaderAsync()83.33%6681.81%
TryDecodeHeader()81.25%181680.55%
ReadFramesAsync()83.33%6685.71%
ReadStreamDataFrameAsync()90.47%524282.29%
TryClose(...)100%66100%
WriteFrame(...)100%44100%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Transports/Slic/Internal/SlicConnection.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Internal;
 4using IceRpc.Transports.Internal;
 5using System.Buffers;
 6using System.Collections.Concurrent;
 7using System.Diagnostics;
 8using System.IO.Pipelines;
 9using System.Security.Authentication;
 10using System.Threading.Channels;
 11using ZeroC.Slice.Codec;
 12
 13namespace IceRpc.Transports.Slic.Internal;
 14
 15/// <summary>The Slic connection implements an <see cref="IMultiplexedConnection" /> on top of a <see
 16/// cref="IDuplexConnection" />.</summary>
 17internal class SlicConnection : IMultiplexedConnection
 18{
 19    /// <summary>Gets a value indicating whether or not this is the server-side of the connection.</summary>
 1732320    internal bool IsServer { get; }
 21
 22    /// <summary>Gets the minimum size of the segment requested from <see cref="Pool" />.</summary>
 570823    internal int MinSegmentSize { get; }
 24
 25    /// <summary>Gets the peer's initial stream window size. This property is set to the <see
 26    /// cref="ParameterKey.InitialStreamWindowSize"/> value carried by the <see cref="FrameType.Initialize" />
 27    /// frame.</summary>
 350428    internal int PeerInitialStreamWindowSize { get; private set; }
 29
 30    /// <summary>Gets the maximum size of stream frames accepted by the peer. This property is set to the <see
 31    /// cref="ParameterKey.MaxStreamFrameSize"/> value carried by the <see cref="FrameType.Initialize" />
 32    /// frame.</summary>
 990033    internal int PeerMaxStreamFrameSize { get; private set; }
 34
 35    /// <summary>Gets the <see cref="MemoryPool{T}" /> used for obtaining memory buffers.</summary>
 570836    internal MemoryPool<byte> Pool { get; }
 37
 38    /// <summary>Gets the initial stream window size.</summary>
 1096539    internal int InitialStreamWindowSize { get; }
 40
 41    /// <summary>Gets the pause writer threshold for the connection's outbound pipe.</summary>
 76942    internal int PauseWriterThreshold { get; }
 43
 44    /// <summary>Gets the window update threshold. When the window size is increased and this threshold reached, a <see
 45    /// cref="FrameType.StreamWindowUpdate" /> frame is sent.</summary>
 744746    internal int StreamWindowUpdateThreshold => InitialStreamWindowSize / StreamWindowUpdateRatio;
 47
 48    // The maximum body size for non-stream frames (Initialize, InitializeAck, Version, Close, Ping, Pong). This
 49    // value is the maximum value that can be encoded as a 2-byte varuint62, which allows WriteFrame to use a 2-byte
 50    // size placeholder. Stream data frames are not subject to this limit; they are gated by per-stream flow control.
 51    private const int MaxControlFrameBodySize = 16_383;
 52
 53    // The ratio used to compute the StreamWindowUpdateThreshold. For now, the stream window update is sent when the
 54    // window size grows over InitialStreamWindowSize / StreamWindowUpdateRatio.
 55    private const int StreamWindowUpdateRatio = 2;
 56
 57    private readonly Channel<IMultiplexedStream> _acceptStreamChannel;
 58    private int _bidirectionalStreamCount;
 59    private SemaphoreSlim? _bidirectionalStreamSemaphore;
 60    private readonly CancellationToken _closedCancellationToken;
 76961    private readonly CancellationTokenSource _closedCts = new();
 62    private string? _closedMessage;
 63    private Task<TransportConnectionInformation>? _connectTask;
 76964    private readonly CancellationTokenSource _disposedCts = new();
 65    private Task? _disposeTask;
 66    private readonly SlicDuplexConnectionDecorator _duplexConnection;
 67    private readonly DuplexConnectionReader _duplexConnectionReader;
 68    private readonly SlicDuplexConnectionWriter _duplexConnectionWriter;
 69
 70    // Invariant: _isClosed only ever transitions false -> true (under _mutex, by TryClose). Every writer site
 71    // (WriteConnectionFrameAsync, WriteStreamFrame, WriteStreamDataFrameAsync, CloseAsync) re-checks _isClosed under
 72    // _mutex *after* acquiring _writeSemaphore, so it bails out before issuing any new Write/WriteFrame on
 73    // _duplexConnectionWriter once _isClosed has been observed true.
 74    private bool _isClosed;
 75    private ulong? _lastRemoteBidirectionalStreamId;
 76    private ulong? _lastRemoteUnidirectionalStreamId;
 77    private readonly TimeSpan _localIdleTimeout;
 78    private readonly int _maxBidirectionalStreams;
 79    private readonly int _maxOutstandingPongs;
 80    private readonly int _maxStreamFrameSize;
 81    private readonly int _maxUnidirectionalStreams;
 82    // _mutex ensure the assignment of _lastRemoteXxx members and the addition of the stream to _streams is
 83    // an atomic operation.
 76984    private readonly Lock _mutex = new();
 85    private ulong _nextBidirectionalId;
 86    private ulong _nextUnidirectionalId;
 87
 88    // The number of Pong frames queued for sending (in response to Ping frames) but not yet written to the duplex
 89    // connection. The connection is aborted when a Ping frame is received while this count has reached
 90    // _maxOutstandingPongs.
 91    private int _outstandingPongCount;
 92    private IceRpcError? _peerCloseError;
 76993    private TimeSpan _peerIdleTimeout = Timeout.InfiniteTimeSpan;
 94
 95    // The number of Ping frames sent to the peer that have not been answered yet by a Pong frame.
 96    private int _pendingPongCount;
 97    private Task? _readFramesTask;
 98
 76999    private readonly ConcurrentDictionary<ulong, SlicStream> _streams = new();
 100    private int _streamSemaphoreWaitCount;
 769101    private readonly TaskCompletionSource _streamSemaphoreWaitClosed =
 769102        new(TaskCreationOptions.RunContinuationsAsynchronously);
 103
 104    private int _unidirectionalStreamCount;
 105    private SemaphoreSlim? _unidirectionalStreamSemaphore;
 106
 107    // Serializes writes to _duplexConnectionWriter so that frame bytes are appended to the outbound pipe in order and
 108    // the pipe's pauseWriterThreshold is observed strictly. This async lock is held across the FlushAsync call, so a
 109    // single parked flush blocks all other connection writers until the background writer task drains enough data.
 110    // Not disposed: background fire-and-forget writes (e.g. StreamWindowUpdate from sync code paths) may attempt to
 111    // acquire it after DisposeAsync, and we don't want to have to handle ObjectDisposedException at every call site.
 112    // Skipping Dispose is harmless here because we never access SemaphoreSlim.AvailableWaitHandle, so no unmanaged
 113    // wait handle is ever allocated.
 114#pragma warning disable CA2213
 769115    private readonly SemaphoreSlim _writeSemaphore = new(1, 1);
 116#pragma warning restore CA2213
 117
 118    // This is only set for server connections to ensure that _duplexConnectionWriter.Write is not called after
 119    // _duplexConnectionWriter.Shutdown. This can occur if the client-side of the connection sends the close frame
 120    // followed by the shutdown of the duplex connection and if CloseAsync is called at the same time on the server
 121    // connection. Guarded by _writeSemaphore.
 122    private bool _writerIsShutdown;
 123
 124    public async ValueTask<IMultiplexedStream> AcceptStreamAsync(CancellationToken cancellationToken)
 2463125    {
 126        lock (_mutex)
 2463127        {
 2463128            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 129
 2462130            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 1131            {
 1132                throw new InvalidOperationException("Cannot accept stream before connecting the Slic connection.");
 133            }
 2461134            if (_isClosed)
 13135            {
 13136                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 137            }
 2448138        }
 139
 140        try
 2448141        {
 2448142            return await _acceptStreamChannel.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 143        }
 118144        catch (ChannelClosedException exception)
 118145        {
 118146            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 117147            Debug.Assert(exception.InnerException is not null);
 148            // The exception given to ChannelWriter.Complete(Exception? exception) is the InnerException.
 117149            throw ExceptionUtil.Throw(exception.InnerException);
 150        }
 2082151    }
 152
 153    public Task<TransportConnectionInformation> ConnectAsync(CancellationToken cancellationToken)
 750154    {
 155        lock (_mutex)
 750156        {
 750157            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 158
 750159            if (_connectTask is not null)
 1160            {
 1161                throw new InvalidOperationException("Cannot connect twice a Slic connection.");
 162            }
 749163            if (_isClosed)
 0164            {
 0165                throw new InvalidOperationException("Cannot connect a closed Slic connection.");
 166            }
 749167            _connectTask = PerformConnectAsync();
 749168        }
 749169        return _connectTask;
 170
 171        async Task<TransportConnectionInformation> PerformConnectAsync()
 749172        {
 749173            await Task.Yield(); // Exit mutex lock
 174
 175            // Connect the duplex connection.
 176            TransportConnectionInformation transportConnectionInformation;
 749177            TimeSpan peerIdleTimeout = TimeSpan.MaxValue;
 178
 179            try
 749180            {
 749181                transportConnectionInformation = await _duplexConnection.ConnectAsync(cancellationToken)
 749182                    .ConfigureAwait(false);
 183
 184                // Initialize the Slic connection.
 727185                if (IsServer)
 370186                {
 187                    // Read the Initialize frame.
 370188                    (ulong version, InitializeBody? initializeBody) = await ReadFrameAsync(
 370189                        DecodeInitialize,
 370190                        cancellationToken).ConfigureAwait(false);
 191
 363192                    if (initializeBody is null)
 2193                    {
 194                        // Unsupported version, try to negotiate another version by sending a Version frame with the
 195                        // Slic versions supported by this server.
 2196                        ulong[] supportedVersions = new ulong[] { SlicDefinitions.V1 };
 197
 2198                        await WriteConnectionFrameAsync(
 2199                            FrameType.Version,
 2200                            new VersionBody(supportedVersions).Encode,
 2201                            cancellationToken).ConfigureAwait(false);
 202
 2203                        (version, initializeBody) = await ReadFrameAsync(
 2204                            (frameType, buffer) =>
 2205                            {
 2206                                if (frameType is null)
 1207                                {
 2208                                    // The client shut down the connection because it doesn't support any of the
 2209                                    // server's supported Slic versions.
 1210                                    throw new IceRpcException(
 1211                                        IceRpcError.ConnectionRefused,
 1212                                        $"The connection was refused because the client Slic version {version} is not su
 2213                                }
 2214                                else
 1215                                {
 1216                                    return DecodeInitialize(frameType, buffer);
 2217                                }
 1218                            },
 2219                            cancellationToken).ConfigureAwait(false);
 1220                    }
 221
 362222                    Debug.Assert(initializeBody is not null);
 223
 362224                    DecodeParameters(initializeBody.Value.Parameters);
 225
 226                    // Write back an InitializeAck frame.
 361227                    await WriteConnectionFrameAsync(
 361228                        FrameType.InitializeAck,
 361229                        new InitializeAckBody(EncodeParameters()).Encode,
 361230                        cancellationToken).ConfigureAwait(false);
 361231                }
 232                else
 357233                {
 234                    // Write the Initialize frame.
 357235                    await WriteConnectionFrameAsync(
 357236                        FrameType.Initialize,
 357237                        (ref SliceEncoder encoder) =>
 357238                        {
 357239                            encoder.EncodeVarUInt62(SlicDefinitions.V1);
 357240                            new InitializeBody(EncodeParameters()).Encode(ref encoder);
 357241                        },
 357242                        cancellationToken).ConfigureAwait(false);
 243
 244                    // Read and decode the InitializeAck or Version frame.
 357245                    (InitializeAckBody? initializeAckBody, VersionBody? versionBody) = await ReadFrameAsync(
 357246                        DecodeInitializeAckOrVersion,
 357247                        cancellationToken).ConfigureAwait(false);
 248
 333249                    Debug.Assert(initializeAckBody is not null || versionBody is not null);
 250
 333251                    if (initializeAckBody is not null)
 331252                    {
 331253                        DecodeParameters(initializeAckBody.Value.Parameters);
 331254                    }
 255
 333256                    if (versionBody is not null)
 2257                    {
 2258                        if (versionBody.Value.Versions.Contains(SlicDefinitions.V1))
 1259                        {
 1260                            throw new InvalidDataException(
 1261                                "The server supported versions include the version initially requested.");
 262                        }
 263                        else
 1264                        {
 265                            // We only support V1 and the peer rejected V1.
 1266                            throw new IceRpcException(
 1267                                IceRpcError.ConnectionRefused,
 1268                                $"The connection was refused because the server only supports Slic version(s) {string.Jo
 269                        }
 270                    }
 331271                }
 692272            }
 8273            catch (InvalidDataException exception)
 8274            {
 8275                throw new IceRpcException(
 8276                    IceRpcError.IceRpcError,
 8277                    "The connection was aborted by a Slic protocol error.",
 8278                    exception);
 279            }
 25280            catch (OperationCanceledException)
 25281            {
 25282                throw;
 283            }
 4284            catch (AuthenticationException)
 4285            {
 4286                throw;
 287            }
 20288            catch (IceRpcException)
 20289            {
 20290                throw;
 291            }
 0292            catch (Exception exception)
 0293            {
 0294                Debug.Fail($"ConnectAsync failed with an unexpected exception: {exception}");
 0295                throw;
 296            }
 297
 298            // Enable the idle timeout checks after the connection establishment. The Ping frames sent by the keep alive
 299            // check are not expected until the Slic connection initialization completes. The idle timeout check uses
 300            // the smallest idle timeout. Timeout.InfiniteTimeSpan is -1 ms so we can't compare it directly with
 301            // positive timeouts.
 302            TimeSpan idleTimeout;
 692303            if (_localIdleTimeout == Timeout.InfiniteTimeSpan)
 2304            {
 2305                idleTimeout = _peerIdleTimeout;
 2306            }
 690307            else if (_peerIdleTimeout == Timeout.InfiniteTimeSpan)
 27308            {
 27309                idleTimeout = _localIdleTimeout;
 27310            }
 311            else
 663312            {
 663313                idleTimeout = _peerIdleTimeout < _localIdleTimeout ? _peerIdleTimeout : _localIdleTimeout;
 663314            }
 315
 692316            if (idleTimeout != Timeout.InfiniteTimeSpan)
 690317            {
 690318                _duplexConnection.Enable(idleTimeout);
 690319            }
 320
 692321            _readFramesTask = ReadFramesAsync(_disposedCts.Token);
 322
 692323            return transportConnectionInformation;
 692324        }
 325
 326        static (ulong, InitializeBody?) DecodeInitialize(FrameType? frameType, ReadOnlySequence<byte> buffer)
 365327        {
 365328            if (frameType != FrameType.Initialize)
 0329            {
 0330                throw new InvalidDataException($"Received unexpected {frameType} frame.");
 331            }
 332
 365333            return buffer.DecodeSliceBuffer<(ulong, InitializeBody?)>(
 365334                (ref SliceDecoder decoder) =>
 365335                {
 365336                    ulong version = decoder.DecodeVarUInt62();
 364337                    if (version == SlicDefinitions.V1)
 362338                    {
 362339                        return (version, new InitializeBody(ref decoder));
 365340                    }
 365341                    else
 2342                    {
 2343                        decoder.Skip((int)(buffer.Length - decoder.Consumed));
 2344                        return (version, null);
 365345                    }
 729346                });
 364347        }
 348
 349        static (InitializeAckBody?, VersionBody?) DecodeInitializeAckOrVersion(
 350            FrameType? frameType,
 351            ReadOnlySequence<byte> buffer) =>
 335352            frameType switch
 335353            {
 332354                FrameType.InitializeAck => (
 332355                    buffer.DecodeSliceBuffer((ref SliceDecoder decoder) => new InitializeAckBody(ref decoder)),
 332356                    null),
 3357                FrameType.Version => (
 3358                    null,
 6359                    buffer.DecodeSliceBuffer((ref SliceDecoder decoder) => new VersionBody(ref decoder))),
 0360                _ => throw new InvalidDataException($"Received unexpected Slic frame: '{frameType}'."),
 335361            };
 362
 363        async ValueTask<T> ReadFrameAsync<T>(
 364            Func<FrameType?, ReadOnlySequence<byte>, T> decodeFunc,
 365            CancellationToken cancellationToken)
 729366        {
 729367            (FrameType FrameType, int FrameSize, ulong?)? header =
 729368                await ReadFrameHeaderAsync(cancellationToken).ConfigureAwait(false);
 369
 370            ReadOnlySequence<byte> buffer;
 701371            if (header is null || header.Value.FrameSize == 0)
 4372            {
 4373                buffer = ReadOnlySequence<byte>.Empty;
 4374            }
 375            else
 697376            {
 697377                buffer = await _duplexConnectionReader.ReadAtLeastAsync(
 697378                    header.Value.FrameSize,
 697379                    cancellationToken).ConfigureAwait(false);
 697380                if (buffer.Length > header.Value.FrameSize)
 0381                {
 0382                    buffer = buffer.Slice(0, header.Value.FrameSize);
 0383                }
 697384            }
 385
 701386            T decodedFrame = decodeFunc(header?.FrameType, buffer);
 697387            _duplexConnectionReader.AdvanceTo(buffer.End);
 697388            return decodedFrame;
 697389        }
 749390    }
 391
 392    public async Task CloseAsync(MultiplexedConnectionCloseError closeError, CancellationToken cancellationToken)
 110393    {
 394        lock (_mutex)
 110395        {
 110396            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 397
 110398            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 1399            {
 1400                throw new InvalidOperationException("Cannot close a Slic connection before connecting it.");
 401            }
 109402        }
 403
 109404        bool waitForWriterShutdown = false;
 109405        if (TryClose(new IceRpcException(IceRpcError.OperationAborted), "The connection was closed."))
 102406        {
 102407            using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 102408            {
 102409                if (IsServer && _writerIsShutdown)
 0410                {
 411                    // ReadFramesAsync already shut down the writer because the client-side sent its Close frame and
 412                    // shut down the duplex connection. Nothing more to send. The client-side is unaffected: it never
 413                    // shuts down the writer from ReadFramesAsync.
 0414                }
 415                else
 102416                {
 102417                    WriteFrame(FrameType.Close, streamId: null, new CloseBody((ulong)closeError).Encode);
 102418                    if (IsServer)
 53419                    {
 420                        // Link with _disposedCts so a concurrent DisposeAsync can break out of a flush parked on
 421                        // PauseWriterThreshold. Without the link, server CloseAsync(None) would deadlock with
 422                        // DisposeAsync: CloseAsync holds _writeSemaphore across this flush while DisposeAsync waits to
 423                        // acquire it before disposing the writer (which is what would otherwise unblock the flush).
 53424                        using var flushCts = CancellationTokenSource.CreateLinkedTokenSource(
 53425                            cancellationToken,
 53426                            _disposedCts.Token);
 427                        try
 53428                        {
 53429                            await _duplexConnectionWriter.FlushAsync(flushCts.Token).ConfigureAwait(false);
 52430                        }
 1431                        catch (OperationCanceledException) when (
 1432                            _disposedCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
 1433                        {
 1434                            throw new IceRpcException(IceRpcError.OperationAborted, "The connection was disposed.");
 435                        }
 52436                    }
 437                    else
 49438                    {
 439                        // The sending of the client-side Close frame is followed by the shutdown of the duplex
 440                        // connection. For TCP, it's important to always shutdown the connection on the client-side firs
 441                        // to avoid TIME_WAIT states on the server-side.
 49442                        _duplexConnectionWriter.Shutdown();
 49443                        waitForWriterShutdown = true;
 49444                    }
 101445                }
 101446            }
 101447        }
 448
 108449        if (waitForWriterShutdown)
 49450        {
 49451            await _duplexConnectionWriter.WriterTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 49452        }
 453
 454        // Now, wait for the peer to close the write side of the connection, which will terminate the read frames task.
 108455        Debug.Assert(_readFramesTask is not null);
 108456        await _readFramesTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 107457    }
 458
 459    public async ValueTask<IMultiplexedStream> CreateStreamAsync(
 460        bool bidirectional,
 461        CancellationToken cancellationToken)
 2145462    {
 463        lock (_mutex)
 2145464        {
 2145465            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 466
 2142467            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 2468            {
 2469                throw new InvalidOperationException("Cannot create stream before connecting the Slic connection.");
 470            }
 2140471            if (_isClosed)
 6472            {
 6473                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 474            }
 475
 2134476            ++_streamSemaphoreWaitCount;
 2134477        }
 478
 479        try
 2134480        {
 2134481            using var createStreamCts = CancellationTokenSource.CreateLinkedTokenSource(
 2134482                _closedCancellationToken,
 2134483                cancellationToken);
 484
 2134485            SemaphoreSlim? streamCountSemaphore = bidirectional ?
 2134486                _bidirectionalStreamSemaphore :
 2134487                _unidirectionalStreamSemaphore;
 488
 2134489            if (streamCountSemaphore is null)
 1490            {
 491                // The stream semaphore is null if the peer's max streams configuration is 0. In this case, we let
 492                // CreateStreamAsync hang indefinitely until the connection is closed.
 1493                await Task.Delay(-1, createStreamCts.Token).ConfigureAwait(false);
 0494            }
 495            else
 2133496            {
 2133497                await streamCountSemaphore.WaitAsync(createStreamCts.Token).ConfigureAwait(false);
 2122498            }
 499
 2122500            return new SlicStream(this, bidirectional, isRemote: false);
 501        }
 12502        catch (OperationCanceledException)
 12503        {
 12504            cancellationToken.ThrowIfCancellationRequested();
 7505            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 6506            Debug.Assert(_isClosed);
 6507            throw new IceRpcException(_peerCloseError ?? IceRpcError.OperationAborted, _closedMessage);
 508        }
 509        finally
 2134510        {
 511            lock (_mutex)
 2134512            {
 2134513                --_streamSemaphoreWaitCount;
 2134514                if (_isClosed && _streamSemaphoreWaitCount == 0)
 7515                {
 7516                    _streamSemaphoreWaitClosed.SetResult();
 7517                }
 2134518            }
 2134519        }
 2122520    }
 521
 522    public ValueTask DisposeAsync()
 1047523    {
 524        lock (_mutex)
 1047525        {
 1047526            _disposeTask ??= PerformDisposeAsync();
 1047527        }
 1047528        return new(_disposeTask);
 529
 530        async Task PerformDisposeAsync()
 768531        {
 532            // Make sure we execute the code below without holding the mutex lock.
 768533            await Task.Yield();
 768534            TryClose(new IceRpcException(IceRpcError.OperationAborted), "The connection was disposed.");
 535
 768536            _disposedCts.Cancel();
 537
 538            try
 768539            {
 768540                await Task.WhenAll(
 768541                    _connectTask ?? Task.CompletedTask,
 768542                    _readFramesTask ?? Task.CompletedTask,
 768543                    _streamSemaphoreWaitClosed.Task).ConfigureAwait(false);
 428544            }
 340545            catch
 340546            {
 547                // Expected if any of these tasks failed or was canceled. Each task takes care of handling unexpected
 548                // exceptions so there's no need to handle them here.
 340549            }
 550
 551            // Clean-up the streams that might still be queued on the channel.
 792552            while (_acceptStreamChannel.Reader.TryRead(out IMultiplexedStream? stream))
 24553            {
 24554                if (stream.IsBidirectional)
 5555                {
 5556                    stream.Output.Complete();
 5557                    stream.Input.Complete();
 5558                }
 19559                else if (stream.IsRemote)
 19560                {
 19561                    stream.Input.Complete();
 19562                }
 563                else
 0564                {
 0565                    stream.Output.Complete();
 0566                }
 24567            }
 568
 569            try
 768570            {
 571                // Prevents unobserved task exceptions.
 768572                await _acceptStreamChannel.Reader.Completion.ConfigureAwait(false);
 0573            }
 768574            catch
 768575            {
 768576            }
 577
 578            // Acquire (and never release) the write semaphore so no in-flight writer (e.g. a stream frame parked on
 579            // FlushAsync due to PauseWriterThreshold) can race with the writer disposal below. The wait is bounded:
 580            // every writer site uses a cancellation token derived from _closedCancellationToken or _disposedCts.Token,
 581            // both of which are cancelled by the time we reach this point.
 768582            await _writeSemaphore.WaitAsync(CancellationToken.None).ConfigureAwait(false);
 583
 768584            await _duplexConnectionWriter.DisposeAsync().ConfigureAwait(false);
 768585            _duplexConnectionReader.Dispose();
 768586            _duplexConnection.Dispose();
 587
 768588            _disposedCts.Dispose();
 768589            _bidirectionalStreamSemaphore?.Dispose();
 768590            _unidirectionalStreamSemaphore?.Dispose();
 768591            _closedCts.Dispose();
 768592        }
 1047593    }
 594
 769595    internal SlicConnection(
 769596        IDuplexConnection duplexConnection,
 769597        MultiplexedConnectionOptions options,
 769598        SlicTransportOptions slicOptions,
 769599        bool isServer)
 769600    {
 769601        IsServer = isServer;
 602
 769603        Pool = options.Pool;
 769604        MinSegmentSize = options.MinSegmentSize;
 769605        _maxBidirectionalStreams = options.MaxBidirectionalStreams;
 769606        _maxUnidirectionalStreams = options.MaxUnidirectionalStreams;
 607
 769608        InitialStreamWindowSize = slicOptions.InitialStreamWindowSize;
 769609        PauseWriterThreshold = slicOptions.PauseWriterThreshold;
 769610        _localIdleTimeout = slicOptions.IdleTimeout;
 769611        _maxOutstandingPongs = slicOptions.MaxOutstandingPongs;
 769612        _maxStreamFrameSize = slicOptions.MaxStreamFrameSize;
 613
 769614        _acceptStreamChannel = Channel.CreateUnbounded<IMultiplexedStream>(new UnboundedChannelOptions
 769615        {
 769616            SingleReader = true,
 769617            SingleWriter = true
 769618        });
 619
 769620        _closedCancellationToken = _closedCts.Token;
 621
 622        // Only the client-side sends pings to keep the connection alive when idle timeout (set later) is not infinite.
 769623        _duplexConnection = IsServer ?
 769624            new SlicDuplexConnectionDecorator(duplexConnection) :
 769625            new SlicDuplexConnectionDecorator(duplexConnection, SendReadPing, SendWritePing);
 626
 769627        _duplexConnectionReader = new DuplexConnectionReader(_duplexConnection, options.Pool, options.MinSegmentSize);
 769628        _duplexConnectionWriter = new SlicDuplexConnectionWriter(
 769629            _duplexConnection,
 769630            options.Pool,
 769631            options.MinSegmentSize,
 769632            PauseWriterThreshold);
 633
 634        // We use the same stream ID numbering scheme as QUIC.
 769635        if (IsServer)
 385636        {
 385637            _nextBidirectionalId = 1;
 385638            _nextUnidirectionalId = 3;
 385639        }
 640        else
 384641        {
 384642            _nextBidirectionalId = 0;
 384643            _nextUnidirectionalId = 2;
 384644        }
 645
 646        async Task SendPingAsync(long payload)
 15647        {
 648            try
 15649            {
 15650                await WriteConnectionFrameAsync(
 15651                    FrameType.Ping,
 15652                    new PingBody(payload).Encode,
 15653                    _closedCancellationToken).ConfigureAwait(false);
 15654            }
 0655            catch (IceRpcException)
 0656            {
 657                // Expected if the connection is closed.
 0658            }
 0659            catch (OperationCanceledException)
 0660            {
 661                // Expected if the connection is closed.
 0662            }
 0663            catch (Exception exception)
 0664            {
 0665                Debug.Fail($"The sending of a Ping frame failed with an unexpected exception: {exception}");
 0666            }
 15667        }
 668
 669        void SendReadPing()
 15670        {
 671            // No-op if there is already a pending Pong.
 15672            if (Interlocked.CompareExchange(ref _pendingPongCount, 1, 0) == 0)
 15673            {
 674                // Timer callbacks cannot await; fire-and-forget. SendPingAsync swallows expected exceptions and
 675                // Debug.Fails on unexpected ones, so the unobserved task carries no exception.
 15676                _ = SendPingAsync(1L);
 15677            }
 15678        }
 679
 680        void SendWritePing()
 0681        {
 682            // _pendingPongCount can be <= 0 if an unexpected pong is received. If it's the case, the connection is
 683            // being torn down and there's no point in sending a ping frame.
 0684            if (Interlocked.Increment(ref _pendingPongCount) > 0)
 0685            {
 0686                _ = SendPingAsync(0L);
 0687            }
 0688        }
 769689    }
 690
 691    /// <summary>Fills the given writer with stream data received on the connection.</summary>
 692    /// <param name="bufferWriter">The destination buffer writer.</param>
 693    /// <param name="byteCount">The amount of stream data to read.</param>
 694    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 695    internal ValueTask FillBufferWriterAsync(
 696        IBufferWriter<byte> bufferWriter,
 697        int byteCount,
 698        CancellationToken cancellationToken) =>
 8703699        _duplexConnectionReader.FillBufferWriterAsync(bufferWriter, byteCount, cancellationToken);
 700
 701    /// <summary>Releases a stream from the connection. The connection stream count is decremented and if this is a
 702    /// client allow a new stream to be started.</summary>
 703    /// <param name="stream">The released stream.</param>
 704    internal void ReleaseStream(SlicStream stream)
 4225705    {
 706        // Only a started stream has an Id and is registered in _streams.
 4225707        if (stream.IsStarted)
 4209708        {
 4209709            _streams.Remove(stream.Id, out SlicStream? _);
 4209710        }
 711
 4225712        if (stream.IsRemote)
 2108713        {
 2108714            if (stream.IsBidirectional)
 688715            {
 688716                Interlocked.Decrement(ref _bidirectionalStreamCount);
 688717            }
 718            else
 1420719            {
 1420720                Interlocked.Decrement(ref _unidirectionalStreamCount);
 1420721            }
 2108722        }
 2117723        else if (!_isClosed)
 1743724        {
 1743725            if (stream.IsBidirectional)
 633726            {
 633727                _bidirectionalStreamSemaphore!.Release();
 633728            }
 729            else
 1110730            {
 1110731                _unidirectionalStreamSemaphore!.Release();
 1110732            }
 1743733        }
 4225734    }
 735
 736    /// <summary>Throws the connection closure exception if the connection is already closed.</summary>
 737    internal void ThrowIfClosed()
 7984738    {
 739        lock (_mutex)
 7984740        {
 7984741            if (_isClosed)
 8742            {
 8743                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 744            }
 7976745        }
 7976746    }
 747
 748    /// <summary>Writes a connection frame.</summary>
 749    /// <param name="frameType">The frame type.</param>
 750    /// <param name="encode">The action to encode the frame.</param>
 751    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 752    internal async ValueTask WriteConnectionFrameAsync(
 753        FrameType frameType,
 754        EncodeAction? encode,
 755        CancellationToken cancellationToken)
 755756    {
 755757        Debug.Assert(frameType < FrameType.Stream);
 758
 755759        using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 753760        {
 761            lock (_mutex)
 753762            {
 753763                if (_isClosed)
 2764                {
 2765                    throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 766                }
 751767            }
 751768            WriteFrame(frameType, streamId: null, encode);
 751769            await _duplexConnectionWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
 751770        }
 751771    }
 772
 773    /// <summary>Writes a stream frame as a fire-and-forget operation. Used by sync code paths (e.g.
 774    /// <see cref="SlicPipeReader.Complete"/>, <see cref="SlicPipeWriter.Complete"/>, window updates) that cannot
 775    /// await.</summary>
 776    /// <param name="stream">The stream to write the frame for.</param>
 777    /// <param name="frameType">The frame type.</param>
 778    /// <param name="encode">The action to encode the frame.</param>
 779    /// <param name="writeReadsClosedFrame"><see langword="true" /> if a <see cref="FrameType.StreamReadsClosed" />
 780    /// frame should be written after the stream frame.</param>
 781    /// <remarks>This method is called by streams and might be called on a closed connection. The connection might
 782    /// also be closed concurrently while it's in progress.</remarks>
 783    internal void WriteStreamFrame(
 784        SlicStream stream,
 785        FrameType frameType,
 786        EncodeAction? encode,
 787        bool writeReadsClosedFrame)
 3249788    {
 789        // Ensure that this method is called for any FrameType.StreamXxx frame type except FrameType.Stream.
 3249790        Debug.Assert(frameType >= FrameType.StreamLast && stream.IsStarted);
 791
 792        // SemaphoreSlim.WaitAsync atomically updates the semaphore state (acquires it or enqueues the waiter)
 793        // synchronously, before the await can yield. Two sequential calls from the same thread therefore enqueue
 794        // in call order, preserving wire ordering of the resulting frames.
 3249795        _ = WriteStreamFrameAsync();
 796
 797        async Task WriteStreamFrameAsync()
 3249798        {
 799            SemaphoreLock semaphoreLock;
 800            try
 3249801            {
 3249802                semaphoreLock = await _writeSemaphore.AcquireAsync(_closedCancellationToken).ConfigureAwait(false);
 3247803            }
 2804            catch (OperationCanceledException)
 2805            {
 806                // The connection was closed while waiting for the semaphore.
 2807                return;
 808            }
 809
 3247810            using (semaphoreLock)
 3247811            {
 812                lock (_mutex)
 3247813                {
 3247814                    if (_isClosed)
 0815                    {
 0816                        return;
 817                    }
 3247818                }
 819
 3247820                WriteFrame(frameType, stream.Id, encode);
 3247821                if (writeReadsClosedFrame)
 101822                {
 101823                    WriteFrame(FrameType.StreamReadsClosed, stream.Id, encode: null);
 101824                }
 3247825                if (frameType == FrameType.StreamLast)
 588826                {
 827                    // Notify the stream that the last stream frame is considered sent at this point. This will
 828                    // close writes on the stream and allow the stream to be released if reads are also closed.
 588829                    stream.WroteLastStreamFrame();
 588830                }
 831
 832                try
 3247833                {
 3247834                    await _duplexConnectionWriter.FlushAsync(_closedCancellationToken).ConfigureAwait(false);
 3242835                }
 4836                catch (OperationCanceledException)
 4837                {
 838                    // The connection was closed while flushing.
 4839                }
 0840                catch (InvalidOperationException)
 0841                {
 842                    // The pipe writer was completed (Shutdown called) — connection is going away.
 0843                }
 1844                catch (IceRpcException exception)
 1845                {
 846                    // The duplex connection write failed. Since this fire-and-forget task has no caller to observe
 847                    // the failure, close the connection.
 1848                    TryClose(exception, "The connection was lost.", IceRpcError.ConnectionAborted);
 1849                }
 0850                catch (Exception exception)
 0851                {
 852                    // A duplex connection is only expected to fail a write with an IceRpcException. Rethrow to
 853                    // generate an unobserved task exception.
 0854                    Debug.Fail($"The stream frame write failed with an unexpected exception: {exception}");
 0855                    throw;
 856                }
 3247857            }
 3249858        }
 3249859    }
 860
 861    /// <summary>Writes a stream data frame.</summary>
 862    /// <param name="stream">The stream to write the frame for.</param>
 863    /// <param name="source1">The first stream frame data source.</param>
 864    /// <param name="source2">The second stream frame data source.</param>
 865    /// <param name="endStream"><see langword="true" /> to write a <see cref="FrameType.StreamLast" /> frame and
 866    /// <see langword="false" /> to write a <see cref="FrameType.Stream" /> frame.</param>
 867    /// <param name="writeReadsClosedFrame"><see langword="true" /> if a <see cref="FrameType.StreamReadsClosed" />
 868    /// frame should be written after the stream frame.</param>
 869    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 870    /// <remarks>This method is called by streams and might be called on a closed connection. The connection might
 871    /// also be closed concurrently while it's in progress.</remarks>
 872    internal async ValueTask<FlushResult> WriteStreamDataFrameAsync(
 873        SlicStream stream,
 874        ReadOnlySequence<byte> source1,
 875        ReadOnlySequence<byte> source2,
 876        bool endStream,
 877        bool writeReadsClosedFrame,
 878        CancellationToken cancellationToken)
 7956879    {
 7956880        Debug.Assert(!source1.IsEmpty || endStream);
 881
 7956882        if (_connectTask is null)
 0883        {
 0884            throw new InvalidOperationException("Cannot send a stream frame before calling ConnectAsync.");
 885        }
 886
 7956887        using var writeCts = CancellationTokenSource.CreateLinkedTokenSource(
 7956888            _closedCancellationToken,
 7956889            cancellationToken);
 890
 891        try
 7956892        {
 893            do
 9239894            {
 895                // Next, ensure send credit is available. If not, this will block until the receiver allows sending
 896                // additional data.
 9239897                int sendCredit = 0;
 9239898                if (!source1.IsEmpty || !source2.IsEmpty)
 9234899                {
 9234900                    sendCredit = await stream.AcquireSendCreditAsync(writeCts.Token).ConfigureAwait(false);
 9201901                    Debug.Assert(sendCredit > 0);
 9201902                }
 903
 904                // Gather data from source1 or source2 up to sendCredit bytes or the peer maximum stream frame size.
 9206905                int sendMaxSize = Math.Min(sendCredit, PeerMaxStreamFrameSize);
 906                ReadOnlySequence<byte> sendSource1;
 907                ReadOnlySequence<byte> sendSource2;
 9206908                if (!source1.IsEmpty)
 8220909                {
 8220910                    int length = Math.Min((int)source1.Length, sendMaxSize);
 8220911                    sendSource1 = source1.Slice(0, length);
 8220912                    source1 = source1.Slice(length);
 8220913                }
 914                else
 986915                {
 986916                    sendSource1 = ReadOnlySequence<byte>.Empty;
 986917                }
 918
 9206919                if (source1.IsEmpty && !source2.IsEmpty)
 2040920                {
 2040921                    int length = Math.Min((int)source2.Length, sendMaxSize - (int)sendSource1.Length);
 2040922                    sendSource2 = source2.Slice(0, length);
 2040923                    source2 = source2.Slice(length);
 2040924                }
 925                else
 7166926                {
 7166927                    sendSource2 = ReadOnlySequence<byte>.Empty;
 7166928                }
 929
 930                // If there's no data left to send and endStream is true, it's the last stream frame.
 9206931                bool lastStreamFrame = endStream && source1.IsEmpty && source2.IsEmpty;
 932
 9206933                using (await _writeSemaphore.AcquireAsync(writeCts.Token).ConfigureAwait(false))
 8230934                {
 935                    lock (_mutex)
 8230936                    {
 8230937                        if (_isClosed)
 0938                        {
 0939                            throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 940                        }
 8230941                    }
 942
 8230943                    if (!stream.IsStarted)
 2103944                    {
 2103945                        if (stream.IsBidirectional)
 682946                        {
 682947                            AddStream(_nextBidirectionalId, stream);
 682948                            _nextBidirectionalId += 4;
 682949                        }
 950                        else
 1421951                        {
 1421952                            AddStream(_nextUnidirectionalId, stream);
 1421953                            _nextUnidirectionalId += 4;
 1421954                        }
 2103955                    }
 956
 957                    // Notify the stream that we're consuming sendSize credit. It's important to call this before
 958                    // sending the stream frame to avoid race conditions where the StreamWindowUpdate frame could
 959                    // be received before the send credit was updated.
 8230960                    if (sendCredit > 0)
 8225961                    {
 8225962                        stream.ConsumedSendCredit((int)(sendSource1.Length + sendSource2.Length));
 8225963                    }
 964
 8230965                    EncodeStreamFrameHeader(stream.Id, sendSource1.Length + sendSource2.Length, lastStreamFrame);
 966
 8230967                    if (lastStreamFrame)
 802968                    {
 969                        // Notify the stream that the last stream frame is considered sent at this point. This
 970                        // will complete writes on the stream and allow the stream to be released if reads are
 971                        // also completed.
 802972                        stream.WroteLastStreamFrame();
 802973                    }
 974
 975                    // Write the stream frame.
 8230976                    if (!sendSource1.IsEmpty)
 8220977                    {
 8220978                        _duplexConnectionWriter.Write(sendSource1);
 8220979                    }
 8230980                    if (!sendSource2.IsEmpty)
 1064981                    {
 1064982                        _duplexConnectionWriter.Write(sendSource2);
 1064983                    }
 984
 8230985                    if (writeReadsClosedFrame)
 381986                    {
 381987                        WriteFrame(FrameType.StreamReadsClosed, stream.Id, encode: null);
 381988                    }
 989
 990                    // Flush the stream frame. This may block if the outbound pipe's pauseWriterThreshold has been
 991                    // reached — the connection's write semaphore is held during the await, so all other connection
 992                    // writers wait until the background writer task drains enough data.
 8230993                    await _duplexConnectionWriter.FlushAsync(writeCts.Token).ConfigureAwait(false);
 8221994                }
 8221995            }
 8221996            while (!source1.IsEmpty || !source2.IsEmpty); // Loop until there's no data left to send.
 6938997        }
 1018998        catch (OperationCanceledException)
 1018999        {
 10181000            cancellationToken.ThrowIfCancellationRequested();
 1001
 01002            Debug.Assert(_isClosed);
 01003            throw new IceRpcException(_peerCloseError ?? IceRpcError.OperationAborted, _closedMessage);
 1004        }
 1005
 69381006        return new FlushResult(isCanceled: false, isCompleted: false);
 1007
 1008        void EncodeStreamFrameHeader(ulong streamId, long size, bool lastStreamFrame)
 82301009        {
 82301010            var encoder = new SliceEncoder(_duplexConnectionWriter);
 82301011            encoder.EncodeFrameType(!lastStreamFrame ? FrameType.Stream : FrameType.StreamLast);
 82301012            Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(4);
 82301013            int startPos = encoder.EncodedByteCount;
 82301014            encoder.EncodeVarUInt62(streamId);
 82301015            SliceEncoder.EncodeVarUInt62((ulong)(encoder.EncodedByteCount - startPos + size), sizePlaceholder);
 82301016        }
 69381017    }
 1018
 1019    private void AddStream(ulong id, SlicStream stream)
 42111020    {
 1021        lock (_mutex)
 42111022        {
 42111023            if (_isClosed)
 21024            {
 21025                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 1026            }
 1027
 42091028            _streams[id] = stream;
 1029
 1030            // Assign the stream ID within the mutex to ensure that the addition of the stream to the connection and the
 1031            // stream ID assignment are atomic.
 42091032            stream.Id = id;
 1033
 1034            // Keep track of the last assigned stream ID. This is used to figure out if the stream is known or unknown.
 42091035            if (stream.IsRemote)
 21061036            {
 21061037                if (stream.IsBidirectional)
 6871038                {
 6871039                    _lastRemoteBidirectionalStreamId = id;
 6871040                }
 1041                else
 14191042                {
 14191043                    _lastRemoteUnidirectionalStreamId = id;
 14191044                }
 21061045            }
 42091046        }
 42091047    }
 1048
 1049    private void DecodeParameters(IDictionary<ParameterKey, IList<byte>> parameters)
 6931050    {
 6931051        int? maxStreamFrameSize = null;
 6931052        int? peerInitialStreamWindowSize = null;
 87201053        foreach ((ParameterKey key, IList<byte> buffer) in parameters)
 33211054        {
 33211055            switch (key)
 1056            {
 1057                case ParameterKey.MaxBidirectionalStreams:
 6081058                {
 6081059                    int value = DecodeParamValue(buffer);
 6081060                    if (value > 0)
 6081061                    {
 6081062                        _bidirectionalStreamSemaphore = new SemaphoreSlim(value, value);
 6081063                    }
 6081064                    break;
 1065                }
 1066                case ParameterKey.MaxUnidirectionalStreams:
 6651067                {
 6651068                    int value = DecodeParamValue(buffer);
 6651069                    if (value > 0)
 6651070                    {
 6651071                        _unidirectionalStreamSemaphore = new SemaphoreSlim(value, value);
 6651072                    }
 6651073                    break;
 1074                }
 1075                case ParameterKey.IdleTimeout:
 6631076                {
 6631077                    _peerIdleTimeout = TimeSpan.FromMilliseconds(DecodeParamValue(buffer));
 6631078                    if (_peerIdleTimeout == TimeSpan.Zero)
 01079                    {
 01080                        throw new InvalidDataException(
 01081                            "The IdleTimeout Slic connection parameter is invalid, it must be greater than 0 s.");
 1082                    }
 6631083                    break;
 1084                }
 1085                case ParameterKey.MaxStreamFrameSize:
 6931086                {
 6931087                    maxStreamFrameSize = DecodeParamValue(buffer);
 6931088                    if (maxStreamFrameSize < 1024)
 01089                    {
 01090                        throw new InvalidDataException(
 01091                            "The MaxStreamFrameSize connection parameter is invalid, it must be at least 1 KB.");
 1092                    }
 6931093                    if (maxStreamFrameSize > SlicTransportOptions.MaxStreamFrameSizeCeiling)
 11094                    {
 11095                        throw new InvalidDataException(
 11096                            $"The MaxStreamFrameSize connection parameter is invalid, it cannot exceed {SlicTransportOpt
 1097                    }
 6921098                    break;
 1099                }
 1100                case ParameterKey.InitialStreamWindowSize:
 6921101                {
 6921102                    peerInitialStreamWindowSize = DecodeParamValue(buffer);
 6921103                    if (peerInitialStreamWindowSize < 1024)
 01104                    {
 01105                        throw new InvalidDataException(
 01106                            "The InitialStreamWindowSize connection parameter is invalid, it must be at least 1 KB.");
 1107                    }
 6921108                    break;
 1109                }
 1110                // Ignore unsupported parameter.
 1111            }
 33201112        }
 1113
 6921114        if (maxStreamFrameSize is null)
 01115        {
 01116            throw new InvalidDataException(
 01117                "The peer didn't send the required MaxStreamFrameSize connection parameter.");
 1118        }
 1119        else
 6921120        {
 6921121            PeerMaxStreamFrameSize = maxStreamFrameSize.Value;
 6921122        }
 1123
 6921124        if (peerInitialStreamWindowSize is null)
 01125        {
 01126            throw new InvalidDataException(
 01127                "The peer didn't send the required InitialStreamWindowSize connection parameter.");
 1128        }
 1129        else
 6921130        {
 6921131            PeerInitialStreamWindowSize = peerInitialStreamWindowSize.Value;
 6921132        }
 1133
 1134        // all parameter values are currently integers in the range 0..Int32Max encoded as varuint62.
 1135        static int DecodeParamValue(IList<byte> buffer)
 33211136        {
 1137            // The IList<byte> decoded by the IceRPC + Slice integration is backed by an array
 33211138            ulong value = new ReadOnlySequence<byte>((byte[])buffer).DecodeSliceBuffer(
 66421139                (ref SliceDecoder decoder) => decoder.DecodeVarUInt62());
 1140            try
 33211141            {
 33211142                return checked((int)value);
 1143            }
 01144            catch (OverflowException exception)
 01145            {
 01146                throw new InvalidDataException("The value is out of the varuint32 accepted range.", exception);
 1147            }
 33211148        }
 6921149    }
 1150
 1151    private Dictionary<ParameterKey, IList<byte>> EncodeParameters()
 7181152    {
 7181153        var parameters = new List<KeyValuePair<ParameterKey, IList<byte>>>
 7181154        {
 7181155            // Required parameters.
 7181156            EncodeParameter(ParameterKey.MaxStreamFrameSize, (ulong)_maxStreamFrameSize),
 7181157            EncodeParameter(ParameterKey.InitialStreamWindowSize, (ulong)InitialStreamWindowSize)
 7181158        };
 1159
 1160        // Optional parameters.
 7181161        if (_localIdleTimeout != Timeout.InfiniteTimeSpan)
 7161162        {
 7161163            parameters.Add(EncodeParameter(ParameterKey.IdleTimeout, (ulong)_localIdleTimeout.TotalMilliseconds));
 7161164        }
 7181165        if (_maxBidirectionalStreams > 0)
 6531166        {
 6531167            parameters.Add(EncodeParameter(ParameterKey.MaxBidirectionalStreams, (ulong)_maxBidirectionalStreams));
 6531168        }
 7181169        if (_maxUnidirectionalStreams > 0)
 7181170        {
 7181171            parameters.Add(EncodeParameter(ParameterKey.MaxUnidirectionalStreams, (ulong)_maxUnidirectionalStreams));
 7181172        }
 1173
 7181174        return new Dictionary<ParameterKey, IList<byte>>(parameters);
 1175
 1176        static KeyValuePair<ParameterKey, IList<byte>> EncodeParameter(ParameterKey key, ulong value)
 35231177        {
 35231178            int sizeLength = SliceEncoder.GetVarUInt62EncodedSize(value);
 35231179            byte[] buffer = new byte[sizeLength];
 35231180            SliceEncoder.EncodeVarUInt62(value, buffer);
 35231181            return new(key, buffer);
 35231182        }
 7181183    }
 1184
 1185    private bool IsUnknownStream(ulong streamId)
 52861186    {
 52861187        bool isRemote = streamId % 2 == (IsServer ? 0ul : 1ul);
 52861188        bool isBidirectional = streamId % 4 < 2;
 52861189        if (isRemote)
 27981190        {
 27981191            if (isBidirectional)
 13491192            {
 13491193                return _lastRemoteBidirectionalStreamId is null || streamId > _lastRemoteBidirectionalStreamId;
 1194            }
 1195            else
 14491196            {
 14491197                return _lastRemoteUnidirectionalStreamId is null || streamId > _lastRemoteUnidirectionalStreamId;
 1198            }
 1199        }
 1200        else
 24881201        {
 24881202            if (isBidirectional)
 13041203            {
 13041204                return streamId >= _nextBidirectionalId;
 1205            }
 1206            else
 11841207            {
 11841208                return streamId >= _nextUnidirectionalId;
 1209            }
 1210        }
 52861211    }
 1212
 1213    private Task ReadFrameAsync(FrameType frameType, int size, ulong? streamId, CancellationToken cancellationToken)
 120781214    {
 120781215        if (frameType >= FrameType.Stream && streamId is null)
 01216        {
 01217            throw new InvalidDataException("Received stream frame without stream ID.");
 1218        }
 1219
 120781220        switch (frameType)
 1221        {
 1222            case FrameType.Close:
 1021223            {
 1021224                return ReadCloseFrameAsync(size, cancellationToken);
 1225            }
 1226            case FrameType.Ping:
 231227            {
 231228                return ReadPingFrameAndWritePongFrameAsync(size, cancellationToken);
 1229            }
 1230            case FrameType.Pong:
 181231            {
 181232                return ReadPongFrameAsync(size, cancellationToken);
 1233            }
 1234            case FrameType.Stream:
 1235            case FrameType.StreamLast:
 88051236            {
 88051237                return ReadStreamDataFrameAsync(frameType, size, streamId!.Value, cancellationToken);
 1238            }
 1239            case FrameType.StreamWindowUpdate:
 12521240            {
 12521241                if (IsUnknownStream(streamId!.Value))
 11242                {
 11243                    throw new InvalidDataException($"Received {frameType} frame for unknown stream.");
 1244                }
 1245
 12511246                return ReadStreamWindowUpdateFrameAsync(size, streamId!.Value, cancellationToken);
 1247            }
 1248            case FrameType.StreamReadsClosed:
 1249            case FrameType.StreamWritesClosed:
 18751250            {
 18751251                if (size > 0)
 21252                {
 21253                    throw new InvalidDataException($"Unexpected body for {frameType} frame.");
 1254                }
 18731255                if (IsUnknownStream(streamId!.Value))
 21256                {
 21257                    throw new InvalidDataException($"Received {frameType} frame for unknown stream.");
 1258                }
 1259
 18711260                if (_streams.TryGetValue(streamId.Value, out SlicStream? stream))
 13751261                {
 13751262                    if (frameType == FrameType.StreamWritesClosed)
 481263                    {
 481264                        stream.ReceivedWritesClosedFrame();
 481265                    }
 1266                    else
 13271267                    {
 13271268                        stream.ReceivedReadsClosedFrame();
 13271269                    }
 13751270                }
 18711271                return Task.CompletedTask;
 1272            }
 1273            default:
 31274            {
 31275                throw new InvalidDataException($"Received unexpected {frameType} frame.");
 1276            }
 1277        }
 1278
 1279        async Task ReadCloseFrameAsync(int size, CancellationToken cancellationToken)
 1021280        {
 1021281            CloseBody closeBody = await ReadFrameBodyAsync(
 1021282                FrameType.Close,
 1021283                size,
 1011284                (ref SliceDecoder decoder) => new CloseBody(ref decoder),
 1021285                cancellationToken).ConfigureAwait(false);
 1286
 1001287            IceRpcError? peerCloseError = closeBody.ApplicationErrorCode switch
 1001288            {
 781289                (ulong)MultiplexedConnectionCloseError.NoError => IceRpcError.ConnectionClosedByPeer,
 41290                (ulong)MultiplexedConnectionCloseError.Refused => IceRpcError.ConnectionRefused,
 81291                (ulong)MultiplexedConnectionCloseError.ServerBusy => IceRpcError.ServerBusy,
 51292                (ulong)MultiplexedConnectionCloseError.Aborted => IceRpcError.ConnectionAborted,
 51293                _ => null
 1001294            };
 1295
 1296            bool notAlreadyClosed;
 1001297            if (peerCloseError is null)
 51298            {
 51299                notAlreadyClosed = TryClose(
 51300                    new IceRpcException(IceRpcError.ConnectionAborted),
 51301                    $"The connection was closed by the peer with an unknown application error code: '{closeBody.Applicat
 51302                    IceRpcError.ConnectionAborted);
 51303            }
 1304            else
 951305            {
 951306                notAlreadyClosed = TryClose(
 951307                    new IceRpcException(peerCloseError.Value),
 951308                    "The connection was closed by the peer.",
 951309                    peerCloseError);
 951310            }
 1311
 1312            // The server-side of the duplex connection is only shutdown once the client-side is shutdown. When using
 1313            // TCP, this ensures that the server TCP connection won't end-up in the TIME_WAIT state on the server-side.
 1001314            if (notAlreadyClosed && !IsServer)
 231315            {
 1316                // DisposeAsync waits for the reads frames task to complete before disposing the writer.
 1317                // _writeSemaphore alone serializes access to the writer.
 231318                using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 231319                {
 231320                    _duplexConnectionWriter.Shutdown();
 231321                }
 231322                await _duplexConnectionWriter.WriterTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 231323            }
 1001324        }
 1325
 1326        async Task ReadPingFrameAndWritePongFrameAsync(int size, CancellationToken cancellationToken)
 231327        {
 1328            // Read the ping frame.
 231329            PingBody pingBody = await ReadFrameBodyAsync(
 231330                FrameType.Ping,
 231331                size,
 221332                (ref SliceDecoder decoder) => new PingBody(ref decoder),
 231333                cancellationToken).ConfigureAwait(false);
 1334
 211335            if (Interlocked.Increment(ref _outstandingPongCount) > _maxOutstandingPongs)
 11336            {
 11337                throw new IceRpcException(
 11338                    IceRpcError.IceRpcError,
 11339                    $"Received a {nameof(FrameType.Ping)} frame while {_maxOutstandingPongs} {nameof(FrameType.Pong)} fr
 1340            }
 1341
 1342            // Return a pong frame with the ping payload, written in the background: writing it from the read frames
 1343            // loop would block the loop when another writer holds _writeSemaphore while parked on a full outbound
 1344            // pipe, suppressing further reads (and idle timeout detection) until the pipe drains.
 201345            _ = WritePongFrameAsync(pingBody.Payload);
 201346        }
 1347
 1348        async Task WritePongFrameAsync(long payload)
 201349        {
 1350            try
 201351            {
 201352                await WriteConnectionFrameAsync(
 201353                    FrameType.Pong,
 201354                    new PongBody(payload).Encode,
 201355                    _closedCancellationToken).ConfigureAwait(false);
 161356            }
 21357            catch (IceRpcException)
 21358            {
 1359                // Expected if the connection is closed.
 21360            }
 21361            catch (OperationCanceledException)
 21362            {
 1363                // Expected if the connection is closed.
 21364            }
 01365            catch (Exception exception)
 01366            {
 01367                Debug.Fail($"The sending of a Pong frame failed with an unexpected exception: {exception}");
 1368
 1369                // Rethrow so in release builds the exception is not swallowed and can be presented to the application
 1370                // as an unobserved task exception.
 01371                throw;
 1372            }
 1373            finally
 201374            {
 201375                Interlocked.Decrement(ref _outstandingPongCount);
 201376            }
 201377        }
 1378
 1379        async Task ReadPongFrameAsync(int size, CancellationToken cancellationToken)
 181380        {
 181381            if (Interlocked.Decrement(ref _pendingPongCount) >= 0)
 151382            {
 1383                // Ensure the pong frame payload value is expected.
 1384
 151385                PongBody pongBody = await ReadFrameBodyAsync(
 151386                    FrameType.Pong,
 151387                    size,
 151388                    (ref SliceDecoder decoder) => new PongBody(ref decoder),
 151389                    cancellationToken).ConfigureAwait(false);
 1390
 1391                // For now, we only send a 0 or 1 payload value (0 for "write ping" and 1 for "read ping").
 151392                if (pongBody.Payload != 0L && pongBody.Payload != 1L)
 01393                {
 01394                    throw new InvalidDataException($"Received {nameof(FrameType.Pong)} with unexpected payload.");
 1395                }
 151396            }
 1397            else
 31398            {
 1399                // If not waiting for a pong frame, this pong frame is unexpected.
 31400                throw new InvalidDataException($"Received unexpected {nameof(FrameType.Pong)} frame.");
 1401            }
 151402        }
 1403
 1404        async Task ReadStreamWindowUpdateFrameAsync(int size, ulong streamId, CancellationToken cancellationToken)
 12511405        {
 12511406            StreamWindowUpdateBody frame = await ReadFrameBodyAsync(
 12511407                FrameType.StreamWindowUpdate,
 12511408                size,
 12511409                (ref SliceDecoder decoder) => new StreamWindowUpdateBody(ref decoder),
 12511410                cancellationToken).ConfigureAwait(false);
 12511411            if (_streams.TryGetValue(streamId, out SlicStream? stream))
 12121412            {
 12121413                stream.ReceivedWindowUpdateFrame(frame);
 12111414            }
 12501415        }
 1416
 1417        async Task<T> ReadFrameBodyAsync<T>(
 1418            FrameType frameType,
 1419            int size,
 1420            DecodeFunc<T> decodeFunc,
 1421            CancellationToken cancellationToken)
 13911422        {
 13911423            if (size <= 0)
 21424            {
 21425                throw new InvalidDataException($"Unexpected empty body for {frameType} frame.");
 1426            }
 1427
 13891428            ReadOnlySequence<byte> buffer = await _duplexConnectionReader.ReadAtLeastAsync(size, cancellationToken)
 13891429                .ConfigureAwait(false);
 1430
 13891431            if (buffer.Length > size)
 9231432            {
 9231433                buffer = buffer.Slice(0, size);
 9231434            }
 1435
 13891436            T decodedFrame = buffer.DecodeSliceBuffer(decodeFunc);
 13871437            _duplexConnectionReader.AdvanceTo(buffer.End);
 13871438            return decodedFrame;
 13871439        }
 120701440    }
 1441
 1442    private async ValueTask<(FrameType FrameType, int FrameSize, ulong? StreamId)?> ReadFrameHeaderAsync(
 1443        CancellationToken cancellationToken)
 134781444    {
 134781445        while (true)
 134781446        {
 1447            // Read data from the pipe reader.
 134781448            if (!_duplexConnectionReader.TryRead(out ReadOnlySequence<byte> buffer))
 93071449            {
 93071450                buffer = await _duplexConnectionReader.ReadAsync(cancellationToken).ConfigureAwait(false);
 87551451            }
 1452
 129261453            if (buffer.IsEmpty)
 1421454            {
 1421455                return null;
 1456            }
 1457
 127841458            if (TryDecodeHeader(
 127841459                buffer,
 127841460                out (FrameType FrameType, int FrameSize, ulong? StreamId) header,
 127841461                out int consumed))
 127781462            {
 127781463                _duplexConnectionReader.AdvanceTo(buffer.GetPosition(consumed));
 127781464                return header;
 1465            }
 1466            else
 01467            {
 01468                _duplexConnectionReader.AdvanceTo(buffer.Start, buffer.End);
 01469            }
 01470        }
 1471
 1472        static bool TryDecodeHeader(
 1473            ReadOnlySequence<byte> buffer,
 1474            out (FrameType FrameType, int FrameSize, ulong? StreamId) header,
 1475            out int consumed)
 127841476        {
 127841477            header = default;
 127841478            consumed = default;
 1479
 127841480            var decoder = new SliceDecoder(buffer);
 1481
 1482            // Decode the frame type and frame size.
 127841483            if (!decoder.TryDecodeUInt8(out byte frameType) || !decoder.TryDecodeVarUInt62(out ulong frameSize))
 01484            {
 01485                return false;
 1486            }
 1487
 127841488            header.FrameType = frameType.AsFrameType();
 1489            try
 127811490            {
 127811491                header.FrameSize = checked((int)frameSize);
 127811492            }
 01493            catch (OverflowException exception)
 01494            {
 01495                throw new InvalidDataException("The frame size can't be larger than int.MaxValue.", exception);
 1496            }
 1497
 1498            // Reject oversized control frame bodies before any buffering occurs.
 127811499            if (header.FrameType < FrameType.Stream && header.FrameSize > MaxControlFrameBodySize)
 11500            {
 11501                throw new InvalidDataException(
 11502                    $"The {header.FrameType} frame body size ({header.FrameSize}) exceeds the maximum allowed size ({Max
 1503            }
 1504
 1505            // If it's a stream frame, try to decode the stream ID
 127801506            if (header.FrameType >= FrameType.Stream)
 119341507            {
 119341508                if (header.FrameSize == 0)
 11509                {
 11510                    throw new InvalidDataException("Invalid stream frame size.");
 1511                }
 1512
 119331513                consumed = (int)decoder.Consumed;
 119331514                if (!decoder.TryDecodeVarUInt62(out ulong streamId))
 01515                {
 01516                    return false;
 1517                }
 119331518                header.StreamId = streamId;
 119331519                header.FrameSize -= (int)decoder.Consumed - consumed;
 1520
 119331521                if (header.FrameSize < 0)
 11522                {
 11523                    throw new InvalidDataException("Invalid stream frame size.");
 1524                }
 119321525            }
 1526
 127781527            consumed = (int)decoder.Consumed;
 127781528            return true;
 127781529        }
 129201530    }
 1531
 1532    private async Task ReadFramesAsync(CancellationToken cancellationToken)
 6921533    {
 1534        try
 6921535        {
 127491536            while (true)
 127491537            {
 127491538                (FrameType Type, int Size, ulong? StreamId)? header = await ReadFrameHeaderAsync(cancellationToken)
 127491539                    .ConfigureAwait(false);
 1540
 122191541                if (header is null)
 1411542                {
 1543                    lock (_mutex)
 1411544                    {
 1411545                        if (!_isClosed)
 01546                        {
 1547                            // Unexpected duplex connection shutdown.
 01548                            throw new IceRpcException(IceRpcError.ConnectionAborted);
 1549                        }
 1411550                    }
 1551                    // The peer has shut down the duplex connection.
 1411552                    break;
 1553                }
 1554
 120781555                await ReadFrameAsync(header.Value.Type, header.Value.Size, header.Value.StreamId, cancellationToken)
 120781556                    .ConfigureAwait(false);
 120571557            }
 1558
 1411559            if (IsServer)
 721560            {
 721561                Debug.Assert(_isClosed);
 1562
 1563                // The server-side of the duplex connection is only shutdown once the client-side is shutdown. When
 1564                // using TCP, this ensures that the server TCP connection won't end-up in the TIME_WAIT state on the
 1565                // server-side.
 1566
 1567                // DisposeAsync waits for the reads frames task to complete before disposing the writer.
 1568                // _writeSemaphore alone serializes access to the writer and guards _writerIsShutdown.
 721569                using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 721570                {
 721571                    _duplexConnectionWriter.Shutdown();
 1572
 1573                    // Make sure that CloseAsync doesn't call Write on the writer if it's called shortly after the peer
 1574                    // shutdown its side of the connection (which triggers ReadFrameHeaderAsync to return null).
 721575                    _writerIsShutdown = true;
 721576                }
 1577
 721578                await _duplexConnectionWriter.WriterTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 711579            }
 1401580        }
 2681581        catch (OperationCanceledException)
 2681582        {
 1583            // Expected, DisposeAsync was called.
 2681584        }
 2621585        catch (IceRpcException exception)
 2621586        {
 2621587            TryClose(exception, "The connection was lost.", IceRpcError.ConnectionAborted);
 2621588            throw;
 1589        }
 221590        catch (InvalidDataException exception)
 221591        {
 221592            var rpcException = new IceRpcException(
 221593                IceRpcError.IceRpcError,
 221594                "The connection was aborted by a Slic protocol error.",
 221595                exception);
 221596            TryClose(rpcException, rpcException.Message, IceRpcError.IceRpcError);
 221597            throw rpcException;
 1598        }
 01599        catch (Exception exception)
 01600        {
 01601            Debug.Fail($"The read frames task completed due to an unhandled exception: {exception}");
 01602            TryClose(exception, "The connection was lost.", IceRpcError.ConnectionAborted);
 01603            throw;
 1604        }
 4081605    }
 1606
 1607    private async Task ReadStreamDataFrameAsync(
 1608        FrameType type,
 1609        int size,
 1610        ulong streamId,
 1611        CancellationToken cancellationToken)
 88051612    {
 88051613        bool endStream = type == FrameType.StreamLast;
 88051614        bool isRemote = streamId % 2 == (IsServer ? 0ul : 1ul);
 88051615        bool isBidirectional = streamId % 4 < 2;
 1616
 88051617        if (!isBidirectional && !isRemote)
 01618        {
 01619            throw new InvalidDataException(
 01620                "Received unexpected stream frame on local unidirectional stream.");
 1621        }
 88051622        else if (size == 0 && !endStream)
 11623        {
 11624            throw new InvalidDataException($"Received invalid {nameof(FrameType.Stream)} frame.");
 1625        }
 88041626        else if (size > _maxStreamFrameSize)
 11627        {
 11628            throw new InvalidDataException(
 11629                $"Received stream frame with size {size} exceeding the advertised maximum of {_maxStreamFrameSize} bytes
 1630        }
 1631
 88031632        if (!_streams.TryGetValue(streamId, out SlicStream? stream) && isRemote && IsUnknownStream(streamId))
 21101633        {
 1634            // Create a new remote stream.
 1635
 21101636            if (size == 0)
 01637            {
 01638                throw new InvalidDataException("Received empty stream frame on new stream.");
 1639            }
 1640
 21101641            if (isBidirectional)
 6891642            {
 6891643                ulong expectedStreamId = _lastRemoteBidirectionalStreamId is ulong lastId
 6891644                    ? lastId + 4
 6891645                    : (IsServer ? 0ul : 1ul);
 6891646                if (streamId != expectedStreamId)
 11647                {
 11648                    throw new InvalidDataException("Invalid stream ID.");
 1649                }
 1650
 6881651                if (_bidirectionalStreamCount == _maxBidirectionalStreams)
 01652                {
 01653                    throw new IceRpcException(
 01654                        IceRpcError.IceRpcError,
 01655                        $"The maximum bidirectional stream count {_maxBidirectionalStreams} was reached.");
 1656                }
 6881657                Interlocked.Increment(ref _bidirectionalStreamCount);
 6881658            }
 1659            else
 14211660            {
 14211661                ulong expectedStreamId = _lastRemoteUnidirectionalStreamId is ulong lastId
 14211662                    ? lastId + 4
 14211663                    : (IsServer ? 2ul : 3ul);
 14211664                if (streamId != expectedStreamId)
 11665                {
 11666                    throw new InvalidDataException("Invalid stream ID.");
 1667                }
 1668
 14201669                if (_unidirectionalStreamCount == _maxUnidirectionalStreams)
 01670                {
 01671                    throw new IceRpcException(
 01672                        IceRpcError.IceRpcError,
 01673                        $"The maximum unidirectional stream count {_maxUnidirectionalStreams} was reached.");
 1674                }
 14201675                Interlocked.Increment(ref _unidirectionalStreamCount);
 14201676            }
 1677
 1678            // The stream is registered with the connection and queued on the channel. The caller of AcceptStreamAsync
 1679            // is responsible for cleaning up the stream.
 21081680            stream = new SlicStream(this, isBidirectional, isRemote: true);
 1681
 1682            try
 21081683            {
 21081684                AddStream(streamId, stream);
 1685
 1686                try
 21061687                {
 21061688                    await _acceptStreamChannel.Writer.WriteAsync(
 21061689                        stream,
 21061690                        cancellationToken).ConfigureAwait(false);
 21061691                }
 01692                catch (ChannelClosedException exception)
 01693                {
 1694                    // The exception given to ChannelWriter.Complete(Exception? exception) is the InnerException.
 01695                    Debug.Assert(exception.InnerException is not null);
 01696                    throw ExceptionUtil.Throw(exception.InnerException);
 1697                }
 21061698            }
 21699            catch (IceRpcException)
 21700            {
 1701                // The two methods above throw IceRpcException if the connection has been closed (either by CloseAsync
 1702                // or because the close frame was received). We cleanup up the stream but don't throw to not abort the
 1703                // reading. The connection graceful closure still needs to read on the connection to figure out when the
 1704                // peer shuts down the duplex connection.
 21705                Debug.Assert(_isClosed);
 21706                stream.Input.Complete();
 21707                if (isBidirectional)
 11708                {
 11709                    stream.Output.Complete();
 11710                }
 21711            }
 21081712        }
 1713
 88011714        bool isDataConsumed = false;
 88011715        if (stream is not null)
 87361716        {
 1717            // Let the stream consume the stream frame data.
 87361718            isDataConsumed = await stream.ReceivedDataFrameAsync(
 87361719                size,
 87361720                endStream,
 87361721                cancellationToken).ConfigureAwait(false);
 87361722        }
 1723
 88011724        if (!isDataConsumed)
 981725        {
 1726            // The stream (if any) didn't consume the data. Read and ignore the data using a helper pipe.
 981727            var pipe = new Pipe(
 981728                new PipeOptions(
 981729                    pool: Pool,
 981730                    pauseWriterThreshold: 0,
 981731                    minimumSegmentSize: MinSegmentSize,
 981732                    useSynchronizationContext: false));
 1733
 981734            await _duplexConnectionReader.FillBufferWriterAsync(
 981735                    pipe.Writer,
 981736                    size,
 981737                    cancellationToken).ConfigureAwait(false);
 1738
 981739            pipe.Writer.Complete();
 981740            pipe.Reader.Complete();
 981741        }
 88011742    }
 1743
 1744    private bool TryClose(Exception exception, string closeMessage, IceRpcError? peerCloseError = null)
 12621745    {
 1746        lock (_mutex)
 12621747        {
 12621748            if (_isClosed)
 4941749            {
 4941750                return false;
 1751            }
 7681752            _isClosed = true;
 7681753            _closedMessage = closeMessage;
 7681754            _peerCloseError = peerCloseError;
 7681755            if (_streamSemaphoreWaitCount == 0)
 7611756            {
 7611757                _streamSemaphoreWaitClosed.SetResult();
 7611758            }
 7681759        }
 1760
 1761        // Cancel pending CreateStreamAsync, AcceptStreamAsync and WriteStreamDataFrameAsync operations.
 7681762        _closedCts.Cancel();
 7681763        _acceptStreamChannel.Writer.TryComplete(exception);
 1764
 1765        // Close streams.
 36461766        foreach (SlicStream stream in _streams.Values)
 6711767        {
 6711768            stream.Close(exception);
 6711769        }
 1770
 7681771        return true;
 12621772    }
 1773
 1774    private void WriteFrame(FrameType frameType, ulong? streamId, EncodeAction? encode)
 45821775    {
 45821776        var encoder = new SliceEncoder(_duplexConnectionWriter);
 45821777        encoder.EncodeFrameType(frameType);
 1778        // 2 bytes is sufficient: control frame bodies are limited to MaxControlFrameBodySize (16,383) and the
 1779        // stream frames encoded by WriteFrame carry at most a stream ID + a small body (e.g., StreamWindowUpdate).
 45821780        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(2);
 45821781        int startPos = encoder.EncodedByteCount;
 45821782        if (streamId is not null)
 37291783        {
 37291784            encoder.EncodeVarUInt62(streamId.Value);
 37291785        }
 45821786        encode?.Invoke(ref encoder);
 45821787        SliceEncoder.EncodeVarUInt62((ulong)(encoder.EncodedByteCount - startPos), sizePlaceholder);
 45821788    }
 1789}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

get_IsServer()
get_MinSegmentSize()
get_PeerInitialStreamWindowSize()
get_PeerMaxStreamFrameSize()
get_Pool()
get_InitialStreamWindowSize()
get_PauseWriterThreshold()
get_StreamWindowUpdateThreshold()
.ctor(IceRpc.Transports.IDuplexConnection,IceRpc.Transports.MultiplexedConnectionOptions,IceRpc.Transports.Slic.SlicTransportOptions,System.Boolean)
AcceptStreamAsync()
ConnectAsync(System.Threading.CancellationToken)
PerformConnectAsync()
DecodeInitialize()
DecodeInitializeAckOrVersion()
ReadFrameAsync()
CloseAsync()
CreateStreamAsync()
DisposeAsync()
PerformDisposeAsync()
SendPingAsync()
SendReadPing()
SendWritePing()
FillBufferWriterAsync(System.Buffers.IBufferWriter`1<System.Byte>,System.Int32,System.Threading.CancellationToken)
ReleaseStream(IceRpc.Transports.Slic.Internal.SlicStream)
ThrowIfClosed()
WriteConnectionFrameAsync()
WriteStreamFrame(IceRpc.Transports.Slic.Internal.SlicStream,IceRpc.Transports.Slic.Internal.FrameType,ZeroC.Slice.Codec.EncodeAction,System.Boolean)
WriteStreamFrameAsync()
WriteStreamDataFrameAsync()
EncodeStreamFrameHeader()
AddStream(System.UInt64,IceRpc.Transports.Slic.Internal.SlicStream)
DecodeParameters(System.Collections.Generic.IDictionary`2<IceRpc.Transports.Slic.Internal.ParameterKey,System.Collections.Generic.IList`1<System.Byte>>)
DecodeParamValue()
EncodeParameters()
EncodeParameter()
IsUnknownStream(System.UInt64)
ReadFrameAsync(IceRpc.Transports.Slic.Internal.FrameType,System.Int32,System.Nullable`1<System.UInt64>,System.Threading.CancellationToken)
ReadCloseFrameAsync()
ReadPingFrameAndWritePongFrameAsync()
WritePongFrameAsync()
ReadPongFrameAsync()
ReadStreamWindowUpdateFrameAsync()
ReadFrameBodyAsync()
ReadFrameHeaderAsync()
TryDecodeHeader()
ReadFramesAsync()
ReadStreamDataFrameAsync()
TryClose(System.Exception,System.String,System.Nullable`1<IceRpc.IceRpcError>)
WriteFrame(IceRpc.Transports.Slic.Internal.FrameType,System.Nullable`1<System.UInt64>,ZeroC.Slice.Codec.EncodeAction)