Compare commits

...

2 Commits

5 changed files with 43 additions and 21 deletions

View File

@ -15,6 +15,7 @@ sealed class Arguments {
public string? DatabaseFile { get; } public string? DatabaseFile { get; }
public ushort? ServerPort { get; } public ushort? ServerPort { get; }
public string? ServerToken { get; } public string? ServerToken { get; }
public byte? ConcurrentDownloads { get; }
public Arguments(IReadOnlyList<string> args) { public Arguments(IReadOnlyList<string> args) {
for (int i = FirstArgument; i < args.Count; i++) { for (int i = FirstArgument; i < args.Count; i++) {
@ -50,11 +51,11 @@ sealed class Arguments {
continue; continue;
case "-port": { case "-port": {
if (ushort.TryParse(value, out var port)) { if (!ushort.TryParse(value, out var port)) {
ServerPort = port; Log.Warn("Invalid port number: " + value);
} }
else { else {
Log.Warn("Invalid port number: " + value); ServerPort = port;
} }
continue; continue;
@ -63,6 +64,20 @@ sealed class Arguments {
case "-token": case "-token":
ServerToken = value; ServerToken = value;
continue; continue;
case "-concurrentdownloads":
if (!ulong.TryParse(value, out var concurrentDownloads) || concurrentDownloads == 0) {
Log.Warn("Invalid concurrent downloads count: " + value);
}
else if (concurrentDownloads > 10) {
Log.Warn("Limiting concurrent downloads to 10");
ConcurrentDownloads = 10;
}
else {
ConcurrentDownloads = (byte) concurrentDownloads;
}
continue;
default: default:
Log.Warn("Unknown command line argument: " + key); Log.Warn("Unknown command line argument: " + key);

View File

@ -30,6 +30,7 @@ sealed partial class MainWindowModel : ObservableObject, IAsyncDisposable {
private MainContentScreenModel? mainContentScreenModel; private MainContentScreenModel? mainContentScreenModel;
private readonly Window window; private readonly Window window;
private readonly int? concurrentDownloads;
private State? state; private State? state;
@ -73,6 +74,8 @@ sealed partial class MainWindowModel : ObservableObject, IAsyncDisposable {
if (args.ServerToken != null) { if (args.ServerToken != null) {
ServerConfiguration.Token = args.ServerToken; ServerConfiguration.Token = args.ServerToken;
} }
concurrentDownloads = args.ConcurrentDownloads;
} }
private async void OnDatabaseSelected(object? sender, IDatabaseFile db) { private async void OnDatabaseSelected(object? sender, IDatabaseFile db) {
@ -80,7 +83,7 @@ sealed partial class MainWindowModel : ObservableObject, IAsyncDisposable {
await DisposeState(); await DisposeState();
state = new State(db); state = new State(db, concurrentDownloads);
try { try {
await state.Server.Start(ServerConfiguration.Port, ServerConfiguration.Token); await state.Server.Start(ServerConfiguration.Port, ServerConfiguration.Token);

View File

@ -11,16 +11,18 @@ public sealed class Downloader {
public bool IsDownloading => current != null; public bool IsDownloading => current != null;
private readonly IDatabaseFile db; private readonly IDatabaseFile db;
private readonly int? concurrentDownloads;
private readonly SemaphoreSlim semaphore = new (1, 1); private readonly SemaphoreSlim semaphore = new (1, 1);
internal Downloader(IDatabaseFile db) { internal Downloader(IDatabaseFile db, int? concurrentDownloads) {
this.db = db; this.db = db;
this.concurrentDownloads = concurrentDownloads;
} }
public async Task<IObservable<DownloadItem>> Start(DownloadItemFilter filter) { public async Task<IObservable<DownloadItem>> Start(DownloadItemFilter filter) {
await semaphore.WaitAsync(); await semaphore.WaitAsync();
try { try {
current ??= new DownloaderTask(db, filter); current ??= new DownloaderTask(db, filter, concurrentDownloads);
return current.FinishedItems; return current.FinishedItems;
} finally { } finally {
semaphore.Release(); semaphore.Release();

View File

@ -15,10 +15,14 @@ namespace DHT.Server.Download;
sealed class DownloaderTask : IAsyncDisposable { sealed class DownloaderTask : IAsyncDisposable {
private static readonly Log Log = Log.ForType<DownloaderTask>(); private static readonly Log Log = Log.ForType<DownloaderTask>();
private const int DownloadTasks = 4; private const int DefaultConcurrentDownloads = 4;
private const int QueueSize = 25; private const int QueueSize = 25;
private const string UserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"; private const string UserAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36";
private static int GetDownloadTaskCount(int? concurrentDownloads) {
return Math.Max(1, concurrentDownloads ?? DefaultConcurrentDownloads);
}
private readonly Channel<DownloadItem> downloadQueue = Channel.CreateBounded<DownloadItem>(new BoundedChannelOptions(QueueSize) { private readonly Channel<DownloadItem> downloadQueue = Channel.CreateBounded<DownloadItem>(new BoundedChannelOptions(QueueSize) {
SingleReader = false, SingleReader = false,
SingleWriter = true, SingleWriter = true,
@ -38,12 +42,12 @@ sealed class DownloaderTask : IAsyncDisposable {
public IObservable<DownloadItem> FinishedItems => finishedItemPublisher; public IObservable<DownloadItem> FinishedItems => finishedItemPublisher;
internal DownloaderTask(IDatabaseFile db, DownloadItemFilter filter) { internal DownloaderTask(IDatabaseFile db, DownloadItemFilter filter, int? concurrentDownloads) {
this.db = db; this.db = db;
this.filter = filter; this.filter = filter;
this.cancellationToken = cancellationTokenSource.Token; this.cancellationToken = cancellationTokenSource.Token;
this.queueWriterTask = Task.Run(RunQueueWriterTask); this.queueWriterTask = Task.Run(RunQueueWriterTask);
this.downloadTasks = Enumerable.Range(1, DownloadTasks).Select(taskIndex => Task.Run(() => RunDownloadTask(taskIndex))).ToArray(); this.downloadTasks = Enumerable.Range(1, GetDownloadTaskCount(concurrentDownloads)).Select(taskIndex => Task.Run(() => RunDownloadTask(taskIndex))).ToArray();
} }
private async Task RunQueueWriterTask() { private async Task RunQueueWriterTask() {
@ -74,8 +78,12 @@ sealed class DownloaderTask : IAsyncDisposable {
try { try {
var downloadedBytes = await client.GetByteArrayAsync(item.DownloadUrl, cancellationToken); var downloadedBytes = await client.GetByteArrayAsync(item.DownloadUrl, cancellationToken);
await db.Downloads.AddDownload(item.ToSuccess(downloadedBytes)); await db.Downloads.AddDownload(item.ToSuccess(downloadedBytes));
} catch (OperationCanceledException) { } catch (OperationCanceledException e) when (e.CancellationToken == cancellationToken) {
// Ignore. // Ignore.
} catch (TaskCanceledException e) {
// HttpClient request timed out.
await db.Downloads.AddDownload(item.ToFailure());
log.Error(e.Message);
} catch (HttpRequestException e) { } catch (HttpRequestException e) {
await db.Downloads.AddDownload(item.ToFailure(e.StatusCode)); await db.Downloads.AddDownload(item.ToFailure(e.StatusCode));
log.Error(e); log.Error(e);

View File

@ -6,18 +6,12 @@ using DHT.Server.Service;
namespace DHT.Server; namespace DHT.Server;
public sealed class State : IAsyncDisposable { public sealed class State(IDatabaseFile db, int? concurrentDownloads) : IAsyncDisposable {
public static State Dummy { get; } = new (DummyDatabaseFile.Instance); public static State Dummy { get; } = new (DummyDatabaseFile.Instance, null);
public IDatabaseFile Db { get; } public IDatabaseFile Db { get; } = db;
public Downloader Downloader { get; } public Downloader Downloader { get; } = new (db, concurrentDownloads);
public ServerManager Server { get; } public ServerManager Server { get; } = new (db);
public State(IDatabaseFile db) {
Db = db;
Downloader = new Downloader(db);
Server = new ServerManager(db);
}
public async ValueTask DisposeAsync() { public async ValueTask DisposeAsync() {
await Downloader.Stop(); await Downloader.Stop();