← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to sql/slave.cc

  • Committer: Brian Aker
  • Date: 2008-07-06 15:03:34 UTC
  • Revision ID: brian@tangent.org-20080706150334-xv3xa202trvs0712
USE_RAID cleanup, along with ftbench tools.

Show diffs side-by-side

added added

removed removed

Lines of Context:
sql/slave.cc
1
 
/* Copyright (C) 2000-2003 DRIZZLE AB
 
1
/* Copyright (C) 2000-2003 MySQL AB
2
2
 
3
3
   This program is free software; you can redistribute it and/or modify
4
4
   it under the terms of the GNU General Public License as published by
23
23
  @brief Code to run the io thread and the sql thread on the
24
24
  replication slave.
25
25
*/
26
 
#include <drizzled/server_includes.h>
27
 
 
28
 
#include <storage/myisam/myisam.h>
 
26
 
 
27
#include "mysql_priv.h"
 
28
 
 
29
#include <mysql.h>
 
30
#include <myisam.h>
 
31
#include "slave.h"
29
32
#include "rpl_mi.h"
30
33
#include "rpl_rli.h"
31
34
#include "sql_repl.h"
32
35
#include "rpl_filter.h"
33
 
#include <mysys/thr_alarm.h>
34
 
#include <libdrizzle/sql_common.h>
35
 
#include <libdrizzle/errmsg.h>
36
 
#include <mysys/mysys_err.h>
37
 
#include <drizzled/drizzled_error_messages.h>
38
 
 
39
 
#if TIME_WITH_SYS_TIME
40
 
# include <sys/time.h>
41
 
# include <time.h>
42
 
#else
43
 
# if HAVE_SYS_TIME_H
44
 
#  include <sys/time.h>
45
 
# else
46
 
#  include <time.h>
47
 
# endif
48
 
#endif
49
 
 
 
36
#include "repl_failsafe.h"
 
37
#include <thr_alarm.h>
 
38
#include <my_dir.h>
 
39
#include <sql_common.h>
 
40
#include <errmsg.h>
 
41
#include <mysys_err.h>
 
42
 
 
43
#ifdef HAVE_REPLICATION
50
44
 
51
45
#include "rpl_tblmap.h"
52
46
 
60
54
 
61
55
char* slave_load_tmpdir = 0;
62
56
Master_info *active_mi= 0;
63
 
bool replicate_same_server_id;
64
 
uint64_t relay_log_space_limit = 0;
 
57
my_bool replicate_same_server_id;
 
58
ulonglong relay_log_space_limit = 0;
65
59
 
66
60
/*
67
61
  When slave thread exits, we need to remember the temporary tables so we
70
64
  TODO: move the vars below under Master_info
71
65
*/
72
66
 
73
 
int32_t disconnect_slave_event_count = 0, abort_slave_event_count = 0;
74
 
int32_t events_till_abort = -1;
 
67
int disconnect_slave_event_count = 0, abort_slave_event_count = 0;
 
68
int events_till_abort = -1;
75
69
 
76
70
enum enum_slave_reconnect_actions
77
71
{
95
89
static const char *reconnect_messages[SLAVE_RECON_ACT_MAX][SLAVE_RECON_MSG_MAX]=
96
90
{
97
91
  {
98
 
    N_("Waiting to reconnect after a failed registration on master"),
99
 
    N_("Slave I/O thread killed while waitnig to reconnect after a "
100
 
                 "failed registration on master"),
101
 
    N_("Reconnecting after a failed registration on master"),
102
 
    N_("failed registering on master, reconnecting to try again, "
103
 
                 "log '%s' at postion %s"),
 
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",
104
98
    "COM_REGISTER_SLAVE",
105
 
    N_("Slave I/O thread killed during or after reconnect")
 
99
    "Slave I/O thread killed during or after reconnect"
106
100
  },
107
101
  {
108
 
    N_("Waiting to reconnect after a failed binlog dump request"),
109
 
    N_("Slave I/O thread killed while retrying master dump"),
110
 
    N_("Reconnecting after a failed binlog dump request"),
111
 
    N_("failed dump request, reconnecting to try again, "
112
 
                 "log '%s' at postion %s"),
 
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",
113
106
    "COM_BINLOG_DUMP",
114
 
    N_("Slave I/O thread killed during or after reconnect")
 
107
    "Slave I/O thread killed during or after reconnect"
115
108
  },
116
109
  {
117
 
    N_("Waiting to reconnect after a failed master event read"),
118
 
    N_("Slave I/O thread killed while waiting to reconnect "
119
 
                 "after a failed read"),
120
 
    N_("Reconnecting after a failed master event read"),
121
 
    N_("Slave I/O thread: Failed reading log event, "
122
 
                 "reconnecting to retry, log '%s' at postion %s"),
 
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",
123
115
    "",
124
 
    N_("Slave I/O thread killed during or after a "
125
 
                 "reconnect done to recover from failed read")
 
116
    "Slave I/O thread killed during or after a reconnect done to recover from \
 
117
failed read"
126
118
  }
127
119
};
128
120
 
129
121
 
130
122
typedef enum { SLAVE_THD_IO, SLAVE_THD_SQL} SLAVE_THD_TYPE;
131
123
 
132
 
static int32_t process_io_rotate(Master_info* mi, Rotate_log_event* rev);
133
 
static int32_t process_io_create_file(Master_info* mi, Create_file_log_event* cev);
 
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);
134
126
static bool wait_for_relay_log_space(Relay_log_info* rli);
135
127
static inline bool io_slave_killed(THD* thd,Master_info* mi);
136
128
static inline bool sql_slave_killed(THD* thd,Relay_log_info* rli);
137
 
static int32_t init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type);
138
 
static int32_t safe_connect(THD* thd, DRIZZLE *drizzle, Master_info* mi);
139
 
static int32_t safe_reconnect(THD* thd, DRIZZLE *drizzle, Master_info* mi,
 
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,
140
132
                          bool suppress_warnings);
141
 
static int32_t connect_to_master(THD* thd, DRIZZLE *drizzle, Master_info* mi,
 
133
static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi,
142
134
                             bool reconnect, bool suppress_warnings);
143
 
static int32_t safe_sleep(THD* thd, int32_t sec, CHECK_KILLED_FUNC thread_killed,
 
135
static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed,
144
136
                      void* thread_killed_arg);
145
 
static int32_t get_master_version_and_clock(DRIZZLE *drizzle, Master_info* mi);
 
137
static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi);
146
138
static Log_event* next_event(Relay_log_info* rli);
147
 
static int32_t queue_event(Master_info* mi,const char* buf,uint32_t event_len);
148
 
static int32_t terminate_slave_thread(THD *thd,
 
139
static int queue_event(Master_info* mi,const char* buf,ulong event_len);
 
140
static int terminate_slave_thread(THD *thd,
149
141
                                  pthread_mutex_t* term_lock,
150
142
                                  pthread_cond_t* term_cond,
151
 
                                  volatile uint32_t *slave_running,
 
143
                                  volatile uint *slave_running,
152
144
                                  bool skip_lock);
153
145
static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info);
154
146
 
170
162
                If inverse == 1, stopped threads
171
163
*/
172
164
 
173
 
void init_thread_mask(int32_t* mask,Master_info* mi,bool inverse)
 
165
void init_thread_mask(int* mask,Master_info* mi,bool inverse)
174
166
{
175
167
  bool set_io = mi->slave_running, set_sql = mi->rli.slave_running;
176
 
  register int32_t tmp_mask=0;
 
168
  register int tmp_mask=0;
 
169
  DBUG_ENTER("init_thread_mask");
177
170
 
178
171
  if (set_io)
179
172
    tmp_mask |= SLAVE_IO;
182
175
  if (inverse)
183
176
    tmp_mask^= (SLAVE_IO | SLAVE_SQL);
184
177
  *mask = tmp_mask;
185
 
  return;
 
178
  DBUG_VOID_RETURN;
186
179
}
187
180
 
188
181
 
192
185
 
193
186
void lock_slave_threads(Master_info* mi)
194
187
{
 
188
  DBUG_ENTER("lock_slave_threads");
 
189
 
195
190
  //TODO: see if we can do this without dual mutex
196
191
  pthread_mutex_lock(&mi->run_lock);
197
192
  pthread_mutex_lock(&mi->rli.run_lock);
198
 
  return;
 
193
  DBUG_VOID_RETURN;
199
194
}
200
195
 
201
196
 
205
200
 
206
201
void unlock_slave_threads(Master_info* mi)
207
202
{
 
203
  DBUG_ENTER("unlock_slave_threads");
 
204
 
208
205
  //TODO: see if we can do this without dual mutex
209
206
  pthread_mutex_unlock(&mi->rli.run_lock);
210
207
  pthread_mutex_unlock(&mi->run_lock);
211
 
  return;
 
208
  DBUG_VOID_RETURN;
212
209
}
213
210
 
214
211
 
215
212
/* Initialize slave structures */
216
213
 
217
 
int32_t init_slave()
 
214
int init_slave()
218
215
{
 
216
  DBUG_ENTER("init_slave");
 
217
 
219
218
  /*
220
219
    This is called when mysqld starts. Before client connections are
221
220
    accepted. However bootstrap may conflict with us if it does START SLAVE.
235
234
  */
236
235
  if (!active_mi)
237
236
  {
238
 
    sql_print_error(_("Failed to allocate memory for the master info structure"));
 
237
    sql_print_error("Failed to allocate memory for the master info structure");
239
238
    goto err;
240
239
  }
241
240
 
242
 
  if (active_mi->init_master_info(master_info_file, relay_log_info_file, (SLAVE_IO | SLAVE_SQL)))
 
241
  if (init_master_info(active_mi,master_info_file,relay_log_info_file,
 
242
                       1, (SLAVE_IO | SLAVE_SQL)))
243
243
  {
244
 
    sql_print_error(_("Failed to initialize the master info structure"));
 
244
    sql_print_error("Failed to initialize the master info structure");
245
245
    goto err;
246
246
  }
247
247
 
256
256
                            relay_log_info_file,
257
257
                            SLAVE_IO | SLAVE_SQL))
258
258
    {
259
 
      sql_print_error(_("Failed to create slave threads"));
 
259
      sql_print_error("Failed to create slave threads");
260
260
      goto err;
261
261
    }
262
262
  }
263
263
  pthread_mutex_unlock(&LOCK_active_mi);
264
 
  return(0);
 
264
  DBUG_RETURN(0);
265
265
 
266
266
err:
267
267
  pthread_mutex_unlock(&LOCK_active_mi);
268
 
  return(1);
 
268
  DBUG_RETURN(1);
269
269
}
270
270
 
271
271
 
283
283
void init_slave_skip_errors(const char* arg)
284
284
{
285
285
  const char *p;
 
286
  DBUG_ENTER("init_slave_skip_errors");
286
287
 
287
288
  if (bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0))
288
289
  {
292
293
  use_slave_mask = 1;
293
294
  for (;my_isspace(system_charset_info,*arg);++arg)
294
295
    /* empty */;
295
 
  if (!my_strnncoll(system_charset_info,(unsigned char*)arg,4,(const unsigned char*)"all",4))
 
296
  if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const uchar*)"all",4))
296
297
  {
297
298
    bitmap_set_all(&slave_error_mask);
298
 
    return;
 
299
    DBUG_VOID_RETURN;
299
300
  }
300
301
  for (p= arg ; *p; )
301
302
  {
303
304
    if (!(p= str2int(p, 10, 0, LONG_MAX, &err_code)))
304
305
      break;
305
306
    if (err_code < MAX_SLAVE_ERROR)
306
 
       bitmap_set_bit(&slave_error_mask,(uint32_t)err_code);
 
307
       bitmap_set_bit(&slave_error_mask,(uint)err_code);
307
308
    while (!my_isdigit(system_charset_info,*p) && *p)
308
309
      p++;
309
310
  }
310
 
  return;
 
311
  DBUG_VOID_RETURN;
311
312
}
312
313
 
313
314
 
314
 
int32_t terminate_slave_threads(Master_info* mi,int32_t thread_mask,bool skip_lock)
 
315
int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)
315
316
{
 
317
  DBUG_ENTER("terminate_slave_threads");
 
318
 
316
319
  if (!mi->inited)
317
 
    return(0); /* successfully do nothing */
318
 
  int32_t error,force_all = (thread_mask & SLAVE_FORCE_ALL);
 
320
    DBUG_RETURN(0); /* successfully do nothing */
 
321
  int error,force_all = (thread_mask & SLAVE_FORCE_ALL);
319
322
  pthread_mutex_t *sql_lock = &mi->rli.run_lock, *io_lock = &mi->run_lock;
320
323
 
321
324
  if ((thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL)))
322
325
  {
 
326
    DBUG_PRINT("info",("Terminating IO thread"));
323
327
    mi->abort_slave=1;
324
328
    if ((error=terminate_slave_thread(mi->io_thd,io_lock,
325
329
                                      &mi->stop_cond,
326
330
                                      &mi->slave_running,
327
331
                                      skip_lock)) &&
328
332
        !force_all)
329
 
      return(error);
 
333
      DBUG_RETURN(error);
330
334
  }
331
335
  if ((thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL)))
332
336
  {
 
337
    DBUG_PRINT("info",("Terminating SQL thread"));
333
338
    mi->rli.abort_slave=1;
334
339
    if ((error=terminate_slave_thread(mi->rli.sql_thd,sql_lock,
335
340
                                      &mi->rli.stop_cond,
336
341
                                      &mi->rli.slave_running,
337
342
                                      skip_lock)) &&
338
343
        !force_all)
339
 
      return(error);
 
344
      DBUG_RETURN(error);
340
345
  }
341
 
  return(0);
 
346
  DBUG_RETURN(0);
342
347
}
343
348
 
344
349
 
371
376
 
372
377
   @retval 0 All OK
373
378
 */
374
 
static int32_t
 
379
static int
375
380
terminate_slave_thread(THD *thd,
376
381
                       pthread_mutex_t* term_lock,
377
382
                       pthread_cond_t* term_cond,
378
 
                       volatile uint32_t *slave_running,
 
383
                       volatile uint *slave_running,
379
384
                       bool skip_lock)
380
385
{
381
 
  int32_t error;
 
386
  int error;
 
387
 
 
388
  DBUG_ENTER("terminate_slave_thread");
382
389
 
383
390
  if (!skip_lock)
384
391
    pthread_mutex_lock(term_lock);
389
396
  {
390
397
    if (!skip_lock)
391
398
      pthread_mutex_unlock(term_lock);
392
 
    return(ER_SLAVE_NOT_RUNNING);
 
399
    DBUG_RETURN(ER_SLAVE_NOT_RUNNING);
393
400
  }
394
 
  assert(thd != 0);
 
401
  DBUG_ASSERT(thd != 0);
395
402
  THD_CHECK_SENTRY(thd);
396
403
 
397
404
  /*
401
408
 
402
409
  while (*slave_running)                        // Should always be true
403
410
  {
 
411
    DBUG_PRINT("loop", ("killing slave thread"));
 
412
 
404
413
    pthread_mutex_lock(&thd->LOCK_delete);
405
414
#ifndef DONT_USE_THR_ALARM
406
415
    /*
408
417
      EINVAL: invalid signal number (can't happen)
409
418
      ESRCH: thread already killed (can happen, should be ignored)
410
419
    */
411
 
    int32_t err= pthread_kill(thd->real_id, thr_client_alarm);
412
 
    assert(err != EINVAL);
 
420
    IF_DBUG(int err= ) pthread_kill(thd->real_id, thr_client_alarm);
 
421
    DBUG_ASSERT(err != EINVAL);
413
422
#endif
414
423
    thd->awake(THD::NOT_KILLED);
415
424
    pthread_mutex_unlock(&thd->LOCK_delete);
421
430
    struct timespec abstime;
422
431
    set_timespec(abstime,2);
423
432
    error= pthread_cond_timedwait(term_cond, term_lock, &abstime);
424
 
    assert(error == ETIMEDOUT || error == 0);
 
433
    DBUG_ASSERT(error == ETIMEDOUT || error == 0);
425
434
  }
426
435
 
427
 
  assert(*slave_running == 0);
 
436
  DBUG_ASSERT(*slave_running == 0);
428
437
 
429
438
  if (!skip_lock)
430
439
    pthread_mutex_unlock(term_lock);
431
 
  return(0);
 
440
  DBUG_RETURN(0);
432
441
}
433
442
 
434
443
 
435
 
int32_t start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,
436
 
                           pthread_mutex_t *cond_lock,
437
 
                           pthread_cond_t *start_cond,
438
 
                           volatile uint32_t *slave_running,
439
 
                           volatile uint32_t *slave_run_id,
440
 
                           Master_info* mi,
441
 
                           bool high_priority)
 
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,
 
449
                       Master_info* mi,
 
450
                       bool high_priority)
442
451
{
443
452
  pthread_t th;
444
 
  uint32_t start_id;
 
453
  ulong start_id;
 
454
  DBUG_ENTER("start_slave_thread");
445
455
 
446
 
  assert(mi->inited);
 
456
  DBUG_ASSERT(mi->inited);
447
457
 
448
458
  if (start_lock)
449
459
    pthread_mutex_lock(start_lock);
453
463
      pthread_cond_broadcast(start_cond);
454
464
    if (start_lock)
455
465
      pthread_mutex_unlock(start_lock);
456
 
    sql_print_error(_("Server id not set, will not start slave"));
457
 
    return(ER_BAD_SLAVE);
 
466
    sql_print_error("Server id not set, will not start slave");
 
467
    DBUG_RETURN(ER_BAD_SLAVE);
458
468
  }
459
469
 
460
470
  if (*slave_running)
463
473
      pthread_cond_broadcast(start_cond);
464
474
    if (start_lock)
465
475
      pthread_mutex_unlock(start_lock);
466
 
    return(ER_SLAVE_MUST_STOP);
 
476
    DBUG_RETURN(ER_SLAVE_MUST_STOP);
467
477
  }
468
478
  start_id= *slave_run_id;
 
479
  DBUG_PRINT("info",("Creating new slave thread"));
469
480
  if (high_priority)
470
481
  {
471
482
    struct sched_param tmp_sched_param;
478
489
  {
479
490
    if (start_lock)
480
491
      pthread_mutex_unlock(start_lock);
481
 
    return(ER_SLAVE_THREAD);
 
492
    DBUG_RETURN(ER_SLAVE_THREAD);
482
493
  }
483
494
  if (start_cond && cond_lock) // caller has cond_lock
484
495
  {
485
496
    THD* thd = current_thd;
486
497
    while (start_id == *slave_run_id)
487
498
    {
 
499
      DBUG_PRINT("sleep",("Waiting for slave thread to start"));
488
500
      const char* old_msg = thd->enter_cond(start_cond,cond_lock,
489
501
                                            "Waiting for slave thread to start");
490
502
      pthread_cond_wait(start_cond,cond_lock);
491
503
      thd->exit_cond(old_msg);
492
504
      pthread_mutex_lock(cond_lock); // re-acquire it as exit_cond() released
493
505
      if (thd->killed)
494
 
        return(thd->killed_errno());
 
506
        DBUG_RETURN(thd->killed_errno());
495
507
    }
496
508
  }
497
509
  if (start_lock)
498
510
    pthread_mutex_unlock(start_lock);
499
 
  return(0);
 
511
  DBUG_RETURN(0);
500
512
}
501
513
 
502
514
 
509
521
    started the threads that were not previously running
510
522
*/
511
523
 
512
 
int32_t start_slave_threads(bool need_slave_mutex, bool wait_for_start,
513
 
                        Master_info* mi,
514
 
                        const char* master_info_fname __attribute__((unused)),
515
 
                        const char* slave_info_fname __attribute__((unused)),
516
 
                        int32_t thread_mask)
 
524
int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
 
525
                        Master_info* mi, const char* master_info_fname,
 
526
                        const char* slave_info_fname, int thread_mask)
517
527
{
518
528
  pthread_mutex_t *lock_io=0,*lock_sql=0,*lock_cond_io=0,*lock_cond_sql=0;
519
529
  pthread_cond_t* cond_io=0,*cond_sql=0;
520
 
  int32_t error=0;
 
530
  int error=0;
 
531
  DBUG_ENTER("start_slave_threads");
521
532
 
522
533
  if (need_slave_mutex)
523
534
  {
533
544
  }
534
545
 
535
546
  if (thread_mask & SLAVE_IO)
536
 
    error= start_slave_thread(handle_slave_io,lock_io,lock_cond_io,
537
 
                              cond_io,
538
 
                              &mi->slave_running, &mi->slave_run_id,
539
 
                              mi, 1); //high priority, to read the most possible
 
547
    error=start_slave_thread(handle_slave_io,lock_io,lock_cond_io,
 
548
                             cond_io,
 
549
                             &mi->slave_running, &mi->slave_run_id,
 
550
                             mi, 1); //high priority, to read the most possible
540
551
  if (!error && (thread_mask & SLAVE_SQL))
541
552
  {
542
 
    error= start_slave_thread(handle_slave_sql,lock_sql,lock_cond_sql,
543
 
                              cond_sql,
544
 
                              &mi->rli.slave_running, &mi->rli.slave_run_id,
545
 
                              mi, 0);
 
553
    error=start_slave_thread(handle_slave_sql,lock_sql,lock_cond_sql,
 
554
                             cond_sql,
 
555
                             &mi->rli.slave_running, &mi->rli.slave_run_id,
 
556
                             mi, 0);
546
557
    if (error)
547
558
      terminate_slave_threads(mi, thread_mask & SLAVE_IO, 0);
548
559
  }
549
 
  return(error);
 
560
  DBUG_RETURN(error);
550
561
}
551
562
 
552
563
 
553
564
#ifdef NOT_USED_YET
554
 
static int32_t end_slave_on_walk(Master_info* mi, unsigned char* /*unused*/)
 
565
static int end_slave_on_walk(Master_info* mi, uchar* /*unused*/)
555
566
{
 
567
  DBUG_ENTER("end_slave_on_walk");
 
568
 
556
569
  end_master_info(mi);
557
 
  return(0);
 
570
  DBUG_RETURN(0);
558
571
}
559
572
#endif
560
573
 
568
581
 
569
582
void end_slave()
570
583
{
 
584
  DBUG_ENTER("end_slave");
 
585
 
571
586
  /*
572
587
    This is called when the server terminates, in close_connections().
573
588
    It terminates slave threads. However, some CHANGE MASTER etc may still be
584
599
      once multi-master code is ready.
585
600
    */
586
601
    terminate_slave_threads(active_mi,SLAVE_FORCE_ALL);
587
 
    active_mi->end_master_info();
 
602
    end_master_info(active_mi);
588
603
    delete active_mi;
589
604
    active_mi= 0;
590
605
  }
591
606
  pthread_mutex_unlock(&LOCK_active_mi);
592
 
  return;
 
607
  DBUG_VOID_RETURN;
593
608
}
594
609
 
595
610
 
596
611
static bool io_slave_killed(THD* thd, Master_info* mi)
597
612
{
598
 
  assert(mi->io_thd == thd);
599
 
  assert(mi->slave_running); // tracking buffer overrun
600
 
  return(mi->abort_slave || abort_loop || thd->killed);
 
613
  DBUG_ENTER("io_slave_killed");
 
614
 
 
615
  DBUG_ASSERT(mi->io_thd == thd);
 
616
  DBUG_ASSERT(mi->slave_running); // tracking buffer overrun
 
617
  DBUG_RETURN(mi->abort_slave || abort_loop || thd->killed);
601
618
}
602
619
 
603
620
 
604
621
static bool sql_slave_killed(THD* thd, Relay_log_info* rli)
605
622
{
606
 
  assert(rli->sql_thd == thd);
607
 
  assert(rli->slave_running == 1);// tracking buffer overrun
 
623
  DBUG_ENTER("sql_slave_killed");
 
624
 
 
625
  DBUG_ASSERT(rli->sql_thd == thd);
 
626
  DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun
608
627
  if (abort_loop || thd->killed || rli->abort_slave)
609
628
  {
610
629
    /*
617
636
      is actively working.
618
637
    */
619
638
    if (rli->last_event_start_time == 0)
620
 
      return(1);
 
639
      DBUG_RETURN(1);
 
640
    DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving "
 
641
                        "it some grace period"));
621
642
    if (difftime(time(0), rli->last_event_start_time) > 60)
622
643
    {
623
644
      rli->report(ERROR_LEVEL, 0,
624
 
                  _("SQL thread had to stop in an unsafe situation, in "
 
645
                  "SQL thread had to stop in an unsafe situation, in "
625
646
                  "the middle of applying updates to a "
626
647
                  "non-transactional table without any primary key. "
627
648
                  "There is a risk of duplicate updates when the slave "
628
649
                  "SQL thread is restarted. Please check your tables' "
629
 
                  "contents after restart."));
630
 
      return(1);
 
650
                  "contents after restart.");
 
651
      DBUG_RETURN(1);
631
652
    }
632
653
  }
633
 
  return(0);
 
654
  DBUG_RETURN(0);
634
655
}
635
656
 
636
657
 
643
664
 
644
665
void skip_load_data_infile(NET *net)
645
666
{
 
667
  DBUG_ENTER("skip_load_data_infile");
 
668
 
646
669
  (void)net_request_file(net, "/dev/null");
647
670
  (void)my_net_read(net);                               // discard response
648
 
  (void)net_write_command(net, 0, (unsigned char*) "", 0, (unsigned char*) "", 0); // ok
649
 
  return;
 
671
  (void)net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0); // ok
 
672
  DBUG_VOID_RETURN;
650
673
}
651
674
 
652
675
 
653
676
bool net_request_file(NET* net, const char* fname)
654
677
{
655
 
  return(net_write_command(net, 251, (unsigned char*) fname, strlen(fname),
656
 
                                (unsigned char*) "", 0));
 
678
  DBUG_ENTER("net_request_file");
 
679
  DBUG_RETURN(net_write_command(net, 251, (uchar*) fname, strlen(fname),
 
680
                                (uchar*) "", 0));
657
681
}
658
682
 
659
683
/*
665
689
 
666
690
const char *print_slave_db_safe(const char* db)
667
691
{
668
 
  return((db ? db : ""));
 
692
  DBUG_ENTER("*print_slave_db_safe");
 
693
 
 
694
  DBUG_RETURN((db ? db : ""));
669
695
}
670
696
 
671
 
int32_t init_strvar_from_file(char *var, int32_t max_size, IO_CACHE *f,
 
697
int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
672
698
                                 const char *default_val)
673
699
{
674
 
  uint32_t length;
 
700
  uint length;
 
701
  DBUG_ENTER("init_strvar_from_file");
675
702
 
676
703
  if ((length=my_b_gets(f,var, max_size)))
677
704
  {
684
711
        If we truncated a line or stopped on last char, remove all chars
685
712
        up to and including newline.
686
713
      */
687
 
      int32_t c;
688
 
      while (((c=my_b_get(f)) != '\n' && c != my_b_EOF)) {};
 
714
      int c;
 
715
      while (((c=my_b_get(f)) != '\n' && c != my_b_EOF));
689
716
    }
690
 
    return(0);
 
717
    DBUG_RETURN(0);
691
718
  }
692
719
  else if (default_val)
693
720
  {
694
721
    strmake(var,  default_val, max_size-1);
695
 
    return(0);
 
722
    DBUG_RETURN(0);
696
723
  }
697
 
  return(1);
 
724
  DBUG_RETURN(1);
698
725
}
699
726
 
700
727
 
701
 
int32_t init_intvar_from_file(int32_t* var, IO_CACHE* f, int32_t default_val)
 
728
int init_intvar_from_file(int* var, IO_CACHE* f, int default_val)
702
729
{
703
730
  char buf[32];
 
731
  DBUG_ENTER("init_intvar_from_file");
704
732
 
705
733
 
706
734
  if (my_b_gets(f, buf, sizeof(buf)))
707
735
  {
708
736
    *var = atoi(buf);
709
 
    return(0);
 
737
    DBUG_RETURN(0);
710
738
  }
711
739
  else if (default_val)
712
740
  {
713
741
    *var = default_val;
714
 
    return(0);
 
742
    DBUG_RETURN(0);
715
743
  }
716
 
  return(1);
 
744
  DBUG_RETURN(1);
717
745
}
718
746
 
719
 
int32_t init_floatvar_from_file(float* var, IO_CACHE* f, float default_val)
 
747
int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val)
720
748
{
721
749
  char buf[16];
 
750
  DBUG_ENTER("init_floatvar_from_file");
722
751
 
723
752
 
724
753
  if (my_b_gets(f, buf, sizeof(buf)))
725
754
  {
726
755
    if (sscanf(buf, "%f", var) != 1)
727
 
      return(1);
 
756
      DBUG_RETURN(1);
728
757
    else
729
 
      return(0);
 
758
      DBUG_RETURN(0);
730
759
  }
731
760
  else if (default_val != 0.0)
732
761
  {
733
762
    *var = default_val;
734
 
    return(0);
 
763
    DBUG_RETURN(0);
735
764
  }
736
 
  return(1);
 
765
  DBUG_RETURN(1);
737
766
}
738
767
 
739
768
static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info)
742
771
  {
743
772
    if (info && global_system_variables.log_warnings)
744
773
      sql_print_information(info);
745
 
    return true;
 
774
    return TRUE;
746
775
  }
747
 
  return false;
 
776
  return FALSE;
748
777
}
749
778
 
750
779
 
762
791
  1       error
763
792
*/
764
793
 
765
 
static int32_t get_master_version_and_clock(DRIZZLE *drizzle, Master_info* mi)
 
794
static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi)
766
795
{
767
796
  char error_buf[512];
768
797
  String err_msg(error_buf, sizeof(error_buf), &my_charset_bin);
769
798
  char err_buff[MAX_SLAVE_ERRMSG];
770
799
  const char* errmsg= 0;
771
 
  int32_t err_code= 0;
772
 
  DRIZZLE_RES *master_res= 0;
773
 
  DRIZZLE_ROW master_row;
 
800
  int err_code= 0;
 
801
  MYSQL_RES *master_res= 0;
 
802
  MYSQL_ROW master_row;
 
803
  DBUG_ENTER("get_master_version_and_clock");
774
804
 
775
805
  err_msg.length(0);
776
806
  /*
780
810
  delete mi->rli.relay_log.description_event_for_queue;
781
811
  mi->rli.relay_log.description_event_for_queue= 0;
782
812
 
783
 
  if (!my_isdigit(&my_charset_bin,*drizzle->server_version))
 
813
  if (!my_isdigit(&my_charset_bin,*mysql->server_version))
784
814
  {
785
 
    errmsg = _("Master reported unrecognized DRIZZLE version");
 
815
    errmsg = "Master reported unrecognized MySQL version";
786
816
    err_code= ER_SLAVE_FATAL_ERROR;
787
817
    sprintf(err_buff, ER(err_code), errmsg);
788
818
    err_msg.append(err_buff);
790
820
  else
791
821
  {
792
822
    /*
793
 
      Note the following switch will bug when we have DRIZZLE branch 30 ;)
 
823
      Note the following switch will bug when we have MySQL branch 30 ;)
794
824
    */
795
 
    switch (*drizzle->server_version)
 
825
    switch (*mysql->server_version)
796
826
    {
797
827
    case '0':
798
828
    case '1':
799
829
    case '2':
800
 
      errmsg = _("Master reported unrecognized DRIZZLE version");
 
830
      errmsg = "Master reported unrecognized MySQL version";
801
831
      err_code= ER_SLAVE_FATAL_ERROR;
802
832
      sprintf(err_buff, ER(err_code), errmsg);
803
833
      err_msg.append(err_buff);
804
834
      break;
805
835
    case '3':
806
836
      mi->rli.relay_log.description_event_for_queue= new
807
 
        Format_description_log_event(1, drizzle->server_version);
 
837
        Format_description_log_event(1, mysql->server_version);
808
838
      break;
809
839
    case '4':
810
840
      mi->rli.relay_log.description_event_for_queue= new
811
 
        Format_description_log_event(3, drizzle->server_version);
 
841
        Format_description_log_event(3, mysql->server_version);
812
842
      break;
813
843
    default:
814
844
      /*
815
 
        Master is DRIZZLE >=5.0. Give a default Format_desc event, so that we can
 
845
        Master is MySQL >=5.0. Give a default Format_desc event, so that we can
816
846
        take the early steps (like tests for "is this a 3.23 master") which we
817
847
        have to take before we receive the real master's Format_desc which will
818
848
        override this one. Note that the Format_desc we create below is garbage
820
850
        master is 3.23, 4.0, etc.
821
851
      */
822
852
      mi->rli.relay_log.description_event_for_queue= new
823
 
        Format_description_log_event(4, drizzle->server_version);
 
853
        Format_description_log_event(4, mysql->server_version);
824
854
      break;
825
855
    }
826
856
  }
838
868
  /* as we are here, we tried to allocate the event */
839
869
  if (!mi->rli.relay_log.description_event_for_queue)
840
870
  {
841
 
    errmsg= _("default Format_description_log_event");
 
871
    errmsg= "default Format_description_log_event";
842
872
    err_code= ER_SLAVE_CREATE_EVENT_FAILURE;
843
873
    sprintf(err_buff, ER(err_code), errmsg);
844
874
    err_msg.append(err_buff);
850
880
    unavailable (very old master not supporting UNIX_TIMESTAMP()?).
851
881
  */
852
882
 
853
 
  if (!drizzle_real_query(drizzle, STRING_WITH_LEN("SELECT UNIX_TIMESTAMP()")) &&
854
 
      (master_res= drizzle_store_result(drizzle)) &&
855
 
      (master_row= drizzle_fetch_row(master_res)))
 
883
  if (!mysql_real_query(mysql, STRING_WITH_LEN("SELECT UNIX_TIMESTAMP()")) &&
 
884
      (master_res= mysql_store_result(mysql)) &&
 
885
      (master_row= mysql_fetch_row(master_res)))
856
886
  {
857
887
    mi->clock_diff_with_master=
858
888
      (long) (time((time_t*) 0) - strtoul(master_row[0], 0, 10));
860
890
  else if (!check_io_slave_killed(mi->io_thd, mi, NULL))
861
891
  {
862
892
    mi->clock_diff_with_master= 0; /* The "most sensible" value */
863
 
    sql_print_warning(_("\"SELECT UNIX_TIMESTAMP()\" failed on master, "
864
 
                        "do not trust column Seconds_Behind_Master of SHOW "
865
 
                        "SLAVE STATUS. Error: %s (%d)"),
866
 
                      drizzle_error(drizzle), drizzle_errno(drizzle));
 
893
    sql_print_warning("\"SELECT UNIX_TIMESTAMP()\" failed on master, "
 
894
                      "do not trust column Seconds_Behind_Master of SHOW "
 
895
                      "SLAVE STATUS. Error: %s (%d)",
 
896
                      mysql_error(mysql), mysql_errno(mysql));
867
897
  }
868
898
  if (master_res)
869
 
    drizzle_free_result(master_res);
 
899
    mysql_free_result(master_res);
870
900
 
871
901
  /*
872
902
    Check that the master's server id and ours are different. Because if they
878
908
    Note: we could have put a @@SERVER_ID in the previous SELECT
879
909
    UNIX_TIMESTAMP() instead, but this would not have worked on 3.23 masters.
880
910
  */
881
 
  if (!drizzle_real_query(drizzle,
 
911
  if (!mysql_real_query(mysql,
882
912
                        STRING_WITH_LEN("SHOW VARIABLES LIKE 'SERVER_ID'")) &&
883
 
      (master_res= drizzle_store_result(drizzle)))
 
913
      (master_res= mysql_store_result(mysql)))
884
914
  {
885
 
    if ((master_row= drizzle_fetch_row(master_res)) &&
 
915
    if ((master_row= mysql_fetch_row(master_res)) &&
886
916
        (::server_id == strtoul(master_row[1], 0, 10)) &&
887
917
        !mi->rli.replicate_same_server_id)
888
918
    {
889
919
      errmsg=
890
 
        _("The slave I/O thread stops because master and slave have equal "
891
 
          "DRIZZLE server ids; these ids must be different "
892
 
          "for replication to work (or "
893
 
          "the --replicate-same-server-id option must be used "
894
 
          "on slave but this does"
895
 
          "not always make sense; please check the manual before using it).");
 
920
        "The slave I/O thread stops because master and slave have equal"
 
921
        " MySQL server ids; these ids must be different for replication to work (or"
 
922
        " the --replicate-same-server-id option must be used on slave but this does"
 
923
        " not always make sense; please check the manual before using it).";
896
924
      err_code= ER_SLAVE_FATAL_ERROR;
897
925
      sprintf(err_buff, ER(err_code), errmsg);
898
926
      err_msg.append(err_buff);
899
927
    }
900
 
    drizzle_free_result(master_res);
 
928
    mysql_free_result(master_res);
901
929
    if (errmsg)
902
930
      goto err;
903
931
  }
920
948
  */
921
949
 
922
950
  /* redundant with rest of code but safer against later additions */
923
 
  if (*drizzle->server_version == '3')
 
951
  if (*mysql->server_version == '3')
924
952
    goto err;
925
953
 
926
 
  if ((*drizzle->server_version == '4') &&
927
 
      !drizzle_real_query(drizzle,
 
954
  if ((*mysql->server_version == '4') &&
 
955
      !mysql_real_query(mysql,
928
956
                        STRING_WITH_LEN("SELECT @@GLOBAL.COLLATION_SERVER")) &&
929
 
      (master_res= drizzle_store_result(drizzle)))
 
957
      (master_res= mysql_store_result(mysql)))
930
958
  {
931
 
    if ((master_row= drizzle_fetch_row(master_res)) &&
 
959
    if ((master_row= mysql_fetch_row(master_res)) &&
932
960
        strcmp(master_row[0], global_system_variables.collation_server->name))
933
961
    {
934
962
      errmsg=
935
 
        _("The slave I/O thread stops because master and slave have"
936
 
          " different values for the COLLATION_SERVER global variable."
937
 
          " The values must be equal for replication to work");
 
963
        "The slave I/O thread stops because master and slave have"
 
964
        " different values for the COLLATION_SERVER global variable."
 
965
        " The values must be equal for replication to work";
938
966
      err_code= ER_SLAVE_FATAL_ERROR;
939
967
      sprintf(err_buff, ER(err_code), errmsg);
940
968
      err_msg.append(err_buff);
941
969
    }
942
 
    drizzle_free_result(master_res);
 
970
    mysql_free_result(master_res);
943
971
    if (errmsg)
944
972
      goto err;
945
973
  }
959
987
    This check is only necessary for 4.x masters (and < 5.0.4 masters but
960
988
    those were alpha).
961
989
  */
962
 
  if ((*drizzle->server_version == '4') &&
963
 
      !drizzle_real_query(drizzle, STRING_WITH_LEN("SELECT @@GLOBAL.TIME_ZONE")) &&
964
 
      (master_res= drizzle_store_result(drizzle)))
 
990
  if ((*mysql->server_version == '4') &&
 
991
      !mysql_real_query(mysql, STRING_WITH_LEN("SELECT @@GLOBAL.TIME_ZONE")) &&
 
992
      (master_res= mysql_store_result(mysql)))
965
993
  {
966
 
    if ((master_row= drizzle_fetch_row(master_res)) &&
 
994
    if ((master_row= mysql_fetch_row(master_res)) &&
967
995
        strcmp(master_row[0],
968
996
               global_system_variables.time_zone->get_name()->ptr()))
969
997
    {
970
998
      errmsg=
971
 
        _("The slave I/O thread stops because master and slave have"
972
 
          " different values for the TIME_ZONE global variable."
973
 
          " The values must be equal for replication to work");
 
999
        "The slave I/O thread stops because master and slave have"
 
1000
        " different values for the TIME_ZONE global variable."
 
1001
        " The values must be equal for replication to work";
974
1002
      err_code= ER_SLAVE_FATAL_ERROR;
975
1003
      sprintf(err_buff, ER(err_code), errmsg);
976
1004
      err_msg.append(err_buff);
977
1005
    }
978
 
    drizzle_free_result(master_res);
 
1006
    mysql_free_result(master_res);
979
1007
 
980
1008
    if (errmsg)
981
1009
      goto err;
987
1015
    const char query_format[]= "SET @master_heartbeat_period= %s";
988
1016
    char query[sizeof(query_format) - 2 + sizeof(llbuf)];
989
1017
    /* 
990
 
       the period is an uint64_t of nano-secs. 
 
1018
       the period is an ulonglong of nano-secs. 
991
1019
    */
992
 
    llstr((uint64_t) (mi->heartbeat_period*1000000000UL), llbuf);
993
 
    sprintf(query, query_format, llbuf);
 
1020
    llstr((ulonglong) (mi->heartbeat_period*1000000000UL), llbuf);
 
1021
    my_sprintf(query, (query, query_format, llbuf));
994
1022
 
995
 
    if (drizzle_real_query(drizzle, query, strlen(query))
 
1023
    if (mysql_real_query(mysql, query, strlen(query))
996
1024
        && !check_io_slave_killed(mi->io_thd, mi, NULL))
997
1025
    {
998
1026
      err_msg.append("The slave I/O thread stops because querying master with '");
999
1027
      err_msg.append(query);
1000
1028
      err_msg.append("' failed;");
1001
1029
      err_msg.append(" error: ");
1002
 
      err_code= drizzle_errno(drizzle);
 
1030
      err_code= mysql_errno(mysql);
1003
1031
      err_msg.qs_append(err_code);
1004
1032
      err_msg.append("  '");
1005
 
      err_msg.append(drizzle_error(drizzle));
 
1033
      err_msg.append(mysql_error(mysql));
1006
1034
      err_msg.append("'");
1007
 
      drizzle_free_result(drizzle_store_result(drizzle));
 
1035
      mysql_free_result(mysql_store_result(mysql));
1008
1036
      goto err;
1009
1037
    }
1010
 
    drizzle_free_result(drizzle_store_result(drizzle));
 
1038
    mysql_free_result(mysql_store_result(mysql));
1011
1039
  }
1012
1040
  
1013
1041
err:
1014
1042
  if (err_msg.length() != 0)
1015
1043
  {
1016
1044
    sql_print_error(err_msg.ptr());
1017
 
    assert(err_code != 0);
 
1045
    DBUG_ASSERT(err_code != 0);
1018
1046
    mi->report(ERROR_LEVEL, err_code, err_msg.ptr());
1019
 
    return(1);
 
1047
    DBUG_RETURN(1);
1020
1048
  }
1021
1049
 
1022
 
  return(0);
 
1050
  DBUG_RETURN(0);
1023
1051
}
1024
1052
 
1025
1053
 
1029
1057
  Master_info* mi = rli->mi;
1030
1058
  const char *save_proc_info;
1031
1059
  THD* thd = mi->io_thd;
 
1060
  DBUG_ENTER("wait_for_relay_log_space");
1032
1061
 
1033
1062
  pthread_mutex_lock(&rli->log_space_lock);
1034
1063
  save_proc_info= thd->enter_cond(&rli->log_space_cond,
1035
1064
                                  &rli->log_space_lock,
1036
 
                                  _("Waiting for the slave SQL thread "
1037
 
                                    "to free enough relay log space"));
 
1065
                                  "\
 
1066
Waiting for the slave SQL thread to free enough relay log space");
1038
1067
  while (rli->log_space_limit < rli->log_space_total &&
1039
1068
         !(slave_killed=io_slave_killed(thd,mi)) &&
1040
1069
         !rli->ignore_log_space_limit)
1041
1070
    pthread_cond_wait(&rli->log_space_cond, &rli->log_space_lock);
1042
1071
  thd->exit_cond(save_proc_info);
1043
 
  return(slave_killed);
 
1072
  DBUG_RETURN(slave_killed);
1044
1073
}
1045
1074
 
1046
1075
 
1057
1086
    ignored events' end position for the use of the slave SQL thread, by
1058
1087
    calling this function. Only that thread can call it (see assertion).
1059
1088
 */
1060
 
static void write_ignored_events_info_to_relay_log(THD *thd __attribute__((unused)),
1061
 
                                                   Master_info *mi)
 
1089
static void write_ignored_events_info_to_relay_log(THD *thd, Master_info *mi)
1062
1090
{
1063
1091
  Relay_log_info *rli= &mi->rli;
1064
1092
  pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
 
1093
  DBUG_ENTER("write_ignored_events_info_to_relay_log");
1065
1094
 
1066
 
  assert(thd == mi->io_thd);
 
1095
  DBUG_ASSERT(thd == mi->io_thd);
1067
1096
  pthread_mutex_lock(log_lock);
1068
1097
  if (rli->ign_master_log_name_end[0])
1069
1098
  {
 
1099
    DBUG_PRINT("info",("writing a Rotate event to track down ignored events"));
1070
1100
    Rotate_log_event *ev= new Rotate_log_event(rli->ign_master_log_name_end,
1071
1101
                                               0, rli->ign_master_log_pos_end,
1072
1102
                                               Rotate_log_event::DUP_NAME);
1079
1109
      if (unlikely(rli->relay_log.append(ev)))
1080
1110
        mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
1081
1111
                   ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
1082
 
                   _("failed to write a Rotate event"
1083
 
                     " to the relay log, SHOW SLAVE STATUS may be"
1084
 
                     " inaccurate"));
 
1112
                   "failed to write a Rotate event"
 
1113
                   " to the relay log, SHOW SLAVE STATUS may be"
 
1114
                   " inaccurate");
1085
1115
      rli->relay_log.harvest_bytes_written(&rli->log_space_total);
1086
 
      if (mi->flush())
1087
 
        sql_print_error(_("Failed to flush master info file"));
 
1116
      if (flush_master_info(mi, 1))
 
1117
        sql_print_error("Failed to flush master info file");
1088
1118
      delete ev;
1089
1119
    }
1090
1120
    else
1091
1121
      mi->report(ERROR_LEVEL, ER_SLAVE_CREATE_EVENT_FAILURE,
1092
1122
                 ER(ER_SLAVE_CREATE_EVENT_FAILURE),
1093
 
                 _("Rotate_event (out of memory?),"
1094
 
                   " SHOW SLAVE STATUS may be inaccurate"));
 
1123
                 "Rotate_event (out of memory?),"
 
1124
                 " SHOW SLAVE STATUS may be inaccurate");
1095
1125
  }
1096
1126
  else
1097
1127
    pthread_mutex_unlock(log_lock);
1098
 
  return;
 
1128
  DBUG_VOID_RETURN;
1099
1129
}
1100
1130
 
1101
1131
 
1102
 
int32_t register_slave_on_master(DRIZZLE *drizzle, Master_info *mi,
 
1132
int register_slave_on_master(MYSQL* mysql, Master_info *mi,
1103
1133
                             bool *suppress_warnings)
1104
1134
{
1105
 
  unsigned char buf[1024], *pos= buf;
1106
 
  uint32_t report_host_len, report_user_len=0, report_password_len=0;
 
1135
  uchar buf[1024], *pos= buf;
 
1136
  uint report_host_len, report_user_len=0, report_password_len=0;
 
1137
  DBUG_ENTER("register_slave_on_master");
1107
1138
 
1108
 
  *suppress_warnings= false;
 
1139
  *suppress_warnings= FALSE;
1109
1140
  if (!report_host)
1110
 
    return(0);
 
1141
    DBUG_RETURN(0);
1111
1142
  report_host_len= strlen(report_host);
1112
1143
  if (report_user)
1113
1144
    report_user_len= strlen(report_user);
1116
1147
  /* 30 is a good safety margin */
1117
1148
  if (report_host_len + report_user_len + report_password_len + 30 >
1118
1149
      sizeof(buf))
1119
 
    return(0);                                     // safety
 
1150
    DBUG_RETURN(0);                                     // safety
1120
1151
 
1121
1152
  int4store(pos, server_id); pos+= 4;
1122
 
  pos= net_store_data(pos, (unsigned char*) report_host, report_host_len);
1123
 
  pos= net_store_data(pos, (unsigned char*) report_user, report_user_len);
1124
 
  pos= net_store_data(pos, (unsigned char*) report_password, report_password_len);
1125
 
  int2store(pos, (uint16_t) report_port); pos+= 2;
1126
 
  int4store(pos, 0);    pos+= 4;
 
1153
  pos= net_store_data(pos, (uchar*) report_host, report_host_len);
 
1154
  pos= net_store_data(pos, (uchar*) report_user, report_user_len);
 
1155
  pos= net_store_data(pos, (uchar*) report_password, report_password_len);
 
1156
  int2store(pos, (uint16) report_port); pos+= 2;
 
1157
  int4store(pos, rpl_recovery_rank);    pos+= 4;
1127
1158
  /* The master will fill in master_id */
1128
1159
  int4store(pos, 0);                    pos+= 4;
1129
1160
 
1130
 
  if (simple_command(drizzle, COM_REGISTER_SLAVE, buf, (size_t) (pos- buf), 0))
 
1161
  if (simple_command(mysql, COM_REGISTER_SLAVE, buf, (size_t) (pos- buf), 0))
1131
1162
  {
1132
 
    if (drizzle_errno(drizzle) == ER_NET_READ_INTERRUPTED)
 
1163
    if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED)
1133
1164
    {
1134
 
      *suppress_warnings= true;                 // Suppress reconnect warning
 
1165
      *suppress_warnings= TRUE;                 // Suppress reconnect warning
1135
1166
    }
1136
1167
    else if (!check_io_slave_killed(mi->io_thd, mi, NULL))
1137
1168
    {
1138
1169
      char buf[256];
1139
 
      snprintf(buf, sizeof(buf), "%s (Errno: %d)", drizzle_error(drizzle), 
1140
 
               drizzle_errno(drizzle));
 
1170
      my_snprintf(buf, sizeof(buf), "%s (Errno: %d)", mysql_error(mysql), 
 
1171
                  mysql_errno(mysql));
1141
1172
      mi->report(ERROR_LEVEL, ER_SLAVE_MASTER_COM_FAILURE,
1142
1173
                 ER(ER_SLAVE_MASTER_COM_FAILURE), "COM_REGISTER_SLAVE", buf);
1143
1174
    }
1144
 
    return(1);
 
1175
    DBUG_RETURN(1);
1145
1176
  }
1146
 
  return(0);
 
1177
  DBUG_RETURN(0);
1147
1178
}
1148
1179
 
1149
1180
 
1152
1183
  // TODO: fix this for multi-master
1153
1184
  List<Item> field_list;
1154
1185
  Protocol *protocol= thd->protocol;
 
1186
  DBUG_ENTER("show_master_info");
1155
1187
 
1156
1188
  field_list.push_back(new Item_empty_string("Slave_IO_State",
1157
1189
                                                     14));
1160
1192
  field_list.push_back(new Item_empty_string("Master_User",
1161
1193
                                                     sizeof(mi->user)));
1162
1194
  field_list.push_back(new Item_return_int("Master_Port", 7,
1163
 
                                           DRIZZLE_TYPE_LONG));
 
1195
                                           MYSQL_TYPE_LONG));
1164
1196
  field_list.push_back(new Item_return_int("Connect_Retry", 10,
1165
 
                                           DRIZZLE_TYPE_LONG));
 
1197
                                           MYSQL_TYPE_LONG));
1166
1198
  field_list.push_back(new Item_empty_string("Master_Log_File",
1167
1199
                                             FN_REFLEN));
1168
1200
  field_list.push_back(new Item_return_int("Read_Master_Log_Pos", 10,
1169
 
                                           DRIZZLE_TYPE_LONGLONG));
 
1201
                                           MYSQL_TYPE_LONGLONG));
1170
1202
  field_list.push_back(new Item_empty_string("Relay_Log_File",
1171
1203
                                             FN_REFLEN));
1172
1204
  field_list.push_back(new Item_return_int("Relay_Log_Pos", 10,
1173
 
                                           DRIZZLE_TYPE_LONGLONG));
 
1205
                                           MYSQL_TYPE_LONGLONG));
1174
1206
  field_list.push_back(new Item_empty_string("Relay_Master_Log_File",
1175
1207
                                             FN_REFLEN));
1176
1208
  field_list.push_back(new Item_empty_string("Slave_IO_Running", 3));
1182
1214
  field_list.push_back(new Item_empty_string("Replicate_Wild_Do_Table", 24));
1183
1215
  field_list.push_back(new Item_empty_string("Replicate_Wild_Ignore_Table",
1184
1216
                                             28));
1185
 
  field_list.push_back(new Item_return_int("Last_Errno", 4, DRIZZLE_TYPE_LONG));
 
1217
  field_list.push_back(new Item_return_int("Last_Errno", 4, MYSQL_TYPE_LONG));
1186
1218
  field_list.push_back(new Item_empty_string("Last_Error", 20));
1187
1219
  field_list.push_back(new Item_return_int("Skip_Counter", 10,
1188
 
                                           DRIZZLE_TYPE_LONG));
 
1220
                                           MYSQL_TYPE_LONG));
1189
1221
  field_list.push_back(new Item_return_int("Exec_Master_Log_Pos", 10,
1190
 
                                           DRIZZLE_TYPE_LONGLONG));
 
1222
                                           MYSQL_TYPE_LONGLONG));
1191
1223
  field_list.push_back(new Item_return_int("Relay_Log_Space", 10,
1192
 
                                           DRIZZLE_TYPE_LONGLONG));
 
1224
                                           MYSQL_TYPE_LONGLONG));
1193
1225
  field_list.push_back(new Item_empty_string("Until_Condition", 6));
1194
1226
  field_list.push_back(new Item_empty_string("Until_Log_File", FN_REFLEN));
1195
1227
  field_list.push_back(new Item_return_int("Until_Log_Pos", 10,
1196
 
                                           DRIZZLE_TYPE_LONGLONG));
 
1228
                                           MYSQL_TYPE_LONGLONG));
 
1229
  field_list.push_back(new Item_empty_string("Master_SSL_Allowed", 7));
 
1230
  field_list.push_back(new Item_empty_string("Master_SSL_CA_File",
 
1231
                                             sizeof(mi->ssl_ca)));
 
1232
  field_list.push_back(new Item_empty_string("Master_SSL_CA_Path",
 
1233
                                             sizeof(mi->ssl_capath)));
 
1234
  field_list.push_back(new Item_empty_string("Master_SSL_Cert",
 
1235
                                             sizeof(mi->ssl_cert)));
 
1236
  field_list.push_back(new Item_empty_string("Master_SSL_Cipher",
 
1237
                                             sizeof(mi->ssl_cipher)));
 
1238
  field_list.push_back(new Item_empty_string("Master_SSL_Key",
 
1239
                                             sizeof(mi->ssl_key)));
1197
1240
  field_list.push_back(new Item_return_int("Seconds_Behind_Master", 10,
1198
 
                                           DRIZZLE_TYPE_LONGLONG));
1199
 
  field_list.push_back(new Item_return_int("Last_IO_Errno", 4, DRIZZLE_TYPE_LONG));
 
1241
                                           MYSQL_TYPE_LONGLONG));
 
1242
  field_list.push_back(new Item_empty_string("Master_SSL_Verify_Server_Cert",
 
1243
                                             3));
 
1244
  field_list.push_back(new Item_return_int("Last_IO_Errno", 4, MYSQL_TYPE_LONG));
1200
1245
  field_list.push_back(new Item_empty_string("Last_IO_Error", 20));
1201
 
  field_list.push_back(new Item_return_int("Last_SQL_Errno", 4, DRIZZLE_TYPE_LONG));
 
1246
  field_list.push_back(new Item_return_int("Last_SQL_Errno", 4, MYSQL_TYPE_LONG));
1202
1247
  field_list.push_back(new Item_empty_string("Last_SQL_Error", 20));
1203
1248
 
1204
1249
  if (protocol->send_fields(&field_list,
1205
1250
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
1206
 
    return(true);
 
1251
    DBUG_RETURN(TRUE);
1207
1252
 
1208
1253
  if (mi->host[0])
1209
1254
  {
 
1255
    DBUG_PRINT("info",("host is set: '%s'", mi->host));
1210
1256
    String *packet= &thd->packet;
1211
1257
    protocol->prepare_for_resend();
1212
1258
 
1215
1261
      non-volotile members like mi->io_thd, which is guarded by the mutex.
1216
1262
    */
1217
1263
    pthread_mutex_lock(&mi->run_lock);
1218
 
    protocol->store(mi->io_thd ? mi->io_thd->get_proc_info() : "", &my_charset_bin);
 
1264
    protocol->store(mi->io_thd ? mi->io_thd->proc_info : "", &my_charset_bin);
1219
1265
    pthread_mutex_unlock(&mi->run_lock);
1220
1266
 
1221
1267
    pthread_mutex_lock(&mi->data_lock);
1222
1268
    pthread_mutex_lock(&mi->rli.data_lock);
1223
 
    protocol->store(mi->getHostname(), &my_charset_bin);
1224
 
    protocol->store(mi->getUsername(), &my_charset_bin);
1225
 
    protocol->store((uint32_t) mi->getPort());
1226
 
    protocol->store(mi->getConnectionRetry());
1227
 
    protocol->store(mi->getLogName(), &my_charset_bin);
1228
 
    protocol->store((uint64_t) mi->getLogPosition());
1229
 
    protocol->store(mi->rli.group_relay_log_name.c_str() +
1230
 
                    dirname_length(mi->rli.group_relay_log_name.c_str()),
 
1269
    protocol->store(mi->host, &my_charset_bin);
 
1270
    protocol->store(mi->user, &my_charset_bin);
 
1271
    protocol->store((uint32) mi->port);
 
1272
    protocol->store((uint32) mi->connect_retry);
 
1273
    protocol->store(mi->master_log_name, &my_charset_bin);
 
1274
    protocol->store((ulonglong) mi->master_log_pos);
 
1275
    protocol->store(mi->rli.group_relay_log_name +
 
1276
                    dirname_length(mi->rli.group_relay_log_name),
1231
1277
                    &my_charset_bin);
1232
 
    protocol->store((uint64_t) mi->rli.group_relay_log_pos);
1233
 
    protocol->store(mi->rli.group_master_log_name.c_str(), &my_charset_bin);
1234
 
    protocol->store(mi->slave_running == DRIZZLE_SLAVE_RUN_CONNECT ?
 
1278
    protocol->store((ulonglong) mi->rli.group_relay_log_pos);
 
1279
    protocol->store(mi->rli.group_master_log_name, &my_charset_bin);
 
1280
    protocol->store(mi->slave_running == MYSQL_SLAVE_RUN_CONNECT ?
1235
1281
                    "Yes" : "No", &my_charset_bin);
1236
1282
    protocol->store(mi->rli.slave_running ? "Yes":"No", &my_charset_bin);
1237
1283
    protocol->store(rpl_filter->get_do_db());
1250
1296
 
1251
1297
    protocol->store(mi->rli.last_error().number);
1252
1298
    protocol->store(mi->rli.last_error().message, &my_charset_bin);
1253
 
    protocol->store((uint32_t) mi->rli.slave_skip_counter);
1254
 
    protocol->store((uint64_t) mi->rli.group_master_log_pos);
1255
 
    protocol->store((uint64_t) mi->rli.log_space_total);
 
1299
    protocol->store((uint32) mi->rli.slave_skip_counter);
 
1300
    protocol->store((ulonglong) mi->rli.group_master_log_pos);
 
1301
    protocol->store((ulonglong) mi->rli.log_space_total);
1256
1302
 
1257
1303
    protocol->store(
1258
1304
      mi->rli.until_condition==Relay_log_info::UNTIL_NONE ? "None":
1259
1305
        ( mi->rli.until_condition==Relay_log_info::UNTIL_MASTER_POS? "Master":
1260
1306
          "Relay"), &my_charset_bin);
1261
1307
    protocol->store(mi->rli.until_log_name, &my_charset_bin);
1262
 
    protocol->store((uint64_t) mi->rli.until_log_pos);
 
1308
    protocol->store((ulonglong) mi->rli.until_log_pos);
 
1309
 
 
1310
    protocol->store(mi->ssl? "Ignored":"No", &my_charset_bin);
 
1311
    protocol->store(mi->ssl_ca, &my_charset_bin);
 
1312
    protocol->store(mi->ssl_capath, &my_charset_bin);
 
1313
    protocol->store(mi->ssl_cert, &my_charset_bin);
 
1314
    protocol->store(mi->ssl_cipher, &my_charset_bin);
 
1315
    protocol->store(mi->ssl_key, &my_charset_bin);
1263
1316
 
1264
1317
    /*
1265
1318
      Seconds_Behind_Master: if SQL thread is running and I/O thread is
1266
1319
      connected, we can compute it otherwise show NULL (i.e. unknown).
1267
1320
    */
1268
 
    if ((mi->slave_running == DRIZZLE_SLAVE_RUN_CONNECT) &&
 
1321
    if ((mi->slave_running == MYSQL_SLAVE_RUN_CONNECT) &&
1269
1322
        mi->rli.slave_running)
1270
1323
    {
1271
1324
      long time_diff= ((long)(time(0) - mi->rli.last_master_timestamp)
1285
1338
        slave is 2. At SHOW SLAVE STATUS time, assume that the difference
1286
1339
        between timestamp of slave and rli->last_master_timestamp is 0
1287
1340
        (i.e. they are in the same second), then we get 0-(2-1)=-1 as a result.
1288
 
        This confuses users, so we don't go below 0: hence the cmax().
 
1341
        This confuses users, so we don't go below 0: hence the max().
1289
1342
 
1290
1343
        last_master_timestamp == 0 (an "impossible" timestamp 1970) is a
1291
1344
        special marker to say "consider we have caught up".
1292
1345
      */
1293
 
      protocol->store((int64_t)(mi->rli.last_master_timestamp ?
1294
 
                                 cmax((long)0, time_diff) : 0));
 
1346
      protocol->store((longlong)(mi->rli.last_master_timestamp ?
 
1347
                                 max(0, time_diff) : 0));
1295
1348
    }
1296
1349
    else
1297
1350
    {
1298
1351
      protocol->store_null();
1299
1352
    }
 
1353
    protocol->store(mi->ssl_verify_server_cert? "Yes":"No", &my_charset_bin);
1300
1354
 
1301
1355
    // Last_IO_Errno
1302
1356
    protocol->store(mi->last_error().number);
1310
1364
    pthread_mutex_unlock(&mi->rli.data_lock);
1311
1365
    pthread_mutex_unlock(&mi->data_lock);
1312
1366
 
1313
 
    if (my_net_write(&thd->net, (unsigned char*) thd->packet.ptr(), packet->length()))
1314
 
      return(true);
 
1367
    if (my_net_write(&thd->net, (uchar*) thd->packet.ptr(), packet->length()))
 
1368
      DBUG_RETURN(TRUE);
1315
1369
  }
1316
1370
  my_eof(thd);
1317
 
  return(false);
 
1371
  DBUG_RETURN(FALSE);
1318
1372
}
1319
1373
 
1320
1374
 
1321
1375
void set_slave_thread_options(THD* thd)
1322
1376
{
 
1377
  DBUG_ENTER("set_slave_thread_options");
1323
1378
  /*
1324
1379
     It's nonsense to constrain the slave threads with max_join_size; if a
1325
1380
     query succeeded on master, we HAVE to execute it. So set
1329
1384
     when max_join_size is 4G, OPTION_BIG_SELECTS is automatically set, but
1330
1385
     only for client threads.
1331
1386
  */
1332
 
  uint64_t options= thd->options | OPTION_BIG_SELECTS;
 
1387
  ulonglong options= thd->options | OPTION_BIG_SELECTS;
1333
1388
  if (opt_log_slave_updates)
1334
1389
    options|= OPTION_BIN_LOG;
1335
1390
  else
1336
1391
    options&= ~OPTION_BIN_LOG;
1337
1392
  thd->options= options;
1338
1393
  thd->variables.completion_type= 0;
1339
 
  return;
 
1394
  DBUG_VOID_RETURN;
 
1395
}
 
1396
 
 
1397
void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli)
 
1398
{
 
1399
  DBUG_ENTER("set_slave_thread_default_charset");
 
1400
 
 
1401
  thd->variables.character_set_client=
 
1402
    global_system_variables.character_set_client;
 
1403
  thd->variables.collation_connection=
 
1404
    global_system_variables.collation_connection;
 
1405
  thd->variables.collation_server=
 
1406
    global_system_variables.collation_server;
 
1407
  thd->update_charset();
 
1408
 
 
1409
  /*
 
1410
    We use a const cast here since the conceptual (and externally
 
1411
    visible) behavior of the function is to set the default charset of
 
1412
    the thread.  That the cache has to be invalidated is a secondary
 
1413
    effect.
 
1414
   */
 
1415
  const_cast<Relay_log_info*>(rli)->cached_charset_invalidate();
 
1416
  DBUG_VOID_RETURN;
1340
1417
}
1341
1418
 
1342
1419
/*
1343
1420
  init_slave_thread()
1344
1421
*/
1345
1422
 
1346
 
static int32_t init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
 
1423
static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
1347
1424
{
1348
 
  int32_t simulate_error= 0;
 
1425
  DBUG_ENTER("init_slave_thread");
 
1426
#if !defined(DBUG_OFF)
 
1427
  int simulate_error= 0;
 
1428
#endif
1349
1429
  thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
1350
1430
    SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO;
1351
1431
  thd->security_ctx->skip_grants();
1358
1438
  thd->variables.max_allowed_packet= global_system_variables.max_allowed_packet
1359
1439
    + MAX_LOG_EVENT_HEADER;  /* note, incr over the global not session var */
1360
1440
  thd->slave_thread = 1;
 
1441
  thd->enable_slow_log= opt_log_slow_slave_statements;
1361
1442
  set_slave_thread_options(thd);
1362
1443
  thd->client_capabilities = CLIENT_LOCAL_FILES;
1363
1444
  pthread_mutex_lock(&LOCK_thread_count);
1364
1445
  thd->thread_id= thd->variables.pseudo_thread_id= thread_id++;
1365
1446
  pthread_mutex_unlock(&LOCK_thread_count);
1366
1447
 
1367
 
 simulate_error|= (1 << SLAVE_THD_IO);
1368
 
 simulate_error|= (1 << SLAVE_THD_SQL);
 
1448
  DBUG_EXECUTE_IF("simulate_io_slave_error_on_init",
 
1449
                  simulate_error|= (1 << SLAVE_THD_IO););
 
1450
  DBUG_EXECUTE_IF("simulate_sql_slave_error_on_init",
 
1451
                  simulate_error|= (1 << SLAVE_THD_SQL););
 
1452
#if !defined(DBUG_OFF)
1369
1453
  if (init_thr_lock() || thd->store_globals() || simulate_error & (1<< thd_type))
 
1454
#else
 
1455
  if (init_thr_lock() || thd->store_globals())
 
1456
#endif
1370
1457
  {
1371
1458
    thd->cleanup();
1372
 
    return(-1);
 
1459
    DBUG_RETURN(-1);
1373
1460
  }
1374
1461
  lex_start(thd);
1375
1462
 
1376
1463
  if (thd_type == SLAVE_THD_SQL)
1377
 
    thd->set_proc_info("Waiting for the next event in relay log");
 
1464
    thd_proc_info(thd, "Waiting for the next event in relay log");
1378
1465
  else
1379
 
    thd->set_proc_info("Waiting for master update");
 
1466
    thd_proc_info(thd, "Waiting for master update");
1380
1467
  thd->version=refresh_version;
1381
1468
  thd->set_time();
1382
 
  return(0);
 
1469
  DBUG_RETURN(0);
1383
1470
}
1384
1471
 
1385
1472
 
1386
 
static int32_t safe_sleep(THD* thd, int32_t sec, CHECK_KILLED_FUNC thread_killed,
 
1473
static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed,
1387
1474
                      void* thread_killed_arg)
1388
1475
{
1389
 
  int32_t nap_time;
 
1476
  int nap_time;
1390
1477
  thr_alarm_t alarmed;
 
1478
  DBUG_ENTER("safe_sleep");
1391
1479
 
1392
1480
  thr_alarm_init(&alarmed);
1393
1481
  time_t start_time= my_time(0);
1394
1482
  time_t end_time= start_time+sec;
1395
1483
 
1396
 
  while ((nap_time= (int32_t) (end_time - start_time)) > 0)
 
1484
  while ((nap_time= (int) (end_time - start_time)) > 0)
1397
1485
  {
1398
1486
    ALARM alarm_buff;
1399
1487
    /*
1406
1494
    thr_end_alarm(&alarmed);
1407
1495
 
1408
1496
    if ((*thread_killed)(thd,thread_killed_arg))
1409
 
      return(1);
 
1497
      DBUG_RETURN(1);
1410
1498
    start_time= my_time(0);
1411
1499
  }
1412
 
  return(0);
 
1500
  DBUG_RETURN(0);
1413
1501
}
1414
1502
 
1415
1503
 
1416
 
static int32_t request_dump(DRIZZLE *drizzle, Master_info* mi,
 
1504
static int request_dump(MYSQL* mysql, Master_info* mi,
1417
1505
                        bool *suppress_warnings)
1418
1506
{
1419
 
  unsigned char buf[FN_REFLEN + 10];
1420
 
  int32_t len;
1421
 
  int32_t binlog_flags = 0; // for now
1422
 
  const char* logname = mi->getLogName();
 
1507
  uchar buf[FN_REFLEN + 10];
 
1508
  int len;
 
1509
  int binlog_flags = 0; // for now
 
1510
  char* logname = mi->master_log_name;
 
1511
  DBUG_ENTER("request_dump");
1423
1512
  
1424
 
  *suppress_warnings= false;
 
1513
  *suppress_warnings= FALSE;
1425
1514
 
1426
1515
  // TODO if big log files: Change next to int8store()
1427
 
  int4store(buf, (uint32_t) mi->getLogPosition());
 
1516
  int4store(buf, (ulong) mi->master_log_pos);
1428
1517
  int2store(buf + 4, binlog_flags);
1429
1518
  int4store(buf + 6, server_id);
1430
 
  len = (uint32_t) strlen(logname);
 
1519
  len = (uint) strlen(logname);
1431
1520
  memcpy(buf + 10, logname,len);
1432
 
  if (simple_command(drizzle, COM_BINLOG_DUMP, buf, len + 10, 1))
 
1521
  if (simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1))
1433
1522
  {
1434
1523
    /*
1435
1524
      Something went wrong, so we will just reconnect and retry later
1436
1525
      in the future, we should do a better error analysis, but for
1437
1526
      now we just fill up the error log :-)
1438
1527
    */
1439
 
    if (drizzle_errno(drizzle) == ER_NET_READ_INTERRUPTED)
1440
 
      *suppress_warnings= true;                 // Suppress reconnect warning
 
1528
    if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED)
 
1529
      *suppress_warnings= TRUE;                 // Suppress reconnect warning
1441
1530
    else
1442
 
      sql_print_error(_("Error on COM_BINLOG_DUMP: %d  %s, will retry in %d secs"),
1443
 
                      drizzle_errno(drizzle), drizzle_error(drizzle),
 
1531
      sql_print_error("Error on COM_BINLOG_DUMP: %d  %s, will retry in %d secs",
 
1532
                      mysql_errno(mysql), mysql_error(mysql),
1444
1533
                      mi->connect_retry);
1445
 
    return(1);
 
1534
    DBUG_RETURN(1);
1446
1535
  }
1447
1536
 
1448
 
  return(0);
 
1537
  DBUG_RETURN(0);
1449
1538
}
1450
1539
 
1451
1540
/*
1453
1542
 
1454
1543
  SYNOPSIS
1455
1544
    read_event()
1456
 
    DRIZZLE               DRIZZLE connection
 
1545
    mysql               MySQL connection
1457
1546
    mi                  Master connection information
1458
1547
    suppress_warnings   TRUE when a normal net read timeout has caused us to
1459
1548
                        try a reconnect.  We do not want to print anything to
1465
1554
    number              Length of packet
1466
1555
*/
1467
1556
 
1468
 
static uint32_t read_event(DRIZZLE *drizzle,
1469
 
                        Master_info *mi __attribute__((unused)),
1470
 
                        bool* suppress_warnings)
 
1557
static ulong read_event(MYSQL* mysql, Master_info *mi, bool* suppress_warnings)
1471
1558
{
1472
 
  uint32_t len;
 
1559
  ulong len;
 
1560
  DBUG_ENTER("read_event");
1473
1561
 
1474
 
  *suppress_warnings= false;
 
1562
  *suppress_warnings= FALSE;
1475
1563
  /*
1476
1564
    my_real_read() will time us out
1477
1565
    We check if we were told to die, and if not, try reading again
1478
1566
  */
 
1567
#ifndef DBUG_OFF
1479
1568
  if (disconnect_slave_event_count && !(mi->events_till_disconnect--))
1480
 
    return(packet_error);
 
1569
    DBUG_RETURN(packet_error);
 
1570
#endif
1481
1571
 
1482
 
  len = cli_safe_read(drizzle);
1483
 
  if (len == packet_error || (int32_t) len < 1)
 
1572
  len = cli_safe_read(mysql);
 
1573
  if (len == packet_error || (long) len < 1)
1484
1574
  {
1485
 
    if (drizzle_errno(drizzle) == ER_NET_READ_INTERRUPTED)
 
1575
    if (mysql_errno(mysql) == ER_NET_READ_INTERRUPTED)
1486
1576
    {
1487
1577
      /*
1488
1578
        We are trying a normal reconnect after a read timeout;
1489
1579
        we suppress prints to .err file as long as the reconnect
1490
1580
        happens without problems
1491
1581
      */
1492
 
      *suppress_warnings= true;
 
1582
      *suppress_warnings= TRUE;
1493
1583
    }
1494
1584
    else
1495
 
      sql_print_error(_("Error reading packet from server: %s ( server_errno=%d)"),
1496
 
                      drizzle_error(drizzle), drizzle_errno(drizzle));
1497
 
    return(packet_error);
 
1585
      sql_print_error("Error reading packet from server: %s ( server_errno=%d)",
 
1586
                      mysql_error(mysql), mysql_errno(mysql));
 
1587
    DBUG_RETURN(packet_error);
1498
1588
  }
1499
1589
 
1500
1590
  /* Check if eof packet */
1501
 
  if (len < 8 && drizzle->net.read_pos[0] == 254)
 
1591
  if (len < 8 && mysql->net.read_pos[0] == 254)
1502
1592
  {
1503
 
    sql_print_information(_("Slave: received end packet from server, apparent "
1504
 
                            "master shutdown: %s"),
1505
 
                     drizzle_error(drizzle));
1506
 
     return(packet_error);
 
1593
    sql_print_information("Slave: received end packet from server, apparent "
 
1594
                          "master shutdown: %s",
 
1595
                     mysql_error(mysql));
 
1596
     DBUG_RETURN(packet_error);
1507
1597
  }
1508
1598
 
1509
 
  return(len - 1);
 
1599
  DBUG_PRINT("exit", ("len: %lu  net->read_pos[4]: %d",
 
1600
                      len, mysql->net.read_pos[4]));
 
1601
  DBUG_RETURN(len - 1);
1510
1602
}
1511
1603
 
1512
1604
 
1513
 
int32_t check_expected_error(THD* thd __attribute__((unused)),
1514
 
                         Relay_log_info const *rli __attribute__((unused)),
1515
 
                         int32_t expected_error)
 
1605
int check_expected_error(THD* thd, Relay_log_info const *rli,
 
1606
                         int expected_error)
1516
1607
{
 
1608
  DBUG_ENTER("check_expected_error");
 
1609
 
1517
1610
  switch (expected_error) {
1518
1611
  case ER_NET_READ_ERROR:
1519
1612
  case ER_NET_ERROR_ON_WRITE:
1520
1613
  case ER_QUERY_INTERRUPTED:
1521
1614
  case ER_SERVER_SHUTDOWN:
1522
1615
  case ER_NEW_ABORTING_CONNECTION:
1523
 
    return(1);
 
1616
    DBUG_RETURN(1);
1524
1617
  default:
1525
 
    return(0);
 
1618
    DBUG_RETURN(0);
1526
1619
  }
1527
1620
}
1528
1621
 
1534
1627
  that the error is temporary by pushing a warning with the error code
1535
1628
  ER_GET_TEMPORARY_ERRMSG, if the originating error is temporary.
1536
1629
*/
1537
 
static int32_t has_temporary_error(THD *thd)
 
1630
static int has_temporary_error(THD *thd)
1538
1631
{
 
1632
  DBUG_ENTER("has_temporary_error");
 
1633
 
1539
1634
  if (thd->is_fatal_error)
1540
 
    return(0);
 
1635
    DBUG_RETURN(0);
1541
1636
 
1542
 
  if (thd->main_da.is_error())
1543
 
  {
1544
 
    thd->clear_error();
1545
 
    my_error(ER_LOCK_DEADLOCK, MYF(0));
1546
 
  }
 
1637
  DBUG_EXECUTE_IF("all_errors_are_temporary_errors",
 
1638
                  if (thd->main_da.is_error())
 
1639
                  {
 
1640
                    thd->clear_error();
 
1641
                    my_error(ER_LOCK_DEADLOCK, MYF(0));
 
1642
                  });
1547
1643
 
1548
1644
  /*
1549
1645
    If there is no message in THD, we can't say if it's a temporary
1551
1647
    which sets no message. Return FALSE.
1552
1648
  */
1553
1649
  if (!thd->is_error())
1554
 
    return(0);
 
1650
    DBUG_RETURN(0);
1555
1651
 
1556
1652
  /*
1557
1653
    Temporary error codes:
1560
1656
  */
1561
1657
  if (thd->main_da.sql_errno() == ER_LOCK_DEADLOCK ||
1562
1658
      thd->main_da.sql_errno() == ER_LOCK_WAIT_TIMEOUT)
1563
 
    return(1);
 
1659
    DBUG_RETURN(1);
1564
1660
 
1565
 
  return(0);
 
1661
  DBUG_RETURN(0);
1566
1662
}
1567
1663
 
1568
1664
 
1592
1688
  @retval 2 No error calling ev->apply_event(), but error calling
1593
1689
  ev->update_pos().
1594
1690
*/
1595
 
int32_t apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli,
 
1691
int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli,
1596
1692
                               bool skip)
1597
1693
{
1598
 
  int32_t exec_res= 0;
 
1694
  int exec_res= 0;
 
1695
 
 
1696
  DBUG_ENTER("apply_event_and_update_pos");
 
1697
 
 
1698
  DBUG_PRINT("exec_event",("%s(type_code: %d; server_id: %d)",
 
1699
                           ev->get_type_str(), ev->get_type_code(),
 
1700
                           ev->server_id));
 
1701
  DBUG_PRINT("info", ("thd->options: %s%s; rli->last_event_start_time: %lu",
 
1702
                      FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT),
 
1703
                      FLAGSTR(thd->options, OPTION_BEGIN),
 
1704
                      rli->last_event_start_time));
1599
1705
 
1600
1706
  /*
1601
1707
    Execute the event to change the database and update the binary
1630
1736
 
1631
1737
  if (skip)
1632
1738
  {
1633
 
    int32_t reason= ev->shall_skip(rli);
 
1739
    int reason= ev->shall_skip(rli);
1634
1740
    if (reason == Log_event::EVENT_SKIP_COUNT)
1635
1741
      --rli->slave_skip_counter;
1636
1742
    pthread_mutex_unlock(&rli->data_lock);
1637
1743
    if (reason == Log_event::EVENT_SKIP_NOT)
1638
1744
      exec_res= ev->apply_event(rli);
 
1745
#ifndef DBUG_OFF
 
1746
    /*
 
1747
      This only prints information to the debug trace.
 
1748
 
 
1749
      TODO: Print an informational message to the error log?
 
1750
    */
 
1751
    static const char *const explain[] = {
 
1752
      // EVENT_SKIP_NOT,
 
1753
      "not skipped",
 
1754
      // EVENT_SKIP_IGNORE,
 
1755
      "skipped because event should be ignored",
 
1756
      // EVENT_SKIP_COUNT
 
1757
      "skipped because event skip counter was non-zero"
 
1758
    };
 
1759
    DBUG_PRINT("info", ("OPTION_BEGIN: %d; IN_STMT: %d",
 
1760
                        thd->options & OPTION_BEGIN ? 1 : 0,
 
1761
                        rli->get_flag(Relay_log_info::IN_STMT)));
 
1762
    DBUG_PRINT("skip_event", ("%s event was %s",
 
1763
                              ev->get_type_str(), explain[reason]));
 
1764
#endif
1639
1765
  }
1640
1766
  else
1641
1767
    exec_res= ev->apply_event(rli);
1642
1768
 
 
1769
  DBUG_PRINT("info", ("apply_event error = %d", exec_res));
1643
1770
  if (exec_res == 0)
1644
1771
  {
1645
 
    int32_t error= ev->update_pos(rli);
 
1772
    int error= ev->update_pos(rli);
 
1773
#ifdef HAVE_purify
 
1774
    if (!rli->is_fake)
 
1775
#endif
 
1776
    {
 
1777
#ifndef DBUG_OFF
 
1778
      char buf[22];
 
1779
#endif
 
1780
      DBUG_PRINT("info", ("update_pos error = %d", error));
 
1781
      DBUG_PRINT("info", ("group %s %s",
 
1782
                          llstr(rli->group_relay_log_pos, buf),
 
1783
                          rli->group_relay_log_name));
 
1784
      DBUG_PRINT("info", ("event %s %s",
 
1785
                          llstr(rli->event_relay_log_pos, buf),
 
1786
                          rli->event_relay_log_name));
 
1787
    }
1646
1788
    /*
1647
1789
      The update should not fail, so print an error message and
1648
1790
      return an error code.
1654
1796
    {
1655
1797
      char buf[22];
1656
1798
      rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR,
1657
 
                  _("It was not possible to update the positions"
 
1799
                  "It was not possible to update the positions"
1658
1800
                  " of the relay log information: the slave may"
1659
1801
                  " be in an inconsistent state."
1660
 
                  " Stopped in %s position %s"),
1661
 
                  rli->group_relay_log_name.c_str(),
 
1802
                  " Stopped in %s position %s",
 
1803
                  rli->group_relay_log_name,
1662
1804
                  llstr(rli->group_relay_log_pos, buf));
1663
 
      return(2);
 
1805
      DBUG_RETURN(2);
1664
1806
    }
1665
1807
  }
1666
1808
 
1667
 
  return(exec_res ? 1 : 0);
 
1809
  DBUG_RETURN(exec_res ? 1 : 0);
1668
1810
}
1669
1811
 
1670
1812
 
1696
1838
 
1697
1839
  @retval 1 The event was not applied.
1698
1840
*/
1699
 
static int32_t exec_relay_log_event(THD* thd, Relay_log_info* rli)
 
1841
static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
1700
1842
{
 
1843
  DBUG_ENTER("exec_relay_log_event");
 
1844
 
1701
1845
  /*
1702
1846
     We acquire this mutex since we need it for all operations except
1703
1847
     event execution. But we will release it in places where we will
1707
1851
 
1708
1852
  Log_event * ev = next_event(rli);
1709
1853
 
1710
 
  assert(rli->sql_thd==thd);
 
1854
  DBUG_ASSERT(rli->sql_thd==thd);
1711
1855
 
1712
1856
  if (sql_slave_killed(thd,rli))
1713
1857
  {
1714
1858
    pthread_mutex_unlock(&rli->data_lock);
1715
1859
    delete ev;
1716
 
    return(1);
 
1860
    DBUG_RETURN(1);
1717
1861
  }
1718
1862
  if (ev)
1719
1863
  {
1720
 
    int32_t exec_res;
 
1864
    int exec_res;
1721
1865
 
1722
1866
    /*
1723
1867
      This tests if the position of the beginning of the current event
1729
1873
                                ev->log_pos - ev->data_written))
1730
1874
    {
1731
1875
      char buf[22];
1732
 
      sql_print_information(_("Slave SQL thread stopped because it reached its"
1733
 
                              " UNTIL position %s"),
1734
 
                            llstr(rli->until_pos(), buf));
 
1876
      sql_print_information("Slave SQL thread stopped because it reached its"
 
1877
                            " UNTIL position %s", llstr(rli->until_pos(), buf));
1735
1878
      /*
1736
1879
        Setting abort_slave flag because we do not want additional message about
1737
1880
        error in query execution to be printed.
1739
1882
      rli->abort_slave= 1;
1740
1883
      pthread_mutex_unlock(&rli->data_lock);
1741
1884
      delete ev;
1742
 
      return(1);
 
1885
      DBUG_RETURN(1);
1743
1886
    }
1744
 
    exec_res= apply_event_and_update_pos(ev, thd, rli, true);
 
1887
    exec_res= apply_event_and_update_pos(ev, thd, rli, TRUE);
1745
1888
 
1746
1889
    /*
1747
1890
      Format_description_log_event should not be deleted because it will be
1750
1893
    */
1751
1894
    if (ev->get_type_code() != FORMAT_DESCRIPTION_EVENT)
1752
1895
    {
 
1896
      DBUG_PRINT("info", ("Deleting the event after it has been executed"));
1753
1897
      delete ev;
1754
1898
    }
1755
1899
 
1758
1902
      retry.
1759
1903
    */
1760
1904
    if (exec_res == 2)
1761
 
      return(1);
 
1905
      DBUG_RETURN(1);
1762
1906
 
1763
1907
    if (slave_trans_retries)
1764
1908
    {
1765
 
      int32_t temp_err= 0;
 
1909
      int temp_err= 0;
1766
1910
      if (exec_res && (temp_err= has_temporary_error(thd)))
1767
1911
      {
1768
1912
        const char *errmsg;
1782
1926
        */
1783
1927
        if (rli->trans_retries < slave_trans_retries)
1784
1928
        {
1785
 
          if (rli->mi->init_master_info(0, 0, SLAVE_SQL))
1786
 
            sql_print_error(_("Failed to initialize the master info structure"));
 
1929
          if (init_master_info(rli->mi, 0, 0, 0, SLAVE_SQL))
 
1930
            sql_print_error("Failed to initialize the master info structure");
1787
1931
          else if (init_relay_log_pos(rli,
1788
 
                                      rli->group_relay_log_name.c_str(),
 
1932
                                      rli->group_relay_log_name,
1789
1933
                                      rli->group_relay_log_pos,
1790
1934
                                      1, &errmsg, 1))
1791
 
            sql_print_error(_("Error initializing relay log position: %s"),
 
1935
            sql_print_error("Error initializing relay log position: %s",
1792
1936
                            errmsg);
1793
1937
          else
1794
1938
          {
1795
1939
            exec_res= 0;
1796
1940
            end_trans(thd, ROLLBACK);
1797
1941
            /* chance for concurrent connection to get more locks */
1798
 
            safe_sleep(thd, cmin(rli->trans_retries, (uint32_t)MAX_SLAVE_RETRY_PAUSE),
 
1942
            safe_sleep(thd, min(rli->trans_retries, MAX_SLAVE_RETRY_PAUSE),
1799
1943
                       (CHECK_KILLED_FUNC)sql_slave_killed, (void*)rli);
1800
1944
            pthread_mutex_lock(&rli->data_lock); // because of SHOW STATUS
1801
1945
            rli->trans_retries++;
1802
1946
            rli->retried_trans++;
1803
1947
            pthread_mutex_unlock(&rli->data_lock);
 
1948
            DBUG_PRINT("info", ("Slave retries transaction "
 
1949
                                "rli->trans_retries: %lu", rli->trans_retries));
1804
1950
          }
1805
1951
        }
1806
1952
        else
1807
 
          sql_print_error(_("Slave SQL thread retried transaction %lu time(s) "
1808
 
                            "in vain, giving up. Consider raising the value of "
1809
 
                            "the slave_transaction_retries variable."),
 
1953
          sql_print_error("Slave SQL thread retried transaction %lu time(s) "
 
1954
                          "in vain, giving up. Consider raising the value of "
 
1955
                          "the slave_transaction_retries variable.",
1810
1956
                          slave_trans_retries);
1811
1957
      }
1812
 
      else if ((exec_res && !temp_err) ||
 
1958
      else if (exec_res && !temp_err ||
1813
1959
               (opt_using_transactions &&
1814
1960
                rli->group_relay_log_pos == rli->event_relay_log_pos))
1815
1961
      {
1820
1966
          non-transient error, the slave will stop with an error.
1821
1967
         */
1822
1968
        rli->trans_retries= 0; // restart from fresh
 
1969
        DBUG_PRINT("info", ("Resetting retry counter, rli->trans_retries: %lu",
 
1970
                            rli->trans_retries));
1823
1971
      }
1824
1972
    }
1825
 
    return(exec_res);
 
1973
    DBUG_RETURN(exec_res);
1826
1974
  }
1827
1975
  pthread_mutex_unlock(&rli->data_lock);
1828
1976
  rli->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_READ_FAILURE,
1829
 
              ER(ER_SLAVE_RELAY_LOG_READ_FAILURE),
1830
 
              _("Could not parse relay log event entry. The possible reasons "
1831
 
                "are: the master's binary log is corrupted (you can check this "
1832
 
                "by running 'mysqlbinlog' on the binary log), the slave's "
1833
 
                "relay log is corrupted (you can check this by running "
1834
 
                "'mysqlbinlog' on the relay log), a network problem, or a bug "
1835
 
                "in the master's or slave's DRIZZLE code. If you want to check "
1836
 
                "the master's binary log or slave's relay log, you will be "
1837
 
                "able to know their names by issuing 'SHOW SLAVE STATUS' "
1838
 
                "on this slave."));
1839
 
  return(1);
 
1977
              ER(ER_SLAVE_RELAY_LOG_READ_FAILURE), "\
 
1978
Could not parse relay log event entry. The possible reasons are: the master's \
 
1979
binary log is corrupted (you can check this by running 'mysqlbinlog' on the \
 
1980
binary log), the slave's relay log is corrupted (you can check this by running \
 
1981
'mysqlbinlog' on the relay log), a network problem, or a bug in the master's \
 
1982
or slave's MySQL code. If you want to check the master's binary log or slave's \
 
1983
relay log, you will be able to know their names by issuing 'SHOW SLAVE STATUS' \
 
1984
on this slave.\
 
1985
");
 
1986
  DBUG_RETURN(1);
1840
1987
}
1841
1988
 
1842
1989
 
1854
2001
  no messages are added to the log.
1855
2002
 
1856
2003
  @param[in]     thd                 Thread context.
1857
 
  @param[in]     DRIZZLE               DRIZZLE connection.
 
2004
  @param[in]     mysql               MySQL connection.
1858
2005
  @param[in]     mi                  Master connection information.
1859
2006
  @param[in,out] retry_count         Number of attempts to reconnect.
1860
2007
  @param[in]     suppress_warnings   TRUE when a normal net read timeout 
1866
2013
  @retval        1                   There was an error.
1867
2014
*/
1868
2015
 
1869
 
static int32_t try_to_reconnect(THD *thd, DRIZZLE *drizzle, Master_info *mi,
1870
 
                                uint32_t *retry_count, bool suppress_warnings,
1871
 
                                const char *messages[SLAVE_RECON_MSG_MAX])
 
2016
static int try_to_reconnect(THD *thd, MYSQL *mysql, Master_info *mi,
 
2017
                            uint *retry_count, bool suppress_warnings,
 
2018
                            const char *messages[SLAVE_RECON_MSG_MAX])
1872
2019
{
1873
 
  mi->slave_running= DRIZZLE_SLAVE_RUN_NOT_CONNECT;
1874
 
  thd->set_proc_info(_(messages[SLAVE_RECON_MSG_WAIT]));
1875
 
  drizzle_disconnect(drizzle);
 
2020
  mi->slave_running= MYSQL_SLAVE_RUN_NOT_CONNECT;
 
2021
  thd->proc_info= messages[SLAVE_RECON_MSG_WAIT];
 
2022
#ifdef SIGNAL_WITH_VIO_CLOSE  
 
2023
  thd->clear_active_vio();
 
2024
#endif
 
2025
  end_server(mysql);
1876
2026
  if ((*retry_count)++)
1877
2027
  {
1878
2028
    if (*retry_count > master_retry_count)
1880
2030
    safe_sleep(thd, mi->connect_retry, (CHECK_KILLED_FUNC) io_slave_killed,
1881
2031
               (void *) mi);
1882
2032
  }
1883
 
  if (check_io_slave_killed(thd, mi,
1884
 
                            _(messages[SLAVE_RECON_MSG_KILLED_WAITING])))
 
2033
  if (check_io_slave_killed(thd, mi, messages[SLAVE_RECON_MSG_KILLED_WAITING]))
1885
2034
    return 1;
1886
 
  thd->set_proc_info(_(messages[SLAVE_RECON_MSG_AFTER]));
1887
 
  if (!suppress_warnings)
 
2035
  thd->proc_info = messages[SLAVE_RECON_MSG_AFTER];
 
2036
  if (!suppress_warnings) 
1888
2037
  {
1889
2038
    char buf[256], llbuff[22];
1890
 
    snprintf(buf, sizeof(buf), _(messages[SLAVE_RECON_MSG_FAILED]),
1891
 
             IO_RPL_LOG_NAME, llstr(mi->getLogPosition(), llbuff));
1892
 
    /*
 
2039
    my_snprintf(buf, sizeof(buf), messages[SLAVE_RECON_MSG_FAILED], 
 
2040
                IO_RPL_LOG_NAME, llstr(mi->master_log_pos, llbuff));
 
2041
    /* 
1893
2042
      Raise a warining during registering on master/requesting dump.
1894
2043
      Log a message reading event.
1895
2044
    */
1896
 
    if (_(messages[SLAVE_RECON_MSG_COMMAND])[0])
 
2045
    if (messages[SLAVE_RECON_MSG_COMMAND][0])
1897
2046
    {
1898
2047
      mi->report(WARNING_LEVEL, ER_SLAVE_MASTER_COM_FAILURE,
1899
 
                 ER(ER_SLAVE_MASTER_COM_FAILURE),
1900
 
                 _(messages[SLAVE_RECON_MSG_COMMAND]), buf);
 
2048
                 ER(ER_SLAVE_MASTER_COM_FAILURE), 
 
2049
                 messages[SLAVE_RECON_MSG_COMMAND], buf);
1901
2050
    }
1902
2051
    else
1903
2052
    {
1904
2053
      sql_print_information(buf);
1905
2054
    }
1906
2055
  }
1907
 
  if (safe_reconnect(thd, drizzle, mi, 1) || io_slave_killed(thd, mi))
 
2056
  if (safe_reconnect(thd, mysql, mi, 1) || io_slave_killed(thd, mi))
1908
2057
  {
1909
2058
    if (global_system_variables.log_warnings)
1910
 
      sql_print_information(_(messages[SLAVE_RECON_MSG_KILLED_AFTER]));
 
2059
      sql_print_information(messages[SLAVE_RECON_MSG_KILLED_AFTER]);
1911
2060
    return 1;
1912
2061
  }
1913
2062
  return 0;
1919
2068
pthread_handler_t handle_slave_io(void *arg)
1920
2069
{
1921
2070
  THD *thd; // needs to be first for thread_stack
1922
 
  DRIZZLE *drizzle;
 
2071
  MYSQL *mysql;
1923
2072
  Master_info *mi = (Master_info*)arg;
1924
2073
  Relay_log_info *rli= &mi->rli;
1925
2074
  char llbuff[22];
1926
 
  uint32_t retry_count;
 
2075
  uint retry_count;
1927
2076
  bool suppress_warnings;
1928
 
  uint32_t retry_count_reg= 0, retry_count_dump= 0, retry_count_event= 0;
 
2077
#ifndef DBUG_OFF
 
2078
  uint retry_count_reg= 0, retry_count_dump= 0, retry_count_event= 0;
 
2079
#endif
 
2080
  // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
1929
2081
  my_thread_init();
 
2082
  DBUG_ENTER("handle_slave_io");
1930
2083
 
1931
 
  assert(mi->inited);
1932
 
  drizzle= NULL ;
 
2084
  DBUG_ASSERT(mi->inited);
 
2085
  mysql= NULL ;
1933
2086
  retry_count= 0;
1934
2087
 
1935
2088
  pthread_mutex_lock(&mi->run_lock);
1936
2089
  /* Inform waiting threads that slave has started */
1937
2090
  mi->slave_run_id++;
1938
2091
 
 
2092
#ifndef DBUG_OFF
1939
2093
  mi->events_till_disconnect = disconnect_slave_event_count;
 
2094
#endif
1940
2095
 
1941
 
  thd= new THD;
 
2096
  thd= new THD; // note that contructor of THD uses DBUG_ !
1942
2097
  THD_CHECK_SENTRY(thd);
1943
2098
  mi->io_thd = thd;
1944
2099
 
1948
2103
  {
1949
2104
    pthread_cond_broadcast(&mi->start_cond);
1950
2105
    pthread_mutex_unlock(&mi->run_lock);
1951
 
    sql_print_error(_("Failed during slave I/O thread initialization"));
 
2106
    sql_print_error("Failed during slave I/O thread initialization");
1952
2107
    goto err;
1953
2108
  }
1954
2109
  pthread_mutex_lock(&LOCK_thread_count);
1959
2114
  pthread_mutex_unlock(&mi->run_lock);
1960
2115
  pthread_cond_broadcast(&mi->start_cond);
1961
2116
 
1962
 
  if (!(mi->drizzle= drizzle = drizzle_create(NULL)))
 
2117
  DBUG_PRINT("master_info",("log_file_name: '%s'  position: %s",
 
2118
                            mi->master_log_name,
 
2119
                            llstr(mi->master_log_pos,llbuff)));
 
2120
 
 
2121
  if (!(mi->mysql = mysql = mysql_init(NULL)))
1963
2122
  {
1964
2123
    mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
1965
 
               ER(ER_SLAVE_FATAL_ERROR), _("error in drizzle_create()"));
 
2124
               ER(ER_SLAVE_FATAL_ERROR), "error in mysql_init()");
1966
2125
    goto err;
1967
2126
  }
1968
2127
 
1969
 
  thd->set_proc_info("Connecting to master");
 
2128
  thd_proc_info(thd, "Connecting to master");
1970
2129
  // we can get killed during safe_connect
1971
 
  if (!safe_connect(thd, drizzle, mi))
 
2130
  if (!safe_connect(thd, mysql, mi))
1972
2131
  {
1973
 
    sql_print_information(_("Slave I/O thread: connected to master '%s@%s:%d',"
1974
 
                            "replication started in log '%s' at position %s"),
1975
 
                          mi->getUsername(), mi->getHostname(), mi->getPort(),
 
2132
    sql_print_information("Slave I/O thread: connected to master '%s@%s:%d',"
 
2133
                          "replication started in log '%s' at position %s",
 
2134
                          mi->user, mi->host, mi->port,
1976
2135
                          IO_RPL_LOG_NAME,
1977
 
                          llstr(mi->getLogPosition(), llbuff));
 
2136
                          llstr(mi->master_log_pos,llbuff));
1978
2137
  /*
1979
2138
    Adding MAX_LOG_EVENT_HEADER_LEN to the max_packet_size on the I/O
1980
2139
    thread, since a replication event can become this much larger than
1981
2140
    the corresponding packet (query) sent from client to master.
1982
2141
  */
1983
 
    drizzle->net.max_packet_size= thd->net.max_packet_size+= MAX_LOG_EVENT_HEADER;
 
2142
    mysql->net.max_packet_size= thd->net.max_packet_size+= MAX_LOG_EVENT_HEADER;
1984
2143
  }
1985
2144
  else
1986
2145
  {
1987
 
    sql_print_information(_("Slave I/O thread killed while connecting to master"));
 
2146
    sql_print_information("Slave I/O thread killed while connecting to master");
1988
2147
    goto err;
1989
2148
  }
1990
2149
 
1991
2150
connected:
1992
2151
 
1993
2152
  // TODO: the assignment below should be under mutex (5.0)
1994
 
  mi->slave_running= DRIZZLE_SLAVE_RUN_CONNECT;
1995
 
  thd->slave_net = &drizzle->net;
1996
 
  thd->set_proc_info("Checking master version");
1997
 
  if (get_master_version_and_clock(drizzle, mi))
 
2153
  mi->slave_running= MYSQL_SLAVE_RUN_CONNECT;
 
2154
  thd->slave_net = &mysql->net;
 
2155
  thd_proc_info(thd, "Checking master version");
 
2156
  if (get_master_version_and_clock(mysql, mi))
1998
2157
    goto err;
1999
 
  
 
2158
 
2000
2159
  if (mi->rli.relay_log.description_event_for_queue->binlog_version > 1)
2001
2160
  {
2002
2161
    /*
2003
2162
      Register ourselves with the master.
2004
2163
    */
2005
 
    thd->set_proc_info("Registering slave on master");
2006
 
    if (register_slave_on_master(drizzle, mi, &suppress_warnings))
 
2164
    thd_proc_info(thd, "Registering slave on master");
 
2165
    if (register_slave_on_master(mysql, mi, &suppress_warnings))
2007
2166
    {
2008
2167
      if (!check_io_slave_killed(thd, mi, "Slave I/O thread killed "
2009
 
                                 "while registering slave on master"))
 
2168
                                "while registering slave on master"))
2010
2169
      {
2011
 
        sql_print_error(_("Slave I/O thread couldn't register on master"));
2012
 
        if (try_to_reconnect(thd, drizzle, mi, &retry_count, suppress_warnings,
 
2170
        sql_print_error("Slave I/O thread couldn't register on master");
 
2171
        if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2013
2172
                             reconnect_messages[SLAVE_RECON_ACT_REG]))
2014
2173
          goto err;
2015
2174
      }
2017
2176
        goto err;
2018
2177
      goto connected;
2019
2178
    }
2020
 
    if (!retry_count_reg)
2021
 
    {
2022
 
      retry_count_reg++;
2023
 
      sql_print_information(_("Forcing to reconnect slave I/O thread"));
2024
 
      if (try_to_reconnect(thd, drizzle, mi, &retry_count, suppress_warnings,
2025
 
                           reconnect_messages[SLAVE_RECON_ACT_REG]))
2026
 
        goto err;
2027
 
      goto connected;
2028
 
    }
 
2179
    DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_REG", 
 
2180
      if (!retry_count_reg)
 
2181
      {
 
2182
        retry_count_reg++;
 
2183
        sql_print_information("Forcing to reconnect slave I/O thread");
 
2184
        if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
 
2185
                             reconnect_messages[SLAVE_RECON_ACT_REG]))
 
2186
          goto err;
 
2187
        goto connected;
 
2188
      });
2029
2189
  }
2030
2190
 
 
2191
  DBUG_PRINT("info",("Starting reading binary log from master"));
2031
2192
  while (!io_slave_killed(thd,mi))
2032
2193
  {
2033
 
    thd->set_proc_info("Requesting binlog dump");
2034
 
    if (request_dump(drizzle, mi, &suppress_warnings))
2035
 
    {
2036
 
      sql_print_error(_("Failed on request_dump()"));
2037
 
      if (check_io_slave_killed(thd, mi, _("Slave I/O thread killed while \
2038
 
requesting master dump")) ||
2039
 
          try_to_reconnect(thd, drizzle, mi, &retry_count, suppress_warnings,
2040
 
                           reconnect_messages[SLAVE_RECON_ACT_DUMP]))
2041
 
        goto err;
2042
 
      goto connected;
2043
 
    }
2044
 
    if (!retry_count_dump)
2045
 
    {
2046
 
      retry_count_dump++;
2047
 
      sql_print_information(_("Forcing to reconnect slave I/O thread"));
2048
 
      if (try_to_reconnect(thd, drizzle, mi, &retry_count, suppress_warnings,
2049
 
                           reconnect_messages[SLAVE_RECON_ACT_DUMP]))
2050
 
        goto err;
2051
 
      goto connected;
2052
 
    }
 
2194
    thd_proc_info(thd, "Requesting binlog dump");
 
2195
    if (request_dump(mysql, mi, &suppress_warnings))
 
2196
    {
 
2197
      sql_print_error("Failed on request_dump()");
 
2198
      if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \
 
2199
requesting master dump") ||
 
2200
          try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
 
2201
                           reconnect_messages[SLAVE_RECON_ACT_DUMP]))
 
2202
        goto err;
 
2203
      goto connected;
 
2204
    }
 
2205
    DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_DUMP", 
 
2206
      if (!retry_count_dump)
 
2207
      {
 
2208
        retry_count_dump++;
 
2209
        sql_print_information("Forcing to reconnect slave I/O thread");
 
2210
        if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
 
2211
                             reconnect_messages[SLAVE_RECON_ACT_DUMP]))
 
2212
          goto err;
 
2213
        goto connected;
 
2214
      });
2053
2215
 
2054
2216
    while (!io_slave_killed(thd,mi))
2055
2217
    {
2056
 
      uint32_t event_len;
 
2218
      ulong event_len;
2057
2219
      /*
2058
 
        We say "waiting" because read_event() will wait if there's nothing to
2059
 
        read. But if there's something to read, it will not wait. The
2060
 
        important thing is to not confuse users by saying "reading" whereas
2061
 
        we're in fact receiving nothing.
 
2220
         We say "waiting" because read_event() will wait if there's nothing to
 
2221
         read. But if there's something to read, it will not wait. The
 
2222
         important thing is to not confuse users by saying "reading" whereas
 
2223
         we're in fact receiving nothing.
2062
2224
      */
2063
 
      thd->set_proc_info(_("Waiting for master to send event"));
2064
 
      event_len= read_event(drizzle, mi, &suppress_warnings);
2065
 
      if (check_io_slave_killed(thd, mi, _("Slave I/O thread killed while "
2066
 
                                           "reading event")))
 
2225
      thd_proc_info(thd, "Waiting for master to send event");
 
2226
      event_len= read_event(mysql, mi, &suppress_warnings);
 
2227
      if (check_io_slave_killed(thd, mi, "Slave I/O thread killed while \
 
2228
reading event"))
2067
2229
        goto err;
2068
 
      if (!retry_count_event)
2069
 
      {
2070
 
        retry_count_event++;
2071
 
        sql_print_information(_("Forcing to reconnect slave I/O thread"));
2072
 
        if (try_to_reconnect(thd, drizzle, mi, &retry_count, suppress_warnings,
2073
 
                             reconnect_messages[SLAVE_RECON_ACT_EVENT]))
2074
 
          goto err;
2075
 
        goto connected;
2076
 
      }
 
2230
      DBUG_EXECUTE_IF("FORCE_SLAVE_TO_RECONNECT_EVENT",
 
2231
        if (!retry_count_event)
 
2232
        {
 
2233
          retry_count_event++;
 
2234
          sql_print_information("Forcing to reconnect slave I/O thread");
 
2235
          if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
 
2236
                               reconnect_messages[SLAVE_RECON_ACT_EVENT]))
 
2237
            goto err;
 
2238
          goto connected;
 
2239
        });
2077
2240
 
2078
2241
      if (event_len == packet_error)
2079
2242
      {
2080
 
        uint32_t drizzle_error_number= drizzle_errno(drizzle);
2081
 
        switch (drizzle_error_number) {
 
2243
        uint mysql_error_number= mysql_errno(mysql);
 
2244
        switch (mysql_error_number) {
2082
2245
        case CR_NET_PACKET_TOO_LARGE:
2083
 
          sql_print_error(_("Log entry on master is longer than "
2084
 
                            "max_allowed_packet (%ld) on "
2085
 
                            "slave. If the entry is correct, restart the "
2086
 
                            "server with a higher value of "
2087
 
                            "max_allowed_packet"),
 
2246
          sql_print_error("\
 
2247
Log entry on master is longer than max_allowed_packet (%ld) on \
 
2248
slave. If the entry is correct, restart the server with a higher value of \
 
2249
max_allowed_packet",
2088
2250
                          thd->variables.max_allowed_packet);
2089
2251
          goto err;
2090
2252
        case ER_MASTER_FATAL_ERROR_READING_BINLOG:
2091
 
          sql_print_error(ER(drizzle_error_number), drizzle_error_number,
2092
 
                          drizzle_error(drizzle));
 
2253
          sql_print_error(ER(mysql_error_number), mysql_error_number,
 
2254
                          mysql_error(mysql));
2093
2255
          goto err;
2094
2256
        case EE_OUTOFMEMORY:
2095
2257
        case ER_OUTOFMEMORY:
2096
 
          sql_print_error(
2097
 
       _("Stopping slave I/O thread due to out-of-memory error from master"));
 
2258
          sql_print_error("\
 
2259
Stopping slave I/O thread due to out-of-memory error from master");
2098
2260
          goto err;
2099
2261
        }
2100
 
        if (try_to_reconnect(thd, drizzle, mi, &retry_count, suppress_warnings,
 
2262
        if (try_to_reconnect(thd, mysql, mi, &retry_count, suppress_warnings,
2101
2263
                             reconnect_messages[SLAVE_RECON_ACT_EVENT]))
2102
2264
          goto err;
2103
2265
        goto connected;
2104
2266
      } // if (event_len == packet_error)
2105
2267
 
2106
2268
      retry_count=0;                    // ok event, reset retry counter
2107
 
      thd->set_proc_info(_("Queueing master event to the relay log"));
2108
 
      if (queue_event(mi,(const char*)drizzle->net.read_pos + 1, event_len))
 
2269
      thd_proc_info(thd, "Queueing master event to the relay log");
 
2270
      if (queue_event(mi,(const char*)mysql->net.read_pos + 1, event_len))
2109
2271
      {
2110
2272
        goto err;
2111
2273
      }
2112
 
      if (mi->flush())
 
2274
      if (flush_master_info(mi, 1))
2113
2275
      {
2114
 
        sql_print_error(_("Failed to flush master info file"));
 
2276
        sql_print_error("Failed to flush master info file");
2115
2277
        goto err;
2116
2278
      }
2117
2279
      /*
2126
2288
        for no reason, but this function will do a clean read, notice the clean
2127
2289
        value and exit immediately.
2128
2290
      */
 
2291
#ifndef DBUG_OFF
 
2292
      {
 
2293
        char llbuf1[22], llbuf2[22];
 
2294
        DBUG_PRINT("info", ("log_space_limit=%s log_space_total=%s \
 
2295
ignore_log_space_limit=%d",
 
2296
                            llstr(rli->log_space_limit,llbuf1),
 
2297
                            llstr(rli->log_space_total,llbuf2),
 
2298
                            (int) rli->ignore_log_space_limit));
 
2299
      }
 
2300
#endif
 
2301
 
2129
2302
      if (rli->log_space_limit && rli->log_space_limit <
2130
2303
          rli->log_space_total &&
2131
2304
          !rli->ignore_log_space_limit)
2132
2305
        if (wait_for_relay_log_space(rli))
2133
2306
        {
2134
 
          sql_print_error(_("Slave I/O thread aborted while waiting for "
2135
 
                            "relay log space"));
 
2307
          sql_print_error("Slave I/O thread aborted while waiting for relay \
 
2308
log space");
2136
2309
          goto err;
2137
2310
        }
2138
2311
    }
2139
2312
  }
2140
2313
 
2141
 
// error = 0;
 
2314
  // error = 0;
2142
2315
err:
2143
 
// print the current replication position
2144
 
  sql_print_information(_("Slave I/O thread exiting, read up to log '%s', "
2145
 
                          "position %s"),
2146
 
                        IO_RPL_LOG_NAME, llstr(mi->getLogPosition(), llbuff));
2147
 
  pthread_mutex_lock(&LOCK_thread_count);
 
2316
  // print the current replication position
 
2317
  sql_print_information("Slave I/O thread exiting, read up to log '%s', position %s",
 
2318
                  IO_RPL_LOG_NAME, llstr(mi->master_log_pos,llbuff));
 
2319
  VOID(pthread_mutex_lock(&LOCK_thread_count));
2148
2320
  thd->query = thd->db = 0; // extra safety
2149
2321
  thd->query_length= thd->db_length= 0;
2150
 
  pthread_mutex_unlock(&LOCK_thread_count);
2151
 
  if (drizzle)
 
2322
  VOID(pthread_mutex_unlock(&LOCK_thread_count));
 
2323
  if (mysql)
2152
2324
  {
2153
2325
    /*
2154
2326
      Here we need to clear the active VIO before closing the
2158
2330
      can be called in the middle of closing the VIO associated with
2159
2331
      the 'mysql' object, causing a crash.
2160
2332
    */
2161
 
    drizzle_close(drizzle);
2162
 
    mi->drizzle=0;
 
2333
#ifdef SIGNAL_WITH_VIO_CLOSE
 
2334
    thd->clear_active_vio();
 
2335
#endif
 
2336
    mysql_close(mysql);
 
2337
    mi->mysql=0;
2163
2338
  }
2164
2339
  write_ignored_events_info_to_relay_log(thd, mi);
2165
 
  thd->set_proc_info(_("Waiting for slave mutex on exit"));
 
2340
  thd_proc_info(thd, "Waiting for slave mutex on exit");
2166
2341
  pthread_mutex_lock(&mi->run_lock);
2167
2342
 
2168
2343
  /* Forget the relay log's format */
2169
2344
  delete mi->rli.relay_log.description_event_for_queue;
2170
2345
  mi->rli.relay_log.description_event_for_queue= 0;
2171
 
  assert(thd->net.buff != 0);
 
2346
  // TODO: make rpl_status part of Master_info
 
2347
  change_rpl_status(RPL_ACTIVE_SLAVE,RPL_IDLE_SLAVE);
 
2348
  DBUG_ASSERT(thd->net.buff != 0);
2172
2349
  net_end(&thd->net); // destructor will not free it, because net.vio is 0
2173
2350
  close_thread_tables(thd);
2174
2351
  pthread_mutex_lock(&LOCK_thread_count);
2187
2364
  pthread_mutex_unlock(&mi->run_lock);
2188
2365
  my_thread_end();
2189
2366
  pthread_exit(0);
2190
 
  return(0);                               // Can't return anything here
 
2367
  DBUG_RETURN(0);                               // Can't return anything here
2191
2368
}
2192
2369
 
2193
2370
 
2201
2378
  Relay_log_info* rli = &((Master_info*)arg)->rli;
2202
2379
  const char *errmsg;
2203
2380
 
 
2381
  // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
2204
2382
  my_thread_init();
 
2383
  DBUG_ENTER("handle_slave_sql");
2205
2384
 
2206
 
  assert(rli->inited);
 
2385
  DBUG_ASSERT(rli->inited);
2207
2386
  pthread_mutex_lock(&rli->run_lock);
2208
 
  assert(!rli->slave_running);
 
2387
  DBUG_ASSERT(!rli->slave_running);
2209
2388
  errmsg= 0;
 
2389
#ifndef DBUG_OFF
2210
2390
  rli->events_till_abort = abort_slave_event_count;
 
2391
#endif
2211
2392
 
2212
 
  thd = new THD;
 
2393
  thd = new THD; // note that contructor of THD uses DBUG_ !
2213
2394
  thd->thread_stack = (char*)&thd; // remember where our stack is
2214
2395
  rli->sql_thd= thd;
2215
2396
  
2226
2407
    */
2227
2408
    pthread_cond_broadcast(&rli->start_cond);
2228
2409
    pthread_mutex_unlock(&rli->run_lock);
2229
 
    sql_print_error(_("Failed during slave thread initialization"));
 
2410
    sql_print_error("Failed during slave thread initialization");
2230
2411
    goto err;
2231
2412
  }
2232
2413
  thd->init_for_queries();
2263
2444
  rli->ignore_log_space_limit= 0;
2264
2445
  pthread_mutex_unlock(&rli->log_space_lock);
2265
2446
  rli->trans_retries= 0; // start from "no error"
 
2447
  DBUG_PRINT("info", ("rli->trans_retries: %lu", rli->trans_retries));
2266
2448
 
2267
2449
  if (init_relay_log_pos(rli,
2268
 
                         rli->group_relay_log_name.c_str(),
 
2450
                         rli->group_relay_log_name,
2269
2451
                         rli->group_relay_log_pos,
2270
2452
                         1 /*need data lock*/, &errmsg,
2271
2453
                         1 /*look for a description_event*/))
2272
2454
  {
2273
 
    sql_print_error(_("Error initializing relay log position: %s"),
 
2455
    sql_print_error("Error initializing relay log position: %s",
2274
2456
                    errmsg);
2275
2457
    goto err;
2276
2458
  }
2277
2459
  THD_CHECK_SENTRY(thd);
2278
 
  assert(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
2279
 
  /*
2280
 
    Wonder if this is correct. I (Guilhem) wonder if my_b_tell() returns the
2281
 
    correct position when it's called just after my_b_seek() (the questionable
2282
 
    stuff is those "seek is done on next read" comments in the my_b_seek()
2283
 
    source code).
2284
 
    The crude reality is that this assertion randomly fails whereas
2285
 
    replication seems to work fine. And there is no easy explanation why it
2286
 
    fails (as we my_b_seek(rli->event_relay_log_pos) at the very end of
2287
 
    init_relay_log_pos() called above). Maybe the assertion would be
2288
 
    meaningful if we held rli->data_lock between the my_b_seek() and the
2289
 
    assert().
2290
 
  */
2291
 
  assert(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
2292
 
  assert(rli->sql_thd == thd);
 
2460
#ifndef DBUG_OFF
 
2461
  {
 
2462
    char llbuf1[22], llbuf2[22];
 
2463
    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
 
2464
                        llstr(my_b_tell(rli->cur_log),llbuf1),
 
2465
                        llstr(rli->event_relay_log_pos,llbuf2)));
 
2466
    DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
 
2467
    /*
 
2468
      Wonder if this is correct. I (Guilhem) wonder if my_b_tell() returns the
 
2469
      correct position when it's called just after my_b_seek() (the questionable
 
2470
      stuff is those "seek is done on next read" comments in the my_b_seek()
 
2471
      source code).
 
2472
      The crude reality is that this assertion randomly fails whereas
 
2473
      replication seems to work fine. And there is no easy explanation why it
 
2474
      fails (as we my_b_seek(rli->event_relay_log_pos) at the very end of
 
2475
      init_relay_log_pos() called above). Maybe the assertion would be
 
2476
      meaningful if we held rli->data_lock between the my_b_seek() and the
 
2477
      DBUG_ASSERT().
 
2478
    */
 
2479
#ifdef SHOULD_BE_CHECKED
 
2480
    DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
 
2481
#endif
 
2482
  }
 
2483
#endif
 
2484
  DBUG_ASSERT(rli->sql_thd == thd);
2293
2485
 
 
2486
  DBUG_PRINT("master_info",("log_file_name: %s  position: %s",
 
2487
                            rli->group_master_log_name,
 
2488
                            llstr(rli->group_master_log_pos,llbuff)));
2294
2489
  if (global_system_variables.log_warnings)
2295
 
    sql_print_information(_("Slave SQL thread initialized, "
2296
 
                            "starting replication in log '%s' at "
2297
 
                            "position %s, relay log '%s' position: %s"),
2298
 
                            RPL_LOG_NAME,
2299
 
                          llstr(rli->group_master_log_pos,llbuff),
2300
 
                          rli->group_relay_log_name.c_str(),
2301
 
                          llstr(rli->group_relay_log_pos,llbuff1));
 
2490
    sql_print_information("Slave SQL thread initialized, starting replication in \
 
2491
log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
 
2492
                    llstr(rli->group_master_log_pos,llbuff),rli->group_relay_log_name,
 
2493
                    llstr(rli->group_relay_log_pos,llbuff1));
2302
2494
 
2303
2495
  /* execute init_slave variable */
2304
2496
  if (sys_init_slave.value_length)
2306
2498
    execute_init_command(thd, &sys_init_slave, &LOCK_sys_init_slave);
2307
2499
    if (thd->is_slave_error)
2308
2500
    {
2309
 
      sql_print_error(_("Slave SQL thread aborted. "
2310
 
                        "Can't execute init_slave query"));
 
2501
      sql_print_error("\
 
2502
Slave SQL thread aborted. Can't execute init_slave query");
2311
2503
      goto err;
2312
2504
    }
2313
2505
  }
2321
2513
      rli->is_until_satisfied(rli->group_master_log_pos))
2322
2514
  {
2323
2515
    char buf[22];
2324
 
    sql_print_information(_("Slave SQL thread stopped because it reached its"
2325
 
                            " UNTIL position %s"), llstr(rli->until_pos(), buf));
 
2516
    sql_print_information("Slave SQL thread stopped because it reached its"
 
2517
                          " UNTIL position %s", llstr(rli->until_pos(), buf));
2326
2518
    pthread_mutex_unlock(&rli->data_lock);
2327
2519
    goto err;
2328
2520
  }
2332
2524
 
2333
2525
  while (!sql_slave_killed(thd,rli))
2334
2526
  {
2335
 
    thd->set_proc_info(_("Reading event from the relay log"));
2336
 
    assert(rli->sql_thd == thd);
 
2527
    thd_proc_info(thd, "Reading event from the relay log");
 
2528
    DBUG_ASSERT(rli->sql_thd == thd);
2337
2529
    THD_CHECK_SENTRY(thd);
2338
2530
    if (exec_relay_log_event(thd,rli))
2339
2531
    {
 
2532
      DBUG_PRINT("info", ("exec_relay_log_event() failed"));
2340
2533
      // do not scare the user if SQL thread was simply killed or stopped
2341
2534
      if (!sql_slave_killed(thd,rli))
2342
2535
      {
2345
2538
          codes and warnings and print this to the error log as to
2346
2539
          allow the user to locate the error
2347
2540
        */
2348
 
        uint32_t const last_errno= rli->last_error().number;
 
2541
        uint32 const last_errno= rli->last_error().number;
2349
2542
 
2350
2543
        if (thd->is_error())
2351
2544
        {
2352
2545
          char const *const errmsg= thd->main_da.message();
2353
2546
 
 
2547
          DBUG_PRINT("info",
 
2548
                     ("thd->main_da.sql_errno()=%d; rli->last_error.number=%d",
 
2549
                      thd->main_da.sql_errno(), last_errno));
2354
2550
          if (last_errno == 0)
2355
2551
          {
2356
2552
            rli->report(ERROR_LEVEL, thd->main_da.sql_errno(), errmsg);
2357
2553
          }
2358
2554
          else if (last_errno != thd->main_da.sql_errno())
2359
2555
          {
2360
 
            sql_print_error(_("Slave (additional info): %s Error_code: %d"),
 
2556
            sql_print_error("Slave (additional info): %s Error_code: %d",
2361
2557
                            errmsg, thd->main_da.sql_errno());
2362
2558
          }
2363
2559
        }
2364
2560
 
2365
2561
        /* Print any warnings issued */
2366
 
        List_iterator_fast<DRIZZLE_ERROR> it(thd->warn_list);
2367
 
        DRIZZLE_ERROR *err;
 
2562
        List_iterator_fast<MYSQL_ERROR> it(thd->warn_list);
 
2563
        MYSQL_ERROR *err;
2368
2564
        /*
2369
2565
          Added controlled slave thread cancel for replication
2370
2566
          of user-defined variables.
2374
2570
        {
2375
2571
          if (err->code == ER_CANT_OPEN_LIBRARY)
2376
2572
            udf_error = true;
2377
 
          sql_print_warning(_("Slave: %s Error_code: %d"),err->msg, err->code);
 
2573
          sql_print_warning("Slave: %s Error_code: %d",err->msg, err->code);
2378
2574
        }
2379
2575
        if (udf_error)
2380
 
          sql_print_error(_("Error loading user-defined library, slave SQL "
2381
 
                            "thread aborted. Install the missing library, "
2382
 
                            "and restart the slave SQL thread with "
2383
 
                            "\"SLAVE START\". We stopped at log '%s' "
2384
 
                            "position %s"),
2385
 
                          RPL_LOG_NAME, llstr(rli->group_master_log_pos,
 
2576
          sql_print_error("Error loading user-defined library, slave SQL "
 
2577
            "thread aborted. Install the missing library, and restart the "
 
2578
            "slave SQL thread with \"SLAVE START\". We stopped at log '%s' "
 
2579
            "position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, 
2386
2580
            llbuff));
2387
2581
        else
2388
 
          sql_print_error(_("Error running query, slave SQL thread aborted. "
2389
 
                            "Fix the problem, and restart "
2390
 
                            "the slave SQL thread with \"SLAVE START\". "
2391
 
                            "We stopped at log '%s' position %s"),
2392
 
                          RPL_LOG_NAME,
2393
 
                          llstr(rli->group_master_log_pos, llbuff));
 
2582
        sql_print_error("\
 
2583
Error running query, slave SQL thread aborted. Fix the problem, and restart \
 
2584
the slave SQL thread with \"SLAVE START\". We stopped at log \
 
2585
'%s' position %s", RPL_LOG_NAME, llstr(rli->group_master_log_pos, llbuff));
2394
2586
      }
2395
2587
      goto err;
2396
2588
    }
2397
2589
  }
2398
2590
 
2399
2591
  /* Thread stopped. Print the current replication position to the log */
2400
 
  sql_print_information(_("Slave SQL thread exiting, replication stopped in "
2401
 
                          "log '%s' at position %s"),
2402
 
                        RPL_LOG_NAME,
2403
 
                        llstr(rli->group_master_log_pos,llbuff));
 
2592
  sql_print_information("Slave SQL thread exiting, replication stopped in log "
 
2593
                        "'%s' at position %s",
 
2594
                        RPL_LOG_NAME, llstr(rli->group_master_log_pos,llbuff));
2404
2595
 
2405
2596
 err:
2406
2597
 
2411
2602
    must "proactively" clear playgrounds:
2412
2603
  */
2413
2604
  rli->cleanup_context(thd, 1);
2414
 
  pthread_mutex_lock(&LOCK_thread_count);
 
2605
  VOID(pthread_mutex_lock(&LOCK_thread_count));
2415
2606
  /*
2416
2607
    Some extra safety, which should not been needed (normally, event deletion
2417
2608
    should already have done these assignments (each event which sets these
2419
2610
  */
2420
2611
  thd->query= thd->db= thd->catalog= 0;
2421
2612
  thd->query_length= thd->db_length= 0;
2422
 
  pthread_mutex_unlock(&LOCK_thread_count);
2423
 
  thd->set_proc_info("Waiting for slave mutex on exit");
 
2613
  VOID(pthread_mutex_unlock(&LOCK_thread_count));
 
2614
  thd_proc_info(thd, "Waiting for slave mutex on exit");
2424
2615
  pthread_mutex_lock(&rli->run_lock);
2425
2616
  /* We need data_lock, at least to wake up any waiting master_pos_wait() */
2426
2617
  pthread_mutex_lock(&rli->data_lock);
2427
 
  assert(rli->slave_running == 1); // tracking buffer overrun
 
2618
  DBUG_ASSERT(rli->slave_running == 1); // tracking buffer overrun
2428
2619
  /* When master_pos_wait() wakes up it will check this and terminate */
2429
2620
  rli->slave_running= 0;
2430
2621
  /* Forget the relay log's format */
2432
2623
  rli->relay_log.description_event_for_exec= 0;
2433
2624
  /* Wake up master_pos_wait() */
2434
2625
  pthread_mutex_unlock(&rli->data_lock);
 
2626
  DBUG_PRINT("info",("Signaling possibly waiting master_pos_wait() functions"));
2435
2627
  pthread_cond_broadcast(&rli->data_cond);
2436
2628
  rli->ignore_log_space_limit= 0; /* don't need any lock */
 
2629
  /* we die so won't remember charset - re-update them on next thread start */
 
2630
  rli->cached_charset_invalidate();
2437
2631
  rli->save_temporary_tables = thd->temporary_tables;
2438
2632
 
2439
2633
  /*
2441
2635
    to avoid unneeded position re-init
2442
2636
  */
2443
2637
  thd->temporary_tables = 0; // remove tempation from destructor to close them
2444
 
  assert(thd->net.buff != 0);
 
2638
  DBUG_ASSERT(thd->net.buff != 0);
2445
2639
  net_end(&thd->net); // destructor will not free it, because we are weird
2446
 
  assert(rli->sql_thd == thd);
 
2640
  DBUG_ASSERT(rli->sql_thd == thd);
2447
2641
  THD_CHECK_SENTRY(thd);
2448
2642
  rli->sql_thd= 0;
2449
2643
  pthread_mutex_lock(&LOCK_thread_count);
2460
2654
  
2461
2655
  my_thread_end();
2462
2656
  pthread_exit(0);
2463
 
  return(0);                               // Can't return anything here
 
2657
  DBUG_RETURN(0);                               // Can't return anything here
2464
2658
}
2465
2659
 
2466
2660
 
2468
2662
  process_io_create_file()
2469
2663
*/
2470
2664
 
2471
 
static int32_t process_io_create_file(Master_info* mi, Create_file_log_event* cev)
 
2665
static int process_io_create_file(Master_info* mi, Create_file_log_event* cev)
2472
2666
{
2473
 
  int32_t error = 1;
2474
 
  uint32_t num_bytes;
 
2667
  int error = 1;
 
2668
  ulong num_bytes;
2475
2669
  bool cev_not_written;
2476
2670
  THD *thd = mi->io_thd;
2477
 
  NET *net = &mi->drizzle->net;
 
2671
  NET *net = &mi->mysql->net;
 
2672
  DBUG_ENTER("process_io_create_file");
2478
2673
 
2479
2674
  if (unlikely(!cev->is_valid()))
2480
 
    return(1);
 
2675
    DBUG_RETURN(1);
2481
2676
 
2482
2677
  if (!rpl_filter->db_ok(cev->db))
2483
2678
  {
2484
2679
    skip_load_data_infile(net);
2485
 
    return(0);
 
2680
    DBUG_RETURN(0);
2486
2681
  }
2487
 
  assert(cev->inited_from_old);
 
2682
  DBUG_ASSERT(cev->inited_from_old);
2488
2683
  thd->file_id = cev->file_id = mi->file_id++;
2489
2684
  thd->server_id = cev->server_id;
2490
2685
  cev_not_written = 1;
2491
2686
 
2492
2687
  if (unlikely(net_request_file(net,cev->fname)))
2493
2688
  {
2494
 
    sql_print_error(_("Slave I/O: failed requesting download of '%s'"),
 
2689
    sql_print_error("Slave I/O: failed requesting download of '%s'",
2495
2690
                    cev->fname);
2496
2691
    goto err;
2497
2692
  }
2508
2703
    {
2509
2704
      if (unlikely((num_bytes=my_net_read(net)) == packet_error))
2510
2705
      {
2511
 
        sql_print_error(_("Network read error downloading '%s' from master"),
 
2706
        sql_print_error("Network read error downloading '%s' from master",
2512
2707
                        cev->fname);
2513
2708
        goto err;
2514
2709
      }
2515
2710
      if (unlikely(!num_bytes)) /* eof */
2516
2711
      {
2517
2712
        /* 3.23 master wants it */
2518
 
        net_write_command(net, 0, (unsigned char*) "", 0, (unsigned char*) "", 0);
 
2713
        net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0);
2519
2714
        /*
2520
2715
          If we wrote Create_file_log_event, then we need to write
2521
2716
          Execute_load_log_event. If we did not write Create_file_log_event,
2530
2725
        {
2531
2726
          mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
2532
2727
                     ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
2533
 
                     _("error writing Exec_load event to relay log"));
 
2728
                     "error writing Exec_load event to relay log");
2534
2729
          goto err;
2535
2730
        }
2536
2731
        mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total);
2544
2739
        {
2545
2740
          mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
2546
2741
                     ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
2547
 
                     _("error writing Create_file event to relay log"));
 
2742
                     "error writing Create_file event to relay log");
2548
2743
          goto err;
2549
2744
        }
2550
2745
        cev_not_written=0;
2559
2754
        {
2560
2755
          mi->report(ERROR_LEVEL, ER_SLAVE_RELAY_LOG_WRITE_FAILURE,
2561
2756
                     ER(ER_SLAVE_RELAY_LOG_WRITE_FAILURE),
2562
 
                     _("error writing Append_block event to relay log"));
 
2757
                     "error writing Append_block event to relay log");
2563
2758
          goto err;
2564
2759
        }
2565
2760
        mi->rli.relay_log.harvest_bytes_written(&mi->rli.log_space_total) ;
2568
2763
  }
2569
2764
  error=0;
2570
2765
err:
2571
 
  return(error);
 
2766
  DBUG_RETURN(error);
2572
2767
}
2573
2768
 
2574
2769
 
2594
2789
 
2595
2790
*/
2596
2791
 
2597
 
static int32_t process_io_rotate(Master_info *mi, Rotate_log_event *rev)
 
2792
static int process_io_rotate(Master_info *mi, Rotate_log_event *rev)
2598
2793
{
 
2794
  DBUG_ENTER("process_io_rotate");
2599
2795
  safe_mutex_assert_owner(&mi->data_lock);
2600
2796
 
2601
2797
  if (unlikely(!rev->is_valid()))
2602
 
    return(1);
 
2798
    DBUG_RETURN(1);
2603
2799
 
2604
2800
  /* Safe copy as 'rev' has been "sanitized" in Rotate_log_event's ctor */
2605
 
  mi->setLogName(rev->new_log_ident);
2606
 
  mi->setLogPosition(rev->pos);
 
2801
  memcpy(mi->master_log_name, rev->new_log_ident, rev->ident_len+1);
 
2802
  mi->master_log_pos= rev->pos;
 
2803
  DBUG_PRINT("info", ("master_log_pos: '%s' %lu",
 
2804
                      mi->master_log_name, (ulong) mi->master_log_pos));
 
2805
#ifndef DBUG_OFF
2607
2806
  /*
2608
2807
    If we do not do this, we will be getting the first
2609
2808
    rotate event forever, so we need to not disconnect after one.
2610
2809
  */
2611
2810
  if (disconnect_slave_event_count)
2612
2811
    mi->events_till_disconnect++;
 
2812
#endif
2613
2813
 
2614
2814
  /*
2615
2815
    If description_event_for_queue is format <4, there is conversion in the
2621
2821
  if (mi->rli.relay_log.description_event_for_queue->binlog_version >= 4)
2622
2822
  {
2623
2823
    delete mi->rli.relay_log.description_event_for_queue;
2624
 
    /* start from format 3 (DRIZZLE 4.0) again */
 
2824
    /* start from format 3 (MySQL 4.0) again */
2625
2825
    mi->rli.relay_log.description_event_for_queue= new
2626
2826
      Format_description_log_event(3);
2627
2827
  }
2630
2830
    start or mysqlbinlog)
2631
2831
  */
2632
2832
  rotate_relay_log(mi); /* will take the right mutexes */
2633
 
  return(0);
 
2833
  DBUG_RETURN(0);
2634
2834
}
2635
2835
 
2636
2836
/*
2637
2837
  Reads a 3.23 event and converts it to the slave's format. This code was
2638
 
  copied from DRIZZLE 4.0.
 
2838
  copied from MySQL 4.0.
2639
2839
*/
2640
 
static int32_t queue_binlog_ver_1_event(Master_info *mi, const char *buf,
2641
 
                           uint32_t event_len)
 
2840
static int queue_binlog_ver_1_event(Master_info *mi, const char *buf,
 
2841
                           ulong event_len)
2642
2842
{
2643
2843
  const char *errmsg = 0;
2644
 
  uint32_t inc_pos;
 
2844
  ulong inc_pos;
2645
2845
  bool ignore_event= 0;
2646
2846
  char *tmp_buf = 0;
2647
2847
  Relay_log_info *rli= &mi->rli;
 
2848
  DBUG_ENTER("queue_binlog_ver_1_event");
2648
2849
 
2649
2850
  /*
2650
2851
    If we get Load event, we need to pass a non-reusable buffer
2655
2856
    if (unlikely(!(tmp_buf=(char*)my_malloc(event_len+1,MYF(MY_WME)))))
2656
2857
    {
2657
2858
      mi->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
2658
 
                 ER(ER_SLAVE_FATAL_ERROR), _("Memory allocation failed"));
2659
 
      return(1);
 
2859
                 ER(ER_SLAVE_FATAL_ERROR), "Memory allocation failed");
 
2860
      DBUG_RETURN(1);
2660
2861
    }
2661
2862
    memcpy(tmp_buf,buf,event_len);
2662
2863
    /*
2680
2881
                                            mi->rli.relay_log.description_event_for_queue);
2681
2882
  if (unlikely(!ev))
2682
2883
  {
2683
 
    sql_print_error(_("Read invalid event from master: '%s', "
2684
 
                      "master could be corrupt but a more likely cause "
2685
 
                      "of this is a bug"),
 
2884
    sql_print_error("Read invalid event from master: '%s',\
 
2885
 master could be corrupt but a more likely cause of this is a bug",
2686
2886
                    errmsg);
2687
 
    free((char*) tmp_buf);
2688
 
    return(1);
 
2887
    my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR));
 
2888
    DBUG_RETURN(1);
2689
2889
  }
2690
2890
 
2691
2891
  pthread_mutex_lock(&mi->data_lock);
2692
 
  ev->log_pos= mi->getLogPosition(); /* 3.23 events don't contain log_pos */
 
2892
  ev->log_pos= mi->master_log_pos; /* 3.23 events don't contain log_pos */
2693
2893
  switch (ev->get_type_code()) {
2694
2894
  case STOP_EVENT:
2695
2895
    ignore_event= 1;
2700
2900
    {
2701
2901
      delete ev;
2702
2902
      pthread_mutex_unlock(&mi->data_lock);
2703
 
      return(1);
 
2903
      DBUG_RETURN(1);
2704
2904
    }
2705
2905
    inc_pos= 0;
2706
2906
    break;
2713
2913
    */
2714
2914
  {
2715
2915
    /* We come here when and only when tmp_buf != 0 */
2716
 
    assert(tmp_buf != 0);
 
2916
    DBUG_ASSERT(tmp_buf != 0);
2717
2917
    inc_pos=event_len;
2718
2918
    ev->log_pos+= inc_pos;
2719
 
    int32_t error = process_io_create_file(mi,(Create_file_log_event*)ev);
 
2919
    int error = process_io_create_file(mi,(Create_file_log_event*)ev);
2720
2920
    delete ev;
2721
 
    mi->incrementLogPosition(inc_pos);
 
2921
    mi->master_log_pos += inc_pos;
 
2922
    DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
2722
2923
    pthread_mutex_unlock(&mi->data_lock);
2723
 
    free((char*)tmp_buf);
2724
 
    return(error);
 
2924
    my_free((char*)tmp_buf, MYF(0));
 
2925
    DBUG_RETURN(error);
2725
2926
  }
2726
2927
  default:
2727
2928
    inc_pos= event_len;
2739
2940
    {
2740
2941
      delete ev;
2741
2942
      pthread_mutex_unlock(&mi->data_lock);
2742
 
      return(1);
 
2943
      DBUG_RETURN(1);
2743
2944
    }
2744
2945
    rli->relay_log.harvest_bytes_written(&rli->log_space_total);
2745
2946
  }
2746
2947
  delete ev;
2747
 
  mi->incrementLogPosition(inc_pos);
 
2948
  mi->master_log_pos+= inc_pos;
 
2949
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
2748
2950
  pthread_mutex_unlock(&mi->data_lock);
2749
 
  return(0);
 
2951
  DBUG_RETURN(0);
2750
2952
}
2751
2953
 
2752
2954
/*
2753
2955
  Reads a 4.0 event and converts it to the slave's format. This code was copied
2754
2956
  from queue_binlog_ver_1_event(), with some affordable simplifications.
2755
2957
*/
2756
 
static int32_t queue_binlog_ver_3_event(Master_info *mi, const char *buf,
2757
 
                           uint32_t event_len)
 
2958
static int queue_binlog_ver_3_event(Master_info *mi, const char *buf,
 
2959
                           ulong event_len)
2758
2960
{
2759
2961
  const char *errmsg = 0;
2760
 
  uint32_t inc_pos;
 
2962
  ulong inc_pos;
2761
2963
  char *tmp_buf = 0;
2762
2964
  Relay_log_info *rli= &mi->rli;
 
2965
  DBUG_ENTER("queue_binlog_ver_3_event");
2763
2966
 
2764
2967
  /* read_log_event() will adjust log_pos to be end_log_pos */
2765
2968
  Log_event *ev = Log_event::read_log_event(buf,event_len, &errmsg,
2766
2969
                                            mi->rli.relay_log.description_event_for_queue);
2767
2970
  if (unlikely(!ev))
2768
2971
  {
2769
 
    sql_print_error(_("Read invalid event from master: '%s', "
2770
 
                      "master could be corrupt but a more likely cause of "
2771
 
                      "this is a bug"),
 
2972
    sql_print_error("Read invalid event from master: '%s',\
 
2973
 master could be corrupt but a more likely cause of this is a bug",
2772
2974
                    errmsg);
2773
 
    free((char*) tmp_buf);
2774
 
    return(1);
 
2975
    my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR));
 
2976
    DBUG_RETURN(1);
2775
2977
  }
2776
2978
  pthread_mutex_lock(&mi->data_lock);
2777
2979
  switch (ev->get_type_code()) {
2782
2984
    {
2783
2985
      delete ev;
2784
2986
      pthread_mutex_unlock(&mi->data_lock);
2785
 
      return(1);
 
2987
      DBUG_RETURN(1);
2786
2988
    }
2787
2989
    inc_pos= 0;
2788
2990
    break;
2794
2996
  {
2795
2997
    delete ev;
2796
2998
    pthread_mutex_unlock(&mi->data_lock);
2797
 
    return(1);
 
2999
    DBUG_RETURN(1);
2798
3000
  }
2799
3001
  rli->relay_log.harvest_bytes_written(&rli->log_space_total);
2800
3002
  delete ev;
2801
 
  mi->incrementLogPosition(inc_pos);
 
3003
  mi->master_log_pos+= inc_pos;
2802
3004
err:
 
3005
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
2803
3006
  pthread_mutex_unlock(&mi->data_lock);
2804
 
  return(0);
 
3007
  DBUG_RETURN(0);
2805
3008
}
2806
3009
 
2807
3010
/*
2816
3019
    setup with 3.23 master or 4.0 master
2817
3020
*/
2818
3021
 
2819
 
static int32_t queue_old_event(Master_info *mi, const char *buf,
2820
 
                           uint32_t event_len)
 
3022
static int queue_old_event(Master_info *mi, const char *buf,
 
3023
                           ulong event_len)
2821
3024
{
 
3025
  DBUG_ENTER("queue_old_event");
 
3026
 
2822
3027
  switch (mi->rli.relay_log.description_event_for_queue->binlog_version)
2823
3028
  {
2824
3029
  case 1:
2825
 
      return(queue_binlog_ver_1_event(mi,buf,event_len));
 
3030
      DBUG_RETURN(queue_binlog_ver_1_event(mi,buf,event_len));
2826
3031
  case 3:
2827
 
      return(queue_binlog_ver_3_event(mi,buf,event_len));
 
3032
      DBUG_RETURN(queue_binlog_ver_3_event(mi,buf,event_len));
2828
3033
  default: /* unsupported format; eg version 2 */
2829
 
    return(1);
 
3034
    DBUG_PRINT("info",("unsupported binlog format %d in queue_old_event()",
 
3035
                       mi->rli.relay_log.description_event_for_queue->binlog_version));
 
3036
    DBUG_RETURN(1);
2830
3037
  }
2831
3038
}
2832
3039
 
2840
3047
  any >=5.0.0 format.
2841
3048
*/
2842
3049
 
2843
 
static int32_t queue_event(Master_info* mi,const char* buf, uint32_t event_len)
 
3050
static int queue_event(Master_info* mi,const char* buf, ulong event_len)
2844
3051
{
2845
 
  int32_t error= 0;
 
3052
  int error= 0;
2846
3053
  String error_msg;
2847
 
  uint32_t inc_pos= 0;
 
3054
  ulong inc_pos;
2848
3055
  Relay_log_info *rli= &mi->rli;
2849
3056
  pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
 
3057
  DBUG_ENTER("queue_event");
2850
3058
 
2851
3059
 
2852
3060
  if (mi->rli.relay_log.description_event_for_queue->binlog_version<4 &&
2853
3061
      buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT /* a way to escape */)
2854
 
    return(queue_old_event(mi,buf,event_len));
 
3062
    DBUG_RETURN(queue_old_event(mi,buf,event_len));
2855
3063
 
2856
3064
  pthread_mutex_lock(&mi->data_lock);
2857
3065
 
2916
3124
       it), i.e. has end_log_pos=0, we do not increment mi->master_log_pos
2917
3125
    */
2918
3126
    inc_pos= uint4korr(buf+LOG_POS_OFFSET) ? event_len : 0;
 
3127
    DBUG_PRINT("info",("binlog format is now %d",
 
3128
                       mi->rli.relay_log.description_event_for_queue->binlog_version));
 
3129
 
2919
3130
  }
2920
3131
  break;
2921
3132
 
2931
3142
      error= ER_SLAVE_HEARTBEAT_FAILURE;
2932
3143
      error_msg.append(STRING_WITH_LEN("inconsistent heartbeat event content;"));
2933
3144
      error_msg.append(STRING_WITH_LEN("the event's data: log_file_name "));
2934
 
      error_msg.append(hb.get_log_ident(), (uint32_t) strlen(hb.get_log_ident()));
 
3145
      error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident()));
2935
3146
      error_msg.append(STRING_WITH_LEN(" log_pos "));
2936
3147
      llstr(hb.log_pos, llbuf);
2937
3148
      error_msg.append(llbuf, strlen(llbuf));
2950
3161
 
2951
3162
       TODO: handling `when' for SHOW SLAVE STATUS' snds behind
2952
3163
    */
2953
 
    if ((mi->setLogName(hb.get_log_ident()) && mi->getLogName() != NULL)
2954
 
        || mi->getLogPosition() != hb.log_pos)
 
3164
    if ((memcmp(mi->master_log_name, hb.get_log_ident(), hb.get_ident_len())
 
3165
         && mi->master_log_name != NULL)
 
3166
        || mi->master_log_pos != hb.log_pos)
2955
3167
    {
2956
3168
      /* missed events of heartbeat from the past */
2957
3169
      error= ER_SLAVE_HEARTBEAT_FAILURE;
2958
3170
      error_msg.append(STRING_WITH_LEN("heartbeat is not compatible with local info;"));
2959
3171
      error_msg.append(STRING_WITH_LEN("the event's data: log_file_name "));
2960
 
      error_msg.append(hb.get_log_ident(), (uint32_t) strlen(hb.get_log_ident()));
 
3172
      error_msg.append(hb.get_log_ident(), (uint) strlen(hb.get_log_ident()));
2961
3173
      error_msg.append(STRING_WITH_LEN(" log_pos "));
2962
3174
      llstr(hb.log_pos, llbuf);
2963
3175
      error_msg.append(llbuf, strlen(llbuf));
3008
3220
        buf[EVENT_TYPE_OFFSET]!=ROTATE_EVENT &&
3009
3221
        buf[EVENT_TYPE_OFFSET]!=STOP_EVENT)
3010
3222
    {
3011
 
      mi->incrementLogPosition(inc_pos);
3012
 
      memcpy(rli->ign_master_log_name_end, mi->getLogName(), FN_REFLEN);
3013
 
      assert(rli->ign_master_log_name_end[0]);
3014
 
      rli->ign_master_log_pos_end= mi->getLogPosition();
 
3223
      mi->master_log_pos+= inc_pos;
 
3224
      memcpy(rli->ign_master_log_name_end, mi->master_log_name, FN_REFLEN);
 
3225
      DBUG_ASSERT(rli->ign_master_log_name_end[0]);
 
3226
      rli->ign_master_log_pos_end= mi->master_log_pos;
3015
3227
    }
3016
3228
    rli->relay_log.signal_update(); // the slave SQL thread needs to re-check
 
3229
    DBUG_PRINT("info", ("master_log_pos: %lu, event originating from the same server, ignored",
 
3230
                        (ulong) mi->master_log_pos));
3017
3231
  }
3018
3232
  else
3019
3233
  {
3020
3234
    /* write the event to the relay log */
3021
3235
    if (likely(!(rli->relay_log.appendv(buf,event_len,0))))
3022
3236
    {
3023
 
      mi->incrementLogPosition(inc_pos);
 
3237
      mi->master_log_pos+= inc_pos;
 
3238
      DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
3024
3239
      rli->relay_log.harvest_bytes_written(&rli->log_space_total);
3025
3240
    }
3026
3241
    else
3035
3250
  
3036
3251
err:
3037
3252
  pthread_mutex_unlock(&mi->data_lock);
 
3253
  DBUG_PRINT("info", ("error: %d", error));
3038
3254
  if (error)
3039
 
    mi->report(ERROR_LEVEL, error, ER(error),
 
3255
    mi->report(ERROR_LEVEL, error, ER(error), 
3040
3256
               (error == ER_SLAVE_RELAY_LOG_WRITE_FAILURE)?
3041
 
               _("could not queue event from master") :
 
3257
               "could not queue event from master" :
3042
3258
               error_msg.ptr());
3043
 
  return(error);
 
3259
  DBUG_RETURN(error);
3044
3260
}
3045
3261
 
3046
3262
 
3047
3263
void end_relay_log_info(Relay_log_info* rli)
3048
3264
{
 
3265
  DBUG_ENTER("end_relay_log_info");
 
3266
 
3049
3267
  if (!rli->inited)
3050
 
    return;
 
3268
    DBUG_VOID_RETURN;
3051
3269
  if (rli->info_fd >= 0)
3052
3270
  {
3053
3271
    end_io_cache(&rli->info_file);
3069
3287
    of slave's temp tables after shutdown.
3070
3288
  */
3071
3289
  rli->close_temporary_tables();
3072
 
  return;
 
3290
  DBUG_VOID_RETURN;
3073
3291
}
3074
3292
 
3075
3293
/*
3078
3296
  SYNPOSIS
3079
3297
    safe_connect()
3080
3298
    thd                 Thread handler for slave
3081
 
    DRIZZLE               DRIZZLE connection handle
 
3299
    mysql               MySQL connection handle
3082
3300
    mi                  Replication handle
3083
3301
 
3084
3302
  RETURN
3086
3304
    #   Error
3087
3305
*/
3088
3306
 
3089
 
static int32_t safe_connect(THD* thd, DRIZZLE *drizzle, Master_info* mi)
 
3307
static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi)
3090
3308
{
3091
 
  return(connect_to_master(thd, drizzle, mi, 0, 0));
 
3309
  DBUG_ENTER("safe_connect");
 
3310
 
 
3311
  DBUG_RETURN(connect_to_master(thd, mysql, mi, 0, 0));
3092
3312
}
3093
3313
 
3094
3314
 
3101
3321
    master_retry_count times
3102
3322
*/
3103
3323
 
3104
 
static int32_t connect_to_master(THD* thd, DRIZZLE *drizzle, Master_info* mi,
3105
 
                                 bool reconnect, bool suppress_warnings)
 
3324
static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi,
 
3325
                             bool reconnect, bool suppress_warnings)
3106
3326
{
3107
 
  int32_t slave_was_killed;
3108
 
  int32_t last_errno= -2;                           // impossible error
3109
 
  uint32_t err_count=0;
 
3327
  int slave_was_killed;
 
3328
  int last_errno= -2;                           // impossible error
 
3329
  ulong err_count=0;
3110
3330
  char llbuff[22];
 
3331
  DBUG_ENTER("connect_to_master");
3111
3332
 
 
3333
#ifndef DBUG_OFF
3112
3334
  mi->events_till_disconnect = disconnect_slave_event_count;
3113
 
  uint32_t client_flag= CLIENT_REMEMBER_OPTIONS;
 
3335
#endif
 
3336
  ulong client_flag= CLIENT_REMEMBER_OPTIONS;
3114
3337
  if (opt_slave_compressed_protocol)
3115
3338
    client_flag=CLIENT_COMPRESS;                /* We will use compression */
3116
3339
 
3117
 
  drizzle_options(drizzle, DRIZZLE_OPT_CONNECT_TIMEOUT, (char *) &slave_net_timeout);
3118
 
  drizzle_options(drizzle, DRIZZLE_OPT_READ_TIMEOUT, (char *) &slave_net_timeout);
 
3340
  mysql_options(mysql, MYSQL_OPT_CONNECT_TIMEOUT, (char *) &slave_net_timeout);
 
3341
  mysql_options(mysql, MYSQL_OPT_READ_TIMEOUT, (char *) &slave_net_timeout);
 
3342
 
 
3343
  mysql_options(mysql, MYSQL_SET_CHARSET_NAME, default_charset_info->csname);
 
3344
  /* This one is not strictly needed but we have it here for completeness */
 
3345
  mysql_options(mysql, MYSQL_SET_CHARSET_DIR, (char *) charsets_dir);
3119
3346
 
3120
3347
  while (!(slave_was_killed = io_slave_killed(thd,mi)) &&
3121
 
         (reconnect ? drizzle_reconnect(drizzle) != 0 :
3122
 
          drizzle_connect(drizzle, mi->getHostname(), mi->getUsername(), mi->getPassword(), 0,
3123
 
                             mi->getPort(), 0, client_flag) == 0))
 
3348
         (reconnect ? mysql_reconnect(mysql) != 0 :
 
3349
          mysql_real_connect(mysql, mi->host, mi->user, mi->password, 0,
 
3350
                             mi->port, 0, client_flag) == 0))
3124
3351
  {
3125
3352
    /* Don't repeat last error */
3126
 
    if ((int32_t)drizzle_errno(drizzle) != last_errno)
 
3353
    if ((int)mysql_errno(mysql) != last_errno)
3127
3354
    {
3128
 
      last_errno=drizzle_errno(drizzle);
 
3355
      last_errno=mysql_errno(mysql);
3129
3356
      suppress_warnings= 0;
3130
3357
      mi->report(ERROR_LEVEL, last_errno,
3131
 
                 _("error %s to master '%s@%s:%d'"
3132
 
                   " - retry-time: %d  retries: %u"),
3133
 
                 (reconnect ? _("reconnecting") : _("connecting")),
3134
 
                 mi->getUsername(), mi->getHostname(), mi->getPort(),
3135
 
                 mi->getConnectionRetry(), master_retry_count);
 
3358
                 "error %s to master '%s@%s:%d'"
 
3359
                 " - retry-time: %d  retries: %lu",
 
3360
                 (reconnect ? "reconnecting" : "connecting"),
 
3361
                 mi->user, mi->host, mi->port,
 
3362
                 mi->connect_retry, master_retry_count);
3136
3363
    }
3137
3364
    /*
3138
3365
      By default we try forever. The reason is that failure will trigger
3143
3370
    if (++err_count == master_retry_count)
3144
3371
    {
3145
3372
      slave_was_killed=1;
 
3373
      if (reconnect)
 
3374
        change_rpl_status(RPL_ACTIVE_SLAVE,RPL_LOST_SOLDIER);
3146
3375
      break;
3147
3376
    }
3148
3377
    safe_sleep(thd,mi->connect_retry,(CHECK_KILLED_FUNC)io_slave_killed,
3154
3383
    if (reconnect)
3155
3384
    {
3156
3385
      if (!suppress_warnings && global_system_variables.log_warnings)
3157
 
        sql_print_information(_("Slave: connected to master '%s@%s:%d', "
3158
 
                                "replication resumed in log '%s' at "
3159
 
                                "position %s"), mi->getUsername(),
3160
 
                                mi->getHostname(), mi->getPort(),
3161
 
                                IO_RPL_LOG_NAME,
3162
 
                                llstr(mi->getLogPosition(),llbuff));
3163
 
    }
 
3386
        sql_print_information("Slave: connected to master '%s@%s:%d',\
 
3387
replication resumed in log '%s' at position %s", mi->user,
 
3388
                        mi->host, mi->port,
 
3389
                        IO_RPL_LOG_NAME,
 
3390
                        llstr(mi->master_log_pos,llbuff));
 
3391
    }
 
3392
    else
 
3393
    {
 
3394
      change_rpl_status(RPL_IDLE_SLAVE,RPL_ACTIVE_SLAVE);
 
3395
      general_log_print(thd, COM_CONNECT_OUT, "%s@%s:%d",
 
3396
                        mi->user, mi->host, mi->port);
 
3397
    }
 
3398
#ifdef SIGNAL_WITH_VIO_CLOSE
 
3399
    thd->set_active_vio(mysql->net.vio);
 
3400
#endif
3164
3401
  }
3165
 
  drizzle->reconnect= 1;
3166
 
  return(slave_was_killed);
 
3402
  mysql->reconnect= 1;
 
3403
  DBUG_PRINT("exit",("slave_was_killed: %d", slave_was_killed));
 
3404
  DBUG_RETURN(slave_was_killed);
3167
3405
}
3168
3406
 
3169
3407
 
3175
3413
    master_retry_count times
3176
3414
*/
3177
3415
 
3178
 
static int32_t safe_reconnect(THD* thd, DRIZZLE *drizzle, Master_info* mi,
 
3416
static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi,
3179
3417
                          bool suppress_warnings)
3180
3418
{
3181
 
  return(connect_to_master(thd, drizzle, mi, 1, suppress_warnings));
 
3419
  DBUG_ENTER("safe_reconnect");
 
3420
  DBUG_RETURN(connect_to_master(thd, mysql, mi, 1, suppress_warnings));
3182
3421
}
3183
3422
 
3184
3423
 
3204
3443
 
3205
3444
  TODO
3206
3445
    - Change the log file information to a binary format to avoid calling
3207
 
      int64_t2str.
 
3446
      longlong2str.
3208
3447
 
3209
3448
  RETURN VALUES
3210
3449
    0   ok
3213
3452
 
3214
3453
bool flush_relay_log_info(Relay_log_info* rli)
3215
3454
{
3216
 
  bool error= 0;
 
3455
  bool error=0;
 
3456
  DBUG_ENTER("flush_relay_log_info");
3217
3457
 
3218
3458
  if (unlikely(rli->no_storage))
3219
 
    return(0);
3220
 
 
3221
 
  return(error);
 
3459
    DBUG_RETURN(0);
 
3460
 
 
3461
  IO_CACHE *file = &rli->info_file;
 
3462
  char buff[FN_REFLEN*2+22*2+4], *pos;
 
3463
 
 
3464
  my_b_seek(file, 0L);
 
3465
  pos=strmov(buff, rli->group_relay_log_name);
 
3466
  *pos++='\n';
 
3467
  pos=longlong2str(rli->group_relay_log_pos, pos, 10);
 
3468
  *pos++='\n';
 
3469
  pos=strmov(pos, rli->group_master_log_name);
 
3470
  *pos++='\n';
 
3471
  pos=longlong2str(rli->group_master_log_pos, pos, 10);
 
3472
  *pos='\n';
 
3473
  if (my_b_write(file, (uchar*) buff, (size_t) (pos-buff)+1))
 
3474
    error=1;
 
3475
  if (flush_io_cache(file))
 
3476
    error=1;
 
3477
 
 
3478
  /* Flushing the relay log is done by the slave I/O thread */
 
3479
  DBUG_RETURN(error);
3222
3480
}
3223
3481
 
3224
3482
 
3228
3486
 
3229
3487
static IO_CACHE *reopen_relay_log(Relay_log_info *rli, const char **errmsg)
3230
3488
{
3231
 
  assert(rli->cur_log != &rli->cache_buf);
3232
 
  assert(rli->cur_log_fd == -1);
 
3489
  DBUG_ENTER("reopen_relay_log");
 
3490
  DBUG_ASSERT(rli->cur_log != &rli->cache_buf);
 
3491
  DBUG_ASSERT(rli->cur_log_fd == -1);
3233
3492
 
3234
3493
  IO_CACHE *cur_log = rli->cur_log=&rli->cache_buf;
3235
 
  if ((rli->cur_log_fd=open_binlog(cur_log, rli->event_relay_log_name.c_str(), errmsg)) < 0)
3236
 
    return(0);
 
3494
  if ((rli->cur_log_fd=open_binlog(cur_log,rli->event_relay_log_name,
 
3495
                                   errmsg)) <0)
 
3496
    DBUG_RETURN(0);
3237
3497
  /*
3238
3498
    We want to start exactly where we was before:
3239
3499
    relay_log_pos       Current log pos
3240
3500
    pending             Number of bytes already processed from the event
3241
3501
  */
3242
 
  rli->event_relay_log_pos= cmax(rli->event_relay_log_pos, (uint64_t)BIN_LOG_HEADER_SIZE);
 
3502
  rli->event_relay_log_pos= max(rli->event_relay_log_pos, BIN_LOG_HEADER_SIZE);
3243
3503
  my_b_seek(cur_log,rli->event_relay_log_pos);
3244
 
  return(cur_log);
 
3504
  DBUG_RETURN(cur_log);
3245
3505
}
3246
3506
 
3247
3507
 
3252
3512
  pthread_mutex_t *log_lock = rli->relay_log.get_log_lock();
3253
3513
  const char* errmsg=0;
3254
3514
  THD* thd = rli->sql_thd;
3255
 
 
3256
 
  assert(thd != 0);
3257
 
 
 
3515
  DBUG_ENTER("next_event");
 
3516
 
 
3517
  DBUG_ASSERT(thd != 0);
 
3518
 
 
3519
#ifndef DBUG_OFF
3258
3520
  if (abort_slave_event_count && !rli->events_till_abort--)
3259
 
    return(0);
 
3521
    DBUG_RETURN(0);
 
3522
#endif
3260
3523
 
3261
3524
  /*
3262
3525
    For most operations we need to protect rli members with data_lock,
3283
3546
    bool hot_log;
3284
3547
    if ((hot_log = (cur_log != &rli->cache_buf)))
3285
3548
    {
3286
 
      assert(rli->cur_log_fd == -1); // foreign descriptor
 
3549
      DBUG_ASSERT(rli->cur_log_fd == -1); // foreign descriptor
3287
3550
      pthread_mutex_lock(log_lock);
3288
3551
 
3289
3552
      /*
3307
3570
    */
3308
3571
    if (!my_b_inited(cur_log))
3309
3572
      goto err;
3310
 
    assert(my_b_tell(cur_log) >= BIN_LOG_HEADER_SIZE);
3311
 
    assert(my_b_tell(cur_log) == rli->event_relay_log_pos);
3312
 
 
 
3573
#ifndef DBUG_OFF
 
3574
    {
 
3575
      /* This is an assertion which sometimes fails, let's try to track it */
 
3576
      char llbuf1[22], llbuf2[22];
 
3577
      DBUG_PRINT("info", ("my_b_tell(cur_log)=%s rli->event_relay_log_pos=%s",
 
3578
                          llstr(my_b_tell(cur_log),llbuf1),
 
3579
                          llstr(rli->event_relay_log_pos,llbuf2)));
 
3580
      DBUG_ASSERT(my_b_tell(cur_log) >= BIN_LOG_HEADER_SIZE);
 
3581
      DBUG_ASSERT(my_b_tell(cur_log) == rli->event_relay_log_pos);
 
3582
    }
 
3583
#endif
3313
3584
    /*
3314
3585
      Relay log is always in new format - if the master is 3.23, the
3315
3586
      I/O thread will convert the format for us.
3320
3591
      When the relay log is created when the I/O thread starts, easy: the
3321
3592
      master will send the description event and we will queue it.
3322
3593
      But if the relay log is created by new_file(): then the solution is:
3323
 
      DRIZZLE_BIN_LOG::open() will write the buffered description event.
 
3594
      MYSQL_BIN_LOG::open() will write the buffered description event.
3324
3595
    */
3325
3596
    if ((ev=Log_event::read_log_event(cur_log,0,
3326
3597
                                      rli->relay_log.description_event_for_exec)))
3327
3598
 
3328
3599
    {
3329
 
      assert(thd==rli->sql_thd);
 
3600
      DBUG_ASSERT(thd==rli->sql_thd);
3330
3601
      /*
3331
3602
        read it while we have a lock, to avoid a mutex lock in
3332
3603
        inc_event_relay_log_pos()
3334
3605
      rli->future_event_relay_log_pos= my_b_tell(cur_log);
3335
3606
      if (hot_log)
3336
3607
        pthread_mutex_unlock(log_lock);
3337
 
      return(ev);
 
3608
      DBUG_RETURN(ev);
3338
3609
    }
3339
 
    assert(thd==rli->sql_thd);
 
3610
    DBUG_ASSERT(thd==rli->sql_thd);
3340
3611
    if (opt_reckless_slave)                     // For mysql-test
3341
3612
      cur_log->error = 0;
3342
3613
    if (cur_log->error < 0)
3379
3650
        time_t save_timestamp= rli->last_master_timestamp;
3380
3651
        rli->last_master_timestamp= 0;
3381
3652
 
3382
 
        assert(rli->relay_log.get_open_count() ==
 
3653
        DBUG_ASSERT(rli->relay_log.get_open_count() ==
3383
3654
                    rli->cur_log_old_open_count);
3384
3655
 
3385
3656
        if (rli->ign_master_log_name_end[0])
3386
3657
        {
3387
3658
          /* We generate and return a Rotate, to make our positions advance */
 
3659
          DBUG_PRINT("info",("seeing an ignored end segment"));
3388
3660
          ev= new Rotate_log_event(rli->ign_master_log_name_end,
3389
3661
                                   0, rli->ign_master_log_pos_end,
3390
3662
                                   Rotate_log_event::DUP_NAME);
3397
3669
            goto err;
3398
3670
          }
3399
3671
          ev->server_id= 0; // don't be ignored by slave SQL thread
3400
 
          return(ev);
 
3672
          DBUG_RETURN(ev);
3401
3673
        }
3402
3674
 
3403
3675
        /*
3454
3726
        cases separately after doing some common initialization
3455
3727
      */
3456
3728
      end_io_cache(cur_log);
3457
 
      assert(rli->cur_log_fd >= 0);
 
3729
      DBUG_ASSERT(rli->cur_log_fd >= 0);
3458
3730
      my_close(rli->cur_log_fd, MYF(MY_WME));
3459
3731
      rli->cur_log_fd = -1;
3460
3732
 
3465
3737
          If the group's coordinates are equal to the event's coordinates
3466
3738
          (i.e. the relay log was not rotated in the middle of a group),
3467
3739
          we can purge this relay log too.
3468
 
          We do uint64_t and string comparisons, this may be slow but
 
3740
          We do ulonglong and string comparisons, this may be slow but
3469
3741
          - purging the last relay log is nice (it can save 1GB of disk), so we
3470
3742
          like to detect the case where we can do it, and given this,
3471
3743
          - I see no better detection method
3474
3746
        if (rli->relay_log.purge_first_log
3475
3747
            (rli,
3476
3748
             rli->group_relay_log_pos == rli->event_relay_log_pos
3477
 
             && !strcmp(rli->group_relay_log_name.c_str(), rli->event_relay_log_name.c_str())))
 
3749
             && !strcmp(rli->group_relay_log_name,rli->event_relay_log_name)))
3478
3750
        {
3479
3751
          errmsg = "Error purging processed logs";
3480
3752
          goto err;
3495
3767
          goto err;
3496
3768
        }
3497
3769
        rli->event_relay_log_pos = BIN_LOG_HEADER_SIZE;
3498
 
        rli->event_relay_log_name.assign(rli->linfo.log_file_name);
 
3770
        strmake(rli->event_relay_log_name,rli->linfo.log_file_name,
 
3771
                sizeof(rli->event_relay_log_name)-1);
3499
3772
        flush_relay_log_info(rli);
3500
3773
      }
3501
3774
 
3509
3782
        To guard against this, we need to have LOCK_log.
3510
3783
      */
3511
3784
 
 
3785
      DBUG_PRINT("info",("hot_log: %d",hot_log));
3512
3786
      if (!hot_log) /* if hot_log, we already have this mutex */
3513
3787
        pthread_mutex_lock(log_lock);
3514
3788
      if (rli->relay_log.is_active(rli->linfo.log_file_name))
3515
3789
      {
3516
3790
#ifdef EXTRA_DEBUG
3517
3791
        if (global_system_variables.log_warnings)
3518
 
          sql_print_information(_("next log '%s' is currently active"),
 
3792
          sql_print_information("next log '%s' is currently active",
3519
3793
                                rli->linfo.log_file_name);
3520
3794
#endif
3521
3795
        rli->cur_log= cur_log= rli->relay_log.get_log_file();
3522
3796
        rli->cur_log_old_open_count= rli->relay_log.get_open_count();
3523
 
        assert(rli->cur_log_fd == -1);
 
3797
        DBUG_ASSERT(rli->cur_log_fd == -1);
3524
3798
 
3525
3799
        /*
3526
3800
          Read pointer has to be at the start since we are the only
3545
3819
      */
3546
3820
#ifdef EXTRA_DEBUG
3547
3821
      if (global_system_variables.log_warnings)
3548
 
        sql_print_information(_("next log '%s' is not active"),
 
3822
        sql_print_information("next log '%s' is not active",
3549
3823
                              rli->linfo.log_file_name);
3550
3824
#endif
3551
3825
      // open_binlog() will check the magic header
3561
3835
      */
3562
3836
      if (hot_log)
3563
3837
        pthread_mutex_unlock(log_lock);
3564
 
      sql_print_error(_("Slave SQL thread: I/O error reading "
3565
 
                        "event(errno: %d  cur_log->error: %d)"),
 
3838
      sql_print_error("Slave SQL thread: I/O error reading \
 
3839
event(errno: %d  cur_log->error: %d)",
3566
3840
                      my_errno,cur_log->error);
3567
3841
      // set read position to the beginning of the event
3568
3842
      my_b_seek(cur_log,rli->event_relay_log_pos);
3569
3843
      /* otherwise, we have had a partial read */
3570
 
      errmsg = _("Aborting slave SQL thread because of partial event read");
 
3844
      errmsg = "Aborting slave SQL thread because of partial event read";
3571
3845
      break;                                    // To end of function
3572
3846
    }
3573
3847
  }
3574
3848
  if (!errmsg && global_system_variables.log_warnings)
3575
3849
  {
3576
 
    sql_print_information(_("Error reading relay log event: %s"),
3577
 
                          _("slave SQL thread was killed"));
3578
 
    return(0);
 
3850
    sql_print_information("Error reading relay log event: %s",
 
3851
                          "slave SQL thread was killed");
 
3852
    DBUG_RETURN(0);
3579
3853
  }
3580
3854
 
3581
3855
err:
3582
3856
  if (errmsg)
3583
 
    sql_print_error(_("Error reading relay log event: %s"), errmsg);
3584
 
  return(0);
 
3857
    sql_print_error("Error reading relay log event: %s", errmsg);
 
3858
  DBUG_RETURN(0);
3585
3859
}
3586
3860
 
3587
3861
/*
3588
3862
  Rotate a relay log (this is used only by FLUSH LOGS; the automatic rotation
3589
3863
  because of size is simpler because when we do it we already have all relevant
3590
3864
  locks; here we don't, so this function is mainly taking locks).
3591
 
  Returns nothing as we cannot catch any error (DRIZZLE_BIN_LOG::new_file()
 
3865
  Returns nothing as we cannot catch any error (MYSQL_BIN_LOG::new_file()
3592
3866
  is void).
3593
3867
*/
3594
3868
 
3595
3869
void rotate_relay_log(Master_info* mi)
3596
3870
{
 
3871
  DBUG_ENTER("rotate_relay_log");
3597
3872
  Relay_log_info* rli= &mi->rli;
3598
3873
 
3599
3874
  /* We don't lock rli->run_lock. This would lead to deadlocks. */
3605
3880
  */
3606
3881
  if (!rli->inited)
3607
3882
  {
 
3883
    DBUG_PRINT("info", ("rli->inited == 0"));
3608
3884
    goto end;
3609
3885
  }
3610
3886
 
3627
3903
  rli->relay_log.harvest_bytes_written(&rli->log_space_total);
3628
3904
end:
3629
3905
  pthread_mutex_unlock(&mi->run_lock);
3630
 
  return;
 
3906
  DBUG_VOID_RETURN;
3631
3907
}
3632
3908
 
3633
3909
 
3637
3913
   @param rli Relay_log_info which tells the master's version
3638
3914
   @param bug_id Number of the bug as found in bugs.mysql.com
3639
3915
   @param report bool report error message, default TRUE
3640
 
   @return true if master has the bug, FALSE if it does not.
 
3916
   @return TRUE if master has the bug, FALSE if it does not.
3641
3917
*/
3642
 
bool rpl_master_has_bug(Relay_log_info *rli, uint32_t bug_id, bool report)
 
3918
bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id, bool report)
3643
3919
{
3644
3920
  struct st_version_range_for_one_bug {
3645
 
    uint32_t        bug_id;
3646
 
    const unsigned char introduced_in[3]; // first version with bug
3647
 
    const unsigned char fixed_in[3];      // first version with fix
 
3921
    uint        bug_id;
 
3922
    const uchar introduced_in[3]; // first version with bug
 
3923
    const uchar fixed_in[3];      // first version with fix
3648
3924
  };
3649
3925
  static struct st_version_range_for_one_bug versions_for_all_bugs[]=
3650
3926
  {
3653
3929
    {33029, { 5, 0,  0 }, { 5, 0, 58 } },
3654
3930
    {33029, { 5, 1,  0 }, { 5, 1, 12 } },
3655
3931
  };
3656
 
  const unsigned char *master_ver=
 
3932
  const uchar *master_ver=
3657
3933
    rli->relay_log.description_event_for_exec->server_version_split;
3658
3934
 
3659
 
  assert(sizeof(rli->relay_log.description_event_for_exec->server_version_split) == 3);
 
3935
  DBUG_ASSERT(sizeof(rli->relay_log.description_event_for_exec->server_version_split) == 3);
3660
3936
 
3661
 
  for (uint32_t i= 0;
 
3937
  for (uint i= 0;
3662
3938
       i < sizeof(versions_for_all_bugs)/sizeof(*versions_for_all_bugs);i++)
3663
3939
  {
3664
 
    const unsigned char *introduced_in= versions_for_all_bugs[i].introduced_in,
 
3940
    const uchar *introduced_in= versions_for_all_bugs[i].introduced_in,
3665
3941
      *fixed_in= versions_for_all_bugs[i].fixed_in;
3666
3942
    if ((versions_for_all_bugs[i].bug_id == bug_id) &&
3667
3943
        (memcmp(introduced_in, master_ver, 3) <= 0) &&
3668
3944
        (memcmp(fixed_in,      master_ver, 3) >  0))
3669
3945
    {
3670
3946
      if (!report)
3671
 
        return true;
3672
 
 
 
3947
        return TRUE;
 
3948
      
3673
3949
      // a short message for SHOW SLAVE STATUS (message length constraints)
3674
 
      my_printf_error(ER_UNKNOWN_ERROR,
3675
 
                      _("master may suffer from"
3676
 
                        " http://bugs.mysql.com/bug.php?id=%u"
3677
 
                        " so slave stops; check error log on slave"
3678
 
                        " for more info"), MYF(0), bug_id);
 
3950
      my_printf_error(ER_UNKNOWN_ERROR, "master may suffer from"
 
3951
                      " http://bugs.mysql.com/bug.php?id=%u"
 
3952
                      " so slave stops; check error log on slave"
 
3953
                      " for more info", MYF(0), bug_id);
3679
3954
      // a verbose message for the error log
3680
3955
      rli->report(ERROR_LEVEL, ER_UNKNOWN_ERROR,
3681
 
                  _("According to the master's version ('%s'),"
3682
 
                    " it is probable that master suffers from this bug:"
3683
 
                    " http://bugs.mysql.com/bug.php?id=%u"
3684
 
                    " and thus replicating the current binary log event"
3685
 
                    " may make the slave's data become different from the"
3686
 
                    " master's data."
3687
 
                    " To take no risk, slave refuses to replicate"
3688
 
                    " this event and stops."
3689
 
                    " We recommend that all updates be stopped on the"
3690
 
                    " master and slave, that the data of both be"
3691
 
                    " manually synchronized,"
3692
 
                    " that master's binary logs be deleted,"
3693
 
                    " that master be upgraded to a version at least"
3694
 
                    " equal to '%d.%d.%d'. Then replication can be"
3695
 
                    " restarted."),
3696
 
                  rli->relay_log.description_event_for_exec->server_version,
3697
 
                  bug_id,
3698
 
                  fixed_in[0], fixed_in[1], fixed_in[2]);
3699
 
      return true;
 
3956
                  "According to the master's version ('%s'),"
 
3957
                  " it is probable that master suffers from this bug:"
 
3958
                      " http://bugs.mysql.com/bug.php?id=%u"
 
3959
                      " and thus replicating the current binary log event"
 
3960
                      " may make the slave's data become different from the"
 
3961
                      " master's data."
 
3962
                      " To take no risk, slave refuses to replicate"
 
3963
                      " this event and stops."
 
3964
                      " We recommend that all updates be stopped on the"
 
3965
                      " master and slave, that the data of both be"
 
3966
                      " manually synchronized,"
 
3967
                      " that master's binary logs be deleted,"
 
3968
                      " that master be upgraded to a version at least"
 
3969
                      " equal to '%d.%d.%d'. Then replication can be"
 
3970
                      " restarted.",
 
3971
                      rli->relay_log.description_event_for_exec->server_version,
 
3972
                      bug_id,
 
3973
                      fixed_in[0], fixed_in[1], fixed_in[2]);
 
3974
      return TRUE;
3700
3975
    }
3701
3976
  }
3702
 
  return false;
 
3977
  return FALSE;
3703
3978
}
3704
3979
 
3705
3980
/**
3717
3992
  if (active_mi && active_mi->rli.sql_thd == thd)
3718
3993
  {
3719
3994
    Relay_log_info *rli= &active_mi->rli;
3720
 
    return rpl_master_has_bug(rli, 33029, false);
 
3995
    return rpl_master_has_bug(rli, 33029, FALSE);
3721
3996
  }
3722
 
  return false;
 
3997
  return FALSE;
3723
3998
}
3724
3999
 
3725
4000
#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
3730
4005
/**
3731
4006
  @} (end of group Replication)
3732
4007
*/
 
4008
 
 
4009
#endif /* HAVE_REPLICATION */