Merge lp:~brianaker/gearmand/misc-fixes2 into lp:gearmand

[Brian Aker]

=== modified file 'libgearman-server/plugins/protocol/gear/protocol.cc'
--- libgearman-server/plugins/protocol/gear/protocol.cc	2013-07-06 09:37:11 +0000
+++ libgearman-server/plugins/protocol/gear/protocol.cc	2013-07-06 17:04:24 +0000
@@ -248,6 +248,15 @@
                          int(packet->data_size),
                          packet->data);
     }
+    else if (packet->command == GEARMAN_COMMAND_OPTION_REQ)
+    {
+      gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
+                         "GEAR length: %" PRIu64 " gearmand_command_t: %s option: %.*s",
+                         uint64_t(packet->data_size),
+                         gearman_strcommand(packet->command),
+                         int(packet->data_size),
+                         packet->data);
+    }
     else
     {
       gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,

=== modified file 'libgearman-server/thread.cc'
--- libgearman-server/thread.cc	2013-05-05 03:23:55 +0000
+++ libgearman-server/thread.cc	2013-07-06 17:04:24 +0000
@@ -309,7 +309,7 @@
 
     gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
                        "Received %s %s:%u",
-                       gearman_command_info(con->packet->packet.command)->name,
+                       gearmand_strcommand(&con->packet->packet),
                        con->_host == NULL ? "-" : con->_host,
                        con->_port == NULL ? "-" : con->_port);
 

=== modified file 'libgearman/connection.cc'
--- libgearman/connection.cc	2013-07-06 00:33:10 +0000
+++ libgearman/connection.cc	2013-07-06 17:04:24 +0000
@@ -75,8 +75,8 @@
 # define FD_CLOEXEC 0
 #endif
 
-#ifndef MSG_DONTWAIT
-# define MSG_DONTWAIT 0
+#ifndef MSG_NOSIGNAL
+# define MSG_NOSIGNAL 0
 #endif
 
 
@@ -1105,7 +1105,7 @@
 #if defined(HAVE_CYASSL) && HAVE_CYASSL
     if (_ssl)
     {
-      read_size= CyaSSL_recv(_ssl, data, data_size, MSG_DONTWAIT);
+      read_size= CyaSSL_recv(_ssl, data, data_size, MSG_NOSIGNAL);
       if (read_size <= 0)
       {
         int sendErr= CyaSSL_get_error(_ssl, read_size);
@@ -1226,7 +1226,7 @@
         int rval;
         do
         { 
-          rval= fcntl (fd, F_SETFD, flags | FD_CLOEXEC);
+          rval= fcntl(fd, F_SETFD, flags | FD_CLOEXEC);
         } while (rval == -1 && (errno == EINTR or errno == EAGAIN));
 
         if (rval == -1)

=== modified file 'libgearman/interface/task.hpp'
--- libgearman/interface/task.hpp	2013-05-07 09:45:37 +0000
+++ libgearman/interface/task.hpp	2013-07-06 17:04:24 +0000
@@ -154,6 +154,11 @@
 
   bool create_result(size_t initial_size);
 
+  void set_state(const enum gearman_task_state_t state_)
+  {
+    state= state_;
+  }
+
 private:
   gearman_task_st* _shell;
   gearman_task_st _owned_shell;

=== modified file 'libgearman/run.cc'
--- libgearman/run.cc	2013-06-27 02:11:30 +0000
+++ libgearman/run.cc	2013-07-06 17:04:24 +0000
@@ -103,7 +103,7 @@
       }
       else if (ret == GEARMAN_IO_WAIT)
       {
-        task->state= GEARMAN_TASK_STATE_SUBMIT;
+        task->set_state(GEARMAN_TASK_STATE_SUBMIT);
         return ret;
       }
       else if (gearman_failed(ret))
@@ -134,12 +134,12 @@
 
           if (ret == GEARMAN_COULD_NOT_CONNECT) // If no connection is found, we will let the user try again
           {
-            task->state= GEARMAN_TASK_STATE_NEW;
+            task->set_state(GEARMAN_TASK_STATE_NEW);
             task->client->impl()->new_tasks++;
           }
           else
           {
-            task->state= GEARMAN_TASK_STATE_FAIL;
+            task->set_state(GEARMAN_TASK_STATE_FAIL);
             task->client->impl()->running_tasks--;
           }
           return ret;
@@ -166,13 +166,13 @@
       gearman_return_t ret= task->func.workload_fn(task->shell());
       if (gearman_failed(ret))
       {
-        task->state= GEARMAN_TASK_STATE_WORKLOAD;
+        task->set_state(GEARMAN_TASK_STATE_WORKLOAD);
         return ret;
       }
     }
 
     task->client->impl()->options.no_new= false;
-    task->state= GEARMAN_TASK_STATE_WORK;
+    task->set_state(GEARMAN_TASK_STATE_WORK);
     task->con->set_events(POLLIN);
     return GEARMAN_SUCCESS;
 
@@ -190,7 +190,7 @@
         gearman_return_t ret= task->func.created_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_CREATED;
+          task->set_state(GEARMAN_TASK_STATE_CREATED);
           return ret;
         }
       }
@@ -201,6 +201,7 @@
           task->send.command == GEARMAN_COMMAND_SUBMIT_JOB_EPOCH ||
           task->send.command == GEARMAN_COMMAND_SUBMIT_REDUCE_JOB_BACKGROUND)
       {
+        task->result_rc= GEARMAN_SUCCESS;
         break;
       }
     }
@@ -215,7 +216,7 @@
         gearman_return_t ret= task->func.data_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_DATA;
+          task->set_state(GEARMAN_TASK_STATE_DATA);
           return ret;
         }
       }
@@ -228,7 +229,7 @@
         gearman_return_t ret= task->func.warning_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_WARNING;
+          task->set_state(GEARMAN_TASK_STATE_WARNING);
           return ret;
         }
       }
@@ -318,7 +319,7 @@
         gearman_return_t ret= task->func.status_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_STATUS;
+          task->set_state(GEARMAN_TASK_STATE_STATUS);
           return ret;
         }
       }
@@ -345,7 +346,7 @@
         gearman_return_t ret= task->func.complete_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_COMPLETE;
+          task->set_state(GEARMAN_TASK_STATE_COMPLETE);
           return ret;
         }
       }
@@ -369,7 +370,7 @@
         gearman_return_t ret= task->func.exception_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_EXCEPTION;
+          task->set_state(GEARMAN_TASK_STATE_EXCEPTION);
           return ret;
         }
       }
@@ -391,7 +392,7 @@
         gearman_return_t ret= task->func.fail_fn(task->shell());
         if (gearman_failed(ret))
         {
-          task->state= GEARMAN_TASK_STATE_FAIL;
+          task->set_state(GEARMAN_TASK_STATE_FAIL);
           return ret;
         }
       }
@@ -399,7 +400,7 @@
       break;
     }
 
-    task->state= GEARMAN_TASK_STATE_WORK;
+    task->set_state(GEARMAN_TASK_STATE_WORK);
     return GEARMAN_SUCCESS;
 
   case GEARMAN_TASK_STATE_FINISHED:
@@ -407,7 +408,10 @@
   }
 
   task->client->impl()->running_tasks--;
-  task->state= GEARMAN_TASK_STATE_FINISHED;
+  task->set_state(GEARMAN_TASK_STATE_FINISHED);
+
+  // @todo this should never happen... but background tasks can signal it.
+  assert(task->result_rc != GEARMAN_UNKNOWN_STATE);
 
   if (task->client->impl()->options.free_tasks and task->type == GEARMAN_TASK_KIND_ADD_TASK)
   {

=== modified file 'tests/libgearman-1.0/worker_test.cc'
--- tests/libgearman-1.0/worker_test.cc	2013-07-06 08:02:19 +0000
+++ tests/libgearman-1.0/worker_test.cc	2013-07-06 17:04:24 +0000
@@ -66,6 +66,19 @@
 #include "tests/workers/v2/call_exception.h"
 #include "tests/workers/v2/check_order.h"
 
+#if 0
+static gearman_return_t exception_fn(gearman_task_st* task)
+{
+  Out << "GEARMAN_WORK_EXCEPTION: Task Handle: " <<  gearman_task_job_handle(task) << " return:" << gearman_strerror(gearman_task_return(task));
+  return GEARMAN_SUCCESS;
+}
+#endif
+
+static void error_logger(const char* message, gearman_verbose_t, void*)
+{
+  Error << message;
+}
+
 static test_return_t init_test(void *)
 {
   gearman_worker_st worker;
@@ -420,7 +433,6 @@
 static test_return_t job_order_TEST(void *)
 {
   libgearman::Client client(libtest::default_port());;
-  ASSERT_EQ(true, gearman_client_set_server_option(&client, test_literal_param("exceptions")));
   gearman_client_add_options(&client, GEARMAN_CLIENT_EXCEPTION);
   gearman_client_add_options(&client, GEARMAN_CLIENT_GENERATE_UNIQUE);
 
@@ -534,13 +546,16 @@
   for (std::vector<gearman_task_st*>::iterator iter= tasks.begin();
        iter != tasks.end(); ++iter)
   {
-    if (gearman_task_return(*iter) != GEARMAN_UNKNOWN_STATE)
+
+    if (gearman_failed(gearman_task_return(*iter)))
     {
-      Error << ":" << gearman_task_error(*iter) << ":";
-      Error << ":" << gearman_strerror(gearman_task_return(*iter)) << ":";
+      if (gearman_task_return(*iter) != GEARMAN_UNKNOWN_STATE)
+      {
+        Error << "gearman_task_error(" << gearman_task_error(*iter) << ") gearman_task_return(" << gearman_strerror(gearman_task_return(*iter)) << ")";
+        ASSERT_EQ(GEARMAN_UNKNOWN_STATE, gearman_task_return(*iter));
+        ASSERT_NULL(gearman_task_error(*iter));
+      }
     }
-    ASSERT_EQ(GEARMAN_UNKNOWN_STATE, gearman_task_return(*iter));
-    ASSERT_NULL(gearman_task_error(*iter));
   }
 
   for (uint32_t x= 0; x < 10; ++x)
@@ -759,12 +774,6 @@
   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);
@@ -779,8 +788,14 @@
   std::vector<gearman_task_st*> tasks;
   libgearman::Client client(libtest::default_port());
 
+#if 0
   gearman_exception_fn *func= exception_fn;
   gearman_client_set_exception_fn(&client, func);
+#endif
+#if 0
+  ASSERT_EQ(true, gearman_client_set_server_option(&client, test_literal_param("exceptions")));
+#endif
+  gearman_client_add_options(&client, GEARMAN_CLIENT_EXCEPTION);
 
   for (size_t x= 0; x < 100; ++x)
   {
@@ -800,33 +815,59 @@
     tasks.push_back(task);
   }
 
-  {
-    gearman_return_t ret;
-    do {
-      ret= gearman_client_run_tasks(&client);
-    } while (gearman_continue(ret));
-  }
-
-  for (std::vector<gearman_task_st*>::iterator iter= tasks.begin();
-       iter != tasks.end(); ++iter)
-  {
-    ASSERT_EQ(GEARMAN_WORK_FAIL, gearman_task_return(*iter));
+  bool more= true;
+  while (more)
+  {
+    for (std::vector<gearman_task_st*>::iterator iter= tasks.begin();
+         iter != tasks.end(); ++iter)
+    {
+      if (gearman_task_return(*iter) == GEARMAN_UNKNOWN_STATE)
+      {
+        {
+          gearman_return_t ret;
+          do {
+            ret= gearman_client_run_tasks(&client);
+          } while (gearman_continue(ret));
+
+          if (gearman_failed(ret))
+          {
+            Error << gearman_strerror(ret);
+          }
+          ASSERT_EQ(GEARMAN_SUCCESS, ret);
+        }
+
+        continue;
+      }
+      else
+      {
+        if (gearman_client_has_option(&client, GEARMAN_CLIENT_EXCEPTION))
+        {
+          ASSERT_EQ(GEARMAN_WORK_EXCEPTION, gearman_task_return(*iter));
+        }
+        else
+        {
+          ASSERT_EQ(GEARMAN_WORK_FAIL, gearman_task_return(*iter));
+        }
+      }
+
+      more= false;
+    }
   }
 
   return TEST_SUCCESS;
 }
 
-static void error_logger(const char* message, gearman_verbose_t, void*)
-{
-  Error << message;
-}
-
 static test_return_t gearman_job_send_exception_TEST(void *)
 {
   libgearman::Client client(libtest::default_port());
   gearman_client_set_log_fn(&client, error_logger, NULL, GEARMAN_VERBOSE_ERROR);
 
-  ASSERT_EQ(true, gearman_client_set_server_option(&client, test_literal_param("exceptions")));
+#if 0
+  gearman_exception_fn *func= exception_fn;
+  gearman_client_set_exception_fn(&client, func);
+#endif
+
+  gearman_client_add_options(&client, GEARMAN_CLIENT_EXCEPTION);
 
   gearman_function_t exception_WORKER_FN= gearman_function_create(exception_WORKER);
   std::auto_ptr<worker_handle_st> handle(test_worker_start(libtest::default_port(),
@@ -848,9 +889,16 @@
     ret= gearman_client_run_tasks(&client);
   } while (gearman_continue(ret));
 
+  if (gearman_client_has_option(&client, GEARMAN_CLIENT_EXCEPTION))
+  {
+    ASSERT_EQ(GEARMAN_WORK_EXCEPTION, gearman_task_return(task));
+  }
+  else
+  {
+    ASSERT_EQ(GEARMAN_WORK_FAIL, gearman_task_return(task));
+  }
+  ASSERT_TRUE(gearman_task_has_exception(task));
   gearman_string_t exception= gearman_task_exception(task);
-  ASSERT_EQ(gearman_task_return(task), GEARMAN_WORK_EXCEPTION);
-  ASSERT_TRUE(gearman_task_has_exception(task));
   ASSERT_STREQ("dog", gearman_c_str(exception));
 
   return TEST_SUCCESS;
@@ -1252,20 +1300,19 @@
   {
     libgearman::Client client(libtest::default_port());
 
-    ASSERT_EQ(gearman_client_do_background(&client, function_name, unique_name,
-                                              test_string_make_from_array(unique_name), NULL),
-                 GEARMAN_SUCCESS);
+    ASSERT_EQ(GEARMAN_SUCCESS,
+              gearman_client_do_background(&client, function_name, unique_name, test_string_make_from_array(unique_name), NULL));
   }
 
   gearman_worker_remove_options(&worker, GEARMAN_WORKER_GRAB_UNIQ);
-  test_false(worker->impl()->options.grab_uniq);
+  ASSERT_FALSE(worker->impl()->options.grab_uniq);
 
   gearman_worker_set_timeout(&worker, 800);
 
   gearman_return_t rc;
   gearman_job_st *job= gearman_worker_grab_job(&worker, NULL, &rc);
   ASSERT_EQ(rc, GEARMAN_SUCCESS);
-  test_truth(job);
+  ASSERT_TRUE(job);
 
   size_t size= 0;
   void *result= no_unique_worker(job, NULL, &size, &rc);