Merge pull request #320 from Tinyakov/feature/jobs_fetching_signals_refact

Fixing Enqueued Job Trigger for Multiple Queues

Author: azygis

src/Hangfire.PostgreSql/PostgreSqlJobQueue.cs

@@ -37,7 +37,7 @@ public class PostgreSqlJobQueue : IPersistentJobQueue
   {
     private const string JobNotificationChannel = "new_job";
 
-    internal static readonly AutoResetEvent _newItemInQueueEvent = new(true);
+    internal static readonly AutoResetEventRegistry _queueEventRegistry = new();
     private readonly PostgreSqlStorage _storage;
 
     public PostgreSqlJobQueue(PostgreSqlStorage storage)
@@ -147,6 +147,12 @@ LIMIT 1
         RETURNING ""id"" AS ""Id"", ""jobid"" AS ""JobId"", ""queue"" AS ""Queue"", ""fetchedat"" AS ""FetchedAt"";
       ";
 
+      WaitHandle[] nextFetchIterationWaitHandles = new[] {
+        cancellationToken.WaitHandle,
+        SignalDequeue,
+        JobQueueNotification,
+      }.Concat(_queueEventRegistry.GetWaitHandles(queues)).ToArray();
+
       do
       {
         cancellationToken.ThrowIfCancellationRequested();
@@ -181,15 +187,7 @@ LIMIT 1
 
         if (fetchedJob == null)
         {
-          WaitHandle.WaitAny(new[] {
-              cancellationToken.WaitHandle,
-              _newItemInQueueEvent,
-              SignalDequeue,
-              JobQueueNotification,
-            },
-            _storage.Options.QueuePollInterval);
-
-          cancellationToken.ThrowIfCancellationRequested();
+          WaitHandle.WaitAny(nextFetchIterationWaitHandles, _storage.Options.QueuePollInterval);
         }
       }
       while (fetchedJob == null);

src/Hangfire.PostgreSql/PostgreSqlStorage.cs

@@ -320,22 +320,12 @@ internal T UseTransaction<T>(DbConnection dedicatedConnection, Func<DbConnection
 
         T result = func(connection, null);
 
-        // TransactionCompleted event is required here, because if this TransactionScope is enlisted within an ambient TransactionScope, the ambient TransactionScope controls when the TransactionScope completes.
-        Transaction.Current.TransactionCompleted += Current_TransactionCompleted;
         transaction.Complete();
 
         return result;
       });
     }
 
-    private static void Current_TransactionCompleted(object sender, TransactionEventArgs e)
-    {
-      if (e.Transaction.TransactionInformation.Status == TransactionStatus.Committed)
-      {
-        PostgreSqlJobQueue._newItemInQueueEvent.Set();
-      }
-    }
-
     internal TransactionScope CreateTransactionScope(IsolationLevel? isolationLevel, TimeSpan? timeout = null)
     {
       return TransactionHelpers.CreateTransactionScope(isolationLevel, Options.EnableTransactionScopeEnlistment, timeout);

src/Hangfire.PostgreSql/PostgreSqlWriteOnlyTransaction.cs

@@ -37,6 +37,7 @@ public class PostgreSqlWriteOnlyTransaction : JobStorageTransaction
   {
     private readonly Queue<Action<IDbConnection>> _commandQueue = new();
     private readonly Func<DbConnection> _dedicatedConnectionFunc;
+    private readonly List<string> _queuesWithAddedJobs = new();
 
     private readonly PostgreSqlStorage _storage;
 
@@ -56,6 +57,10 @@ public override void Commit()
           command(connection);
         }
       }, CreateTransactionScope);
+      
+      // Triggers signals for all queues to which jobs have been added in this transaction
+      _queuesWithAddedJobs.ForEach(PostgreSqlJobQueue._queueEventRegistry.Set);
+      _queuesWithAddedJobs.Clear();
     }
 
     private TransactionScope CreateTransactionScope()
@@ -134,6 +139,8 @@ public override void AddToQueue(string queue, string jobId)
       IPersistentJobQueue persistentQueue = provider.GetJobQueue();
 
       QueueCommand(con => persistentQueue.Enqueue(con, queue, jobId));
+      
+      _queuesWithAddedJobs.Add(queue);
     }
 
     public override void IncrementCounter(string key)

src/Hangfire.PostgreSql/Utils/AutoResetEventRegistry.cs

@@ -0,0 +1,40 @@
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace Hangfire.PostgreSql.Utils
+{
+  /// <summary>
+  /// Represents a registry for managing AutoResetEvent instances using event keys.
+  /// </summary>
+  public class AutoResetEventRegistry
+  {
+    private readonly ConcurrentDictionary<string, AutoResetEvent> _events = new();
+
+    /// <summary>
+    /// Retrieves the wait handles associated with the specified event keys.
+    /// </summary>
+    /// <param name="eventKeys">The event keys.</param>
+    /// <returns>An enumerable of wait handles.</returns>
+    public IEnumerable<WaitHandle> GetWaitHandles(IEnumerable<string> eventKeys)
+    {
+      foreach (string eventKey in eventKeys)
+      {
+          AutoResetEvent newHandle = _events.GetOrAdd(eventKey, _ => new AutoResetEvent(false));
+          yield return newHandle;
+      }
+    }
+
+    /// <summary>
+    /// Sets the specified event.
+    /// </summary>
+    /// <param name="eventKey">The event key.</param>
+    public void Set(string eventKey)
+    {
+      if (_events.TryGetValue(eventKey, out AutoResetEvent handle))
+      {
+        handle.Set();
+      }
+    }
+  }
+}