Merge lp:~brianaker/gearmand/ref-client-fix into lp:gearmand

[Brian Aker]

=== modified file 'libgearman/actions.cc'
--- libgearman/actions.cc	2013-07-06 08:02:19 +0000
+++ libgearman/actions.cc	2013-07-06 19:18:28 +0000
@@ -68,7 +68,7 @@
     {
       if (task->create_result(gearman_task_data_size(shell)) == false)
       {
-        return gearman_error(task->client->impl()->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "Failed to create result_st");
+        return gearman_error(task->client->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "Failed to create result_st");
       }
     }
     assert_msg(task->result(), "programmer error, result_ptr has not been allocated for task");
@@ -137,7 +137,7 @@
     {
       if (task->create_result(gearman_task_data_size(shell)) == false)
       {
-        return gearman_error(task->client->impl()->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "Failed to create result_st");
+        return gearman_error(task->client->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "Failed to create result_st");
       }
     }
 
@@ -158,7 +158,7 @@
     {
       if (task->create_result(gearman_task_data_size(shell)) == false)
       {
-        return gearman_error(task->client->impl()->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "Failed to create result_st");
+        return gearman_error(task->client->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "Failed to create result_st");
       }
     }
 
 
=== modified file 'libgearman/add.cc'
--- libgearman/add.cc	2013-07-03 09:00:14 +0000
+++ libgearman/add.cc	2013-07-06 19:18:28 +0000
@@ -202,7 +202,6 @@
     return NULL;
   }
   assert(task_shell->impl()->client);
-  assert(task_shell->impl()->client == &client);
 
   Task* task= task_shell->impl();
 
@@ -229,7 +228,7 @@
     {
       if (safe_uuid_generate(task->unique, task->unique_length) == -1)
       {
-        gearman_log_debug(task->client->impl()->universal, "uuid_generate_time_safe() failed or does not exist on this platform");
+        gearman_log_debug(task->client->universal, "uuid_generate_time_safe() failed or does not exist on this platform");
       }
     }
     else
@@ -242,7 +241,6 @@
   gearman_unique_t final_unique= gearman_unique_make(task->unique, task->unique_length);
 
   assert(task->client);
-  assert(task->client == &client);
 
   gearman_return_t rc= GEARMAN_INVALID_ARGUMENT;
   switch (command)
@@ -250,7 +248,7 @@
   case GEARMAN_COMMAND_SUBMIT_JOB:
   case GEARMAN_COMMAND_SUBMIT_JOB_LOW:
   case GEARMAN_COMMAND_SUBMIT_JOB_HIGH:
-    rc= libgearman::protocol::submit(task->client->impl()->universal,
+    rc= libgearman::protocol::submit(task->client->universal,
                                      task->send,
                                      final_unique,
                                      command,
@@ -259,7 +257,7 @@
     break;
 
   case GEARMAN_COMMAND_SUBMIT_JOB_EPOCH:
-    rc= libgearman::protocol::submit_epoch(task->client->impl()->universal,
+    rc= libgearman::protocol::submit_epoch(task->client->universal,
                                            task->send,
                                            final_unique,
                                            function,
@@ -270,7 +268,7 @@
   case GEARMAN_COMMAND_SUBMIT_JOB_BG:
   case GEARMAN_COMMAND_SUBMIT_JOB_LOW_BG:
   case GEARMAN_COMMAND_SUBMIT_JOB_HIGH_BG:
-    rc= libgearman::protocol::submit_background(task->client->impl()->universal,
+    rc= libgearman::protocol::submit_background(task->client->universal,
                                                 task->send,
                                                 final_unique,
                                                 command,
@@ -379,15 +377,16 @@
     return NULL;
   }
 
-  gearman_task_st *task= gearman_task_internal_create(*client, NULL);
-  if (task == NULL)
+  gearman_task_st *task_shell= gearman_task_internal_create(*client, NULL);
+  if (task_shell == NULL)
   {
     assert(client->impl()->universal.error_code());
     return NULL;
   }
 
-  task->impl()->context= context;
-  task->impl()->func= actions;
+  Task* task= task_shell->impl();
+  task->context= context;
+  task->func= actions;
 
   /**
     @todo fix it so that NULL is done by default by the API not by happenstance.
@@ -413,33 +412,33 @@
 
   if (gearman_unique_is_hash(unique))
   {
-    task->impl()->unique_length= snprintf(task->impl()->unique, GEARMAN_MAX_UNIQUE_SIZE, "%u", libhashkit_murmur3(gearman_string_param(workload)));
+    task->unique_length= snprintf(task->unique, GEARMAN_MAX_UNIQUE_SIZE, "%u", libhashkit_murmur3(gearman_string_param(workload)));
   }
-  else if ((task->impl()->unique_length= gearman_size(unique)))
+  else if ((task->unique_length= gearman_size(unique)))
   {
-    if (task->impl()->unique_length >= GEARMAN_MAX_UNIQUE_SIZE)
+    if (task->unique_length >= GEARMAN_MAX_UNIQUE_SIZE)
     {
-      task->impl()->unique_length= GEARMAN_MAX_UNIQUE_SIZE -1; // Leave space for NULL byte
+      task->unique_length= GEARMAN_MAX_UNIQUE_SIZE -1; // Leave space for NULL byte
     }
 
-    strncpy(task->impl()->unique, gearman_c_str(unique), GEARMAN_MAX_UNIQUE_SIZE);
-    task->impl()->unique[task->impl()->unique_length]= 0;
+    strncpy(task->unique, gearman_c_str(unique), GEARMAN_MAX_UNIQUE_SIZE);
+    task->unique[task->unique_length]= 0;
   }
   else
   {
     if (client->impl()->options.generate_unique or is_background(command))
     {
-      safe_uuid_generate(task->impl()->unique, task->impl()->unique_length);
+      safe_uuid_generate(task->unique, task->unique_length);
     }
     else
     {
-      task->impl()->unique_length= 0;
-      task->impl()->unique[0]= 0;
+      task->unique_length= 0;
+      task->unique[0]= 0;
     }
   }
 
-  args[1]= task->impl()->unique;
-  args_size[1]= task->impl()->unique_length +1; // +1 is for the needed null
+  args[1]= task->unique;
+  args_size[1]= task->unique_length +1; // +1 is for the needed null
 
   assert_msg(command == GEARMAN_COMMAND_SUBMIT_REDUCE_JOB or command == GEARMAN_COMMAND_SUBMIT_REDUCE_JOB_BACKGROUND,
              "Command was not appropriate for request");
@@ -474,14 +473,14 @@
   args_size[4]= gearman_size(workload);
 
   gearman_return_t rc;
-  if (gearman_success(rc= gearman_packet_create_args(client->impl()->universal, task->impl()->send,
+  if (gearman_success(rc= gearman_packet_create_args(client->impl()->universal, task->send,
                                                      GEARMAN_MAGIC_REQUEST, command,
                                                      args, args_size,
                                                      5)))
   {
     client->impl()->new_tasks++;
     client->impl()->running_tasks++;
-    task->impl()->options.send_in_use= true;
+    task->options.send_in_use= true;
   }
   else
   {
@@ -489,7 +488,7 @@
     gearman_task_free(task);
     task= NULL;
   }
-  task->impl()->type= GEARMAN_TASK_KIND_EXECUTE;
+  task->type= GEARMAN_TASK_KIND_EXECUTE;
 
-  return task;
+  return task->shell();
 }
 
=== modified file 'libgearman/client.cc'
--- libgearman/client.cc	2013-07-06 08:02:19 +0000
+++ libgearman/client.cc	2013-07-06 19:18:28 +0000
@@ -89,7 +89,7 @@
 /**
  * Real do function.
  */
-static void *_client_do(gearman_client_st *client, gearman_command_t command,
+static void *_client_do(gearman_client_st *client_shell, gearman_command_t command,
                         const char *function_name,
                         const char *unique,
                         const void *workload_str, size_t workload_size,
@@ -101,13 +101,15 @@
     ret_ptr= &unused;
   }
 
-  if (client == NULL or client->impl() == NULL)
+  if (client_shell == NULL or client_shell->impl() == NULL)
   {
     *ret_ptr= GEARMAN_INVALID_ARGUMENT;
     return NULL;
   }
 
-  client->impl()->universal.reset_error();
+  Client* client= client_shell->impl();
+
+  client->universal.reset_error();
 
   size_t unused_size;
   if (result_size == NULL)
@@ -122,17 +124,17 @@
 
   gearman_task_st do_task;
   {
-    client->impl()->universal.options.no_new_data= true;
-    gearman_task_st *do_task_ptr= add_task(*client, &do_task, NULL, command,
+    client->universal.options.no_new_data= true;
+    gearman_task_st *do_task_ptr= add_task(*(client->shell()), &do_task, NULL, command,
                                            function,
                                            local_unique,
                                            workload,
                                            time_t(0),
                                            gearman_actions_do_default());
-    client->impl()->universal.options.no_new_data= false;
+    client->universal.options.no_new_data= false;
     if (do_task_ptr == NULL)
     {
-      *ret_ptr= client->impl()->universal.error_code();
+      *ret_ptr= client->universal.error_code();
       return NULL;
     }
     assert_msg(do_task.impl(), "Bad return by add_task()");
@@ -141,10 +143,10 @@
 
   do_task.impl()->type= GEARMAN_TASK_KIND_DO;
 
-  gearman_return_t ret= gearman_client_run_block_tasks(client, &do_task);
+  gearman_return_t ret= gearman_client_run_block_tasks(client->shell(), &do_task);
 
   // gearman_client_run_tasks failed
-  assert(client->impl()->task_list); // Programmer error, we should always have the task that we used for do
+  assert(client->task_list); // Programmer error, we should always have the task that we used for do
 
   char *returnable= NULL;
   if (gearman_failed(ret))
@@ -153,7 +155,7 @@
     { }
     else
     {
-      gearman_error(client->impl()->universal, ret, "occured during gearman_client_run_tasks()");
+      gearman_error(client->universal, ret, "occured during gearman_client_run_tasks()");
     }
 
     *ret_ptr= ret;
@@ -164,13 +166,13 @@
     *ret_ptr= do_task.impl()->result_rc;
     if (gearman_task_result(&do_task))
     {
-      if (gearman_has_allocator(client->impl()->universal))
+      if (gearman_has_allocator(client->universal))
       {
         gearman_string_t result= gearman_result_string(do_task.impl()->result());
-        returnable= static_cast<char *>(gearman_malloc(client->impl()->universal, gearman_size(result) +1));
+        returnable= static_cast<char *>(gearman_malloc(client->universal, gearman_size(result) +1));
         if (returnable == NULL)
         {
-          gearman_error(client->impl()->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "custom workload_fn failed to allocate memory");
+          gearman_error(client->universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "custom workload_fn failed to allocate memory");
           *result_size= 0;
         }
         else // NULL terminate
@@ -194,15 +196,15 @@
   }
   else // gearman_client_run_tasks() was successful, but the task was not
   {
-    gearman_error(client->impl()->universal, do_task.impl()->result_rc, "occured during gearman_client_run_tasks()");
+    gearman_error(client->universal, do_task.impl()->result_rc, "occured during gearman_client_run_tasks()");
 
     *ret_ptr= do_task.impl()->result_rc;
     *result_size= 0;
   }
 
   gearman_task_free(&do_task);
-  client->impl()->new_tasks= 0;
-  client->impl()->running_tasks= 0;
+  client->new_tasks= 0;
+  client->running_tasks= 0;
 
   return returnable;
 }
@@ -210,31 +212,32 @@
 /*
   Real background do function.
 */
-static gearman_return_t _client_do_background(gearman_client_st *client,
+static gearman_return_t _client_do_background(gearman_client_st* client_shell,
                                               gearman_command_t command,
                                               gearman_string_t &function,
                                               gearman_unique_t &unique,
                                               gearman_string_t &workload,
                                               gearman_job_handle_t job_handle)
 {
-  if (client == NULL)
+  if (client_shell == NULL or client_shell->impl() == NULL)
   {
     return GEARMAN_INVALID_ARGUMENT;
   }
 
-  client->impl()->universal.reset_error();
+  Client* client= client_shell->impl();
+  client->universal.reset_error();
 
   if (gearman_size(function) == 0)
   {
-    return gearman_error(client->impl()->universal, GEARMAN_INVALID_ARGUMENT, "function argument was empty");
+    return gearman_error(client->universal, GEARMAN_INVALID_ARGUMENT, "function argument was empty");
   }
 
-  client->impl()->_do_handle[0]= 0; // Reset the job_handle we store in client
+  client->_do_handle[0]= 0; // Reset the job_handle we store in client
 
   gearman_task_st do_task;
   {
-    client->impl()->universal.options.no_new_data= true;
-    gearman_task_st* do_task_ptr= add_task(*client, &do_task, 
+    client->universal.options.no_new_data= true;
+    gearman_task_st* do_task_ptr= add_task(*client_shell, &do_task, 
                                            client, 
                                            command,
                                            function,
@@ -242,26 +245,26 @@
                                            workload,
                                            time_t(0),
                                            gearman_actions_do_default());
-    client->impl()->universal.options.no_new_data= false;
+    client->universal.options.no_new_data= false;
 
     if (do_task_ptr == NULL)
     {
-      return client->impl()->universal.error_code();
+      return client->universal.error_code();
     }
     assert(do_task_ptr);
     assert(&do_task == do_task_ptr);
   }
   do_task.impl()->type= GEARMAN_TASK_KIND_DO;
 
-  gearman_return_t ret= gearman_client_run_block_tasks(client, &do_task);
+  gearman_return_t ret= gearman_client_run_block_tasks(client_shell, &do_task);
 
   if (job_handle)
   {
     strncpy(job_handle, do_task.impl()->job_handle, GEARMAN_JOB_HANDLE_SIZE);
   }
-  strncpy(client->impl()->_do_handle, do_task.impl()->job_handle, GEARMAN_JOB_HANDLE_SIZE);
-  client->impl()->new_tasks= 0;
-  client->impl()->running_tasks= 0;
+  strncpy(client->_do_handle, do_task.impl()->job_handle, GEARMAN_JOB_HANDLE_SIZE);
+  client->new_tasks= 0;
+  client->running_tasks= 0;
   gearman_task_free(&do_task);
 
   return ret;
@@ -393,31 +396,33 @@
 {
   if (client_shell and client_shell->impl())
   {
+    Client* client= client_shell->impl();
+
     switch (option)
     {
     case GEARMAN_CLIENT_ALLOCATED:
       return gearman_is_allocated(client_shell);
 
     case GEARMAN_CLIENT_NON_BLOCKING:
-      return client_shell->impl()->options.non_blocking;
+      return client->options.non_blocking;
 
     case GEARMAN_CLIENT_UNBUFFERED_RESULT:
-      return client_shell->impl()->options.unbuffered_result;
+      return client->options.unbuffered_result;
 
     case GEARMAN_CLIENT_NO_NEW:
-      return client_shell->impl()->options.no_new;
+      return client->options.no_new;
 
     case GEARMAN_CLIENT_FREE_TASKS:
-      return client_shell->impl()->options.free_tasks;
+      return client->options.free_tasks;
 
     case GEARMAN_CLIENT_GENERATE_UNIQUE:
-      return client_shell->impl()->options.generate_unique;
+      return client->options.generate_unique;
 
     case GEARMAN_CLIENT_EXCEPTION:
-      return client_shell->impl()->options.exceptions;
+      return client->options.exceptions;
 
     case GEARMAN_CLIENT_SSL:
-      return client_shell->impl()->ssl();
+      return client->ssl();
 
     default:
     case GEARMAN_CLIENT_TASK_IN_USE:
@@ -463,35 +468,36 @@
 {
   if (client_shell and client_shell->impl())
   {
+    Client* client= client_shell->impl();
     if (options & GEARMAN_CLIENT_NON_BLOCKING)
     {
-      gearman_universal_add_options(client_shell->impl()->universal, GEARMAN_UNIVERSAL_NON_BLOCKING);
-      client_shell->impl()->options.non_blocking= true;
+      gearman_universal_add_options(client->universal, GEARMAN_UNIVERSAL_NON_BLOCKING);
+      client->options.non_blocking= true;
     }
 
     if (options & GEARMAN_CLIENT_UNBUFFERED_RESULT)
     {
-      client_shell->impl()->options.unbuffered_result= true;
+      client->options.unbuffered_result= true;
     }
 
     if (options & GEARMAN_CLIENT_FREE_TASKS)
     {
-      client_shell->impl()->options.free_tasks= true;
+      client->options.free_tasks= true;
     }
 
     if (options & GEARMAN_CLIENT_GENERATE_UNIQUE)
     {
-      client_shell->impl()->options.generate_unique= true;
+      client->options.generate_unique= true;
     }
 
     if (options & GEARMAN_CLIENT_EXCEPTION)
     {
-      client_shell->impl()->options.exceptions= gearman_client_set_server_option(client_shell, gearman_literal_param("exceptions"));
+      client->options.exceptions= gearman_client_set_server_option(client_shell, gearman_literal_param("exceptions"));
     }
 
     if (options & GEARMAN_CLIENT_SSL)
     {
-      client_shell->impl()->ssl(true);
+      client->ssl(true);
     }
   }
 }
@@ -501,25 +507,27 @@
 {
   if (client_shell and client_shell->impl())
   {
+    Client* client= client_shell->impl();
+
     if (options & GEARMAN_CLIENT_NON_BLOCKING)
     {
-      gearman_universal_remove_options(client_shell->impl()->universal, GEARMAN_UNIVERSAL_NON_BLOCKING);
-      client_shell->impl()->options.non_blocking= false;
+      gearman_universal_remove_options(client->universal, GEARMAN_UNIVERSAL_NON_BLOCKING);
+      client->options.non_blocking= false;
     }
 
     if (options & GEARMAN_CLIENT_UNBUFFERED_RESULT)
     {
-      client_shell->impl()->options.unbuffered_result= false;
+      client->options.unbuffered_result= false;
     }
 
     if (options & GEARMAN_CLIENT_FREE_TASKS)
     {
-      client_shell->impl()->options.free_tasks= false;
+      client->options.free_tasks= false;
     }
 
     if (options & GEARMAN_CLIENT_GENERATE_UNIQUE)
     {
-      client_shell->impl()->options.generate_unique= false;
+      client->options.generate_unique= false;
     }
   }
 }
@@ -988,13 +996,14 @@
   return gearman_echo(client->impl()->universal, workload, workload_size);
 }
 
-void gearman_client_task_free_all(gearman_client_st *client)
+void gearman_client_task_free_all(gearman_client_st *client_shell)
 {
-  if (client and client->impl() and client->impl()->task_list)
+  if (client_shell and client_shell->impl() and client_shell->impl()->task_list)
   {
-    while (client->impl()->task_list)
+    Client* client= client_shell->impl();
+    while (client->task_list)
     {
-      gearman_task_free(client->impl()->task_list);
+      gearman_task_free(client->task_list);
     }
   }
 }
@@ -1397,96 +1406,96 @@
   }
 }
 
-static inline gearman_return_t _client_run_tasks(gearman_client_st *client, gearman_task_st* exit_task)
+static inline gearman_return_t _client_run_tasks(gearman_client_st *client_shell, gearman_task_st* exit_task)
 {
   gearman_return_t ret= GEARMAN_MAX_RETURN;
 
-  switch(client->impl()->state)
+  Client* client= client_shell->impl();
+
+  switch(client->state)
   {
   case GEARMAN_CLIENT_STATE_IDLE:
     while (1)
     {
       /* Start any new tasks. */
-      if (client->impl()->new_tasks > 0 && ! (client->impl()->options.no_new))
+      if (client->new_tasks > 0 && ! (client->options.no_new))
       {
-        for (client->impl()->task= client->impl()->task_list; client->impl()->task;
-             client->impl()->task= client->impl()->task->impl()->next)
+        for (client->task= client->task_list; client->task;
+             client->task= client->task->impl()->next)
         {
-          if (client->impl()->task->impl()->state != GEARMAN_TASK_STATE_NEW)
+          if (client->task->impl()->state != GEARMAN_TASK_STATE_NEW)
           {
             continue;
           }
 
   case GEARMAN_CLIENT_STATE_NEW:
-          if (client->impl()->task == NULL)
+          if (client->task == NULL)
           {
-            client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+            client->state= GEARMAN_CLIENT_STATE_IDLE;
             break;
           }
 
-          assert_msg(client == client->impl()->task->impl()->client, "Programmer error, client and task member client are not the same");
-          gearman_return_t local_ret= _client_run_task(client->impl()->task->impl());
+          gearman_return_t local_ret= _client_run_task(client->task->impl());
           if (gearman_failed(local_ret) and local_ret != GEARMAN_IO_WAIT)
           {
-            client->impl()->state= GEARMAN_CLIENT_STATE_NEW;
+            client->state= GEARMAN_CLIENT_STATE_NEW;
 
             return local_ret;
           }
         }
 
-        if (client->impl()->new_tasks == 0)
+        if (client->new_tasks == 0)
         {
-          gearman_flush_all(client->impl()->universal);
+          gearman_flush_all(client->universal);
         }
       }
 
       /* See if there are any connections ready for I/O. */
-      while ((client->impl()->con= gearman_ready(client->impl()->universal)))
+      while ((client->con= gearman_ready(client->universal)))
       {
-        if (client->impl()->con->revents & (POLLOUT | POLLERR | POLLHUP | POLLNVAL))
+        if (client->con->revents & (POLLOUT | POLLERR | POLLHUP | POLLNVAL))
         {
           /* Socket is ready for writing, continue submitting jobs. */
-          for (client->impl()->task= client->impl()->task_list; client->impl()->task;
-               client->impl()->task= client->impl()->task->impl()->next)
+          for (client->task= client->task_list; client->task;
+               client->task= client->task->impl()->next)
           {
-            if (client->impl()->task->impl()->con != client->impl()->con or
-                (client->impl()->task->impl()->state != GEARMAN_TASK_STATE_SUBMIT and
-                 client->impl()->task->impl()->state != GEARMAN_TASK_STATE_WORKLOAD))
+            if (client->task->impl()->con != client->con or
+                (client->task->impl()->state != GEARMAN_TASK_STATE_SUBMIT and
+                 client->task->impl()->state != GEARMAN_TASK_STATE_WORKLOAD))
             {
               continue;
             }
 
   case GEARMAN_CLIENT_STATE_SUBMIT:
-            if (client->impl()->task == NULL)
+            if (client->task == NULL)
             {
-              client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+              client->state= GEARMAN_CLIENT_STATE_IDLE;
               break;
             }
-            assert_msg(client == client->impl()->task->impl()->client, "Programmer error, client and task member client are not the same");
-            gearman_return_t local_ret= _client_run_task(client->impl()->task->impl());
+            gearman_return_t local_ret= _client_run_task(client->task->impl());
             if (local_ret == GEARMAN_COULD_NOT_CONNECT)
             {
-              client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+              client->state= GEARMAN_CLIENT_STATE_IDLE;
               return local_ret;
             }
             else if (gearman_failed(local_ret) and local_ret != GEARMAN_IO_WAIT)
             {
-              client->impl()->state= GEARMAN_CLIENT_STATE_SUBMIT;
+              client->state= GEARMAN_CLIENT_STATE_SUBMIT;
               return local_ret;
             }
           }
 
           /* Connection errors are fatal. */
-          if (client->impl()->con->revents & (POLLERR | POLLHUP | POLLNVAL))
+          if (client->con->revents & (POLLERR | POLLHUP | POLLNVAL))
           {
-            gearman_error(client->impl()->universal, GEARMAN_LOST_CONNECTION, "detected lost connection in _client_run_tasks()");
-            client->impl()->con->close_socket();
-            client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+            gearman_error(client->universal, GEARMAN_LOST_CONNECTION, "detected lost connection in _client_run_tasks()");
+            client->con->close_socket();
+            client->state= GEARMAN_CLIENT_STATE_IDLE;
             return GEARMAN_LOST_CONNECTION;
           }
         }
 
-        if ((client->impl()->con->revents & POLLIN) == 0)
+        if ((client->con->revents & POLLIN) == 0)
         {
           continue;
         }
@@ -1495,17 +1504,17 @@
         while (1)
         {
           /* Read packet on connection and find which task it belongs to. */
-          if (client->impl()->options.unbuffered_result)
+          if (client->options.unbuffered_result)
           {
             /* If client is handling the data read, make sure it's complete. */
-            if (client->impl()->con->recv_state == GEARMAN_CON_RECV_STATE_READ_DATA)
+            if (client->con->recv_state == GEARMAN_CON_RECV_STATE_READ_DATA)
             {
-              for (client->impl()->task= client->impl()->task_list; client->impl()->task;
-                   client->impl()->task= client->impl()->task->impl()->next)
+              for (client->task= client->task_list; client->task;
+                   client->task= client->task->impl()->next)
               {
-                if (client->impl()->task->impl()->con == client->impl()->con &&
-                    (client->impl()->task->impl()->state == GEARMAN_TASK_STATE_DATA or
-                     client->impl()->task->impl()->state == GEARMAN_TASK_STATE_COMPLETE))
+                if (client->task->impl()->con == client->con &&
+                    (client->task->impl()->state == GEARMAN_TASK_STATE_DATA or
+                     client->task->impl()->state == GEARMAN_TASK_STATE_COMPLETE))
                 {
                   break;
                 }
@@ -1515,24 +1524,24 @@
                 Someone has set GEARMAN_CLIENT_UNBUFFERED_RESULT but hasn't setup the client to fetch data correctly.
                 Fatal error :(
               */
-              return gearman_universal_set_error(client->impl()->universal, GEARMAN_INVALID_ARGUMENT, GEARMAN_AT,
+              return gearman_universal_set_error(client->universal, GEARMAN_INVALID_ARGUMENT, GEARMAN_AT,
                                                  "client created with GEARMAN_CLIENT_UNBUFFERED_RESULT, but was not setup to use it. %s", __func__);
             }
             else
             {
               /* Read the next packet, without buffering the data part. */
-              client->impl()->task= NULL;
-              (void)client->impl()->con->receiving(client->impl()->con->_packet, ret, false);
+              client->task= NULL;
+              (void)client->con->receiving(client->con->_packet, ret, false);
             }
           }
           else
           {
             /* Read the next packet, buffering the data part. */
-            client->impl()->task= NULL;
-            (void)client->impl()->con->receiving(client->impl()->con->_packet, ret, true);
+            client->task= NULL;
+            (void)client->con->receiving(client->con->_packet, ret, true);
           }
 
-          if (client->impl()->task == NULL)
+          if (client->task == NULL)
           {
             assert(ret != GEARMAN_MAX_RETURN);
 
@@ -1544,71 +1553,71 @@
                 break;
               }
 
-              client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+              client->state= GEARMAN_CLIENT_STATE_IDLE;
               return ret;
             }
 
-            client->impl()->con->options.packet_in_use= true;
+            client->con->options.packet_in_use= true;
 
             /* We have a packet, see which task it belongs to. */
-            for (client->impl()->task= client->impl()->task_list; client->impl()->task;
-                 client->impl()->task= client->impl()->task->impl()->next)
+            for (client->task= client->task_list; client->task;
+                 client->task= client->task->impl()->next)
             {
-              if (client->impl()->task->impl()->con != client->impl()->con)
+              if (client->task->impl()->con != client->con)
               {
                 continue;
               }
 
-              if (client->impl()->con->_packet.command == GEARMAN_COMMAND_JOB_CREATED)
+              if (client->con->_packet.command == GEARMAN_COMMAND_JOB_CREATED)
               {
-                if (client->impl()->task->impl()->created_id != client->impl()->con->created_id)
+                if (client->task->impl()->created_id != client->con->created_id)
                 {
                   continue;
                 }
 
                 /* New job created, drop through below and notify task. */
-                client->impl()->con->created_id++;
+                client->con->created_id++;
               }
-              else if (client->impl()->con->_packet.command == GEARMAN_COMMAND_ERROR)
+              else if (client->con->_packet.command == GEARMAN_COMMAND_ERROR)
               {
-                gearman_return_t maybe_server_error= string2return_code(static_cast<char *>(client->impl()->con->_packet.arg[0]), int(client->impl()->con->_packet.arg_size[0]));
+                gearman_return_t maybe_server_error= string2return_code(static_cast<char *>(client->con->_packet.arg[0]), int(client->con->_packet.arg_size[0]));
 
                 if (maybe_server_error == GEARMAN_MAX_RETURN)
                 {
                   maybe_server_error= GEARMAN_SERVER_ERROR;
                 }
 
-                gearman_universal_set_error(client->impl()->universal, maybe_server_error, GEARMAN_AT,
+                gearman_universal_set_error(client->universal, maybe_server_error, GEARMAN_AT,
                                             "%s:%.*s",
-                                            static_cast<char *>(client->impl()->con->_packet.arg[0]),
-                                            int(client->impl()->con->_packet.arg_size[1]),
-                                            static_cast<char *>(client->impl()->con->_packet.arg[1]));
+                                            static_cast<char *>(client->con->_packet.arg[0]),
+                                            int(client->con->_packet.arg_size[1]),
+                                            static_cast<char *>(client->con->_packet.arg[1]));
 
                 /* 
                   Packet cleanup copied from "Clean up the packet" below, and must
                   remain in sync with its reference.
                 */
-                gearman_packet_free(&(client->impl()->con->_packet));
-                client->impl()->con->options.packet_in_use= false;
+                gearman_packet_free(&(client->con->_packet));
+                client->con->options.packet_in_use= false;
 
                 /* This step copied from _client_run_tasks() above: */
                 /* Increment this value because new job created then failed. */
-                client->impl()->con->created_id++;
+                client->con->created_id++;
 
                 return maybe_server_error;
               }
-              else if (client->impl()->con->_packet.command == GEARMAN_COMMAND_STATUS_RES_UNIQUE and
-                       (strncmp(gearman_task_unique(client->impl()->task),
-                               static_cast<char *>(client->impl()->con->_packet.arg[0]),
-                               client->impl()->con->_packet.arg_size[0]) == 0))
+              else if (client->con->_packet.command == GEARMAN_COMMAND_STATUS_RES_UNIQUE and
+                       (strncmp(gearman_task_unique(client->task),
+                               static_cast<char *>(client->con->_packet.arg[0]),
+                               client->con->_packet.arg_size[0]) == 0))
               { }
-              else if (strncmp(client->impl()->task->impl()->job_handle,
-                               static_cast<char *>(client->impl()->con->_packet.arg[0]),
-                               client->impl()->con->_packet.arg_size[0]) ||
-                       (client->impl()->con->_packet.failed() == false &&
-                        strlen(client->impl()->task->impl()->job_handle) != client->impl()->con->_packet.arg_size[0] - 1) ||
-                       (client->impl()->con->_packet.failed() &&
-                        strlen(client->impl()->task->impl()->job_handle) != client->impl()->con->_packet.arg_size[0]))
+              else if (strncmp(client->task->impl()->job_handle,
+                               static_cast<char *>(client->con->_packet.arg[0]),
+                               client->con->_packet.arg_size[0]) ||
+                       (client->con->_packet.failed() == false &&
+                        strlen(client->task->impl()->job_handle) != client->con->_packet.arg_size[0] - 1) ||
+                       (client->con->_packet.failed() &&
+                        strlen(client->task->impl()->job_handle) != client->con->_packet.arg_size[0]))
               {
                 continue;
               }
@@ -1618,20 +1627,19 @@
               break;
             }
 
-            if (client->impl()->task == NULL)
+            if (client->task == NULL)
             {
               /* The client has stopped waiting for the response, ignore it. */
-              client->impl()->con->free_private_packet();
+              client->con->free_private_packet();
               continue;
             }
 
-            client->impl()->task->impl()->recv= &(client->impl()->con->_packet);
+            client->task->impl()->recv= &(client->con->_packet);
           }
 
   case GEARMAN_CLIENT_STATE_PACKET:
           /* Let task process job created or result packet. */
-          assert_msg(client == client->impl()->task->impl()->client, "Programmer error, client and task member client are not the same");
-          gearman_return_t local_ret= _client_run_task(client->impl()->task->impl());
+          gearman_return_t local_ret= _client_run_task(client->task->impl());
           if (local_ret == GEARMAN_IO_WAIT)
           {
             break;
@@ -1639,56 +1647,56 @@
 
           if (gearman_failed(local_ret))
           {
-            client->impl()->state= GEARMAN_CLIENT_STATE_PACKET;
+            client->state= GEARMAN_CLIENT_STATE_PACKET;
             return local_ret;
           }
 
           /* Clean up the packet. */
-          client->impl()->con->free_private_packet();
+          client->con->free_private_packet();
 
           /* If exit task is set and matched, exit */
           if (exit_task)
           {
             if (exit_task->impl()->result_rc != GEARMAN_UNKNOWN_STATE)
             {
-              client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+              client->state= GEARMAN_CLIENT_STATE_IDLE;
               return GEARMAN_SUCCESS;
             }
           }
 
           /* If all tasks are done, return. */
-          if (client->impl()->running_tasks == 0)
+          if (client->running_tasks == 0)
           {
-            client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+            client->state= GEARMAN_CLIENT_STATE_IDLE;
             return GEARMAN_SUCCESS;
           }
         }
       }
 
       /* If all tasks are done, return. */
-      if (client->impl()->running_tasks == 0)
+      if (client->running_tasks == 0)
       {
         break;
       }
 
-      if (client->impl()->new_tasks > 0 and ! (client->impl()->options.no_new))
+      if (client->new_tasks > 0 and ! (client->options.no_new))
       {
         continue;
       }
 
-      if (client->impl()->options.non_blocking)
+      if (client->options.non_blocking)
       {
         /* Let the caller wait for activity. */
-        client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+        client->state= GEARMAN_CLIENT_STATE_IDLE;
 
-        return gearman_gerror(client->impl()->universal, GEARMAN_IO_WAIT);
+        return gearman_gerror(client->universal, GEARMAN_IO_WAIT);
       }
 
       /* Wait for activity on one of the connections. */
-      gearman_return_t local_ret= gearman_wait(client->impl()->universal);
+      gearman_return_t local_ret= gearman_wait(client->universal);
       if (gearman_failed(local_ret) and local_ret != GEARMAN_IO_WAIT)
       {
-        client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+        client->state= GEARMAN_CLIENT_STATE_IDLE;
 
         return local_ret;
       }
@@ -1697,73 +1705,76 @@
     break;
   }
 
-  client->impl()->state= GEARMAN_CLIENT_STATE_IDLE;
+  client->state= GEARMAN_CLIENT_STATE_IDLE;
 
   return GEARMAN_SUCCESS;
 }
 
-gearman_return_t gearman_client_run_tasks(gearman_client_st *client)
-{
-  if (client == NULL or client->impl() == NULL)
-  {
-    return GEARMAN_INVALID_ARGUMENT;
-  }
-
-  if (client->impl()->task_list == NULL) // We are immediatly successful if all tasks are completed
-  {
-    return GEARMAN_SUCCESS;
-  }
-
-  gearman_return_t rc;
-  {
-    PUSH_NON_BLOCKING(client->impl()->universal);
-
-    rc= _client_run_tasks(client, NULL);
-  }
-
-  if (rc == GEARMAN_COULD_NOT_CONNECT)
-  {
-    gearman_reset(client->impl()->universal);
-  }
-
-  return rc;
-}
-
-gearman_return_t gearman_client_run_block_tasks(gearman_client_st *shell, gearman_task_st* exit_task)
-{
-  if (shell == NULL)
-  {
-    return GEARMAN_INVALID_ARGUMENT;
-  }
-  Client *client= shell->impl();
-
-  if (client->task_list == NULL) // We are immediatly successful if all tasks are completed
-  {
-    return GEARMAN_SUCCESS;
-  }
-
-
-  gearman_return_t rc;
-  {
-    PUSH_BLOCKING(client->universal);
-
-    rc= _client_run_tasks(shell, exit_task);
-  }
-
-  if (gearman_failed(rc))
-  {
+gearman_return_t gearman_client_run_tasks(gearman_client_st *client_shell)
+{
+  if (client_shell and client_shell->impl())
+  {
+    Client* client= client_shell->impl();
+
+    if (client->task_list == NULL) // We are immediatly successful if all tasks are completed
+    {
+      return GEARMAN_SUCCESS;
+    }
+
+    gearman_return_t rc;
+    {
+      PUSH_NON_BLOCKING(client->universal);
+
+      rc= _client_run_tasks(client_shell, NULL);
+    }
+
     if (rc == GEARMAN_COULD_NOT_CONNECT)
     {
       gearman_reset(client->universal);
     }
 
-    if (client->universal.error_code() != rc and rc != GEARMAN_COULD_NOT_CONNECT)
-    {
-      assert(client->universal.error_code() == rc);
-    }
-  }
-
-  return rc;
+    return rc;
+  }
+
+  return GEARMAN_INVALID_ARGUMENT;
+}
+
+gearman_return_t gearman_client_run_block_tasks(gearman_client_st *client_shell, gearman_task_st* exit_task)
+{
+  if (client_shell and client_shell->impl())
+  {
+    Client *client= client_shell->impl();
+
+    if (client->task_list == NULL) // We are immediatly successful if all tasks are completed
+    {
+      return GEARMAN_SUCCESS;
+    }
+
+
+    gearman_return_t rc;
+    {
+      PUSH_BLOCKING(client->universal);
+
+      rc= _client_run_tasks(client_shell, exit_task);
+    }
+
+    if (gearman_failed(rc))
+    {
+      if (rc == GEARMAN_COULD_NOT_CONNECT)
+      {
+        gearman_reset(client->universal);
+      }
+
+      if (client->universal.error_code() != rc and rc != GEARMAN_COULD_NOT_CONNECT)
+      {
+        assert(client->universal.error_code() == rc);
+      }
+    }
+
+    return rc;
+  }
+
+  return GEARMAN_INVALID_ARGUMENT;
 }
 
 /*
 
=== modified file 'libgearman/interface/task.hpp'
--- libgearman/interface/task.hpp	2013-07-06 01:22:31 +0000
+++ libgearman/interface/task.hpp	2013-07-06 19:18:28 +0000
@@ -2,7 +2,7 @@
  * 
  *  Gearmand client and server library.
  *
- *  Copyright (C) 2012 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2012-2013 Data Differential, http://datadifferential.com/
  *  All rights reserved.
  *
  *  Redistribution and use in source and binary forms, with or without
@@ -69,7 +69,7 @@
   uint32_t numerator;
   uint32_t denominator;
   uint32_t client_count;
-  gearman_client_st *client;
+  Client *client;
   gearman_task_st *next;
   gearman_task_st *prev;
   void *context;
@@ -97,7 +97,7 @@
     numerator(0),
     denominator(0),
     client_count(0),
-    client(&client_),
+    client(client_.impl()),
     next(NULL),
     prev(NULL),
     context(NULL),
 
=== modified file 'libgearman/run.cc'
--- libgearman/run.cc	2013-07-06 17:00:51 +0000
+++ libgearman/run.cc	2013-07-06 19:18:28 +0000
@@ -51,7 +51,7 @@
   assert_msg(task->client, "Programmer error, somehow an invalid task was specified");
   if (task->client == NULL)
   {
-    return gearman_universal_set_error(task->client->impl()->universal, GEARMAN_INVALID_ARGUMENT, GEARMAN_AT,
+    return gearman_universal_set_error(task->client->universal, GEARMAN_INVALID_ARGUMENT, GEARMAN_AT,
                                        "Programmer error, somehow an invalid task was specified");
   }
 
@@ -59,16 +59,16 @@
   {
   case GEARMAN_TASK_STATE_NEW:
     
-    if (task->client->impl()->universal.has_connections() == false)
+    if (task->client->universal.has_connections() == false)
     {
-      assert(task->client->impl()->universal.con_count == 0);
-      assert(task->client->impl()->universal.con_list == NULL);
-      task->client->impl()->new_tasks--;
-      task->client->impl()->running_tasks--;
-      return gearman_universal_set_error(task->client->impl()->universal, GEARMAN_NO_SERVERS, GEARMAN_AT, "no servers provided");
+      assert(task->client->universal.con_count == 0);
+      assert(task->client->universal.con_list == NULL);
+      task->client->new_tasks--;
+      task->client->running_tasks--;
+      return gearman_universal_set_error(task->client->universal, GEARMAN_NO_SERVERS, GEARMAN_AT, "no servers provided");
     }
 
-    for (task->con= task->client->impl()->universal.con_list; task->con;
+    for (task->con= task->client->universal.con_list; task->con;
          task->con= task->con->next_connection())
     {
       if (task->con->send_state == GEARMAN_CON_SEND_STATE_NONE)
@@ -79,11 +79,11 @@
 
     if (task->con == NULL)
     {
-      task->client->impl()->options.no_new= true;
-      return gearman_gerror(task->client->impl()->universal, GEARMAN_IO_WAIT);
+      task->client->options.no_new= true;
+      return gearman_gerror(task->client->universal, GEARMAN_IO_WAIT);
     }
 
-    task->client->impl()->new_tasks--;
+    task->client->new_tasks--;
 
     if (task->send.command != GEARMAN_COMMAND_GET_STATUS)
     {
@@ -95,7 +95,7 @@
     while (1)
     {
       assert(task->con);
-      gearman_return_t ret= task->con->send_packet(task->send, task->client->impl()->new_tasks == 0 ? true : false);
+      gearman_return_t ret= task->con->send_packet(task->send, task->client->new_tasks == 0 ? true : false);
 
       if (gearman_success(ret))
       {
@@ -134,13 +134,23 @@
 
           if (ret == GEARMAN_COULD_NOT_CONNECT) // If no connection is found, we will let the user try again
           {
+<<<<<<< TREE
             task->set_state(GEARMAN_TASK_STATE_NEW);
             task->client->impl()->new_tasks++;
+=======
+            task->set_state(GEARMAN_TASK_STATE_NEW);
+            task->client->new_tasks++;
+>>>>>>> MERGE-SOURCE
           }
           else
           {
+<<<<<<< TREE
             task->set_state(GEARMAN_TASK_STATE_FAIL);
             task->client->impl()->running_tasks--;
+=======
+            task->set_state(GEARMAN_TASK_STATE_FAIL);
+            task->client->running_tasks--;
+>>>>>>> MERGE-SOURCE
           }
           return ret;
         }
@@ -157,7 +167,7 @@
     {
       if (not task->func.workload_fn)
       {
-        gearman_error(task->client->impl()->universal, GEARMAN_NEED_WORKLOAD_FN,
+        gearman_error(task->client->universal, GEARMAN_NEED_WORKLOAD_FN,
                       "workload size > 0, but no data pointer or workload_fn was given");
         return GEARMAN_NEED_WORKLOAD_FN;
       }
@@ -171,8 +181,13 @@
       }
     }
 
+<<<<<<< TREE
     task->client->impl()->options.no_new= false;
     task->set_state(GEARMAN_TASK_STATE_WORK);
+=======
+    task->client->options.no_new= false;
+    task->set_state(GEARMAN_TASK_STATE_WORK);
+>>>>>>> MERGE-SOURCE
     task->con->set_events(POLLIN);
     return GEARMAN_SUCCESS;
 
@@ -407,6 +422,7 @@
     break;
   }
 
+<<<<<<< TREE
   task->client->impl()->running_tasks--;
   task->set_state(GEARMAN_TASK_STATE_FINISHED);
 
@@ -414,6 +430,14 @@
   assert(task->result_rc != GEARMAN_UNKNOWN_STATE);
 
   if (task->client->impl()->options.free_tasks and task->type == GEARMAN_TASK_KIND_ADD_TASK)
+=======
+  task->client->running_tasks--;
+  task->set_state(GEARMAN_TASK_STATE_FINISHED);
+
+  assert(task->result_rc != GEARMAN_UNKNOWN_STATE);
+
+  if (task->client->options.free_tasks and task->type == GEARMAN_TASK_KIND_ADD_TASK)
+>>>>>>> MERGE-SOURCE
   {
     gearman_task_free(task->shell());
   }
 
=== modified file 'libgearman/task.cc'
--- libgearman/task.cc	2013-05-11 11:15:03 +0000
+++ libgearman/task.cc	2013-07-06 19:18:28 +0000
@@ -59,7 +59,6 @@
   Task* task= new (std::nothrow) Task(client, task_shell);
   if (task)
   {
-    assert(task->client == &client);
     return task->shell();
   }
 
@@ -69,6 +68,10 @@
   return NULL;
 }
 
+void gearman_task_free(Task* task)
+{
+  gearman_task_free(task->shell());
+}
 
 void gearman_task_free(gearman_task_st *task_shell)
 {
@@ -91,14 +94,14 @@
             gearman_packet_free(&(task->send));
           }
 
-          if (task->type != GEARMAN_TASK_KIND_DO  and task->context and  task->client->impl()->task_context_free_fn)
+          if (task->type != GEARMAN_TASK_KIND_DO  and task->context and  task->client->task_context_free_fn)
           {
-            task->client->impl()->task_context_free_fn(task_shell, static_cast<void *>(task->context));
+            task->client->task_context_free_fn(task_shell, static_cast<void *>(task->context));
           }
 
-          if (task->client->impl()->task_list == task_shell)
+          if (task->client->task_list == task_shell)
           {
-            task->client->impl()->task_list= task->next;
+            task->client->task_list= task->next;
           }
 
           if (task->prev)
@@ -111,13 +114,13 @@
             task->next->impl()->prev= task->prev;
           }
 
-          task->client->impl()->task_count--;
+          task->client->task_count--;
 
           // If the task we are removing is a current task, remove it from the client
           // structures.
-          if (task->client->impl()->task == task_shell)
+          if (task->client->task == task_shell)
           {
-            task->client->impl()->task= NULL;
+            task->client->task= NULL;
           }
           task->client= NULL;
         }
 
=== modified file 'libgearman/task.hpp'
--- libgearman/task.hpp	2013-01-31 05:27:46 +0000
+++ libgearman/task.hpp	2013-07-06 19:18:28 +0000
@@ -49,6 +49,8 @@
 gearman_task_st *gearman_task_internal_create(gearman_client_st& client,
                                               gearman_task_st *task);
 
+void gearman_task_free(Task* task);
+
 void gearman_task_clear_fn(gearman_task_st *task);
 
 bool gearman_task_is_active(const gearman_task_st *self);