Skip to content

M
MariaDB
Commit ce79dae2 authored by unknown's avatar unknown
Browse files
Options

Merge work:/home/bk/mysql 

into mysql.sashanet.com:/home/sasha/src/bk/mysql


sql/slave.cc:
  Auto merged
sql/sql_class.cc:
  Auto merged
parents 6578b60b 156e31aa
Hide whitespace changes
Inline Side-by-side
Showing with 209 additions and 587 deletions
repl-tests/test-dump/run.test
View file @ ce79dae2
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; ...@@ -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;
......
sql/Makefile.am
View file @ ce79dae2
...@@ -48,7 +48,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \ ...@@ -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 \ ...@@ -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)
......
sql/log_event.cc
View file @ ce79dae2
...@@ -22,16 +22,6 @@ ...@@ -22,16 +22,6 @@
#include "mysql_priv.h" #include "mysql_priv.h"
#endif /* MYSQL_CLIENT */ #endif /* MYSQL_CLIENT */
#define LOG_EVENT_HEADER_LEN 9
#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 5
#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)
{ {
...@@ -71,6 +61,8 @@ int Log_event::write_header(FILE* file) ...@@ -71,6 +61,8 @@ int Log_event::write_header(FILE* file)
int4store(pos, when); // timestamp int4store(pos, when); // timestamp
pos += 4; pos += 4;
*pos++ = get_type_code(); // event type code *pos++ = get_type_code(); // event type code
int4store(pos, server_id);
pos += 4;
int4store(pos, get_data_size() + LOG_EVENT_HEADER_LEN); int4store(pos, get_data_size() + LOG_EVENT_HEADER_LEN);
pos += 4; pos += 4;
return (my_fwrite(file, (byte*) buf, (uint) (pos - buf), return (my_fwrite(file, (byte*) buf, (uint) (pos - buf),
...@@ -106,16 +98,19 @@ int Log_event::read_log_event(FILE* file, String* packet) ...@@ -106,16 +98,19 @@ int Log_event::read_log_event(FILE* file, String* packet)
Log_event* Log_event::read_log_event(FILE* file) Log_event* Log_event::read_log_event(FILE* file)
{ {
time_t timestamp; time_t timestamp;
char buf[5]; uint32 server_id;
char buf[LOG_EVENT_HEADER_LEN-4];
if (my_fread(file, (byte *) buf, sizeof(buf), MY_NABP)) if (my_fread(file, (byte *) buf, sizeof(buf), MY_NABP))
return NULL; return NULL;
timestamp = uint4korr(buf); timestamp = uint4korr(buf);
server_id = uint4korr(buf + 5);
switch(buf[EVENT_TYPE_OFFSET]) switch(buf[EVENT_TYPE_OFFSET])
{ {
case QUERY_EVENT: case QUERY_EVENT:
{ {
Query_log_event* q = new Query_log_event(file, timestamp); Query_log_event* q = new Query_log_event(file, timestamp, server_id);
if (!q->query) if (!q->query)
{ {
delete q; delete q;
...@@ -127,7 +122,7 @@ Log_event* Log_event::read_log_event(FILE* file) ...@@ -127,7 +122,7 @@ Log_event* Log_event::read_log_event(FILE* file)
case LOAD_EVENT: case LOAD_EVENT:
{ {
Load_log_event* l = new Load_log_event(file, timestamp); Load_log_event* l = new Load_log_event(file, timestamp, server_id);
if (!l->table_name) if (!l->table_name)
{ {
delete l; delete l;
...@@ -140,7 +135,7 @@ Log_event* Log_event::read_log_event(FILE* file) ...@@ -140,7 +135,7 @@ Log_event* Log_event::read_log_event(FILE* file)
case ROTATE_EVENT: case ROTATE_EVENT:
{ {
Rotate_log_event* r = new Rotate_log_event(file, timestamp); Rotate_log_event* r = new Rotate_log_event(file, timestamp, server_id);
if (!r->new_log_ident) if (!r->new_log_ident)
{ {
delete r; delete r;
...@@ -152,7 +147,7 @@ Log_event* Log_event::read_log_event(FILE* file) ...@@ -152,7 +147,7 @@ Log_event* Log_event::read_log_event(FILE* file)
case INTVAR_EVENT: case INTVAR_EVENT:
{ {
Intvar_log_event* e = new Intvar_log_event(file, timestamp); Intvar_log_event* e = new Intvar_log_event(file, timestamp, server_id);
if (e->type == INVALID_INT_EVENT) if (e->type == INVALID_INT_EVENT)
{ {
delete e; delete e;
...@@ -162,8 +157,8 @@ Log_event* Log_event::read_log_event(FILE* file) ...@@ -162,8 +157,8 @@ Log_event* Log_event::read_log_event(FILE* file)
return e; return e;
} }
case START_EVENT: return new Start_log_event(file, timestamp); case START_EVENT: return new Start_log_event(file, timestamp, server_id);
case STOP_EVENT: return new Stop_log_event(file, timestamp); case STOP_EVENT: return new Stop_log_event(file, timestamp, server_id);
default: return NULL; default: return NULL;
} }
...@@ -221,12 +216,23 @@ Log_event* Log_event::read_log_event(const char* buf, int max_buf) ...@@ -221,12 +216,23 @@ Log_event* Log_event::read_log_event(const char* buf, int max_buf)
return NULL; return NULL;
} }
void Log_event::print_timestamp(FILE* file) void Log_event::print_header(FILE* file)
{
fputc('#', file);
print_timestamp(file);
fprintf(file, " server id %d ", server_id);
}
void Log_event::print_timestamp(FILE* file, time_t* ts = 0)
{ {
struct tm tm_tmp; struct tm tm_tmp;
localtime_r(&when,&tm_tmp); if(!ts)
{
ts = &when;
}
localtime_r(ts,&tm_tmp);
fprintf(file,"#%02d%02d%02d %2d:%02d:%02d", fprintf(file,"%02d%02d%02d %2d:%02d:%02d",
tm_tmp.tm_year % 100, tm_tmp.tm_year % 100,
tm_tmp.tm_mon+1, tm_tmp.tm_mon+1,
tm_tmp.tm_mday, tm_tmp.tm_mday,
...@@ -241,8 +247,11 @@ void Start_log_event::print(FILE* file, bool short_form) ...@@ -241,8 +247,11 @@ void Start_log_event::print(FILE* file, bool short_form)
if (short_form) if (short_form)
return; return;
print_timestamp(file); print_header(file);
fprintf(file, "\tStart\n"); fprintf(file, "\tStart: binlog v %d, server v %s created ", binlog_version,
server_version);
print_timestamp(file, (time_t*)&created);
fputc('\n', file);
fflush(file); fflush(file);
} }
...@@ -251,7 +260,7 @@ void Stop_log_event::print(FILE* file, bool short_form) ...@@ -251,7 +260,7 @@ void Stop_log_event::print(FILE* file, bool short_form)
if (short_form) if (short_form)
return; return;
print_timestamp(file); print_header(file);
fprintf(file, "\tStop\n"); fprintf(file, "\tStop\n");
fflush(file); fflush(file);
} }
...@@ -261,7 +270,7 @@ void Rotate_log_event::print(FILE* file, bool short_form) ...@@ -261,7 +270,7 @@ void Rotate_log_event::print(FILE* file, bool short_form)
if (short_form) if (short_form)
return; return;
print_timestamp(file); print_header(file);
fprintf(file, "\tRotate to "); fprintf(file, "\tRotate to ");
if (new_log_ident) if (new_log_ident)
my_fwrite(file, (byte*) new_log_ident, (uint)ident_len, my_fwrite(file, (byte*) new_log_ident, (uint)ident_len,
...@@ -270,8 +279,9 @@ void Rotate_log_event::print(FILE* file, bool short_form) ...@@ -270,8 +279,9 @@ void Rotate_log_event::print(FILE* file, bool short_form)
fflush(file); fflush(file);
} }
Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg): Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg,
Log_event(when_arg),new_log_ident(NULL),alloced(0) uint32 server_id):
Log_event(when_arg, 0, 0, server_id),new_log_ident(NULL),alloced(0)
{ {
char *tmp_ident; char *tmp_ident;
char buf[4]; char buf[4];
...@@ -298,6 +308,14 @@ Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg): ...@@ -298,6 +308,14 @@ Rotate_log_event::Rotate_log_event(FILE* file, time_t when_arg):
alloced = 1; alloced = 1;
} }
Start_log_event::Start_log_event(const char* buf) :Log_event(buf)
{
buf += EVENT_LEN_OFFSET + 4; // skip even length
binlog_version = uint2korr(buf);
memcpy(server_version, buf + 2, sizeof(server_version));
created = uint4korr(buf + 2 + sizeof(server_version));
}
Rotate_log_event::Rotate_log_event(const char* buf, int max_buf): Rotate_log_event::Rotate_log_event(const char* buf, int max_buf):
Log_event(buf),new_log_ident(NULL),alloced(0) Log_event(buf),new_log_ident(NULL),alloced(0)
{ {
...@@ -322,8 +340,9 @@ int Rotate_log_event::write_data(FILE* file) ...@@ -322,8 +340,9 @@ int Rotate_log_event::write_data(FILE* file)
return 0; return 0;
} }
Query_log_event::Query_log_event(FILE* file, time_t when_arg): Query_log_event::Query_log_event(FILE* file, time_t when_arg,
Log_event(when_arg),data_buf(0),query(NULL),db(NULL) uint32 server_id):
Log_event(when_arg,0,0,server_id),data_buf(0),query(NULL),db(NULL)
{ {
char buf[QUERY_HEADER_LEN + 4]; char buf[QUERY_HEADER_LEN + 4];
ulong data_len; ulong data_len;
...@@ -382,7 +401,7 @@ void Query_log_event::print(FILE* file, bool short_form) ...@@ -382,7 +401,7 @@ void Query_log_event::print(FILE* file, bool short_form)
{ {
if (!short_form) if (!short_form)
{ {
print_timestamp(file); print_header(file);
fprintf(file, "\tQuery\tthread_id=%lu\texec_time=%lu\n", fprintf(file, "\tQuery\tthread_id=%lu\texec_time=%lu\n",
(ulong) thread_id, (ulong) exec_time); (ulong) thread_id, (ulong) exec_time);
} }
...@@ -414,8 +433,9 @@ int Query_log_event::write_data(FILE* file) ...@@ -414,8 +433,9 @@ int Query_log_event::write_data(FILE* file)
return 0; return 0;
} }
Intvar_log_event:: Intvar_log_event(FILE* file, time_t when_arg) Intvar_log_event:: Intvar_log_event(FILE* file, time_t when_arg,
:Log_event(when_arg), type(INVALID_INT_EVENT) uint32 server_id)
:Log_event(when_arg,0,0,server_id), type(INVALID_INT_EVENT)
{ {
my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
char buf[9]; char buf[9];
...@@ -444,7 +464,7 @@ void Intvar_log_event::print(FILE* file, bool short_form) ...@@ -444,7 +464,7 @@ void Intvar_log_event::print(FILE* file, bool short_form)
char llbuff[22]; char llbuff[22];
if(!short_form) if(!short_form)
{ {
print_timestamp(file); print_header(file);
fprintf(file, "\tIntvar\n"); fprintf(file, "\tIntvar\n");
} }
...@@ -493,8 +513,9 @@ int Load_log_event::write_data(FILE* file __attribute__((unused))) ...@@ -493,8 +513,9 @@ int Load_log_event::write_data(FILE* file __attribute__((unused)))
return 0; return 0;
} }
Load_log_event::Load_log_event(FILE* file, time_t when): Load_log_event::Load_log_event(FILE* file, time_t when, uint32 server_id):
Log_event(when),data_buf(0),num_fields(0),fields(0),field_lens(0),field_block_len(0), Log_event(when,0,0,server_id),data_buf(0),num_fields(0),
fields(0),field_lens(0),field_block_len(0),
table_name(0),db(0),fname(0) table_name(0),db(0),fname(0)
{ {
...@@ -539,7 +560,8 @@ Load_log_event::Load_log_event(FILE* file, time_t when): ...@@ -539,7 +560,8 @@ Load_log_event::Load_log_event(FILE* file, time_t when):
} }
Load_log_event::Load_log_event(const char* buf, int max_buf): Load_log_event::Load_log_event(const char* buf, int max_buf):
Log_event(when),data_buf(0),num_fields(0),fields(0),field_lens(0),field_block_len(0), Log_event(when,0,0,server_id),data_buf(0),num_fields(0),fields(0),
field_lens(0),field_block_len(0),
table_name(0),db(0),fname(0) table_name(0),db(0),fname(0)
{ {
...@@ -594,7 +616,7 @@ void Load_log_event::print(FILE* file, bool short_form) ...@@ -594,7 +616,7 @@ void Load_log_event::print(FILE* file, bool short_form)
{ {
if (!short_form) if (!short_form)
{ {
print_timestamp(file); print_header(file);
fprintf(file, "\tQuery\tthread_id=%d\texec_time=%ld\n", fprintf(file, "\tQuery\tthread_id=%d\texec_time=%ld\n",
thread_id, exec_time); thread_id, exec_time);
} }
......
sql/log_event.h
View file @ ce79dae2
...@@ -28,6 +28,18 @@ ...@@ -28,6 +28,18 @@
#define LOG_READ_MEM -5 #define LOG_READ_MEM -5
#define LOG_EVENT_OFFSET 4 #define LOG_EVENT_OFFSET 4
#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,
...@@ -40,25 +52,32 @@ enum Int_event_type { INVALID_INT_EVENT = 0, LAST_INSERT_ID_EVENT = 1, INSERT_ID ...@@ -40,25 +52,32 @@ enum Int_event_type { INVALID_INT_EVENT = 0, LAST_INSERT_ID_EVENT = 1, INSERT_ID
class String; class String;
#endif #endif
extern uint32 server_id;
class Log_event class Log_event
{ {
public: public:
time_t when; time_t when;
ulong exec_time; ulong exec_time;
int valid_exec_time; // if false, the exec time setting is bogus and needs int valid_exec_time; // if false, the exec time setting is bogus
uint32 server_id;
int write(FILE* file); int write(FILE* file);
int write_header(FILE* file); int write_header(FILE* file);
virtual int write_data(FILE* file __attribute__((unused))) { return 0; } virtual int write_data(FILE* file __attribute__((unused))) { return 0; }
virtual Log_event_type get_type_code() = 0; virtual Log_event_type get_type_code() = 0;
Log_event(time_t when_arg, ulong exec_time_arg = 0, Log_event(time_t when_arg, ulong exec_time_arg = 0,
int valid_exec_time_arg = 0): when(when_arg), int valid_exec_time_arg = 0, uint32 server_id = 0): when(when_arg),
exec_time(exec_time_arg), valid_exec_time(valid_exec_time_arg) {} exec_time(exec_time_arg), valid_exec_time(valid_exec_time_arg)
{
if(!server_id) this->server_id = ::server_id;
else this->server_id = server_id;
}
Log_event(const char* buf): valid_exec_time(1) Log_event(const char* buf): valid_exec_time(0)
{ {
when = uint4korr(buf); when = uint4korr(buf);
exec_time = uint4korr(buf + 5); server_id = uint4korr(buf + 5);
} }
virtual ~Log_event() {} virtual ~Log_event() {}
...@@ -66,7 +85,8 @@ class Log_event ...@@ -66,7 +85,8 @@ class Log_event
virtual int get_data_size() { return 0;} virtual int get_data_size() { return 0;}
virtual void print(FILE* file, bool short_form = 0) = 0; virtual void print(FILE* file, bool short_form = 0) = 0;
void print_timestamp(FILE* file); void print_timestamp(FILE* file, time_t *ts = 0);
void print_header(FILE* file);
static Log_event* read_log_event(FILE* file); static Log_event* read_log_event(FILE* file);
static Log_event* read_log_event(const char* buf, int max_buf); static Log_event* read_log_event(const char* buf, int max_buf);
...@@ -93,7 +113,7 @@ class Query_log_event: public Log_event ...@@ -93,7 +113,7 @@ class Query_log_event: public Log_event
#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), 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)
{ {
...@@ -105,7 +125,7 @@ class Query_log_event: public Log_event ...@@ -105,7 +125,7 @@ class Query_log_event: public Log_event
} }
#endif #endif
Query_log_event(FILE* file, time_t when); Query_log_event(FILE* file, time_t when, uint32 server_id);
Query_log_event(const char* buf, int max_buf); Query_log_event(const char* buf, int max_buf);
~Query_log_event() ~Query_log_event()
{ {
...@@ -244,7 +264,7 @@ class Load_log_event: public Log_event ...@@ -244,7 +264,7 @@ class Load_log_event: public Log_event
void set_fields(List<Item> &fields); void set_fields(List<Item> &fields);
#endif #endif
Load_log_event(FILE* file, time_t when); Load_log_event(FILE* file, time_t when, uint32 server_id);
Load_log_event(const char* buf, int max_buf); Load_log_event(const char* buf, int max_buf);
~Load_log_event() ~Load_log_event()
{ {
...@@ -269,21 +289,52 @@ class Load_log_event: public Log_event ...@@ -269,21 +289,52 @@ class Load_log_event: public Log_event
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0);
}; };
extern char server_version[50];
class Start_log_event: public Log_event class Start_log_event: public Log_event
{ {
public: public:
Start_log_event() :Log_event(time(NULL)) uint16 binlog_version;
{} char server_version[50];
Start_log_event(FILE* file, time_t when_arg) :Log_event(when_arg) uint32 created;
Start_log_event() :Log_event(time(NULL)),binlog_version(BINLOG_VERSION)
{ {
my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length created = when;
memcpy(server_version, ::server_version, sizeof(server_version));
} }
Start_log_event(const char* buf) :Log_event(buf) Start_log_event(FILE* file, time_t when_arg, uint32 server_id) :
Log_event(when_arg, 0, 0, server_id)
{ {
char buf[sizeof(server_version) + sizeof(binlog_version) +
sizeof(created)];
my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
if (my_fread(file, (byte*) buf, sizeof(buf), MYF(MY_NABP | MY_WME)))
return;
binlog_version = uint2korr(buf);
memcpy(server_version, buf + 2, sizeof(server_version));
created = uint4korr(buf + 2 + sizeof(server_version));
} }
Start_log_event(const char* buf);
~Start_log_event() {} ~Start_log_event() {}
Log_event_type get_type_code() { return START_EVENT;} Log_event_type get_type_code() { return START_EVENT;}
int write_data(FILE* file)
{
if(my_fwrite(file, (byte*) &binlog_version, sizeof(binlog_version),
MYF(MY_NABP | MY_WME)) ||
my_fwrite(file, (byte*) server_version, sizeof(server_version),
MYF(MY_NABP | MY_WME)) ||
my_fwrite(file, (byte*) &created, sizeof(created),
MYF(MY_NABP | MY_WME)))
return -1;
return 0;
}
int get_data_size()
{
return sizeof(binlog_version) + sizeof(server_version) +
sizeof(created);
}
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0);
}; };
...@@ -295,7 +346,7 @@ class Intvar_log_event: public Log_event ...@@ -295,7 +346,7 @@ class Intvar_log_event: public Log_event
Intvar_log_event(uchar type_arg, ulonglong val_arg) Intvar_log_event(uchar type_arg, ulonglong val_arg)
:Log_event(time(NULL)),val(val_arg),type(type_arg) :Log_event(time(NULL)),val(val_arg),type(type_arg)
{} {}
Intvar_log_event(FILE* file, time_t when); Intvar_log_event(FILE* file, time_t when, uint32 server_id);
Intvar_log_event(const char* buf); Intvar_log_event(const char* buf);
~Intvar_log_event() {} ~Intvar_log_event() {}
Log_event_type get_type_code() { return INTVAR_EVENT;} Log_event_type get_type_code() { return INTVAR_EVENT;}
...@@ -311,7 +362,8 @@ class Stop_log_event: public Log_event ...@@ -311,7 +362,8 @@ class Stop_log_event: public Log_event
public: public:
Stop_log_event() :Log_event(time(NULL)) Stop_log_event() :Log_event(time(NULL))
{} {}
Stop_log_event(FILE* file, time_t when_arg): Log_event(when_arg) Stop_log_event(FILE* file, time_t when_arg, uint32 server_id):
Log_event(when_arg,0,0,server_id)
{ {
my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length my_fseek(file, 4L, MY_SEEK_CUR, MYF(MY_WME)); // skip the event length
} }
...@@ -337,7 +389,7 @@ class Rotate_log_event: public Log_event ...@@ -337,7 +389,7 @@ class Rotate_log_event: public Log_event
alloced(0) alloced(0)
{} {}
Rotate_log_event(FILE* file, time_t when) ; Rotate_log_event(FILE* file, time_t when, uint32 server_id) ;
Rotate_log_event(const char* buf, int max_buf); Rotate_log_event(const char* buf, int max_buf);
~Rotate_log_event() ~Rotate_log_event()
{ {
......
sql/mysql_priv.h
View file @ ce79dae2
...@@ -437,6 +437,7 @@ void sql_perror(const char *message); ...@@ -437,6 +437,7 @@ void sql_perror(const char *message);
void sql_print_error(const char *format,...) void sql_print_error(const char *format,...)
__attribute__ ((format (printf, 1, 2))); __attribute__ ((format (printf, 1, 2)));
extern uint32 server_id;
extern char mysql_data_home[2],server_version[50],max_sort_char, extern char mysql_data_home[2],server_version[50],max_sort_char,
mysql_real_data_home[]; mysql_real_data_home[];
extern my_string mysql_unix_port,mysql_tmpdir; extern my_string mysql_unix_port,mysql_tmpdir;
...@@ -463,7 +464,7 @@ extern pthread_mutex_t LOCK_mysql_create_db,LOCK_Acl,LOCK_open, ...@@ -463,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;
......
sql/mysqlbinlog.cc
View file @ ce79dae2
...@@ -29,6 +29,8 @@ ...@@ -29,6 +29,8 @@
#define CLIENT_CAPABILITIES (CLIENT_LONG_PASSWORD | CLIENT_LONG_FLAG | CLIENT_LOCAL_FILES) #define CLIENT_CAPABILITIES (CLIENT_LONG_PASSWORD | CLIENT_LONG_FLAG | CLIENT_LOCAL_FILES)
char server_version[50];
uint32 server_id = 0;
// needed by net_serv.c // needed by net_serv.c
ulong bytes_sent = 0L, bytes_received = 0L; ulong bytes_sent = 0L, bytes_received = 0L;
...@@ -268,8 +270,9 @@ static void dump_remote_log_entries(const char* logname) ...@@ -268,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(;;)
......
sql/mysqld.cc
View file @ ce79dae2
...@@ -142,6 +142,7 @@ static uint handler_count; ...@@ -142,6 +142,7 @@ static uint handler_count;
static bool opt_console=0; static bool opt_console=0;
#endif #endif
static bool opt_skip_slave_start = 0; // if set, slave is not autostarted
static ulong opt_specialflag=SPECIAL_ENGLISH; static ulong opt_specialflag=SPECIAL_ENGLISH;
static my_socket unix_sock= INVALID_SOCKET,ip_sock= INVALID_SOCKET; static my_socket unix_sock= INVALID_SOCKET,ip_sock= INVALID_SOCKET;
static ulong back_log,connect_timeout,concurrency; static ulong back_log,connect_timeout,concurrency;
...@@ -180,6 +181,7 @@ I_List<i_string> replicate_do_db, replicate_ignore_db; ...@@ -180,6 +181,7 @@ I_List<i_string> replicate_do_db, replicate_ignore_db;
// allow the user to tell us which db to replicate and which to ignore // allow the user to tell us which db to replicate and which to ignore
I_List<i_string> binlog_do_db, binlog_ignore_db; I_List<i_string> binlog_do_db, binlog_ignore_db;
uint32 server_id = 0; // server id for replication
uint mysql_port; uint mysql_port;
uint test_flags, select_errors=0, dropping_tables=0,ha_open_options=0; uint test_flags, select_errors=0, dropping_tables=0,ha_open_options=0;
uint volatile thread_count=0, thread_running=0, kill_cached_threads=0, uint volatile thread_count=0, thread_running=0, kill_cached_threads=0,
...@@ -246,7 +248,7 @@ pthread_mutex_t LOCK_mysql_create_db, LOCK_Acl, LOCK_open, LOCK_thread_count, ...@@ -246,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;
...@@ -1393,6 +1395,7 @@ int main(int argc, char **argv) ...@@ -1393,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);
...@@ -1482,17 +1485,23 @@ int main(int argc, char **argv) ...@@ -1482,17 +1485,23 @@ int main(int argc, char **argv)
LOG_NEW); LOG_NEW);
if (opt_bin_log) if (opt_bin_log)
{ {
if (!opt_bin_logname) if(server_id)
{ {
char tmp[FN_REFLEN]; if (!opt_bin_logname)
strnmov(tmp,hostname,FN_REFLEN-5); {
strmov(strcend(tmp,'.'),"-bin"); char tmp[FN_REFLEN];
opt_bin_logname=my_strdup(tmp,MYF(MY_WME)); strnmov(tmp,hostname,FN_REFLEN-5);
} strmov(strcend(tmp,'.'),"-bin");
mysql_bin_log.set_index_file_name(opt_binlog_index_name); opt_bin_logname=my_strdup(tmp,MYF(MY_WME));
open_log(&mysql_bin_log, hostname, opt_bin_logname, "-bin", }
LOG_BIN); mysql_bin_log.set_index_file_name(opt_binlog_index_name);
open_log(&mysql_bin_log, hostname, opt_bin_logname, "-bin",
LOG_BIN);
}
else
sql_print_error("Server id is not set - binary logging disabled");
} }
if (opt_slow_log) if (opt_slow_log)
open_log(&mysql_slow_log, hostname, opt_slow_logname, "-slow.log", open_log(&mysql_slow_log, hostname, opt_slow_logname, "-slow.log",
LOG_NORMAL); LOG_NORMAL);
...@@ -1596,10 +1605,15 @@ int main(int argc, char **argv) ...@@ -1596,10 +1605,15 @@ int main(int argc, char **argv)
// slave thread // slave thread
if(master_host) if(master_host)
{ {
pthread_t hThread; if(server_id)
if(pthread_create(&hThread, &connection_attrib, handle_slave, 0)) {
sql_print_error("Warning: Can't create thread to handle slave"); pthread_t hThread;
if(!opt_skip_slave_start &&
pthread_create(&hThread, &connection_attrib, handle_slave, 0))
sql_print_error("Warning: Can't create thread to handle slave");
}
else
sql_print_error("Server id is not set, slave thread will not be started");
} }
printf(ER(ER_READY),my_progname,server_version,""); printf(ER(ER_READY),my_progname,server_version,"");
...@@ -2201,7 +2215,8 @@ enum options { ...@@ -2201,7 +2215,8 @@ enum options {
OPT_REPLICATE_DO_DB, OPT_REPLICATE_IGNORE_DB, OPT_REPLICATE_DO_DB, OPT_REPLICATE_IGNORE_DB,
OPT_LOG_SLAVE_UPDATES, OPT_BINLOG_DO_DB, OPT_LOG_SLAVE_UPDATES, OPT_BINLOG_DO_DB,
OPT_BINLOG_IGNORE_DB, OPT_WANT_CORE, OPT_BINLOG_IGNORE_DB, OPT_WANT_CORE,
OPT_SKIP_CONCURRENT_INSERT, OPT_MEMLOCK, OPT_MYISAM_RECOVER OPT_SKIP_CONCURRENT_INSERT, OPT_MEMLOCK, OPT_MYISAM_RECOVER,
OPT_REPLICATE_REWRITE_DB, OPT_SERVER_ID, OPT_SKIP_SLAVE_START
}; };
static struct option long_options[] = { static struct option long_options[] = {
...@@ -2266,8 +2281,11 @@ static struct option long_options[] = { ...@@ -2266,8 +2281,11 @@ static struct option long_options[] = {
{"port", required_argument, 0, 'P'}, {"port", required_argument, 0, 'P'},
{"replicate-do-db", required_argument, 0, (int) OPT_REPLICATE_DO_DB}, {"replicate-do-db", required_argument, 0, (int) OPT_REPLICATE_DO_DB},
{"replicate-ignore-db", required_argument, 0, (int) OPT_REPLICATE_IGNORE_DB}, {"replicate-ignore-db", required_argument, 0, (int) OPT_REPLICATE_IGNORE_DB},
{"replicate-rewrite-db", required_argument, 0,
(int) OPT_REPLICATE_REWRITE_DB},
{"safe-mode", no_argument, 0, (int) OPT_SAFE}, {"safe-mode", no_argument, 0, (int) OPT_SAFE},
{"socket", required_argument, 0, (int) OPT_SOCKET}, {"socket", required_argument, 0, (int) OPT_SOCKET},
{"server-id", required_argument, 0, (int)OPT_SERVER_ID},
{"set-variable", required_argument, 0, 'O'}, {"set-variable", required_argument, 0, 'O'},
#ifdef HAVE_BERKELEY_DB #ifdef HAVE_BERKELEY_DB
{"skip-bdb", no_argument, 0, (int) OPT_BDB_SKIP}, {"skip-bdb", no_argument, 0, (int) OPT_BDB_SKIP},
...@@ -2280,6 +2298,7 @@ static struct option long_options[] = { ...@@ -2280,6 +2298,7 @@ static struct option long_options[] = {
{"skip-name-resolve", no_argument, 0, (int) OPT_SKIP_RESOLVE}, {"skip-name-resolve", no_argument, 0, (int) OPT_SKIP_RESOLVE},
{"skip-new", no_argument, 0, (int) OPT_SKIP_NEW}, {"skip-new", no_argument, 0, (int) OPT_SKIP_NEW},
{"skip-show-database", no_argument, 0, (int) OPT_SKIP_SHOW_DB}, {"skip-show-database", no_argument, 0, (int) OPT_SKIP_SHOW_DB},
{"skip-slave-start", no_argument, 0, (int) OPT_SKIP_SLAVE_START},
{"skip-networking", no_argument, 0, (int) OPT_SKIP_NETWORKING}, {"skip-networking", no_argument, 0, (int) OPT_SKIP_NETWORKING},
{"skip-thread-priority", no_argument, 0, (int) OPT_SKIP_PRIOR}, {"skip-thread-priority", no_argument, 0, (int) OPT_SKIP_PRIOR},
{"sql-bin-update-same", no_argument, 0, (int) OPT_SQL_BIN_UPDATE_SAME}, {"sql-bin-update-same", no_argument, 0, (int) OPT_SQL_BIN_UPDATE_SAME},
...@@ -2428,6 +2447,7 @@ struct show_var_st init_vars[]= { ...@@ -2428,6 +2447,7 @@ struct show_var_st init_vars[]= {
{"port", (char*) &mysql_port, SHOW_INT}, {"port", (char*) &mysql_port, SHOW_INT},
{"protocol_version", (char*) &protocol_version, SHOW_INT}, {"protocol_version", (char*) &protocol_version, SHOW_INT},
{"record_buffer", (char*) &my_default_record_cache_size,SHOW_LONG}, {"record_buffer", (char*) &my_default_record_cache_size,SHOW_LONG},
{"server_id", (char*) &server_id, SHOW_LONG},
{"skip_locking", (char*) &my_disable_locking, SHOW_MY_BOOL}, {"skip_locking", (char*) &my_disable_locking, SHOW_MY_BOOL},
{"skip_networking", (char*) &opt_disable_networking, SHOW_BOOL}, {"skip_networking", (char*) &opt_disable_networking, SHOW_BOOL},
{"skip_show_database", (char*) &opt_skip_show_db, SHOW_BOOL}, {"skip_show_database", (char*) &opt_skip_show_db, SHOW_BOOL},
...@@ -2851,6 +2871,9 @@ static void get_options(int argc,char **argv) ...@@ -2851,6 +2871,9 @@ static void get_options(int argc,char **argv)
opt_slow_log=1; opt_slow_log=1;
opt_slow_logname=optarg; opt_slow_logname=optarg;
break; break;
case (int)OPT_SKIP_SLAVE_START:
opt_skip_slave_start = 1;
break;
case (int) OPT_SKIP_NEW: case (int) OPT_SKIP_NEW:
opt_specialflag|= SPECIAL_NO_NEW_FUNC; opt_specialflag|= SPECIAL_NO_NEW_FUNC;
default_table_type=DB_TYPE_ISAM; default_table_type=DB_TYPE_ISAM;
...@@ -2972,6 +2995,9 @@ static void get_options(int argc,char **argv) ...@@ -2972,6 +2995,9 @@ static void get_options(int argc,char **argv)
default_table_type= (enum db_type) type; default_table_type= (enum db_type) type;
break; break;
} }
case OPT_SERVER_ID:
server_id = atoi(optarg);
break;
case OPT_DELAY_KEY_WRITE: case OPT_DELAY_KEY_WRITE:
ha_open_options|=HA_OPEN_DELAY_KEY_WRITE; ha_open_options|=HA_OPEN_DELAY_KEY_WRITE;
myisam_delay_key_write=1; myisam_delay_key_write=1;
......
sql/slave.cc
View file @ ce79dae2
...@@ -463,15 +463,16 @@ static int safe_sleep(THD* thd, int sec) ...@@ -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 :-)
...@@ -795,11 +796,17 @@ pthread_handler_decl(handle_slave,arg __attribute__((unused))) ...@@ -795,11 +796,17 @@ pthread_handler_decl(handle_slave,arg __attribute__((unused)))
THD *thd;; // needs to be first for thread_stack THD *thd;; // needs to be first for thread_stack
MYSQL *mysql = NULL ; MYSQL *mysql = NULL ;
if(!server_id)
{
sql_print_error("Server id not set, will not start slave");
pthread_exit(1);
}
pthread_mutex_lock(&LOCK_slave); pthread_mutex_lock(&LOCK_slave);
if(slave_running) if(slave_running)
{ {
pthread_mutex_unlock(&LOCK_slave); pthread_mutex_unlock(&LOCK_slave);
return 0; // safety just in case pthread_exit(1); // safety just in case
} }
slave_running = 1; slave_running = 1;
abort_slave = 0; abort_slave = 0;
...@@ -936,7 +943,7 @@ static void safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) ...@@ -936,7 +943,7 @@ static void safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi)
while(!slave_killed(thd) && mc_mysql_reconnect(mysql)) while(!slave_killed(thd) && mc_mysql_reconnect(mysql))
{ {
sql_print_error( sql_print_error(
"Slave thread: error connecting to slave:%s, retry in %d sec", "Slave thread: error connecting to master:%s, retry in %d sec",
mc_mysql_error(mysql), mi->connect_retry); mc_mysql_error(mysql), mi->connect_retry);
safe_sleep(thd, mi->connect_retry); safe_sleep(thd, mi->connect_retry);
} }
......
sql/sql_base.cc
View file @ ce79dae2
...@@ -42,71 +42,6 @@ static void mysql_rm_tmp_tables(void); ...@@ -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) ...@@ -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;
}
/****************************************************************************** /******************************************************************************
......
sql/sql_class.cc
View file @ ce79dae2
...@@ -83,6 +83,7 @@ THD::THD() ...@@ -83,6 +83,7 @@ THD::THD()
user_time=bootstrap=in_lock_tables=global_read_lock=0; user_time=bootstrap=in_lock_tables=global_read_lock=0;
query_length=col_access=0; query_length=col_access=0;
query_error=0; query_error=0;
server_id = ::server_id;
server_status=SERVER_STATUS_AUTOCOMMIT; server_status=SERVER_STATUS_AUTOCOMMIT;
next_insert_id=last_insert_id=0; next_insert_id=last_insert_id=0;
open_tables=temporary_tables=0; open_tables=temporary_tables=0;
......
sql/sql_class.h
View file @ ce79dae2
...@@ -243,6 +243,7 @@ class THD :public ilink { ...@@ -243,6 +243,7 @@ class THD :public ilink {
ULL *ull; ULL *ull;
struct st_my_thread_var *mysys_var; struct st_my_thread_var *mysys_var;
enum enum_server_command command; enum enum_server_command command;
uint32 server_id;
const char *where; const char *where;
char* last_nx_table; // last non-existent table, we need this for replication char* last_nx_table; // last non-existent table, we need this for replication
char* last_nx_db; // database of the last nx table char* last_nx_db; // database of the last nx table
......
sql/sql_parse.cc
View file @ ce79dae2
...@@ -17,6 +17,7 @@ ...@@ -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 @@ ...@@ -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); ...@@ -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) ...@@ -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) ...@@ -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);
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7