Restart solid 7z decoders after failures and read mode changes

This commit is contained in:
Adam Hathcock
2026-10-22 09:11:14 +01:00
parent 2c29a6bfab
commit 366fa4ab34
6 changed files with 235 additions and 167 deletions

View File

@@ -919,8 +919,8 @@ of an entry does not corrupt the entries that follow: the decoder skips forward
over the bytes that were not read when the next entry is opened. Read an entry
before opening the next entry of the same folder - the shared decoder cannot
serve two open entry streams of the same folder at the same time. Mixing
synchronous and asynchronous entry reads of the same folder is supported; the
decoder keeps the read mode it was first used with.
synchronous and asynchronous entry APIs restarts the folder decoder when needed;
same-mode sequential access reuses it. Failed decoder reads also invalidate reuse.
## 7z Encrypted Decoding Errors

View File

@@ -49,11 +49,12 @@ internal sealed partial class ArchiveDatabase
CancellationToken cancellationToken
)
{
cancellationToken.ThrowIfCancellationRequested();
var tracker = await GetCachedFolderTrackerAsync(stream, folder, pw, cancellationToken)
.ConfigureAwait(false);
if (skipSize < tracker.Position)
if (tracker.IsFaulted || tracker.IsAsync == false || skipSize < tracker.Position)
{
// Non-sequential (backward) access within the same folder requires restarting.
// Failed reads and mode changes invalidate reuse, as does backward access.
await DisposeCachedFolderStreamAsync().ConfigureAwait(false);
tracker = await GetCachedFolderTrackerAsync(stream, folder, pw, cancellationToken)
.ConfigureAwait(false);

View File

@@ -351,9 +351,10 @@ internal partial class ArchiveDatabase
)
{
var tracker = GetCachedFolderTracker(stream, folder, pw, enableParallelism);
if (skipSize < tracker.Position)
if (tracker.IsFaulted || tracker.IsAsync == true || skipSize < tracker.Position)
{
// Non-sequential (backward) access within the same folder requires restarting.
// Restart for backward access, a failed read, or a mode change. LZMA cannot mix
// sync and async reads; do not hide synchronous I/O inside the async API (or vice versa).
DisposeCachedFolderStream();
tracker = GetCachedFolderTracker(stream, folder, pw, enableParallelism);
}

View File

@@ -6,164 +6,142 @@ using SharpCompress.IO;
namespace SharpCompress.Common.SevenZip;
/// <summary>
/// Keeps track of how much of a solid folder has actually been consumed from its shared decoder.
/// </summary>
/// <remarks>
/// <para>
/// Every entry of a solid folder is decoded by the same forward-only stream, and that stream
/// cannot be rewound. The position therefore has to be tracked explicitly: it tells the caller
/// whether the next entry can be reached by skipping forward over the bytes that were not read
/// (the common case) or whether the folder has to be decoded again from its start.
/// </para>
/// <para>
/// The tracker also pins the decoder to a single read mode. LzmaStream refuses to mix synchronous
/// and asynchronous reads on the same instance, and the decoder is shared by every entry of the
/// folder, so the first read decides the mode used by all later reads. Reads that arrive in the
/// other mode are routed to the mode the decoder was established with instead of failing.
/// </para>
/// <para>
/// The tracker does not own the decoder: the ArchiveDatabase creates and disposes it. Disposing
/// the tracker therefore leaves the shared decoder alone.
/// </para>
/// </remarks>
internal sealed class FolderStreamTracker : Stream, IStreamStack
// Observes a borrowed decoder: the database, not this wrapper or an entry, owns its lifetime.
internal sealed class FolderStreamTracker(Stream stream) : Stream, IStreamStack
{
Stream IStreamStack.BaseStream() => _stream;
private readonly Stream _stream;
private readonly byte[] _singleByte = new byte[1];
private long _position;
private bool _asyncMode;
private bool _modeEstablished;
public FolderStreamTracker(Stream stream) =>
_stream = stream ?? throw new ArgumentNullException(nameof(stream));
internal bool? IsAsync { get; private set; }
internal bool IsFaulted { get; private set; }
/// <summary>
/// The number of decompressed bytes consumed from the folder so far.
/// </summary>
Stream IStreamStack.BaseStream() => stream;
public override bool CanRead => stream.CanRead;
public override bool CanSeek => stream.CanSeek;
public override bool CanWrite => false;
public override long Length => stream.Length;
public override long Position
{
get => _position;
set
{
if (!_stream.CanSeek)
{
throw new NotSupportedException();
}
_stream.Position = value;
_position = value;
}
set => _position = stream.Position = value;
}
public override bool CanRead => _stream.CanRead;
public override bool CanSeek => _stream.CanSeek;
public override bool CanWrite => false;
public override void Flush() { }
public override long Length => _stream.Length;
public override int Read(byte[] buffer, int offset, int count)
{
if (count <= 0)
try
{
return 0;
if (count > 0)
{
IsAsync = false;
}
var read = stream.Read(buffer, offset, count);
_position += read;
return read;
}
if (_modeEstablished && _asyncMode)
catch
{
// The decoder is already used asynchronously, so it must not be read synchronously.
#pragma warning disable VSTHRD002 // The decoder cannot be read synchronously in this mode.
return ReadAsyncCore(buffer, offset, count, CancellationToken.None)
.GetAwaiter()
.GetResult();
#pragma warning restore VSTHRD002
IsFaulted = true;
throw;
}
_modeEstablished = true;
_asyncMode = false;
var read = _stream.Read(buffer, offset, count);
_position += read;
return read;
}
public override int ReadByte()
{
if (_modeEstablished && _asyncMode)
try
{
// The decoder is already used asynchronously, so it must not be read synchronously.
#pragma warning disable VSTHRD002 // The decoder cannot be read synchronously in this mode.
var read = ReadAsyncCore(_singleByte, 0, 1, CancellationToken.None)
.GetAwaiter()
.GetResult();
#pragma warning restore VSTHRD002
return read == 0 ? -1 : _singleByte[0];
IsAsync = false;
var value = stream.ReadByte();
if (value >= 0)
{
_position++;
}
return value;
}
_modeEstablished = true;
_asyncMode = false;
var value = _stream.ReadByte();
if (value != -1)
catch
{
_position++;
IsFaulted = true;
throw;
}
return value;
}
public override Task<int> ReadAsync(
public override async Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
if (count <= 0)
try
{
return Task.FromResult(0);
if (count > 0)
{
IsAsync = true;
}
var read = await stream
.ReadAsync(buffer, offset, count, cancellationToken)
.ConfigureAwait(false);
_position += read;
return read;
}
if (_modeEstablished && !_asyncMode)
catch
{
// The decoder is already used synchronously, so it must not be read asynchronously.
cancellationToken.ThrowIfCancellationRequested();
return Task.FromResult(Read(buffer, offset, count));
// A decoder may advance before throwing, including on cancellation. Its position and
// internal state are no longer safe to reuse, even if no bytes reached the caller.
IsFaulted = true;
throw;
}
_modeEstablished = true;
_asyncMode = true;
return ReadAsyncCore(buffer, offset, count, cancellationToken);
}
private async Task<int> ReadAsyncCore(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
#if !LEGACY_DOTNET
public override int Read(Span<byte> buffer)
{
try
{
if (!buffer.IsEmpty)
{
IsAsync = false;
}
var read = stream.Read(buffer);
_position += read;
return read;
}
catch
{
IsFaulted = true;
throw;
}
}
public override async ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
var read = await _stream
.ReadAsync(buffer, offset, count, cancellationToken)
.ConfigureAwait(false);
_position += read;
return read;
try
{
if (!buffer.IsEmpty)
{
IsAsync = true;
}
var read = await stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
_position += read;
return read;
}
catch
{
IsFaulted = true;
throw;
}
}
#endif
public override long Seek(long offset, SeekOrigin origin)
{
var position = _stream.Seek(offset, origin);
_position = position;
return position;
}
public override long Seek(long offset, SeekOrigin origin) =>
_position = stream.Seek(offset, origin);
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) =>
throw new NotSupportedException();
public override void WriteByte(byte value) => throw new NotSupportedException();
}

View File

@@ -83,22 +83,24 @@ public class FolderStreamTrackerTests
}
[Fact]
public async Task MixingReadModes_KeepsTheDecoderInASingleMode()
public async Task Reads_RecordActualMode_AndCancellationInvalidatesReuse()
{
var payload = Payload(64);
var tracker = new FolderStreamTracker(new MemoryStream(payload));
var buffer = new byte[10];
Assert.Equal(10, tracker.Read(buffer, 0, buffer.Length));
// An asynchronous read after a synchronous one must not fail and must keep counting.
Assert.Equal(10, await tracker.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None));
Assert.Equal(20, tracker.Position);
Assert.Equal(false, tracker.IsAsync);
// ... and the other way around.
var other = new FolderStreamTracker(new MemoryStream(payload));
Assert.Equal(10, await other.ReadAsync(buffer, 0, buffer.Length, CancellationToken.None));
Assert.Equal(10, other.Read(buffer, 0, buffer.Length));
Assert.Equal(20, other.Position);
Assert.Equal(true, other.IsAsync);
using var cancellation = new CancellationTokenSource();
cancellation.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(() =>
other.ReadAsync(buffer, 0, buffer.Length, cancellation.Token)
);
Assert.True(other.IsFaulted);
}
}

View File

@@ -165,17 +165,49 @@ public class SevenZipSolidCacheTests : TestBase
[Fact]
public void PartialRead_ThenLaterEntry_ReturnsEntryContent()
{
var expected = ExpectedContent(SolidArchive, 2);
const string archiveName = "7Zip.delta.7z";
var expected = ExpectedContent(archiveName, 2);
using var archive = SevenZipArchive.OpenArchive(PathOf(SolidArchive));
using var archive = SevenZipArchive.OpenArchive(PathOf(archiveName));
var entries = FileEntries(archive);
Assert.True(entries.Count >= 3);
Assert.Same(
((SevenZipArchiveEntry)entries[0]).FilePart.Folder,
((SevenZipArchiveEntry)entries[2]).FilePart.Folder
);
ReadPartial(entries[0], 16);
Assert.Equal(expected, ReadAll(entries[2]));
}
[Fact]
public async Task PartialRead_ThenSkippedEntryAsync_ReusesFolder()
{
const string archiveName = "7Zip.delta.7z";
await using var archive = await SevenZipArchive.OpenAsyncArchive(PathOf(archiveName));
var entries = await FileEntriesAsync(archive);
Assert.Same(
((SevenZipArchiveEntry)entries[0]).FilePart.Folder,
((SevenZipArchiveEntry)entries[2]).FilePart.Folder
);
await ReadPartialAsync(entries[0], 16);
var decoder = CachedFolderStream(archive);
Assert.Equal(ExpectedContent(archiveName, 2), await ReadAllAsync(entries[2]));
Assert.Same(decoder, CachedFolderStream(archive));
}
[Fact]
public async Task FullyReadEntriesAsync_ReuseFolder()
{
await using var archive = await SevenZipArchive.OpenAsyncArchive(PathOf(SolidArchive));
var entries = await FileEntriesAsync(archive);
await ReadAllAsync(entries[0]);
var decoder = CachedFolderStream(archive);
Assert.Equal(ExpectedContent(SolidArchive, 1), await ReadAllAsync(entries[1]));
Assert.Same(decoder, CachedFolderStream(archive));
}
[Theory]
[InlineData(SolidArchive)]
[InlineData(SingleBlockSolidArchive)]
@@ -233,58 +265,112 @@ public class SevenZipSolidCacheTests : TestBase
Assert.Equal(expected, await ReadAllAsync(entries[0]).ConfigureAwait(false));
}
[Fact]
public void AlternatingSyncAndAsyncEntryReads_ReturnEntryContent()
[Theory]
[InlineData(true, 0)]
[InlineData(true, 16)]
[InlineData(true, -1)]
[InlineData(false, 0)]
[InlineData(false, 16)]
[InlineData(false, -1)]
public async Task AlternatingEntryReadModes_ReturnEntryContent(bool syncFirst, int count)
{
var expected = FileEntries(SevenZipArchive.OpenArchive(PathOf(SolidArchive)))
.Select((_, index) => ExpectedContent(SolidArchive, index))
.ToList();
using var archive = SevenZipArchive.OpenArchive(PathOf(SolidArchive));
var entries = FileEntries(archive);
Assert.True(entries.Count >= 3);
ReadPartial(entries[0], 16);
var second = entries[1].OpenEntryStream();
using (second)
var expected = ExpectedContent(SolidArchive, 1);
await using var archive = await SevenZipArchive.OpenAsyncArchive(PathOf(SolidArchive));
var entries = await FileEntriesAsync(archive);
if (syncFirst)
{
using var content = new MemoryStream();
second.CopyTo(content);
Assert.Equal(expected[1], content.ToArray());
if (count < 0)
{
ReadAll(entries[0]);
}
else
{
ReadPartial(entries[0], count);
}
var decoder = CachedFolderStream(archive);
Assert.Equal(expected, await ReadAllAsync(entries[1]));
if (count != 0)
{
Assert.NotSame(decoder, CachedFolderStream(archive));
}
}
else
{
if (count < 0)
{
await ReadAllAsync(entries[0]);
}
else
{
await ReadPartialAsync(entries[0], count);
}
var decoder = CachedFolderStream(archive);
Assert.Equal(expected, ReadAll(entries[1]));
if (count != 0)
{
Assert.NotSame(decoder, CachedFolderStream(archive));
}
}
Assert.Equal(expected[2], ReadAll(entries[2]));
}
[Fact]
public async Task AlternatingSyncAndAsyncEntryReads_ReturnEntryContentAsync()
public async Task CancellationAfterSourceProgress_RestartsCachedDecoder()
{
var expected = new List<byte[]>();
for (var index = 0; index < 3; index++)
using var source = new CancelAfterReadStream(File.ReadAllBytes(PathOf(SolidArchive)));
await using var archive = await SevenZipArchive.OpenAsyncArchive(
source,
ReaderOptions.ForExternalStream
);
var entries = await FileEntriesAsync(archive);
var stream = await entries[0].OpenEntryStreamAsync(CancellationToken.None);
var decoder = CachedFolderStream(archive);
source.CancelNextRead = true;
await using (stream.DisposeAsyncScope())
{
expected.Add(await ExpectedContentAsync(SolidArchive, index).ConfigureAwait(false));
await Assert.ThrowsAsync<OperationCanceledException>(() =>
stream.CopyToAsync(Stream.Null, Constants.BufferSize, CancellationToken.None)
);
}
Assert.False(source.CancelNextRead);
Assert.Equal(ExpectedContent(SolidArchive, 1), await ReadAllAsync(entries[1]));
Assert.NotSame(decoder, CachedFolderStream(archive));
}
private sealed class CancelAfterReadStream(byte[] bytes) : MemoryStream(bytes)
{
internal bool CancelNextRead { get; set; }
public override Task<int> ReadAsync(
byte[] buffer,
int offset,
int count,
CancellationToken cancellationToken
)
{
var read = base.Read(buffer, offset, Math.Min(count, 64));
if (CancelNextRead)
{
CancelNextRead = false;
throw new OperationCanceledException();
}
return Task.FromResult(read);
}
await using var archive = await SevenZipArchive
.OpenAsyncArchive(PathOf(SolidArchive))
.ConfigureAwait(false);
var entries = await FileEntriesAsync(archive).ConfigureAwait(false);
Assert.True(entries.Count >= 3);
await ReadPartialAsync(entries[0], 16).ConfigureAwait(false);
var second = entries[1].OpenEntryStream();
using (second)
#if !LEGACY_DOTNET
public override ValueTask<int> ReadAsync(
Memory<byte> buffer,
CancellationToken cancellationToken = default
)
{
using var content = new MemoryStream();
await second
.CopyToAsync(content, Constants.BufferSize, CancellationToken.None)
.ConfigureAwait(false);
Assert.Equal(expected[1], content.ToArray());
var read = base.Read(buffer.Span.Slice(0, Math.Min(buffer.Length, 64)));
if (CancelNextRead)
{
CancelNextRead = false;
throw new OperationCanceledException();
}
return new ValueTask<int>(read);
}
Assert.Equal(expected[2], await ReadAllAsync(entries[2]).ConfigureAwait(false));
#endif
}
[Fact]