From 2c7ff6f73a61a4777ae15ac1e744b0a9815583b8 Mon Sep 17 00:00:00 2001 From: rootdarkarchon Date: Sat, 13 Jan 2024 12:08:49 +0100 Subject: [PATCH] some file server fixes I guess --- .../Metrics/MetricsAPI.cs | 3 + .../Controllers/RequestController.cs | 12 ++- .../Services/RequestQueueService.cs | 81 ++++++++----------- .../Startup.cs | 3 + .../Utils/UserRequest.cs | 5 +- 5 files changed, 53 insertions(+), 51 deletions(-) diff --git a/MareSynchronosServer/MareSynchronosShared/Metrics/MetricsAPI.cs b/MareSynchronosServer/MareSynchronosShared/Metrics/MetricsAPI.cs index 87838d0..180bde9 100644 --- a/MareSynchronosServer/MareSynchronosShared/Metrics/MetricsAPI.cs +++ b/MareSynchronosServer/MareSynchronosShared/Metrics/MetricsAPI.cs @@ -31,6 +31,9 @@ public class MetricsAPI public const string GaugeQueueActive = "mare_download_queue_active"; public const string GaugeQueueInactive = "mare_download_queue_inactive"; public const string GaugeDownloadQueue = "mare_download_queue"; + public const string GaugeDownloadQueueCancelled = "mare_download_queue_cancelled"; + public const string GaugeDownloadPriorityQueue = "mare_download_priority_queue"; + public const string GaugeDownloadPriorityQueueCancelled = "mare_download_priority_queue_cancelled"; public const string CounterFileRequests = "mare_files_requests"; public const string CounterFileRequestSize = "mare_files_request_size"; public const string CounterUserPairCacheHit = "mare_pairscache_hit"; diff --git a/MareSynchronosServer/MareSynchronosStaticFilesServer/Controllers/RequestController.cs b/MareSynchronosServer/MareSynchronosStaticFilesServer/Controllers/RequestController.cs index e22e08c..b4368bd 100644 --- a/MareSynchronosServer/MareSynchronosStaticFilesServer/Controllers/RequestController.cs +++ b/MareSynchronosServer/MareSynchronosStaticFilesServer/Controllers/RequestController.cs @@ -24,9 +24,10 @@ public class RequestController : ControllerBase [Route(MareFiles.Request_Cancel)] public async Task CancelQueueRequest(Guid requestId) { + await _parallelRequestSemaphore.WaitAsync(HttpContext.RequestAborted); + try { - await _parallelRequestSemaphore.WaitAsync(HttpContext.RequestAborted); _requestQueue.RemoveFromQueue(requestId, MareUser); return Ok(); } @@ -41,9 +42,10 @@ public class RequestController : ControllerBase [Route(MareFiles.Request_Enqueue)] public async Task PreRequestFilesAsync([FromBody] IEnumerable files) { + await _parallelRequestSemaphore.WaitAsync(HttpContext.RequestAborted); + try { - await _parallelRequestSemaphore.WaitAsync(HttpContext.RequestAborted); foreach (var file in files) { _logger.LogDebug("Prerequested file: " + file); @@ -66,6 +68,8 @@ public class RequestController : ControllerBase [Route(MareFiles.Request_Check)] public async Task CheckQueueAsync(Guid requestId, [FromBody] IEnumerable files) { + await _parallelRequestSemaphore.WaitAsync(HttpContext.RequestAborted); + try { if (!await _requestQueue.StillEnqueued(requestId, MareUser, _mareDbContext)) @@ -73,5 +77,9 @@ public class RequestController : ControllerBase return Ok(); } catch (OperationCanceledException) { return BadRequest(); } + finally + { + _parallelRequestSemaphore.Release(); + } } } \ No newline at end of file diff --git a/MareSynchronosServer/MareSynchronosStaticFilesServer/Services/RequestQueueService.cs b/MareSynchronosServer/MareSynchronosStaticFilesServer/Services/RequestQueueService.cs index e63f6d4..b6c858f 100644 --- a/MareSynchronosServer/MareSynchronosStaticFilesServer/Services/RequestQueueService.cs +++ b/MareSynchronosServer/MareSynchronosStaticFilesServer/Services/RequestQueueService.cs @@ -22,7 +22,6 @@ public class RequestQueueService : IHostedService private readonly ConcurrentQueue _priorityQueue = new(); private readonly int _queueExpirationSeconds; private readonly SemaphoreSlim _queueProcessingSemaphore = new(1); - private readonly ConcurrentDictionary _queueRemoval = new(); private readonly SemaphoreSlim _queueSemaphore = new(1); private readonly UserQueueEntry[] _userQueueRequests; private readonly ConcurrentDictionary _priorityCache = new(StringComparer.Ordinal); @@ -118,7 +117,9 @@ public class RequestQueueService : IHostedService public void RemoveFromQueue(Guid requestId, string user) { - if (!_queue.Any(f => f.RequestId == requestId && string.Equals(f.User, user, StringComparison.Ordinal))) + var existingRequest = _priorityQueue.FirstOrDefault(f => f.RequestId == requestId && string.Equals(f.User, user, StringComparison.Ordinal)) + ?? _queue.FirstOrDefault(f => f.RequestId == requestId && string.Equals(f.User, user, StringComparison.Ordinal)); + if (existingRequest == null) { var activeSlot = _userQueueRequests.FirstOrDefault(r => r != null && string.Equals(r.UserRequest.User, user, StringComparison.Ordinal) && r.UserRequest.RequestId == requestId); if (activeSlot != null) @@ -129,10 +130,11 @@ public class RequestQueueService : IHostedService _userQueueRequests[idx] = null; } } - - return; } - _queueRemoval[requestId] = user; + else + { + existingRequest.IsCancelled = true; + } } public Task StartAsync(CancellationToken cancellationToken) @@ -175,7 +177,7 @@ public class RequestQueueService : IHostedService try { - if (_queue.Count > _queueLimitForReset) + if (_queue.Count(c => !c.IsCancelled) > _queueLimitForReset) { _queue.Clear(); return; @@ -189,56 +191,36 @@ public class RequestQueueService : IHostedService { try { - if (_userQueueRequests[i] != null && ((!_userQueueRequests[i].IsActive && _userQueueRequests[i].ExpirationDate < DateTime.UtcNow))) + if (_userQueueRequests[i] != null + && (((!_userQueueRequests[i].IsActive && _userQueueRequests[i].ExpirationDate < DateTime.UtcNow)) + || (_userQueueRequests[i].IsActive && _userQueueRequests[i].ActivationDate < DateTime.UtcNow.Subtract(TimeSpan.FromSeconds(_queueReleaseSeconds)))) + ) { - _logger.LogDebug("Expiring inactive request {guid} slot {slot}", _userQueueRequests[i].UserRequest.RequestId, i); + _logger.LogDebug("Expiring request {guid} slot {slot}", _userQueueRequests[i].UserRequest.RequestId, i); _userQueueRequests[i] = null; } - if (_userQueueRequests[i] != null && (_userQueueRequests[i].IsActive && _userQueueRequests[i].ActivationDate < DateTime.UtcNow.Subtract(TimeSpan.FromSeconds(_queueReleaseSeconds)))) - { - _logger.LogDebug("Expiring active request {guid} slot {slot}", _userQueueRequests[i].UserRequest.RequestId, i); - _userQueueRequests[i] = null; - } + if ((!_queue.Any() && !_priorityQueue.Any()) || _userQueueRequests[i] != null) return; - if (!_queue.Any() && !_priorityQueue.Any()) return; - - if (_userQueueRequests[i] == null) + while (true) { - bool enqueued = false; - while (!enqueued) + if (_priorityQueue.TryDequeue(out var prioRequest)) { - if (_priorityQueue.TryDequeue(out var prioRequest)) - { - if (_queueRemoval.TryGetValue(prioRequest.RequestId, out string user) && string.Equals(user, prioRequest.User, StringComparison.Ordinal)) - { - _logger.LogDebug("Request cancelled: {requestId} by {user}", prioRequest.RequestId, user); - _queueRemoval.Remove(prioRequest.RequestId, out _); - continue; - } + if (prioRequest.IsCancelled) continue; - await DequeueIntoSlotAsync(prioRequest, i).ConfigureAwait(false); - enqueued = true; - break; - } - - if (_queue.TryDequeue(out var request)) - { - if (_queueRemoval.TryGetValue(request.RequestId, out string user) && string.Equals(user, request.User, StringComparison.Ordinal)) - { - _logger.LogDebug("Request cancelled: {requestId} by {user}", request.RequestId, user); - _queueRemoval.Remove(request.RequestId, out _); - continue; - } - - await DequeueIntoSlotAsync(request, i).ConfigureAwait(false); - enqueued = true; - } - else - { - enqueued = true; - } + await DequeueIntoSlotAsync(prioRequest, i).ConfigureAwait(false); + break; } + + if (_queue.TryDequeue(out var request)) + { + if (request.IsCancelled) continue; + + await DequeueIntoSlotAsync(request, i).ConfigureAwait(false); + break; + } + + break; } } catch (Exception ex) @@ -259,6 +241,9 @@ public class RequestQueueService : IHostedService _metrics.SetGaugeTo(MetricsAPI.GaugeQueueFree, _userQueueRequests.Count(c => c == null)); _metrics.SetGaugeTo(MetricsAPI.GaugeQueueActive, _userQueueRequests.Count(c => c != null && c.IsActive)); _metrics.SetGaugeTo(MetricsAPI.GaugeQueueInactive, _userQueueRequests.Count(c => c != null && !c.IsActive)); - _metrics.SetGaugeTo(MetricsAPI.GaugeDownloadQueue, _queue.Count); + _metrics.SetGaugeTo(MetricsAPI.GaugeDownloadQueue, _queue.Count(q => !q.IsCancelled)); + _metrics.SetGaugeTo(MetricsAPI.GaugeDownloadQueueCancelled, _queue.Count(q => q.IsCancelled)); + _metrics.SetGaugeTo(MetricsAPI.GaugeDownloadPriorityQueue, _priorityQueue.Count(q => !q.IsCancelled)); + _metrics.SetGaugeTo(MetricsAPI.GaugeDownloadPriorityQueueCancelled, _priorityQueue.Count(q => q.IsCancelled)); } } \ No newline at end of file diff --git a/MareSynchronosServer/MareSynchronosStaticFilesServer/Startup.cs b/MareSynchronosServer/MareSynchronosStaticFilesServer/Startup.cs index 9afb427..5c7a503 100644 --- a/MareSynchronosServer/MareSynchronosStaticFilesServer/Startup.cs +++ b/MareSynchronosServer/MareSynchronosStaticFilesServer/Startup.cs @@ -65,6 +65,9 @@ public class Startup MetricsAPI.GaugeFilesUniquePastHourSize, MetricsAPI.GaugeCurrentDownloads, MetricsAPI.GaugeDownloadQueue, + MetricsAPI.GaugeDownloadQueueCancelled, + MetricsAPI.GaugeDownloadPriorityQueue, + MetricsAPI.GaugeDownloadPriorityQueueCancelled, MetricsAPI.GaugeQueueFree, MetricsAPI.GaugeQueueInactive, MetricsAPI.GaugeQueueActive, diff --git a/MareSynchronosServer/MareSynchronosStaticFilesServer/Utils/UserRequest.cs b/MareSynchronosServer/MareSynchronosStaticFilesServer/Utils/UserRequest.cs index 4640c09..f64def1 100644 --- a/MareSynchronosServer/MareSynchronosStaticFilesServer/Utils/UserRequest.cs +++ b/MareSynchronosServer/MareSynchronosStaticFilesServer/Utils/UserRequest.cs @@ -1,3 +1,6 @@ namespace MareSynchronosStaticFilesServer.Utils; -public record UserRequest(Guid RequestId, string User, List FileIds); +public record UserRequest(Guid RequestId, string User, List FileIds) +{ + public bool IsCancelled { get; set; } = false; +} \ No newline at end of file