Handle the case when zero byte reads are cancelled with an empty buff…

…er (#53456)

- When there's no buffer allocated and the zero byte reads is cancelled, it'll try to make a read only sequence with the current buffer, if that buffer is null because it wasn't yet allocated, it'll end up throwing a null reference exception. This adds a check to make sure were return an empty ReadOnlySequence if the backing buffer is null.
- Added tests

src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/StreamPipeReader.cs

@@ -513,7 +513,7 @@ private bool TryReadInternal(CancellationTokenSource source, out ReadResult resu
                     ClearCancellationToken();
                 }
 
-                ReadOnlySequence<byte> buffer = _readHead == null ? default : GetCurrentReadOnlySequence();
+                ReadOnlySequence<byte> buffer = GetCurrentReadOnlySequence();
 
                 result = new ReadResult(buffer, isCancellationRequested, _isStreamCompleted);
                 return true;
@@ -525,8 +525,8 @@ private bool TryReadInternal(CancellationTokenSource source, out ReadResult resu
 
         private ReadOnlySequence<byte> GetCurrentReadOnlySequence()
         {
-            Debug.Assert(_readHead != null && _readTail != null);
-            return new ReadOnlySequence<byte>(_readHead, _readIndex, _readTail, _readTail.End);
+            // If _readHead is null then _readTail is also null
+            return _readHead is null ? default : new ReadOnlySequence<byte>(_readHead, _readIndex, _readTail!, _readTail!.End);
         }
 
         private void AllocateReadTail(int? minimumSize = null)

src/libraries/System.IO.Pipelines/tests/StreamPipeReaderTests.cs

@@ -267,11 +267,13 @@ public async Task ReadCanBeCancelledViaProvidedCancellationToken()
             reader.Complete();
         }
 
-        [ConditionalFact(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
-        public async Task ReadCanBeCanceledViaCancelPendingReadWhenReadIsAsync()
+        [ConditionalTheory(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
+        [InlineData(false)]
+        [InlineData(true)]
+        public async Task ReadCanBeCanceledViaCancelPendingReadWhenReadAsync(bool useZeroByteReads)
         {
             var stream = new CancelledReadsStream();
-            PipeReader reader = PipeReader.Create(stream);
+            PipeReader reader = PipeReader.Create(stream, new StreamPipeReaderOptions(useZeroByteReads: useZeroByteReads));
 
             ValueTask<ReadResult> task = reader.ReadAsync();
 
@@ -284,6 +286,25 @@ public async Task ReadCanBeCanceledViaCancelPendingReadWhenReadIsAsync()
             reader.Complete();
         }
 
+        [ConditionalTheory(typeof(PlatformDetection), nameof(PlatformDetection.IsThreadingSupported))]
+        [InlineData(false)]
+        [InlineData(true)]
+        public async Task ReadCanBeCanceledViaCancelPendingReadWhenReadAtLeastAsync(bool useZeroByteReads)
+        {
+            var stream = new CancelledReadsStream();
+            PipeReader reader = PipeReader.Create(stream, new StreamPipeReaderOptions(useZeroByteReads: useZeroByteReads));
+
+            ValueTask<ReadResult> task = reader.ReadAtLeastAsync(1);
+
+            reader.CancelPendingRead();
+
+            stream.WaitForReadTask.TrySetResult(null);
+
+            ReadResult readResult = await task;
+            Assert.True(readResult.IsCanceled);
+            reader.Complete();
+        }
+
         [Fact]
         public async Task ReadAsyncReturnsCanceledIfCanceledBeforeRead()
         {