Fix for lp:1098413

libgearman-1.0/return.h

@@ -94,6 +94,7 @@ enum gearman_return_t
   GEARMAN_ARGUMENT_TOO_LARGE,
   GEARMAN_INVALID_ARGUMENT,
   GEARMAN_IN_PROGRESS, // See gearman_client_job_status()
+  GEARMAN_INVALID_SERVER_OPTION, // Bad server option sent to server
   GEARMAN_MAX_RETURN, /* Always add new error code before */
   GEARMAN_FAIL= GEARMAN_WORK_FAIL,
   GEARMAN_FATAL= GEARMAN_WORK_FAIL,

libgearman-1.0/universal.h

@@ -46,6 +46,7 @@
 typedef struct gearman_connection_st gearman_connection_st;
 typedef gearman_return_t (gearman_event_watch_fn)(gearman_connection_st *con,
                                                   short events, void *context);
+typedef struct gearman_server_options_st gearman_server_options_st;
 
 /**
  * @ingroup gearman_universal
@@ -63,6 +64,7 @@ struct gearman_universal_st
   uint32_t sending;
   int timeout; // Connection timeout.
   gearman_connection_st *con_list;
+  gearman_server_options_st *server_options_list;
   gearman_packet_st *packet_list;
   struct pollfd *pfds;
   gearman_log_fn *log_fn;

libgearman/common.h

@@ -67,6 +67,7 @@ struct gearman_result_st;
 #include "libgearman/task.hpp"
 #include "libgearman/job.h"
 #include "libgearman/client.h"
+#include "libgearman/server_options.hpp"
 
 #include "libgearman/status.hpp"
 

libgearman/connection.cc

@@ -141,6 +141,7 @@ gearman_connection_st::gearman_connection_st(gearman_universal_st &universal_arg
 {
   options.ready= false;
   options.packet_in_use= false;
+  options.server_options_sent= false;
 
   if (options_args)
   {
@@ -321,6 +322,8 @@ void gearman_connection_st::close_socket()
   recv_buffer_ptr= recv_buffer;
   recv_buffer_size= 0;
 
+  options.server_options_sent= false;
+
   // created_id_next is incremented for every outbound packet (except status).
   // created_id is incremented for every response packet received, and also when
   // no packets are received due to an error. There are lots of such error paths
@@ -350,7 +353,83 @@ void gearman_connection_st::reset_addrinfo()
   addrinfo_next= NULL;
 }
 
+
+/*
+ * The send_packet() method does not only send the passed-in packet_arg. If there are any server options
+ * established, and they haven't yet been sent over, then these options are sent over first.
+ * Only if that succeeds is the packet_arg sent.
+ * The reason for this is server options are only set once by the client/worker. In the older code, this
+ * resulted in them being sent over exactly once. If the connection was dropped and rebuilt, then the options
+ * were not sent over again, rendering them moot. This way, we're guaranteed that the options are always sent
+ * at least once to a connected server.
+ */
 gearman_return_t gearman_connection_st::send_packet(const gearman_packet_st& packet_arg, const bool flush_buffer)
+{
+  if (!options.server_options_sent)
+  {
+    for (gearman_server_options_st* head= universal.server_options_list;
+         head;
+         head= head->next)
+    {
+      gearman_packet_st message;
+      const void *args[]= { head->option };
+      size_t args_size[]= { head->option_length };
+      gearman_return_t ret= gearman_packet_create_args(universal, message, GEARMAN_MAGIC_REQUEST,
+                                                       GEARMAN_COMMAND_OPTION_REQ, args, args_size, 1);
+
+      if (gearman_failed(ret))
+      {
+        gearman_packet_free(&message);
+        gearman_error(universal, GEARMAN_MEMORY_ALLOCATION_FAILURE, "gearman_packet_create_args()");
+        return GEARMAN_MEMORY_ALLOCATION_FAILURE;
+      }
+
+      PUSH_BLOCKING(universal);
+      OptionCheck check(universal);
+      ret= _send_packet(message, true);
+      if (gearman_failed(ret))
+      {
+        gearman_packet_free(&message);
+        gearman_error(universal, ret, "Failed to send server-options packet");
+        return ret;
+      }
+
+      options.packet_in_use= true;
+      gearman_packet_st *packet_ptr= receiving(_packet, ret, true);
+      if (packet_ptr == NULL)
+      {
+        gearman_packet_free(&message);
+        options.packet_in_use= false;
+        gearman_error(universal, ret, "Failed in receiving()");
+        return ret;
+      }
+
+      if (gearman_failed(ret) ||
+          gearman_failed(ret= check.success(this)))
+      {
+        gearman_packet_free(&message);
+        free_private_packet();
+        reset_recv_packet();
+        gearman_error(universal, ret, "receiving()");
+        return ret;
+      }
+
+      free_private_packet();
+      reset_recv_packet();
+      gearman_packet_free(&message);
+    }
+
+    options.server_options_sent= true;
+  }
+
+  return _send_packet(packet_arg, flush_buffer);
+}
+
+/*
+ * This is the real implementation that actually sends a packet. Read the comments for send_packet() for why
+ * that is. Note that this is a private method. External callers should only call send_packet().
+ */
+gearman_return_t gearman_connection_st::_send_packet(const gearman_packet_st& packet_arg, const bool flush_buffer)
 {
   switch (send_state)
   {

libgearman/connection.hpp

@@ -43,6 +43,7 @@
 struct gearman_connection_st
 {
   struct {
+    bool server_options_sent;
     bool ready;
     bool packet_in_use;
   } options;
@@ -122,6 +123,7 @@ public:
   }
 
 private:
+  gearman_return_t _send_packet(const gearman_packet_st&, const bool flush_buffer);
   gearman_return_t set_socket_options();
   size_t recv_socket(void *data, size_t data_size, gearman_return_t&);
   gearman_return_t connect_poll();

libgearman/function/function_v2.cc

@@ -118,6 +118,7 @@ gearman_function_error_t FunctionV2::callback(gearman_job_st* job, void *context
   case GEARMAN_ARGUMENT_TOO_LARGE:
   case GEARMAN_INVALID_ARGUMENT:
   case GEARMAN_IN_PROGRESS:
+  case GEARMAN_INVALID_SERVER_OPTION:
   case GEARMAN_MAX_RETURN:
     break;
   }

libgearman/function/partition.cc

@@ -118,6 +118,7 @@ gearman_function_error_t Partition::callback(gearman_job_st* job, void *context_
   case GEARMAN_ARGUMENT_TOO_LARGE:
   case GEARMAN_INVALID_ARGUMENT:
   case GEARMAN_IN_PROGRESS:
+  case GEARMAN_INVALID_SERVER_OPTION:
   case GEARMAN_MAX_RETURN:
     break;
   }

libgearman/include.am

@@ -47,6 +47,7 @@ noinst_HEADERS+= \
 		 libgearman/task.hpp \
 		 libgearman/task_attr.h \
 		 libgearman/unique.hpp \
+		 libgearman/server_options.hpp \
 		 libgearman/universal.hpp \
 		 libgearman/vector.hpp
 
@@ -63,6 +64,7 @@ libgearman_libgearmancore_la_SOURCES= \
 				      libgearman/packet.cc \
 				      libgearman/strcommand.cc \
 				      libgearman/strerror.cc \
+				      libgearman/server_options.cc \
 				      libgearman/universal.cc \
 				      libgearman/vector.cc
 
@@ -110,6 +112,7 @@ libgearman_libgearman_la_SOURCES+= \
 				  libgearman/task.cc \
 				  libgearman/task_attr.cc \
 				  libgearman/unique.cc \
+				  libgearman/server_options.cc \
 				  libgearman/universal.cc \
 				  libgearman/vector.cc \
 				  libgearman/worker.cc

libgearman/server_options.cc

@@ -0,0 +1,86 @@
+/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ * 
+ *  Gearmand client and server library.
+ *
+ *  Copyright (C) 2013 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2013 Keyur Govande
+ *  All rights reserved.
+ *
+ *  Redistribution and use in source and binary forms, with or without
+ *  modification, are permitted provided that the following conditions are
+ *  met:
+ *
+ *      * Redistributions of source code must retain the above copyright
+ *  notice, this list of conditions and the following disclaimer.
+ *
+ *      * Redistributions in binary form must reproduce the above
+ *  copyright notice, this list of conditions and the following disclaimer
+ *  in the documentation and/or other materials provided with the
+ *  distribution.
+ *
+ *      * The names of its contributors may not be used to endorse or
+ *  promote products derived from this software without specific prior
+ *  written permission.
+ *
+ *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/**
+ * @file
+ * @brief Server options
+ */
+
+#include "gear_config.h"
+#include <libgearman/common.h>
+
+#include "libgearman/server_options.hpp"
+
+gearman_server_options_st::gearman_server_options_st(gearman_universal_st &universal_arg,
+                                                     const char* option_arg, const size_t option_arg_size) : 
+  next(NULL), prev(NULL), option(option_arg), 
+  option_length(option_arg_size),
+  universal(universal_arg)
+{
+  if (universal.server_options_list)
+  {
+    universal.server_options_list->prev= this;
+  }
+  next= universal.server_options_list;
+  universal.server_options_list= this;
+}
+
+gearman_server_options_st::~gearman_server_options_st()
+{
+  if (option)
+  {
+    free((void*)option);
+  }
+
+  { // Remove from universal list
+    if (universal.server_options_list == this)
+    {
+      universal.server_options_list= next;
+    }
+
+    if (prev)
+    {
+      prev->next= next;
+    }
+
+    if (next)
+    {
+      next->prev= prev;
+    }
+  }
+}

libgearman/server_options.hpp

@@ -0,0 +1,57 @@
+/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ * 
+ *  Gearmand client and server library.
+ *
+ *  Copyright (C) 2013 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2013 Keyur Govande
+ *  All rights reserved.
+ *
+ *  Redistribution and use in source and binary forms, with or without
+ *  modification, are permitted provided that the following conditions are
+ *  met:
+ *
+ *      * Redistributions of source code must retain the above copyright
+ *  notice, this list of conditions and the following disclaimer.
+ *
+ *      * Redistributions in binary form must reproduce the above
+ *  copyright notice, this list of conditions and the following disclaimer
+ *  in the documentation and/or other materials provided with the
+ *  distribution.
+ *
+ *      * The names of its contributors may not be used to endorse or
+ *  promote products derived from this software without specific prior
+ *  written permission.
+ *
+ *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#pragma once
+
+#include <cstdlib>
+
+struct gearman_universal_st;
+
+struct gearman_server_options_st
+{
+  gearman_server_options_st *next;
+  gearman_server_options_st *prev;
+  const char* option;
+  const size_t option_length;
+  gearman_universal_st &universal;
+
+public:
+  gearman_server_options_st(gearman_universal_st &universal_arg,
+                            const char* option_arg, const size_t option_arg_size);
+  ~gearman_server_options_st();
+};