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

Merge pull request #1477 from adamhathcock/perf/xz-async-decoding

Browse Source 
Optimize XZ async reads and buffered LZMA2 decoding
This commit is contained in:
Adam Hathcock
2026-10-26 08:27:00 +00:00
committed by GitHub
parent 0a716d4236 0c6b575135
commit 8fa17607d8
23 changed files with 1283 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
docs/API.md
View File

@@ -888,12 +888,45 @@ 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.
`LzmaStream` also rejects reads and writes after disposal. Failed initialization releases internal buffers without closing the supplied stream; successful construction retains the specified `leaveOpen` semantics. Disposal releases internal buffers even if finalization or underlying stream disposal throws.
On modern .NET targets, async LZMA2 reads may prefetch the current compressed
chunk (at most 64 KiB) using async I/O before producing output. They never prefetch
the following chunk header. Decoding the buffered payload uses CPU-only batches
of at most 16 KiB of output between cancellation checks. Cancellation or failure
during a partial fill does not permit unsafe decoding of uninitialized bytes;
disposing the stream returns the input lease. Raw LZMA and legacy-target async
decoding retain their existing streaming paths.
`SharpCompress.Compressors.LZMA.Utilities.CrcCheckStream` validates its expected CRC on disposal. A mismatch throws `ArchiveOperationException`, but pooled buffers are still released. Repeated disposal is safe, including after failed CRC validation. Once disposed, `CanWrite` is `false`, and `Write` / `WriteAsync` throw `ObjectDisposedException`.
## Related Documentation

114
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
@@ -215,6 +218,117 @@ not explain the adverse elapsed result by themselves, nor justify a lock rewrite
no evidence here to justify pooling them. Checksum work must remain enabled in
future throughput comparisons.
## Follow-up: remove duplicate BCJ result tasks
On `perf/xz-async-decoding`, the first follow-up keeps the internal BCJ read as
a `ValueTask<int>` and creates a task only at the existing public array boundary.
Already-buffered reads complete without entering an async state machine. Custom
subclasses installing a synchronously throwing input stream retain the previous
faulted/canceled-task behavior; sync signatures are unchanged.
A fresh before/after comparison against the pooled baseline uses the same isolated
1 MiB x86 workload, 16-byte array reads, two warmups, seven measured drains,
Release .NET 10, and disabled tiered compilation. Median allocations fall from
**9,438,376 to 4,719,784 bytes**; median elapsed time is **1.695 to 0.963 ms**.
This is a small-read wrapper result, not a complete XZ throughput claim. Logs and
the follow-up harness are in the parent artifact directory as `fix1-before.log`,
`fix1-after.log`, and `fix1-harness/`.
All **728 targeted Release tests** pass, including allocation guards for all six
architectures and fault/cancellation regressions. All six library targets build,
and net48 test compilation passes. The remaining public `Task<int>` result cannot
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.
## Follow-up: bounded async LZMA2 fast decoding
The third follow-up changes modern-target LZMA2 async decoding, not raw LZMA or
legacy-target decoding. Each compressed LZMA2 chunk declares at most 64 KiB of
compressed data. That payload is filled asynchronously into the range decoder's
existing pooled fast-input lease. The existing optimized decoder then performs
CPU-only work on that buffer; it never makes a synchronous call to the supplied
input stream. There is no second request-sized compressed-input allocation.
CPU-only batches are capped at **16 KiB of output between cancellation checks**,
even when the caller asks for much more. The fast probability bank is reused
across continuation chunks and reinitialized on property/state resets. Scalar
models are not also allocated for that path. The decoder still validates LZMA2
compressed/uncompressed counts, end markers, pending output, and outer XZ checks.
Input refills are bounded by the declared payload, not the rented array's
capacity. Incomplete fills cannot enter unsafe decoding; truncation retains the
`DataErrorException` category, and disposal returns partially filled leases.
Cancellation after a complete payload fill can resume without losing buffered
data. Tests exercise yielding async-only non-seekable input, one-byte/short reads,
following-chunk sentinels, malformed/truncated payloads, retries, oversized
poisoned rents, cancellation, model reuse, and disposal.
A fresh comparison against `31f3476` reuses the complete **32 MiB** mixed XZ
workload and 64 KiB native-memory requests. Each version/input runs in a fresh
process with two warmups and seven measured complete drains; version order is
reversed for the file scenario. Release .NET 10 uses disabled tiered compilation.
No builds or tests compete with these measurements.
| Input / memory-async API | Before ms | After ms | Before process CPU ms | After process CPU ms | Before allocated bytes | After allocated bytes |
| --- | ---: | ---: | ---: | ---: | ---: | ---: |
| Memory | 5,963.0 | 769.2 | 5,961.2 | 769.2 | 80,672 | 72,600 |
| File | 6,812.9 | 768.7 | 6,932.0 | 780.6 | 6,195,096 | 3,721,104 |
This workload's elapsed medians improve by approximately **7.8x memory-backed**
and **8.9x file-backed**; process CPU reductions support the decoding explanation,
not merely reduced waiting. These results do not establish a speedup for raw
LZMA, legacy targets, other data distributions, or cold storage. File allocations
still include suspension costs and pool misses. All original checksum work remains.
Artifacts and hash-checked before/after DLLs are in `fix3-harness/` in the parent
directory. Its `evaluate.py`, `results.json`, `library-hashes.json`, and raw logs
record the comparison. Both versions decode to the same byte count; the updated
build additionally passes full SHA-256 verification of this workload through all
three read APIs and all 24 byte-for-byte issue-sample combinations.
All **969 targeted Release tests**, including 7z, pass. All six library targets
and net48 test compilation pass; all 84 benchmark cases still smoke-test cleanly.
The final complete **.NET 10 Release suite passes: 3,629 tests, no failures or skips**.
A final measured-phase async memory trace confirms the intended path change:
**81.4%** of workload leaf samples are in `CodeFastCore`, **8.2%** in CRC64,
and **5.5%** in the fast matched-literal helper. Per-bit async decoder helpers no
longer dominate. This trace contains 1,478 workload samples, one complete measured
phase, and zero lost events. Its raw trace, Speedscope file, and analysis JSON are
`fix3-harness/after-memory.*` / `after-memory-analysis.json`. Sampling limitations
remain the same as in the original evaluation.
The checksum tables remain shared and unchanged: the profiling report identified
no per-block table-allocation problem to fix.
## Artifacts and reproduction
The controlled comparison is isolated from preliminary sample-file measurements:

3
src/SharpCompress/Compressors/LZMA/LzmaDecoder.Fast.cs
View File

@@ -20,7 +20,8 @@ namespace SharpCompress.Compressors.LZMA;
// call per byte.
//
// These tables are separate from (and not kept in sync with) the BitDecoder-based tables used
// by the async decode path (LzmaDecoder.Async.cs), which remains unchanged on every target.
// by scalar/per-symbol async decoding (LzmaDecoder.Async.cs). Modern LZMA2 async streams also
// use this fast bank after asynchronously buffering a bounded compressed chunk.
// Both sets are kept up to date by SetDecoderProperties/Init, so either decode path can be used
// on a given Decoder instance, as long as sync and async decoding are not interleaved on the
// same instance.

80
src/SharpCompress/Compressors/LZMA/LzmaStream.Async.cs
View File

@@ -91,6 +91,9 @@ public partial class LzmaStream
private async ValueTask DecodeChunkHeaderAsync(CancellationToken cancellationToken = default)
{
#if !LEGACY_DOTNET
_bufferedAsyncChunkReady = false;
#endif
var headerBuffer = GetAsyncHeaderBuffer();
await _inputStream!
.ReadExactAsync(headerBuffer, 0, 1, cancellationToken)
@@ -159,7 +162,18 @@ public partial class LzmaStream
_decoder.SetDecoderProperties(Properties);
}
#if !LEGACY_DOTNET
await _rangeDecoder
.InitLzma2ChunkAsync(
_inputStream,
checked((int)_rangeDecoderLimit),
cancellationToken
)
.ConfigureAwait(false);
_bufferedAsyncChunkReady = true;
#else
await _rangeDecoder.InitAsync(_inputStream, cancellationToken).ConfigureAwait(false);
#endif
}
else if (control > 0x02)
{
@@ -176,6 +190,25 @@ public partial class LzmaStream
}
}
private ValueTask<bool> DecodeBlockAsync(CancellationToken cancellationToken)
{
#if !LEGACY_DOTNET
if (_isLzma2)
{
cancellationToken.ThrowIfCancellationRequested();
if (!_bufferedAsyncChunkReady)
{
// Do not enter the pointer-based decoder after a canceled/failed payload fill.
throw new DataErrorException();
}
#pragma warning disable VSTHRD103 // The bounded LZMA2 payload is already in memory; this call performs CPU work only.
return new ValueTask<bool>(_decoder!.Code(_dictionarySize, _outWindow, _rangeDecoder));
#pragma warning restore VSTHRD103
}
#endif
return _decoder!.CodeAsync(_dictionarySize, _outWindow, _rangeDecoder, cancellationToken);
}
public override async Task<int> ReadAsync(
byte[] buffer,
int offset,
@@ -225,6 +258,13 @@ public partial class LzmaStream
{
toProcess = (int)_availableBytes;
}
#if !LEGACY_DOTNET
if (_isLzma2 && !_uncompressedChunk)
{
// Bound CPU-only work between cancellation checks, even for huge caller buffers.
toProcess = Math.Min(toProcess, 16 * 1024);
}
#endif
_outWindow.SetLimit(toProcess);
if (_uncompressedChunk)
@@ -233,11 +273,7 @@ public partial class LzmaStream
.CopyStreamAsync(_inputStream, toProcess, cancellationToken)
.ConfigureAwait(false);
}
else if (
await _decoder!
.CodeAsync(_dictionarySize, _outWindow, _rangeDecoder, cancellationToken)
.ConfigureAwait(false)
)
else if (await DecodeBlockAsync(cancellationToken).ConfigureAwait(false))
{
HandleEndMarker();
}
@@ -265,16 +301,7 @@ public partial class LzmaStream
)
{
_outWindow.SetLimit(toProcess + 1);
if (
!await _decoder!
.CodeAsync(
_dictionarySize,
_outWindow,
_rangeDecoder,
cancellationToken
)
.ConfigureAwait(false)
)
if (!await DecodeBlockAsync(cancellationToken).ConfigureAwait(false))
{
_rangeDecoder.ReleaseStream();
throw new DataErrorException();
@@ -356,6 +383,12 @@ public partial class LzmaStream
{
toProcess = (int)_availableBytes;
}
#if !LEGACY_DOTNET
if (_isLzma2 && !_uncompressedChunk)
{
toProcess = Math.Min(toProcess, 16 * 1024);
}
#endif
_outWindow.SetLimit(toProcess);
if (_uncompressedChunk)
@@ -364,11 +397,7 @@ public partial class LzmaStream
.CopyStreamAsync(_inputStream, toProcess, cancellationToken)
.ConfigureAwait(false);
}
else if (
await _decoder!
.CodeAsync(_dictionarySize, _outWindow, _rangeDecoder, cancellationToken)
.ConfigureAwait(false)
)
else if (await DecodeBlockAsync(cancellationToken).ConfigureAwait(false))
{
HandleEndMarker();
}
@@ -396,16 +425,7 @@ public partial class LzmaStream
)
{
_outWindow.SetLimit(toProcess + 1);
if (
!await _decoder!
.CodeAsync(
_dictionarySize,
_outWindow,
_rangeDecoder,
cancellationToken
)
.ConfigureAwait(false)
)
if (!await DecodeBlockAsync(cancellationToken).ConfigureAwait(false))
{
_rangeDecoder.ReleaseStream();
throw new DataErrorException();

3
src/SharpCompress/Compressors/LZMA/LzmaStream.cs
View File

@@ -32,6 +32,9 @@ public partial class LzmaStream : Stream, IStreamStack, IAsyncDisposable
private bool _uncompressedChunk;
private bool _needDictReset = true;
private bool _needProps = true;
#if !LEGACY_DOTNET
private bool _bufferedAsyncChunkReady;
#endif
private readonly Encoder? _encoder;
private byte[]? _asyncHeaderBuffer;

50
src/SharpCompress/Compressors/LZMA/RangeCoder/RangeCoder.Async.cs
View File

@@ -1,5 +1,6 @@
#nullable disable
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
@@ -81,6 +82,55 @@ internal partial class Encoder
internal partial class Decoder
{
#if !LEGACY_DOTNET
internal async ValueTask InitLzma2ChunkAsync(
Stream stream,
int compressedSize,
CancellationToken cancellationToken
)
{
// LZMA2 explicitly bounds each compressed payload to at most 64 KiB. Fill the
// existing fast input lease asynchronously; decoding can then be CPU-only,
// without a synchronous read from the caller's stream or the next chunk header.
if (compressedSize < 5 || compressedSize > FastBufferSize)
{
throw new DataErrorException();
}
cancellationToken.ThrowIfCancellationRequested();
_fastBuffer ??= System.Buffers.ArrayPool<byte>.Shared.Rent(FastBufferSize);
var offset = 0;
while (offset < compressedSize)
{
cancellationToken.ThrowIfCancellationRequested();
var read = await stream
.ReadAsync(_fastBuffer.AsMemory(offset, compressedSize - offset), cancellationToken)
.ConfigureAwait(false);
if (read == 0)
{
// A truncated compressed payload is LZMA2 data corruption, not an
// incomplete outer header. Keep the decoder's existing error category.
throw new DataErrorException();
}
offset += read;
}
_stream = null;
_range = uint.MaxValue;
_code = 0;
for (var i = 0; i < 5; i++)
{
_code = (_code << 8) | _fastBuffer[i];
}
_total = 5;
_fastLimit = compressedSize;
_fastBufferPos = 5;
_fastBufferLen = compressedSize;
// A corrupt payload may request more range bytes. Preserve the fast decoder's
// EOF sentinel behavior, but never fetch those bytes from a following chunk.
_fastEndOfStream = true;
_fastBufferSafeUnbounded = false;
}
#endif
private byte[] SingleByteBuffer => _singleByteBuffer ??= new byte[1];
public async ValueTask InitAsync(Stream stream, CancellationToken cancellationToken = default)

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

@@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters;
public partial class ArmFilter
{
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => await BaseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
) => ReadBaseAsync(buffer, offset, count, cancellationToken);
#if !LEGACY_DOTNET
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
}

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

@@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters;
public partial class ArmThumbFilter
{
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => await BaseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
) => ReadBaseAsync(buffer, offset, count, cancellationToken);
#if !LEGACY_DOTNET
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
}

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

@@ -6,24 +6,22 @@ namespace SharpCompress.Compressors.Xz.Filters;
internal sealed partial class BcjFilterStream
{
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => ReadCoreAsync(buffer, offset, count, cancellationToken).AsTask();
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
internal async ValueTask<int> ReadCoreAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
ThrowIfDisposed();
ThrowHelper.ThrowIfNull(buffer);
if (offset < 0 || offset > buffer.Length)
{
throw new ArgumentOutOfRangeException(nameof(offset));
}
if (count < 0 || count > buffer.Length - offset)
{
throw new ArgumentOutOfRangeException(nameof(count));
}
ValidateRead(buffer, offset, count);
cancellationToken.ThrowIfCancellationRequested();
var total = 0;
@@ -43,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);
}
}

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

@@ -1,6 +1,7 @@
using System;
using System.Buffers;
using System.IO;
using System.Threading;
namespace SharpCompress.Compressors.Xz.Filters;
@@ -34,6 +35,9 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream
public override bool CanRead => !_isDisposed && BaseStream.CanRead;
public override int Read(byte[] buffer, int offset, int count) =>
ReadCore(buffer, offset, count);
protected override void Dispose(bool disposing)
{
if (disposing && !_isDisposed)
@@ -53,6 +57,58 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream
}
}
private void ValidateRead(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();
ThrowHelper.ThrowIfNull(buffer);
if (offset < 0 || offset > buffer.Length)
{
throw new ArgumentOutOfRangeException(nameof(offset));
}
if (count < 0 || count > buffer.Length - offset)
{
throw new ArgumentOutOfRangeException(nameof(count));
}
}
internal bool TryReadBuffered(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken,
out int read
)
{
read = 0;
if (count < 0 || count > _filtered - _position)
{
return false;
}
ValidateRead(buffer, offset, count);
cancellationToken.ThrowIfCancellationRequested();
read = CopyBuffered(buffer.AsSpan(offset, count));
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);
@@ -71,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;

49
src/SharpCompress/Compressors/Xz/Filters/BlockFilter.Async.cs Normal file
View File

@@ -0,0 +1,49 @@
using System;
using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
namespace SharpCompress.Compressors.Xz.Filters;
public abstract partial class BlockFilter
{
internal Task<int> ReadBaseAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
try
{
// A single adapter belongs at the public Task boundary, not at every wrapper.
if (BaseStream is BcjFilterStream filter)
{
// Buffered reads need no async state machine, only the public result task.
return filter.TryReadBuffered(
buffer,
offset,
count,
cancellationToken,
out var read
)
? Task.FromResult(read)
: filter.ReadCoreAsync(buffer, offset, count, cancellationToken).AsTask();
}
return BaseStream.ReadAsync(buffer, offset, count, cancellationToken);
}
catch (Exception exception)
{
// A custom subclass may install a stream that throws before returning a task.
// Preserve the old async method's faulted/canceled-task semantics in that case.
return RethrowReadExceptionAsync(exception);
}
}
private static async Task<int> RethrowReadExceptionAsync(Exception exception)
{
await Task.CompletedTask.ConfigureAwait(false);
ExceptionDispatchInfo.Capture(exception).Throw();
return 0;
}
}

5
src/SharpCompress/Compressors/Xz/Filters/BlockFilter.cs
View File

@@ -5,7 +5,7 @@ using SharpCompress.Common;
namespace SharpCompress.Compressors.Xz.Filters;
public abstract class BlockFilter : ReadOnlyStream
public abstract partial class BlockFilter : ReadOnlyStream
{
private Stream? _ownedStream;
@@ -15,6 +15,9 @@ public abstract class BlockFilter : ReadOnlyStream
BaseStream = _ownedStream = stream;
}
internal int ReadBase(byte[] buffer, int offset, int count) =>
BaseStream.Read(buffer, offset, count);
protected override void Dispose(bool disposing)
{
if (disposing)

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

@@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters;
public partial class IA64Filter
{
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => await BaseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
) => ReadBaseAsync(buffer, offset, count, cancellationToken);
#if !LEGACY_DOTNET
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
}

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

@@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters;
public partial class PowerPCFilter
{
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => await BaseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
) => ReadBaseAsync(buffer, offset, count, cancellationToken);
#if !LEGACY_DOTNET
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
}

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

@@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters;
public partial class SparcFilter
{
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => await BaseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
) => ReadBaseAsync(buffer, offset, count, cancellationToken);
#if !LEGACY_DOTNET
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
}

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

@@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters;
public partial class X86Filter
{
[Zomp.SyncMethodGenerator.CreateSyncVersion]
public override async Task<int> ReadAsync(
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => await BaseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
) => ReadBaseAsync(buffer, offset, count, cancellationToken);
#if !LEGACY_DOTNET
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>

296
tests/SharpCompress.Test/Streams/Lzma2AsyncBufferingTests.cs Normal file
View File

@@ -0,0 +1,296 @@
#if !LEGACY_DOTNET
using System;
using System.Buffers;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using SharpCompress.Compressors.LZMA;
using Xunit;
namespace SharpCompress.Test.Streams;
public class Lzma2AsyncBufferingTests
{
// liblzma raw LZMA2, 4 KiB dictionary, one compressed chunk followed by its end marker.
private static byte[] Packed() =>
Convert.FromBase64String(
"4Ii3AGldADEbyxGbBQY80wYkJNZUZojJCx/dH+R7NYxybfjsA6YFO6vj+t6Pjqr8q+3nmIKYLQdQf3feK0ZBZu3t1pE+S0M9VTCcsH3/PD9kncWqHcTIHqGAu441fiPlqtCCWEf8CABLZsj8KdrAAAA="
);
private static byte[] Expected() =>
Encoding.ASCII.GetBytes(
string.Concat(Enumerable.Repeat("bounded asynchronous LZMA2 payload\n", 1000))
);
[Theory]
[InlineData(false, 1)]
[InlineData(true, 1)]
[InlineData(false, 7)]
[InlineData(true, 7)]
[InlineData(false, 65536)]
[InlineData(true, 65536)]
public async Task UsesOnlyAsyncInputAndStopsAtChunkBoundary(bool memory, int maxRead)
{
var packed = Packed();
var expected = Expected();
using var input = new RecordingInput(
packed.Concat(new byte[] { 0xDE, 0xAD, 0xBE, 0xEF }).ToArray(),
maxRead
);
await using var decoder = await LzmaStream.CreateAsync(
new byte[] { 0 },
input,
packed.Length,
expected.Length,
leaveOpen: true
);
using var actual = new MemoryStream();
var buffer = new byte[17000];
int read;
while (
(
read = memory
? await decoder.ReadAsync(buffer.AsMemory(), CancellationToken.None)
: await decoder.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None)
) != 0
)
{
actual.Write(buffer, 0, read);
}
Assert.Equal(expected, actual.ToArray());
Assert.Equal(packed.Length, input.BytesRead);
Assert.InRange(input.MaxRequested, 5, 65536);
if (maxRead == 65536)
{
// Per-byte range decoding would issue more than 100 source reads here.
Assert.InRange(input.ReadCalls, 1, 12);
}
await decoder.DisposeAsync();
Assert.True(input.CanRead);
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task InvalidShortPayloadNeverReadsFollowingChunk(bool memory)
{
// The declared four-byte compressed payload is too short for range initialization.
var bytes = new byte[] { 0xE0, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0xDE, 0xAD };
using var input = new RecordingInput(bytes, 65536);
await using var decoder = await LzmaStream.CreateAsync(
new byte[] { 0 },
input,
-1,
-1,
leaveOpen: true
);
var buffer = new byte[1];
await Assert.ThrowsAsync<DataErrorException>(async () =>
{
if (memory)
{
await decoder.ReadAsync(buffer.AsMemory(), CancellationToken.None);
}
else
{
await decoder.ReadAsync(buffer, 0, 1, CancellationToken.None);
}
});
Assert.Equal(6, input.BytesRead);
// Retrying after a failed fill must not enter unsafe decoding with an uninitialized buffer.
await Assert.ThrowsAsync<DataErrorException>(() =>
decoder.ReadAsync(buffer, 0, 1, CancellationToken.None)
);
Assert.Equal(6, input.BytesRead);
}
[Fact]
public async Task FailedPayloadFillReturnsInputLeaseOnDisposal()
{
var packed = Packed();
using var cancellation = new CancellationTokenSource();
using var input = new RecordingInput(packed, 7);
input.AfterRead = position =>
{
if (position > 6)
{
cancellation.Cancel();
}
};
var decoder = await LzmaStream.CreateAsync(
new byte[] { 0 },
input,
-1,
-1,
leaveOpen: true
);
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
decoder.ReadAsync(new byte[1], 0, 1, cancellation.Token)
);
var range = GetField(decoder, "_rangeDecoder")!;
Assert.NotNull(GetField(range, "_fastBuffer"));
await decoder.DisposeAsync();
decoder.Dispose();
Assert.Null(GetField(range, "_fastBuffer"));
Assert.True(input.CanRead);
}
[Fact]
public async Task TruncatedPayloadCannotRetryUnsafeDecodingAndReturnsLease()
{
using var input = new RecordingInput(Packed().Take(20).ToArray(), 7);
var decoder = await LzmaStream.CreateAsync(
new byte[] { 0 },
input,
-1,
-1,
leaveOpen: true
);
var buffer = new byte[1];
await Assert.ThrowsAsync<DataErrorException>(() =>
decoder.ReadAsync(buffer, 0, 1, CancellationToken.None)
);
var range = GetField(decoder, "_rangeDecoder")!;
Assert.NotNull(GetField(range, "_fastBuffer"));
await Assert.ThrowsAsync<DataErrorException>(() =>
decoder.ReadAsync(buffer, 0, 1, CancellationToken.None)
);
Assert.Equal(20, input.BytesRead);
await decoder.DisposeAsync();
Assert.Null(GetField(range, "_fastBuffer"));
Assert.True(input.CanRead);
}
[Fact]
public async Task CancellationAfterPayloadBufferingCanResumeWithoutLosingData()
{
var packed = Packed();
var payloadEnd = 6 + ((packed[3] << 8) | packed[4]) + 1;
using var cancellation = new CancellationTokenSource();
using var input = new RecordingInput(packed, 65536);
input.AfterRead = position =>
{
if (position == payloadEnd)
{
cancellation.Cancel();
}
};
await using var decoder = await LzmaStream.CreateAsync(
new byte[] { 0 },
input,
-1,
-1,
leaveOpen: true
);
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
decoder.ReadAsync(new byte[1], 0, 1, cancellation.Token)
);
input.AfterRead = null;
using var output = new MemoryStream();
await decoder.CopyToAsync(output, 65536, CancellationToken.None);
Assert.Equal(Expected(), output.ToArray());
Assert.Equal(packed.Length, input.BytesRead);
}
[Fact]
public async Task OversizedPoisonedRentUsesOnlyDeclaredPayloadBytes()
{
var packed = Packed();
using var input = new RecordingInput(packed, 7);
await using var decoder = await LzmaStream.CreateAsync(
new byte[] { 0 },
input,
-1,
-1,
leaveOpen: true
);
var range = GetField(decoder, "_rangeDecoder")!;
var oversized = ArrayPool<byte>.Shared.Rent(128 * 1024);
Array.Fill(oversized, (byte)0xCC);
range
.GetType()
.GetField("_fastBuffer", BindingFlags.Instance | BindingFlags.NonPublic)!
.SetValue(range, oversized);
var first = new byte[1];
Assert.Equal(1, await decoder.ReadAsync(first.AsMemory(), CancellationToken.None));
Assert.Same(oversized, GetField(range, "_fastBuffer"));
Assert.Equal(((packed[3] << 8) | packed[4]) + 1, GetField(range, "_fastBufferLen"));
using var output = new MemoryStream();
output.Write(first);
await decoder.CopyToAsync(output, 65536, CancellationToken.None);
Assert.Equal(Expected(), output.ToArray());
Assert.Null(GetField(range, "_fastBuffer"));
}
private static object? GetField(object instance, string name) =>
instance
.GetType()
.GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!
.GetValue(instance);
private sealed class RecordingInput(byte[] bytes, int maxRead) : Stream
{
private bool disposed;
public int BytesRead { get; private set; }
public int MaxRequested { get; private set; }
public int ReadCalls { get; private set; }
public Action<int>? AfterRead { get; set; }
public override bool CanRead => !disposed;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => throw new NotSupportedException();
public override long Position
{
get => throw new NotSupportedException();
set => throw new NotSupportedException();
}
public override int Read(byte[] buffer, int offset, int count) =>
throw new NotSupportedException("No synchronous I/O");
public override int ReadByte() => throw new NotSupportedException("No synchronous I/O");
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) => ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
await Task.Yield();
cancellationToken.ThrowIfCancellationRequested();
MaxRequested = Math.Max(MaxRequested, buffer.Length);
ReadCalls++;
var count = Math.Min(Math.Min(buffer.Length, maxRead), bytes.Length - BytesRead);
bytes.AsMemory(BytesRead, count).CopyTo(buffer);
BytesRead += count;
AfterRead?.Invoke(BytesRead);
return count;
}
public override void Flush() => throw new NotSupportedException();
public override long Seek(long offset, SeekOrigin origin) =>
throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) =>
throw new NotSupportedException();
protected override void Dispose(bool disposing)
{
disposed = true;
base.Dispose(disposing);
}
}
}
#endif

12
tests/SharpCompress.Test/Streams/LzmaModelReuseTests.cs
View File

@@ -128,14 +128,10 @@ public class LzmaModelReuseTests
}
previous = current;
#if !LEGACY_DOTNET
if (useAsync)
{
Assert.Null(GetField(current, "_fLiteral"));
}
else
{
Assert.Null(GetField(GetField(current, "_literalDecoder")!, "_models"));
}
// Both modern read paths now use the same fast probability bank. Neither
// should additionally allocate the scalar/per-symbol async model bank.
Assert.NotNull(GetField(current, "_fLiteral"));
Assert.Null(GetField(GetField(current, "_literalDecoder")!, "_models"));
#endif
}
var end = useAsync

134
tests/SharpCompress.Test/Xz/Filters/BcjAsyncAllocationTests.cs Normal file
View File

@@ -0,0 +1,134 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using SharpCompress.Compressors.Xz.Filters;
using Xunit;
namespace SharpCompress.Test.Xz.Filters;
public class BcjAsyncAllocationTests
{
#if !LEGACY_DOTNET && !DEBUG
// Debug async state machines allocate independently of Task result adapters.
// This guard measures the Release behavior used by the profiling harness.
[Theory]
[InlineData(0)]
[InlineData(1)]
[InlineData(2)]
[InlineData(3)]
[InlineData(4)]
[InlineData(5)]
public async Task ArrayReadsAllocateOnlyOnePublicTaskPerResult(int filterId)
{
using var input = new MemoryStream(new byte[1024 * 1024]);
using var filter = CreateFilter(filterId);
filter.SetBaseStream(input);
var buffer = new byte[16];
Assert.Equal(16, await filter.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None));
var start = GC.GetAllocatedBytesForCurrentThread();
var total = 16;
int read;
while (
(read = await filter.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None)) != 0
)
{
total += read;
}
var allocated = GC.GetAllocatedBytesForCurrentThread() - start;
Assert.Equal(1024 * 1024, total);
// One 72-byte Task per 16-byte result is about 4.5 MiB; two are about 9 MiB.
Assert.InRange(allocated, 0, 6 * 1024 * 1024);
}
private static BlockFilter CreateFilter(int filterId) =>
filterId switch
{
0 => new X86Filter(),
1 => new ArmFilter(),
2 => new ArmThumbFilter(),
3 => new PowerPCFilter(),
4 => new SparcFilter(),
5 => new IA64Filter(),
_ => throw new ArgumentOutOfRangeException(nameof(filterId)),
};
#endif
[Fact]
public async Task ArgumentAndDisposalErrorsRemainFaultedTasks()
{
using var input = new MemoryStream(new byte[32]);
using var filter = new X86Filter();
filter.SetBaseStream(input);
var invalid = filter.ReadAsync(new byte[1], 2, 1, CancellationToken.None);
await Assert.ThrowsAsync<ArgumentOutOfRangeException>(() => invalid);
Assert.True(invalid.IsFaulted);
filter.Dispose();
var disposed = filter.ReadAsync(new byte[1], 0, 1, CancellationToken.None);
await Assert.ThrowsAsync<ObjectDisposedException>(() => disposed);
Assert.True(disposed.IsFaulted);
}
[Fact]
public async Task CanceledBufferedReadDoesNotConsumeOutput()
{
using var input = new MemoryStream(new byte[32]);
using var filter = new X86Filter();
filter.SetBaseStream(input);
var buffer = new byte[1];
Assert.Equal(1, await filter.ReadAsync(buffer, 0, 1, CancellationToken.None));
using var cancellation = new CancellationTokenSource();
cancellation.Cancel();
var canceled = filter.ReadAsync(buffer, 0, 1, cancellation.Token);
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => canceled);
Assert.True(canceled.IsCanceled);
var remaining = 0;
while (await filter.ReadAsync(buffer, 0, 1, CancellationToken.None) != 0)
{
remaining++;
}
Assert.Equal(31, remaining);
}
[Theory]
[InlineData(false)]
[InlineData(true)]
public async Task SynchronouslyThrowingCustomBaseStreamPreservesAsyncSemantics(bool canceled)
{
using var input = new ThrowingInput(canceled);
using var filter = new CustomX86Filter();
filter.SetBaseStream(input);
var task = filter.ReadAsync(new byte[1], 0, 1, CancellationToken.None);
if (canceled)
{
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => task);
Assert.True(task.IsCanceled);
}
else
{
await Assert.ThrowsAsync<IOException>(() => task);
Assert.True(task.IsFaulted);
}
}
private sealed class CustomX86Filter : X86Filter
{
public override void SetBaseStream(Stream stream) => BaseStream = stream;
}
private sealed class ThrowingInput(bool canceled) : MemoryStream
{
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
) =>
throw (
canceled
? new OperationCanceledException(new CancellationToken(true))
: new IOException("Injected synchronous read failure")
);
}
}

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