starred/sharpcompress
1
0
Fork 0
mirror of https://github.com/adamhathcock/sharpcompress.git synced 2026-11-15 17:02:03 +00:00
Code Issues 975 Packages Projects Releases 25 Wiki Activity

Avoid duplicate task results in XZ BCJ array reads

Browse Source
This commit is contained in:
Adam Hathcock
2026-10-24 12:54:39 +01:00
parent d755de5170
commit de7a5d9ad4
12 changed files with 267 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
docs/XZ_PROFILING.md
View File

@@ -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<int>` and creates a task only at the existing public array boundary.
Already-buffered reads complete without entering an async state machine. Custom
subclasses installing a synchronously throwing input stream retain the previous
faulted/canceled-task behavior; sync signatures are unchanged.
A fresh before/after comparison against the pooled baseline uses the same isolated
1 MiB x86 workload, 16-byte array reads, two warmups, seven measured drains,
Release .NET 10, and disabled tiered compilation. Median allocations fall from
**9,438,376 to 4,719,784 bytes**; median elapsed time is **1.695 to 0.963 ms**.
This is a small-read wrapper result, not a complete XZ throughput claim. Logs and
the follow-up harness are in the parent artifact directory as `fix1-before.log`,
`fix1-after.log`, and `fix1-harness/`.
All **728 targeted Release tests** pass, including allocation guards for all six
architectures and fault/cancellation regressions. All six library targets build,
and net48 test compilation passes. The remaining public `Task<int>` result cannot
be removed without changing that API; a native memory-based XZ chain remains a
separate follow-up.
## Artifacts and reproduction
The controlled comparison is isolated from preliminary sample-file measurements:

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

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

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

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

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

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

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

@@ -1,6 +1,7 @@
using System;
using System.Buffers;
using System.IO;
using System.Threading;
namespace SharpCompress.Compressors.Xz.Filters;
@@ -34,6 +35,9 @@ internal sealed partial class BcjFilterStream : ReadOnlyStream
public override bool CanRead => !_isDisposed && BaseStream.CanRead;
public override int Read(byte[] buffer, int offset, int count) =>
ReadCore(buffer, offset, count);
protected override void Dispose(bool disposing)
{
if (disposing && !_isDisposed)
@@ -53,6 +57,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<byte> destination)
{
var count = Math.Min(destination.Length, _filtered - _position);

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

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

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

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

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

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

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

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

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

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

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

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

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

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