Merge lp:~sven-nierlein/gearmand/gearmand into lp:gearmand

[Sven Nierlein]

added retention queue plugin

The retention queue plugin will save the queue to disk on shutdown. Therefor
it does not slow down the gearmand during runtime and still preserves the
queue during restarts.

[Brian Aker (brianaker)]

Sorry, just looking at this now.

[Sven Nierlein (sven-nierlein)]

so what do you think? I noticed the status changed to "merged" but i cannot find these changes in gearmand yet.

[Brian Aker (brianaker)]

Hi,

What I did to test this was to extend the sqlite plugin to record on shutdown to see how this works out. In testing,... so far so good.

Look in lp:gearmand

Cheers,
 -Bian

On Nov 25, 2012, at 1:46 AM, Sven Nierlein <email address hidden> wrote:

> so what do you think? I noticed the status changed to "merged" but i cannot find these changes in gearmand yet.
> --
> https://code.launchpad.net/~sven-nierlein/gearmand/gearmand/+merge/133809
> Your team Tangent is subscribed to branch lp:gearmand.

[Sven Nierlein (sven-nierlein)]

I can't get this working:
when i start gearmand with either this:
-q libsqlite3 --libsqlite3-db=/tmp/gearmand.sqlite --libsqlite3-store-on-shutdown
then it stores all jobs in sqlite3, not only on shutdown. This is exactly what i wanted
to avoid.

When i start gearmand with:
--libsqlite3-db=/tmp/gearmand.sqlite --libsqlite3-store-on-shutdown
only, then nothing is stored, not even on shutdown.

Besides that, this adds a dependency to sqlite3 which wasn't necessary with my merge proposal.

[Brian Aker (brianaker)]

Hi,

On Nov 25, 2012, at 3:28 AM, Sven Nierlein <email address hidden> wrote:

> I can't get this working:
> when i start gearmand with either this:
> -q libsqlite3 --libsqlite3-db=/tmp/gearmand.sqlite --libsqlite3-store-on-shutdown
> then it stores all jobs in sqlite3, not only on shutdown. This is exactly what i wanted
> to avoid.

Let me see what is happening with it then. The test case should be covering this problem.

> Besides that, this adds a dependency to sqlite3 which wasn't necessary with my merge proposal.

The creation of a new queue plugin in separate from adding the ability for a queue to store on shutdown.

Which brings up the question of what do you want to achieve with a new queue type?

Cheers,
 -Brian

[Sven Nierlein (sven-nierlein)]

The intention of the new queue type is to create a queue plugin, which adds no overhead
during runtime and just stores the queue to disk on shutdown.

I am "sni" in the freenode #gearman channel if you want to discuss details.

[Brian Aker (brianaker)]

Hi!

https://bugs.launchpad.net/gearmand/+bug/1087654

I've created a bug to track this, and I have linked a tree that provides a test case.

The test case creates eight jobs and then checks to before shutting down the server that sqlite is empty. The test then shuts down the server and checks to see if they jobs are there. It then starts up the server and verifies that the jobs are processed.

Cheers,
 -Brian

On Nov 25, 2012, at 11:40 PM, Brian Aker <email address hidden> wrote:

> Hi,
>
> On Nov 25, 2012, at 3:28 AM, Sven Nierlein <email address hidden> wrote:
>
>> I can't get this working:
>> when i start gearmand with either this:
>> -q libsqlite3 --libsqlite3-db=/tmp/gearmand.sqlite --libsqlite3-store-on-shutdown
>> then it stores all jobs in sqlite3, not only on shutdown. This is exactly what i wanted
>> to avoid.
>
> Let me see what is happening with it then. The test case should be covering this problem.
>
>> Besides that, this adds a dependency to sqlite3 which wasn't necessary with my merge proposal.
>
> The creation of a new queue plugin in separate from adding the ability for a queue to store on shutdown.
>
> Which brings up the question of what do you want to achieve with a new queue type?
>
> Cheers,
> -Brian
>
>
> --
> https://code.launchpad.net/~sven-nierlein/gearmand/gearmand/+merge/133809
> Your team Tangent is subscribed to branch lp:gearmand.

=== modified file 'libgearman-server/constants.h'
--- libgearman-server/constants.h	2012-08-18 06:26:36 +0000
+++ libgearman-server/constants.h	2012-11-11 01:10:27 +0000
@@ -155,6 +155,8 @@
                                                    void *context,
                                                    gearman_queue_add_fn *add_fn,
                                                    void *add_context);
+typedef gearmand_error_t (gearman_queue_shutdown_fn)(gearman_server_st *server,
+                                                   void *context);
 
 typedef gearmand_error_t (gearmand_connection_add_fn)(gearman_server_con_st *con);
 

=== modified file 'libgearman-server/gearmand.cc'
--- libgearman-server/gearmand.cc	2012-10-29 07:06:01 +0000
+++ libgearman-server/gearmand.cc	2012-11-11 01:10:27 +0000
@@ -1,5 +1,5 @@
 /*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
- * 
+ *
  *  Gearmand client and server library.
  *
  *  Copyright (C) 2011-2012 Data Differential, http://datadifferential.com/
@@ -100,6 +100,10 @@
   /* All threads should be cleaned up before calling this. */
   assert(server.thread_list == NULL);
 
+  gearmand_debug("shutdown queue: begin");
+  gearman_server_queue_shutdown(server);
+  gearmand_debug("shutdown queue: end");
+
   for (uint32_t key= 0; key < GEARMAND_JOB_HASH_SIZE; key++)
   {
     while (server.job_hash[key] != NULL)
@@ -335,7 +339,7 @@
   /* Initialize server components. */
   if (gearmand->base == NULL)
   {
-    gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, "Starting up, verbose set to %s", 
+    gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, "Starting up, verbose set to %s",
                       gearmand_verbose_name(gearmand->verbose));
 
     if (gearmand->threads > 0)
@@ -425,7 +429,7 @@
     }
     else
     {
-      gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, 
+      gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM,
                          "gearmand_wakeup() incorrectly wrote %lu bytes of data.", (unsigned long)written);
     }
   }
@@ -587,7 +591,7 @@
         }
 
         // We are in single user threads, so strerror() is fine.
-        gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Retrying bind(%s) on %s:%s %u >= %u", 
+        gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "Retrying bind(%s) on %s:%s %u >= %u",
                            strerror(errno), host, port->port,
                            waited, bind_timeout);
         this_wait= retry * retry / 3 + 1;
@@ -722,7 +726,7 @@
     {
       for (uint32_t y= 0; y < gearmand->port_list[x].listen_count; y++)
       {
-        gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, 
+        gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM,
                           "Clearing event for listening socket (%d)",
                           gearmand->port_list[x].listen_fd[y]);
 
@@ -768,7 +772,7 @@
     return;
   }
 
-  /* 
+  /*
     Since this is numeric, it should never fail. Even if it did we don't want to really error from it.
   */
   char host[NI_MAXHOST];
@@ -922,7 +926,7 @@
         gearmand_debug("Received SHUTDOWN_GRACEFUL wakeup event");
         _listen_close(gearmand);
 
-        for (gearmand_thread_st* thread= gearmand->thread_list; 
+        for (gearmand_thread_st* thread= gearmand->thread_list;
              thread != NULL;
              thread= thread->next)
         {
@@ -1081,7 +1085,7 @@
   return success;
 }
 
-static bool gearman_server_create(gearman_server_st& server, 
+static bool gearman_server_create(gearman_server_st& server,
                                   uint8_t job_retries_arg,
                                   uint8_t worker_wakeup_arg,
                                   bool round_robin_arg)

=== modified file 'libgearman-server/plugins.cc'
--- libgearman-server/plugins.cc	2012-10-28 20:04:35 +0000
+++ libgearman-server/plugins.cc	2012-11-11 01:10:27 +0000
@@ -50,6 +50,8 @@
 {
   queue::initialize_default();
 
+  queue::initialize_retention();
+
   if (HAVE_LIBDRIZZLE)
   {
     queue::initialize_drizzle();

=== modified file 'libgearman-server/plugins/base.h'
--- libgearman-server/plugins/base.h	2012-10-08 03:37:53 +0000
+++ libgearman-server/plugins/base.h	2012-11-11 01:10:27 +0000
@@ -96,6 +96,8 @@
 
   virtual gearmand_error_t replay(gearman_server_st *server)= 0;
 
+  virtual gearmand_error_t shutdown(gearman_server_st *server)= 0;
+
   static gearmand_error_t replay_add(gearman_server_st *server,
                                      void *context __attribute__ ((unused)),
                                      const char *unique, size_t unique_size,

=== modified file 'libgearman-server/plugins/queue.h'
--- libgearman-server/plugins/queue.h	2012-06-12 18:02:10 +0000
+++ libgearman-server/plugins/queue.h	2012-11-11 01:10:27 +0000
@@ -52,3 +52,5 @@
 #include <libgearman-server/plugins/queue/redis/queue.h>
 
 #include <libgearman-server/plugins/queue/mysql/queue.h>
+
+#include <libgearman-server/plugins/queue/retention/queue.h>

=== modified file 'libgearman-server/plugins/queue/include.am'
--- libgearman-server/plugins/queue/include.am	2012-08-11 01:50:29 +0000
+++ libgearman-server/plugins/queue/include.am	2012-11-11 01:10:27 +0000
@@ -21,3 +21,4 @@
 include libgearman-server/plugins/queue/sqlite/include.am
 include libgearman-server/plugins/queue/tokyocabinet/include.am
 include libgearman-server/plugins/queue/mysql/include.am
+include libgearman-server/plugins/queue/retention/include.am

=== added directory 'libgearman-server/plugins/queue/retention'
=== added file 'libgearman-server/plugins/queue/retention/include.am'
--- libgearman-server/plugins/queue/retention/include.am	1970-01-01 00:00:00 +0000
+++ libgearman-server/plugins/queue/retention/include.am	2012-11-11 01:10:27 +0000
@@ -0,0 +1,15 @@
+# vim:ft=automake
+# Gearman
+# Copyright (C) 2012 Data Differential, http://datadifferential.com/
+# Copyright (C) 2012 Sven Nierlein, [e-mail]
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license.  See
+# the COPYING file in the parent directory for full text.
+#
+# All paths should be given relative to the root
+#
+
+noinst_HEADERS+= libgearman-server/plugins/queue/retention/queue.h
+
+libgearman_server_libgearman_server_la_SOURCES+= libgearman-server/plugins/queue/retention/queue.cc

=== added file 'libgearman-server/plugins/queue/retention/queue.cc'
--- libgearman-server/plugins/queue/retention/queue.cc	1970-01-01 00:00:00 +0000
+++ libgearman-server/plugins/queue/retention/queue.cc	2012-11-11 01:10:27 +0000
@@ -0,0 +1,230 @@
+/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Gearmand client and server library.
+ *
+ *  Copyright (C) 2012 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2012 Sven Nierlein, [e-mail]
+ *  All rights reserved.
+ *
+ */
+
+
+/**
+ * @file
+ * @brief Retention Queue Storage Definitions
+ */
+
+#include <gear_config.h>
+#include <libgearman-server/common.h>
+
+#include <libgearman-server/plugins/queue/retention/queue.h>
+#include <libgearman-server/plugins/queue/base.h>
+
+#include <stdio.h>
+
+namespace gearmand { namespace plugins { namespace queue { class Retention; }}}
+
+#pragma GCC diagnostic ignored "-Wold-style-cast"
+
+static gearmand_error_t _initialize(gearman_server_st& server, gearmand::plugins::queue::Retention *queue);
+
+namespace gearmand {
+namespace plugins {
+namespace queue {
+
+class Retention : public gearmand::plugins::Queue {
+public:
+  Retention();
+  ~Retention();
+  gearmand_error_t initialize();
+  std::string retention_file;
+
+private:
+};
+
+Retention::Retention() :
+  Queue("retention")
+{
+    command_line_options().add_options()
+      ("retention-file", boost::program_options::value(&retention_file)->default_value("/tmp/retention.dat"), "retention file to save jobs on shutdown");
+}
+
+Retention::~Retention()
+{
+}
+
+gearmand_error_t Retention::initialize()
+{
+  return _initialize(Gearmand()->server, this);
+}
+
+
+void initialize_retention()
+{
+  static Retention local_instance;
+}
+
+} // namespace queue
+} // namespace plugins
+} // namespace gearmand
+
+
+
+/**
+ * @addtogroup gearman_queue_retention_static Static retention Queue Storage Definitions
+ * @ingroup gearman_queue_retention
+ * @{
+ */
+
+/* Queue callback functions. */
+static gearmand_error_t __add(gearman_server_st *server __attribute__((unused)),
+                              void *context __attribute__((unused)),
+                              const char *unique __attribute__((unused)), size_t unique_size __attribute__((unused)),
+                              const char *function_name __attribute__((unused)),
+                              size_t function_name_size __attribute__((unused)),
+                              const void *data __attribute__((unused)), size_t data_size __attribute__((unused)),
+                              gearman_job_priority_t priority __attribute__((unused)),
+                              int64_t when __attribute__((unused)))
+{
+  gearmand_debug(__func__);
+  return GEARMAN_SUCCESS;
+}
+
+
+static gearmand_error_t __flush(gearman_server_st *server __attribute__((unused)),
+                                void *context __attribute__((unused)))
+{
+  gearmand_debug(__func__);
+  return GEARMAN_SUCCESS;
+}
+
+static gearmand_error_t __done(gearman_server_st *server __attribute__((unused)),
+                               void *context __attribute__((unused)),
+                               const char *unique __attribute__((unused)),
+                               size_t unique_size __attribute__((unused)),
+                               const char *function_name __attribute__((unused)),
+                               size_t function_name_size __attribute__((unused)))
+{
+  gearmand_debug(__func__);
+  return GEARMAN_SUCCESS;
+}
+
+
+static gearmand_error_t __replay(gearman_server_st *server,
+                                 void *context,
+                                 gearman_queue_add_fn *add_fn,
+                                 void *add_context)
+{
+  int read;
+  int when;
+  int size;
+  char *data_line;
+  char *data;
+  char *function;
+  char *unique;
+  char *prio_s;
+  FILE * fp;
+  gearman_job_priority_t priority= (gearman_job_priority_t)0;
+  gearmand_error_t ret = GEARMAN_SUCCESS;
+  gearmand::plugins::queue::Retention *queue= (gearmand::plugins::queue::Retention *)context;
+
+  gearmand_debug(__func__);
+  gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "reading retention file: %s", queue->retention_file.c_str());
+
+  fp = fopen(queue->retention_file.c_str(), "r");
+  if(!fp)
+    return GEARMAN_SUCCESS;
+
+  data_line = (char*)malloc(1024);
+  if(data_line == NULL){
+    gearmand_perror("malloc failed");
+    return GEARMAN_MEMORY_ALLOCATION_FAILURE;
+  }
+  while(!feof(fp)) {
+    if(fgets(data_line, 1024, fp) == NULL)
+      break; // empty files
+
+    function = strsep( &data_line, ";" );
+    unique   = strsep( &data_line, ";" );
+    prio_s   = strsep( &data_line, ";" );
+    priority = (gearman_job_priority_t) atoi(prio_s);
+    when     = atoi(strsep( &data_line, ";" ));
+    size     = atoi(strsep( &data_line, "\n" ));
+    data     = (char*)malloc(size+1);
+    if (data == NULL){
+      gearmand_perror("malloc failed");
+      return GEARMAN_MEMORY_ALLOCATION_FAILURE;
+    }
+    read = fread(data, size, 1, fp);
+    ret= (*add_fn)(server, add_context,
+                   unique, (size_t) strlen(unique),
+                   function, (size_t) strlen(function),
+                   data, size,
+                   priority,
+                   when);
+    if (ret != GEARMAN_SUCCESS && ret != GEARMAN_JOB_EXISTS)
+      return ret;
+
+    // complete rest of line till newline
+    if(fgets(data_line, 1024, fp) == NULL)
+      break; // no more jobs
+  }
+
+  fclose(fp);
+
+  // cleanup retention file
+  if(unlink(queue->retention_file.c_str()) == -1)
+    perror(queue->retention_file.c_str());
+
+  return GEARMAN_SUCCESS;
+}
+
+static gearmand_error_t __shutdown(gearman_server_st *server,
+                                 void *context)
+{
+  FILE * fp;
+  gearmand::plugins::queue::Retention *queue= (gearmand::plugins::queue::Retention *)context;
+
+  gearmand_debug(__func__);
+  gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "writing retention file: %s for %i jobs", queue->retention_file.c_str(), server->job_count);
+
+  fp = fopen(queue->retention_file.c_str(), "w+");
+  if(fp == NULL) {
+    gearmand_perror(queue->retention_file.c_str());
+    return GEARMAN_ERRNO;
+  }
+
+  for (uint32_t key= 0; key < GEARMAND_JOB_HASH_SIZE; key++) {
+    if(server->job_hash[key] != NULL) {
+      gearman_server_job_st *server_job = server->job_hash[key];
+      fprintf(fp, "%s;%s;%i;%" PRId64 ";%i\n%.*s\n",
+                  server_job->function->function_name,
+                  server_job->unique,
+                  (int)server_job->priority,
+                  server_job->when,
+                  server_job->data_size,
+                  server_job->data_size, (const char*)server_job->data);
+      while (server_job->unique_next != NULL) {
+        server_job = server_job->unique_next;
+        fprintf(fp, "%s;%s;%i;%" PRId64 ";%i\n%.*s\n",
+                    server_job->function->function_name,
+                    server_job->unique,
+                    (int)server_job->priority,
+                    server_job->when,
+                    server_job->data_size,
+                    server_job->data_size, (const char*)server_job->data);
+      }
+    }
+  }
+
+  fclose(fp);
+
+  return GEARMAN_SUCCESS;
+}
+
+gearmand_error_t _initialize(gearman_server_st& server, gearmand::plugins::queue::Retention *queue)
+{
+  gearman_server_set_queue(server, queue, __add, __flush, __done, __replay, __shutdown);
+
+  return GEARMAN_SUCCESS;
+}

=== added file 'libgearman-server/plugins/queue/retention/queue.h'
--- libgearman-server/plugins/queue/retention/queue.h	1970-01-01 00:00:00 +0000
+++ libgearman-server/plugins/queue/retention/queue.h	2012-11-11 01:10:27 +0000
@@ -0,0 +1,22 @@
+/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Gearmand client and server library.
+ *
+ *  Copyright (C) 2012 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2012 Sven Nierlein, [e-mail]
+ *  All rights reserved.
+ *
+ */
+
+#pragma once
+
+
+namespace gearmand {
+namespace plugins {
+namespace queue {
+
+void initialize_retention();
+
+} // namespace queue
+} // namespace plugin
+} // namespace gearmand

=== modified file 'libgearman-server/queue.cc'
--- libgearman-server/queue.cc	2012-10-28 20:04:35 +0000
+++ libgearman-server/queue.cc	2012-11-11 01:10:27 +0000
@@ -137,6 +137,29 @@
 }
 
 void gearman_server_set_queue(gearman_server_st& server,
+                              void *context,
+                              gearman_queue_add_fn *add,
+                              gearman_queue_flush_fn *flush,
+                              gearman_queue_done_fn *done,
+                              gearman_queue_replay_fn *replay,
+                              gearman_queue_shutdown_fn *shutdown)
+{
+  server.queue_version= QUEUE_VERSION_FUNCTION;
+  server.queue.functions= new queue_st();
+  if (server.queue.functions)
+  {
+    server.queue.functions->_context= context;
+    server.queue.functions->_add_fn= add;
+    server.queue.functions->_flush_fn= flush;
+    server.queue.functions->_done_fn= done;
+    server.queue.functions->_replay_fn= replay;
+    server.queue.functions->_shutdown_fn= shutdown;
+  }
+  assert(server.queue.functions);
+}
+
+
+void gearman_server_set_queue(gearman_server_st& server,
                               gearmand::queue::Context* context)
 {
   assert(context);

=== modified file 'libgearman-server/queue.hpp'
--- libgearman-server/queue.hpp	2012-10-08 03:37:53 +0000
+++ libgearman-server/queue.hpp	2012-11-11 01:10:27 +0000
@@ -70,4 +70,12 @@
                               gearman_queue_replay_fn *replay);
 
 void gearman_server_set_queue(gearman_server_st& server,
+                              void *context,
+                              gearman_queue_add_fn *add,
+                              gearman_queue_flush_fn *flush,
+                              gearman_queue_done_fn *done,
+                              gearman_queue_replay_fn *replay,
+                              gearman_queue_shutdown_fn *shutdown);
+
+void gearman_server_set_queue(gearman_server_st& server,
                               gearmand::queue::Context* context);

=== modified file 'libgearman-server/server.cc'
--- libgearman-server/server.cc	2012-10-28 20:04:35 +0000
+++ libgearman-server/server.cc	2012-11-11 01:10:27 +0000
@@ -1,5 +1,5 @@
 /*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
- * 
+ *
  *  Gearmand client and server library.
  *
  *  Copyright (C) 2011-2012 Data Differential, http://datadifferential.com/
@@ -98,7 +98,7 @@
   gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
                      "%15s:%5s packet command  %s",
                      server_con->con.context == NULL || server_con->con.context->host == NULL ? "-" : server_con->con.context->host,
-                     server_con->con.context == NULL || server_con->con.context->port == NULL ? "-" : server_con->con.context->port, 
+                     server_con->con.context == NULL || server_con->con.context->port == NULL ? "-" : server_con->con.context->port,
                      gearmand_strcommand(packet));
 
   switch (packet->command)
@@ -242,7 +242,7 @@
         {
           return gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "strtoul(%ul)", when);
         }
-        gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, 
+        gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
                            "Received EPOCH job submission, function:%.*s unique:%.*s with data for %jd at %jd, args %d",
                            packet->arg_size[0], packet->arg[0],
                            packet->arg_size[1], packet->arg[1],
@@ -576,8 +576,8 @@
       }
       else if (packet->command == GEARMAN_COMMAND_GRAB_JOB_UNIQ)
       {
-        /* 
-          We found a runnable job, queue job assigned packet and take the job off the queue. 
+        /*
+          We found a runnable job, queue job assigned packet and take the job off the queue.
         */
         ret= gearman_server_io_packet_add(server_con, false,
                                           GEARMAN_MAGIC_RESPONSE,
@@ -596,8 +596,8 @@
                            strlen(server_job->reducer), server_job->reducer, strlen(server_job->reducer),
                            server_job->unique_length, server_job->unique, server_job->unique_length,
                            (unsigned long)server_job->data_size);
-        /* 
-          We found a runnable job, queue job assigned packet and take the job off the queue. 
+        /*
+          We found a runnable job, queue job assigned packet and take the job off the queue.
         */
         ret= gearman_server_io_packet_add(server_con, false,
                                           GEARMAN_MAGIC_RESPONSE,
@@ -611,8 +611,8 @@
       }
       else if (packet->command == GEARMAN_COMMAND_GRAB_JOB_ALL)
       {
-        /* 
-          We found a runnable job, queue job assigned packet and take the job off the queue. 
+        /*
+          We found a runnable job, queue job assigned packet and take the job off the queue.
         */
         ret= gearman_server_io_packet_add(server_con, false,
                                           GEARMAN_MAGIC_RESPONSE,
@@ -922,6 +922,26 @@
   return ret;
 }
 
+static gearmand_error_t gearman_queue_shutdown(gearman_server_st& server)
+{
+  if(server.queue.functions->_shutdown_fn == NULL)
+    return GEARMAN_SUCCESS;
+  if (server.queue_version == QUEUE_VERSION_FUNCTION)
+  {
+    return (*(server.queue.functions->_shutdown_fn))(&server,
+                                                    (void *)server.queue.functions->_context);
+  }
+
+  assert(server.queue.object);
+  return server.queue.object->shutdown(&server);
+}
+
+gearmand_error_t gearman_server_queue_shutdown(gearman_server_st& server)
+{
+  gearmand_error_t ret= gearman_queue_shutdown(server);
+  return ret;
+}
+
 void *gearman_server_queue_context(const gearman_server_st *server)
 {
   if (server->queue_version == QUEUE_VERSION_FUNCTION)

=== modified file 'libgearman-server/server.h'
--- libgearman-server/server.h	2012-09-28 09:55:33 +0000
+++ libgearman-server/server.h	2012-11-11 01:10:27 +0000
@@ -90,6 +90,18 @@
 #endif
 
 /**
+ * shutdown the persistent queue to save all remaining jobs.
+ * should only be run at shutdown.
+ * @param server Server structure previously initialized with
+ *        gearman_server_create.
+ * @return Standard gearman return value.
+ */
+#ifdef __cplusplus
+GEARMAN_API
+gearmand_error_t gearman_server_queue_shutdown(gearman_server_st& server);
+#endif
+
+/**
  * Get persistent queue context.
  */
 GEARMAN_API

=== modified file 'libgearman-server/struct/server.h'
--- libgearman-server/struct/server.h	2012-08-11 01:53:36 +0000
+++ libgearman-server/struct/server.h	2012-11-11 01:10:27 +0000
@@ -43,6 +43,7 @@
   gearman_queue_flush_fn *_flush_fn;
   gearman_queue_done_fn *_done_fn;
   gearman_queue_replay_fn *_replay_fn;
+  gearman_queue_shutdown_fn *_shutdown_fn;
 
 #ifdef __cplusplus
   queue_st() :
@@ -50,7 +51,8 @@
     _add_fn(NULL),
     _flush_fn(NULL),
     _done_fn(NULL),
-    _replay_fn(NULL)
+    _replay_fn(NULL),
+    _shutdown_fn(NULL)
   {
   }
 #endif