diff --git a/docs/API.md b/docs/API.md index e9febe9c..3d7aa8c2 100644 --- a/docs/API.md +++ b/docs/API.md @@ -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 diff --git a/src/SharpCompress/Common/SevenZip/ArchiveDatabase.Async.cs b/src/SharpCompress/Common/SevenZip/ArchiveDatabase.Async.cs index 4d70cd21..8197ae17 100644 --- a/src/SharpCompress/Common/SevenZip/ArchiveDatabase.Async.cs +++ b/src/SharpCompress/Common/SevenZip/ArchiveDatabase.Async.cs @@ -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); diff --git a/src/SharpCompress/Common/SevenZip/ArchiveDatabase.cs b/src/SharpCompress/Common/SevenZip/ArchiveDatabase.cs index 7499071d..b2103fe9 100644 --- a/src/SharpCompress/Common/SevenZip/ArchiveDatabase.cs +++ b/src/SharpCompress/Common/SevenZip/ArchiveDatabase.cs @@ -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); } diff --git a/src/SharpCompress/Common/SevenZip/FolderStreamTracker.cs b/src/SharpCompress/Common/SevenZip/FolderStreamTracker.cs index 4d69bc42..4b9f3ebb 100644 --- a/src/SharpCompress/Common/SevenZip/FolderStreamTracker.cs +++ b/src/SharpCompress/Common/SevenZip/FolderStreamTracker.cs @@ -6,164 +6,142 @@ using SharpCompress.IO; namespace SharpCompress.Common.SevenZip; -/// -/// Keeps track of how much of a solid folder has actually been consumed from its shared decoder. -/// -/// -/// -/// 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. -/// -/// -/// 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. -/// -/// -/// The tracker does not own the decoder: the ArchiveDatabase creates and disposes it. Disposing -/// the tracker therefore leaves the shared decoder alone. -/// -/// -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; } - /// - /// The number of decompressed bytes consumed from the folder so far. - /// + 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 ReadAsync( + public override async Task 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 ReadAsyncCore( - byte[] buffer, - int offset, - int count, - CancellationToken cancellationToken +#if !LEGACY_DOTNET + public override int Read(Span buffer) + { + try + { + if (!buffer.IsEmpty) + { + IsAsync = false; + } + var read = stream.Read(buffer); + _position += read; + return read; + } + catch + { + IsFaulted = true; + throw; + } + } + + public override async ValueTask ReadAsync( + Memory 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(); } diff --git a/tests/SharpCompress.Test/SevenZip/FolderStreamTrackerTests.cs b/tests/SharpCompress.Test/SevenZip/FolderStreamTrackerTests.cs index 99c2fe47..b6b9c13f 100644 --- a/tests/SharpCompress.Test/SevenZip/FolderStreamTrackerTests.cs +++ b/tests/SharpCompress.Test/SevenZip/FolderStreamTrackerTests.cs @@ -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(() => + other.ReadAsync(buffer, 0, buffer.Length, cancellation.Token) + ); + Assert.True(other.IsFaulted); } } diff --git a/tests/SharpCompress.Test/SevenZip/SevenZipSolidCacheTests.cs b/tests/SharpCompress.Test/SevenZip/SevenZipSolidCacheTests.cs index c90938c2..9a53c72b 100644 --- a/tests/SharpCompress.Test/SevenZip/SevenZipSolidCacheTests.cs +++ b/tests/SharpCompress.Test/SevenZip/SevenZipSolidCacheTests.cs @@ -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(); - 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(() => + 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 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 ReadAsync( + Memory 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(read); } - - Assert.Equal(expected[2], await ReadAllAsync(entries[2]).ConfigureAwait(false)); +#endif } [Fact]