diff --git a/repl-tests/test-dump/run.test b/repl-tests/test-dump/run.test index 612db3fc6cf..22fa37e5ffd 100644 --- a/repl-tests/test-dump/run.test +++ b/repl-tests/test-dump/run.test @@ -1,4 +1,11 @@ source ../include/master-slave.inc; +connection slave; +!slave stop; +flush slave; +connection master; +flush master; +connection slave; +slave start; connection master; use test; drop table if exists words; @@ -7,6 +14,7 @@ load data infile '/usr/dict/words' into table words; drop table if exists words1; create table words1 (word char(20) not null); load data infile '/usr/dict/words' into table words1; +sleep 5; connection slave; use test; drop table if exists words; diff --git a/sql/Makefile.am b/sql/Makefile.am index 58ab8f69304..20208b10da5 100644 --- a/sql/Makefile.am +++ b/sql/Makefile.am @@ -48,7 +48,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \ opt_range.h \ sql_select.h structs.h table.h sql_udf.h hash_filo.h\ lex.h lex_symbol.h sql_acl.h sql_crypt.h md5.h \ - log_event.h mini_client.h + log_event.h mini_client.h sql_repl.h mysqld_SOURCES = sql_lex.cc \ item.cc item_sum.cc item_buff.cc item_func.cc \ item_cmpfunc.cc item_strfunc.cc item_timefunc.cc \ @@ -69,7 +69,7 @@ mysqld_SOURCES = sql_lex.cc \ sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \ sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \ sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \ - slave.cc \ + slave.cc sql_repl.cc \ md5.c log_event.cc mini_client.cc mini_client_errors.c gen_lex_hash_SOURCES = gen_lex_hash.cc gen_lex_hash_LDADD = $(LDADD) $(CXXLDFLAGS) diff --git a/sql/log_event.cc b/sql/log_event.cc index 45a3e749e9e..1d45a82648a 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -22,16 +22,6 @@ #include "mysql_priv.h" #endif /* MYSQL_CLIENT */ -#define LOG_EVENT_HEADER_LEN 9 -#define QUERY_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + sizeof(uchar)) -#define LOAD_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + \ - + sizeof(uint32) + 2 + sizeof(uint32)) -#define EVENT_LEN_OFFSET 5 -#define EVENT_TYPE_OFFSET 4 -#define MAX_EVENT_LEN 4*1024*1024 -#define QUERY_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN+QUERY_HEADER_LEN -#define ROTATE_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN -#define LOAD_EVENT_OVERHEAD (LOG_EVENT_HEADER_LEN+LOAD_HEADER_LEN+sizeof(sql_ex_info)) static void pretty_print_char(FILE* file, int c) { @@ -71,6 +61,8 @@ int Log_event::write_header(FILE* file) int4store(pos, when); // timestamp pos += 4; *pos++ = get_type_code(); // event type code + int4store(pos, server_id); + pos += 4; int4store(pos, get_data_size() + LOG_EVENT_HEADER_LEN); pos += 4; return (my_fwrite(file, (byte*) buf, (uint) (pos - buf), @@ -106,16 +98,19 @@ int Log_event::read_log_event(FILE* file, String* packet) Log_event* Log_event::read_log_event(FILE* file) { time_t timestamp; - char buf[5]; + uint32 server_id; + + char buf[LOG_EVENT_HEADER_LEN-4]; if (my_fread(file, (byte *) buf, sizeof(buf), MY_NABP)) return NULL; timestamp = uint4korr(buf); - + server_id = uint4korr(buf + 5); + switch(buf[EVENT_TYPE_OFFSET]) { case QUERY_EVENT: { - Query_log_event* q = new Query_log_event(file, timestamp); + Query_log_event* q = new Query_log_event(file, timestamp, server_id); if (!q->query) { delete q; @@ -127,7 +122,7 @@ Log_event* Log_event::read_log_event(FILE* file) case LOAD_EVENT: { - Load_log_event* l = new Load_log_event(file, timestamp); + Load_log_event* l = new Load_log_event(file, timestamp, server_id); if (!l->table_name) { delete l; @@ -140,7 +135,7 @@ Log_event* Log_event::read_log_event(FILE* file) case ROTATE_EVENT: { - Rotate_log_event* r = new Rotate_log_event(file, timestamp); + Rotate_log_event* r = new Rotate_log_event(file, timestamp, server_id); if (!r->new_log_ident) { delete r; @@ -152,7 +147,7 @@ Log_event* Log_event::read_log_event(FILE* file) case INTVAR_EVENT: { - Intvar_log_event* e = new Intvar_log_event(file, timestamp); + Intvar_log_event* e = new Intvar_log_event(file, timestamp, server_id); if (e->type == INVALID_INT_EVENT) { delete e; @@ -162,8 +157,8 @@ Log_event* Log_event::read_log_event(FILE* file) return e; } - case START_EVENT: return new Start_log_event(file, timestamp); - case STOP_EVENT: return new Stop_log_event(file, timestamp); + case START_EVENT: return new Start_log_event(file, timestamp, server_id); + case STOP_EVENT: return new Stop_log_event(file, timestamp, server_id); default: return NULL; } @@ -221,12 +216,23 @@ Log_event* Log_event::read_log_event(const char* buf, int max_buf) return NULL; } -void Log_event::print_timestamp(FILE* file) +void Log_event::print_header(FILE* file) +{ + fputc('#', file); + print_timestamp(file); + fprintf(file, " server id %d ", server_id); +} + +void Log_event::print_timestamp(FILE* file, time_t* ts = 0) { struct tm tm_tmp; - localtime_r(&when,&tm_tmp); + if(!ts) + { + ts = &when; + } + localtime_r(ts,&tm_tmp); - fprintf(file,"#%02d%02d%02d %2d:%02d:%02d", + fprintf(file,"%02d%02d%02d %2d:%02d:%02d", tm_tmp.tm_year % 100, tm_tmp.tm_mon+1, tm_tmp.tm_mday, @@ -241,8 +247,11 @@ void Start_log_event::print(FILE* file, bool short_form) if (short_form) return; - print_timestamp(file); - fprintf(file, "\tStart\n"); + print_header(file); + fprintf(file, "\tStart: binlog v %d, server v %s created ", binlog_version, + server_version); + print_timestamp(file, (time_t*)&created); + fputc('\n', file); fflush(file); } @@ -251,7 +260,7 @@ void Stop_log_event::print(FILE* file, bool short_form) if (short_form) return; - print_timestamp(file); + print_header(file); fprintf(file, "\tStop\n"); fflush(file); } @@ -261,7 +270,7 @@ void Rotate_log_event::print(FILE* file, bool short_form) if (short_form) return; - print_timestamp(file); + print_header(file); fprintf(file, "\tRotate to "); if (new_log_ident) my_fwrite(file, (byte*) new_log_ident, (uint)ident_len, @@ -270,8 +279,9 @@ void Rotate_log_event::print(FILE* file, bool short_form) fflush(file); } -Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg): - Log_event(when_arg),new_log_ident(NULL),alloced(0) +Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg, + uint32 server_id): + Log_event(when_arg, 0, 0, server_id),new_log_ident(NULL),alloced(0) { char *tmp_ident; char buf[4]; @@ -298,6 +308,14 @@ Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg): alloced = 1; } +Start_log_event::Start_log_event(const char* buf) :Log_event(buf) +{ + buf += EVENT_LEN_OFFSET + 4; // skip even length + binlog_version = uint2korr(buf); + memcpy(server_version, buf + 2, sizeof(server_version)); + created = uint4korr(buf + 2 + sizeof(server_version)); +} + Rotate_log_event::Rotate_log_event(const char* buf, int max_buf): Log_event(buf),new_log_ident(NULL),alloced(0) { @@ -322,8 +340,9 @@ int Rotate_log_event::write_data(FILE* file) return 0; } -Query_log_event::Query_log_event(FILE* file, time_t when_arg): - Log_event(when_arg),data_buf(0),query(NULL),db(NULL) +Query_log_event::Query_log_event(FILE* file, time_t when_arg, + uint32 server_id): + Log_event(when_arg,0,0,server_id),data_buf(0),query(NULL),db(NULL) { char buf[QUERY_HEADER_LEN + 4]; ulong data_len; @@ -382,7 +401,7 @@ void Query_log_event::print(FILE* file, bool short_form) { if (!short_form) { - print_timestamp(file); + print_header(file); fprintf(file, "\tQuery\tthread_id=%lu\texec_time=%lu\n", (ulong) thread_id, (ulong) exec_time); } @@ -414,8 +433,9 @@ int Query_log_event::write_data(FILE* file) return 0; } -Intvar_log_event:: Intvar_log_event(FILE* file, time_t when_arg) - :Log_event(when_arg), type(INVALID_INT_EVENT) +Intvar_log_event:: Intvar_log_event(FILE* file, time_t when_arg, + uint32 server_id) + :Log_event(when_arg,0,0,server_id), type(INVALID_INT_EVENT) { my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length char buf[9]; @@ -444,7 +464,7 @@ void Intvar_log_event::print(FILE* file, bool short_form) char llbuff[22]; if(!short_form) { - print_timestamp(file); + print_header(file); fprintf(file, "\tIntvar\n"); } @@ -493,8 +513,9 @@ int Load_log_event::write_data(FILE* file __attribute__((unused))) return 0; } -Load_log_event::Load_log_event(FILE* file, time_t when): - Log_event(when),data_buf(0),num_fields(0),fields(0),field_lens(0),field_block_len(0), +Load_log_event::Load_log_event(FILE* file, time_t when, uint32 server_id): + Log_event(when,0,0,server_id),data_buf(0),num_fields(0), + fields(0),field_lens(0),field_block_len(0), table_name(0),db(0),fname(0) { @@ -539,7 +560,8 @@ Load_log_event::Load_log_event(FILE* file, time_t when): } Load_log_event::Load_log_event(const char* buf, int max_buf): - Log_event(when),data_buf(0),num_fields(0),fields(0),field_lens(0),field_block_len(0), + Log_event(when,0,0,server_id),data_buf(0),num_fields(0),fields(0), + field_lens(0),field_block_len(0), table_name(0),db(0),fname(0) { @@ -594,7 +616,7 @@ void Load_log_event::print(FILE* file, bool short_form) { if (!short_form) { - print_timestamp(file); + print_header(file); fprintf(file, "\tQuery\tthread_id=%d\texec_time=%ld\n", thread_id, exec_time); } diff --git a/sql/log_event.h b/sql/log_event.h index 82ab462bdf5..871258885cf 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -28,6 +28,18 @@ #define LOG_READ_MEM -5 #define LOG_EVENT_OFFSET 4 +#define BINLOG_VERSION 1 + +#define LOG_EVENT_HEADER_LEN 13 +#define QUERY_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + sizeof(uchar)) +#define LOAD_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + \ + + sizeof(uint32) + 2 + sizeof(uint32)) +#define EVENT_LEN_OFFSET 9 +#define EVENT_TYPE_OFFSET 4 +#define MAX_EVENT_LEN 4*1024*1024 +#define QUERY_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN+QUERY_HEADER_LEN +#define ROTATE_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN +#define LOAD_EVENT_OVERHEAD (LOG_EVENT_HEADER_LEN+LOAD_HEADER_LEN+sizeof(sql_ex_info)) enum Log_event_type { START_EVENT = 1, QUERY_EVENT =2, @@ -40,25 +52,32 @@ enum Int_event_type { INVALID_INT_EVENT = 0, LAST_INSERT_ID_EVENT = 1, INSERT_ID class String; #endif +extern uint32 server_id; + class Log_event { public: time_t when; ulong exec_time; - int valid_exec_time; // if false, the exec time setting is bogus and needs + int valid_exec_time; // if false, the exec time setting is bogus + uint32 server_id; int write(FILE* file); int write_header(FILE* file); virtual int write_data(FILE* file __attribute__((unused))) { return 0; } virtual Log_event_type get_type_code() = 0; Log_event(time_t when_arg, ulong exec_time_arg = 0, - int valid_exec_time_arg = 0): when(when_arg), - exec_time(exec_time_arg), valid_exec_time(valid_exec_time_arg) {} + int valid_exec_time_arg = 0, uint32 server_id = 0): when(when_arg), + exec_time(exec_time_arg), valid_exec_time(valid_exec_time_arg) + { + if(!server_id) this->server_id = ::server_id; + else this->server_id = server_id; + } - Log_event(const char* buf): valid_exec_time(1) + Log_event(const char* buf): valid_exec_time(0) { when = uint4korr(buf); - exec_time = uint4korr(buf + 5); + server_id = uint4korr(buf + 5); } virtual ~Log_event() {} @@ -66,7 +85,8 @@ public: virtual int get_data_size() { return 0;} virtual void print(FILE* file, bool short_form = 0) = 0; - void print_timestamp(FILE* file); + void print_timestamp(FILE* file, time_t *ts = 0); + void print_header(FILE* file); static Log_event* read_log_event(FILE* file); static Log_event* read_log_event(const char* buf, int max_buf); @@ -93,7 +113,7 @@ public: #if !defined(MYSQL_CLIENT) THD* thd; Query_log_event(THD* thd_arg, const char* query_arg): - Log_event(thd_arg->start_time), data_buf(0), + Log_event(thd_arg->start_time,0,0,thd_arg->server_id), data_buf(0), query(query_arg), db(thd_arg->db), q_len(thd_arg->query_length), thread_id(thd_arg->thread_id), thd(thd_arg) { @@ -105,7 +125,7 @@ public: } #endif - Query_log_event(FILE* file, time_t when); + Query_log_event(FILE* file, time_t when, uint32 server_id); Query_log_event(const char* buf, int max_buf); ~Query_log_event() { @@ -244,7 +264,7 @@ public: void set_fields(List &fields); #endif - Load_log_event(FILE* file, time_t when); + Load_log_event(FILE* file, time_t when, uint32 server_id); Load_log_event(const char* buf, int max_buf); ~Load_log_event() { @@ -269,21 +289,52 @@ public: void print(FILE* file, bool short_form = 0); }; +extern char server_version[50]; class Start_log_event: public Log_event { public: - Start_log_event() :Log_event(time(NULL)) - {} - Start_log_event(FILE* file, time_t when_arg) :Log_event(when_arg) + uint16 binlog_version; + char server_version[50]; + uint32 created; + + Start_log_event() :Log_event(time(NULL)),binlog_version(BINLOG_VERSION) { + created = when; + memcpy(server_version, ::server_version, sizeof(server_version)); + } + Start_log_event(FILE* file, time_t when_arg, uint32 server_id) : + Log_event(when_arg, 0, 0, server_id) + { + char buf[sizeof(server_version) + sizeof(binlog_version) + + sizeof(created)]; my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length + if (my_fread(file, (byte*) buf, sizeof(buf), MYF(MY_NABP | MY_WME))) + return; + binlog_version = uint2korr(buf); + memcpy(server_version, buf + 2, sizeof(server_version)); + created = uint4korr(buf + 2 + sizeof(server_version)); } - Start_log_event(const char* buf) :Log_event(buf) - { - } + Start_log_event(const char* buf); + ~Start_log_event() {} Log_event_type get_type_code() { return START_EVENT;} + int write_data(FILE* file) + { + if(my_fwrite(file, (byte*) &binlog_version, sizeof(binlog_version), + MYF(MY_NABP | MY_WME)) || + my_fwrite(file, (byte*) server_version, sizeof(server_version), + MYF(MY_NABP | MY_WME)) || + my_fwrite(file, (byte*) &created, sizeof(created), + MYF(MY_NABP | MY_WME))) + return -1; + return 0; + } + int get_data_size() + { + return sizeof(binlog_version) + sizeof(server_version) + + sizeof(created); + } void print(FILE* file, bool short_form = 0); }; @@ -295,7 +346,7 @@ public: Intvar_log_event(uchar type_arg, ulonglong val_arg) :Log_event(time(NULL)),val(val_arg),type(type_arg) {} - Intvar_log_event(FILE* file, time_t when); + Intvar_log_event(FILE* file, time_t when, uint32 server_id); Intvar_log_event(const char* buf); ~Intvar_log_event() {} Log_event_type get_type_code() { return INTVAR_EVENT;} @@ -311,7 +362,8 @@ class Stop_log_event: public Log_event public: Stop_log_event() :Log_event(time(NULL)) {} - Stop_log_event(FILE* file, time_t when_arg): Log_event(when_arg) + Stop_log_event(FILE* file, time_t when_arg, uint32 server_id): + Log_event(when_arg,0,0,server_id) { my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length } @@ -337,7 +389,7 @@ public: alloced(0) {} - Rotate_log_event(FILE* file, time_t when) ; + Rotate_log_event(FILE* file, time_t when, uint32 server_id) ; Rotate_log_event(const char* buf, int max_buf); ~Rotate_log_event() { diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h index dd87f40efe9..1ebfd011a1c 100644 --- a/sql/mysql_priv.h +++ b/sql/mysql_priv.h @@ -437,6 +437,7 @@ void sql_perror(const char *message); void sql_print_error(const char *format,...) __attribute__ ((format (printf, 1, 2))); +extern uint32 server_id; extern char mysql_data_home[2],server_version[50],max_sort_char, mysql_real_data_home[]; extern my_string mysql_unix_port,mysql_tmpdir; @@ -463,7 +464,7 @@ extern pthread_mutex_t LOCK_mysql_create_db,LOCK_Acl,LOCK_open, LOCK_thread_count,LOCK_mapped_file,LOCK_user_locks, LOCK_status, LOCK_grant, LOCK_error_log, LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone, - LOCK_binlog_update, LOCK_slave; + LOCK_binlog_update, LOCK_slave, LOCK_server_id; extern pthread_cond_t COND_refresh,COND_thread_count, COND_binlog_update, COND_slave_stopped; extern pthread_attr_t connection_attrib; diff --git a/sql/mysqlbinlog.cc b/sql/mysqlbinlog.cc index dd34230bc1a..2d1bde21f3f 100644 --- a/sql/mysqlbinlog.cc +++ b/sql/mysqlbinlog.cc @@ -29,6 +29,8 @@ #define CLIENT_CAPABILITIES (CLIENT_LONG_PASSWORD | CLIENT_LONG_FLAG | CLIENT_LOCAL_FILES) +char server_version[50]; +uint32 server_id = 0; // needed by net_serv.c ulong bytes_sent = 0L, bytes_received = 0L; @@ -268,8 +270,9 @@ static void dump_remote_log_entries(const char* logname) int4store(buf, position); int2store(buf + 4, binlog_flags); len = (uint) strlen(logname); - memcpy(buf + 6, logname,len); - if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 6, 1)) + int4store(buf + 6, 0); + memcpy(buf + 10, logname,len); + if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1)) die("Error sending the log dump command"); for(;;) diff --git a/sql/mysqld.cc b/sql/mysqld.cc index aa3aa0b0bf4..38d8b5ddcab 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -142,6 +142,7 @@ static uint handler_count; static bool opt_console=0; #endif +static bool opt_skip_slave_start = 0; // if set, slave is not autostarted static ulong opt_specialflag=SPECIAL_ENGLISH; static my_socket unix_sock= INVALID_SOCKET,ip_sock= INVALID_SOCKET; static ulong back_log,connect_timeout,concurrency; @@ -180,6 +181,7 @@ I_List replicate_do_db, replicate_ignore_db; // allow the user to tell us which db to replicate and which to ignore I_List binlog_do_db, binlog_ignore_db; +uint32 server_id = 0; // server id for replication uint mysql_port; uint test_flags, select_errors=0, dropping_tables=0,ha_open_options=0; uint volatile thread_count=0, thread_running=0, kill_cached_threads=0, @@ -246,7 +248,7 @@ pthread_mutex_t LOCK_mysql_create_db, LOCK_Acl, LOCK_open, LOCK_thread_count, LOCK_error_log, LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create, LOCK_flush, LOCK_crypt, LOCK_bytes_sent, LOCK_bytes_received, - LOCK_binlog_update, LOCK_slave; + LOCK_binlog_update, LOCK_slave, LOCK_server_id; pthread_cond_t COND_refresh,COND_thread_count,COND_flush, COND_binlog_update, COND_slave_stopped; @@ -1393,6 +1395,7 @@ int main(int argc, char **argv) (void) pthread_mutex_init(&LOCK_timezone,NULL); (void) pthread_mutex_init(&LOCK_binlog_update, NULL); (void) pthread_mutex_init(&LOCK_slave, NULL); + (void) pthread_mutex_init(&LOCK_server_id, NULL); (void) pthread_cond_init(&COND_binlog_update, NULL); (void) pthread_cond_init(&COND_slave_stopped, NULL); @@ -1482,17 +1485,23 @@ int main(int argc, char **argv) LOG_NEW); if (opt_bin_log) { - if (!opt_bin_logname) - { - char tmp[FN_REFLEN]; - strnmov(tmp,hostname,FN_REFLEN-5); - strmov(strcend(tmp,'.'),"-bin"); - opt_bin_logname=my_strdup(tmp,MYF(MY_WME)); - } - mysql_bin_log.set_index_file_name(opt_binlog_index_name); - open_log(&mysql_bin_log, hostname, opt_bin_logname, "-bin", - LOG_BIN); + if(server_id) + { + if (!opt_bin_logname) + { + char tmp[FN_REFLEN]; + strnmov(tmp,hostname,FN_REFLEN-5); + strmov(strcend(tmp,'.'),"-bin"); + opt_bin_logname=my_strdup(tmp,MYF(MY_WME)); + } + mysql_bin_log.set_index_file_name(opt_binlog_index_name); + open_log(&mysql_bin_log, hostname, opt_bin_logname, "-bin", + LOG_BIN); + } + else + sql_print_error("Server id is not set - binary logging disabled"); } + if (opt_slow_log) open_log(&mysql_slow_log, hostname, opt_slow_logname, "-slow.log", LOG_NORMAL); @@ -1596,10 +1605,15 @@ int main(int argc, char **argv) // slave thread if(master_host) { - pthread_t hThread; - if(pthread_create(&hThread, &connection_attrib, handle_slave, 0)) - sql_print_error("Warning: Can't create thread to handle slave"); - + if(server_id) + { + pthread_t hThread; + if(!opt_skip_slave_start && + pthread_create(&hThread, &connection_attrib, handle_slave, 0)) + sql_print_error("Warning: Can't create thread to handle slave"); + } + else + sql_print_error("Server id is not set, slave thread will not be started"); } printf(ER(ER_READY),my_progname,server_version,""); @@ -2201,7 +2215,8 @@ enum options { OPT_REPLICATE_DO_DB, OPT_REPLICATE_IGNORE_DB, OPT_LOG_SLAVE_UPDATES, OPT_BINLOG_DO_DB, OPT_BINLOG_IGNORE_DB, OPT_WANT_CORE, - OPT_SKIP_CONCURRENT_INSERT, OPT_MEMLOCK, OPT_MYISAM_RECOVER + OPT_SKIP_CONCURRENT_INSERT, OPT_MEMLOCK, OPT_MYISAM_RECOVER, + OPT_REPLICATE_REWRITE_DB, OPT_SERVER_ID, OPT_SKIP_SLAVE_START }; static struct option long_options[] = { @@ -2266,8 +2281,11 @@ static struct option long_options[] = { {"port", required_argument, 0, 'P'}, {"replicate-do-db", required_argument, 0, (int) OPT_REPLICATE_DO_DB}, {"replicate-ignore-db", required_argument, 0, (int) OPT_REPLICATE_IGNORE_DB}, + {"replicate-rewrite-db", required_argument, 0, + (int) OPT_REPLICATE_REWRITE_DB}, {"safe-mode", no_argument, 0, (int) OPT_SAFE}, {"socket", required_argument, 0, (int) OPT_SOCKET}, + {"server-id", required_argument, 0, (int)OPT_SERVER_ID}, {"set-variable", required_argument, 0, 'O'}, #ifdef HAVE_BERKELEY_DB {"skip-bdb", no_argument, 0, (int) OPT_BDB_SKIP}, @@ -2280,6 +2298,7 @@ static struct option long_options[] = { {"skip-name-resolve", no_argument, 0, (int) OPT_SKIP_RESOLVE}, {"skip-new", no_argument, 0, (int) OPT_SKIP_NEW}, {"skip-show-database", no_argument, 0, (int) OPT_SKIP_SHOW_DB}, + {"skip-slave-start", no_argument, 0, (int) OPT_SKIP_SLAVE_START}, {"skip-networking", no_argument, 0, (int) OPT_SKIP_NETWORKING}, {"skip-thread-priority", no_argument, 0, (int) OPT_SKIP_PRIOR}, {"sql-bin-update-same", no_argument, 0, (int) OPT_SQL_BIN_UPDATE_SAME}, @@ -2428,6 +2447,7 @@ struct show_var_st init_vars[]= { {"port", (char*) &mysql_port, SHOW_INT}, {"protocol_version", (char*) &protocol_version, SHOW_INT}, {"record_buffer", (char*) &my_default_record_cache_size,SHOW_LONG}, + {"server_id", (char*) &server_id, SHOW_LONG}, {"skip_locking", (char*) &my_disable_locking, SHOW_MY_BOOL}, {"skip_networking", (char*) &opt_disable_networking, SHOW_BOOL}, {"skip_show_database", (char*) &opt_skip_show_db, SHOW_BOOL}, @@ -2851,6 +2871,9 @@ static void get_options(int argc,char **argv) opt_slow_log=1; opt_slow_logname=optarg; break; + case (int)OPT_SKIP_SLAVE_START: + opt_skip_slave_start = 1; + break; case (int) OPT_SKIP_NEW: opt_specialflag|= SPECIAL_NO_NEW_FUNC; default_table_type=DB_TYPE_ISAM; @@ -2972,6 +2995,9 @@ static void get_options(int argc,char **argv) default_table_type= (enum db_type) type; break; } + case OPT_SERVER_ID: + server_id = atoi(optarg); + break; case OPT_DELAY_KEY_WRITE: ha_open_options|=HA_OPEN_DELAY_KEY_WRITE; myisam_delay_key_write=1; diff --git a/sql/share/romanian/errmsg.sys b/sql/share/romanian/errmsg.sys index 20870bb1fa3..a40c10f7062 100644 Binary files a/sql/share/romanian/errmsg.sys and b/sql/share/romanian/errmsg.sys differ diff --git a/sql/slave.cc b/sql/slave.cc index f8746c78e49..f1055be2d38 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -463,15 +463,16 @@ static int safe_sleep(THD* thd, int sec) static int request_dump(MYSQL* mysql, MASTER_INFO* mi) { - char buf[FN_REFLEN + 6]; + char buf[FN_REFLEN + 10]; int len; int binlog_flags = 0; // for now char* logname = mi->log_file_name; int4store(buf, mi->pos); int2store(buf + 4, binlog_flags); + int4store(buf + 6, server_id); len = (uint) strlen(logname); - memcpy(buf + 6, logname,len); - if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 6, 1)) + memcpy(buf + 10, logname,len); + if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1)) // something went wrong, so we will just reconnect and retry later // in the future, we should do a better error analysis, but for // now we just fill up the error log :-) @@ -795,11 +796,17 @@ pthread_handler_decl(handle_slave,arg __attribute__((unused))) THD *thd;; // needs to be first for thread_stack MYSQL *mysql = NULL ; + if(!server_id) + { + sql_print_error("Server id not set, will not start slave"); + pthread_exit(1); + } + pthread_mutex_lock(&LOCK_slave); if(slave_running) { pthread_mutex_unlock(&LOCK_slave); - return 0; // safety just in case + pthread_exit(1); // safety just in case } slave_running = 1; abort_slave = 0; @@ -936,7 +943,7 @@ static void safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) while(!slave_killed(thd) && mc_mysql_reconnect(mysql)) { sql_print_error( - "Slave thread: error connecting to slave:%s, retry in %d sec", + "Slave thread: error connecting to master:%s, retry in %d sec", mc_mysql_error(mysql), mi->connect_retry); safe_sleep(thd, mi->connect_retry); } diff --git a/sql/sql_base.cc b/sql/sql_base.cc index eb3952be281..4a05f2370f6 100644 --- a/sql/sql_base.cc +++ b/sql/sql_base.cc @@ -42,71 +42,6 @@ static void mysql_rm_tmp_tables(void); static key_map get_key_map_from_key_list(THD *thd, TABLE *table, List *index_list); -static int send_file(THD *thd) -{ - NET* net = &thd->net; - int fd = -1,bytes, error = 1; - char fname[FN_REFLEN+1]; - char buf[IO_SIZE*15]; - const char *errmsg = 0; - int old_timeout; - DBUG_ENTER("send_file"); - - // the client might be slow loading the data, give him wait_timeout to do - // the job - old_timeout = thd->net.timeout; - thd->net.timeout = thd->inactive_timeout; - - // we need net_flush here because the client will not know it needs to send - // us the file name until it has processed the load event entry - if (net_flush(net) || my_net_read(net) == packet_error) - { - errmsg = "Failed reading file name"; - goto err; - } - - fn_format(fname, (char*)net->read_pos + 1, "", "", 4); - // this is needed to make replicate-ignore-db - if (!strcmp(fname,"/dev/null")) - goto end; - // TODO: work on the well-known system that does not have a /dev/null :-) - - if ((fd = my_open(fname, O_RDONLY, MYF(MY_WME))) < 0) - { - errmsg = "Failed on my_open()"; - goto err; - } - - while ((bytes = (int) my_read(fd, (byte*) buf, sizeof(buf), - MYF(MY_WME))) > 0) - { - if (my_net_write(net, buf, bytes)) - { - errmsg = "Failed on my_net_write()"; - goto err; - } - } - - end: - if (my_net_write(net, "", 0) || net_flush(net) || - (my_net_read(net) == packet_error)) - { - errmsg = "failed negotiating file transfer close"; - goto err; - } - error = 0; - - err: - thd->net.timeout = old_timeout; - if(fd >= 0) - (void) my_close(fd, MYF(MY_WME)); - if (errmsg) - { - sql_print_error("failed in send_file() : %s", errmsg); - DBUG_PRINT("error", (errmsg)); - } - DBUG_RETURN(error); -} static byte *cache_key(const byte *record,uint *length, my_bool not_used __attribute__((unused))) @@ -177,228 +112,6 @@ static void check_unused(void) #define check_unused() #endif -void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) -{ - LOG_INFO linfo; - char *log_file_name = linfo.log_file_name; - char search_file_name[FN_REFLEN]; - FILE* log = NULL; - String* packet = &thd->packet; - int error; - const char *errmsg = "Unknown error"; - NET* net = &thd->net; - - DBUG_ENTER("mysql_binlog_send"); - - if(!mysql_bin_log.is_open()) - { - errmsg = "Binary log is not open"; - goto err; - } - - if(log_ident[0]) - mysql_bin_log.make_log_name(search_file_name, log_ident); - else - search_file_name[0] = 0; - - if(mysql_bin_log.find_first_log(&linfo, search_file_name)) - { - errmsg = "Could not find first log"; - goto err; - } - log = my_fopen(log_file_name, O_RDONLY, MYF(MY_WME)); - - if(!log) - { - errmsg = "Could not open log file"; - goto err; - } - - if(my_fseek(log, pos, MY_SEEK_SET, MYF(MY_WME)) == MY_FILEPOS_ERROR ) - { - errmsg = "Error on fseek()"; - goto err; - } - - - packet->length(0); - packet->append("\0", 1); // we need to start a packet with something other than 255 - // to distiquish it from error - - while(!net->error && net->vio != 0 && !thd->killed) - { - while(!(error = Log_event::read_log_event(log, packet))) - { - if(my_net_write(net, (char*)packet->ptr(), packet->length()) ) - { - errmsg = "Failed on my_net_write()"; - goto err; - } - DBUG_PRINT("info", ("log event code %d",(*packet)[LOG_EVENT_OFFSET+1] )); - if((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) - { - if(send_file(thd)) - { - errmsg = "failed in send_file()"; - goto err; - } - } - packet->length(0); - packet->append("\0",1); - } - if(error != LOG_READ_EOF) - { - errmsg = "error reading log event"; - goto err; - } - - if(!(flags & BINLOG_DUMP_NON_BLOCK) && mysql_bin_log.is_active(log_file_name)) - // block until there is more data in the log - // unless non-blocking mode requested - { - if(net_flush(net)) - { - errmsg = "failed on net_flush()"; - goto err; - } - - // we may have missed the update broadcast from the log - // that has just happened, let's try to catch it if it did - // if we did not miss anything, we just wait for other threads - // to signal us - { - pthread_mutex_t *log_lock = mysql_bin_log.get_log_lock(); - clearerr(log); - - // tell the kill thread how to wake us up - pthread_mutex_lock(&thd->mysys_var->mutex); - thd->mysys_var->current_mutex = log_lock; - thd->mysys_var->current_cond = &COND_binlog_update; - const char* proc_info = thd->proc_info; - thd->proc_info = "Waiting for update"; - pthread_mutex_unlock(&thd->mysys_var->mutex); - - bool read_packet = 0, fatal_error = 0; - - pthread_mutex_lock(log_lock); // no one will update the log while we are reading - // now, but we'll be quick and just read one record - - - switch(Log_event::read_log_event(log, packet)) - { - case 0: - read_packet = 1; // we read successfully, so we'll need to send it to the - // slave - break; - case LOG_READ_EOF: - pthread_cond_wait(&COND_binlog_update, log_lock); - break; - - default: - fatal_error = 1; - break; - } - - pthread_mutex_unlock(log_lock); - - pthread_mutex_lock(&thd->mysys_var->mutex); - thd->mysys_var->current_mutex= 0; - thd->mysys_var->current_cond= 0; - thd->proc_info= proc_info; - pthread_mutex_unlock(&thd->mysys_var->mutex); - - if(read_packet) - { - if(my_net_write(net, (char*)packet->ptr(), packet->length()) ) - { - errmsg = "Failed on my_net_write()"; - goto err; - } - - if((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) - { - if(send_file(thd)) - { - errmsg = "failed in send_file()"; - goto err; - } - } - packet->length(0); - packet->append("\0",1); - // no need to net_flush because we will get to flush later when - // we hit EOF pretty quick - } - - if(fatal_error) - { - errmsg = "error reading log entry"; - goto err; - } - - clearerr(log); - } - } - else - { - bool loop_breaker = 0; // need this to break out of the for loop from switch - - switch(mysql_bin_log.find_next_log(&linfo)) - { - case LOG_INFO_EOF: - loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK); - break; - case 0: - break; - default: - errmsg = "could not find next log"; - goto err; - } - - if(loop_breaker) - break; - - (void) my_fclose(log, MYF(MY_WME)); - log = my_fopen(log_file_name, O_RDONLY, MYF(MY_WME)); - if(!log) - goto err; - // fake Rotate_log event just in case it did not make it to the log - // otherwise the slave make get confused about the offset - { - char header[9]; - memset(header, 0, 4); // when does not matter - header[4] = ROTATE_EVENT; - char* p = strrchr(log_file_name, FN_LIBCHAR); // find the last slash - if(p) - p++; - else - p = log_file_name; - - uint ident_len = (uint) strlen(p); - ulong event_len = ident_len + sizeof(header); - int4store(header + 5, event_len); - packet->append(header, sizeof(header)); - packet->append(p,ident_len); - if(my_net_write(net, (char*)packet->ptr(), packet->length())) - { - errmsg = "failed on my_net_write()"; - goto err; - } - packet->length(0); - packet->append("\0",1); - } - } - } - - (void)my_fclose(log, MYF(MY_WME)); - - send_eof(&thd->net); - DBUG_VOID_RETURN; - err: - if(log) - (void) my_fclose(log, MYF(MY_WME)); - send_error(&thd->net, 0, errmsg); - DBUG_VOID_RETURN; -} /****************************************************************************** diff --git a/sql/sql_class.cc b/sql/sql_class.cc index 2db08045d89..64fd281ee3a 100644 --- a/sql/sql_class.cc +++ b/sql/sql_class.cc @@ -83,6 +83,7 @@ THD::THD() user_time=bootstrap=in_lock_tables=global_read_lock=0; query_length=col_access=0; query_error=0; + server_id = ::server_id; server_status=SERVER_STATUS_AUTOCOMMIT; next_insert_id=last_insert_id=0; open_tables=temporary_tables=0; diff --git a/sql/sql_class.h b/sql/sql_class.h index 9e4293c7b62..212a641f335 100644 --- a/sql/sql_class.h +++ b/sql/sql_class.h @@ -243,6 +243,7 @@ public: ULL *ull; struct st_my_thread_var *mysys_var; enum enum_server_command command; + uint32 server_id; const char *where; char* last_nx_table; // last non-existent table, we need this for replication char* last_nx_db; // database of the last nx table diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index d8504cf5fd6..5871232c0c5 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -17,6 +17,7 @@ #include "mysql_priv.h" #include "sql_acl.h" +#include "sql_repl.h" #include #include #include @@ -24,13 +25,6 @@ #define SCRAMBLE_LENGTH 8 -extern pthread_handler_decl(handle_slave,arg); -extern bool slave_running; -extern char* master_host; -extern pthread_t slave_real_id; -extern MASTER_INFO glob_mi; -extern my_string opt_bin_logname, master_info_file; -extern I_List binlog_do_db, binlog_ignore_db; extern int yyparse(void); extern "C" pthread_mutex_t THR_LOCK_keycache; @@ -47,15 +41,8 @@ static void mysql_init_query(THD *thd); static void remove_escape(char *name); static void kill_one_thread(THD *thd, ulong thread); static void refresh_status(void); -static int start_slave(THD* thd = 0, bool net_report = 1); -static int stop_slave(THD* thd = 0, bool net_report = 1); -static int change_master(THD* thd); -static void reset_slave(); -static void reset_master(); -extern int init_master_info(MASTER_INFO* mi); - -static const char *any_db="*any*"; // Special symbol for check_access +const char *any_db="*any*"; // Special symbol for check_access const char *command_name[]={ "Sleep", "Quit", "Init DB", "Query", "Field List", "Create DB", @@ -717,9 +704,14 @@ bool do_command(THD *thd) ulong pos; ushort flags; + uint32 slave_server_id; pos = uint4korr(packet + 1); flags = uint2korr(packet + 5); - mysql_binlog_send(thd, thd->strdup(packet + 7), pos, flags); + pthread_mutex_lock(&LOCK_server_id); + kill_zombie_dump_threads(slave_server_id = uint4korr(packet+7)); + thd->server_id = slave_server_id; + pthread_mutex_unlock(&LOCK_server_id); + mysql_binlog_send(thd, strdup(packet + 11), pos, flags); break; } case COM_REFRESH: @@ -2516,207 +2508,3 @@ static void refresh_status(void) pthread_mutex_unlock(&THR_LOCK_keycache); } -static int start_slave(THD* thd , bool net_report) -{ - if(!thd) thd = current_thd; - NET* net = &thd->net; - const char* err = 0; - if (check_access(thd, PROCESS_ACL, any_db)) - return 1; - pthread_mutex_lock(&LOCK_slave); - if(!slave_running) - if(glob_mi.inited && glob_mi.host) - { - pthread_t hThread; - if(pthread_create(&hThread, &connection_attrib, handle_slave, 0)) - { - err = "cannot create slave thread"; - } - } - else - err = "Master host not set or master info not initialized"; - else - err = "Slave already running"; - - pthread_mutex_unlock(&LOCK_slave); - if(err) - { - if(net_report) send_error(net, 0, err); - return 1; - } - else if(net_report) - send_ok(net); - - return 0; -} - -static int stop_slave(THD* thd, bool net_report ) -{ - if(!thd) thd = current_thd; - NET* net = &thd->net; - const char* err = 0; - - if (check_access(thd, PROCESS_ACL, any_db)) - return 1; - - pthread_mutex_lock(&LOCK_slave); - if (slave_running) - { - abort_slave = 1; - thr_alarm_kill(slave_real_id); - // do not abort the slave in the middle of a query, so we do not set - // thd->killed for the slave thread - thd->proc_info = "waiting for slave to die"; - pthread_cond_wait(&COND_slave_stopped, &LOCK_slave); - } - else - err = "Slave is not running"; - - pthread_mutex_unlock(&LOCK_slave); - thd->proc_info = 0; - - if(err) - { - if(net_report) send_error(net, 0, err); - return 1; - } - else if(net_report) - send_ok(net); - - return 0; -} - -static void reset_slave() -{ - MY_STAT stat_area; - char fname[FN_REFLEN]; - bool slave_was_running = slave_running; - - if(slave_running) - stop_slave(0,0); - - fn_format(fname, master_info_file, mysql_data_home, "", 4+16+32); - if(my_stat(fname, &stat_area, MYF(0))) - if(my_delete(fname, MYF(MY_WME))) - return; - - if(slave_was_running) - start_slave(0,0); -} - -static int change_master(THD* thd) -{ - bool slave_was_running; - // kill slave thread - pthread_mutex_lock(&LOCK_slave); - if((slave_was_running = slave_running)) - { - abort_slave = 1; - thr_alarm_kill(slave_real_id); - thd->proc_info = "waiting for slave to die"; - pthread_cond_wait(&COND_slave_stopped, &LOCK_slave); // wait until done - } - pthread_mutex_unlock(&LOCK_slave); - thd->proc_info = "changing master"; - LEX_MASTER_INFO* lex_mi = &thd->lex.mi; - - if(!glob_mi.inited) - init_master_info(&glob_mi); - - pthread_mutex_lock(&glob_mi.lock); - if((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos) - { - // if we change host or port, we must reset the postion - glob_mi.log_file_name[0] = 0; - glob_mi.pos = 0; - } - - if(lex_mi->log_file_name) - strmake(glob_mi.log_file_name, lex_mi->log_file_name, - sizeof(glob_mi.log_file_name)); - if(lex_mi->pos) - glob_mi.pos = lex_mi->pos; - - if(lex_mi->host) - strmake(glob_mi.host, lex_mi->host, sizeof(glob_mi.host)); - if(lex_mi->user) - strmake(glob_mi.user, lex_mi->user, sizeof(glob_mi.user)); - if(lex_mi->password) - strmake(glob_mi.password, lex_mi->password, sizeof(glob_mi.password)); - if(lex_mi->port) - glob_mi.port = lex_mi->port; - if(lex_mi->connect_retry) - glob_mi.connect_retry = lex_mi->connect_retry; - - flush_master_info(&glob_mi); - pthread_mutex_unlock(&glob_mi.lock); - thd->proc_info = "starting slave"; - if(slave_was_running) - start_slave(0,0); - thd->proc_info = 0; - - send_ok(&thd->net); - return 0; -} - -static void reset_master() -{ - if(!mysql_bin_log.is_open()) - { - my_error(ER_FLUSH_MASTER_BINLOG_CLOSED, MYF(ME_BELL+ME_WAITTANG)); - return; - } - - LOG_INFO linfo; - if (mysql_bin_log.find_first_log(&linfo, "")) - return; - - for(;;) - { - my_delete(linfo.log_file_name, MYF(MY_WME)); - if (mysql_bin_log.find_next_log(&linfo)) - break; - } - mysql_bin_log.close(1); // exiting close - my_delete(mysql_bin_log.get_index_fname(), MYF(MY_WME)); - mysql_bin_log.open(opt_bin_logname,LOG_BIN); - -} - -int show_binlog_info(THD* thd) -{ - DBUG_ENTER("show_binlog_info"); - List field_list; - field_list.push_back(new Item_empty_string("File", FN_REFLEN)); - field_list.push_back(new Item_empty_string("Position",20)); - field_list.push_back(new Item_empty_string("Binlog_do_db",20)); - field_list.push_back(new Item_empty_string("Binlog_ignore_db",20)); - - if(send_fields(thd, field_list, 1)) - DBUG_RETURN(-1); - String* packet = &thd->packet; - packet->length(0); - - if(mysql_bin_log.is_open()) - { - LOG_INFO li; - mysql_bin_log.get_current_log(&li); - net_store_data(packet, li.log_file_name); - net_store_data(packet, (longlong)li.pos); - net_store_data(packet, &binlog_do_db); - net_store_data(packet, &binlog_ignore_db); - } - else - { - net_store_null(packet); - net_store_null(packet); - net_store_null(packet); - net_store_null(packet); - } - - if(my_net_write(&thd->net, (char*)thd->packet.ptr(), packet->length())) - DBUG_RETURN(-1); - - send_eof(&thd->net); - DBUG_RETURN(0); -}