database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-27 18:02:13 +03:00
Code Activity

merge mysql-next-mr --> mysql-5.1-rpl-merge

Browse Source 
Conflicts:
  Text conflict in sql/log.cc
  Text conflict in sql/slave.cc
  Text conflict in sql/sql_base.cc
This commit is contained in:
Alfranio Correia
2010-01-13 12:22:34 +00:00
parent 231773b449 357496c091
commit 46d1689b7c
166 changed files with 5710 additions and 2415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

324
sql/slave.cc
View File

@ -1,4 +1,4 @@
/* Copyright (C) 2000-2003 MySQL AB
/* Copyright (C) 2000-2003 MySQL AB, 2008-2009 Sun Microsystems, Inc
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
@ -148,8 +148,8 @@ 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,
mysql_mutex_t *term_lock,
mysql_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);
@ -198,8 +198,8 @@ 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);
mysql_mutex_lock(&mi->run_lock);
mysql_mutex_lock(&mi->rli.run_lock);
DBUG_VOID_RETURN;
}
@ -213,11 +213,32 @@ 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);
mysql_mutex_unlock(&mi->rli.run_lock);
mysql_mutex_unlock(&mi->run_lock);
DBUG_VOID_RETURN;
}
#ifdef HAVE_PSI_INTERFACE
static PSI_thread_key key_thread_slave_io, key_thread_slave_sql;
static PSI_thread_info all_slave_threads[]=
{
{ &key_thread_slave_io, "slave_io", PSI_FLAG_GLOBAL},
{ &key_thread_slave_sql, "slave_sql", PSI_FLAG_GLOBAL}
};
static void init_slave_psi_keys(void)
{
const char* category= "sql";
int count;
if (PSI_server == NULL)
return;
count= array_elements(all_slave_threads);
PSI_server->register_thread(category, all_slave_threads, count);
}
#endif /* HAVE_PSI_INTERFACE */
/* Initialize slave structures */
@ -226,12 +247,16 @@ int init_slave()
DBUG_ENTER("init_slave");
int error= 0;
#ifdef HAVE_PSI_INTERFACE
init_slave_psi_keys();
#endif
/*
This is called when mysqld starts. Before client connections are
accepted. However bootstrap may conflict with us if it does START SLAVE.
So it's safer to take the lock.
*/
pthread_mutex_lock(&LOCK_active_mi);
mysql_mutex_lock(&LOCK_active_mi);
/*
TODO: re-write this to interate through the list of files
for multi-master
@ -288,7 +313,7 @@ int init_slave()
}
err:
pthread_mutex_unlock(&LOCK_active_mi);
mysql_mutex_unlock(&LOCK_active_mi);
DBUG_RETURN(error);
}
@ -466,8 +491,8 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
if (!mi->inited)
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 *log_lock= mi->rli.relay_log.get_log_lock();
mysql_mutex_t *sql_lock = &mi->rli.run_lock, *io_lock = &mi->run_lock;
mysql_mutex_t *log_lock= mi->rli.relay_log.get_log_lock();
if (thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL))
{
@ -480,7 +505,7 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
!force_all)
DBUG_RETURN(error);
pthread_mutex_lock(log_lock);
mysql_mutex_lock(log_lock);
DBUG_PRINT("info",("Flushing relay log and master info file."));
if (current_thd)
@ -494,7 +519,7 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
if (my_sync(mi->fd, MYF(MY_WME)))
DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS);
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
}
if (thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL))
{
@ -507,7 +532,7 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
!force_all)
DBUG_RETURN(error);
pthread_mutex_lock(log_lock);
mysql_mutex_lock(log_lock);
DBUG_PRINT("info",("Flushing relay-log info file."));
if (current_thd)
@ -518,7 +543,7 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
if (my_sync(mi->rli.info_fd, MYF(MY_WME)))
DBUG_RETURN(ER_ERROR_DURING_FLUSH_LOGS);
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
}
DBUG_RETURN(0);
}
@ -560,19 +585,19 @@ int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
*/
static int
terminate_slave_thread(THD *thd,
pthread_mutex_t *term_lock,
pthread_cond_t *term_cond,
mysql_mutex_t *term_lock,
mysql_cond_t *term_cond,
volatile uint *slave_running,
bool skip_lock)
{
DBUG_ENTER("terminate_slave_thread");
if (!skip_lock)
{
pthread_mutex_lock(term_lock);
mysql_mutex_lock(term_lock);
}
else
{
safe_mutex_assert_owner(term_lock);
mysql_mutex_assert_owner(term_lock);
}
if (!*slave_running)
{
@ -582,7 +607,7 @@ terminate_slave_thread(THD *thd,
if run_lock (term_lock) is acquired locally then either
slave_running status is fine
*/
pthread_mutex_unlock(term_lock);
mysql_mutex_unlock(term_lock);
DBUG_RETURN(0);
}
else
@ -603,7 +628,7 @@ terminate_slave_thread(THD *thd,
int error;
DBUG_PRINT("loop", ("killing slave thread"));
pthread_mutex_lock(&thd->LOCK_thd_data);
mysql_mutex_lock(&thd->LOCK_thd_data);
#ifndef DONT_USE_THR_ALARM
/*
Error codes from pthread_kill are:
@ -614,7 +639,7 @@ terminate_slave_thread(THD *thd,
DBUG_ASSERT(err != EINVAL);
#endif
thd->awake(THD::NOT_KILLED);
pthread_mutex_unlock(&thd->LOCK_thd_data);
mysql_mutex_unlock(&thd->LOCK_thd_data);
/*
There is a small chance that slave thread might miss the first
@ -622,21 +647,25 @@ terminate_slave_thread(THD *thd,
*/
struct timespec abstime;
set_timespec(abstime,2);
error= pthread_cond_timedwait(term_cond, term_lock, &abstime);
error= mysql_cond_timedwait(term_cond, term_lock, &abstime);
DBUG_ASSERT(error == ETIMEDOUT || error == 0);
}
DBUG_ASSERT(*slave_running == 0);
if (!skip_lock)
pthread_mutex_unlock(term_lock);
mysql_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,
int start_slave_thread(
#ifdef HAVE_PSI_INTERFACE
PSI_thread_key thread_key,
#endif
pthread_handler h_func, mysql_mutex_t *start_lock,
mysql_mutex_t *cond_lock,
mysql_cond_t *start_cond,
volatile uint *slave_running,
volatile ulong *slave_run_id,
Master_info* mi)
@ -648,13 +677,13 @@ int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,
DBUG_ASSERT(mi->inited);
if (start_lock)
pthread_mutex_lock(start_lock);
mysql_mutex_lock(start_lock);
if (!server_id)
{
if (start_cond)
pthread_cond_broadcast(start_cond);
mysql_cond_broadcast(start_cond);
if (start_lock)
pthread_mutex_unlock(start_lock);
mysql_mutex_unlock(start_lock);
sql_print_error("Server id not set, will not start slave");
DBUG_RETURN(ER_BAD_SLAVE);
}
@ -662,17 +691,18 @@ int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,
if (*slave_running)
{
if (start_cond)
pthread_cond_broadcast(start_cond);
mysql_cond_broadcast(start_cond);
if (start_lock)
pthread_mutex_unlock(start_lock);
mysql_mutex_unlock(start_lock);
DBUG_RETURN(ER_SLAVE_MUST_STOP);
}
start_id= *slave_run_id;
DBUG_PRINT("info",("Creating new slave thread"));
if (pthread_create(&th, &connection_attrib, h_func, (void*)mi))
if (mysql_thread_create(thread_key,
&th, &connection_attrib, h_func, (void*)mi))
{
if (start_lock)
pthread_mutex_unlock(start_lock);
mysql_mutex_unlock(start_lock);
DBUG_RETURN(ER_SLAVE_THREAD);
}
if (start_cond && cond_lock) // caller has cond_lock
@ -683,19 +713,19 @@ 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");
pthread_cond_wait(start_cond, cond_lock);
mysql_cond_wait(start_cond, cond_lock);
thd->exit_cond(old_msg);
pthread_mutex_lock(cond_lock); // re-acquire it as exit_cond() released
mysql_mutex_lock(cond_lock); // re-acquire it as exit_cond() released
if (thd->killed)
{
if (start_lock)
pthread_mutex_unlock(start_lock);
mysql_mutex_unlock(start_lock);
DBUG_RETURN(thd->killed_errno());
}
}
}
if (start_lock)
pthread_mutex_unlock(start_lock);
mysql_mutex_unlock(start_lock);
DBUG_RETURN(0);
}
@ -713,8 +743,8 @@ 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)
{
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;
mysql_mutex_t *lock_io=0, *lock_sql=0, *lock_cond_io=0, *lock_cond_sql=0;
mysql_cond_t* cond_io=0, *cond_sql=0;
int error=0;
DBUG_ENTER("start_slave_threads");
@ -732,16 +762,24 @@ 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);
error= start_slave_thread(
#ifdef HAVE_PSI_INTERFACE
key_thread_slave_io,
#endif
handle_slave_io, lock_io, lock_cond_io,
cond_io,
&mi->slave_running, &mi->slave_run_id,
mi);
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);
error= start_slave_thread(
#ifdef HAVE_PSI_INTERFACE
key_thread_slave_sql,
#endif
handle_slave_sql, lock_sql, lock_cond_sql,
cond_sql,
&mi->rli.slave_running, &mi->rli.slave_run_id,
mi);
if (error)
terminate_slave_threads(mi, thread_mask & SLAVE_IO, !need_slave_mutex);
}
@ -778,7 +816,7 @@ void end_slave()
will make us wait until slave threads have started, and START SLAVE
returns, then we terminate them here.
*/
pthread_mutex_lock(&LOCK_active_mi);
mysql_mutex_lock(&LOCK_active_mi);
if (active_mi)
{
/*
@ -788,7 +826,7 @@ void end_slave()
*/
terminate_slave_threads(active_mi,SLAVE_FORCE_ALL);
}
pthread_mutex_unlock(&LOCK_active_mi);
mysql_mutex_unlock(&LOCK_active_mi);
DBUG_VOID_RETURN;
}
@ -803,14 +841,14 @@ void end_slave()
*/
void close_active_mi()
{
pthread_mutex_lock(&LOCK_active_mi);
mysql_mutex_lock(&LOCK_active_mi);
if (active_mi)
{
end_master_info(active_mi);
delete active_mi;
active_mi= 0;
}
pthread_mutex_unlock(&LOCK_active_mi);
mysql_mutex_unlock(&LOCK_active_mi);
}
static bool io_slave_killed(THD* thd, Master_info* mi)
@ -1509,7 +1547,7 @@ static bool wait_for_relay_log_space(Relay_log_info* rli)
THD* thd = mi->io_thd;
DBUG_ENTER("wait_for_relay_log_space");
pthread_mutex_lock(&rli->log_space_lock);
mysql_mutex_lock(&rli->log_space_lock);
save_proc_info= thd->enter_cond(&rli->log_space_cond,
&rli->log_space_lock,
"\
@ -1517,7 +1555,7 @@ 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)) &&
!rli->ignore_log_space_limit)
pthread_cond_wait(&rli->log_space_cond, &rli->log_space_lock);
mysql_cond_wait(&rli->log_space_cond, &rli->log_space_lock);
thd->exit_cond(save_proc_info);
DBUG_RETURN(slave_killed);
}
@ -1539,11 +1577,11 @@ Waiting for the slave SQL thread to free enough relay log space");
static void write_ignored_events_info_to_relay_log(THD *thd, Master_info *mi)
{
Relay_log_info *rli= &mi->rli;
pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
mysql_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);
mysql_mutex_lock(log_lock);
if (rli->ign_master_log_name_end[0])
{
DBUG_PRINT("info",("writing a Rotate event to track down ignored events"));
@ -1552,7 +1590,7 @@ static void write_ignored_events_info_to_relay_log(THD *thd, Master_info *mi)
Rotate_log_event::DUP_NAME);
rli->ign_master_log_name_end[0]= 0;
/* can unlock before writing as slave SQL thd will soon see our Rotate */
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
if (likely((bool)ev))
{
ev->server_id= 0; // don't be ignored by slave SQL thread
@ -1574,7 +1612,7 @@ static void write_ignored_events_info_to_relay_log(THD *thd, Master_info *mi)
" SHOW SLAVE STATUS may be inaccurate");
}
else
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
DBUG_VOID_RETURN;
}
@ -1751,14 +1789,14 @@ bool show_master_info(THD* thd, Master_info* mi)
slave_running can be accessed without run_lock but not other
non-volotile members like mi->io_thd, which is guarded by the mutex.
*/
pthread_mutex_lock(&mi->run_lock);
mysql_mutex_lock(&mi->run_lock);
protocol->store(mi->io_thd ? mi->io_thd->proc_info : "", &my_charset_bin);
pthread_mutex_unlock(&mi->run_lock);
mysql_mutex_unlock(&mi->run_lock);
pthread_mutex_lock(&mi->data_lock);
pthread_mutex_lock(&mi->rli.data_lock);
pthread_mutex_lock(&mi->err_lock);
pthread_mutex_lock(&mi->rli.err_lock);
mysql_mutex_lock(&mi->data_lock);
mysql_mutex_lock(&mi->rli.data_lock);
mysql_mutex_lock(&mi->err_lock);
mysql_mutex_lock(&mi->rli.err_lock);
protocol->store(mi->host, &my_charset_bin);
protocol->store(mi->user, &my_charset_bin);
protocol->store((uint32) mi->port);
@ -1885,10 +1923,10 @@ bool show_master_info(THD* thd, Master_info* mi)
// Master_Server_id
protocol->store((uint32) mi->master_id);
pthread_mutex_unlock(&mi->rli.err_lock);
pthread_mutex_unlock(&mi->err_lock);
pthread_mutex_unlock(&mi->rli.data_lock);
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->rli.err_lock);
mysql_mutex_unlock(&mi->err_lock);
mysql_mutex_unlock(&mi->rli.data_lock);
mysql_mutex_unlock(&mi->data_lock);
if (my_net_write(&thd->net, (uchar*) thd->packet.ptr(), packet->length()))
DBUG_RETURN(TRUE);
@ -2264,7 +2302,7 @@ int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli)
int reason= ev->shall_skip(rli);
if (reason == Log_event::EVENT_SKIP_COUNT)
sql_slave_skip_counter= --rli->slave_skip_counter;
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
if (reason == Log_event::EVENT_SKIP_NOT)
exec_res= ev->apply_event(rli);
@ -2370,7 +2408,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
event execution. But we will release it in places where we will
wait for something for example inside of next_event().
*/
pthread_mutex_lock(&rli->data_lock);
mysql_mutex_lock(&rli->data_lock);
Log_event * ev = next_event(rli);
@ -2378,7 +2416,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
if (sql_slave_killed(thd,rli))
{
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
delete ev;
DBUG_RETURN(1);
}
@ -2401,7 +2439,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
error in query execution to be printed.
*/
rli->abort_slave= 1;
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
delete ev;
DBUG_RETURN(1);
}
@ -2419,7 +2457,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
{
DBUG_ASSERT(thd->transaction.all.modified_non_trans_table);
rli->abort_slave= 1;
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
delete ev;
rli->inc_event_relay_log_pos();
DBUG_RETURN(0);
@ -2483,10 +2521,10 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
/* 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);
pthread_mutex_lock(&rli->data_lock); // because of SHOW STATUS
mysql_mutex_lock(&rli->data_lock); // because of SHOW STATUS
rli->trans_retries++;
rli->retried_trans++;
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
DBUG_PRINT("info", ("Slave retries transaction "
"rli->trans_retries: %lu", rli->trans_retries));
}
@ -2514,7 +2552,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
}
DBUG_RETURN(exec_res);
}
pthread_mutex_unlock(&rli->data_lock);
mysql_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 \
@ -2645,7 +2683,7 @@ pthread_handler_t handle_slave_io(void *arg)
mysql= NULL ;
retry_count= 0;
pthread_mutex_lock(&mi->run_lock);
mysql_mutex_lock(&mi->run_lock);
/* Inform waiting threads that slave has started */
mi->slave_run_id++;
@ -2662,8 +2700,8 @@ pthread_handler_t handle_slave_io(void *arg)
mi->clear_error();
if (init_slave_thread(thd, SLAVE_THD_IO))
{
pthread_cond_broadcast(&mi->start_cond);
pthread_mutex_unlock(&mi->run_lock);
mysql_cond_broadcast(&mi->start_cond);
mysql_mutex_unlock(&mi->run_lock);
sql_print_error("Failed during slave I/O thread initialization");
goto err;
}
@ -2672,8 +2710,8 @@ pthread_handler_t handle_slave_io(void *arg)
pthread_mutex_unlock(&LOCK_thread_count);
mi->slave_running = 1;
mi->abort_slave = 0;
pthread_mutex_unlock(&mi->run_lock);
pthread_cond_broadcast(&mi->start_cond);
mysql_mutex_unlock(&mi->run_lock);
mysql_cond_broadcast(&mi->start_cond);
DBUG_PRINT("master_info",("log_file_name: '%s' position: %s",
mi->master_log_name,
@ -2956,7 +2994,7 @@ err:
}
write_ignored_events_info_to_relay_log(thd, mi);
thd_proc_info(thd, "Waiting for slave mutex on exit");
pthread_mutex_lock(&mi->run_lock);
mysql_mutex_lock(&mi->run_lock);
/* Forget the relay log's format */
delete mi->rli.relay_log.description_event_for_queue;
@ -2978,9 +3016,9 @@ err:
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(&mi->stop_cond); // tell the world we are done
mysql_cond_broadcast(&mi->stop_cond); // tell the world we are done
DBUG_EXECUTE_IF("simulate_slave_delay_at_terminate_bug38694", sleep(5););
pthread_mutex_unlock(&mi->run_lock);
mysql_mutex_unlock(&mi->run_lock);
DBUG_LEAVE; // Must match DBUG_ENTER()
my_thread_end();
@ -3017,16 +3055,17 @@ int check_temp_dir(char* tmp_file)
/*
Check permissions to create a file.
*/
if ((fd= my_create(tmp_file, CREATE_MODE,
O_WRONLY | O_BINARY | O_EXCL | O_NOFOLLOW,
MYF(MY_WME))) < 0)
if ((fd= mysql_file_create(key_file_misc,
tmp_file, CREATE_MODE,
O_WRONLY | O_BINARY | O_EXCL | O_NOFOLLOW,
MYF(MY_WME))) < 0)
DBUG_RETURN(1);
/*
Clean up.
*/
my_close(fd, MYF(0));
my_delete(tmp_file, MYF(0));
mysql_file_close(fd, MYF(0));
mysql_file_delete(key_file_misc, tmp_file, MYF(0));
DBUG_RETURN(0);
}
@ -3052,7 +3091,7 @@ pthread_handler_t handle_slave_sql(void *arg)
DBUG_ENTER("handle_slave_sql");
DBUG_ASSERT(rli->inited);
pthread_mutex_lock(&rli->run_lock);
mysql_mutex_lock(&rli->run_lock);
DBUG_ASSERT(!rli->slave_running);
errmsg= 0;
#ifndef DBUG_OFF
@ -3074,8 +3113,8 @@ pthread_handler_t handle_slave_sql(void *arg)
TODO: this is currently broken - slave start and change master
will be stuck if we fail here
*/
pthread_cond_broadcast(&rli->start_cond);
pthread_mutex_unlock(&rli->run_lock);
mysql_cond_broadcast(&rli->start_cond);
mysql_mutex_unlock(&rli->run_lock);
rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
"Failed during slave thread initialization");
goto err;
@ -3095,8 +3134,8 @@ pthread_handler_t handle_slave_sql(void *arg)
Seconds_Behind_Master grows. No big deal.
*/
rli->abort_slave = 0;
pthread_mutex_unlock(&rli->run_lock);
pthread_cond_broadcast(&rli->start_cond);
mysql_mutex_unlock(&rli->run_lock);
mysql_cond_broadcast(&rli->start_cond);
/*
Reset errors for a clean start (otherwise, if the master is idle, the SQL
@ -3111,9 +3150,9 @@ pthread_handler_t handle_slave_sql(void *arg)
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);
mysql_mutex_lock(&rli->log_space_lock);
rli->ignore_log_space_limit= 0;
pthread_mutex_unlock(&rli->log_space_lock);
mysql_mutex_unlock(&rli->log_space_lock);
rli->trans_retries= 0; // start from "no error"
DBUG_PRINT("info", ("rli->trans_retries: %lu", rli->trans_retries));
@ -3187,17 +3226,17 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
First check until condition - probably there is nothing to execute. We
do not want to wait for next event in this case.
*/
pthread_mutex_lock(&rli->data_lock);
mysql_mutex_lock(&rli->data_lock);
if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
rli->is_until_satisfied(thd, NULL))
{
char buf[22];
sql_print_information("Slave SQL thread stopped because it reached its"
" UNTIL position %s", llstr(rli->until_pos(), buf));
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
goto err;
}
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
/* Read queries from the IO/THREAD until this thread is killed */
@ -3300,9 +3339,9 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \
thd->set_query(NULL, 0);
thd->reset_db(NULL, 0);
thd_proc_info(thd, "Waiting for slave mutex on exit");
pthread_mutex_lock(&rli->run_lock);
mysql_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);
mysql_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;
@ -3310,9 +3349,9 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \
delete rli->relay_log.description_event_for_exec;
rli->relay_log.description_event_for_exec= 0;
/* Wake up master_pos_wait() */
pthread_mutex_unlock(&rli->data_lock);
mysql_mutex_unlock(&rli->data_lock);
DBUG_PRINT("info",("Signaling possibly waiting master_pos_wait() functions"));
pthread_cond_broadcast(&rli->data_cond);
mysql_cond_broadcast(&rli->data_cond);
rli->ignore_log_space_limit= 0; /* don't need any lock */
/* we die so won't remember charset - re-update them on next thread start */
rli->cached_charset_invalidate();
@ -3338,9 +3377,9 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \
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);
mysql_cond_broadcast(&rli->stop_cond);
DBUG_EXECUTE_IF("simulate_slave_delay_at_terminate_bug38694", sleep(5););
pthread_mutex_unlock(&rli->run_lock); // tell the world we are done
mysql_mutex_unlock(&rli->run_lock); // tell the world we are done
DBUG_LEAVE; // Must match DBUG_ENTER()
my_thread_end();
@ -3483,7 +3522,7 @@ err:
static int process_io_rotate(Master_info *mi, Rotate_log_event *rev)
{
DBUG_ENTER("process_io_rotate");
safe_mutex_assert_owner(&mi->data_lock);
mysql_mutex_assert_owner(&mi->data_lock);
if (unlikely(!rev->is_valid()))
DBUG_RETURN(1);
@ -3579,7 +3618,7 @@ static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
DBUG_RETURN(1);
}
pthread_mutex_lock(&mi->data_lock);
mysql_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()) {
case STOP_EVENT:
@ -3590,7 +3629,7 @@ static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
if (unlikely(process_io_rotate(mi,(Rotate_log_event*)ev)))
{
delete ev;
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_RETURN(1);
}
inc_pos= 0;
@ -3611,7 +3650,7 @@ static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
delete ev;
mi->master_log_pos += inc_pos;
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
my_free((char*)tmp_buf, MYF(0));
DBUG_RETURN(error);
}
@ -3630,7 +3669,7 @@ static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
if (unlikely(rli->relay_log.append(ev)))
{
delete ev;
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_RETURN(1);
}
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
@ -3638,7 +3677,7 @@ static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
delete ev;
mi->master_log_pos+= inc_pos;
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_RETURN(0);
}
@ -3666,7 +3705,7 @@ static int queue_binlog_ver_3_event(Master_info *mi, const char *buf,
my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR));
DBUG_RETURN(1);
}
pthread_mutex_lock(&mi->data_lock);
mysql_mutex_lock(&mi->data_lock);
switch (ev->get_type_code()) {
case STOP_EVENT:
goto err;
@ -3674,7 +3713,7 @@ static int queue_binlog_ver_3_event(Master_info *mi, const char *buf,
if (unlikely(process_io_rotate(mi,(Rotate_log_event*)ev)))
{
delete ev;
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_RETURN(1);
}
inc_pos= 0;
@ -3686,7 +3725,7 @@ static int queue_binlog_ver_3_event(Master_info *mi, const char *buf,
if (unlikely(rli->relay_log.append(ev)))
{
delete ev;
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_RETURN(1);
}
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
@ -3694,7 +3733,7 @@ static int queue_binlog_ver_3_event(Master_info *mi, const char *buf,
mi->master_log_pos+= inc_pos;
err:
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_RETURN(0);
}
@ -3744,7 +3783,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
String error_msg;
ulong inc_pos;
Relay_log_info *rli= &mi->rli;
pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
mysql_mutex_t *log_lock= rli->relay_log.get_log_lock();
ulong s_id;
DBUG_ENTER("queue_event");
@ -3755,7 +3794,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
DBUG_RETURN(queue_old_event(mi,buf,event_len));
LINT_INIT(inc_pos);
pthread_mutex_lock(&mi->data_lock);
mysql_mutex_lock(&mi->data_lock);
switch (buf[EVENT_TYPE_OFFSET]) {
case STOP_EVENT:
@ -3891,7 +3930,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
direct master (an unsupported, useless setup!).
*/
pthread_mutex_lock(log_lock);
mysql_mutex_lock(log_lock);
s_id= uint4korr(buf + SERVER_ID_OFFSET);
if ((s_id == ::server_id && !mi->rli.replicate_same_server_id) ||
/*
@ -3953,12 +3992,12 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
}
rli->ign_master_log_name_end[0]= 0; // last event is not ignored
}
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
skip_relay_logging:
err:
pthread_mutex_unlock(&mi->data_lock);
mysql_mutex_unlock(&mi->data_lock);
DBUG_PRINT("info", ("error: %d", error));
if (error)
mi->report(ERROR_LEVEL, error, ER(error),
@ -3978,13 +4017,13 @@ void end_relay_log_info(Relay_log_info* rli)
if (rli->info_fd >= 0)
{
end_io_cache(&rli->info_file);
(void) my_close(rli->info_fd, MYF(MY_WME));
mysql_file_close(rli->info_fd, MYF(MY_WME));
rli->info_fd = -1;
}
if (rli->cur_log_fd >= 0)
{
end_io_cache(&rli->cache_buf);
(void)my_close(rli->cur_log_fd, MYF(MY_WME));
mysql_file_close(rli->cur_log_fd, MYF(MY_WME));
rli->cur_log_fd = -1;
}
rli->inited = 0;
@ -4344,7 +4383,7 @@ 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();
mysql_mutex_t *log_lock = rli->relay_log.get_log_lock();
const char* errmsg=0;
THD* thd = rli->sql_thd;
DBUG_ENTER("next_event");
@ -4361,9 +4400,9 @@ static Log_event* next_event(Relay_log_info* rli)
so we assume calling function acquired this mutex for us and we will
hold it for the most of the loop below However, we will release it
whenever it is worth the hassle, and in the cases when we go into a
pthread_cond_wait() with the non-data_lock mutex
mysql_cond_wait() with the non-data_lock mutex
*/
safe_mutex_assert_owner(&rli->data_lock);
mysql_mutex_assert_owner(&rli->data_lock);
while (!sql_slave_killed(thd,rli))
{
@ -4382,7 +4421,7 @@ static Log_event* next_event(Relay_log_info* rli)
if ((hot_log = (cur_log != &rli->cache_buf)))
{
DBUG_ASSERT(rli->cur_log_fd == -1); // foreign descriptor
pthread_mutex_lock(log_lock);
mysql_mutex_lock(log_lock);
/*
Reading xxx_file_id is safe because the log will only
@ -4392,7 +4431,7 @@ static Log_event* next_event(Relay_log_info* rli)
{
// 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);
mysql_mutex_unlock(log_lock);
if (!cur_log) // No more log files
goto err;
hot_log=0; // Using old binary log
@ -4439,7 +4478,7 @@ static 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);
mysql_mutex_unlock(log_lock);
DBUG_RETURN(ev);
}
DBUG_ASSERT(thd==rli->sql_thd);
@ -4449,7 +4488,7 @@ static Log_event* next_event(Relay_log_info* rli)
{
errmsg = "slave SQL thread aborted because of I/O error";
if (hot_log)
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
goto err;
}
if (!cur_log->error) /* EOF */
@ -4496,7 +4535,7 @@ static Log_event* next_event(Relay_log_info* rli)
0, rli->ign_master_log_pos_end,
Rotate_log_event::DUP_NAME);
rli->ign_master_log_name_end[0]= 0;
pthread_mutex_unlock(log_lock);
mysql_mutex_unlock(log_lock);
if (unlikely(!ev))
{
errmsg= "Slave SQL thread failed to create a Rotate event "
@ -4511,7 +4550,7 @@ static Log_event* next_event(Relay_log_info* rli)
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);
mysql_mutex_unlock(&rli->data_lock);
/*
Possible deadlock :
@ -4537,7 +4576,7 @@ static Log_event* next_event(Relay_log_info* rli)
be stopped, and the SQL thread sets ignore_log_space_limit to 0 when
it stops.
*/
pthread_mutex_lock(&rli->log_space_lock);
mysql_mutex_lock(&rli->log_space_lock);
// prevent the I/O thread from blocking next times
rli->ignore_log_space_limit= 1;
/*
@ -4546,12 +4585,12 @@ static Log_event* next_event(Relay_log_info* rli)
~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);
mysql_mutex_unlock(&rli->log_space_lock);
mysql_cond_broadcast(&rli->log_space_cond);
// Note that wait_for_update_relay_log unlocks lock_log !
rli->relay_log.wait_for_update_relay_log(rli->sql_thd);
// re-acquire data lock since we released it earlier
pthread_mutex_lock(&rli->data_lock);
mysql_mutex_lock(&rli->data_lock);
rli->last_master_timestamp= save_timestamp;
continue;
}
@ -4562,7 +4601,7 @@ static Log_event* next_event(Relay_log_info* rli)
*/
end_io_cache(cur_log);
DBUG_ASSERT(rli->cur_log_fd >= 0);
my_close(rli->cur_log_fd, MYF(MY_WME));
mysql_file_close(rli->cur_log_fd, MYF(MY_WME));
rli->cur_log_fd = -1;
if (relay_log_purge)
@ -4619,7 +4658,7 @@ static Log_event* next_event(Relay_log_info* rli)
DBUG_PRINT("info",("hot_log: %d",hot_log));
if (!hot_log) /* if hot_log, we already have this mutex */
pthread_mutex_lock(log_lock);
mysql_mutex_lock(log_lock);
if (rli->relay_log.is_active(rli->linfo.log_file_name))
{
#ifdef EXTRA_DEBUG
@ -4640,13 +4679,16 @@ static Log_event* next_event(Relay_log_info* rli)
*/
if (check_binlog_magic(cur_log,&errmsg))
{
if (!hot_log) pthread_mutex_unlock(log_lock);
if (!hot_log)
mysql_mutex_unlock(log_lock);
goto err;
}
if (!hot_log) pthread_mutex_unlock(log_lock);
if (!hot_log)
mysql_mutex_unlock(log_lock);
continue;
}
if (!hot_log) pthread_mutex_unlock(log_lock);
if (!hot_log)
mysql_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
@ -4669,7 +4711,7 @@ static Log_event* next_event(Relay_log_info* rli)
TODO: come up with something better to handle this error
*/
if (hot_log)
pthread_mutex_unlock(log_lock);
mysql_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);