Merged PR 135566: [2.1] Fix a bug where cancelling read while pipe is…

… empty could result in infinite read loop

MSRC 46845

The bug was caused by `examined.Segment == _commitHead` comparison.

When pipe is empty and read is cancelld cursors are returned with `null` segment value.

Then if memory is allocated from pipe before AdvanceTo is called `examined.Segment == _commitHead` condition would return false resulting in pipe never being observed entirely until the first Flush.

src/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs

@@ -395,15 +395,27 @@ internal void AdvanceReader(BufferSegment consumedSegment, int consumedIndex, Bu
 
             lock (_sync)
             {
+                bool isEmpty = _readHead == null;
                 var examinedEverything = false;
-                if (examinedSegment == _commitHead)
+
+                if (examinedSegment != null)
+                {
+                    if (isEmpty)
+                    {
+                        ThrowHelper.ThrowInvalidOperationException_AdvanceToInvalidCursor();
+                        return;
+                    }
+
+                    examinedEverything = examinedSegment == _commitHead && examinedIndex == _commitHeadIndex - _commitHead.Start;
+                }
+                else
                 {
-                    examinedEverything = _commitHead != null ? examinedIndex == _commitHeadIndex - _commitHead.Start : examinedIndex == 0;
+                    examinedEverything = isEmpty;
                 }
 
                 if (consumedSegment != null)
                 {
-                    if (_readHead == null)
+                    if (isEmpty)
                     {
                         ThrowHelper.ThrowInvalidOperationException_AdvanceToInvalidCursor();
                         return;

src/System.IO.Pipelines/tests/ReadAsyncCancellationTests.cs

@@ -430,5 +430,26 @@ public async Task WriteAndCancellingPendingReadBeforeReadAsync()
             Assert.Equal("Hello World", Encoding.ASCII.GetString(array));
             Pipe.Reader.AdvanceTo(buffer.End, buffer.End);
         }
+
+        [Fact]
+        public async Task AdvanceTo_AfterCancelledRead_ReadAsyncNotCompleted()
+        {
+           var pipe = new Pipe();
+            var b = new byte[100];
+            await pipe.Writer.WriteAsync(b);
+
+            var result = await pipe.Reader.ReadAsync();
+            pipe.Reader.AdvanceTo(result.Buffer.End);
+
+            var read = pipe.Reader.ReadAsync();
+            pipe.Reader.CancelPendingRead();
+
+            result = await read;
+            var memory = pipe.Writer.GetMemory();
+
+            pipe.Reader.AdvanceTo(result.Buffer.End);
+
+            Assert.False(pipe.Reader.ReadAsync().IsCompleted);
+        }
     }
 }