Merge lp:~brianaker/gearmand/job-cleanup-for-exception into lp:gearmand

[Brian Aker]

=== modified file 'libgearman-server/queue.cc'
--- libgearman-server/queue.cc	2013-06-05 21:12:38 +0000
+++ libgearman-server/queue.cc	2013-07-05 22:08:31 +0000
@@ -61,7 +61,11 @@
 {
   assert(server->state.queue_startup == false);
   gearmand_error_t ret;
-  if (server->queue_version == QUEUE_VERSION_FUNCTION)
+  if (server->queue_version == QUEUE_VERSION_NONE)
+  {
+    return GEARMAND_SUCCESS;
+  }
+  else if (server->queue_version == QUEUE_VERSION_FUNCTION)
   {
     assert(server->queue.functions->_add_fn);
     ret= (*(server->queue.functions->_add_fn))(server,
@@ -114,7 +118,11 @@
                                     const char *function_name,
                                     size_t function_name_size)
 {
-  if (server->queue_version == QUEUE_VERSION_FUNCTION)
+  if (server->queue_version == QUEUE_VERSION_NONE)
+  {
+    return GEARMAND_SUCCESS;
+  }
+  else if (server->queue_version == QUEUE_VERSION_FUNCTION)
   {
     assert(server->queue.functions->_done_fn);
     return (*(server->queue.functions->_done_fn))(server,
@@ -123,12 +131,14 @@
                                                   function_name,
                                                   function_name_size);
   }
-
-  assert(server->queue.object);
-  return server->queue.object->done(server,
-                                    unique, unique_size,
-                                    function_name,
-                                    function_name_size);
+  else
+  {
+    assert(server->queue.object);
+    return server->queue.object->done(server,
+                                      unique, unique_size,
+                                      function_name,
+                                      function_name_size);
+  }
 }
 
 void gearman_server_save_job(gearman_server_st& server,
 
=== modified file 'libgearman-server/server.cc'
--- libgearman-server/server.cc	2013-06-27 22:00:23 +0000
+++ libgearman-server/server.cc	2013-07-05 22:08:31 +0000
@@ -827,6 +827,26 @@
       {
         return gearmand_gerror("_server_queue_work_data", ret);
       }
+
+#if 0
+      /* Remove from persistent queue if one exists. */
+      if (server_job->job_queued)
+      {
+        ret= gearman_queue_done(Server,
+                                server_job->unique,
+                                server_job->unique_length,
+                                server_job->function->function_name,
+                                server_job->function->function_name_size);
+        if (gearmand_failed(ret))
+        {
+          gearmand_gerror("Remove from persistent queue", ret);
+          return ret;
+        }
+      }
+
+      /* Job is done, remove it. */
+      gearman_server_job_free(server_job);
+#endif
     }
 
     break;
 
=== modified file 'libgearman/add.cc'
--- libgearman/add.cc	2013-06-23 08:23:26 +0000
+++ libgearman/add.cc	2013-07-05 22:08:31 +0000
@@ -139,15 +139,9 @@
                               const char *unique,
                               const void *workload_str, size_t workload_size,
                               time_t when,
-                              gearman_return_t *ret_ptr,
+                              gearman_return_t& ret_ptr,
                               const gearman_actions_t &actions)
 {
-  gearman_return_t unused;
-  if (ret_ptr == NULL)
-  {
-    ret_ptr= &unused;
-  }
-
   gearman_string_t function= { gearman_string_param_cstr(function_name) };
   gearman_unique_t local_unique= gearman_unique_make(unique, unique ? strlen(unique) : 0);
   gearman_string_t workload= { static_cast<const char *>(workload_str), workload_size };
@@ -155,11 +149,11 @@
   task= add_task(client, task, context, command, function, local_unique, workload, when, actions);
   if (task == NULL)
   {
-    *ret_ptr= client.impl()->universal.error_code();
+    ret_ptr= client.impl()->universal.error_code();
     return NULL;
   }
 
-  *ret_ptr= GEARMAN_SUCCESS;
+  ret_ptr= GEARMAN_SUCCESS;
 
   return task;
 }
 
=== modified file 'libgearman/add.hpp'
--- libgearman/add.hpp	2012-05-13 23:19:56 +0000
+++ libgearman/add.hpp	2013-07-05 22:08:31 +0000
@@ -59,7 +59,7 @@
                               const char *unique,
                               const void *workload_str, size_t workload_size,
                               time_t when,
-                              gearman_return_t *ret_ptr,
+                              gearman_return_t& ret_ptr,
                               const gearman_actions_t &actions);
 
 gearman_task_st *add_task(gearman_client_st& client,
 
=== modified file 'libgearman/client.cc'
--- libgearman/client.cc	2013-06-27 22:00:23 +0000
+++ libgearman/client.cc	2013-07-05 22:08:31 +0000
@@ -1052,7 +1052,7 @@
                       unique,
                       workload, workload_size,
                       time_t(0),
-                      ret_ptr,
+                      *ret_ptr,
                       client->impl()->actions);
 }
 
@@ -1082,7 +1082,7 @@
                       unique,
                       workload, workload_size,
                       time_t(0),
-                      ret_ptr,
+                      *ret_ptr,
                       client->impl()->actions);
 }
 
@@ -1111,7 +1111,7 @@
                       unique,
                       workload, workload_size,
                       time_t(0),
-                      ret_ptr,
+                      *ret_ptr,
                       client->impl()->actions);
 }
 
@@ -1140,7 +1140,7 @@
                       unique,
                       workload, workload_size,
                       time_t(0),
-                      ret_ptr,
+                      *ret_ptr,
                       client->impl()->actions);
 }
 
@@ -1171,7 +1171,7 @@
                       unique,
                       workload, workload_size,
                       time_t(0),
-                      ret_ptr,
+                      *ret_ptr,
                       client->impl()->actions);
 }
 
@@ -1201,7 +1201,7 @@
                       unique,
                       workload, workload_size,
                       time_t(0),
-                      ret_ptr,
+                      *ret_ptr,
                       client->impl()->actions);
 
 }
 
=== modified file 'libgearman/function/function_v1.hpp'
--- libgearman/function/function_v1.hpp	2013-07-02 23:51:10 +0000
+++ libgearman/function/function_v1.hpp	2013-07-05 22:08:31 +0000
@@ -62,20 +62,20 @@
       gearman_job_build_reducer(job, NULL);
     }
 
-    job->error_code= GEARMAN_SUCCESS;
-    job->_worker.work_result= _worker_fn(job, context_arg, &(job->_worker.work_result_size), &job->error_code);
+    job->_error_code= GEARMAN_SUCCESS;
+    job->_worker.work_result= _worker_fn(job, context_arg, &(job->_worker.work_result_size), &job->_error_code);
 
-    if (job->error_code == GEARMAN_LOST_CONNECTION)
+    if (job->_error_code == GEARMAN_LOST_CONNECTION)
     {
       return GEARMAN_FUNCTION_ERROR;
     }
 
-    if (job->error_code == GEARMAN_SHUTDOWN)
+    if (job->_error_code == GEARMAN_SHUTDOWN)
     {
       return GEARMAN_FUNCTION_SHUTDOWN;
     }
 
-    if (gearman_failed(job->error_code))
+    if (gearman_failed(job->_error_code))
     {
       return GEARMAN_FUNCTION_FATAL;
     }
 
=== modified file 'libgearman/function/function_v2.cc'
--- libgearman/function/function_v2.cc	2013-02-03 13:13:26 +0000
+++ libgearman/function/function_v2.cc	2013-07-05 22:08:31 +0000
@@ -56,19 +56,19 @@
   switch (error)
   {
   case GEARMAN_SHUTDOWN:
-    job->error_code= GEARMAN_SUCCESS;
+    job->_error_code= GEARMAN_SUCCESS;
     return GEARMAN_FUNCTION_SHUTDOWN;
 
   case GEARMAN_FATAL:
-    job->error_code= GEARMAN_FATAL;
+    job->_error_code= GEARMAN_FATAL;
     return GEARMAN_FUNCTION_FATAL;
 
   case GEARMAN_ERROR:
-    job->error_code= GEARMAN_ERROR;
+    job->_error_code= GEARMAN_ERROR;
     return GEARMAN_FUNCTION_ERROR;
 
   case GEARMAN_SUCCESS:
-    job->error_code= GEARMAN_SUCCESS;
+    job->_error_code= GEARMAN_SUCCESS;
     return GEARMAN_FUNCTION_SUCCESS;
 
   case GEARMAN_IO_WAIT:
 
=== modified file 'libgearman/function/partition.cc'
--- libgearman/function/partition.cc	2013-02-03 13:13:26 +0000
+++ libgearman/function/partition.cc	2013-07-05 22:08:31 +0000
@@ -56,19 +56,19 @@
   switch (error)
   {
   case GEARMAN_FATAL:
-    job->error_code= GEARMAN_FATAL;
+    job->_error_code= GEARMAN_FATAL;
     return GEARMAN_FUNCTION_FATAL;
 
   case GEARMAN_SHUTDOWN:
-    job->error_code= GEARMAN_SUCCESS;
+    job->_error_code= GEARMAN_SUCCESS;
     return GEARMAN_FUNCTION_SHUTDOWN;
 
   case GEARMAN_ERROR:
-    job->error_code= GEARMAN_ERROR;
+    job->_error_code= GEARMAN_ERROR;
     return GEARMAN_FUNCTION_ERROR;
 
   case GEARMAN_SUCCESS:
-    job->error_code= GEARMAN_SUCCESS;
+    job->_error_code= GEARMAN_SUCCESS;
     return GEARMAN_FUNCTION_SUCCESS;
 
   case GEARMAN_IO_WAIT:
 
=== modified file 'libgearman/interface/worker.hpp'
--- libgearman/interface/worker.hpp	2013-07-02 23:51:10 +0000
+++ libgearman/interface/worker.hpp	2013-07-05 22:08:31 +0000
@@ -132,6 +132,11 @@
     return universal.error();
   }
 
+  gearman_return_t error_code() const
+  {
+    return universal.error_code();
+  }
+
 private:
   gearman_worker_st* _shell;
   gearman_worker_st _owned_shell;
 
=== modified file 'libgearman/job.cc'
--- libgearman/job.cc	2013-07-03 04:42:18 +0000
+++ libgearman/job.cc	2013-07-05 22:08:31 +0000
@@ -191,7 +191,7 @@
     job->options.finished= false;
 
     job->reducer= NULL;
-    job->error_code= GEARMAN_UNKNOWN_STATE;
+    job->_error_code= GEARMAN_UNKNOWN_STATE;
 
     if (job->_worker.job_list)
     {
@@ -456,7 +456,6 @@
       {
         return ret;
       }
-
       job->finished(true);
     }
 
@@ -488,7 +487,13 @@
         job->options.work_in_use= true;
       }
 
-      return _job_send(job);
+      if (gearman_failed(_job_send(job)))
+      {
+        return job->error_code();
+      }
+#if 0
+      job->finished(true);
+#endif
     }
 
     return GEARMAN_SUCCESS;
 
=== modified file 'libgearman/job.hpp'
--- libgearman/job.hpp	2013-07-02 23:51:10 +0000
+++ libgearman/job.hpp	2013-07-05 22:08:31 +0000
@@ -69,10 +69,20 @@
   gearman_packet_st assigned;
   gearman_packet_st work;
   struct gearman_job_reducer_st *reducer;
-  gearman_return_t error_code;
+  gearman_return_t _error_code;
 
   gearman_universal_st& universal()
   {
     return _worker.universal;
   }
+
+  gearman_universal_st& universal() const
+  {
+    return _worker.universal;
+  }
+
+  gearman_return_t error_code() const
+  {
+    return universal().error_code();
+  }
 };
 
=== modified file 'libgearman/worker.cc'
--- libgearman/worker.cc	2013-07-02 23:51:10 +0000
+++ libgearman/worker.cc	2013-07-05 22:08:31 +0000
@@ -1043,19 +1043,19 @@
                                                           static_cast<void *>(worker->impl()->work_function->context)))
           {
             case GEARMAN_FUNCTION_INVALID_ARGUMENT:
-              worker->impl()->work_job->error_code= gearman_error(worker->impl()->universal, GEARMAN_INVALID_ARGUMENT, "worker returned an invalid response, gearman_return_t");
+              worker->impl()->work_job->_error_code= gearman_error(worker->impl()->universal, GEARMAN_INVALID_ARGUMENT, "worker returned an invalid response, gearman_return_t");
             case GEARMAN_FUNCTION_FATAL:
               if (gearman_job_send_fail_fin(worker->impl()->work_job) == GEARMAN_LOST_CONNECTION) // If we fail this, we have no connection, @note this causes us to lose the current error
               {
-                worker->impl()->work_job->error_code= GEARMAN_LOST_CONNECTION;
+                worker->impl()->work_job->_error_code= GEARMAN_LOST_CONNECTION;
                 break;
               }
               worker->impl()->work_state= GEARMAN_WORKER_WORK_UNIVERSAL_FAIL;
-              return worker->impl()->work_job->error_code;
+              return worker->impl()->work_job->_error_code;
 
             case GEARMAN_FUNCTION_ERROR: // retry 
               gearman_reset(worker->impl()->universal);
-              worker->impl()->work_job->error_code= GEARMAN_LOST_CONNECTION;
+              worker->impl()->work_job->_error_code= GEARMAN_LOST_CONNECTION;
               break;
 
             case GEARMAN_FUNCTION_SHUTDOWN:
@@ -1065,7 +1065,7 @@
               break;
           }
 
-          if (worker->impl()->work_job->error_code == GEARMAN_LOST_CONNECTION)
+          if (worker->impl()->work_job->_error_code == GEARMAN_LOST_CONNECTION)
           {
             break;
           }
@@ -1073,12 +1073,12 @@
 
       case GEARMAN_WORKER_WORK_UNIVERSAL_COMPLETE:
         {
-          worker->impl()->work_job->error_code= gearman_job_send_complete_fin(worker->impl()->work_job,
+          worker->impl()->work_job->_error_code= gearman_job_send_complete_fin(worker->impl()->work_job,
                                                                               worker->impl()->work_result, worker->impl()->work_result_size);
-          if (worker->impl()->work_job->error_code == GEARMAN_IO_WAIT)
+          if (worker->impl()->work_job->_error_code == GEARMAN_IO_WAIT)
           {
             worker->impl()->work_state= GEARMAN_WORKER_WORK_UNIVERSAL_COMPLETE;
-            return gearman_error(worker->impl()->universal, worker->impl()->work_job->error_code,
+            return gearman_error(worker->impl()->universal, worker->impl()->work_job->_error_code,
                                  "A failure occurred after worker had successful complete, unless gearman_job_send_complete() was called directly by worker, client has not been informed of success.");
           }
 
@@ -1089,31 +1089,31 @@
           }
 
           // If we lost the connection, we retry the work, otherwise we error
-          if (worker->impl()->work_job->error_code == GEARMAN_LOST_CONNECTION)
+          if (worker->impl()->work_job->_error_code == GEARMAN_LOST_CONNECTION)
           {
             break;
           }
-          else if (worker->impl()->work_job->error_code == GEARMAN_SHUTDOWN)
+          else if (worker->impl()->work_job->_error_code == GEARMAN_SHUTDOWN)
           { }
-          else if (gearman_failed(worker->impl()->work_job->error_code))
+          else if (gearman_failed(worker->impl()->work_job->_error_code))
           {
             worker->impl()->work_state= GEARMAN_WORKER_WORK_UNIVERSAL_FAIL;
 
-            return worker->impl()->work_job->error_code;
+            return worker->impl()->work_job->_error_code;
           }
         }
         break;
 
       case GEARMAN_WORKER_WORK_UNIVERSAL_FAIL:
         {
-          if (gearman_failed(worker->impl()->work_job->error_code= gearman_job_send_fail_fin(worker->impl()->work_job)))
+          if (gearman_failed(worker->impl()->work_job->_error_code= gearman_job_send_fail_fin(worker->impl()->work_job)))
           {
-            if (worker->impl()->work_job->error_code == GEARMAN_LOST_CONNECTION)
+            if (worker->impl()->work_job->_error_code == GEARMAN_LOST_CONNECTION)
             {
               break;
             }
 
-            return worker->impl()->work_job->error_code;
+            return worker->impl()->work_job->_error_code;
           }
         }
         break;
 
=== modified file 'tests/libgearman-1.0/worker_test.cc'
--- tests/libgearman-1.0/worker_test.cc	2013-06-27 22:00:23 +0000
+++ tests/libgearman-1.0/worker_test.cc	2013-07-05 22:08:31 +0000
@@ -759,6 +759,12 @@
   return TEST_SUCCESS;
 }
 
+static gearman_return_t exception_fn(gearman_task_st* task)
+{
+  Out << "Task Handle: " <<  gearman_task_job_handle(task);
+  return GEARMAN_SUCCESS;
+}
+
 static test_return_t gearman_job_send_exception_mass_TEST(void *)
 {
   gearman_function_t call_exception_WORKER_FN= gearman_function_create(call_exception_WORKER);
@@ -772,6 +778,10 @@
 
   std::vector<gearman_task_st*> tasks;
   libgearman::Client client(libtest::default_port());
+
+  gearman_exception_fn *func= exception_fn;
+  gearman_client_set_exception_fn(&client, func);
+
   for (size_t x= 0; x < 100; ++x)
   {
     char buffer[GEARMAN_MAXIMUM_INTEGER_DISPLAY_LENGTH];
@@ -1371,14 +1381,17 @@
     max_block_size= 24;
   }
 
+  libtest::vchar_t workload;
+  libtest::vchar::make(workload, block_size);
+
   for (size_t x= 1; x < max_block_size; ++x)
   {
     if (valgrind_is_caller() and (x * block_size) > 15728640)
     {
       continue;
     }
-    libtest::vchar_t workload;
-    libtest::vchar::make(workload, x * block_size);
+
+    workload.resize(x * block_size);
 
     gearman_argument_t value= gearman_argument_make(0, 0, vchar_param(workload));
 
@@ -1403,7 +1416,7 @@
                  gearman_task_return(task));
 
     gearman_result_st *result= gearman_task_result(task);
-    test_true(result);
+    ASSERT_TRUE(result);
     ASSERT_EQ(gearman_result_size(result), workload.size());
   }