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

Add native ValueTask memory reads through XZ filter chains

Browse Source
This commit is contained in:
Adam Hathcock
2026-10-24 13:09:51 +01:00
parent de7a5d9ad4
commit 31f3476fc9
14 changed files with 542 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
docs/API.md
View File

@@ -888,6 +888,31 @@ using (var reader = ReaderFactory.OpenReader(stream))
---
## Low-Level XZ Memory Reads
On .NET 6 and later, `SharpCompress.Compressors.Xz.XZStream` and `XZBlock`
override `ReadAsync(Memory<byte>, CancellationToken)` with a native ValueTask
read path. Array-backed and custom/native memory are supported; CRC32, CRC64,
and SHA-256 checks remain enabled. SHA-256 uses a bounded pooled scratch buffer
when the destination memory cannot expose an array. Read through EOF to validate
the complete stream, including its final block check.
```csharp
using var input = File.OpenRead("input.tar.xz");
using var xz = new SharpCompress.Compressors.Xz.XZStream(input);
using var output = File.Create("output.tar");
var buffer = new byte[65536];
int read;
while ((read = await xz.ReadAsync(buffer.AsMemory(), cancellationToken)) != 0)
{
await output.WriteAsync(buffer.AsMemory(0, read), cancellationToken);
}
```
Disposing these low-level XZ streams releases their owned filter buffers without
closing the caller-provided input; dispose the input separately. The existing
array-based async signatures are unchanged.
## CRC-Checking Stream Lifetime
`SharpCompressStream` rejects reading and starting recording after disposal. Its pooled ring buffer is released even when disposal of the underlying stream fails.

31
docs/XZ_PROFILING.md
View File

@@ -198,6 +198,9 @@ not explain the adverse elapsed result by themselves, nor justify a lock rewrite
## Remaining opportunities
These describe the profiled pooling baseline; completed follow-ups are recorded
separately below rather than rewriting the historical measurements.
- Async LZMA2 currently calls the per-symbol async decoder, whereas sync decoding
can use the optimized range-decoder path. A future experiment should separate
asynchronous input refills from CPU-only decoding. It must not replace async I/O
@@ -237,6 +240,34 @@ and net48 test compilation passes. The remaining public `Task<int>` result canno
be removed without changing that API; a native memory-based XZ chain remains a
separate follow-up.
## Follow-up: native XZ memory read chain
The next change adds native memory/ValueTask reads to `XZStream` and `XZBlock`
on modern targets, carries memory-based I/O through BCJ refills, and completes
already-buffered reads without traversing multiple async wrappers. All buffered
bytes still update the block check exactly once. Array API signatures and sync
array-based I/O remain unchanged. Native/custom destination memory works with
all four XZ check types; SHA-256 uses bounded pooled scratch when required.
A fresh comparison against the first follow-up (`de7a5d9`) uses the built-in
1 MiB x86+LZMA2 fixture, 16-byte memory reads at offset 5, two warmups, seven
measured complete drains, Release .NET 10, and disabled tiered compilation.
Median allocations fall from **14,180,048 to 15,056 bytes**, with elapsed medians
**22.399 to 19.422 ms**. Both builds retain checksum verification and return the
expected output length. This does not yet address the async range-decoder CPU
bottleneck or establish file-backed throughput improvements.
Artifacts are `fix2-before.log`, `fix2-after.log`, and `fix2-harness/` in the parent
directory. The harness's before/after DLLs were explicitly bound and hash-checked
after observing that MSBuild reused an earlier reference when switching HintPath
properties; `fix2-harness/comparison.json` records the actual measured hashes.
All **764 targeted Release tests** pass, including yielding async-only input,
seekable/forward-only streams, native/custom memory, mixed async overloads,
bounded allocations, empty/canceled reads, and checksum corruption. All six
library targets and net48 test compilation pass. The issue sample still matches
liblzma in all 24 read-size/overload/offset combinations.
## Artifacts and reproduction
The controlled comparison is isolated from preliminary sample-file measurements:

12
src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs
View File

@@ -18,6 +18,16 @@ public partial class ArmFilter
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
)
{
if (
BaseStream is BcjFilterStream filter
&& filter.TryReadBuffered(buffer, cancellationToken, out var read)
)
{
return read;
}
return await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
}
#endif
}

12
src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs
View File

@@ -18,6 +18,16 @@ public partial class ArmThumbFilter
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
)
{
if (
BaseStream is BcjFilterStream filter
&& filter.TryReadBuffered(buffer, cancellationToken, out var read)
)
{
return read;
}
return await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
}
#endif
}

7
src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs
View File

@@ -41,15 +41,20 @@ internal sealed partial class BcjFilterStream
return total;
}
[Zomp.SyncMethodGenerator.CreateSyncVersion]
private async ValueTask FillAsync(CancellationToken cancellationToken)
{
KeepUnfilteredTail();
while (_filtered == 0 && !_endReached)
{
#if LEGACY_DOTNET
var bytesRead = await BaseStream
.ReadAsync(_buffer, _size, BufferSize - _size, cancellationToken)
.ConfigureAwait(false);
#else
var bytesRead = await BaseStream
.ReadAsync(_buffer.AsMemory(_size, BufferSize - _size), cancellationToken)
.ConfigureAwait(false);
#endif
ConvertBuffered(bytesRead);
}
}

30
src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs
View File

@@ -90,6 +90,25 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream
return true;
}
#if !LEGACY_DOTNET
internal bool TryReadBuffered(
Memory<byte> buffer,
CancellationToken cancellationToken,
out int read
)
{
read = 0;
if (buffer.Length > _filtered - _position)
{
return false;
}
ThrowIfDisposed();
cancellationToken.ThrowIfCancellationRequested();
read = CopyBuffered(buffer.Span);
return true;
}
#endif
private int CopyBuffered(Span<byte> destination)
{
var count = Math.Min(destination.Length, _filtered - _position);
@@ -108,6 +127,17 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream
_filtered = 0;
}
private void Fill()
{
KeepUnfilteredTail();
while (_filtered == 0 && !_endReached)
{
// Preserve the sync array-based I/O contract; only async reads prefer Memory.
var bytesRead = BaseStream.Read(_buffer, _size, BufferSize - _size);
ConvertBuffered(bytesRead);
}
}
private void ConvertBuffered(int bytesRead)
{
_size += bytesRead;

12
src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs
View File

@@ -18,6 +18,16 @@ public partial class IA64Filter
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
)
{
if (
BaseStream is BcjFilterStream filter
&& filter.TryReadBuffered(buffer, cancellationToken, out var read)
)
{
return read;
}
return await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
}
#endif
}

12
src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs
View File

@@ -18,6 +18,16 @@ public partial class PowerPCFilter
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
)
{
if (
BaseStream is BcjFilterStream filter
&& filter.TryReadBuffered(buffer, cancellationToken, out var read)
)
{
return read;
}
return await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
}
#endif
}

12
src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs
View File

@@ -18,6 +18,16 @@ public partial class SparcFilter
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
)
{
if (
BaseStream is BcjFilterStream filter
&& filter.TryReadBuffered(buffer, cancellationToken, out var read)
)
{
return read;
}
return await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
}
#endif
}

12
src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs
View File

@@ -18,6 +18,16 @@ public partial class X86Filter
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
)
{
if (
BaseStream is BcjFilterStream filter
&& filter.TryReadBuffered(buffer, cancellationToken, out var read)
)
{
return read;
}
return await BaseStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
}
#endif
}

113
src/SharpCompress/Compressors/Xz/XZBlock.Async.cs
View File

@@ -2,6 +2,7 @@ using System;
using System.Buffers;
using System.IO;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using SharpCompress.Common;
@@ -56,6 +57,118 @@ public sealed partial class XZBlock
return bytesRead;
}
#if !LEGACY_DOTNET
internal bool TryReadBuffered(
Memory<byte> buffer,
CancellationToken cancellationToken,
out int read
)
{
read = 0;
if (
_streamConnected
&& !_endOfStream
&& _decomStream is Filters.BlockFilter filter
&& filter.BaseStream is Filters.BcjFilterStream buffered
&& buffered.TryReadBuffered(buffer, cancellationToken, out read)
)
{
// Buffered data still contributes to the same block check, exactly once.
UpdateCheck(buffer.Slice(0, read));
return true;
}
return false;
}
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
ThrowIfDisposed();
cancellationToken.ThrowIfCancellationRequested();
if (buffer.IsEmpty)
{
return 0;
}
if (!HeaderIsLoaded)
{
await LoadHeaderAsync(cancellationToken).ConfigureAwait(false);
}
if (!_streamConnected)
{
ConnectStream();
}
var bytesRead = 0;
if (!_endOfStream)
{
bytesRead = await _decomStream
.NotNull()
.ReadAsync(buffer, cancellationToken)
.ConfigureAwait(false);
UpdateCheck(buffer.Slice(0, bytesRead));
}
if (bytesRead != buffer.Length)
{
_endOfStream = true;
}
if (_endOfStream && !_paddingSkipped)
{
await SkipPaddingAsync(cancellationToken).ConfigureAwait(false);
}
if (_endOfStream && !_crcChecked)
{
await CheckCrcAsync(cancellationToken).ConfigureAwait(false);
}
return bytesRead;
}
private void UpdateCheck(ReadOnlyMemory<byte> bytes)
{
if (bytes.IsEmpty || _checkType == CheckType.NONE)
{
return;
}
switch (_checkType)
{
case CheckType.CRC32:
_crc32 = Crc32.Update(_crc32, bytes.Span);
break;
case CheckType.CRC64:
_crc64 = Crc64.UpdateXz(_crc64, bytes.Span);
break;
case CheckType.SHA256:
if (MemoryMarshal.TryGetArray(bytes, out var array))
{
_sha256
.NotNull()
.TransformBlock(array.Array!, array.Offset, array.Count, null, 0);
}
else
{
// HashAlgorithm's streaming API needs arrays. Bound the fallback copy for
// native/custom memory instead of allocating a request-sized temporary array.
var scratch = ArrayPool<byte>.Shared.Rent(8192);
try
{
while (!bytes.IsEmpty)
{
var count = Math.Min(bytes.Length, 8192);
bytes.Span.Slice(0, count).CopyTo(scratch);
_sha256.NotNull().TransformBlock(scratch, 0, count, null, 0);
bytes = bytes.Slice(count);
}
}
finally
{
ArrayPool<byte>.Shared.Return(scratch);
}
}
break;
}
}
#endif
private async ValueTask SkipPaddingAsync(CancellationToken cancellationToken = default)
{
var bytes = (BaseStream.Position - _startPosition) % 4;

69
src/SharpCompress/Compressors/Xz/XZStream.Async.cs
View File

@@ -61,6 +61,75 @@ public sealed partial class XZStream
return bytesRead;
}
#if !LEGACY_DOTNET
/// <summary>
/// Read and validate XZ data without adapting memory reads through Task-based array overloads.
/// </summary>
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
ThrowIfDisposed();
cancellationToken.ThrowIfCancellationRequested();
if (buffer.IsEmpty || _endOfStream)
{
return 0;
}
if (!HeaderIsRead)
{
await ReadHeaderAsync(cancellationToken).ConfigureAwait(false);
}
if (
_currentBlock is not null
&& _currentBlock.TryReadBuffered(buffer, cancellationToken, out var buffered)
)
{
return buffered;
}
var bytesRead = await ReadBlocksAsync(buffer, cancellationToken).ConfigureAwait(false);
if (bytesRead < buffer.Length)
{
_endOfStream = true;
await ReadIndexAsync(cancellationToken).ConfigureAwait(false);
await ReadFooterAsync(cancellationToken).ConfigureAwait(false);
}
return bytesRead;
}
private async ValueTask<int> ReadBlocksAsync(
Memory<byte> buffer,
CancellationToken cancellationToken
)
{
var bytesRead = 0;
if (_currentBlock is null)
{
NextBlock();
}
while (bytesRead < buffer.Length)
{
try
{
var remaining = buffer.Slice(bytesRead);
var justRead = await _currentBlock
.ReadAsync(remaining, cancellationToken)
.ConfigureAwait(false);
bytesRead += justRead;
if (justRead < remaining.Length)
{
NextBlock();
}
}
catch (XZIndexMarkerReachedException)
{
break;
}
}
return bytesRead;
}
#endif
/// <summary>
/// Asynchronously reads and validates the XZ header.
/// </summary>

7
tests/SharpCompress.Test/Xz/Filters/BcjBufferLifetimeTests.cs
View File

@@ -210,5 +210,12 @@ public class BcjBufferLifetimeTests
int count,
CancellationToken cancellationToken
) => Task.FromException<int>(new IOException("Injected read failure"));
#if !LEGACY_DOTNET
public override ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
) => new(Task.FromException<int>(new IOException("Injected read failure")));
#endif
}
}

195
tests/SharpCompress.Test/Xz/XZMemoryReadTests.cs Normal file
View File

@@ -0,0 +1,195 @@
#if !LEGACY_DOTNET
using System;
using System.Buffers;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using SharpCompress.Common;
using SharpCompress.Compressors.Xz;
using SharpCompress.IO;
using SharpCompress.Test.Mocks;
using Xunit;
namespace SharpCompress.Test.Xz;
public class XzMemoryReadTests
{
// liblzma-generated x86+LZMA2 streams containing 2,000 repetitions of the expected line.
public static TheoryData<string> Checks =>
new()
{
"/Td6WFoAAAD/EtlBAgEEACEBAADaM60D4LOvAGBdACaZSach4xxvjplweW8sBThpghKnl3v62Kohix9VuFZ6Y6UNYeuBVsU4MUwSbr1yebmoUdjYLCUnhap0oivonKNF3AVwgPWqZ6UcPnRRZ9QNV+m11evb/jb/zCgx+5gpAAAAAXSw5wIAAPE+DQuoAAr8AgAAAAAAWVo=",
"/Td6WFoAAAFpIt42AgEEACEBAADaM60D4LOvAGBdACaZSach4xxvjplweW8sBThpghKnl3v62Kohix9VuFZ6Y6UNYeuBVsU4MUwSbr1yebmoUdjYLCUnhap0oivonKNF3AVwgPWqZ6UcPnRRZ9QNV+m11evb/jb/zCgx+5gpAAAuAAwEAAF4sOcCAACK/s98PjANiwIAAAAAAVla",
"/Td6WFoAAATm1rRGAgEEACEBAADaM60D4LOvAGBdACaZSach4xxvjplweW8sBThpghKnl3v62Kohix9VuFZ6Y6UNYeuBVsU4MUwSbr1yebmoUdjYLCUnhap0oivonKNF3AVwgPWqZ6UcPnRRZ9QNV+m11evb/jb/zCgx+5gpAADdy3Bd9bawwgABfLDnAgAAnLxe57HEZ/sCAAAAAARZWg==",
"/Td6WFoAAArh+wyhAgEEACEBAADaM60D4LOvAGBdACaZSach4xxvjplweW8sBThpghKnl3v62Kohix9VuFZ6Y6UNYeuBVsU4MUwSbr1yebmoUdjYLCUnhap0oivonKNF3AVwgPWqZ6UcPnRRZ9QNV+m11evb/jb/zCgx+5gpAABZo8OamypsYGhitd9QtvP1Bfo7M36ti3jv0rEHlcfmJAABlAGw5wIABUMr+7bp3xwCAAAAAApZWg==",
};
public static IEnumerable<object[]> Cases()
{
foreach (var check in Checks)
{
foreach (var nonArray in new[] { false, true })
{
foreach (var forwardOnly in new[] { false, true })
{
foreach (var suspend in new[] { false, true })
{
yield return new object[] { check.Data, nonArray, forwardOnly, suspend };
}
}
}
}
}
[Theory]
[MemberData(nameof(Cases))]
public async Task MemoryReadsValidateAllChecksAndMemoryKinds(
string compressed,
bool nonArray,
bool forwardOnly,
bool suspend
)
{
using var input = suspend
? new YieldingInput(Convert.FromBase64String(compressed))
: new MemoryStream(Convert.FromBase64String(compressed));
using var asyncOnly = new AsyncOnlyStream(input, disposeStream: false);
using var forward = new ForwardOnlyStream(asyncOnly);
using var tracked = SharpCompressStream.Create(forward);
using var xz = new XZStream(forwardOnly ? tracked : asyncOnly);
var bytes = new byte[10013];
using var manager = new NonArrayMemory(bytes);
var memory = (nonArray ? manager.Memory : bytes.AsMemory()).Slice(5, 10000);
using var output = new MemoryStream();
while (true)
{
Array.Fill(bytes, (byte)0xCC);
var read = await xz.ReadAsync(memory, CancellationToken.None);
Assert.True(bytes.AsSpan(0, 5).IndexOfAnyExcept((byte)0xCC) < 0);
Assert.True(bytes.AsSpan(5 + read).IndexOfAnyExcept((byte)0xCC) < 0);
if (read == 0)
{
break;
}
output.Write(bytes, 5, read);
}
Assert.Equal(Expected(), output.ToArray());
xz.Dispose();
Assert.True(input.CanRead);
await Assert.ThrowsAsync<ObjectDisposedException>(async () => await xz.ReadAsync(memory));
}
[Fact]
public async Task MemoryAndArrayReadsCanBeInterleaved()
{
using var input = new MemoryStream(Convert.FromBase64String(Checks.Last().Data));
using var xz = new XZStream(input);
var buffer = new byte[17];
using var output = new MemoryStream();
for (var iteration = 0; ; iteration++)
{
var read =
iteration % 2 == 0
? await xz.ReadAsync(buffer.AsMemory(), CancellationToken.None)
: await xz.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None);
if (read == 0)
{
break;
}
output.Write(buffer, 0, read);
}
Assert.Equal(Expected(), output.ToArray());
}
[Fact]
public async Task EmptyAndCanceledMemoryReadsDoNotConsumeInput()
{
using var input = new MemoryStream(Convert.FromBase64String(Checks.First().Data));
using var xz = new XZStream(input);
Assert.Equal(0, await xz.ReadAsync(Memory<byte>.Empty));
Assert.Equal(0, input.Position);
using var cancellation = new CancellationTokenSource();
cancellation.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () =>
await xz.ReadAsync(new byte[1].AsMemory(), cancellation.Token)
);
Assert.Equal(0, input.Position);
}
[Fact]
public async Task NativeMemoryPathRejectsCorruptChecksum()
{
var compressed = Convert.FromBase64String(Checks.Last().Data);
compressed[compressed.Length - 29] ^= 1;
using var input = new MemoryStream(compressed);
using var xz = new XZStream(input);
using var manager = new NonArrayMemory(new byte[8192]);
await Assert.ThrowsAsync<InvalidFormatException>(async () =>
{
while (await xz.ReadAsync(manager.Memory) != 0) { }
});
}
#if !DEBUG
[Fact]
public async Task SmallMemoryReadsAvoidPerReadTaskAllocations()
{
using var input = new MemoryStream(Convert.FromBase64String(Checks.ElementAt(2).Data));
using var xz = new XZStream(input);
using var manager = new NonArrayMemory(new byte[16]);
var memory = manager.Memory;
var total = await xz.ReadAsync(memory);
var start = GC.GetAllocatedBytesForCurrentThread();
int read;
while ((read = await xz.ReadAsync(memory)) != 0)
{
total += read;
}
var allocated = GC.GetAllocatedBytesForCurrentThread() - start;
Assert.Equal(46000, total);
Assert.InRange(allocated, 0, 65536);
}
#endif
private static byte[] Expected() =>
Encoding.ASCII.GetBytes(string.Concat(Enumerable.Repeat("Memory-backed XZ reads\n", 2000)));
private sealed class NonArrayMemory(byte[] bytes) : MemoryManager<byte>
{
public override Span<byte> GetSpan() => bytes;
public override MemoryHandle Pin(int elementIndex = 0) => throw new NotSupportedException();
public override void Unpin() { }
protected override void Dispose(bool disposing) { }
}
private sealed class YieldingInput(byte[] bytes) : MemoryStream(bytes)
{
public override async Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
await Task.Yield();
cancellationToken.ThrowIfCancellationRequested();
return base.Read(buffer, offset, count);
}
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
await Task.Yield();
return await base.ReadAsync(buffer, cancellationToken);
}
}
}
#endif