1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

skip updates with the same server id as self

Browse Source 
kill zombie COM_BINLOG_DUMP with the same server id on connect


sql/mysql_priv.h:
  made kill_one_thread global scope
sql/slave.cc:
  skip updates with the same server id as self
  fixed compiler warning
sql/sql_parse.cc:
  COM_BINLOG_DUMP is followed by a simulation of COM_QUIT for proper thread clean up
sql/sql_repl.cc:
  kill zombie COM_BINLOG_DUMP with the same server id on connect,
  more verbose binlog_send process status
This commit is contained in:
unknown 2000-10-02 17:59:12 -06:00
parent a9f2a439b4
commit f97de74c30
4 changed files with 69 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
sql/mysql_priv.h
View File

@ -41,6 +41,7 @@ char *sql_strdup(const char *str);
char *sql_strmake(const char *str,uint len);
gptr sql_memdup(const void * ptr,unsigned size);
void sql_element_free(void *ptr);
void kill_one_thread(THD *thd, ulong id);
#define x_free(A) { my_free((gptr) (A),MYF(MY_WME | MY_FAE | MY_ALLOW_ZERO_PTR)); }
#define safeFree(x) { if(x) { my_free((gptr) x,MYF(0)); x = NULL; } }

38
sql/slave.cc
View File

@ -38,6 +38,7 @@ MASTER_INFO glob_mi;
extern bool opt_log_slave_updates ;
static inline void skip_load_data_infile(NET* net);
static inline bool slave_killed(THD* thd);
static int init_slave_thread(THD* thd);
int init_master_info(MASTER_INFO* mi);
@ -53,6 +54,15 @@ static inline bool slave_killed(THD* thd)
return abort_slave || abort_loop || thd->killed;
}
static inline void skip_load_data_infile(NET* net)
{
(void)my_net_write(net, "\xfb/dev/null", 10);
(void)net_flush(net);
(void)my_net_read(net); // discard response
send_ok(net); // the master expects it
}
int db_ok(const char* db, I_List<i_string> &do_list,
I_List<i_string> &ignore_list )
{
@ -553,9 +563,26 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
{
Log_event * ev = Log_event::read_log_event((const char*)net->read_pos + 1,
event_len);
if (ev)
{
switch(ev->get_type_code())
int type_code = ev->get_type_code();
if(ev->server_id == ::server_id)
{
if(type_code == LOAD_EVENT)
skip_load_data_infile(net);
mi->inc_pos(event_len);
flush_master_info(mi);
delete ev;
return 0; // avoid infinite update loops
}
thd->server_id = ev->server_id; // use the original server id for logging
thd->set_time(); // time the query
ev->when = time(NULL);
switch(type_code)
{
case QUERY_EVENT:
{
@ -706,10 +733,7 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
else // we will just ask the master to send us /dev/null if we do not want to
// load the data :-)
{
(void)my_net_write(net, "\xfb/dev/null", 10);
(void)net_flush(net);
(void)my_net_read(net); // discard response
send_ok(net); // the master expects it
skip_load_data_infile(net);
}
thd->net.vio = 0;
@ -799,14 +823,14 @@ pthread_handler_decl(handle_slave,arg __attribute__((unused)))
if(!server_id)
{
sql_print_error("Server id not set, will not start slave");
pthread_exit(1);
pthread_exit((void*)1);
}
pthread_mutex_lock(&LOCK_slave);
if(slave_running)
{
pthread_mutex_unlock(&LOCK_slave);
pthread_exit(1); // safety just in case
pthread_exit((void*)1); // safety just in case
}
slave_running = 1;
abort_slave = 0;

6
sql/sql_parse.cc
View File

@ -39,7 +39,6 @@ static bool check_dup(THD *thd,const char *db,const char *name,
TABLE_LIST *tables);
static void mysql_init_query(THD *thd);
static void remove_escape(char *name);
static void kill_one_thread(THD *thd, ulong thread);
static void refresh_status(void);
const char *any_db="*any*"; // Special symbol for check_access
@ -712,6 +711,9 @@ bool do_command(THD *thd)
thd->server_id = slave_server_id;
pthread_mutex_unlock(&LOCK_server_id);
mysql_binlog_send(thd, strdup(packet + 11), pos, flags);
// fake COM_QUIT -- if we get here, the thread needs to terminate
error = TRUE;
net->error = 0;
break;
}
case COM_REFRESH:
@ -2451,7 +2453,7 @@ bool reload_acl_and_cache(THD *thd, uint options, TABLE_LIST *tables)
}
static void kill_one_thread(THD *thd, ulong id)
void kill_one_thread(THD *thd, ulong id)
{
VOID(pthread_mutex_lock(&LOCK_thread_count)); // For unlink from list
I_List_iterator<THD> it(threads);

35
sql/sql_repl.cc
View File

@ -225,6 +225,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags)
if(read_packet)
{
thd->proc_info = "sending update to slave";
if(my_net_write(net, (char*)packet->ptr(), packet->length()) )
{
errmsg = "Failed on my_net_write()";
@ -257,7 +258,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags)
else
{
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))
{
case LOG_INFO_EOF:
@ -307,10 +308,12 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags)
}
(void)my_fclose(log, MYF(MY_WME));
send_eof(&thd->net);
thd->proc_info = "waiting to finalize termination";
DBUG_VOID_RETURN;
err:
thd->proc_info = "waiting to finalize termination";
if(log)
(void) my_fclose(log, MYF(MY_WME));
send_error(&thd->net, 0, errmsg);
@ -408,6 +411,34 @@ void reset_slave()
void kill_zombie_dump_threads(uint32 slave_server_id)
{
pthread_mutex_lock(&LOCK_thread_count);
I_List_iterator<THD> it(threads);
THD *tmp;
while((tmp=it++))
{
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;
pthread_mutex_lock(&tmp->mysys_var->mutex);
tmp->mysys_var->abort = 1;
if(tmp->mysys_var->current_mutex)
{
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);
}
}
pthread_mutex_unlock(&LOCK_thread_count);
}