Merge pull request #140 from bezzad/feature/inmemory_buffering_limita…

…tion

Feature/inmemory buffering limitation

README.md

@@ -52,6 +52,7 @@ Downloader is compatible with .NET Standard 2.0 and above, running on Windows, L
 - Live streaming support, suitable for playing music at the same time as downloading.
 - Ability to download just a certain range of bytes of a large file.
 - Code is tiny, fast and does not depend on external libraries.
+- Control the amount of system memroy (RAM) that the Downloader consumes during downloading.
 
 ---
 
@@ -93,6 +94,8 @@ var downloadOpt = new DownloadConfiguration()
     MaximumBytesPerSecond = 1024*1024*2, 
     // the maximum number of times to fail
     MaxTryAgainOnFailover = 5,    
+    // release memory buffer after each 50 MB
+    MaximumMemoryBufferBytes = 1024 * 1024 * 50, 
     // download parts of file as parallel or not. Default value is false
     ParallelDownload = true,
     // number of parallel downloads. The default value is the same as the chunk count

src/Downloader.Test/Downloader.Test.csproj

@@ -24,19 +24,19 @@
     <PackageReference Include="AssertMessage.Fody" Version="2.1.0">
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="Fody" Version="6.6.4">
+    <PackageReference Include="Fody" Version="6.7.0">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.3.2" />
-    <PackageReference Include="Moq" Version="4.18.2" />
-    <PackageReference Include="MSTest.TestAdapter" Version="2.2.10" />
-    <PackageReference Include="MSTest.TestFramework" Version="2.2.10" />
-    <PackageReference Include="coverlet.collector" Version="3.2.0">
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.6.0" />
+    <PackageReference Include="Moq" Version="4.18.4" />
+    <PackageReference Include="MSTest.TestAdapter" Version="3.0.3" />
+    <PackageReference Include="MSTest.TestFramework" Version="3.0.3" />
+    <PackageReference Include="coverlet.collector" Version="6.0.0">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-    <PackageReference Include="Newtonsoft.Json" Version="13.0.2" />
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
   </ItemGroup>
 
   <ItemGroup>

src/Downloader.Test/IntegrationTests/DownloadServiceTest.cs

@@ -124,7 +124,7 @@ public void TestPackageSituationAfterDispose()
             Package.TotalFileSize = sampleDataLength * 64;
             Options.ChunkCount = 1;
             new ChunkHub(Options).SetFileChunks(Package);
-            Package.BuildStorage(false);
+            Package.BuildStorage(false, 1024 * 1024);
             Package.Storage.WriteAsync(0, sampleData, sampleDataLength);
             Package.Storage.Flush();
 
@@ -145,7 +145,7 @@ public async Task TestPackageChunksDataAfterDispose()
             var dummyData = DummyData.GenerateOrderedBytes(chunkSize);
             Options.ChunkCount = 64;
             Package.TotalFileSize = chunkSize * 64;
-            Package.BuildStorage(false);
+            Package.BuildStorage(false, 1024 * 1024);
             new ChunkHub(Options).SetFileChunks(Package);
             for (int i = 0; i < Package.Chunks.Length; i++)
             {

src/Downloader.Test/UnitTests/DownloadPackageTest.cs

@@ -19,7 +19,7 @@ public virtual void Initial()
         {
             Config = new DownloadConfiguration() { ChunkCount = 8 };
             Data = DummyData.GenerateOrderedBytes(DummyFileHelper.FileSize16Kb);
-            Package.BuildStorage(false);
+            Package.BuildStorage(false, 1024 * 1024);
             new ChunkHub(Config).SetFileChunks(Package);
             Package.Storage.WriteAsync(0, Data, DummyFileHelper.FileSize16Kb);
             Package.Storage.Flush();

src/Downloader.Test/UnitTests/DownloadPackageTestOnFile.cs

@@ -53,7 +53,7 @@ private void BuildStorageTest(bool reserveSpace)
             };
 
             // act
-            Package.BuildStorage(reserveSpace);
+            Package.BuildStorage(reserveSpace, 1024*1024);
 
             // assert
             Assert.IsInstanceOfType(Package.Storage.OpenRead(), typeof(FileStream));

src/Downloader.Test/UnitTests/RequestTest.cs

@@ -389,7 +389,7 @@ public void GetRedirectUrlByLocationTest()
 
             // assert
             Assert.AreNotEqual(url, redirectUrl);
-            Assert.AreNotEqual(request.Address, redirectUrl);
+            Assert.AreNotEqual(request.Address.ToString(), redirectUrl);
             Assert.AreEqual(redirectUrl, actualRedirectUrl.AbsoluteUri);
         }
 
@@ -408,7 +408,7 @@ public void GetRedirectUrlWithoutLocationTest()
 
             // assert
             Assert.AreNotEqual(url, redirectUrl);
-            Assert.AreNotEqual(request.Address, redirectUrl);
+            Assert.AreNotEqual(request.Address.ToString(), redirectUrl);
             Assert.AreEqual(redirectUrl, actualRedirectUrl.AbsoluteUri);
         }
 

src/Downloader/ConcurrentStream.cs

@@ -8,11 +8,11 @@ namespace Downloader
 {
     public class ConcurrentStream : IDisposable
     {
-        private readonly SemaphoreSlim _queueCheckerSemaphore = new SemaphoreSlim(0);
+        private readonly SemaphoreSlim _queueConsumerLocker = new SemaphoreSlim(0);
         private readonly ManualResetEventSlim _completionEvent = new ManualResetEventSlim(true);
+        private readonly ManualResetEventSlim _stopWriteNewPacketEvent = new ManualResetEventSlim(true);
         private readonly ConcurrentBag<Packet> _inputBag = new ConcurrentBag<Packet>();
-        private int? _resourceReleaseThreshold;
-        private long _packetCounter = 0;
+        private long _maxMemoryBufferBytes = 0;
         private bool _disposed;
         private Stream _stream;
         private string _path;
@@ -29,7 +29,7 @@ public string Path
                 }
             }
         }
-        
+
         public byte[] Data
         {
             get
@@ -49,24 +49,35 @@ public byte[] Data
 
         public long Length => _stream?.Length ?? 0;
 
-        public ConcurrentStream(Stream stream)
+        public long MaxMemoryBufferBytes
+        {
+            get => _maxMemoryBufferBytes;
+            set
+            {
+                _maxMemoryBufferBytes = (value <= 0) ? long.MaxValue : value;
+            }
+        }
+
+        public ConcurrentStream(Stream stream, long maxMemoryBufferBytes = 0)
         {
             _stream = stream;
+            MaxMemoryBufferBytes = maxMemoryBufferBytes;
             Initial();
         }
 
-        public ConcurrentStream(string filename, long initSize)
+        public ConcurrentStream(string filename, long initSize, long maxMemoryBufferBytes = 0)
         {
             _path = filename;
             _stream = new FileStream(filename, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.Read);
+            MaxMemoryBufferBytes = maxMemoryBufferBytes;
 
             if (initSize > 0)
                 _stream.SetLength(initSize);
 
             Initial();
         }
 
-        public ConcurrentStream()
+        public ConcurrentStream() // parameterless constructor for deserialization
         {
             _stream = new MemoryStream();
             Initial();
@@ -88,44 +99,46 @@ public Stream OpenRead()
 
         public void WriteAsync(long position, byte[] bytes, int length)
         {
+            _stopWriteNewPacketEvent.Wait();
             _inputBag.Add(new Packet(position, bytes, length));
             _completionEvent.Reset();
-            _queueCheckerSemaphore.Release();
+            _queueConsumerLocker.Release();
+            ReleaseQueue(length);
         }
 
         private async Task Watcher()
         {
             while (!_disposed)
             {
-                await _queueCheckerSemaphore.WaitAsync().ConfigureAwait(false);
+                await _queueConsumerLocker.WaitAsync().ConfigureAwait(false);
                 if (_inputBag.TryTake(out var packet))
                 {
                     await WritePacket(packet).ConfigureAwait(false);
-                    ReleasePackets(packet.Data.Length);
                     packet.Dispose();
                 }
             }
         }
+
         private async Task WritePacket(Packet packet)
         {
             if (_stream.CanSeek)
             {
                 _stream.Position = packet.Position;
                 await _stream.WriteAsync(packet.Data, 0, packet.Length).ConfigureAwait(false);
-                _packetCounter++;
             }
 
             if (_inputBag.IsEmpty)
                 _completionEvent.Set();
         }
 
-        private void ReleasePackets(int packetSize)
+        private void ReleaseQueue(int packetSize)
         {
-            _resourceReleaseThreshold ??= 1024 * 1024 * 50 / packetSize; // 50MB / a packet size
-
             // Clean up RAM every _resourceReleaseThreshold packet
-            if (_packetCounter % _resourceReleaseThreshold == 0)
-                GC.Collect();
+            if (MaxMemoryBufferBytes < packetSize * _inputBag.Count)
+            {
+                _stopWriteNewPacketEvent.Set();
+                Flush();
+            }
         }
 
         public void Flush()
@@ -141,7 +154,7 @@ public void Dispose()
             {
                 Flush();
                 _disposed = true;
-                _queueCheckerSemaphore.Dispose();
+                _queueConsumerLocker.Dispose();
                 _stream.Dispose();
             }
         }

src/Downloader/DownloadConfiguration.cs

@@ -9,8 +9,9 @@ public class DownloadConfiguration : ICloneable, INotifyPropertyChanged
         private int _bufferBlockSize;
         private int _chunkCount;
         private long _maximumBytesPerSecond;
+        private int _maximumTryAgainOnFailover;
+        private long _maximumMemoryBufferBytes;
         private bool _checkDiskSizeBeforeDownload;
-        private int _maxTryAgainOnFailover;
         private bool _parallelDownload;
         private int _parallelCount;
         private int _timeout;
@@ -26,7 +27,7 @@ public class DownloadConfiguration : ICloneable, INotifyPropertyChanged
         public DownloadConfiguration()
         {
             RequestConfiguration = new RequestConfiguration(); // default requests configuration
-            _maxTryAgainOnFailover = int.MaxValue; // the maximum number of times to fail.
+            _maximumTryAgainOnFailover = int.MaxValue; // the maximum number of times to fail.
             _parallelDownload = false; // download parts of file as parallel or not
             _parallelCount = 0; // number of parallel downloads
             _chunkCount = 1; // file parts to download
@@ -117,10 +118,10 @@ public long MaximumBytesPerSecond
         /// </summary>
         public int MaxTryAgainOnFailover
         {
-            get => _maxTryAgainOnFailover;
+            get => _maximumTryAgainOnFailover;
             set
             {
-                _maxTryAgainOnFailover = value;
+                _maximumTryAgainOnFailover = value;
                 OnPropertyChanged();
             }
         }
@@ -249,6 +250,29 @@ public bool ReserveStorageSpaceBeforeStartingDownload
             }
         }
 
+        /// <summary>
+        /// Gets or sets the maximum amount of memory, in bytes, that the Downloader library is allowed
+        /// to allocate for buffering downloaded content. Once this limit is reached, the library will
+        /// stop downloading and start writing the buffered data to a file stream before continuing.
+        /// The default value for is 0, which indicates unlimited buffering.
+        /// </summary>
+        /// <example>
+        /// The following example sets the maximum memory buffer to 50 MB, causing the library to release
+        /// the memory buffer after each 50 MB of downloaded content:
+        /// <code>
+        /// MaximumMemoryBufferBytes = 1024 * 1024 * 50
+        /// </code>
+        /// </example>
+        public long MaximumMemoryBufferBytes
+        {
+            get => _maximumMemoryBufferBytes;
+            set
+            {
+                _maximumMemoryBufferBytes = value;
+                OnPropertyChanged();
+            }
+        }
+
         public object Clone()
         {
             return MemberwiseClone();

src/Downloader/DownloadPackage.cs

@@ -50,12 +50,12 @@ public void Validate()
             }
         }
 
-        public void BuildStorage(bool reserveFileSize)
+        public void BuildStorage(bool reserveFileSize, long maxMemoryBufferBytes = 0)
         {
             if (string.IsNullOrWhiteSpace(FileName))
-                Storage = new ConcurrentStream();
+                Storage = new ConcurrentStream() { MaxMemoryBufferBytes = maxMemoryBufferBytes };
             else
-                Storage = new ConcurrentStream(FileName, reserveFileSize ? TotalFileSize : 0);
+                Storage = new ConcurrentStream(FileName, reserveFileSize ? TotalFileSize : 0, maxMemoryBufferBytes);
         }
     }
 }

src/Downloader/DownloadService.cs

@@ -178,7 +178,7 @@ private async Task<Stream> StartDownload()
                 await _singleInstanceSemaphore.WaitAsync();
                 Package.TotalFileSize = await _requestInstance.GetFileSize().ConfigureAwait(false);
                 Package.IsSupportDownloadInRange = await _requestInstance.IsSupportDownloadInRange().ConfigureAwait(false);
-                Package.BuildStorage(Options.ReserveStorageSpaceBeforeStartingDownload);
+                Package.BuildStorage(Options.ReserveStorageSpaceBeforeStartingDownload, Options.MaximumMemoryBufferBytes);
                 ValidateBeforeChunking();
                 _chunkHub.SetFileChunks(Package);
 

src/Downloader/Downloader.csproj

@@ -10,7 +10,7 @@
     <PackageProjectUrl>https://github.com/bezzad/Downloader</PackageProjectUrl>
     <RepositoryUrl>https://github.com/bezzad/Downloader</RepositoryUrl>
     <PackageTags>download-manager, downloader, download, idm, internet, streaming, download-file, stream-downloader, multipart-download</PackageTags>
-    <PackageReleaseNotes>Added task async method for the download cancel operation. #133</PackageReleaseNotes>
+    <PackageReleaseNotes>Added task async method for the download cancel operation. #132</PackageReleaseNotes>
     <SignAssembly>true</SignAssembly>
     <AssemblyOriginatorKeyFile>Downloader.snk</AssemblyOriginatorKeyFile>
     <Copyright>Copyright (C) 2019-2022 Behzad Khosravifar</Copyright>

src/Samples/Downloader.Sample/DownloadList.json

@@ -8,8 +8,8 @@
   //  "Url": "https://c2rsetup.officeapps.live.com/c2r/downloadVS.aspx?sku=community&channel=Release&version=VS2022&source=VSLandingPage&includeRecommended=true&cid=2030:9bf2104738684908988ca7dcd5dafed1"
   //},
   {
-    "FileName": "D:\\TestDownload\\LocalFile100MB_Raw.dat",
-    "Url": "http://localhost:3333/dummyfile/file/size/104857600"
+    "FileName": "D:\\TestDownload\\LocalFile1GB_Raw.dat",
+    "Url": "http://localhost:3333/dummyfile/file/size/1073741824"
   },
   {
     "FolderPath": "D:\\TestDownload",

src/Samples/Downloader.Sample/Downloader.Sample.csproj

@@ -6,7 +6,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageReference Include="ShellProgressBar" Version="5.2.0" />
   </ItemGroup>
 

src/Samples/Downloader.Sample/Helper.cs

@@ -1,9 +1,12 @@
 using System;
+using System.Diagnostics;
 
 namespace Downloader.Sample
 {
     public static class Helper
     {
+        private static Process CurrentProcess = Process.GetCurrentProcess();
+
         public static string CalcMemoryMensurableUnit(this long bytes)
         {
             return CalcMemoryMensurableUnit((double)bytes);
@@ -51,12 +54,14 @@ public static void UpdateTitleInfo(this DownloadProgressChangedEventArgs e, bool
             string bytesReceived = e.ReceivedBytesSize.CalcMemoryMensurableUnit();
             string totalBytesToReceive = e.TotalBytesToReceive.CalcMemoryMensurableUnit();
             string progressPercentage = $"{e.ProgressPercentage:F3}".Replace("/", ".");
+            string usedMemory = CurrentProcess.WorkingSet64.CalcMemoryMensurableUnit();
 
             Console.Title = $"{progressPercentage}%  -  " +
                             $"{speed}/s (avg: {avgSpeed}/s)  -  " +
                             $"{estimateTime} {timeLeftUnit} left   -  " +
                             $"Active Chunks: {e.ActiveChunks}   -   " +
                             $"[{bytesReceived} of {totalBytesToReceive}]   " +
+                            $"[{usedMemory} memory buffer]   " +
                             (isPaused ? " - Paused" : "");
         }
     }

src/Samples/Downloader.Sample/Program.cs

@@ -117,6 +117,7 @@ private static DownloadConfiguration GetDownloadConfiguration()
                 ChunkCount = 8,             // file parts to download, default value is 1
                 MaximumBytesPerSecond = 1024 * 1024 * 10, // download speed limited to 10MB/s, default values is zero or unlimited
                 MaxTryAgainOnFailover = 5,  // the maximum number of times to fail
+                MaximumMemoryBufferBytes = 1024 * 1024 * 50, // release memory buffer after each 50 MB
                 ParallelDownload = true,    // download parts of file as parallel or not. Default value is false
                 ParallelCount = 4,          // number of parallel downloads. The default value is the same as the chunk count
                 Timeout = 3000,             // timeout (millisecond) per stream block reader, default value is 1000