Merge pull request #1003 from adamhathcock/adam/async-lzma

async lzma

Author: adamhathcock

Directory.Packages.props

@@ -14,6 +14,7 @@
     <PackageVersion Include="xunit" Version="2.9.3" />
     <PackageVersion Include="xunit.runner.visualstudio" Version="3.1.5" />
     <PackageVersion Include="ZstdSharp.Port" Version="0.8.6" />
+    <PackageVersion Include="Microsoft.NET.ILLink.Tasks" Version="8.0.21" />
     <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="8.0.0" />
     <PackageVersion Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.3" />
   </ItemGroup>

src/SharpCompress/Compressors/LZMA/AesDecoderStream.cs

@@ -2,6 +2,8 @@
 using System.IO;
 using System.Security.Cryptography;
 using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.Compressors.LZMA.Utilites;
 using SharpCompress.IO;
 
@@ -283,5 +285,70 @@ private int HandleUnderflow(byte[] buffer, int offset, int count)
         return count;
     }
 
+    public override async Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        if (count == 0 || mWritten == mLimit)
+        {
+            return 0;
+        }
+
+        if (mUnderflow > 0)
+        {
+            return HandleUnderflow(buffer, offset, count);
+        }
+
+        // Need at least 16 bytes to proceed.
+        if (mEnding - mOffset < 16)
+        {
+            Buffer.BlockCopy(mBuffer, mOffset, mBuffer, 0, mEnding - mOffset);
+            mEnding -= mOffset;
+            mOffset = 0;
+
+            do
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                var read = await mStream
+                    .ReadAsync(mBuffer, mEnding, mBuffer.Length - mEnding, cancellationToken)
+                    .ConfigureAwait(false);
+                if (read == 0)
+                {
+                    // We are not done decoding and have less than 16 bytes.
+                    throw new EndOfStreamException();
+                }
+
+                mEnding += read;
+            } while (mEnding - mOffset < 16);
+        }
+
+        // We shouldn't return more data than we are limited to.
+        if (count > mLimit - mWritten)
+        {
+            count = (int)(mLimit - mWritten);
+        }
+
+        // We cannot transform less than 16 bytes into the target buffer,
+        // but we also cannot return zero, so we need to handle this.
+        if (count < 16)
+        {
+            return HandleUnderflow(buffer, offset, count);
+        }
+
+        if (count > mEnding - mOffset)
+        {
+            count = mEnding - mOffset;
+        }
+
+        // Otherwise we transform directly into the target buffer.
+        var processed = mDecoder.TransformBlock(mBuffer, mOffset, count & ~15, buffer, offset);
+        mOffset += processed;
+        mWritten += processed;
+        return processed;
+    }
+
     #endregion
 }

src/SharpCompress/Compressors/LZMA/Bcj2DecoderStream.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.IO;
 
 namespace SharpCompress.Compressors.LZMA;
@@ -191,6 +193,18 @@ public override int Read(byte[] buffer, int offset, int count)
         return count;
     }
 
+    public override Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    )
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        // Bcj2DecoderStream uses complex state machine with multiple streams
+        return Task.FromResult(Read(buffer, offset, count));
+    }
+
     public override int ReadByte()
     {
         if (_mFinished)

src/SharpCompress/Compressors/LZMA/LZ/LzOutWindow.cs

@@ -3,6 +3,8 @@
 using System;
 using System.Buffers;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace SharpCompress.Compressors.LZMA.LZ;
 
@@ -85,6 +87,12 @@ public void ReleaseStream()
         _stream = null;
     }
 
+    public async Task ReleaseStreamAsync(CancellationToken cancellationToken = default)
+    {
+        await FlushAsync(cancellationToken).ConfigureAwait(false);
+        _stream = null;
+    }
+
     private void Flush()
     {
         if (_stream is null)
@@ -104,6 +112,27 @@ private void Flush()
         _streamPos = _pos;
     }
 
+    private async Task FlushAsync(CancellationToken cancellationToken = default)
+    {
+        if (_stream is null)
+        {
+            return;
+        }
+        var size = _pos - _streamPos;
+        if (size == 0)
+        {
+            return;
+        }
+        await _stream
+            .WriteAsync(_buffer, _streamPos, size, cancellationToken)
+            .ConfigureAwait(false);
+        if (_pos >= _windowSize)
+        {
+            _pos = 0;
+        }
+        _streamPos = _pos;
+    }
+
     public void CopyPending()
     {
         if (_pendingLen < 1)
@@ -124,6 +153,26 @@ public void CopyPending()
         _pendingLen = rem;
     }
 
+    public async Task CopyPendingAsync(CancellationToken cancellationToken = default)
+    {
+        if (_pendingLen < 1)
+        {
+            return;
+        }
+        var rem = _pendingLen;
+        var pos = (_pendingDist < _pos ? _pos : _pos + _windowSize) - _pendingDist - 1;
+        while (rem > 0 && HasSpace)
+        {
+            if (pos >= _windowSize)
+            {
+                pos = 0;
+            }
+            await PutByteAsync(_buffer[pos++], cancellationToken).ConfigureAwait(false);
+            rem--;
+        }
+        _pendingLen = rem;
+    }
+
     public void CopyBlock(int distance, int len)
     {
         var rem = len;
@@ -157,6 +206,43 @@ public void CopyBlock(int distance, int len)
         _pendingDist = distance;
     }
 
+    public async Task CopyBlockAsync(
+        int distance,
+        int len,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var rem = len;
+        var pos = (distance < _pos ? _pos : _pos + _windowSize) - distance - 1;
+        var targetSize = HasSpace ? (int)Math.Min(rem, _limit - _total) : 0;
+        var sizeUntilWindowEnd = Math.Min(_windowSize - _pos, _windowSize - pos);
+        var sizeUntilOverlap = Math.Abs(pos - _pos);
+        var fastSize = Math.Min(Math.Min(sizeUntilWindowEnd, sizeUntilOverlap), targetSize);
+        if (fastSize >= 2)
+        {
+            _buffer.AsSpan(pos, fastSize).CopyTo(_buffer.AsSpan(_pos, fastSize));
+            _pos += fastSize;
+            pos += fastSize;
+            _total += fastSize;
+            if (_pos >= _windowSize)
+            {
+                await FlushAsync(cancellationToken).ConfigureAwait(false);
+            }
+            rem -= fastSize;
+        }
+        while (rem > 0 && HasSpace)
+        {
+            if (pos >= _windowSize)
+            {
+                pos = 0;
+            }
+            await PutByteAsync(_buffer[pos++], cancellationToken).ConfigureAwait(false);
+            rem--;
+        }
+        _pendingLen = rem;
+        _pendingDist = distance;
+    }
+
     public void PutByte(byte b)
     {
         _buffer[_pos++] = b;
@@ -167,6 +253,16 @@ public void PutByte(byte b)
         }
     }
 
+    public async Task PutByteAsync(byte b, CancellationToken cancellationToken = default)
+    {
+        _buffer[_pos++] = b;
+        _total++;
+        if (_pos >= _windowSize)
+        {
+            await FlushAsync(cancellationToken).ConfigureAwait(false);
+        }
+    }
+
     public byte GetByte(int distance)
     {
         var pos = _pos - distance - 1;
@@ -207,6 +303,44 @@ public int CopyStream(Stream stream, int len)
         return len - size;
     }
 
+    public async Task<int> CopyStreamAsync(
+        Stream stream,
+        int len,
+        CancellationToken cancellationToken = default
+    )
+    {
+        var size = len;
+        while (size > 0 && _pos < _windowSize && _total < _limit)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            var curSize = _windowSize - _pos;
+            if (curSize > _limit - _total)
+            {
+                curSize = (int)(_limit - _total);
+            }
+            if (curSize > size)
+            {
+                curSize = size;
+            }
+            var numReadBytes = await stream
+                .ReadAsync(_buffer, _pos, curSize, cancellationToken)
+                .ConfigureAwait(false);
+            if (numReadBytes == 0)
+            {
+                throw new DataErrorException();
+            }
+            size -= numReadBytes;
+            _pos += numReadBytes;
+            _total += numReadBytes;
+            if (_pos >= _windowSize)
+            {
+                await FlushAsync(cancellationToken).ConfigureAwait(false);
+            }
+        }
+        return len - size;
+    }
+
     public void SetLimit(long size) => _limit = _total + size;
 
     public bool HasSpace => _pos < _windowSize && _total < _limit;

src/SharpCompress/Compressors/LZMA/LZipStream.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Buffers.Binary;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 using SharpCompress.Common;
 using SharpCompress.Crypto;
 using SharpCompress.IO;
@@ -157,6 +159,11 @@ public override int Read(byte[] buffer, int offset, int count) =>
 
 #if !NETFRAMEWORK && !NETSTANDARD2_0
 
+    public override ValueTask<int> ReadAsync(
+        Memory<byte> buffer,
+        CancellationToken cancellationToken = default
+    ) => _stream.ReadAsync(buffer, cancellationToken);
+
     public override int Read(Span<byte> buffer) => _stream.Read(buffer);
 
     public override void Write(ReadOnlySpan<byte> buffer)
@@ -179,6 +186,25 @@ public override void WriteByte(byte value)
         ++_writeCount;
     }
 
+    public override Task<int> ReadAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken = default
+    ) => _stream.ReadAsync(buffer, offset, count, cancellationToken);
+
+    public override async Task WriteAsync(
+        byte[] buffer,
+        int offset,
+        int count,
+        CancellationToken cancellationToken
+    )
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        await _stream.WriteAsync(buffer, offset, count, cancellationToken);
+        _writeCount += count;
+    }
+
     #endregion
 
     /// <summary>