Merge lp:~coryb/gearmand/time-order into lp:gearmand/1.0

[coryb]

[coryb (coryb)]

This branch adds the --time-order option to gearmand to dispatch jobs in FIFO order irregardless of queue the job is in.

I have merged in all the changes made to trunk so now this branch should merge cleanly to lp:gearmand.

I have also added a test case: tests/time_order_test

The context for this patch was discussed on the gearman google group:
http://groups.google.com/group/gearman/browse_thread/thread/f945197af727262d/f518dfe496724476

[Brian Aker (brianaker)]

If you take a look at http://hudson.drizzle.org/ at the Gearman queue you can see that your test is sometimes failing. Any ideas?

It is currently in the lp:gearmand/build tree

Thanks!

[coryb (coryb)]

Thanks Brian!

> If you take a look at http://hudson.drizzle.org/ at the Gearman queue you can
> see that your test is sometimes failing. Any ideas?

Could you point me to one of the builds that failed due to the time_order_test failing?
I spent some time looking at the builds for Jan-21 in the Gearman-build group, and I didn't see any of them where time_order_test failed. There were several that failed on the sqlite_test, but I didn't change that code, so I am not sure how it could be related exactly.

At one point when I was testing my code, the tests started failing due to a rogue test running in the background due do the fork in test_gearmand_start not getting cleaned up. I suspect some test called test_gearmand_start and then failed to call test_gearmand_stop. I also noticed that in world_create in both sqlite_test.c and my time_order_test.c (which was mostly copied from sqlite_test.c) the world_create calls test_gearmand_start but then there several error cases where we return but fail to call
test_gearmand_stop before returning. It is probably an edge-case but I think that could be triggering the sqlite_test failures. Since I cant log into the build machine I don't think I can debug this test case, but maybe you can log in and check to see if anything is bound to port 32123?

> It is currently in the lp:gearmand/build tree

Awesome, thanks!
-Cory

=== modified file 'gearmand/gearmand.c'
--- gearmand/gearmand.c	2009-12-02 05:22:18 +0000
+++ gearmand/gearmand.c	2010-01-20 21:27:13 +0000
@@ -110,6 +110,7 @@
   uint32_t threads= 0;
   const char *user= NULL;
   uint8_t verbose= 0;
+  uint8_t time_order = 0;
   gearman_return_t ret;
   gearmand_log_info_st log_info;
   bool close_stdio= false;
@@ -155,6 +156,7 @@
   MCO("protocol", 'r', "PROTOCOL", "Load protocol module.")
   MCO("queue-type", 'q', "QUEUE", "Persistent queue type to use.")
   MCO("threads", 't', "THREADS", "Number of I/O threads to use. Default=0.")
+  MCO("time-order", 'T', NULL, "Dispatch jobs in order received")
   MCO("user", 'u', "USER", "Switch to given user after startup.")
   MCO("verbose", 'v', NULL, "Increase verbosity level by one.")
   MCO("version", 'V', NULL, "Display the version of gearmand and exit.")
@@ -278,6 +280,8 @@
       queue_type= value;
     else if (!strcmp(name, "threads"))
       threads= (uint32_t)atoi(value);
+    else if (!strcmp(name, "time-order"))
+      time_order++;
     else if (!strcmp(name, "user"))
       user= value;
     else if (!strcmp(name, "verbose"))
@@ -336,6 +340,7 @@
 
   gearmand_set_backlog(_gearmand, backlog);
   gearmand_set_threads(_gearmand, threads);
+  gearmand_set_time_order(_gearmand, time_order);
   gearmand_set_job_retries(_gearmand, job_retries);
   gearmand_set_worker_wakeup(_gearmand, worker_wakeup);
   gearmand_set_log_fn(_gearmand, _log, &log_info, verbose);

=== modified file 'libgearman-server/gearmand.c'
--- libgearman-server/gearmand.c	2010-01-02 19:57:26 +0000
+++ libgearman-server/gearmand.c	2010-01-20 21:27:13 +0000
@@ -165,6 +165,11 @@
   gearmand->verbose= verbose;
 }
 
+void gearmand_set_time_order(gearmand_st *gearmand, uint8_t time_order)
+{
+  gearman_server_set_time_order(&gearmand->server,time_order);
+}
+
 gearman_return_t gearmand_port_add(gearmand_st *gearmand, in_port_t port,
                                    gearman_connection_add_fn *function)
 {

=== modified file 'libgearman-server/gearmand.h'
--- libgearman-server/gearmand.h	2010-01-02 19:57:26 +0000
+++ libgearman-server/gearmand.h	2010-01-20 21:27:13 +0000
@@ -131,6 +131,15 @@
 void gearmand_set_threads(gearmand_st *gearmand, uint32_t threads);
 
 /**
+ * Set option to order jobs by time
+ * @param gearmand Server instance structure previously initialized with
+ *        gearmand_create.
+ * @param boolean value, 0 (default) will not order jobs by time
+ */
+GEARMAN_API
+void gearmand_set_time_order(gearmand_st *gearmand, uint8_t time_order);
+
+/**
  * Set logging callback for server instance.
  * @param gearmand Server instance structure previously initialized with
  *        gearmand_create.

=== modified file 'libgearman-server/job.c'
--- libgearman-server/job.c	2009-10-29 22:01:39 +0000
+++ libgearman-server/job.c	2010-01-20 21:27:13 +0000
@@ -110,6 +110,9 @@
       return NULL;
     }
 
+    if( server->time_order )
+      gettimeofday(&server_job->insert_time, NULL);
+
     server_job->priority= priority;
 
     server_job->function= server_function;
@@ -218,6 +221,8 @@
   else
     server_job->options= 0;
 
+  server_job->insert_time.tv_sec= 0;
+  server_job->insert_time.tv_usec= 0;
   server_job->retries= 0;
   server_job->priority= 0;
   server_job->job_handle_key= 0;
@@ -298,12 +303,12 @@
   return NULL;
 }
 
-gearman_server_job_st *
-gearman_server_job_peek(gearman_server_con_st *server_con)
-{
+static gearman_server_job_st *
+gearman_server_next_job(gearman_server_con_st *server_con) {
   gearman_server_worker_st *server_worker;
+  gearman_server_job_st *server_job=NULL, *job_cursor = NULL;
   gearman_job_priority_t priority;
-
+  
   for (server_worker= server_con->worker_list; server_worker != NULL;
        server_worker= server_worker->con_next)
   {
@@ -314,50 +319,64 @@
       {
         if (server_worker->function->job_list[priority] != NULL)
         {
-          if (server_worker->function->job_list[priority]->options &
-              GEARMAN_SERVER_JOB_IGNORE)
-          {
-            /* This is only happens when a client disconnects from a foreground
-               job. We do this because we don't want to run the job anymore. */
-            server_worker->function->job_list[priority]->options&=
-                       (gearman_server_job_options_t)~GEARMAN_SERVER_JOB_IGNORE;
-            gearman_server_job_free(gearman_server_job_take(server_con));
-            return gearman_server_job_peek(server_con);
-          }
-          return server_worker->function->job_list[priority];
+          job_cursor = server_worker->function->job_list[priority];
+          if (!server_job)
+          {
+            server_job = job_cursor;
+            server_job->worker =  server_worker;
+            if( !server_job->server->time_order ) {
+              /* if not order by time, just return on the first
+                 found job */
+              return server_job;
+            }
+          }
+          else if (job_cursor->priority < server_job->priority ||
+                   (job_cursor->priority == server_job->priority &&
+                    job_cursor->insert_time.tv_sec < server_job->insert_time.tv_sec) ||
+                   (job_cursor->priority == server_job->priority &&
+                    job_cursor->insert_time.tv_sec == server_job->insert_time.tv_sec &&
+                    job_cursor->insert_time.tv_usec < server_job->insert_time.tv_usec))
+          {
+            /* job is either higher priority or has been sitting in the queue longer,
+               so do it first */
+            server_job = job_cursor;
+            server_job->worker = server_worker;
+            continue;
+          }
         }
       }
     }
   }
+  return server_job;
+}
 
-  return NULL;
+gearman_server_job_st *
+gearman_server_job_peek(gearman_server_con_st *server_con)
+{
+  gearman_server_job_st * server_job = gearman_server_next_job(server_con);
+  if(!server_job)
+    return NULL;
+  if (server_job->options & GEARMAN_SERVER_JOB_IGNORE)
+  {
+    server_job->options &= (gearman_server_job_options_t)~GEARMAN_SERVER_JOB_IGNORE;
+    gearman_server_job_free(gearman_server_job_take(server_con));
+    return gearman_server_job_peek(server_con);
+  }
+  return server_job;
 }
 
 gearman_server_job_st *
 gearman_server_job_take(gearman_server_con_st *server_con)
 {
   gearman_server_worker_st *server_worker;
-  gearman_server_job_st *server_job;
   gearman_job_priority_t priority;
-
-  for (server_worker= server_con->worker_list; server_worker != NULL;
-       server_worker= server_worker->con_next)
-  {
-    if (server_worker->function->job_count != 0)
-      break;
-  }
-
-  if (server_worker == NULL)
+  gearman_server_job_st * server_job = gearman_server_next_job(server_con);
+  if(!server_job)
     return NULL;
 
-  for (priority= GEARMAN_JOB_PRIORITY_HIGH;
-       priority != GEARMAN_JOB_PRIORITY_MAX; priority++)
-  {
-    if (server_worker->function->job_list[priority] != NULL)
-      break;
-  }
+  server_worker = server_job->worker;
+  priority = server_job->priority;
 
-  server_job= server_worker->function->job_list[priority];
   server_job->function->job_list[priority]= server_job->function_next;
   if (server_job->function->job_end[priority] == server_job)
     server_job->function->job_end[priority]= NULL;

=== modified file 'libgearman-server/job.h'
--- libgearman-server/job.h	2009-12-30 19:17:03 +0000
+++ libgearman-server/job.h	2010-01-20 21:27:13 +0000
@@ -53,6 +53,7 @@
   gearman_server_worker_st *worker;
   char job_handle[GEARMAN_JOB_HANDLE_SIZE];
   char unique[GEARMAN_UNIQUE_SIZE];
+  struct timeval insert_time;
 };
 
 /**

=== modified file 'libgearman-server/server.c'
--- libgearman-server/server.c	2010-01-05 18:45:42 +0000
+++ libgearman-server/server.c	2010-01-20 21:27:13 +0000
@@ -101,6 +101,7 @@
   server->free_worker_list= NULL;
   server->log_fn= NULL;
   server->log_context= NULL;
+  server->time_order= 0;
   server->queue_context= NULL;
   server->queue_add_fn= NULL;
   server->queue_flush_fn= NULL;
@@ -207,6 +208,12 @@
   gearman_set_log_fn(server->gearman, _log, server, verbose);
 }
 
+void gearman_server_set_time_order(gearman_server_st *server,
+                                   uint8_t time_order)
+{
+  server->time_order= time_order;
+}
+
 gearman_return_t gearman_server_run_command(gearman_server_con_st *server_con,
                                             gearman_packet_st *packet)
 {

=== modified file 'libgearman-server/server.h'
--- libgearman-server/server.h	2010-01-02 19:57:26 +0000
+++ libgearman-server/server.h	2010-01-20 21:27:13 +0000
@@ -63,6 +63,7 @@
   char job_handle_prefix[GEARMAN_JOB_HANDLE_SIZE];
   gearman_server_job_st *job_hash[GEARMAN_JOB_HASH_SIZE];
   gearman_server_job_st *unique_hash[GEARMAN_JOB_HASH_SIZE];
+  uint8_t time_order;
 };
 
 /**
@@ -117,6 +118,17 @@
                                void *context, gearman_verbose_t verbose);
 
 /**
+ * Set option to order jobs by time
+ * @param server Server structure previously initialized with
+ *        gearman_server_create.
+ * @param boolean value, 0 (default) will not order jobs by time
+ */
+GEARMAN_API
+void gearman_server_set_time_order(gearman_server_st *server,
+                                   uint8_t time_order);
+
+
+/**
  * Process commands for a connection.
  * @param server_con Server connection that has a packet to process.
  * @param packet The packet that needs processing.

=== modified file 'tests/include.am'
--- tests/include.am	2010-01-07 17:46:44 +0000
+++ tests/include.am	2010-01-20 21:27:13 +0000
@@ -50,7 +50,8 @@
 		tests/cpp_test \
 		tests/internals_test \
 		tests/regression_test \
-		tests/worker_test
+		tests/worker_test \
+	    tests/time_order_test
 
 noinst_HEADERS+= \
 	      tests/test.h \
@@ -69,6 +70,9 @@
 tests_worker_test_SOURCES= tests/worker_test.c
 tests_worker_test_LDADD= ${TEST_LDADD}
 
+tests_time_order_test_SOURCES= tests/time_order_test.c
+tests_time_order_test_LDADD= ${TEST_LDADD}
+
 # Test linking with C++ application
 tests_cpp_test_SOURCES= tests/cpp_test.cc
 tests_cpp_test_LDADD= ${TEST_LDADD}
@@ -79,6 +83,9 @@
 test-worker:
 	tests/worker_test $(ARG1) $(ARG2)
 
+test-time-order:
+	tests/time_order_test $(ARG1) $(ARG2)
+
 test-internals:
 	tests/internals_test $(ARG1) $(ARG2)
 
@@ -91,7 +98,7 @@
 	$(SQLITE_TEST) $(ARG1) $(ARG2)
 	$(SQLITE_RM)
 
-check-local: test-client test-worker test-internals test-libmemcached test-sqlite
+check-local: test-client test-worker test-time-order test-internals test-libmemcached test-sqlite
 
 gdb-client: ${noinst_PROGRAMS}
 	$(LIBTOOL) --mode=execute gdb tests/client_test

=== modified file 'tests/test_gearmand.c'
--- tests/test_gearmand.c	2010-01-05 18:11:04 +0000
+++ tests/test_gearmand.c	2010-01-20 21:27:13 +0000
@@ -42,6 +42,10 @@
   pid_t gearmand_pid;
   gearmand_st *gearmand;
   gearman_conf_st conf;
+  gearman_conf_module_st module;
+  const char *name;
+  const char *value;
+  uint8_t time_order = 0;
 
   gearmand_pid= fork();
   assert(gearmand_pid != -1);
@@ -53,6 +57,13 @@
 
     conf_ptr= gearman_conf_create(&conf);
     assert(conf_ptr);
+  
+    assert( gearman_conf_module_create(&conf, &module, NULL) );
+  
+#define MCO(__name, __short, __value, __help) \
+    gearman_conf_module_add_option(&module, __name, __short, __value, __help);
+    MCO("time-order", 'T', NULL, "Dispatch jobs in order received")
+
 #ifdef HAVE_LIBDRIZZLE
     ret= gearman_server_queue_libdrizzle_conf(&conf);
     assert(ret == GEARMAN_SUCCESS);
@@ -69,9 +80,18 @@
     ret= gearman_conf_parse_args(&conf, argc, argv);
     assert(ret == GEARMAN_SUCCESS);
 
+    /* Check for option values that were given. */
+    while (gearman_conf_module_value(&module, &name, &value))
+    {
+      if (!strcmp(name, "time-order"))
+        time_order++;
+    }
+
     gearmand= gearmand_create(NULL, port);
     assert(gearmand != NULL);
 
+    gearmand_set_time_order(gearmand, time_order);
+
     if (queue_type != NULL)
     {
       assert(argc);

=== added file 'tests/time_order_test.c'
--- tests/time_order_test.c	1970-01-01 00:00:00 +0000
+++ tests/time_order_test.c	2010-01-20 21:27:13 +0000
@@ -0,0 +1,197 @@
+/* Gearman server and library
+ * Copyright (C) 2008 Brian Aker, Eric Day
+ * All rights reserved.
+ *
+ * Use and distribution licensed under the BSD license.  See
+ * the COPYING file in the parent directory for full text.
+ */
+
+#include "config.h"
+
+#if defined(NDEBUG)
+# undef NDEBUG
+#endif
+
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <libgearman/gearman.h>
+#include "test.h"
+#include "test_gearmand.h"
+
+#define WORKER_TEST_PORT 32123
+
+typedef struct
+{
+  pid_t gearmand_pid;
+  gearman_worker_st worker;
+  bool run_worker;
+} worker_test_st;
+
+/* Prototypes */
+test_return_t queue_add(void *object);
+test_return_t queue_worker(void *object);
+
+test_return_t pre(void *object);
+test_return_t post(void *object);
+
+void *world_create(test_return_t *error);
+test_return_t world_destroy(void *object);
+
+/* append test for worker */
+static void *append_function(gearman_job_st *job __attribute__((unused)),
+                             void *context, size_t *result_size,
+                             gearman_return_t *ret_ptr __attribute__((unused)))
+{
+  /* this will will set the last char in the context (buffer) to the */
+  /* first char of the work */
+  char * buf = (char *)context;
+  char * work = (char *)gearman_job_workload(job);
+  buf += strlen(buf);
+  *buf = *work;
+  *result_size= 0;
+  return NULL;
+}
+
+test_return_t queue_add(void *object)
+{
+  worker_test_st *test= (worker_test_st *)object;
+  gearman_client_st client;
+  char job_handle[GEARMAN_JOB_HANDLE_SIZE];
+
+  uint8_t * value= (uint8_t *)strdup("0");
+  size_t value_length= 1;
+  uint8_t i;
+
+  test->run_worker= false;
+
+  if (gearman_client_create(&client) == NULL)
+    return TEST_FAILURE;
+
+  if (gearman_client_add_server(&client, NULL,
+                                WORKER_TEST_PORT) != GEARMAN_SUCCESS)
+  {
+    return TEST_FAILURE;
+  }
+
+  /* send strings "0", "1" ... "9" to alternating between 2 queues */
+  /* queue1 = 1,3,5,7,9 */
+  /* queue2 = 0,2,4,6,8 */
+  for (i=0; i<10; i++) {
+    if (gearman_client_do_background(&client, i % 2 ? "queue1" : "queue2", NULL, value,
+                                     value_length, job_handle) != GEARMAN_SUCCESS)
+    {
+      return TEST_FAILURE;
+    }
+    *value = (uint8_t)(*value + 1);
+  }
+
+  gearman_client_free(&client);
+  free(value);
+
+  test->run_worker= true;
+  return TEST_SUCCESS;
+}
+
+test_return_t queue_worker(void *object)
+{
+  worker_test_st *test= (worker_test_st *)object;
+  gearman_worker_st *worker= &(test->worker);
+  char buffer[11];
+  int i;
+  memset(buffer, 0, 11);
+
+  if (!test->run_worker)
+    return TEST_FAILURE;
+
+  if (gearman_worker_add_function(worker, "queue1", 5, append_function,
+                                  buffer) != GEARMAN_SUCCESS)
+  {
+    return TEST_FAILURE;
+  }
+
+  if (gearman_worker_add_function(worker, "queue2", 5, append_function,
+                                  buffer) != GEARMAN_SUCCESS)
+  {
+    return TEST_FAILURE;
+  }
+
+  for (i=0; i<10; i++) {
+    if (gearman_worker_work(worker) != GEARMAN_SUCCESS)
+      return TEST_FAILURE;
+  }
+
+  // expect buffer to be reassembled in order
+  // of insert, not the default "sticky queue" order
+  if( strcmp(buffer, "0123456789") ) 
+    return TEST_FAILURE;
+
+  return TEST_SUCCESS;
+}
+
+
+void *world_create(test_return_t *error)
+{
+  worker_test_st *test;
+  const char *argv[2]= { "test_gearmand", "--time-order"};
+  pid_t gearmand_pid;
+
+  gearmand_pid= test_gearmand_start(WORKER_TEST_PORT, NULL, (char **)argv, 2);
+
+  test= malloc(sizeof(worker_test_st));
+  if (! test)
+  {
+    *error= TEST_MEMORY_ALLOCATION_FAILURE;
+    return NULL;
+  }
+
+  memset(test, 0, sizeof(worker_test_st));
+  if (gearman_worker_create(&(test->worker)) == NULL)
+  {
+    *error= TEST_FAILURE;
+    return NULL;
+  }
+
+  if (gearman_worker_add_server(&(test->worker), NULL, WORKER_TEST_PORT) != GEARMAN_SUCCESS)
+  {
+    *error= TEST_FAILURE;
+    return NULL;
+  }
+
+  test->gearmand_pid= gearmand_pid;
+
+  *error= TEST_SUCCESS;
+
+  return (void *)test;
+}
+
+test_return_t world_destroy(void *object)
+{
+  worker_test_st *test= (worker_test_st *)object;
+  gearman_worker_free(&(test->worker));
+  test_gearmand_stop(test->gearmand_pid);
+  free(test);
+
+  return TEST_SUCCESS;
+}
+
+test_st tests[] ={
+  {"add", 0, queue_add },
+  {"worker", 0, queue_worker },
+  {0, 0, 0}
+};
+
+collection_st collection[] ={
+  {"time_order", 0, 0, tests},
+  {0, 0, 0, 0}
+};
+
+void get_world(world_st *world)
+{
+  world->collections= collection;
+  world->create= world_create;
+  world->destroy= world_destroy;
+}