notfire/Iceshrimp.NET
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[backend/queue] Fix QueueService race condition (properly this time)

Browse source 
While the previous fix was likely enough, there was still a razor-thin theoretical race condition remaining. This commit fixes said race condition, and simplifies some if statements across the file.
This commit is contained in:
Laura Hausmann 2024-07-25 00:23:09 +02:00
parent 4690d5c1fb
commit a845405c45
No known key found for this signature in database
GPG key ID: D044E84C5BE01605
2 changed files with 14 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Iceshrimp.Backend/Core/Helpers/SemaphorePlus.cs
View file

@ -4,4 +4,10 @@ public class SemaphorePlus(int maxCount) : SemaphoreSlim(maxCount, maxCount)
{ {
private readonly int _maxCount = maxCount; private readonly int _maxCount = maxCount;
public int ActiveCount => _maxCount - CurrentCount; public int ActiveCount => _maxCount - CurrentCount;
public async Task WaitAndReleaseAsync(CancellationToken token)
{
await WaitAsync(token);
Release();
}
} }

17
Iceshrimp.Backend/Core/Services/QueueService.cs
View file

@ -188,24 +188,23 @@ public class PostgresJobQueue<T>(
await using var db = GetDbContext(scope); await using var db = GetDbContext(scope);
var queuedCount = await db.GetJobQueuedCount(name, token); var queuedCount = await db.GetJobQueuedCount(name, token);
var actualParallelism = Math.Min(parallelism - _semaphore.ActiveCount, queuedCount); var actualParallelism = Math.Min(_semaphore.CurrentCount, queuedCount);
if (actualParallelism <= 0) if (actualParallelism == 0)
{ {
await _queuedChannel.WaitAsync(token).SafeWaitAsync(queueToken); if (_semaphore.CurrentCount == 0 && queuedCount > 0)
await _semaphore.WaitAndReleaseAsync(token).SafeWaitAsync(queueToken);
else
await _queuedChannel.WaitAsync(token).SafeWaitAsync(queueToken);
continue; continue;
} }
// ReSharper disable MethodSupportsCancellation // ReSharper disable MethodSupportsCancellation
var queuedChannelCts = new CancellationTokenSource(); var queuedChannelCts = new CancellationTokenSource();
if (_semaphore.ActiveCount + queuedCount < parallelism) if (_semaphore.CurrentCount - queuedCount > 0)
{ {
_ = _queuedChannel.WaitWithoutResetAsync() _ = _queuedChannel.WaitWithoutResetAsync()
.ContinueWith(_ => .ContinueWith(_ => queuedChannelCts.Cancel())
{
if (_semaphore.ActiveCount < parallelism)
queuedChannelCts.Cancel();
})
.SafeWaitAsync(queueToken); .SafeWaitAsync(queueToken);
} }
// ReSharper restore MethodSupportsCancellation // ReSharper restore MethodSupportsCancellation