Add Stream ReadAtLeast and ReadExactly (#69272)

* Add Stream ReadAtLeast and ReadExactly

Adds methods to Stream to read at least a minimum amount of bytes, or a full buffer, of data from the stream.
ReadAtLeast allows for the caller to specify whether an exception should be thrown or not on the end of the stream.

Make use of the new methods where appropriate in net7.0.

Fix #16598

* Add ReadAtLeast and ReadExactly unit tests

* Add XML docs to the new APIs

* Preserve behavior in StreamReader.FillBuffer when passed 0.

* Handle ReadExactly with an empty buffer, and ReadAtLeast with minimumBytes == 0.

Both of these cases are a no-op. No exception is thrown. A read won't be issued to the underlying stream. They won't block until data is available.

Author: eerhardt

src/libraries/System.Formats.Tar/src/System/Formats/Tar/TarHeader.Read.cs

@@ -26,7 +26,7 @@ internal bool TryGetNextHeader(Stream archiveStream, bool copyData)
             Span<byte> buffer = rented.AsSpan(0, TarHelpers.RecordSize); // minimumLength means the array could've been larger
             buffer.Clear(); // Rented arrays aren't clean
 
-            TarHelpers.ReadOrThrow(archiveStream, buffer);
+            archiveStream.ReadExactly(buffer);
 
             try
             {
@@ -486,10 +486,7 @@ private void ReadExtendedAttributesBlock(Stream archiveStream)
             }
 
             byte[] buffer = new byte[(int)_size];
-            if (archiveStream.Read(buffer.AsSpan()) != _size)
-            {
-                throw new EndOfStreamException();
-            }
+            archiveStream.ReadExactly(buffer);
 
             string dataAsString = TarHelpers.GetTrimmedUtf8String(buffer);
 
@@ -520,11 +517,7 @@ private void ReadGnuLongPathDataBlock(Stream archiveStream)
             }
 
             byte[] buffer = new byte[(int)_size];
-
-            if (archiveStream.Read(buffer.AsSpan()) != _size)
-            {
-                throw new EndOfStreamException();
-            }
+            archiveStream.ReadExactly(buffer);
 
             string longPath = TarHelpers.GetTrimmedUtf8String(buffer);
 

src/libraries/System.Formats.Tar/src/System/Formats/Tar/TarHelpers.cs

@@ -151,22 +151,6 @@ private static string GetTrimmedString(ReadOnlySpan<byte> buffer, Encoding encod
         // removing the trailing null or space chars.
         internal static string GetTrimmedUtf8String(ReadOnlySpan<byte> buffer) => GetTrimmedString(buffer, Encoding.UTF8);
 
-        // Reads the specified number of bytes and stores it in the byte buffer passed by reference.
-        // Throws if end of stream is reached.
-        internal static void ReadOrThrow(Stream archiveStream, Span<byte> buffer)
-        {
-            int totalRead = 0;
-            while (totalRead < buffer.Length)
-            {
-                int bytesRead = archiveStream.Read(buffer.Slice(totalRead));
-                if (bytesRead == 0)
-                {
-                    throw new EndOfStreamException();
-                }
-                totalRead += bytesRead;
-            }
-        }
-
         // Returns true if it successfully converts the specified string to a DateTimeOffset, false otherwise.
         internal static bool TryConvertToDateTimeOffset(string value, out DateTimeOffset timestamp)
         {

src/libraries/System.IO.Compression/src/System/IO/Compression/ZipHelper.cs

@@ -40,17 +40,10 @@ internal static Encoding GetEncoding(string text)
         /// </summary>
         internal static void ReadBytes(Stream stream, byte[] buffer, int bytesToRead)
         {
-            int bytesLeftToRead = bytesToRead;
-
-            int totalBytesRead = 0;
-
-            while (bytesLeftToRead > 0)
+            int bytesRead = stream.ReadAtLeast(buffer.AsSpan(0, bytesToRead), bytesToRead, throwOnEndOfStream: false);
+            if (bytesRead < bytesToRead)
             {
-                int bytesRead = stream.Read(buffer, totalBytesRead, bytesLeftToRead);
-                if (bytesRead == 0) throw new IOException(SR.UnexpectedEndOfStream);
-
-                totalBytesRead += bytesRead;
-                bytesLeftToRead -= bytesRead;
+                throw new IOException(SR.UnexpectedEndOfStream);
             }
         }
 

src/libraries/System.IO/tests/BinaryReader/BinaryReaderTests.cs

@@ -437,5 +437,28 @@ public void Read_CharSpan_ThrowIfDisposed()
                 Assert.Throws<ObjectDisposedException>(() => binaryReader.Read(new Span<char>()));
             }
         }
+
+        private class DerivedBinaryReader : BinaryReader
+        {
+            public DerivedBinaryReader(Stream input) : base(input) { }
+
+            public void CallFillBuffer0()
+            {
+                FillBuffer(0);
+            }
+        }
+
+        [Fact]
+        public void FillBuffer_Zero_Throws()
+        {
+            using Stream stream = CreateStream();
+
+            string hello = "Hello";
+            stream.Write(Encoding.ASCII.GetBytes(hello));
+            stream.Position = 0;
+
+            using var derivedReader = new DerivedBinaryReader(stream);
+            Assert.Throws<EndOfStreamException>(derivedReader.CallFillBuffer0);
+        }
     }
 }

src/libraries/System.IO/tests/Stream/Stream.ReadAtLeast.cs

@@ -0,0 +1,263 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace System.IO.Tests
+{
+    public class Stream_ReadAtLeast
+    {
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task DelegatesToRead_Success(bool async)
+        {
+            bool readInvoked = false;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvoked = true;
+                    Assert.NotNull(array);
+                    Assert.Equal(0, offset);
+                    Assert.Equal(30, count);
+
+                    for (int i = 0; i < 10; i++) array[offset + i] = (byte)i;
+                    return 10;
+                });
+
+            byte[] buffer = new byte[30];
+
+            Assert.Equal(10, async ? await s.ReadAtLeastAsync(buffer, 10) : s.ReadAtLeast(buffer, 10));
+            Assert.True(readInvoked);
+            for (int i = 0; i < 10; i++) Assert.Equal(i, buffer[i]);
+            for (int i = 10; i < 30; i++) Assert.Equal(0, buffer[i]);
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task ReadMoreThanOnePage(bool async)
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvokedCount++;
+
+                    for (int i = 0; i < 10; i++) array[offset + i] = (byte)i;
+                    return 10;
+                });
+
+            byte[] buffer = new byte[30];
+
+            Assert.Equal(20, async ? await s.ReadAtLeastAsync(buffer, 20) : s.ReadAtLeast(buffer, 20));
+            Assert.Equal(2, readInvokedCount);
+            for (int i = 0; i < 10; i++) Assert.Equal(i, buffer[i]);
+            for (int i = 10; i < 20; i++) Assert.Equal(i - 10, buffer[i]);
+            for (int i = 20; i < 30; i++) Assert.Equal(0, buffer[i]);
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task ReadMoreThanMinimumBytes(bool async)
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvokedCount++;
+
+                    int byteCount = Math.Min(count, 10);
+                    for (int i = 0; i < byteCount; i++) array[offset + i] = (byte)i;
+                    return byteCount;
+                });
+
+            // first try with a buffer that doesn't fill 3 full pages
+            byte[] buffer = new byte[28];
+
+            Assert.Equal(28, async ? await s.ReadAtLeastAsync(buffer, 22) : s.ReadAtLeast(buffer, 22));
+            Assert.Equal(3, readInvokedCount);
+            for (int i = 0; i < 10; i++) Assert.Equal(i, buffer[i]);
+            for (int i = 10; i < 20; i++) Assert.Equal(i - 10, buffer[i]);
+            for (int i = 20; i < 28; i++) Assert.Equal(i - 20, buffer[i]);
+
+            // now try with a buffer that is bigger than 3 pages
+            readInvokedCount = 0;
+            buffer = new byte[32];
+
+            Assert.Equal(30, async ? await s.ReadAtLeastAsync(buffer, 22) : s.ReadAtLeast(buffer, 22));
+            Assert.Equal(3, readInvokedCount);
+            for (int i = 0; i < 10; i++) Assert.Equal(i, buffer[i]);
+            for (int i = 10; i < 20; i++) Assert.Equal(i - 10, buffer[i]);
+            for (int i = 20; i < 30; i++) Assert.Equal(i - 20, buffer[i]);
+            for (int i = 30; i < 32; i++) Assert.Equal(0, buffer[i]);
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task ReadAtLeastZero(bool async)
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvokedCount++;
+
+                    int byteCount = Math.Min(count, 10);
+                    for (int i = 0; i < byteCount; i++) array[offset + i] = (byte)i;
+                    return byteCount;
+                });
+
+            byte[] buffer = new byte[20];
+
+            // ReadAtLeast minimumBytes=0 is a no-op
+            Assert.Equal(0, async ? await s.ReadAtLeastAsync(buffer, 0) : s.ReadAtLeast(buffer, 0));
+            Assert.Equal(0, readInvokedCount);
+
+            // now try with an empty buffer
+            byte[] emptyBuffer = Array.Empty<byte>();
+
+            Assert.Equal(0, async ? await s.ReadAtLeastAsync(emptyBuffer, 0) : s.ReadAtLeast(emptyBuffer, 0));
+            Assert.Equal(0, readInvokedCount);
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task NegativeMinimumBytes(bool async)
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvokedCount++;
+
+                    int byteCount = Math.Min(count, 10);
+                    for (int i = 0; i < byteCount; i++) array[offset + i] = (byte)i;
+                    return byteCount;
+                });
+
+            byte[] buffer = new byte[10];
+            if (async)
+            {
+                await Assert.ThrowsAsync<ArgumentOutOfRangeException>(async () => await s.ReadAtLeastAsync(buffer, -1));
+                await Assert.ThrowsAsync<ArgumentOutOfRangeException>(async () => await s.ReadAtLeastAsync(buffer, -10));
+            }
+            else
+            {
+                Assert.Throws<ArgumentOutOfRangeException>(() => s.ReadAtLeast(buffer, -1));
+                Assert.Throws<ArgumentOutOfRangeException>(() => s.ReadAtLeast(buffer, -10));
+            }
+            Assert.Equal(0, readInvokedCount);
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task BufferSmallerThanMinimumBytes(bool async)
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvokedCount++;
+
+                    int byteCount = Math.Min(count, 10);
+                    for (int i = 0; i < byteCount; i++) array[offset + i] = (byte)i;
+                    return byteCount;
+                });
+
+            byte[] buffer = new byte[20];
+            byte[] emptyBuffer = Array.Empty<byte>();
+            if (async)
+            {
+                await Assert.ThrowsAsync<ArgumentOutOfRangeException>(async () => await s.ReadAtLeastAsync(buffer, 21));
+                await Assert.ThrowsAsync<ArgumentOutOfRangeException>(async () => await s.ReadAtLeastAsync(emptyBuffer, 1));
+            }
+            else
+            {
+                Assert.Throws<ArgumentOutOfRangeException>(() => s.ReadAtLeast(buffer, 21));
+                Assert.Throws<ArgumentOutOfRangeException>(() => s.ReadAtLeast(emptyBuffer, 1));
+            }
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task HandleEndOfStream(bool async)
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readFunc: (array, offset, count) =>
+                {
+                    readInvokedCount++;
+
+                    if (readInvokedCount == 1)
+                    {
+                        int byteCount = Math.Min(count, 10);
+                        for (int i = 0; i < byteCount; i++) array[offset + i] = (byte)i;
+                        return byteCount;
+                    }
+                    else
+                    {
+                        return 0;
+                    }
+                });
+
+            byte[] buffer = new byte[20];
+            if (async)
+            {
+                await Assert.ThrowsAsync<EndOfStreamException>(async () => await s.ReadAtLeastAsync(buffer, 11));
+            }
+            else
+            {
+                Assert.Throws<EndOfStreamException>(() => s.ReadAtLeast(buffer, 11));
+            }
+            Assert.Equal(2, readInvokedCount);
+
+            readInvokedCount = 0;
+
+            Assert.Equal(10, async ? await s.ReadAtLeastAsync(buffer, 11, throwOnEndOfStream: false) : s.ReadAtLeast(buffer, 11, throwOnEndOfStream: false));
+            for (int i = 0; i < 10; i++) Assert.Equal(i, buffer[i]);
+            for (int i = 10; i < 20; i++) Assert.Equal(0, buffer[i]);
+            Assert.Equal(2, readInvokedCount);
+        }
+
+        [Fact]
+        public async Task CancellationTokenIsPassedThrough()
+        {
+            int readInvokedCount = 0;
+            var s = new DelegateStream(
+                canReadFunc: () => true,
+                readAsyncFunc: (array, offset, count, cancellationToken) =>
+                {
+                    readInvokedCount++;
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    int byteCount = Math.Min(count, 10);
+                    for (int i = 0; i < byteCount; i++) array[offset + i] = (byte)i;
+                    return Task.FromResult(10);
+                });
+
+            byte[] buffer = new byte[20];
+
+            using CancellationTokenSource cts = new CancellationTokenSource();
+            CancellationToken token = cts.Token;
+            cts.Cancel();
+
+            await Assert.ThrowsAsync<OperationCanceledException>(async () => await s.ReadAtLeastAsync(buffer, 10, cancellationToken: token));
+            Assert.Equal(1, readInvokedCount);
+        }
+    }
+}