From de7a5d9ad4058dd0c900453b948e605fa277be2e Mon Sep 17 00:00:00 2001 From: Adam Hathcock Date: Sat, 24 Oct 2026 12:54:39 +0100 Subject: [PATCH] Avoid duplicate task results in XZ BCJ array reads --- docs/XZ_PROFILING.md | 22 +++ .../Compressors/Xz/Filters/ArmFilter.Async.cs | 4 +- .../Xz/Filters/ArmThumbFilter.Async.cs | 4 +- .../Xz/Filters/BcjFilterStream.Async.cs | 20 ++- .../Compressors/Xz/Filters/BcjFilterStream.cs | 37 +++++ .../Xz/Filters/BlockFilter.Async.cs | 49 +++++++ .../Compressors/Xz/Filters/BlockFilter.cs | 5 +- .../Xz/Filters/IA64Filter.Async.cs | 4 +- .../Xz/Filters/PowerPCFilter.Async.cs | 4 +- .../Xz/Filters/SparcFilter.Async.cs | 4 +- .../Compressors/Xz/Filters/X86Filter.Async.cs | 4 +- .../Xz/Filters/BcjAsyncAllocationTests.cs | 134 ++++++++++++++++++ 12 files changed, 267 insertions(+), 24 deletions(-) create mode 100644 src/SharpCompress/Compressors/Xz/Filters/BlockFilter.Async.cs create mode 100644 tests/SharpCompress.Test/Xz/Filters/BcjAsyncAllocationTests.cs diff --git a/docs/XZ_PROFILING.md b/docs/XZ_PROFILING.md index 2156ce75..fa81f6c3 100644 --- a/docs/XZ_PROFILING.md +++ b/docs/XZ_PROFILING.md @@ -215,6 +215,28 @@ 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. + ## 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 9a6ab002..a66e9595 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/ArmFilter.Async.cs @@ -7,12 +7,12 @@ 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( diff --git a/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs index 735c3972..051846bc 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/ArmThumbFilter.Async.cs @@ -7,12 +7,12 @@ 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( diff --git a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.Async.cs index a14ab12d..dff21bf9 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; diff --git a/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs b/src/SharpCompress/Compressors/Xz/Filters/BcjFilterStream.cs index ac040463..8171397d 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,39 @@ 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; + } + private int CopyBuffered(Span destination) { var count = Math.Min(destination.Length, _filtered - _position); 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..8a9d7a7f 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/IA64Filter.Async.cs @@ -7,12 +7,12 @@ 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( diff --git a/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs index 11e1f51a..6ff3cf92 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/PowerPCFilter.Async.cs @@ -7,12 +7,12 @@ 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( diff --git a/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs index 92c1d478..617cb697 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/SparcFilter.Async.cs @@ -7,12 +7,12 @@ 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( diff --git a/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs b/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs index 9aa3c198..2da396b2 100644 --- a/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs +++ b/src/SharpCompress/Compressors/Xz/Filters/X86Filter.Async.cs @@ -7,12 +7,12 @@ 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( 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") + ); + } +}