< 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: 2300_35243572715
Line coverage
89%
Covered lines: 1007
Uncovered lines: 116
Coverable lines: 1123
Total lines: 1797
Line coverage: 89.6%
Branch coverage
90%
Covered branches: 319
Total branches: 354
Branch coverage: 90.1%
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()100%88100%
CloseAsync()92.85%141495.45%
CreateStreamAsync()100%141497.61%
DisposeAsync()100%22100%
PerformDisposeAsync()93.75%161690.69%
SendPingAsync()100%1161.11%
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(...)83.33%6689.47%
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%66100%
ReadFrameBodyAsync()100%44100%
ReadFrameHeaderAsync()83.33%6681.81%
TryDecodeHeader()83.33%201881.08%
ReadFramesAsync()83.33%6685.71%
ReadStreamDataFrameAsync()85.71%774272.91%
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>
 1743520    internal bool IsServer { get; }
 21
 22    /// <summary>Gets the minimum size of the segment requested from <see cref="Pool" />.</summary>
 575623    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>
 356528    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>
 996033    internal int PeerMaxStreamFrameSize { get; private set; }
 34
 35    /// <summary>Gets the <see cref="MemoryPool{T}" /> used for obtaining memory buffers.</summary>
 575636    internal MemoryPool<byte> Pool { get; }
 37
 38    /// <summary>Gets the initial stream window size.</summary>
 1086339    internal int InitialStreamWindowSize { get; }
 40
 41    /// <summary>Gets the pause writer threshold for the connection's outbound pipe.</summary>
 79942    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>
 728346    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;
 79961    private readonly CancellationTokenSource _closedCts = new();
 62    private string? _closedMessage;
 63    private Task<TransportConnectionInformation>? _connectTask;
 79964    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.
 79984    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;
 79993    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
 79999    private readonly ConcurrentDictionary<ulong, SlicStream> _streams = new();
 100    private int _streamSemaphoreWaitCount;
 799101    private readonly TaskCompletionSource _streamSemaphoreWaitClosed =
 799102        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
 799115    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)
 2502125    {
 126        lock (_mutex)
 2502127        {
 2502128            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 129
 2501130            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 1131            {
 1132                throw new InvalidOperationException("Cannot accept stream before connecting the Slic connection.");
 133            }
 2500134            if (_isClosed)
 10135            {
 10136                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 137            }
 2490138        }
 139
 140        try
 2490141        {
 2490142            return await _acceptStreamChannel.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
 143        }
 158144        catch (ChannelClosedException exception)
 158145        {
 158146            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 157147            Debug.Assert(exception.InnerException is not null);
 148            // The exception given to ChannelWriter.Complete(Exception? exception) is the InnerException.
 157149            throw ExceptionUtil.Throw(exception.InnerException);
 150        }
 2107151    }
 152
 153    public Task<TransportConnectionInformation> ConnectAsync(CancellationToken cancellationToken)
 780154    {
 155        lock (_mutex)
 780156        {
 780157            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 158
 780159            if (_connectTask is not null)
 1160            {
 1161                throw new InvalidOperationException("Cannot connect twice a Slic connection.");
 162            }
 779163            if (_isClosed)
 0164            {
 0165                throw new InvalidOperationException("Cannot connect a closed Slic connection.");
 166            }
 779167            _connectTask = PerformConnectAsync();
 779168        }
 779169        return _connectTask;
 170
 171        async Task<TransportConnectionInformation> PerformConnectAsync()
 779172        {
 779173            await Task.Yield(); // Exit mutex lock
 174
 175            // Connect the duplex connection.
 176            TransportConnectionInformation transportConnectionInformation;
 779177            TimeSpan peerIdleTimeout = TimeSpan.MaxValue;
 178
 179            try
 779180            {
 779181                transportConnectionInformation = await _duplexConnection.ConnectAsync(cancellationToken)
 779182                    .ConfigureAwait(false);
 183
 184                // Initialize the Slic connection.
 757185                if (IsServer)
 387186                {
 187                    // Read the Initialize frame.
 387188                    (ulong version, InitializeBody? initializeBody) = await ReadFrameAsync(
 387189                        DecodeInitialize,
 387190                        cancellationToken).ConfigureAwait(false);
 191
 379192                    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
 378222                    Debug.Assert(initializeBody is not null);
 223
 378224                    DecodeParameters(initializeBody.Value.Parameters);
 225
 226                    // Write back an InitializeAck frame.
 377227                    await WriteConnectionFrameAsync(
 377228                        FrameType.InitializeAck,
 377229                        new InitializeAckBody(EncodeParameters()).Encode,
 377230                        cancellationToken).ConfigureAwait(false);
 377231                }
 232                else
 370233                {
 234                    // Write the Initialize frame.
 370235                    await WriteConnectionFrameAsync(
 370236                        FrameType.Initialize,
 370237                        (ref SliceEncoder encoder) =>
 370238                        {
 370239                            encoder.EncodeVarUInt62(SlicDefinitions.V1);
 370240                            new InitializeBody(EncodeParameters()).Encode(ref encoder);
 370241                        },
 370242                        cancellationToken).ConfigureAwait(false);
 243
 244                    // Read and decode the InitializeAck or Version frame.
 370245                    (InitializeAckBody? initializeAckBody, VersionBody? versionBody) = await ReadFrameAsync(
 370246                        DecodeInitializeAckOrVersion,
 370247                        cancellationToken).ConfigureAwait(false);
 248
 346249                    Debug.Assert(initializeAckBody is not null || versionBody is not null);
 250
 346251                    if (initializeAckBody is not null)
 344252                    {
 344253                        DecodeParameters(initializeAckBody.Value.Parameters);
 344254                    }
 255
 346256                    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                    }
 344271                }
 721272            }
 9273            catch (InvalidDataException exception)
 9274            {
 9275                throw new IceRpcException(
 9276                    IceRpcError.IceRpcError,
 9277                    "The connection was aborted by a Slic protocol error.",
 9278                    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;
 721303            if (_localIdleTimeout == Timeout.InfiniteTimeSpan)
 2304            {
 2305                idleTimeout = _peerIdleTimeout;
 2306            }
 719307            else if (_peerIdleTimeout == Timeout.InfiniteTimeSpan)
 30308            {
 30309                idleTimeout = _localIdleTimeout;
 30310            }
 311            else
 689312            {
 689313                idleTimeout = _peerIdleTimeout < _localIdleTimeout ? _peerIdleTimeout : _localIdleTimeout;
 689314            }
 315
 721316            if (idleTimeout != Timeout.InfiniteTimeSpan)
 719317            {
 719318                _duplexConnection.Enable(idleTimeout);
 719319            }
 320
 721321            _readFramesTask = ReadFramesAsync(_disposedCts.Token);
 322
 721323            return transportConnectionInformation;
 721324        }
 325
 326        static (ulong, InitializeBody?) DecodeInitialize(FrameType? frameType, ReadOnlySequence<byte> buffer)
 381327        {
 381328            if (frameType != FrameType.Initialize)
 0329            {
 0330                throw new InvalidDataException($"Received unexpected {frameType} frame.");
 331            }
 332
 381333            return buffer.DecodeSliceBuffer<(ulong, InitializeBody?)>(
 381334                (ref SliceDecoder decoder) =>
 381335                {
 381336                    ulong version = decoder.DecodeVarUInt62();
 380337                    if (version == SlicDefinitions.V1)
 378338                    {
 378339                        return (version, new InitializeBody(ref decoder));
 381340                    }
 381341                    else
 2342                    {
 2343                        decoder.Skip((int)(buffer.Length - decoder.Consumed));
 2344                        return (version, null);
 381345                    }
 761346                });
 380347        }
 348
 349        static (InitializeAckBody?, VersionBody?) DecodeInitializeAckOrVersion(
 350            FrameType? frameType,
 351            ReadOnlySequence<byte> buffer) =>
 348352            frameType switch
 348353            {
 345354                FrameType.InitializeAck => (
 345355                    buffer.DecodeSliceBuffer((ref SliceDecoder decoder) => new InitializeAckBody(ref decoder)),
 345356                    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}'."),
 348361            };
 362
 363        async ValueTask<T> ReadFrameAsync<T>(
 364            Func<FrameType?, ReadOnlySequence<byte>, T> decodeFunc,
 365            CancellationToken cancellationToken)
 759366        {
 759367            (FrameType FrameType, int FrameSize, ulong?)? header =
 759368                await ReadFrameHeaderAsync(cancellationToken).ConfigureAwait(false);
 369
 370            ReadOnlySequence<byte> buffer;
 730371            if (header is null || header.Value.FrameSize == 0)
 4372            {
 4373                buffer = ReadOnlySequence<byte>.Empty;
 4374            }
 375            else
 726376            {
 726377                buffer = await _duplexConnectionReader.ReadAtLeastAsync(
 726378                    header.Value.FrameSize,
 726379                    cancellationToken).ConfigureAwait(false);
 726380                if (buffer.Length > header.Value.FrameSize)
 6381                {
 6382                    buffer = buffer.Slice(0, header.Value.FrameSize);
 6383                }
 726384            }
 385
 730386            T decodedFrame = decodeFunc(header?.FrameType, buffer);
 726387            _duplexConnectionReader.AdvanceTo(buffer.End);
 726388            return decodedFrame;
 726389        }
 779390    }
 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."))
 84406        {
 84407            using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 84408            {
 84409                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
 84416                {
 84417                    WriteFrame(FrameType.Close, streamId: null, new CloseBody((ulong)closeError).Encode);
 84418                    if (IsServer)
 45419                    {
 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).
 45424                        using var flushCts = CancellationTokenSource.CreateLinkedTokenSource(
 45425                            cancellationToken,
 45426                            _disposedCts.Token);
 427                        try
 45428                        {
 45429                            await _duplexConnectionWriter.FlushAsync(flushCts.Token).ConfigureAwait(false);
 44430                        }
 1431                        catch (OperationCanceledException) when (
 1432                            _disposedCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
 1433                        {
 1434                            throw new IceRpcException(IceRpcError.OperationAborted, "The connection was disposed.");
 435                        }
 44436                    }
 437                    else
 39438                    {
 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 shut down the connection on the client-side
 441                        // first to avoid TIME_WAIT states on the server-side.
 39442                        _duplexConnectionWriter.Shutdown();
 39443                        waitForWriterShutdown = true;
 39444                    }
 83445                }
 83446            }
 83447        }
 448
 108449        if (waitForWriterShutdown)
 39450        {
 39451            await _duplexConnectionWriter.WriterTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 39452        }
 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);
 106457    }
 458
 459    public async ValueTask<IMultiplexedStream> CreateStreamAsync(
 460        bool bidirectional,
 461        CancellationToken cancellationToken)
 2173462    {
 463        lock (_mutex)
 2173464        {
 2173465            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 466
 2170467            if (_connectTask is null || !_connectTask.IsCompletedSuccessfully)
 2468            {
 2469                throw new InvalidOperationException("Cannot create stream before connecting the Slic connection.");
 470            }
 2168471            if (_isClosed)
 8472            {
 8473                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 474            }
 475
 2160476            ++_streamSemaphoreWaitCount;
 2160477        }
 478
 479        try
 2160480        {
 2160481            using var createStreamCts = CancellationTokenSource.CreateLinkedTokenSource(
 2160482                _closedCancellationToken,
 2160483                cancellationToken);
 484
 2160485            SemaphoreSlim? streamCountSemaphore = bidirectional ?
 2160486                _bidirectionalStreamSemaphore :
 2160487                _unidirectionalStreamSemaphore;
 488
 2160489            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
 2159496            {
 2159497                await streamCountSemaphore.WaitAsync(createStreamCts.Token).ConfigureAwait(false);
 2146498            }
 499
 2146500            return new SlicStream(this, bidirectional, isRemote: false);
 501        }
 14502        catch (OperationCanceledException)
 14503        {
 14504            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
 2160510        {
 511            lock (_mutex)
 2160512            {
 2160513                --_streamSemaphoreWaitCount;
 2160514                if (_isClosed && _streamSemaphoreWaitCount == 0)
 7515                {
 7516                    _streamSemaphoreWaitClosed.SetResult();
 7517                }
 2160518            }
 2160519        }
 2146520    }
 521
 522    public ValueTask DisposeAsync()
 1091523    {
 524        lock (_mutex)
 1091525        {
 1091526            _disposeTask ??= PerformDisposeAsync();
 1091527        }
 1091528        return new(_disposeTask);
 529
 530        async Task PerformDisposeAsync()
 798531        {
 532            // Make sure we execute the code below without holding the mutex lock.
 798533            await Task.Yield();
 798534            TryClose(new IceRpcException(IceRpcError.OperationAborted), "The connection was disposed.");
 535
 798536            _disposedCts.Cancel();
 537
 538            try
 798539            {
 798540                await Task.WhenAll(
 798541                    _connectTask ?? Task.CompletedTask,
 798542                    _readFramesTask ?? Task.CompletedTask,
 798543                    _streamSemaphoreWaitClosed.Task).ConfigureAwait(false);
 442544            }
 356545            catch
 356546            {
 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.
 356549            }
 550
 551            // Clean-up the streams that might still be queued on the channel.
 821552            while (_acceptStreamChannel.Reader.TryRead(out IMultiplexedStream? stream))
 23553            {
 23554                if (stream.IsBidirectional)
 5555                {
 5556                    stream.Output.Complete();
 5557                    stream.Input.Complete();
 5558                }
 18559                else if (stream.IsRemote)
 18560                {
 18561                    stream.Input.Complete();
 18562                }
 563                else
 0564                {
 0565                    stream.Output.Complete();
 0566                }
 23567            }
 568
 569            try
 798570            {
 571                // Prevents unobserved task exceptions.
 798572                await _acceptStreamChannel.Reader.Completion.ConfigureAwait(false);
 0573            }
 798574            catch
 798575            {
 798576            }
 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.
 798582            await _writeSemaphore.WaitAsync(CancellationToken.None).ConfigureAwait(false);
 583
 798584            await _duplexConnectionWriter.DisposeAsync().ConfigureAwait(false);
 798585            _duplexConnectionReader.Dispose();
 798586            _duplexConnection.Dispose();
 587
 798588            _disposedCts.Dispose();
 798589            _bidirectionalStreamSemaphore?.Dispose();
 798590            _unidirectionalStreamSemaphore?.Dispose();
 798591            _closedCts.Dispose();
 798592        }
 1091593    }
 594
 799595    internal SlicConnection(
 799596        IDuplexConnection duplexConnection,
 799597        MultiplexedConnectionOptions options,
 799598        SlicTransportOptions slicOptions,
 799599        bool isServer)
 799600    {
 799601        IsServer = isServer;
 602
 799603        Pool = options.Pool;
 799604        MinSegmentSize = options.MinSegmentSize;
 799605        _maxBidirectionalStreams = options.MaxBidirectionalStreams;
 799606        _maxUnidirectionalStreams = options.MaxUnidirectionalStreams;
 607
 799608        InitialStreamWindowSize = slicOptions.InitialStreamWindowSize;
 799609        PauseWriterThreshold = slicOptions.PauseWriterThreshold;
 799610        _localIdleTimeout = slicOptions.IdleTimeout;
 799611        _maxOutstandingPongs = slicOptions.MaxOutstandingPongs;
 799612        _maxStreamFrameSize = slicOptions.MaxStreamFrameSize;
 613
 799614        _acceptStreamChannel = Channel.CreateUnbounded<IMultiplexedStream>(new UnboundedChannelOptions
 799615        {
 799616            SingleReader = true,
 799617            SingleWriter = true
 799618        });
 619
 799620        _closedCancellationToken = _closedCts.Token;
 621
 622        // Only the client-side sends pings to keep the connection alive when idle timeout (set later) is not infinite.
 799623        _duplexConnection = IsServer ?
 799624            new SlicDuplexConnectionDecorator(duplexConnection) :
 799625            new SlicDuplexConnectionDecorator(duplexConnection, SendReadPing, SendWritePing);
 626
 799627        _duplexConnectionReader = new DuplexConnectionReader(_duplexConnection, options.Pool, options.MinSegmentSize);
 799628        _duplexConnectionWriter = new SlicDuplexConnectionWriter(
 799629            _duplexConnection,
 799630            options.Pool,
 799631            options.MinSegmentSize,
 799632            PauseWriterThreshold);
 633
 634        // We use the same stream ID numbering scheme as QUIC.
 799635        if (IsServer)
 402636        {
 402637            _nextBidirectionalId = 1;
 402638            _nextUnidirectionalId = 3;
 402639        }
 640        else
 397641        {
 397642            _nextBidirectionalId = 0;
 397643            _nextUnidirectionalId = 2;
 397644        }
 645
 646        async Task SendPingAsync(long payload)
 14647        {
 648            try
 14649            {
 14650                await WriteConnectionFrameAsync(
 14651                    FrameType.Ping,
 14652                    new PingBody(payload).Encode,
 14653                    _closedCancellationToken).ConfigureAwait(false);
 13654            }
 0655            catch (IceRpcException)
 0656            {
 657                // Expected if the connection is closed.
 0658            }
 1659            catch (OperationCanceledException)
 1660            {
 661                // Expected if the connection is closed.
 1662            }
 0663            catch (Exception exception)
 0664            {
 0665                Debug.Fail($"The sending of a Ping frame failed with an unexpected exception: {exception}");
 0666            }
 14667        }
 668
 669        void SendReadPing()
 14670        {
 671            // No-op if there is already a pending Pong.
 14672            if (Interlocked.CompareExchange(ref _pendingPongCount, 1, 0) == 0)
 14673            {
 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.
 14676                _ = SendPingAsync(1L);
 14677            }
 14678        }
 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        }
 799689    }
 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) =>
 8758699        _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)
 4271705    {
 706        // Only a started stream has an Id and is registered in _streams.
 4271707        if (stream.IsStarted)
 4253708        {
 4253709            _streams.Remove(stream.Id, out SlicStream? _);
 4253710        }
 711
 4271712        if (stream.IsRemote)
 2130713        {
 2130714            if (stream.IsBidirectional)
 696715            {
 696716                Interlocked.Decrement(ref _bidirectionalStreamCount);
 696717            }
 718            else
 1434719            {
 1434720                Interlocked.Decrement(ref _unidirectionalStreamCount);
 1434721            }
 2130722        }
 2141723        else if (!_isClosed)
 1732724        {
 1732725            if (stream.IsBidirectional)
 620726            {
 620727                _bidirectionalStreamSemaphore!.Release();
 620728            }
 729            else
 1112730            {
 1112731                _unidirectionalStreamSemaphore!.Release();
 1112732            }
 1732733        }
 4271734    }
 735
 736    /// <summary>Throws the connection closure exception if the connection is already closed.</summary>
 737    internal void ThrowIfClosed()
 8012738    {
 739        lock (_mutex)
 8012740        {
 8012741            if (_isClosed)
 8742            {
 8743                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 744            }
 8004745        }
 8004746    }
 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)
 781756    {
 781757        Debug.Assert(frameType < FrameType.Stream);
 758
 781759        using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 778760        {
 761            lock (_mutex)
 778762            {
 778763                if (_isClosed)
 2764                {
 2765                    throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 766                }
 776767            }
 776768            WriteFrame(frameType, streamId: null, encode);
 776769            await _duplexConnectionWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
 776770        }
 776771    }
 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)
 3247788    {
 789        // Ensure that this method is called for any FrameType.StreamXxx frame type except FrameType.Stream.
 3247790        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.
 3247795        _ = WriteStreamFrameAsync();
 796
 797        async Task WriteStreamFrameAsync()
 3247798        {
 799            SemaphoreLock semaphoreLock;
 800            try
 3247801            {
 3247802                semaphoreLock = await _writeSemaphore.AcquireAsync(_closedCancellationToken).ConfigureAwait(false);
 3243803            }
 4804            catch (OperationCanceledException)
 4805            {
 806                // The connection was closed while waiting for the semaphore.
 4807                return;
 808            }
 809
 3243810            using (semaphoreLock)
 3243811            {
 812                lock (_mutex)
 3243813                {
 3243814                    if (_isClosed)
 0815                    {
 0816                        return;
 817                    }
 3243818                }
 819
 3243820                WriteFrame(frameType, stream.Id, encode);
 3243821                if (writeReadsClosedFrame)
 106822                {
 106823                    WriteFrame(FrameType.StreamReadsClosed, stream.Id, encode: null);
 106824                }
 3243825                if (frameType == FrameType.StreamLast)
 594826                {
 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.
 594829                    stream.WroteLastStreamFrame();
 594830                }
 831
 832                try
 3243833                {
 3243834                    await _duplexConnectionWriter.FlushAsync(_closedCancellationToken).ConfigureAwait(false);
 3240835                }
 2836                catch (OperationCanceledException)
 2837                {
 838                    // The connection was closed while flushing.
 2839                }
 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                }
 3243857            }
 3247858        }
 3247859    }
 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 last 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)
 7985879    {
 7985880        Debug.Assert(!source1.IsEmpty || endStream);
 881
 7985882        if (_connectTask is null)
 0883        {
 0884            throw new InvalidOperationException("Cannot send a stream frame before calling ConnectAsync.");
 885        }
 886
 7985887        using var writeCts = CancellationTokenSource.CreateLinkedTokenSource(
 7985888            _closedCancellationToken,
 7985889            cancellationToken);
 890
 891        try
 7985892        {
 893            do
 9303894            {
 895                // Next, ensure send credit is available. If not, this will block until the receiver allows sending
 896                // additional data.
 9303897                int sendCredit = 0;
 9303898                if (!source1.IsEmpty || !source2.IsEmpty)
 9298899                {
 9298900                    sendCredit = await stream.AcquireSendCreditAsync(writeCts.Token).ConfigureAwait(false);
 9232901                    Debug.Assert(sendCredit > 0);
 9232902                }
 903
 904                // Gather data from source1 or source2 up to sendCredit bytes or the peer maximum stream frame size.
 9237905                int sendMaxSize = Math.Min(sendCredit, PeerMaxStreamFrameSize);
 906                ReadOnlySequence<byte> sendSource1;
 907                ReadOnlySequence<byte> sendSource2;
 9237908                if (!source1.IsEmpty)
 8249909                {
 8249910                    int length = Math.Min((int)source1.Length, sendMaxSize);
 8249911                    sendSource1 = source1.Slice(0, length);
 8249912                    source1 = source1.Slice(length);
 8249913                }
 914                else
 988915                {
 988916                    sendSource1 = ReadOnlySequence<byte>.Empty;
 988917                }
 918
 9237919                if (source1.IsEmpty && !source2.IsEmpty)
 2042920                {
 2042921                    int length = Math.Min((int)source2.Length, sendMaxSize - (int)sendSource1.Length);
 2042922                    sendSource2 = source2.Slice(0, length);
 2042923                    source2 = source2.Slice(length);
 2042924                }
 925                else
 7195926                {
 7195927                    sendSource2 = ReadOnlySequence<byte>.Empty;
 7195928                }
 929
 930                // If there's no data left to send and endStream is true, it's the last stream frame.
 9237931                bool lastStreamFrame = endStream && source1.IsEmpty && source2.IsEmpty;
 932
 9237933                using (await _writeSemaphore.AcquireAsync(writeCts.Token).ConfigureAwait(false))
 8292934                {
 935                    lock (_mutex)
 8292936                    {
 8292937                        if (_isClosed)
 0938                        {
 0939                            throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 940                        }
 8292941                    }
 942
 8292943                    if (!stream.IsStarted)
 2123944                    {
 2123945                        if (stream.IsBidirectional)
 689946                        {
 689947                            AddStream(_nextBidirectionalId, stream);
 689948                            _nextBidirectionalId += 4;
 689949                        }
 950                        else
 1434951                        {
 1434952                            AddStream(_nextUnidirectionalId, stream);
 1434953                            _nextUnidirectionalId += 4;
 1434954                        }
 2123955                    }
 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.
 8292960                    if (sendCredit > 0)
 8287961                    {
 8287962                        stream.ConsumedSendCredit((int)(sendSource1.Length + sendSource2.Length));
 8287963                    }
 964
 8292965                    EncodeStreamFrameHeader(stream.Id, sendSource1.Length + sendSource2.Length, lastStreamFrame);
 966
 8292967                    if (lastStreamFrame)
 807968                    {
 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.
 807972                        stream.WroteLastStreamFrame();
 807973                    }
 974
 975                    // Write the stream frame.
 8292976                    if (!sendSource1.IsEmpty)
 8249977                    {
 8249978                        _duplexConnectionWriter.Write(sendSource1);
 8249979                    }
 8292980                    if (!sendSource2.IsEmpty)
 1097981                    {
 1097982                        _duplexConnectionWriter.Write(sendSource2);
 1097983                    }
 984
 8292985                    if (writeReadsClosedFrame && lastStreamFrame)
 382986                    {
 382987                        WriteFrame(FrameType.StreamReadsClosed, stream.Id, encode: null);
 382988                    }
 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.
 8292993                    await _duplexConnectionWriter.FlushAsync(writeCts.Token).ConfigureAwait(false);
 8281994                }
 8281995            }
 8281996            while (!source1.IsEmpty || !source2.IsEmpty); // Loop until there's no data left to send.
 6963997        }
 1022998        catch (OperationCanceledException)
 1022999        {
 10221000            cancellationToken.ThrowIfCancellationRequested();
 1001
 01002            Debug.Assert(_isClosed);
 01003            throw new IceRpcException(_peerCloseError ?? IceRpcError.OperationAborted, _closedMessage);
 1004        }
 1005
 69631006        return new FlushResult(isCanceled: false, isCompleted: false);
 1007
 1008        void EncodeStreamFrameHeader(ulong streamId, long size, bool lastStreamFrame)
 82921009        {
 82921010            var encoder = new SliceEncoder(_duplexConnectionWriter);
 82921011            encoder.EncodeFrameType(!lastStreamFrame ? FrameType.Stream : FrameType.StreamLast);
 82921012            Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(4);
 82921013            int startPos = encoder.EncodedByteCount;
 82921014            encoder.EncodeVarUInt62(streamId);
 82921015            SliceEncoder.EncodeVarUInt62((ulong)(encoder.EncodedByteCount - startPos + size), sizePlaceholder);
 82921016        }
 69631017    }
 1018
 1019    private void AddStream(ulong id, SlicStream stream)
 42531020    {
 1021        lock (_mutex)
 42531022        {
 42531023            if (_isClosed)
 01024            {
 01025                throw new IceRpcException(_peerCloseError ?? IceRpcError.ConnectionAborted, _closedMessage);
 1026            }
 1027
 42531028            _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.
 42531032            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.
 42531035            if (stream.IsRemote)
 21301036            {
 21301037                if (stream.IsBidirectional)
 6961038                {
 6961039                    _lastRemoteBidirectionalStreamId = id;
 6961040                }
 1041                else
 14341042                {
 14341043                    _lastRemoteUnidirectionalStreamId = id;
 14341044                }
 21301045            }
 42531046        }
 42531047    }
 1048
 1049    private void DecodeParameters(IDictionary<ParameterKey, IList<byte>> parameters)
 7221050    {
 7221051        int? maxStreamFrameSize = null;
 7221052        int? peerInitialStreamWindowSize = null;
 90791053        foreach ((ParameterKey key, IList<byte> buffer) in parameters)
 34571054        {
 34571055            switch (key)
 1056            {
 1057                case ParameterKey.MaxBidirectionalStreams:
 6341058                {
 6341059                    int value = DecodeParamValue(buffer);
 6341060                    if (value > 0)
 6341061                    {
 6341062                        _bidirectionalStreamSemaphore = new SemaphoreSlim(value, value);
 6341063                    }
 6341064                    break;
 1065                }
 1066                case ParameterKey.MaxUnidirectionalStreams:
 6911067                {
 6911068                    int value = DecodeParamValue(buffer);
 6911069                    if (value > 0)
 6911070                    {
 6911071                        _unidirectionalStreamSemaphore = new SemaphoreSlim(value, value);
 6911072                    }
 6911073                    break;
 1074                }
 1075                case ParameterKey.IdleTimeout:
 6891076                {
 6891077                    _peerIdleTimeout = TimeSpan.FromMilliseconds(DecodeParamValue(buffer));
 6891078                    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                    }
 6891083                    break;
 1084                }
 1085                case ParameterKey.MaxStreamFrameSize:
 7221086                {
 7221087                    maxStreamFrameSize = DecodeParamValue(buffer);
 7221088                    if (maxStreamFrameSize < 1024)
 01089                    {
 01090                        throw new InvalidDataException(
 01091                            "The MaxStreamFrameSize connection parameter is invalid, it must be at least 1 KB.");
 1092                    }
 7221093                    if (maxStreamFrameSize > SlicTransportOptions.MaxStreamFrameSizeCeiling)
 11094                    {
 11095                        throw new InvalidDataException(
 11096                            $"The MaxStreamFrameSize connection parameter is invalid, it cannot exceed {SlicTransportOpt
 1097                    }
 7211098                    break;
 1099                }
 1100                case ParameterKey.InitialStreamWindowSize:
 7211101                {
 7211102                    peerInitialStreamWindowSize = DecodeParamValue(buffer);
 7211103                    if (peerInitialStreamWindowSize < 1024)
 01104                    {
 01105                        throw new InvalidDataException(
 01106                            "The InitialStreamWindowSize connection parameter is invalid, it must be at least 1 KB.");
 1107                    }
 7211108                    break;
 1109                }
 1110                // Ignore unsupported parameter.
 1111            }
 34561112        }
 1113
 7211114        if (maxStreamFrameSize is null)
 01115        {
 01116            throw new InvalidDataException(
 01117                "The peer didn't send the required MaxStreamFrameSize connection parameter.");
 1118        }
 1119        else
 7211120        {
 7211121            PeerMaxStreamFrameSize = maxStreamFrameSize.Value;
 7211122        }
 1123
 7211124        if (peerInitialStreamWindowSize is null)
 01125        {
 01126            throw new InvalidDataException(
 01127                "The peer didn't send the required InitialStreamWindowSize connection parameter.");
 1128        }
 1129        else
 7211130        {
 7211131            PeerInitialStreamWindowSize = peerInitialStreamWindowSize.Value;
 7211132        }
 1133
 1134        // all parameter values are currently integers in the range 0..Int32Max encoded as varuint62.
 1135        static int DecodeParamValue(IList<byte> buffer)
 34571136        {
 1137            // The IList<byte> decoded by the IceRPC + Slice integration is backed by an array
 34571138            ulong value = new ReadOnlySequence<byte>((byte[])buffer).DecodeSliceBuffer(
 69141139                (ref SliceDecoder decoder) => decoder.DecodeVarUInt62());
 1140            try
 34571141            {
 34571142                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            }
 34571148        }
 7211149    }
 1150
 1151    private Dictionary<ParameterKey, IList<byte>> EncodeParameters()
 7471152    {
 7471153        var parameters = new List<KeyValuePair<ParameterKey, IList<byte>>>
 7471154        {
 7471155            // Required parameters.
 7471156            EncodeParameter(ParameterKey.MaxStreamFrameSize, (ulong)_maxStreamFrameSize),
 7471157            EncodeParameter(ParameterKey.InitialStreamWindowSize, (ulong)InitialStreamWindowSize)
 7471158        };
 1159
 1160        // Optional parameters.
 7471161        if (_localIdleTimeout != Timeout.InfiniteTimeSpan)
 7451162        {
 7451163            parameters.Add(EncodeParameter(ParameterKey.IdleTimeout, (ulong)_localIdleTimeout.TotalMilliseconds));
 7451164        }
 7471165        if (_maxBidirectionalStreams > 0)
 6821166        {
 6821167            parameters.Add(EncodeParameter(ParameterKey.MaxBidirectionalStreams, (ulong)_maxBidirectionalStreams));
 6821168        }
 7471169        if (_maxUnidirectionalStreams > 0)
 7471170        {
 7471171            parameters.Add(EncodeParameter(ParameterKey.MaxUnidirectionalStreams, (ulong)_maxUnidirectionalStreams));
 7471172        }
 1173
 7471174        return new Dictionary<ParameterKey, IList<byte>>(parameters);
 1175
 1176        static KeyValuePair<ParameterKey, IList<byte>> EncodeParameter(ParameterKey key, ulong value)
 36681177        {
 36681178            int sizeLength = SliceEncoder.GetVarUInt62EncodedSize(value);
 36681179            byte[] buffer = new byte[sizeLength];
 36681180            SliceEncoder.EncodeVarUInt62(value, buffer);
 36681181            return new(key, buffer);
 36681182        }
 7471183    }
 1184
 1185    private bool IsUnknownStream(ulong streamId)
 52681186    {
 52681187        bool isRemote = streamId % 2 == (IsServer ? 0ul : 1ul);
 52681188        bool isBidirectional = streamId % 4 < 2;
 52681189        if (isRemote)
 28341190        {
 28341191            if (isBidirectional)
 13701192            {
 13701193                return _lastRemoteBidirectionalStreamId is null || streamId > _lastRemoteBidirectionalStreamId;
 1194            }
 1195            else
 14641196            {
 14641197                return _lastRemoteUnidirectionalStreamId is null || streamId > _lastRemoteUnidirectionalStreamId;
 1198            }
 1199        }
 1200        else
 24341201        {
 24341202            if (isBidirectional)
 12741203            {
 12741204                return streamId >= _nextBidirectionalId;
 1205            }
 1206            else
 11601207            {
 11601208                return streamId >= _nextUnidirectionalId;
 1209            }
 1210        }
 52681211    }
 1212
 1213    private Task ReadFrameAsync(FrameType frameType, int size, ulong? streamId, CancellationToken cancellationToken)
 120491214    {
 120491215        if (frameType >= FrameType.Stream && streamId is null)
 01216        {
 01217            throw new InvalidDataException("Received stream frame without stream ID.");
 1218        }
 1219
 120491220        switch (frameType)
 1221        {
 1222            case FrameType.Close:
 841223            {
 841224                return ReadCloseFrameAsync(size, cancellationToken);
 1225            }
 1226            case FrameType.Ping:
 211227            {
 211228                return ReadPingFrameAndWritePongFrameAsync(size, cancellationToken);
 1229            }
 1230            case FrameType.Pong:
 161231            {
 161232                return ReadPongFrameAsync(size, cancellationToken);
 1233            }
 1234            case FrameType.Stream:
 1235            case FrameType.StreamLast:
 88431236            {
 88431237                return ReadStreamDataFrameAsync(frameType, size, streamId!.Value, cancellationToken);
 1238            }
 1239            case FrameType.StreamWindowUpdate:
 12461240            {
 12461241                if (IsUnknownStream(streamId!.Value))
 11242                {
 11243                    throw new InvalidDataException($"Received {frameType} frame for unknown stream.");
 1244                }
 1245
 12451246                return ReadStreamWindowUpdateFrameAsync(size, streamId.Value, cancellationToken);
 1247            }
 1248            case FrameType.StreamReadsClosed:
 1249            case FrameType.StreamWritesClosed:
 18361250            {
 18361251                if (size > 0)
 21252                {
 21253                    throw new InvalidDataException($"Unexpected body for {frameType} frame.");
 1254                }
 18341255                if (IsUnknownStream(streamId!.Value))
 21256                {
 21257                    throw new InvalidDataException($"Received {frameType} frame for unknown stream.");
 1258                }
 1259
 18321260                if (_streams.TryGetValue(streamId.Value, out SlicStream? stream))
 17721261                {
 17721262                    if (frameType == FrameType.StreamWritesClosed)
 491263                    {
 491264                        stream.ReceivedWritesClosedFrame();
 491265                    }
 1266                    else
 17231267                    {
 17231268                        stream.ReceivedReadsClosedFrame();
 17231269                    }
 17721270                }
 18321271                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)
 841280        {
 841281            CloseBody closeBody = await ReadFrameBodyAsync(
 841282                FrameType.Close,
 841283                size,
 831284                (ref SliceDecoder decoder) => new CloseBody(ref decoder),
 841285                cancellationToken).ConfigureAwait(false);
 1286
 821287            IceRpcError? peerCloseError = closeBody.ApplicationErrorCode switch
 821288            {
 601289                (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
 821294            };
 1295
 1296            bool notAlreadyClosed;
 821297            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
 771305            {
 771306                notAlreadyClosed = TryClose(
 771307                    new IceRpcException(peerCloseError.Value),
 771308                    "The connection was closed by the peer.",
 771309                    peerCloseError);
 771310            }
 1311
 1312            // The server-side of the duplex connection is only shut down once the client-side is shut down. When using
 1313            // TCP, this ensures that the server TCP connection won't end-up in the TIME_WAIT state on the server-side.
 821314            if (notAlreadyClosed && !IsServer)
 331315            {
 1316                // DisposeAsync waits for the reads frames task to complete before disposing the writer.
 1317                // _writeSemaphore alone serializes access to the writer.
 331318                using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 331319                {
 331320                    _duplexConnectionWriter.Shutdown();
 331321                }
 331322                await _duplexConnectionWriter.WriterTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 331323            }
 821324        }
 1325
 1326        async Task ReadPingFrameAndWritePongFrameAsync(int size, CancellationToken cancellationToken)
 211327        {
 1328            // Read the ping frame.
 211329            PingBody pingBody = await ReadFrameBodyAsync(
 211330                FrameType.Ping,
 211331                size,
 201332                (ref SliceDecoder decoder) => new PingBody(ref decoder),
 211333                cancellationToken).ConfigureAwait(false);
 1334
 191335            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.
 181345            _ = WritePongFrameAsync(pingBody.Payload);
 181346        }
 1347
 1348        async Task WritePongFrameAsync(long payload)
 181349        {
 1350            try
 181351            {
 181352                await WriteConnectionFrameAsync(
 181353                    FrameType.Pong,
 181354                    new PongBody(payload).Encode,
 181355                    _closedCancellationToken).ConfigureAwait(false);
 141356            }
 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
 181374            {
 181375                Interlocked.Decrement(ref _outstandingPongCount);
 181376            }
 181377        }
 1378
 1379        async Task ReadPongFrameAsync(int size, CancellationToken cancellationToken)
 161380        {
 161381            if (Interlocked.Decrement(ref _pendingPongCount) >= 0)
 131382            {
 1383                // Ensure the pong frame payload value is expected.
 1384
 131385                PongBody pongBody = await ReadFrameBodyAsync(
 131386                    FrameType.Pong,
 131387                    size,
 131388                    (ref SliceDecoder decoder) => new PongBody(ref decoder),
 131389                    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").
 131392                if (pongBody.Payload != 0L && pongBody.Payload != 1L)
 01393                {
 01394                    throw new InvalidDataException($"Received {nameof(FrameType.Pong)} with unexpected payload.");
 1395                }
 131396            }
 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            }
 131402        }
 1403
 1404        async Task ReadStreamWindowUpdateFrameAsync(int size, ulong streamId, CancellationToken cancellationToken)
 12451405        {
 12451406            StreamWindowUpdateBody frame = await ReadFrameBodyAsync(
 12451407                FrameType.StreamWindowUpdate,
 12451408                size,
 12451409                (ref SliceDecoder decoder) => new StreamWindowUpdateBody(ref decoder),
 12451410                cancellationToken).ConfigureAwait(false);
 12451411            if (_streams.TryGetValue(streamId, out SlicStream? stream))
 11991412            {
 11991413                if (stream.IsRemote && !stream.IsBidirectional)
 11414                {
 1415                    // The local side doesn't write on a remote unidirectional stream, so there is no window to update.
 11416                    throw new InvalidDataException(
 11417                        $"Received unexpected {nameof(FrameType.StreamWindowUpdate)} frame on remote unidirectional stre
 1418                }
 11981419                stream.ReceivedWindowUpdateFrame(frame);
 11961420            }
 12421421        }
 1422
 1423        async Task<T> ReadFrameBodyAsync<T>(
 1424            FrameType frameType,
 1425            int size,
 1426            DecodeFunc<T> decodeFunc,
 1427            CancellationToken cancellationToken)
 13631428        {
 13631429            if (size <= 0)
 21430            {
 21431                throw new InvalidDataException($"Unexpected empty body for {frameType} frame.");
 1432            }
 1433
 13611434            ReadOnlySequence<byte> buffer = await _duplexConnectionReader.ReadAtLeastAsync(size, cancellationToken)
 13611435                .ConfigureAwait(false);
 1436
 13611437            if (buffer.Length > size)
 9411438            {
 9411439                buffer = buffer.Slice(0, size);
 9411440            }
 1441
 13611442            T decodedFrame = buffer.DecodeSliceBuffer(decodeFunc);
 13591443            _duplexConnectionReader.AdvanceTo(buffer.End);
 13591444            return decodedFrame;
 13591445        }
 120411446    }
 1447
 1448    private async ValueTask<(FrameType FrameType, int FrameSize, ulong? StreamId)?> ReadFrameHeaderAsync(
 1449        CancellationToken cancellationToken)
 135051450    {
 135051451        while (true)
 135051452        {
 1453            // Read data from the pipe reader.
 135051454            if (!_duplexConnectionReader.TryRead(out ReadOnlySequence<byte> buffer))
 85431455            {
 85431456                buffer = await _duplexConnectionReader.ReadAsync(cancellationToken).ConfigureAwait(false);
 79631457            }
 1458
 129251459            if (buffer.IsEmpty)
 1401460            {
 1401461                return null;
 1462            }
 1463
 127851464            if (TryDecodeHeader(
 127851465                buffer,
 127851466                out (FrameType FrameType, int FrameSize, ulong? StreamId) header,
 127851467                out int consumed))
 127781468            {
 127781469                _duplexConnectionReader.AdvanceTo(buffer.GetPosition(consumed));
 127781470                return header;
 1471            }
 1472            else
 01473            {
 01474                _duplexConnectionReader.AdvanceTo(buffer.Start, buffer.End);
 01475            }
 01476        }
 1477
 1478        static bool TryDecodeHeader(
 1479            ReadOnlySequence<byte> buffer,
 1480            out (FrameType FrameType, int FrameSize, ulong? StreamId) header,
 1481            out int consumed)
 127851482        {
 127851483            header = default;
 127851484            consumed = default;
 1485
 127851486            var decoder = new SliceDecoder(buffer);
 1487
 1488            // Decode the frame type and frame size.
 127851489            if (!decoder.TryDecodeUInt8(out byte frameType) || !decoder.TryDecodeVarUInt62(out ulong frameSize))
 01490            {
 01491                return false;
 1492            }
 1493
 127851494            header.FrameType = frameType.AsFrameType();
 1495            try
 127821496            {
 127821497                header.FrameSize = checked((int)frameSize);
 127821498            }
 01499            catch (OverflowException exception)
 01500            {
 01501                throw new InvalidDataException("The frame size can't be larger than int.MaxValue.", exception);
 1502            }
 1503
 1504            // Reject oversized control frame bodies before any buffering occurs. Only the stream data frames are
 1505            // exempt: their size is bounded by the stream's flow control window.
 127821506            if (header.FrameType is not (FrameType.Stream or FrameType.StreamLast) &&
 127821507                header.FrameSize > MaxControlFrameBodySize)
 21508            {
 21509                throw new InvalidDataException(
 21510                    $"The {header.FrameType} frame body size ({header.FrameSize}) exceeds the maximum allowed size ({Max
 1511            }
 1512
 1513            // If it's a stream frame, try to decode the stream ID
 127801514            if (header.FrameType >= FrameType.Stream)
 119271515            {
 119271516                if (header.FrameSize == 0)
 11517                {
 11518                    throw new InvalidDataException("Invalid stream frame size.");
 1519                }
 1520
 119261521                consumed = (int)decoder.Consumed;
 119261522                if (!decoder.TryDecodeVarUInt62(out ulong streamId))
 01523                {
 01524                    return false;
 1525                }
 119261526                header.StreamId = streamId;
 119261527                header.FrameSize -= (int)decoder.Consumed - consumed;
 1528
 119261529                if (header.FrameSize < 0)
 11530                {
 11531                    throw new InvalidDataException("Invalid stream frame size.");
 1532                }
 119251533            }
 1534
 127781535            consumed = (int)decoder.Consumed;
 127781536            return true;
 127781537        }
 129181538    }
 1539
 1540    private async Task ReadFramesAsync(CancellationToken cancellationToken)
 7211541    {
 1542        try
 7211543        {
 127461544            while (true)
 127461545            {
 127461546                (FrameType Type, int Size, ulong? StreamId)? header = await ReadFrameHeaderAsync(cancellationToken)
 127461547                    .ConfigureAwait(false);
 1548
 121881549                if (header is null)
 1391550                {
 1551                    lock (_mutex)
 1391552                    {
 1391553                        if (!_isClosed)
 01554                        {
 1555                            // Unexpected duplex connection shutdown.
 01556                            throw new IceRpcException(IceRpcError.ConnectionAborted);
 1557                        }
 1391558                    }
 1559                    // The peer has shut down the duplex connection.
 1391560                    break;
 1561                }
 1562
 120491563                await ReadFrameAsync(header.Value.Type, header.Value.Size, header.Value.StreamId, cancellationToken)
 120491564                    .ConfigureAwait(false);
 120251565            }
 1566
 1391567            if (IsServer)
 721568            {
 721569                Debug.Assert(_isClosed);
 1570
 1571                // The server-side of the duplex connection is only shut down once the client-side is shut down. When
 1572                // using TCP, this ensures that the server TCP connection won't end-up in the TIME_WAIT state on the
 1573                // server-side.
 1574
 1575                // DisposeAsync waits for the reads frames task to complete before disposing the writer.
 1576                // _writeSemaphore alone serializes access to the writer and guards _writerIsShutdown.
 721577                using (await _writeSemaphore.AcquireAsync(cancellationToken).ConfigureAwait(false))
 721578                {
 721579                    _duplexConnectionWriter.Shutdown();
 1580
 1581                    // Make sure that CloseAsync doesn't call Write on the writer if it's called shortly after the peer
 1582                    // shutdown its side of the connection (which triggers ReadFrameHeaderAsync to return null).
 721583                    _writerIsShutdown = true;
 721584                }
 1585
 721586                await _duplexConnectionWriter.WriterTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 711587            }
 1381588        }
 2841589        catch (OperationCanceledException)
 2841590        {
 1591            // Expected, DisposeAsync was called.
 2841592        }
 2741593        catch (IceRpcException exception)
 2741594        {
 2741595            TryClose(exception, "The connection was lost.", IceRpcError.ConnectionAborted);
 2741596            throw;
 1597        }
 251598        catch (InvalidDataException exception)
 251599        {
 251600            var rpcException = new IceRpcException(
 251601                IceRpcError.IceRpcError,
 251602                "The connection was aborted by a Slic protocol error.",
 251603                exception);
 251604            TryClose(rpcException, rpcException.Message, IceRpcError.IceRpcError);
 251605            throw rpcException;
 1606        }
 01607        catch (Exception exception)
 01608        {
 01609            Debug.Fail($"The read frames task completed due to an unhandled exception: {exception}");
 01610            TryClose(exception, "The connection was lost.", IceRpcError.ConnectionAborted);
 01611            throw;
 1612        }
 4221613    }
 1614
 1615    private async Task ReadStreamDataFrameAsync(
 1616        FrameType type,
 1617        int size,
 1618        ulong streamId,
 1619        CancellationToken cancellationToken)
 88431620    {
 88431621        bool endStream = type == FrameType.StreamLast;
 88431622        bool isRemote = streamId % 2 == (IsServer ? 0ul : 1ul);
 88431623        bool isBidirectional = streamId % 4 < 2;
 1624
 88431625        if (!isBidirectional && !isRemote)
 01626        {
 01627            throw new InvalidDataException(
 01628                "Received unexpected stream frame on local unidirectional stream.");
 1629        }
 88431630        else if (size == 0 && !endStream)
 11631        {
 11632            throw new InvalidDataException($"Received invalid {nameof(FrameType.Stream)} frame.");
 1633        }
 88421634        else if (size > _maxStreamFrameSize)
 11635        {
 11636            throw new InvalidDataException(
 11637                $"Received stream frame with size {size} exceeding the advertised maximum of {_maxStreamFrameSize} bytes
 1638        }
 1639
 88411640        if (!_streams.TryGetValue(streamId, out SlicStream? stream) && isRemote && IsUnknownStream(streamId))
 21321641        {
 1642            // Create a new remote stream.
 1643
 21321644            if (size == 0)
 01645            {
 01646                throw new InvalidDataException("Received empty stream frame on new stream.");
 1647            }
 1648
 21321649            if (isBidirectional)
 6971650            {
 6971651                ulong expectedStreamId = _lastRemoteBidirectionalStreamId is ulong lastId
 6971652                    ? lastId + 4
 6971653                    : (IsServer ? 0ul : 1ul);
 6971654                if (streamId != expectedStreamId)
 11655                {
 11656                    throw new InvalidDataException("Invalid stream ID.");
 1657                }
 1658
 6961659                if (_bidirectionalStreamCount == _maxBidirectionalStreams)
 01660                {
 01661                    throw new IceRpcException(
 01662                        IceRpcError.IceRpcError,
 01663                        $"The maximum bidirectional stream count {_maxBidirectionalStreams} was reached.");
 1664                }
 6961665                Interlocked.Increment(ref _bidirectionalStreamCount);
 6961666            }
 1667            else
 14351668            {
 14351669                ulong expectedStreamId = _lastRemoteUnidirectionalStreamId is ulong lastId
 14351670                    ? lastId + 4
 14351671                    : (IsServer ? 2ul : 3ul);
 14351672                if (streamId != expectedStreamId)
 11673                {
 11674                    throw new InvalidDataException("Invalid stream ID.");
 1675                }
 1676
 14341677                if (_unidirectionalStreamCount == _maxUnidirectionalStreams)
 01678                {
 01679                    throw new IceRpcException(
 01680                        IceRpcError.IceRpcError,
 01681                        $"The maximum unidirectional stream count {_maxUnidirectionalStreams} was reached.");
 1682                }
 14341683                Interlocked.Increment(ref _unidirectionalStreamCount);
 14341684            }
 1685
 1686            // The stream is registered with the connection and queued on the channel. The caller of AcceptStreamAsync
 1687            // is responsible for cleaning up the stream.
 21301688            stream = new SlicStream(this, isBidirectional, isRemote: true);
 1689
 1690            try
 21301691            {
 21301692                AddStream(streamId, stream);
 1693
 1694                try
 21301695                {
 21301696                    await _acceptStreamChannel.Writer.WriteAsync(
 21301697                        stream,
 21301698                        cancellationToken).ConfigureAwait(false);
 21301699                }
 01700                catch (ChannelClosedException exception)
 01701                {
 1702                    // The exception given to ChannelWriter.Complete(Exception? exception) is the InnerException.
 01703                    Debug.Assert(exception.InnerException is not null);
 01704                    throw ExceptionUtil.Throw(exception.InnerException);
 1705                }
 21301706            }
 01707            catch (IceRpcException)
 01708            {
 1709                // The two methods above throw IceRpcException if the connection has been closed (either by CloseAsync
 1710                // or because the close frame was received). We cleanup up the stream but don't throw to not abort the
 1711                // reading. The connection graceful closure still needs to read on the connection to figure out when the
 1712                // peer shuts down the duplex connection.
 01713                Debug.Assert(_isClosed);
 01714                stream.Input.Complete();
 01715                if (isBidirectional)
 01716                {
 01717                    stream.Output.Complete();
 01718                }
 01719            }
 21301720        }
 1721
 88391722        bool isDataConsumed = false;
 88391723        if (stream is not null)
 87751724        {
 1725            // Let the stream consume the stream frame data.
 87751726            isDataConsumed = await stream.ReceivedDataFrameAsync(
 87751727                size,
 87751728                endStream,
 87751729                cancellationToken).ConfigureAwait(false);
 87751730        }
 1731
 88391732        if (!isDataConsumed)
 811733        {
 1734            // The stream (if any) didn't consume the data. Read and ignore the data using a helper pipe.
 811735            var pipe = new Pipe(
 811736                new PipeOptions(
 811737                    pool: Pool,
 811738                    pauseWriterThreshold: 0,
 811739                    minimumSegmentSize: MinSegmentSize,
 811740                    useSynchronizationContext: false));
 1741
 811742            await _duplexConnectionReader.FillBufferWriterAsync(
 811743                    pipe.Writer,
 811744                    size,
 811745                    cancellationToken).ConfigureAwait(false);
 1746
 801747            pipe.Writer.Complete();
 801748            pipe.Reader.Complete();
 801749        }
 88381750    }
 1751
 1752    private bool TryClose(Exception exception, string closeMessage, IceRpcError? peerCloseError = null)
 12891753    {
 1754        lock (_mutex)
 12891755        {
 12891756            if (_isClosed)
 4911757            {
 4911758                return false;
 1759            }
 7981760            _isClosed = true;
 7981761            _closedMessage = closeMessage;
 7981762            _peerCloseError = peerCloseError;
 7981763            if (_streamSemaphoreWaitCount == 0)
 7911764            {
 7911765                _streamSemaphoreWaitClosed.SetResult();
 7911766            }
 7981767        }
 1768
 1769        // Cancel pending CreateStreamAsync, AcceptStreamAsync and WriteStreamDataFrameAsync operations.
 7981770        _closedCts.Cancel();
 7981771        _acceptStreamChannel.Writer.TryComplete(exception);
 1772
 1773        // Close streams.
 38881774        foreach (SlicStream stream in _streams.Values)
 7471775        {
 7471776            stream.Close(exception);
 7471777        }
 1778
 7981779        return true;
 12891780    }
 1781
 1782    private void WriteFrame(FrameType frameType, ulong? streamId, EncodeAction? encode)
 45911783    {
 45911784        var encoder = new SliceEncoder(_duplexConnectionWriter);
 45911785        encoder.EncodeFrameType(frameType);
 1786        // 2 bytes is sufficient: control frame bodies are limited to MaxControlFrameBodySize (16,383) and the
 1787        // stream frames encoded by WriteFrame carry at most a stream ID + a small body (e.g., StreamWindowUpdate).
 45911788        Span<byte> sizePlaceholder = encoder.GetPlaceholderSpan(2);
 45911789        int startPos = encoder.EncodedByteCount;
 45911790        if (streamId is not null)
 37311791        {
 37311792            encoder.EncodeVarUInt62(streamId.Value);
 37311793        }
 45911794        encode?.Invoke(ref encoder);
 45911795        SliceEncoder.EncodeVarUInt62((ulong)(encoder.EncodedByteCount - startPos), sizePlaceholder);
 45911796    }
 1797}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
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)