< Summary

Information
Class: IceRpc.Transports.Slic.Internal.SlicPipeReader
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Transports/Slic/Internal/SlicPipeReader.cs
Tag: 2300_35243572715
Line coverage
79%
Covered lines: 106
Uncovered lines: 28
Coverable lines: 134
Total lines: 263
Line coverage: 79.1%
Branch coverage
73%
Covered branches: 28
Total branches: 38
Branch coverage: 73.6%
Method coverage
100%
Covered methods: 11
Fully covered methods: 7
Total methods: 11
Method coverage: 100%
Full method coverage: 63.6%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
AdvanceTo(...)100%11100%
AdvanceTo(...)100%22100%
CancelPendingRead()100%11100%
Complete(...)100%22100%
ReadAsync()100%22100%
TryRead(...)50%5461.53%
.ctor(...)100%11100%
CompleteReads(...)100%44100%
ReceivedDataFrameAsync()71.42%211467.5%
ProcessReadResult(...)62.5%12860%
ThrowIfCompleted()50%2260%

File(s)

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

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Internal;
 4using IceRpc.Transports.Internal;
 5using System.IO.Pipelines;
 6
 7namespace IceRpc.Transports.Slic.Internal;
 8
 9// The SlicPipeReader doesn't override ReadAtLeastAsyncCore or CopyToAsync methods because:
 10// - we can't forward the calls to the internal pipe reader since reading relies on the AdvanceTo implementation to send
 11//   the StreamWindowUpdate frame once the data is examined,
 12// - the default implementation can't be much optimized.
 13internal class SlicPipeReader : PipeReader
 14{
 15    private int _examined;
 16    private volatile Exception? _exception;
 17    private long _lastExaminedOffset;
 18    private readonly Pipe _pipe;
 19    private ReadResult _readResult;
 20    // FlagEnumExtensions operations are used to update the state. These operations are atomic and don't require mutex
 21    // locking.
 22    private int _state;
 23    private readonly SlicStream _stream;
 24    private int _windowSize;
 25
 357226    public override void AdvanceTo(SequencePosition consumed) => AdvanceTo(consumed, consumed);
 27
 28    public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
 728329    {
 728330        ThrowIfCompleted();
 31
 728332        long startOffset = _readResult.Buffer.GetOffset(_readResult.Buffer.Start);
 728333        long consumedOffset = _readResult.Buffer.GetOffset(consumed) - startOffset;
 728334        long examinedOffset = _readResult.Buffer.GetOffset(examined) - startOffset;
 35
 36        // Add the additional examined bytes to the examined bytes total.
 728337        _examined += (int)(examinedOffset - _lastExaminedOffset);
 728338        _lastExaminedOffset = examinedOffset - consumedOffset;
 39
 40        // If the number of examined bytes is superior to the window update threshold, notifies the stream of the window
 41        // update. This will trigger the sending of a window update frame and allow the sender to send additional data.
 728342        if (_examined >= _stream.WindowUpdateThreshold)
 124843        {
 124844            Interlocked.Add(ref _windowSize, _examined);
 124845            _stream.WindowUpdate(_examined);
 124846            _examined = 0;
 124847        }
 48
 728349        _pipe.Reader.AdvanceTo(consumed, examined);
 728350    }
 51
 552    public override void CancelPendingRead() => _pipe.Reader.CancelPendingRead();
 53
 54    public override void Complete(Exception? exception = null)
 281555    {
 281556        if (_state.TrySetFlag(State.Completed))
 279257        {
 58            // Forcefully close the stream reads if reads were not already gracefully closed by ReadAsync or TryRead.
 279259            _stream.CloseReads(graceful: false);
 60
 279261            CompleteReads(exception: null);
 62
 279263            _pipe.Reader.Complete();
 279264        }
 281565    }
 66
 67    public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
 760468    {
 760469        ThrowIfCompleted();
 70
 760471        if (_exception is not null)
 1672        {
 1673            _stream.ThrowIfConnectionClosed();
 1374        }
 75
 760176        return ProcessReadResult(await _pipe.Reader.ReadAsync(cancellationToken).ConfigureAwait(false));
 730277    }
 78
 79    public override bool TryRead(out ReadResult result)
 7780    {
 7781        ThrowIfCompleted();
 82
 7783        if (_exception is not null)
 084        {
 085            _stream.ThrowIfConnectionClosed();
 086        }
 87
 7788        if (_pipe.Reader.TryRead(out result))
 7789        {
 7790            result = ProcessReadResult(result);
 7791            return true;
 92        }
 93        else
 094        {
 095            return false;
 96        }
 7797    }
 98
 283399    internal SlicPipeReader(SlicStream stream, SlicConnection connection)
 2833100    {
 2833101        _stream = stream;
 2833102        _windowSize = connection.InitialStreamWindowSize;
 103
 104        // We keep the default readerScheduler (ThreadPool) because the _pipe.Writer.FlushAsync executes in the
 105        // "read loop task" and we don't want this task to continue into application code. The writerScheduler
 106        // doesn't matter since _pipe.Writer.FlushAsync never blocks.
 2833107        _pipe = new(new PipeOptions(
 2833108            pool: connection.Pool,
 2833109            pauseWriterThreshold: 0,
 2833110            minimumSegmentSize: connection.MinSegmentSize,
 2833111            useSynchronizationContext: false));
 2833112    }
 113
 114    /// <summary>Completes reads.</summary>
 115    /// <param name="exception">The exception that will be raised by <see cref="ReadAsync" /> or <see cref="TryRead" />
 116    /// operation.</param>
 117    internal void CompleteReads(Exception? exception)
 3201118    {
 3201119        Interlocked.CompareExchange(ref _exception, exception, null);
 120
 3201121        if (_state.TrySetFlag(State.PipeWriterCompleted))
 2829122        {
 2829123            if (_state.HasFlag(State.PipeWriterInUse))
 3124            {
 3125                _pipe.Reader.CancelPendingRead();
 3126            }
 127            else
 2826128            {
 2826129                _pipe.Writer.Complete(exception);
 2826130            }
 2829131        }
 3201132    }
 133
 134    /// <summary>Notifies the reader of the reception of a <see cref="FrameType.Stream" /> or <see
 135    /// cref="FrameType.StreamLast" /> frame. The stream data is consumed from the connection and buffered by this
 136    /// reader on its internal pipe.</summary>
 137    /// <returns><see langword="true" /> if the data was consumed; otherwise, <see langword="false"/> if the reader was
 138    /// completed by the application.</returns>
 139    internal async ValueTask<bool> ReceivedDataFrameAsync(
 140        int dataSize,
 141        bool endStream,
 142        CancellationToken cancellationToken)
 8758143    {
 8758144        if (dataSize == 0 && !endStream)
 0145        {
 0146            throw new IceRpcException(
 0147                IceRpcError.IceRpcError,
 0148                "An empty Slic stream frame is not allowed unless endStream is true.");
 149        }
 150
 8758151        if (!_state.TrySetFlag(State.PipeWriterInUse))
 0152        {
 0153            throw new InvalidOperationException(
 0154                $"The {nameof(ReceivedDataFrameAsync)} operation is not thread safe.");
 155        }
 156
 157        try
 8758158        {
 8758159            if (_state.HasFlag(State.PipeWriterCompleted))
 0160            {
 0161                return false; // No bytes consumed because the application completed the stream input.
 162            }
 163
 8758164            int newWindowSize = Interlocked.Add(ref _windowSize, -dataSize);
 8758165            if (newWindowSize < 0)
 0166            {
 0167                throw new IceRpcException(
 0168                    IceRpcError.IceRpcError,
 0169                    "Received more data than flow control permits.");
 170            }
 171
 172            // Fill the pipe writer with dataSize bytes.
 8758173            await _stream.FillBufferWriterAsync(
 8758174                _pipe.Writer,
 8758175                dataSize,
 8758176                cancellationToken).ConfigureAwait(false);
 177
 8758178            if (endStream)
 1284179            {
 1284180                _pipe.Writer.Complete();
 1284181            }
 182            else
 7474183            {
 7474184                _ = await _pipe.Writer.FlushAsync(CancellationToken.None).ConfigureAwait(false);
 7474185            }
 186
 8758187            return true;
 188        }
 189        finally
 8758190        {
 8758191            if (_state.HasFlag(State.PipeWriterCompleted))
 3192            {
 193                // If the pipe writer has been completed while we were reading the data from the stream, we make sure to
 194                // complete the writer now since Complete or CompleteWriter didn't do it.
 3195                _pipe.Writer.Complete(_exception);
 3196            }
 8758197            _state.ClearFlag(State.PipeWriterInUse);
 8758198        }
 8758199    }
 200
 201    private ReadResult ProcessReadResult(ReadResult result)
 7379202    {
 203        // This method is called by ReadAsync or TryRead with the read result returned by the _pipe.Reader read
 204        // operation.
 7379205        if (result.IsCanceled)
 5206        {
 207            // The _pipe.Reader ReadAsync/TryRead operations can return a canceled read result for two different
 208            // reasons:
 209            // - the application called CancelPendingRead
 210            // - the connection is closed while data is written on _pipe.Writer
 5211            if (_state.HasFlag(State.PipeWriterCompleted))
 0212            {
 213                // The connection was closed while the pipe writer was in use. Either throw or return a non-canceled
 214                // result depending on the completion exception.
 0215                if (_exception is null)
 0216                {
 0217                    result = new ReadResult(result.Buffer, isCanceled: false, isCompleted: true);
 0218                }
 219                else
 0220                {
 0221                    throw ExceptionUtil.Throw(_exception);
 222                }
 0223            }
 224            // else: the application called CancelPendingRead, return the canceled read result as-is.
 5225        }
 7374226        else if (result.IsCompleted)
 1413227        {
 228            // All the data from the peer is considered read at this point. It's time to close reads on the stream. This
 229            // will write the StreamReadsClosed frame to the peer and allow it to release the stream semaphore.
 1413230            _stream.CloseReads(graceful: true);
 1413231        }
 232
 233        // Cache the read result returned to the application: AdvanceTo computes the consumed and examined offsets
 234        // against the cached buffer.
 7379235        _readResult = result;
 7379236        return result;
 7379237    }
 238
 239    private void ThrowIfCompleted()
 14964240    {
 14964241        if (_state.HasFlag(State.Completed))
 0242        {
 243            // If the reader is completed, the caller is bogus, it shouldn't call read operations after completing the
 244            // pipe reader.
 0245            throw new InvalidOperationException("Reading is not allowed once the reader is completed.");
 246        }
 14964247    }
 248
 249    /// <summary>The state enumeration is used to ensure the reader is not used after it's completed and to ensure that
 250    /// the internal pipe writer isn't completed concurrently when it's being used by <see
 251    /// cref="ReceivedDataFrameAsync" />.</summary>
 252    private enum State : int
 253    {
 254        /// <summary><see cref="Complete" /> was called on this Slic pipe reader.</summary>
 255        Completed = 1,
 256
 257        /// <summary>Data is being written to the internal pipe writer.</summary>
 258        PipeWriterInUse = 2,
 259
 260        /// <summary>The internal pipe writer was completed by <see cref="CompleteReads" />.</summary>
 261        PipeWriterCompleted = 4,
 262    }
 263}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
GitHub | reportgenerator.io

Methods/Properties

AdvanceTo(System.SequencePosition)
AdvanceTo(System.SequencePosition,System.SequencePosition)
CancelPendingRead()
Complete(System.Exception)
ReadAsync()
TryRead(System.IO.Pipelines.ReadResult&)
.ctor(IceRpc.Transports.Slic.Internal.SlicStream,IceRpc.Transports.Slic.Internal.SlicConnection)
CompleteReads(System.Exception)
ReceivedDataFrameAsync()
ProcessReadResult(System.IO.Pipelines.ReadResult)
ThrowIfCompleted()