diff --git a/src/Hangfire.SqlServer/SqlServerJobQueue.cs b/src/Hangfire.SqlServer/SqlServerJobQueue.cs index 83dda505d..5a51a98f8 100644 --- a/src/Hangfire.SqlServer/SqlServerJobQueue.cs +++ b/src/Hangfire.SqlServer/SqlServerJobQueue.cs @@ -34,7 +34,7 @@ internal class SqlServerJobQueue : IPersistentJobQueue // This is an optimization that helps to overcome the polling delay, when // both client and server reside in the same process. Everything is working // without these events, but it helps to reduce the delays in processing. - internal static readonly ConcurrentDictionary NewItemInQueueEvents = new(); + internal static readonly ConcurrentDictionary, AutoResetEvent> NewItemInQueueEvents = new(); private static readonly Func, SemaphoreSlim> CreateSemaphoreFunc = CreateSemaphore; private static readonly TimeSpan LongPollingThreshold = TimeSpan.FromSeconds(1); @@ -123,7 +123,7 @@ private SqlServerTimeoutJob DequeueUsingSlidingInvisibilityTimeout(string[] queu var queuesString = String.Join("_", queues.OrderBy(static x => x)); var resource = Tuple.Create(_storage, queuesString); - var waitArray = GetWaitArrayForQueueSignals(queues, cancellationToken); + var waitArray = GetWaitArrayForQueueSignals(_storage, queues, cancellationToken); SemaphoreSlim semaphore = null; @@ -210,7 +210,7 @@ private SqlServerTransactionJob DequeueUsingTransaction(string[] queues, Cancell ? _options.QueuePollInterval : TimeSpan.FromSeconds(1); - var waitArray = GetWaitArrayForQueueSignals(queues, cancellationToken); + var waitArray = GetWaitArrayForQueueSignals(_storage, queues, cancellationToken); while (!cancellationToken.IsCancellationRequested) { @@ -271,7 +271,7 @@ private SqlServerTransactionJob DequeueUsingTransaction(string[] queues, Cancell return null; } - private static WaitHandle[] GetWaitArrayForQueueSignals(string[] queues, CancellationToken cancellationToken) + private static WaitHandle[] GetWaitArrayForQueueSignals(SqlServerStorage storage, string[] queues, CancellationToken cancellationToken) { var waitList = new List(capacity: queues.Length + 1) { @@ -280,7 +280,7 @@ private static WaitHandle[] GetWaitArrayForQueueSignals(string[] queues, Cancell foreach (var queue in queues) { - waitList.Add(NewItemInQueueEvents.GetOrAdd(queue, static _ => new AutoResetEvent(initialState: false))); + waitList.Add(NewItemInQueueEvents.GetOrAdd(Tuple.Create(storage, queue), static _ => new AutoResetEvent(initialState: false))); } return waitList.ToArray(); diff --git a/src/Hangfire.SqlServer/SqlServerWriteOnlyTransaction.cs b/src/Hangfire.SqlServer/SqlServerWriteOnlyTransaction.cs index 9e54ee0f2..3d21f2699 100644 --- a/src/Hangfire.SqlServer/SqlServerWriteOnlyTransaction.cs +++ b/src/Hangfire.SqlServer/SqlServerWriteOnlyTransaction.cs @@ -622,7 +622,7 @@ private void TrySignalListeningWorkers() { foreach (var queue in _queuesToSignal) { - if (SqlServerJobQueue.NewItemInQueueEvents.TryGetValue(queue, out var signal)) + if (SqlServerJobQueue.NewItemInQueueEvents.TryGetValue(Tuple.Create(_storage, queue), out var signal)) { signal.Set(); }