some file server fixes I guess
This commit is contained in:
@@ -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";
|
||||
|
||||
@@ -24,9 +24,10 @@ public class RequestController : ControllerBase
|
||||
[Route(MareFiles.Request_Cancel)]
|
||||
public async Task<IActionResult> 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<IActionResult> PreRequestFilesAsync([FromBody] IEnumerable<string> 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<IActionResult> CheckQueueAsync(Guid requestId, [FromBody] IEnumerable<string> 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -22,7 +22,6 @@ public class RequestQueueService : IHostedService
|
||||
private readonly ConcurrentQueue<UserRequest> _priorityQueue = new();
|
||||
private readonly int _queueExpirationSeconds;
|
||||
private readonly SemaphoreSlim _queueProcessingSemaphore = new(1);
|
||||
private readonly ConcurrentDictionary<Guid, string> _queueRemoval = new();
|
||||
private readonly SemaphoreSlim _queueSemaphore = new(1);
|
||||
private readonly UserQueueEntry[] _userQueueRequests;
|
||||
private readonly ConcurrentDictionary<string, PriorityEntry> _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));
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
namespace MareSynchronosStaticFilesServer.Utils;
|
||||
|
||||
public record UserRequest(Guid RequestId, string User, List<string> FileIds);
|
||||
public record UserRequest(Guid RequestId, string User, List<string> FileIds)
|
||||
{
|
||||
public bool IsCancelled { get; set; } = false;
|
||||
}
|
||||
Reference in New Issue
Block a user