diff options
Diffstat (limited to 'sql/slave.cc')
| -rw-r--r-- | sql/slave.cc | 3984 |
1 files changed, 1370 insertions, 2614 deletions
diff --git a/sql/slave.cc b/sql/slave.cc index e8bdf655115..837ae89294b 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -13,14 +13,26 @@ along with this program; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ -#include "mysql_priv.h" -#ifdef HAVE_REPLICATION +/** + @addtogroup Replication + @{ + + @file + + @brief Code to run the io thread and the sql thread on the + replication slave. +*/ + +#include "mysql_priv.h" #include <mysql.h> #include <myisam.h> #include "slave.h" +#include "rpl_mi.h" +#include "rpl_rli.h" #include "sql_repl.h" +#include "rpl_filter.h" #include "repl_failsafe.h" #include <thr_alarm.h> #include <my_dir.h> @@ -28,20 +40,20 @@ #include <errmsg.h> #include <mysys_err.h> +#ifdef HAVE_REPLICATION + +#include "rpl_tblmap.h" + +#define FLAGSTR(V,F) ((V)&(F)?#F" ":"") + #define MAX_SLAVE_RETRY_PAUSE 5 bool use_slave_mask = 0; MY_BITMAP slave_error_mask; typedef bool (*CHECK_KILLED_FUNC)(THD*,void*); -volatile bool slave_sql_running = 0, slave_io_running = 0; char* slave_load_tmpdir = 0; -MASTER_INFO *active_mi; -HASH replicate_do_table, replicate_ignore_table; -DYNAMIC_ARRAY replicate_wild_do_table, replicate_wild_ignore_table; -bool do_table_inited = 0, ignore_table_inited = 0; -bool wild_do_table_inited = 0, wild_ignore_table_inited = 0; -bool table_rules_on= 0; +Master_info *active_mi= 0; my_bool replicate_same_server_id; ulonglong relay_log_space_limit = 0; @@ -49,58 +61,116 @@ ulonglong relay_log_space_limit = 0; When slave thread exits, we need to remember the temporary tables so we can re-use them on slave start. - TODO: move the vars below under MASTER_INFO + TODO: move the vars below under Master_info */ int disconnect_slave_event_count = 0, abort_slave_event_count = 0; int events_till_abort = -1; -#ifndef DBUG_OFF -static int events_till_disconnect = -1; -#endif + +enum enum_slave_reconnect_actions +{ + SLAVE_RECON_ACT_REG= 0, + SLAVE_RECON_ACT_DUMP= 1, + SLAVE_RECON_ACT_EVENT= 2, + SLAVE_RECON_ACT_MAX +}; + +enum enum_slave_reconnect_messages +{ + SLAVE_RECON_MSG_WAIT= 0, + SLAVE_RECON_MSG_KILLED_WAITING= 1, + SLAVE_RECON_MSG_AFTER= 2, + SLAVE_RECON_MSG_FAILED= 3, + SLAVE_RECON_MSG_COMMAND= 4, + SLAVE_RECON_MSG_KILLED_AFTER= 5, + SLAVE_RECON_MSG_MAX +}; + +static const char *reconnect_messages[SLAVE_RECON_ACT_MAX][SLAVE_RECON_MSG_MAX]= +{ + { + "Waiting to reconnect after a failed registration on master", + "Slave I/O thread killed while waitnig to reconnect after a failed \ +registration on master", + "Reconnecting after a failed registration on master", + "failed registering on master, reconnecting to try again, \ +log '%s' at postion %s", + "COM_REGISTER_SLAVE", + "Slave I/O thread killed during or after reconnect" + }, + { + "Waiting to reconnect after a failed binlog dump request", + "Slave I/O thread killed while retrying master dump", + "Reconnecting after a failed binlog dump request", + "failed dump request, reconnecting to try again, log '%s' at postion %s", + "COM_BINLOG_DUMP", + "Slave I/O thread killed during or after reconnect" + }, + { + "Waiting to reconnect after a failed master event read", + "Slave I/O thread killed while waiting to reconnect after a failed read", + "Reconnecting after a failed master event read", + "Slave I/O thread: Failed reading log event, reconnecting to retry, \ +log '%s' at postion %s", + "", + "Slave I/O thread killed during or after a reconnect done to recover from \ +failed read" + } +}; + typedef enum { SLAVE_THD_IO, SLAVE_THD_SQL} SLAVE_THD_TYPE; -static int process_io_rotate(MASTER_INFO* mi, Rotate_log_event* rev); -static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev); -static bool wait_for_relay_log_space(RELAY_LOG_INFO* rli); -static inline bool io_slave_killed(THD* thd,MASTER_INFO* mi); -static inline bool sql_slave_killed(THD* thd,RELAY_LOG_INFO* rli); -static int count_relay_log_space(RELAY_LOG_INFO* rli); +static int process_io_rotate(Master_info* mi, Rotate_log_event* rev); +static int process_io_create_file(Master_info* mi, Create_file_log_event* cev); +static bool wait_for_relay_log_space(Relay_log_info* rli); +static inline bool io_slave_killed(THD* thd,Master_info* mi); +static inline bool sql_slave_killed(THD* thd,Relay_log_info* rli); static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type); -static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi); -static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi, - bool suppress_warnings); -static int connect_to_master(THD* thd, MYSQL* mysql, MASTER_INFO* mi, - bool reconnect, bool suppress_warnings); +static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi); +static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi, + bool suppress_warnings); +static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi, + bool reconnect, bool suppress_warnings); static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed, - void* thread_killed_arg); + void* thread_killed_arg); static int request_table_dump(MYSQL* mysql, const char* db, const char* table); static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, - const char* table_name, bool overwrite); -static int get_master_version_and_clock(MYSQL* mysql, MASTER_INFO* mi); + const char* table_name, bool overwrite); +static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi); +static Log_event* next_event(Relay_log_info* rli); +static int queue_event(Master_info* mi,const char* buf,ulong event_len); +static int terminate_slave_thread(THD *thd, + pthread_mutex_t* term_lock, + pthread_cond_t* term_cond, + volatile uint *slave_running, + bool skip_lock); +static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info); /* Find out which replications threads are running SYNOPSIS init_thread_mask() - mask Return value here - mi master_info for slave - inverse If set, returns which threads are not running + mask Return value here + mi master_info for slave + inverse If set, returns which threads are not running IMPLEMENTATION Get a bit mask for which threads are running so that we can later restart these threads. RETURN - mask If inverse == 0, running threads - If inverse == 1, stopped threads + mask If inverse == 0, running threads + If inverse == 1, stopped threads */ -void init_thread_mask(int* mask,MASTER_INFO* mi,bool inverse) +void init_thread_mask(int* mask,Master_info* mi,bool inverse) { bool set_io = mi->slave_running, set_sql = mi->rli.slave_running; register int tmp_mask=0; + DBUG_ENTER("init_thread_mask"); + if (set_io) tmp_mask |= SLAVE_IO; if (set_sql) @@ -108,6 +178,7 @@ void init_thread_mask(int* mask,MASTER_INFO* mi,bool inverse) if (inverse) tmp_mask^= (SLAVE_IO | SLAVE_SQL); *mask = tmp_mask; + DBUG_VOID_RETURN; } @@ -115,11 +186,14 @@ void init_thread_mask(int* mask,MASTER_INFO* mi,bool inverse) lock_slave_threads() */ -void lock_slave_threads(MASTER_INFO* mi) +void lock_slave_threads(Master_info* mi) { + DBUG_ENTER("lock_slave_threads"); + //TODO: see if we can do this without dual mutex pthread_mutex_lock(&mi->run_lock); pthread_mutex_lock(&mi->rli.run_lock); + DBUG_VOID_RETURN; } @@ -127,11 +201,14 @@ void lock_slave_threads(MASTER_INFO* mi) unlock_slave_threads() */ -void unlock_slave_threads(MASTER_INFO* mi) +void unlock_slave_threads(Master_info* mi) { + DBUG_ENTER("unlock_slave_threads"); + //TODO: see if we can do this without dual mutex pthread_mutex_unlock(&mi->rli.run_lock); pthread_mutex_unlock(&mi->run_lock); + DBUG_VOID_RETURN; } @@ -151,7 +228,7 @@ int init_slave() TODO: re-write this to interate through the list of files for multi-master */ - active_mi= new MASTER_INFO; + active_mi= new Master_info; /* If master_host is not specified, try to read it from the master_info file. @@ -165,7 +242,7 @@ int init_slave() } if (init_master_info(active_mi,master_info_file,relay_log_info_file, - !master_host, (SLAVE_IO | SLAVE_SQL))) + !master_host, (SLAVE_IO | SLAVE_SQL))) { sql_print_error("Failed to initialize the master info structure"); goto err; @@ -179,11 +256,11 @@ int init_slave() if (master_host && !opt_skip_slave_start) { if (start_slave_threads(1 /* need mutex */, - 0 /* no wait for start*/, - active_mi, - master_info_file, - relay_log_info_file, - SLAVE_IO | SLAVE_SQL)) + 0 /* no wait for start*/, + active_mi, + master_info_file, + relay_log_info_file, + SLAVE_IO | SLAVE_SQL)) { sql_print_error("Failed to create slave threads"); goto err; @@ -198,243 +275,12 @@ err: } -static void free_table_ent(TABLE_RULE_ENT* e) -{ - my_free((gptr) e, MYF(0)); -} - - -static byte* get_table_key(TABLE_RULE_ENT* e, uint* len, - my_bool not_used __attribute__((unused))) -{ - *len = e->key_len; - return (byte*)e->db; -} - - -/* - Open the given relay log - - SYNOPSIS - init_relay_log_pos() - rli Relay information (will be initialized) - log Name of relay log file to read from. NULL = First log - pos Position in relay log file - need_data_lock Set to 1 if this functions should do mutex locks - errmsg Store pointer to error message here - look_for_description_event - 1 if we should look for such an event. We only need - this when the SQL thread starts and opens an existing - relay log and has to execute it (possibly from an - offset >4); then we need to read the first event of - the relay log to be able to parse the events we have - to execute. - - DESCRIPTION - - Close old open relay log files. - - If we are using the same relay log as the running IO-thread, then set - rli->cur_log to point to the same IO_CACHE entry. - - If not, open the 'log' binary file. - - TODO - - check proper initialization of group_master_log_name/group_master_log_pos - - RETURN VALUES - 0 ok - 1 error. errmsg is set to point to the error message -*/ - -int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log, - ulonglong pos, bool need_data_lock, - const char** errmsg, - bool look_for_description_event) -{ - DBUG_ENTER("init_relay_log_pos"); - DBUG_PRINT("info", ("pos: %lu", (long) pos)); - - *errmsg=0; - pthread_mutex_t *log_lock=rli->relay_log.get_log_lock(); - - if (need_data_lock) - pthread_mutex_lock(&rli->data_lock); - - /* - Slave threads are not the only users of init_relay_log_pos(). CHANGE MASTER - is, too, and init_slave() too; these 2 functions allocate a description - event in init_relay_log_pos, which is not freed by the terminating SQL slave - thread as that thread is not started by these functions. So we have to free - the description_event here, in case, so that there is no memory leak in - running, say, CHANGE MASTER. - */ - delete rli->relay_log.description_event_for_exec; - /* - By default the relay log is in binlog format 3 (4.0). - Even if format is 4, this will work enough to read the first event - (Format_desc) (remember that format 4 is just lenghtened compared to format - 3; format 3 is a prefix of format 4). - */ - rli->relay_log.description_event_for_exec= new - Format_description_log_event(3); - - pthread_mutex_lock(log_lock); - - /* Close log file and free buffers if it's already open */ - if (rli->cur_log_fd >= 0) - { - end_io_cache(&rli->cache_buf); - my_close(rli->cur_log_fd, MYF(MY_WME)); - rli->cur_log_fd = -1; - } - - rli->group_relay_log_pos = rli->event_relay_log_pos = pos; - - /* - Test to see if the previous run was with the skip of purging - If yes, we do not purge when we restart - */ - if (rli->relay_log.find_log_pos(&rli->linfo, NullS, 1)) - { - *errmsg="Could not find first log during relay log initialization"; - goto err; - } - - if (log && rli->relay_log.find_log_pos(&rli->linfo, log, 1)) - { - *errmsg="Could not find target log during relay log initialization"; - goto err; - } - strmake(rli->group_relay_log_name,rli->linfo.log_file_name, - sizeof(rli->group_relay_log_name)-1); - strmake(rli->event_relay_log_name,rli->linfo.log_file_name, - sizeof(rli->event_relay_log_name)-1); - if (rli->relay_log.is_active(rli->linfo.log_file_name)) - { - /* - The IO thread is using this log file. - In this case, we will use the same IO_CACHE pointer to - read data as the IO thread is using to write data. - */ - my_b_seek((rli->cur_log=rli->relay_log.get_log_file()), (off_t)0); - if (check_binlog_magic(rli->cur_log,errmsg)) - goto err; - rli->cur_log_old_open_count=rli->relay_log.get_open_count(); - } - else - { - /* - Open the relay log and set rli->cur_log to point at this one - */ - if ((rli->cur_log_fd=open_binlog(&rli->cache_buf, - rli->linfo.log_file_name,errmsg)) < 0) - goto err; - rli->cur_log = &rli->cache_buf; - } - /* - In all cases, check_binlog_magic() has been called so we're at offset 4 for - sure. - */ - if (pos > BIN_LOG_HEADER_SIZE) /* If pos<=4, we stay at 4 */ - { - Log_event* ev; - while (look_for_description_event) - { - /* - Read the possible Format_description_log_event; if position - was 4, no need, it will be read naturally. - */ - DBUG_PRINT("info",("looking for a Format_description_log_event")); - - if (my_b_tell(rli->cur_log) >= pos) - break; - - /* - Because of we have rli->data_lock and log_lock, we can safely read an - event - */ - if (!(ev=Log_event::read_log_event(rli->cur_log,0, - rli->relay_log.description_event_for_exec))) - { - DBUG_PRINT("info",("could not read event, rli->cur_log->error=%d", - rli->cur_log->error)); - if (rli->cur_log->error) /* not EOF */ - { - *errmsg= "I/O error reading event at position 4"; - goto err; - } - break; - } - else if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT) - { - DBUG_PRINT("info",("found Format_description_log_event")); - delete rli->relay_log.description_event_for_exec; - rli->relay_log.description_event_for_exec= (Format_description_log_event*) ev; - /* - As ev was returned by read_log_event, it has passed is_valid(), so - my_malloc() in ctor worked, no need to check again. - */ - /* - Ok, we found a Format_description event. But it is not sure that this - describes the whole relay log; indeed, one can have this sequence - (starting from position 4): - Format_desc (of slave) - Rotate (of master) - Format_desc (of master) - So the Format_desc which really describes the rest of the relay log - is the 3rd event (it can't be further than that, because we rotate - the relay log when we queue a Rotate event from the master). - But what describes the Rotate is the first Format_desc. - So what we do is: - go on searching for Format_description events, until you exceed the - position (argument 'pos') or until you find another event than Rotate - or Format_desc. - */ - } - else - { - DBUG_PRINT("info",("found event of another type=%d", - ev->get_type_code())); - look_for_description_event= (ev->get_type_code() == ROTATE_EVENT); - delete ev; - } - } - my_b_seek(rli->cur_log,(off_t)pos); -#ifndef DBUG_OFF - { - char llbuf1[22], llbuf2[22]; - DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s", - llstr(my_b_tell(rli->cur_log),llbuf1), - llstr(rli->event_relay_log_pos,llbuf2))); - } -#endif - - } - -err: - /* - If we don't purge, we can't honour relay_log_space_limit ; - silently discard it - */ - if (!relay_log_purge) - rli->log_space_limit= 0; - pthread_cond_broadcast(&rli->data_cond); - - pthread_mutex_unlock(log_lock); - - if (need_data_lock) - pthread_mutex_unlock(&rli->data_lock); - if (!rli->relay_log.description_event_for_exec->is_valid() && !*errmsg) - *errmsg= "Invalid Format_description log event; could be out of memory"; - - DBUG_RETURN ((*errmsg) ? 1 : 0); -} - - /* Init function to set up array for errors that should be skipped for slave SYNOPSIS init_slave_skip_errors() - arg List of errors numbers to skip, separated with ',' + arg List of errors numbers to skip, separated with ',' NOTES Called from get_options() in mysqld.cc on start-up @@ -443,6 +289,8 @@ err: void init_slave_skip_errors(const char* arg) { const char *p; + DBUG_ENTER("init_slave_skip_errors"); + if (bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0)) { fprintf(stderr, "Badly out of memory, please check your system status\n"); @@ -454,7 +302,7 @@ void init_slave_skip_errors(const char* arg) if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const uchar*)"all",4)) { bitmap_set_all(&slave_error_mask); - return; + DBUG_VOID_RETURN; } for (p= arg ; *p; ) { @@ -466,237 +314,105 @@ void init_slave_skip_errors(const char* arg) while (!my_isdigit(system_charset_info,*p) && *p) p++; } + DBUG_VOID_RETURN; } -void st_relay_log_info::inc_group_relay_log_pos(ulonglong log_pos, - bool skip_lock) -{ - if (!skip_lock) - pthread_mutex_lock(&data_lock); - inc_event_relay_log_pos(); - group_relay_log_pos= event_relay_log_pos; - strmake(group_relay_log_name,event_relay_log_name, - sizeof(group_relay_log_name)-1); - - notify_group_relay_log_name_update(); - - /* - If the slave does not support transactions and replicates a transaction, - users should not trust group_master_log_pos (which they can display with - SHOW SLAVE STATUS or read from relay-log.info), because to compute - group_master_log_pos the slave relies on log_pos stored in the master's - binlog, but if we are in a master's transaction these positions are always - the BEGIN's one (excepted for the COMMIT), so group_master_log_pos does - not advance as it should on the non-transactional slave (it advances by - big leaps, whereas it should advance by small leaps). - */ - /* - In 4.x we used the event's len to compute the positions here. This is - wrong if the event was 3.23/4.0 and has been converted to 5.0, because - then the event's len is not what is was in the master's binlog, so this - will make a wrong group_master_log_pos (yes it's a bug in 3.23->4.0 - replication: Exec_master_log_pos is wrong). Only way to solve this is to - have the original offset of the end of the event the relay log. This is - what we do in 5.0: log_pos has become "end_log_pos" (because the real use - of log_pos in 4.0 was to compute the end_log_pos; so better to store - end_log_pos instead of begin_log_pos. - If we had not done this fix here, the problem would also have appeared - when the slave and master are 5.0 but with different event length (for - example the slave is more recent than the master and features the event - UID). It would give false MASTER_POS_WAIT, false Exec_master_log_pos in - SHOW SLAVE STATUS, and so the user would do some CHANGE MASTER using this - value which would lead to badly broken replication. - Even the relay_log_pos will be corrupted in this case, because the len is - the relay log is not "val". - With the end_log_pos solution, we avoid computations involving lengthes. - */ - DBUG_PRINT("info", ("log_pos: %lu group_master_log_pos: %lu", - (long) log_pos, (long) group_master_log_pos)); - if (log_pos) // 3.23 binlogs don't have log_posx - { - group_master_log_pos= log_pos; - } - pthread_cond_broadcast(&data_cond); - if (!skip_lock) - pthread_mutex_unlock(&data_lock); -} - - -void st_relay_log_info::close_temporary_tables() -{ - TABLE *table,*next; - - for (table=save_temporary_tables ; table ; table=next) - { - next=table->next; - /* - Don't ask for disk deletion. For now, anyway they will be deleted when - slave restarts, but it is a better intention to not delete them. - */ - close_temporary(table, 0); - } - save_temporary_tables= 0; - slave_open_temp_tables= 0; -} - -/* - purge_relay_logs() - - NOTES - Assumes to have a run lock on rli and that no slave thread are running. -*/ - -int purge_relay_logs(RELAY_LOG_INFO* rli, THD *thd, bool just_reset, - const char** errmsg) +int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock) { - int error=0; - DBUG_ENTER("purge_relay_logs"); - - /* - Even if rli->inited==0, we still try to empty rli->master_log_* variables. - Indeed, rli->inited==0 does not imply that they already are empty. - It could be that slave's info initialization partly succeeded : - for example if relay-log.info existed but *relay-bin*.* - have been manually removed, init_relay_log_info reads the old - relay-log.info and fills rli->master_log_*, then init_relay_log_info - checks for the existence of the relay log, this fails and - init_relay_log_info leaves rli->inited to 0. - In that pathological case, rli->master_log_pos* will be properly reinited - at the next START SLAVE (as RESET SLAVE or CHANGE - MASTER, the callers of purge_relay_logs, will delete bogus *.info files - or replace them with correct files), however if the user does SHOW SLAVE - STATUS before START SLAVE, he will see old, confusing rli->master_log_*. - In other words, we reinit rli->master_log_* for SHOW SLAVE STATUS - to display fine in any case. - */ - - rli->group_master_log_name[0]= 0; - rli->group_master_log_pos= 0; - - if (!rli->inited) - { - DBUG_PRINT("info", ("rli->inited == 0")); - DBUG_RETURN(0); - } - - DBUG_ASSERT(rli->slave_running == 0); - DBUG_ASSERT(rli->mi->slave_running == 0); - - rli->slave_skip_counter=0; - pthread_mutex_lock(&rli->data_lock); - - /* - we close the relay log fd possibly left open by the slave SQL thread, - to be able to delete it; the relay log fd possibly left open by the slave - I/O thread will be closed naturally in reset_logs() by the - close(LOG_CLOSE_TO_BE_OPENED) call - */ - if (rli->cur_log_fd >= 0) - { - end_io_cache(&rli->cache_buf); - my_close(rli->cur_log_fd, MYF(MY_WME)); - rli->cur_log_fd= -1; - } - - if (rli->relay_log.reset_logs(thd)) - { - *errmsg = "Failed during log reset"; - error=1; - goto err; - } - /* Save name of used relay log file */ - strmake(rli->group_relay_log_name, rli->relay_log.get_log_fname(), - sizeof(rli->group_relay_log_name)-1); - strmake(rli->event_relay_log_name, rli->relay_log.get_log_fname(), - sizeof(rli->event_relay_log_name)-1); - rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE; - if (count_relay_log_space(rli)) - { - *errmsg= "Error counting relay log space"; - goto err; - } - if (!just_reset) - error= init_relay_log_pos(rli, rli->group_relay_log_name, - rli->group_relay_log_pos, - 0 /* do not need data lock */, errmsg, 0); - -err: -#ifndef DBUG_OFF - char buf[22]; -#endif - DBUG_PRINT("info",("log_space_total: %s",llstr(rli->log_space_total,buf))); - pthread_mutex_unlock(&rli->data_lock); - DBUG_RETURN(error); -} - + DBUG_ENTER("terminate_slave_threads"); -int terminate_slave_threads(MASTER_INFO* mi,int thread_mask,bool skip_lock) -{ if (!mi->inited) - return 0; /* successfully do nothing */ + DBUG_RETURN(0); /* successfully do nothing */ int error,force_all = (thread_mask & SLAVE_FORCE_ALL); pthread_mutex_t *sql_lock = &mi->rli.run_lock, *io_lock = &mi->run_lock; - pthread_mutex_t *sql_cond_lock,*io_cond_lock; - DBUG_ENTER("terminate_slave_threads"); - sql_cond_lock=sql_lock; - io_cond_lock=io_lock; - - if (skip_lock) - { - sql_lock = io_lock = 0; - } - if ((thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL)) && mi->slave_running) + if ((thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL))) { DBUG_PRINT("info",("Terminating IO thread")); mi->abort_slave=1; if ((error=terminate_slave_thread(mi->io_thd,io_lock, - io_cond_lock, - &mi->stop_cond, - &mi->slave_running)) && - !force_all) + &mi->stop_cond, + &mi->slave_running, + skip_lock)) && + !force_all) DBUG_RETURN(error); } - if ((thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL)) && mi->rli.slave_running) + if ((thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL))) { DBUG_PRINT("info",("Terminating SQL thread")); - DBUG_ASSERT(mi->rli.sql_thd != 0) ; mi->rli.abort_slave=1; if ((error=terminate_slave_thread(mi->rli.sql_thd,sql_lock, - sql_cond_lock, - &mi->rli.stop_cond, - &mi->rli.slave_running)) && - !force_all) + &mi->rli.stop_cond, + &mi->rli.slave_running, + skip_lock)) && + !force_all) DBUG_RETURN(error); } DBUG_RETURN(0); } -int terminate_slave_thread(THD* thd, pthread_mutex_t* term_lock, - pthread_mutex_t *cond_lock, - pthread_cond_t* term_cond, - volatile uint *slave_running) +/** + Wait for a slave thread to terminate. + + This function is called after requesting the thread to terminate + (by setting @c abort_slave member of @c Relay_log_info or @c + Master_info structure to 1). Termination of the thread is + controlled with the the predicate <code>*slave_running</code>. + + Function will acquire @c term_lock before waiting on the condition + unless @c skip_lock is true in which case the mutex should be owned + by the caller of this function and will remain acquired after + return from the function. + + @param term_lock + Associated lock to use when waiting for @c term_cond + + @param term_cond + Condition that is signalled when the thread has terminated + + @param slave_running + Pointer to predicate to check for slave thread termination + + @param skip_lock + If @c true the lock will not be acquired before waiting on + the condition. In this case, it is assumed that the calling + function acquires the lock before calling this function. + + @retval 0 All OK + */ +static int +terminate_slave_thread(THD *thd, + pthread_mutex_t* term_lock, + pthread_cond_t* term_cond, + volatile uint *slave_running, + bool skip_lock) { + int error; + DBUG_ENTER("terminate_slave_thread"); - if (term_lock) - { + + if (!skip_lock) pthread_mutex_lock(term_lock); - if (!*slave_running) - { + + safe_mutex_assert_owner(term_lock); + + if (!*slave_running) + { + if (!skip_lock) pthread_mutex_unlock(term_lock); - DBUG_RETURN(ER_SLAVE_NOT_RUNNING); - } + DBUG_RETURN(ER_SLAVE_NOT_RUNNING); } DBUG_ASSERT(thd != 0); THD_CHECK_SENTRY(thd); + /* Is is critical to test if the slave is running. Otherwise, we might be referening freed memory trying to kick it */ - while (*slave_running) // Should always be true + while (*slave_running) // Should always be true { DBUG_PRINT("loop", ("killing slave thread")); @@ -719,27 +435,32 @@ int terminate_slave_thread(THD* thd, pthread_mutex_t* term_lock, */ struct timespec abstime; set_timespec(abstime,2); - pthread_cond_timedwait(term_cond, cond_lock, &abstime); + error= pthread_cond_timedwait(term_cond, term_lock, &abstime); + DBUG_ASSERT(error == ETIMEDOUT || error == 0); } - if (term_lock) + + DBUG_ASSERT(*slave_running == 0); + + if (!skip_lock) pthread_mutex_unlock(term_lock); DBUG_RETURN(0); } int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock, - pthread_mutex_t *cond_lock, - pthread_cond_t *start_cond, - volatile uint *slave_running, - volatile ulong *slave_run_id, - MASTER_INFO* mi, + pthread_mutex_t *cond_lock, + pthread_cond_t *start_cond, + volatile uint *slave_running, + volatile ulong *slave_run_id, + Master_info* mi, bool high_priority) { pthread_t th; ulong start_id; - DBUG_ASSERT(mi->inited); DBUG_ENTER("start_slave_thread"); + DBUG_ASSERT(mi->inited); + if (start_lock) pthread_mutex_lock(start_lock); if (!server_id) @@ -751,7 +472,7 @@ int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock, sql_print_error("Server id not set, will not start slave"); DBUG_RETURN(ER_BAD_SLAVE); } - + if (*slave_running) { if (start_cond) @@ -777,12 +498,12 @@ int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock, { DBUG_PRINT("sleep",("Waiting for slave thread to start")); const char* old_msg = thd->enter_cond(start_cond,cond_lock, - "Waiting for slave thread to start"); + "Waiting for slave thread to start"); pthread_cond_wait(start_cond,cond_lock); thd->exit_cond(old_msg); pthread_mutex_lock(cond_lock); // re-acquire it as exit_cond() released if (thd->killed) - DBUG_RETURN(thd->killed_errno()); + DBUG_RETURN(thd->killed_errno()); } } if (start_lock) @@ -801,14 +522,14 @@ int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock, */ 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) + Master_info* mi, const char* master_info_fname, + const char* slave_info_fname, int thread_mask) { pthread_mutex_t *lock_io=0,*lock_sql=0,*lock_cond_io=0,*lock_cond_sql=0; pthread_cond_t* cond_io=0,*cond_sql=0; int error=0; DBUG_ENTER("start_slave_threads"); - + if (need_slave_mutex) { lock_io = &mi->run_lock; @@ -824,15 +545,15 @@ int start_slave_threads(bool need_slave_mutex, bool wait_for_start, if (thread_mask & SLAVE_IO) error=start_slave_thread(handle_slave_io,lock_io,lock_cond_io, - cond_io, - &mi->slave_running, &mi->slave_run_id, - mi, 1); //high priority, to read the most possible + cond_io, + &mi->slave_running, &mi->slave_run_id, + mi, 1); //high priority, to read the most possible if (!error && (thread_mask & SLAVE_SQL)) { error=start_slave_thread(handle_slave_sql,lock_sql,lock_cond_sql, - cond_sql, - &mi->rli.slave_running, &mi->rli.slave_run_id, - mi, 0); + cond_sql, + &mi->rli.slave_running, &mi->rli.slave_run_id, + mi, 0); if (error) terminate_slave_threads(mi, thread_mask & SLAVE_IO, 0); } @@ -840,242 +561,13 @@ int start_slave_threads(bool need_slave_mutex, bool wait_for_start, } -void init_table_rule_hash(HASH* h, bool* h_inited) -{ - hash_init(h, system_charset_info,TABLE_RULE_HASH_SIZE,0,0, - (hash_get_key) get_table_key, - (hash_free_key) free_table_ent, 0); - *h_inited = 1; -} - - -void init_table_rule_array(DYNAMIC_ARRAY* a, bool* a_inited) -{ - my_init_dynamic_array(a, sizeof(TABLE_RULE_ENT*), TABLE_RULE_ARR_SIZE, - TABLE_RULE_ARR_SIZE); - *a_inited = 1; -} - - -static TABLE_RULE_ENT* find_wild(DYNAMIC_ARRAY *a, const char* key, int len) -{ - uint i; - const char* key_end = key + len; - - for (i = 0; i < a->elements; i++) - { - TABLE_RULE_ENT* e ; - get_dynamic(a, (gptr)&e, i); - if (!my_wildcmp(system_charset_info, key, key_end, - (const char*)e->db, - (const char*)(e->db + e->key_len), - '\\',wild_one,wild_many)) - return e; - } - - return 0; -} - - -/* - Checks whether tables match some (wild_)do_table and (wild_)ignore_table - rules (for replication) - - SYNOPSIS - tables_ok() - thd thread (SQL slave thread normally). Mustn't be null. - tables list of tables to check - - NOTES - Note that changing the order of the tables in the list can lead to - different results. Note also the order of precedence of the do/ignore - rules (see code below). For that reason, users should not set conflicting - rules because they may get unpredicted results (precedence order is - explained in the manual). - - Thought which arose from a question of a big customer "I want to include - all tables like "abc.%" except the "%.EFG"". This can't be done now. If we - supported Perl regexps we could do it with this pattern: /^abc\.(?!EFG)/ - (I could not find an equivalent in the regex library MySQL uses). - - RETURN VALUES - 0 should not be logged/replicated - 1 should be logged/replicated -*/ - -bool tables_ok(THD* thd, TABLE_LIST* tables) -{ - bool some_tables_updating= 0; - DBUG_ENTER("tables_ok"); - - /* - In routine, can't reliably pick and choose substatements, so always - replicate. - We can't reliably know if one substatement should be executed or not: - consider the case of this substatement: a SELECT on a non-replicated - constant table; if we don't execute it maybe it was going to fill a - variable which was going to be used by the next substatement to update - a replicated table? If we execute it maybe the constant non-replicated - table does not exist (and so we'll fail) while there was no need to - execute this as this SELECT does not influence replicated tables in the - rest of the routine? In other words: users are used to replicate-*-table - specifying how to handle updates to tables, these options don't say - anything about reads to tables; we can't guess. - */ - if (thd->spcont) - DBUG_RETURN(1); - - for (; tables; tables= tables->next_global) - { - char hash_key[2*NAME_LEN+2]; - char *end; - uint len; - - if (!tables->updating) - continue; - some_tables_updating= 1; - end= strmov(hash_key, tables->db ? tables->db : thd->db); - *end++= '.'; - len= (uint) (strmov(end, tables->table_name) - hash_key); - if (do_table_inited) // if there are any do's - { - if (hash_search(&replicate_do_table, (byte*) hash_key, len)) - DBUG_RETURN(1); - } - if (ignore_table_inited) // if there are any ignores - { - if (hash_search(&replicate_ignore_table, (byte*) hash_key, len)) - DBUG_RETURN(0); - } - if (wild_do_table_inited && find_wild(&replicate_wild_do_table, - hash_key, len)) - DBUG_RETURN(1); - if (wild_ignore_table_inited && find_wild(&replicate_wild_ignore_table, - hash_key, len)) - DBUG_RETURN(0); - } - - /* - If no table was to be updated, ignore statement (no reason we play it on - slave, slave is supposed to replicate _changes_ only). - If no explicit rule found and there was a do list, do not replicate. - If there was no do list, go ahead - */ - DBUG_RETURN(some_tables_updating && - !do_table_inited && !wild_do_table_inited); -} - - -/* - Checks whether a db matches wild_do_table and wild_ignore_table - rules (for replication) - - SYNOPSIS - db_ok_with_wild_table() - db name of the db to check. - Is tested with check_db_name() before calling this function. - - NOTES - Here is the reason for this function. - We advise users who want to exclude a database 'db1' safely to do it - with replicate_wild_ignore_table='db1.%' instead of binlog_ignore_db or - replicate_ignore_db because the two lasts only check for the selected db, - which won't work in that case: - USE db2; - UPDATE db1.t SET ... #this will be replicated and should not - whereas replicate_wild_ignore_table will work in all cases. - With replicate_wild_ignore_table, we only check tables. When - one does 'DROP DATABASE db1', tables are not involved and the - statement will be replicated, while users could expect it would not (as it - rougly means 'DROP db1.first_table, DROP db1.second_table...'). - In other words, we want to interpret 'db1.%' as "everything touching db1". - That is why we want to match 'db1' against 'db1.%' wild table rules. - - RETURN VALUES - 0 should not be logged/replicated - 1 should be logged/replicated - */ - -int db_ok_with_wild_table(const char *db) -{ - char hash_key[NAME_LEN+2]; - char *end; - int len; - end= strmov(hash_key, db); - *end++= '.'; - len= end - hash_key ; - if (wild_do_table_inited && find_wild(&replicate_wild_do_table, - hash_key, len)) - return 1; - if (wild_ignore_table_inited && find_wild(&replicate_wild_ignore_table, - hash_key, len)) - return 0; - - /* - If no explicit rule found and there was a do list, do not replicate. - If there was no do list, go ahead - */ - return !wild_do_table_inited; -} - - -int add_table_rule(HASH* h, const char* table_spec) -{ - const char* dot = strchr(table_spec, '.'); - if (!dot) return 1; - // len is always > 0 because we know the there exists a '.' - uint len = (uint)strlen(table_spec); - TABLE_RULE_ENT* e = (TABLE_RULE_ENT*)my_malloc(sizeof(TABLE_RULE_ENT) - + len, MYF(MY_WME)); - if (!e) return 1; - e->db = (char*)e + sizeof(TABLE_RULE_ENT); - e->tbl_name = e->db + (dot - table_spec) + 1; - e->key_len = len; - memcpy(e->db, table_spec, len); - (void)my_hash_insert(h, (byte*)e); - return 0; -} - - -/* - Add table expression with wildcards to dynamic array -*/ - -int add_wild_table_rule(DYNAMIC_ARRAY* a, const char* table_spec) -{ - const char* dot = strchr(table_spec, '.'); - if (!dot) return 1; - uint len = (uint)strlen(table_spec); - TABLE_RULE_ENT* e = (TABLE_RULE_ENT*)my_malloc(sizeof(TABLE_RULE_ENT) - + len, MYF(MY_WME)); - if (!e) return 1; - e->db = (char*)e + sizeof(TABLE_RULE_ENT); - e->tbl_name = e->db + (dot - table_spec) + 1; - e->key_len = len; - memcpy(e->db, table_spec, len); - insert_dynamic(a, (gptr)&e); - return 0; -} - - -static void free_string_array(DYNAMIC_ARRAY *a) -{ - uint i; - for (i = 0; i < a->elements; i++) - { - char* p; - get_dynamic(a, (gptr) &p, i); - my_free(p, MYF(MY_WME)); - } - delete_dynamic(a); -} - - #ifdef NOT_USED_YET -static int end_slave_on_walk(MASTER_INFO* mi, gptr /*unused*/) +static int end_slave_on_walk(Master_info* mi, uchar* /*unused*/) { + DBUG_ENTER("end_slave_on_walk"); + end_master_info(mi); - return 0; + DBUG_RETURN(0); } #endif @@ -1089,6 +581,8 @@ static int end_slave_on_walk(MASTER_INFO* mi, gptr /*unused*/) void end_slave() { + DBUG_ENTER("end_slave"); + /* This is called when the server terminates, in close_connections(). It terminates slave threads. However, some CHANGE MASTER etc may still be @@ -1106,72 +600,62 @@ void end_slave() */ terminate_slave_threads(active_mi,SLAVE_FORCE_ALL); end_master_info(active_mi); - if (do_table_inited) - hash_free(&replicate_do_table); - if (ignore_table_inited) - hash_free(&replicate_ignore_table); - if (wild_do_table_inited) - free_string_array(&replicate_wild_do_table); - if (wild_ignore_table_inited) - free_string_array(&replicate_wild_ignore_table); delete active_mi; active_mi= 0; } pthread_mutex_unlock(&LOCK_active_mi); + DBUG_VOID_RETURN; } -static bool io_slave_killed(THD* thd, MASTER_INFO* mi) +static bool io_slave_killed(THD* thd, Master_info* mi) { + DBUG_ENTER("io_slave_killed"); + DBUG_ASSERT(mi->io_thd == thd); DBUG_ASSERT(mi->slave_running); // tracking buffer overrun - return mi->abort_slave || abort_loop || thd->killed; + DBUG_RETURN(mi->abort_slave || abort_loop || thd->killed); } -static bool sql_slave_killed(THD* thd, RELAY_LOG_INFO* rli) +static bool sql_slave_killed(THD* thd, Relay_log_info* rli) { + DBUG_ENTER("sql_slave_killed"); + DBUG_ASSERT(rli->sql_thd == thd); DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun - return rli->abort_slave || abort_loop || thd->killed; + if (abort_loop || thd->killed || rli->abort_slave) + { + /* + If we are in an unsafe situation (stopping could corrupt replication), + we give one minute to the slave SQL thread of grace before really + terminating, in the hope that it will be able to read more events and + the unsafe situation will soon be left. Note that this one minute starts + from the last time anything happened in the slave SQL thread. So it's + really one minute of idleness, we don't timeout if the slave SQL thread + is actively working. + */ + if (rli->last_event_start_time == 0) + DBUG_RETURN(1); + DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving " + "it some grace period")); + if (difftime(time(0), rli->last_event_start_time) > 60) + { + rli->report(ERROR_LEVEL, 0, + "SQL thread had to stop in an unsafe situation, in " + "the middle of applying updates to a " + "non-transactional table without any primary key. " + "There is a risk of duplicate updates when the slave " + "SQL thread is restarted. Please check your tables' " + "contents after restart."); + DBUG_RETURN(1); + } + } + DBUG_RETURN(0); } /* - Writes an error message to rli->last_slave_error and rli->last_slave_errno - (which will be displayed by SHOW SLAVE STATUS), and prints it to stderr. - - SYNOPSIS - slave_print_error() - rli - err_code The error code - msg The error message (usually related to the error code, but can - contain more information). - ... (this is printf-like format, with % symbols in msg) - - RETURN VALUES - void -*/ - -void slave_print_error(RELAY_LOG_INFO* rli, int err_code, const char* msg, ...) -{ - va_list args; - va_start(args,msg); - my_vsnprintf(rli->last_slave_error, - sizeof(rli->last_slave_error), msg, args); - rli->last_slave_errno = err_code; - /* If the error string ends with '.', do not add a ',' it would be ugly */ - if (rli->last_slave_error[0] && - (*(strend(rli->last_slave_error)-1) == '.')) - sql_print_error("Slave: %s Error_code: %d", rli->last_slave_error, - err_code); - else - sql_print_error("Slave: %s, Error_code: %d", rli->last_slave_error, - err_code); - -} - -/* skip_load_data_infile() NOTES @@ -1180,19 +664,22 @@ void slave_print_error(RELAY_LOG_INFO* rli, int err_code, const char* msg, ...) void skip_load_data_infile(NET *net) { + DBUG_ENTER("skip_load_data_infile"); + (void)net_request_file(net, "/dev/null"); - (void)my_net_read(net); // discard response - (void)net_write_command(net, 0, "", 0, "", 0); // Send ok + (void)my_net_read(net); // discard response + (void)net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0); // ok + DBUG_VOID_RETURN; } bool net_request_file(NET* net, const char* fname) { DBUG_ENTER("net_request_file"); - DBUG_RETURN(net_write_command(net, 251, fname, strlen(fname), "", 0)); + DBUG_RETURN(net_write_command(net, 251, (uchar*) fname, strlen(fname), + (uchar*) "", 0)); } - /* From other comments and tests in code, it looks like sometimes Query_log_event and Load_log_event can have db == 0 @@ -1202,68 +689,17 @@ bool net_request_file(NET* net, const char* fname) const char *print_slave_db_safe(const char* db) { - return (db ? db : ""); -} - -/* - Checks whether a db matches some do_db and ignore_db rules - (for logging or replication) - - SYNOPSIS - db_ok() - db name of the db to check - do_list either binlog_do_db or replicate_do_db - ignore_list either binlog_ignore_db or replicate_ignore_db - - RETURN VALUES - 0 should not be logged/replicated - 1 should be logged/replicated -*/ - -int db_ok(const char* db, I_List<i_string> &do_list, - I_List<i_string> &ignore_list ) -{ - if (do_list.is_empty() && ignore_list.is_empty()) - return 1; // ok to replicate if the user puts no constraints - - /* - If the user has specified restrictions on which databases to replicate - and db was not selected, do not replicate. - */ - if (!db) - return 0; - - if (!do_list.is_empty()) // if the do's are not empty - { - I_List_iterator<i_string> it(do_list); - i_string* tmp; - - while ((tmp=it++)) - { - if (!strcmp(tmp->ptr, db)) - return 1; // match - } - return 0; - } - else // there are some elements in the don't, otherwise we cannot get here - { - I_List_iterator<i_string> it(ignore_list); - i_string* tmp; + DBUG_ENTER("*print_slave_db_safe"); - while ((tmp=it++)) - { - if (!strcmp(tmp->ptr, db)) - return 0; // match - } - return 1; - } + DBUG_RETURN((db ? db : "")); } - -static int init_strvar_from_file(char *var, int max_size, IO_CACHE *f, - const char *default_val) +int init_strvar_from_file(char *var, int max_size, IO_CACHE *f, + const char *default_val) { uint length; + DBUG_ENTER("init_strvar_from_file"); + if ((length=my_b_gets(f,var, max_size))) { char* last_p = var + length -1; @@ -1272,38 +708,40 @@ static int init_strvar_from_file(char *var, int max_size, IO_CACHE *f, else { /* - If we truncated a line or stopped on last char, remove all chars - up to and including newline. + If we truncated a line or stopped on last char, remove all chars + up to and including newline. */ int c; while (((c=my_b_get(f)) != '\n' && c != my_b_EOF)); } - return 0; + DBUG_RETURN(0); } else if (default_val) { strmake(var, default_val, max_size-1); - return 0; + DBUG_RETURN(0); } - return 1; + DBUG_RETURN(1); } -static int init_intvar_from_file(int* var, IO_CACHE* f, int default_val) +int init_intvar_from_file(int* var, IO_CACHE* f, int default_val) { char buf[32]; - - if (my_b_gets(f, buf, sizeof(buf))) + DBUG_ENTER("init_intvar_from_file"); + + + if (my_b_gets(f, buf, sizeof(buf))) { *var = atoi(buf); - return 0; + DBUG_RETURN(0); } else if (default_val) { *var = default_val; - return 0; + DBUG_RETURN(0); } - return 1; + DBUG_RETURN(1); } /* @@ -1314,15 +752,20 @@ static int init_intvar_from_file(int* var, IO_CACHE* f, int default_val) when people upgrade a 3.23 master to 4.0 without doing RESET MASTER: 4.0 slaves are fooled. So we do this only to distinguish between 3.23 and more recent masters (it's too late to change things for 3.23). - + RETURNS 0 ok 1 error */ -static int get_master_version_and_clock(MYSQL* mysql, MASTER_INFO* mi) +static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi) { + char err_buff[MAX_SLAVE_ERRMSG]; const char* errmsg= 0; + int err_code= 0; + MYSQL_RES *master_res= 0; + MYSQL_ROW master_row; + DBUG_ENTER("get_master_version_and_clock"); /* Free old description_event_for_queue (that is needed if we are in @@ -1330,30 +773,36 @@ static int get_master_version_and_clock(MYSQL* mysql, MASTER_INFO* mi) */ delete mi->rli.relay_log.description_event_for_queue; mi->rli.relay_log.description_event_for_queue= 0; - + if (!my_isdigit(&my_charset_bin,*mysql->server_version)) + { errmsg = "Master reported unrecognized MySQL version"; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, ER(err_code), errmsg); + } else { /* Note the following switch will bug when we have MySQL branch 30 ;) */ - switch (*mysql->server_version) + switch (*mysql->server_version) { case '0': case '1': case '2': errmsg = "Master reported unrecognized MySQL version"; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, ER(err_code), errmsg); break; case '3': mi->rli.relay_log.description_event_for_queue= new - Format_description_log_event(1, mysql->server_version); + Format_description_log_event(1, mysql->server_version); break; case '4': mi->rli.relay_log.description_event_for_queue= new - Format_description_log_event(3, mysql->server_version); + Format_description_log_event(3, mysql->server_version); break; - default: + default: /* Master is MySQL >=5.0. Give a default Format_desc event, so that we can take the early steps (like tests for "is this a 3.23 master") which we @@ -1363,54 +812,53 @@ static int get_master_version_and_clock(MYSQL* mysql, MASTER_INFO* mi) master is 3.23, 4.0, etc. */ mi->rli.relay_log.description_event_for_queue= new - Format_description_log_event(4, mysql->server_version); + Format_description_log_event(4, mysql->server_version); break; } } - - /* + + /* This does not mean that a 5.0 slave will be able to read a 6.0 master; but as we don't know yet, we don't want to forbid this for now. If a 5.0 slave can't read a 6.0 master, this will show up when the slave can't read some events sent by the master, and there will be error messages. */ - + if (errmsg) - { - sql_print_error(errmsg); - return 1; - } + goto err; /* as we are here, we tried to allocate the event */ if (!mi->rli.relay_log.description_event_for_queue) { - sql_print_error("Slave I/O thread failed to create a default Format_description_log_event"); - return 1; + errmsg= "default Format_description_log_event"; + err_code= ER_SLAVE_CREATE_EVENT_FAILURE; + sprintf(err_buff, ER(err_code), errmsg); + goto err; } /* Compare the master and slave's clock. Do not die if master's clock is unavailable (very old master not supporting UNIX_TIMESTAMP()?). */ - MYSQL_RES *master_res= 0; - MYSQL_ROW master_row; - + if (!mysql_real_query(mysql, STRING_WITH_LEN("SELECT UNIX_TIMESTAMP()")) && (master_res= mysql_store_result(mysql)) && (master_row= mysql_fetch_row(master_res))) { - mi->clock_diff_with_master= + mi->clock_diff_with_master= (long) (time((time_t*) 0) - strtoul(master_row[0], 0, 10)); } - else + else if (!check_io_slave_killed(mi->io_thd, mi, NULL)) { mi->clock_diff_with_master= 0; /* The "most sensible" value */ - sql_print_warning("\"SELECT UNIX_TIMESTAMP()\" failed on master, \ -do not trust column Seconds_Behind_Master of SHOW SLAVE STATUS"); + sql_print_warning("\"SELECT UNIX_TIMESTAMP()\" failed on master, " + "do not trust column Seconds_Behind_Master of SHOW " + "SLAVE STATUS. Error: %s (%d)", + mysql_error(mysql), mysql_errno(mysql)); } if (master_res) - mysql_free_result(master_res); - + mysql_free_result(master_res); + /* Check that the master's server id and ours are different. Because if they are equal (which can result from a simple copy of master's datadir to slave, @@ -1427,12 +875,18 @@ do not trust column Seconds_Behind_Master of SHOW SLAVE STATUS"); { if ((master_row= mysql_fetch_row(master_res)) && (::server_id == strtoul(master_row[1], 0, 10)) && - !replicate_same_server_id) + !mi->rli.replicate_same_server_id) + { errmsg= "The slave I/O thread stops because master and slave have equal \ MySQL server ids; these ids must be different for replication to work (or \ the --replicate-same-server-id option must be used on slave but this does \ not always make sense; please check the manual before using it)."; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, ER(err_code), errmsg); + } mysql_free_result(master_res); + if (errmsg) + goto err; } /* @@ -1463,10 +917,16 @@ not always make sense; please check the manual before using it)."; { if ((master_row= mysql_fetch_row(master_res)) && strcmp(master_row[0], global_system_variables.collation_server->name)) + { errmsg= "The slave I/O thread stops because master and slave have \ different values for the COLLATION_SERVER global variable. The values must \ be equal for replication to work"; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, ER(err_code), errmsg); + } mysql_free_result(master_res); + if (errmsg) + goto err; } /* @@ -1477,9 +937,9 @@ be equal for replication to work"; time and so could differ for slave and master even if they are really in the same system time zone. So we are omiting this check and just relying on documentation. Also according to Monty there are many users - who are using replication between servers in various time zones. Hence - such check will broke everything for them. (And now everything will - work for them because by default both their master and slave will have + who are using replication between servers in various time zones. Hence + such check will broke everything for them. (And now everything will + work for them because by default both their master and slave will have 'SYSTEM' time zone). This check is only necessary for 4.x masters (and < 5.0.4 masters but those were alpha). @@ -1489,22 +949,30 @@ be equal for replication to work"; (master_res= mysql_store_result(mysql))) { if ((master_row= mysql_fetch_row(master_res)) && - strcmp(master_row[0], + strcmp(master_row[0], global_system_variables.time_zone->get_name()->ptr())) + { errmsg= "The slave I/O thread stops because master and slave have \ different values for the TIME_ZONE global variable. The values must \ be equal for replication to work"; + err_code= ER_SLAVE_FATAL_ERROR; + sprintf(err_buff, ER(err_code), errmsg); + } mysql_free_result(master_res); + + if (errmsg) + goto err; } err: if (errmsg) { - sql_print_error(errmsg); - return 1; + DBUG_ASSERT(err_code != 0); + mi->report(ERROR_LEVEL, err_code, err_buff); + DBUG_RETURN(1); } - return 0; + DBUG_RETURN(0); } /* @@ -1519,7 +987,7 @@ err: */ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, - const char* table_name, bool overwrite) + const char* table_name, bool overwrite) { ulong packet_len; char *query, *save_db; @@ -1532,7 +1000,7 @@ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, ulonglong save_options; NET *net= &mysql->net; const char *found_semicolon= NULL; - DBUG_ENTER("create_table_from_dump"); + DBUG_ENTER("create_table_from_dump"); packet_len= my_net_read(net); // read create table statement if (packet_len == packet_error) @@ -1542,10 +1010,10 @@ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, } if (net->read_pos[0] == 255) // error from master { - char *err_msg; + char *err_msg; err_msg= (char*) net->read_pos + ((mysql->server_capabilities & - CLIENT_PROTOCOL_41) ? - 3+SQLSTATE_LENGTH+1 : 3); + CLIENT_PROTOCOL_41) ? + 3+SQLSTATE_LENGTH+1 : 3); my_error(ER_MASTER, MYF(0), err_msg); DBUG_RETURN(1); } @@ -1559,47 +1027,62 @@ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, DBUG_RETURN(1); } thd->query= query; - thd->query_error = 0; - thd->net.no_send_ok = 1; + thd->is_slave_error = 0; bzero((char*) &tables,sizeof(tables)); tables.db = (char*)db; tables.alias= tables.table_name= (char*)table_name; /* Drop the table if 'overwrite' is true */ - if (overwrite && mysql_rm_table(thd,&tables,1,0)) /* drop if exists */ + if (overwrite) { - sql_print_error("create_table_from_dump: failed to drop the table"); - goto err; + if (mysql_rm_table(thd,&tables,1,0)) /* drop if exists */ + { + sql_print_error("create_table_from_dump: failed to drop the table"); + goto err; + } + else + { + /* Clear the OK result of mysql_rm_table(). */ + thd->main_da.reset_diagnostics_area(); + } } /* Create the table. We do not want to log the "create table" statement */ save_options = thd->options; - thd->options &= ~(ulong) (OPTION_BIN_LOG); - thd->proc_info = "Creating table from master dump"; + thd->options &= ~ (OPTION_BIN_LOG); + thd_proc_info(thd, "Creating table from master dump"); // save old db in case we are creating in a different database save_db = thd->db; save_db_length= thd->db_length; - DBUG_ASSERT(db != 0); - thd->reset_db((char*)db, strlen(db)); + thd->db = (char*)db; + DBUG_ASSERT(thd->db != 0); + thd->db_length= strlen(thd->db); mysql_parse(thd, thd->query, packet_len, &found_semicolon); // run create table - thd->db = save_db; // leave things the way the were before + thd->db = save_db; // leave things the way the were before thd->db_length= save_db_length; thd->options = save_options; - - if (thd->query_error) - goto err; // mysql_parse took care of the error send - thd->proc_info = "Opening master dump table"; + if (thd->is_slave_error) + goto err; // mysql_parse took care of the error send + + thd_proc_info(thd, "Opening master dump table"); + thd->main_da.reset_diagnostics_area(); /* cleanup from CREATE_TABLE */ + /* + Note: If this function starts to fail for MERGE tables, + change the next two lines to these: + tables.table= NULL; // was set by mysql_rm_table() + if (!open_n_lock_single_table(thd, &tables, TL_WRITE)) + */ tables.lock_type = TL_WRITE; - if (!open_ltable(thd, &tables, TL_WRITE)) + if (!open_ltable(thd, &tables, TL_WRITE, 0)) { sql_print_error("create_table_from_dump: could not open created table"); goto err; } - + file = tables.table->file; - thd->proc_info = "Reading master dump table data"; + thd_proc_info(thd, "Reading master dump table data"); /* Copy the data file */ if (file->net_read_dump(net)) { @@ -1611,7 +1094,7 @@ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, check_opt.init(); check_opt.flags|= T_VERY_SILENT | T_CALC_CHECKSUM | T_QUICK; - thd->proc_info = "Rebuilding the index on master dump table"; + thd_proc_info(thd, "Rebuilding the index on master dump table"); /* We do not want repair() to spam us with messages just send them to the error log, and report the failure in case of @@ -1623,27 +1106,26 @@ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db, error=file->ha_repair(thd,&check_opt) != 0; thd->net.vio = save_vio; if (error) - my_error(ER_INDEX_REBUILD, MYF(0), tables.table->s->table_name); + my_error(ER_INDEX_REBUILD, MYF(0), tables.table->s->table_name.str); err: close_thread_tables(thd); - thd->net.no_send_ok = 0; - DBUG_RETURN(error); + DBUG_RETURN(error); } int fetch_master_table(THD *thd, const char *db_name, const char *table_name, - MASTER_INFO *mi, MYSQL *mysql, bool overwrite) + Master_info *mi, MYSQL *mysql, bool overwrite) { int error= 1; const char *errmsg=0; bool called_connected= (mysql != NULL); DBUG_ENTER("fetch_master_table"); DBUG_PRINT("enter", ("db_name: '%s' table_name: '%s'", - db_name,table_name)); + db_name,table_name)); if (!called_connected) - { + { if (!(mysql = mysql_init(NULL))) { DBUG_RETURN(1); @@ -1674,288 +1156,34 @@ int fetch_master_table(THD *thd, const char *db_name, const char *table_name, goto err; } if (create_table_from_dump(thd, mysql, db_name, - table_name, overwrite)) + table_name, overwrite)) goto err; // create_table_from_dump have sent the error already error = 0; err: - thd->net.no_send_ok = 0; // Clear up garbage after create_table_from_dump if (!called_connected) mysql_close(mysql); if (errmsg && thd->vio_ok()) my_message(error, errmsg, MYF(0)); - DBUG_RETURN(test(error)); // Return 1 on error -} - - -void end_master_info(MASTER_INFO* mi) -{ - DBUG_ENTER("end_master_info"); - - if (!mi->inited) - DBUG_VOID_RETURN; - end_relay_log_info(&mi->rli); - if (mi->fd >= 0) - { - end_io_cache(&mi->file); - (void)my_close(mi->fd, MYF(MY_WME)); - mi->fd = -1; - } - mi->inited = 0; - - DBUG_VOID_RETURN; -} - - -static int init_relay_log_info(RELAY_LOG_INFO* rli, - const char* info_fname) -{ - char fname[FN_REFLEN+128]; - int info_fd; - const char* msg = 0; - int error; - DBUG_ENTER("init_relay_log_info"); - - if (rli->inited) // Set if this function called - DBUG_RETURN(0); - fn_format(fname, info_fname, mysql_data_home, "", 4+32); - pthread_mutex_lock(&rli->data_lock); - info_fd = rli->info_fd; - rli->cur_log_fd = -1; - rli->slave_skip_counter=0; - rli->abort_pos_wait=0; - rli->log_space_limit= relay_log_space_limit; - rli->log_space_total= 0; - - /* - The relay log will now be opened, as a SEQ_READ_APPEND IO_CACHE. - Note that the I/O thread flushes it to disk after writing every - event, in flush_master_info(mi, 1). - */ - - /* - For the maximum log size, we choose max_relay_log_size if it is - non-zero, max_binlog_size otherwise. If later the user does SET - GLOBAL on one of these variables, fix_max_binlog_size and - fix_max_relay_log_size will reconsider the choice (for example - if the user changes max_relay_log_size to zero, we have to - switch to using max_binlog_size for the relay log) and update - rli->relay_log.max_size (and mysql_bin_log.max_size). - */ - { - char buf[FN_REFLEN]; - const char *ln; - static bool name_warning_sent= 0; - ln= rli->relay_log.generate_name(opt_relay_logname, "-relay-bin", - 1, buf); - /* We send the warning only at startup, not after every RESET SLAVE */ - if (!opt_relay_logname && !opt_relaylog_index_name && !name_warning_sent) - { - /* - User didn't give us info to name the relay log index file. - Picking `hostname`-relay-bin.index like we do, causes replication to - fail if this slave's hostname is changed later. So, we would like to - instead require a name. But as we don't want to break many existing - setups, we only give warning, not error. - */ - sql_print_warning("Neither --relay-log nor --relay-log-index were used;" - " so replication " - "may break when this MySQL server acts as a " - "slave and has his hostname changed!! Please " - "use '--relay-log=%s' to avoid this problem.", ln); - name_warning_sent= 1; - } - /* - note, that if open() fails, we'll still have index file open - but a destructor will take care of that - */ - if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln) || - rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND, 0, - (max_relay_log_size ? max_relay_log_size : - max_binlog_size), 1)) - { - pthread_mutex_unlock(&rli->data_lock); - sql_print_error("Failed in open_log() called from init_relay_log_info()"); - DBUG_RETURN(1); - } - } - - /* if file does not exist */ - if (access(fname,F_OK)) - { - /* - If someone removed the file from underneath our feet, just close - the old descriptor and re-create the old file - */ - if (info_fd >= 0) - my_close(info_fd, MYF(MY_WME)); - if ((info_fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0) - { - sql_print_error("Failed to create a new relay log info file (\ -file '%s', errno %d)", fname, my_errno); - msg= current_thd->net.last_error; - goto err; - } - if (init_io_cache(&rli->info_file, info_fd, IO_SIZE*2, READ_CACHE, 0L,0, - MYF(MY_WME))) - { - sql_print_error("Failed to create a cache on relay log info file '%s'", - fname); - msg= current_thd->net.last_error; - goto err; - } - - /* Init relay log with first entry in the relay index file */ - if (init_relay_log_pos(rli,NullS,BIN_LOG_HEADER_SIZE,0 /* no data lock */, - &msg, 0)) - { - sql_print_error("Failed to open the relay log 'FIRST' (relay_log_pos 4)"); - goto err; - } - rli->group_master_log_name[0]= 0; - rli->group_master_log_pos= 0; - rli->info_fd= info_fd; - } - else // file exists - { - error= 0; - if (info_fd >= 0) - reinit_io_cache(&rli->info_file, READ_CACHE, 0L,0,0); - else - { - if ((info_fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0) - { - sql_print_error("\ -Failed to open the existing relay log info file '%s' (errno %d)", - fname, my_errno); - error= 1; - } - else if (init_io_cache(&rli->info_file, info_fd, - IO_SIZE*2, READ_CACHE, 0L, 0, MYF(MY_WME))) - { - sql_print_error("Failed to create a cache on relay log info file '%s'", - fname); - error= 1; - } - if (error) - { - if (info_fd >= 0) - my_close(info_fd, MYF(0)); - rli->info_fd= -1; - rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT); - pthread_mutex_unlock(&rli->data_lock); - DBUG_RETURN(1); - } - } - - rli->info_fd = info_fd; - int relay_log_pos, master_log_pos; - if (init_strvar_from_file(rli->group_relay_log_name, - sizeof(rli->group_relay_log_name), - &rli->info_file, "") || - init_intvar_from_file(&relay_log_pos, - &rli->info_file, BIN_LOG_HEADER_SIZE) || - init_strvar_from_file(rli->group_master_log_name, - sizeof(rli->group_master_log_name), - &rli->info_file, "") || - init_intvar_from_file(&master_log_pos, &rli->info_file, 0)) - { - msg="Error reading slave log configuration"; - goto err; - } - strmake(rli->event_relay_log_name,rli->group_relay_log_name, - sizeof(rli->event_relay_log_name)-1); - rli->group_relay_log_pos= rli->event_relay_log_pos= relay_log_pos; - rli->group_master_log_pos= master_log_pos; - - if (init_relay_log_pos(rli, - rli->group_relay_log_name, - rli->group_relay_log_pos, - 0 /* no data lock*/, - &msg, 0)) - { - char llbuf[22]; - sql_print_error("Failed to open the relay log '%s' (relay_log_pos %s)", - rli->group_relay_log_name, - llstr(rli->group_relay_log_pos, llbuf)); - goto err; - } - } - -#ifndef DBUG_OFF - { - char llbuf1[22], llbuf2[22]; - DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s", - llstr(my_b_tell(rli->cur_log),llbuf1), - llstr(rli->event_relay_log_pos,llbuf2))); - DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE); - DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos); - } -#endif - - /* - Now change the cache from READ to WRITE - must do this - before flush_relay_log_info - */ - reinit_io_cache(&rli->info_file, WRITE_CACHE,0L,0,1); - if ((error= flush_relay_log_info(rli))) - sql_print_error("Failed to flush relay log info file"); - if (count_relay_log_space(rli)) - { - msg="Error counting relay log space"; - goto err; - } - rli->inited= 1; - pthread_mutex_unlock(&rli->data_lock); - DBUG_RETURN(error); - -err: - sql_print_error(msg); - end_io_cache(&rli->info_file); - if (info_fd >= 0) - my_close(info_fd, MYF(0)); - rli->info_fd= -1; - rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT); - pthread_mutex_unlock(&rli->data_lock); - DBUG_RETURN(1); + DBUG_RETURN(test(error)); // Return 1 on error } -static inline int add_relay_log(RELAY_LOG_INFO* rli,LOG_INFO* linfo) -{ - MY_STAT s; - DBUG_ENTER("add_relay_log"); - if (!my_stat(linfo->log_file_name,&s,MYF(0))) - { - sql_print_error("log %s listed in the index, but failed to stat", - linfo->log_file_name); - DBUG_RETURN(1); - } - rli->log_space_total += s.st_size; -#ifndef DBUG_OFF - char buf[22]; - DBUG_PRINT("info",("log_space_total: %s", llstr(rli->log_space_total,buf))); -#endif - DBUG_RETURN(0); -} - - -static bool wait_for_relay_log_space(RELAY_LOG_INFO* rli) +static bool wait_for_relay_log_space(Relay_log_info* rli) { bool slave_killed=0; - MASTER_INFO* mi = rli->mi; + Master_info* mi = rli->mi; const char *save_proc_info; THD* thd = mi->io_thd; - DBUG_ENTER("wait_for_relay_log_space"); pthread_mutex_lock(&rli->log_space_lock); save_proc_info= thd->enter_cond(&rli->log_space_cond, - &rli->log_space_lock, - "\ + &rli->log_space_lock, + "\ Waiting for the slave SQL thread to free enough relay log space"); while (rli->log_space_limit < rli->log_space_total && - !(slave_killed=io_slave_killed(thd,mi)) && + !(slave_killed=io_slave_killed(thd,mi)) && !rli->ignore_log_space_limit) pthread_cond_wait(&rli->log_space_cond, &rli->log_space_lock); thd->exit_cond(save_proc_info); @@ -1963,31 +1191,6 @@ Waiting for the slave SQL thread to free enough relay log space"); } -static int count_relay_log_space(RELAY_LOG_INFO* rli) -{ - LOG_INFO linfo; - DBUG_ENTER("count_relay_log_space"); - rli->log_space_total= 0; - if (rli->relay_log.find_log_pos(&linfo, NullS, 1)) - { - sql_print_error("Could not find first log while counting relay log space"); - DBUG_RETURN(1); - } - do - { - if (add_relay_log(rli,&linfo)) - DBUG_RETURN(1); - } while (!rli->relay_log.find_next_log(&linfo, 1)); - /* - As we have counted everything, including what may have written in a - preceding write, we must reset bytes_written, or we may count some space - twice. - */ - rli->relay_log.reset_bytes_written(); - DBUG_RETURN(0); -} - - /* Builds a Rotate from the ignored events' info and writes it to relay log. @@ -2001,16 +1204,18 @@ static int count_relay_log_space(RELAY_LOG_INFO* rli) ignored events' end position for the use of the slave SQL thread, by calling this function. Only that thread can call it (see assertion). */ -static void write_ignored_events_info_to_relay_log(THD *thd, MASTER_INFO *mi) +static void write_ignored_events_info_to_relay_log(THD *thd, Master_info *mi) { - RELAY_LOG_INFO *rli= &mi->rli; + Relay_log_info *rli= &mi->rli; pthread_mutex_t *log_lock= rli->relay_log.get_log_lock(); + DBUG_ENTER("write_ignored_events_info_to_relay_log"); + DBUG_ASSERT(thd == mi->io_thd); pthread_mutex_lock(log_lock); if (rli->ign_master_log_name_end[0]) { DBUG_PRINT("info",("writing a Rotate event to track down ignored events")); - Rotate_log_event *ev= new Rotate_log_event(thd, rli->ign_master_log_name_end, + Rotate_log_event *ev= new Rotate_log_event(rli->ign_master_log_name_end, 0, rli->ign_master_log_pos_end, Rotate_log_event::DUP_NAME); rli->ign_master_log_name_end[0]= 0; @@ -2020,300 +1225,38 @@ static void write_ignored_events_info_to_relay_log(THD *thd, MASTER_INFO *mi) { ev->server_id= 0; // don't be ignored by slave SQL thread if (unlikely(rli->relay_log.append(ev))) - sql_print_error("Slave I/O thread failed to write a Rotate event" - " to the relay log, " - "SHOW SLAVE STATUS may be inaccurate"); + mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE, + ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE), + "failed to write a Rotate event" + " to the relay log, SHOW SLAVE STATUS may be" + " inaccurate"); rli->relay_log.harvest_bytes_written(&rli->log_space_total); if (flush_master_info(mi, 1)) sql_print_error("Failed to flush master info file"); delete ev; } else - sql_print_error("Slave I/O thread failed to create a Rotate event" - " (out of memory?), " - "SHOW SLAVE STATUS may be inaccurate"); + mi->report(ERROR_LEVEL, ER_SLAVE_CREATE_EVENT_FAILURE, + ER(ER_SLAVE_CREATE_EVENT_FAILURE), + "Rotate_event (out of memory?)," + " SHOW SLAVE STATUS may be inaccurate"); } else pthread_mutex_unlock(log_lock); + DBUG_VOID_RETURN; } -void init_master_info_with_options(MASTER_INFO* mi) -{ - mi->master_log_name[0] = 0; - mi->master_log_pos = BIN_LOG_HEADER_SIZE; // skip magic number - - if (master_host) - strmake(mi->host, master_host, sizeof(mi->host) - 1); - if (master_user) - strmake(mi->user, master_user, sizeof(mi->user) - 1); - if (master_password) - strmake(mi->password, master_password, MAX_PASSWORD_LENGTH); - mi->port = master_port; - mi->connect_retry = master_connect_retry; - - mi->ssl= master_ssl; - if (master_ssl_ca) - strmake(mi->ssl_ca, master_ssl_ca, sizeof(mi->ssl_ca)-1); - if (master_ssl_capath) - strmake(mi->ssl_capath, master_ssl_capath, sizeof(mi->ssl_capath)-1); - if (master_ssl_cert) - strmake(mi->ssl_cert, master_ssl_cert, sizeof(mi->ssl_cert)-1); - if (master_ssl_cipher) - strmake(mi->ssl_cipher, master_ssl_cipher, sizeof(mi->ssl_cipher)-1); - if (master_ssl_key) - strmake(mi->ssl_key, master_ssl_key, sizeof(mi->ssl_key)-1); -} - -void clear_slave_error(RELAY_LOG_INFO* rli) -{ - /* Clear the errors displayed by SHOW SLAVE STATUS */ - rli->last_slave_error[0]= 0; - rli->last_slave_errno= 0; -} - -/* - Reset UNTIL condition for RELAY_LOG_INFO - SYNOPSYS - clear_until_condition() - rli - RELAY_LOG_INFO structure where UNTIL condition should be reset - */ -void clear_until_condition(RELAY_LOG_INFO* rli) -{ - rli->until_condition= RELAY_LOG_INFO::UNTIL_NONE; - rli->until_log_name[0]= 0; - rli->until_log_pos= 0; -} - - -#define LINES_IN_MASTER_INFO_WITH_SSL 14 - - -int init_master_info(MASTER_INFO* mi, const char* master_info_fname, - const char* slave_info_fname, - bool abort_if_no_master_info_file, - int thread_mask) -{ - int fd,error; - char fname[FN_REFLEN+128]; - DBUG_ENTER("init_master_info"); - - if (mi->inited) - { - /* - We have to reset read position of relay-log-bin as we may have - already been reading from 'hotlog' when the slave was stopped - last time. If this case pos_in_file would be set and we would - get a crash when trying to read the signature for the binary - relay log. - - We only rewind the read position if we are starting the SQL - thread. The handle_slave_sql thread assumes that the read - position is at the beginning of the file, and will read the - "signature" and then fast-forward to the last position read. - */ - if (thread_mask & SLAVE_SQL) - { - my_b_seek(mi->rli.cur_log, (my_off_t) 0); - } - DBUG_RETURN(0); - } - - mi->mysql=0; - mi->file_id=1; - fn_format(fname, master_info_fname, mysql_data_home, "", 4+32); - - /* - We need a mutex while we are changing master info parameters to - keep other threads from reading bogus info - */ - - pthread_mutex_lock(&mi->data_lock); - fd = mi->fd; - - /* does master.info exist ? */ - - if (access(fname,F_OK)) - { - if (abort_if_no_master_info_file) - { - pthread_mutex_unlock(&mi->data_lock); - DBUG_RETURN(0); - } - /* - if someone removed the file from underneath our feet, just close - the old descriptor and re-create the old file - */ - if (fd >= 0) - my_close(fd, MYF(MY_WME)); - if ((fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0 ) - { - sql_print_error("Failed to create a new master info file (\ -file '%s', errno %d)", fname, my_errno); - goto err; - } - if (init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,0, - MYF(MY_WME))) - { - sql_print_error("Failed to create a cache on master info file (\ -file '%s')", fname); - goto err; - } - - mi->fd = fd; - init_master_info_with_options(mi); - - } - else // file exists - { - if (fd >= 0) - reinit_io_cache(&mi->file, READ_CACHE, 0L,0,0); - else - { - if ((fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0 ) - { - sql_print_error("Failed to open the existing master info file (\ -file '%s', errno %d)", fname, my_errno); - goto err; - } - if (init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L, - 0, MYF(MY_WME))) - { - sql_print_error("Failed to create a cache on master info file (\ -file '%s')", fname); - goto err; - } - } - - mi->fd = fd; - int port, connect_retry, master_log_pos, ssl= 0, lines; - char *first_non_digit; - - /* - Starting from 4.1.x master.info has new format. Now its - first line contains number of lines in file. By reading this - number we will be always distinguish to which version our - master.info corresponds to. We can't simply count lines in - file since versions before 4.1.x could generate files with more - lines than needed. - If first line doesn't contain a number or contain number less than - 14 then such file is treated like file from pre 4.1.1 version. - There is no ambiguity when reading an old master.info, as before - 4.1.1, the first line contained the binlog's name, which is either - empty or has an extension (contains a '.'), so can't be confused - with an integer. - - So we're just reading first line and trying to figure which version - is this. - */ - - /* - The first row is temporarily stored in mi->master_log_name, - if it is line count and not binlog name (new format) it will be - overwritten by the second row later. - */ - if (init_strvar_from_file(mi->master_log_name, - sizeof(mi->master_log_name), &mi->file, - "")) - goto errwithmsg; - - lines= strtoul(mi->master_log_name, &first_non_digit, 10); - - if (mi->master_log_name[0]!='\0' && - *first_non_digit=='\0' && lines >= LINES_IN_MASTER_INFO_WITH_SSL) - { // Seems to be new format - if (init_strvar_from_file(mi->master_log_name, - sizeof(mi->master_log_name), &mi->file, "")) - goto errwithmsg; - } - else - lines= 7; - - if (init_intvar_from_file(&master_log_pos, &mi->file, 4) || - init_strvar_from_file(mi->host, sizeof(mi->host), &mi->file, - master_host) || - init_strvar_from_file(mi->user, sizeof(mi->user), &mi->file, - master_user) || - init_strvar_from_file(mi->password, SCRAMBLED_PASSWORD_CHAR_LENGTH+1, - &mi->file, master_password) || - init_intvar_from_file(&port, &mi->file, master_port) || - init_intvar_from_file(&connect_retry, &mi->file, - master_connect_retry)) - goto errwithmsg; - - /* - If file has ssl part use it even if we have server without - SSL support. But these option will be ignored later when - slave will try connect to master, so in this case warning - is printed. - */ - if (lines >= LINES_IN_MASTER_INFO_WITH_SSL && - (init_intvar_from_file(&ssl, &mi->file, master_ssl) || - init_strvar_from_file(mi->ssl_ca, sizeof(mi->ssl_ca), - &mi->file, master_ssl_ca) || - init_strvar_from_file(mi->ssl_capath, sizeof(mi->ssl_capath), - &mi->file, master_ssl_capath) || - init_strvar_from_file(mi->ssl_cert, sizeof(mi->ssl_cert), - &mi->file, master_ssl_cert) || - init_strvar_from_file(mi->ssl_cipher, sizeof(mi->ssl_cipher), - &mi->file, master_ssl_cipher) || - init_strvar_from_file(mi->ssl_key, sizeof(mi->ssl_key), - &mi->file, master_ssl_key))) - goto errwithmsg; -#ifndef HAVE_OPENSSL - if (ssl) - sql_print_warning("SSL information in the master info file " - "('%s') are ignored because this MySQL slave was compiled " - "without SSL support.", fname); -#endif /* HAVE_OPENSSL */ - - /* - This has to be handled here as init_intvar_from_file can't handle - my_off_t types - */ - mi->master_log_pos= (my_off_t) master_log_pos; - mi->port= (uint) port; - mi->connect_retry= (uint) connect_retry; - mi->ssl= (my_bool) ssl; - } - DBUG_PRINT("master_info",("log_file_name: %s position: %ld", - mi->master_log_name, - (ulong) mi->master_log_pos)); - - mi->rli.mi = mi; - if (init_relay_log_info(&mi->rli, slave_info_fname)) - goto err; - - mi->inited = 1; - // now change cache READ -> WRITE - must do this before flush_master_info - reinit_io_cache(&mi->file, WRITE_CACHE, 0L, 0, 1); - if ((error=test(flush_master_info(mi, 1)))) - sql_print_error("Failed to flush master info file"); - pthread_mutex_unlock(&mi->data_lock); - DBUG_RETURN(error); - -errwithmsg: - sql_print_error("Error reading master configuration"); - -err: - if (fd >= 0) - { - my_close(fd, MYF(0)); - end_io_cache(&mi->file); - } - mi->fd= -1; - pthread_mutex_unlock(&mi->data_lock); - DBUG_RETURN(1); -} - - -int register_slave_on_master(MYSQL* mysql) +int register_slave_on_master(MYSQL* mysql, Master_info *mi, + bool *suppress_warnings) { - char buf[1024], *pos= buf; + uchar buf[1024], *pos= buf; uint report_host_len, report_user_len=0, report_password_len=0; + DBUG_ENTER("register_slave_on_master"); + *suppress_warnings= FALSE; if (!report_host) - return 0; + DBUG_RETURN(0); report_host_len= strlen(report_host); if (report_user) report_user_len= strlen(report_user); @@ -2322,72 +1265,38 @@ int register_slave_on_master(MYSQL* mysql) /* 30 is a good safety margin */ if (report_host_len + report_user_len + report_password_len + 30 > sizeof(buf)) - return 0; // safety + DBUG_RETURN(0); // safety int4store(pos, server_id); pos+= 4; - pos= net_store_data(pos, report_host, report_host_len); - pos= net_store_data(pos, report_user, report_user_len); - pos= net_store_data(pos, report_password, report_password_len); + pos= net_store_data(pos, (uchar*) report_host, report_host_len); + pos= net_store_data(pos, (uchar*) report_user, report_user_len); + pos= net_store_data(pos, (uchar*) report_password, report_password_len); int2store(pos, (uint16) report_port); pos+= 2; - int4store(pos, rpl_recovery_rank); pos+= 4; + int4store(pos, rpl_recovery_rank); pos+= 4; /* The master will fill in master_id */ - int4store(pos, 0); pos+= 4; - - if (simple_command(mysql, COM_REGISTER_SLAVE, (char*) buf, - (uint) (pos- buf), 0)) - { - sql_print_error("Error on COM_REGISTER_SLAVE: %d '%s'", - mysql_errno(mysql), - mysql_error(mysql)); - return 1; - } - return 0; -} + int4store(pos, 0); pos+= 4; - -/* - Builds a String from a HASH of TABLE_RULE_ENT. Cannot be used for any other - hash, as it assumes that the hash entries are TABLE_RULE_ENT. - - SYNOPSIS - table_rule_ent_hash_to_str() - s pointer to the String to fill - h pointer to the HASH to read - - RETURN VALUES - none -*/ - -void table_rule_ent_hash_to_str(String* s, HASH* h) -{ - s->length(0); - for (uint i=0 ; i < h->records ; i++) + if (simple_command(mysql, COM_REGISTER_SLAVE, buf, (size_t) (pos- buf), 0)) { - TABLE_RULE_ENT* e= (TABLE_RULE_ENT*) hash_element(h, i); - if (s->length()) - s->append(','); - s->append(e->db,e->key_len); + if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED) + { + *suppress_warnings= TRUE; // Suppress reconnect warning + } + else if (!check_io_slave_killed(mi->io_thd, mi, NULL)) + { + char buf[256]; + my_snprintf(buf, sizeof(buf), "%s (Errno: %d)", mysql_error(mysql), + mysql_errno(mysql)); + mi->report(ERROR_LEVEL, ER_SLAVE_MASTER_COM_FAILURE, + ER(ER_SLAVE_MASTER_COM_FAILURE), "COM_REGISTER_SLAVE", buf); + } + DBUG_RETURN(1); } + DBUG_RETURN(0); } -/* - Mostly the same thing as above -*/ - -void table_rule_ent_dynamic_array_to_str(String* s, DYNAMIC_ARRAY* a) -{ - s->length(0); - for (uint i=0 ; i < a->elements ; i++) - { - TABLE_RULE_ENT* e; - get_dynamic(a, (gptr)&e, i); - if (s->length()) - s->append(','); - s->append(e->db,e->key_len); - } -} -bool show_master_info(THD* thd, MASTER_INFO* mi) +bool show_master_info(THD* thd, Master_info* mi) { // TODO: fix this for multi-master List<Item> field_list; @@ -2395,25 +1304,25 @@ bool show_master_info(THD* thd, MASTER_INFO* mi) DBUG_ENTER("show_master_info"); field_list.push_back(new Item_empty_string("Slave_IO_State", - 14)); + 14)); field_list.push_back(new Item_empty_string("Master_Host", - sizeof(mi->host))); + sizeof(mi->host))); field_list.push_back(new Item_empty_string("Master_User", - sizeof(mi->user))); + sizeof(mi->user))); field_list.push_back(new Item_return_int("Master_Port", 7, - MYSQL_TYPE_LONG)); + MYSQL_TYPE_LONG)); field_list.push_back(new Item_return_int("Connect_Retry", 10, - MYSQL_TYPE_LONG)); + MYSQL_TYPE_LONG)); field_list.push_back(new Item_empty_string("Master_Log_File", - FN_REFLEN)); + FN_REFLEN)); field_list.push_back(new Item_return_int("Read_Master_Log_Pos", 10, - MYSQL_TYPE_LONGLONG)); + MYSQL_TYPE_LONGLONG)); field_list.push_back(new Item_empty_string("Relay_Log_File", - FN_REFLEN)); + FN_REFLEN)); field_list.push_back(new Item_return_int("Relay_Log_Pos", 10, - MYSQL_TYPE_LONGLONG)); + MYSQL_TYPE_LONGLONG)); field_list.push_back(new Item_empty_string("Relay_Master_Log_File", - FN_REFLEN)); + FN_REFLEN)); field_list.push_back(new Item_empty_string("Slave_IO_Running", 3)); field_list.push_back(new Item_empty_string("Slave_SQL_Running", 3)); field_list.push_back(new Item_empty_string("Replicate_Do_DB", 20)); @@ -2422,33 +1331,39 @@ bool show_master_info(THD* thd, MASTER_INFO* mi) field_list.push_back(new Item_empty_string("Replicate_Ignore_Table", 23)); field_list.push_back(new Item_empty_string("Replicate_Wild_Do_Table", 24)); field_list.push_back(new Item_empty_string("Replicate_Wild_Ignore_Table", - 28)); + 28)); field_list.push_back(new Item_return_int("Last_Errno", 4, MYSQL_TYPE_LONG)); field_list.push_back(new Item_empty_string("Last_Error", 20)); field_list.push_back(new Item_return_int("Skip_Counter", 10, - MYSQL_TYPE_LONG)); + MYSQL_TYPE_LONG)); field_list.push_back(new Item_return_int("Exec_Master_Log_Pos", 10, - MYSQL_TYPE_LONGLONG)); + MYSQL_TYPE_LONGLONG)); field_list.push_back(new Item_return_int("Relay_Log_Space", 10, - MYSQL_TYPE_LONGLONG)); + MYSQL_TYPE_LONGLONG)); field_list.push_back(new Item_empty_string("Until_Condition", 6)); field_list.push_back(new Item_empty_string("Until_Log_File", FN_REFLEN)); - field_list.push_back(new Item_return_int("Until_Log_Pos", 10, + field_list.push_back(new Item_return_int("Until_Log_Pos", 10, MYSQL_TYPE_LONGLONG)); field_list.push_back(new Item_empty_string("Master_SSL_Allowed", 7)); field_list.push_back(new Item_empty_string("Master_SSL_CA_File", sizeof(mi->ssl_ca))); - field_list.push_back(new Item_empty_string("Master_SSL_CA_Path", + field_list.push_back(new Item_empty_string("Master_SSL_CA_Path", sizeof(mi->ssl_capath))); - field_list.push_back(new Item_empty_string("Master_SSL_Cert", + field_list.push_back(new Item_empty_string("Master_SSL_Cert", sizeof(mi->ssl_cert))); - field_list.push_back(new Item_empty_string("Master_SSL_Cipher", + field_list.push_back(new Item_empty_string("Master_SSL_Cipher", sizeof(mi->ssl_cipher))); - field_list.push_back(new Item_empty_string("Master_SSL_Key", + field_list.push_back(new Item_empty_string("Master_SSL_Key", sizeof(mi->ssl_key))); field_list.push_back(new Item_return_int("Seconds_Behind_Master", 10, MYSQL_TYPE_LONGLONG)); - + field_list.push_back(new Item_empty_string("Master_SSL_Verify_Server_Cert", + 3)); + field_list.push_back(new Item_return_int("Last_IO_Errno", 4, MYSQL_TYPE_LONG)); + field_list.push_back(new Item_empty_string("Last_IO_Error", 20)); + field_list.push_back(new Item_return_int("Last_SQL_Errno", 4, MYSQL_TYPE_LONG)); + field_list.push_back(new Item_empty_string("Last_SQL_Error", 20)); + if (protocol->send_fields(&field_list, Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF)) DBUG_RETURN(TRUE); @@ -2458,7 +1373,7 @@ bool show_master_info(THD* thd, MASTER_INFO* mi) DBUG_PRINT("info",("host is set: '%s'", mi->host)); String *packet= &thd->packet; protocol->prepare_for_resend(); - + /* slave_running can be accessed without run_lock but not other non-volotile members like mi->io_thd, which is guarded by the mutex. @@ -2476,46 +1391,41 @@ bool show_master_info(THD* thd, MASTER_INFO* mi) protocol->store(mi->master_log_name, &my_charset_bin); protocol->store((ulonglong) mi->master_log_pos); protocol->store(mi->rli.group_relay_log_name + - dirname_length(mi->rli.group_relay_log_name), - &my_charset_bin); + dirname_length(mi->rli.group_relay_log_name), + &my_charset_bin); protocol->store((ulonglong) mi->rli.group_relay_log_pos); protocol->store(mi->rli.group_master_log_name, &my_charset_bin); protocol->store(mi->slave_running == MYSQL_SLAVE_RUN_CONNECT ? "Yes" : "No", &my_charset_bin); protocol->store(mi->rli.slave_running ? "Yes":"No", &my_charset_bin); - protocol->store(&replicate_do_db); - protocol->store(&replicate_ignore_db); - /* - We can't directly use some protocol->store for - replicate_*_table, - as Protocol doesn't know the TABLE_RULE_ENT struct. - We first build Strings and then pass them to protocol->store. - */ + protocol->store(rpl_filter->get_do_db()); + protocol->store(rpl_filter->get_ignore_db()); + char buf[256]; String tmp(buf, sizeof(buf), &my_charset_bin); - table_rule_ent_hash_to_str(&tmp, &replicate_do_table); + rpl_filter->get_do_table(&tmp); protocol->store(&tmp); - table_rule_ent_hash_to_str(&tmp, &replicate_ignore_table); + rpl_filter->get_ignore_table(&tmp); protocol->store(&tmp); - table_rule_ent_dynamic_array_to_str(&tmp, &replicate_wild_do_table); + rpl_filter->get_wild_do_table(&tmp); protocol->store(&tmp); - table_rule_ent_dynamic_array_to_str(&tmp, &replicate_wild_ignore_table); + rpl_filter->get_wild_ignore_table(&tmp); protocol->store(&tmp); - protocol->store((uint32) mi->rli.last_slave_errno); - protocol->store(mi->rli.last_slave_error, &my_charset_bin); + protocol->store(mi->rli.last_error().number); + protocol->store(mi->rli.last_error().message, &my_charset_bin); protocol->store((uint32) mi->rli.slave_skip_counter); protocol->store((ulonglong) mi->rli.group_master_log_pos); protocol->store((ulonglong) mi->rli.log_space_total); protocol->store( - mi->rli.until_condition==RELAY_LOG_INFO::UNTIL_NONE ? "None": - ( mi->rli.until_condition==RELAY_LOG_INFO::UNTIL_MASTER_POS? "Master": + mi->rli.until_condition==Relay_log_info::UNTIL_NONE ? "None": + ( mi->rli.until_condition==Relay_log_info::UNTIL_MASTER_POS? "Master": "Relay"), &my_charset_bin); protocol->store(mi->rli.until_log_name, &my_charset_bin); protocol->store((ulonglong) mi->rli.until_log_pos); - -#ifdef HAVE_OPENSSL + +#ifdef HAVE_OPENSSL protocol->store(mi->ssl? "Yes":"No", &my_charset_bin); #else protocol->store(mi->ssl? "Ignored":"No", &my_charset_bin); @@ -2533,8 +1443,7 @@ bool show_master_info(THD* thd, MASTER_INFO* mi) if ((mi->slave_running == MYSQL_SLAVE_RUN_CONNECT) && mi->rli.slave_running) { - long time_diff= ((long)((time_t)time((time_t*) 0) - - mi->rli.last_master_timestamp) + long time_diff= ((long)(time(0) - mi->rli.last_master_timestamp) - mi->clock_diff_with_master); /* Apparently on some systems time_diff can be <0. Here are possible @@ -2560,318 +1469,34 @@ bool show_master_info(THD* thd, MASTER_INFO* mi) max(0, time_diff) : 0)); } else + { protocol->store_null(); + } + protocol->store(mi->ssl_verify_server_cert? "Yes":"No", &my_charset_bin); + + // Last_IO_Errno + protocol->store(mi->last_error().number); + // Last_IO_Error + protocol->store(mi->last_error().message, &my_charset_bin); + // Last_SQL_Errno + protocol->store(mi->rli.last_error().number); + // Last_SQL_Error + protocol->store(mi->rli.last_error().message, &my_charset_bin); pthread_mutex_unlock(&mi->rli.data_lock); pthread_mutex_unlock(&mi->data_lock); - if (my_net_write(&thd->net, (char*)thd->packet.ptr(), packet->length())) + if (my_net_write(&thd->net, (uchar*) thd->packet.ptr(), packet->length())) DBUG_RETURN(TRUE); } - send_eof(thd); + my_eof(thd); DBUG_RETURN(FALSE); } -/* - RETURN - 2 - flush relay log failed - 1 - flush master info failed - 0 - all ok -*/ -int flush_master_info(MASTER_INFO* mi, bool flush_relay_log_cache) -{ - IO_CACHE* file = &mi->file; - char lbuf[22]; - DBUG_ENTER("flush_master_info"); - DBUG_PRINT("enter",("master_pos: %ld", (long) mi->master_log_pos)); - - /* - Flush the relay log to disk. If we don't do it, then the relay log while - have some part (its last kilobytes) in memory only, so if the slave server - dies now, with, say, from master's position 100 to 150 in memory only (not - on disk), and with position 150 in master.info, then when the slave - restarts, the I/O thread will fetch binlogs from 150, so in the relay log - we will have "[0, 100] U [150, infinity[" and nobody will notice it, so the - SQL thread will jump from 100 to 150, and replication will silently break. - - When we come to this place in code, relay log may or not be initialized; - the caller is responsible for setting 'flush_relay_log_cache' accordingly. - */ - if (flush_relay_log_cache && - flush_io_cache(mi->rli.relay_log.get_log_file())) - DBUG_RETURN(2); - - /* - We flushed the relay log BEFORE the master.info file, because if we crash - now, we will get a duplicate event in the relay log at restart. If we - flushed in the other order, we would get a hole in the relay log. - And duplicate is better than hole (with a duplicate, in later versions we - can add detection and scrap one event; with a hole there's nothing we can - do). - */ - - /* - In certain cases this code may create master.info files that seems - corrupted, because of extra lines filled with garbage in the end - file (this happens if new contents take less space than previous - contents of file). But because of number of lines in the first line - of file we don't care about this garbage. - */ - - my_b_seek(file, 0L); - my_b_printf(file, "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n", - LINES_IN_MASTER_INFO_WITH_SSL, - mi->master_log_name, llstr(mi->master_log_pos, lbuf), - mi->host, mi->user, - mi->password, mi->port, mi->connect_retry, - (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert, - mi->ssl_cipher, mi->ssl_key); - DBUG_RETURN(-flush_io_cache(file)); -} - - -st_relay_log_info::st_relay_log_info() - :info_fd(-1), cur_log_fd(-1), save_temporary_tables(0), - cur_log_old_open_count(0), group_master_log_pos(0), log_space_total(0), - ignore_log_space_limit(0), last_master_timestamp(0), slave_skip_counter(0), - abort_pos_wait(0), slave_run_id(0), sql_thd(0), last_slave_errno(0), - inited(0), abort_slave(0), slave_running(0), until_condition(UNTIL_NONE), - until_log_pos(0), retried_trans(0) -{ - group_relay_log_name[0]= event_relay_log_name[0]= - group_master_log_name[0]= 0; - last_slave_error[0]= until_log_name[0]= ign_master_log_name_end[0]= 0; - - bzero((char*) &info_file, sizeof(info_file)); - bzero((char*) &cache_buf, sizeof(cache_buf)); - cached_charset_invalidate(); - pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST); - pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST); - pthread_mutex_init(&log_space_lock, MY_MUTEX_INIT_FAST); - pthread_cond_init(&data_cond, NULL); - pthread_cond_init(&start_cond, NULL); - pthread_cond_init(&stop_cond, NULL); - pthread_cond_init(&log_space_cond, NULL); - relay_log.init_pthread_objects(); -} - - -st_relay_log_info::~st_relay_log_info() -{ - pthread_mutex_destroy(&run_lock); - pthread_mutex_destroy(&data_lock); - pthread_mutex_destroy(&log_space_lock); - pthread_cond_destroy(&data_cond); - pthread_cond_destroy(&start_cond); - pthread_cond_destroy(&stop_cond); - pthread_cond_destroy(&log_space_cond); - relay_log.cleanup(); -} - -/* - Waits until the SQL thread reaches (has executed up to) the - log/position or timed out. - - SYNOPSIS - wait_for_pos() - thd client thread that sent SELECT MASTER_POS_WAIT - log_name log name to wait for - log_pos position to wait for - timeout timeout in seconds before giving up waiting - - NOTES - timeout is longlong whereas it should be ulong ; but this is - to catch if the user submitted a negative timeout. - - RETURN VALUES - -2 improper arguments (log_pos<0) - or slave not running, or master info changed - during the function's execution, - or client thread killed. -2 is translated to NULL by caller - -1 timed out - >=0 number of log events the function had to wait - before reaching the desired log/position - */ - -int st_relay_log_info::wait_for_pos(THD* thd, String* log_name, - longlong log_pos, - longlong timeout) -{ - if (!inited) - return -2; - int event_count = 0; - ulong init_abort_pos_wait; - int error=0; - struct timespec abstime; // for timeout checking - const char *msg; - DBUG_ENTER("wait_for_pos"); - DBUG_PRINT("enter",("log_name: '%s' log_pos: %lu timeout: %lu", - log_name->c_ptr(), (ulong) log_pos, (ulong) timeout)); - - set_timespec(abstime,timeout); - pthread_mutex_lock(&data_lock); - msg= thd->enter_cond(&data_cond, &data_lock, - "Waiting for the slave SQL thread to " - "advance position"); - /* - This function will abort when it notices that some CHANGE MASTER or - RESET MASTER has changed the master info. - To catch this, these commands modify abort_pos_wait ; We just monitor - abort_pos_wait and see if it has changed. - Why do we have this mechanism instead of simply monitoring slave_running - in the loop (we do this too), as CHANGE MASTER/RESET SLAVE require that - the SQL thread be stopped? - This is becasue if someones does: - STOP SLAVE;CHANGE MASTER/RESET SLAVE; START SLAVE; - the change may happen very quickly and we may not notice that - slave_running briefly switches between 1/0/1. - */ - init_abort_pos_wait= abort_pos_wait; - - /* - We'll need to - handle all possible log names comparisons (e.g. 999 vs 1000). - We use ulong for string->number conversion ; this is no - stronger limitation than in find_uniq_filename in sql/log.cc - */ - ulong log_name_extension; - char log_name_tmp[FN_REFLEN]; //make a char[] from String - - strmake(log_name_tmp, log_name->ptr(), min(log_name->length(), FN_REFLEN-1)); - - char *p= fn_ext(log_name_tmp); - char *p_end; - if (!*p || log_pos<0) - { - error= -2; //means improper arguments - goto err; - } - // Convert 0-3 to 4 - log_pos= max(log_pos, BIN_LOG_HEADER_SIZE); - /* p points to '.' */ - log_name_extension= strtoul(++p, &p_end, 10); - /* - p_end points to the first invalid character. - If it equals to p, no digits were found, error. - If it contains '\0' it means conversion went ok. - */ - if (p_end==p || *p_end) - { - error= -2; - goto err; - } - - /* The "compare and wait" main loop */ - while (!thd->killed && - init_abort_pos_wait == abort_pos_wait && - slave_running) - { - bool pos_reached; - int cmp_result= 0; - - DBUG_PRINT("info", - ("init_abort_pos_wait: %ld abort_pos_wait: %ld", - init_abort_pos_wait, abort_pos_wait)); - DBUG_PRINT("info",("group_master_log_name: '%s' pos: %lu", - group_master_log_name, (ulong) group_master_log_pos)); - - /* - group_master_log_name can be "", if we are just after a fresh - replication start or after a CHANGE MASTER TO MASTER_HOST/PORT - (before we have executed one Rotate event from the master) or - (rare) if the user is doing a weird slave setup (see next - paragraph). If group_master_log_name is "", we assume we don't - have enough info to do the comparison yet, so we just wait until - more data. In this case master_log_pos is always 0 except if - somebody (wrongly) sets this slave to be a slave of itself - without using --replicate-same-server-id (an unsupported - configuration which does nothing), then group_master_log_pos - will grow and group_master_log_name will stay "". - */ - if (*group_master_log_name) - { - char *basename= (group_master_log_name + - dirname_length(group_master_log_name)); - /* - First compare the parts before the extension. - Find the dot in the master's log basename, - and protect against user's input error : - if the names do not match up to '.' included, return error - */ - char *q= (char*)(fn_ext(basename)+1); - if (strncmp(basename, log_name_tmp, (int)(q-basename))) - { - error= -2; - break; - } - // Now compare extensions. - char *q_end; - ulong group_master_log_name_extension= strtoul(q, &q_end, 10); - if (group_master_log_name_extension < log_name_extension) - cmp_result= -1 ; - else - cmp_result= (group_master_log_name_extension > log_name_extension) ? 1 : 0 ; - - pos_reached= ((!cmp_result && group_master_log_pos >= (ulonglong)log_pos) || - cmp_result > 0); - if (pos_reached || thd->killed) - break; - } - - //wait for master update, with optional timeout. - - DBUG_PRINT("info",("Waiting for master update")); - /* - We are going to pthread_cond_(timed)wait(); if the SQL thread stops it - will wake us up. - */ - if (timeout > 0) - { - /* - Note that pthread_cond_timedwait checks for the timeout - before for the condition ; i.e. it returns ETIMEDOUT - if the system time equals or exceeds the time specified by abstime - before the condition variable is signaled or broadcast, _or_ if - the absolute time specified by abstime has already passed at the time - of the call. - For that reason, pthread_cond_timedwait will do the "timeoutting" job - even if its condition is always immediately signaled (case of a loaded - master). - */ - error=pthread_cond_timedwait(&data_cond, &data_lock, &abstime); - } - else - pthread_cond_wait(&data_cond, &data_lock); - DBUG_PRINT("info",("Got signal of master update or timed out")); - if (error == ETIMEDOUT || error == ETIME) - { - error= -1; - break; - } - error=0; - event_count++; - DBUG_PRINT("info",("Testing if killed or SQL thread not running")); - } - -err: - thd->exit_cond(msg); - DBUG_PRINT("exit",("killed: %d abort: %d slave_running: %d \ -improper_arguments: %d timed_out: %d", - thd->killed_errno(), - (int) (init_abort_pos_wait != abort_pos_wait), - (int) slave_running, - (int) (error == -2), - (int) (error == -1))); - if (thd->killed || init_abort_pos_wait != abort_pos_wait || - !slave_running) - { - error= -2; - } - DBUG_RETURN( error ? error : event_count ); -} void set_slave_thread_options(THD* thd) { + DBUG_ENTER("set_slave_thread_options"); /* It's nonsense to constrain the slave threads with max_join_size; if a query succeeded on master, we HAVE to execute it. So set @@ -2888,10 +1513,13 @@ void set_slave_thread_options(THD* thd) options&= ~OPTION_BIN_LOG; thd->options= options; thd->variables.completion_type= 0; + DBUG_VOID_RETURN; } -void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli) +void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli) { + DBUG_ENTER("set_slave_thread_default_charset"); + thd->variables.character_set_client= global_system_variables.character_set_client; thd->variables.collation_connection= @@ -2899,7 +1527,15 @@ void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli) thd->variables.collation_server= global_system_variables.collation_server; thd->update_charset(); - rli->cached_charset_invalidate(); + + /* + We use a const cast here since the conceptual (and externally + visible) behavior of the function is to set the default charset of + the thread. That the cache has to be invalidated is a secondary + effect. + */ + const_cast<Relay_log_info*>(rli)->cached_charset_invalidate(); + DBUG_VOID_RETURN; } /* @@ -2913,7 +1549,7 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type) int simulate_error= 0; #endif thd->system_thread = (thd_type == SLAVE_THD_SQL) ? - SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO; + SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO; thd->security_ctx->skip_grants(); my_net_init(&thd->net, 0); /* @@ -2924,11 +1560,11 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type) thd->variables.max_allowed_packet= global_system_variables.max_allowed_packet + MAX_LOG_EVENT_HEADER; /* note, incr over the global not session var */ thd->slave_thread = 1; + thd->enable_slow_log= opt_log_slow_slave_statements; set_slave_thread_options(thd); thd->client_capabilities = CLIENT_LOCAL_FILES; - thd->real_id=pthread_self(); pthread_mutex_lock(&LOCK_thread_count); - thd->thread_id = thread_id++; + thd->thread_id= thd->variables.pseudo_thread_id= thread_id++; pthread_mutex_unlock(&LOCK_thread_count); DBUG_EXECUTE_IF("simulate_io_slave_error_on_init", @@ -2944,17 +1580,12 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type) thd->cleanup(); DBUG_RETURN(-1); } - -#if !defined(__WIN__) && !defined(OS2) && !defined(__NETWARE__) - sigset_t set; - VOID(sigemptyset(&set)); // Get mask in use - VOID(pthread_sigmask(SIG_UNBLOCK,&set,&thd->block_signals)); -#endif + lex_start(thd); if (thd_type == SLAVE_THD_SQL) - thd->proc_info= "Waiting for the next event in relay log"; + thd_proc_info(thd, "Waiting for the next event in relay log"); else - thd->proc_info= "Waiting for master update"; + thd_proc_info(thd, "Waiting for master update"); thd->version=refresh_version; thd->set_time(); DBUG_RETURN(0); @@ -2962,12 +1593,14 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type) static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed, - void* thread_killed_arg) + void* thread_killed_arg) { int nap_time; thr_alarm_t alarmed; + DBUG_ENTER("safe_sleep"); + thr_alarm_init(&alarmed); - time_t start_time= time((time_t*) 0); + time_t start_time= my_time(0); time_t end_time= start_time+sec; while ((nap_time= (int) (end_time - start_time)) > 0) @@ -2981,23 +1614,25 @@ static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed, thr_alarm(&alarmed, 2 * nap_time, &alarm_buff); sleep(nap_time); thr_end_alarm(&alarmed); - + if ((*thread_killed)(thd,thread_killed_arg)) - return 1; - start_time=time((time_t*) 0); + DBUG_RETURN(1); + start_time= my_time(0); } - return 0; + DBUG_RETURN(0); } -static int request_dump(MYSQL* mysql, MASTER_INFO* mi, - bool *suppress_warnings) +static int request_dump(MYSQL* mysql, Master_info* mi, + bool *suppress_warnings) { - char buf[FN_REFLEN + 10]; + uchar buf[FN_REFLEN + 10]; int len; int binlog_flags = 0; // for now char* logname = mi->master_log_name; DBUG_ENTER("request_dump"); + + *suppress_warnings= FALSE; // TODO if big log files: Change next to int8store() int4store(buf, (ulong) mi->master_log_pos); @@ -3013,11 +1648,11 @@ static int request_dump(MYSQL* mysql, MASTER_INFO* mi, now we just fill up the error log :-) */ if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED) - *suppress_warnings= 1; // Suppress reconnect warning + *suppress_warnings= TRUE; // Suppress reconnect warning else sql_print_error("Error on COM_BINLOG_DUMP: %d %s, will retry in %d secs", - mysql_errno(mysql), mysql_error(mysql), - master_connect_retry); + mysql_errno(mysql), mysql_error(mysql), + master_connect_retry); DBUG_RETURN(1); } @@ -3027,82 +1662,82 @@ static int request_dump(MYSQL* mysql, MASTER_INFO* mi, static int request_table_dump(MYSQL* mysql, const char* db, const char* table) { - char buf[1024]; - char * p = buf; + uchar buf[1024], *p = buf; + DBUG_ENTER("request_table_dump"); + uint table_len = (uint) strlen(table); uint db_len = (uint) strlen(db); if (table_len + db_len > sizeof(buf) - 2) { sql_print_error("request_table_dump: Buffer overrun"); - return 1; - } - + DBUG_RETURN(1); + } + *p++ = db_len; memcpy(p, db, db_len); p += db_len; *p++ = table_len; memcpy(p, table, table_len); - + if (simple_command(mysql, COM_TABLE_DUMP, buf, p - buf + table_len, 1)) { sql_print_error("request_table_dump: Error sending the table dump \ command"); - return 1; + DBUG_RETURN(1); } - return 0; + DBUG_RETURN(0); } /* Read one event from the master - + SYNOPSIS read_event() - mysql MySQL connection - mi Master connection information - suppress_warnings TRUE when a normal net read timeout has caused us to - try a reconnect. We do not want to print anything to - the error log in this case because this a anormal - event in an idle server. + mysql MySQL connection + mi Master connection information + suppress_warnings TRUE when a normal net read timeout has caused us to + try a reconnect. We do not want to print anything to + the error log in this case because this a anormal + event in an idle server. RETURN VALUES - 'packet_error' Error - number Length of packet + 'packet_error' Error + number Length of packet */ -static ulong read_event(MYSQL* mysql, MASTER_INFO *mi, bool* suppress_warnings) +static ulong read_event(MYSQL* mysql, Master_info *mi, bool* suppress_warnings) { ulong len; + DBUG_ENTER("read_event"); - *suppress_warnings= 0; + *suppress_warnings= FALSE; /* my_real_read() will time us out We check if we were told to die, and if not, try reading again - - TODO: Move 'events_till_disconnect' to the MASTER_INFO structure */ #ifndef DBUG_OFF - if (disconnect_slave_event_count && !(events_till_disconnect--)) - return packet_error; + if (disconnect_slave_event_count && !(mi->events_till_disconnect--)) + DBUG_RETURN(packet_error); #endif - + len = cli_safe_read(mysql); if (len == packet_error || (long) len < 1) { if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED) { /* - We are trying a normal reconnect after a read timeout; - we suppress prints to .err file as long as the reconnect - happens without problems + We are trying a normal reconnect after a read timeout; + we suppress prints to .err file as long as the reconnect + happens without problems */ *suppress_warnings= TRUE; } else sql_print_error("Error reading packet from server: %s ( server_errno=%d)", - mysql_error(mysql), mysql_errno(mysql)); - return packet_error; + mysql_error(mysql), mysql_errno(mysql)); + DBUG_RETURN(packet_error); } /* Check if eof packet */ @@ -3110,146 +1745,274 @@ static ulong read_event(MYSQL* mysql, MASTER_INFO *mi, bool* suppress_warnings) { sql_print_information("Slave: received end packet from server, apparent " "master shutdown: %s", - mysql_error(mysql)); - return packet_error; + mysql_error(mysql)); + DBUG_RETURN(packet_error); } - - DBUG_PRINT("info",( "len: %lu net->read_pos[4]: %d\n", - len, mysql->net.read_pos[4])); - return len - 1; + + DBUG_PRINT("exit", ("len: %lu net->read_pos[4]: %d", + len, mysql->net.read_pos[4])); + DBUG_RETURN(len - 1); } -int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int expected_error) +int check_expected_error(THD* thd, Relay_log_info const *rli, + int expected_error) { + DBUG_ENTER("check_expected_error"); + switch (expected_error) { case ER_NET_READ_ERROR: - case ER_NET_ERROR_ON_WRITE: + case ER_NET_ERROR_ON_WRITE: case ER_QUERY_INTERRUPTED: - case ER_SERVER_SHUTDOWN: + case ER_SERVER_SHUTDOWN: case ER_NEW_ABORTING_CONNECTION: - return 1; + DBUG_RETURN(1); default: - return 0; + DBUG_RETURN(0); } } + /* - Check if condition stated in UNTIL clause of START SLAVE is reached. - SYNOPSYS - st_relay_log_info::is_until_satisfied() - master_beg_pos position of the beginning of to be executed event - (not log_pos member of the event that points to the - beginning of the following event) - - - DESCRIPTION - Checks if UNTIL condition is reached. Uses caching result of last - comparison of current log file name and target log file name. So cached - value should be invalidated if current log file name changes - (see st_relay_log_info::notify_... functions). - - This caching is needed to avoid of expensive string comparisons and - strtol() conversions needed for log names comparison. We don't need to - compare them each time this function is called, we only need to do this - when current log name changes. If we have UNTIL_MASTER_POS condition we - need to do this only after Rotate_log_event::exec_event() (which is - rare, so caching gives real benifit), and if we have UNTIL_RELAY_POS - condition then we should invalidate cached comarison value after - inc_group_relay_log_pos() which called for each group of events (so we - have some benefit if we have something like queries that use - autoincrement or if we have transactions). - - Should be called ONLY if until_condition != UNTIL_NONE ! - RETURN VALUE - true - condition met or error happened (condition seems to have - bad log file name) - false - condition not met + Check if the current error is of temporary nature of not. + Some errors are temporary in nature, such as + ER_LOCK_DEADLOCK and ER_LOCK_WAIT_TIMEOUT. Ndb also signals + that the error is temporary by pushing a warning with the error code + ER_GET_TEMPORARY_ERRMSG, if the originating error is temporary. */ +static int has_temporary_error(THD *thd) +{ + DBUG_ENTER("has_temporary_error"); + + if (thd->is_fatal_error) + DBUG_RETURN(0); -bool st_relay_log_info::is_until_satisfied(my_off_t master_beg_pos) + DBUG_EXECUTE_IF("all_errors_are_temporary_errors", + if (thd->main_da.is_error()) + { + thd->clear_error(); + my_error(ER_LOCK_DEADLOCK, MYF(0)); + }); + + /* + If there is no message in THD, we can't say if it's a temporary + error or not. This is currently the case for Incident_log_event, + which sets no message. Return FALSE. + */ + if (!thd->is_error()) + DBUG_RETURN(0); + + /* + Temporary error codes: + currently, InnoDB deadlock detected by InnoDB or lock + wait timeout (innodb_lock_wait_timeout exceeded + */ + if (thd->main_da.sql_errno() == ER_LOCK_DEADLOCK || + thd->main_da.sql_errno() == ER_LOCK_WAIT_TIMEOUT) + DBUG_RETURN(1); + +#ifdef HAVE_NDB_BINLOG + /* + currently temporary error set in ndbcluster + */ + List_iterator_fast<MYSQL_ERROR> it(thd->warn_list); + MYSQL_ERROR *err; + while ((err= it++)) + { + DBUG_PRINT("info", ("has warning %d %s", err->code, err->msg)); + switch (err->code) + { + case ER_GET_TEMPORARY_ERRMSG: + DBUG_RETURN(1); + default: + break; + } + } +#endif + DBUG_RETURN(0); +} + + +/** + Applies the given event and advances the relay log position. + + In essence, this function does: + + @code + ev->apply_event(rli); + ev->update_pos(rli); + @endcode + + But it also does some maintainance, such as skipping events if + needed and reporting errors. + + If the @c skip flag is set, then it is tested whether the event + should be skipped, by looking at the slave_skip_counter and the + server id. The skip flag should be set when calling this from a + replication thread but not set when executing an explicit BINLOG + statement. + + @retval 0 OK. + + @retval 1 Error calling ev->apply_event(). + + @retval 2 No error calling ev->apply_event(), but error calling + ev->update_pos(). +*/ +int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli, + bool skip) { - const char *log_name; - ulonglong log_pos; + int exec_res= 0; - DBUG_ASSERT(until_condition != UNTIL_NONE); - - if (until_condition == UNTIL_MASTER_POS) + DBUG_ENTER("apply_event_and_update_pos"); + + DBUG_PRINT("exec_event",("%s(type_code: %d; server_id: %d)", + ev->get_type_str(), ev->get_type_code(), + ev->server_id)); + DBUG_PRINT("info", ("thd->options: %s%s; rli->last_event_start_time: %lu", + FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT), + FLAGSTR(thd->options, OPTION_BEGIN), + rli->last_event_start_time)); + + /* + Execute the event to change the database and update the binary + log coordinates, but first we set some data that is needed for + the thread. + + The event will be executed unless it is supposed to be skipped. + + Queries originating from this server must be skipped. Low-level + events (Format_description_log_event, Rotate_log_event, + Stop_log_event) from this server must also be skipped. But for + those we don't want to modify 'group_master_log_pos', because + these events did not exist on the master. + Format_description_log_event is not completely skipped. + + Skip queries specified by the user in 'slave_skip_counter'. We + can't however skip events that has something to do with the log + files themselves. + + Filtering on own server id is extremely important, to ignore + execution of events created by the creation/rotation of the relay + log (remember that now the relay log starts with its Format_desc, + has a Rotate etc). + */ + + thd->server_id = ev->server_id; // use the original server id for logging + thd->set_time(); // time the query + thd->lex->current_select= 0; + if (!ev->when) + ev->when= my_time(0); + ev->thd = thd; // because up to this point, ev->thd == 0 + + if (skip) { - log_name= group_master_log_name; - log_pos= master_beg_pos; + int reason= ev->shall_skip(rli); + if (reason == Log_event::EVENT_SKIP_COUNT) + --rli->slave_skip_counter; + pthread_mutex_unlock(&rli->data_lock); + if (reason == Log_event::EVENT_SKIP_NOT) + exec_res= ev->apply_event(rli); +#ifndef DBUG_OFF + /* + This only prints information to the debug trace. + + TODO: Print an informational message to the error log? + */ + static const char *const explain[] = { + // EVENT_SKIP_NOT, + "not skipped", + // EVENT_SKIP_IGNORE, + "skipped because event should be ignored", + // EVENT_SKIP_COUNT + "skipped because event skip counter was non-zero" + }; + DBUG_PRINT("info", ("OPTION_BEGIN: %d; IN_STMT: %d", + thd->options & OPTION_BEGIN ? 1 : 0, + rli->get_flag(Relay_log_info::IN_STMT))); + DBUG_PRINT("skip_event", ("%s event was %s", + ev->get_type_str(), explain[reason])); +#endif } else - { /* until_condition == UNTIL_RELAY_POS */ - log_name= group_relay_log_name; - log_pos= group_relay_log_pos; - } - - if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN) + exec_res= ev->apply_event(rli); + + DBUG_PRINT("info", ("apply_event error = %d", exec_res)); + if (exec_res == 0) { + int error= ev->update_pos(rli); +#ifdef HAVE_purify + if (!rli->is_fake) +#endif + { +#ifndef DBUG_OFF + char buf[22]; +#endif + DBUG_PRINT("info", ("update_pos error = %d", error)); + DBUG_PRINT("info", ("group %s %s", + llstr(rli->group_relay_log_pos, buf), + rli->group_relay_log_name)); + DBUG_PRINT("info", ("event %s %s", + llstr(rli->event_relay_log_pos, buf), + rli->event_relay_log_name)); + } /* - We have no cached comparison results so we should compare log names - and cache result. - If we are after RESET SLAVE, and the SQL slave thread has not processed - any event yet, it could be that group_master_log_name is "". In that case, - just wait for more events (as there is no sensible comparison to do). - */ + The update should not fail, so print an error message and + return an error code. - if (*log_name) + TODO: Replace this with a decent error message when merged + with BUG#24954 (which adds several new error message). + */ + if (error) { - const char *basename= log_name + dirname_length(log_name); - - const char *q= (const char*)(fn_ext(basename)+1); - if (strncmp(basename, until_log_name, (int)(q-basename)) == 0) - { - /* Now compare extensions. */ - char *q_end; - ulong log_name_extension= strtoul(q, &q_end, 10); - if (log_name_extension < until_log_name_extension) - until_log_names_cmp_result= UNTIL_LOG_NAMES_CMP_LESS; - else - until_log_names_cmp_result= - (log_name_extension > until_log_name_extension) ? - UNTIL_LOG_NAMES_CMP_GREATER : UNTIL_LOG_NAMES_CMP_EQUAL ; - } - else - { - /* Probably error so we aborting */ - sql_print_error("Slave SQL thread is stopped because UNTIL " - "condition is bad."); - return TRUE; - } + char buf[22]; + rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR, + "It was not possible to update the positions" + " of the relay log information: the slave may" + " be in an inconsistent state." + " Stopped in %s position %s", + rli->group_relay_log_name, + llstr(rli->group_relay_log_pos, buf)); + DBUG_RETURN(2); } - else - return until_log_pos == 0; } - - return ((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL && - log_pos >= until_log_pos) || - until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER); + + DBUG_RETURN(exec_res ? 1 : 0); } -void st_relay_log_info::cached_charset_invalidate() -{ - /* Full of zeroes means uninitialized. */ - bzero(cached_charset, sizeof(cached_charset)); -} +/** + Top-level function for executing the next event from the relay log. + This function reads the event from the relay log, executes it, and + advances the relay log position. It also handles errors, etc. -bool st_relay_log_info::cached_charset_compare(char *charset) -{ - if (bcmp(cached_charset, charset, sizeof(cached_charset))) - { - memcpy(cached_charset, charset, sizeof(cached_charset)); - return 1; - } - return 0; -} + This function may fail to apply the event for the following reasons: + + - The position specfied by the UNTIL condition of the START SLAVE + command is reached. + + - It was not possible to read the event from the log. + + - The slave is killed. + + - An error occurred when applying the event, and the event has been + tried slave_trans_retries times. If the event has been retried + fewer times, 0 is returned. + - init_master_info or init_relay_log_pos failed. (These are called + if a failure occurs when applying the event.)</li> -static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) + - An error occurred when updating the binlog position. + + @retval 0 The event was applied. + + @retval 1 The event was not applied. +*/ +static int exec_relay_log_event(THD* thd, Relay_log_info* rli) { + DBUG_ENTER("exec_relay_log_event"); + /* We acquire this mutex since we need it for all operations except event execution. But we will release it in places where we will @@ -3265,19 +2028,18 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) { pthread_mutex_unlock(&rli->data_lock); delete ev; - return 1; + DBUG_RETURN(1); } if (ev) { - int type_code = ev->get_type_code(); int exec_res; /* This tests if the position of the beginning of the current event hits the UNTIL barrier. */ - if (rli->until_condition != RELAY_LOG_INFO::UNTIL_NONE && - rli->is_until_satisfied((thd->options & OPTION_BEGIN || !ev->log_pos) ? + if (rli->until_condition != Relay_log_info::UNTIL_NONE && + rli->is_until_satisfied((rli->is_in_group() || !ev->log_pos) ? rli->group_master_log_pos : ev->log_pos - ev->data_written)) { @@ -3291,157 +2053,40 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) rli->abort_slave= 1; pthread_mutex_unlock(&rli->data_lock); delete ev; - return 1; - } - /* - Queries originating from this server must be skipped. - Low-level events (Format_desc, Rotate, Stop) from this server - must also be skipped. But for those we don't want to modify - group_master_log_pos, because these events did not exist on the master. - Format_desc is not completely skipped. - Skip queries specified by the user in slave_skip_counter. - We can't however skip events that has something to do with the - log files themselves. - Filtering on own server id is extremely important, to ignore execution of - events created by the creation/rotation of the relay log (remember that - now the relay log starts with its Format_desc, has a Rotate etc). - */ - - DBUG_PRINT("info",("type_code: %d; server_id: %d; slave_skip_counter: %d", - type_code, ev->server_id, rli->slave_skip_counter)); - - /* - If the slave skip counter is positive, we still need to set the - OPTION_BEGIN flag correctly and not skip the log events that - start or end a transaction. If we do this, the slave will not - notice that it is inside a transaction, and happily start - executing from inside the transaction. - - Note that the code block below is strictly 5.0. - */ -#if MYSQL_VERSION_ID < 50100 - if (unlikely(rli->slave_skip_counter > 0)) - { - switch (type_code) - { - case QUERY_EVENT: - { - Query_log_event* const qev= (Query_log_event*) ev; - DBUG_PRINT("info", ("QUERY_EVENT { query: '%s', q_len: %u }", - qev->query, qev->q_len)); - if (memcmp("BEGIN", qev->query, qev->q_len+1) == 0) - thd->options|= OPTION_BEGIN; - else if (memcmp("COMMIT", qev->query, qev->q_len+1) == 0 || - memcmp("ROLLBACK", qev->query, qev->q_len+1) == 0) - thd->options&= ~OPTION_BEGIN; - } - break; - - case XID_EVENT: - DBUG_PRINT("info", ("XID_EVENT")); - thd->options&= ~OPTION_BEGIN; - break; - } - } -#endif - - if ((ev->server_id == (uint32) ::server_id && - !replicate_same_server_id && - type_code != FORMAT_DESCRIPTION_EVENT) || - (rli->slave_skip_counter && - type_code != ROTATE_EVENT && type_code != STOP_EVENT && - type_code != START_EVENT_V3 && type_code!= FORMAT_DESCRIPTION_EVENT)) - { - DBUG_PRINT("info", ("event skipped")); - if (thd->options & OPTION_BEGIN) - rli->inc_event_relay_log_pos(); - else - { - rli->inc_group_relay_log_pos((type_code == ROTATE_EVENT || - type_code == STOP_EVENT || - type_code == FORMAT_DESCRIPTION_EVENT) ? - LL(0) : ev->log_pos, - 1/* skip lock*/); - flush_relay_log_info(rli); - } - - DBUG_PRINT("info", ("thd->options: %s", - (thd->options & OPTION_BEGIN) ? "OPTION_BEGIN" : "")) - - /* - Protect against common user error of setting the counter to 1 - instead of 2 while recovering from an insert which used auto_increment, - rand or user var. - */ - if (rli->slave_skip_counter && - !((type_code == INTVAR_EVENT || - type_code == RAND_EVENT || - type_code == USER_VAR_EVENT || - type_code == BEGIN_LOAD_QUERY_EVENT || - type_code == APPEND_BLOCK_EVENT || - type_code == CREATE_FILE_EVENT) && - rli->slave_skip_counter == 1) && -#if MYSQL_VERSION_ID < 50100 - /* - Decrease the slave skip counter only if we are not inside - a transaction or the slave skip counter is more than - 1. The slave skip counter will be decreased from 1 to 0 - when reaching the final ROLLBACK, COMMIT, or XID_EVENT. - */ - (!(thd->options & OPTION_BEGIN) || rli->slave_skip_counter > 1) && -#endif - /* - The events from ourselves which have something to do with the relay - log itself must be skipped, true, but they mustn't decrement - rli->slave_skip_counter, because the user is supposed to not see - these events (they are not in the master's binlog) and if we - decremented, START SLAVE would for example decrement when it sees - the Rotate, so the event which the user probably wanted to skip - would not be skipped. - */ - !(ev->server_id == (uint32) ::server_id && - (type_code == ROTATE_EVENT || - type_code == STOP_EVENT || - type_code == START_EVENT_V3 || - type_code == FORMAT_DESCRIPTION_EVENT))) - --rli->slave_skip_counter; - pthread_mutex_unlock(&rli->data_lock); - delete ev; - return 0; // avoid infinite update loops + DBUG_RETURN(1); } - pthread_mutex_unlock(&rli->data_lock); + exec_res= apply_event_and_update_pos(ev, thd, rli, TRUE); - thd->server_id = ev->server_id; // use the original server id for logging - thd->set_time(); // time the query - thd->lex->current_select= 0; - if (!ev->when) - ev->when = time(NULL); - ev->thd = thd; - exec_res = ev->exec_event(rli); - DBUG_ASSERT(rli->sql_thd==thd); /* - Format_description_log_event should not be deleted because it will be - used to read info about the relay log's format; it will be deleted when - the SQL thread does not need it, i.e. when this thread terminates. + Format_description_log_event should not be deleted because it will be + used to read info about the relay log's format; it will be deleted when + the SQL thread does not need it, i.e. when this thread terminates. */ if (ev->get_type_code() != FORMAT_DESCRIPTION_EVENT) { DBUG_PRINT("info", ("Deleting the event after it has been executed")); delete ev; } + + /* + update_log_pos failed: this should not happen, so we don't + retry. + */ + if (exec_res == 2) + DBUG_RETURN(1); + if (slave_trans_retries) { - if (exec_res && - (thd->net.last_errno == ER_LOCK_DEADLOCK || - thd->net.last_errno == ER_LOCK_WAIT_TIMEOUT) && - !thd->is_fatal_error) + int temp_err; + if (exec_res && (temp_err= has_temporary_error(thd))) { const char *errmsg; /* We were in a transaction which has been rolled back because of a - Sonera deadlock. if lock wait timeout (innodb_lock_wait_timeout exceeded) + temporary error; + let's seek back to BEGIN log event and retry it all again. + Note, if lock wait timeout (innodb_lock_wait_timeout exceeded) there is no rollback since 5.0.13 (ref: manual). - let's seek back to BEGIN log event and retry it all again. We have to not only seek but also a) init_master_info(), to seek back to hot relay log's start for later (for when we will come back to this hot log after re-processing the @@ -3463,17 +2108,17 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) else { exec_res= 0; - end_trans(thd, ROLLBACK); - /* chance for concurrent connection to get more locks */ + end_trans(thd, ROLLBACK); + /* chance for concurrent connection to get more locks */ safe_sleep(thd, min(rli->trans_retries, MAX_SLAVE_RETRY_PAUSE), - (CHECK_KILLED_FUNC)sql_slave_killed, (void*)rli); + (CHECK_KILLED_FUNC)sql_slave_killed, (void*)rli); pthread_mutex_lock(&rli->data_lock); // because of SHOW STATUS - rli->trans_retries++; + rli->trans_retries++; rli->retried_trans++; pthread_mutex_unlock(&rli->data_lock); DBUG_PRINT("info", ("Slave retries transaction " "rli->trans_retries: %lu", rli->trans_retries)); - } + } } else sql_print_error("Slave SQL thread retried transaction %lu time(s) " @@ -3481,23 +2126,26 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli) "the slave_transaction_retries variable.", slave_trans_retries); } - else if (!((thd->options & OPTION_BEGIN) && opt_using_transactions)) + else if (exec_res && !temp_err || + (opt_using_transactions && + rli->group_relay_log_pos == rli->event_relay_log_pos)) { /* - Only reset the retry counter if the event succeeded or - failed with a non-transient error. On a successful event, - the execution will proceed as usual; in the case of a + Only reset the retry counter if the entire group succeeded + or failed with a non-transient error. On a successful + event, the execution will proceed as usual; in the case of a non-transient error, the slave will stop with an error. - */ + */ rli->trans_retries= 0; // restart from fresh + DBUG_PRINT("info", ("Resetting retry counter, rli->trans_retries: %lu", + rli->trans_retries)); } } - return exec_res; + DBUG_RETURN(exec_res); } - else - { - pthread_mutex_unlock(&rli->data_lock); - slave_print_error(rli, 0, "\ + pthread_mutex_unlock(&rli->data_lock); + rli->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_READ_FAILURE, + ER(ER_SLAVE_RELAY_LOG_READ_FAILURE), "\ Could not parse relay log event entry. The possible reasons are: the master's \ binary log is corrupted (you can check this by running 'mysqlbinlog' on the \ binary log), the slave's relay log is corrupted (you can check this by running \ @@ -3506,8 +2154,95 @@ or slave's MySQL code. If you want to check the master's binary log or slave's \ relay log, you will be able to know their names by issuing 'SHOW SLAVE STATUS' \ on this slave.\ "); + DBUG_RETURN(1); +} + + +static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info) +{ + if (io_slave_killed(thd, mi)) + { + if (info && global_system_variables.log_warnings) + sql_print_information(info); + return TRUE; + } + return FALSE; +} + + +/** + @brief Try to reconnect slave IO thread. + + @details Terminates current connection to master, sleeps for + @c mi->connect_retry msecs and initiates new connection with + @c safe_reconnect(). Variable pointed by @c retry_count is increased - + if it exceeds @c master_retry_count then connection is not re-established + and function signals error. + Unless @c suppres_warnings is TRUE, a warning is put in the server error log + when reconnecting. The warning message and messages used to report errors + are taken from @c messages array. In case @c master_retry_count is exceeded, + no messages are added to the log. + + @param[in] thd Thread context. + @param[in] mysql MySQL connection. + @param[in] mi Master connection information. + @param[in,out] retry_count Number of attempts to reconnect. + @param[in] suppress_warnings TRUE when a normal net read timeout + has caused to reconnecting. + @param[in] messages Messages to print/log, see + reconnect_messages[] array. + + @retval 0 OK. + @retval 1 There was an error. +*/ + +static int try_to_reconnect(THD *thd, MYSQL *mysql, Master_info *mi, + uint *retry_count, bool suppress_warnings, + const char *messages[SLAVE_RECON_MSG_MAX]) +{ + mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT; + thd->proc_info= messages[SLAVE_RECON_MSG_WAIT]; +#ifdef SIGNAL_WITH_VIO_CLOSE + thd->clear_active_vio(); +#endif + end_server(mysql); + if ((*retry_count)++) + { + if (*retry_count > master_retry_count) + return 1; // Don't retry forever + safe_sleep(thd, mi->connect_retry, (CHECK_KILLED_FUNC) io_slave_killed, + (void *) mi); + } + if (check_io_slave_killed(thd, mi, messages[SLAVE_RECON_MSG_KILLED_WAITING])) + return 1; + thd->proc_info = messages[SLAVE_RECON_MSG_AFTER]; + if (!suppress_warnings) + { + char buf[256], llbuff[22]; + my_snprintf(buf, sizeof(buf), messages[SLAVE_RECON_MSG_FAILED], + IO_RPL_LOG_NAME, llstr(mi->master_log_pos, llbuff)); + /* + Raise a warining during registering on master/requesting dump. + Log a message reading event. + */ + if (messages[SLAVE_RECON_MSG_COMMAND][0]) + { + mi->report(WARNING_LEVEL, ER_SLAVE_MASTER_COM_FAILURE, + ER(ER_SLAVE_MASTER_COM_FAILURE), + messages[SLAVE_RECON_MSG_COMMAND], buf); + } + else + { + sql_print_information(buf); + } + } + if (safe_reconnect(thd, mysql, mi, 1) || io_slave_killed(thd, mi)) + { + if (global_system_variables.log_warnings) + sql_print_information(messages[SLAVE_RECON_MSG_KILLED_AFTER]); return 1; } + return 0; } @@ -3517,18 +2252,18 @@ pthread_handler_t handle_slave_io(void *arg) { THD *thd; // needs to be first for thread_stack MYSQL *mysql; - MASTER_INFO *mi = (MASTER_INFO*)arg; - RELAY_LOG_INFO *rli= &mi->rli; + Master_info *mi = (Master_info*)arg; + Relay_log_info *rli= &mi->rli; char llbuff[22]; uint retry_count; - + bool suppress_warnings; +#ifndef DBUG_OFF + uint retry_count_reg= 0, retry_count_dump= 0, retry_count_event= 0; +#endif // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff my_thread_init(); DBUG_ENTER("handle_slave_io"); -#ifndef DBUG_OFF -slave_begin: -#endif DBUG_ASSERT(mi->inited); mysql= NULL ; retry_count= 0; @@ -3538,7 +2273,7 @@ slave_begin: mi->slave_run_id++; #ifndef DBUG_OFF - mi->events_till_abort = abort_slave_event_count; + mi->events_till_disconnect = disconnect_slave_event_count; #endif thd= new THD; // note that contructor of THD uses DBUG_ ! @@ -3563,22 +2298,23 @@ slave_begin: pthread_cond_broadcast(&mi->start_cond); DBUG_PRINT("master_info",("log_file_name: '%s' position: %s", - mi->master_log_name, - llstr(mi->master_log_pos,llbuff))); + mi->master_log_name, + llstr(mi->master_log_pos,llbuff))); if (!(mi->mysql = mysql = mysql_init(NULL))) { - sql_print_error("Slave I/O thread: error in mysql_init()"); + mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, + ER(ER_SLAVE_FATAL_ERROR), "error in mysql_init()"); goto err; } - thd->proc_info = "Connecting to master"; + thd_proc_info(thd, "Connecting to master"); // we can get killed during safe_connect if (!safe_connect(thd, mysql, mi)) { - sql_print_information("Slave I/O thread: connected to master '%s@%s:%d',\ - replication started in log '%s' at position %s", mi->user, - mi->host, mi->port, + sql_print_information("Slave I/O thread: connected to master '%s@%s:%d'," + "replication started in log '%s' at position %s", + mi->user, mi->host, mi->port, IO_RPL_LOG_NAME, llstr(mi->master_log_pos,llbuff)); /* @@ -3599,7 +2335,7 @@ connected: // TODO: the assignment below should be under mutex (5.0) mi->slave_running= MYSQL_SLAVE_RUN_CONNECT; thd->slave_net = &mysql->net; - thd->proc_info = "Checking master version"; + thd_proc_info(thd, "Checking master version"); if (get_master_version_and_clock(mysql, mi)) goto err; @@ -3607,89 +2343,83 @@ connected: { /* Register ourselves with the master. - If fails, this is not fatal - we just print the error message and go - on with life. */ - thd->proc_info = "Registering slave on master"; - if (register_slave_on_master(mysql)) - goto err; + thd_proc_info(thd, "Registering slave on master"); + if (register_slave_on_master(mysql, mi, &suppress_warnings)) + { + if (!check_io_slave_killed(thd, mi, "Slave I/O thread killed " + "while registering slave on master")) + { + sql_print_error("Slave I/O thread couldn't register on master"); + if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings, + reconnect_messages[SLAVE_RECON_ACT_REG])) + goto err; + } + else + goto err; + goto connected; + } + DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_REG", + if (!retry_count_reg) + { + retry_count_reg++; + sql_print_information("Forcing to reconnect slave I/O thread"); + if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings, + reconnect_messages[SLAVE_RECON_ACT_REG])) + goto err; + goto connected; + }); } DBUG_PRINT("info",("Starting reading binary log from master")); while (!io_slave_killed(thd,mi)) { - bool suppress_warnings= 0; - thd->proc_info = "Requesting binlog dump"; + thd_proc_info(thd, "Requesting binlog dump"); if (request_dump(mysql, mi, &suppress_warnings)) { sql_print_error("Failed on request_dump()"); - if (io_slave_killed(thd,mi)) - { - sql_print_information("Slave I/O thread killed while requesting master \ -dump"); - goto err; - } - - mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT; - thd->proc_info= "Waiting to reconnect after a failed binlog dump request"; -#ifdef SIGNAL_WITH_VIO_CLOSE - thd->clear_active_vio(); -#endif - end_server(mysql); - /* - First time retry immediately, assuming that we can recover - right away - if first time fails, sleep between re-tries - hopefuly the admin can fix the problem sometime - */ - if (retry_count++) - { - if (retry_count > master_retry_count) - goto err; // Don't retry forever - safe_sleep(thd,mi->connect_retry,(CHECK_KILLED_FUNC)io_slave_killed, - (void*)mi); - } - if (io_slave_killed(thd,mi)) - { - sql_print_information("Slave I/O thread killed while retrying master \ -dump"); - goto err; - } - - thd->proc_info = "Reconnecting after a failed binlog dump request"; - if (!suppress_warnings) - sql_print_error("Slave I/O thread: failed dump request, \ -reconnecting to try again, log '%s' at postion %s", IO_RPL_LOG_NAME, - llstr(mi->master_log_pos,llbuff)); - if (safe_reconnect(thd, mysql, mi, suppress_warnings) || - io_slave_killed(thd,mi)) - { - sql_print_information("Slave I/O thread killed during or \ -after reconnect"); - goto err; - } - + if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \ +requesting master dump") || + try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings, + reconnect_messages[SLAVE_RECON_ACT_DUMP])) + goto err; goto connected; } + DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_DUMP", + if (!retry_count_dump) + { + retry_count_dump++; + sql_print_information("Forcing to reconnect slave I/O thread"); + if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings, + reconnect_messages[SLAVE_RECON_ACT_DUMP])) + goto err; + goto connected; + }); while (!io_slave_killed(thd,mi)) { ulong event_len; - - suppress_warnings= 0; /* We say "waiting" because read_event() will wait if there's nothing to read. But if there's something to read, it will not wait. The important thing is to not confuse users by saying "reading" whereas we're in fact receiving nothing. */ - thd->proc_info= "Waiting for master to send event"; + thd_proc_info(thd, "Waiting for master to send event"); event_len= read_event(mysql, mi, &suppress_warnings); - if (io_slave_killed(thd,mi)) - { - if (global_system_variables.log_warnings) - sql_print_information("Slave I/O thread killed while reading event"); - goto err; - } + if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \ +reading event")) + goto err; + DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_EVENT", + if (!retry_count_event) + { + retry_count_event++; + sql_print_information("Forcing to reconnect slave I/O thread"); + if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings, + reconnect_messages[SLAVE_RECON_ACT_EVENT])) + goto err; + goto connected; + }); if (event_len == packet_error) { @@ -3712,49 +2442,21 @@ max_allowed_packet", Stopping slave I/O thread due to out-of-memory error from master"); goto err; } - mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT; - thd->proc_info = "Waiting to reconnect after a failed master event read"; -#ifdef SIGNAL_WITH_VIO_CLOSE - thd->clear_active_vio(); -#endif - end_server(mysql); - if (retry_count++) - { - if (retry_count > master_retry_count) - goto err; // Don't retry forever - safe_sleep(thd,mi->connect_retry,(CHECK_KILLED_FUNC)io_slave_killed, - (void*) mi); - } - if (io_slave_killed(thd,mi)) - { - if (global_system_variables.log_warnings) - sql_print_information("Slave I/O thread killed while waiting to \ -reconnect after a failed read"); - goto err; - } - thd->proc_info = "Reconnecting after a failed master event read"; - if (!suppress_warnings) - sql_print_information("Slave I/O thread: Failed reading log event, \ -reconnecting to retry, log '%s' position %s", IO_RPL_LOG_NAME, - llstr(mi->master_log_pos, llbuff)); - if (safe_reconnect(thd, mysql, mi, suppress_warnings) || - io_slave_killed(thd,mi)) - { - if (global_system_variables.log_warnings) - sql_print_information("Slave I/O thread killed during or after a \ -reconnect done to recover from failed read"); - goto err; - } - goto connected; + if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings, + reconnect_messages[SLAVE_RECON_ACT_EVENT])) + goto err; + goto connected; } // if (event_len == packet_error) - retry_count=0; // ok event, reset retry counter - thd->proc_info = "Queueing master event to the relay log"; + retry_count=0; // ok event, reset retry counter + thd_proc_info(thd, "Queueing master event to the relay log"); if (queue_event(mi,(const char*)mysql->net.read_pos + 1, - event_len)) + event_len)) { - sql_print_error("Slave I/O thread could not queue event from master"); - goto err; + mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE, + ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE), + "could not queue event from master"); + goto err; } if (flush_master_info(mi, 1)) { @@ -3780,39 +2482,30 @@ reconnect done to recover from failed read"); ignore_log_space_limit=%d", llstr(rli->log_space_limit,llbuf1), llstr(rli->log_space_total,llbuf2), - (int) rli->ignore_log_space_limit)); + (int) rli->ignore_log_space_limit)); } #endif if (rli->log_space_limit && rli->log_space_limit < - rli->log_space_total && + rli->log_space_total && !rli->ignore_log_space_limit) - if (wait_for_relay_log_space(rli)) - { - sql_print_error("Slave I/O thread aborted while waiting for relay \ + if (wait_for_relay_log_space(rli)) + { + sql_print_error("Slave I/O thread aborted while waiting for relay \ log space"); - goto err; - } - // TODO: check debugging abort code -#ifndef DBUG_OFF - if (abort_slave_event_count && !--events_till_abort) - { - sql_print_error("Slave I/O thread: debugging abort"); - goto err; - } -#endif - } + goto err; + } + } } // error = 0; err: // print the current replication position sql_print_information("Slave I/O thread exiting, read up to log '%s', position %s", - IO_RPL_LOG_NAME, llstr(mi->master_log_pos,llbuff)); + IO_RPL_LOG_NAME, llstr(mi->master_log_pos,llbuff)); VOID(pthread_mutex_lock(&LOCK_thread_count)); - thd->query= 0; // extra safety - thd->query_length= 0; - thd->reset_db(NULL, 0); + thd->query = thd->db = 0; // extra safety + thd->query_length= thd->db_length= 0; VOID(pthread_mutex_unlock(&LOCK_thread_count)); if (mysql) { @@ -3831,17 +2524,17 @@ err: mi->mysql=0; } write_ignored_events_info_to_relay_log(thd, mi); - thd->proc_info = "Waiting for slave mutex on exit"; + thd_proc_info(thd, "Waiting for slave mutex on exit"); pthread_mutex_lock(&mi->run_lock); /* Forget the relay log's format */ delete mi->rli.relay_log.description_event_for_queue; mi->rli.relay_log.description_event_for_queue= 0; - // TODO: make rpl_status part of MASTER_INFO + // TODO: make rpl_status part of Master_info change_rpl_status(RPL_ACTIVE_SLAVE,RPL_IDLE_SLAVE); DBUG_ASSERT(thd->net.buff != 0); net_end(&thd->net); // destructor will not free it, because net.vio is 0 - close_thread_tables(thd, 0); + close_thread_tables(thd); pthread_mutex_lock(&LOCK_thread_count); THD_CHECK_SENTRY(thd); delete thd; @@ -3856,13 +2549,9 @@ err: */ pthread_cond_broadcast(&mi->stop_cond); // tell the world we are done pthread_mutex_unlock(&mi->run_lock); -#ifndef DBUG_OFF - if (abort_slave_event_count && !events_till_abort) - goto slave_begin; -#endif my_thread_end(); pthread_exit(0); - DBUG_RETURN(0); // Can't return anything here + DBUG_RETURN(0); // Can't return anything here } @@ -3870,26 +2559,23 @@ err: pthread_handler_t handle_slave_sql(void *arg) { - THD *thd; /* needs to be first for thread_stack */ + THD *thd; /* needs to be first for thread_stack */ char llbuff[22],llbuff1[22]; - RELAY_LOG_INFO* rli = &((MASTER_INFO*)arg)->rli; + + Relay_log_info* rli = &((Master_info*)arg)->rli; const char *errmsg; // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff my_thread_init(); DBUG_ENTER("handle_slave_sql"); -#ifndef DBUG_OFF -slave_begin: -#endif - DBUG_ASSERT(rli->inited); pthread_mutex_lock(&rli->run_lock); DBUG_ASSERT(!rli->slave_running); errmsg= 0; -#ifndef DBUG_OFF +#ifndef DBUG_OFF rli->events_till_abort = abort_slave_event_count; -#endif +#endif thd = new THD; // note that contructor of THD uses DBUG_ ! thd->thread_stack = (char*)&thd; // remember where our stack is @@ -3938,22 +2624,23 @@ slave_begin: now. But the master timestamp is reset by RESET SLAVE & CHANGE MASTER. */ - clear_slave_error(rli); + rli->clear_error(); //tell the I/O thread to take relay_log_space_limit into account from now on pthread_mutex_lock(&rli->log_space_lock); rli->ignore_log_space_limit= 0; pthread_mutex_unlock(&rli->log_space_lock); rli->trans_retries= 0; // start from "no error" + DBUG_PRINT("info", ("rli->trans_retries: %lu", rli->trans_retries)); if (init_relay_log_pos(rli, - rli->group_relay_log_name, - rli->group_relay_log_pos, - 1 /*need data lock*/, &errmsg, + rli->group_relay_log_name, + rli->group_relay_log_pos, + 1 /*need data lock*/, &errmsg, 1 /*look for a description_event*/)) { sql_print_error("Error initializing relay log position: %s", - errmsg); + errmsg); goto err; } THD_CHECK_SENTRY(thd); @@ -3961,7 +2648,7 @@ slave_begin: { char llbuf1[22], llbuf2[22]; DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s", - llstr(my_b_tell(rli->cur_log),llbuf1), + llstr(my_b_tell(rli->cur_log),llbuf1), llstr(rli->event_relay_log_pos,llbuf2))); DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE); /* @@ -3984,19 +2671,19 @@ slave_begin: DBUG_ASSERT(rli->sql_thd == thd); DBUG_PRINT("master_info",("log_file_name: %s position: %s", - rli->group_master_log_name, - llstr(rli->group_master_log_pos,llbuff))); + rli->group_master_log_name, + llstr(rli->group_master_log_pos,llbuff))); if (global_system_variables.log_warnings) sql_print_information("Slave SQL thread initialized, starting replication in \ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME, - llstr(rli->group_master_log_pos,llbuff),rli->group_relay_log_name, - llstr(rli->group_relay_log_pos,llbuff1)); + llstr(rli->group_master_log_pos,llbuff),rli->group_relay_log_name, + llstr(rli->group_relay_log_pos,llbuff1)); /* execute init_slave variable */ if (sys_init_slave.value_length) { execute_init_command(thd, &sys_init_slave, &LOCK_sys_init_slave); - if (thd->query_error) + if (thd->is_slave_error) { sql_print_error("\ Slave SQL thread aborted. Can't execute init_slave query"); @@ -4009,7 +2696,7 @@ Slave SQL thread aborted. Can't execute init_slave query"); do not want to wait for next event in this case. */ pthread_mutex_lock(&rli->data_lock); - if (rli->until_condition != RELAY_LOG_INFO::UNTIL_NONE && + if (rli->until_condition != Relay_log_info::UNTIL_NONE && rli->is_until_satisfied(rli->group_master_log_pos)) { char buf[22]; @@ -4024,45 +2711,100 @@ Slave SQL thread aborted. Can't execute init_slave query"); while (!sql_slave_killed(thd,rli)) { - thd->proc_info = "Reading event from the relay log"; + thd_proc_info(thd, "Reading event from the relay log"); DBUG_ASSERT(rli->sql_thd == thd); THD_CHECK_SENTRY(thd); if (exec_relay_log_event(thd,rli)) { + DBUG_PRINT("info", ("exec_relay_log_event() failed")); // do not scare the user if SQL thread was simply killed or stopped if (!sql_slave_killed(thd,rli)) - sql_print_error("\ + { + /* + retrieve as much info as possible from the thd and, error + codes and warnings and print this to the error log as to + allow the user to locate the error + */ + uint32 const last_errno= rli->last_error().number; + + if (thd->is_error()) + { + char const *const errmsg= thd->main_da.message(); + + DBUG_PRINT("info", + ("thd->main_da.sql_errno()=%d; rli->last_error.number=%d", + thd->main_da.sql_errno(), last_errno)); + if (last_errno == 0) + { + rli->report(ERROR_LEVEL, thd->main_da.sql_errno(), errmsg); + } + else if (last_errno != thd->main_da.sql_errno()) + { + sql_print_error("Slave (additional info): %s Error_code: %d", + errmsg, thd->main_da.sql_errno()); + } + } + + /* Print any warnings issued */ + List_iterator_fast<MYSQL_ERROR> it(thd->warn_list); + MYSQL_ERROR *err; + /* + Added controlled slave thread cancel for replication + of user-defined variables. + */ + bool udf_error = false; + while ((err= it++)) + { + if (err->code == ER_CANT_OPEN_LIBRARY) + udf_error = true; + sql_print_warning("Slave: %s Error_code: %d",err->msg, err->code); + } + if (udf_error) + sql_print_error("Error loading user-defined library, slave SQL " + "thread aborted. Install the missing library, and restart the " + "slave SQL thread with \"SLAVE START\". We stopped at log '%s' " + "position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, + llbuff)); + else + sql_print_error("\ Error running query, slave SQL thread aborted. Fix the problem, and restart \ the slave SQL thread with \"SLAVE START\". We stopped at log \ '%s' position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, llbuff)); + } goto err; } } /* Thread stopped. Print the current replication position to the log */ sql_print_information("Slave SQL thread exiting, replication stopped in log " - "'%s' at position %s", - RPL_LOG_NAME, llstr(rli->group_master_log_pos,llbuff)); + "'%s' at position %s", + RPL_LOG_NAME, llstr(rli->group_master_log_pos,llbuff)); err: + + /* + Some events set some playgrounds, which won't be cleared because thread + stops. Stopping of this thread may not be known to these events ("stop" + request is detected only by the present function, not by events), so we + must "proactively" clear playgrounds: + */ + rli->cleanup_context(thd, 1); VOID(pthread_mutex_lock(&LOCK_thread_count)); /* Some extra safety, which should not been needed (normally, event deletion should already have done these assignments (each event which sets these variables is supposed to set them to 0 before terminating)). */ - thd->catalog= 0; - thd->reset_db(NULL, 0); - thd->query= 0; - thd->query_length= 0; + thd->query= thd->db= thd->catalog= 0; + thd->query_length= thd->db_length= 0; VOID(pthread_mutex_unlock(&LOCK_thread_count)); - thd->proc_info = "Waiting for slave mutex on exit"; + thd_proc_info(thd, "Waiting for slave mutex on exit"); pthread_mutex_lock(&rli->run_lock); /* We need data_lock, at least to wake up any waiting master_pos_wait() */ pthread_mutex_lock(&rli->data_lock); DBUG_ASSERT(rli->slave_running == 1); // tracking buffer overrun /* When master_pos_wait() wakes up it will check this and terminate */ - rli->slave_running= 0; + rli->slave_running= 0; /* Forget the relay log's format */ delete rli->relay_log.description_event_for_exec; rli->relay_log.description_event_for_exec= 0; @@ -4089,29 +2831,17 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \ THD_CHECK_SENTRY(thd); delete thd; pthread_mutex_unlock(&LOCK_thread_count); - /* Note: the order of the broadcast and unlock calls below (first broadcast, then unlock) is important. Otherwise a killer_thread can execute between the calls and delete the mi structure leading to a crash! (see BUG#25306 for details) */ pthread_cond_broadcast(&rli->stop_cond); -#ifndef DBUG_OFF - /* - Bug #19938 Valgrind error (race) in handle_slave_sql() - Read the value of rli->event_till_abort before releasing the mutex - */ - const int eta= rli->events_till_abort; -#endif pthread_mutex_unlock(&rli->run_lock); // tell the world we are done -#ifndef DBUG_OFF // TODO: reconsider the code below - if (abort_slave_event_count && !eta) - goto slave_begin; -#endif my_thread_end(); pthread_exit(0); - DBUG_RETURN(0); // Can't return anything here + DBUG_RETURN(0); // Can't return anything here } @@ -4119,7 +2849,7 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \ process_io_create_file() */ -static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev) +static int process_io_create_file(Master_info* mi, Create_file_log_event* cev) { int error = 1; ulong num_bytes; @@ -4130,10 +2860,8 @@ static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev) if (unlikely(!cev->is_valid())) DBUG_RETURN(1); - /* - TODO: fix to honor table rules, not only db rules - */ - if (!db_ok(cev->db, replicate_do_db, replicate_ignore_db)) + + if (!rpl_filter->db_ok(cev->db)) { skip_load_data_infile(net); DBUG_RETURN(0); @@ -4142,11 +2870,11 @@ static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev) thd->file_id = cev->file_id = mi->file_id++; thd->server_id = cev->server_id; cev_not_written = 1; - + if (unlikely(net_request_file(net,cev->fname))) { sql_print_error("Slave I/O: failed requesting download of '%s'", - cev->fname); + cev->fname); goto err; } @@ -4157,18 +2885,19 @@ static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev) */ { Append_block_log_event aev(thd,0,0,0,0); - + for (;;) { if (unlikely((num_bytes=my_net_read(net)) == packet_error)) { - sql_print_error("Network read error downloading '%s' from master", - cev->fname); - goto err; + sql_print_error("Network read error downloading '%s' from master", + cev->fname); + goto err; } if (unlikely(!num_bytes)) /* eof */ { - net_write_command(net, 0, "", 0, "", 0);/* 3.23 master wants it */ + /* 3.23 master wants it */ + net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0); /* If we wrote Create_file_log_event, then we need to write Execute_load_log_event. If we did not write Create_file_log_event, @@ -4176,43 +2905,46 @@ static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev) INFILE had not existed, i.e. write nothing. */ if (unlikely(cev_not_written)) - break; - Execute_load_log_event xev(thd,0,0); - xev.log_pos = cev->log_pos; - if (unlikely(mi->rli.relay_log.append(&xev))) - { - sql_print_error("Slave I/O: error writing Exec_load event to \ -relay log"); - goto err; - } - mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total); - break; + break; + Execute_load_log_event xev(thd,0,0); + xev.log_pos = cev->log_pos; + if (unlikely(mi->rli.relay_log.append(&xev))) + { + mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE, + ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE), + "error writing Exec_load event to relay log"); + goto err; + } + mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total); + break; } if (unlikely(cev_not_written)) { - cev->block = (char*)net->read_pos; - cev->block_len = num_bytes; - if (unlikely(mi->rli.relay_log.append(cev))) - { - sql_print_error("Slave I/O: error writing Create_file event to \ -relay log"); - goto err; - } - cev_not_written=0; - mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total); + cev->block = net->read_pos; + cev->block_len = num_bytes; + if (unlikely(mi->rli.relay_log.append(cev))) + { + mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE, + ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE), + "error writing Create_file event to relay log"); + goto err; + } + cev_not_written=0; + mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total); } else { - aev.block = (char*)net->read_pos; - aev.block_len = num_bytes; - aev.log_pos = cev->log_pos; - if (unlikely(mi->rli.relay_log.append(&aev))) - { - sql_print_error("Slave I/O: error writing Append_block event to \ -relay log"); - goto err; - } - mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total) ; + aev.block = net->read_pos; + aev.block_len = num_bytes; + aev.log_pos = cev->log_pos; + if (unlikely(mi->rli.relay_log.append(&aev))) + { + mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE, + ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE), + "error writing Append_block event to relay log"); + goto err; + } + mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total) ; } } } @@ -4227,8 +2959,8 @@ err: SYNOPSIS process_io_rotate() - mi master_info for the slave - rev The rotate log event read from the binary log + mi master_info for the slave + rev The rotate log event read from the binary log DESCRIPTION Updates the master info with the place in the next binary @@ -4239,12 +2971,12 @@ err: We assume we already locked mi->data_lock RETURN VALUES - 0 ok - 1 Log event is illegal + 0 ok + 1 Log event is illegal */ -static int process_io_rotate(MASTER_INFO *mi, Rotate_log_event *rev) +static int process_io_rotate(Master_info *mi, Rotate_log_event *rev) { DBUG_ENTER("process_io_rotate"); safe_mutex_assert_owner(&mi->data_lock); @@ -4256,14 +2988,14 @@ static int process_io_rotate(MASTER_INFO *mi, Rotate_log_event *rev) memcpy(mi->master_log_name, rev->new_log_ident, rev->ident_len+1); mi->master_log_pos= rev->pos; DBUG_PRINT("info", ("master_log_pos: '%s' %lu", - mi->master_log_name, (ulong) mi->master_log_pos)); + mi->master_log_name, (ulong) mi->master_log_pos)); #ifndef DBUG_OFF /* If we do not do this, we will be getting the first rotate event forever, so we need to not disconnect after one. */ if (disconnect_slave_event_count) - events_till_disconnect++; + mi->events_till_disconnect++; #endif /* @@ -4292,14 +3024,14 @@ static int process_io_rotate(MASTER_INFO *mi, Rotate_log_event *rev) Reads a 3.23 event and converts it to the slave's format. This code was copied from MySQL 4.0. */ -static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf, - ulong event_len) +static int queue_binlog_ver_1_event(Master_info *mi, const char *buf, + ulong event_len) { const char *errmsg = 0; ulong inc_pos; bool ignore_event= 0; char *tmp_buf = 0; - RELAY_LOG_INFO *rli= &mi->rli; + Relay_log_info *rli= &mi->rli; DBUG_ENTER("queue_binlog_ver_1_event"); /* @@ -4310,7 +3042,8 @@ static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf, { if (unlikely(!(tmp_buf=(char*)my_malloc(event_len+1,MYF(MY_WME))))) { - sql_print_error("Slave I/O: out of memory for Load event"); + mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, + ER(ER_SLAVE_FATAL_ERROR), "Memory allocation failed"); DBUG_RETURN(1); } memcpy(tmp_buf,buf,event_len); @@ -4337,10 +3070,11 @@ static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf, { sql_print_error("Read invalid event from master: '%s',\ master could be corrupt but a more likely cause of this is a bug", - errmsg); + errmsg); my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR)); DBUG_RETURN(1); } + pthread_mutex_lock(&mi->data_lock); ev->log_pos= mi->master_log_pos; /* 3.23 events don't contain log_pos */ switch (ev->get_type_code()) { @@ -4383,8 +3117,8 @@ static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf, } if (likely(!ignore_event)) { - if (ev->log_pos) - /* + if (ev->log_pos) + /* Don't do it for fake Rotate events (see comment in Log_event::Log_event(const char* buf...) in log_event.cc). */ @@ -4408,13 +3142,13 @@ static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf, Reads a 4.0 event and converts it to the slave's format. This code was copied from queue_binlog_ver_1_event(), with some affordable simplifications. */ -static int queue_binlog_ver_3_event(MASTER_INFO *mi, const char *buf, - ulong event_len) +static int queue_binlog_ver_3_event(Master_info *mi, const char *buf, + ulong event_len) { const char *errmsg = 0; ulong inc_pos; char *tmp_buf = 0; - RELAY_LOG_INFO *rli= &mi->rli; + Relay_log_info *rli= &mi->rli; DBUG_ENTER("queue_binlog_ver_3_event"); /* read_log_event() will adjust log_pos to be end_log_pos */ @@ -4424,7 +3158,7 @@ static int queue_binlog_ver_3_event(MASTER_INFO *mi, const char *buf, { sql_print_error("Read invalid event from master: '%s',\ master could be corrupt but a more likely cause of this is a bug", - errmsg); + errmsg); my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR)); DBUG_RETURN(1); } @@ -4467,24 +3201,26 @@ err: (exactly, slave's) format. To do the conversion, we create a 5.0 event from the 3.23/4.0 bytes, then write this event to the relay log. - TODO: + TODO: Test this code before release - it has to be tested on a separate setup with 3.23 master or 4.0 master */ -static int queue_old_event(MASTER_INFO *mi, const char *buf, - ulong event_len) +static int queue_old_event(Master_info *mi, const char *buf, + ulong event_len) { + DBUG_ENTER("queue_old_event"); + switch (mi->rli.relay_log.description_event_for_queue->binlog_version) { case 1: - return queue_binlog_ver_1_event(mi,buf,event_len); + DBUG_RETURN(queue_binlog_ver_1_event(mi,buf,event_len)); case 3: - return queue_binlog_ver_3_event(mi,buf,event_len); + DBUG_RETURN(queue_binlog_ver_3_event(mi,buf,event_len)); default: /* unsupported format; eg version 2 */ DBUG_PRINT("info",("unsupported binlog format %d in queue_old_event()", - mi->rli.relay_log.description_event_for_queue->binlog_version)); - return 1; + mi->rli.relay_log.description_event_for_queue->binlog_version)); + DBUG_RETURN(1); } } @@ -4498,11 +3234,11 @@ static int queue_old_event(MASTER_INFO *mi, const char *buf, any >=5.0.0 format. */ -int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) +static int queue_event(Master_info* mi,const char* buf, ulong event_len) { int error= 0; ulong inc_pos; - RELAY_LOG_INFO *rli= &mi->rli; + Relay_log_info *rli= &mi->rli; pthread_mutex_t *log_lock= rli->relay_log.get_log_lock(); DBUG_ENTER("queue_event"); @@ -4512,6 +3248,7 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT /* a way to escape */) DBUG_RETURN(queue_old_event(mi,buf,event_len)); + LINT_INIT(inc_pos); pthread_mutex_lock(&mi->data_lock); switch (buf[EVENT_TYPE_OFFSET]) { @@ -4522,7 +3259,7 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) cleaning is already done on a per-master-thread basis (as the master server is shutting down cleanly, it has written all DROP TEMPORARY TABLE prepared statements' deletion are TODO only when we binlog prep stmts). - + We don't even increment mi->master_log_pos, because we may be just after a Rotate event. Btw, in a few milliseconds we are going to have a Start event from the next binlog (unless the master is presently running @@ -4531,7 +3268,7 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) goto err; case ROTATE_EVENT: { - Rotate_log_event rev(buf,event_len,mi->rli.relay_log.description_event_for_queue); + Rotate_log_event rev(buf,event_len,mi->rli.relay_log.description_event_for_queue); if (unlikely(process_io_rotate(mi,&rev))) { error= 1; @@ -4566,17 +3303,17 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) } delete mi->rli.relay_log.description_event_for_queue; mi->rli.relay_log.description_event_for_queue= tmp; - /* + /* Though this does some conversion to the slave's format, this will - preserve the master's binlog format version, and number of event types. + preserve the master's binlog format version, and number of event types. */ - /* + /* If the event was not requested by the slave (the slave did not ask for - it), i.e. has end_log_pos=0, we do not increment mi->master_log_pos + it), i.e. has end_log_pos=0, we do not increment mi->master_log_pos */ inc_pos= uint4korr(buf+LOG_POS_OFFSET) ? event_len : 0; DBUG_PRINT("info",("binlog format is now %d", - mi->rli.relay_log.description_event_for_queue->binlog_version)); + mi->rli.relay_log.description_event_for_queue->binlog_version)); } break; @@ -4585,8 +3322,8 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) break; } - /* - If this event is originating from this server, don't queue it. + /* + If this event is originating from this server, don't queue it. We don't check this for 3.23 events because it's simpler like this; 3.23 will be filtered anyway by the SQL slave thread which also tests the server id (we must also keep this test in the SQL thread, in case somebody @@ -4601,7 +3338,7 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) pthread_mutex_lock(log_lock); if ((uint4korr(buf + SERVER_ID_OFFSET) == ::server_id) && - !replicate_same_server_id) + !mi->rli.replicate_same_server_id) { /* Do not write it to the relay log. @@ -4627,9 +3364,9 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len) rli->ign_master_log_pos_end= mi->master_log_pos; } rli->relay_log.signal_update(); // the slave SQL thread needs to re-check - DBUG_PRINT("info", ("master_log_pos: %lu event originating from the same server, ignored", + DBUG_PRINT("info", ("master_log_pos: %lu, event originating from the same server, ignored", (ulong) mi->master_log_pos)); - } + } else { /* write the event to the relay log */ @@ -4653,7 +3390,7 @@ err: } -void end_relay_log_info(RELAY_LOG_INFO* rli) +void end_relay_log_info(Relay_log_info* rli) { DBUG_ENTER("end_relay_log_info"); @@ -4688,18 +3425,20 @@ void end_relay_log_info(RELAY_LOG_INFO* rli) SYNPOSIS safe_connect() - thd Thread handler for slave - mysql MySQL connection handle - mi Replication handle + thd Thread handler for slave + mysql MySQL connection handle + mi Replication handle RETURN - 0 ok - # Error + 0 ok + # Error */ -static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) +static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi) { - return connect_to_master(thd, mysql, mi, 0, 0); + DBUG_ENTER("safe_connect"); + + DBUG_RETURN(connect_to_master(thd, mysql, mi, 0, 0)); } @@ -4712,33 +3451,37 @@ static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) master_retry_count times */ -static int connect_to_master(THD* thd, MYSQL* mysql, MASTER_INFO* mi, - bool reconnect, bool suppress_warnings) +static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi, + bool reconnect, bool suppress_warnings) { int slave_was_killed; - int last_errno= -2; // impossible error + int last_errno= -2; // impossible error ulong err_count=0; char llbuff[22]; DBUG_ENTER("connect_to_master"); #ifndef DBUG_OFF - events_till_disconnect = disconnect_slave_event_count; + mi->events_till_disconnect = disconnect_slave_event_count; #endif ulong client_flag= CLIENT_REMEMBER_OPTIONS; if (opt_slave_compressed_protocol) - client_flag=CLIENT_COMPRESS; /* We will use compression */ + client_flag=CLIENT_COMPRESS; /* We will use compression */ mysql_options(mysql, MYSQL_OPT_CONNECT_TIMEOUT, (char *) &slave_net_timeout); mysql_options(mysql, MYSQL_OPT_READ_TIMEOUT, (char *) &slave_net_timeout); - + #ifdef HAVE_OPENSSL if (mi->ssl) - mysql_ssl_set(mysql, + { + mysql_ssl_set(mysql, mi->ssl_key[0]?mi->ssl_key:0, - mi->ssl_cert[0]?mi->ssl_cert:0, + mi->ssl_cert[0]?mi->ssl_cert:0, mi->ssl_ca[0]?mi->ssl_ca:0, mi->ssl_capath[0]?mi->ssl_capath:0, mi->ssl_cipher[0]?mi->ssl_cipher:0); + mysql_options(mysql, MYSQL_OPT_SSL_VERIFY_SERVER_CERT, + &mi->ssl_verify_server_cert); + } #endif mysql_options(mysql, MYSQL_SET_CHARSET_NAME, default_charset_info->csname); @@ -4746,23 +3489,21 @@ static int connect_to_master(THD* thd, MYSQL* mysql, MASTER_INFO* mi, mysql_options(mysql, MYSQL_SET_CHARSET_DIR, (char *) charsets_dir); while (!(slave_was_killed = io_slave_killed(thd,mi)) && - (reconnect ? mysql_reconnect(mysql) != 0 : - mysql_real_connect(mysql, mi->host, mi->user, mi->password, 0, - mi->port, 0, client_flag) == 0)) + (reconnect ? mysql_reconnect(mysql) != 0 : + mysql_real_connect(mysql, mi->host, mi->user, mi->password, 0, + mi->port, 0, client_flag) == 0)) { /* Don't repeat last error */ if ((int)mysql_errno(mysql) != last_errno) { last_errno=mysql_errno(mysql); suppress_warnings= 0; - sql_print_error("Slave I/O thread: error %s to master \ -'%s@%s:%d': \ -Error: '%s' errno: %d retry-time: %d retries: %lu", - (reconnect ? "reconnecting" : "connecting"), - mi->user,mi->host,mi->port, - mysql_error(mysql), last_errno, - mi->connect_retry, - master_retry_count); + mi->report(ERROR_LEVEL, last_errno, + "error %s to master '%s@%s:%d'" + " - retry-time: %d retries: %lu", + (reconnect ? "reconnecting" : "connecting"), + mi->user, mi->host, mi->port, + mi->connect_retry, master_retry_count); } /* By default we try forever. The reason is that failure will trigger @@ -4778,29 +3519,29 @@ Error: '%s' errno: %d retry-time: %d retries: %lu", break; } safe_sleep(thd,mi->connect_retry,(CHECK_KILLED_FUNC)io_slave_killed, - (void*)mi); + (void*)mi); } if (!slave_was_killed) { if (reconnect) - { + { if (!suppress_warnings && global_system_variables.log_warnings) - sql_print_information("Slave: connected to master '%s@%s:%d',\ + sql_print_information("Slave: connected to master '%s@%s:%d',\ replication resumed in log '%s' at position %s", mi->user, - mi->host, mi->port, - IO_RPL_LOG_NAME, - llstr(mi->master_log_pos,llbuff)); + mi->host, mi->port, + IO_RPL_LOG_NAME, + llstr(mi->master_log_pos,llbuff)); } else { change_rpl_status(RPL_IDLE_SLAVE,RPL_ACTIVE_SLAVE); - mysql_log.write(thd, COM_CONNECT_OUT, "%s@%s:%d", - mi->user, mi->host, mi->port); + general_log_print(thd, COM_CONNECT_OUT, "%s@%s:%d", + mi->user, mi->host, mi->port); } #ifdef SIGNAL_WITH_VIO_CLOSE thd->set_active_vio(mysql->net.vio); -#endif +#endif } mysql->reconnect= 1; DBUG_PRINT("exit",("slave_was_killed: %d", slave_was_killed)); @@ -4816,8 +3557,8 @@ replication resumed in log '%s' at position %s", mi->user, master_retry_count times */ -static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi, - bool suppress_warnings) +static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi, + bool suppress_warnings) { DBUG_ENTER("safe_reconnect"); DBUG_RETURN(connect_to_master(thd, mysql, mi, 1, suppress_warnings)); @@ -4830,7 +3571,7 @@ static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi, SYNOPSIS flush_relay_log_info() - rli Relay log information + rli Relay log information NOTES - As this is only called by the slave thread, we don't need to @@ -4849,13 +3590,18 @@ static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi, longlong2str. RETURN VALUES - 0 ok - 1 write error + 0 ok + 1 write error */ -bool flush_relay_log_info(RELAY_LOG_INFO* rli) +bool flush_relay_log_info(Relay_log_info* rli) { bool error=0; + DBUG_ENTER("flush_relay_log_info"); + + if (unlikely(rli->no_storage)) + DBUG_RETURN(0); + IO_CACHE *file = &rli->info_file; char buff[FN_REFLEN*2+22*2+4], *pos; @@ -4868,12 +3614,13 @@ bool flush_relay_log_info(RELAY_LOG_INFO* rli) *pos++='\n'; pos=longlong2str(rli->group_master_log_pos, pos, 10); *pos='\n'; - if (my_b_write(file, (byte*) buff, (ulong) (pos-buff)+1)) + if (my_b_write(file, (uchar*) buff, (size_t) (pos-buff)+1)) error=1; if (flush_io_cache(file)) error=1; + /* Flushing the relay log is done by the slave I/O thread */ - return error; + DBUG_RETURN(error); } @@ -4881,20 +3628,20 @@ bool flush_relay_log_info(RELAY_LOG_INFO* rli) Called when we notice that the current "hot" log got rotated under our feet. */ -static IO_CACHE *reopen_relay_log(RELAY_LOG_INFO *rli, const char **errmsg) +static IO_CACHE *reopen_relay_log(Relay_log_info *rli, const char **errmsg) { + DBUG_ENTER("reopen_relay_log"); DBUG_ASSERT(rli->cur_log != &rli->cache_buf); DBUG_ASSERT(rli->cur_log_fd == -1); - DBUG_ENTER("reopen_relay_log"); IO_CACHE *cur_log = rli->cur_log=&rli->cache_buf; if ((rli->cur_log_fd=open_binlog(cur_log,rli->event_relay_log_name, - errmsg)) <0) + errmsg)) <0) DBUG_RETURN(0); /* We want to start exactly where we was before: - relay_log_pos Current log pos - pending Number of bytes already processed from the event + relay_log_pos Current log pos + pending Number of bytes already processed from the event */ rli->event_relay_log_pos= max(rli->event_relay_log_pos, BIN_LOG_HEADER_SIZE); my_b_seek(cur_log,rli->event_relay_log_pos); @@ -4902,17 +3649,22 @@ static IO_CACHE *reopen_relay_log(RELAY_LOG_INFO *rli, const char **errmsg) } -Log_event* next_event(RELAY_LOG_INFO* rli) +static Log_event* next_event(Relay_log_info* rli) { Log_event* ev; IO_CACHE* cur_log = rli->cur_log; - pthread_mutex_t *log_lock = rli->relay_log.get_log_lock(); + pthread_mutex_t *log_lock = rli->relay_log.get_log_lock(); const char* errmsg=0; THD* thd = rli->sql_thd; - DBUG_ENTER("next_event"); + DBUG_ASSERT(thd != 0); +#ifndef DBUG_OFF + if (abort_slave_event_count && !rli->events_till_abort--) + DBUG_RETURN(0); +#endif + /* For most operations we need to protect rli members with data_lock, so we assume calling function acquired this mutex for us and we will @@ -4921,7 +3673,7 @@ Log_event* next_event(RELAY_LOG_INFO* rli) pthread_cond_wait() with the non-data_lock mutex */ safe_mutex_assert_owner(&rli->data_lock); - + while (!sql_slave_killed(thd,rli)) { /* @@ -4942,17 +3694,17 @@ Log_event* next_event(RELAY_LOG_INFO* rli) pthread_mutex_lock(log_lock); /* - Reading xxx_file_id is safe because the log will only - be rotated when we hold relay_log.LOCK_log + Reading xxx_file_id is safe because the log will only + be rotated when we hold relay_log.LOCK_log */ if (rli->relay_log.get_open_count() != rli->cur_log_old_open_count) { - // The master has switched to a new log file; Reopen the old log file - cur_log=reopen_relay_log(rli, &errmsg); - pthread_mutex_unlock(log_lock); - if (!cur_log) // No more log files - goto err; - hot_log=0; // Using old binary log + // The master has switched to a new log file; Reopen the old log file + cur_log=reopen_relay_log(rli, &errmsg); + pthread_mutex_unlock(log_lock); + if (!cur_log) // No more log files + goto err; + hot_log=0; // Using old binary log } } /* @@ -4983,7 +3735,7 @@ Log_event* next_event(RELAY_LOG_INFO* rli) When the relay log is created when the I/O thread starts, easy: the master will send the description event and we will queue it. But if the relay log is created by new_file(): then the solution is: - MYSQL_LOG::open() will write the buffered description event. + MYSQL_BIN_LOG::open() will write the buffered description event. */ if ((ev=Log_event::read_log_event(cur_log,0, rli->relay_log.description_event_for_exec))) @@ -4996,25 +3748,25 @@ Log_event* next_event(RELAY_LOG_INFO* rli) */ rli->future_event_relay_log_pos= my_b_tell(cur_log); if (hot_log) - pthread_mutex_unlock(log_lock); + pthread_mutex_unlock(log_lock); DBUG_RETURN(ev); } DBUG_ASSERT(thd==rli->sql_thd); - if (opt_reckless_slave) // For mysql-test + if (opt_reckless_slave) // For mysql-test cur_log->error = 0; if (cur_log->error < 0) { errmsg = "slave SQL thread aborted because of I/O error"; if (hot_log) - pthread_mutex_unlock(log_lock); + pthread_mutex_unlock(log_lock); goto err; } if (!cur_log->error) /* EOF */ { /* - On a hot log, EOF means that there are no more updates to - process and we must block until I/O thread adds some and - signals us to continue + On a hot log, EOF means that there are no more updates to + process and we must block until I/O thread adds some and + signals us to continue */ if (hot_log) { @@ -5042,14 +3794,14 @@ Log_event* next_event(RELAY_LOG_INFO* rli) time_t save_timestamp= rli->last_master_timestamp; rli->last_master_timestamp= 0; - DBUG_ASSERT(rli->relay_log.get_open_count() == + DBUG_ASSERT(rli->relay_log.get_open_count() == rli->cur_log_old_open_count); if (rli->ign_master_log_name_end[0]) { /* We generate and return a Rotate, to make our positions advance */ DBUG_PRINT("info",("seeing an ignored end segment")); - ev= new Rotate_log_event(thd, rli->ign_master_log_name_end, + ev= new Rotate_log_event(rli->ign_master_log_name_end, 0, rli->ign_master_log_pos_end, Rotate_log_event::DUP_NAME); rli->ign_master_log_name_end[0]= 0; @@ -5064,14 +3816,14 @@ Log_event* next_event(RELAY_LOG_INFO* rli) DBUG_RETURN(ev); } - /* - We can, and should release data_lock while we are waiting for - update. If we do not, show slave status will block - */ - pthread_mutex_unlock(&rli->data_lock); + /* + We can, and should release data_lock while we are waiting for + update. If we do not, show slave status will block + */ + pthread_mutex_unlock(&rli->data_lock); /* - Possible deadlock : + Possible deadlock : - the I/O thread has reached log_space_limit - the SQL thread has read all relay logs, but cannot purge for some reason: @@ -5083,10 +3835,10 @@ Log_event* next_event(RELAY_LOG_INFO* rli) the I/O thread to temporarily ignore the log_space_limit constraint, because we do not want the I/O thread to block because of space (it's ok if it blocks for any other reason (e.g. because the - master does not send anything). Then the I/O thread stops waiting + master does not send anything). Then the I/O thread stops waiting and reads more events. The SQL thread decides when the I/O thread should take log_space_limit - into account again : ignore_log_space_limit is reset to 0 + into account again : ignore_log_space_limit is reset to 0 in purge_first_log (when the SQL thread purges the just-read relay log), and also when the SQL thread starts. We should also reset ignore_log_space_limit to 0 when the user does RESET SLAVE, but in @@ -5096,12 +3848,12 @@ Log_event* next_event(RELAY_LOG_INFO* rli) */ pthread_mutex_lock(&rli->log_space_lock); // prevent the I/O thread from blocking next times - rli->ignore_log_space_limit= 1; + rli->ignore_log_space_limit= 1; /* - If the I/O thread is blocked, unblock it. - Ok to broadcast after unlock, because the mutex is only destroyed in - ~st_relay_log_info(), i.e. when rli is destroyed, and rli will not be - destroyed before we exit the present function. + If the I/O thread is blocked, unblock it. Ok to broadcast + after unlock, because the mutex is only destroyed in + ~Relay_log_info(), i.e. when rli is destroyed, and rli will + not be destroyed before we exit the present function. */ pthread_mutex_unlock(&rli->log_space_lock); pthread_cond_broadcast(&rli->log_space_cond); @@ -5110,21 +3862,21 @@ Log_event* next_event(RELAY_LOG_INFO* rli) // re-acquire data lock since we released it earlier pthread_mutex_lock(&rli->data_lock); rli->last_master_timestamp= save_timestamp; - continue; + continue; } /* - If the log was not hot, we need to move to the next log in - sequence. The next log could be hot or cold, we deal with both - cases separately after doing some common initialization + If the log was not hot, we need to move to the next log in + sequence. The next log could be hot or cold, we deal with both + cases separately after doing some common initialization */ end_io_cache(cur_log); DBUG_ASSERT(rli->cur_log_fd >= 0); my_close(rli->cur_log_fd, MYF(MY_WME)); rli->cur_log_fd = -1; - + if (relay_log_purge) { - /* + /* purge_first_log will properly set up relay log coordinates in rli. If the group's coordinates are equal to the event's coordinates (i.e. the relay log was not rotated in the middle of a group), @@ -5135,33 +3887,33 @@ Log_event* next_event(RELAY_LOG_INFO* rli) - I see no better detection method - purge_first_log is not called that often */ - if (rli->relay_log.purge_first_log + if (rli->relay_log.purge_first_log (rli, rli->group_relay_log_pos == rli->event_relay_log_pos && !strcmp(rli->group_relay_log_name,rli->event_relay_log_name))) - { - errmsg = "Error purging processed logs"; - goto err; - } + { + errmsg = "Error purging processed logs"; + goto err; + } } else { - /* - If hot_log is set, then we already have a lock on - LOCK_log. If not, we have to get the lock. - - According to Sasha, the only time this code will ever be executed - is if we are recovering from a bug. - */ - if (rli->relay_log.find_next_log(&rli->linfo, !hot_log)) - { - errmsg = "error switching to the next log"; - goto err; - } - rli->event_relay_log_pos = BIN_LOG_HEADER_SIZE; - strmake(rli->event_relay_log_name,rli->linfo.log_file_name, - sizeof(rli->event_relay_log_name)-1); - flush_relay_log_info(rli); + /* + If hot_log is set, then we already have a lock on + LOCK_log. If not, we have to get the lock. + + According to Sasha, the only time this code will ever be executed + is if we are recovering from a bug. + */ + if (rli->relay_log.find_next_log(&rli->linfo, !hot_log)) + { + errmsg = "error switching to the next log"; + goto err; + } + rli->event_relay_log_pos = BIN_LOG_HEADER_SIZE; + strmake(rli->event_relay_log_name,rli->linfo.log_file_name, + sizeof(rli->event_relay_log_name)-1); + flush_relay_log_info(rli); } /* @@ -5180,66 +3932,66 @@ Log_event* next_event(RELAY_LOG_INFO* rli) if (rli->relay_log.is_active(rli->linfo.log_file_name)) { #ifdef EXTRA_DEBUG - if (global_system_variables.log_warnings) - sql_print_information("next log '%s' is currently active", + if (global_system_variables.log_warnings) + sql_print_information("next log '%s' is currently active", rli->linfo.log_file_name); -#endif - rli->cur_log= cur_log= rli->relay_log.get_log_file(); - rli->cur_log_old_open_count= rli->relay_log.get_open_count(); - DBUG_ASSERT(rli->cur_log_fd == -1); - - /* - Read pointer has to be at the start since we are the only - reader. +#endif + rli->cur_log= cur_log= rli->relay_log.get_log_file(); + rli->cur_log_old_open_count= rli->relay_log.get_open_count(); + DBUG_ASSERT(rli->cur_log_fd == -1); + + /* + Read pointer has to be at the start since we are the only + reader. We must keep the LOCK_log to read the 4 first bytes, as this is a hot log (same as when we call read_log_event() above: for a hot log we take the mutex). - */ - if (check_binlog_magic(cur_log,&errmsg)) + */ + if (check_binlog_magic(cur_log,&errmsg)) { if (!hot_log) pthread_mutex_unlock(log_lock); - goto err; + goto err; } if (!hot_log) pthread_mutex_unlock(log_lock); - continue; + continue; } if (!hot_log) pthread_mutex_unlock(log_lock); /* - if we get here, the log was not hot, so we will have to open it - ourselves. We are sure that the log is still not hot now (a log can get - from hot to cold, but not from cold to hot). No need for LOCK_log. + if we get here, the log was not hot, so we will have to open it + ourselves. We are sure that the log is still not hot now (a log can get + from hot to cold, but not from cold to hot). No need for LOCK_log. */ #ifdef EXTRA_DEBUG if (global_system_variables.log_warnings) - sql_print_information("next log '%s' is not active", + sql_print_information("next log '%s' is not active", rli->linfo.log_file_name); -#endif +#endif // open_binlog() will check the magic header if ((rli->cur_log_fd=open_binlog(cur_log,rli->linfo.log_file_name, - &errmsg)) <0) - goto err; + &errmsg)) <0) + goto err; } else { /* - Read failed with a non-EOF error. - TODO: come up with something better to handle this error + Read failed with a non-EOF error. + TODO: come up with something better to handle this error */ if (hot_log) - pthread_mutex_unlock(log_lock); + pthread_mutex_unlock(log_lock); sql_print_error("Slave SQL thread: I/O error reading \ event(errno: %d cur_log->error: %d)", - my_errno,cur_log->error); + my_errno,cur_log->error); // set read position to the beginning of the event my_b_seek(cur_log,rli->event_relay_log_pos); /* otherwise, we have had a partial read */ errmsg = "Aborting slave SQL thread because of partial event read"; - break; // To end of function + break; // To end of function } } if (!errmsg && global_system_variables.log_warnings) { - sql_print_information("Error reading relay log event: %s", + sql_print_information("Error reading relay log event: %s", "slave SQL thread was killed"); DBUG_RETURN(0); } @@ -5253,19 +4005,20 @@ err: /* Rotate a relay log (this is used only by FLUSH LOGS; the automatic rotation because of size is simpler because when we do it we already have all relevant - locks; here we don't, so this function is mainly taking locks). - Returns nothing as we cannot catch any error (MYSQL_LOG::new_file() is void). + locks; here we don't, so this function is mainly taking locks). + Returns nothing as we cannot catch any error (MYSQL_BIN_LOG::new_file() + is void). */ -void rotate_relay_log(MASTER_INFO* mi) +void rotate_relay_log(Master_info* mi) { DBUG_ENTER("rotate_relay_log"); - RELAY_LOG_INFO* rli= &mi->rli; + Relay_log_info* rli= &mi->rli; /* We don't lock rli->run_lock. This would lead to deadlocks. */ pthread_mutex_lock(&mi->run_lock); - /* + /* We need to test inited because otherwise, new_file() will attempt to lock LOCK_log, which may not be inited (if we're not a slave). */ @@ -5276,7 +4029,7 @@ void rotate_relay_log(MASTER_INFO* mi) } /* If the relay log is closed, new_file() will do nothing. */ - rli->relay_log.new_file(1); + rli->relay_log.new_file(); /* We harvest now, because otherwise BIN_LOG_HEADER_SIZE will not immediately @@ -5301,11 +4054,11 @@ end: /** Detects, based on master's version (as found in the relay log), if master has a certain bug. - @param rli RELAY_LOG_INFO which tells the master's version + @param rli Relay_log_info which tells the master's version @param bug_id Number of the bug as found in bugs.mysql.com @return TRUE if master has the bug, FALSE if it does not. */ -bool rpl_master_has_bug(RELAY_LOG_INFO *rli, uint bug_id) +bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id) { struct st_version_range_for_one_bug { uint bug_id; @@ -5331,31 +4084,31 @@ bool rpl_master_has_bug(RELAY_LOG_INFO *rli, uint bug_id) (memcmp(introduced_in, master_ver, 3) <= 0) && (memcmp(fixed_in, master_ver, 3) > 0)) { - // a verbose message for the error log - slave_print_error(rli, ER_UNKNOWN_ERROR, - "According to the master's version ('%s')," - " it is probable that master suffers from this bug:" - " http://bugs.mysql.com/bug.php?id=%u" - " and thus replicating the current binary log event" - " may make the slave's data become different from the" - " master's data." - " To take no risk, slave refuses to replicate" - " this event and stops." - " We recommend that all updates be stopped on the" - " master and slave, that the data of both be" - " manually synchronized," - " that master's binary logs be deleted," - " that master be upgraded to a version at least" - " equal to '%d.%d.%d'. Then replication can be" - " restarted.", - rli->relay_log.description_event_for_exec->server_version, - bug_id, - fixed_in[0], fixed_in[1], fixed_in[2]); // a short message for SHOW SLAVE STATUS (message length constraints) my_printf_error(ER_UNKNOWN_ERROR, "master may suffer from" " http://bugs.mysql.com/bug.php?id=%u" " so slave stops; check error log on slave" " for more info", MYF(0), bug_id); + // a verbose message for the error log + rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR, + "According to the master's version ('%s')," + " it is probable that master suffers from this bug:" + " http://bugs.mysql.com/bug.php?id=%u" + " and thus replicating the current binary log event" + " may make the slave's data become different from the" + " master's data." + " To take no risk, slave refuses to replicate" + " this event and stops." + " We recommend that all updates be stopped on the" + " master and slave, that the data of both be" + " manually synchronized," + " that master's binary logs be deleted," + " that master be upgraded to a version at least" + " equal to '%d.%d.%d'. Then replication can be" + " restarted.", + rli->relay_log.description_event_for_exec->server_version, + bug_id, + fixed_in[0], fixed_in[1], fixed_in[2]); return TRUE; } } @@ -5367,5 +4120,8 @@ template class I_List_iterator<i_string>; template class I_List_iterator<i_string_pair>; #endif +/** + @} (end of group Replication) +*/ #endif /* HAVE_REPLICATION */ |