Job assign priority (#431) [BREAKING CHANGE]

* make sure higher priority job assigned first across all functions

This may break anyone not expecting for priority in one queue to affect priority in another. As such this will trigger a major version bump.

---------

Co-authored-by: huamxu <huamin.xu@nokia-sbell.com>

Author: pythonerdog

libgearman-server/gearmand_con.cc

@@ -359,14 +359,34 @@ gearman_server_job_st * gearman_server_job_peek(gearman_server_con_st *server_co
 
 gearman_server_job_st *gearman_server_job_take(gearman_server_con_st *server_con)
 {
-  for (gearman_server_worker_st *server_worker= server_con->worker_list; server_worker; server_worker= server_worker->con_next)
+  /* Select jobs by global priority across all workers. This ensures a high
+     priority job on any worker is preferred to lower priority jobs on earlier
+     workers in the list. */
+  for (gearman_job_priority_t priority= GEARMAN_JOB_PRIORITY_HIGH;
+       priority != GEARMAN_JOB_PRIORITY_MAX;
+       priority= gearman_job_priority_t(int(priority) +1))
   {
-    if (server_worker->function and server_worker->function->job_count)
+    for (gearman_server_worker_st *server_worker= server_con->worker_list;
+         server_worker; server_worker= server_worker->con_next)
     {
+      if (server_worker->function == NULL || server_worker->function->job_count == 0)
+      {
+        continue;
+      }
+
+      /* Only consider workers that have jobs for this priority. */
+      if (server_worker->function->job_list[priority] == NULL)
+      {
+        continue;
+      }
+
       gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Jobs available for %.*s: %lu",
                          (int)server_worker->function->function_name_size, server_worker->function->function_name,
                          (unsigned long)(server_worker->function->job_count));
 
+      /* Preserve the original round-robin behavior: when we select a worker
+         to hand out a job, move it to the end of the connection's worker
+         list so subsequent selection is fair. */
       if (Server->flags.round_robin)
       {
         GEARMAND_LIST_DEL(server_con->worker, server_worker, con_)
@@ -378,16 +398,6 @@ gearman_server_job_st *gearman_server_job_take(gearman_server_con_st *server_con
         }
       }
 
-      gearman_job_priority_t priority;
-      for (priority= GEARMAN_JOB_PRIORITY_HIGH; priority < GEARMAN_JOB_PRIORITY_LOW;
-           priority= gearman_job_priority_t(int(priority) +1))
-      {
-        if (server_worker->function->job_list[priority])
-        {
-          break;
-        }
-      }
-
       gearman_server_job_st *server_job= server_worker->function->job_list[priority];
       gearman_server_job_st *previous_job= server_job;
 

tests/libgearman-1.0/client_test.cc

@@ -2457,6 +2457,9 @@ test_st limit_tests[] ={
   {0, 0, 0}
 };
 
+// Forward declaration for priority tests
+test_st *test_gearman_worker_priority(void);
+
 collection_st collection[] ={
   {"gearman_return_t", 0, 0, gearman_return_t_TESTS},
   {"init", 0, 0, gearman_client_st_init_TESTS},
@@ -2509,6 +2512,7 @@ collection_st collection[] ={
   {"regression2", reset_SETUP, 0, regression2_TESTS },
   {"gearman_worker_timeout()", default_v2_SETUP, 0, gearman_worker_timeout_TESTS },
   {"gearman_client_set_identifier()", default_v2_SETUP, 0, gearman_client_set_identifier_TESTS },
+  {"gearman_worker_priority", 0, 0, test_gearman_worker_priority() },
   {0, 0, 0, 0}
 };
 

tests/libgearman-1.0/include.am

@@ -28,6 +28,7 @@ t_client_SOURCES+= tests/libgearman-1.0/protocol.cc
 t_client_SOURCES+= tests/libgearman-1.0/server_options.cc 
 t_client_SOURCES+= tests/libgearman-1.0/task.cc 
 t_client_SOURCES+= tests/libgearman-1.0/unique.cc 
+t_client_SOURCES+= tests/libgearman-1.0/priority_test.cc
 t_client_SOURCES+= tests/workers/aggregator/cat.cc
 t_client_SOURCES+= tests/workers/v1/echo_or_react.cc
 t_client_SOURCES+= tests/workers/v1/echo_or_react_chunk.cc