diff --git a/docs/API.md b/docs/API.md index ec01d5d9..1702a82d 100644 --- a/docs/API.md +++ b/docs/API.md @@ -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, 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 diff --git a/docs/XZ_PROFILING.md b/docs/XZ_PROFILING.md index 2156ce75..e4e40b00 100644 --- a/docs/XZ_PROFILING.md +++ b/docs/XZ_PROFILING.md @@ -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` 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` 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: diff --git a/src/SharpCompress/Compressors/LZMA/LzmaDecoder.Fast.cs b/src/SharpCompress/Compressors/LZMA/LzmaDecoder.Fast.cs index 0b6339e3..3b6b52a5 100644 --- a/src/SharpCompress/Compressors/LZMA/LzmaDecoder.Fast.cs +++ b/src/SharpCompress/Compressors/LZMA/LzmaDecoder.Fast.cs @@ -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. diff --git a/src/SharpCompress/Compressors/LZMA/LzmaStream.Async.cs b/src/SharpCompress/Compressors/LZMA/LzmaStream.Async.cs index 4b8a8583..c5e8b4ec 100644 --- a/src/SharpCompress/Compressors/LZMA/LzmaStream.Async.cs +++ b/src/SharpCompress/Compressors/LZMA/LzmaStream.Async.cs @@ -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 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(_decoder!.Code(_dictionarySize, _outWindow, _rangeDecoder)); +#pragma warning restore VSTHRD103 + } +#endif + return _decoder!.CodeAsync(_dictionarySize, _outWindow, _rangeDecoder, cancellationToken); + } + public override async Task 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(); diff --git a/src/SharpCompress/Compressors/LZMA/LzmaStream.cs b/src/SharpCompress/Compressors/LZMA/LzmaStream.cs index f068ae0d..615ea7d6 100644 --- a/src/SharpCompress/Compressors/LZMA/LzmaStream.cs +++ b/src/SharpCompress/Compressors/LZMA/LzmaStream.cs @@ -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; diff --git a/src/SharpCompress/Compressors/LZMA/RangeCoder/RangeCoder.Async.cs b/src/SharpCompress/Compressors/LZMA/RangeCoder/RangeCoder.Async.cs index 61db6eae..1e63c557 100644 --- a/src/SharpCompress/Compressors/LZMA/RangeCoder/RangeCoder.Async.cs +++ b/src/SharpCompress/Compressors/LZMA/RangeCoder/RangeCoder.Async.cs @@ -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.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) diff --git a/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs index 9a6ab002..bffb2068 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs @@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters; public partial class ArmFilter { [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + public override Task 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 ReadAsync( Memory 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 } diff --git a/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs index 735c3972..587c73fd 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs @@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters; public partial class ArmThumbFilter { [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + public override Task 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 ReadAsync( Memory 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 } diff --git a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs index a14ab12d..d757204d 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs @@ -6,24 +6,22 @@ namespace SharpCompress.Compressors.Xz.Filters; internal sealed partial class BcjFilterStream { + public override Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken + ) => ReadCoreAsync(buffer, offset, count, cancellationToken).AsTask(); + [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + internal async ValueTask 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); } } diff --git a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs index ac040463..6548c7c8 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs @@ -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 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 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; diff --git a/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.Async.cs new file mode 100644 index 00000000..a83829ca --- /dev/null +++ b/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.Async.cs @@ -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 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 RethrowReadExceptionAsync(Exception exception) + { + await Task.CompletedTask.ConfigureAwait(false); + ExceptionDispatchInfo.Capture(exception).Throw(); + return 0; + } +} diff --git a/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.cs b/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.cs index 7d5a8190..e06afcaf 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/BlockFilter.cs @@ -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) diff --git a/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs index c41ae5b1..700bbe87 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs @@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters; public partial class IA64Filter { [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + public override Task 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 ReadAsync( Memory 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 } diff --git a/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs index 11e1f51a..11ebb3b3 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs @@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters; public partial class PowerPCFilter { [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + public override Task 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 ReadAsync( Memory 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 } diff --git a/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs index 92c1d478..33f957f9 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs @@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters; public partial class SparcFilter { [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + public override Task 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 ReadAsync( Memory 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 } diff --git a/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs index 9aa3c198..4f63f636 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs @@ -7,17 +7,27 @@ namespace SharpCompress.Compressors.Xz.Filters; public partial class X86Filter { [Zomp.SyncMethodGenerator.CreateSyncVersion] - public override async Task ReadAsync( + public override Task 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 ReadAsync( Memory 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 } diff --git a/src/SharpCompress/Compressors/Xz/XZBlock.Async.cs b/src/SharpCompress/Compressors/Xz/XZBlock.Async.cs index 139aa737..2a67c73c 100644 --- a/src/SharpCompress/Compressors/Xz/XZBlock.Async.cs +++ b/src/SharpCompress/Compressors/Xz/XZBlock.Async.cs @@ -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 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 ReadAsync( + Memory 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 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.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.Shared.Return(scratch); + } + } + break; + } + } +#endif + private async ValueTask SkipPaddingAsync(CancellationToken cancellationToken = default) { var bytes = (BaseStream.Position - _startPosition) % 4; diff --git a/src/SharpCompress/Compressors/Xz/XZStream.Async.cs b/src/SharpCompress/Compressors/Xz/XZStream.Async.cs index 961ebc8c..ea84cadb 100644 --- a/src/SharpCompress/Compressors/Xz/XZStream.Async.cs +++ b/src/SharpCompress/Compressors/Xz/XZStream.Async.cs @@ -61,6 +61,75 @@ public sealed partial class XZStream return bytesRead; } +#if !LEGACY_DOTNET + /// + /// Read and validate XZ data without adapting memory reads through Task-based array overloads. + /// + public override async ValueTask ReadAsync( + Memory 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 ReadBlocksAsync( + Memory 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 + /// /// Asynchronously reads and validates the XZ header. /// diff --git a/tests/SharpCompress.Test/Streams/Lzma2AsyncBufferingTests.cs b/tests/SharpCompress.Test/Streams/Lzma2AsyncBufferingTests.cs new file mode 100644 index 00000000..72b57ccc --- /dev/null +++ b/tests/SharpCompress.Test/Streams/Lzma2AsyncBufferingTests.cs @@ -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(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(() => + 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(() => + 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(() => + decoder.ReadAsync(buffer, 0, 1, CancellationToken.None) + ); + var range = GetField(decoder, "_rangeDecoder")!; + Assert.NotNull(GetField(range, "_fastBuffer")); + await Assert.ThrowsAsync(() => + 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(() => + 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.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? 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 ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken + ) => ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + + public override async ValueTask ReadAsync( + Memory 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 diff --git a/tests/SharpCompress.Test/Streams/LzmaModelReuseTests.cs b/tests/SharpCompress.Test/Streams/LzmaModelReuseTests.cs index 87f36512..4f4a6374 100644 --- a/tests/SharpCompress.Test/Streams/LzmaModelReuseTests.cs +++ b/tests/SharpCompress.Test/Streams/LzmaModelReuseTests.cs @@ -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 diff --git a/tests/SharpCompress.Test/Xz/Filters/BcjAsyncAllocationTests.cs b/tests/SharpCompress.Test/Xz/Filters/BcjAsyncAllocationTests.cs new file mode 100644 index 00000000..c5da7ce5 --- /dev/null +++ b/tests/SharpCompress.Test/Xz/Filters/BcjAsyncAllocationTests.cs @@ -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(() => invalid); + Assert.True(invalid.IsFaulted); + + filter.Dispose(); + var disposed = filter.ReadAsync(new byte[1], 0, 1, CancellationToken.None); + await Assert.ThrowsAsync(() => 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(() => 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(() => task); + Assert.True(task.IsCanceled); + } + else + { + await Assert.ThrowsAsync(() => 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 ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken + ) => + throw ( + canceled + ? new OperationCanceledException(new CancellationToken(true)) + : new IOException("Injected synchronous read failure") + ); + } +} diff --git a/tests/SharpCompress.Test/Xz/Filters/BcjBufferLifetimeTests.cs b/tests/SharpCompress.Test/Xz/Filters/BcjBufferLifetimeTests.cs index 1e83675b..bc4d2233 100644 --- a/tests/SharpCompress.Test/Xz/Filters/BcjBufferLifetimeTests.cs +++ b/tests/SharpCompress.Test/Xz/Filters/BcjBufferLifetimeTests.cs @@ -210,5 +210,12 @@ public class BcjBufferLifetimeTests int count, CancellationToken cancellationToken ) => Task.FromException(new IOException("Injected read failure")); + +#if !LEGACY_DOTNET + public override ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default + ) => new(Task.FromException(new IOException("Injected read failure"))); +#endif } } diff --git a/tests/SharpCompress.Test/Xz/XZMemoryReadTests.cs b/tests/SharpCompress.Test/Xz/XZMemoryReadTests.cs new file mode 100644 index 00000000..8f998db4 --- /dev/null +++ b/tests/SharpCompress.Test/Xz/XZMemoryReadTests.cs @@ -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 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 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(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.Empty)); + Assert.Equal(0, input.Position); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(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(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 + { + public override Span 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 ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken + ) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + return base.Read(buffer, offset, count); + } + + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default + ) + { + await Task.Yield(); + return await base.ReadAsync(buffer, cancellationToken); + } + } +} +#endif