sql_repl.cc 62.5 KB
Newer Older
1
/* Copyright (c) 2000, 2012, Oracle and/or its affiliates. All rights reserved.
2 3 4

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
unknown's avatar
unknown committed
5
   the Free Software Foundation; version 2 of the License.
6 7 8 9 10 11 12 13

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
14
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
15

16 17 18
#include "sql_priv.h"
#include "unireg.h"
#include "sql_parse.h"                          // check_access
unknown's avatar
SCRUM  
unknown committed
19 20
#ifdef HAVE_REPLICATION

21
#include "rpl_mi.h"
22
#include "sql_repl.h"
23
#include "sql_acl.h"                            // SUPER_ACL
24
#include "log_event.h"
unknown's avatar
unknown committed
25
#include "rpl_filter.h"
unknown's avatar
unknown committed
26
#include <my_dir.h>
He Zhenxing's avatar
He Zhenxing committed
27
#include "rpl_handler.h"
28
#include "debug_sync.h"
29

30
int max_binlog_dump_events = 0; // unlimited
31
my_bool opt_sporadic_binlog_dump_fail = 0;
unknown's avatar
unknown committed
32
#ifndef DBUG_OFF
33
static int binlog_dump_count = 0;
unknown's avatar
unknown committed
34
#endif
unknown's avatar
unknown committed
35

36 37 38 39 40 41 42 43
/**
  a copy of active_mi->rli->slave_skip_counter, for showing in SHOW VARIABLES,
  INFORMATION_SCHEMA.GLOBAL_VARIABLES and @@sql_slave_skip_counter without
  taking all the mutexes needed to access active_mi->rli->slave_skip_counter
  properly.
*/
uint sql_slave_skip_counter;

44
/*
45 46 47 48
    fake_rotate_event() builds a fake (=which does not exist physically in any
    binlog) Rotate event, which contains the name of the binlog we are going to
    send to the slave (because the slave may not know it if it just asked for
    MASTER_LOG_FILE='', MASTER_LOG_POS=4).
49 50
    < 4.0.14, fake_rotate_event() was called only if the requested pos was 4.
    After this version we always call it, so that a 3.23.58 slave can rely on
51 52 53 54 55
    it to detect if the master is 4.0 (and stop) (the _fake_ Rotate event has
    zeros in the good positions which, by chance, make it possible for the 3.23
    slave to detect that this event is unexpected) (this is luck which happens
    because the master and slave disagree on the size of the header of
    Log_event).
56 57 58 59

    Relying on the event length of the Rotate event instead of these
    well-placed zeros was not possible as Rotate events have a variable-length
    part.
60 61
*/

unknown's avatar
unknown committed
62
static int fake_rotate_event(NET* net, String* packet, char* log_file_name,
63
                             ulonglong position, const char** errmsg)
unknown's avatar
unknown committed
64
{
65
  DBUG_ENTER("fake_rotate_event");
66 67 68 69 70 71
  char header[LOG_EVENT_HEADER_LEN], buf[ROTATE_HEADER_LEN+100];
  /*
    'when' (the timestamp) is set to 0 so that slave could distinguish between
    real and fake Rotate events (if necessary)
  */
  memset(header, 0, 4);
unknown's avatar
unknown committed
72 73
  header[EVENT_TYPE_OFFSET] = ROTATE_EVENT;

74
  char* p = log_file_name+dirname_length(log_file_name);
unknown's avatar
unknown committed
75
  uint ident_len = (uint) strlen(p);
76
  ulong event_len = ident_len + LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN;
77
  int4store(header + SERVER_ID_OFFSET, server_id);
unknown's avatar
unknown committed
78
  int4store(header + EVENT_LEN_OFFSET, event_len);
79
  int2store(header + FLAGS_OFFSET, LOG_EVENT_ARTIFICIAL_F);
80

81 82
  // TODO: check what problems this may cause and fix them
  int4store(header + LOG_POS_OFFSET, 0);
83

unknown's avatar
unknown committed
84
  packet->append(header, sizeof(header));
85
  int8store(buf+R_POS_OFFSET,position);
86
  packet->append(buf, ROTATE_HEADER_LEN);
unknown's avatar
unknown committed
87
  packet->append(p,ident_len);
88
  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
89 90
  {
    *errmsg = "failed on my_net_write()";
91
    DBUG_RETURN(-1);
92
  }
93
  DBUG_RETURN(0);
unknown's avatar
unknown committed
94 95
}

He Zhenxing's avatar
He Zhenxing committed
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121
/*
  Reset thread transmit packet buffer for event sending

  This function allocates header bytes for event transmission, and
  should be called before store the event data to the packet buffer.
*/
static int reset_transmit_packet(THD *thd, ushort flags,
                                 ulong *ev_offset, const char **errmsg)
{
  int ret= 0;
  String *packet= &thd->packet;

  /* reserve and set default header */
  packet->length(0);
  packet->set("\0", 1, &my_charset_bin);

  if (RUN_HOOK(binlog_transmit, reserve_header, (thd, flags, packet)))
  {
    *errmsg= "Failed to run hook 'reserve_header'";
    my_errno= ER_UNKNOWN_ERROR;
    ret= 1;
  }
  *ev_offset= packet->length();
  return ret;
}

122 123 124
static int send_file(THD *thd)
{
  NET* net = &thd->net;
125 126
  int fd = -1, error = 1;
  size_t bytes;
127 128
  char fname[FN_REFLEN+1];
  const char *errmsg = 0;
unknown's avatar
unknown committed
129
  int old_timeout;
unknown's avatar
unknown committed
130
  unsigned long packet_len;
131
  uchar buf[IO_SIZE];				// It's safe to alloc this
132 133
  DBUG_ENTER("send_file");

unknown's avatar
unknown committed
134 135 136 137
  /*
    The client might be slow loading the data, give him wait_timeout to do
    the job
  */
138
  old_timeout= net->read_timeout;
139
  my_net_set_read_timeout(net, thd->variables.net_wait_timeout);
unknown's avatar
unknown committed
140 141 142 143 144

  /*
    We need net_flush here because the client will not know it needs to send
    us the file name until it has processed the load event entry
  */
unknown's avatar
unknown committed
145
  if (net_flush(net) || (packet_len = my_net_read(net)) == packet_error)
146
  {
147
    errmsg = "while reading file name";
148 149 150
    goto err;
  }

151 152 153
  // terminate with \0 for fn_format
  *((char*)net->read_pos +  packet_len) = 0;
  fn_format(fname, (char*) net->read_pos + 1, "", "", 4);
154 155 156 157
  // this is needed to make replicate-ignore-db
  if (!strcmp(fname,"/dev/null"))
    goto end;

Marc Alff's avatar
Marc Alff committed
158 159
  if ((fd= mysql_file_open(key_file_send_file,
                           fname, O_RDONLY, MYF(0))) < 0)
160
  {
161
    errmsg = "on open of file";
162 163 164
    goto err;
  }

Marc Alff's avatar
Marc Alff committed
165
  while ((long) (bytes= mysql_file_read(fd, buf, IO_SIZE, MYF(0))) > 0)
166 167 168
  {
    if (my_net_write(net, buf, bytes))
    {
169
      errmsg = "while writing data to client";
170 171 172 173 174
      goto err;
    }
  }

 end:
175
  if (my_net_write(net, (uchar*) "", 0) || net_flush(net) ||
176 177
      (my_net_read(net) == packet_error))
  {
178
    errmsg = "while negotiating file transfer close";
179 180 181 182 183
    goto err;
  }
  error = 0;

 err:
184
  my_net_set_read_timeout(net, old_timeout);
185
  if (fd >= 0)
Marc Alff's avatar
Marc Alff committed
186
    mysql_file_close(fd, MYF(0));
187 188
  if (errmsg)
  {
189
    sql_print_error("Failed in send_file() %s", errmsg);
Staale Smedseng's avatar
Staale Smedseng committed
190
    DBUG_PRINT("error", ("%s", errmsg));
191 192 193 194
  }
  DBUG_RETURN(error);
}

195

196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216
/*
  Adjust the position pointer in the binary log file for all running slaves

  SYNOPSIS
    adjust_linfo_offsets()
    purge_offset	Number of bytes removed from start of log index file

  NOTES
    - This is called when doing a PURGE when we delete lines from the
      index log file

  REQUIREMENTS
    - Before calling this function, we have to ensure that no threads are
      using any binary log file before purge_offset.a

  TODO
    - Inform the slave threads that they should sync the position
      in the binary log file with flush_relay_log_info.
      Now they sync is done for next read.
*/

unknown's avatar
unknown committed
217 218 219
void adjust_linfo_offsets(my_off_t purge_offset)
{
  THD *tmp;
220

Marc Alff's avatar
Marc Alff committed
221
  mysql_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
222 223
  I_List_iterator<THD> it(threads);

224 225 226 227 228
  while ((tmp=it++))
  {
    LOG_INFO* linfo;
    if ((linfo = tmp->current_linfo))
    {
Marc Alff's avatar
Marc Alff committed
229
      mysql_mutex_lock(&linfo->lock);
230 231 232 233
      /*
	Index file offset can be less that purge offset only if
	we just started reading the index file. In that case
	we have nothing to adjust
234
      */
235 236 237 238
      if (linfo->index_file_offset < purge_offset)
	linfo->fatal = (linfo->index_file_offset != 0);
      else
	linfo->index_file_offset -= purge_offset;
Marc Alff's avatar
Marc Alff committed
239
      mysql_mutex_unlock(&linfo->lock);
240 241
    }
  }
Marc Alff's avatar
Marc Alff committed
242
  mysql_mutex_unlock(&LOCK_thread_count);
unknown's avatar
unknown committed
243 244
}

unknown's avatar
merge  
unknown committed
245

unknown's avatar
unknown committed
246 247
bool log_in_use(const char* log_name)
{
248
  size_t log_name_len = strlen(log_name) + 1;
unknown's avatar
unknown committed
249 250
  THD *tmp;
  bool result = 0;
251

Marc Alff's avatar
Marc Alff committed
252
  mysql_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
253
  I_List_iterator<THD> it(threads);
254 255 256 257 258

  while ((tmp=it++))
  {
    LOG_INFO* linfo;
    if ((linfo = tmp->current_linfo))
unknown's avatar
unknown committed
259
    {
Marc Alff's avatar
Marc Alff committed
260
      mysql_mutex_lock(&linfo->lock);
261
      result = !memcmp(log_name, linfo->log_file_name, log_name_len);
Marc Alff's avatar
Marc Alff committed
262
      mysql_mutex_unlock(&linfo->lock);
263 264
      if (result)
	break;
265 266
    }
  }
unknown's avatar
unknown committed
267

Marc Alff's avatar
Marc Alff committed
268
  mysql_mutex_unlock(&LOCK_thread_count);
unknown's avatar
unknown committed
269 270 271
  return result;
}

unknown's avatar
unknown committed
272
bool purge_error_message(THD* thd, int res)
unknown's avatar
unknown committed
273
{
274
  uint errcode;
275

276
  if ((errcode= purge_log_get_error_code(res)) != 0)
277
  {
278
    my_message(errcode, ER(errcode), MYF(0));
unknown's avatar
unknown committed
279
    return TRUE;
280
  }
281
  my_ok(thd);
unknown's avatar
unknown committed
282
  return FALSE;
unknown's avatar
unknown committed
283 284
}

unknown's avatar
unknown committed
285

286 287 288 289 290 291 292 293 294 295 296
/**
  Execute a PURGE BINARY LOGS TO <log> command.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @param to_log Name of the last log to purge.

  @retval FALSE success
  @retval TRUE failure
*/
unknown's avatar
unknown committed
297
bool purge_master_logs(THD* thd, const char* to_log)
298 299
{
  char search_file_name[FN_REFLEN];
unknown's avatar
unknown committed
300 301
  if (!mysql_bin_log.is_open())
  {
302
    my_ok(thd);
unknown's avatar
unknown committed
303
    return FALSE;
unknown's avatar
unknown committed
304
  }
305 306

  mysql_bin_log.make_log_name(search_file_name, to_log);
unknown's avatar
unknown committed
307 308
  return purge_error_message(thd,
			     mysql_bin_log.purge_logs(search_file_name, 0, 1,
unknown's avatar
unknown committed
309
						      1, NULL));
310 311 312
}


313 314 315 316 317 318 319 320 321 322 323
/**
  Execute a PURGE BINARY LOGS BEFORE <date> command.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @param purge_time Date before which logs should be purged.

  @retval FALSE success
  @retval TRUE failure
*/
unknown's avatar
unknown committed
324
bool purge_master_logs_before_date(THD* thd, time_t purge_time)
325
{
326 327
  if (!mysql_bin_log.is_open())
  {
328
    my_ok(thd);
329 330 331 332
    return 0;
  }
  return purge_error_message(thd,
                             mysql_bin_log.purge_logs_before_date(purge_time));
333 334
}

335
int test_for_non_eof_log_read_errors(int error, const char **errmsg)
336
{
337
  if (error == LOG_READ_EOF)
338 339 340 341
    return 0;
  my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
  switch (error) {
  case LOG_READ_BOGUS:
342
    *errmsg = "bogus data in log event";
343 344
    break;
  case LOG_READ_TOO_LARGE:
345
    *errmsg = "log event entry exceeded max_allowed_packet; \
346 347 348
Increase max_allowed_packet on master";
    break;
  case LOG_READ_IO:
349
    *errmsg = "I/O error reading log event";
350 351
    break;
  case LOG_READ_MEM:
352
    *errmsg = "memory allocation failed reading log event";
353 354
    break;
  case LOG_READ_TRUNC:
355
    *errmsg = "binlog truncated in the middle of event; consider out of disk space on master";
356 357
    break;
  default:
358
    *errmsg = "unknown error reading log event on the master";
359 360 361 362 363 364
    break;
  }
  return error;
}


Andrei Elkin's avatar
Andrei Elkin committed
365 366 367 368 369 370 371 372 373 374 375 376 377 378
/**
  An auxiliary function for calling in mysql_binlog_send
  to initialize the heartbeat timeout in waiting for a binlogged event.

  @param[in]    thd  THD to access a user variable

  @return        heartbeat period an ulonglong of nanoseconds
                 or zero if heartbeat was not demanded by slave
*/ 
static ulonglong get_heartbeat_period(THD * thd)
{
  my_bool null_value;
  LEX_STRING name=  { C_STRING_WITH_LEN("master_heartbeat_period")};
  user_var_entry *entry= 
379
    (user_var_entry*) my_hash_search(&thd->user_vars, (uchar*) name.str,
Andrei Elkin's avatar
Andrei Elkin committed
380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431
                                  name.length);
  return entry? entry->val_int(&null_value) : 0;
}

/*
  Function prepares and sends repliation heartbeat event.

  @param net                net object of THD
  @param packet             buffer to store the heartbeat instance
  @param event_coordinates  binlog file name and position of the last
                            real event master sent from binlog

  @note 
    Among three essential pieces of heartbeat data Log_event::when
    is computed locally.
    The  error to send is serious and should force terminating
    the dump thread.
*/
static int send_heartbeat_event(NET* net, String* packet,
                                const struct event_coordinates *coord)
{
  DBUG_ENTER("send_heartbeat_event");
  char header[LOG_EVENT_HEADER_LEN];
  /*
    'when' (the timestamp) is set to 0 so that slave could distinguish between
    real and fake Rotate events (if necessary)
  */
  memset(header, 0, 4);  // when

  header[EVENT_TYPE_OFFSET] = HEARTBEAT_LOG_EVENT;

  char* p= coord->file_name + dirname_length(coord->file_name);

  uint ident_len = strlen(p);
  ulong event_len = ident_len + LOG_EVENT_HEADER_LEN;
  int4store(header + SERVER_ID_OFFSET, server_id);
  int4store(header + EVENT_LEN_OFFSET, event_len);
  int2store(header + FLAGS_OFFSET, 0);

  int4store(header + LOG_POS_OFFSET, coord->pos);  // log_pos

  packet->append(header, sizeof(header));
  packet->append(p, ident_len);             // log_file_name

  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) ||
      net_flush(net))
  {
    DBUG_RETURN(-1);
  }
  DBUG_RETURN(0);
}

432 433 434
/*
  TODO: Clean up loop to only have one call to send_file()
*/
unknown's avatar
merge  
unknown committed
435

436 437
void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
		       ushort flags)
438 439 440
{
  LOG_INFO linfo;
  char *log_file_name = linfo.log_file_name;
441
  char search_file_name[FN_REFLEN], *name;
He Zhenxing's avatar
He Zhenxing committed
442 443 444

  ulong ev_offset;

445 446
  IO_CACHE log;
  File file = -1;
447 448 449
  String* packet = &thd->packet;
  int error;
  const char *errmsg = "Unknown error";
450
  char error_text[MAX_SLAVE_ERRMSG]; // to be send to slave via my_message()
451
  NET* net = &thd->net;
Marc Alff's avatar
Marc Alff committed
452
  mysql_mutex_t *log_lock;
453 454
  mysql_cond_t *log_cond;

455
  bool binlog_can_be_corrupted= FALSE;
456 457
#ifndef DBUG_OFF
  int left_events = max_binlog_dump_events;
458
#endif
459
  int old_max_allowed_packet= thd->variables.max_allowed_packet;
460
  DBUG_ENTER("mysql_binlog_send");
461 462
  DBUG_PRINT("enter",("log_ident: '%s'  pos: %ld", log_ident, (long) pos));

463
  bzero((char*) &log,sizeof(log));
Andrei Elkin's avatar
Andrei Elkin committed
464 465 466 467 468 469
  /* 
     heartbeat_period from @master_heartbeat_period user variable
  */
  ulonglong heartbeat_period= get_heartbeat_period(thd);
  struct timespec heartbeat_buf;
  struct timespec *heartbeat_ts= NULL;
470 471 472 473
  const LOG_POS_COORD start_coord= { log_ident, pos },
    *p_start_coord= &start_coord;
  LOG_POS_COORD coord_buf= { log_file_name, BIN_LOG_HEADER_SIZE },
    *p_coord= &coord_buf;
Andrei Elkin's avatar
Andrei Elkin committed
474 475 476 477 478
  if (heartbeat_period != LL(0))
  {
    heartbeat_ts= &heartbeat_buf;
    set_timespec_nsec(*heartbeat_ts, 0);
  }
479 480
  if (global_system_variables.log_warnings > 1)
    sql_print_information("Start binlog_dump to slave_server(%d), pos(%s, %lu)",
He Zhenxing's avatar
He Zhenxing committed
481 482 483 484 485 486 487
                        thd->server_id, log_ident, (ulong)pos);
  if (RUN_HOOK(binlog_transmit, transmit_start, (thd, flags, log_ident, pos)))
  {
    errmsg= "Failed to run hook 'transmit_start'";
    my_errno= ER_UNKNOWN_ERROR;
    goto err;
  }
488

unknown's avatar
unknown committed
489
#ifndef DBUG_OFF
490 491 492
  if (opt_sporadic_binlog_dump_fail && (binlog_dump_count++ % 2))
  {
    errmsg = "Master failed COM_BINLOG_DUMP to test if slave can recover";
493
    my_errno= ER_UNKNOWN_ERROR;
494 495
    goto err;
  }
496
#endif
497

498
  if (!mysql_bin_log.is_open())
499 500
  {
    errmsg = "Binary log is not open";
501
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
502 503
    goto err;
  }
504 505 506
  if (!server_id_supplied)
  {
    errmsg = "Misconfigured master - server id was not set";
507
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
508 509 510
    goto err;
  }

511
  name=search_file_name;
512
  if (log_ident[0])
513 514
    mysql_bin_log.make_log_name(search_file_name, log_ident);
  else
515
    name=0;					// Find first log
516

unknown's avatar
unknown committed
517
  linfo.index_file_offset = 0;
518

519
  if (mysql_bin_log.find_log_pos(&linfo, name, 1))
520
  {
521
    errmsg = "Could not find first log file name in binary log index file";
522
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
523 524
    goto err;
  }
525

Marc Alff's avatar
Marc Alff committed
526
  mysql_mutex_lock(&LOCK_thread_count);
527
  thd->current_linfo = &linfo;
Marc Alff's avatar
Marc Alff committed
528
  mysql_mutex_unlock(&LOCK_thread_count);
529

530
  if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)
531 532
  {
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
533
    goto err;
534 535
  }
  if (pos < BIN_LOG_HEADER_SIZE || pos > my_b_filelength(&log))
536
  {
537
    errmsg= "Client requested master to start replication from \
unknown's avatar
unknown committed
538
impossible position";
539
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
540 541
    goto err;
  }
542

He Zhenxing's avatar
He Zhenxing committed
543 544 545
  /* reset transmit packet for the fake rotate event below */
  if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
    goto err;
546

547
  /*
548
    Tell the client about the log name with a fake Rotate event;
549 550 551 552 553 554 555 556 557
    this is needed even if we also send a Format_description_log_event
    just after, because that event does not contain the binlog's name.
    Note that as this Rotate event is sent before
    Format_description_log_event, the slave cannot have any info to
    understand this event's format, so the header len of
    Rotate_log_event is FROZEN (so in 5.0 it will have a header shorter
    than other events except FORMAT_DESCRIPTION_EVENT).
    Before 4.0.14 we called fake_rotate_event below only if (pos ==
    BIN_LOG_HEADER_SIZE), because if this is false then the slave
558
    already knows the binlog's name.
559 560 561 562 563 564 565
    Since, we always call fake_rotate_event; if the slave already knew
    the log's name (ex: CHANGE MASTER TO MASTER_LOG_FILE=...) this is
    useless but does not harm much. It is nice for 3.23 (>=.58) slaves
    which test Rotate events to see if the master is 4.0 (then they
    choose to stop because they can't replicate 4.0); by always calling
    fake_rotate_event we are sure that 3.23.58 and newer will detect the
    problem as soon as replication starts (BUG#198).
566
    Always calling fake_rotate_event makes sending of normal
567 568 569 570 571 572 573
    (=from-binlog) Rotate events a priori unneeded, but it is not so
    simple: the 2 Rotate events are not equivalent, the normal one is
    before the Stop event, the fake one is after. If we don't send the
    normal one, then the Stop event will be interpreted (by existing 4.0
    slaves) as "the master stopped", which is wrong. So for safety,
    given that we want minimum modification of 4.0, we send the normal
    and fake Rotates.
574
  */
575
  if (fake_rotate_event(net, packet, log_file_name, pos, &errmsg))
unknown's avatar
unknown committed
576
  {
577 578 579 580
    /*
       This error code is not perfect, as fake_rotate_event() does not
       read anything from the binlog; if it fails it's because of an
       error in my_net_write(), fortunately it will say so in errmsg.
581
    */
582 583
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
    goto err;
unknown's avatar
unknown committed
584
  }
He Zhenxing's avatar
He Zhenxing committed
585

586 587 588 589 590
  /*
    Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become
    this larger than the corresponding packet (query) sent 
    from client to master.
  */
591
  thd->variables.max_allowed_packet= MAX_MAX_ALLOWED_PACKET;
unknown's avatar
unknown committed
592

593
  /*
594 595 596
    We can set log_lock now, it does not move (it's a member of
    mysql_bin_log, and it's already inited, and it will be destroyed
    only at shutdown).
597
  */
598
  p_coord->pos= pos; // the first hb matches the slave's last seen value
599 600
  log_lock= mysql_bin_log.get_log_lock();
  log_cond= mysql_bin_log.get_log_cond();
601
  if (pos > BIN_LOG_HEADER_SIZE)
602
  {
He Zhenxing's avatar
He Zhenxing committed
603 604 605 606 607
    /* reset transmit packet for the event read from binary log
       file */
    if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
      goto err;

608 609 610 611
     /*
       Try to find a Format_description_log_event at the beginning of
       the binlog
     */
612 613 614
     if (!(error = Log_event::read_log_event(&log, packet, log_lock)))
     {
       /*
He Zhenxing's avatar
He Zhenxing committed
615 616 617
         The packet has offsets equal to the normal offsets in a
         binlog event + ev_offset (the first ev_offset characters are
         the header (default \0)).
618 619 620
       */
       DBUG_PRINT("info",
                  ("Looked for a Format_description_log_event, found event type %d",
He Zhenxing's avatar
He Zhenxing committed
621 622
                   (*packet)[EVENT_TYPE_OFFSET+ev_offset]));
       if ((*packet)[EVENT_TYPE_OFFSET+ev_offset] == FORMAT_DESCRIPTION_EVENT)
623
       {
He Zhenxing's avatar
He Zhenxing committed
624
         binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+ev_offset] &
625
                                       LOG_EVENT_BINLOG_IN_USE_F);
He Zhenxing's avatar
He Zhenxing committed
626
         (*packet)[FLAGS_OFFSET+ev_offset] &= ~LOG_EVENT_BINLOG_IN_USE_F;
627 628 629 630 631
         /*
           mark that this event with "log_pos=0", so the slave
           should not increment master's binlog position
           (rli->group_master_log_pos)
         */
He Zhenxing's avatar
He Zhenxing committed
632
         int4store((char*) packet->ptr()+LOG_POS_OFFSET+ev_offset, 0);
633 634 635 636 637
         /*
           if reconnect master sends FD event with `created' as 0
           to avoid destroying temp tables.
          */
         int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+
He Zhenxing's avatar
He Zhenxing committed
638
                   ST_CREATED_OFFSET+ev_offset, (ulong) 0);
639
         /* send it */
640
         if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
641 642 643 644 645
         {
           errmsg = "Failed on my_net_write()";
           my_errno= ER_UNKNOWN_ERROR;
           goto err;
         }
unknown's avatar
unknown committed
646

647
         /*
648 649 650 651
           No need to save this event. We are only doing simple reads
           (no real parsing of the events) so we don't need it. And so
           we don't need the artificial Format_description_log_event of
           3.23&4.x.
652 653 654 655
         */
       }
     }
     else
656
     {
657
       if (test_for_non_eof_log_read_errors(error, &errmsg))
658
         goto err;
659 660 661 662 663 664 665 666 667 668
       /*
         It's EOF, nothing to do, go on reading next events, the
         Format_description_log_event will be found naturally if it is written.
       */
     }
  } /* end of if (pos > BIN_LOG_HEADER_SIZE); */
  else
  {
    /* The Format_description_log_event event will be found naturally. */
  }
669

670 671 672
  /* seek to the requested position, to start the requested dump */
  my_b_seek(&log, pos);			// Seek will done on next read

unknown's avatar
unknown committed
673
  while (!net->error && net->vio != 0 && !thd->killed)
674
  {
He Zhenxing's avatar
He Zhenxing committed
675 676 677 678 679 680
    Log_event_type event_type= UNKNOWN_EVENT;

    /* reset the transmit packet for the event read from binary log
       file */
    if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
      goto err;
681

682
    while (!(error= Log_event::read_log_event(&log, packet, log_lock)))
683
    {
684
#ifndef DBUG_OFF
685
      if (max_binlog_dump_events && !left_events--)
686 687 688
      {
	net_flush(net);
	errmsg = "Debugging binlog dump abort";
689
	my_errno= ER_UNKNOWN_ERROR;
690 691
	goto err;
      }
692
#endif
Andrei Elkin's avatar
Andrei Elkin committed
693 694 695
      /*
        log's filename does not change while it's active
      */
696
      p_coord->pos= uint4korr(packet->ptr() + ev_offset + LOG_POS_OFFSET);
697

He Zhenxing's avatar
He Zhenxing committed
698
      event_type= (Log_event_type)((*packet)[LOG_EVENT_OFFSET+ev_offset]);
699 700
      DBUG_EXECUTE_IF("dump_thread_wait_before_send_xid",
                      {
Libing Song's avatar
Libing Song committed
701
                        if (event_type == XID_EVENT)
702 703 704 705 706 707 708 709 710 711
                        {
                          net_flush(net);
                          const char act[]=
                            "now "
                            "wait_for signal.continue";
                          DBUG_ASSERT(opt_debug_sync_timeout > 0);
                          DBUG_ASSERT(!debug_sync_set_action(current_thd,
                                                             STRING_WITH_LEN(act)));
                        }
                      });
He Zhenxing's avatar
He Zhenxing committed
712
      if (event_type == FORMAT_DESCRIPTION_EVENT)
unknown's avatar
unknown committed
713
      {
He Zhenxing's avatar
He Zhenxing committed
714
        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+ev_offset] &
715
                                      LOG_EVENT_BINLOG_IN_USE_F);
He Zhenxing's avatar
He Zhenxing committed
716
        (*packet)[FLAGS_OFFSET+ev_offset] &= ~LOG_EVENT_BINLOG_IN_USE_F;
unknown's avatar
unknown committed
717
      }
He Zhenxing's avatar
He Zhenxing committed
718
      else if (event_type == STOP_EVENT)
719 720
        binlog_can_be_corrupted= FALSE;

721 722 723 724 725 726 727 728 729 730 731
      /*
        Introduced this code to make the gcc 4.6.1 compiler happy. When
        warnings are converted to errors, the compiler complains about
        the fact that binlog_can_be_corrupted is defined but never used.

        We need to check if this is a dead code or if someone removed any
        code by mistake.

        /Alfranio
      */
      if (binlog_can_be_corrupted)
732 733 734 735 736 737 738 739 740
      {
        /*
           Don't try to print out warning messages because this generates
           erroneous messages in the error log and causes performance
           problems.

           /Alfranio
        */
      }
741
      
He Zhenxing's avatar
He Zhenxing committed
742 743 744 745 746 747 748 749 750
      pos = my_b_tell(&log);
      if (RUN_HOOK(binlog_transmit, before_send_event,
                   (thd, flags, packet, log_file_name, pos)))
      {
        my_errno= ER_UNKNOWN_ERROR;
        errmsg= "run 'before_send_event' hook failed";
        goto err;
      }

751
      if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
752 753
      {
	errmsg = "Failed on my_net_write()";
754
	my_errno= ER_UNKNOWN_ERROR;
755 756
	goto err;
      }
unknown's avatar
unknown committed
757

758 759
      DBUG_EXECUTE_IF("dump_thread_wait_before_send_xid",
                      {
Libing Song's avatar
Libing Song committed
760
                        if (event_type == XID_EVENT)
761 762 763 764 765
                        {
                          net_flush(net);
                        }
                      });

He Zhenxing's avatar
He Zhenxing committed
766 767
      DBUG_PRINT("info", ("log event code %d", event_type));
      if (event_type == LOAD_EVENT)
768
      {
769
	if (send_file(thd))
770
	{
771
	  errmsg = "failed in send_file()";
772
	  my_errno= ER_UNKNOWN_ERROR;
773 774
	  goto err;
	}
775
      }
He Zhenxing's avatar
He Zhenxing committed
776 777 778 779 780 781 782 783 784 785 786

      if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet)))
      {
        errmsg= "Failed to run hook 'after_send_event'";
        my_errno= ER_UNKNOWN_ERROR;
        goto err;
      }

      /* reset transmit packet for next loop */
      if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
        goto err;
787
    }
788

789 790
    /*
      TODO: now that we are logging the offset, check to make sure
791
      the recorded offset and the actual match.
792 793 794
      Guilhem 2003-06: this is not true if this master is a slave
      <4.0.15 running with --log-slave-updates, because then log_pos may
      be the offset in the-master-of-this-master's binlog.
795
    */
796
    if (test_for_non_eof_log_read_errors(error, &errmsg))
797
      goto err;
798

799
    if (!(flags & BINLOG_DUMP_NON_BLOCK) &&
800
        mysql_bin_log.is_active(log_file_name))
801
    {
802 803 804
      /*
	Block until there is more data in the log
      */
805
      if (net_flush(net))
806 807
      {
	errmsg = "failed on net_flush()";
808
	my_errno= ER_UNKNOWN_ERROR;
809 810
	goto err;
      }
811

812 813 814 815 816 817
      /*
	We may have missed the update broadcast from the log
	that has just happened, let's try to catch it if it did.
	If we did not miss anything, we just wait for other threads
	to signal us.
      */
818 819
      {
	log.error=0;
820
	bool read_packet = 0;
821

822
#ifndef DBUG_OFF
823
	if (max_binlog_dump_events && !left_events--)
824 825
	{
	  errmsg = "Debugging binlog dump abort";
826
	  my_errno= ER_UNKNOWN_ERROR;
827 828
	  goto err;
	}
829
#endif
830

He Zhenxing's avatar
He Zhenxing committed
831 832 833 834 835
        /* reset the transmit packet for the event read from binary log
           file */
        if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
          goto err;
        
836 837 838 839
	/*
	  No one will update the log while we are reading
	  now, but we'll be quick and just read one record

840
	  TODO:
841 842 843
          Add an counter that is incremented for each time we update the
          binary log.  We can avoid the following read if the counter
          has not been updated since last read.
844
	*/
845

Marc Alff's avatar
Marc Alff committed
846 847
        mysql_mutex_lock(log_lock);
        switch (error= Log_event::read_log_event(&log, packet, (mysql_mutex_t*) 0)) {
848
	case 0:
849
	  /* we read successfully, so we'll need to send it to the slave */
Marc Alff's avatar
Marc Alff committed
850
          mysql_mutex_unlock(log_lock);
851
	  read_packet = 1;
852
          p_coord->pos= uint4korr(packet->ptr() + ev_offset + LOG_POS_OFFSET);
He Zhenxing's avatar
He Zhenxing committed
853
          event_type= (Log_event_type)((*packet)[LOG_EVENT_OFFSET+ev_offset]);
854
	  break;
855

856
	case LOG_READ_EOF:
Andrei Elkin's avatar
Andrei Elkin committed
857 858 859
        {
          int ret;
          ulong signal_cnt;
860
	  DBUG_PRINT("wait",("waiting for data in binary log"));
unknown's avatar
unknown committed
861
	  if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0)
unknown's avatar
unknown committed
862
	  {
Marc Alff's avatar
Marc Alff committed
863
            mysql_mutex_unlock(log_lock);
unknown's avatar
unknown committed
864 865
	    goto end;
	  }
866

Andrei Elkin's avatar
Andrei Elkin committed
867 868 869
#ifndef DBUG_OFF
          ulong hb_info_counter= 0;
#endif
870
          const char* old_msg= thd->proc_info;
Andrei Elkin's avatar
Andrei Elkin committed
871 872 873
          signal_cnt= mysql_bin_log.signal_cnt;
          do 
          {
874
            if (heartbeat_period != 0)
Andrei Elkin's avatar
Andrei Elkin committed
875
            {
876
              DBUG_ASSERT(heartbeat_ts);
Andrei Elkin's avatar
Andrei Elkin committed
877 878
              set_timespec_nsec(*heartbeat_ts, heartbeat_period);
            }
879 880 881
            thd->enter_cond(log_cond, log_lock,
                            "Master has sent all binlog to slave; "
                            "waiting for binlog to be updated");
Andrei Elkin's avatar
Andrei Elkin committed
882
            ret= mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts);
883
            DBUG_ASSERT(ret == 0 || (heartbeat_period != 0));
Andrei Elkin's avatar
Andrei Elkin committed
884 885 886 887 888 889 890 891 892 893 894
            if (ret == ETIMEDOUT || ret == ETIME)
            {
#ifndef DBUG_OFF
              if (hb_info_counter < 3)
              {
                sql_print_information("master sends heartbeat message");
                hb_info_counter++;
                if (hb_info_counter == 3)
                  sql_print_information("the rest of heartbeat info skipped ...");
              }
#endif
895 896
              /* reset transmit packet for the heartbeat event */
              if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
897 898
              {
                thd->exit_cond(old_msg);
899
                goto err;
900
              }
901
              if (send_heartbeat_event(net, packet, p_coord))
Andrei Elkin's avatar
Andrei Elkin committed
902 903 904
              {
                errmsg = "Failed on my_net_write()";
                my_errno= ER_UNKNOWN_ERROR;
905
                thd->exit_cond(old_msg);
Andrei Elkin's avatar
Andrei Elkin committed
906 907 908 909 910
                goto err;
              }
            }
            else
            {
911
              DBUG_PRINT("wait",("binary log received update or a broadcast signal caught"));
Andrei Elkin's avatar
Andrei Elkin committed
912 913
            }
          } while (signal_cnt == mysql_bin_log.signal_cnt && !thd->killed);
914
          thd->exit_cond(old_msg);
Andrei Elkin's avatar
Andrei Elkin committed
915 916 917 918
        }
        break;
            
        default:
Marc Alff's avatar
Marc Alff committed
919
          mysql_mutex_unlock(log_lock);
920 921
          test_for_non_eof_log_read_errors(error, &errmsg);
          goto err;
922
	}
923

924
	if (read_packet)
He Zhenxing's avatar
He Zhenxing committed
925 926 927 928 929 930 931 932 933 934 935
        {
          thd_proc_info(thd, "Sending binlog event to slave");
          pos = my_b_tell(&log);
          if (RUN_HOOK(binlog_transmit, before_send_event,
                       (thd, flags, packet, log_file_name, pos)))
          {
            my_errno= ER_UNKNOWN_ERROR;
            errmsg= "run 'before_send_event' hook failed";
            goto err;
          }
	  
936
	  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) )
937 938
	  {
	    errmsg = "Failed on my_net_write()";
939
	    my_errno= ER_UNKNOWN_ERROR;
940 941
	    goto err;
	  }
unknown's avatar
unknown committed
942

He Zhenxing's avatar
He Zhenxing committed
943
	  if (event_type == LOAD_EVENT)
944
	  {
945
	    if (send_file(thd))
unknown's avatar
unknown committed
946
	    {
947
	      errmsg = "failed in send_file()";
948
	      my_errno= ER_UNKNOWN_ERROR;
unknown's avatar
unknown committed
949 950
	      goto err;
	    }
951
	  }
He Zhenxing's avatar
He Zhenxing committed
952 953 954 955 956 957 958

          if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet)))
          {
            my_errno= ER_UNKNOWN_ERROR;
            errmsg= "Failed to run hook 'after_send_event'";
            goto err;
          }
959
	}
960 961 962

	log.error=0;
      }
963
    }
964 965 966
    else
    {
      bool loop_breaker = 0;
967
      /* need this to break out of the for loop from switch */
968

969
      thd_proc_info(thd, "Finished reading one binlog; switching to next binlog");
970
      switch (mysql_bin_log.find_next_log(&linfo, 1)) {
971 972
      case 0:
	break;
973 974 975 976 977 978
      case LOG_INFO_EOF:
        if (mysql_bin_log.is_active(log_file_name))
        {
          loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
          break;
        }
979 980
      default:
	errmsg = "could not find next log";
981
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
982 983 984
	goto err;
      }

985 986
      if (loop_breaker)
        break;
unknown's avatar
foo1  
unknown committed
987

988
      end_io_cache(&log);
Marc Alff's avatar
Marc Alff committed
989
      mysql_file_close(file, MYF(MY_WME));
990

He Zhenxing's avatar
He Zhenxing committed
991 992 993 994
      /* reset transmit packet for the possible fake rotate event */
      if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
        goto err;
      
995
      /*
996 997 998 999 1000 1001 1002
        Call fake_rotate_event() in case the previous log (the one which
        we have just finished reading) did not contain a Rotate event
        (for example (I don't know any other example) the previous log
        was the last one before the master was shutdown & restarted).
        This way we tell the slave about the new log's name and
        position.  If the binlog is 5.0, the next event we are going to
        read and send is Format_description_log_event.
1003
      */
1004
      if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||
1005 1006
	  fake_rotate_event(net, packet, log_file_name, BIN_LOG_HEADER_SIZE,
                            &errmsg))
1007 1008
      {
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
unknown's avatar
unknown committed
1009
	goto err;
1010
      }
unknown's avatar
unknown committed
1011

1012
      p_coord->file_name= log_file_name; // reset to the next
1013
    }
1014
  }
1015

unknown's avatar
unknown committed
1016
end:
1017
  end_io_cache(&log);
Marc Alff's avatar
Marc Alff committed
1018
  mysql_file_close(file, MYF(MY_WME));
1019

He Zhenxing's avatar
He Zhenxing committed
1020
  RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags));
1021
  my_eof(thd);
1022
  thd_proc_info(thd, "Waiting to finalize termination");
Marc Alff's avatar
Marc Alff committed
1023
  mysql_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
1024
  thd->current_linfo = 0;
Marc Alff's avatar
Marc Alff committed
1025
  mysql_mutex_unlock(&LOCK_thread_count);
1026
  thd->variables.max_allowed_packet= old_max_allowed_packet;
1027
  DBUG_VOID_RETURN;
1028

unknown's avatar
unknown committed
1029
err:
1030
  thd_proc_info(thd, "Waiting to finalize termination");
1031
  if (my_errno == ER_MASTER_FATAL_ERROR_READING_BINLOG && my_b_inited(&log))
1032
  {
1033 1034 1035 1036
    /* 
       detailing the fatal error message with coordinates 
       of the last position read.
    */
Andrei Elkin's avatar
Andrei Elkin committed
1037 1038 1039 1040 1041 1042 1043 1044
    my_snprintf(error_text, sizeof(error_text),
                "%s; the first event '%s' at %lld, "
                "the last event read from '%s' at %lld, "
                "the last byte read from '%s' at %lld.",
                errmsg,
                p_start_coord->file_name, p_start_coord->pos,
                p_coord->file_name, p_coord->pos,
                log_file_name, my_b_tell(&log));
1045
  }
1046 1047
  else
    strcpy(error_text, errmsg);
1048
  end_io_cache(&log);
He Zhenxing's avatar
He Zhenxing committed
1049
  RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags));
unknown's avatar
unknown committed
1050 1051 1052 1053 1054 1055 1056
  /*
    Exclude  iteration through thread list
    this is needed for purge_logs() - it will iterate through
    thread list and update thd->current_linfo->index_file_offset
    this mutex will make sure that it never tried to update our linfo
    after we return from this stack frame
  */
Marc Alff's avatar
Marc Alff committed
1057
  mysql_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
1058
  thd->current_linfo = 0;
Marc Alff's avatar
Marc Alff committed
1059
  mysql_mutex_unlock(&LOCK_thread_count);
1060
  if (file >= 0)
Marc Alff's avatar
Marc Alff committed
1061
    mysql_file_close(file, MYF(MY_WME));
1062
  thd->variables.max_allowed_packet= old_max_allowed_packet;
1063

1064
  my_message(my_errno, error_text, MYF(0));
1065 1066 1067
  DBUG_VOID_RETURN;
}

1068 1069 1070 1071 1072 1073 1074 1075 1076

/**
  Execute a START SLAVE statement.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @param mi Pointer to Master_info object for the slave's IO thread.

Marc Alff's avatar
Marc Alff committed
1077
  @param net_report If true, saves the exit status into thd->stmt_da.
1078 1079 1080 1081

  @retval 0 success
  @retval 1 error
*/
1082
int start_slave(THD* thd , Master_info* mi,  bool net_report)
1083
{
1084
  int slave_errno= 0;
1085
  int thread_mask;
1086
  DBUG_ENTER("start_slave");
1087

Marc Alff's avatar
Marc Alff committed
1088
  if (check_access(thd, SUPER_ACL, any_db, NULL, NULL, 0, 0))
1089
    DBUG_RETURN(1);
1090
  lock_slave_threads(mi);  // this allows us to cleanly read slave_running
1091
  // Get a mask of _stopped_ threads
1092
  init_thread_mask(&thread_mask,mi,1 /* inverse */);
1093
  /*
1094 1095 1096 1097
    Below we will start all stopped threads.  But if the user wants to
    start only one thread, do as if the other thread was running (as we
    don't wan't to touch the other thread), so set the bit to 0 for the
    other thread
1098
  */
unknown's avatar
unknown committed
1099
  if (thd->lex->slave_thd_opt)
1100
    thread_mask&= thd->lex->slave_thd_opt;
1101
  if (thread_mask) //some threads are stopped, start them
1102
  {
1103 1104
    if (init_master_info(mi,master_info_file,relay_log_info_file, 0,
			 thread_mask))
1105 1106
      slave_errno=ER_MASTER_INFO;
    else if (server_id_supplied && *mi->host)
1107
    {
1108
      /*
1109 1110 1111
        If we will start SQL thread we will care about UNTIL options If
        not and they are specified we will ignore them and warn user
        about this fact.
1112 1113 1114
      */
      if (thread_mask & SLAVE_SQL)
      {
Marc Alff's avatar
Marc Alff committed
1115
        mysql_mutex_lock(&mi->rli.data_lock);
1116

unknown's avatar
unknown committed
1117
        if (thd->lex->mi.pos)
1118
        {
1119
          mi->rli.until_condition= Relay_log_info::UNTIL_MASTER_POS;
unknown's avatar
unknown committed
1120
          mi->rli.until_log_pos= thd->lex->mi.pos;
1121 1122
          /*
             We don't check thd->lex->mi.log_file_name for NULL here
1123 1124
             since it is checked in sql_yacc.yy
          */
unknown's avatar
unknown committed
1125
          strmake(mi->rli.until_log_name, thd->lex->mi.log_file_name,
1126
                  sizeof(mi->rli.until_log_name)-1);
1127
        }
unknown's avatar
unknown committed
1128
        else if (thd->lex->mi.relay_log_pos)
1129
        {
1130
          mi->rli.until_condition= Relay_log_info::UNTIL_RELAY_POS;
unknown's avatar
unknown committed
1131 1132
          mi->rli.until_log_pos= thd->lex->mi.relay_log_pos;
          strmake(mi->rli.until_log_name, thd->lex->mi.relay_log_name,
1133
                  sizeof(mi->rli.until_log_name)-1);
1134 1135
        }
        else
1136
          mi->rli.clear_until_condition();
1137

1138
        if (mi->rli.until_condition != Relay_log_info::UNTIL_NONE)
1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150
        {
          /* Preparing members for effective until condition checking */
          const char *p= fn_ext(mi->rli.until_log_name);
          char *p_end;
          if (*p)
          {
            //p points to '.'
            mi->rli.until_log_name_extension= strtoul(++p,&p_end, 10);
            /*
              p_end points to the first invalid character. If it equals
              to p, no digits were found, error. If it contains '\0' it
              means  conversion went ok.
1151
            */
unknown's avatar
unknown committed
1152
            if (p_end==p || *p_end)
1153 1154 1155 1156
              slave_errno=ER_BAD_SLAVE_UNTIL_COND;
          }
          else
            slave_errno=ER_BAD_SLAVE_UNTIL_COND;
1157

1158
          /* mark the cached result of the UNTIL comparison as "undefined" */
1159
          mi->rli.until_log_names_cmp_result=
1160
            Relay_log_info::UNTIL_LOG_NAMES_CMP_UNKNOWN;
1161 1162 1163

          /* Issuing warning then started without --skip-slave-start */
          if (!opt_skip_slave_start)
1164
            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
unknown's avatar
foo1  
unknown committed
1165
                         ER_MISSING_SKIP_SLAVE,
1166 1167
                         ER(ER_MISSING_SKIP_SLAVE));
        }
1168

Marc Alff's avatar
Marc Alff committed
1169
        mysql_mutex_unlock(&mi->rli.data_lock);
1170
      }
unknown's avatar
unknown committed
1171
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
1172
        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
1173
                     ER(ER_UNTIL_COND_IGNORED));
1174

unknown's avatar
unknown committed
1175
      if (!slave_errno)
1176
        slave_errno = start_slave_threads(0 /*no mutex */,
1177 1178 1179 1180
					1 /* wait for start */,
					mi,
					master_info_file,relay_log_info_file,
					thread_mask);
1181
    }
1182 1183 1184
    else
      slave_errno = ER_BAD_SLAVE;
  }
1185
  else
1186 1187
  {
    /* no error if all threads are already started, only a warning */
1188 1189
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_RUNNING,
                 ER(ER_SLAVE_WAS_RUNNING));
1190
  }
unknown's avatar
foo1  
unknown committed
1191

1192
  unlock_slave_threads(mi);
1193

1194 1195
  if (slave_errno)
  {
1196
    if (net_report)
unknown's avatar
unknown committed
1197
      my_message(slave_errno, ER(slave_errno), MYF(0));
1198
    DBUG_RETURN(1);
1199 1200
  }
  else if (net_report)
1201
    my_ok(thd);
1202

1203
  DBUG_RETURN(0);
1204 1205
}

1206

1207 1208 1209 1210 1211 1212 1213 1214
/**
  Execute a STOP SLAVE statement.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @param mi Pointer to Master_info object for the slave's IO thread.

Marc Alff's avatar
Marc Alff committed
1215
  @param net_report If true, saves the exit status into thd->stmt_da.
1216 1217 1218 1219

  @retval 0 success
  @retval 1 error
*/
1220
int stop_slave(THD* thd, Master_info* mi, bool net_report )
1221
{
1222 1223
  DBUG_ENTER("stop_slave");
  
1224
  int slave_errno;
1225 1226
  if (!thd)
    thd = current_thd;
1227

Marc Alff's avatar
Marc Alff committed
1228
  if (check_access(thd, SUPER_ACL, any_db, NULL, NULL, 0, 0))
1229
    DBUG_RETURN(1);
1230
  thd_proc_info(thd, "Killing slave");
1231 1232
  int thread_mask;
  lock_slave_threads(mi);
1233
  // Get a mask of _running_ threads
1234
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
1235 1236 1237 1238 1239 1240
  /*
    Below we will stop all running threads.
    But if the user wants to stop only one thread, do as if the other thread
    was stopped (as we don't wan't to touch the other thread), so set the
    bit to 0 for the other thread
  */
unknown's avatar
unknown committed
1241 1242
  if (thd->lex->slave_thd_opt)
    thread_mask &= thd->lex->slave_thd_opt;
1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255

  if (thread_mask)
  {
    slave_errno= terminate_slave_threads(mi,thread_mask,
                                         1 /*skip lock */);
  }
  else
  {
    //no error if both threads are already stopped, only a warning
    slave_errno= 0;
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_NOT_RUNNING,
                 ER(ER_SLAVE_WAS_NOT_RUNNING));
  }
1256
  unlock_slave_threads(mi);
1257
  thd_proc_info(thd, 0);
1258

1259 1260 1261
  if (slave_errno)
  {
    if (net_report)
unknown's avatar
unknown committed
1262
      my_message(slave_errno, ER(slave_errno), MYF(0));
1263
    DBUG_RETURN(1);
1264 1265
  }
  else if (net_report)
1266
    my_ok(thd);
1267

1268
  DBUG_RETURN(0);
1269 1270
}

1271

1272 1273
/**
  Execute a RESET SLAVE statement.
1274

1275 1276
  @param thd Pointer to THD object of the client thread executing the
  statement.
1277

1278
  @param mi Pointer to Master_info object for the slave.
1279

1280 1281 1282
  @retval 0 success
  @retval 1 error
*/
1283
int reset_slave(THD *thd, Master_info* mi)
1284 1285 1286
{
  MY_STAT stat_area;
  char fname[FN_REFLEN];
1287
  int thread_mask= 0, error= 0;
He Zhenxing's avatar
He Zhenxing committed
1288 1289
  uint sql_errno=ER_UNKNOWN_ERROR;
  const char* errmsg= "Unknown error occured while reseting slave";
unknown's avatar
unknown committed
1290 1291
  DBUG_ENTER("reset_slave");

1292
  lock_slave_threads(mi);
1293 1294 1295 1296 1297 1298 1299
  init_thread_mask(&thread_mask,mi,0 /* not inverse */);
  if (thread_mask) // We refuse if any slave thread is running
  {
    sql_errno= ER_SLAVE_MUST_STOP;
    error=1;
    goto err;
  }
unknown's avatar
unknown committed
1300 1301 1302

  ha_reset_slave(thd);

1303
  // delete relay logs, clear relay log coordinates
1304 1305 1306
  if ((error= purge_relay_logs(&mi->rli, thd,
			       1 /* just reset */,
			       &errmsg)))
1307 1308
  {
    sql_errno= ER_RELAY_LOG_FAIL;
1309
    goto err;
1310
  }
1311

1312 1313 1314
  /* Clear master's log coordinates and associated information */
  mi->clear_in_memory_info(thd->lex->reset_slave_info.all);

1315
  /*
1316
     Reset errors (the idea is that we forget about the
unknown's avatar
unknown committed
1317 1318
     old master).
  */
1319
  mi->clear_error();
1320
  mi->rli.clear_error();
1321
  mi->rli.clear_until_condition();
1322

1323
  // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
1324
  end_master_info(mi);
1325
  // and delete these two files
1326
  fn_format(fname, master_info_file, mysql_data_home, "", 4+32);
Marc Alff's avatar
Marc Alff committed
1327 1328
  if (mysql_file_stat(key_file_master_info, fname, &stat_area, MYF(0)) &&
      mysql_file_delete(key_file_master_info, fname, MYF(MY_WME)))
1329
  {
1330 1331
    error=1;
    goto err;
1332
  }
1333
  // delete relay_log_info_file
1334
  fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
Marc Alff's avatar
Marc Alff committed
1335 1336
  if (mysql_file_stat(key_file_relay_log_info, fname, &stat_area, MYF(0)) &&
      mysql_file_delete(key_file_relay_log_info, fname, MYF(MY_WME)))
1337 1338 1339 1340
  {
    error=1;
    goto err;
  }
1341

He Zhenxing's avatar
He Zhenxing committed
1342
  RUN_HOOK(binlog_relay_io, after_reset_slave, (thd, mi));
1343 1344
err:
  unlock_slave_threads(mi);
unknown's avatar
unknown committed
1345
  if (error)
1346
    my_error(sql_errno, MYF(0), errmsg);
unknown's avatar
unknown committed
1347
  DBUG_RETURN(error);
1348 1349
}

1350 1351 1352 1353
/*

  Kill all Binlog_dump threads which previously talked to the same slave
  ("same" means with the same server id). Indeed, if the slave stops, if the
Marc Alff's avatar
Marc Alff committed
1354
  Binlog_dump thread is waiting (mysql_cond_wait) for binlog update, then it
1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367
  will keep existing until a query is written to the binlog. If the master is
  idle, then this could last long, and if the slave reconnects, we could have 2
  Binlog_dump threads in SHOW PROCESSLIST, until a query is written to the
  binlog. To avoid this, when the slave reconnects and sends COM_BINLOG_DUMP,
  the master kills any existing thread with the slave's server id (if this id is
  not zero; it will be true for real slaves, but false for mysqlbinlog when it
  sends COM_BINLOG_DUMP to get a remote binlog dump).

  SYNOPSIS
    kill_zombie_dump_threads()
    slave_server_id     the slave's server id

*/
unknown's avatar
foo1  
unknown committed
1368

1369

1370 1371
void kill_zombie_dump_threads(uint32 slave_server_id)
{
Marc Alff's avatar
Marc Alff committed
1372
  mysql_mutex_lock(&LOCK_thread_count);
1373 1374 1375
  I_List_iterator<THD> it(threads);
  THD *tmp;

1376 1377 1378 1379
  while ((tmp=it++))
  {
    if (tmp->command == COM_BINLOG_DUMP &&
       tmp->server_id == slave_server_id)
1380
    {
Marc Alff's avatar
Marc Alff committed
1381
      mysql_mutex_lock(&tmp->LOCK_thd_data);    // Lock from delete
1382
      break;
1383 1384
    }
  }
Marc Alff's avatar
Marc Alff committed
1385
  mysql_mutex_unlock(&LOCK_thread_count);
1386 1387 1388 1389 1390 1391 1392
  if (tmp)
  {
    /*
      Here we do not call kill_one_thread() as
      it will be slow because it will iterate through the list
      again. We just to do kill the thread ourselves.
    */
unknown's avatar
SCRUM  
unknown committed
1393
    tmp->awake(THD::KILL_QUERY);
Marc Alff's avatar
Marc Alff committed
1394
    mysql_mutex_unlock(&tmp->LOCK_thd_data);
1395
  }
1396 1397
}

1398

1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410
/**
  Execute a CHANGE MASTER statement.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @param mi Pointer to Master_info object belonging to the slave's IO
  thread.

  @retval FALSE success
  @retval TRUE error
*/
1411
bool change_master(THD* thd, Master_info* mi)
1412
{
1413
  int thread_mask;
unknown's avatar
unknown committed
1414 1415
  const char* errmsg= 0;
  bool need_relay_log_purge= 1;
Andrei Elkin's avatar
Andrei Elkin committed
1416
  bool ret= FALSE;
1417 1418 1419 1420
  char saved_host[HOSTNAME_LENGTH + 1];
  uint saved_port;
  char saved_log_name[FN_REFLEN];
  my_off_t saved_log_pos;
1421 1422
  DBUG_ENTER("change_master");

1423
  lock_slave_threads(mi);
1424
  init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
Andrei Elkin's avatar
Andrei Elkin committed
1425
  LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
1426
  if (thread_mask) // We refuse if any slave thread is running
1427
  {
unknown's avatar
unknown committed
1428
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
Andrei Elkin's avatar
Andrei Elkin committed
1429 1430
    ret= TRUE;
    goto err;
1431
  }
1432

1433
  thd_proc_info(thd, "Changing master");
1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446
  /* 
    We need to check if there is an empty master_host. Otherwise
    change master succeeds, a master.info file is created containing 
    empty master_host string and when issuing: start slave; an error
    is thrown stating that the server is not configured as slave.
    (See BUG#28796).
  */
  if(lex_mi->host && !*lex_mi->host) 
  {
    my_error(ER_WRONG_ARGUMENTS, MYF(0), "MASTER_HOST");
    unlock_slave_threads(mi);
    DBUG_RETURN(TRUE);
  }
1447
  // TODO: see if needs re-write
1448 1449
  if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
		       thread_mask))
1450
  {
unknown's avatar
unknown committed
1451
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
Andrei Elkin's avatar
Andrei Elkin committed
1452 1453
    ret= TRUE;
    goto err;
1454 1455
  }

1456 1457 1458 1459
  /*
    Data lock not needed since we have already stopped the running threads,
    and we have the hold on the run locks which will keep all threads that
    could possibly modify the data structures from running
1460
  */
1461

1462 1463 1464
  /*
    Before processing the command, save the previous state.
  */
1465
  strmake(saved_host, mi->host, HOSTNAME_LENGTH);
1466
  saved_port= mi->port;
1467
  strmake(saved_log_name, mi->master_log_name, FN_REFLEN - 1);
1468 1469
  saved_log_pos= mi->master_log_pos;

1470
  /*
unknown's avatar
foo1  
unknown committed
1471
    If the user specified host or port without binlog or position,
1472
    reset binlog's name to FIRST and position to 4.
unknown's avatar
foo1  
unknown committed
1473
  */
1474

1475 1476
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
  {
1477
    mi->master_log_name[0] = 0;
1478
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
1479
  }
1480

1481
  if (lex_mi->log_file_name)
1482
    strmake(mi->master_log_name, lex_mi->log_file_name,
1483
	    sizeof(mi->master_log_name)-1);
1484
  if (lex_mi->pos)
unknown's avatar
unknown committed
1485
  {
1486
    mi->master_log_pos= lex_mi->pos;
unknown's avatar
unknown committed
1487
  }
unknown's avatar
unknown committed
1488
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
1489 1490

  if (lex_mi->host)
1491
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
1492
  if (lex_mi->user)
1493
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
1494
  if (lex_mi->password)
1495
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
1496
  if (lex_mi->port)
1497
    mi->port = lex_mi->port;
1498
  if (lex_mi->connect_retry)
1499
    mi->connect_retry = lex_mi->connect_retry;
Andrei Elkin's avatar
Andrei Elkin committed
1500 1501 1502 1503 1504 1505
  if (lex_mi->heartbeat_opt != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->heartbeat_period = lex_mi->heartbeat_period;
  else
    mi->heartbeat_period= (float) min(SLAVE_MAX_HEARTBEAT_PERIOD,
                                      (slave_net_timeout/2.0));
  mi->received_heartbeats= LL(0); // counter lives until master is CHANGEd
Andrei Elkin's avatar
Andrei Elkin committed
1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517
  /*
    reset the last time server_id list if the current CHANGE MASTER 
    is mentioning IGNORE_SERVER_IDS= (...)
  */
  if (lex_mi->repl_ignore_server_ids_opt == LEX_MASTER_INFO::LEX_MI_ENABLE)
    reset_dynamic(&mi->ignore_server_ids);
  for (uint i= 0; i < lex_mi->repl_ignore_server_ids.elements; i++)
  {
    ulong s_id;
    get_dynamic(&lex_mi->repl_ignore_server_ids, (uchar*) &s_id, i);
    if (s_id == ::server_id && replicate_same_server_id)
    {
Guilhem Bichot's avatar
Guilhem Bichot committed
1518
      my_error(ER_SLAVE_IGNORE_SERVER_IDS, MYF(0), static_cast<int>(s_id));
Andrei Elkin's avatar
Andrei Elkin committed
1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532
      ret= TRUE;
      goto err;
    }
    else
    {
      if (bsearch((const ulong *) &s_id,
                  mi->ignore_server_ids.buffer,
                  mi->ignore_server_ids.elements, sizeof(ulong),
                  (int (*) (const void*, const void*))
                  change_master_server_id_cmp) == NULL)
        insert_dynamic(&mi->ignore_server_ids, (uchar*) &s_id);
    }
  }
  sort_dynamic(&mi->ignore_server_ids, (qsort_cmp) change_master_server_id_cmp);
unknown's avatar
foo1  
unknown committed
1533

Andrei Elkin's avatar
Andrei Elkin committed
1534 1535
  if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE);
1536

Andrei Elkin's avatar
Andrei Elkin committed
1537
  if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
1538
    mi->ssl_verify_server_cert=
Andrei Elkin's avatar
Andrei Elkin committed
1539
      (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::LEX_MI_ENABLE);
1540

unknown's avatar
unknown committed
1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552
  if (lex_mi->ssl_ca)
    strmake(mi->ssl_ca, lex_mi->ssl_ca, sizeof(mi->ssl_ca)-1);
  if (lex_mi->ssl_capath)
    strmake(mi->ssl_capath, lex_mi->ssl_capath, sizeof(mi->ssl_capath)-1);
  if (lex_mi->ssl_cert)
    strmake(mi->ssl_cert, lex_mi->ssl_cert, sizeof(mi->ssl_cert)-1);
  if (lex_mi->ssl_cipher)
    strmake(mi->ssl_cipher, lex_mi->ssl_cipher, sizeof(mi->ssl_cipher)-1);
  if (lex_mi->ssl_key)
    strmake(mi->ssl_key, lex_mi->ssl_key, sizeof(mi->ssl_key)-1);
#ifndef HAVE_OPENSSL
  if (lex_mi->ssl || lex_mi->ssl_ca || lex_mi->ssl_capath ||
1553 1554
      lex_mi->ssl_cert || lex_mi->ssl_cipher || lex_mi->ssl_key ||
      lex_mi->ssl_verify_server_cert )
unknown's avatar
foo1  
unknown committed
1555
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
unknown's avatar
unknown committed
1556 1557
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
#endif
1558

1559 1560
  if (lex_mi->relay_log_name)
  {
1561
    need_relay_log_purge= 0;
1562 1563 1564
    char relay_log_name[FN_REFLEN];
    mi->rli.relay_log.make_log_name(relay_log_name, lex_mi->relay_log_name);
    strmake(mi->rli.group_relay_log_name, relay_log_name,
1565
	    sizeof(mi->rli.group_relay_log_name)-1);
1566
    strmake(mi->rli.event_relay_log_name, relay_log_name,
1567
	    sizeof(mi->rli.event_relay_log_name)-1);
1568 1569 1570 1571
  }

  if (lex_mi->relay_log_pos)
  {
1572
    need_relay_log_purge= 0;
1573
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
1574 1575
  }

unknown's avatar
unknown committed
1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600
  /*
    If user did specify neither host nor port nor any log name nor any log
    pos, i.e. he specified only user/password/master_connect_retry, he probably
    wants replication to resume from where it had left, i.e. from the
    coordinates of the **SQL** thread (imagine the case where the I/O is ahead
    of the SQL; restarting from the coordinates of the I/O would lose some
    events which is probably unwanted when you are just doing minor changes
    like changing master_connect_retry).
    A side-effect is that if only the I/O thread was started, this thread may
    restart from ''/4 after the CHANGE MASTER. That's a minor problem (it is a
    much more unlikely situation than the one we are fixing here).
    Note: coordinates of the SQL thread must be read here, before the
    'if (need_relay_log_purge)' block which resets them.
  */
  if (!lex_mi->host && !lex_mi->port &&
      !lex_mi->log_file_name && !lex_mi->pos &&
      need_relay_log_purge)
   {
     /*
       Sometimes mi->rli.master_log_pos == 0 (it happens when the SQL thread is
       not initialized), so we use a max().
       What happens to mi->rli.master_log_pos during the initialization stages
       of replication is not 100% clear, so we guard against problems using
       max().
      */
unknown's avatar
unknown committed
1601 1602 1603
     mi->master_log_pos = max(BIN_LOG_HEADER_SIZE,
			      mi->rli.group_master_log_pos);
     strmake(mi->master_log_name, mi->rli.group_master_log_name,
unknown's avatar
unknown committed
1604 1605
             sizeof(mi->master_log_name)-1);
  }
unknown's avatar
unknown committed
1606 1607 1608 1609
  /*
    Relay log's IO_CACHE may not be inited, if rli->inited==0 (server was never
    a slave before).
  */
1610
  if (flush_master_info(mi, FALSE, FALSE))
1611 1612
  {
    my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
Andrei Elkin's avatar
Andrei Elkin committed
1613 1614
    ret= TRUE;
    goto err;
1615
  }
1616
  if (need_relay_log_purge)
1617
  {
1618
    relay_log_purge= 1;
1619
    thd_proc_info(thd, "Purging old relay logs");
1620 1621
    if (purge_relay_logs(&mi->rli, thd,
			 0 /* not only reset, but also reinit */,
1622 1623
			 &errmsg))
    {
1624
      my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
Andrei Elkin's avatar
Andrei Elkin committed
1625 1626
      ret= TRUE;
      goto err;
1627 1628 1629 1630 1631
    }
  }
  else
  {
    const char* msg;
1632
    relay_log_purge= 0;
1633 1634
    /* Relay log is already initialized */
    if (init_relay_log_pos(&mi->rli,
1635 1636
			   mi->rli.group_relay_log_name,
			   mi->rli.group_relay_log_pos,
1637
			   0 /*no data lock*/,
1638
			   &msg, 0))
1639
    {
1640
      my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
Andrei Elkin's avatar
Andrei Elkin committed
1641 1642
      ret= TRUE;
      goto err;
1643
    }
1644
  }
unknown's avatar
unknown committed
1645 1646 1647 1648 1649 1650 1651 1652 1653 1654
  /*
    Coordinates in rli were spoilt by the 'if (need_relay_log_purge)' block,
    so restore them to good values. If we left them to ''/0, that would work;
    but that would fail in the case of 2 successive CHANGE MASTER (without a
    START SLAVE in between): because first one would set the coords in mi to
    the good values of those in rli, the set those in rli to ''/0, then
    second CHANGE MASTER would set the coords in mi to those of rli, i.e. to
    ''/0: we have lost all copies of the original good coordinates.
    That's why we always save good coords in rli.
  */
unknown's avatar
unknown committed
1655
  mi->rli.group_master_log_pos= mi->master_log_pos;
1656
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
1657 1658
  strmake(mi->rli.group_master_log_name,mi->master_log_name,
	  sizeof(mi->rli.group_master_log_name)-1);
unknown's avatar
unknown committed
1659

1660 1661
  if (!mi->rli.group_master_log_name[0]) // uninitialized case
    mi->rli.group_master_log_pos=0;
1662

Marc Alff's avatar
Marc Alff committed
1663
  mysql_mutex_lock(&mi->rli.data_lock);
1664
  mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
1665
  /* Clear the errors, for a clean start */
1666
  mi->rli.clear_error();
1667
  mi->rli.clear_until_condition();
1668 1669 1670 1671 1672 1673 1674 1675 1676

  sql_print_information("'CHANGE MASTER TO executed'. "
    "Previous state master_host='%s', master_port='%u', master_log_file='%s', "
    "master_log_pos='%ld'. "
    "New state master_host='%s', master_port='%u', master_log_file='%s', "
    "master_log_pos='%ld'.", saved_host, saved_port, saved_log_name,
    (ulong) saved_log_pos, mi->host, mi->port, mi->master_log_name,
    (ulong) mi->master_log_pos);

unknown's avatar
unknown committed
1677 1678 1679 1680 1681 1682
  /*
    If we don't write new coordinates to disk now, then old will remain in
    relay-log.info until START SLAVE is issued; but if mysqld is shutdown
    before START SLAVE, then old will remain in relay-log.info, and will be the
    in-memory value at restart (thus causing errors, as the old relay log does
    not exist anymore).
1683
  */
unknown's avatar
unknown committed
1684
  flush_relay_log_info(&mi->rli);
Marc Alff's avatar
Marc Alff committed
1685 1686
  mysql_cond_broadcast(&mi->data_cond);
  mysql_mutex_unlock(&mi->rli.data_lock);
1687

Andrei Elkin's avatar
Andrei Elkin committed
1688
err:
1689
  unlock_slave_threads(mi);
1690
  thd_proc_info(thd, 0);
Andrei Elkin's avatar
Andrei Elkin committed
1691 1692 1693
  if (ret == FALSE)
    my_ok(thd);
  DBUG_RETURN(ret);
1694 1695
}

1696 1697 1698 1699 1700 1701 1702 1703 1704 1705

/**
  Execute a RESET MASTER statement.

  @param thd Pointer to THD object of the client thread executing the
  statement.

  @retval 0 success
  @retval 1 error
*/
1706
int reset_master(THD* thd)
1707
{
1708
  if (!mysql_bin_log.is_open())
1709
  {
unknown's avatar
unknown committed
1710 1711
    my_message(ER_FLUSH_MASTER_BINLOG_CLOSED,
               ER(ER_FLUSH_MASTER_BINLOG_CLOSED), MYF(ME_BELL+ME_WAITTANG));
1712
    return 1;
1713
  }
He Zhenxing's avatar
He Zhenxing committed
1714 1715 1716 1717 1718

  if (mysql_bin_log.reset_logs(thd))
    return 1;
  RUN_HOOK(binlog_transmit, after_reset_master, (thd, 0 /* flags */));
  return 0;
1719 1720
}

unknown's avatar
unknown committed
1721

1722 1723 1724 1725 1726 1727 1728 1729 1730
/**
  Execute a SHOW BINLOG EVENTS statement.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @retval FALSE success
  @retval TRUE failure
*/
1731
bool mysql_show_binlog_events(THD* thd)
unknown's avatar
unknown committed
1732
{
1733
  Protocol *protocol= thd->protocol;
unknown's avatar
unknown committed
1734
  List<Item> field_list;
1735
  const char *errmsg = 0;
1736
  bool ret = TRUE;
unknown's avatar
unknown committed
1737 1738
  IO_CACHE log;
  File file = -1;
1739
  MYSQL_BIN_LOG *binary_log= NULL;
1740
  int old_max_allowed_packet= thd->variables.max_allowed_packet;
1741 1742
  LOG_INFO linfo;

1743
  DBUG_ENTER("mysql_show_binlog_events");
1744 1745

  Log_event::init_show_field_list(&field_list);
1746
  if (protocol->send_result_set_metadata(&field_list,
1747
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1748
    DBUG_RETURN(TRUE);
1749

1750 1751 1752
  Format_description_log_event *description_event= new
    Format_description_log_event(3); /* MySQL 4.0 by default */

1753 1754
  DBUG_ASSERT(thd->lex->sql_command == SQLCOM_SHOW_BINLOG_EVENTS ||
              thd->lex->sql_command == SQLCOM_SHOW_RELAYLOG_EVENTS);
unknown's avatar
unknown committed
1755

1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774
  /* select wich binary log to use: binlog or relay */
  if ( thd->lex->sql_command == SQLCOM_SHOW_BINLOG_EVENTS )
  {
    /*
      Wait for handlers to insert any pending information
      into the binlog.  For e.g. ndb which updates the binlog asynchronously
      this is needed so that the uses sees all its own commands in the binlog
    */
    ha_binlog_wait(thd);

    binary_log= &mysql_bin_log;
  }
  else  /* showing relay log contents */
  {
    if (!active_mi)
      DBUG_RETURN(TRUE);

    binary_log= &(active_mi->rli.relay_log);
  }
unknown's avatar
unknown committed
1775

1776
  if (binary_log->is_open())
unknown's avatar
unknown committed
1777
  {
1778
    LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
1779
    SELECT_LEX_UNIT *unit= &thd->lex->unit;
1780
    ha_rows event_count, limit_start, limit_end;
1781
    my_off_t pos = max(BIN_LOG_HEADER_SIZE, lex_mi->pos); // user-friendly
1782 1783
    char search_file_name[FN_REFLEN], *name;
    const char *log_file_name = lex_mi->log_file_name;
Marc Alff's avatar
Marc Alff committed
1784
    mysql_mutex_t *log_lock = binary_log->get_log_lock();
1785
    Log_event* ev;
1786

1787 1788 1789
    unit->set_limit(thd->lex->current_select);
    limit_start= unit->offset_limit_cnt;
    limit_end= unit->select_limit_cnt;
unknown's avatar
unknown committed
1790

1791
    name= search_file_name;
unknown's avatar
unknown committed
1792
    if (log_file_name)
1793
      binary_log->make_log_name(search_file_name, log_file_name);
unknown's avatar
unknown committed
1794
    else
1795
      name=0;					// Find first log
unknown's avatar
unknown committed
1796 1797

    linfo.index_file_offset = 0;
1798

1799
    if (binary_log->find_log_pos(&linfo, name, 1))
unknown's avatar
unknown committed
1800 1801 1802 1803 1804
    {
      errmsg = "Could not find target log";
      goto err;
    }

Marc Alff's avatar
Marc Alff committed
1805
    mysql_mutex_lock(&LOCK_thread_count);
1806
    thd->current_linfo = &linfo;
Marc Alff's avatar
Marc Alff committed
1807
    mysql_mutex_unlock(&LOCK_thread_count);
1808

unknown's avatar
unknown committed
1809 1810 1811
    if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0)
      goto err;

1812 1813 1814 1815 1816
    /*
      to account binlog event header size
    */
    thd->variables.max_allowed_packet += MAX_LOG_EVENT_HEADER;

Marc Alff's avatar
Marc Alff committed
1817
    mysql_mutex_lock(log_lock);
1818

1819
    /*
1820 1821 1822 1823 1824 1825
      open_binlog() sought to position 4.
      Read the first event in case it's a Format_description_log_event, to
      know the format. If there's no such event, we are 3.23 or 4.x. This
      code, like before, can't read 3.23 binlogs.
      This code will fail on a mixed relay log (one which has Format_desc then
      Rotate then Format_desc).
1826
    */
Marc Alff's avatar
Marc Alff committed
1827
    ev= Log_event::read_log_event(&log, (mysql_mutex_t*)0, description_event);
1828 1829 1830
    if (ev)
    {
      if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
1831 1832
      {
        delete description_event;
1833
        description_event= (Format_description_log_event*) ev;
1834
      }
1835 1836 1837 1838
      else
        delete ev;
    }

unknown's avatar
unknown committed
1839 1840
    my_b_seek(&log, pos);

1841 1842 1843 1844 1845
    if (!description_event->is_valid())
    {
      errmsg="Invalid Format_description event; could be out of memory";
      goto err;
    }
1846

1847
    for (event_count = 0;
Marc Alff's avatar
Marc Alff committed
1848
         (ev = Log_event::read_log_event(&log, (mysql_mutex_t*) 0,
1849
                                         description_event)); )
unknown's avatar
unknown committed
1850 1851
    {
      if (event_count >= limit_start &&
1852
	  ev->net_send(protocol, linfo.log_file_name, pos))
1853 1854 1855
      {
	errmsg = "Net error";
	delete ev;
Marc Alff's avatar
Marc Alff committed
1856
        mysql_mutex_unlock(log_lock);
1857 1858 1859 1860
	goto err;
      }

      pos = my_b_tell(&log);
unknown's avatar
unknown committed
1861 1862 1863 1864 1865 1866 1867 1868 1869
      delete ev;

      if (++event_count >= limit_end)
	break;
    }

    if (event_count < limit_end && log.error)
    {
      errmsg = "Wrong offset or I/O error";
Marc Alff's avatar
Marc Alff committed
1870
      mysql_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1871 1872
      goto err;
    }
1873

Marc Alff's avatar
Marc Alff committed
1874
    mysql_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1875
  }
1876 1877
  // Check that linfo is still on the function scope.
  DEBUG_SYNC(thd, "after_show_binlog_events");
unknown's avatar
unknown committed
1878

1879 1880
  ret= FALSE;

unknown's avatar
unknown committed
1881
err:
1882
  delete description_event;
unknown's avatar
unknown committed
1883 1884 1885
  if (file >= 0)
  {
    end_io_cache(&log);
Marc Alff's avatar
Marc Alff committed
1886
    mysql_file_close(file, MYF(MY_WME));
unknown's avatar
unknown committed
1887
  }
1888

unknown's avatar
unknown committed
1889
  if (errmsg)
1890 1891
    my_error(ER_ERROR_WHEN_EXECUTING_COMMAND, MYF(0),
             "SHOW BINLOG EVENTS", errmsg);
1892
  else
1893
    my_eof(thd);
unknown's avatar
unknown committed
1894

Marc Alff's avatar
Marc Alff committed
1895
  mysql_mutex_lock(&LOCK_thread_count);
1896
  thd->current_linfo = 0;
Marc Alff's avatar
Marc Alff committed
1897
  mysql_mutex_unlock(&LOCK_thread_count);
1898
  thd->variables.max_allowed_packet= old_max_allowed_packet;
1899
  DBUG_RETURN(ret);
unknown's avatar
unknown committed
1900 1901
}

unknown's avatar
unknown committed
1902

1903 1904 1905 1906 1907 1908 1909 1910 1911
/**
  Execute a SHOW MASTER STATUS statement.

  @param thd Pointer to THD object for the client thread executing the
  statement.

  @retval FALSE success
  @retval TRUE failure
*/
unknown's avatar
unknown committed
1912
bool show_binlog_info(THD* thd)
1913
{
1914
  Protocol *protocol= thd->protocol;
1915 1916 1917
  DBUG_ENTER("show_binlog_info");
  List<Item> field_list;
  field_list.push_back(new Item_empty_string("File", FN_REFLEN));
1918 1919
  field_list.push_back(new Item_return_int("Position",20,
					   MYSQL_TYPE_LONGLONG));
1920 1921
  field_list.push_back(new Item_empty_string("Binlog_Do_DB",255));
  field_list.push_back(new Item_empty_string("Binlog_Ignore_DB",255));
1922

1923
  if (protocol->send_result_set_metadata(&field_list,
1924
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1925
    DBUG_RETURN(TRUE);
1926 1927
  protocol->prepare_for_resend();

1928 1929 1930 1931 1932
  if (mysql_bin_log.is_open())
  {
    LOG_INFO li;
    mysql_bin_log.get_current_log(&li);
    int dir_len = dirname_length(li.log_file_name);
1933
    protocol->store(li.log_file_name + dir_len, &my_charset_bin);
1934
    protocol->store((ulonglong) li.pos);
1935 1936
    protocol->store(binlog_filter->get_do_db());
    protocol->store(binlog_filter->get_ignore_db());
1937
    if (protocol->write())
unknown's avatar
unknown committed
1938
      DBUG_RETURN(TRUE);
1939
  }
1940
  my_eof(thd);
unknown's avatar
unknown committed
1941
  DBUG_RETURN(FALSE);
1942
}
unknown's avatar
unknown committed
1943

1944

1945 1946
/**
  Execute a SHOW BINARY LOGS statement.
1947

1948 1949
  @param thd Pointer to THD object for the client thread executing the
  statement.
1950

1951 1952
  @retval FALSE success
  @retval TRUE failure
1953
*/
unknown's avatar
unknown committed
1954
bool show_binlogs(THD* thd)
unknown's avatar
unknown committed
1955
{
1956
  IO_CACHE *index_file;
1957 1958
  LOG_INFO cur;
  File file;
unknown's avatar
unknown committed
1959 1960
  char fname[FN_REFLEN];
  List<Item> field_list;
1961
  uint length;
unknown's avatar
unknown committed
1962
  int cur_dir_len;
1963 1964
  Protocol *protocol= thd->protocol;
  DBUG_ENTER("show_binlogs");
1965 1966

  if (!mysql_bin_log.is_open())
1967
  {
Luis Soares's avatar
Luis Soares committed
1968
    my_error(ER_NO_BINARY_LOGGING, MYF(0));
1969
    DBUG_RETURN(TRUE);
1970
  }
unknown's avatar
unknown committed
1971

1972
  field_list.push_back(new Item_empty_string("Log_name", 255));
unknown's avatar
foo1  
unknown committed
1973
  field_list.push_back(new Item_return_int("File_size", 20,
1974
                                           MYSQL_TYPE_LONGLONG));
1975
  if (protocol->send_result_set_metadata(&field_list,
1976
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1977
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1978
  
Marc Alff's avatar
Marc Alff committed
1979
  mysql_mutex_lock(mysql_bin_log.get_log_lock());
unknown's avatar
unknown committed
1980
  mysql_bin_log.lock_index();
1981
  index_file=mysql_bin_log.get_index_file();
unknown's avatar
unknown committed
1982 1983
  
  mysql_bin_log.raw_get_current_log(&cur); // dont take mutex
Marc Alff's avatar
Marc Alff committed
1984
  mysql_mutex_unlock(mysql_bin_log.get_log_lock()); // lockdep, OK
unknown's avatar
unknown committed
1985
  
unknown's avatar
unknown committed
1986
  cur_dir_len= dirname_length(cur.log_file_name);
1987

1988 1989 1990 1991
  reinit_io_cache(index_file, READ_CACHE, (my_off_t) 0, 0, 0);

  /* The file ends with EOF or empty line */
  while ((length=my_b_gets(index_file, fname, sizeof(fname))) > 1)
1992
  {
unknown's avatar
unknown committed
1993 1994 1995
    int dir_len;
    ulonglong file_length= 0;                   // Length if open fails
    fname[--length] = '\0';                     // remove the newline
1996

1997
    protocol->prepare_for_resend();
unknown's avatar
unknown committed
1998 1999 2000 2001 2002 2003 2004
    dir_len= dirname_length(fname);
    length-= dir_len;
    protocol->store(fname + dir_len, length, &my_charset_bin);

    if (!(strncmp(fname+dir_len, cur.log_file_name+cur_dir_len, length)))
      file_length= cur.pos;  /* The active log, use the active position */
    else
2005 2006
    {
      /* this is an old log, open it and find the size */
Marc Alff's avatar
Marc Alff committed
2007 2008 2009
      if ((file= mysql_file_open(key_file_binlog,
                                 fname, O_RDONLY | O_SHARE | O_BINARY,
                                 MYF(0))) >= 0)
2010
      {
Marc Alff's avatar
Marc Alff committed
2011 2012
        file_length= (ulonglong) mysql_file_seek(file, 0L, MY_SEEK_END, MYF(0));
        mysql_file_close(file, MYF(0));
2013 2014
      }
    }
unknown's avatar
unknown committed
2015
    protocol->store(file_length);
2016
    if (protocol->write())
2017
      goto err;
2018
  }
unknown's avatar
unknown committed
2019
  mysql_bin_log.unlock_index();
2020
  my_eof(thd);
unknown's avatar
unknown committed
2021
  DBUG_RETURN(FALSE);
2022

2023 2024
err:
  mysql_bin_log.unlock_index();
unknown's avatar
unknown committed
2025
  DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
2026 2027
}

2028 2029 2030 2031 2032 2033 2034
/**
   Load data's io cache specific hook to be executed
   before a chunk of data is being read into the cache's buffer
   The fuction instantianates and writes into the binlog
   replication events along LOAD DATA processing.
   
   @param file  pointer to io-cache
He Zhenxing's avatar
He Zhenxing committed
2035 2036
   @retval 0 success
   @retval 1 failure
2037
*/
2038 2039
int log_loaded_block(IO_CACHE* file)
{
2040
  DBUG_ENTER("log_loaded_block");
2041
  LOAD_FILE_INFO *lf_info;
2042 2043
  uint block_len;
  /* buffer contains position where we started last read */
2044
  uchar* buffer= (uchar*) my_b_get_buffer_start(file);
2045 2046
  uint max_event_size= current_thd->variables.max_allowed_packet;
  lf_info= (LOAD_FILE_INFO*) file->arg;
2047
  if (lf_info->thd->is_current_stmt_binlog_format_row())
2048
    DBUG_RETURN(0);
2049
  if (lf_info->last_pos_in_file != HA_POS_ERROR &&
2050
      lf_info->last_pos_in_file >= my_b_get_pos_in_file(file))
2051
    DBUG_RETURN(0);
2052
  
2053
  for (block_len= (uint) (my_b_get_bytes_in_buffer(file)); block_len > 0;
2054 2055
       buffer += min(block_len, max_event_size),
       block_len -= min(block_len, max_event_size))
2056
  {
2057 2058 2059 2060 2061 2062
    lf_info->last_pos_in_file= my_b_get_pos_in_file(file);
    if (lf_info->wrote_create_file)
    {
      Append_block_log_event a(lf_info->thd, lf_info->thd->db, buffer,
                               min(block_len, max_event_size),
                               lf_info->log_delayed);
He Zhenxing's avatar
He Zhenxing committed
2063 2064
      if (mysql_bin_log.write(&a))
        DBUG_RETURN(1);
2065 2066 2067 2068 2069 2070 2071
    }
    else
    {
      Begin_load_query_log_event b(lf_info->thd, lf_info->thd->db,
                                   buffer,
                                   min(block_len, max_event_size),
                                   lf_info->log_delayed);
He Zhenxing's avatar
He Zhenxing committed
2072 2073
      if (mysql_bin_log.write(&b))
        DBUG_RETURN(1);
2074 2075
      lf_info->wrote_create_file= 1;
    }
2076
  }
2077
  DBUG_RETURN(0);
2078
}
2079

unknown's avatar
SCRUM  
unknown committed
2080
#endif /* HAVE_REPLICATION */