MySQL数据库

开发平台：

Visual C++

slave.cpp：源码内容

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
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
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
#include <mysql.h>
#include <myisam.h>
#include "slave.h"
#include "sql_repl.h"
#include "repl_failsafe.h"
#include <thr_alarm.h>
#include <my_dir.h>
#include <sql_common.h>
#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, replicate_same_server_id;
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
*/
int disconnect_slave_event_count = 0, abort_slave_event_count = 0;
int events_till_abort = -1;
static int events_till_disconnect = -1;
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 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_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed,
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);
/*
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
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
*/
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;
if (set_io)
tmp_mask |= SLAVE_IO;
if (set_sql)
tmp_mask |= SLAVE_SQL;
if (inverse)
tmp_mask^= (SLAVE_IO | SLAVE_SQL);
*mask = tmp_mask;
}
/*
lock_slave_threads()
*/
void lock_slave_threads(MASTER_INFO* mi)
{
//TODO: see if we can do this without dual mutex
pthread_mutex_lock(&mi->run_lock);
pthread_mutex_lock(&mi->rli.run_lock);
}
/*
unlock_slave_threads()
*/
void unlock_slave_threads(MASTER_INFO* mi)
{
//TODO: see if we can do this without dual mutex
pthread_mutex_unlock(&mi->rli.run_lock);
pthread_mutex_unlock(&mi->run_lock);
}
/* Initialize slave structures */
int init_slave()
{
DBUG_ENTER("init_slave");
/*
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);
/*
TODO: re-write this to interate through the list of files
for multi-master
*/
active_mi= new MASTER_INFO;
/*
If master_host is not specified, try to read it from the master_info file.
If master_host is specified, create the master_info file if it doesn't
exists.
*/
if (!active_mi)
{
sql_print_error("Failed to allocate memory for the master info structure");
goto err;
}
if (init_master_info(active_mi,master_info_file,relay_log_info_file,
!master_host, (SLAVE_IO | SLAVE_SQL)))
{
sql_print_error("Failed to initialize the master info structure");
goto err;
}
if (server_id && !master_host && active_mi->host[0])
master_host= active_mi->host;
/* If server id is not set, start_slave_thread() will say it */
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))
{
sql_print_error("Failed to create slave threads");
goto err;
}
}
pthread_mutex_unlock(&LOCK_active_mi);
DBUG_RETURN(0);
err:
pthread_mutex_unlock(&LOCK_active_mi);
DBUG_RETURN(1);
}
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
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)
{
DBUG_ENTER("init_relay_log_pos");
*errmsg=0;
pthread_mutex_t *log_lock=rli->relay_log.get_log_lock();
if (need_data_lock)
pthread_mutex_lock(&rli->data_lock);
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.
*/
rli->cur_log= rli->relay_log.get_log_file();
if (my_b_tell(rli->cur_log) == 0 &&
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;
}
if (pos >= BIN_LOG_HEADER_SIZE)
my_b_seek(rli->cur_log,(off_t)pos);
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);
DBUG_RETURN ((*errmsg) ? 1 : 0);
}
/*
Init functio 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 ','
NOTES
Called from get_options() in mysqld.cc on start-up
*/
void init_slave_skip_errors(const char* arg)
{
const char *p;
if (bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0))
{
fprintf(stderr, "Badly out of memory, please check your system statusn");
exit(1);
}
use_slave_mask = 1;
for (;my_isspace(system_charset_info,*arg);++arg)
/* empty */;
if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const uchar*)"all",4))
{
bitmap_set_all(&slave_error_mask);
return;
}
for (p= arg ; *p; )
{
long err_code;
if (!(p= str2int(p, 10, 0, LONG_MAX, &err_code)))
break;
if (err_code < MAX_SLAVE_ERROR)
bitmap_set_bit(&slave_error_mask,(uint)err_code);
while (!my_isdigit(system_charset_info,*p) && *p)
p++;
}
}
void st_relay_log_info::inc_group_relay_log_pos(ulonglong val,
ulonglong log_pos,
bool skip_lock)
{
if (!skip_lock)
pthread_mutex_lock(&data_lock);
inc_event_relay_log_pos(val);
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).
*/
if (log_pos) // 3.23 binlogs don't have log_posx
{
#if MYSQL_VERSION_ID < 50000
/*
If the event was converted from a 3.23 format, get_event_len() has
grown by 6 bytes (at least for most events, except LOAD DATA INFILE
which is already a big problem for 3.23->4.0 replication); 6 bytes is
the difference between the header's size in 4.0 (LOG_EVENT_HEADER_LEN)
and the header's size in 3.23 (OLD_HEADER_LEN). Note that using
mi->old_format will not help if the I/O thread has not started yet.
Yes this is a hack but it's just to make 3.23->4.x replication work;
3.23->5.0 replication is working much better.
*/
group_master_log_pos= log_pos + val -
(mi->old_format ? (LOG_EVENT_HEADER_LEN - OLD_HEADER_LEN) : 0);
#else
group_master_log_pos= log_pos+ val;
#endif /* MYSQL_VERSION_ID < 5000 */
}
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 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);
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);
// Just first log with magic number and nothing else
rli->log_space_total= BIN_LOG_HEADER_SIZE;
rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE;
rli->relay_log.reset_bytes_written();
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);
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);
}
int terminate_slave_threads(MASTER_INFO* mi,int thread_mask,bool skip_lock)
{
if (!mi->inited)
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)
{
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)
DBUG_RETURN(error);
}
if ((thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL)) && mi->rli.slave_running)
{
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)
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)
{
if (term_lock)
{
pthread_mutex_lock(term_lock);
if (!*slave_running)
{
pthread_mutex_unlock(term_lock);
return ER_SLAVE_NOT_RUNNING;
}
}
DBUG_ASSERT(thd != 0);
/*
Is is criticate to test if the slave is running. Otherwise, we might
be referening freed memory trying to kick it
*/
THD_CHECK_SENTRY(thd);
while (*slave_running) // Should always be true
{
KICK_SLAVE(thd);
/*
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;
set_timespec(abstime,2);
pthread_cond_timedwait(term_cond, cond_lock, &abstime);
}
if (term_lock)
pthread_mutex_unlock(term_lock);
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,
bool high_priority)
{
pthread_t th;
ulong start_id;
DBUG_ASSERT(mi->inited);
DBUG_ENTER("start_slave_thread");
if (start_lock)
pthread_mutex_lock(start_lock);
if (!server_id)
{
if (start_cond)
pthread_cond_broadcast(start_cond);
if (start_lock)
pthread_mutex_unlock(start_lock);
sql_print_error("Server id not set, will not start slave");
DBUG_RETURN(ER_BAD_SLAVE);
}
if (*slave_running)
{
if (start_cond)
pthread_cond_broadcast(start_cond);
if (start_lock)
pthread_mutex_unlock(start_lock);
DBUG_RETURN(ER_SLAVE_MUST_STOP);
}
start_id= *slave_run_id;
DBUG_PRINT("info",("Creating new slave thread"));
if (high_priority)
my_pthread_attr_setprio(&connection_attrib,CONNECT_PRIOR);
if (pthread_create(&th, &connection_attrib, h_func, (void*)mi))
{
if (start_lock)
pthread_mutex_unlock(start_lock);
DBUG_RETURN(ER_SLAVE_THREAD);
}
if (start_cond && cond_lock) // caller has cond_lock
{
THD* thd = current_thd;
while (start_id == *slave_run_id)
{
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);
thd->exit_cond(old_msg);
pthread_mutex_lock(cond_lock); // re-acquire it as exit_cond() released
if (thd->killed)
DBUG_RETURN(ER_SERVER_SHUTDOWN);
}
}
if (start_lock)
pthread_mutex_unlock(start_lock);
DBUG_RETURN(0);
}
/*
start_slave_threads()
NOTES
SLAVE_FORCE_ALL is not implemented here on purpose since it does not make
sense to do that for starting a slave--we always care if it actually
started the threads that were not previously running
*/
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;
int error=0;
DBUG_ENTER("start_slave_threads");
if (need_slave_mutex)
{
lock_io = &mi->run_lock;
lock_sql = &mi->rli.run_lock;
}
if (wait_for_start)
{
cond_io = &mi->start_cond;
cond_sql = &mi->rli.start_cond;
lock_cond_io = &mi->run_lock;
lock_cond_sql = &mi->rli.run_lock;
}
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
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);
if (error)
terminate_slave_threads(mi, thread_mask & SLAVE_IO, 0);
}
DBUG_RETURN(error);
}
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)
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).
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");
for (; tables; tables = tables->next)
{
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->real_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*/)
{
end_master_info(mi);
return 0;
}
#endif
/*
Free all resources used by slave
SYNOPSIS
end_slave()
*/
void end_slave()
{
/*
This is called when the server terminates, in close_connections().
It terminates slave threads. However, some CHANGE MASTER etc may still be
running presently. If a START SLAVE was in progress, the mutex lock below
will make us wait until slave threads have started, and START SLAVE
returns, then we terminate them here.
*/
pthread_mutex_lock(&LOCK_active_mi);
if (active_mi)
{
/*
TODO: replace the line below with
list_walk(&master_list, (list_walk_action)end_slave_on_walk,0);
once multi-master code is ready.
*/
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);
}
static bool io_slave_killed(THD* thd, MASTER_INFO* mi)
{
DBUG_ASSERT(mi->io_thd == thd);
DBUG_ASSERT(mi->slave_running); // tracking buffer overrun
return mi->abort_slave || abort_loop || thd->killed;
}
static bool sql_slave_killed(THD* thd, RELAY_LOG_INFO* rli)
{
DBUG_ASSERT(rli->sql_thd == thd);
DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun
return rli->abort_slave || abort_loop || thd->killed;
}
/*
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
This is used to tell a 3.23 master to break send_file()
*/
void skip_load_data_infile(NET *net)
{
(void)net_request_file(net, "/dev/null");
(void)my_net_read(net); // discard response
(void)net_write_command(net, 0, "", 0, "", 0); // Send ok
}
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));
}
const char *rewrite_db(const char* db, uint32 *new_len)
{
if (replicate_rewrite_db.is_empty() || !db)
return db;
I_List_iterator<i_string_pair> it(replicate_rewrite_db);
i_string_pair* tmp;
while ((tmp=it++))
{
if (!strcmp(tmp->key, db))
{
*new_len= (uint32)strlen(tmp->val);
return tmp->val;
}
}
return db;
}
/*
From other comments and tests in code, it looks like
sometimes Query_log_event and Load_log_event can have db == 0
(see rewrite_db() above for example)
(cases where this happens are unclear; it may be when the master is 3.23).
*/
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;
while ((tmp=it++))
{
if (!strcmp(tmp->ptr, db))
return 0; // match
}
return 1;
}
}
static int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
const char *default_val)
{
uint length;
if ((length=my_b_gets(f,var, max_size)))
{
char* last_p = var + length -1;
if (*last_p == 'n')
*last_p = 0; // if we stopped on newline, kill it
else
{
/*
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;
}
else if (default_val)
{
strmake(var, default_val, max_size-1);
return 0;
}
return 1;
}
static int init_intvar_from_file(int* var, IO_CACHE* f, int default_val)
{
char buf[32];
if (my_b_gets(f, buf, sizeof(buf)))
{
*var = atoi(buf);
return 0;
}
else if (default_val)
{
*var = default_val;
return 0;
}
return 1;
}
static int get_master_version_and_clock(MYSQL* mysql, MASTER_INFO* mi)
{
const char* errmsg= 0;
/*
Note the following switch will bug when we have MySQL branch 30 ;)
*/
switch (*mysql->server_version) {
case '3':
mi->old_format =
(strncmp(mysql->server_version, "3.23.57", 7) < 0) /* < .57 */ ?
BINLOG_FORMAT_323_LESS_57 :
BINLOG_FORMAT_323_GEQ_57 ;
break;
case '4':
mi->old_format = BINLOG_FORMAT_CURRENT;
break;
default:
/* 5.0 is not supported */
errmsg = "Master reported an unrecognized MySQL version. Note that 4.1
slaves can't replicate a 5.0 or newer master.";
break;
}
/*
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, "SELECT UNIX_TIMESTAMP()", 23) &&
(master_res= mysql_store_result(mysql)) &&
(master_row= mysql_fetch_row(master_res)))
{
mi->clock_diff_with_master=
(long) (time((time_t*) 0) - strtoul(master_row[0], 0, 10));
}
else
{
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");
}
if (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,
thus copying some my.cnf), replication will work but all events will be
skipped.
Do not die if SHOW VARIABLES LIKE 'SERVER_ID' fails on master (very old
master?).
Note: we could have put a @@SERVER_ID in the previous SELECT
UNIX_TIMESTAMP() instead, but this would not have worked on 3.23 masters.
*/
if (!mysql_real_query(mysql, "SHOW VARIABLES LIKE 'SERVER_ID'", 31) &&
(master_res= mysql_store_result(mysql)))
{
if ((master_row= mysql_fetch_row(master_res)) &&
(::server_id == strtoul(master_row[1], 0, 10)) &&
!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).";
mysql_free_result(master_res);
}
/*
Check that the master's global character_set_server and ours are the same.
Not fatal if query fails (old master?).
Note that we don't check for equality of global character_set_client and
collation_connection (neither do we prevent their setting in
set_var.cc). That's because from what I (Guilhem) have tested, the global
values of these 2 are never used (new connections don't use them).
We don't test equality of global collation_database either as it's is
going to be deprecated (made read-only) in 4.1 very soon.
We don't do it for <3.23.57 because masters <3.23.50 hang on
SELECT @@unknown_var (BUG#7965 - see changelog of 3.23.50).
*/
if (mi->old_format == BINLOG_FORMAT_323_LESS_57)
goto err;
if (!mysql_real_query(mysql, "SELECT @@GLOBAL.COLLATION_SERVER", 32) &&
(master_res= mysql_store_result(mysql)))
{
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";
mysql_free_result(master_res);
}
/*
Perform analogous check for time zone. Theoretically we also should
perform check here to verify that SYSTEM time zones are the same on
slave and master, but we can't rely on value of @@system_time_zone
variable (it is time zone abbreviation) since it determined at start
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
'SYSTEM' time zone).
*/
if (!mysql_real_query(mysql, "SELECT @@GLOBAL.TIME_ZONE", 25) &&
(master_res= mysql_store_result(mysql)))
{
if ((master_row= mysql_fetch_row(master_res)) &&
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";
mysql_free_result(master_res);
}
err:
if (errmsg)
{
sql_print_error(errmsg);
return 1;
}
return 0;
}
/*
Used by fetch_master_table (used by LOAD TABLE tblname FROM MASTER and LOAD
DATA FROM MASTER). Drops the table (if 'overwrite' is true) and recreates it
from the dump. Honours replication inclusion/exclusion rules.
db must be non-zero (guarded by assertion).
RETURN VALUES
0 success
1 error
*/
static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db,
const char* table_name, bool overwrite)
{
ulong packet_len;
char *query, *save_db;
uint32 save_db_length;
Vio* save_vio;
HA_CHECK_OPT check_opt;
TABLE_LIST tables;
int error= 1;
handler *file;
ulong save_options;
NET *net= &mysql->net;
DBUG_ENTER("create_table_from_dump");
packet_len= my_net_read(net); // read create table statement
if (packet_len == packet_error)
{
send_error(thd, ER_MASTER_NET_READ);
DBUG_RETURN(1);
}
if (net->read_pos[0] == 255) // error from master
{
char *err_msg;
err_msg= (char*) net->read_pos + ((mysql->server_capabilities &
CLIENT_PROTOCOL_41) ?
3+SQLSTATE_LENGTH+1 : 3);
net_printf(thd, ER_MASTER, err_msg);
DBUG_RETURN(1);
}
thd->command = COM_TABLE_DUMP;
thd->query_length= packet_len;
/* Note that we should not set thd->query until the area is initalized */
if (!(query = thd->strmake((char*) net->read_pos, packet_len)))
{
sql_print_error("create_table_from_dump: out of memory");
net_printf(thd, ER_GET_ERRNO, "Out of memory");
DBUG_RETURN(1);
}
thd->query= query;
thd->query_error = 0;
thd->net.no_send_ok = 1;
bzero((char*) &tables,sizeof(tables));
tables.db = (char*)db;
tables.alias= tables.real_name= (char*)table_name;
/* Drop the table if 'overwrite' is true */
if (overwrite && mysql_rm_table(thd,&tables,1,0)) /* drop if exists */
{
send_error(thd);
sql_print_error("create_table_from_dump: failed to drop the table");
goto err;
}
/* 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";
// save old db in case we are creating in a different database
save_db = thd->db;
save_db_length= thd->db_length;
thd->db = (char*)db;
DBUG_ASSERT(thd->db);
thd->db_length= strlen(thd->db);
mysql_parse(thd, thd->query, packet_len); // run create table
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";
tables.lock_type = TL_WRITE;
if (!open_ltable(thd, &tables, TL_WRITE))
{
send_error(thd,0,0); // Send error from open_ltable
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";
/* Copy the data file */
if (file->net_read_dump(net))
{
net_printf(thd, ER_MASTER_NET_READ);
sql_print_error("create_table_from_dump: failed in
handler::net_read_dump()");
goto err;
}
check_opt.init();
check_opt.flags|= T_VERY_SILENT | T_CALC_CHECKSUM | T_QUICK;
thd->proc_info = "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
problems.
*/
save_vio = thd->net.vio;
thd->net.vio = 0;
/* Rebuild the index file from the copied data file (with REPAIR) */
error=file->repair(thd,&check_opt) != 0;
thd->net.vio = save_vio;
if (error)
net_printf(thd, ER_INDEX_REBUILD,tables.table->real_name);
err:
close_thread_tables(thd);
thd->net.no_send_ok = 0;
DBUG_RETURN(error);
}
int fetch_master_table(THD *thd, const char *db_name, const char *table_name,
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));
if (!called_connected)
{
if (!(mysql = mysql_init(NULL)))
{
send_error(thd); // EOM
DBUG_RETURN(1);
}
if (connect_to_master(thd, mysql, mi))
{
net_printf(thd, ER_CONNECT_TO_MASTER, mysql_error(mysql));
mysql_close(mysql);
DBUG_RETURN(1);
}
if (thd->killed)
goto err;
}
if (request_table_dump(mysql, db_name, table_name))
{
error= ER_UNKNOWN_ERROR;
errmsg= "Failed on table dump request";
goto err;
}
if (create_table_from_dump(thd, mysql, db_name,
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())
send_error(thd, error, errmsg);
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;
}
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 = 0;
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;
// TODO: make this work with multi-master
if (!opt_relay_logname)
{
char tmp[FN_REFLEN];
/*
TODO: The following should be using fn_format(); We just need to
first change fn_format() to cut the file name if it's too long.
*/
strmake(tmp,glob_hostname,FN_REFLEN-5);
strmov(strcend(tmp,'.'),"-relay-bin");
opt_relay_logname=my_strdup(tmp,MYF(MY_WME));
}
/*
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).
*/
if (open_log(&rli->relay_log, glob_hostname, opt_relay_logname,
"-relay-bin", opt_relaylog_index_name,
LOG_BIN, 1 /* read_append cache */,
1 /* no auto events */,
max_relay_log_size ? max_relay_log_size : max_binlog_size))
{
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))
{
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
{
if (info_fd >= 0)
reinit_io_cache(&rli->info_file, READ_CACHE, 0L,0,0);
else
{
int error=0;
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))
{
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;
}
}
DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
/*
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);
}
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)
{
bool slave_killed=0;
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,
"
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);
thd->exit_cond(save_proc_info);
DBUG_RETURN(slave_killed);
}
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.
SYNOPSIS
write_ignored_events_info_to_relay_log()
thd pointer to I/O thread's thd
mi
DESCRIPTION
Slave I/O thread, going to die, must leave a durable trace of the
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)
{
RELAY_LOG_INFO *rli= &mi->rli;
pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
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,
0, rli->ign_master_log_pos_end,
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);
if (likely((bool)ev))
{
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");
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
flush_master_info(mi, 1);
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");
}
else
pthread_mutex_unlock(log_lock);
}
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]!='' &&
*first_non_digit=='' && 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_error("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)
{
char buf[1024], *pos= buf;
uint report_host_len, report_user_len=0, report_password_len=0;
if (!report_host)
return 0;
report_host_len= strlen(report_host);
if (report_user)
report_user_len= strlen(report_user);
if (report_password)
report_password_len= strlen(report_password);
/* 30 is a good safety margin */
if (report_host_len + report_user_len + report_password_len + 30 >
sizeof(buf))
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);
int2store(pos, (uint16) report_port); pos+= 2;
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;
}
/*
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++)
{
TABLE_RULE_ENT* e= (TABLE_RULE_ENT*) hash_element(h, i);
if (s->length())
s->append(',');
s->append(e->db,e->key_len);
}
}
/*
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);
}
}
int show_master_info(THD* thd, MASTER_INFO* mi)
{
// TODO: fix this for multi-master
List<Item> field_list;
Protocol *protocol= thd->protocol;
DBUG_ENTER("show_master_info");
field_list.push_back(new Item_empty_string("Slave_IO_State",
14));
field_list.push_back(new Item_empty_string("Master_Host",
sizeof(mi->host)));
field_list.push_back(new Item_empty_string("Master_User",
sizeof(mi->user)));
field_list.push_back(new Item_return_int("Master_Port", 7,
MYSQL_TYPE_LONG));
field_list.push_back(new Item_return_int("Connect_Retry", 10,
MYSQL_TYPE_LONG));
field_list.push_back(new Item_empty_string("Master_Log_File",
FN_REFLEN));
field_list.push_back(new Item_return_int("Read_Master_Log_Pos", 10,
MYSQL_TYPE_LONGLONG));
field_list.push_back(new Item_empty_string("Relay_Log_File",
FN_REFLEN));
field_list.push_back(new Item_return_int("Relay_Log_Pos", 10,
MYSQL_TYPE_LONGLONG));
field_list.push_back(new Item_empty_string("Relay_Master_Log_File",
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));
field_list.push_back(new Item_empty_string("Replicate_Ignore_DB", 20));
field_list.push_back(new Item_empty_string("Replicate_Do_Table", 20));
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));
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));
field_list.push_back(new Item_return_int("Exec_Master_Log_Pos", 10,
MYSQL_TYPE_LONGLONG));
field_list.push_back(new Item_return_int("Relay_Log_Space", 10,
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,
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",
sizeof(mi->ssl_capath)));
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",
sizeof(mi->ssl_cipher)));
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));
if (protocol->send_fields(&field_list, 1))
DBUG_RETURN(-1);
if (mi->host[0])
{
DBUG_PRINT("info",("host is set: '%s'", mi->host));
String *packet= &thd->packet;
protocol->prepare_for_resend();
/*
TODO: we read slave_running without run_lock, whereas these variables
are updated under run_lock and not data_lock. In 5.0 we should lock
run_lock on top of data_lock (with good order).
*/
pthread_mutex_lock(&mi->data_lock);
pthread_mutex_lock(&mi->rli.data_lock);
protocol->store(mi->io_thd ? mi->io_thd->proc_info : "", &my_charset_bin);
protocol->store(mi->host, &my_charset_bin);
protocol->store(mi->user, &my_charset_bin);
protocol->store((uint32) mi->port);
protocol->store((uint32) mi->connect_retry);
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);
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.
*/
char buf[256];
String tmp(buf, sizeof(buf), &my_charset_bin);
table_rule_ent_hash_to_str(&tmp, &replicate_do_table);
protocol->store(&tmp);
table_rule_ent_hash_to_str(&tmp, &replicate_ignore_table);
protocol->store(&tmp);
table_rule_ent_dynamic_array_to_str(&tmp, &replicate_wild_do_table);
protocol->store(&tmp);
table_rule_ent_dynamic_array_to_str(&tmp, &replicate_wild_ignore_table);
protocol->store(&tmp);
protocol->store((uint32) mi->rli.last_slave_errno);
protocol->store(mi->rli.last_slave_error, &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":
"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
protocol->store(mi->ssl? "Yes":"No", &my_charset_bin);
#else
protocol->store(mi->ssl? "Ignored":"No", &my_charset_bin);
#endif
protocol->store(mi->ssl_ca, &my_charset_bin);
protocol->store(mi->ssl_capath, &my_charset_bin);
protocol->store(mi->ssl_cert, &my_charset_bin);
protocol->store(mi->ssl_cipher, &my_charset_bin);
protocol->store(mi->ssl_key, &my_charset_bin);
/*
Seconds_Behind_Master: if SQL thread is running and I/O thread is
connected, we can compute it otherwise show NULL (i.e. unknown).
*/
if ((mi->slave_running == MYSQL_SLAVE_RUN_CONNECT) &&
mi->rli.slave_running)
{
long tmp= (long)((time_t)time((time_t*) 0)
- mi->rli.last_master_timestamp)
- mi->clock_diff_with_master;
/*
Apparently on some systems tmp can be <0. Here are possible reasons
related to MySQL:
- the master is itself a slave of another master whose time is ahead.
- somebody used an explicit SET TIMESTAMP on the master.
Possible reason related to granularity-to-second of time functions
(nothing to do with MySQL), which can explain a value of -1:
assume the master's and slave's time are perfectly synchronized, and
that at slave's connection time, when the master's timestamp is read,
it is at the very end of second 1, and (a very short time later) when
the slave's timestamp is read it is at the very beginning of second