From c09616448603f8d92e87c70f2e481d1f9fc8be25 Mon Sep 17 00:00:00 2001 From: Adam Hathcock Date: Sat, 31 Jan 2026 11:18:16 +0000 Subject: [PATCH] add shrink stream async --- .../Common/Zip/ZipFilePart.Async.cs | 15 +- .../Compressors/Shrink/ShrinkStream.Async.cs | 138 ++++++++++++++++++ .../Compressors/Shrink/ShrinkStream.cs | 59 +++++--- 3 files changed, 182 insertions(+), 30 deletions(-) create mode 100644 src/SharpCompress/Compressors/Shrink/ShrinkStream.Async.cs diff --git a/src/SharpCompress/Common/Zip/ZipFilePart.Async.cs b/src/SharpCompress/Common/Zip/ZipFilePart.Async.cs index d2860781..e222a835 100644 --- a/src/SharpCompress/Common/Zip/ZipFilePart.Async.cs +++ b/src/SharpCompress/Common/Zip/ZipFilePart.Async.cs @@ -136,12 +136,15 @@ internal abstract partial class ZipFilePart } case ZipCompressionMethod.Shrink: { - return new ShrinkStream( - stream, - CompressionMode.Decompress, - Header.CompressedSize, - Header.UncompressedSize - ); + return await ShrinkStream + .CreateAsync( + stream, + CompressionMode.Decompress, + Header.CompressedSize, + Header.UncompressedSize, + cancellationToken + ) + .ConfigureAwait(false); } case ZipCompressionMethod.Reduce1: { diff --git a/src/SharpCompress/Compressors/Shrink/ShrinkStream.Async.cs b/src/SharpCompress/Compressors/Shrink/ShrinkStream.Async.cs new file mode 100644 index 00000000..b34cdcc6 --- /dev/null +++ b/src/SharpCompress/Compressors/Shrink/ShrinkStream.Async.cs @@ -0,0 +1,138 @@ +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using SharpCompress.IO; + +namespace SharpCompress.Compressors.Shrink; + +internal partial class ShrinkStream : Stream, IStreamStack +{ + internal static async ValueTask CreateAsync( + Stream stream, + CompressionMode compressionMode, + long compressedSize, + long uncompressedSize, + CancellationToken cancellationToken = default + ) + { + var shrinkStream = new ShrinkStream( + stream, + compressionMode, + compressedSize, + uncompressedSize + ); + await shrinkStream.DecompressAsync(cancellationToken).ConfigureAwait(false); + return shrinkStream; + } + + private async Task DecompressAsync(CancellationToken cancellationToken) + { + if (_decompressed) + { + return; + } + + if (inStream.Position == (long)_compressedSize) + { + return; + } + + // Read all compressed data asynchronously + var src = new byte[_compressedSize]; + int bytesRead = 0; + int totalBytesRead = 0; + + while (totalBytesRead < (int)_compressedSize) + { + bytesRead = await inStream + .ReadAsync( + src, + totalBytesRead, + (int)_compressedSize - totalBytesRead, + cancellationToken + ) + .ConfigureAwait(false); + if (bytesRead == 0) + { + throw new EndOfStreamException( + "Unexpected end of stream while reading compressed data" + ); + } + totalBytesRead += bytesRead; + } + + // Decompress synchronously (CPU-bound operation) + var srcUsed = 0; + var dstUsed = 0; + + HwUnshrink.Unshrink( + src, + (int)_compressedSize, + out srcUsed, + _byteOut, + (int)_uncompressedSize, + out dstUsed + ); + _outBytesCount = dstUsed; + _decompressed = true; + } + + public override async Task ReadAsync( + byte[] buffer, + int offset, + int count, + CancellationToken cancellationToken + ) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (!_decompressed) + { + await DecompressAsync(cancellationToken).ConfigureAwait(false); + } + + // Copy from decompressed buffer + long remaining = _outBytesCount - _position; + if (remaining <= 0) + { + return 0; + } + + int toCopy = (int)Math.Min(count, remaining); + Buffer.BlockCopy(_byteOut, (int)_position, buffer, offset, toCopy); + _position += toCopy; + return toCopy; + } + +#if !LEGACY_DOTNET + public override async ValueTask ReadAsync( + Memory buffer, + CancellationToken cancellationToken = default + ) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (!_decompressed) + { + await DecompressAsync(cancellationToken).ConfigureAwait(false); + } + + if (buffer.IsEmpty) + { + return 0; + } + + long remaining = _outBytesCount - _position; + if (remaining <= 0) + { + return 0; + } + + int toCopy = (int)Math.Min(buffer.Length, remaining); + _byteOut.AsMemory((int)_position, toCopy).CopyTo(buffer); + _position += toCopy; + return toCopy; + } +#endif +} diff --git a/src/SharpCompress/Compressors/Shrink/ShrinkStream.cs b/src/SharpCompress/Compressors/Shrink/ShrinkStream.cs index 8e304896..d40a6c5f 100644 --- a/src/SharpCompress/Compressors/Shrink/ShrinkStream.cs +++ b/src/SharpCompress/Compressors/Shrink/ShrinkStream.cs @@ -4,7 +4,7 @@ using SharpCompress.IO; namespace SharpCompress.Compressors.Shrink; -internal class ShrinkStream : Stream, IStreamStack +internal partial class ShrinkStream : Stream, IStreamStack { #if DEBUG_STREAMS long IStreamStack.InstanceId { get; set; } @@ -33,6 +33,8 @@ internal class ShrinkStream : Stream, IStreamStack private long _uncompressedSize; private byte[] _byteOut; private long _outBytesCount; + private bool _decompressed; + private long _position; public ShrinkStream( Stream stream, @@ -72,7 +74,7 @@ internal class ShrinkStream : Stream, IStreamStack public override long Position { - get => _outBytesCount; + get => _position; set => throw new NotImplementedException(); } @@ -80,32 +82,41 @@ internal class ShrinkStream : Stream, IStreamStack public override int Read(byte[] buffer, int offset, int count) { - if (inStream.Position == (long)_compressedSize) + if (!_decompressed) + { + if (inStream.Position == (long)_compressedSize) + { + return 0; + } + + var src = new byte[_compressedSize]; + inStream.Read(src, 0, (int)_compressedSize); + var srcUsed = 0; + var dstUsed = 0; + + HwUnshrink.Unshrink( + src, + (int)_compressedSize, + out srcUsed, + _byteOut, + (int)_uncompressedSize, + out dstUsed + ); + _outBytesCount = dstUsed; + _decompressed = true; + _position = 0; + } + + long remaining = _outBytesCount - _position; + if (remaining <= 0) { return 0; } - var src = new byte[_compressedSize]; - inStream.Read(src, offset, (int)_compressedSize); - var srcUsed = 0; - var dstUsed = 0; - HwUnshrink.Unshrink( - src, - (int)_compressedSize, - out srcUsed, - _byteOut, - (int)_uncompressedSize, - out dstUsed - ); - _outBytesCount = _byteOut.Length; - - for (var index = 0; index < _outBytesCount; ++index) - { - buffer[offset + index] = _byteOut[index]; - } - var tmp = _outBytesCount; - _outBytesCount = 0; - return (int)tmp; + int toCopy = (int)Math.Min(count, remaining); + Buffer.BlockCopy(_byteOut, (int)_position, buffer, offset, toCopy); + _position += toCopy; + return toCopy; } public override long Seek(long offset, SeekOrigin origin) =>