From b9889dbbc3c1390182f348fe79f586693c9f0a66 Mon Sep 17 00:00:00 2001 From: Roger Far Date: Sat, 30 Oct 2021 20:00:20 -0600 Subject: [PATCH] Add download speed limiter on the Simple downloader. --- CHANGELOG.md | 4 + client/src/app/navbar/navbar.component.html | 2 +- .../src/app/settings/settings.component.html | 2 +- package.json | 2 +- .../Services/Downloaders/SimpleDownloader.cs | 67 +++--- .../Services/ThrottledStream.cs | 208 ++++++++++++++++++ server/RdtClient.Web/RdtClient.Web.csproj | 2 +- 7 files changed, 253 insertions(+), 34 deletions(-) create mode 100644 server/RdtClient.Service/Services/ThrottledStream.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 7582f96..b103e2d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.9.8] - 2021-10-30 +### Added +- Add speed limit setting on the Simple downloader. + ## [1.9.7] - 2021-10-30 ### Added - Add AllDebrid support. diff --git a/client/src/app/navbar/navbar.component.html b/client/src/app/navbar/navbar.component.html index 26e71b2..4d4f2eb 100644 --- a/client/src/app/navbar/navbar.component.html +++ b/client/src/app/navbar/navbar.component.html @@ -55,7 +55,7 @@ Profile Logout
-
Version 1.9.7
+
Version 1.9.8
diff --git a/client/src/app/settings/settings.component.html b/client/src/app/settings/settings.component.html index 3c70271..2484ccd 100644 --- a/client/src/app/settings/settings.component.html +++ b/client/src/app/settings/settings.component.html @@ -147,7 +147,7 @@

-
+
diff --git a/package.json b/package.json index bb1f232..49a9622 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "rdt-client", - "version": "1.9.7", + "version": "1.9.8", "description": "This is a web interface to manage your torrents on Real-Debrid.", "main": "index.js", "dependencies": { diff --git a/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs b/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs index c2e56b0..d8444f7 100644 --- a/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs +++ b/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs @@ -1,6 +1,7 @@ using System; using System.IO; using System.Net; +using System.Threading; using System.Threading.Tasks; using Serilog; @@ -8,6 +9,8 @@ namespace RdtClient.Service.Services.Downloaders { public class SimpleDownloader : IDownloader { + private const Int32 BufferSize = 8 * 1024; + public event EventHandler DownloadComplete; public event EventHandler DownloadProgress; @@ -81,47 +84,51 @@ namespace RdtClient.Service.Services.Downloaders var request = WebRequest.Create(_uri); using var response = await request.GetResponseAsync(); - await using var stream = response.GetResponseStream(); + await using var responseStream = response.GetResponseStream(); - if (stream == null) + if (responseStream == null) { throw new IOException("No stream"); } - await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write); - var buffer = new Byte[64 * 1024]; - - while (fileStream.Length < response.ContentLength && !_cancelled) + var speedLimit = Settings.Get.DownloadMaxSpeed * BufferSize * 1024L; + + if (speedLimit == 0) { - var read = await stream.ReadAsync(buffer.AsMemory(0, buffer.Length)); + speedLimit = ThrottledStream.Infinite; + } - if (read > 0) + await using var destinationStream = new ThrottledStream(responseStream, speedLimit); + + await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write); + + var readSize = 1; + while (readSize > 0 && !_cancelled) + { + using var innerCts = new CancellationTokenSource(1000); + var buffer = new Byte[BufferSize * 8]; + readSize = await destinationStream.ReadAsync(buffer, 0, buffer.Length, innerCts.Token).ConfigureAwait(false); + + await fileStream.WriteAsync(buffer.AsMemory(0, readSize), innerCts.Token); + + BytesDone = fileStream.Length; + BytesTotal = responseLength; + + if (DateTime.UtcNow > _nextUpdate) { - await fileStream.WriteAsync(buffer.AsMemory(0, read)); + Speed = fileStream.Length - _bytesLastUpdate; - BytesDone = fileStream.Length; - BytesTotal = responseLength; + _nextUpdate = DateTime.UtcNow.AddSeconds(1); + _bytesLastUpdate = fileStream.Length; - if (DateTime.UtcNow > _nextUpdate) + timeout = DateTimeOffset.UtcNow.AddHours(1); + + DownloadProgress?.Invoke(this, new DownloadProgressEventArgs { - Speed = fileStream.Length - _bytesLastUpdate; - - _nextUpdate = DateTime.UtcNow.AddSeconds(1); - _bytesLastUpdate = fileStream.Length; - - timeout = DateTimeOffset.UtcNow.AddHours(1); - - DownloadProgress?.Invoke(this, new DownloadProgressEventArgs - { - Speed = Speed, - BytesDone = BytesDone, - BytesTotal = BytesTotal - }); - } - } - else - { - break; + Speed = Speed, + BytesDone = BytesDone, + BytesTotal = BytesTotal + }); } } diff --git a/server/RdtClient.Service/Services/ThrottledStream.cs b/server/RdtClient.Service/Services/ThrottledStream.cs new file mode 100644 index 0000000..3087ce9 --- /dev/null +++ b/server/RdtClient.Service/Services/ThrottledStream.cs @@ -0,0 +1,208 @@ +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace RdtClient.Service.Services +{ + /// + /// Class for streaming data with throttling support. + /// Taken from Downloader: https://github.com/bezzad/Downloader + /// + public class ThrottledStream : Stream + { + private readonly Bandwidth _bandwidth; + private readonly Stream _baseStream; + private Int64 _bandwidthLimit; + + /// + /// Initializes a new instance of the class. + /// + /// The base stream. + /// The maximum bytes per second that can be transferred through the base stream. + /// Thrown when is a null reference. + /// Thrown when is a negative value. + public ThrottledStream(Stream baseStream, Int64 bandwidthLimit) + { + if (bandwidthLimit < 0) + { + throw new ArgumentOutOfRangeException(nameof(bandwidthLimit), + bandwidthLimit, + "The maximum number of bytes per second can't be negative."); + } + + _baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream)); + BandwidthLimit = bandwidthLimit; + + _bandwidth = new Bandwidth + { + BandwidthLimit = BandwidthLimit + }; + } + + public static Int64 Infinite => Int64.MaxValue; + + /// + /// Bandwidth Limit (in B/s) + /// + /// The maximum bytes per second. + public Int64 BandwidthLimit + { + get => _bandwidthLimit; + set => _bandwidthLimit = value <= 0 ? Infinite : value; + } + + /// + public override Boolean CanRead => _baseStream.CanRead; + + /// + public override Boolean CanSeek => _baseStream.CanSeek; + + /// + public override Boolean CanWrite => _baseStream.CanWrite; + + /// + public override Int64 Length => _baseStream.Length; + + /// + public override Int64 Position + { + get => _baseStream.Position; + set => _baseStream.Position = value; + } + + /// + public override void Flush() + { + _baseStream.Flush(); + } + + /// + public override Int64 Seek(Int64 offset, SeekOrigin origin) + { + return _baseStream.Seek(offset, origin); + } + + /// + public override void SetLength(Int64 value) + { + _baseStream.SetLength(value); + } + + /// + public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count) + { + Throttle(count).Wait(); + + return _baseStream.Read(buffer, offset, count); + } + + public override async Task ReadAsync(Byte[] buffer, + Int32 offset, + Int32 count, + CancellationToken cancellationToken) + { + await Throttle(count).ConfigureAwait(false); + + return await _baseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); + } + + /// + public override void Write(Byte[] buffer, Int32 offset, Int32 count) + { + Throttle(count).Wait(); + _baseStream.Write(buffer, offset, count); + } + + /// + public override async Task WriteAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken) + { + await Throttle(count).ConfigureAwait(false); + await _baseStream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); + } + + private async Task Throttle(Int32 transmissionVolume) + { + // Make sure the buffer isn't empty. + if (BandwidthLimit > 0 && transmissionVolume > 0) + { + // Calculate the time to sleep. + _bandwidth.CalculateSpeed(transmissionVolume); + await Sleep(_bandwidth.PopSpeedRetrieveTime()).ConfigureAwait(false); + } + } + + private async Task Sleep(Int32 time) + { + if (time > 0) + { + await Task.Delay(time).ConfigureAwait(false); + } + } + + /// + public override String ToString() + { + return _baseStream.ToString(); + } + } + + public class Bandwidth + { + private const Double OneSecond = 1000; // millisecond + private Int64 _count; + private Int32 _lastSecondCheckpoint; + private Int64 _lastTransferredBytesCount; + private Int32 _speedRetrieveTime; + + public Bandwidth() + { + BandwidthLimit = Int64.MaxValue; + Reset(); + } + + public Double Speed { get; private set; } + public Double AverageSpeed { get; private set; } + public Int64 BandwidthLimit { get; set; } + + public void CalculateSpeed(Int64 receivedBytesCount) + { + var elapsedTime = (Environment.TickCount - _lastSecondCheckpoint) + 1; + receivedBytesCount = Interlocked.Add(ref _lastTransferredBytesCount, receivedBytesCount); + var momentSpeed = (receivedBytesCount * OneSecond) / elapsedTime; // B/s + + if (OneSecond < elapsedTime) + { + Speed = momentSpeed; + AverageSpeed = ((AverageSpeed * _count) + Speed) / (_count + 1); + _count++; + SecondCheckpoint(); + } + + if (momentSpeed >= BandwidthLimit) + { + var expectedTime = (receivedBytesCount * OneSecond) / BandwidthLimit; + Interlocked.Add(ref _speedRetrieveTime, (Int32)expectedTime - elapsedTime); + } + } + + public Int32 PopSpeedRetrieveTime() + { + return Interlocked.Exchange(ref _speedRetrieveTime, 0); + } + + public void Reset() + { + SecondCheckpoint(); + _count = 0; + Speed = 0; + AverageSpeed = 0; + } + + private void SecondCheckpoint() + { + Interlocked.Exchange(ref _lastSecondCheckpoint, Environment.TickCount); + Interlocked.Exchange(ref _lastTransferredBytesCount, 0); + } + } +} diff --git a/server/RdtClient.Web/RdtClient.Web.csproj b/server/RdtClient.Web/RdtClient.Web.csproj index e1960d3..1432bfa 100644 --- a/server/RdtClient.Web/RdtClient.Web.csproj +++ b/server/RdtClient.Web/RdtClient.Web.csproj @@ -4,7 +4,7 @@ net5.0 Exe 94c24cba-f03f-4453-a671-3640b517c573 - 1.9.7 + 1.9.8