bug fixes in server-id, moved replication functionality to
sql_repl.* will push this one, since the replication code now works
This commit is contained in:
parent
079d859072
commit
62a6a0fdf1
@ -1,4 +1,11 @@
|
|||||||
source ../include/master-slave.inc;
|
source ../include/master-slave.inc;
|
||||||
|
connection slave;
|
||||||
|
!slave stop;
|
||||||
|
flush slave;
|
||||||
|
connection master;
|
||||||
|
flush master;
|
||||||
|
connection slave;
|
||||||
|
slave start;
|
||||||
connection master;
|
connection master;
|
||||||
use test;
|
use test;
|
||||||
drop table if exists words;
|
drop table if exists words;
|
||||||
@ -7,6 +14,7 @@ load data infile '/usr/dict/words' into table words;
|
|||||||
drop table if exists words1;
|
drop table if exists words1;
|
||||||
create table words1 (word char(20) not null);
|
create table words1 (word char(20) not null);
|
||||||
load data infile '/usr/dict/words' into table words1;
|
load data infile '/usr/dict/words' into table words1;
|
||||||
|
sleep 5;
|
||||||
connection slave;
|
connection slave;
|
||||||
use test;
|
use test;
|
||||||
drop table if exists words;
|
drop table if exists words;
|
||||||
|
@ -48,7 +48,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \
|
|||||||
opt_range.h \
|
opt_range.h \
|
||||||
sql_select.h structs.h table.h sql_udf.h hash_filo.h\
|
sql_select.h structs.h table.h sql_udf.h hash_filo.h\
|
||||||
lex.h lex_symbol.h sql_acl.h sql_crypt.h md5.h \
|
lex.h lex_symbol.h sql_acl.h sql_crypt.h md5.h \
|
||||||
log_event.h mini_client.h
|
log_event.h mini_client.h sql_repl.h
|
||||||
mysqld_SOURCES = sql_lex.cc \
|
mysqld_SOURCES = sql_lex.cc \
|
||||||
item.cc item_sum.cc item_buff.cc item_func.cc \
|
item.cc item_sum.cc item_buff.cc item_func.cc \
|
||||||
item_cmpfunc.cc item_strfunc.cc item_timefunc.cc \
|
item_cmpfunc.cc item_strfunc.cc item_timefunc.cc \
|
||||||
@ -69,7 +69,7 @@ mysqld_SOURCES = sql_lex.cc \
|
|||||||
sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
|
sql_db.cc sql_table.cc sql_rename.cc sql_crypt.cc \
|
||||||
sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
|
sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
|
||||||
sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
|
sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
|
||||||
slave.cc \
|
slave.cc sql_repl.cc \
|
||||||
md5.c log_event.cc mini_client.cc mini_client_errors.c
|
md5.c log_event.cc mini_client.cc mini_client_errors.c
|
||||||
gen_lex_hash_SOURCES = gen_lex_hash.cc
|
gen_lex_hash_SOURCES = gen_lex_hash.cc
|
||||||
gen_lex_hash_LDADD = $(LDADD) $(CXXLDFLAGS)
|
gen_lex_hash_LDADD = $(LDADD) $(CXXLDFLAGS)
|
||||||
|
@ -22,16 +22,6 @@
|
|||||||
#include "mysql_priv.h"
|
#include "mysql_priv.h"
|
||||||
#endif /* MYSQL_CLIENT */
|
#endif /* MYSQL_CLIENT */
|
||||||
|
|
||||||
#define LOG_EVENT_HEADER_LEN 13
|
|
||||||
#define QUERY_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + sizeof(uchar))
|
|
||||||
#define LOAD_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + \
|
|
||||||
+ sizeof(uint32) + 2 + sizeof(uint32))
|
|
||||||
#define EVENT_LEN_OFFSET 9
|
|
||||||
#define EVENT_TYPE_OFFSET 4
|
|
||||||
#define MAX_EVENT_LEN 4*1024*1024
|
|
||||||
#define QUERY_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN+QUERY_HEADER_LEN
|
|
||||||
#define ROTATE_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN
|
|
||||||
#define LOAD_EVENT_OVERHEAD (LOG_EVENT_HEADER_LEN+LOAD_HEADER_LEN+sizeof(sql_ex_info))
|
|
||||||
|
|
||||||
static void pretty_print_char(FILE* file, int c)
|
static void pretty_print_char(FILE* file, int c)
|
||||||
{
|
{
|
||||||
|
@ -30,6 +30,18 @@
|
|||||||
#define LOG_EVENT_OFFSET 4
|
#define LOG_EVENT_OFFSET 4
|
||||||
#define BINLOG_VERSION 1
|
#define BINLOG_VERSION 1
|
||||||
|
|
||||||
|
#define LOG_EVENT_HEADER_LEN 13
|
||||||
|
#define QUERY_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + sizeof(uchar))
|
||||||
|
#define LOAD_HEADER_LEN (sizeof(uint32) + sizeof(uint32) + \
|
||||||
|
+ sizeof(uint32) + 2 + sizeof(uint32))
|
||||||
|
#define EVENT_LEN_OFFSET 9
|
||||||
|
#define EVENT_TYPE_OFFSET 4
|
||||||
|
#define MAX_EVENT_LEN 4*1024*1024
|
||||||
|
#define QUERY_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN+QUERY_HEADER_LEN
|
||||||
|
#define ROTATE_EVENT_OVERHEAD LOG_EVENT_HEADER_LEN
|
||||||
|
#define LOAD_EVENT_OVERHEAD (LOG_EVENT_HEADER_LEN+LOAD_HEADER_LEN+sizeof(sql_ex_info))
|
||||||
|
|
||||||
|
|
||||||
enum Log_event_type { START_EVENT = 1, QUERY_EVENT =2,
|
enum Log_event_type { START_EVENT = 1, QUERY_EVENT =2,
|
||||||
STOP_EVENT=3, ROTATE_EVENT = 4, INTVAR_EVENT=5,
|
STOP_EVENT=3, ROTATE_EVENT = 4, INTVAR_EVENT=5,
|
||||||
LOAD_EVENT=6};
|
LOAD_EVENT=6};
|
||||||
@ -101,7 +113,7 @@ public:
|
|||||||
#if !defined(MYSQL_CLIENT)
|
#if !defined(MYSQL_CLIENT)
|
||||||
THD* thd;
|
THD* thd;
|
||||||
Query_log_event(THD* thd_arg, const char* query_arg):
|
Query_log_event(THD* thd_arg, const char* query_arg):
|
||||||
Log_event(thd_arg->start_time,0,0,thd->server_id), data_buf(0),
|
Log_event(thd_arg->start_time,0,0,thd_arg->server_id), data_buf(0),
|
||||||
query(query_arg), db(thd_arg->db), q_len(thd_arg->query_length),
|
query(query_arg), db(thd_arg->db), q_len(thd_arg->query_length),
|
||||||
thread_id(thd_arg->thread_id), thd(thd_arg)
|
thread_id(thd_arg->thread_id), thd(thd_arg)
|
||||||
{
|
{
|
||||||
|
@ -464,7 +464,7 @@ extern pthread_mutex_t LOCK_mysql_create_db,LOCK_Acl,LOCK_open,
|
|||||||
LOCK_thread_count,LOCK_mapped_file,LOCK_user_locks, LOCK_status,
|
LOCK_thread_count,LOCK_mapped_file,LOCK_user_locks, LOCK_status,
|
||||||
LOCK_grant, LOCK_error_log, LOCK_delayed_insert,
|
LOCK_grant, LOCK_error_log, LOCK_delayed_insert,
|
||||||
LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
|
LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
|
||||||
LOCK_binlog_update, LOCK_slave;
|
LOCK_binlog_update, LOCK_slave, LOCK_server_id;
|
||||||
extern pthread_cond_t COND_refresh,COND_thread_count, COND_binlog_update,
|
extern pthread_cond_t COND_refresh,COND_thread_count, COND_binlog_update,
|
||||||
COND_slave_stopped;
|
COND_slave_stopped;
|
||||||
extern pthread_attr_t connection_attrib;
|
extern pthread_attr_t connection_attrib;
|
||||||
|
@ -270,8 +270,9 @@ static void dump_remote_log_entries(const char* logname)
|
|||||||
int4store(buf, position);
|
int4store(buf, position);
|
||||||
int2store(buf + 4, binlog_flags);
|
int2store(buf + 4, binlog_flags);
|
||||||
len = (uint) strlen(logname);
|
len = (uint) strlen(logname);
|
||||||
memcpy(buf + 6, logname,len);
|
int4store(buf + 6, 0);
|
||||||
if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 6, 1))
|
memcpy(buf + 10, logname,len);
|
||||||
|
if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1))
|
||||||
die("Error sending the log dump command");
|
die("Error sending the log dump command");
|
||||||
|
|
||||||
for(;;)
|
for(;;)
|
||||||
|
@ -248,7 +248,7 @@ pthread_mutex_t LOCK_mysql_create_db, LOCK_Acl, LOCK_open, LOCK_thread_count,
|
|||||||
LOCK_error_log,
|
LOCK_error_log,
|
||||||
LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
|
LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
|
||||||
LOCK_flush, LOCK_crypt, LOCK_bytes_sent, LOCK_bytes_received,
|
LOCK_flush, LOCK_crypt, LOCK_bytes_sent, LOCK_bytes_received,
|
||||||
LOCK_binlog_update, LOCK_slave;
|
LOCK_binlog_update, LOCK_slave, LOCK_server_id;
|
||||||
|
|
||||||
pthread_cond_t COND_refresh,COND_thread_count,COND_flush, COND_binlog_update,
|
pthread_cond_t COND_refresh,COND_thread_count,COND_flush, COND_binlog_update,
|
||||||
COND_slave_stopped;
|
COND_slave_stopped;
|
||||||
@ -1395,6 +1395,7 @@ int main(int argc, char **argv)
|
|||||||
(void) pthread_mutex_init(&LOCK_timezone,NULL);
|
(void) pthread_mutex_init(&LOCK_timezone,NULL);
|
||||||
(void) pthread_mutex_init(&LOCK_binlog_update, NULL);
|
(void) pthread_mutex_init(&LOCK_binlog_update, NULL);
|
||||||
(void) pthread_mutex_init(&LOCK_slave, NULL);
|
(void) pthread_mutex_init(&LOCK_slave, NULL);
|
||||||
|
(void) pthread_mutex_init(&LOCK_server_id, NULL);
|
||||||
(void) pthread_cond_init(&COND_binlog_update, NULL);
|
(void) pthread_cond_init(&COND_binlog_update, NULL);
|
||||||
(void) pthread_cond_init(&COND_slave_stopped, NULL);
|
(void) pthread_cond_init(&COND_slave_stopped, NULL);
|
||||||
|
|
||||||
|
@ -463,15 +463,16 @@ static int safe_sleep(THD* thd, int sec)
|
|||||||
|
|
||||||
static int request_dump(MYSQL* mysql, MASTER_INFO* mi)
|
static int request_dump(MYSQL* mysql, MASTER_INFO* mi)
|
||||||
{
|
{
|
||||||
char buf[FN_REFLEN + 6];
|
char buf[FN_REFLEN + 10];
|
||||||
int len;
|
int len;
|
||||||
int binlog_flags = 0; // for now
|
int binlog_flags = 0; // for now
|
||||||
char* logname = mi->log_file_name;
|
char* logname = mi->log_file_name;
|
||||||
int4store(buf, mi->pos);
|
int4store(buf, mi->pos);
|
||||||
int2store(buf + 4, binlog_flags);
|
int2store(buf + 4, binlog_flags);
|
||||||
|
int4store(buf + 6, server_id);
|
||||||
len = (uint) strlen(logname);
|
len = (uint) strlen(logname);
|
||||||
memcpy(buf + 6, logname,len);
|
memcpy(buf + 10, logname,len);
|
||||||
if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 6, 1))
|
if(mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1))
|
||||||
// something went wrong, so we will just reconnect and retry later
|
// something went wrong, so we will just reconnect and retry later
|
||||||
// in the future, we should do a better error analysis, but for
|
// in the future, we should do a better error analysis, but for
|
||||||
// now we just fill up the error log :-)
|
// now we just fill up the error log :-)
|
||||||
|
287
sql/sql_base.cc
287
sql/sql_base.cc
@ -42,71 +42,6 @@ static void mysql_rm_tmp_tables(void);
|
|||||||
static key_map get_key_map_from_key_list(THD *thd, TABLE *table,
|
static key_map get_key_map_from_key_list(THD *thd, TABLE *table,
|
||||||
List<String> *index_list);
|
List<String> *index_list);
|
||||||
|
|
||||||
static int send_file(THD *thd)
|
|
||||||
{
|
|
||||||
NET* net = &thd->net;
|
|
||||||
int fd = -1,bytes, error = 1;
|
|
||||||
char fname[FN_REFLEN+1];
|
|
||||||
char buf[IO_SIZE*15];
|
|
||||||
const char *errmsg = 0;
|
|
||||||
int old_timeout;
|
|
||||||
DBUG_ENTER("send_file");
|
|
||||||
|
|
||||||
// the client might be slow loading the data, give him wait_timeout to do
|
|
||||||
// the job
|
|
||||||
old_timeout = thd->net.timeout;
|
|
||||||
thd->net.timeout = thd->inactive_timeout;
|
|
||||||
|
|
||||||
// we need net_flush here because the client will not know it needs to send
|
|
||||||
// us the file name until it has processed the load event entry
|
|
||||||
if (net_flush(net) || my_net_read(net) == packet_error)
|
|
||||||
{
|
|
||||||
errmsg = "Failed reading file name";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
fn_format(fname, (char*)net->read_pos + 1, "", "", 4);
|
|
||||||
// this is needed to make replicate-ignore-db
|
|
||||||
if (!strcmp(fname,"/dev/null"))
|
|
||||||
goto end;
|
|
||||||
// TODO: work on the well-known system that does not have a /dev/null :-)
|
|
||||||
|
|
||||||
if ((fd = my_open(fname, O_RDONLY, MYF(MY_WME))) < 0)
|
|
||||||
{
|
|
||||||
errmsg = "Failed on my_open()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
while ((bytes = (int) my_read(fd, (byte*) buf, sizeof(buf),
|
|
||||||
MYF(MY_WME))) > 0)
|
|
||||||
{
|
|
||||||
if (my_net_write(net, buf, bytes))
|
|
||||||
{
|
|
||||||
errmsg = "Failed on my_net_write()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
end:
|
|
||||||
if (my_net_write(net, "", 0) || net_flush(net) ||
|
|
||||||
(my_net_read(net) == packet_error))
|
|
||||||
{
|
|
||||||
errmsg = "failed negotiating file transfer close";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
error = 0;
|
|
||||||
|
|
||||||
err:
|
|
||||||
thd->net.timeout = old_timeout;
|
|
||||||
if(fd >= 0)
|
|
||||||
(void) my_close(fd, MYF(MY_WME));
|
|
||||||
if (errmsg)
|
|
||||||
{
|
|
||||||
sql_print_error("failed in send_file() : %s", errmsg);
|
|
||||||
DBUG_PRINT("error", (errmsg));
|
|
||||||
}
|
|
||||||
DBUG_RETURN(error);
|
|
||||||
}
|
|
||||||
|
|
||||||
static byte *cache_key(const byte *record,uint *length,
|
static byte *cache_key(const byte *record,uint *length,
|
||||||
my_bool not_used __attribute__((unused)))
|
my_bool not_used __attribute__((unused)))
|
||||||
@ -177,228 +112,6 @@ static void check_unused(void)
|
|||||||
#define check_unused()
|
#define check_unused()
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags)
|
|
||||||
{
|
|
||||||
LOG_INFO linfo;
|
|
||||||
char *log_file_name = linfo.log_file_name;
|
|
||||||
char search_file_name[FN_REFLEN];
|
|
||||||
FILE* log = NULL;
|
|
||||||
String* packet = &thd->packet;
|
|
||||||
int error;
|
|
||||||
const char *errmsg = "Unknown error";
|
|
||||||
NET* net = &thd->net;
|
|
||||||
|
|
||||||
DBUG_ENTER("mysql_binlog_send");
|
|
||||||
|
|
||||||
if(!mysql_bin_log.is_open())
|
|
||||||
{
|
|
||||||
errmsg = "Binary log is not open";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
if(log_ident[0])
|
|
||||||
mysql_bin_log.make_log_name(search_file_name, log_ident);
|
|
||||||
else
|
|
||||||
search_file_name[0] = 0;
|
|
||||||
|
|
||||||
if(mysql_bin_log.find_first_log(&linfo, search_file_name))
|
|
||||||
{
|
|
||||||
errmsg = "Could not find first log";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
log = my_fopen(log_file_name, O_RDONLY, MYF(MY_WME));
|
|
||||||
|
|
||||||
if(!log)
|
|
||||||
{
|
|
||||||
errmsg = "Could not open log file";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
if(my_fseek(log, pos, MY_SEEK_SET, MYF(MY_WME)) == MY_FILEPOS_ERROR )
|
|
||||||
{
|
|
||||||
errmsg = "Error on fseek()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
packet->length(0);
|
|
||||||
packet->append("\0", 1); // we need to start a packet with something other than 255
|
|
||||||
// to distiquish it from error
|
|
||||||
|
|
||||||
while(!net->error && net->vio != 0 && !thd->killed)
|
|
||||||
{
|
|
||||||
while(!(error = Log_event::read_log_event(log, packet)))
|
|
||||||
{
|
|
||||||
if(my_net_write(net, (char*)packet->ptr(), packet->length()) )
|
|
||||||
{
|
|
||||||
errmsg = "Failed on my_net_write()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
DBUG_PRINT("info", ("log event code %d",(*packet)[LOG_EVENT_OFFSET+1] ));
|
|
||||||
if((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
|
|
||||||
{
|
|
||||||
if(send_file(thd))
|
|
||||||
{
|
|
||||||
errmsg = "failed in send_file()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
packet->length(0);
|
|
||||||
packet->append("\0",1);
|
|
||||||
}
|
|
||||||
if(error != LOG_READ_EOF)
|
|
||||||
{
|
|
||||||
errmsg = "error reading log event";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
if(!(flags & BINLOG_DUMP_NON_BLOCK) && mysql_bin_log.is_active(log_file_name))
|
|
||||||
// block until there is more data in the log
|
|
||||||
// unless non-blocking mode requested
|
|
||||||
{
|
|
||||||
if(net_flush(net))
|
|
||||||
{
|
|
||||||
errmsg = "failed on net_flush()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
// we may have missed the update broadcast from the log
|
|
||||||
// that has just happened, let's try to catch it if it did
|
|
||||||
// if we did not miss anything, we just wait for other threads
|
|
||||||
// to signal us
|
|
||||||
{
|
|
||||||
pthread_mutex_t *log_lock = mysql_bin_log.get_log_lock();
|
|
||||||
clearerr(log);
|
|
||||||
|
|
||||||
// tell the kill thread how to wake us up
|
|
||||||
pthread_mutex_lock(&thd->mysys_var->mutex);
|
|
||||||
thd->mysys_var->current_mutex = log_lock;
|
|
||||||
thd->mysys_var->current_cond = &COND_binlog_update;
|
|
||||||
const char* proc_info = thd->proc_info;
|
|
||||||
thd->proc_info = "Waiting for update";
|
|
||||||
pthread_mutex_unlock(&thd->mysys_var->mutex);
|
|
||||||
|
|
||||||
bool read_packet = 0, fatal_error = 0;
|
|
||||||
|
|
||||||
pthread_mutex_lock(log_lock); // no one will update the log while we are reading
|
|
||||||
// now, but we'll be quick and just read one record
|
|
||||||
|
|
||||||
|
|
||||||
switch(Log_event::read_log_event(log, packet))
|
|
||||||
{
|
|
||||||
case 0:
|
|
||||||
read_packet = 1; // we read successfully, so we'll need to send it to the
|
|
||||||
// slave
|
|
||||||
break;
|
|
||||||
case LOG_READ_EOF:
|
|
||||||
pthread_cond_wait(&COND_binlog_update, log_lock);
|
|
||||||
break;
|
|
||||||
|
|
||||||
default:
|
|
||||||
fatal_error = 1;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
pthread_mutex_unlock(log_lock);
|
|
||||||
|
|
||||||
pthread_mutex_lock(&thd->mysys_var->mutex);
|
|
||||||
thd->mysys_var->current_mutex= 0;
|
|
||||||
thd->mysys_var->current_cond= 0;
|
|
||||||
thd->proc_info= proc_info;
|
|
||||||
pthread_mutex_unlock(&thd->mysys_var->mutex);
|
|
||||||
|
|
||||||
if(read_packet)
|
|
||||||
{
|
|
||||||
if(my_net_write(net, (char*)packet->ptr(), packet->length()) )
|
|
||||||
{
|
|
||||||
errmsg = "Failed on my_net_write()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
if((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
|
|
||||||
{
|
|
||||||
if(send_file(thd))
|
|
||||||
{
|
|
||||||
errmsg = "failed in send_file()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
packet->length(0);
|
|
||||||
packet->append("\0",1);
|
|
||||||
// no need to net_flush because we will get to flush later when
|
|
||||||
// we hit EOF pretty quick
|
|
||||||
}
|
|
||||||
|
|
||||||
if(fatal_error)
|
|
||||||
{
|
|
||||||
errmsg = "error reading log entry";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
clearerr(log);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
bool loop_breaker = 0; // need this to break out of the for loop from switch
|
|
||||||
|
|
||||||
switch(mysql_bin_log.find_next_log(&linfo))
|
|
||||||
{
|
|
||||||
case LOG_INFO_EOF:
|
|
||||||
loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
|
|
||||||
break;
|
|
||||||
case 0:
|
|
||||||
break;
|
|
||||||
default:
|
|
||||||
errmsg = "could not find next log";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
|
|
||||||
if(loop_breaker)
|
|
||||||
break;
|
|
||||||
|
|
||||||
(void) my_fclose(log, MYF(MY_WME));
|
|
||||||
log = my_fopen(log_file_name, O_RDONLY, MYF(MY_WME));
|
|
||||||
if(!log)
|
|
||||||
goto err;
|
|
||||||
// fake Rotate_log event just in case it did not make it to the log
|
|
||||||
// otherwise the slave make get confused about the offset
|
|
||||||
{
|
|
||||||
char header[9];
|
|
||||||
memset(header, 0, 4); // when does not matter
|
|
||||||
header[4] = ROTATE_EVENT;
|
|
||||||
char* p = strrchr(log_file_name, FN_LIBCHAR); // find the last slash
|
|
||||||
if(p)
|
|
||||||
p++;
|
|
||||||
else
|
|
||||||
p = log_file_name;
|
|
||||||
|
|
||||||
uint ident_len = (uint) strlen(p);
|
|
||||||
ulong event_len = ident_len + sizeof(header);
|
|
||||||
int4store(header + 5, event_len);
|
|
||||||
packet->append(header, sizeof(header));
|
|
||||||
packet->append(p,ident_len);
|
|
||||||
if(my_net_write(net, (char*)packet->ptr(), packet->length()))
|
|
||||||
{
|
|
||||||
errmsg = "failed on my_net_write()";
|
|
||||||
goto err;
|
|
||||||
}
|
|
||||||
packet->length(0);
|
|
||||||
packet->append("\0",1);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
(void)my_fclose(log, MYF(MY_WME));
|
|
||||||
|
|
||||||
send_eof(&thd->net);
|
|
||||||
DBUG_VOID_RETURN;
|
|
||||||
err:
|
|
||||||
if(log)
|
|
||||||
(void) my_fclose(log, MYF(MY_WME));
|
|
||||||
send_error(&thd->net, 0, errmsg);
|
|
||||||
DBUG_VOID_RETURN;
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
/******************************************************************************
|
/******************************************************************************
|
||||||
|
228
sql/sql_parse.cc
228
sql/sql_parse.cc
@ -17,6 +17,7 @@
|
|||||||
|
|
||||||
#include "mysql_priv.h"
|
#include "mysql_priv.h"
|
||||||
#include "sql_acl.h"
|
#include "sql_acl.h"
|
||||||
|
#include "sql_repl.h"
|
||||||
#include <m_ctype.h>
|
#include <m_ctype.h>
|
||||||
#include <thr_alarm.h>
|
#include <thr_alarm.h>
|
||||||
#include <myisam.h>
|
#include <myisam.h>
|
||||||
@ -24,13 +25,6 @@
|
|||||||
|
|
||||||
#define SCRAMBLE_LENGTH 8
|
#define SCRAMBLE_LENGTH 8
|
||||||
|
|
||||||
extern pthread_handler_decl(handle_slave,arg);
|
|
||||||
extern bool slave_running;
|
|
||||||
extern char* master_host;
|
|
||||||
extern pthread_t slave_real_id;
|
|
||||||
extern MASTER_INFO glob_mi;
|
|
||||||
extern my_string opt_bin_logname, master_info_file;
|
|
||||||
extern I_List<i_string> binlog_do_db, binlog_ignore_db;
|
|
||||||
|
|
||||||
extern int yyparse(void);
|
extern int yyparse(void);
|
||||||
extern "C" pthread_mutex_t THR_LOCK_keycache;
|
extern "C" pthread_mutex_t THR_LOCK_keycache;
|
||||||
@ -47,15 +41,8 @@ static void mysql_init_query(THD *thd);
|
|||||||
static void remove_escape(char *name);
|
static void remove_escape(char *name);
|
||||||
static void kill_one_thread(THD *thd, ulong thread);
|
static void kill_one_thread(THD *thd, ulong thread);
|
||||||
static void refresh_status(void);
|
static void refresh_status(void);
|
||||||
static int start_slave(THD* thd = 0, bool net_report = 1);
|
|
||||||
static int stop_slave(THD* thd = 0, bool net_report = 1);
|
|
||||||
static int change_master(THD* thd);
|
|
||||||
static void reset_slave();
|
|
||||||
static void reset_master();
|
|
||||||
|
|
||||||
extern int init_master_info(MASTER_INFO* mi);
|
const char *any_db="*any*"; // Special symbol for check_access
|
||||||
|
|
||||||
static const char *any_db="*any*"; // Special symbol for check_access
|
|
||||||
|
|
||||||
const char *command_name[]={
|
const char *command_name[]={
|
||||||
"Sleep", "Quit", "Init DB", "Query", "Field List", "Create DB",
|
"Sleep", "Quit", "Init DB", "Query", "Field List", "Create DB",
|
||||||
@ -717,9 +704,14 @@ bool do_command(THD *thd)
|
|||||||
|
|
||||||
ulong pos;
|
ulong pos;
|
||||||
ushort flags;
|
ushort flags;
|
||||||
|
uint32 slave_server_id;
|
||||||
pos = uint4korr(packet + 1);
|
pos = uint4korr(packet + 1);
|
||||||
flags = uint2korr(packet + 5);
|
flags = uint2korr(packet + 5);
|
||||||
mysql_binlog_send(thd, thd->strdup(packet + 7), pos, flags);
|
pthread_mutex_lock(&LOCK_server_id);
|
||||||
|
kill_zombie_dump_threads(slave_server_id = uint4korr(packet+7));
|
||||||
|
thd->server_id = slave_server_id;
|
||||||
|
pthread_mutex_unlock(&LOCK_server_id);
|
||||||
|
mysql_binlog_send(thd, strdup(packet + 11), pos, flags);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case COM_REFRESH:
|
case COM_REFRESH:
|
||||||
@ -2516,207 +2508,3 @@ static void refresh_status(void)
|
|||||||
pthread_mutex_unlock(&THR_LOCK_keycache);
|
pthread_mutex_unlock(&THR_LOCK_keycache);
|
||||||
}
|
}
|
||||||
|
|
||||||
static int start_slave(THD* thd , bool net_report)
|
|
||||||
{
|
|
||||||
if(!thd) thd = current_thd;
|
|
||||||
NET* net = &thd->net;
|
|
||||||
const char* err = 0;
|
|
||||||
if (check_access(thd, PROCESS_ACL, any_db))
|
|
||||||
return 1;
|
|
||||||
pthread_mutex_lock(&LOCK_slave);
|
|
||||||
if(!slave_running)
|
|
||||||
if(glob_mi.inited && glob_mi.host)
|
|
||||||
{
|
|
||||||
pthread_t hThread;
|
|
||||||
if(pthread_create(&hThread, &connection_attrib, handle_slave, 0))
|
|
||||||
{
|
|
||||||
err = "cannot create slave thread";
|
|
||||||
}
|
|
||||||
}
|
|
||||||
else
|
|
||||||
err = "Master host not set or master info not initialized";
|
|
||||||
else
|
|
||||||
err = "Slave already running";
|
|
||||||
|
|
||||||
pthread_mutex_unlock(&LOCK_slave);
|
|
||||||
if(err)
|
|
||||||
{
|
|
||||||
if(net_report) send_error(net, 0, err);
|
|
||||||
return 1;
|
|
||||||
}
|
|
||||||
else if(net_report)
|
|
||||||
send_ok(net);
|
|
||||||
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
static int stop_slave(THD* thd, bool net_report )
|
|
||||||
{
|
|
||||||
if(!thd) thd = current_thd;
|
|
||||||
NET* net = &thd->net;
|
|
||||||
const char* err = 0;
|
|
||||||
|
|
||||||
if (check_access(thd, PROCESS_ACL, any_db))
|
|
||||||
return 1;
|
|
||||||
|
|
||||||
pthread_mutex_lock(&LOCK_slave);
|
|
||||||
if (slave_running)
|
|
||||||
{
|
|
||||||
abort_slave = 1;
|
|
||||||
thr_alarm_kill(slave_real_id);
|
|
||||||
// do not abort the slave in the middle of a query, so we do not set
|
|
||||||
// thd->killed for the slave thread
|
|
||||||
thd->proc_info = "waiting for slave to die";
|
|
||||||
pthread_cond_wait(&COND_slave_stopped, &LOCK_slave);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
err = "Slave is not running";
|
|
||||||
|
|
||||||
pthread_mutex_unlock(&LOCK_slave);
|
|
||||||
thd->proc_info = 0;
|
|
||||||
|
|
||||||
if(err)
|
|
||||||
{
|
|
||||||
if(net_report) send_error(net, 0, err);
|
|
||||||
return 1;
|
|
||||||
}
|
|
||||||
else if(net_report)
|
|
||||||
send_ok(net);
|
|
||||||
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
static void reset_slave()
|
|
||||||
{
|
|
||||||
MY_STAT stat_area;
|
|
||||||
char fname[FN_REFLEN];
|
|
||||||
bool slave_was_running = slave_running;
|
|
||||||
|
|
||||||
if(slave_running)
|
|
||||||
stop_slave(0,0);
|
|
||||||
|
|
||||||
fn_format(fname, master_info_file, mysql_data_home, "", 4+16+32);
|
|
||||||
if(my_stat(fname, &stat_area, MYF(0)))
|
|
||||||
if(my_delete(fname, MYF(MY_WME)))
|
|
||||||
return;
|
|
||||||
|
|
||||||
if(slave_was_running)
|
|
||||||
start_slave(0,0);
|
|
||||||
}
|
|
||||||
|
|
||||||
static int change_master(THD* thd)
|
|
||||||
{
|
|
||||||
bool slave_was_running;
|
|
||||||
// kill slave thread
|
|
||||||
pthread_mutex_lock(&LOCK_slave);
|
|
||||||
if((slave_was_running = slave_running))
|
|
||||||
{
|
|
||||||
abort_slave = 1;
|
|
||||||
thr_alarm_kill(slave_real_id);
|
|
||||||
thd->proc_info = "waiting for slave to die";
|
|
||||||
pthread_cond_wait(&COND_slave_stopped, &LOCK_slave); // wait until done
|
|
||||||
}
|
|
||||||
pthread_mutex_unlock(&LOCK_slave);
|
|
||||||
thd->proc_info = "changing master";
|
|
||||||
LEX_MASTER_INFO* lex_mi = &thd->lex.mi;
|
|
||||||
|
|
||||||
if(!glob_mi.inited)
|
|
||||||
init_master_info(&glob_mi);
|
|
||||||
|
|
||||||
pthread_mutex_lock(&glob_mi.lock);
|
|
||||||
if((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
|
|
||||||
{
|
|
||||||
// if we change host or port, we must reset the postion
|
|
||||||
glob_mi.log_file_name[0] = 0;
|
|
||||||
glob_mi.pos = 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
if(lex_mi->log_file_name)
|
|
||||||
strmake(glob_mi.log_file_name, lex_mi->log_file_name,
|
|
||||||
sizeof(glob_mi.log_file_name));
|
|
||||||
if(lex_mi->pos)
|
|
||||||
glob_mi.pos = lex_mi->pos;
|
|
||||||
|
|
||||||
if(lex_mi->host)
|
|
||||||
strmake(glob_mi.host, lex_mi->host, sizeof(glob_mi.host));
|
|
||||||
if(lex_mi->user)
|
|
||||||
strmake(glob_mi.user, lex_mi->user, sizeof(glob_mi.user));
|
|
||||||
if(lex_mi->password)
|
|
||||||
strmake(glob_mi.password, lex_mi->password, sizeof(glob_mi.password));
|
|
||||||
if(lex_mi->port)
|
|
||||||
glob_mi.port = lex_mi->port;
|
|
||||||
if(lex_mi->connect_retry)
|
|
||||||
glob_mi.connect_retry = lex_mi->connect_retry;
|
|
||||||
|
|
||||||
flush_master_info(&glob_mi);
|
|
||||||
pthread_mutex_unlock(&glob_mi.lock);
|
|
||||||
thd->proc_info = "starting slave";
|
|
||||||
if(slave_was_running)
|
|
||||||
start_slave(0,0);
|
|
||||||
thd->proc_info = 0;
|
|
||||||
|
|
||||||
send_ok(&thd->net);
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
static void reset_master()
|
|
||||||
{
|
|
||||||
if(!mysql_bin_log.is_open())
|
|
||||||
{
|
|
||||||
my_error(ER_FLUSH_MASTER_BINLOG_CLOSED, MYF(ME_BELL+ME_WAITTANG));
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
LOG_INFO linfo;
|
|
||||||
if (mysql_bin_log.find_first_log(&linfo, ""))
|
|
||||||
return;
|
|
||||||
|
|
||||||
for(;;)
|
|
||||||
{
|
|
||||||
my_delete(linfo.log_file_name, MYF(MY_WME));
|
|
||||||
if (mysql_bin_log.find_next_log(&linfo))
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
mysql_bin_log.close(1); // exiting close
|
|
||||||
my_delete(mysql_bin_log.get_index_fname(), MYF(MY_WME));
|
|
||||||
mysql_bin_log.open(opt_bin_logname,LOG_BIN);
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
int show_binlog_info(THD* thd)
|
|
||||||
{
|
|
||||||
DBUG_ENTER("show_binlog_info");
|
|
||||||
List<Item> field_list;
|
|
||||||
field_list.push_back(new Item_empty_string("File", FN_REFLEN));
|
|
||||||
field_list.push_back(new Item_empty_string("Position",20));
|
|
||||||
field_list.push_back(new Item_empty_string("Binlog_do_db",20));
|
|
||||||
field_list.push_back(new Item_empty_string("Binlog_ignore_db",20));
|
|
||||||
|
|
||||||
if(send_fields(thd, field_list, 1))
|
|
||||||
DBUG_RETURN(-1);
|
|
||||||
String* packet = &thd->packet;
|
|
||||||
packet->length(0);
|
|
||||||
|
|
||||||
if(mysql_bin_log.is_open())
|
|
||||||
{
|
|
||||||
LOG_INFO li;
|
|
||||||
mysql_bin_log.get_current_log(&li);
|
|
||||||
net_store_data(packet, li.log_file_name);
|
|
||||||
net_store_data(packet, (longlong)li.pos);
|
|
||||||
net_store_data(packet, &binlog_do_db);
|
|
||||||
net_store_data(packet, &binlog_ignore_db);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
net_store_null(packet);
|
|
||||||
net_store_null(packet);
|
|
||||||
net_store_null(packet);
|
|
||||||
net_store_null(packet);
|
|
||||||
}
|
|
||||||
|
|
||||||
if(my_net_write(&thd->net, (char*)thd->packet.ptr(), packet->length()))
|
|
||||||
DBUG_RETURN(-1);
|
|
||||||
|
|
||||||
send_eof(&thd->net);
|
|
||||||
DBUG_RETURN(0);
|
|
||||||
}
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user