Fix DeflateStream's handling of partial reads

src/libraries/Common/tests/StreamConformanceTests/System/IO/StreamConformanceTests.cs

@@ -1562,11 +1562,6 @@ public abstract class ConnectedStreamConformanceTests : StreamConformanceTests
         /// Gets whether the stream guarantees that all data written to it will be flushed as part of Flush{Async}.
         /// </summary>
         protected virtual bool FlushGuaranteesAllDataWritten => true;
-        /// <summary>
-        /// Gets whether a stream implements an aggressive read that tries to fill the supplied buffer and only
-        /// stops when it does so or hits EOF.
-        /// </summary>
-        protected virtual bool ReadsMayBlockUntilBufferFullOrEOF => false;
         /// <summary>Gets whether reads for a count of 0 bytes block if no bytes are available to read.</summary>
         protected virtual bool BlocksOnZeroByteReads => false;
         /// <summary>
@@ -1709,6 +1704,10 @@ public virtual async Task ReadWriteByte_Success()
             }
         }
 
+        public static IEnumerable<object[]> ReadWrite_Modes =>
+            from mode in Enum.GetValues<ReadWriteMode>()
+            select new object[] { mode };
+
         public static IEnumerable<object[]> ReadWrite_Success_MemberData() =>
             from mode in Enum.GetValues<ReadWriteMode>()
             from writeSize in new[] { 1, 42, 10 * 1024 }
@@ -1785,6 +1784,54 @@ public virtual async Task ReadWrite_Success(ReadWriteMode mode, int writeSize, b
             }
         }
 
+        [Theory]
+        [MemberData(nameof(ReadWrite_Modes))]
+        [ActiveIssue("https://github.com/dotnet/runtime/issues/51371", TestPlatforms.iOS | TestPlatforms.tvOS | TestPlatforms.MacCatalyst)]
+        public virtual async Task ReadWrite_MessagesSmallerThanReadBuffer_Success(ReadWriteMode mode)
+        {
+            if (!FlushGuaranteesAllDataWritten)
+            {
+                return;
+            }
+
+            foreach (CancellationToken nonCanceledToken in new[] { CancellationToken.None, new CancellationTokenSource().Token })
+            {
+                using StreamPair streams = await CreateConnectedStreamsAsync();
+
+                foreach ((Stream writeable, Stream readable) in GetReadWritePairs(streams))
+                {
+                    byte[] writerBytes = RandomNumberGenerator.GetBytes(512);
+                    var readerBytes = new byte[writerBytes.Length * 2];
+
+                    // Repeatedly write then read a message smaller in size than the read buffer
+                    for (int i = 0; i < 5; i++)
+                    {
+                        Task writes = Task.Run(async () =>
+                        {
+                            await WriteAsync(mode, writeable, writerBytes, 0, writerBytes.Length, nonCanceledToken);
+                            if (FlushRequiredToWriteData)
+                            {
+                                await writeable.FlushAsync();
+                            }
+                        });
+
+                        int n = 0;
+                        while (n < writerBytes.Length)
+                        {
+                            int r = await ReadAsync(mode, readable, readerBytes, n, readerBytes.Length - n);
+                            Assert.InRange(r, 1, writerBytes.Length - n);
+                            n += r;
+                        }
+
+                        Assert.Equal(writerBytes.Length, n);
+                        AssertExtensions.SequenceEqual(writerBytes, readerBytes.AsSpan(0, writerBytes.Length));
+
+                        await writes;
+                    }
+                }
+            }
+        }
+
         [Theory]
         [MemberData(nameof(AllReadWriteModesAndValue), false)]
         [MemberData(nameof(AllReadWriteModesAndValue), true)]
@@ -2160,6 +2207,10 @@ public virtual async Task ZeroByteRead_BlocksUntilDataAvailableOrNops(ReadWriteM
                         });
                         Assert.Equal(0, await zeroByteRead);
 
+                        // Perform a second zero-byte read.
+                        await Task.Run(() => ReadAsync(mode, readable, Array.Empty<byte>(), 0, 0));
+
+                        // Now consume all the data.
                         var readBytes = new byte[5];
                         int count = 0;
                         while (count < readBytes.Length)
@@ -2684,7 +2735,7 @@ public virtual async Task Flush_FlushesUnderlyingStream(bool flushAsync)
         [InlineData(true, true)]
         public virtual async Task Dispose_Flushes(bool useAsync, bool leaveOpen)
         {
-            if (leaveOpen && (!SupportsLeaveOpen || ReadsMayBlockUntilBufferFullOrEOF))
+            if (leaveOpen && !SupportsLeaveOpen)
             {
                 return;
             }

src/libraries/Common/tests/System/IO/Compression/CompressionStreamTestBase.cs

@@ -54,6 +54,6 @@ protected override Task<StreamPair> CreateWrappedConnectedStreamsAsync(StreamPai
         protected override Type UnsupportedReadWriteExceptionType => typeof(InvalidOperationException);
         protected override bool WrappedUsableAfterClose => false;
         protected override bool FlushRequiredToWriteData => true;
-        protected override bool FlushGuaranteesAllDataWritten => false;
+        protected override bool BlocksOnZeroByteReads => true;
     }
 }

src/libraries/Common/tests/System/IO/Compression/ZipTestHelper.cs

@@ -65,6 +65,17 @@ public static void ReadBytes(Stream stream, byte[] buffer, long bytesToRead)
             }
         }
 
+        public static int ReadAllBytes(Stream stream, byte[] buffer, int offset, int count)
+        {
+            int bytesRead;
+            int totalRead = 0;
+            while ((bytesRead = stream.Read(buffer, offset + totalRead, count - totalRead)) != 0)
+            {
+                totalRead += bytesRead;
+            }
+            return totalRead;
+        }
+
         public static bool ArraysEqual<T>(T[] a, T[] b) where T : IComparable<T>
         {
             if (a.Length != b.Length) return false;
@@ -111,8 +122,8 @@ public static void StreamsEqual(Stream ast, Stream bst, int blocksToRead)
                 if (blocksToRead != -1 && blocksRead >= blocksToRead)
                     break;
 
-                ac = ast.Read(ad, 0, 4096);
-                bc = bst.Read(bd, 0, 4096);
+                ac = ReadAllBytes(ast, ad, 0, 4096);
+                bc = ReadAllBytes(bst, bd, 0, 4096);
 
                 if (ac != bc)
                 {
@@ -170,7 +181,7 @@ public static void IsZipSameAsDir(Stream archiveFile, string directory, ZipArchi
                         var buffer = new byte[entry.Length];
                         using (Stream entrystream = entry.Open())
                         {
-                            entrystream.Read(buffer, 0, buffer.Length);
+                            ReadAllBytes(entrystream, buffer, 0, buffer.Length);
 #if NETCOREAPP
                             uint zipcrc = entry.Crc32;
                             Assert.Equal(CRC.CalculateCRC(buffer), zipcrc);

src/libraries/System.IO.Compression.Brotli/src/System/IO/Compression/BrotliStream.cs

@@ -173,21 +173,19 @@ private void EnsureNoActiveAsyncOperation()
 
         private void AsyncOperationStarting()
         {
-            if (Interlocked.CompareExchange(ref _activeAsyncOperation, 1, 0) != 0)
+            if (Interlocked.Exchange(ref _activeAsyncOperation, 1) != 0)
             {
                 ThrowInvalidBeginCall();
             }
         }
 
         private void AsyncOperationCompleting()
         {
-            int oldValue = Interlocked.CompareExchange(ref _activeAsyncOperation, 0, 1);
-            Debug.Assert(oldValue == 1, $"Expected {nameof(_activeAsyncOperation)} to be 1, got {oldValue}");
+            Debug.Assert(_activeAsyncOperation == 1);
+            Volatile.Write(ref _activeAsyncOperation, 0);
         }
 
-        private static void ThrowInvalidBeginCall()
-        {
+        private static void ThrowInvalidBeginCall() =>
             throw new InvalidOperationException(SR.InvalidBeginCall);
-        }
     }
 }