Merge lp:~daniel-nichter/drizzle/fix-slave-reconnect-bug-956200 into lp:drizzle

[Daniel Nichter]

Fixes bug 956200 (slave config file option max-reconnects doesn't work) and bug 956214 (slave doesn't set status to STOPPED after max-reconnecct failures).

[David Shrewsbury (dshrews)]

Although I whole-heartedly agree with the changes you are trying to make, I think parts of it need reworked a bit.

We need to keep the concept of when a thread is actually alive (i.e., RUNNING) vs. when the thread has terminated (STOPPED). You've essentially redefined those states here, and that's the part I disagree with. What we need instead is to add additional thread states. E.g.,

  * running and initializing (INITIALIZING). This would likely replace RUNNING.
  * running and disconnected and will not attempt reconnect (DISCONNECTED)
  * running and awaiting reconnect (RECONNECTING)
  * running and connected normally (CONNECTED)
  * running but in some awful error state (ERROR)
  * thread has terminated normally (STOPPED)
  * etc.

Something along those lines. Those states are just off the top of my head and I'm open to what is defined.

And although I was the one to move max-reconnects to each [masterN] section for some inexplicable reason, wouldn't we want the max-reconnect, seconds-between-reconnects, and io-thread-sleep values to be common values among all defined masters, rather than defined separately for each master?

review: Disapprove

[Daniel Nichter (daniel-nichter)]

Dave,

I agree: we need additional states, and I had already planned to tackle that too after further discussion. For right now, however, slave is broken and this patch is just a first step to fix it. More work is needed, but who knows when that work will be done. Since the patch doesn't break or regress any features, I think it's worth accepting now, and we'll continue to improve it when we can.

Le 1 mai 2012 à 10:23, David Shrewsbury a écrit :

> Review: Disapprove
>
> Although I whole-heartedly agree with the changes you are trying to make, I think parts of it need reworked a bit.
>
> We need to keep the concept of when a thread is actually alive (i.e., RUNNING) vs. when the thread has terminated (STOPPED). You've essentially redefined those states here, and that's the part I disagree with. What we need instead is to add additional thread states. E.g.,
>
> * running and initializing (INITIALIZING). This would likely replace RUNNING.
> * running and disconnected and will not attempt reconnect (DISCONNECTED)
> * running and awaiting reconnect (RECONNECTING)
> * running and connected normally (CONNECTED)
> * running but in some awful error state (ERROR)
> * thread has terminated normally (STOPPED)
> * etc.
>
> Something along those lines. Those states are just off the top of my head and I'm open to what is defined.
>
> And although I was the one to move max-reconnects to each [masterN] section for some inexplicable reason, wouldn't we want the max-reconnect, seconds-between-reconnects, and io-thread-sleep values to be common values among all defined masters, rather than defined separately for each master?
> --
> https://code.launchpad.net/~daniel-nichter/drizzle/fix-slave-reconnect-bug-956200/+merge/104046
> You are the owner of lp:~daniel-nichter/drizzle/fix-slave-reconnect-bug-956200.

[Brian Aker (brianaker)]

Any thoughts on how to proceed?

[Daniel Nichter (daniel-nichter)]

Yes, I sent another email to the discussion list about the additional status states. We can put this merge on hold for now until I implement these addition states.

Le 3 mai 2012 à 23:39, Brian Aker a écrit :

> Any thoughts on how to proceed?
> --
> https://code.launchpad.net/~daniel-nichter/drizzle/fix-slave-reconnect-bug-956200/+merge/104046
> You are the owner of lp:~daniel-nichter/drizzle/fix-slave-reconnect-bug-956200.

[lp:~daniel-nichter/drizzle/fix-slave-reconnect-bug-956200]

2552. By Daniel Nichter on 2012-05-06

Add extra IO thread states: INITIALIZING, DISCONNECTED, RECONNECTING, CONNECTING, OK, STOPPED, ERROR. Change setIOState() params. Use C++ str instead of C str for err msg.

Unmerged revisions

2552. By Daniel Nichter on 2012-05-06

Add extra IO thread states: INITIALIZING, DISCONNECTED, RECONNECTING, CONNECTING, OK, STOPPED, ERROR. Change setIOState() params. Use C++ str instead of C str for err msg.

2551. By Daniel Nichter on 2012-04-29

Make status=STOPPED until IO thread connects. Set error_msg if all reconnects fail.

2550. By Daniel Nichter on 2012-04-29

Fix max-reconnects. Make seconds-between-reconnects and io-thread-sleep per-master options.

=== modified file 'plugin/slave/queue_producer.cc'
--- plugin/slave/queue_producer.cc	2012-04-14 20:43:20 +0000
+++ plugin/slave/queue_producer.cc	2012-05-06 02:51:19 +0000
@@ -27,6 +27,7 @@
 #include <drizzled/gettext.h>
 #include <drizzled/message/transaction.pb.h>
 #include <boost/lexical_cast.hpp>
+#include <boost/format.hpp>
 #include <google/protobuf/text_format.h>
 #include <string>
 #include <vector>
@@ -45,7 +46,7 @@
 
 bool QueueProducer::init()
 {
-  setIOState("", true);
+  setIOState("INITIALIZING");
   return reconnect(true);
 }
 
@@ -104,7 +105,11 @@
 
 void QueueProducer::shutdown()
 {
-  setIOState(_last_error_message, false);
+  if (_last_error_message.empty())
+    setIOState("STOPPED");
+  else
+    setIOState("ERROR", _last_error_message);
+
   if (_is_connected)
     closeConnection();
 }
@@ -121,28 +126,45 @@
   _last_error_message.clear();
   boost::posix_time::seconds duration(_seconds_between_reconnects);
 
-  uint32_t attempts= 1;
-
-  setIOState("Connecting...", true);
-
+  uint32_t attempts= 0;
   while (not openConnection())
   {
-    char buf[250];
-    snprintf(buf, sizeof(buf),_("Connection attempt %d of %d failed, sleeping for %d seconds and retrying. %s"), attempts, _max_reconnects, _seconds_between_reconnects, _last_error_message.c_str());
+    if (++attempts == _max_reconnects)
+      break;
 
-    setIOState(buf, true);
-    if (attempts++ == _max_reconnects)
-      break;
+    std::string err_msg= boost::str(
+      boost::format(_("Connection attempt %1% of %2% failed. "
+                      "Sleeping for %3% seconds and retrying. %4%"))
+      % attempts
+      % _max_reconnects
+      % _seconds_between_reconnects
+      % _last_error_message
+    );
+    setIOState("RECONNECTING", err_msg);
     boost::this_thread::sleep(duration);
   }
 
-  setIOState(_is_connected ? _("Connected") : _("Disconnected"), true);
+  if (_is_connected)
+  {
+    setIOState("OK");
+  }
+  else
+  {
+    std::string err_msg= boost::str(
+      boost::format(_("Failed to connect to master after %1% attempts. %2%"))
+      % _max_reconnects
+      % _last_error_message.c_str()
+    );
+    setIOState("DISCONNECTED", err_msg);
+  }
 
   return _is_connected;
 }
 
 bool QueueProducer::openConnection()
 {
+  setIOState("CONNECTING");
+
   if ((_drizzle= drizzle_create()) == NULL)
   {
     _last_return= DRIZZLE_RETURN_INTERNAL_ERROR;
@@ -476,41 +498,33 @@
   return EE_OK;
 }
 
-
-void QueueProducer::setIOState(const string &err_msg, bool status)
+void QueueProducer::setIOState(const std::string &state, const std::string &err_msg)
 {
   vector<string> statements;
-  string sql;
-  string msg(err_msg);
-
-  if (not status)
-  {
-    sql= "UPDATE `sys_replication`.`io_state` SET `status` = 'STOPPED'";
-  }
-  else
-  {
-    sql= "UPDATE `sys_replication`.`io_state` SET `status` = 'RUNNING'";
-  }
-  
-  sql.append(", `error_msg` = '", 17);
+  string sql= "UPDATE `sys_replication`.`io_state` SET `status` = '"
+            + state + "', `error_msg` = '";
 
   /* Escape embedded quotes and statement terminators */
-  string::iterator it;
-  for (it= msg.begin(); it != msg.end(); ++it)
+  if (not err_msg.empty())
   {
-    if (*it == '\'')
-    {
-      it= msg.insert(it, '\'');
-      ++it;  /* advance back to the quote */
-    }
-    else if (*it == ';')
-    {
-      it= msg.insert(it, '\\');
-      ++it;  /* advance back to the semicolon */
-    }
+    string msg(err_msg);
+    string::iterator it;
+    for (it= msg.begin(); it != msg.end(); ++it)
+    {
+      if (*it == '\'')
+      {
+        it= msg.insert(it, '\'');
+        ++it;  /* advance back to the quote */
+      }
+      else if (*it == ';')
+      {
+        it= msg.insert(it, '\\');
+        ++it;  /* advance back to the semicolon */
+      }
+    }
+    sql.append(msg);
   }
-  
-  sql.append(msg);
+
   sql.append("' WHERE `master_id` = ");
   sql.append(boost::lexical_cast<string>(masterId()));
 

=== modified file 'plugin/slave/queue_producer.h'
--- plugin/slave/queue_producer.h	2011-11-26 23:14:59 +0000
+++ plugin/slave/queue_producer.h	2012-05-06 02:51:19 +0000
@@ -193,10 +193,10 @@
   /**
    * Update IO thread status in state table.
    *
+   * @param state State of IO thread
    * @param err_msg Error message string
-   * @param status false = STOPPED, true = RUNNING
    */
-  void setIOState(const std::string &err_msg, bool status);
+  void setIOState(const std::string &state, const std::string &err_msg="");
 
 };
 

=== modified file 'plugin/slave/replication_slave.cc'
--- plugin/slave/replication_slave.cc	2011-07-06 23:13:00 +0000
+++ plugin/slave/replication_slave.cc	2012-05-06 02:51:19 +0000
@@ -63,14 +63,19 @@
   po::variables_map vm;
   po::options_description slave_options("Options for the slave plugin");
 
-  /* Common options */
+  /*
+   * Define the slave's applier (consumer) thread options.  These are
+   * the valid options at the start of the slave config file before
+   * any [masterN] headers.
+   */
   slave_options.add_options()
-    ("seconds-between-reconnects", po::value<uint32_t>()->default_value(30))
-    ("io-thread-sleep", po::value<uint32_t>()->default_value(5))
     ("applier-thread-sleep", po::value<uint32_t>()->default_value(5))
     ("ignore-errors", po::value<bool>()->default_value(false)->zero_tokens());
 
-  /* Master defining options */
+  /*
+   * Define per-master IO (producer) thread options.  These are
+   * the valid options under each [masterN] header in the slave config file.
+   */
   for (size_t num= 1; num <= 10; num++)
   {
     string section("master");
@@ -81,6 +86,8 @@
       ((section + ".master-user").c_str(), po::value<string>()->default_value(""))
       ((section + ".master-pass").c_str(), po::value<string>()->default_value(""))
       ((section + ".max-reconnects").c_str(), po::value<uint32_t>()->default_value(10))
+      ((section + ".seconds-between-reconnects").c_str(), po::value<uint32_t>()->default_value(30))
+      ((section + ".io-thread-sleep").c_str(), po::value<uint32_t>()->default_value(5))
       ((section + ".max-commit-id").c_str(), po::value<uint64_t>());
    }
 
@@ -119,6 +126,7 @@
 
     _masters[num]= new (std::nothrow) Master(num);
 
+    /* Set the IO (consumer) thread options for this master. */
     if (vm.count(section + ".master-host"))
       master(num).producer().setMasterHost(vm[section + ".master-host"].as<string>());
 
@@ -133,28 +141,24 @@
 
     if (vm.count(section + ".max-commit-id"))
       master(num).producer().setCachedMaxCommitId(vm[section + ".max-commit-id"].as<uint64_t>());
-  }
-
-  boost::unordered_map<uint32_t, Master *>::const_iterator it;
-
-  for (it= _masters.begin(); it != _masters.end(); ++it)
-  {
-    if (vm.count("max-reconnects"))
-      it->second->producer().setMaxReconnectAttempts(vm["max-reconnects"].as<uint32_t>());
-
-    if (vm.count("seconds-between-reconnects"))
-      it->second->producer().setSecondsBetweenReconnects(vm["seconds-between-reconnects"].as<uint32_t>());
-
-    if (vm.count("io-thread-sleep"))
-      it->second->producer().setSleepInterval(vm["io-thread-sleep"].as<uint32_t>());
-  }
-
+
+    if (vm.count(section + ".max-reconnects"))
+      master(num).producer().setMaxReconnectAttempts(vm[section + ".max-reconnects"].as<uint32_t>());
+
+    if (vm.count(section + ".seconds-between-reconnects"))
+      master(num).producer().setSecondsBetweenReconnects(vm[section + ".seconds-between-reconnects"].as<uint32_t>());
+
+    if (vm.count(section + ".io-thread-sleep"))
+      master(num).producer().setSleepInterval(vm[section + ".io-thread-sleep"].as<uint32_t>());
+  }
+
+  /* Set the slave's applier (consumer) thread options. */
   if (vm.count("applier-thread-sleep"))
     _consumer.setSleepInterval(vm["applier-thread-sleep"].as<uint32_t>());
   if (vm.count("ignore-errors"))
     _consumer.setIgnoreErrors(vm["ignore-errors"].as<bool>());
 
-  /* setup schema and tables */
+  /* Create the sys_replication schema and its tables. */
   ReplicationSchema rs;
   if (not rs.create())
   {
@@ -162,9 +166,10 @@
     return false;
   }
 
+  /* Create a row in the sys_replication tables for each master. */
+  boost::unordered_map<uint32_t, Master *>::const_iterator it;
   for (it= _masters.begin(); it != _masters.end(); ++it)
   {
-    /* make certain a row exists for each master */
     rs.createInitialApplierRow(it->first);
     rs.createInitialIORow(it->first);