Adjust SimpleZipArchiveReader changes + Pool Stream on Zip extractor

Author: neon-nyan

CollapseLauncher/Classes/GameManagement/WpfPackage/WpfPackageContext.Methods.cs

@@ -95,7 +95,7 @@ private async ValueTask PerformDownloadFromZipOnTheFly(
         string gamePath = GamePath;
 
         ZipArchiveReader reader =
-            await ZipArchiveReader.CreateFromRemoteAsync(url, _localCts.Token);
+            await ZipArchiveReader.CreateFromAsync(url, _localCts.Token);
 
         long totalSizeUncompressed = reader.Sum(x => x.Size);
         ProgressAllSizeTotal     = totalSizeUncompressed;
@@ -147,8 +147,7 @@ await IsPackageHashMatchWithCrc32(stream, entry.Crc32, token))
             byte[] buffer = ArrayPool<byte>.Shared.Rent(bufferSize);
             try
             {
-                await using Stream deflateStream =
-                    await entry.OpenStreamFromFactoryAsync(CreateStreamFromUrl, token);
+                await using Stream deflateStream = await entry.OpenStreamFromAsync(CreateStreamFromUrl, token);
 
                 int read;
                 while ((read = await deflateStream.ReadAsync(buffer, token)) > 0)

CollapseLauncher/Classes/Helper/ThreadObjectPool.cs

@@ -0,0 +1,102 @@
+using System;
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+#nullable enable
+namespace CollapseLauncher.Helper;
+
+internal class ThreadObjectPool<T> : IDisposable where T : class
+{
+    private ConcurrentQueue<T> Items = new();
+    private SemaphoreSlim Semaphore;
+    private Func<object?> _factory;
+
+    private int _countUsed;
+    private readonly int _capacity;
+    private readonly bool _isDisposeObjects;
+    private bool _isDisposed;
+
+    internal ThreadObjectPool(Func<T> factory, int capacity = 0, bool isDisposeObjects = true)
+    {
+        _capacity = capacity == 0 ? Environment.ProcessorCount : capacity;
+        _factory = factory;
+        _isDisposeObjects = isDisposeObjects;
+        Semaphore = new SemaphoreSlim(_capacity, _capacity);
+    }
+
+    internal ThreadObjectPool(Task<T> factoryAsync, int capacity = 0, bool isDisposeObjects = true)
+    {
+        _capacity = capacity == 0 ? Environment.ProcessorCount : capacity;
+        _factory = () => factoryAsync;
+        _isDisposeObjects = isDisposeObjects;
+        Semaphore = new SemaphoreSlim(_capacity, _capacity);
+    }
+
+    internal async Task<T> GetOrCreateObjectAsync(CancellationToken token = default)
+    {
+        await Semaphore.WaitAsync(token);
+        Interlocked.Increment(ref _countUsed);
+
+        if (Items.TryDequeue(out T? pooled))
+            return pooled;
+
+        if (_factory is Func<Task<T>> asyncFactory)
+            return await asyncFactory();
+
+        return ((Func<T>)_factory)();
+    }
+
+    internal T GetOrCreateObject()
+    {
+        Semaphore.Wait();
+        Interlocked.Increment(ref _countUsed);
+
+        if (Items.TryDequeue(out T? pooled))
+            return pooled;
+
+        if (_factory is Func<Task<T>> asyncFactory)
+            return asyncFactory().Result;
+
+        return ((Func<T>)_factory)();
+    }
+
+    internal void Return(T item)
+    {
+        Items.Enqueue(item);
+        Interlocked.Decrement(ref _countUsed);
+        Semaphore.Release();
+    }
+
+    public void Dispose()
+    {
+        if (_isDisposed)
+        {
+            return;
+        }
+
+        Semaphore.Dispose();
+
+        try
+        {
+            if (!_isDisposeObjects)
+            {
+                return;
+            }
+
+            foreach (IDisposable item in Items.OfType<IDisposable>())
+            {
+                item.Dispose();
+            }
+        }
+        finally
+        {
+            _isDisposed = true;
+            Items.Clear();
+            _factory = null!;
+            Items = null!;
+            Semaphore = null!;
+        }
+    }
+}

CollapseLauncher/Classes/InstallManagement/Genshin/GenshinInstall.cs

@@ -203,7 +203,7 @@ protected override GameInstallFileInfo GetGameInstallFileInfo()
                 (sdkApi.Data?.TryFindByBizOrId(gameBiz, gameId, out HypChannelSdkData sdkData) ?? false) &&
                 sdkData.SdkPackageDetail?.Url is { } sdkZipUrl)
             {
-                ZipArchiveReader reader = await ZipArchiveReader.CreateFromRemoteAsync(sdkZipUrl);
+                ZipArchiveReader reader = await ZipArchiveReader.CreateFromAsync(sdkZipUrl);
                 localFileInfos.AddRange(reader
                                        .Where(x => !x.IsDirectory)
                                        .Select(ConvertZipEntry));
@@ -214,7 +214,7 @@ protected override GameInstallFileInfo GetGameInstallFileInfo()
                 (wpfApi.Data?.TryFindByBizOrId(gameBiz, gameId, out HypWpfPackageData wpfData) ?? false) &&
                 wpfData.PackageInfo?.Url is { } wpfZipUrl)
             {
-                ZipArchiveReader reader = await ZipArchiveReader.CreateFromRemoteAsync(wpfZipUrl);
+                ZipArchiveReader reader = await ZipArchiveReader.CreateFromAsync(wpfZipUrl);
                 localFileInfos.AddRange(reader
                                        .Where(x => !x.IsDirectory)
                                        .Select(ConvertZipEntry));

CollapseLauncher/Classes/Interfaces/Class/ProgressBase.cs

@@ -1007,7 +1007,7 @@ protected async Task FetchBilibiliSdk(CancellationToken token)
         string url      = sdkData.SdkPackageDetail?.Url ?? throw new NullReferenceException();
 
         // Create ZipArchiveReader and get the remote stream of the zip file
-        ZipArchiveReader reader = await ZipArchiveReader.CreateFromRemoteAsync(url, token);
+        ZipArchiveReader reader = await ZipArchiveReader.CreateFromAsync(url, token);
         HttpClient       client = FallbackCDNUtil.GetGlobalHttpClient(true);
 
         await Parallel.ForEachAsync(reader.Where(x => !x.IsDirectory),
@@ -1041,7 +1041,7 @@ async ValueTask Impl(ZipArchiveEntry entry, CancellationToken innerToken)
                 return;
             }
 
-            await using Stream entryStream = await entry.OpenStreamFromFactoryAsync(CreateStreamFromPosUrl, innerToken);
+            await using Stream entryStream = await entry.OpenStreamFromAsync(CreateStreamFromPosUrl, innerToken);
             // Reset the SDK DLL stream pos and write the data
             sdkDllStream.Position = 0;
             await entryStream.CopyToAsync(sdkDllStream, innerToken);
@@ -1359,11 +1359,11 @@ internal static ValueTask<FileStream> NaivelyOpenFileStreamAsync(FileInfo info,
     protected virtual long GetArchiveUncompressedSizeManaged(Stream archiveStream)
     {
         ZipArchiveReader archive = ZipArchiveReader
-           .CreateFromStreamFactory((pos, _) =>
-                                    {
-                                        archiveStream.Position = pos ?? 0;
-                                        return archiveStream;
-                                    });
+           .CreateFrom((pos, _) =>
+           {
+               archiveStream.Position = pos ?? 0;
+               return archiveStream;
+           });
 
         return archive.Sum(x => x.Size);
     }
@@ -1424,9 +1424,24 @@ protected virtual async Task ExtractUsingManagedZip(
         string            outputDir,
         CancellationToken token)
     {
+        // Use ThreadObjectPool to cache the Streams and re-using it.
+        using ThreadObjectPool<Stream> streamPool = new(
+            () => CreateStreamWithPos(0, 0, token),
+            capacity: ThreadCount * 2, // Double the thread count for spare capacity
+            isDisposeObjects: true);
+
         int threadCounts = ThreadCount;
-        ZipArchiveReader archive = await ZipArchiveReader
-           .CreateFromStreamFactoryAsync(CreateStreamWithPos, token);
+        Stream zipInitialStream = await streamPool.GetOrCreateObjectAsync(token);
+        ZipArchiveReader archive;
+        
+        try
+        {
+            archive = await ZipArchiveReader.CreateFromAsync(zipInitialStream, token);
+        }
+        finally
+        {
+            streamPool.Return(zipInitialStream);
+        }
 
         // Run the workers
         await Parallel.ForEachAsync(archive,
@@ -1439,45 +1454,53 @@ await Parallel.ForEachAsync(archive,
 
         return;
 
-        ValueTask Impl(ZipArchiveEntry entry, CancellationToken innerToken)
+        async ValueTask Impl(ZipArchiveEntry entry, CancellationToken innerToken)
         {
-            return ExtractUsingManagedZipWorker(entry,
-                                                CreateStreamWithPos,
-                                                outputDir,
-                                                innerToken);
+            string outputPath = Path.Combine(outputDir, entry.Filename);
+            if (entry.IsDirectory)
+            {
+                _ = Directory.CreateDirectory(outputPath);
+                return;
+            }
+
+            Stream reusableStream = await streamPool.GetOrCreateObjectAsync(innerToken);
+            try
+            {
+                await ExtractUsingManagedZipWorker(entry,
+                                                   reusableStream,
+                                                   outputDir,
+                                                   innerToken);
+            }
+            finally
+            {
+                streamPool.Return(reusableStream);
+            }
         }
 
-        Task<Stream> CreateStreamWithPos(long? start, long? _, CancellationToken innerToken)
-            => Task.Factory.StartNew(() =>
-                                     {
-                                         Stream stream = streamFactory();
-                                         stream.Position = start ?? 0;
-                                         return stream;
-                                     }, innerToken);
+        Stream CreateStreamWithPos(long? start, long? _, CancellationToken innerToken)
+        {
+            Stream stream = streamFactory();
+            stream.Position = start ?? 0;
+            return stream;
+        }
     }
 
     protected virtual async ValueTask ExtractUsingManagedZipWorker(
-        ZipArchiveEntry    entry,
-        StreamFactoryAsync streamFactory,
-        string             outputDir,
-        CancellationToken  token)
+        ZipArchiveEntry   entry,
+        Stream            sourceStream,
+        string            outputDir,
+        CancellationToken token)
     {
         byte[] buffer = ArrayPool<byte>.Shared.Rent(128 << 10);
 
         try
         {
-            if (token.IsCancellationRequested ||
-                entry.IsDirectory)
-            {
-                return;
-            }
-
             string outputPath = Path.Combine(outputDir, entry.Filename);
             FileInfo outputFile = new FileInfo(outputPath).EnsureCreationOfDirectory()
                                                           .StripAlternateDataStream()
                                                           .EnsureNoReadOnly();
 
-            await using Stream deflateStream = await entry.OpenStreamFromFactoryAsync(streamFactory, token);
+            await using Stream deflateStream = await entry.OpenStreamFromAsync(sourceStream, false, false, token);
             await using FileStream fileStream = outputFile.Open(FileMode.Create,
                                                                   FileAccess.Write,
                                                                   FileShare.Write);