< Summary

Information
Class: IceRpc.Transports.Slic.Internal.SlicStream
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Transports/Slic/Internal/SlicStream.cs
Tag: 1986_28452893481
Line coverage
92%
Covered lines: 225
Uncovered lines: 18
Coverable lines: 243
Total lines: 475
Line coverage: 92.5%
Branch coverage
92%
Covered branches: 76
Total branches: 82
Branch coverage: 92.6%
Method coverage
100%
Covered methods: 29
Fully covered methods: 24
Total methods: 29
Method coverage: 100%
Full method coverage: 82.7%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_Id()50%2271.42%
set_Id(...)100%11100%
get_Input()50%22100%
get_IsBidirectional()100%11100%
get_IsRemote()100%11100%
get_IsStarted()100%11100%
get_Output()50%22100%
get_WritesClosed()100%11100%
get_WindowUpdateThreshold()100%11100%
.ctor(...)100%1212100%
AcquireSendCreditAsync(...)100%11100%
Close(...)100%44100%
CloseReads(...)100%242492.5%
CloseWrites(...)100%151486.04%
ConsumedSendCredit(...)100%11100%
FillBufferWriterAsync(...)100%11100%
ReceivedDataFrameAsync(...)100%66100%
ReceivedReadsClosedFrame()50%22100%
ReceivedWindowUpdateFrame(...)50%3250%
ReceivedWritesClosedFrame()50%22100%
WindowUpdate(...)100%1172.72%
WriteStreamFrameAsync(...)100%22100%
WriteLastStreamFrameAsync()100%11100%
WroteLastStreamFrame()100%22100%
ThrowIfConnectionClosed()100%11100%
TrySetReadsClosed()100%11100%
TrySetWritesClosed()100%22100%
TrySetState(...)100%44100%
WriteStreamFrame(...)100%11100%

File(s)

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

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Transports.Internal;
 4using System.Buffers;
 5using System.Diagnostics;
 6using System.IO.Pipelines;
 7using ZeroC.Slice.Codec;
 8
 9namespace IceRpc.Transports.Slic.Internal;
 10
 11/// <summary>The stream implementation for Slic.</summary>
 12/// <remarks>The stream implementation implements flow control to ensure data isn't buffered indefinitely if the
 13/// application doesn't consume it.</remarks>
 14internal class SlicStream : IMultiplexedStream
 15{
 16    public ulong Id
 17    {
 18        get
 1772019        {
 1772020            ulong id = Volatile.Read(ref _id);
 1772021            if (id == ulong.MaxValue)
 022            {
 023                throw new InvalidOperationException("The stream ID isn't allocated yet.");
 24            }
 1772025            return id;
 1772026        }
 27
 28        set
 420929        {
 420930            Debug.Assert(_id == ulong.MaxValue);
 420931            Volatile.Write(ref _id, value);
 420932        }
 33    }
 34
 35    public PipeReader Input =>
 1391636        _inputPipeReader ?? throw new InvalidOperationException("A local unidirectional stream has no Input.");
 37
 38    /// <inheritdoc/>
 3566739    public bool IsBidirectional { get; }
 40
 41    /// <inheritdoc/>
 3237942    public bool IsRemote { get; }
 43
 44    /// <inheritdoc/>
 2320445    public bool IsStarted => Volatile.Read(ref _id) != ulong.MaxValue;
 46
 47    public PipeWriter Output =>
 912348        _outputPipeWriter ?? throw new InvalidOperationException("A remote unidirectional stream has no Output.");
 49
 227850    public Task WritesClosed => _writesClosedTcs.Task;
 51
 744752    internal int WindowUpdateThreshold => _connection.StreamWindowUpdateThreshold;
 53
 54    private bool _closeReadsOnWritesClosure;
 55    private readonly SlicConnection _connection;
 423056    private ulong _id = ulong.MaxValue;
 57    private readonly SlicPipeReader? _inputPipeReader;
 58    // This mutex protects _writesClosePending, _closeReadsOnWritesClosure.
 423059    private readonly Lock _mutex = new();
 60    private readonly SlicPipeWriter? _outputPipeWriter;
 61    // FlagEnumExtensions operations are used to update the state. These operations are atomic and don't require mutex
 62    // locking.
 63    private int _state;
 423064    private readonly TaskCompletionSource _writesClosedTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
 65    private bool _writesClosePending;
 66
 423067    internal SlicStream(SlicConnection connection, bool isBidirectional, bool isRemote)
 423068    {
 423069        _connection = connection;
 70
 423071        IsBidirectional = isBidirectional;
 423072        IsRemote = isRemote;
 73
 423074        if (!IsBidirectional)
 285075        {
 285076            if (IsRemote)
 142077            {
 78                // Write-side of remote unidirectional stream is marked as closed.
 142079                TrySetWritesClosed();
 142080            }
 81            else
 143082            {
 83                // Read-side of local unidirectional stream is marked as closed.
 143084                TrySetReadsClosed();
 143085            }
 285086        }
 87
 423088        if (IsRemote || IsBidirectional)
 280089        {
 280090            _inputPipeReader = new SlicPipeReader(this, _connection);
 280091        }
 92
 423093        if (!IsRemote || IsBidirectional)
 281094        {
 281095            _outputPipeWriter = new SlicPipeWriter(this, _connection);
 281096        }
 423097    }
 98
 99    /// <summary>Acquires send credit.</summary>
 100    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 101    /// <returns>The available send credit.</returns>
 102    /// <remarks>This method should be called before sending a <see cref="FrameType.Stream"/> or <see
 103    /// cref="FrameType.StreamLast"/> frame to ensure enough send credit is available. If no send credit is available,
 104    /// it will block until send credit is available. The send credit matches the size of the peer's flow-control
 105    /// window.</remarks>
 106    internal ValueTask<int> AcquireSendCreditAsync(CancellationToken cancellationToken) =>
 9234107        _outputPipeWriter!.AcquireSendCreditAsync(cancellationToken);
 108
 109    /// <summary>Closes the read and write sides of the stream and notifies the stream <see cref="Input" /> and <see
 110    /// cref="Output" /> of the reads and writes closure.</summary>
 111    internal void Close(Exception closeException)
 671112    {
 671113        if (TrySetReadsClosed())
 326114        {
 326115            Debug.Assert(_inputPipeReader is not null);
 326116            _inputPipeReader.CompleteReads(closeException);
 326117        }
 671118        if (TrySetWritesClosed())
 383119        {
 383120            Debug.Assert(_outputPipeWriter is not null);
 383121            _outputPipeWriter.CompleteWrites(closeException);
 383122        }
 671123    }
 124
 125    /// <summary>Closes the read-side of the stream. It's only called by <see cref="SlicPipeReader.Complete" />, <see
 126    /// cref="SlicPipeReader.TryRead" /> or <see cref="SlicPipeReader.ReadAsync" /> and never called concurrently.
 127    /// </summary>
 128    /// <param name="graceful"><see langword="true" /> if the application consumed all the stream data from the stream
 129    /// <see cref="Input" />; otherwise, <see langword="false" />.</param>
 130    internal void CloseReads(bool graceful)
 4706131    {
 4706132        bool writeReadsClosedFrame = false;
 133
 134        lock (_mutex)
 4706135        {
 4706136            if (IsStarted && !_state.HasFlag(State.ReadsClosed))
 2415137            {
 138                // As an optimization, if reads are gracefully closed once the buffered data is consumed but before
 139                // writes are closed, we don't send the StreamReadsClosed frame just yet. Instead, when writes are
 140                // closed, CloseWrites will bundle the sending of the StreamReadsClosed with the sending of the
 141                // StreamLast or StreamWritesClosed frame. This allows to send both frames with a single write on the
 142                // duplex connection.
 2415143                if (graceful &&
 2415144                    IsBidirectional &&
 2415145                    IsRemote &&
 2415146                    !_state.HasFlag(State.WritesClosed) &&
 2415147                    !_writesClosePending)
 488148                {
 488149                    _closeReadsOnWritesClosure = true;
 488150                }
 1927151                else if (!graceful || IsRemote)
 1360152                {
 153                    // If forcefully closed because the input was completed before the data was fully read or if writes
 154                    // are already closed and the stream is a remote stream, we send the StreamReadsClosed frame to
 155                    // notify the peer that reads are closed.
 1360156                    writeReadsClosedFrame = true;
 1360157                }
 2415158            }
 4706159        }
 160
 4706161        if (writeReadsClosedFrame)
 1360162        {
 1360163            if (IsRemote)
 1286164            {
 165                // If it's a remote stream, we close writes before sending the StreamReadsClosed frame to ensure
 166                // _connection._bidirectionalStreamCount or _connection._unidirectionalStreamCount is decreased before
 167                // the peer receives the frame. This is necessary to prevent a race condition where the peer could
 168                // release the connection's bidirectional or unidirectional stream semaphore before this connection's
 169                // stream count is actually decreased.
 1286170                TrySetReadsClosed();
 1286171            }
 172
 173            try
 1360174            {
 1360175                WriteStreamFrame(FrameType.StreamReadsClosed, encode: null, writeReadsClosedFrame: false);
 1360176            }
 0177            catch (IceRpcException)
 0178            {
 179                // Ignore connection failures.
 0180            }
 181
 1360182            if (!IsRemote)
 74183            {
 184                // We can now close reads to permit a new stream to be started. The peer will receive the
 185                // StreamReadsClosed frame before the new stream sends a Stream frame.
 74186                TrySetReadsClosed();
 74187            }
 1360188        }
 189        else
 3346190        {
 3346191            TrySetReadsClosed();
 3346192        }
 4706193    }
 194
 195    /// <summary>Closes the write-side of the stream. It's only called by <see cref="SlicPipeWriter.Complete" /> and
 196    /// never called concurrently.</summary>
 197    /// <param name="graceful"><see langword="true" /> if the application wrote all the stream data on the stream <see
 198    /// cref="Output" />; otherwise, <see langword="false" />.</param>
 199    internal void CloseWrites(bool graceful)
 2779200    {
 2779201        bool writeWritesClosedFrame = false;
 2779202        bool writeReadsClosedFrame = false;
 203
 204        lock (_mutex)
 2779205        {
 2779206            if (IsStarted && !_state.HasFlag(State.WritesClosed) && !_writesClosePending)
 637207            {
 637208                writeReadsClosedFrame = _closeReadsOnWritesClosure;
 637209                _writesClosePending = true;
 637210                writeWritesClosedFrame = true;
 637211            }
 2779212        }
 213
 2779214        if (writeWritesClosedFrame)
 637215        {
 637216            if (IsRemote)
 252217            {
 218                // If it's a remote stream, we close writes before sending the StreamLast or StreamWritesClosed
 219                // frame to ensure _connection._bidirectionalStreamCount or _connection._unidirectionalStreamCount
 220                // is decreased before the peer receives the frame. This is necessary to prevent a race condition
 221                // where the peer could release the connection's bidirectional or unidirectional stream semaphore
 222                // before this connection's stream count is actually decreased.
 252223                TrySetWritesClosed();
 252224            }
 225
 637226            if (graceful)
 588227            {
 228                try
 588229                {
 588230                    WriteStreamFrame(FrameType.StreamLast, encode: null, writeReadsClosedFrame);
 588231                }
 0232                catch (IceRpcException)
 0233                {
 234                    // Ignore connection failures.
 0235                }
 236
 237                // If the stream is a local stream, writes are not closed until the StreamReadsClosed frame is
 238                // received from the peer (see ReceivedReadsClosedFrame). This ensures that the connection's
 239                // bidirectional or unidirectional stream semaphore is released only once the peer consumed the
 240                // buffered data.
 588241            }
 242            else
 49243            {
 244                try
 49245                {
 49246                    WriteStreamFrame(FrameType.StreamWritesClosed, encode: null, writeReadsClosedFrame);
 49247                }
 0248                catch (IceRpcException)
 0249                {
 250                    // Ignore connection failures.
 0251                }
 252
 49253                if (!IsRemote)
 28254                {
 255                    // We can now close writes to allow starting a new stream. Since the sending of frames is
 256                    // serialized over the connection, the peer will receive this StreamWritesClosed frame before
 257                    // a new stream sends a StreamFrame frame.
 28258                    TrySetWritesClosed();
 28259                }
 49260            }
 637261        }
 262        else
 2142263        {
 2142264            TrySetWritesClosed();
 2142265        }
 2779266    }
 267
 268    /// <summary>Notifies the stream of the amount of data consumed by the connection to send a <see
 269    /// cref="FrameType.Stream" /> or <see cref="FrameType.StreamLast" /> frame.</summary>
 270    /// <param name="size">The size of the stream frame.</param>
 8225271    internal void ConsumedSendCredit(int size) => _outputPipeWriter!.ConsumedSendCredit(size);
 272
 273    /// <summary>Fills the given writer with stream data received on the connection.</summary>
 274    /// <param name="bufferWriter">The destination buffer writer.</param>
 275    /// <param name="byteCount">The amount of stream data to read.</param>
 276    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 277    internal ValueTask FillBufferWriterAsync(
 278        IBufferWriter<byte> bufferWriter,
 279        int byteCount,
 280        CancellationToken cancellationToken) =>
 8703281        _connection.FillBufferWriterAsync(bufferWriter, byteCount, cancellationToken);
 282
 283    /// <summary>Notifies the stream of the reception of a <see cref="FrameType.Stream" /> or <see
 284    /// cref="FrameType.StreamLast" /> frame.</summary>
 285    /// <param name="size">The size of the data carried by the stream frame.</param>
 286    /// <param name="endStream"><see langword="true" /> if the received stream frame is the <see
 287    /// cref="FrameType.StreamLast" /> frame; otherwise, <see langword="false" />.</param>
 288    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 289    internal ValueTask<bool> ReceivedDataFrameAsync(int size, bool endStream, CancellationToken cancellationToken)
 8736290    {
 8736291        Debug.Assert(_inputPipeReader is not null);
 8736292        if (_state.HasFlag(State.ReadsClosed))
 33293        {
 33294            return new(false);
 295        }
 296        else
 8703297        {
 8703298            if (endStream && !IsRemote)
 567299            {
 300                // For a local stream we can close reads after we have received the StreamLast frame. For remote
 301                // streams reads are closed after the application has consumed all the data.
 567302                CloseReads(graceful: true);
 567303            }
 8703304            return _inputPipeReader.ReceivedDataFrameAsync(size, endStream, cancellationToken);
 305        }
 8736306    }
 307
 308    /// <summary>Notifies the stream of the reception of a <see cref="FrameType.StreamReadsClosed" /> frame.</summary>
 309    internal void ReceivedReadsClosedFrame()
 1327310    {
 1327311        TrySetWritesClosed();
 1327312        _outputPipeWriter?.CompleteWrites(exception: null);
 1327313    }
 314
 315    /// <summary>Notifies the stream of the reception of a <see cref="FrameType.StreamWindowUpdate" /> frame.</summary>
 316    /// <param name="frame">The body of the <see cref="FrameType.StreamWindowUpdate" /> frame.</param>
 317    internal void ReceivedWindowUpdateFrame(StreamWindowUpdateBody frame)
 1212318    {
 1212319        if (frame.WindowSizeIncrement > SlicTransportOptions.MaxWindowSize)
 0320        {
 0321            throw new IceRpcException(
 0322                IceRpcError.IceRpcError,
 0323                $"The window update is trying to increase the window size to a value larger than allowed.");
 324        }
 1212325        _outputPipeWriter!.ReceivedWindowUpdateFrame((int)frame.WindowSizeIncrement);
 1211326    }
 327
 328    /// <summary>Notifies the stream of the reception of a <see cref="FrameType.StreamWritesClosed" /> frame.</summary>
 329    internal void ReceivedWritesClosedFrame()
 48330    {
 48331        TrySetReadsClosed();
 332
 333        // Read operations will return a TruncatedData error if the peer closed writes.
 48334        _inputPipeReader?.CompleteReads(new IceRpcException(IceRpcError.TruncatedData));
 48335    }
 336
 337    /// <summary>Notifies the stream of the window update.</summary>
 338    /// <param name="size">The amount of data consumed by the application on the stream <see cref="Input" />.</param>
 339    internal void WindowUpdate(int size)
 1252340    {
 341        try
 1252342        {
 343            // Notify the sender of the window update to permit the sending of additional data.
 1252344            WriteStreamFrame(
 1252345                FrameType.StreamWindowUpdate,
 1252346                new StreamWindowUpdateBody((ulong)size).Encode,
 1252347                writeReadsClosedFrame: false);
 1252348        }
 0349        catch (IceRpcException)
 0350        {
 351            // Ignore connection failures.
 0352        }
 1252353    }
 354
 355    /// <summary>Writes a <see cref="FrameType.Stream" /> or <see cref="FrameType.StreamLast" /> frame on the
 356    /// connection.</summary>
 357    /// <param name="source1">The first stream frame data source.</param>
 358    /// <param name="source2">The second stream frame data source.</param>
 359    /// <param name="endStream"><see langword="true" /> to write a <see cref="FrameType.StreamLast" /> frame; otherwise,
 360    /// <see langword="false" />.</param>
 361    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 362    internal ValueTask<FlushResult> WriteStreamFrameAsync(
 363        ReadOnlySequence<byte> source1,
 364        ReadOnlySequence<byte> source2,
 365        bool endStream,
 366        CancellationToken cancellationToken)
 7956367    {
 7956368        if (!endStream)
 6150369        {
 370            // Hot path: a non-final stream frame requires no writes-closure bookkeeping, so forward the inner
 371            // ValueTask directly without an async state machine.
 6150372            return _connection.WriteStreamDataFrameAsync(
 6150373                this,
 6150374                source1,
 6150375                source2,
 6150376                endStream: false,
 6150377                writeReadsClosedFrame: false,
 6150378                cancellationToken);
 379        }
 380
 1806381        return WriteLastStreamFrameAsync();
 382
 383        async ValueTask<FlushResult> WriteLastStreamFrameAsync()
 1806384        {
 385            bool writeReadsClosedFrame;
 386            lock (_mutex)
 1806387            {
 1806388                writeReadsClosedFrame = _closeReadsOnWritesClosure;
 1806389                _writesClosePending = true;
 1806390            }
 391
 392            try
 1806393            {
 1806394                return await _connection.WriteStreamDataFrameAsync(
 1806395                    this,
 1806396                    source1,
 1806397                    source2,
 1806398                    endStream: true,
 1806399                    writeReadsClosedFrame,
 1806400                    cancellationToken).ConfigureAwait(false);
 401            }
 1005402            catch when (!_writesClosedTcs.Task.IsCompleted)
 2403            {
 404                // The write failed before the StreamLast frame was queued on the connection (WroteLastStreamFrame
 405                // completes _writesClosedTcs when the frame is queued), so roll back _writesClosePending so that
 406                // completing the stream output can still close writes and notify the peer with a StreamWritesClosed
 407                // frame. The bundled reads closure isn't lost either since _closeReadsOnWritesClosure wasn't cleared.
 408                lock (_mutex)
 2409                {
 2410                    _writesClosePending = false;
 2411                }
 2412                throw;
 413            }
 801414        }
 7956415    }
 416
 417    /// <summary>Notifies the stream that the <see cref="FrameType.StreamLast" /> was written by the
 418    /// connection.</summary>
 419    internal void WroteLastStreamFrame()
 1390420    {
 1390421        if (IsRemote)
 621422        {
 621423            TrySetWritesClosed();
 621424        }
 425        // For local streams, writes will be closed only once the peer sends the StreamReadsClosed frame.
 426
 1390427        _writesClosedTcs.TrySetResult();
 1390428    }
 429
 430    /// <summary>Throws the connection closure exception if the connection is closed.</summary>
 7984431    internal void ThrowIfConnectionClosed() => _connection.ThrowIfClosed();
 432
 6855433    private bool TrySetReadsClosed() => TrySetState(State.ReadsClosed);
 434
 435    private bool TrySetWritesClosed()
 6461436    {
 6461437        if (TrySetState(State.WritesClosed))
 4225438        {
 4225439            _writesClosedTcs.TrySetResult();
 4225440            return true;
 441        }
 442        else
 2236443        {
 2236444            return false;
 445        }
 6461446    }
 447
 448    private bool TrySetState(State state)
 13316449    {
 13316450        if (_state.TrySetFlag(state, out int newState))
 8453451        {
 8453452            if (newState.HasFlag(State.ReadsClosed | State.WritesClosed))
 4225453            {
 454                // The stream reads and writes are closed, it's time to release the stream to either allow creating or
 455                // accepting a new stream.
 4225456                _connection.ReleaseStream(this);
 4225457            }
 8453458            return true;
 459        }
 460        else
 4863461        {
 4863462            return false;
 463        }
 13316464    }
 465
 466    private void WriteStreamFrame(FrameType frameType, EncodeAction? encode, bool writeReadsClosedFrame) =>
 3249467        _connection.WriteStreamFrame(stream: this, frameType, encode, writeReadsClosedFrame);
 468
 469    [Flags]
 470    private enum State : int
 471    {
 472        ReadsClosed = 1,
 473        WritesClosed = 2
 474    }
 475}
Generated by: ReportGenerator 5.5.10.0
06/30/2026 - 14:46:58
GitHub | reportgenerator.io

Methods/Properties

get_Id()
set_Id(System.UInt64)
get_Input()
get_IsBidirectional()
get_IsRemote()
get_IsStarted()
get_Output()
get_WritesClosed()
get_WindowUpdateThreshold()
.ctor(IceRpc.Transports.Slic.Internal.SlicConnection,System.Boolean,System.Boolean)
AcquireSendCreditAsync(System.Threading.CancellationToken)
Close(System.Exception)
CloseReads(System.Boolean)
CloseWrites(System.Boolean)
ConsumedSendCredit(System.Int32)
FillBufferWriterAsync(System.Buffers.IBufferWriter`1<System.Byte>,System.Int32,System.Threading.CancellationToken)
ReceivedDataFrameAsync(System.Int32,System.Boolean,System.Threading.CancellationToken)
ReceivedReadsClosedFrame()
ReceivedWindowUpdateFrame(IceRpc.Transports.Slic.Internal.StreamWindowUpdateBody)
ReceivedWritesClosedFrame()
WindowUpdate(System.Int32)
WriteStreamFrameAsync(System.Buffers.ReadOnlySequence`1<System.Byte>,System.Buffers.ReadOnlySequence`1<System.Byte>,System.Boolean,System.Threading.CancellationToken)
WriteLastStreamFrameAsync()
WroteLastStreamFrame()
ThrowIfConnectionClosed()
TrySetReadsClosed()
TrySetWritesClosed()
TrySetState(IceRpc.Transports.Slic.Internal.SlicStream/State)
WriteStreamFrame(IceRpc.Transports.Slic.Internal.FrameType,ZeroC.Slice.Codec.EncodeAction,System.Boolean)