Merge lp:~fallenpegasus/drizzle/multimaster into lp:~drizzle-trunk/drizzle/development

[Mark Atwood]

=== modified file 'plugin/slave/module.cc'
--- plugin/slave/module.cc	2011-04-02 01:32:02 +0000
+++ plugin/slave/module.cc	2011-07-06 23:18:34 +0000
@@ -45,8 +45,6 @@
   const module::option_map &vm= context.getOptions();
 
   ReplicationSlave *slave= new ReplicationSlave(vm["config-file"].as<string>());
-  if (vm.count("max-commit-id"))
-    slave->setMaxCommitId(vm["max-commit-id"].as<uint64_t>());
   context.add(slave);
   return 0;
 }
@@ -56,9 +54,6 @@
   context("config-file",
           po::value<string>()->default_value(DEFAULT_SLAVE_CFG_FILE.string()),
           N_("Path to the slave configuration file"));
-  context("max-commit-id",
-          po::value<uint64_t>(),
-          N_("Value to use as the maximum commit ID stored on the slave"));
 }
 
 } /* namespace slave */

=== modified file 'plugin/slave/queue_consumer.cc'
--- plugin/slave/queue_consumer.cc	2011-04-27 18:16:25 +0000
+++ plugin/slave/queue_consumer.cc	2011-07-06 23:18:34 +0000
@@ -20,10 +20,11 @@
 
 #include <config.h>
 #include <plugin/slave/queue_consumer.h>
+#include <drizzled/errmsg_print.h>
+#include <drizzled/execute.h>
 #include <drizzled/message/transaction.pb.h>
 #include <drizzled/message/statement_transform.h>
 #include <drizzled/sql/result_set.h>
-#include <drizzled/execute.h>
 #include <string>
 #include <vector>
 #include <boost/thread.hpp>
@@ -51,9 +52,25 @@
 
 bool QueueConsumer::process()
 {
+  for (size_t index= 0; index < _master_ids.size(); index++)
+  {
+    /* We go ahead and get the string version of the master ID
+     * so we don't have to keep converting it from int to string.
+     */
+    const string master_id= boost::lexical_cast<string>(_master_ids[index]);
+
+    if (not processSingleMaster(master_id))
+      return false;
+  }
+
+  return true;
+}
+
+bool QueueConsumer::processSingleMaster(const string &master_id)
+{
   TrxIdList completedTransactionIds;
 
-  getListOfCompletedTransactions(completedTransactionIds);
+  getListOfCompletedTransactions(master_id, completedTransactionIds);
 
   for (size_t x= 0; x < completedTransactionIds.size(); x++)
   {
@@ -68,7 +85,7 @@
     message::Transaction transaction;
     uint32_t segment_id= 1;
 
-    while (getMessage(transaction, commit_id, trx_id, originating_server_uuid,
+    while (getMessage(transaction, commit_id, master_id, trx_id, originating_server_uuid,
                       originating_commit_id, segment_id++))
     {
       convertToSQL(transaction, aggregate_sql, segmented_sql);
@@ -120,12 +137,30 @@
 
     if (not executeSQLWithCommitId(aggregate_sql, commit_id, 
                                    originating_server_uuid, 
-                                   originating_commit_id))
+                                   originating_commit_id,
+                                   master_id))
     {
-      return false;
+      if (_ignore_errors)
+      {
+        clearErrorState();
+
+        /* Still need to record that we handled this trx */
+        vector<string> sql;
+        string tmp("UPDATE `sys_replication`.`applier_state`"
+                   " SET `last_applied_commit_id` = ");
+        tmp.append(commit_id);
+        tmp.append(" WHERE `master_id` = ");
+        tmp.append(master_id);
+        sql.push_back(tmp);
+        executeSQL(sql);
+      }
+      else
+      {
+        return false;
+      }
     }
 
-    if (not deleteFromQueue(trx_id))
+    if (not deleteFromQueue(master_id, trx_id))
     {
       return false;
     }
@@ -137,6 +172,7 @@
 
 bool QueueConsumer::getMessage(message::Transaction &transaction,
                               string &commit_id,
+                              const string &master_id,
                               uint64_t trx_id,
                               string &originating_server_uuid,
                               uint64_t &originating_commit_id,
@@ -148,6 +184,8 @@
   sql.append(boost::lexical_cast<string>(trx_id));
   sql.append(" AND `seg_id` = ", 16);
   sql.append(boost::lexical_cast<string>(segment_id));
+  sql.append(" AND `master_id` = ", 19),
+  sql.append(master_id);
 
   sql::ResultSet result_set(4);
   Execute execute(*(_session.get()), true);
@@ -189,13 +227,16 @@
   return true;
 }
 
-bool QueueConsumer::getListOfCompletedTransactions(TrxIdList &list)
+bool QueueConsumer::getListOfCompletedTransactions(const string &master_id,
+                                                   TrxIdList &list)
 {
   Execute execute(*(_session.get()), true);
   
   string sql("SELECT `trx_id` FROM `sys_replication`.`queue`"
              " WHERE `commit_order` IS NOT NULL AND `commit_order` > 0"
-             " ORDER BY `commit_order` ASC");
+             " AND `master_id` = "
+             + master_id
+             + " ORDER BY `commit_order` ASC");
   
   /* ResultSet size must match column count */
   sql::ResultSet result_set(1);
@@ -329,6 +370,13 @@
 }
 
 
+/*
+ * TODO: This currently updates every row in the applier_state table.
+ * This use to be a single row. With multi-master support, we now need
+ * a row for every master so we can track the last applied commit ID
+ * value for each. Eventually, we may want multiple consumer threads,
+ * so then we'd need to update each row independently.
+ */
 void QueueConsumer::setApplierState(const string &err_msg, bool status)
 {
   vector<string> statements;
@@ -373,7 +421,8 @@
 bool QueueConsumer::executeSQLWithCommitId(vector<string> &sql,
                                            const string &commit_id, 
                                            const string &originating_server_uuid, 
-                                           uint64_t originating_commit_id)
+                                           uint64_t originating_commit_id,
+                                           const string &master_id)
 {
   string tmp("UPDATE `sys_replication`.`applier_state`"
              " SET `last_applied_commit_id` = ");
@@ -383,6 +432,9 @@
   tmp.append("' , `originating_commit_id` = ");
   tmp.append(boost::lexical_cast<string>(originating_commit_id));
 
+  tmp.append(" WHERE `master_id` = ");
+  tmp.append(master_id);
+
   sql.push_back(tmp);
  
   _session->setOriginatingServerUUID(originating_server_uuid);
@@ -392,11 +444,14 @@
 }
 
 
-bool QueueConsumer::deleteFromQueue(uint64_t trx_id)
+bool QueueConsumer::deleteFromQueue(const string &master_id, uint64_t trx_id)
 {
   string sql("DELETE FROM `sys_replication`.`queue` WHERE `trx_id` = ");
   sql.append(boost::lexical_cast<std::string>(trx_id));
 
+  sql.append(" AND `master_id` = ");
+  sql.append(master_id);
+
   vector<string> sql_vect;
   sql_vect.push_back(sql);
 

=== modified file 'plugin/slave/queue_consumer.h'
--- plugin/slave/queue_consumer.h	2011-04-27 18:16:25 +0000
+++ plugin/slave/queue_consumer.h	2011-07-06 23:18:34 +0000
@@ -23,6 +23,7 @@
 #include <plugin/slave/queue_thread.h>
 #include <plugin/slave/sql_executor.h>
 #include <drizzled/session.h>
+#include <string>
 
 namespace drizzled
 {
@@ -41,7 +42,8 @@
   QueueConsumer() :
     QueueThread(),
     SQLExecutor("slave", "replication"),
-    _check_interval(5)
+    _check_interval(5),
+    _ignore_errors(false)
   { }
 
   bool init();
@@ -59,6 +61,14 @@
   }
   
   /**
+   * Determines if we should ignore errors from statements pulled from masters.
+   */
+  void setIgnoreErrors(bool value)
+  {
+    _ignore_errors= value;
+  }
+
+  /**
    * Update applier status in state table.
    *
    * @param err_msg Error message string
@@ -66,16 +76,41 @@
    */
   void setApplierState(const std::string &err_msg, bool status);
 
+  void addMasterId(uint32_t id)
+  {
+    _master_ids.push_back(id);
+  }
+
+  bool processSingleMaster(const std::string &master_id);
+
 private:
   typedef std::vector<uint64_t> TrxIdList;
 
   /** Number of seconds to sleep between checking queue for messages */
   uint32_t _check_interval;
 
-  bool getListOfCompletedTransactions(TrxIdList &list);
+  std::vector<uint32_t> _master_ids;
+
+  bool _ignore_errors;
+
+  /**
+   * Get a list of transaction IDs from the queue that are complete.
+   *
+   * A "complete" transaction is one in which we have received the end
+   * segment of the transaction.
+   *
+   * @param[in] master_id Identifier of the master we are interested in.
+   * @param[out] list The list to populate with transaction IDs.
+   *
+   * @retval true Success
+   * @retval false Error
+   */
+  bool getListOfCompletedTransactions(const std::string &master_id,
+                                      TrxIdList &list);
 
   bool getMessage(drizzled::message::Transaction &transaction,
                   std::string &commit_id,
+                  const std::string &master_id,
                   uint64_t trx_id,
                   std::string &originating_server_uuid,
                   uint64_t &originating_commit_id,
@@ -111,7 +146,8 @@
   bool executeSQLWithCommitId(std::vector<std::string> &sql,
                               const std::string &commit_id,
                               const std::string &originating_server_uuid,
-                              uint64_t originating_commit_id);
+                              uint64_t originating_commit_id,
+                              const std::string &master_id);
   
   /**
    * Remove messages for a given transaction from the queue.
@@ -121,7 +157,7 @@
    * @retval true Success
    * @retval false Failure
    */
-  bool deleteFromQueue(uint64_t trx_id);
+  bool deleteFromQueue(const std::string &master_id, uint64_t trx_id);
 
   /**
    * Determine if a Statement message is an end message.

=== modified file 'plugin/slave/queue_producer.cc'
--- plugin/slave/queue_producer.cc	2011-04-27 18:16:25 +0000
+++ plugin/slave/queue_producer.cc	2011-07-06 23:18:34 +0000
@@ -18,7 +18,6 @@
  *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  */
 
-#include <config.h>
 #include <plugin/slave/queue_producer.h>
 #include <drizzled/errmsg_print.h>
 #include <drizzled/sql/result_set.h>
@@ -27,6 +26,8 @@
 #include <drizzled/message/transaction.pb.h>
 #include <boost/lexical_cast.hpp>
 #include <google/protobuf/text_format.h>
+#include <string>
+#include <vector>
 
 using namespace std;
 using namespace drizzled;
@@ -199,8 +200,12 @@
    */
   string sql("SELECT MAX(x.cid) FROM"
              " (SELECT MAX(`commit_order`) AS cid FROM `sys_replication`.`queue`"
-             "  UNION ALL SELECT `last_applied_commit_id` AS cid"
-             "  FROM `sys_replication`.`applier_state`) AS x");
+             "  WHERE `master_id` = "
+             + boost::lexical_cast<string>(masterId())
+             + "  UNION ALL SELECT `last_applied_commit_id` AS cid"
+             + "  FROM `sys_replication`.`applier_state` WHERE `master_id` = "
+             + boost::lexical_cast<string>(masterId())
+             + ") AS x");
 
   sql::ResultSet result_set(1);
   Execute execute(*(_session.get()), true);
@@ -304,8 +309,10 @@
    * The SQL to insert our results into the local queue.
    */
   string sql= "INSERT INTO `sys_replication`.`queue`"
-              " (`trx_id`, `seg_id`, `commit_order`,"
+              " (`master_id`, `trx_id`, `seg_id`, `commit_order`,"
               "  `originating_server_uuid`, `originating_commit_id`, `msg`) VALUES (";
+  sql.append(boost::lexical_cast<string>(masterId()));
+  sql.append(", ", 2);
   sql.append(trx_id);
   sql.append(", ", 2);
   sql.append(seg_id);
@@ -490,7 +497,8 @@
   }
   
   sql.append(msg);
-  sql.append("'", 1);
+  sql.append("' WHERE `master_id` = ");
+  sql.append(boost::lexical_cast<string>(masterId()));
 
   statements.push_back(sql);
   executeSQL(statements);

=== modified file 'plugin/slave/queue_producer.h'
--- plugin/slave/queue_producer.h	2011-04-26 18:08:44 +0000
+++ plugin/slave/queue_producer.h	2011-07-06 23:18:34 +0000
@@ -37,6 +37,7 @@
     SQLExecutor("slave", "replication"),
     _check_interval(5),
     _master_port(3306),
+    _master_id(0),
     _last_return(DRIZZLE_RETURN_OK),
     _is_connected(false),
     _saved_max_commit_id(0),
@@ -95,6 +96,21 @@
     _saved_max_commit_id= value;
   }
 
+  uint64_t cachedMaxCommitId()
+  {
+    return _saved_max_commit_id;
+  }
+
+  void setMasterId(uint32_t value)
+  {
+    _master_id= value;
+  }
+
+  uint32_t masterId()
+  {
+    return _master_id;
+  }
+
 private:
   /** Number of seconds to sleep between checking queue for messages */
   uint32_t _check_interval;
@@ -105,6 +121,8 @@
   std::string _master_user;
   std::string _master_pass;
 
+  uint32_t _master_id;
+
   drizzle_st _drizzle;
   drizzle_con_st _connection;
   drizzle_return_t _last_return;

=== modified file 'plugin/slave/replication_schema.cc'
--- plugin/slave/replication_schema.cc	2011-04-27 18:16:25 +0000
+++ plugin/slave/replication_schema.cc	2011-07-06 23:18:34 +0000
@@ -44,19 +44,28 @@
     return false;
 
   /*
-   * Create our IO thread state information table if we need to.
+    Create our IO thread state information table if we need to.
    */
 
+  /*
+    Table: io_state
+    Version 1.0: Initial definition
+    Version 1.1: Added master_id and PK on master_id
+  */
+
+
   sql.clear();
   sql.push_back("COMMIT");
   sql.push_back("CREATE TABLE IF NOT EXISTS `sys_replication`.`io_state` ("
+                " `master_id` BIGINT NOT NULL,"
                 " `status` VARCHAR(20) NOT NULL,"
                 " `error_msg` VARCHAR(250))"
-                " COMMENT = 'VERSION 1.0'");
+                " COMMENT = 'VERSION 1.1'");
 
   if (not executeSQL(sql))
     return false;
 
+#if 0
   sql.clear();
   sql.push_back("SELECT COUNT(*) FROM `sys_replication`.`io_state`");
 
@@ -77,6 +86,7 @@
         return false;
     }
   }
+#endif
 
   /*
    * Create our applier thread state information table if we need to.
@@ -86,21 +96,24 @@
    * Table: applier_state
    * Version 1.0: Initial definition
    * Version 1.1: Added originating_server_uuid and originating_commit_id
+   * Version 1.2: Added master_id and changed PK to master_id
    */
 
   sql.clear();
   sql.push_back("COMMIT");
   sql.push_back("CREATE TABLE IF NOT EXISTS `sys_replication`.`applier_state`"
                 " (`last_applied_commit_id` BIGINT NOT NULL PRIMARY KEY,"
+                " `master_id` BIGINT NOT NULL,"
                 " `originating_server_uuid` VARCHAR(36) NOT NULL,"
                 " `originating_commit_id` BIGINT NOT NULL,"
                 " `status` VARCHAR(20) NOT NULL,"
                 " `error_msg` VARCHAR(250))"
-                " COMMENT = 'VERSION 1.1'");
+                " COMMENT = 'VERSION 1.2'");
 
   if (not executeSQL(sql))
     return false;
 
+#if 0
   sql.clear();
   sql.push_back("SELECT COUNT(*) FROM `sys_replication`.`applier_state`");
 
@@ -123,11 +136,13 @@
         return false;
     }
   }
+#endif
 
   /*
    * Create our message queue table if we need to.
    * Version 1.0: Initial definition
    * Version 1.1: Added originating_server_uuid and originating_commit_id
+   * Version 1.2: Added master_id and changed PK to (master_id, trx_id, seg_id)
    */
 
   sql.clear();
@@ -138,21 +153,76 @@
                 " `originating_server_uuid` VARCHAR(36) NOT NULL,"
                 " `originating_commit_id` BIGINT NOT NULL,"
                 " `msg` BLOB,"
+                " `master_id` BIGINT NOT NULL,"
                 " PRIMARY KEY(`trx_id`, `seg_id`))"
-                " COMMENT = 'VERSION 1.1'");
+                " COMMENT = 'VERSION 1.2'");
   if (not executeSQL(sql))
     return false;
 
   return true;
 }
 
-bool ReplicationSchema::setInitialMaxCommitId(uint64_t value)
+bool ReplicationSchema::createInitialIORow(uint32_t master_id)
+{
+  vector<string> sql;
+
+  sql.push_back("SELECT COUNT(*) FROM `sys_replication`.`io_state` WHERE `master_id` = " + boost::lexical_cast<string>(master_id));
+
+  sql::ResultSet result_set(1);
+  Execute execute(*(_session.get()), true);
+  execute.run(sql[0], result_set);
+  result_set.next();
+  string count= result_set.getString(0);
+
+  if (count == "0")
+  {
+    sql.clear();
+    sql.push_back("INSERT INTO `sys_replication`.`io_state` (`master_id`, `status`) VALUES ("
+                  + boost::lexical_cast<string>(master_id)
+                  + ", 'STOPPED')");
+    if (not executeSQL(sql))
+      return false;
+  }
+
+  return true;
+}
+
+bool ReplicationSchema::createInitialApplierRow(uint32_t master_id)
+{
+  vector<string> sql;
+
+  sql.push_back("SELECT COUNT(*) FROM `sys_replication`.`applier_state` WHERE `master_id` = " + boost::lexical_cast<string>(master_id));
+
+  sql::ResultSet result_set(1);
+  Execute execute(*(_session.get()), true);
+  execute.run(sql[0], result_set);
+  result_set.next();
+  string count= result_set.getString(0);
+
+  if (count == "0")
+  {
+    sql.clear();
+    sql.push_back("INSERT INTO `sys_replication`.`applier_state`"
+                  " (`master_id`, `last_applied_commit_id`, `status`) VALUES ("
+                  + boost::lexical_cast<string>(master_id)
+                  + ",0 , 'STOPPED')");
+    if (not executeSQL(sql))
+      return false;
+  }
+
+  return true;
+}
+
+
+bool ReplicationSchema::setInitialMaxCommitId(uint32_t master_id, uint64_t value)
 {
   vector<string> sql;
 
   sql.push_back("UPDATE `sys_replication`.`applier_state`"
                 " SET `last_applied_commit_id` = "
-                + lexical_cast<string>(value));
+                + lexical_cast<string>(value)
+                + " WHERE `master_id` = "
+                + lexical_cast<string>(master_id));
 
   return executeSQL(sql);
 }

=== modified file 'plugin/slave/replication_schema.h'
--- plugin/slave/replication_schema.h	2011-04-02 01:32:02 +0000
+++ plugin/slave/replication_schema.h	2011-07-06 23:18:34 +0000
@@ -44,9 +44,13 @@
    * another event from the master (which causes persistence of the value).
    * An edge case, but still possible.
    *
+   * @param[in] master_id Unique master identifier.
    * @param[in] value The initial value.
    */
-  bool setInitialMaxCommitId(uint64_t value);
+  bool setInitialMaxCommitId(uint32_t master_id, uint64_t value);
+
+  bool createInitialApplierRow(uint32_t master_id);
+  bool createInitialIORow(uint32_t master_id);
 };
 
 } /* namespace slave */

=== modified file 'plugin/slave/replication_slave.cc'
--- plugin/slave/replication_slave.cc	2011-04-02 01:32:02 +0000
+++ plugin/slave/replication_slave.cc	2011-07-06 23:18:34 +0000
@@ -20,8 +20,9 @@
 
 #include <config.h>
 #include <plugin/slave/replication_slave.h>
+#include <drizzled/errmsg_print.h>
 #include <drizzled/program_options/config_file.h>
-#include <drizzled/errmsg_print.h>
+#include <boost/lexical_cast.hpp>
 #include <boost/program_options.hpp>
 #include <fstream>
 #include <drizzled/plugin.h>
@@ -34,20 +35,26 @@
 namespace slave
 {
 
-/* Gets called after all plugins are initialized. */
 void ReplicationSlave::startup(Session &session)
 {
   (void)session;
   if (not initWithConfig())
   {
-    errmsg_printf(error::ERROR,
-                  _("Could not start slave services: %s\n"),
-                  getError().c_str());
+    errmsg_printf(error::ERROR, _("Could not start slave services: %s\n"),
+                  _error.c_str());
   }
   else
   {
+    /* Start the IO threads */
+    boost::unordered_map<uint32_t, Master *>::const_iterator it;
+    for (it= _masters.begin(); it != _masters.end(); ++it)
+    {
+      it->second->start();
+      /* Consumer must know server IDs */
+      _consumer.addMasterId(it->first);
+    }
+
     _consumer_thread= boost::thread(&QueueConsumer::run, &_consumer);
-    _producer_thread= boost::thread(&QueueProducer::run, &_producer);
   }
 }
 
@@ -56,15 +63,26 @@
   po::variables_map vm;
   po::options_description slave_options("Options for the slave plugin");
 
+  /* Common options */
   slave_options.add_options()
-    ("master-host", po::value<string>()->default_value(""))
-    ("master-port", po::value<uint16_t>()->default_value(3306))
-    ("master-user", po::value<string>()->default_value(""))
-    ("master-pass", po::value<string>()->default_value(""))
-    ("max-reconnects", po::value<uint32_t>()->default_value(10))
     ("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));
+    ("applier-thread-sleep", po::value<uint32_t>()->default_value(5))
+    ("ignore-errors", po::value<bool>()->default_value(false)->zero_tokens());
+
+  /* Master defining options */
+  for (size_t num= 1; num <= 10; num++)
+  {
+    string section("master");
+    section.append(boost::lexical_cast<string>(num));
+    slave_options.add_options()
+      ((section + ".master-host").c_str(), po::value<string>()->default_value(""))
+      ((section + ".master-port").c_str(), po::value<uint16_t>()->default_value(3306))
+      ((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 + ".max-commit-id").c_str(), po::value<uint64_t>());
+   }
 
   ifstream cf_stream(_config_file.c_str());
 
@@ -79,29 +97,62 @@
 
   po::notify(vm);
 
-  if (vm.count("master-host"))
-    _producer.setMasterHost(vm["master-host"].as<string>());
-
-  if (vm.count("master-port"))
-    _producer.setMasterPort(vm["master-port"].as<uint16_t>());
-
-  if (vm.count("master-user"))
-    _producer.setMasterUser(vm["master-user"].as<string>());
-
-  if (vm.count("master-pass"))
-    _producer.setMasterPassword(vm["master-pass"].as<string>());
-
-  if (vm.count("max-reconnects"))
-    _producer.setMaxReconnectAttempts(vm["max-reconnects"].as<uint32_t>());
-
-  if (vm.count("seconds-between-reconnects"))
-    _producer.setSecondsBetweenReconnects(vm["seconds-between-reconnects"].as<uint32_t>());
-
-  if (vm.count("io-thread-sleep"))
-    _producer.setSleepInterval(vm["io-thread-sleep"].as<uint32_t>());
+  /*
+   * We will support 10 masters. This loope effectively creates the Master
+   * objects as they are referenced.
+   *
+   * @todo Support a variable number of master hosts.
+   */
+  for (size_t num= 1; num <= 10; num++)
+  {
+    string section("master");
+    section.append(boost::lexical_cast<string>(num));
+
+    /* WARNING! Hack!
+     * We need to be able to determine when a master host is actually defined
+     * by the user vs. we are just using defaults. So if the hostname is ever
+     * the default value of "", then we'll assume that this section was not
+     * user defined.
+     */
+    if (vm[section + ".master-host"].as<string>() == "")
+      continue;
+
+    _masters[num]= new (std::nothrow) Master(num);
+
+    if (vm.count(section + ".master-host"))
+      master(num).producer().setMasterHost(vm[section + ".master-host"].as<string>());
+
+    if (vm.count(section + ".master-port"))
+      master(num).producer().setMasterPort(vm[section + ".master-port"].as<uint16_t>());
+
+    if (vm.count(section + ".master-user"))
+      master(num).producer().setMasterUser(vm[section + ".master-user"].as<string>());
+
+    if (vm.count(section + ".master-pass"))
+      master(num).producer().setMasterPassword(vm[section + ".master-pass"].as<string>());
+
+    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("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 */
   ReplicationSchema rs;
@@ -111,12 +162,20 @@
     return false;
   }
 
-  if (_initial_max_commit_id)
+  for (it= _masters.begin(); it != _masters.end(); ++it)
   {
-    if (not rs.setInitialMaxCommitId(_initial_max_commit_id))
+    /* make certain a row exists for each master */
+    rs.createInitialApplierRow(it->first);
+    rs.createInitialIORow(it->first);
+
+    uint64_t cachedValue= it->second->producer().cachedMaxCommitId();
+    if (cachedValue)
     {
-      _error= rs.getErrorMessage();
-      return false;
+      if (not rs.setInitialMaxCommitId(it->first, cachedValue))
+      {
+        _error= rs.getErrorMessage();
+        return false;
+      }
     }
   }
 

=== modified file 'plugin/slave/replication_slave.h'
--- plugin/slave/replication_slave.h	2011-04-02 01:32:02 +0000
+++ plugin/slave/replication_slave.h	2011-07-06 23:18:34 +0000
@@ -25,6 +25,7 @@
 #include <plugin/slave/replication_schema.h>
 #include <drizzled/plugin/daemon.h>
 #include <boost/thread.hpp>
+#include <boost/unordered_map.hpp>
 
 namespace drizzled
 {
@@ -40,65 +41,88 @@
 
   ReplicationSlave(const std::string &config)
     : drizzled::plugin::Daemon("Replication Slave"),
-      _config_file(config),
-      _initial_max_commit_id(0)
+      _config_file(config)
   {}
   
   ~ReplicationSlave()
   {
     _consumer_thread.interrupt();
-    _producer_thread.interrupt();
+
+    boost::unordered_map<uint32_t, Master *>::const_iterator it;
+
+    for (it= _masters.begin(); it != _masters.end(); ++it)
+    {
+      it->second->thread().interrupt();
+    }
   }
 
+  /** Gets called after all plugins are initialized */
   void startup(drizzled::Session &session);
 
-  /**
-   * Get the error message describing what went wrong during setup.
-   */
-  const std::string &getError() const
-  {
-    return _error;
-  }
-
-  /**
-   * Set the initial value for the slave's maximum commit ID.
-   *
-   * This value basically determines where to start retrieving events from
-   * the master. Normally this is computed automatically based on the
-   * contents of the queue and/or the last applied commit ID. This allows
-   * us to override those values and start from another point. E.g., new
-   * slave provisioning or skipping a trouble statement.
-   *
-   * @param[in] value The commit ID value.
-   */
-  void setMaxCommitId(uint64_t value)
-  {
-    /* must tell producer to set its cached value */
-    _producer.setCachedMaxCommitId(value);
-    /* setting this indicates that we should store it permanently */
-    _initial_max_commit_id= value;
-  }
-
 private:
+
+  /**
+   * Class representing a master server.
+   */
+  class Master
+  {
+  public:
+    Master(uint32_t id)
+    {
+      _producer.setMasterId(id);
+    }
+
+    QueueProducer &producer()
+    {
+      return _producer;
+    }
+  
+    boost::thread &thread()
+    {
+      return _producer_thread;
+    }
+
+    void start()
+    {
+      _producer_thread= boost::thread(&QueueProducer::run, &_producer);
+    }
+
+  private:
+    /** Manages a single master */
+    QueueProducer _producer;
+
+    /** I/O thread that will populate the work queue */
+    boost::thread _producer_thread;
+  };
+
+  /** Configuration file containing master info */
   std::string _config_file;
+
   std::string _error;
-
+  
+  /** Object to use with the consumer thread */
   QueueConsumer _consumer;
-  QueueProducer _producer;
 
-  /** Applier thread that will drain the work queue */
+  /**
+   * Applier thread that will drain the work queue.
+   * @todo Support more than one consumer thread.
+   */
   boost::thread _consumer_thread;
 
-  /** I/O thread that will populate the work queue */
-  boost::thread _producer_thread;
+  /** List of master objects, one per master */
+  boost::unordered_map<uint32_t, Master *> _masters;
 
-  uint64_t _initial_max_commit_id;
+  /** Convenience method to get object reference */
+  Master &master(size_t index)
+  {
+    return *(_masters[index]);
+  }
 
   /**
    * Initialize slave services with the given configuration file.
    *
-   * In case of an error during initialization, you can call the getError()
-   * method to get a string describing what went wrong.
+   * In case of an error during initialization, _error contains a
+   * string describing what went wrong.
    *
    * @retval true Success
    * @retval false Failure

=== modified file 'tests/lib/server_mgmt/drizzled.py'
--- tests/lib/server_mgmt/drizzled.py	2011-06-10 21:33:49 +0000
+++ tests/lib/server_mgmt/drizzled.py	2011-07-06 23:18:34 +0000
@@ -197,7 +197,8 @@
 
        """
 
-       config_data = [ "master-host=127.0.0.1"
+       config_data = [ "[master1]"
+                     , "master-host=127.0.0.1"
                      , "master-port=%d" %self.master_port
                      , "master-user=root"
                      , "master-pass=''"