1
/* Copyright (C) 2000-2003 MySQL AB
3
This program is free software; you can redistribute it and/or modify
4
it under the terms of the GNU General Public License as published by
5
the Free Software Foundation; version 2 of the License.
7
This program is distributed in the hope that it will be useful,
8
but WITHOUT ANY WARRANTY; without even the implied warranty of
9
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10
GNU General Public License for more details.
12
You should have received a copy of the GNU General Public License
13
along with this program; if not, write to the Free Software
14
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
18
@addtogroup Replication
23
@brief Code to run the io thread and the sql thread on the
27
#include "mysql_priv.h"
35
#include "rpl_filter.h"
36
#include "repl_failsafe.h"
37
#include <thr_alarm.h>
39
#include <sql_common.h>
41
#include <mysys_err.h>
43
#ifdef HAVE_REPLICATION
45
#include "rpl_tblmap.h"
47
#define FLAGSTR(V,F) ((V)&(F)?#F" ":"")
49
#define MAX_SLAVE_RETRY_PAUSE 5
50
bool use_slave_mask = 0;
51
MY_BITMAP slave_error_mask;
53
typedef bool (*CHECK_KILLED_FUNC)(THD*,void*);
55
char* slave_load_tmpdir = 0;
56
Master_info *active_mi= 0;
57
my_bool replicate_same_server_id;
58
ulonglong relay_log_space_limit = 0;
61
When slave thread exits, we need to remember the temporary tables so we
62
can re-use them on slave start.
64
TODO: move the vars below under Master_info
67
int disconnect_slave_event_count = 0, abort_slave_event_count = 0;
68
int events_till_abort = -1;
70
enum enum_slave_reconnect_actions
72
SLAVE_RECON_ACT_REG= 0,
73
SLAVE_RECON_ACT_DUMP= 1,
74
SLAVE_RECON_ACT_EVENT= 2,
78
enum enum_slave_reconnect_messages
80
SLAVE_RECON_MSG_WAIT= 0,
81
SLAVE_RECON_MSG_KILLED_WAITING= 1,
82
SLAVE_RECON_MSG_AFTER= 2,
83
SLAVE_RECON_MSG_FAILED= 3,
84
SLAVE_RECON_MSG_COMMAND= 4,
85
SLAVE_RECON_MSG_KILLED_AFTER= 5,
89
static const char *reconnect_messages[SLAVE_RECON_ACT_MAX][SLAVE_RECON_MSG_MAX]=
92
"Waiting to reconnect after a failed registration on master",
93
"Slave I/O thread killed while waitnig to reconnect after a failed \
94
registration on master",
95
"Reconnecting after a failed registration on master",
96
"failed registering on master, reconnecting to try again, \
97
log '%s' at postion %s",
99
"Slave I/O thread killed during or after reconnect"
102
"Waiting to reconnect after a failed binlog dump request",
103
"Slave I/O thread killed while retrying master dump",
104
"Reconnecting after a failed binlog dump request",
105
"failed dump request, reconnecting to try again, log '%s' at postion %s",
107
"Slave I/O thread killed during or after reconnect"
110
"Waiting to reconnect after a failed master event read",
111
"Slave I/O thread killed while waiting to reconnect after a failed read",
112
"Reconnecting after a failed master event read",
113
"Slave I/O thread: Failed reading log event, reconnecting to retry, \
114
log '%s' at postion %s",
116
"Slave I/O thread killed during or after a reconnect done to recover from \
122
typedef enum { SLAVE_THD_IO, SLAVE_THD_SQL} SLAVE_THD_TYPE;
124
static int process_io_rotate(Master_info* mi, Rotate_log_event* rev);
125
static int process_io_create_file(Master_info* mi, Create_file_log_event* cev);
126
static bool wait_for_relay_log_space(Relay_log_info* rli);
127
static inline bool io_slave_killed(THD* thd,Master_info* mi);
128
static inline bool sql_slave_killed(THD* thd,Relay_log_info* rli);
129
static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type);
130
static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi);
131
static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi,
132
bool suppress_warnings);
133
static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi,
134
bool reconnect, bool suppress_warnings);
135
static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed,
136
void* thread_killed_arg);
137
static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi);
138
static Log_event* next_event(Relay_log_info* rli);
139
static int queue_event(Master_info* mi,const char* buf,ulong event_len);
140
static int terminate_slave_thread(THD *thd,
141
pthread_mutex_t* term_lock,
142
pthread_cond_t* term_cond,
143
volatile uint *slave_running,
145
static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info);
148
Find out which replications threads are running
152
mask Return value here
153
mi master_info for slave
154
inverse If set, returns which threads are not running
157
Get a bit mask for which threads are running so that we can later restart
161
mask If inverse == 0, running threads
162
If inverse == 1, stopped threads
165
void init_thread_mask(int* mask,Master_info* mi,bool inverse)
167
bool set_io = mi->slave_running, set_sql = mi->rli.slave_running;
168
register int tmp_mask=0;
169
DBUG_ENTER("init_thread_mask");
172
tmp_mask |= SLAVE_IO;
174
tmp_mask |= SLAVE_SQL;
176
tmp_mask^= (SLAVE_IO | SLAVE_SQL);
186
void lock_slave_threads(Master_info* mi)
188
DBUG_ENTER("lock_slave_threads");
190
//TODO: see if we can do this without dual mutex
191
pthread_mutex_lock(&mi->run_lock);
192
pthread_mutex_lock(&mi->rli.run_lock);
198
unlock_slave_threads()
201
void unlock_slave_threads(Master_info* mi)
203
DBUG_ENTER("unlock_slave_threads");
205
//TODO: see if we can do this without dual mutex
206
pthread_mutex_unlock(&mi->rli.run_lock);
207
pthread_mutex_unlock(&mi->run_lock);
212
/* Initialize slave structures */
216
DBUG_ENTER("init_slave");
219
This is called when mysqld starts. Before client connections are
220
accepted. However bootstrap may conflict with us if it does START SLAVE.
221
So it's safer to take the lock.
223
pthread_mutex_lock(&LOCK_active_mi);
225
TODO: re-write this to interate through the list of files
228
active_mi= new Master_info;
231
If master_host is not specified, try to read it from the master_info file.
232
If master_host is specified, create the master_info file if it doesn't
237
sql_print_error("Failed to allocate memory for the master info structure");
241
if (init_master_info(active_mi,master_info_file,relay_log_info_file,
242
1, (SLAVE_IO | SLAVE_SQL)))
244
sql_print_error("Failed to initialize the master info structure");
248
/* If server id is not set, start_slave_thread() will say it */
250
if (active_mi->host[0] && !opt_skip_slave_start)
252
if (start_slave_threads(1 /* need mutex */,
253
0 /* no wait for start*/,
257
SLAVE_IO | SLAVE_SQL))
259
sql_print_error("Failed to create slave threads");
263
pthread_mutex_unlock(&LOCK_active_mi);
267
pthread_mutex_unlock(&LOCK_active_mi);
273
Init function to set up array for errors that should be skipped for slave
276
init_slave_skip_errors()
277
arg List of errors numbers to skip, separated with ','
280
Called from get_options() in mysqld.cc on start-up
283
void init_slave_skip_errors(const char* arg)
286
DBUG_ENTER("init_slave_skip_errors");
288
if (bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0))
290
fprintf(stderr, "Badly out of memory, please check your system status\n");
294
for (;my_isspace(system_charset_info,*arg);++arg)
296
if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const uchar*)"all",4))
298
bitmap_set_all(&slave_error_mask);
304
if (!(p= str2int(p, 10, 0, LONG_MAX, &err_code)))
306
if (err_code < MAX_SLAVE_ERROR)
307
bitmap_set_bit(&slave_error_mask,(uint)err_code);
308
while (!my_isdigit(system_charset_info,*p) && *p)
315
int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
317
DBUG_ENTER("terminate_slave_threads");
320
DBUG_RETURN(0); /* successfully do nothing */
321
int error,force_all = (thread_mask & SLAVE_FORCE_ALL);
322
pthread_mutex_t *sql_lock = &mi->rli.run_lock, *io_lock = &mi->run_lock;
324
if ((thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL)))
326
DBUG_PRINT("info",("Terminating IO thread"));
328
if ((error=terminate_slave_thread(mi->io_thd,io_lock,
335
if ((thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL)))
337
DBUG_PRINT("info",("Terminating SQL thread"));
338
mi->rli.abort_slave=1;
339
if ((error=terminate_slave_thread(mi->rli.sql_thd,sql_lock,
341
&mi->rli.slave_running,
351
Wait for a slave thread to terminate.
353
This function is called after requesting the thread to terminate
354
(by setting @c abort_slave member of @c Relay_log_info or @c
355
Master_info structure to 1). Termination of the thread is
356
controlled with the the predicate <code>*slave_running</code>.
358
Function will acquire @c term_lock before waiting on the condition
359
unless @c skip_lock is true in which case the mutex should be owned
360
by the caller of this function and will remain acquired after
361
return from the function.
364
Associated lock to use when waiting for @c term_cond
367
Condition that is signalled when the thread has terminated
370
Pointer to predicate to check for slave thread termination
373
If @c true the lock will not be acquired before waiting on
374
the condition. In this case, it is assumed that the calling
375
function acquires the lock before calling this function.
380
terminate_slave_thread(THD *thd,
381
pthread_mutex_t* term_lock,
382
pthread_cond_t* term_cond,
383
volatile uint *slave_running,
388
DBUG_ENTER("terminate_slave_thread");
391
pthread_mutex_lock(term_lock);
393
safe_mutex_assert_owner(term_lock);
398
pthread_mutex_unlock(term_lock);
399
DBUG_RETURN(ER_SLAVE_NOT_RUNNING);
401
DBUG_ASSERT(thd != 0);
402
THD_CHECK_SENTRY(thd);
405
Is is critical to test if the slave is running. Otherwise, we might
406
be referening freed memory trying to kick it
409
while (*slave_running) // Should always be true
411
DBUG_PRINT("loop", ("killing slave thread"));
413
pthread_mutex_lock(&thd->LOCK_delete);
414
#ifndef DONT_USE_THR_ALARM
416
Error codes from pthread_kill are:
417
EINVAL: invalid signal number (can't happen)
418
ESRCH: thread already killed (can happen, should be ignored)
420
IF_DBUG(int err= ) pthread_kill(thd->real_id, thr_client_alarm);
421
DBUG_ASSERT(err != EINVAL);
423
thd->awake(THD::NOT_KILLED);
424
pthread_mutex_unlock(&thd->LOCK_delete);
427
There is a small chance that slave thread might miss the first
428
alarm. To protect againts it, resend the signal until it reacts
430
struct timespec abstime;
431
set_timespec(abstime,2);
432
error= pthread_cond_timedwait(term_cond, term_lock, &abstime);
433
DBUG_ASSERT(error == ETIMEDOUT || error == 0);
436
DBUG_ASSERT(*slave_running == 0);
439
pthread_mutex_unlock(term_lock);
444
int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,
445
pthread_mutex_t *cond_lock,
446
pthread_cond_t *start_cond,
447
volatile uint *slave_running,
448
volatile ulong *slave_run_id,
454
DBUG_ENTER("start_slave_thread");
456
DBUG_ASSERT(mi->inited);
459
pthread_mutex_lock(start_lock);
463
pthread_cond_broadcast(start_cond);
465
pthread_mutex_unlock(start_lock);
466
sql_print_error("Server id not set, will not start slave");
467
DBUG_RETURN(ER_BAD_SLAVE);
473
pthread_cond_broadcast(start_cond);
475
pthread_mutex_unlock(start_lock);
476
DBUG_RETURN(ER_SLAVE_MUST_STOP);
478
start_id= *slave_run_id;
479
DBUG_PRINT("info",("Creating new slave thread"));
481
my_pthread_attr_setprio(&connection_attrib,CONNECT_PRIOR);
482
if (pthread_create(&th, &connection_attrib, h_func, (void*)mi))
485
pthread_mutex_unlock(start_lock);
486
DBUG_RETURN(ER_SLAVE_THREAD);
488
if (start_cond && cond_lock) // caller has cond_lock
490
THD* thd = current_thd;
491
while (start_id == *slave_run_id)
493
DBUG_PRINT("sleep",("Waiting for slave thread to start"));
494
const char* old_msg = thd->enter_cond(start_cond,cond_lock,
495
"Waiting for slave thread to start");
496
pthread_cond_wait(start_cond,cond_lock);
497
thd->exit_cond(old_msg);
498
pthread_mutex_lock(cond_lock); // re-acquire it as exit_cond() released
500
DBUG_RETURN(thd->killed_errno());
504
pthread_mutex_unlock(start_lock);
510
start_slave_threads()
513
SLAVE_FORCE_ALL is not implemented here on purpose since it does not make
514
sense to do that for starting a slave--we always care if it actually
515
started the threads that were not previously running
518
int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
519
Master_info* mi, const char* master_info_fname,
520
const char* slave_info_fname, int thread_mask)
522
pthread_mutex_t *lock_io=0,*lock_sql=0,*lock_cond_io=0,*lock_cond_sql=0;
523
pthread_cond_t* cond_io=0,*cond_sql=0;
525
DBUG_ENTER("start_slave_threads");
527
if (need_slave_mutex)
529
lock_io = &mi->run_lock;
530
lock_sql = &mi->rli.run_lock;
534
cond_io = &mi->start_cond;
535
cond_sql = &mi->rli.start_cond;
536
lock_cond_io = &mi->run_lock;
537
lock_cond_sql = &mi->rli.run_lock;
540
if (thread_mask & SLAVE_IO)
541
error=start_slave_thread(handle_slave_io,lock_io,lock_cond_io,
543
&mi->slave_running, &mi->slave_run_id,
544
mi, 1); //high priority, to read the most possible
545
if (!error && (thread_mask & SLAVE_SQL))
547
error=start_slave_thread(handle_slave_sql,lock_sql,lock_cond_sql,
549
&mi->rli.slave_running, &mi->rli.slave_run_id,
552
terminate_slave_threads(mi, thread_mask & SLAVE_IO, 0);
559
static int end_slave_on_walk(Master_info* mi, uchar* /*unused*/)
561
DBUG_ENTER("end_slave_on_walk");
570
Free all resources used by slave
578
DBUG_ENTER("end_slave");
581
This is called when the server terminates, in close_connections().
582
It terminates slave threads. However, some CHANGE MASTER etc may still be
583
running presently. If a START SLAVE was in progress, the mutex lock below
584
will make us wait until slave threads have started, and START SLAVE
585
returns, then we terminate them here.
587
pthread_mutex_lock(&LOCK_active_mi);
591
TODO: replace the line below with
592
list_walk(&master_list, (list_walk_action)end_slave_on_walk,0);
593
once multi-master code is ready.
595
terminate_slave_threads(active_mi,SLAVE_FORCE_ALL);
596
end_master_info(active_mi);
600
pthread_mutex_unlock(&LOCK_active_mi);
605
static bool io_slave_killed(THD* thd, Master_info* mi)
607
DBUG_ENTER("io_slave_killed");
609
DBUG_ASSERT(mi->io_thd == thd);
610
DBUG_ASSERT(mi->slave_running); // tracking buffer overrun
611
DBUG_RETURN(mi->abort_slave || abort_loop || thd->killed);
615
static bool sql_slave_killed(THD* thd, Relay_log_info* rli)
617
DBUG_ENTER("sql_slave_killed");
619
DBUG_ASSERT(rli->sql_thd == thd);
620
DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun
621
if (abort_loop || thd->killed || rli->abort_slave)
624
If we are in an unsafe situation (stopping could corrupt replication),
625
we give one minute to the slave SQL thread of grace before really
626
terminating, in the hope that it will be able to read more events and
627
the unsafe situation will soon be left. Note that this one minute starts
628
from the last time anything happened in the slave SQL thread. So it's
629
really one minute of idleness, we don't timeout if the slave SQL thread
632
if (rli->last_event_start_time == 0)
634
DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving "
635
"it some grace period"));
636
if (difftime(time(0), rli->last_event_start_time) > 60)
638
rli->report(ERROR_LEVEL, 0,
639
"SQL thread had to stop in an unsafe situation, in "
640
"the middle of applying updates to a "
641
"non-transactional table without any primary key. "
642
"There is a risk of duplicate updates when the slave "
643
"SQL thread is restarted. Please check your tables' "
644
"contents after restart.");
653
skip_load_data_infile()
656
This is used to tell a 3.23 master to break send_file()
659
void skip_load_data_infile(NET *net)
661
DBUG_ENTER("skip_load_data_infile");
663
(void)net_request_file(net, "/dev/null");
664
(void)my_net_read(net); // discard response
665
(void)net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0); // ok
670
bool net_request_file(NET* net, const char* fname)
672
DBUG_ENTER("net_request_file");
673
DBUG_RETURN(net_write_command(net, 251, (uchar*) fname, strlen(fname),
678
From other comments and tests in code, it looks like
679
sometimes Query_log_event and Load_log_event can have db == 0
680
(see rewrite_db() above for example)
681
(cases where this happens are unclear; it may be when the master is 3.23).
684
const char *print_slave_db_safe(const char* db)
686
DBUG_ENTER("*print_slave_db_safe");
688
DBUG_RETURN((db ? db : ""));
691
int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
692
const char *default_val)
695
DBUG_ENTER("init_strvar_from_file");
697
if ((length=my_b_gets(f,var, max_size)))
699
char* last_p = var + length -1;
701
*last_p = 0; // if we stopped on newline, kill it
705
If we truncated a line or stopped on last char, remove all chars
706
up to and including newline.
709
while (((c=my_b_get(f)) != '\n' && c != my_b_EOF));
713
else if (default_val)
715
strmake(var, default_val, max_size-1);
722
int init_intvar_from_file(int* var, IO_CACHE* f, int default_val)
725
DBUG_ENTER("init_intvar_from_file");
728
if (my_b_gets(f, buf, sizeof(buf)))
733
else if (default_val)
741
int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val)
744
DBUG_ENTER("init_floatvar_from_file");
747
if (my_b_gets(f, buf, sizeof(buf)))
749
if (sscanf(buf, "%f", var) != 1)
754
else if (default_val != 0.0)
762
static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info)
764
if (io_slave_killed(thd, mi))
766
if (info && global_system_variables.log_warnings)
767
sql_print_information(info);
775
Note that we rely on the master's version (3.23, 4.0.14 etc) instead of
776
relying on the binlog's version. This is not perfect: imagine an upgrade
777
of the master without waiting that all slaves are in sync with the master;
778
then a slave could be fooled about the binlog's format. This is what happens
779
when people upgrade a 3.23 master to 4.0 without doing RESET MASTER: 4.0
780
slaves are fooled. So we do this only to distinguish between 3.23 and more
781
recent masters (it's too late to change things for 3.23).
788
static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi)
791
String err_msg(error_buf, sizeof(error_buf), &my_charset_bin);
792
char err_buff[MAX_SLAVE_ERRMSG];
793
const char* errmsg= 0;
795
MYSQL_RES *master_res= 0;
796
MYSQL_ROW master_row;
797
DBUG_ENTER("get_master_version_and_clock");
801
Free old description_event_for_queue (that is needed if we are in
804
delete mi->rli.relay_log.description_event_for_queue;
805
mi->rli.relay_log.description_event_for_queue= 0;
807
if (!my_isdigit(&my_charset_bin,*mysql->server_version))
809
errmsg = "Master reported unrecognized MySQL version";
810
err_code= ER_SLAVE_FATAL_ERROR;
811
sprintf(err_buff, ER(err_code), errmsg);
812
err_msg.append(err_buff);
817
Note the following switch will bug when we have MySQL branch 30 ;)
819
switch (*mysql->server_version)
824
errmsg = "Master reported unrecognized MySQL version";
825
err_code= ER_SLAVE_FATAL_ERROR;
826
sprintf(err_buff, ER(err_code), errmsg);
827
err_msg.append(err_buff);
830
mi->rli.relay_log.description_event_for_queue= new
831
Format_description_log_event(1, mysql->server_version);
834
mi->rli.relay_log.description_event_for_queue= new
835
Format_description_log_event(3, mysql->server_version);
839
Master is MySQL >=5.0. Give a default Format_desc event, so that we can
840
take the early steps (like tests for "is this a 3.23 master") which we
841
have to take before we receive the real master's Format_desc which will
842
override this one. Note that the Format_desc we create below is garbage
843
(it has the format of the *slave*); it's only good to help know if the
844
master is 3.23, 4.0, etc.
846
mi->rli.relay_log.description_event_for_queue= new
847
Format_description_log_event(4, mysql->server_version);
853
This does not mean that a 5.0 slave will be able to read a 6.0 master; but
854
as we don't know yet, we don't want to forbid this for now. If a 5.0 slave
855
can't read a 6.0 master, this will show up when the slave can't read some
856
events sent by the master, and there will be error messages.
859
if (err_msg.length() != 0)
862
/* as we are here, we tried to allocate the event */
863
if (!mi->rli.relay_log.description_event_for_queue)
865
errmsg= "default Format_description_log_event";
866
err_code= ER_SLAVE_CREATE_EVENT_FAILURE;
867
sprintf(err_buff, ER(err_code), errmsg);
868
err_msg.append(err_buff);
873
Compare the master and slave's clock. Do not die if master's clock is
874
unavailable (very old master not supporting UNIX_TIMESTAMP()?).
877
if (!mysql_real_query(mysql, STRING_WITH_LEN("SELECT UNIX_TIMESTAMP()")) &&
878
(master_res= mysql_store_result(mysql)) &&
879
(master_row= mysql_fetch_row(master_res)))
881
mi->clock_diff_with_master=
882
(long) (time((time_t*) 0) - strtoul(master_row[0], 0, 10));
884
else if (!check_io_slave_killed(mi->io_thd, mi, NULL))
886
mi->clock_diff_with_master= 0; /* The "most sensible" value */
887
sql_print_warning("\"SELECT UNIX_TIMESTAMP()\" failed on master, "
888
"do not trust column Seconds_Behind_Master of SHOW "
889
"SLAVE STATUS. Error: %s (%d)",
890
mysql_error(mysql), mysql_errno(mysql));
893
mysql_free_result(master_res);
896
Check that the master's server id and ours are different. Because if they
897
are equal (which can result from a simple copy of master's datadir to slave,
898
thus copying some my.cnf), replication will work but all events will be
900
Do not die if SHOW VARIABLES LIKE 'SERVER_ID' fails on master (very old
902
Note: we could have put a @@SERVER_ID in the previous SELECT
903
UNIX_TIMESTAMP() instead, but this would not have worked on 3.23 masters.
905
if (!mysql_real_query(mysql,
906
STRING_WITH_LEN("SHOW VARIABLES LIKE 'SERVER_ID'")) &&
907
(master_res= mysql_store_result(mysql)))
909
if ((master_row= mysql_fetch_row(master_res)) &&
910
(::server_id == strtoul(master_row[1], 0, 10)) &&
911
!mi->rli.replicate_same_server_id)
914
"The slave I/O thread stops because master and slave have equal"
915
" MySQL server ids; these ids must be different for replication to work (or"
916
" the --replicate-same-server-id option must be used on slave but this does"
917
" not always make sense; please check the manual before using it).";
918
err_code= ER_SLAVE_FATAL_ERROR;
919
sprintf(err_buff, ER(err_code), errmsg);
920
err_msg.append(err_buff);
922
mysql_free_result(master_res);
928
Check that the master's global character_set_server and ours are the same.
929
Not fatal if query fails (old master?).
930
Note that we don't check for equality of global character_set_client and
931
collation_connection (neither do we prevent their setting in
932
set_var.cc). That's because from what I (Guilhem) have tested, the global
933
values of these 2 are never used (new connections don't use them).
934
We don't test equality of global collation_database either as it's is
935
going to be deprecated (made read-only) in 4.1 very soon.
936
The test is only relevant if master < 5.0.3 (we'll test only if it's older
937
than the 5 branch; < 5.0.3 was alpha...), as >= 5.0.3 master stores
938
charset info in each binlog event.
939
We don't do it for 3.23 because masters <3.23.50 hang on
940
SELECT @@unknown_var (BUG#7965 - see changelog of 3.23.50). So finally we
941
test only if master is 4.x.
944
/* redundant with rest of code but safer against later additions */
945
if (*mysql->server_version == '3')
948
if ((*mysql->server_version == '4') &&
949
!mysql_real_query(mysql,
950
STRING_WITH_LEN("SELECT @@GLOBAL.COLLATION_SERVER")) &&
951
(master_res= mysql_store_result(mysql)))
953
if ((master_row= mysql_fetch_row(master_res)) &&
954
strcmp(master_row[0], global_system_variables.collation_server->name))
957
"The slave I/O thread stops because master and slave have"
958
" different values for the COLLATION_SERVER global variable."
959
" The values must be equal for replication to work";
960
err_code= ER_SLAVE_FATAL_ERROR;
961
sprintf(err_buff, ER(err_code), errmsg);
962
err_msg.append(err_buff);
964
mysql_free_result(master_res);
970
Perform analogous check for time zone. Theoretically we also should
971
perform check here to verify that SYSTEM time zones are the same on
972
slave and master, but we can't rely on value of @@system_time_zone
973
variable (it is time zone abbreviation) since it determined at start
974
time and so could differ for slave and master even if they are really
975
in the same system time zone. So we are omiting this check and just
976
relying on documentation. Also according to Monty there are many users
977
who are using replication between servers in various time zones. Hence
978
such check will broke everything for them. (And now everything will
979
work for them because by default both their master and slave will have
981
This check is only necessary for 4.x masters (and < 5.0.4 masters but
984
if ((*mysql->server_version == '4') &&
985
!mysql_real_query(mysql, STRING_WITH_LEN("SELECT @@GLOBAL.TIME_ZONE")) &&
986
(master_res= mysql_store_result(mysql)))
988
if ((master_row= mysql_fetch_row(master_res)) &&
989
strcmp(master_row[0],
990
global_system_variables.time_zone->get_name()->ptr()))
993
"The slave I/O thread stops because master and slave have"
994
" different values for the TIME_ZONE global variable."
995
" The values must be equal for replication to work";
996
err_code= ER_SLAVE_FATAL_ERROR;
997
sprintf(err_buff, ER(err_code), errmsg);
998
err_msg.append(err_buff);
1000
mysql_free_result(master_res);
1006
if (mi->heartbeat_period != 0.0)
1009
const char query_format[]= "SET @master_heartbeat_period= %s";
1010
char query[sizeof(query_format) - 2 + sizeof(llbuf)];
1012
the period is an ulonglong of nano-secs.
1014
llstr((ulonglong) (mi->heartbeat_period*1000000000UL), llbuf);
1015
my_sprintf(query, (query, query_format, llbuf));
1017
if (mysql_real_query(mysql, query, strlen(query))
1018
&& !check_io_slave_killed(mi->io_thd, mi, NULL))
1020
err_msg.append("The slave I/O thread stops because querying master with '");
1021
err_msg.append(query);
1022
err_msg.append("' failed;");
1023
err_msg.append(" error: ");
1024
err_code= mysql_errno(mysql);
1025
err_msg.qs_append(err_code);
1026
err_msg.append(" '");
1027
err_msg.append(mysql_error(mysql));
1028
err_msg.append("'");
1029
mysql_free_result(mysql_store_result(mysql));
1032
mysql_free_result(mysql_store_result(mysql));
1036
if (err_msg.length() != 0)
1038
sql_print_error(err_msg.ptr());
1039
DBUG_ASSERT(err_code != 0);
1040
mi->report(ERROR_LEVEL, err_code, err_msg.ptr());
1048
static bool wait_for_relay_log_space(Relay_log_info* rli)
1050
bool slave_killed=0;
1051
Master_info* mi = rli->mi;
1052
const char *save_proc_info;
1053
THD* thd = mi->io_thd;
1054
DBUG_ENTER("wait_for_relay_log_space");
1056
pthread_mutex_lock(&rli->log_space_lock);
1057
save_proc_info= thd->enter_cond(&rli->log_space_cond,
1058
&rli->log_space_lock,
1060
Waiting for the slave SQL thread to free enough relay log space");
1061
while (rli->log_space_limit < rli->log_space_total &&
1062
!(slave_killed=io_slave_killed(thd,mi)) &&
1063
!rli->ignore_log_space_limit)
1064
pthread_cond_wait(&rli->log_space_cond, &rli->log_space_lock);
1065
thd->exit_cond(save_proc_info);
1066
DBUG_RETURN(slave_killed);
1071
Builds a Rotate from the ignored events' info and writes it to relay log.
1074
write_ignored_events_info_to_relay_log()
1075
thd pointer to I/O thread's thd
1079
Slave I/O thread, going to die, must leave a durable trace of the
1080
ignored events' end position for the use of the slave SQL thread, by
1081
calling this function. Only that thread can call it (see assertion).
1083
static void write_ignored_events_info_to_relay_log(THD *thd, Master_info *mi)
1085
Relay_log_info *rli= &mi->rli;
1086
pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
1087
DBUG_ENTER("write_ignored_events_info_to_relay_log");
1089
DBUG_ASSERT(thd == mi->io_thd);
1090
pthread_mutex_lock(log_lock);
1091
if (rli->ign_master_log_name_end[0])
1093
DBUG_PRINT("info",("writing a Rotate event to track down ignored events"));
1094
Rotate_log_event *ev= new Rotate_log_event(rli->ign_master_log_name_end,
1095
0, rli->ign_master_log_pos_end,
1096
Rotate_log_event::DUP_NAME);
1097
rli->ign_master_log_name_end[0]= 0;
1098
/* can unlock before writing as slave SQL thd will soon see our Rotate */
1099
pthread_mutex_unlock(log_lock);
1100
if (likely((bool)ev))
1102
ev->server_id= 0; // don't be ignored by slave SQL thread
1103
if (unlikely(rli->relay_log.append(ev)))
1104
mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
1105
ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
1106
"failed to write a Rotate event"
1107
" to the relay log, SHOW SLAVE STATUS may be"
1109
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
1110
if (flush_master_info(mi, 1))
1111
sql_print_error("Failed to flush master info file");
1115
mi->report(ERROR_LEVEL, ER_SLAVE_CREATE_EVENT_FAILURE,
1116
ER(ER_SLAVE_CREATE_EVENT_FAILURE),
1117
"Rotate_event (out of memory?),"
1118
" SHOW SLAVE STATUS may be inaccurate");
1121
pthread_mutex_unlock(log_lock);
1126
int register_slave_on_master(MYSQL* mysql, Master_info *mi,
1127
bool *suppress_warnings)
1129
uchar buf[1024], *pos= buf;
1130
uint report_host_len, report_user_len=0, report_password_len=0;
1131
DBUG_ENTER("register_slave_on_master");
1133
*suppress_warnings= FALSE;
1136
report_host_len= strlen(report_host);
1138
report_user_len= strlen(report_user);
1139
if (report_password)
1140
report_password_len= strlen(report_password);
1141
/* 30 is a good safety margin */
1142
if (report_host_len + report_user_len + report_password_len + 30 >
1144
DBUG_RETURN(0); // safety
1146
int4store(pos, server_id); pos+= 4;
1147
pos= net_store_data(pos, (uchar*) report_host, report_host_len);
1148
pos= net_store_data(pos, (uchar*) report_user, report_user_len);
1149
pos= net_store_data(pos, (uchar*) report_password, report_password_len);
1150
int2store(pos, (uint16) report_port); pos+= 2;
1151
int4store(pos, rpl_recovery_rank); pos+= 4;
1152
/* The master will fill in master_id */
1153
int4store(pos, 0); pos+= 4;
1155
if (simple_command(mysql, COM_REGISTER_SLAVE, buf, (size_t) (pos- buf), 0))
1157
if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED)
1159
*suppress_warnings= TRUE; // Suppress reconnect warning
1161
else if (!check_io_slave_killed(mi->io_thd, mi, NULL))
1164
my_snprintf(buf, sizeof(buf), "%s (Errno: %d)", mysql_error(mysql),
1165
mysql_errno(mysql));
1166
mi->report(ERROR_LEVEL, ER_SLAVE_MASTER_COM_FAILURE,
1167
ER(ER_SLAVE_MASTER_COM_FAILURE), "COM_REGISTER_SLAVE", buf);
1175
bool show_master_info(THD* thd, Master_info* mi)
1177
// TODO: fix this for multi-master
1178
List<Item> field_list;
1179
Protocol *protocol= thd->protocol;
1180
DBUG_ENTER("show_master_info");
1182
field_list.push_back(new Item_empty_string("Slave_IO_State",
1184
field_list.push_back(new Item_empty_string("Master_Host",
1186
field_list.push_back(new Item_empty_string("Master_User",
1188
field_list.push_back(new Item_return_int("Master_Port", 7,
1190
field_list.push_back(new Item_return_int("Connect_Retry", 10,
1192
field_list.push_back(new Item_empty_string("Master_Log_File",
1194
field_list.push_back(new Item_return_int("Read_Master_Log_Pos", 10,
1195
MYSQL_TYPE_LONGLONG));
1196
field_list.push_back(new Item_empty_string("Relay_Log_File",
1198
field_list.push_back(new Item_return_int("Relay_Log_Pos", 10,
1199
MYSQL_TYPE_LONGLONG));
1200
field_list.push_back(new Item_empty_string("Relay_Master_Log_File",
1202
field_list.push_back(new Item_empty_string("Slave_IO_Running", 3));
1203
field_list.push_back(new Item_empty_string("Slave_SQL_Running", 3));
1204
field_list.push_back(new Item_empty_string("Replicate_Do_DB", 20));
1205
field_list.push_back(new Item_empty_string("Replicate_Ignore_DB", 20));
1206
field_list.push_back(new Item_empty_string("Replicate_Do_Table", 20));
1207
field_list.push_back(new Item_empty_string("Replicate_Ignore_Table", 23));
1208
field_list.push_back(new Item_empty_string("Replicate_Wild_Do_Table", 24));
1209
field_list.push_back(new Item_empty_string("Replicate_Wild_Ignore_Table",
1211
field_list.push_back(new Item_return_int("Last_Errno", 4, MYSQL_TYPE_LONG));
1212
field_list.push_back(new Item_empty_string("Last_Error", 20));
1213
field_list.push_back(new Item_return_int("Skip_Counter", 10,
1215
field_list.push_back(new Item_return_int("Exec_Master_Log_Pos", 10,
1216
MYSQL_TYPE_LONGLONG));
1217
field_list.push_back(new Item_return_int("Relay_Log_Space", 10,
1218
MYSQL_TYPE_LONGLONG));
1219
field_list.push_back(new Item_empty_string("Until_Condition", 6));
1220
field_list.push_back(new Item_empty_string("Until_Log_File", FN_REFLEN));
1221
field_list.push_back(new Item_return_int("Until_Log_Pos", 10,
1222
MYSQL_TYPE_LONGLONG));
1223
field_list.push_back(new Item_empty_string("Master_SSL_Allowed", 7));
1224
field_list.push_back(new Item_empty_string("Master_SSL_CA_File",
1225
sizeof(mi->ssl_ca)));
1226
field_list.push_back(new Item_empty_string("Master_SSL_CA_Path",
1227
sizeof(mi->ssl_capath)));
1228
field_list.push_back(new Item_empty_string("Master_SSL_Cert",
1229
sizeof(mi->ssl_cert)));
1230
field_list.push_back(new Item_empty_string("Master_SSL_Cipher",
1231
sizeof(mi->ssl_cipher)));
1232
field_list.push_back(new Item_empty_string("Master_SSL_Key",
1233
sizeof(mi->ssl_key)));
1234
field_list.push_back(new Item_return_int("Seconds_Behind_Master", 10,
1235
MYSQL_TYPE_LONGLONG));
1236
field_list.push_back(new Item_empty_string("Master_SSL_Verify_Server_Cert",
1238
field_list.push_back(new Item_return_int("Last_IO_Errno", 4, MYSQL_TYPE_LONG));
1239
field_list.push_back(new Item_empty_string("Last_IO_Error", 20));
1240
field_list.push_back(new Item_return_int("Last_SQL_Errno", 4, MYSQL_TYPE_LONG));
1241
field_list.push_back(new Item_empty_string("Last_SQL_Error", 20));
1243
if (protocol->send_fields(&field_list,
1244
Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
1249
DBUG_PRINT("info",("host is set: '%s'", mi->host));
1250
String *packet= &thd->packet;
1251
protocol->prepare_for_resend();
1254
slave_running can be accessed without run_lock but not other
1255
non-volotile members like mi->io_thd, which is guarded by the mutex.
1257
pthread_mutex_lock(&mi->run_lock);
1258
protocol->store(mi->io_thd ? mi->io_thd->proc_info : "", &my_charset_bin);
1259
pthread_mutex_unlock(&mi->run_lock);
1261
pthread_mutex_lock(&mi->data_lock);
1262
pthread_mutex_lock(&mi->rli.data_lock);
1263
protocol->store(mi->host, &my_charset_bin);
1264
protocol->store(mi->user, &my_charset_bin);
1265
protocol->store((uint32) mi->port);
1266
protocol->store((uint32) mi->connect_retry);
1267
protocol->store(mi->master_log_name, &my_charset_bin);
1268
protocol->store((ulonglong) mi->master_log_pos);
1269
protocol->store(mi->rli.group_relay_log_name +
1270
dirname_length(mi->rli.group_relay_log_name),
1272
protocol->store((ulonglong) mi->rli.group_relay_log_pos);
1273
protocol->store(mi->rli.group_master_log_name, &my_charset_bin);
1274
protocol->store(mi->slave_running == MYSQL_SLAVE_RUN_CONNECT ?
1275
"Yes" : "No", &my_charset_bin);
1276
protocol->store(mi->rli.slave_running ? "Yes":"No", &my_charset_bin);
1277
protocol->store(rpl_filter->get_do_db());
1278
protocol->store(rpl_filter->get_ignore_db());
1281
String tmp(buf, sizeof(buf), &my_charset_bin);
1282
rpl_filter->get_do_table(&tmp);
1283
protocol->store(&tmp);
1284
rpl_filter->get_ignore_table(&tmp);
1285
protocol->store(&tmp);
1286
rpl_filter->get_wild_do_table(&tmp);
1287
protocol->store(&tmp);
1288
rpl_filter->get_wild_ignore_table(&tmp);
1289
protocol->store(&tmp);
1291
protocol->store(mi->rli.last_error().number);
1292
protocol->store(mi->rli.last_error().message, &my_charset_bin);
1293
protocol->store((uint32) mi->rli.slave_skip_counter);
1294
protocol->store((ulonglong) mi->rli.group_master_log_pos);
1295
protocol->store((ulonglong) mi->rli.log_space_total);
1298
mi->rli.until_condition==Relay_log_info::UNTIL_NONE ? "None":
1299
( mi->rli.until_condition==Relay_log_info::UNTIL_MASTER_POS? "Master":
1300
"Relay"), &my_charset_bin);
1301
protocol->store(mi->rli.until_log_name, &my_charset_bin);
1302
protocol->store((ulonglong) mi->rli.until_log_pos);
1304
protocol->store(mi->ssl? "Ignored":"No", &my_charset_bin);
1305
protocol->store(mi->ssl_ca, &my_charset_bin);
1306
protocol->store(mi->ssl_capath, &my_charset_bin);
1307
protocol->store(mi->ssl_cert, &my_charset_bin);
1308
protocol->store(mi->ssl_cipher, &my_charset_bin);
1309
protocol->store(mi->ssl_key, &my_charset_bin);
1312
Seconds_Behind_Master: if SQL thread is running and I/O thread is
1313
connected, we can compute it otherwise show NULL (i.e. unknown).
1315
if ((mi->slave_running == MYSQL_SLAVE_RUN_CONNECT) &&
1316
mi->rli.slave_running)
1318
long time_diff= ((long)(time(0) - mi->rli.last_master_timestamp)
1319
- mi->clock_diff_with_master);
1321
Apparently on some systems time_diff can be <0. Here are possible
1322
reasons related to MySQL:
1323
- the master is itself a slave of another master whose time is ahead.
1324
- somebody used an explicit SET TIMESTAMP on the master.
1325
Possible reason related to granularity-to-second of time functions
1326
(nothing to do with MySQL), which can explain a value of -1:
1327
assume the master's and slave's time are perfectly synchronized, and
1328
that at slave's connection time, when the master's timestamp is read,
1329
it is at the very end of second 1, and (a very short time later) when
1330
the slave's timestamp is read it is at the very beginning of second
1331
2. Then the recorded value for master is 1 and the recorded value for
1332
slave is 2. At SHOW SLAVE STATUS time, assume that the difference
1333
between timestamp of slave and rli->last_master_timestamp is 0
1334
(i.e. they are in the same second), then we get 0-(2-1)=-1 as a result.
1335
This confuses users, so we don't go below 0: hence the max().
1337
last_master_timestamp == 0 (an "impossible" timestamp 1970) is a
1338
special marker to say "consider we have caught up".
1340
protocol->store((longlong)(mi->rli.last_master_timestamp ?
1341
max(0, time_diff) : 0));
1345
protocol->store_null();
1347
protocol->store(mi->ssl_verify_server_cert? "Yes":"No", &my_charset_bin);
1350
protocol->store(mi->last_error().number);
1352
protocol->store(mi->last_error().message, &my_charset_bin);
1354
protocol->store(mi->rli.last_error().number);
1356
protocol->store(mi->rli.last_error().message, &my_charset_bin);
1358
pthread_mutex_unlock(&mi->rli.data_lock);
1359
pthread_mutex_unlock(&mi->data_lock);
1361
if (my_net_write(&thd->net, (uchar*) thd->packet.ptr(), packet->length()))
1369
void set_slave_thread_options(THD* thd)
1371
DBUG_ENTER("set_slave_thread_options");
1373
It's nonsense to constrain the slave threads with max_join_size; if a
1374
query succeeded on master, we HAVE to execute it. So set
1375
OPTION_BIG_SELECTS. Setting max_join_size to HA_POS_ERROR is not enough
1376
(and it's not needed if we have OPTION_BIG_SELECTS) because an INSERT
1377
SELECT examining more than 4 billion rows would still fail (yes, because
1378
when max_join_size is 4G, OPTION_BIG_SELECTS is automatically set, but
1379
only for client threads.
1381
ulonglong options= thd->options | OPTION_BIG_SELECTS;
1382
if (opt_log_slave_updates)
1383
options|= OPTION_BIN_LOG;
1385
options&= ~OPTION_BIN_LOG;
1386
thd->options= options;
1387
thd->variables.completion_type= 0;
1391
void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli)
1393
DBUG_ENTER("set_slave_thread_default_charset");
1395
thd->variables.character_set_client=
1396
global_system_variables.character_set_client;
1397
thd->variables.collation_connection=
1398
global_system_variables.collation_connection;
1399
thd->variables.collation_server=
1400
global_system_variables.collation_server;
1401
thd->update_charset();
1404
We use a const cast here since the conceptual (and externally
1405
visible) behavior of the function is to set the default charset of
1406
the thread. That the cache has to be invalidated is a secondary
1409
const_cast<Relay_log_info*>(rli)->cached_charset_invalidate();
1417
static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
1419
DBUG_ENTER("init_slave_thread");
1420
#if !defined(DBUG_OFF)
1421
int simulate_error= 0;
1423
thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
1424
SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO;
1425
thd->security_ctx->skip_grants();
1426
my_net_init(&thd->net, 0);
1428
Adding MAX_LOG_EVENT_HEADER_LEN to the max_allowed_packet on all
1429
slave threads, since a replication event can become this much larger
1430
than the corresponding packet (query) sent from client to master.
1432
thd->variables.max_allowed_packet= global_system_variables.max_allowed_packet
1433
+ MAX_LOG_EVENT_HEADER; /* note, incr over the global not session var */
1434
thd->slave_thread = 1;
1435
thd->enable_slow_log= opt_log_slow_slave_statements;
1436
set_slave_thread_options(thd);
1437
thd->client_capabilities = CLIENT_LOCAL_FILES;
1438
pthread_mutex_lock(&LOCK_thread_count);
1439
thd->thread_id= thd->variables.pseudo_thread_id= thread_id++;
1440
pthread_mutex_unlock(&LOCK_thread_count);
1442
DBUG_EXECUTE_IF("simulate_io_slave_error_on_init",
1443
simulate_error|= (1 << SLAVE_THD_IO););
1444
DBUG_EXECUTE_IF("simulate_sql_slave_error_on_init",
1445
simulate_error|= (1 << SLAVE_THD_SQL););
1446
#if !defined(DBUG_OFF)
1447
if (init_thr_lock() || thd->store_globals() || simulate_error & (1<< thd_type))
1449
if (init_thr_lock() || thd->store_globals())
1457
if (thd_type == SLAVE_THD_SQL)
1458
thd_proc_info(thd, "Waiting for the next event in relay log");
1460
thd_proc_info(thd, "Waiting for master update");
1461
thd->version=refresh_version;
1467
static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed,
1468
void* thread_killed_arg)
1471
thr_alarm_t alarmed;
1472
DBUG_ENTER("safe_sleep");
1474
thr_alarm_init(&alarmed);
1475
time_t start_time= my_time(0);
1476
time_t end_time= start_time+sec;
1478
while ((nap_time= (int) (end_time - start_time)) > 0)
1482
The only reason we are asking for alarm is so that
1483
we will be woken up in case of murder, so if we do not get killed,
1484
set the alarm so it goes off after we wake up naturally
1486
thr_alarm(&alarmed, 2 * nap_time, &alarm_buff);
1488
thr_end_alarm(&alarmed);
1490
if ((*thread_killed)(thd,thread_killed_arg))
1492
start_time= my_time(0);
1498
static int request_dump(MYSQL* mysql, Master_info* mi,
1499
bool *suppress_warnings)
1501
uchar buf[FN_REFLEN + 10];
1503
int binlog_flags = 0; // for now
1504
char* logname = mi->master_log_name;
1505
DBUG_ENTER("request_dump");
1507
*suppress_warnings= FALSE;
1509
// TODO if big log files: Change next to int8store()
1510
int4store(buf, (ulong) mi->master_log_pos);
1511
int2store(buf + 4, binlog_flags);
1512
int4store(buf + 6, server_id);
1513
len = (uint) strlen(logname);
1514
memcpy(buf + 10, logname,len);
1515
if (simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1))
1518
Something went wrong, so we will just reconnect and retry later
1519
in the future, we should do a better error analysis, but for
1520
now we just fill up the error log :-)
1522
if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED)
1523
*suppress_warnings= TRUE; // Suppress reconnect warning
1525
sql_print_error("Error on COM_BINLOG_DUMP: %d %s, will retry in %d secs",
1526
mysql_errno(mysql), mysql_error(mysql),
1535
Read one event from the master
1539
mysql MySQL connection
1540
mi Master connection information
1541
suppress_warnings TRUE when a normal net read timeout has caused us to
1542
try a reconnect. We do not want to print anything to
1543
the error log in this case because this a anormal
1544
event in an idle server.
1547
'packet_error' Error
1548
number Length of packet
1551
static ulong read_event(MYSQL* mysql, Master_info *mi, bool* suppress_warnings)
1554
DBUG_ENTER("read_event");
1556
*suppress_warnings= FALSE;
1558
my_real_read() will time us out
1559
We check if we were told to die, and if not, try reading again
1562
if (disconnect_slave_event_count && !(mi->events_till_disconnect--))
1563
DBUG_RETURN(packet_error);
1566
len = cli_safe_read(mysql);
1567
if (len == packet_error || (long) len < 1)
1569
if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED)
1572
We are trying a normal reconnect after a read timeout;
1573
we suppress prints to .err file as long as the reconnect
1574
happens without problems
1576
*suppress_warnings= TRUE;
1579
sql_print_error("Error reading packet from server: %s ( server_errno=%d)",
1580
mysql_error(mysql), mysql_errno(mysql));
1581
DBUG_RETURN(packet_error);
1584
/* Check if eof packet */
1585
if (len < 8 && mysql->net.read_pos[0] == 254)
1587
sql_print_information("Slave: received end packet from server, apparent "
1588
"master shutdown: %s",
1589
mysql_error(mysql));
1590
DBUG_RETURN(packet_error);
1593
DBUG_PRINT("exit", ("len: %lu net->read_pos[4]: %d",
1594
len, mysql->net.read_pos[4]));
1595
DBUG_RETURN(len - 1);
1599
int check_expected_error(THD* thd, Relay_log_info const *rli,
1602
DBUG_ENTER("check_expected_error");
1604
switch (expected_error) {
1605
case ER_NET_READ_ERROR:
1606
case ER_NET_ERROR_ON_WRITE:
1607
case ER_QUERY_INTERRUPTED:
1608
case ER_SERVER_SHUTDOWN:
1609
case ER_NEW_ABORTING_CONNECTION:
1618
Check if the current error is of temporary nature of not.
1619
Some errors are temporary in nature, such as
1620
ER_LOCK_DEADLOCK and ER_LOCK_WAIT_TIMEOUT. Ndb also signals
1621
that the error is temporary by pushing a warning with the error code
1622
ER_GET_TEMPORARY_ERRMSG, if the originating error is temporary.
1624
static int has_temporary_error(THD *thd)
1626
DBUG_ENTER("has_temporary_error");
1628
if (thd->is_fatal_error)
1631
DBUG_EXECUTE_IF("all_errors_are_temporary_errors",
1632
if (thd->main_da.is_error())
1635
my_error(ER_LOCK_DEADLOCK, MYF(0));
1639
If there is no message in THD, we can't say if it's a temporary
1640
error or not. This is currently the case for Incident_log_event,
1641
which sets no message. Return FALSE.
1643
if (!thd->is_error())
1647
Temporary error codes:
1648
currently, InnoDB deadlock detected by InnoDB or lock
1649
wait timeout (innodb_lock_wait_timeout exceeded
1651
if (thd->main_da.sql_errno() == ER_LOCK_DEADLOCK ||
1652
thd->main_da.sql_errno() == ER_LOCK_WAIT_TIMEOUT)
1660
Applies the given event and advances the relay log position.
1662
In essence, this function does:
1665
ev->apply_event(rli);
1666
ev->update_pos(rli);
1669
But it also does some maintainance, such as skipping events if
1670
needed and reporting errors.
1672
If the @c skip flag is set, then it is tested whether the event
1673
should be skipped, by looking at the slave_skip_counter and the
1674
server id. The skip flag should be set when calling this from a
1675
replication thread but not set when executing an explicit BINLOG
1680
@retval 1 Error calling ev->apply_event().
1682
@retval 2 No error calling ev->apply_event(), but error calling
1685
int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli,
1690
DBUG_ENTER("apply_event_and_update_pos");
1692
DBUG_PRINT("exec_event",("%s(type_code: %d; server_id: %d)",
1693
ev->get_type_str(), ev->get_type_code(),
1695
DBUG_PRINT("info", ("thd->options: %s%s; rli->last_event_start_time: %lu",
1696
FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT),
1697
FLAGSTR(thd->options, OPTION_BEGIN),
1698
rli->last_event_start_time));
1701
Execute the event to change the database and update the binary
1702
log coordinates, but first we set some data that is needed for
1705
The event will be executed unless it is supposed to be skipped.
1707
Queries originating from this server must be skipped. Low-level
1708
events (Format_description_log_event, Rotate_log_event,
1709
Stop_log_event) from this server must also be skipped. But for
1710
those we don't want to modify 'group_master_log_pos', because
1711
these events did not exist on the master.
1712
Format_description_log_event is not completely skipped.
1714
Skip queries specified by the user in 'slave_skip_counter'. We
1715
can't however skip events that has something to do with the log
1718
Filtering on own server id is extremely important, to ignore
1719
execution of events created by the creation/rotation of the relay
1720
log (remember that now the relay log starts with its Format_desc,
1724
thd->server_id = ev->server_id; // use the original server id for logging
1725
thd->set_time(); // time the query
1726
thd->lex->current_select= 0;
1728
ev->when= my_time(0);
1729
ev->thd = thd; // because up to this point, ev->thd == 0
1733
int reason= ev->shall_skip(rli);
1734
if (reason == Log_event::EVENT_SKIP_COUNT)
1735
--rli->slave_skip_counter;
1736
pthread_mutex_unlock(&rli->data_lock);
1737
if (reason == Log_event::EVENT_SKIP_NOT)
1738
exec_res= ev->apply_event(rli);
1741
This only prints information to the debug trace.
1743
TODO: Print an informational message to the error log?
1745
static const char *const explain[] = {
1748
// EVENT_SKIP_IGNORE,
1749
"skipped because event should be ignored",
1751
"skipped because event skip counter was non-zero"
1753
DBUG_PRINT("info", ("OPTION_BEGIN: %d; IN_STMT: %d",
1754
thd->options & OPTION_BEGIN ? 1 : 0,
1755
rli->get_flag(Relay_log_info::IN_STMT)));
1756
DBUG_PRINT("skip_event", ("%s event was %s",
1757
ev->get_type_str(), explain[reason]));
1761
exec_res= ev->apply_event(rli);
1763
DBUG_PRINT("info", ("apply_event error = %d", exec_res));
1766
int error= ev->update_pos(rli);
1774
DBUG_PRINT("info", ("update_pos error = %d", error));
1775
DBUG_PRINT("info", ("group %s %s",
1776
llstr(rli->group_relay_log_pos, buf),
1777
rli->group_relay_log_name));
1778
DBUG_PRINT("info", ("event %s %s",
1779
llstr(rli->event_relay_log_pos, buf),
1780
rli->event_relay_log_name));
1783
The update should not fail, so print an error message and
1784
return an error code.
1786
TODO: Replace this with a decent error message when merged
1787
with BUG#24954 (which adds several new error message).
1792
rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR,
1793
"It was not possible to update the positions"
1794
" of the relay log information: the slave may"
1795
" be in an inconsistent state."
1796
" Stopped in %s position %s",
1797
rli->group_relay_log_name,
1798
llstr(rli->group_relay_log_pos, buf));
1803
DBUG_RETURN(exec_res ? 1 : 0);
1808
Top-level function for executing the next event from the relay log.
1810
This function reads the event from the relay log, executes it, and
1811
advances the relay log position. It also handles errors, etc.
1813
This function may fail to apply the event for the following reasons:
1815
- The position specfied by the UNTIL condition of the START SLAVE
1818
- It was not possible to read the event from the log.
1820
- The slave is killed.
1822
- An error occurred when applying the event, and the event has been
1823
tried slave_trans_retries times. If the event has been retried
1824
fewer times, 0 is returned.
1826
- init_master_info or init_relay_log_pos failed. (These are called
1827
if a failure occurs when applying the event.)</li>
1829
- An error occurred when updating the binlog position.
1831
@retval 0 The event was applied.
1833
@retval 1 The event was not applied.
1835
static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
1837
DBUG_ENTER("exec_relay_log_event");
1840
We acquire this mutex since we need it for all operations except
1841
event execution. But we will release it in places where we will
1842
wait for something for example inside of next_event().
1844
pthread_mutex_lock(&rli->data_lock);
1846
Log_event * ev = next_event(rli);
1848
DBUG_ASSERT(rli->sql_thd==thd);
1850
if (sql_slave_killed(thd,rli))
1852
pthread_mutex_unlock(&rli->data_lock);
1861
This tests if the position of the beginning of the current event
1862
hits the UNTIL barrier.
1864
if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
1865
rli->is_until_satisfied((rli->is_in_group() || !ev->log_pos) ?
1866
rli->group_master_log_pos :
1867
ev->log_pos - ev->data_written))
1870
sql_print_information("Slave SQL thread stopped because it reached its"
1871
" UNTIL position %s", llstr(rli->until_pos(), buf));
1873
Setting abort_slave flag because we do not want additional message about
1874
error in query execution to be printed.
1876
rli->abort_slave= 1;
1877
pthread_mutex_unlock(&rli->data_lock);
1881
exec_res= apply_event_and_update_pos(ev, thd, rli, TRUE);
1884
Format_description_log_event should not be deleted because it will be
1885
used to read info about the relay log's format; it will be deleted when
1886
the SQL thread does not need it, i.e. when this thread terminates.
1888
if (ev->get_type_code() != FORMAT_DESCRIPTION_EVENT)
1890
DBUG_PRINT("info", ("Deleting the event after it has been executed"));
1895
update_log_pos failed: this should not happen, so we don't
1901
if (slave_trans_retries)
1904
if (exec_res && (temp_err= has_temporary_error(thd)))
1908
We were in a transaction which has been rolled back because of a
1910
let's seek back to BEGIN log event and retry it all again.
1911
Note, if lock wait timeout (innodb_lock_wait_timeout exceeded)
1912
there is no rollback since 5.0.13 (ref: manual).
1913
We have to not only seek but also
1914
a) init_master_info(), to seek back to hot relay log's start for later
1915
(for when we will come back to this hot log after re-processing the
1916
possibly existing old logs where BEGIN is: check_binlog_magic() will
1917
then need the cache to be at position 0 (see comments at beginning of
1918
init_master_info()).
1919
b) init_relay_log_pos(), because the BEGIN may be an older relay log.
1921
if (rli->trans_retries < slave_trans_retries)
1923
if (init_master_info(rli->mi, 0, 0, 0, SLAVE_SQL))
1924
sql_print_error("Failed to initialize the master info structure");
1925
else if (init_relay_log_pos(rli,
1926
rli->group_relay_log_name,
1927
rli->group_relay_log_pos,
1929
sql_print_error("Error initializing relay log position: %s",
1934
end_trans(thd, ROLLBACK);
1935
/* chance for concurrent connection to get more locks */
1936
safe_sleep(thd, min(rli->trans_retries, MAX_SLAVE_RETRY_PAUSE),
1937
(CHECK_KILLED_FUNC)sql_slave_killed, (void*)rli);
1938
pthread_mutex_lock(&rli->data_lock); // because of SHOW STATUS
1939
rli->trans_retries++;
1940
rli->retried_trans++;
1941
pthread_mutex_unlock(&rli->data_lock);
1942
DBUG_PRINT("info", ("Slave retries transaction "
1943
"rli->trans_retries: %lu", rli->trans_retries));
1947
sql_print_error("Slave SQL thread retried transaction %lu time(s) "
1948
"in vain, giving up. Consider raising the value of "
1949
"the slave_transaction_retries variable.",
1950
slave_trans_retries);
1952
else if (exec_res && !temp_err ||
1953
(opt_using_transactions &&
1954
rli->group_relay_log_pos == rli->event_relay_log_pos))
1957
Only reset the retry counter if the entire group succeeded
1958
or failed with a non-transient error. On a successful
1959
event, the execution will proceed as usual; in the case of a
1960
non-transient error, the slave will stop with an error.
1962
rli->trans_retries= 0; // restart from fresh
1963
DBUG_PRINT("info", ("Resetting retry counter, rli->trans_retries: %lu",
1964
rli->trans_retries));
1967
DBUG_RETURN(exec_res);
1969
pthread_mutex_unlock(&rli->data_lock);
1970
rli->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_READ_FAILURE,
1971
ER(ER_SLAVE_RELAY_LOG_READ_FAILURE), "\
1972
Could not parse relay log event entry. The possible reasons are: the master's \
1973
binary log is corrupted (you can check this by running 'mysqlbinlog' on the \
1974
binary log), the slave's relay log is corrupted (you can check this by running \
1975
'mysqlbinlog' on the relay log), a network problem, or a bug in the master's \
1976
or slave's MySQL code. If you want to check the master's binary log or slave's \
1977
relay log, you will be able to know their names by issuing 'SHOW SLAVE STATUS' \
1985
@brief Try to reconnect slave IO thread.
1987
@details Terminates current connection to master, sleeps for
1988
@c mi->connect_retry msecs and initiates new connection with
1989
@c safe_reconnect(). Variable pointed by @c retry_count is increased -
1990
if it exceeds @c master_retry_count then connection is not re-established
1991
and function signals error.
1992
Unless @c suppres_warnings is TRUE, a warning is put in the server error log
1993
when reconnecting. The warning message and messages used to report errors
1994
are taken from @c messages array. In case @c master_retry_count is exceeded,
1995
no messages are added to the log.
1997
@param[in] thd Thread context.
1998
@param[in] mysql MySQL connection.
1999
@param[in] mi Master connection information.
2000
@param[in,out] retry_count Number of attempts to reconnect.
2001
@param[in] suppress_warnings TRUE when a normal net read timeout
2002
has caused to reconnecting.
2003
@param[in] messages Messages to print/log, see
2004
reconnect_messages[] array.
2007
@retval 1 There was an error.
2010
static int try_to_reconnect(THD *thd, MYSQL *mysql, Master_info *mi,
2011
uint *retry_count, bool suppress_warnings,
2012
const char *messages[SLAVE_RECON_MSG_MAX])
2014
mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT;
2015
thd->proc_info= messages[SLAVE_RECON_MSG_WAIT];
2016
#ifdef SIGNAL_WITH_VIO_CLOSE
2017
thd->clear_active_vio();
2020
if ((*retry_count)++)
2022
if (*retry_count > master_retry_count)
2023
return 1; // Don't retry forever
2024
safe_sleep(thd, mi->connect_retry, (CHECK_KILLED_FUNC) io_slave_killed,
2027
if (check_io_slave_killed(thd, mi, messages[SLAVE_RECON_MSG_KILLED_WAITING]))
2029
thd->proc_info = messages[SLAVE_RECON_MSG_AFTER];
2030
if (!suppress_warnings)
2032
char buf[256], llbuff[22];
2033
my_snprintf(buf, sizeof(buf), messages[SLAVE_RECON_MSG_FAILED],
2034
IO_RPL_LOG_NAME, llstr(mi->master_log_pos, llbuff));
2036
Raise a warining during registering on master/requesting dump.
2037
Log a message reading event.
2039
if (messages[SLAVE_RECON_MSG_COMMAND][0])
2041
mi->report(WARNING_LEVEL, ER_SLAVE_MASTER_COM_FAILURE,
2042
ER(ER_SLAVE_MASTER_COM_FAILURE),
2043
messages[SLAVE_RECON_MSG_COMMAND], buf);
2047
sql_print_information(buf);
2050
if (safe_reconnect(thd, mysql, mi, 1) || io_slave_killed(thd, mi))
2052
if (global_system_variables.log_warnings)
2053
sql_print_information(messages[SLAVE_RECON_MSG_KILLED_AFTER]);
2060
/* Slave I/O Thread entry point */
2062
pthread_handler_t handle_slave_io(void *arg)
2064
THD *thd; // needs to be first for thread_stack
2066
Master_info *mi = (Master_info*)arg;
2067
Relay_log_info *rli= &mi->rli;
2070
bool suppress_warnings;
2072
uint retry_count_reg= 0, retry_count_dump= 0, retry_count_event= 0;
2074
// needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
2076
DBUG_ENTER("handle_slave_io");
2078
DBUG_ASSERT(mi->inited);
2082
pthread_mutex_lock(&mi->run_lock);
2083
/* Inform waiting threads that slave has started */
2087
mi->events_till_disconnect = disconnect_slave_event_count;
2090
thd= new THD; // note that contructor of THD uses DBUG_ !
2091
THD_CHECK_SENTRY(thd);
2094
pthread_detach_this_thread();
2095
thd->thread_stack= (char*) &thd; // remember where our stack is
2096
if (init_slave_thread(thd, SLAVE_THD_IO))
2098
pthread_cond_broadcast(&mi->start_cond);
2099
pthread_mutex_unlock(&mi->run_lock);
2100
sql_print_error("Failed during slave I/O thread initialization");
2103
pthread_mutex_lock(&LOCK_thread_count);
2104
threads.append(thd);
2105
pthread_mutex_unlock(&LOCK_thread_count);
2106
mi->slave_running = 1;
2107
mi->abort_slave = 0;
2108
pthread_mutex_unlock(&mi->run_lock);
2109
pthread_cond_broadcast(&mi->start_cond);
2111
DBUG_PRINT("master_info",("log_file_name: '%s' position: %s",
2112
mi->master_log_name,
2113
llstr(mi->master_log_pos,llbuff)));
2115
if (!(mi->mysql = mysql = mysql_init(NULL)))
2117
mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
2118
ER(ER_SLAVE_FATAL_ERROR), "error in mysql_init()");
2122
thd_proc_info(thd, "Connecting to master");
2123
// we can get killed during safe_connect
2124
if (!safe_connect(thd, mysql, mi))
2126
sql_print_information("Slave I/O thread: connected to master '%s@%s:%d',"
2127
"replication started in log '%s' at position %s",
2128
mi->user, mi->host, mi->port,
2130
llstr(mi->master_log_pos,llbuff));
2132
Adding MAX_LOG_EVENT_HEADER_LEN to the max_packet_size on the I/O
2133
thread, since a replication event can become this much larger than
2134
the corresponding packet (query) sent from client to master.
2136
mysql->net.max_packet_size= thd->net.max_packet_size+= MAX_LOG_EVENT_HEADER;
2140
sql_print_information("Slave I/O thread killed while connecting to master");
2146
// TODO: the assignment below should be under mutex (5.0)
2147
mi->slave_running= MYSQL_SLAVE_RUN_CONNECT;
2148
thd->slave_net = &mysql->net;
2149
thd_proc_info(thd, "Checking master version");
2150
if (get_master_version_and_clock(mysql, mi))
2153
if (mi->rli.relay_log.description_event_for_queue->binlog_version > 1)
2156
Register ourselves with the master.
2158
thd_proc_info(thd, "Registering slave on master");
2159
if (register_slave_on_master(mysql, mi, &suppress_warnings))
2161
if (!check_io_slave_killed(thd, mi, "Slave I/O thread killed "
2162
"while registering slave on master"))
2164
sql_print_error("Slave I/O thread couldn't register on master");
2165
if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2166
reconnect_messages[SLAVE_RECON_ACT_REG]))
2173
DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_REG",
2174
if (!retry_count_reg)
2177
sql_print_information("Forcing to reconnect slave I/O thread");
2178
if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2179
reconnect_messages[SLAVE_RECON_ACT_REG]))
2185
DBUG_PRINT("info",("Starting reading binary log from master"));
2186
while (!io_slave_killed(thd,mi))
2188
thd_proc_info(thd, "Requesting binlog dump");
2189
if (request_dump(mysql, mi, &suppress_warnings))
2191
sql_print_error("Failed on request_dump()");
2192
if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \
2193
requesting master dump") ||
2194
try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2195
reconnect_messages[SLAVE_RECON_ACT_DUMP]))
2199
DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_DUMP",
2200
if (!retry_count_dump)
2203
sql_print_information("Forcing to reconnect slave I/O thread");
2204
if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2205
reconnect_messages[SLAVE_RECON_ACT_DUMP]))
2210
while (!io_slave_killed(thd,mi))
2214
We say "waiting" because read_event() will wait if there's nothing to
2215
read. But if there's something to read, it will not wait. The
2216
important thing is to not confuse users by saying "reading" whereas
2217
we're in fact receiving nothing.
2219
thd_proc_info(thd, "Waiting for master to send event");
2220
event_len= read_event(mysql, mi, &suppress_warnings);
2221
if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \
2224
DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_EVENT",
2225
if (!retry_count_event)
2227
retry_count_event++;
2228
sql_print_information("Forcing to reconnect slave I/O thread");
2229
if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2230
reconnect_messages[SLAVE_RECON_ACT_EVENT]))
2235
if (event_len == packet_error)
2237
uint mysql_error_number= mysql_errno(mysql);
2238
switch (mysql_error_number) {
2239
case CR_NET_PACKET_TOO_LARGE:
2241
Log entry on master is longer than max_allowed_packet (%ld) on \
2242
slave. If the entry is correct, restart the server with a higher value of \
2243
max_allowed_packet",
2244
thd->variables.max_allowed_packet);
2246
case ER_MASTER_FATAL_ERROR_READING_BINLOG:
2247
sql_print_error(ER(mysql_error_number), mysql_error_number,
2248
mysql_error(mysql));
2250
case EE_OUTOFMEMORY:
2251
case ER_OUTOFMEMORY:
2253
Stopping slave I/O thread due to out-of-memory error from master");
2256
if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2257
reconnect_messages[SLAVE_RECON_ACT_EVENT]))
2260
} // if (event_len == packet_error)
2262
retry_count=0; // ok event, reset retry counter
2263
thd_proc_info(thd, "Queueing master event to the relay log");
2264
if (queue_event(mi,(const char*)mysql->net.read_pos + 1, event_len))
2268
if (flush_master_info(mi, 1))
2270
sql_print_error("Failed to flush master info file");
2274
See if the relay logs take too much space.
2275
We don't lock mi->rli.log_space_lock here; this dirty read saves time
2276
and does not introduce any problem:
2277
- if mi->rli.ignore_log_space_limit is 1 but becomes 0 just after (so
2278
the clean value is 0), then we are reading only one more event as we
2279
should, and we'll block only at the next event. No big deal.
2280
- if mi->rli.ignore_log_space_limit is 0 but becomes 1 just after (so
2281
the clean value is 1), then we are going into wait_for_relay_log_space()
2282
for no reason, but this function will do a clean read, notice the clean
2283
value and exit immediately.
2287
char llbuf1[22], llbuf2[22];
2288
DBUG_PRINT("info", ("log_space_limit=%s log_space_total=%s \
2289
ignore_log_space_limit=%d",
2290
llstr(rli->log_space_limit,llbuf1),
2291
llstr(rli->log_space_total,llbuf2),
2292
(int) rli->ignore_log_space_limit));
2296
if (rli->log_space_limit && rli->log_space_limit <
2297
rli->log_space_total &&
2298
!rli->ignore_log_space_limit)
2299
if (wait_for_relay_log_space(rli))
2301
sql_print_error("Slave I/O thread aborted while waiting for relay \
2310
// print the current replication position
2311
sql_print_information("Slave I/O thread exiting, read up to log '%s', position %s",
2312
IO_RPL_LOG_NAME, llstr(mi->master_log_pos,llbuff));
2313
VOID(pthread_mutex_lock(&LOCK_thread_count));
2314
thd->query = thd->db = 0; // extra safety
2315
thd->query_length= thd->db_length= 0;
2316
VOID(pthread_mutex_unlock(&LOCK_thread_count));
2320
Here we need to clear the active VIO before closing the
2321
connection with the master. The reason is that THD::awake()
2322
might be called from terminate_slave_thread() because somebody
2323
issued a STOP SLAVE. If that happends, the close_active_vio()
2324
can be called in the middle of closing the VIO associated with
2325
the 'mysql' object, causing a crash.
2327
#ifdef SIGNAL_WITH_VIO_CLOSE
2328
thd->clear_active_vio();
2333
write_ignored_events_info_to_relay_log(thd, mi);
2334
thd_proc_info(thd, "Waiting for slave mutex on exit");
2335
pthread_mutex_lock(&mi->run_lock);
2337
/* Forget the relay log's format */
2338
delete mi->rli.relay_log.description_event_for_queue;
2339
mi->rli.relay_log.description_event_for_queue= 0;
2340
// TODO: make rpl_status part of Master_info
2341
change_rpl_status(RPL_ACTIVE_SLAVE,RPL_IDLE_SLAVE);
2342
DBUG_ASSERT(thd->net.buff != 0);
2343
net_end(&thd->net); // destructor will not free it, because net.vio is 0
2344
close_thread_tables(thd);
2345
pthread_mutex_lock(&LOCK_thread_count);
2346
THD_CHECK_SENTRY(thd);
2348
pthread_mutex_unlock(&LOCK_thread_count);
2350
mi->slave_running= 0;
2353
Note: the order of the two following calls (first broadcast, then unlock)
2354
is important. Otherwise a killer_thread can execute between the calls and
2355
delete the mi structure leading to a crash! (see BUG#25306 for details)
2357
pthread_cond_broadcast(&mi->stop_cond); // tell the world we are done
2358
pthread_mutex_unlock(&mi->run_lock);
2361
DBUG_RETURN(0); // Can't return anything here
2365
/* Slave SQL Thread entry point */
2367
pthread_handler_t handle_slave_sql(void *arg)
2369
THD *thd; /* needs to be first for thread_stack */
2370
char llbuff[22],llbuff1[22];
2372
Relay_log_info* rli = &((Master_info*)arg)->rli;
2375
// needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
2377
DBUG_ENTER("handle_slave_sql");
2379
DBUG_ASSERT(rli->inited);
2380
pthread_mutex_lock(&rli->run_lock);
2381
DBUG_ASSERT(!rli->slave_running);
2384
rli->events_till_abort = abort_slave_event_count;
2387
thd = new THD; // note that contructor of THD uses DBUG_ !
2388
thd->thread_stack = (char*)&thd; // remember where our stack is
2391
/* Inform waiting threads that slave has started */
2392
rli->slave_run_id++;
2393
rli->slave_running = 1;
2395
pthread_detach_this_thread();
2396
if (init_slave_thread(thd, SLAVE_THD_SQL))
2399
TODO: this is currently broken - slave start and change master
2400
will be stuck if we fail here
2402
pthread_cond_broadcast(&rli->start_cond);
2403
pthread_mutex_unlock(&rli->run_lock);
2404
sql_print_error("Failed during slave thread initialization");
2407
thd->init_for_queries();
2408
thd->temporary_tables = rli->save_temporary_tables; // restore temp tables
2409
pthread_mutex_lock(&LOCK_thread_count);
2410
threads.append(thd);
2411
pthread_mutex_unlock(&LOCK_thread_count);
2413
We are going to set slave_running to 1. Assuming slave I/O thread is
2414
alive and connected, this is going to make Seconds_Behind_Master be 0
2415
i.e. "caught up". Even if we're just at start of thread. Well it's ok, at
2416
the moment we start we can think we are caught up, and the next second we
2417
start receiving data so we realize we are not caught up and
2418
Seconds_Behind_Master grows. No big deal.
2420
rli->abort_slave = 0;
2421
pthread_mutex_unlock(&rli->run_lock);
2422
pthread_cond_broadcast(&rli->start_cond);
2425
Reset errors for a clean start (otherwise, if the master is idle, the SQL
2426
thread may execute no Query_log_event, so the error will remain even
2427
though there's no problem anymore). Do not reset the master timestamp
2428
(imagine the slave has caught everything, the STOP SLAVE and START SLAVE:
2429
as we are not sure that we are going to receive a query, we want to
2430
remember the last master timestamp (to say how many seconds behind we are
2432
But the master timestamp is reset by RESET SLAVE & CHANGE MASTER.
2436
//tell the I/O thread to take relay_log_space_limit into account from now on
2437
pthread_mutex_lock(&rli->log_space_lock);
2438
rli->ignore_log_space_limit= 0;
2439
pthread_mutex_unlock(&rli->log_space_lock);
2440
rli->trans_retries= 0; // start from "no error"
2441
DBUG_PRINT("info", ("rli->trans_retries: %lu", rli->trans_retries));
2443
if (init_relay_log_pos(rli,
2444
rli->group_relay_log_name,
2445
rli->group_relay_log_pos,
2446
1 /*need data lock*/, &errmsg,
2447
1 /*look for a description_event*/))
2449
sql_print_error("Error initializing relay log position: %s",
2453
THD_CHECK_SENTRY(thd);
2456
char llbuf1[22], llbuf2[22];
2457
DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
2458
llstr(my_b_tell(rli->cur_log),llbuf1),
2459
llstr(rli->event_relay_log_pos,llbuf2)));
2460
DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
2462
Wonder if this is correct. I (Guilhem) wonder if my_b_tell() returns the
2463
correct position when it's called just after my_b_seek() (the questionable
2464
stuff is those "seek is done on next read" comments in the my_b_seek()
2466
The crude reality is that this assertion randomly fails whereas
2467
replication seems to work fine. And there is no easy explanation why it
2468
fails (as we my_b_seek(rli->event_relay_log_pos) at the very end of
2469
init_relay_log_pos() called above). Maybe the assertion would be
2470
meaningful if we held rli->data_lock between the my_b_seek() and the
2473
#ifdef SHOULD_BE_CHECKED
2474
DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
2478
DBUG_ASSERT(rli->sql_thd == thd);
2480
DBUG_PRINT("master_info",("log_file_name: %s position: %s",
2481
rli->group_master_log_name,
2482
llstr(rli->group_master_log_pos,llbuff)));
2483
if (global_system_variables.log_warnings)
2484
sql_print_information("Slave SQL thread initialized, starting replication in \
2485
log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
2486
llstr(rli->group_master_log_pos,llbuff),rli->group_relay_log_name,
2487
llstr(rli->group_relay_log_pos,llbuff1));
2489
/* execute init_slave variable */
2490
if (sys_init_slave.value_length)
2492
execute_init_command(thd, &sys_init_slave, &LOCK_sys_init_slave);
2493
if (thd->is_slave_error)
2496
Slave SQL thread aborted. Can't execute init_slave query");
2502
First check until condition - probably there is nothing to execute. We
2503
do not want to wait for next event in this case.
2505
pthread_mutex_lock(&rli->data_lock);
2506
if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
2507
rli->is_until_satisfied(rli->group_master_log_pos))
2510
sql_print_information("Slave SQL thread stopped because it reached its"
2511
" UNTIL position %s", llstr(rli->until_pos(), buf));
2512
pthread_mutex_unlock(&rli->data_lock);
2515
pthread_mutex_unlock(&rli->data_lock);
2517
/* Read queries from the IO/THREAD until this thread is killed */
2519
while (!sql_slave_killed(thd,rli))
2521
thd_proc_info(thd, "Reading event from the relay log");
2522
DBUG_ASSERT(rli->sql_thd == thd);
2523
THD_CHECK_SENTRY(thd);
2524
if (exec_relay_log_event(thd,rli))
2526
DBUG_PRINT("info", ("exec_relay_log_event() failed"));
2527
// do not scare the user if SQL thread was simply killed or stopped
2528
if (!sql_slave_killed(thd,rli))
2531
retrieve as much info as possible from the thd and, error
2532
codes and warnings and print this to the error log as to
2533
allow the user to locate the error
2535
uint32 const last_errno= rli->last_error().number;
2537
if (thd->is_error())
2539
char const *const errmsg= thd->main_da.message();
2542
("thd->main_da.sql_errno()=%d; rli->last_error.number=%d",
2543
thd->main_da.sql_errno(), last_errno));
2544
if (last_errno == 0)
2546
rli->report(ERROR_LEVEL, thd->main_da.sql_errno(), errmsg);
2548
else if (last_errno != thd->main_da.sql_errno())
2550
sql_print_error("Slave (additional info): %s Error_code: %d",
2551
errmsg, thd->main_da.sql_errno());
2555
/* Print any warnings issued */
2556
List_iterator_fast<MYSQL_ERROR> it(thd->warn_list);
2559
Added controlled slave thread cancel for replication
2560
of user-defined variables.
2562
bool udf_error = false;
2565
if (err->code == ER_CANT_OPEN_LIBRARY)
2567
sql_print_warning("Slave: %s Error_code: %d",err->msg, err->code);
2570
sql_print_error("Error loading user-defined library, slave SQL "
2571
"thread aborted. Install the missing library, and restart the "
2572
"slave SQL thread with \"SLAVE START\". We stopped at log '%s' "
2573
"position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos,
2577
Error running query, slave SQL thread aborted. Fix the problem, and restart \
2578
the slave SQL thread with \"SLAVE START\". We stopped at log \
2579
'%s' position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, llbuff));
2585
/* Thread stopped. Print the current replication position to the log */
2586
sql_print_information("Slave SQL thread exiting, replication stopped in log "
2587
"'%s' at position %s",
2588
RPL_LOG_NAME, llstr(rli->group_master_log_pos,llbuff));
2593
Some events set some playgrounds, which won't be cleared because thread
2594
stops. Stopping of this thread may not be known to these events ("stop"
2595
request is detected only by the present function, not by events), so we
2596
must "proactively" clear playgrounds:
2598
rli->cleanup_context(thd, 1);
2599
VOID(pthread_mutex_lock(&LOCK_thread_count));
2601
Some extra safety, which should not been needed (normally, event deletion
2602
should already have done these assignments (each event which sets these
2603
variables is supposed to set them to 0 before terminating)).
2605
thd->query= thd->db= thd->catalog= 0;
2606
thd->query_length= thd->db_length= 0;
2607
VOID(pthread_mutex_unlock(&LOCK_thread_count));
2608
thd_proc_info(thd, "Waiting for slave mutex on exit");
2609
pthread_mutex_lock(&rli->run_lock);
2610
/* We need data_lock, at least to wake up any waiting master_pos_wait() */
2611
pthread_mutex_lock(&rli->data_lock);
2612
DBUG_ASSERT(rli->slave_running == 1); // tracking buffer overrun
2613
/* When master_pos_wait() wakes up it will check this and terminate */
2614
rli->slave_running= 0;
2615
/* Forget the relay log's format */
2616
delete rli->relay_log.description_event_for_exec;
2617
rli->relay_log.description_event_for_exec= 0;
2618
/* Wake up master_pos_wait() */
2619
pthread_mutex_unlock(&rli->data_lock);
2620
DBUG_PRINT("info",("Signaling possibly waiting master_pos_wait() functions"));
2621
pthread_cond_broadcast(&rli->data_cond);
2622
rli->ignore_log_space_limit= 0; /* don't need any lock */
2623
/* we die so won't remember charset - re-update them on next thread start */
2624
rli->cached_charset_invalidate();
2625
rli->save_temporary_tables = thd->temporary_tables;
2628
TODO: see if we can do this conditionally in next_event() instead
2629
to avoid unneeded position re-init
2631
thd->temporary_tables = 0; // remove tempation from destructor to close them
2632
DBUG_ASSERT(thd->net.buff != 0);
2633
net_end(&thd->net); // destructor will not free it, because we are weird
2634
DBUG_ASSERT(rli->sql_thd == thd);
2635
THD_CHECK_SENTRY(thd);
2637
pthread_mutex_lock(&LOCK_thread_count);
2638
THD_CHECK_SENTRY(thd);
2640
pthread_mutex_unlock(&LOCK_thread_count);
2642
Note: the order of the broadcast and unlock calls below (first broadcast, then unlock)
2643
is important. Otherwise a killer_thread can execute between the calls and
2644
delete the mi structure leading to a crash! (see BUG#25306 for details)
2646
pthread_cond_broadcast(&rli->stop_cond);
2647
pthread_mutex_unlock(&rli->run_lock); // tell the world we are done
2651
DBUG_RETURN(0); // Can't return anything here
2656
process_io_create_file()
2659
static int process_io_create_file(Master_info* mi, Create_file_log_event* cev)
2663
bool cev_not_written;
2664
THD *thd = mi->io_thd;
2665
NET *net = &mi->mysql->net;
2666
DBUG_ENTER("process_io_create_file");
2668
if (unlikely(!cev->is_valid()))
2671
if (!rpl_filter->db_ok(cev->db))
2673
skip_load_data_infile(net);
2676
DBUG_ASSERT(cev->inited_from_old);
2677
thd->file_id = cev->file_id = mi->file_id++;
2678
thd->server_id = cev->server_id;
2679
cev_not_written = 1;
2681
if (unlikely(net_request_file(net,cev->fname)))
2683
sql_print_error("Slave I/O: failed requesting download of '%s'",
2689
This dummy block is so we could instantiate Append_block_log_event
2690
once and then modify it slightly instead of doing it multiple times
2694
Append_block_log_event aev(thd,0,0,0,0);
2698
if (unlikely((num_bytes=my_net_read(net)) == packet_error))
2700
sql_print_error("Network read error downloading '%s' from master",
2704
if (unlikely(!num_bytes)) /* eof */
2706
/* 3.23 master wants it */
2707
net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0);
2709
If we wrote Create_file_log_event, then we need to write
2710
Execute_load_log_event. If we did not write Create_file_log_event,
2711
then this is an empty file and we can just do as if the LOAD DATA
2712
INFILE had not existed, i.e. write nothing.
2714
if (unlikely(cev_not_written))
2716
Execute_load_log_event xev(thd,0,0);
2717
xev.log_pos = cev->log_pos;
2718
if (unlikely(mi->rli.relay_log.append(&xev)))
2720
mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
2721
ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
2722
"error writing Exec_load event to relay log");
2725
mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total);
2728
if (unlikely(cev_not_written))
2730
cev->block = net->read_pos;
2731
cev->block_len = num_bytes;
2732
if (unlikely(mi->rli.relay_log.append(cev)))
2734
mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
2735
ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
2736
"error writing Create_file event to relay log");
2740
mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total);
2744
aev.block = net->read_pos;
2745
aev.block_len = num_bytes;
2746
aev.log_pos = cev->log_pos;
2747
if (unlikely(mi->rli.relay_log.append(&aev)))
2749
mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
2750
ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
2751
"error writing Append_block event to relay log");
2754
mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total) ;
2765
Start using a new binary log on the master
2769
mi master_info for the slave
2770
rev The rotate log event read from the binary log
2773
Updates the master info with the place in the next binary
2774
log where we should start reading.
2775
Rotate the relay log to avoid mixed-format relay logs.
2778
We assume we already locked mi->data_lock
2782
1 Log event is illegal
2786
static int process_io_rotate(Master_info *mi, Rotate_log_event *rev)
2788
DBUG_ENTER("process_io_rotate");
2789
safe_mutex_assert_owner(&mi->data_lock);
2791
if (unlikely(!rev->is_valid()))
2794
/* Safe copy as 'rev' has been "sanitized" in Rotate_log_event's ctor */
2795
memcpy(mi->master_log_name, rev->new_log_ident, rev->ident_len+1);
2796
mi->master_log_pos= rev->pos;
2797
DBUG_PRINT("info", ("master_log_pos: '%s' %lu",
2798
mi->master_log_name, (ulong) mi->master_log_pos));
2801
If we do not do this, we will be getting the first
2802
rotate event forever, so we need to not disconnect after one.
2804
if (disconnect_slave_event_count)
2805
mi->events_till_disconnect++;
2809
If description_event_for_queue is format <4, there is conversion in the
2810
relay log to the slave's format (4). And Rotate can mean upgrade or
2811
nothing. If upgrade, it's to 5.0 or newer, so we will get a Format_desc, so
2812
no need to reset description_event_for_queue now. And if it's nothing (same
2813
master version as before), no need (still using the slave's format).
2815
if (mi->rli.relay_log.description_event_for_queue->binlog_version >= 4)
2817
delete mi->rli.relay_log.description_event_for_queue;
2818
/* start from format 3 (MySQL 4.0) again */
2819
mi->rli.relay_log.description_event_for_queue= new
2820
Format_description_log_event(3);
2823
Rotate the relay log makes binlog format detection easier (at next slave
2824
start or mysqlbinlog)
2826
rotate_relay_log(mi); /* will take the right mutexes */
2831
Reads a 3.23 event and converts it to the slave's format. This code was
2832
copied from MySQL 4.0.
2834
static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
2837
const char *errmsg = 0;
2839
bool ignore_event= 0;
2841
Relay_log_info *rli= &mi->rli;
2842
DBUG_ENTER("queue_binlog_ver_1_event");
2845
If we get Load event, we need to pass a non-reusable buffer
2846
to read_log_event, so we do a trick
2848
if (buf[EVENT_TYPE_OFFSET] == LOAD_EVENT)
2850
if (unlikely(!(tmp_buf=(char*)my_malloc(event_len+1,MYF(MY_WME)))))
2852
mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
2853
ER(ER_SLAVE_FATAL_ERROR), "Memory allocation failed");
2856
memcpy(tmp_buf,buf,event_len);
2858
Create_file constructor wants a 0 as last char of buffer, this 0 will
2859
serve as the string-termination char for the file's name (which is at the
2861
We must increment event_len, otherwise the event constructor will not see
2862
this end 0, which leads to segfault.
2864
tmp_buf[event_len++]=0;
2865
int4store(tmp_buf+EVENT_LEN_OFFSET, event_len);
2866
buf = (const char*)tmp_buf;
2869
This will transform LOAD_EVENT into CREATE_FILE_EVENT, ask the master to
2870
send the loaded file, and write it to the relay log in the form of
2871
Append_block/Exec_load (the SQL thread needs the data, as that thread is not
2872
connected to the master).
2874
Log_event *ev = Log_event::read_log_event(buf,event_len, &errmsg,
2875
mi->rli.relay_log.description_event_for_queue);
2878
sql_print_error("Read invalid event from master: '%s',\
2879
master could be corrupt but a more likely cause of this is a bug",
2881
my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR));
2885
pthread_mutex_lock(&mi->data_lock);
2886
ev->log_pos= mi->master_log_pos; /* 3.23 events don't contain log_pos */
2887
switch (ev->get_type_code()) {
2893
if (unlikely(process_io_rotate(mi,(Rotate_log_event*)ev)))
2896
pthread_mutex_unlock(&mi->data_lock);
2901
case CREATE_FILE_EVENT:
2903
Yes it's possible to have CREATE_FILE_EVENT here, even if we're in
2904
queue_old_event() which is for 3.23 events which don't comprise
2905
CREATE_FILE_EVENT. This is because read_log_event() above has just
2906
transformed LOAD_EVENT into CREATE_FILE_EVENT.
2909
/* We come here when and only when tmp_buf != 0 */
2910
DBUG_ASSERT(tmp_buf != 0);
2912
ev->log_pos+= inc_pos;
2913
int error = process_io_create_file(mi,(Create_file_log_event*)ev);
2915
mi->master_log_pos += inc_pos;
2916
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
2917
pthread_mutex_unlock(&mi->data_lock);
2918
my_free((char*)tmp_buf, MYF(0));
2925
if (likely(!ignore_event))
2929
Don't do it for fake Rotate events (see comment in
2930
Log_event::Log_event(const char* buf...) in log_event.cc).
2932
ev->log_pos+= event_len; /* make log_pos be the pos of the end of the event */
2933
if (unlikely(rli->relay_log.append(ev)))
2936
pthread_mutex_unlock(&mi->data_lock);
2939
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
2942
mi->master_log_pos+= inc_pos;
2943
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
2944
pthread_mutex_unlock(&mi->data_lock);
2949
Reads a 4.0 event and converts it to the slave's format. This code was copied
2950
from queue_binlog_ver_1_event(), with some affordable simplifications.
2952
static int queue_binlog_ver_3_event(Master_info *mi, const char *buf,
2955
const char *errmsg = 0;
2958
Relay_log_info *rli= &mi->rli;
2959
DBUG_ENTER("queue_binlog_ver_3_event");
2961
/* read_log_event() will adjust log_pos to be end_log_pos */
2962
Log_event *ev = Log_event::read_log_event(buf,event_len, &errmsg,
2963
mi->rli.relay_log.description_event_for_queue);
2966
sql_print_error("Read invalid event from master: '%s',\
2967
master could be corrupt but a more likely cause of this is a bug",
2969
my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR));
2972
pthread_mutex_lock(&mi->data_lock);
2973
switch (ev->get_type_code()) {
2977
if (unlikely(process_io_rotate(mi,(Rotate_log_event*)ev)))
2980
pthread_mutex_unlock(&mi->data_lock);
2989
if (unlikely(rli->relay_log.append(ev)))
2992
pthread_mutex_unlock(&mi->data_lock);
2995
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
2997
mi->master_log_pos+= inc_pos;
2999
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
3000
pthread_mutex_unlock(&mi->data_lock);
3007
Writes a 3.23 or 4.0 event to the relay log, after converting it to the 5.0
3008
(exactly, slave's) format. To do the conversion, we create a 5.0 event from
3009
the 3.23/4.0 bytes, then write this event to the relay log.
3012
Test this code before release - it has to be tested on a separate
3013
setup with 3.23 master or 4.0 master
3016
static int queue_old_event(Master_info *mi, const char *buf,
3019
DBUG_ENTER("queue_old_event");
3021
switch (mi->rli.relay_log.description_event_for_queue->binlog_version)
3024
DBUG_RETURN(queue_binlog_ver_1_event(mi,buf,event_len));
3026
DBUG_RETURN(queue_binlog_ver_3_event(mi,buf,event_len));
3027
default: /* unsupported format; eg version 2 */
3028
DBUG_PRINT("info",("unsupported binlog format %d in queue_old_event()",
3029
mi->rli.relay_log.description_event_for_queue->binlog_version));
3037
If the event is 3.23/4.0, passes it to queue_old_event() which will convert
3038
it. Otherwise, writes a 5.0 (or newer) event to the relay log. Then there is
3039
no format conversion, it's pure read/write of bytes.
3040
So a 5.0.0 slave's relay log can contain events in the slave's format or in
3044
static int queue_event(Master_info* mi,const char* buf, ulong event_len)
3049
Relay_log_info *rli= &mi->rli;
3050
pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
3051
DBUG_ENTER("queue_event");
3054
if (mi->rli.relay_log.description_event_for_queue->binlog_version<4 &&
3055
buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT /* a way to escape */)
3056
DBUG_RETURN(queue_old_event(mi,buf,event_len));
3058
pthread_mutex_lock(&mi->data_lock);
3060
switch (buf[EVENT_TYPE_OFFSET]) {
3063
We needn't write this event to the relay log. Indeed, it just indicates a
3064
master server shutdown. The only thing this does is cleaning. But
3065
cleaning is already done on a per-master-thread basis (as the master
3066
server is shutting down cleanly, it has written all DROP TEMPORARY TABLE
3067
prepared statements' deletion are TODO only when we binlog prep stmts).
3069
We don't even increment mi->master_log_pos, because we may be just after
3070
a Rotate event. Btw, in a few milliseconds we are going to have a Start
3071
event from the next binlog (unless the master is presently running
3077
Rotate_log_event rev(buf,event_len,mi->rli.relay_log.description_event_for_queue);
3078
if (unlikely(process_io_rotate(mi,&rev)))
3080
error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
3084
Now the I/O thread has just changed its mi->master_log_name, so
3085
incrementing mi->master_log_pos is nonsense.
3090
case FORMAT_DESCRIPTION_EVENT:
3093
Create an event, and save it (when we rotate the relay log, we will have
3094
to write this event again).
3097
We are the only thread which reads/writes description_event_for_queue.
3098
The relay_log struct does not move (though some members of it can
3099
change), so we needn't any lock (no rli->data_lock, no log lock).
3101
Format_description_log_event* tmp;
3103
if (!(tmp= (Format_description_log_event*)
3104
Log_event::read_log_event(buf, event_len, &errmsg,
3105
mi->rli.relay_log.description_event_for_queue)))
3107
error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
3110
delete mi->rli.relay_log.description_event_for_queue;
3111
mi->rli.relay_log.description_event_for_queue= tmp;
3113
Though this does some conversion to the slave's format, this will
3114
preserve the master's binlog format version, and number of event types.
3117
If the event was not requested by the slave (the slave did not ask for
3118
it), i.e. has end_log_pos=0, we do not increment mi->master_log_pos
3120
inc_pos= uint4korr(buf+LOG_POS_OFFSET) ? event_len : 0;
3121
DBUG_PRINT("info",("binlog format is now %d",
3122
mi->rli.relay_log.description_event_for_queue->binlog_version));
3127
case HEARTBEAT_LOG_EVENT:
3130
HB (heartbeat) cannot come before RL (Relay)
3133
Heartbeat_log_event hb(buf, event_len, mi->rli.relay_log.description_event_for_queue);
3136
error= ER_SLAVE_HEARTBEAT_FAILURE;
3137
error_msg.append(STRING_WITH_LEN("inconsistent heartbeat event content;"));
3138
error_msg.append(STRING_WITH_LEN("the event's data: log_file_name "));
3139
error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident()));
3140
error_msg.append(STRING_WITH_LEN(" log_pos "));
3141
llstr(hb.log_pos, llbuf);
3142
error_msg.append(llbuf, strlen(llbuf));
3145
mi->received_heartbeats++;
3147
compare local and event's versions of log_file, log_pos.
3149
Heartbeat is sent only after an event corresponding to the corrdinates
3150
the heartbeat carries.
3151
Slave can not have a difference in coordinates except in the only
3152
special case when mi->master_log_name, master_log_pos have never
3153
been updated by Rotate event i.e when slave does not have any history
3154
with the master (and thereafter mi->master_log_pos is NULL).
3156
TODO: handling `when' for SHOW SLAVE STATUS' snds behind
3158
if ((memcmp(mi->master_log_name, hb.get_log_ident(), hb.get_ident_len())
3159
&& mi->master_log_name != NULL)
3160
|| mi->master_log_pos != hb.log_pos)
3162
/* missed events of heartbeat from the past */
3163
error= ER_SLAVE_HEARTBEAT_FAILURE;
3164
error_msg.append(STRING_WITH_LEN("heartbeat is not compatible with local info;"));
3165
error_msg.append(STRING_WITH_LEN("the event's data: log_file_name "));
3166
error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident()));
3167
error_msg.append(STRING_WITH_LEN(" log_pos "));
3168
llstr(hb.log_pos, llbuf);
3169
error_msg.append(llbuf, strlen(llbuf));
3172
goto skip_relay_logging;
3182
If this event is originating from this server, don't queue it.
3183
We don't check this for 3.23 events because it's simpler like this; 3.23
3184
will be filtered anyway by the SQL slave thread which also tests the
3185
server id (we must also keep this test in the SQL thread, in case somebody
3186
upgrades a 4.0 slave which has a not-filtered relay log).
3188
ANY event coming from ourselves can be ignored: it is obvious for queries;
3189
for STOP_EVENT/ROTATE_EVENT/START_EVENT: these cannot come from ourselves
3190
(--log-slave-updates would not log that) unless this slave is also its
3191
direct master (an unsupported, useless setup!).
3194
pthread_mutex_lock(log_lock);
3196
if ((uint4korr(buf + SERVER_ID_OFFSET) == ::server_id) &&
3197
!mi->rli.replicate_same_server_id)
3200
Do not write it to the relay log.
3201
a) We still want to increment mi->master_log_pos, so that we won't
3202
re-read this event from the master if the slave IO thread is now
3203
stopped/restarted (more efficient if the events we are ignoring are big
3205
b) We want to record that we are skipping events, for the information of
3206
the slave SQL thread, otherwise that thread may let
3207
rli->group_relay_log_pos stay too small if the last binlog's event is
3209
But events which were generated by this slave and which do not exist in
3210
the master's binlog (i.e. Format_desc, Rotate & Stop) should not increment
3213
if (buf[EVENT_TYPE_OFFSET]!=FORMAT_DESCRIPTION_EVENT &&
3214
buf[EVENT_TYPE_OFFSET]!=ROTATE_EVENT &&
3215
buf[EVENT_TYPE_OFFSET]!=STOP_EVENT)
3217
mi->master_log_pos+= inc_pos;
3218
memcpy(rli->ign_master_log_name_end, mi->master_log_name, FN_REFLEN);
3219
DBUG_ASSERT(rli->ign_master_log_name_end[0]);
3220
rli->ign_master_log_pos_end= mi->master_log_pos;
3222
rli->relay_log.signal_update(); // the slave SQL thread needs to re-check
3223
DBUG_PRINT("info", ("master_log_pos: %lu, event originating from the same server, ignored",
3224
(ulong) mi->master_log_pos));
3228
/* write the event to the relay log */
3229
if (likely(!(rli->relay_log.appendv(buf,event_len,0))))
3231
mi->master_log_pos+= inc_pos;
3232
DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
3233
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
3237
error= ER_SLAVE_RELAY_LOG_WRITE_FAILURE;
3239
rli->ign_master_log_name_end[0]= 0; // last event is not ignored
3241
pthread_mutex_unlock(log_lock);
3246
pthread_mutex_unlock(&mi->data_lock);
3247
DBUG_PRINT("info", ("error: %d", error));
3249
mi->report(ERROR_LEVEL, error, ER(error),
3250
(error == ER_SLAVE_RELAY_LOG_WRITE_FAILURE)?
3251
"could not queue event from master" :
3257
void end_relay_log_info(Relay_log_info* rli)
3259
DBUG_ENTER("end_relay_log_info");
3263
if (rli->info_fd >= 0)
3265
end_io_cache(&rli->info_file);
3266
(void) my_close(rli->info_fd, MYF(MY_WME));
3269
if (rli->cur_log_fd >= 0)
3271
end_io_cache(&rli->cache_buf);
3272
(void)my_close(rli->cur_log_fd, MYF(MY_WME));
3273
rli->cur_log_fd = -1;
3276
rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
3277
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
3279
Delete the slave's temporary tables from memory.
3280
In the future there will be other actions than this, to ensure persistance
3281
of slave's temp tables after shutdown.
3283
rli->close_temporary_tables();
3288
Try to connect until successful or slave killed
3292
thd Thread handler for slave
3293
mysql MySQL connection handle
3294
mi Replication handle
3301
static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi)
3303
DBUG_ENTER("safe_connect");
3305
DBUG_RETURN(connect_to_master(thd, mysql, mi, 0, 0));
3314
Try to connect until successful or slave killed or we have retried
3315
master_retry_count times
3318
static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi,
3319
bool reconnect, bool suppress_warnings)
3321
int slave_was_killed;
3322
int last_errno= -2; // impossible error
3325
DBUG_ENTER("connect_to_master");
3328
mi->events_till_disconnect = disconnect_slave_event_count;
3330
ulong client_flag= CLIENT_REMEMBER_OPTIONS;
3331
if (opt_slave_compressed_protocol)
3332
client_flag=CLIENT_COMPRESS; /* We will use compression */
3334
mysql_options(mysql, MYSQL_OPT_CONNECT_TIMEOUT, (char *) &slave_net_timeout);
3335
mysql_options(mysql, MYSQL_OPT_READ_TIMEOUT, (char *) &slave_net_timeout);
3337
mysql_options(mysql, MYSQL_SET_CHARSET_NAME, default_charset_info->csname);
3338
/* This one is not strictly needed but we have it here for completeness */
3339
mysql_options(mysql, MYSQL_SET_CHARSET_DIR, (char *) charsets_dir);
3341
while (!(slave_was_killed = io_slave_killed(thd,mi)) &&
3342
(reconnect ? mysql_reconnect(mysql) != 0 :
3343
mysql_real_connect(mysql, mi->host, mi->user, mi->password, 0,
3344
mi->port, 0, client_flag) == 0))
3346
/* Don't repeat last error */
3347
if ((int)mysql_errno(mysql) != last_errno)
3349
last_errno=mysql_errno(mysql);
3350
suppress_warnings= 0;
3351
mi->report(ERROR_LEVEL, last_errno,
3352
"error %s to master '%s@%s:%d'"
3353
" - retry-time: %d retries: %lu",
3354
(reconnect ? "reconnecting" : "connecting"),
3355
mi->user, mi->host, mi->port,
3356
mi->connect_retry, master_retry_count);
3359
By default we try forever. The reason is that failure will trigger
3360
master election, so if the user did not set master_retry_count we
3361
do not want to have election triggered on the first failure to
3364
if (++err_count == master_retry_count)
3368
change_rpl_status(RPL_ACTIVE_SLAVE,RPL_LOST_SOLDIER);
3371
safe_sleep(thd,mi->connect_retry,(CHECK_KILLED_FUNC)io_slave_killed,
3375
if (!slave_was_killed)
3379
if (!suppress_warnings && global_system_variables.log_warnings)
3380
sql_print_information("Slave: connected to master '%s@%s:%d',\
3381
replication resumed in log '%s' at position %s", mi->user,
3384
llstr(mi->master_log_pos,llbuff));
3388
change_rpl_status(RPL_IDLE_SLAVE,RPL_ACTIVE_SLAVE);
3389
general_log_print(thd, COM_CONNECT_OUT, "%s@%s:%d",
3390
mi->user, mi->host, mi->port);
3392
#ifdef SIGNAL_WITH_VIO_CLOSE
3393
thd->set_active_vio(mysql->net.vio);
3396
mysql->reconnect= 1;
3397
DBUG_PRINT("exit",("slave_was_killed: %d", slave_was_killed));
3398
DBUG_RETURN(slave_was_killed);
3406
Try to connect until successful or slave killed or we have retried
3407
master_retry_count times
3410
static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi,
3411
bool suppress_warnings)
3413
DBUG_ENTER("safe_reconnect");
3414
DBUG_RETURN(connect_to_master(thd, mysql, mi, 1, suppress_warnings));
3419
Store the file and position where the execute-slave thread are in the
3423
flush_relay_log_info()
3424
rli Relay log information
3427
- As this is only called by the slave thread, we don't need to
3428
have a lock on this.
3429
- If there is an active transaction, then we don't update the position
3430
in the relay log. This is to ensure that we re-execute statements
3431
if we die in the middle of an transaction that was rolled back.
3432
- As a transaction never spans binary logs, we don't have to handle the
3433
case where we do a relay-log-rotation in the middle of the transaction.
3434
If this would not be the case, we would have to ensure that we
3435
don't delete the relay log file where the transaction started when
3436
we switch to a new relay log file.
3439
- Change the log file information to a binary format to avoid calling
3447
bool flush_relay_log_info(Relay_log_info* rli)
3450
DBUG_ENTER("flush_relay_log_info");
3452
if (unlikely(rli->no_storage))
3455
IO_CACHE *file = &rli->info_file;
3456
char buff[FN_REFLEN*2+22*2+4], *pos;
3458
my_b_seek(file, 0L);
3459
pos=strmov(buff, rli->group_relay_log_name);
3461
pos=longlong2str(rli->group_relay_log_pos, pos, 10);
3463
pos=strmov(pos, rli->group_master_log_name);
3465
pos=longlong2str(rli->group_master_log_pos, pos, 10);
3467
if (my_b_write(file, (uchar*) buff, (size_t) (pos-buff)+1))
3469
if (flush_io_cache(file))
3472
/* Flushing the relay log is done by the slave I/O thread */
3478
Called when we notice that the current "hot" log got rotated under our feet.
3481
static IO_CACHE *reopen_relay_log(Relay_log_info *rli, const char **errmsg)
3483
DBUG_ENTER("reopen_relay_log");
3484
DBUG_ASSERT(rli->cur_log != &rli->cache_buf);
3485
DBUG_ASSERT(rli->cur_log_fd == -1);
3487
IO_CACHE *cur_log = rli->cur_log=&rli->cache_buf;
3488
if ((rli->cur_log_fd=open_binlog(cur_log,rli->event_relay_log_name,
3492
We want to start exactly where we was before:
3493
relay_log_pos Current log pos
3494
pending Number of bytes already processed from the event
3496
rli->event_relay_log_pos= max(rli->event_relay_log_pos, BIN_LOG_HEADER_SIZE);
3497
my_b_seek(cur_log,rli->event_relay_log_pos);
3498
DBUG_RETURN(cur_log);
3502
static Log_event* next_event(Relay_log_info* rli)
3505
IO_CACHE* cur_log = rli->cur_log;
3506
pthread_mutex_t *log_lock = rli->relay_log.get_log_lock();
3507
const char* errmsg=0;
3508
THD* thd = rli->sql_thd;
3509
DBUG_ENTER("next_event");
3511
DBUG_ASSERT(thd != 0);
3514
if (abort_slave_event_count && !rli->events_till_abort--)
3519
For most operations we need to protect rli members with data_lock,
3520
so we assume calling function acquired this mutex for us and we will
3521
hold it for the most of the loop below However, we will release it
3522
whenever it is worth the hassle, and in the cases when we go into a
3523
pthread_cond_wait() with the non-data_lock mutex
3525
safe_mutex_assert_owner(&rli->data_lock);
3527
while (!sql_slave_killed(thd,rli))
3530
We can have two kinds of log reading:
3532
rli->cur_log points at the IO_CACHE of relay_log, which
3533
is actively being updated by the I/O thread. We need to be careful
3534
in this case and make sure that we are not looking at a stale log that
3535
has already been rotated. If it has been, we reopen the log.
3537
The other case is much simpler:
3538
We just have a read only log that nobody else will be updating.
3541
if ((hot_log = (cur_log != &rli->cache_buf)))
3543
DBUG_ASSERT(rli->cur_log_fd == -1); // foreign descriptor
3544
pthread_mutex_lock(log_lock);
3547
Reading xxx_file_id is safe because the log will only
3548
be rotated when we hold relay_log.LOCK_log
3550
if (rli->relay_log.get_open_count() != rli->cur_log_old_open_count)
3552
// The master has switched to a new log file; Reopen the old log file
3553
cur_log=reopen_relay_log(rli, &errmsg);
3554
pthread_mutex_unlock(log_lock);
3555
if (!cur_log) // No more log files
3557
hot_log=0; // Using old binary log
3561
As there is no guarantee that the relay is open (for example, an I/O
3562
error during a write by the slave I/O thread may have closed it), we
3565
if (!my_b_inited(cur_log))
3569
/* This is an assertion which sometimes fails, let's try to track it */
3570
char llbuf1[22], llbuf2[22];
3571
DBUG_PRINT("info", ("my_b_tell(cur_log)=%s rli->event_relay_log_pos=%s",
3572
llstr(my_b_tell(cur_log),llbuf1),
3573
llstr(rli->event_relay_log_pos,llbuf2)));
3574
DBUG_ASSERT(my_b_tell(cur_log) >= BIN_LOG_HEADER_SIZE);
3575
DBUG_ASSERT(my_b_tell(cur_log) == rli->event_relay_log_pos);
3579
Relay log is always in new format - if the master is 3.23, the
3580
I/O thread will convert the format for us.
3581
A problem: the description event may be in a previous relay log. So if
3582
the slave has been shutdown meanwhile, we would have to look in old relay
3583
logs, which may even have been deleted. So we need to write this
3584
description event at the beginning of the relay log.
3585
When the relay log is created when the I/O thread starts, easy: the
3586
master will send the description event and we will queue it.
3587
But if the relay log is created by new_file(): then the solution is:
3588
MYSQL_BIN_LOG::open() will write the buffered description event.
3590
if ((ev=Log_event::read_log_event(cur_log,0,
3591
rli->relay_log.description_event_for_exec)))
3594
DBUG_ASSERT(thd==rli->sql_thd);
3596
read it while we have a lock, to avoid a mutex lock in
3597
inc_event_relay_log_pos()
3599
rli->future_event_relay_log_pos= my_b_tell(cur_log);
3601
pthread_mutex_unlock(log_lock);
3604
DBUG_ASSERT(thd==rli->sql_thd);
3605
if (opt_reckless_slave) // For mysql-test
3607
if (cur_log->error < 0)
3609
errmsg = "slave SQL thread aborted because of I/O error";
3611
pthread_mutex_unlock(log_lock);
3614
if (!cur_log->error) /* EOF */
3617
On a hot log, EOF means that there are no more updates to
3618
process and we must block until I/O thread adds some and
3619
signals us to continue
3624
We say in Seconds_Behind_Master that we have "caught up". Note that
3625
for example if network link is broken but I/O slave thread hasn't
3626
noticed it (slave_net_timeout not elapsed), then we'll say "caught
3627
up" whereas we're not really caught up. Fixing that would require
3628
internally cutting timeout in smaller pieces in network read, no
3629
thanks. Another example: SQL has caught up on I/O, now I/O has read
3630
a new event and is queuing it; the false "0" will exist until SQL
3631
finishes executing the new event; it will be look abnormal only if
3632
the events have old timestamps (then you get "many", 0, "many").
3634
Transient phases like this can be fixed with implemeting
3635
Heartbeat event which provides the slave the status of the
3636
master at time the master does not have any new update to send.
3637
Seconds_Behind_Master would be zero only when master has no
3638
more updates in binlog for slave. The heartbeat can be sent
3639
in a (small) fraction of slave_net_timeout. Until it's done
3640
rli->last_master_timestamp is temporarely (for time of
3641
waiting for the following event) reset whenever EOF is
3644
time_t save_timestamp= rli->last_master_timestamp;
3645
rli->last_master_timestamp= 0;
3647
DBUG_ASSERT(rli->relay_log.get_open_count() ==
3648
rli->cur_log_old_open_count);
3650
if (rli->ign_master_log_name_end[0])
3652
/* We generate and return a Rotate, to make our positions advance */
3653
DBUG_PRINT("info",("seeing an ignored end segment"));
3654
ev= new Rotate_log_event(rli->ign_master_log_name_end,
3655
0, rli->ign_master_log_pos_end,
3656
Rotate_log_event::DUP_NAME);
3657
rli->ign_master_log_name_end[0]= 0;
3658
pthread_mutex_unlock(log_lock);
3661
errmsg= "Slave SQL thread failed to create a Rotate event "
3662
"(out of memory?), SHOW SLAVE STATUS may be inaccurate";
3665
ev->server_id= 0; // don't be ignored by slave SQL thread
3670
We can, and should release data_lock while we are waiting for
3671
update. If we do not, show slave status will block
3673
pthread_mutex_unlock(&rli->data_lock);
3677
- the I/O thread has reached log_space_limit
3678
- the SQL thread has read all relay logs, but cannot purge for some
3680
* it has already purged all logs except the current one
3681
* there are other logs than the current one but they're involved in
3682
a transaction that finishes in the current one (or is not finished)
3684
Wake up the possibly waiting I/O thread, and set a boolean asking
3685
the I/O thread to temporarily ignore the log_space_limit
3686
constraint, because we do not want the I/O thread to block because of
3687
space (it's ok if it blocks for any other reason (e.g. because the
3688
master does not send anything). Then the I/O thread stops waiting
3689
and reads more events.
3690
The SQL thread decides when the I/O thread should take log_space_limit
3691
into account again : ignore_log_space_limit is reset to 0
3692
in purge_first_log (when the SQL thread purges the just-read relay
3693
log), and also when the SQL thread starts. We should also reset
3694
ignore_log_space_limit to 0 when the user does RESET SLAVE, but in
3695
fact, no need as RESET SLAVE requires that the slave
3696
be stopped, and the SQL thread sets ignore_log_space_limit to 0 when
3699
pthread_mutex_lock(&rli->log_space_lock);
3700
// prevent the I/O thread from blocking next times
3701
rli->ignore_log_space_limit= 1;
3703
If the I/O thread is blocked, unblock it. Ok to broadcast
3704
after unlock, because the mutex is only destroyed in
3705
~Relay_log_info(), i.e. when rli is destroyed, and rli will
3706
not be destroyed before we exit the present function.
3708
pthread_mutex_unlock(&rli->log_space_lock);
3709
pthread_cond_broadcast(&rli->log_space_cond);
3710
// Note that wait_for_update_relay_log unlocks lock_log !
3711
rli->relay_log.wait_for_update_relay_log(rli->sql_thd);
3712
// re-acquire data lock since we released it earlier
3713
pthread_mutex_lock(&rli->data_lock);
3714
rli->last_master_timestamp= save_timestamp;
3718
If the log was not hot, we need to move to the next log in
3719
sequence. The next log could be hot or cold, we deal with both
3720
cases separately after doing some common initialization
3722
end_io_cache(cur_log);
3723
DBUG_ASSERT(rli->cur_log_fd >= 0);
3724
my_close(rli->cur_log_fd, MYF(MY_WME));
3725
rli->cur_log_fd = -1;
3727
if (relay_log_purge)
3730
purge_first_log will properly set up relay log coordinates in rli.
3731
If the group's coordinates are equal to the event's coordinates
3732
(i.e. the relay log was not rotated in the middle of a group),
3733
we can purge this relay log too.
3734
We do ulonglong and string comparisons, this may be slow but
3735
- purging the last relay log is nice (it can save 1GB of disk), so we
3736
like to detect the case where we can do it, and given this,
3737
- I see no better detection method
3738
- purge_first_log is not called that often
3740
if (rli->relay_log.purge_first_log
3742
rli->group_relay_log_pos == rli->event_relay_log_pos
3743
&& !strcmp(rli->group_relay_log_name,rli->event_relay_log_name)))
3745
errmsg = "Error purging processed logs";
3752
If hot_log is set, then we already have a lock on
3753
LOCK_log. If not, we have to get the lock.
3755
According to Sasha, the only time this code will ever be executed
3756
is if we are recovering from a bug.
3758
if (rli->relay_log.find_next_log(&rli->linfo, !hot_log))
3760
errmsg = "error switching to the next log";
3763
rli->event_relay_log_pos = BIN_LOG_HEADER_SIZE;
3764
strmake(rli->event_relay_log_name,rli->linfo.log_file_name,
3765
sizeof(rli->event_relay_log_name)-1);
3766
flush_relay_log_info(rli);
3770
Now we want to open this next log. To know if it's a hot log (the one
3771
being written by the I/O thread now) or a cold log, we can use
3772
is_active(); if it is hot, we use the I/O cache; if it's cold we open
3773
the file normally. But if is_active() reports that the log is hot, this
3774
may change between the test and the consequence of the test. So we may
3775
open the I/O cache whereas the log is now cold, which is nonsense.
3776
To guard against this, we need to have LOCK_log.
3779
DBUG_PRINT("info",("hot_log: %d",hot_log));
3780
if (!hot_log) /* if hot_log, we already have this mutex */
3781
pthread_mutex_lock(log_lock);
3782
if (rli->relay_log.is_active(rli->linfo.log_file_name))
3785
if (global_system_variables.log_warnings)
3786
sql_print_information("next log '%s' is currently active",
3787
rli->linfo.log_file_name);
3789
rli->cur_log= cur_log= rli->relay_log.get_log_file();
3790
rli->cur_log_old_open_count= rli->relay_log.get_open_count();
3791
DBUG_ASSERT(rli->cur_log_fd == -1);
3794
Read pointer has to be at the start since we are the only
3796
We must keep the LOCK_log to read the 4 first bytes, as this is a hot
3797
log (same as when we call read_log_event() above: for a hot log we
3800
if (check_binlog_magic(cur_log,&errmsg))
3802
if (!hot_log) pthread_mutex_unlock(log_lock);
3805
if (!hot_log) pthread_mutex_unlock(log_lock);
3808
if (!hot_log) pthread_mutex_unlock(log_lock);
3810
if we get here, the log was not hot, so we will have to open it
3811
ourselves. We are sure that the log is still not hot now (a log can get
3812
from hot to cold, but not from cold to hot). No need for LOCK_log.
3815
if (global_system_variables.log_warnings)
3816
sql_print_information("next log '%s' is not active",
3817
rli->linfo.log_file_name);
3819
// open_binlog() will check the magic header
3820
if ((rli->cur_log_fd=open_binlog(cur_log,rli->linfo.log_file_name,
3827
Read failed with a non-EOF error.
3828
TODO: come up with something better to handle this error
3831
pthread_mutex_unlock(log_lock);
3832
sql_print_error("Slave SQL thread: I/O error reading \
3833
event(errno: %d cur_log->error: %d)",
3834
my_errno,cur_log->error);
3835
// set read position to the beginning of the event
3836
my_b_seek(cur_log,rli->event_relay_log_pos);
3837
/* otherwise, we have had a partial read */
3838
errmsg = "Aborting slave SQL thread because of partial event read";
3839
break; // To end of function
3842
if (!errmsg && global_system_variables.log_warnings)
3844
sql_print_information("Error reading relay log event: %s",
3845
"slave SQL thread was killed");
3851
sql_print_error("Error reading relay log event: %s", errmsg);
3856
Rotate a relay log (this is used only by FLUSH LOGS; the automatic rotation
3857
because of size is simpler because when we do it we already have all relevant
3858
locks; here we don't, so this function is mainly taking locks).
3859
Returns nothing as we cannot catch any error (MYSQL_BIN_LOG::new_file()
3863
void rotate_relay_log(Master_info* mi)
3865
DBUG_ENTER("rotate_relay_log");
3866
Relay_log_info* rli= &mi->rli;
3868
/* We don't lock rli->run_lock. This would lead to deadlocks. */
3869
pthread_mutex_lock(&mi->run_lock);
3872
We need to test inited because otherwise, new_file() will attempt to lock
3873
LOCK_log, which may not be inited (if we're not a slave).
3877
DBUG_PRINT("info", ("rli->inited == 0"));
3881
/* If the relay log is closed, new_file() will do nothing. */
3882
rli->relay_log.new_file();
3885
We harvest now, because otherwise BIN_LOG_HEADER_SIZE will not immediately
3886
be counted, so imagine a succession of FLUSH LOGS and assume the slave
3887
threads are started:
3888
relay_log_space decreases by the size of the deleted relay log, but does
3889
not increase, so flush-after-flush we may become negative, which is wrong.
3890
Even if this will be corrected as soon as a query is replicated on the
3891
slave (because the I/O thread will then call harvest_bytes_written() which
3892
will harvest all these BIN_LOG_HEADER_SIZE we forgot), it may give strange
3893
output in SHOW SLAVE STATUS meanwhile. So we harvest now.
3894
If the log is closed, then this will just harvest the last writes, probably
3895
0 as they probably have been harvested.
3897
rli->relay_log.harvest_bytes_written(&rli->log_space_total);
3899
pthread_mutex_unlock(&mi->run_lock);
3905
Detects, based on master's version (as found in the relay log), if master
3907
@param rli Relay_log_info which tells the master's version
3908
@param bug_id Number of the bug as found in bugs.mysql.com
3909
@param report bool report error message, default TRUE
3910
@return TRUE if master has the bug, FALSE if it does not.
3912
bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id, bool report)
3914
struct st_version_range_for_one_bug {
3916
const uchar introduced_in[3]; // first version with bug
3917
const uchar fixed_in[3]; // first version with fix
3919
static struct st_version_range_for_one_bug versions_for_all_bugs[]=
3921
{24432, { 5, 0, 24 }, { 5, 0, 38 } },
3922
{24432, { 5, 1, 12 }, { 5, 1, 17 } },
3923
{33029, { 5, 0, 0 }, { 5, 0, 58 } },
3924
{33029, { 5, 1, 0 }, { 5, 1, 12 } },
3926
const uchar *master_ver=
3927
rli->relay_log.description_event_for_exec->server_version_split;
3929
DBUG_ASSERT(sizeof(rli->relay_log.description_event_for_exec->server_version_split) == 3);
3932
i < sizeof(versions_for_all_bugs)/sizeof(*versions_for_all_bugs);i++)
3934
const uchar *introduced_in= versions_for_all_bugs[i].introduced_in,
3935
*fixed_in= versions_for_all_bugs[i].fixed_in;
3936
if ((versions_for_all_bugs[i].bug_id == bug_id) &&
3937
(memcmp(introduced_in, master_ver, 3) <= 0) &&
3938
(memcmp(fixed_in, master_ver, 3) > 0))
3943
// a short message for SHOW SLAVE STATUS (message length constraints)
3944
my_printf_error(ER_UNKNOWN_ERROR, "master may suffer from"
3945
" http://bugs.mysql.com/bug.php?id=%u"
3946
" so slave stops; check error log on slave"
3947
" for more info", MYF(0), bug_id);
3948
// a verbose message for the error log
3949
rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR,
3950
"According to the master's version ('%s'),"
3951
" it is probable that master suffers from this bug:"
3952
" http://bugs.mysql.com/bug.php?id=%u"
3953
" and thus replicating the current binary log event"
3954
" may make the slave's data become different from the"
3956
" To take no risk, slave refuses to replicate"
3957
" this event and stops."
3958
" We recommend that all updates be stopped on the"
3959
" master and slave, that the data of both be"
3960
" manually synchronized,"
3961
" that master's binary logs be deleted,"
3962
" that master be upgraded to a version at least"
3963
" equal to '%d.%d.%d'. Then replication can be"
3965
rli->relay_log.description_event_for_exec->server_version,
3967
fixed_in[0], fixed_in[1], fixed_in[2]);
3975
BUG#33029, For all 5.0 up to 5.0.58 exclusive, and 5.1 up to 5.1.12
3976
exclusive, if one statement in a SP generated AUTO_INCREMENT value
3977
by the top statement, all statements after it would be considered
3978
generated AUTO_INCREMENT value by the top statement, and a
3979
erroneous INSERT_ID value might be associated with these statement,
3980
which could cause duplicate entry error and stop the slave.
3982
Detect buggy master to work around.
3984
bool rpl_master_erroneous_autoinc(THD *thd)
3986
if (active_mi && active_mi->rli.sql_thd == thd)
3988
Relay_log_info *rli= &active_mi->rli;
3989
return rpl_master_has_bug(rli, 33029, FALSE);
3994
#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
3995
template class I_List_iterator<i_string>;
3996
template class I_List_iterator<i_string_pair>;
4000
@} (end of group Replication)
4003
#endif /* HAVE_REPLICATION */
added
removed
sql/slave.cc
