ScanAndRepair() now allow to set the number of concurrent downloads (#12)

* ScanAndRepair() now allow to set the number of concurrent downloads

---------------------
GameScanner
---------------------
Add concurrentDownload "setting":

	- if = 0 (default) then the amount of concurrentDownload woulbe ProcessorCount * 5.
	- if = 1 then then "SimpleFileDownloader" will be used not "ChunkFileDownload".
	- if > 1 then the amount of concurrentDownload woulbe the value passed.

In all case the amount of concurrentDownload would be limited at max to 10 (that more than enough to get an great speed without abusing too much server bp).

Usage: "int concurrentDownload = 0" has been add as last paramater in "ScanAndRepair()".

---------------------
ChunckFileDownloader
---------------------

- The amount of concurrent worker can now be configured (see upper -> GameScanner: concurrentDownload).
- Enforce usage of the "cancellation token" on download operation so if throw they are canceled and we don't have to wait for them to complete.
- Worker now start one by one with a delay of 1sec.
- Now if an chunk fail to download instead of failling the "whole download" we retry to download it once in the same worker and if it fail again the worker stop and the item is re-enqueue so it can be processed by another worker, so in the end if an error append we will reduce the amount of concurent worker one by one until none left (their is an fallback in case an chunk is re-enqueue but all worker have completed already).

Author: Ryddag

src/FileDownloader/ChunkDownload.cs

@@ -1,15 +1,16 @@
 using System;
 using System.IO;
 using System.Net;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ProjectCeleste.GameFiles.GameScanner.FileDownloader
 {
     internal class ChunkDownload
     {
-        private const int ChunkBufferSize = 32 * 1024; // 32Kb
+        internal const int ChunkBufferSize = 32 * 1024; // 32Kb
 
-        internal string DownloadTmpFileName { get; private set; }
+        internal string DownloadTmpFileName { get; }
 
         private readonly string _fileToDownload;
         private readonly FileRange _fileRange;
@@ -38,28 +39,26 @@ private HttpWebRequest CreateHttpWebRequest()
             return downloadRequest;
         }
 
-        internal async Task<bool> TryDownloadAsync(Action<int> progressCallback)
+        internal async Task<bool> TryDownloadAsync(Action<int> progressCallback, CancellationToken ct = default)
         {
             var downloadRequest = CreateHttpWebRequest();
 
             try
             {
-                using (var downloadResponse = (HttpWebResponse) downloadRequest.GetResponse())
-                using (var downloadSource = downloadResponse.GetResponseStream())
-                using (var downloadTarget = new FileStream(DownloadTmpFileName, FileMode.Create, FileAccess.Write))
-                {
-                    int bytesRead;
-                    var buffer = new byte[ChunkBufferSize];
+                using var downloadResponse = (HttpWebResponse) downloadRequest.GetResponse();
+                using var downloadSource = downloadResponse.GetResponseStream();
+                using var downloadTarget = new FileStream(DownloadTmpFileName, FileMode.Create, FileAccess.Write);
+                int bytesRead;
+                var buffer = new byte[ChunkBufferSize];
 
-                    do
-                    {
-                        bytesRead = await downloadSource.ReadAsync(buffer, 0, ChunkBufferSize);
-                        downloadTarget.Write(buffer, 0, bytesRead);
+                do
+                {
+                    bytesRead = await downloadSource.ReadAsync(buffer, 0, ChunkBufferSize, ct);
+                    await downloadTarget.WriteAsync(buffer, 0, bytesRead, ct);
 
-                        progressCallback(bytesRead);
-                        _bytesDownloaded += bytesRead;
-                    } while (bytesRead > 0);
-                }
+                    progressCallback(bytesRead);
+                    _bytesDownloaded += bytesRead;
+                } while (bytesRead > 0);
             }
             catch
             {

src/FileDownloader/ChunkFileDownloader.cs

@@ -13,22 +13,29 @@ namespace ProjectCeleste.GameFiles.GameScanner.FileDownloader
     public class ChunkFileDownloader : IFileDownloader
     {
         internal const int MaxChunkSize = 10 * 1024 * 1024; //10Mb
-        private const int MaxConcurrentDownloads = 6;
+        public const int MaxConcurrentDownloads = 10;
 
+        private readonly int _concurrentDownloads;
         private readonly Stopwatch _downloadSpeedStopwatch;
         private readonly string _downloadTempFolder;
-
         private readonly ConcurrentDictionary<long, string> _completedChunks = new ConcurrentDictionary<long, string>();
-        private ConcurrentQueue<FileRange> _chunkDownloadQueue;
 
+        private ConcurrentQueue<FileRange> _chunkDownloadQueue;
         private long _downloadSizeCompleted;
-        private int _activeDownloads = 1;
-        private bool _downloadFailed = false;
 
-        public ChunkFileDownloader(string httpLink, string outputFileName, string tmpFolder)
+        public ChunkFileDownloader(string httpLink, string outputFileName, string tmpFolder, int concurrentDownload = 0)
         {
             DownloadUrl = httpLink;
             FilePath = outputFileName;
+
+            if (concurrentDownload <= 0)
+                concurrentDownload = 5;
+
+            if (concurrentDownload > MaxConcurrentDownloads)
+                concurrentDownload = MaxConcurrentDownloads;
+
+            _concurrentDownloads = concurrentDownload;
+
             _downloadTempFolder = tmpFolder;
             _downloadSpeedStopwatch = new Stopwatch();
 
@@ -37,7 +44,7 @@ public ChunkFileDownloader(string httpLink, string outputFileName, string tmpFol
             ServicePointManager.MaxServicePointIdleTime = 1000;
         }
 
-        public FileDownloaderState State { get; private set; } = FileDownloaderState.Invalid;
+        public FileDownloaderState State { get; private set; }
 
         public double DownloadProgress => DownloadSize > 0 ? (double) BytesDownloaded / DownloadSize * 100 : 0;
 
@@ -80,31 +87,36 @@ public async Task DownloadAsync(CancellationToken ct = default)
             try
             {
                 _downloadSpeedStopwatch.Start();
-
-                OnProgressChanged();
-
                 DownloadSize = await GetDownloadSizeAsync();
 
-                var readRanges = CalculateFileChunkRanges();
+                ReportProgress(null); //Forced
 
-                //Parallel download
-                using (new Timer(ReportProgress, null, 100, 100))
+                using (var reportProgressTimer = new Timer(ReportProgress, null, 500, 500))
                 {
-                    await OrchestrateDownloadWorkersAsync(readRanges, ct);
+                    await StartDownloadAsync(ct);
                     _downloadSpeedStopwatch.Stop();
+                    reportProgressTimer.Change(Timeout.Infinite, Timeout.Infinite);
+                }
 
-                    if (BytesDownloaded != DownloadSize)
-                        throw new Exception(
-                            $"Download was completed ({BytesDownloaded} bytes), but did not receive expected size of {DownloadSize} bytes");
+                ReportProgress(null); //Forced
 
-                    State = FileDownloaderState.Finalize;
-                    ReportProgress(null); //Forced
+                if (_chunkDownloadQueue.Count > 0)
+                    throw new Exception(
+                        $"Download was incomplete ({_chunkDownloadQueue.Count} missing chunks)");
 
-                    WriteChunksToFile(_completedChunks);
+                if (BytesDownloaded != DownloadSize)
+                    throw new Exception(
+                        $"Download was incomplete ({BytesDownloaded}/{DownloadSize} bytes)");
 
-                    State = FileDownloaderState.Complete;
-                    ReportProgress(null); //Forced
-                }
+                State = FileDownloaderState.Finalize;
+                
+                ReportProgress(null); //Forced
+
+                ct.ThrowIfCancellationRequested();
+
+                WriteChunksToFile(_completedChunks);
+
+                State = FileDownloaderState.Complete;
             }
             catch (Exception e)
             {
@@ -115,46 +127,61 @@ public async Task DownloadAsync(CancellationToken ct = default)
             }
             finally
             {
-                OnProgressChanged();
+                ReportProgress(null); //Forced
             }
         }
 
-        private async Task OrchestrateDownloadWorkersAsync(IEnumerable<FileRange> chunks, CancellationToken ct)
+        private async Task StartDownloadAsync(CancellationToken ct = default)
         {
-            var downloaderWorkers = new ConcurrentQueue<Task>();
-            _chunkDownloadQueue = new ConcurrentQueue<FileRange>(chunks);
+            var readRanges = CalculateFileChunkRanges();
+            var fileRanges = readRanges as FileRange[] ?? readRanges.ToArray();
+            _chunkDownloadQueue = new ConcurrentQueue<FileRange>(fileRanges);
 
-            while (_activeDownloads < MaxConcurrentDownloads && _chunkDownloadQueue.Count > 0 && !_downloadFailed)
-            {
-                _activeDownloads++;
+            var tasks = Enumerable.Range(1, Math.Min(_concurrentDownloads, _chunkDownloadQueue.Count)).Select(
+                async workerIndex =>
+                {
+                    if (workerIndex > 1)
+                        await Task.Delay(1000 * (workerIndex - 1), ct);
 
-                downloaderWorkers.Enqueue(Task.Run(DequeueAndDownloadChunksAsync, ct));
-                await Task.Delay(1000, ct);
-            }
+                    await DequeueAndDownloadChunksAsync(ct);
+                });
 
-            await Task.WhenAll(downloaderWorkers);
+            await Task.WhenAll(tasks); //Start parallel download
+
+            if (_completedChunks.Count > 0 && _chunkDownloadQueue.Count > 0)
+            {
+                //Try to get missing chunk if any
+                await DequeueAndDownloadChunksAsync(ct);
+            }
         }
 
-        private async Task DequeueAndDownloadChunksAsync()
+        private async Task DequeueAndDownloadChunksAsync(CancellationToken ct = default)
         {
-            var workerFailedDownloading = false;
-            var taskId = _activeDownloads;
-
-            while (_chunkDownloadQueue.TryDequeue(out var fileChunk) && !workerFailedDownloading)
+            while (_chunkDownloadQueue.TryDequeue(out var fileChunk))
             {
+                var workerFailedDownloadingOnce = false;
+
+                retry:
                 var chunkDownload = new ChunkDownload(DownloadUrl, fileChunk, _downloadTempFolder);
-                var downloadSuccesfullyCompleted = await chunkDownload.TryDownloadAsync(IncrementTotalDownloadProgress);
+                var downloadSuccessfullyCompleted =
+                    await chunkDownload.TryDownloadAsync(IncrementTotalDownloadProgress, ct);
 
-                if (!downloadSuccesfullyCompleted)
-                {
-                    _chunkDownloadQueue.Enqueue(fileChunk);
-                    workerFailedDownloading = true;
-                    _downloadFailed = true;
-                }
-                else
+                if (!downloadSuccessfullyCompleted)
                 {
-                    _completedChunks.TryAdd(fileChunk.Start, chunkDownload.DownloadTmpFileName);
+                    if (workerFailedDownloadingOnce)
+                    {
+                        _chunkDownloadQueue.Enqueue(fileChunk);
+                        break;
+                    }
+
+                    workerFailedDownloadingOnce = true;
+                    await Task.Delay(1000, ct);
+                    goto retry;
                 }
+
+                _completedChunks.TryAdd(fileChunk.Start, chunkDownload.DownloadTmpFileName);
+
+                await Task.Delay(1000, ct);
             }
         }
 
@@ -176,23 +203,28 @@ private async Task<long> GetDownloadSizeAsync()
             var sizeDownloadRequest = WebRequest.Create(DownloadUrl);
             sizeDownloadRequest.Method = "HEAD";
 
-            using (var response = await sizeDownloadRequest.GetResponseAsync())
-            {
-                return long.Parse(response.Headers.Get("Content-Length"));
-            }
+            using var response = await sizeDownloadRequest.GetResponseAsync();
+            return long.Parse(response.Headers.Get("Content-Length"));
         }
 
         private void WriteChunksToFile(IDictionary<long, string> fileChunks)
         {
-            using (var targetFile = new BufferedStream(new FileStream(FilePath, FileMode.Create, FileAccess.Write)))
+            using var targetFile = new BufferedStream(new FileStream(FilePath, FileMode.Create, FileAccess.Write),
+                ChunkDownload.ChunkBufferSize);
+            foreach (var tempFile in fileChunks.ToArray().OrderBy(b => b.Key))
             {
-                foreach (var tempFile in fileChunks.ToArray().OrderBy(b => b.Key))
-                {
-                    using (var sourceChunks = new BufferedStream(File.OpenRead(tempFile.Value)))
-                        sourceChunks.CopyTo(targetFile);
+                using (var sourceChunks =
+                    new BufferedStream(File.OpenRead(tempFile.Value), ChunkDownload.ChunkBufferSize))
+                    sourceChunks.CopyTo(targetFile);
 
+                try
+                {
                     File.Delete(tempFile.Value);
                 }
+                catch
+                {
+                    //
+                }
             }
         }
 
@@ -203,14 +235,9 @@ private void IncrementTotalDownloadProgress(int bytesDownloaded)
 
         public event EventHandler ProgressChanged;
 
-        protected virtual void OnProgressChanged()
-        {
-            ProgressChanged?.Invoke(this, null);
-        }
-
         private void ReportProgress(object state)
         {
-            OnProgressChanged();
+            ProgressChanged?.Invoke(this, null);
         }
     }
 }

src/FileDownloader/IFileDownloader.cs

@@ -17,6 +17,6 @@ public interface IFileDownloader
 
         event EventHandler ProgressChanged;
 
-        Task DownloadAsync(CancellationToken ct = default(CancellationToken));
+        Task DownloadAsync(CancellationToken ct = default);
     }
 }