Commit cb9d706d authored by unknown's avatar unknown

Merge work:/home/bk/mysql

into mysql.sashanet.com:/home/sasha/src/bk/mysql
parents b8838c04 fab8da32
...@@ -1766,6 +1766,7 @@ static void init_var_hash() ...@@ -1766,6 +1766,7 @@ static void init_var_hash()
die("Variable hash initialization failed"); die("Variable hash initialization failed");
var_from_env("MASTER_MYPORT", "9306"); var_from_env("MASTER_MYPORT", "9306");
var_from_env("SLAVE_MYPORT", "9307"); var_from_env("SLAVE_MYPORT", "9307");
var_from_env("MYSQL_TEST_DIR", "");
} }
int main(int argc, char** argv) int main(int argc, char** argv)
......
...@@ -490,7 +490,7 @@ start_slave() ...@@ -490,7 +490,7 @@ start_slave()
--core \ --core \
--tmpdir=$MYSQL_TMP_DIR \ --tmpdir=$MYSQL_TMP_DIR \
--language=english \ --language=english \
--skip-innodb \ --skip-innodb --skip-slave-start \
$SMALL_SERVER \ $SMALL_SERVER \
$EXTRA_SLAVE_OPT $EXTRA_SLAVE_MYSQLD_OPT" $EXTRA_SLAVE_OPT $EXTRA_SLAVE_MYSQLD_OPT"
if [ x$DO_DDD = x1 ] if [ x$DO_DDD = x1 ]
......
reset master;
drop table if exists t1;
create table t1(n int);
insert into t1 values (1),(2),(3),(4);
flush logs;
update t1 set n = n + 10;
save_master_pos;
flush tables with read lock;
system rm -rf var/tmp/backup;
system mkdir -p var/tmp/backup;
system cp var/master-data/master-bin.* var/tmp/backup;
unlock tables;
drop table t1;
eval change master to master_host='$MYSQL_TEST_DIR/var/tmp/backup/master-bin';
slave start;
sync_with_master;
select * from t1;
...@@ -103,7 +103,7 @@ MYSQL_LOG::~MYSQL_LOG() ...@@ -103,7 +103,7 @@ MYSQL_LOG::~MYSQL_LOG()
void MYSQL_LOG::set_index_file_name(const char* index_file_name) void MYSQL_LOG::set_index_file_name(const char* index_file_name)
{ {
if (index_file_name) if (index_file_name)
fn_format(this->index_file_name,index_file_name,mysql_data_home,"-index", fn_format(this->index_file_name,index_file_name,mysql_data_home,".index",
4); 4);
else else
this->index_file_name[0] = 0; this->index_file_name[0] = 0;
...@@ -129,6 +129,32 @@ int MYSQL_LOG::generate_new_name(char *new_name, const char *log_name) ...@@ -129,6 +129,32 @@ int MYSQL_LOG::generate_new_name(char *new_name, const char *log_name)
return 0; return 0;
} }
bool MYSQL_LOG::open_index( int options)
{
return (index_file < 0 &&
(index_file = my_open(index_file_name, options | O_BINARY ,
MYF(MY_WME))) < 0);
}
void MYSQL_LOG::init(enum_log_type log_type_arg)
{
log_type = log_type_arg;
if (!inited)
{
inited=1;
(void) pthread_mutex_init(&LOCK_log,MY_MUTEX_INIT_SLOW);
(void) pthread_mutex_init(&LOCK_index, MY_MUTEX_INIT_SLOW);
}
}
void MYSQL_LOG::close_index()
{
if(index_file >= 0)
{
my_close(index_file, MYF(0));
index_file = -1;
}
}
void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg, void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg,
const char *new_name) const char *new_name)
...@@ -137,17 +163,11 @@ void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg, ...@@ -137,17 +163,11 @@ void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg,
char buff[512]; char buff[512];
File file= -1; File file= -1;
bool do_magic; bool do_magic;
if (!inited) if (!inited && log_type_arg == LOG_BIN && *fn_ext(log_name))
{
inited=1;
(void) pthread_mutex_init(&LOCK_log,MY_MUTEX_INIT_SLOW);
(void) pthread_mutex_init(&LOCK_index, MY_MUTEX_INIT_SLOW);
if (log_type_arg == LOG_BIN && *fn_ext(log_name))
no_rotate = 1; no_rotate = 1;
} init(log_type_arg);
log_type=log_type_arg;
if (!(name=my_strdup(log_name,MYF(MY_WME)))) if (!(name=my_strdup(log_name,MYF(MY_WME))))
goto err; goto err;
if (new_name) if (new_name)
...@@ -208,10 +228,7 @@ void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg, ...@@ -208,10 +228,7 @@ void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg,
clean up if failed clean up if failed
*/ */
if ((do_magic && my_b_write(&log_file, (byte*) BINLOG_MAGIC, 4)) || if ((do_magic && my_b_write(&log_file, (byte*) BINLOG_MAGIC, 4)) ||
(index_file < 0 && open_index(O_APPEND | O_RDWR | O_CREAT))
(index_file = my_open(index_file_name,
O_APPEND | O_BINARY | O_RDWR | O_CREAT,
MYF(MY_WME))) < 0))
goto err; goto err;
Start_log_event s; Start_log_event s;
bool error; bool error;
...@@ -224,8 +241,7 @@ void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg, ...@@ -224,8 +241,7 @@ void MYSQL_LOG::open(const char *log_name, enum_log_type log_type_arg,
pthread_mutex_unlock(&LOCK_index); pthread_mutex_unlock(&LOCK_index);
if (error) if (error)
{ {
my_close(index_file,MYF(0)); close_index();
index_file= -1;
goto err; goto err;
} }
} }
......
...@@ -280,7 +280,7 @@ void Log_event::print_timestamp(FILE* file, time_t* ts) ...@@ -280,7 +280,7 @@ void Log_event::print_timestamp(FILE* file, time_t* ts)
} }
void Start_log_event::print(FILE* file, bool short_form) void Start_log_event::print(FILE* file, bool short_form, char* last_db)
{ {
if (short_form) if (short_form)
return; return;
...@@ -293,7 +293,7 @@ void Start_log_event::print(FILE* file, bool short_form) ...@@ -293,7 +293,7 @@ void Start_log_event::print(FILE* file, bool short_form)
fflush(file); fflush(file);
} }
void Stop_log_event::print(FILE* file, bool short_form) void Stop_log_event::print(FILE* file, bool short_form, char* last_db)
{ {
if (short_form) if (short_form)
return; return;
...@@ -303,7 +303,7 @@ void Stop_log_event::print(FILE* file, bool short_form) ...@@ -303,7 +303,7 @@ void Stop_log_event::print(FILE* file, bool short_form)
fflush(file); fflush(file);
} }
void Rotate_log_event::print(FILE* file, bool short_form) void Rotate_log_event::print(FILE* file, bool short_form, char* last_db)
{ {
if (short_form) if (short_form)
return; return;
...@@ -441,7 +441,7 @@ Query_log_event::Query_log_event(const char* buf, int event_len): ...@@ -441,7 +441,7 @@ Query_log_event::Query_log_event(const char* buf, int event_len):
*((char*)query+q_len) = 0; *((char*)query+q_len) = 0;
} }
void Query_log_event::print(FILE* file, bool short_form) void Query_log_event::print(FILE* file, bool short_form, char* last_db)
{ {
char buff[40],*end; // Enough for SET TIMESTAMP char buff[40],*end; // Enough for SET TIMESTAMP
if (!short_form) if (!short_form)
...@@ -451,7 +451,15 @@ void Query_log_event::print(FILE* file, bool short_form) ...@@ -451,7 +451,15 @@ void Query_log_event::print(FILE* file, bool short_form)
(ulong) thread_id, (ulong) exec_time, error_code); (ulong) thread_id, (ulong) exec_time, error_code);
} }
if (db && db[0]) bool same_db = 0;
if(db && last_db)
{
if(!(same_db = !memcmp(last_db, db, db_len)))
memcpy(last_db, db, db_len + 1);
}
if (db && db[0] && !same_db)
fprintf(file, "use %s;\n", db); fprintf(file, "use %s;\n", db);
end=int10_to_str((long) when, strmov(buff,"SET TIMESTAMP="),10); end=int10_to_str((long) when, strmov(buff,"SET TIMESTAMP="),10);
*end++=';'; *end++=';';
...@@ -507,7 +515,7 @@ int Intvar_log_event::write_data(IO_CACHE* file) ...@@ -507,7 +515,7 @@ int Intvar_log_event::write_data(IO_CACHE* file)
return my_b_write(file, (byte*) buf, sizeof(buf)); return my_b_write(file, (byte*) buf, sizeof(buf));
} }
void Intvar_log_event::print(FILE* file, bool short_form) void Intvar_log_event::print(FILE* file, bool short_form, char* last_db)
{ {
char llbuff[22]; char llbuff[22];
if(!short_form) if(!short_form)
...@@ -625,7 +633,7 @@ void Load_log_event::copy_log_event(const char *buf, ulong data_len) ...@@ -625,7 +633,7 @@ void Load_log_event::copy_log_event(const char *buf, ulong data_len)
} }
void Load_log_event::print(FILE* file, bool short_form) void Load_log_event::print(FILE* file, bool short_form, char* last_db)
{ {
if (!short_form) if (!short_form)
{ {
...@@ -634,7 +642,15 @@ void Load_log_event::print(FILE* file, bool short_form) ...@@ -634,7 +642,15 @@ void Load_log_event::print(FILE* file, bool short_form)
thread_id, exec_time); thread_id, exec_time);
} }
if(db && db[0]) bool same_db = 0;
if(db && last_db)
{
if(!(same_db = !memcmp(last_db, db, db_len)))
memcpy(last_db, db, db_len + 1);
}
if(db && db[0] && !same_db)
fprintf(file, "use %s;\n", db); fprintf(file, "use %s;\n", db);
fprintf(file, "LOAD DATA INFILE '%s' ", fname); fprintf(file, "LOAD DATA INFILE '%s' ", fname);
......
...@@ -100,7 +100,7 @@ class Log_event ...@@ -100,7 +100,7 @@ class Log_event
virtual ~Log_event() {} virtual ~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, char* last_db = 0) = 0;
void print_timestamp(FILE* file, time_t *ts = 0); void print_timestamp(FILE* file, time_t *ts = 0);
void print_header(FILE* file); void print_header(FILE* file);
...@@ -169,7 +169,7 @@ class Query_log_event: public Log_event ...@@ -169,7 +169,7 @@ class Query_log_event: public Log_event
; ;
} }
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
}; };
#define DUMPFILE_FLAG 0x1 #define DUMPFILE_FLAG 0x1
...@@ -312,7 +312,7 @@ class Load_log_event: public Log_event ...@@ -312,7 +312,7 @@ class Load_log_event: public Log_event
; ;
} }
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
}; };
extern char server_version[SERVER_VERSION_LENGTH]; extern char server_version[SERVER_VERSION_LENGTH];
...@@ -350,7 +350,7 @@ class Start_log_event: public Log_event ...@@ -350,7 +350,7 @@ class Start_log_event: public Log_event
// sizeof(binlog_version) + sizeof(server_version) sizeof(created) // sizeof(binlog_version) + sizeof(server_version) sizeof(created)
return 2 + sizeof(server_version) + 4; return 2 + sizeof(server_version) + 4;
} }
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
}; };
class Intvar_log_event: public Log_event class Intvar_log_event: public Log_event
...@@ -369,7 +369,7 @@ class Intvar_log_event: public Log_event ...@@ -369,7 +369,7 @@ class Intvar_log_event: public Log_event
int write_data(IO_CACHE* file); int write_data(IO_CACHE* file);
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
}; };
class Stop_log_event: public Log_event class Stop_log_event: public Log_event
...@@ -388,7 +388,7 @@ class Stop_log_event: public Log_event ...@@ -388,7 +388,7 @@ class Stop_log_event: public Log_event
} }
~Stop_log_event() {} ~Stop_log_event() {}
Log_event_type get_type_code() { return STOP_EVENT;} Log_event_type get_type_code() { return STOP_EVENT;}
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
}; };
class Rotate_log_event: public Log_event class Rotate_log_event: public Log_event
...@@ -416,7 +416,7 @@ class Rotate_log_event: public Log_event ...@@ -416,7 +416,7 @@ class Rotate_log_event: public Log_event
int get_data_size() { return ident_len;} int get_data_size() { return ident_len;}
int write_data(IO_CACHE* file); int write_data(IO_CACHE* file);
void print(FILE* file, bool short_form = 0); void print(FILE* file, bool short_form = 0, char* last_db = 0);
}; };
#endif #endif
...@@ -290,6 +290,7 @@ static void dump_remote_table(NET* net, const char* db, const char* table) ...@@ -290,6 +290,7 @@ static void dump_remote_table(NET* net, const char* db, const char* table)
static void dump_remote_log_entries(const char* logname) static void dump_remote_log_entries(const char* logname)
{ {
char buf[128]; char buf[128];
char last_db[FN_REFLEN+1] = "";
uint len; uint len;
NET* net = &mysql->net; NET* net = &mysql->net;
if(!position) position = 4; // protect the innocent from spam if(!position) position = 4; // protect the innocent from spam
...@@ -323,7 +324,7 @@ Unfortunately, no sweepstakes today, adjusted position to 4\n"); ...@@ -323,7 +324,7 @@ Unfortunately, no sweepstakes today, adjusted position to 4\n");
len - 1); len - 1);
if(ev) if(ev)
{ {
ev->print(stdout, short_form); ev->print(stdout, short_form, last_db);
if(ev->get_type_code() == LOAD_EVENT) if(ev->get_type_code() == LOAD_EVENT)
dump_remote_file(net, ((Load_log_event*)ev)->fname); dump_remote_file(net, ((Load_log_event*)ev)->fname);
delete ev; delete ev;
...@@ -338,6 +339,7 @@ static void dump_local_log_entries(const char* logname) ...@@ -338,6 +339,7 @@ static void dump_local_log_entries(const char* logname)
File fd = -1; File fd = -1;
IO_CACHE cache,*file= &cache; IO_CACHE cache,*file= &cache;
ulonglong rec_count = 0; ulonglong rec_count = 0;
char last_db[FN_REFLEN+1] = "";
if (logname && logname[0] != '-') if (logname && logname[0] != '-')
{ {
...@@ -397,7 +399,7 @@ Could not read entry at offset %s : Error in log format or read error", ...@@ -397,7 +399,7 @@ Could not read entry at offset %s : Error in log format or read error",
if (!short_form) if (!short_form)
printf("# at %s\n",llstr(old_off,llbuff)); printf("# at %s\n",llstr(old_off,llbuff));
ev->print(stdout, short_form); ev->print(stdout, short_form, last_db);
} }
rec_count++; rec_count++;
delete ev; delete ev;
......
...@@ -20,6 +20,7 @@ ...@@ -20,6 +20,7 @@
#include <myisam.h> #include <myisam.h>
#include "mini_client.h" #include "mini_client.h"
#include "slave.h" #include "slave.h"
#include "sql_repl.h"
#include <thr_alarm.h> #include <thr_alarm.h>
#include <my_dir.h> #include <my_dir.h>
...@@ -441,6 +442,101 @@ int fetch_nx_table(THD* thd, MASTER_INFO* mi) ...@@ -441,6 +442,101 @@ int fetch_nx_table(THD* thd, MASTER_INFO* mi)
return error; return error;
} }
void MASTER_INFO::close_virtual_master()
{
vm_binlog.close_index();
end_io_cache(&vm_cache);
if(vm_fd >= 0)
{
my_close(vm_fd, MYF(0));
vm_fd = -1;
}
}
int MASTER_INFO::setup_virtual_master()
{
vm_binlog.init(LOG_BIN);
vm_binlog.set_index_file_name(host);
if(vm_binlog.open_index(O_RDONLY))
{
sql_print_error("virtual master: could not open index file '%s': \
(%d)", host, my_errno);
return 1;
}
if(vm_binlog.find_first_log(&vm_linfo,log_file_name))
{
sql_print_error("virtual master: could not find first log");
return 1;
}
if(open_log())
return 1;
return 0;
}
int MASTER_INFO::open_log()
{
const char* errmsg = "Unknown error";
if(vm_fd >= 0)
{
end_io_cache(&vm_cache);
my_close(vm_fd, MYF(0));
}
// if backup-up logs have relative paths, assume they are relative to
// the directory that has the log index, not cwd
char logname_buf[FN_REFLEN+1], *logname;
if(vm_linfo.log_file_name[0] == FN_LIBCHAR)
logname = vm_linfo.log_file_name;
else
{
char* end = strnmov(logname_buf, host,
sizeof(logname_buf));
for(; *end != FN_LIBCHAR; --end); // we will always find it, first
// char of host is always FN_LIBCHAR for virtual master
strncpy(end + 1, vm_linfo.log_file_name,
sizeof(logname_buf) - (end - logname_buf));
logname = logname_buf;
}
if((vm_fd = open_binlog(&vm_cache, logname, &errmsg)) < 0)
{
sql_print_error("virtual master: error opening binlog '%s': %s",
vm_linfo.log_file_name, errmsg);
return 1;
}
strncpy(log_file_name, vm_linfo.log_file_name, sizeof(log_file_name));
return 0;
}
uint MASTER_INFO::read_event()
{
for(;!(vm_ev = Log_event::read_log_event(&vm_cache, 0));)
{
if(!vm_cache.error) // eof - try next log
{
switch(vm_binlog.find_next_log(&vm_linfo))
{
case LOG_INFO_EOF:
return 0;
case 0:
if(open_log())
return packet_error;
continue;
default:
sql_print_error("virtual master: could not read next log");
return packet_error;
}
}
}
return vm_ev->get_data_size() + LOG_EVENT_HEADER_LEN;
}
void end_master_info(MASTER_INFO* mi) void end_master_info(MASTER_INFO* mi)
{ {
if(mi->fd >= 0) if(mi->fd >= 0)
...@@ -450,6 +546,8 @@ void end_master_info(MASTER_INFO* mi) ...@@ -450,6 +546,8 @@ void end_master_info(MASTER_INFO* mi)
mi->fd = -1; mi->fd = -1;
} }
mi->inited = 0; mi->inited = 0;
if(mi->virtual_master)
mi->close_virtual_master();
} }
int init_master_info(MASTER_INFO* mi) int init_master_info(MASTER_INFO* mi)
...@@ -545,6 +643,7 @@ int init_master_info(MASTER_INFO* mi) ...@@ -545,6 +643,7 @@ int init_master_info(MASTER_INFO* mi)
} }
mi->inited = 1; mi->inited = 1;
mi->virtual_master = (mi->host[0] == FN_LIBCHAR);
// now change the cache from READ to WRITE - must do this // now change the cache from READ to WRITE - must do this
// before flush_master_info // before flush_master_info
reinit_io_cache(&mi->file, WRITE_CACHE, 0L,0,1); reinit_io_cache(&mi->file, WRITE_CACHE, 0L,0,1);
...@@ -742,6 +841,9 @@ static int safe_sleep(THD* thd, int sec) ...@@ -742,6 +841,9 @@ 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)
{ {
if(mi->virtual_master)
return 0;
char buf[FN_REFLEN + 10]; char buf[FN_REFLEN + 10];
int len; int len;
int binlog_flags = 0; // for now int binlog_flags = 0; // for now
...@@ -795,6 +897,9 @@ command"); ...@@ -795,6 +897,9 @@ command");
static uint read_event(MYSQL* mysql, MASTER_INFO *mi) static uint read_event(MYSQL* mysql, MASTER_INFO *mi)
{ {
if(mi->virtual_master)
return mi->read_event();
uint len = packet_error; uint len = packet_error;
// for convinience lets think we start by // for convinience lets think we start by
// being in the interrupted state :-) // being in the interrupted state :-)
...@@ -860,16 +965,18 @@ point. If you are sure that your master is ok, run this query manually on the\ ...@@ -860,16 +965,18 @@ point. If you are sure that your master is ok, run this query manually on the\
static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len) static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
{ {
Log_event * ev = Log_event::read_log_event((const char*)net->read_pos + 1, Log_event * ev = (mi->virtual_master) ? mi->vm_ev :
event_len); Log_event::read_log_event((const char*)net->read_pos + 1,
event_len) ;
char llbuff[22]; char llbuff[22];
if (ev) if (ev)
{ {
int type_code = ev->get_type_code(); int type_code = ev->get_type_code();
if (ev->server_id == ::server_id || slave_skip_counter) if ((!mi->virtual_master && ev->server_id == ::server_id)
|| slave_skip_counter)
{ {
if(type_code == LOAD_EVENT) if(type_code == LOAD_EVENT && !mi->virtual_master)
skip_load_data_infile(net); skip_load_data_infile(net);
mi->inc_pos(event_len); mi->inc_pos(event_len);
...@@ -971,6 +1078,14 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len) ...@@ -971,6 +1078,14 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
case LOAD_EVENT: case LOAD_EVENT:
{ {
if(mi->virtual_master)
{
delete ev;
sql_print_error("LOAD DATA INFILE does not yet work with virtual \
master. Perform in manually, then restart slave with SET SQL_SKIP_COUNTER=1;\
SLAVE START");
return 1;
}
Load_log_event* lev = (Load_log_event*)ev; Load_log_event* lev = (Load_log_event*)ev;
init_sql_alloc(&thd->mem_root, 8192,0); init_sql_alloc(&thd->mem_root, 8192,0);
thd->db = rewrite_db((char*)lev->db); thd->db = rewrite_db((char*)lev->db);
...@@ -993,7 +1108,8 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len) ...@@ -993,7 +1108,8 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
// the table will be opened in mysql_load // the table will be opened in mysql_load
if(table_rules_on && !tables_ok(thd, &tables)) if(table_rules_on && !tables_ok(thd, &tables))
{ {
skip_load_data_infile(net); if(!mi->virtual_master)
skip_load_data_infile(net);
} }
else else
{ {
...@@ -1057,7 +1173,8 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len) ...@@ -1057,7 +1173,8 @@ static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
{ {
// we will just ask the master to send us /dev/null if we do not // we will just ask the master to send us /dev/null if we do not
// want to load the data :-) // want to load the data :-)
skip_load_data_infile(net); if(!mi->virtual_master)
skip_load_data_infile(net);
} }
thd->net.vio = 0; thd->net.vio = 0;
...@@ -1293,9 +1410,21 @@ try again, log '%s' at postion %s", RPL_LOG_NAME, ...@@ -1293,9 +1410,21 @@ try again, log '%s' at postion %s", RPL_LOG_NAME,
sql_print_error("Slave thread killed while reading event"); sql_print_error("Slave thread killed while reading event");
goto err; goto err;
} }
if(!event_len && glob_mi.virtual_master)
{
sql_print_error("Virtual master replication finished");
goto err;
}
if (event_len == packet_error) if (event_len == packet_error)
{ {
if(glob_mi.virtual_master)
{
sql_print_error("Virtual master replication encountered \
error while reading event, replication terminated");
goto err;
}
thd->proc_info = "Waiting to reconnect after a failed read"; thd->proc_info = "Waiting to reconnect after a failed read";
if(mysql->net.vio) if(mysql->net.vio)
vio_close(mysql->net.vio); vio_close(mysql->net.vio);
...@@ -1403,6 +1532,8 @@ position %s", ...@@ -1403,6 +1532,8 @@ position %s",
static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi)
{ {
if(mi->virtual_master)
return mi->setup_virtual_master();
int slave_was_killed; int slave_was_killed;
#ifndef DBUG_OFF #ifndef DBUG_OFF
events_till_disconnect = disconnect_slave_event_count; events_till_disconnect = disconnect_slave_event_count;
...@@ -1432,6 +1563,9 @@ static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) ...@@ -1432,6 +1563,9 @@ static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi)
static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi) static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi)
{ {
if(mi->virtual_master)
return mi->setup_virtual_master();
int slave_was_killed; int slave_was_killed;
char llbuff[22]; char llbuff[22];
......
...@@ -16,8 +16,14 @@ typedef struct st_master_info ...@@ -16,8 +16,14 @@ typedef struct st_master_info
pthread_mutex_t lock; pthread_mutex_t lock;
pthread_cond_t cond; pthread_cond_t cond;
bool inited; bool inited;
bool virtual_master; // for replay of binlogs from a directory
MYSQL_LOG vm_binlog;
LOG_INFO vm_linfo;
IO_CACHE vm_cache;
int vm_fd;
Log_event* vm_ev;
st_master_info():pending(0),fd(-1),inited(0) st_master_info():pending(0),fd(-1),inited(0),virtual_master(0),vm_fd(-1)
{ {
host[0] = 0; user[0] = 0; password[0] = 0; host[0] = 0; user[0] = 0; password[0] = 0;
pthread_mutex_init(&lock, MY_MUTEX_INIT_FAST); pthread_mutex_init(&lock, MY_MUTEX_INIT_FAST);
...@@ -28,6 +34,8 @@ typedef struct st_master_info ...@@ -28,6 +34,8 @@ typedef struct st_master_info
{ {
pthread_mutex_destroy(&lock); pthread_mutex_destroy(&lock);
pthread_cond_destroy(&cond); pthread_cond_destroy(&cond);
if(virtual_master)
close_virtual_master();
} }
inline void inc_pending(ulonglong val) inline void inc_pending(ulonglong val)
{ {
...@@ -51,6 +59,11 @@ typedef struct st_master_info ...@@ -51,6 +59,11 @@ typedef struct st_master_info
} }
int wait_for_pos(THD* thd, String* log_name, ulonglong log_pos); int wait_for_pos(THD* thd, String* log_name, ulonglong log_pos);
int setup_virtual_master();
void close_virtual_master();
uint read_event();
int open_log();
} MASTER_INFO; } MASTER_INFO;
typedef struct st_table_rule_ent typedef struct st_table_rule_ent
......
...@@ -71,9 +71,12 @@ class MYSQL_LOG { ...@@ -71,9 +71,12 @@ class MYSQL_LOG {
~MYSQL_LOG(); ~MYSQL_LOG();
pthread_mutex_t* get_log_lock() { return &LOCK_log; } pthread_mutex_t* get_log_lock() { return &LOCK_log; }
void set_index_file_name(const char* index_file_name = 0); void set_index_file_name(const char* index_file_name = 0);
void init(enum_log_type log_type_arg);
void open(const char *log_name,enum_log_type log_type, void open(const char *log_name,enum_log_type log_type,
const char *new_name=0); const char *new_name=0);
void new_file(void); void new_file(void);
bool open_index(int options);
void close_index();
bool write(THD *thd, enum enum_server_command command,const char *format,...); bool write(THD *thd, enum enum_server_command command,const char *format,...);
bool write(THD *thd, const char *query, uint query_length, bool write(THD *thd, const char *query, uint query_length,
time_t query_start=0); time_t query_start=0);
......
...@@ -132,7 +132,7 @@ static int send_file(THD *thd) ...@@ -132,7 +132,7 @@ static int send_file(THD *thd)
} }
static File open_log(IO_CACHE *log, const char *log_file_name, File open_binlog(IO_CACHE *log, const char *log_file_name,
const char **errmsg) const char **errmsg)
{ {
File file; File file;
...@@ -294,7 +294,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags) ...@@ -294,7 +294,7 @@ void mysql_binlog_send(THD* thd, char* log_ident, ulong pos, ushort flags)
goto err; goto err;
} }
if ((file=open_log(&log, log_file_name, &errmsg)) < 0) if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)
goto err; goto err;
if(pos < 4) if(pos < 4)
...@@ -483,7 +483,7 @@ sweepstakes if you report the bug"; ...@@ -483,7 +483,7 @@ sweepstakes if you report the bug";
// fake Rotate_log event just in case it did not make it to the log // fake Rotate_log event just in case it did not make it to the log
// otherwise the slave make get confused about the offset // otherwise the slave make get confused about the offset
if ((file=open_log(&log, log_file_name, &errmsg)) < 0 || if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||
fake_rotate_event(net, packet, log_file_name, &errmsg)) fake_rotate_event(net, packet, log_file_name, &errmsg))
goto err; goto err;
...@@ -694,7 +694,12 @@ int change_master(THD* thd) ...@@ -694,7 +694,12 @@ int change_master(THD* thd)
glob_mi.pos = lex_mi->pos; glob_mi.pos = lex_mi->pos;
if(lex_mi->host) if(lex_mi->host)
strmake(glob_mi.host, lex_mi->host, sizeof(glob_mi.host)); {
if(glob_mi.virtual_master)
glob_mi.close_virtual_master();
strmake(glob_mi.host, lex_mi->host, sizeof(glob_mi.host));
glob_mi.virtual_master = (glob_mi.host[0] == FN_LIBCHAR);
}
if(lex_mi->user) if(lex_mi->user)
strmake(glob_mi.user, lex_mi->user, sizeof(glob_mi.user)); strmake(glob_mi.user, lex_mi->user, sizeof(glob_mi.user));
if(lex_mi->password) if(lex_mi->password)
......
...@@ -9,6 +9,9 @@ extern uint32 server_id; ...@@ -9,6 +9,9 @@ extern uint32 server_id;
extern bool server_id_supplied; extern bool server_id_supplied;
extern I_List<i_string> binlog_do_db, binlog_ignore_db; extern I_List<i_string> binlog_do_db, binlog_ignore_db;
File open_binlog(IO_CACHE *log, const char *log_file_name,
const char **errmsg);
int start_slave(THD* thd = 0, bool net_report = 1); int start_slave(THD* thd = 0, bool net_report = 1);
int stop_slave(THD* thd = 0, bool net_report = 1); int stop_slave(THD* thd = 0, bool net_report = 1);
int change_master(THD* thd); int change_master(THD* thd);
......
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