From c78cc86e04e8266e2429253b14bb8dd6b6142cfe Mon Sep 17 00:00:00 2001 From: Adam Hathcock Date: Tue, 20 Oct 2026 10:52:15 +0100 Subject: [PATCH] Avoid temporary allocations in ring-buffered async reads --- .../IO/SharpCompressStream.Async.cs | 54 +++++++- .../Streams/DeflateReadAllocationTests.cs | 124 ++++++++++++++++++ 2 files changed, 171 insertions(+), 7 deletions(-) diff --git a/src/SharpCompress/IO/SharpCompressStream.Async.cs b/src/SharpCompress/IO/SharpCompressStream.Async.cs index 47ba651e..3dc208ff 100644 --- a/src/SharpCompress/IO/SharpCompressStream.Async.cs +++ b/src/SharpCompress/IO/SharpCompressStream.Async.cs @@ -1,5 +1,7 @@ using System; +using System.Buffers; using System.IO; +using System.Runtime.InteropServices; using System.Threading; using System.Threading.Tasks; using SharpCompress.Common; @@ -146,9 +148,11 @@ public partial class SharpCompressStream int totalRead = 0; int count = buffer.Length; int offset = 0; + // Deflate refills use array-backed memory. Preserve the ValueTask input read + // without allocating temporary arrays just to access the ring buffer. + var arrayBacked = MemoryMarshal.TryGetArray((ReadOnlyMemory)buffer, out var segment); // If logical position is behind stream position, read from ring buffer first - // Note: We need to use a temporary byte array because RingBuffer.ReadFromEnd expects byte[] while (count > 0 && _logicalPosition < streamPosition) { long bytesFromEnd = streamPosition - _logicalPosition; @@ -162,9 +166,30 @@ public partial class SharpCompressStream ); } - var tempBuffer = new byte[Math.Min(count, (int)bytesFromEnd)]; - int available = _ringBuffer.ReadFromEnd(bytesFromEnd, tempBuffer, 0, tempBuffer.Length); - tempBuffer.AsSpan(0, available).CopyTo(buffer.Span.Slice(offset)); + int available; + if (arrayBacked) + { + available = _ringBuffer.ReadFromEnd( + bytesFromEnd, + segment.Array!, + segment.Offset + offset, + count + ); + } + else + { + int replayCount = Math.Min(count, (int)bytesFromEnd); + var tempBuffer = ArrayPool.Shared.Rent(replayCount); + try + { + available = _ringBuffer.ReadFromEnd(bytesFromEnd, tempBuffer, 0, replayCount); + tempBuffer.AsSpan(0, available).CopyTo(buffer.Span.Slice(offset)); + } + finally + { + ArrayPool.Shared.Return(tempBuffer); + } + } totalRead += available; offset += available; @@ -180,9 +205,24 @@ public partial class SharpCompressStream .ConfigureAwait(false); if (read > 0) { - // RingBuffer.Write expects byte[], so we need to copy - var tempBuffer = buffer.Slice(offset, read).ToArray(); - _ringBuffer!.Write(tempBuffer, 0, read); + if (arrayBacked) + { + _ringBuffer!.Write(segment.Array!, segment.Offset + offset, read); + } + else + { + // RingBuffer.Write requires an array for non-array memory. + var tempBuffer = ArrayPool.Shared.Rent(read); + try + { + buffer.Slice(offset, read).Span.CopyTo(tempBuffer); + _ringBuffer!.Write(tempBuffer, 0, read); + } + finally + { + ArrayPool.Shared.Return(tempBuffer); + } + } streamPosition += read; _logicalPosition += read; totalRead += read; diff --git a/tests/SharpCompress.Test/Streams/DeflateReadAllocationTests.cs b/tests/SharpCompress.Test/Streams/DeflateReadAllocationTests.cs index 194ff9a6..1494e2a6 100644 --- a/tests/SharpCompress.Test/Streams/DeflateReadAllocationTests.cs +++ b/tests/SharpCompress.Test/Streams/DeflateReadAllocationTests.cs @@ -7,6 +7,9 @@ using System.Threading; using System.Threading.Tasks; using SharpCompress.Compressors; using SharpCompress.Compressors.Deflate; +using SharpCompress.IO; +using SharpCompress.Readers; +using SharpCompress.Test.Mocks; using Xunit; namespace SharpCompress.Test.Streams; @@ -80,6 +83,127 @@ public class DeflateReadAllocationTests } #if !LEGACY_DOTNET + [Fact] + public async Task NonSeekable_Reader_Does_Not_Allocate_Compressed_Payload_Copies() + { + var expected = new byte[2 * 1024 * 1024]; + new Random(1234).NextBytes(expected); + using var packed = new MemoryStream(); + using ( + var zip = new System.IO.Compression.ZipArchive( + packed, + System.IO.Compression.ZipArchiveMode.Create, + leaveOpen: true + ) + ) + { + using var entry = zip.CreateEntry("random.bin").Open(); + entry.Write(expected); + } + packed.Position = 0; + using var input = new ForwardOnlyStream(packed); + await using var reader = await ReaderFactory.OpenAsyncReader( + input, + ReaderOptions.ForExternalStream, + CancellationToken.None + ); + Assert.True(await reader.MoveToNextEntryAsync(CancellationToken.None)); + await using var decoded = await reader.OpenEntryStreamAsync(CancellationToken.None); + var buffer = new byte[32768]; + var total = await decoded.ReadAsync(buffer.AsMemory(), CancellationToken.None); + Assert.True(expected.AsSpan(0, total).SequenceEqual(buffer.AsSpan(0, total))); + var start = GC.GetAllocatedBytesForCurrentThread(); + int count; + while ((count = await decoded.ReadAsync(buffer.AsMemory(), CancellationToken.None)) != 0) + { + Assert.True(expected.AsSpan(total, count).SequenceEqual(buffer.AsSpan(0, count))); + total += count; + } + var allocated = GC.GetAllocatedBytesForCurrentThread() - start; + Assert.Equal(expected.Length, total); + // Memory-backed I/O completes synchronously, keeping this counter on one thread. + // A temporary array per compressed refill would allocate about 2 MiB. + Assert.InRange(allocated, 0, 131072); + Assert.True(input.CanRead); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task RingBuffered_Memory_Read_Preserves_Slices_Across_Replay_And_Fresh_Input( + bool nonArray + ) + { + var expected = Enumerable.Range(0, 64).Select(i => (byte)i).ToArray(); + using var input = new ForwardOnlyStream(new MemoryStream(expected)); + using var stream = SharpCompressStream.Create(input, 16); + var bytes = Enumerable.Repeat((byte)0xCC, 22).ToArray(); + using var owner = new NonArrayMemory(bytes); + var memory = (nonArray ? owner.Memory : bytes.AsMemory()).Slice(5, 12); + Assert.Equal(12, await stream.ReadAsync(memory, CancellationToken.None)); + Assert.Equal(12, await stream.ReadAsync(memory, CancellationToken.None)); + // The ring has wrapped. Combine six replayed bytes with six fresh bytes. + stream.Position = 18; + Assert.Equal(12, await stream.ReadAsync(memory, CancellationToken.None)); + Assert.Equal(expected.Skip(18).Take(12), bytes.Skip(5).Take(12)); + Assert.Equal(30, stream.Position); + // Replay the fresh bytes too, verifying the source offset used when recording. + stream.Position = 24; + Assert.Equal(12, await stream.ReadAsync(memory, CancellationToken.None)); + Assert.Equal(expected.Skip(24).Take(12), bytes.Skip(5).Take(12)); + Assert.Equal(36, stream.Position); + Assert.All(bytes.Take(5), b => Assert.Equal(0xCC, b)); + Assert.All(bytes.Skip(17), b => Assert.Equal(0xCC, b)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task RingBuffer_Fresh_Input_Does_Not_Allocate_Per_Read(bool nonArray) + { + const int length = 1024 * 1024; + using var input = new ForwardOnlyStream(new MemoryStream(new byte[length])); + using var stream = SharpCompressStream.Create(input, 8192); + var buffer = new byte[4106]; + using var owner = new NonArrayMemory(buffer); + var memory = (nonArray ? owner.Memory : buffer.AsMemory()).Slice(5, 4096); + var total = await stream.ReadAsync(memory, CancellationToken.None); + var start = GC.GetAllocatedBytesForCurrentThread(); + int count; + while ((count = await stream.ReadAsync(memory, CancellationToken.None)) != 0) + { + total += count; + } + var allocated = GC.GetAllocatedBytesForCurrentThread() - start; + Assert.Equal(length, total); + Assert.InRange(allocated, 0, 65536); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task RingBuffer_Replay_Does_Not_Allocate_Per_Read(bool nonArray) + { + using var input = new ForwardOnlyStream(new MemoryStream(new byte[32])); + using var stream = SharpCompressStream.Create(input, 16); + var buffer = new byte[22]; + using var owner = new NonArrayMemory(buffer); + var memory = (nonArray ? owner.Memory : buffer.AsMemory()).Slice(5, 12); + await stream.ReadAsync(memory, CancellationToken.None); + stream.Position = 0; + await stream.ReadAsync(memory, CancellationToken.None); + var start = GC.GetAllocatedBytesForCurrentThread(); + var total = 0; + for (var i = 0; i < 10000; i++) + { + stream.Position = 0; + total += await stream.ReadAsync(memory, CancellationToken.None); + } + var allocated = GC.GetAllocatedBytesForCurrentThread() - start; + Assert.Equal(120000, total); + Assert.InRange(allocated, 0, 65536); + } + [Fact] public async Task NonArray_Fallback_Reuses_Its_Rental_Across_Many_Reads() {