From d4cb09e1a39ffb3ade91d0a3e9f627a5f0e2c936 Mon Sep 17 00:00:00 2001 From: Roger Far Date: Sat, 19 Mar 2022 14:19:50 -0600 Subject: [PATCH] Fixed simple downloader speed indication. --- .../Services/Downloaders/SimpleDownloader.cs | 64 +++++++++---------- .../Services/ThrottledStream.cs | 31 +++++---- server/RdtClient.Service/Services/Torrents.cs | 4 +- 3 files changed, 51 insertions(+), 48 deletions(-) diff --git a/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs b/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs index 637f0ef..9a19b19 100644 --- a/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs +++ b/server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs @@ -18,13 +18,11 @@ namespace RdtClient.Service.Services.Downloaders private readonly String _uri; private readonly String _filePath; - private Int64 Speed { get; set; } - private Int64 BytesTotal { get; set; } - private Int64 BytesDone { get; set; } + private Int64 _bytesTotal; + private Int64 _bytesDone; private Boolean _cancelled; - private Int64 _bytesLastUpdate; private DateTime _nextUpdate; private readonly ILogger _logger; @@ -62,15 +60,14 @@ namespace RdtClient.Service.Services.Downloaders { try { - _bytesLastUpdate = 0; _nextUpdate = DateTime.UtcNow.AddSeconds(1); - BytesTotal = await GetContentSize(); + _bytesTotal = await GetContentSize(); var timeout = DateTimeOffset.UtcNow.AddHours(1); var httpClient = new HttpClient(); - + while (timeout > DateTimeOffset.UtcNow && !_cancelled) { try @@ -82,43 +79,46 @@ namespace RdtClient.Service.Services.Downloaders throw new IOException("No stream"); } - var speedLimit = Settings.Get.DownloadMaxSpeed * BufferSize * 1024L; + var speedLimit = Settings.Get.DownloadMaxSpeed; - if (speedLimit == 0) - { - speedLimit = ThrottledStream.Infinite; - } - - await using var destinationStream = new ThrottledStream(responseStream, speedLimit); + await using var destinationStream = new ThrottledStream(responseStream, speedLimit * 1000L * 1000L); await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write); var readSize = 1; + var buffer = new Byte[BufferSize * 8]; + while (readSize > 0 && !_cancelled) { - using var innerCts = new CancellationTokenSource(1000); - var buffer = new Byte[BufferSize * 8]; - readSize = await destinationStream.ReadAsync(buffer.AsMemory(0, buffer.Length), innerCts.Token).ConfigureAwait(false); - - await fileStream.WriteAsync(buffer.AsMemory(0, readSize), innerCts.Token); - - BytesDone = fileStream.Length; - - if (DateTime.UtcNow > _nextUpdate) + // ReSharper disable once ConvertToUsingDeclaration + using (var innerCts = new CancellationTokenSource(1000)) { - Speed = fileStream.Length - _bytesLastUpdate; + readSize = await destinationStream.ReadAsync(buffer.AsMemory(0, buffer.Length), innerCts.Token).ConfigureAwait(false); - _nextUpdate = DateTime.UtcNow.AddSeconds(1); - _bytesLastUpdate = fileStream.Length; + await fileStream.WriteAsync(buffer.AsMemory(0, readSize), innerCts.Token); - timeout = DateTimeOffset.UtcNow.AddHours(1); + _bytesDone = fileStream.Length; - DownloadProgress?.Invoke(this, new DownloadProgressEventArgs + if (DateTime.UtcNow > _nextUpdate) { - Speed = Speed, - BytesDone = BytesDone, - BytesTotal = BytesTotal - }); + _nextUpdate = DateTime.UtcNow.AddSeconds(1); + + timeout = DateTimeOffset.UtcNow.AddHours(1); + + DownloadProgress?.Invoke(this, + new DownloadProgressEventArgs + { + Speed = destinationStream.Speed, + BytesDone = _bytesDone, + BytesTotal = _bytesTotal + }); + + if (Settings.Get.DownloadMaxSpeed != speedLimit) + { + speedLimit = Settings.Get.DownloadMaxSpeed; + destinationStream.BandwidthLimit = speedLimit * 1000L * 1000L; + } + } } } diff --git a/server/RdtClient.Service/Services/ThrottledStream.cs b/server/RdtClient.Service/Services/ThrottledStream.cs index 0e72c05..0853645 100644 --- a/server/RdtClient.Service/Services/ThrottledStream.cs +++ b/server/RdtClient.Service/Services/ThrottledStream.cs @@ -11,9 +11,11 @@ namespace RdtClient.Service.Services /// public class ThrottledStream : Stream { - private readonly Bandwidth _bandwidth; - private readonly Stream _baseStream; + public Int64 Speed => (Int64)_bandwidth.AverageSpeed; + + private Bandwidth _bandwidth; private Int64 _bandwidthLimit; + private readonly Stream _baseStream; /// /// Initializes a new instance of the class. @@ -33,11 +35,6 @@ namespace RdtClient.Service.Services _baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream)); BandwidthLimit = bandwidthLimit; - - _bandwidth = new Bandwidth - { - BandwidthLimit = BandwidthLimit - }; } public static Int64 Infinite => Int64.MaxValue; @@ -49,7 +46,12 @@ namespace RdtClient.Service.Services public Int64 BandwidthLimit { get => _bandwidthLimit; - set => _bandwidthLimit = value <= 0 ? Infinite : value; + set + { + _bandwidthLimit = value <= 0 ? Infinite : value; + _bandwidth ??= new Bandwidth(); + _bandwidth.BandwidthLimit = _bandwidthLimit; + } } /// @@ -70,7 +72,7 @@ namespace RdtClient.Service.Services get => _baseStream.Position; set => _baseStream.Position = value; } - + /// public override void Flush() { @@ -88,7 +90,7 @@ namespace RdtClient.Service.Services { _baseStream.SetLength(value); } - + /// public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count) { @@ -104,7 +106,7 @@ namespace RdtClient.Service.Services { await Throttle(count).ConfigureAwait(false); - return await _baseStream.ReadAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false); + return await _baseStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } /// @@ -118,7 +120,7 @@ namespace RdtClient.Service.Services public override async Task WriteAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken) { await Throttle(count).ConfigureAwait(false); - await _baseStream.WriteAsync(buffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false); + await _baseStream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); } private async Task Throttle(Int32 transmissionVolume) @@ -128,6 +130,7 @@ namespace RdtClient.Service.Services { // Calculate the time to sleep. _bandwidth.CalculateSpeed(transmissionVolume); + await Sleep(_bandwidth.PopSpeedRetrieveTime()).ConfigureAwait(false); } } @@ -147,7 +150,7 @@ namespace RdtClient.Service.Services } } - public class Bandwidth + internal class Bandwidth { private const Double OneSecond = 1000; // millisecond private Int64 _count; @@ -182,7 +185,7 @@ namespace RdtClient.Service.Services if (momentSpeed >= BandwidthLimit) { var expectedTime = (receivedBytesCount * OneSecond) / BandwidthLimit; - Interlocked.Add(ref _speedRetrieveTime, (Int32)expectedTime - elapsedTime); + Interlocked.Add(ref _speedRetrieveTime, (Int32) expectedTime - elapsedTime); } } diff --git a/server/RdtClient.Service/Services/Torrents.cs b/server/RdtClient.Service/Services/Torrents.cs index f584a3f..33c97df 100644 --- a/server/RdtClient.Service/Services/Torrents.cs +++ b/server/RdtClient.Service/Services/Torrents.cs @@ -483,14 +483,14 @@ namespace RdtClient.Service.Services Log($"Retrying Download", download, download.Torrent); - while (TorrentRunner.ActiveDownloadClients.TryGetValue(download.DownloadId, out var downloadClient)) + while (TorrentRunner.ActiveDownloadClients.TryRemove(download.DownloadId, out var downloadClient)) { await downloadClient.Cancel(); await Task.Delay(100); } - while (TorrentRunner.ActiveUnpackClients.TryGetValue(download.DownloadId, out var unpackClient)) + while (TorrentRunner.ActiveUnpackClients.TryRemove(download.DownloadId, out var unpackClient)) { unpackClient.Cancel();