starred/sharpcompress
1
0
Fork 0
mirror of https://github.com/adamhathcock/sharpcompress.git synced 2026-11-15 17:02:03 +00:00
Code Issues 975 Packages Projects Releases 25 Wiki Activity

Avoid temporary allocations in ring-buffered async reads

Browse Source
This commit is contained in:
Adam Hathcock
2026-10-20 10:52:15 +01:00
parent 7731843e25
commit c78cc86e04
2 changed files with 171 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
src/SharpCompress/IO/SharpCompressStream.Async.cs
View File

@@ -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<byte>)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<byte>.Shared.Rent(replayCount);
try
{
available = _ringBuffer.ReadFromEnd(bytesFromEnd, tempBuffer, 0, replayCount);
tempBuffer.AsSpan(0, available).CopyTo(buffer.Span.Slice(offset));
}
finally
{
ArrayPool<byte>.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<byte>.Shared.Rent(read);
try
{
buffer.Slice(offset, read).Span.CopyTo(tempBuffer);
_ringBuffer!.Write(tempBuffer, 0, read);
}
finally
{
ArrayPool<byte>.Shared.Return(tempBuffer);
}
}
streamPosition += read;
_logicalPosition += read;
totalRead += read;

124
tests/SharpCompress.Test/Streams/DeflateReadAllocationTests.cs
View File

@@ -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()
{