Here comes a nasty patch, although I am not ready to push it yet. I will

first pull, merge,test, and get it to work.

The main change is the new replication code - now we have two slave threads
SQL thread and I/O thread. I have also re-written a lot of the code to 
prepare for multi-master implementation. 

I also documented IO_CACHE quite extensively and to some extend, THD class.


Makefile.am:
  moved tags target script into a separate file
include/my_sys.h:
  fixes in IO_CACHE for SEQ_READ_APPEND + some documentation
libmysqld/lib_sql.cc:
  updated replication locks, but now I see I did it wrong and it won't compile. Will fix
  before the push.
mysql-test/r/rpl000014.result:
  test result update
mysql-test/r/rpl000015.result:
  test result update
mysql-test/r/rpl000016.result:
  test result update
mysql-test/r/rpl_log.result:
  test result update
mysql-test/t/rpl000016-slave.sh:
  remove relay logs
mysql-test/t/rpl000017-slave.sh:
  remove relay logs
mysql-test/t/rpl_log.test:
  updated test
mysys/mf_iocache.c:
  IO_CACHE updates to make replication work
mysys/mf_iocache2.c:
  IO_CACHE update to make replication work
mysys/thr_mutex.c:
  cosmetic change
sql/item_func.cc:
  new replication code
sql/lex.h:
  new replication
sql/log.cc:
  new replication
sql/log_event.cc:
  new replication
sql/log_event.h:
  new replication
sql/mini_client.cc:
  new replication
sql/mini_client.h:
  new replication
sql/mysql_priv.h:
  new replication
sql/mysqld.cc:
  new replication
sql/repl_failsafe.cc:
  new replication
sql/slave.cc:
  new replication
sql/slave.h:
  new replication
sql/sql_class.cc:
  new replication
sql/sql_class.h:
  new replication
sql/sql_lex.h:
  new replication
sql/sql_parse.cc:
  new replication
sql/sql_repl.cc:
  new replication
sql/sql_repl.h:
  new replication
sql/sql_show.cc:
  new replication
sql/sql_yacc.yy:
  new replication
sql/stacktrace.c:
  more robust stack tracing
sql/structs.h:
  new replication code
BitKeeper/etc/ignore:
  Added mysql-test/r/rpl000002.eval mysql-test/r/rpl000014.eval mysql-test/r/rpl000015.eval mysql-test/r/rpl000016.eval mysql-test/r/slave-running.eval mysql-test/r/slave-stopped.eval to the ignore list

1 files changed, 287 insertions, 56 deletions

diff --git a/sql/slave.h b/sql/slave.h
index a5bc4d61309..9ad5c75a556 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -2,67 +2,265 @@
 #define SLAVE_H
 
 #include "mysql.h"
+#include "my_list.h"
 #define SLAVE_NET_TIMEOUT  3600
+#define MAX_SLAVE_ERRMSG   1024
+
+/*
+  The replication is accomplished by starting two threads - I/O
+  thread, and SQL thread. I/O thread is associated with its
+  MASTER_INFO struct, so MASTER_INFO can be viewed as I/O thread
+  descriptor. SQL thread is associated with RELAY_LOG_INFO struct.
+
+  I/O thread reads maintains a connection to the master, and reads log
+  events from the master as they arrive, queueing them by writing them
+  out into the temporary slave binary log (relay log). The SQL thread,
+  in turn, reads the slave binary log executing each event.
+
+  Relay log is needed to be able to handle situations when there is a large
+  backlog of unprocessed events from the master (eg. one particular update
+  takes a day to finish), and to be able to restart the slave server without
+  having to re-read the master updates.
+ */
 
 extern ulong slave_net_timeout, master_retry_count;
 extern char* slave_load_tmpdir;
+extern my_string master_info_file,relay_log_info_file;
+extern my_string opt_relay_logname, opt_relaylog_index_name;
+extern bool opt_skip_slave_start;
+struct st_master_info;
+
+#define LOCK_ACTIVE_MI { pthread_mutex_lock(&LOCK_active_mi); \
+ ++active_mi_in_use; \
+ pthread_mutex_unlock(&LOCK_active_mi);}
+
+#define UNLOCK_ACTIVE_MI { pthread_mutex_lock(&LOCK_active_mi); \
+ --active_mi_in_use; \
+ pthread_mutex_unlock(&LOCK_active_mi); }
+
+/*
+  st_relay_log_info contains information on the current relay log and
+  relay log offset, and master log name and log sequence corresponding to the
+  last update. Additionally, misc information specific to the SQL thread is
+  included.
+
+  st_relay_log_info is initialized from the slave.info file if such exists.
+  Otherwise, data members are intialized with defaults. The initialization is
+  done with init_relay_log_info() call.
+
+  The format of slave.info file:
+
+  relay_log_name
+  relay_log_pos
+  master_log_name
+  master_log_pos
+
+  To clean up, call end_relay_log_info()
+ */
+typedef struct st_relay_log_info
+{
+  // info_fd - file descriptor of the info file. set only during
+  //  initialization or clean up - safe to read anytime
+  // cur_log_fd - file descriptor of the current read  relay log, protected by
+  //  data_lock
+  File info_fd,cur_log_fd;
+  
+  // IO_CACHE of the info file - set only during init or end, safe to read
+  //  anytime
+  IO_CACHE info_file;
+
+  // name of current read relay log - protected by data_lock
+  char relay_log_name[FN_REFLEN];
+
+  // master log name corresponding to current read position - protected by
+  // data lock
+  char master_log_name[FN_REFLEN];
+
+  // original log position of last processed event - protected by data_lock
+  volatile uint32 master_log_pos;
 
+  // when we restart slave thread we need to have access to the previously
+  // created temporary tables. Modified only on init/end and by the SQL
+  // thread, read only by SQL thread, need no mutex
+  TABLE* save_temporary_tables;
+
+  // relay_log_pos - current offset in the relay log - protected by data_lock
+  // pending - in some cases we do not increment offset immediately after
+  //  processing an event, because the following event needs to be processed
+  // atomically together with this one ( so far, there is only one type of
+  // such event - Intvar_event that sets auto_increment value). However, once
+  // both events have been processed, we need to increment by the cumulative
+  // offset. pending stored the extra offset to be added to the position.
+  ulonglong relay_log_pos,pending;
+
+  // standard lock acquistion order to avoid deadlocks:
+  //  run_lock, data_lock, relay_log.LOCK_log,relay_log.LOCK_index
+  pthread_mutex_t data_lock,run_lock;
+
+  // start_cond is broadcast when SQL thread is started
+  // stop_cond - when stopped
+  // data_cond - when data protected by data_lock changes
+  pthread_cond_t start_cond,stop_cond,data_cond;
+
+  // if not set, the value of other members of the structure are undefined
+  bool inited;
+
+  // parent master info structure
+  struct st_master_info *mi;
+
+  // protected with internal locks
+  // must get data_lock when resetting the logs
+  MYSQL_LOG relay_log;
+  LOG_INFO linfo;
+  IO_CACHE cache_buf,*cur_log;
+  
+  /* needed to deal properly with cur_log getting closed and re-opened with
+     a different log under our feet
+  */
+  int cur_log_init_count;
+  
+  volatile bool abort_slave, slave_running;
+// needed for problems when slave stops and
+// we want to restart it skipping one or more events in the master log that
+// have caused errors, and have been manually applied by DBA already
+  volatile uint32 slave_skip_counter;
+#ifndef DBUG_OFF
+  int events_till_abort;
+#endif  
+  int last_slave_errno;
+  char last_slave_error[MAX_SLAVE_ERRMSG];
+  THD* sql_thd;
+  bool log_pos_current;
+  
+  st_relay_log_info():info_fd(-1),cur_log_fd(-1),inited(0),
+		      cur_log_init_count(0),
+		      log_pos_current(0)
+    {
+      relay_log_name[0] = master_log_name[0] = 0;
+      pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
+      pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
+      pthread_cond_init(&data_cond, NULL);
+      pthread_cond_init(&start_cond, NULL);
+      pthread_cond_init(&stop_cond, NULL);
+    }
+  ~st_relay_log_info()
+    {
+      pthread_mutex_destroy(&run_lock);
+      pthread_mutex_destroy(&data_lock);
+      pthread_cond_destroy(&data_cond);
+      pthread_cond_destroy(&start_cond);
+      pthread_cond_destroy(&stop_cond);
+    }
+  inline void inc_pending(ulonglong val)
+  {
+    pending += val;
+  }
+  // TODO: this probably needs to be fixed
+  inline void inc_pos(ulonglong val, uint32 log_pos, bool skip_lock=0)
+  {
+    if (!skip_lock)
+      pthread_mutex_lock(&data_lock);
+    relay_log_pos += val+pending;
+    pending = 0;
+    if (log_pos)
+      master_log_pos = log_pos+val;
+    pthread_cond_broadcast(&data_cond);
+    if (!skip_lock)
+      pthread_mutex_unlock(&data_lock);
+  }
+  // thread safe read of position - not needed if we are in the slave thread,
+  // but required otherwise
+  inline void read_pos(ulonglong& var)
+  {
+    pthread_mutex_lock(&data_lock);
+    var = relay_log_pos;
+    pthread_mutex_unlock(&data_lock);
+  }
+
+  int wait_for_pos(THD* thd, String* log_name, ulonglong log_pos);
+} RELAY_LOG_INFO;
+
+// repopen_relay_log() is called when we notice that the current "hot" log
+// got rotated under our feet
+IO_CACHE* reopen_relay_log(RELAY_LOG_INFO* rli, const char** errmsg);
+Log_event* next_event(RELAY_LOG_INFO* rli);
+
+/* st_master_info contains information about how to connect to a master,
+   current master log name, and current log offset, as well as misc
+   control variables
+
+   st_master_info is initialized once from the master.info file if such
+   exists. Otherwise, data members corresponding to master.info fields are
+   initialized with defaults specified by master-* options. The initialization
+   is done through init_master_info() call.
+
+   The format of master.info file:
+
+   log_name
+   log_pos
+   master_host
+   master_user
+   master_pass
+   master_port
+   master_connect_retry
+
+   To write out the contents of master.info file to disk ( needed every
+   time we read and queue data from the master ), a call to
+   flush_master_info() is required.
+
+   To clean up, call end_master_info()
+   
+*/
+   
 typedef struct st_master_info
 {
-  char log_file_name[FN_REFLEN];
-  ulonglong pos,pending;
-  File fd; // we keep the file open, so we need to remember the file pointer
+  char master_log_name[FN_REFLEN];
+  
+  ulonglong master_log_pos;
+  File fd; 
   IO_CACHE file;
+  
   // the variables below are needed because we can change masters on the fly
   char host[HOSTNAME_LENGTH+1];
   char user[USERNAME_LENGTH+1];
   char password[HASH_PASSWORD_LENGTH+1];
   uint port;
   uint connect_retry;
-  uint32 last_log_seq; // log sequence number of last processed event
-  pthread_mutex_t lock;
-  pthread_cond_t cond;
+  pthread_mutex_t data_lock,run_lock;
+  pthread_cond_t data_cond,start_cond,stop_cond;
   bool inited;
   bool old_format; /* master binlog is in 3.23 format */
+  RELAY_LOG_INFO rli;
+#ifndef DBUG_OFF
+  int events_till_abort;
+#endif
+  volatile bool abort_slave, slave_running;
+  THD* io_thd;
   
-  st_master_info():pending(0),fd(-1),last_log_seq(0),inited(0),
-		   old_format(0)
+  st_master_info():fd(-1),inited(0),
+		   old_format(0),io_thd(0)
   {
     host[0] = 0; user[0] = 0; password[0] = 0;
-    pthread_mutex_init(&lock, MY_MUTEX_INIT_FAST);
-    pthread_cond_init(&cond, NULL);
+    pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
+    pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
+    pthread_cond_init(&data_cond, NULL);
+    pthread_cond_init(&start_cond, NULL);
+    pthread_cond_init(&stop_cond, NULL);
   }
 
   ~st_master_info()
   {
-    pthread_mutex_destroy(&lock);
-    pthread_cond_destroy(&cond);
-  }
-  inline void inc_pending(ulonglong val)
-  {
-    pending += val;
-  }
-  inline void inc_pos(ulonglong val, uint32 log_seq)
-  {
-    pthread_mutex_lock(&lock);
-    pos += val + pending;
-    pending = 0;
-    last_log_seq = log_seq;
-    pthread_cond_broadcast(&cond);
-    pthread_mutex_unlock(&lock);
-  }
-  // thread safe read of position - not needed if we are in the slave thread,
-  // but required otherwise
-  inline void read_pos(ulonglong& var)
-  {
-    pthread_mutex_lock(&lock);
-    var = pos;
-    pthread_mutex_unlock(&lock);
+    pthread_mutex_destroy(&run_lock);
+    pthread_mutex_destroy(&data_lock);
+    pthread_cond_destroy(&data_cond);
+    pthread_cond_destroy(&start_cond);
+    pthread_cond_destroy(&stop_cond);
   }
 
-  int wait_for_pos(THD* thd, String* log_name, ulonglong log_pos);
 } MASTER_INFO;
 
+int queue_event(MASTER_INFO* mi,const char* buf,uint event_len);
+
 typedef struct st_table_rule_ent
 {
   char* db;
@@ -74,22 +272,52 @@ typedef struct st_table_rule_ent
 #define TABLE_RULE_ARR_SIZE   16
 #define MAX_SLAVE_ERRMSG      1024
 
-#define RPL_LOG_NAME (glob_mi.log_file_name[0] ? glob_mi.log_file_name :\
+#define RPL_LOG_NAME (rli->master_log_name[0] ? rli->master_log_name :\
+ "FIRST")
+#define IO_RPL_LOG_NAME (mi->master_log_name[0] ? mi->master_log_name :\
  "FIRST")
 
+/* masks for start/stop operations on io and sql slave threads */
+#define SLAVE_IO  1
+#define SLAVE_SQL 2
+#define SLAVE_FORCE_ALL 4 /* if this is set, if  first gives an
+				   error, second will be tried. Otherwise,
+				   if first fails, we fail
+				*/
 
+int init_slave();
 int flush_master_info(MASTER_INFO* mi);
+int flush_relay_log_info(RELAY_LOG_INFO* rli);
 int register_slave_on_master(MYSQL* mysql);
+int terminate_slave_threads(MASTER_INFO* mi, int thread_mask,
+			     bool skip_lock = 0);
+int terminate_slave_thread(THD* thd, pthread_mutex_t* term_mutex,
+			   pthread_mutex_t* cond_lock,
+			   pthread_cond_t* term_cond,
+			   volatile bool* slave_running);
+int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
+			MASTER_INFO* mi, const char* master_info_fname,
+			const char* slave_info_fname, int thread_mask);
+/* cond_lock is usually same as start_lock. It is needed for the case when
+   start_lock is 0 which happens if start_slave_thread() is called already
+   inside the start_lock section, but at the same time we want a
+   pthread_cond_wait() on start_cond,start_lock
+*/
+int start_slave_thread(pthread_handler h_func, pthread_mutex_t* start_lock,
+		       pthread_mutex_t *cond_lock,
+		       pthread_cond_t* start_cond,
+		       volatile bool* slave_running,
+		       MASTER_INFO* mi);
 
 int mysql_table_dump(THD* thd, const char* db,
 		     const char* tbl_name, int fd = -1);
 // if fd is -1, dump to NET
 
-int fetch_nx_table(THD* thd, const char* db_name, const char* table_name,
+int fetch_master_table(THD* thd, const char* db_name, const char* table_name,
 		   MASTER_INFO* mi, MYSQL* mysql);
 // retrieve non-exitent table from master
 
-int show_master_info(THD* thd);
+int show_master_info(THD* thd, MASTER_INFO* mi);
 int show_binlog_info(THD* thd);
 
 int tables_ok(THD* thd, TABLE_LIST* tables);
@@ -105,29 +333,31 @@ int add_wild_table_rule(DYNAMIC_ARRAY* a, const char* table_spec);
 void init_table_rule_hash(HASH* h, bool* h_inited);
 void init_table_rule_array(DYNAMIC_ARRAY* a, bool* a_inited);
 char* rewrite_db(char* db);
-int check_expected_error(THD* thd, int error_code);
+int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int error_code);
 void skip_load_data_infile(NET* net);
-void slave_print_error(int err_code, const char* msg, ...);
+void slave_print_error(RELAY_LOG_INFO* rli,int err_code, const char* msg, ...);
 
 void end_slave(); // clean up
-int init_master_info(MASTER_INFO* mi);
+int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
+		     const char* slave_info_fname);
 void end_master_info(MASTER_INFO* mi);
-extern bool opt_log_slave_updates ;
-pthread_handler_decl(handle_slave,arg);
-extern bool volatile abort_loop, abort_slave, slave_running;
-extern uint32 slave_skip_counter;
-// needed for problems when slave stops and
-// we want to restart it skipping one or more events in the master log that
-// have caused errors, and have been manually applied by DBA already
+int init_relay_log_info(RELAY_LOG_INFO* rli, const char* info_fname);
+void end_relay_log_info(RELAY_LOG_INFO* rli);
+void lock_slave_threads(MASTER_INFO* mi);
+void unlock_slave_threads(MASTER_INFO* mi);
+void init_thread_mask(int* mask,MASTER_INFO* mi,bool inverse);
+int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log,ulonglong pos,
+		       bool need_data_lock, const char** errmsg);
 
-extern int last_slave_errno;
-#ifndef DBUG_OFF
-extern int events_till_abort;
-#endif
-extern char last_slave_error[MAX_SLAVE_ERRMSG];
-extern pthread_t slave_real_id;
-extern THD* slave_thd;
-extern MASTER_INFO glob_mi;
+int purge_relay_logs(RELAY_LOG_INFO* rli,bool just_reset,const char** errmsg);
+
+extern bool opt_log_slave_updates ;
+pthread_handler_decl(handle_slave_io,arg);
+pthread_handler_decl(handle_slave_sql,arg);
+extern bool volatile abort_loop;
+extern MASTER_INFO main_mi, *active_mi; // active_mi for multi-master
+extern volatile int active_mi_in_use;
+extern LIST master_list;
 extern HASH replicate_do_table, replicate_ignore_table;
 extern DYNAMIC_ARRAY  replicate_wild_do_table, replicate_wild_ignore_table;
 extern bool do_table_inited, ignore_table_inited,
@@ -141,7 +371,8 @@ extern int disconnect_slave_event_count, abort_slave_event_count ;
 // the master variables are defaults read from my.cnf or command line
 extern uint master_port, master_connect_retry, report_port;
 extern my_string master_user, master_password, master_host,
-  master_info_file, report_user, report_host, report_password;
+  master_info_file, relay_log_info_file, report_user, report_host,
+  report_password;
 
 extern I_List<i_string> replicate_do_db, replicate_ignore_db;
 extern I_List<i_string_pair> replicate_rewrite_db;