diff options
Diffstat (limited to 'sql/sql_repl.cc')
| -rw-r--r-- | sql/sql_repl.cc | 963 |
1 files changed, 570 insertions, 393 deletions
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc index 1940ff360c2..3e652a4d5a6 100644 --- a/sql/sql_repl.cc +++ b/sql/sql_repl.cc @@ -1,4 +1,4 @@ -/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB +/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB & Sasha This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -15,102 +15,118 @@ Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ // Sasha Pachev <sasha@mysql.com> is currently in charge of this file -// Do not mess with it without his permission! #include "mysql_priv.h" #include "sql_repl.h" #include "sql_acl.h" #include "log_event.h" -#include <thr_alarm.h> +#include "mini_client.h" #include <my_dir.h> +#include <assert.h> extern const char* any_db; -extern pthread_handler_decl(handle_slave,arg); -#ifndef DBUG_OFF int max_binlog_dump_events = 0; // unlimited bool opt_sporadic_binlog_dump_fail = 0; static int binlog_dump_count = 0; -#endif + +int check_binlog_magic(IO_CACHE* log, const char** errmsg) +{ + char magic[4]; + DBUG_ASSERT(my_b_tell(log) == 0); + + if (my_b_read(log, (byte*) magic, sizeof(magic))) + { + *errmsg = "I/O error reading the header from the binary log"; + sql_print_error("%s, errno=%d, io cache code=%d", *errmsg, my_errno, + log->error); + return 1; + } + if (memcmp(magic, BINLOG_MAGIC, sizeof(magic))) + { + *errmsg = "Binlog has bad magic number; It's not a binary log file that can be used by this version of MySQL"; + return 1; + } + return 0; +} static int fake_rotate_event(NET* net, String* packet, char* log_file_name, const char**errmsg) { - char header[LOG_EVENT_HEADER_LEN]; + char header[LOG_EVENT_HEADER_LEN], buf[ROTATE_HEADER_LEN]; memset(header, 0, 4); // when does not matter header[EVENT_TYPE_OFFSET] = ROTATE_EVENT; - char* p = strrchr(log_file_name, FN_LIBCHAR); - // find the last slash - if(p) - p++; - else - p = log_file_name; + char* p = log_file_name+dirname_length(log_file_name); uint ident_len = (uint) strlen(p); - ulong event_len = ident_len + sizeof(header); - int4store(header + EVENT_TYPE_OFFSET + 1, server_id); + ulong event_len = ident_len + ROTATE_EVENT_OVERHEAD; + int4store(header + SERVER_ID_OFFSET, server_id); int4store(header + EVENT_LEN_OFFSET, event_len); + int2store(header + FLAGS_OFFSET, 0); + + // TODO: check what problems this may cause and fix them + int4store(header + LOG_POS_OFFSET, 0); + packet->append(header, sizeof(header)); + /* We need to split the next statement because of problem with cxx */ + int4store(buf,4); // tell slave to skip magic number + int4store(buf+4,0); + packet->append(buf, ROTATE_HEADER_LEN); packet->append(p,ident_len); - if(my_net_write(net, (char*)packet->ptr(), packet->length())) - { - *errmsg = "failed on my_net_write()"; - return -1; - } + if (my_net_write(net, (char*)packet->ptr(), packet->length())) + { + *errmsg = "failed on my_net_write()"; + return -1; + } return 0; } - static int send_file(THD *thd) { NET* net = &thd->net; int fd = -1,bytes, error = 1; char fname[FN_REFLEN+1]; - char *buf; const char *errmsg = 0; int old_timeout; uint packet_len; + char buf[IO_SIZE]; // It's safe to alloc this DBUG_ENTER("send_file"); - // the client might be slow loading the data, give him wait_timeout to do - // the job - old_timeout = thd->net.timeout; - thd->net.timeout = thd->inactive_timeout; - - // spare the stack - if(!(buf = alloc_root(&thd->mem_root,IO_SIZE))) - { - errmsg = "Out of memory"; - goto err; - } - - // we need net_flush here because the client will not know it needs to send - // us the file name until it has processed the load event entry + /* + The client might be slow loading the data, give him wait_timeout to do + the job + */ + old_timeout = thd->net.read_timeout; + thd->net.read_timeout = thd->variables.net_wait_timeout; + + /* + We need net_flush here because the client will not know it needs to send + us the file name until it has processed the load event entry + */ if (net_flush(net) || (packet_len = my_net_read(net)) == packet_error) { - errmsg = "Failed reading file name"; + errmsg = "while reading file name"; goto err; } - *((char*)net->read_pos + packet_len) = 0; // terminate with \0 - //for fn_format - fn_format(fname, (char*)net->read_pos + 1, "", "", 4); + // terminate with \0 for fn_format + *((char*)net->read_pos + packet_len) = 0; + fn_format(fname, (char*) net->read_pos + 1, "", "", 4); // this is needed to make replicate-ignore-db if (!strcmp(fname,"/dev/null")) goto end; - if ((fd = my_open(fname, O_RDONLY, MYF(MY_WME))) < 0) + if ((fd = my_open(fname, O_RDONLY, MYF(0))) < 0) { - errmsg = "Failed on my_open()"; + errmsg = "on open of file"; goto err; } - while ((bytes = (int) my_read(fd, (byte*) buf, IO_SIZE, - MYF(MY_WME))) > 0) + while ((bytes = (int) my_read(fd, (byte*) buf, IO_SIZE, MYF(0))) > 0) { if (my_net_write(net, buf, bytes)) { - errmsg = "Failed on my_net_write()"; + errmsg = "while writing data to client"; goto err; } } @@ -119,18 +135,18 @@ static int send_file(THD *thd) if (my_net_write(net, "", 0) || net_flush(net) || (my_net_read(net) == packet_error)) { - errmsg = "failed negotiating file transfer close"; + errmsg = "while negotiating file transfer close"; goto err; } error = 0; err: - thd->net.timeout = old_timeout; - if(fd >= 0) - (void) my_close(fd, MYF(MY_WME)); + thd->net.read_timeout = old_timeout; + if (fd >= 0) + (void) my_close(fd, MYF(0)); if (errmsg) { - sql_print_error("failed in send_file() : %s", errmsg); + sql_print_error("Failed in send_file() %s", errmsg); DBUG_PRINT("error", (errmsg)); } DBUG_RETURN(error); @@ -138,61 +154,56 @@ static int send_file(THD *thd) File open_binlog(IO_CACHE *log, const char *log_file_name, - const char **errmsg) + const char **errmsg) { File file; - char magic[4]; + DBUG_ENTER("open_binlog"); + if ((file = my_open(log_file_name, O_RDONLY | O_BINARY, MYF(MY_WME))) < 0 || init_io_cache(log, file, IO_SIZE*2, READ_CACHE, 0, 0, - MYF(MY_WME))) + MYF(MY_WME | MY_DONT_CHECK_FILESIZE))) { - *errmsg = "Could not open log file"; // This will not be sent + *errmsg = "Could not open log file"; // This will not be sent goto err; } - - if (my_b_read(log, (byte*) magic, sizeof(magic))) - { - *errmsg = "I/O error reading binlog magic number"; - goto err; - } - if (memcmp(magic, BINLOG_MAGIC, 4)) - { - *errmsg = "Binlog has bad magic number, fire your magician"; + if (check_binlog_magic(log,errmsg)) goto err; - } - return file; + DBUG_RETURN(file); err: - if (file > 0) + if (file >= 0) + { my_close(file,MYF(0)); - end_io_cache(log); - return -1; + end_io_cache(log); + } + DBUG_RETURN(-1); } void adjust_linfo_offsets(my_off_t purge_offset) { THD *tmp; - + pthread_mutex_lock(&LOCK_thread_count); I_List_iterator<THD> it(threads); - - while((tmp=it++)) - { - LOG_INFO* linfo; - if((linfo = tmp->current_linfo)) - { - pthread_mutex_lock(&linfo->lock); - // no big deal if we just started reading the log - // nothing to adjust - if(linfo->index_file_offset < purge_offset) - linfo->fatal = (linfo->index_file_offset != 0); - else - linfo->index_file_offset -= purge_offset; - pthread_mutex_unlock(&linfo->lock); - } - } + while ((tmp=it++)) + { + LOG_INFO* linfo; + if ((linfo = tmp->current_linfo)) + { + pthread_mutex_lock(&linfo->lock); + /* index file offset can be less that purge offset + only if we just started reading the index file. In that case + we have nothing to adjust + */ + if (linfo->index_file_offset < purge_offset) + linfo->fatal = (linfo->index_file_offset != 0); + else + linfo->index_file_offset -= purge_offset; + pthread_mutex_unlock(&linfo->lock); + } + } pthread_mutex_unlock(&LOCK_thread_count); } @@ -202,21 +213,21 @@ bool log_in_use(const char* log_name) int log_name_len = strlen(log_name) + 1; THD *tmp; bool result = 0; - + pthread_mutex_lock(&LOCK_thread_count); I_List_iterator<THD> it(threads); - - while((tmp=it++)) + + while ((tmp=it++)) + { + LOG_INFO* linfo; + if ((linfo = tmp->current_linfo)) { - LOG_INFO* linfo; - if((linfo = tmp->current_linfo)) - { - pthread_mutex_lock(&linfo->lock); - result = !memcmp(log_name, linfo->log_file_name, log_name_len); - pthread_mutex_unlock(&linfo->lock); - if(result) break; - } - } + pthread_mutex_lock(&linfo->lock); + result = !memcmp(log_name, linfo->log_file_name, log_name_len); + pthread_mutex_unlock(&linfo->lock); + if (result) break; + } + } pthread_mutex_unlock(&LOCK_thread_count); return result; @@ -226,35 +237,35 @@ bool log_in_use(const char* log_name) int purge_master_logs(THD* thd, const char* to_log) { char search_file_name[FN_REFLEN]; + const char* errmsg = 0; + mysql_bin_log.make_log_name(search_file_name, to_log); int res = mysql_bin_log.purge_logs(thd, search_file_name); - const char* errmsg = 0; - switch(res) - { - case 0: break; - case LOG_INFO_EOF: errmsg = "Target log not found in binlog index"; break; - case LOG_INFO_IO: errmsg = "I/O error reading log index file"; break; - case LOG_INFO_INVALID: errmsg = "Server configuration does not permit \ + + switch(res) { + case 0: break; + case LOG_INFO_EOF: errmsg = "Target log not found in binlog index"; break; + case LOG_INFO_IO: errmsg = "I/O error reading log index file"; break; + case LOG_INFO_INVALID: errmsg = "Server configuration does not permit \ binlog purge"; break; - case LOG_INFO_SEEK: errmsg = "Failed on fseek()"; break; - case LOG_INFO_PURGE_NO_ROTATE: errmsg = "Cannot purge unrotatable log"; - break; - case LOG_INFO_MEM: errmsg = "Out of memory"; break; - case LOG_INFO_FATAL: errmsg = "Fatal error during purge"; break; - case LOG_INFO_IN_USE: errmsg = "A purgeable log is in use, will not purge"; - break; - default: - errmsg = "Unknown error during purge"; break; - } - - if(errmsg) - { - send_error(&thd->net, 0, errmsg); - return 1; - } + case LOG_INFO_SEEK: errmsg = "Failed on fseek()"; break; + case LOG_INFO_PURGE_NO_ROTATE: errmsg = "Cannot purge unrotatable log"; + break; + case LOG_INFO_MEM: errmsg = "Out of memory"; break; + case LOG_INFO_FATAL: errmsg = "Fatal error during purge"; break; + case LOG_INFO_IN_USE: errmsg = "A purgeable log is in use, will not purge"; + break; + default: errmsg = "Unknown error during purge"; break; + } + + if (errmsg) + { + send_error(&thd->net, 0, errmsg); + return 1; + } else send_ok(&thd->net); - + return 0; } @@ -272,7 +283,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) NET* net = &thd->net; #ifndef DBUG_OFF int left_events = max_binlog_dump_events; -#endif +#endif DBUG_ENTER("mysql_binlog_send"); bzero((char*) &log,sizeof(log)); @@ -282,25 +293,25 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) errmsg = "Master failed COM_BINLOG_DUMP to test if slave can recover"; goto err; } -#endif +#endif - if(!mysql_bin_log.is_open()) + if (!mysql_bin_log.is_open()) { errmsg = "Binary log is not open"; goto err; } - if(!server_id_supplied) - { - errmsg = "Misconfigured master - server id was not set"; - goto err; - } - + if (!server_id_supplied) + { + errmsg = "Misconfigured master - server id was not set"; + goto err; + } + if (log_ident[0]) mysql_bin_log.make_log_name(search_file_name, log_ident); else search_file_name[0] = 0; - + linfo.index_file_offset = 0; thd->current_linfo = &linfo; @@ -315,20 +326,21 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) if (pos < 4) { - errmsg = "Client requested master to start repliction from impossible position.\n"; + errmsg= "Client requested master to start replication from \ +impossible position"; goto err; } - + my_b_seek(&log, pos); // Seek will done on next read packet->length(0); - packet->append("\0", 1); // we need to start a packet with something other than 255 // to distiquish it from error + packet->append("\0", 1); - // tell the client log name with a fake rotate_event // if we are at the start of the log - if(pos == 4) + if (pos == 4) { + // tell the client log name with a fake rotate_event if (fake_rotate_event(net, packet, log_file_name, &errmsg)) goto err; packet->length(0); @@ -338,17 +350,17 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) while (!net->error && net->vio != 0 && !thd->killed) { pthread_mutex_t *log_lock = mysql_bin_log.get_log_lock(); - + while (!(error = Log_event::read_log_event(&log, packet, log_lock))) { #ifndef DBUG_OFF - if(max_binlog_dump_events && !left_events--) + if (max_binlog_dump_events && !left_events--) { net_flush(net); errmsg = "Debugging binlog dump abort"; goto err; } -#endif +#endif if (my_net_write(net, (char*)packet->ptr(), packet->length()) ) { errmsg = "Failed on my_net_write()"; @@ -358,7 +370,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) (*packet)[LOG_EVENT_OFFSET+1] )); if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) { - if(send_file(thd)) + if (send_file(thd)) { errmsg = "failed in send_file()"; goto err; @@ -367,15 +379,15 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) packet->length(0); packet->append("\0",1); } - + // TODO: now that we are logging the offset, check to make sure + // the recorded offset and the actual match if (error != LOG_READ_EOF) { - switch(error) - { - case LOG_READ_BOGUS: + switch(error) { + case LOG_READ_BOGUS: errmsg = "bogus data in log event"; break; - case LOG_READ_TOO_LARGE: + case LOG_READ_TOO_LARGE: errmsg = "log event entry exceeded max_allowed_packet -\ increase max_allowed_packet on master"; break; @@ -395,12 +407,12 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) goto err; } - if(!(flags & BINLOG_DUMP_NON_BLOCK) && + if (!(flags & BINLOG_DUMP_NON_BLOCK) && mysql_bin_log.is_active(log_file_name)) + { // block until there is more data in the log // unless non-blocking mode requested - { - if(net_flush(net)) + if (net_flush(net)) { errmsg = "failed on net_flush()"; goto err; @@ -412,64 +424,56 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) // to signal us { log.error=0; - - // tell the kill thread how to wake us up - thd->mysys_var->current_mutex = log_lock; - thd->mysys_var->current_cond = &COND_binlog_update; - const char* proc_info = thd->proc_info; - thd->proc_info = "Slave connection: waiting for binlog update"; - bool read_packet = 0, fatal_error = 0; #ifndef DBUG_OFF - if(max_binlog_dump_events && !left_events--) + if (max_binlog_dump_events && !left_events--) { net_flush(net); errmsg = "Debugging binlog dump abort"; goto err; } -#endif +#endif // no one will update the log while we are reading // now, but we'll be quick and just read one record pthread_mutex_lock(log_lock); - switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*) 0)) + switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*)0)) { case 0: + pthread_mutex_unlock(log_lock); read_packet = 1; // we read successfully, so we'll need to send it to the // slave break; case LOG_READ_EOF: - DBUG_PRINT("wait",("waiting for data on binary log")); + DBUG_PRINT("wait",("waiting for data in binary log")); + // wait_for_update unlocks the log lock - needed to avoid race if (!thd->killed) - pthread_cond_wait(&COND_binlog_update, log_lock); + mysql_bin_log.wait_for_update(thd); + else + pthread_mutex_unlock(log_lock); + DBUG_PRINT("wait",("binary log received update")); break; default: + pthread_mutex_unlock(log_lock); fatal_error = 1; break; } - pthread_mutex_unlock(log_lock); - - pthread_mutex_lock(&thd->mysys_var->mutex); - thd->mysys_var->current_mutex= 0; - thd->mysys_var->current_cond= 0; - thd->proc_info= proc_info; - pthread_mutex_unlock(&thd->mysys_var->mutex); - - if(read_packet) + + if (read_packet) { thd->proc_info = "sending update to slave"; - if(my_net_write(net, (char*)packet->ptr(), packet->length()) ) + if (my_net_write(net, (char*)packet->ptr(), packet->length()) ) { errmsg = "Failed on my_net_write()"; goto err; } - if((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) + if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT) { - if(send_file(thd)) + if (send_file(thd)) { errmsg = "failed in send_file()"; goto err; @@ -481,7 +485,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) // we hit EOF pretty quick } - if(fatal_error) + if (fatal_error) { errmsg = "error reading log entry"; goto err; @@ -494,8 +498,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) bool loop_breaker = 0; // need this to break out of the for loop from switch thd->proc_info = "switching to next log"; - switch(mysql_bin_log.find_next_log(&linfo)) - { + switch (mysql_bin_log.find_next_log(&linfo)) { case LOG_INFO_EOF: loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK); break; @@ -506,12 +509,12 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) goto err; } - if(loop_breaker) + if (loop_breaker) break; end_io_cache(&log); (void) my_close(file, MYF(MY_WME)); - + // fake Rotate_log event just in case it did not make it to the log // otherwise the slave make get confused about the offset if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 || @@ -525,22 +528,25 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) end_io_cache(&log); (void)my_close(file, MYF(MY_WME)); - + send_eof(&thd->net); thd->proc_info = "waiting to finalize termination"; pthread_mutex_lock(&LOCK_thread_count); thd->current_linfo = 0; pthread_mutex_unlock(&LOCK_thread_count); DBUG_VOID_RETURN; + err: thd->proc_info = "waiting to finalize termination"; end_io_cache(&log); pthread_mutex_lock(&LOCK_thread_count); - // exclude iteration through thread list - // this is needed for purge_logs() - it will iterate through - // thread list and update thd->current_linfo->index_file_offset - // this mutex will make sure that it never tried to update our linfo - // after we return from this stack frame + /* + Exclude iteration through thread list + this is needed for purge_logs() - it will iterate through + thread list and update thd->current_linfo->index_file_offset + this mutex will make sure that it never tried to update our linfo + after we return from this stack frame + */ thd->current_linfo = 0; pthread_mutex_unlock(&LOCK_thread_count); if (file >= 0) @@ -549,126 +555,115 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) DBUG_VOID_RETURN; } -int start_slave(THD* thd , bool net_report) +int start_slave(THD* thd , MASTER_INFO* mi, bool net_report) { - if(!thd) thd = current_thd; - NET* net = &thd->net; int slave_errno = 0; - if (check_access(thd, PROCESS_ACL, any_db)) + if (!thd) thd = current_thd; + NET* net = &thd->net; + int thread_mask; + + if (check_access(thd, SUPER_ACL, any_db)) return 1; - pthread_mutex_lock(&LOCK_slave); - if(!slave_running) - { - if(init_master_info(&glob_mi)) - slave_errno = ER_MASTER_INFO; - else if(server_id_supplied && *glob_mi.host) - { - pthread_t hThread; - if(pthread_create(&hThread, &connection_attrib, handle_slave, 0)) - { - slave_errno = ER_SLAVE_THREAD; - } - while(!slave_running) // slave might already be running by now - pthread_cond_wait(&COND_slave_start, &LOCK_slave); - } - else - slave_errno = ER_BAD_SLAVE; - } + lock_slave_threads(mi); // this allows us to cleanly read slave_running + init_thread_mask(&thread_mask,mi,1 /* inverse */); + if (thd->lex.slave_thd_opt) + thread_mask &= thd->lex.slave_thd_opt; + if (thread_mask) + { + if (server_id_supplied && (!mi->inited || (mi->inited && *mi->host))) + slave_errno = start_slave_threads(0 /*no mutex */, + 1 /* wait for start */, + mi, + master_info_file,relay_log_info_file, + thread_mask); + else + slave_errno = ER_BAD_SLAVE; + } else slave_errno = ER_SLAVE_MUST_STOP; - - pthread_mutex_unlock(&LOCK_slave); - if(slave_errno) - { - if(net_report) send_error(net, slave_errno); - return 1; - } - else if(net_report) + + unlock_slave_threads(mi); + + if (slave_errno) + { + if (net_report) + send_error(net, slave_errno); + return 1; + } + else if (net_report) send_ok(net); return 0; } -int stop_slave(THD* thd, bool net_report ) +int stop_slave(THD* thd, MASTER_INFO* mi, bool net_report ) { - if(!thd) thd = current_thd; - NET* net = &thd->net; int slave_errno = 0; - - if (check_access(thd, PROCESS_ACL, any_db)) + if (!thd) thd = current_thd; + NET* net = &thd->net; + + if (check_access(thd, SUPER_ACL, any_db)) return 1; + thd->proc_info = "Killing slave"; + int thread_mask; + lock_slave_threads(mi); + init_thread_mask(&thread_mask,mi,0 /* not inverse*/); + if (thd->lex.slave_thd_opt) + thread_mask &= thd->lex.slave_thd_opt; + slave_errno = (thread_mask) ? + terminate_slave_threads(mi,thread_mask, + 1 /*skip lock */) : ER_SLAVE_NOT_RUNNING; + unlock_slave_threads(mi); + thd->proc_info = 0; - pthread_mutex_lock(&LOCK_slave); - if (slave_running) + if (slave_errno) { - abort_slave = 1; - KICK_SLAVE; - // do not abort the slave in the middle of a query, so we do not set - // thd->killed for the slave thread - thd->proc_info = "waiting for slave to die"; - while(slave_running) - { - /* there is a small chance that slave thread might miss the first - alarm. To protect againts it, resend the signal until it reacts - */ - - struct timespec abstime; -#ifdef HAVE_TIMESPEC_TS_SEC - abstime.ts_sec=time(NULL)+2; - abstime.ts_nsec=0; -#elif defined(__WIN__) - abstime.tv_sec=time((time_t*) 0)+2; - abstime.tv_nsec=0; -#else - struct timeval tv; - gettimeofday(&tv,0); - abstime.tv_sec=tv.tv_sec+2; - abstime.tv_nsec=tv.tv_usec*1000; -#endif - pthread_cond_timedwait(&COND_slave_stopped, &LOCK_slave, &abstime); - if (slave_running) - KICK_SLAVE; - } + if (net_report) + send_error(net, slave_errno); + return 1; } - else - slave_errno = ER_SLAVE_NOT_RUNNING; - - pthread_mutex_unlock(&LOCK_slave); - thd->proc_info = 0; - - if(slave_errno) - { - if(net_report) send_error(net, slave_errno); - return 1; - } - else if(net_report) + else if (net_report) send_ok(net); return 0; } -void reset_slave() +int reset_slave(MASTER_INFO* mi) { MY_STAT stat_area; char fname[FN_REFLEN]; - bool slave_was_running ; - - pthread_mutex_lock(&LOCK_slave); - if((slave_was_running = slave_running)) - { - pthread_mutex_unlock(&LOCK_slave); - stop_slave(0,0); - } - else - pthread_mutex_unlock(&LOCK_slave); + int restart_thread_mask = 0,error=0; + const char* errmsg=0; + DBUG_ENTER("reset_slave"); + + lock_slave_threads(mi); + init_thread_mask(&restart_thread_mask,mi,0 /* not inverse */); + if ((error=terminate_slave_threads(mi,restart_thread_mask,1 /*skip lock*/)) + || (error=purge_relay_logs(&mi->rli,1 /*just reset*/,&errmsg))) + goto err; - end_master_info(&glob_mi); - fn_format(fname, master_info_file, mysql_data_home, "", 4+16+32); - if(my_stat(fname, &stat_area, MYF(0))) - if(my_delete(fname, MYF(MY_WME))) - return; - if(slave_was_running) - start_slave(0,0); + end_master_info(mi); + fn_format(fname, master_info_file, mysql_data_home, "", 4+32); + if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME))) + { + error=1; + goto err; + } + fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32); + if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME))) + { + error=1; + goto err; + } + if (restart_thread_mask) + error=start_slave_threads(0 /* mutex not needed */, + 1 /* wait for start*/, + mi,master_info_file,relay_log_info_file, + restart_thread_mask); + // TODO: fix error messages so they get to the client +err: + unlock_slave_threads(mi); + DBUG_RETURN(error); } void kill_zombie_dump_threads(uint32 slave_server_id) @@ -677,121 +672,273 @@ void kill_zombie_dump_threads(uint32 slave_server_id) I_List_iterator<THD> it(threads); THD *tmp; - while((tmp=it++)) + while ((tmp=it++)) + { + if (tmp->command == COM_BINLOG_DUMP && + tmp->server_id == slave_server_id) { - if(tmp->command == COM_BINLOG_DUMP && - tmp->server_id == slave_server_id) - { - // here we do not call kill_one_thread() - // it will be slow because it will iterate through the list - // again. Plus it double-locks LOCK_thread_count, which - // make safe_mutex complain and abort - // so we just to our own thread murder - - thr_alarm_kill(tmp->real_id); - tmp->killed = 1; - tmp->mysys_var->abort = 1; - pthread_mutex_lock(&tmp->mysys_var->mutex); - if(tmp->mysys_var->current_cond) - { - pthread_mutex_lock(tmp->mysys_var->current_mutex); - pthread_cond_broadcast(tmp->mysys_var->current_cond); - pthread_mutex_unlock(tmp->mysys_var->current_mutex); - } - pthread_mutex_unlock(&tmp->mysys_var->mutex); - } - } - + /* + Here we do not call kill_one_thread() as + it will be slow because it will iterate through the list + again. Plus it double-locks LOCK_tread_count, which + make safe_mutex complain and abort. + We just to do kill the thread ourselves. + */ + tmp->awake(1/*prepare to die*/); + } + } pthread_mutex_unlock(&LOCK_thread_count); } -int change_master(THD* thd) + +int change_master(THD* thd, MASTER_INFO* mi) { - bool slave_was_running; + int error=0,restart_thread_mask; + const char* errmsg=0; + bool need_relay_log_purge=1; + DBUG_ENTER("change_master"); + // kill slave thread - pthread_mutex_lock(&LOCK_slave); - if((slave_was_running = slave_running)) - { - abort_slave = 1; - KICK_SLAVE; - thd->proc_info = "waiting for slave to die"; - while(slave_running) - pthread_cond_wait(&COND_slave_stopped, &LOCK_slave); // wait until done - } - pthread_mutex_unlock(&LOCK_slave); + lock_slave_threads(mi); + init_thread_mask(&restart_thread_mask,mi,0 /*not inverse*/); + if (restart_thread_mask && + (error=terminate_slave_threads(mi, + restart_thread_mask, + 1 /*skip lock*/))) + { + send_error(&thd->net,error); + unlock_slave_threads(mi); + DBUG_RETURN(1); + } thd->proc_info = "changing master"; LEX_MASTER_INFO* lex_mi = &thd->lex.mi; + // TODO: see if needs re-write + if (init_master_info(mi,master_info_file,relay_log_info_file)) + { + send_error(&thd->net, 0, "Could not initialize master info"); + unlock_slave_threads(mi); + DBUG_RETURN(1); + } - if(init_master_info(&glob_mi)) - { - send_error(&thd->net, 0, "Could not initialize master info"); - return 1; - } - - pthread_mutex_lock(&glob_mi.lock); - if((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos) - { - // if we change host or port, we must reset the postion - glob_mi.log_file_name[0] = 0; - glob_mi.pos = 4; // skip magic number - glob_mi.pending = 0; - } + /* data lock not needed since we have already stopped the running threads, + and we have the hold on the run locks which will keep all threads that + could possibly modify the data structures from running + */ + if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos) + { + // if we change host or port, we must reset the postion + mi->master_log_name[0] = 0; + mi->master_log_pos= BIN_LOG_HEADER_SIZE; + mi->rli.pending = 0; + } - if(lex_mi->log_file_name) - strmake(glob_mi.log_file_name, lex_mi->log_file_name, - sizeof(glob_mi.log_file_name)); - if(lex_mi->pos) + if (lex_mi->log_file_name) + strmake(mi->master_log_name, lex_mi->log_file_name, + sizeof(mi->master_log_name)); + if (lex_mi->pos) { - glob_mi.pos = lex_mi->pos; - glob_mi.pending = 0; + mi->master_log_pos= lex_mi->pos; + mi->rli.pending = 0; } - - if(lex_mi->host) + DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos)); + + if (lex_mi->host) + strmake(mi->host, lex_mi->host, sizeof(mi->host)); + if (lex_mi->user) + strmake(mi->user, lex_mi->user, sizeof(mi->user)); + if (lex_mi->password) + strmake(mi->password, lex_mi->password, sizeof(mi->password)); + if (lex_mi->port) + mi->port = lex_mi->port; + if (lex_mi->connect_retry) + mi->connect_retry = lex_mi->connect_retry; + + if (lex_mi->relay_log_name) + { + need_relay_log_purge = 0; + mi->rli.skip_log_purge=1; + strmake(mi->rli.relay_log_name,lex_mi->relay_log_name, + sizeof(mi->rli.relay_log_name)-1); + } + + if (lex_mi->relay_log_pos) + { + need_relay_log_purge=0; + mi->rli.relay_log_pos=lex_mi->relay_log_pos; + } + + flush_master_info(mi); + if (need_relay_log_purge) + { + mi->rli.skip_log_purge=0; + thd->proc_info="purging old relay logs"; + if (purge_relay_logs(&mi->rli,0 /* not only reset, but also reinit*/, + &errmsg)) + { + net_printf(&thd->net, 0, "Failed purging old relay logs: %s",errmsg); + DBUG_RETURN(1); + } + } + else + { + const char* msg; + if (init_relay_log_pos(&mi->rli,0/*log already inited*/, + 0 /*pos already inited*/, + 0 /*no data lock*/, + &msg)) { - strmake(glob_mi.host, lex_mi->host, sizeof(glob_mi.host)); + //Sasha: note that I had to change net_printf() to make this work + net_printf(&thd->net,0,"Failed initializing relay log position: %s",msg); + unlock_slave_threads(mi); + DBUG_RETURN(1); } - if(lex_mi->user) - strmake(glob_mi.user, lex_mi->user, sizeof(glob_mi.user)); - if(lex_mi->password) - strmake(glob_mi.password, lex_mi->password, sizeof(glob_mi.password)); - if(lex_mi->port) - glob_mi.port = lex_mi->port; - if(lex_mi->connect_retry) - glob_mi.connect_retry = lex_mi->connect_retry; - - flush_master_info(&glob_mi); - pthread_mutex_unlock(&glob_mi.lock); + } + mi->rli.master_log_pos = mi->master_log_pos; + DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos)); + strmake(mi->rli.master_log_name,mi->master_log_name, + sizeof(mi->rli.master_log_name)-1); + if (!mi->rli.master_log_name[0]) // uninitialized case + mi->rli.master_log_pos=0; + + pthread_mutex_lock(&mi->rli.data_lock); + mi->rli.abort_pos_wait = 1; + pthread_cond_broadcast(&mi->data_cond); + pthread_mutex_unlock(&mi->rli.data_lock); + thd->proc_info = "starting slave"; - if(slave_was_running) - start_slave(0,0); + if (restart_thread_mask) + error=start_slave_threads(0 /* mutex not needed*/, + 1 /* wait for start*/, + mi,master_info_file,relay_log_info_file, + restart_thread_mask); + unlock_slave_threads(mi); thd->proc_info = 0; - - send_ok(&thd->net); - return 0; + if (error) + send_error(&thd->net,error); + else + send_ok(&thd->net); + DBUG_RETURN(0); } -void reset_master() +int reset_master(THD* thd) { - if(!mysql_bin_log.is_open()) + if (!mysql_bin_log.is_open()) { my_error(ER_FLUSH_MASTER_BINLOG_CLOSED, MYF(ME_BELL+ME_WAITTANG)); - return; + return 1; } + return mysql_bin_log.reset_logs(thd); +} - LOG_INFO linfo; - if (mysql_bin_log.find_first_log(&linfo, "")) - return; +int cmp_master_pos(const char* log_file_name1, ulonglong log_pos1, + const char* log_file_name2, ulonglong log_pos2) +{ + int res; + if ((res = strcmp(log_file_name1, log_file_name2))) + return res; + if (log_pos1 > log_pos2) + return 1; + else if (log_pos1 == log_pos2) + return 0; + return -1; +} - for(;;) +int show_binlog_events(THD* thd) +{ + DBUG_ENTER("show_binlog_events"); + List<Item> field_list; + const char* errmsg = 0; + IO_CACHE log; + File file = -1; + + Log_event::init_show_field_list(&field_list); + if (send_fields(thd, field_list, 1)) + DBUG_RETURN(-1); + + if (mysql_bin_log.is_open()) { - my_delete(linfo.log_file_name, MYF(MY_WME)); - if (mysql_bin_log.find_next_log(&linfo)) - break; + LOG_INFO linfo; + char search_file_name[FN_REFLEN]; + LEX_MASTER_INFO* lex_mi = &thd->lex.mi; + uint event_count, limit_start, limit_end; + const char* log_file_name = lex_mi->log_file_name; + Log_event* ev; + my_off_t pos = lex_mi->pos; + + limit_start = thd->lex.select->offset_limit; + limit_end = thd->lex.select->select_limit + limit_start; + + if (log_file_name) + mysql_bin_log.make_log_name(search_file_name, log_file_name); + else + search_file_name[0] = 0; + + linfo.index_file_offset = 0; + thd->current_linfo = &linfo; + + if (mysql_bin_log.find_first_log(&linfo, search_file_name)) + { + errmsg = "Could not find target log"; + goto err; + } + + if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0) + goto err; + + if (pos < 4) + { + errmsg = "Invalid log position"; + goto err; + } + + pthread_mutex_lock(mysql_bin_log.get_log_lock()); + my_b_seek(&log, pos); + + for (event_count = 0; + (ev = Log_event::read_log_event(&log,(pthread_mutex_t*)0,0)); ) + { + if (event_count >= limit_start && + ev->net_send(thd, linfo.log_file_name, pos)) + { + errmsg = "Net error"; + delete ev; + pthread_mutex_unlock(mysql_bin_log.get_log_lock()); + goto err; + } + + pos = my_b_tell(&log); + delete ev; + + if (++event_count >= limit_end) + break; + } + + if (event_count < limit_end && log.error) + { + errmsg = "Wrong offset or I/O error"; + pthread_mutex_unlock(mysql_bin_log.get_log_lock()); + goto err; + } + + pthread_mutex_unlock(mysql_bin_log.get_log_lock()); } - mysql_bin_log.close(1); // exiting close - my_delete(mysql_bin_log.get_index_fname(), MYF(MY_WME)); - mysql_bin_log.open(opt_bin_logname,LOG_BIN); +err: + if (file >= 0) + { + end_io_cache(&log); + (void) my_close(file, MYF(MY_WME)); + } + + if (errmsg) + { + net_printf(&thd->net, ER_ERROR_WHEN_EXECUTING_COMMAND, + "SHOW BINLOG EVENTS", errmsg); + DBUG_RETURN(1); + } + + send_eof(&thd->net); + DBUG_RETURN(0); } int show_binlog_info(THD* thd) @@ -803,36 +950,37 @@ int show_binlog_info(THD* thd) field_list.push_back(new Item_empty_string("Binlog_do_db",20)); field_list.push_back(new Item_empty_string("Binlog_ignore_db",20)); - if(send_fields(thd, field_list, 1)) + if (send_fields(thd, field_list, 1)) DBUG_RETURN(-1); String* packet = &thd->packet; packet->length(0); - if(mysql_bin_log.is_open()) - { - LOG_INFO li; - mysql_bin_log.get_current_log(&li); - int dir_len = dirname_length(li.log_file_name); - net_store_data(packet, li.log_file_name + dir_len); - net_store_data(packet, (longlong)li.pos); - net_store_data(packet, &binlog_do_db); - net_store_data(packet, &binlog_ignore_db); - } + if (mysql_bin_log.is_open()) + { + LOG_INFO li; + mysql_bin_log.get_current_log(&li); + int dir_len = dirname_length(li.log_file_name); + net_store_data(packet, li.log_file_name + dir_len); + net_store_data(packet, (longlong)li.pos); + net_store_data(packet, &binlog_do_db); + net_store_data(packet, &binlog_ignore_db); + } else - { - net_store_null(packet); - net_store_null(packet); - net_store_null(packet); - net_store_null(packet); - } + { + net_store_null(packet); + net_store_null(packet); + net_store_null(packet); + net_store_null(packet); + } - if(my_net_write(&thd->net, (char*)thd->packet.ptr(), packet->length())) + if (my_net_write(&thd->net, (char*)thd->packet.ptr(), packet->length())) DBUG_RETURN(-1); send_eof(&thd->net); DBUG_RETURN(0); } + int show_binlogs(THD* thd) { const char* errmsg = 0; @@ -843,20 +991,20 @@ int show_binlogs(THD* thd) String* packet = &thd->packet; IO_CACHE io_cache; uint length; - - if(!mysql_bin_log.is_open()) + + if (!mysql_bin_log.is_open()) { errmsg = "binlog is not open"; goto err; } field_list.push_back(new Item_empty_string("Log_name", 128)); - if(send_fields(thd, field_list, 1)) + if (send_fields(thd, field_list, 1)) { sql_print_error("Failed in send_fields"); return 1; } - + mysql_bin_log.lock_index(); index_file = mysql_bin_log.get_index_file(); if (index_file < 0) @@ -876,7 +1024,7 @@ int show_binlogs(THD* thd) int dir_len = dirname_length(fname); packet->length(0); net_store_data(packet, fname + dir_len, length-dir_len); - if(my_net_write(net, (char*) packet->ptr(), packet->length())) + if (my_net_write(net, (char*) packet->ptr(), packet->length())) { sql_print_error("Failed in my_net_write"); end_io_cache(&io_cache); @@ -884,10 +1032,10 @@ int show_binlogs(THD* thd) return 1; } } - + mysql_bin_log.unlock_index(); end_io_cache(&io_cache); - send_eof(net); + send_eof(net); return 0; err2: @@ -898,5 +1046,34 @@ err: return 1; } - - +int log_loaded_block(IO_CACHE* file) +{ + LOAD_FILE_INFO* lf_info; + uint block_len ; + + /* file->request_pos contains position where we started last read */ + char* buffer = (char*) file->request_pos; + if (!(block_len = (char*) file->read_end - (char*) buffer)) + return 0; + lf_info = (LOAD_FILE_INFO*) file->arg; + if (lf_info->last_pos_in_file != HA_POS_ERROR && + lf_info->last_pos_in_file >= file->pos_in_file) + return 0; + lf_info->last_pos_in_file = file->pos_in_file; + if (lf_info->wrote_create_file) + { + Append_block_log_event a(lf_info->thd, buffer, block_len); + mysql_bin_log.write(&a); + } + else + { + Create_file_log_event c(lf_info->thd,lf_info->ex,lf_info->db, + lf_info->table_name, *lf_info->fields, + lf_info->handle_dup, buffer, + block_len); + mysql_bin_log.write(&c); + lf_info->wrote_create_file = 1; + DBUG_SYNC_POINT("debug_lock.created_file_event",10); + } + return 0; +} |