From 31f3476fc9afaa0aea01764bbc37e42ab8cc0cd0 Mon Sep 17 00:00:00 2001 From: Adam Hathcock Date: Sat, 24 Oct 2026 13:09:51 +0100 Subject: [PATCH] Add native ValueTask memory reads through XZ filter chains --- docs/API.md | 25 +++ docs/XZ_PROFILING.md | 31 +++ .../Compressors/Xz/Filters/ArmFilter.Async.cs | 12 +- .../Xz/Filters/ArmThumbFilter.Async.cs | 12 +- .../Xz/Filters/BcjFilterStream.Async.cs | 7 +- .../Compressors/Xz/Filters/BcjFilterStream.cs | 30 +++ .../Xz/Filters/IA64Filter.Async.cs | 12 +- .../Xz/Filters/PowerPCFilter.Async.cs | 12 +- .../Xz/Filters/SparcFilter.Async.cs | 12 +- .../Compressors/Xz/Filters/X86Filter.Async.cs | 12 +- .../Compressors/Xz/XZBlock.Async.cs | 113 ++++++++++ .../Compressors/Xz/XZStream.Async.cs | 69 +++++++ .../Xz/Filters/BcjBufferLifetimeTests.cs | 7 + .../Xz/XZMemoryReadTests.cs | 195 ++++++++++++++++++ 14 files changed, 542 insertions(+), 7 deletions(-) create mode 100644 tests/SharpCompress.Test/Xz/XZMemoryReadTests.cs diff --git a/docs/API.md b/docs/API.md index ec01d5d9..d47fa444 100644 --- a/docs/API.md +++ b/docs/API.md @@ -888,6 +888,31 @@ using (var reader = ReaderFactory.OpenReader(stream)) --- +## Low-Level XZ Memory Reads + +On .NET 6 and later, `SharpCompress.Compressors.Xz.XZStream` and `XZBlock` +override `ReadAsync(Memory, 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. diff --git a/docs/XZ_PROFILING.md b/docs/XZ_PROFILING.md index fa81f6c3..d6706dcc 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 @@ -237,6 +240,34 @@ and net48 test compilation passes. The remaining public `Task` result canno be removed without changing that API; a native memory-based XZ chain remains a separate follow-up. +## Follow-up: native XZ memory read chain + +The next change adds native memory/ValueTask reads to `XZStream` and `XZBlock` +on modern targets, carries memory-based I/O through BCJ refills, and completes +already-buffered reads without traversing multiple async wrappers. All buffered +bytes still update the block check exactly once. Array API signatures and sync +array-based I/O remain unchanged. Native/custom destination memory works with +all four XZ check types; SHA-256 uses bounded pooled scratch when required. + +A fresh comparison against the first follow-up (`de7a5d9`) uses the built-in +1 MiB x86+LZMA2 fixture, 16-byte memory reads at offset 5, two warmups, seven +measured complete drains, Release .NET 10, and disabled tiered compilation. +Median allocations fall from **14,180,048 to 15,056 bytes**, with elapsed medians +**22.399 to 19.422 ms**. Both builds retain checksum verification and return the +expected output length. This does not yet address the async range-decoder CPU +bottleneck or establish file-backed throughput improvements. + +Artifacts are `fix2-before.log`, `fix2-after.log`, and `fix2-harness/` in the parent +directory. The harness's before/after DLLs were explicitly bound and hash-checked +after observing that MSBuild reused an earlier reference when switching HintPath +properties; `fix2-harness/comparison.json` records the actual measured hashes. + +All **764 targeted Release tests** pass, including yielding async-only input, +seekable/forward-only streams, native/custom memory, mixed async overloads, +bounded allocations, empty/canceled reads, and checksum corruption. All six +library targets and net48 test compilation pass. The issue sample still matches +liblzma in all 24 read-size/overload/offset combinations. + ## Artifacts and reproduction The controlled comparison is isolated from preliminary sample-file measurements: diff --git a/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs index a66e9595..bffb2068 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs @@ -18,6 +18,16 @@ public partial class ArmFilter 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 051846bc..587c73fd 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs @@ -18,6 +18,16 @@ public partial class ArmThumbFilter 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 dff21bf9..d757204d 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs @@ -41,15 +41,20 @@ internal sealed partial class BcjFilterStream return total; } - [Zomp.SyncMethodGenerator.CreateSyncVersion] private async ValueTask FillAsync(CancellationToken cancellationToken) { KeepUnfilteredTail(); while (_filtered == 0 && !_endReached) { +#if LEGACY_DOTNET var bytesRead = await BaseStream .ReadAsync(_buffer, _size, BufferSize - _size, cancellationToken) .ConfigureAwait(false); +#else + var bytesRead = await BaseStream + .ReadAsync(_buffer.AsMemory(_size, BufferSize - _size), cancellationToken) + .ConfigureAwait(false); +#endif ConvertBuffered(bytesRead); } } diff --git a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs index 8171397d..6548c7c8 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs @@ -90,6 +90,25 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream 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); @@ -108,6 +127,17 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream _filtered = 0; } + private void Fill() + { + KeepUnfilteredTail(); + while (_filtered == 0 && !_endReached) + { + // Preserve the sync array-based I/O contract; only async reads prefer Memory. + var bytesRead = BaseStream.Read(_buffer, _size, BufferSize - _size); + ConvertBuffered(bytesRead); + } + } + private void ConvertBuffered(int bytesRead) { _size += bytesRead; diff --git a/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs index 8a9d7a7f..700bbe87 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs @@ -18,6 +18,16 @@ public partial class IA64Filter 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 6ff3cf92..11ebb3b3 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs @@ -18,6 +18,16 @@ public partial class PowerPCFilter 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 617cb697..33f957f9 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs @@ -18,6 +18,16 @@ public partial class SparcFilter 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 2da396b2..4f63f636 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs @@ -18,6 +18,16 @@ public partial class X86Filter 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/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