danvics/rdt-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixed simple downloader speed indication.

Browse source
This commit is contained in:
Roger Far 2022-03-19 14:19:50 -06:00
parent f75ac43e49
commit d4cb09e1a3
3 changed files with 51 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
server/RdtClient.Service/Services/Downloaders/SimpleDownloader.cs
View file

@ -18,13 +18,11 @@ namespace RdtClient.Service.Services.Downloaders
private readonly String _uri; private readonly String _uri;
private readonly String _filePath; private readonly String _filePath;
private Int64 Speed { get; set; } private Int64 _bytesTotal;
private Int64 BytesTotal { get; set; } private Int64 _bytesDone;
private Int64 BytesDone { get; set; }
private Boolean _cancelled; private Boolean _cancelled;
private Int64 _bytesLastUpdate;
private DateTime _nextUpdate; private DateTime _nextUpdate;
private readonly ILogger _logger; private readonly ILogger _logger;
@ -62,15 +60,14 @@ namespace RdtClient.Service.Services.Downloaders
{ {
try try
{ {
_bytesLastUpdate = 0;
_nextUpdate = DateTime.UtcNow.AddSeconds(1); _nextUpdate = DateTime.UtcNow.AddSeconds(1);
BytesTotal = await GetContentSize(); _bytesTotal = await GetContentSize();
var timeout = DateTimeOffset.UtcNow.AddHours(1); var timeout = DateTimeOffset.UtcNow.AddHours(1);
var httpClient = new HttpClient(); var httpClient = new HttpClient();
while (timeout > DateTimeOffset.UtcNow && !_cancelled) while (timeout > DateTimeOffset.UtcNow && !_cancelled)
{ {
try try
@ -82,43 +79,46 @@ namespace RdtClient.Service.Services.Downloaders
throw new IOException("No stream"); throw new IOException("No stream");
} }
var speedLimit = Settings.Get.DownloadMaxSpeed * BufferSize * 1024L; var speedLimit = Settings.Get.DownloadMaxSpeed;
if (speedLimit == 0) await using var destinationStream = new ThrottledStream(responseStream, speedLimit * 1000L * 1000L);
{
speedLimit = ThrottledStream.Infinite;
}
await using var destinationStream = new ThrottledStream(responseStream, speedLimit);
await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write); await using var fileStream = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Write);
var readSize = 1; var readSize = 1;
var buffer = new Byte[BufferSize * 8];
while (readSize > 0 && !_cancelled) while (readSize > 0 && !_cancelled)
{ {
using var innerCts = new CancellationTokenSource(1000); // ReSharper disable once ConvertToUsingDeclaration
var buffer = new Byte[BufferSize * 8]; using (var innerCts = new CancellationTokenSource(1000))
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)
{ {
Speed = fileStream.Length - _bytesLastUpdate; readSize = await destinationStream.ReadAsync(buffer.AsMemory(0, buffer.Length), innerCts.Token).ConfigureAwait(false);
_nextUpdate = DateTime.UtcNow.AddSeconds(1); await fileStream.WriteAsync(buffer.AsMemory(0, readSize), innerCts.Token);
_bytesLastUpdate = fileStream.Length;
timeout = DateTimeOffset.UtcNow.AddHours(1); _bytesDone = fileStream.Length;
DownloadProgress?.Invoke(this, new DownloadProgressEventArgs if (DateTime.UtcNow > _nextUpdate)
{ {
Speed = Speed, _nextUpdate = DateTime.UtcNow.AddSeconds(1);
BytesDone = BytesDone,
BytesTotal = BytesTotal 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;
}
}
} }
} }

31
server/RdtClient.Service/Services/ThrottledStream.cs
View file

@ -11,9 +11,11 @@ namespace RdtClient.Service.Services
/// </summary> /// </summary>
public class ThrottledStream : Stream public class ThrottledStream : Stream
{ {
private readonly Bandwidth _bandwidth; public Int64 Speed => (Int64)_bandwidth.AverageSpeed;
private readonly Stream _baseStream;
private Bandwidth _bandwidth;
private Int64 _bandwidthLimit; private Int64 _bandwidthLimit;
private readonly Stream _baseStream;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="T:ThrottledStream" /> class. /// Initializes a new instance of the <see cref="T:ThrottledStream" /> class.
@ -33,11 +35,6 @@ namespace RdtClient.Service.Services
_baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream)); _baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream));
BandwidthLimit = bandwidthLimit; BandwidthLimit = bandwidthLimit;
_bandwidth = new Bandwidth
{
BandwidthLimit = BandwidthLimit
};
} }
public static Int64 Infinite => Int64.MaxValue; public static Int64 Infinite => Int64.MaxValue;
@ -49,7 +46,12 @@ namespace RdtClient.Service.Services
public Int64 BandwidthLimit public Int64 BandwidthLimit
{ {
get => _bandwidthLimit; get => _bandwidthLimit;
set => _bandwidthLimit = value <= 0 ? Infinite : value; set
{
_bandwidthLimit = value <= 0 ? Infinite : value;
_bandwidth ??= new Bandwidth();
_bandwidth.BandwidthLimit = _bandwidthLimit;
}
} }
/// <inheritdoc /> /// <inheritdoc />
@ -70,7 +72,7 @@ namespace RdtClient.Service.Services
get => _baseStream.Position; get => _baseStream.Position;
set => _baseStream.Position = value; set => _baseStream.Position = value;
} }
/// <inheritdoc /> /// <inheritdoc />
public override void Flush() public override void Flush()
{ {
@ -88,7 +90,7 @@ namespace RdtClient.Service.Services
{ {
_baseStream.SetLength(value); _baseStream.SetLength(value);
} }
/// <inheritdoc /> /// <inheritdoc />
public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count) public override Int32 Read(Byte[] buffer, Int32 offset, Int32 count)
{ {
@ -104,7 +106,7 @@ namespace RdtClient.Service.Services
{ {
await Throttle(count).ConfigureAwait(false); 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);
} }
/// <inheritdoc /> /// <inheritdoc />
@ -118,7 +120,7 @@ namespace RdtClient.Service.Services
public override async Task WriteAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken) public override async Task WriteAsync(Byte[] buffer, Int32 offset, Int32 count, CancellationToken cancellationToken)
{ {
await Throttle(count).ConfigureAwait(false); 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) private async Task Throttle(Int32 transmissionVolume)
@ -128,6 +130,7 @@ namespace RdtClient.Service.Services
{ {
// Calculate the time to sleep. // Calculate the time to sleep.
_bandwidth.CalculateSpeed(transmissionVolume); _bandwidth.CalculateSpeed(transmissionVolume);
await Sleep(_bandwidth.PopSpeedRetrieveTime()).ConfigureAwait(false); 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 const Double OneSecond = 1000; // millisecond
private Int64 _count; private Int64 _count;
@ -182,7 +185,7 @@ namespace RdtClient.Service.Services
if (momentSpeed >= BandwidthLimit) if (momentSpeed >= BandwidthLimit)
{ {
var expectedTime = (receivedBytesCount * OneSecond) / BandwidthLimit; var expectedTime = (receivedBytesCount * OneSecond) / BandwidthLimit;
Interlocked.Add(ref _speedRetrieveTime, (Int32)expectedTime - elapsedTime); Interlocked.Add(ref _speedRetrieveTime, (Int32) expectedTime - elapsedTime);
} }
} }

4
server/RdtClient.Service/Services/Torrents.cs
View file

@ -483,14 +483,14 @@ namespace RdtClient.Service.Services
Log($"Retrying Download", download, download.Torrent); 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 downloadClient.Cancel();
await Task.Delay(100); await Task.Delay(100);
} }
while (TorrentRunner.ActiveUnpackClients.TryGetValue(download.DownloadId, out var unpackClient)) while (TorrentRunner.ActiveUnpackClients.TryRemove(download.DownloadId, out var unpackClient))
{ {
unpackClient.Cancel(); unpackClient.Cancel();