Merge lp:~dshrews/drizzle/bug729397_optimize_io into lp:drizzle/7.0

[David Shrewsbury]

Changes the IO (producer) thread to continually pull from the master until it is caught up, at which point it will then sleep.

=== modified file 'plugin/slave/queue_producer.cc'
--- plugin/slave/queue_producer.cc	2011-03-04 22:49:58 +0000
+++ plugin/slave/queue_producer.cc	2011-03-07 17:40:50 +0000
@@ -71,7 +71,12 @@
     }
   }
 
-  if (not queryForReplicationEvents(_saved_max_commit_id))
+  /* Keep getting events until caught up */
+  enum drizzled::error_t err;
+  while ((err= (queryForReplicationEvents(_saved_max_commit_id))) == EE_OK)
+  {}
+
+  if (err == ER_YES)  /* We encountered an error */
   {
     if (_last_return == DRIZZLE_RETURN_LOST_CONNECTION)
     {
@@ -368,16 +373,16 @@
 }
 
 
-bool QueueProducer::queryForReplicationEvents(uint64_t max_commit_id)
+enum drizzled::error_t QueueProducer::queryForReplicationEvents(uint64_t max_commit_id)
 {
   vector<uint64_t> trx_id_list;
 
   if (not queryForTrxIdList(max_commit_id, trx_id_list))
-    return false;
+    return ER_YES;
 
   if (trx_id_list.size() == 0)    /* nothing to get from the master */
   {
-    return true;
+    return ER_NO;
   }
 
   /*
@@ -406,7 +411,7 @@
     _last_error_message.append(drizzle_error(&_drizzle));
     errmsg_printf(error::ERROR, _("%s"), _last_error_message.c_str());
     drizzle_result_free(&result);
-    return false;
+    return ER_YES;
   }
 
   /* TODO: Investigate 1-row-at-a-time buffering */
@@ -420,7 +425,7 @@
     _last_error_message.append(drizzle_error(&_drizzle));
     errmsg_printf(error::ERROR, _("%s"), _last_error_message.c_str());
     drizzle_result_free(&result);
-    return false;
+    return ER_YES;
   }
 
   drizzle_row_t row;
@@ -432,13 +437,13 @@
       errmsg_printf(error::ERROR,
                     _("Replication slave: Unable to insert into queue."));
       drizzle_result_free(&result);
-      return false;
+      return ER_YES;
     }
   }
 
   drizzle_result_free(&result);
 
-  return true;
+  return EE_OK;
 }
 
 

=== modified file 'plugin/slave/queue_producer.h'
--- plugin/slave/queue_producer.h	2011-02-28 14:33:56 +0000
+++ plugin/slave/queue_producer.h	2011-03-07 17:40:50 +0000
@@ -21,9 +21,10 @@
 #ifndef PLUGIN_SLAVE_QUEUE_PRODUCER_H
 #define PLUGIN_SLAVE_QUEUE_PRODUCER_H
 
+#include <client/client_priv.h>
+#include <drizzled/error_t.h>
 #include <plugin/slave/queue_thread.h>
 #include <plugin/slave/sql_executor.h>
-#include <client/client_priv.h>
 #include <string>
 #include <vector>
 
@@ -132,8 +133,31 @@
    */
   bool reconnect(bool initial_connection);
 
+  /**
+   * Get maximum commit ID that we have stored locally on the slave.
+   *
+   * This method determines where this slave is in relation to the master,
+   * or, in other words, how "caught up" we are.
+   *
+   * @param[out] max_commit_id Maximum commit ID we have on this slave.
+   */
   bool queryForMaxCommitId(uint64_t *max_commit_id);
-  bool queryForReplicationEvents(uint64_t max_commit_id);
+
+  /**
+   * Get replication events/messages from the master.
+   *
+   * Calling this method will a limited number of events from the master.
+   * It should be repeatedly called until it returns -1, which means there
+   * were no more events to retrieve.
+   *
+   * @param[in] max_commit_id Largest commit ID we have stored locally.
+   *
+   * @retval EE_OK  Successfully retrieved events
+   * @retval ER_NO  No errors, but no more events to retrieve
+   * @retval ER_YES Error
+   */
+  enum drizzled::error_t queryForReplicationEvents(uint64_t max_commit_id);
+
   bool queryForTrxIdList(uint64_t max_commit_id, std::vector<uint64_t> &list);
   bool queueInsert(const char *trx_id,
                    const char *seg_id,