sql_repl.cc 54.5 KB
Newer Older
unknown's avatar
unknown committed
1
/* Copyright (C) 2000-2006 MySQL AB & Sasha
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 14 15 16

   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
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include "mysql_priv.h"
unknown's avatar
SCRUM  
unknown committed
17 18
#ifdef HAVE_REPLICATION

19
#include "rpl_mi.h"
20 21
#include "sql_repl.h"
#include "log_event.h"
unknown's avatar
unknown committed
22
#include "rpl_filter.h"
unknown's avatar
unknown committed
23
#include <my_dir.h>
24

25
int max_binlog_dump_events = 0; // unlimited
26
my_bool opt_sporadic_binlog_dump_fail = 0;
unknown's avatar
unknown committed
27
#ifndef DBUG_OFF
28
static int binlog_dump_count = 0;
unknown's avatar
unknown committed
29
#endif
unknown's avatar
unknown committed
30

31
/*
32 33 34 35
    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).
36 37
    < 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
38 39 40 41 42
    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).
43 44 45 46

    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.
47 48
*/

unknown's avatar
unknown committed
49
static int fake_rotate_event(NET* net, String* packet, char* log_file_name,
50
                             ulonglong position, const char** errmsg)
unknown's avatar
unknown committed
51
{
52
  DBUG_ENTER("fake_rotate_event");
53 54 55 56 57 58
  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
59 60
  header[EVENT_TYPE_OFFSET] = ROTATE_EVENT;

61
  char* p = log_file_name+dirname_length(log_file_name);
unknown's avatar
unknown committed
62
  uint ident_len = (uint) strlen(p);
63
  ulong event_len = ident_len + LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN;
64
  int4store(header + SERVER_ID_OFFSET, server_id);
unknown's avatar
unknown committed
65
  int4store(header + EVENT_LEN_OFFSET, event_len);
66
  int2store(header + FLAGS_OFFSET, LOG_EVENT_ARTIFICIAL_F);
67

68 69
  // TODO: check what problems this may cause and fix them
  int4store(header + LOG_POS_OFFSET, 0);
70

unknown's avatar
unknown committed
71
  packet->append(header, sizeof(header));
72
  int8store(buf+R_POS_OFFSET,position);
73
  packet->append(buf, ROTATE_HEADER_LEN);
unknown's avatar
unknown committed
74
  packet->append(p,ident_len);
75
  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
76 77
  {
    *errmsg = "failed on my_net_write()";
78
    DBUG_RETURN(-1);
79
  }
80
  DBUG_RETURN(0);
unknown's avatar
unknown committed
81 82
}

83 84 85
static int send_file(THD *thd)
{
  NET* net = &thd->net;
86 87
  int fd = -1, error = 1;
  size_t bytes;
88 89
  char fname[FN_REFLEN+1];
  const char *errmsg = 0;
unknown's avatar
unknown committed
90
  int old_timeout;
unknown's avatar
unknown committed
91
  unsigned long packet_len;
92
  uchar buf[IO_SIZE];				// It's safe to alloc this
93 94
  DBUG_ENTER("send_file");

unknown's avatar
unknown committed
95 96 97 98
  /*
    The client might be slow loading the data, give him wait_timeout to do
    the job
  */
99
  old_timeout= net->read_timeout;
100
  my_net_set_read_timeout(net, thd->variables.net_wait_timeout);
unknown's avatar
unknown committed
101 102 103 104 105

  /*
    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
106
  if (net_flush(net) || (packet_len = my_net_read(net)) == packet_error)
107
  {
108
    errmsg = "while reading file name";
109 110 111
    goto err;
  }

112 113 114
  // terminate with \0 for fn_format
  *((char*)net->read_pos +  packet_len) = 0;
  fn_format(fname, (char*) net->read_pos + 1, "", "", 4);
115 116 117 118
  // this is needed to make replicate-ignore-db
  if (!strcmp(fname,"/dev/null"))
    goto end;

119
  if ((fd = my_open(fname, O_RDONLY, MYF(0))) < 0)
120
  {
121
    errmsg = "on open of file";
122 123 124
    goto err;
  }

125
  while ((long) (bytes= my_read(fd, buf, IO_SIZE, MYF(0))) > 0)
126 127 128
  {
    if (my_net_write(net, buf, bytes))
    {
129
      errmsg = "while writing data to client";
130 131 132 133 134
      goto err;
    }
  }

 end:
135
  if (my_net_write(net, (uchar*) "", 0) || net_flush(net) ||
136 137
      (my_net_read(net) == packet_error))
  {
138
    errmsg = "while negotiating file transfer close";
139 140 141 142 143
    goto err;
  }
  error = 0;

 err:
144
  my_net_set_read_timeout(net, old_timeout);
145 146
  if (fd >= 0)
    (void) my_close(fd, MYF(0));
147 148
  if (errmsg)
  {
149
    sql_print_error("Failed in send_file() %s", errmsg);
Staale Smedseng's avatar
Staale Smedseng committed
150
    DBUG_PRINT("error", ("%s", errmsg));
151 152 153 154
  }
  DBUG_RETURN(error);
}

155

156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176
/*
  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
177 178 179
void adjust_linfo_offsets(my_off_t purge_offset)
{
  THD *tmp;
180

unknown's avatar
unknown committed
181 182 183
  pthread_mutex_lock(&LOCK_thread_count);
  I_List_iterator<THD> it(threads);

184 185 186 187 188 189
  while ((tmp=it++))
  {
    LOG_INFO* linfo;
    if ((linfo = tmp->current_linfo))
    {
      pthread_mutex_lock(&linfo->lock);
190 191 192 193
      /*
	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
194
      */
195 196 197 198 199 200 201
      if (linfo->index_file_offset < purge_offset)
	linfo->fatal = (linfo->index_file_offset != 0);
      else
	linfo->index_file_offset -= purge_offset;
      pthread_mutex_unlock(&linfo->lock);
    }
  }
unknown's avatar
unknown committed
202 203 204
  pthread_mutex_unlock(&LOCK_thread_count);
}

unknown's avatar
merge  
unknown committed
205

unknown's avatar
unknown committed
206 207
bool log_in_use(const char* log_name)
{
208
  size_t log_name_len = strlen(log_name) + 1;
unknown's avatar
unknown committed
209 210
  THD *tmp;
  bool result = 0;
211

unknown's avatar
unknown committed
212 213
  pthread_mutex_lock(&LOCK_thread_count);
  I_List_iterator<THD> it(threads);
214 215 216 217 218

  while ((tmp=it++))
  {
    LOG_INFO* linfo;
    if ((linfo = tmp->current_linfo))
unknown's avatar
unknown committed
219
    {
220
      pthread_mutex_lock(&linfo->lock);
221 222
      result = !bcmp((uchar*) log_name, (uchar*) linfo->log_file_name,
                     log_name_len);
223
      pthread_mutex_unlock(&linfo->lock);
224 225
      if (result)
	break;
226 227
    }
  }
unknown's avatar
unknown committed
228 229 230 231 232

  pthread_mutex_unlock(&LOCK_thread_count);
  return result;
}

unknown's avatar
unknown committed
233
bool purge_error_message(THD* thd, int res)
unknown's avatar
unknown committed
234
{
unknown's avatar
unknown committed
235
  uint errmsg= 0;
236

unknown's avatar
unknown committed
237
  switch (res)  {
238
  case 0: break;
unknown's avatar
unknown committed
239 240 241 242 243 244 245
  case LOG_INFO_EOF:	errmsg= ER_UNKNOWN_TARGET_BINLOG; break;
  case LOG_INFO_IO:	errmsg= ER_IO_ERR_LOG_INDEX_READ; break;
  case LOG_INFO_INVALID:errmsg= ER_BINLOG_PURGE_PROHIBITED; break;
  case LOG_INFO_SEEK:	errmsg= ER_FSEEK_FAIL; break;
  case LOG_INFO_MEM:	errmsg= ER_OUT_OF_RESOURCES; break;
  case LOG_INFO_FATAL:	errmsg= ER_BINLOG_PURGE_FATAL_ERR; break;
  case LOG_INFO_IN_USE: errmsg= ER_LOG_IN_USE; break;
246
  case LOG_INFO_EMFILE: errmsg= ER_BINLOG_PURGE_EMFILE; break;
unknown's avatar
unknown committed
247
  default:		errmsg= ER_LOG_PURGE_UNKNOWN_ERR; break;
248 249 250 251
  }

  if (errmsg)
  {
unknown's avatar
unknown committed
252
    my_message(errmsg, ER(errmsg), MYF(0));
unknown's avatar
unknown committed
253
    return TRUE;
254
  }
255
  my_ok(thd);
unknown's avatar
unknown committed
256
  return FALSE;
unknown's avatar
unknown committed
257 258
}

unknown's avatar
unknown committed
259

260 261 262 263 264 265 266 267 268 269 270
/**
  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
271
bool purge_master_logs(THD* thd, const char* to_log)
272 273
{
  char search_file_name[FN_REFLEN];
unknown's avatar
unknown committed
274 275
  if (!mysql_bin_log.is_open())
  {
276
    my_ok(thd);
unknown's avatar
unknown committed
277
    return FALSE;
unknown's avatar
unknown committed
278
  }
279 280

  mysql_bin_log.make_log_name(search_file_name, to_log);
unknown's avatar
unknown committed
281 282
  return purge_error_message(thd,
			     mysql_bin_log.purge_logs(search_file_name, 0, 1,
unknown's avatar
unknown committed
283
						      1, NULL));
284 285 286
}


287 288 289 290 291 292 293 294 295 296 297
/**
  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
298
bool purge_master_logs_before_date(THD* thd, time_t purge_time)
299
{
300 301
  if (!mysql_bin_log.is_open())
  {
302
    my_ok(thd);
303 304 305 306
    return 0;
  }
  return purge_error_message(thd,
                             mysql_bin_log.purge_logs_before_date(purge_time));
307 308
}

309
int test_for_non_eof_log_read_errors(int error, const char **errmsg)
310
{
311
  if (error == LOG_READ_EOF)
312 313 314 315
    return 0;
  my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
  switch (error) {
  case LOG_READ_BOGUS:
316
    *errmsg = "bogus data in log event";
317 318
    break;
  case LOG_READ_TOO_LARGE:
319
    *errmsg = "log event entry exceeded max_allowed_packet; \
320 321 322
Increase max_allowed_packet on master";
    break;
  case LOG_READ_IO:
323
    *errmsg = "I/O error reading log event";
324 325
    break;
  case LOG_READ_MEM:
326
    *errmsg = "memory allocation failed reading log event";
327 328
    break;
  case LOG_READ_TRUNC:
329
    *errmsg = "binlog truncated in the middle of event";
330 331
    break;
  default:
332
    *errmsg = "unknown error reading log event on the master";
333 334 335 336 337 338
    break;
  }
  return error;
}


339 340 341
/*
  TODO: Clean up loop to only have one call to send_file()
*/
unknown's avatar
merge  
unknown committed
342

343 344
void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
		       ushort flags)
345 346 347
{
  LOG_INFO linfo;
  char *log_file_name = linfo.log_file_name;
348
  char search_file_name[FN_REFLEN], *name;
349 350
  IO_CACHE log;
  File file = -1;
351 352 353 354
  String* packet = &thd->packet;
  int error;
  const char *errmsg = "Unknown error";
  NET* net = &thd->net;
355
  pthread_mutex_t *log_lock;
356
  bool binlog_can_be_corrupted= FALSE;
357 358
#ifndef DBUG_OFF
  int left_events = max_binlog_dump_events;
359
#endif
360
  DBUG_ENTER("mysql_binlog_send");
361 362
  DBUG_PRINT("enter",("log_ident: '%s'  pos: %ld", log_ident, (long) pos));

363
  bzero((char*) &log,sizeof(log));
364

unknown's avatar
unknown committed
365
#ifndef DBUG_OFF
366 367 368
  if (opt_sporadic_binlog_dump_fail && (binlog_dump_count++ % 2))
  {
    errmsg = "Master failed COM_BINLOG_DUMP to test if slave can recover";
369
    my_errno= ER_UNKNOWN_ERROR;
370 371
    goto err;
  }
372
#endif
373

374
  if (!mysql_bin_log.is_open())
375 376
  {
    errmsg = "Binary log is not open";
377
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
378 379
    goto err;
  }
380 381 382
  if (!server_id_supplied)
  {
    errmsg = "Misconfigured master - server id was not set";
383
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
384 385 386
    goto err;
  }

387
  name=search_file_name;
388
  if (log_ident[0])
389 390
    mysql_bin_log.make_log_name(search_file_name, log_ident);
  else
391
    name=0;					// Find first log
392

unknown's avatar
unknown committed
393
  linfo.index_file_offset = 0;
394

395
  if (mysql_bin_log.find_log_pos(&linfo, name, 1))
396
  {
397
    errmsg = "Could not find first log file name in binary log index file";
398
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
399 400
    goto err;
  }
401

402 403 404 405
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = &linfo;
  pthread_mutex_unlock(&LOCK_thread_count);

406
  if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)
407 408
  {
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
409
    goto err;
410 411
  }
  if (pos < BIN_LOG_HEADER_SIZE || pos > my_b_filelength(&log))
412
  {
413
    errmsg= "Client requested master to start replication from \
unknown's avatar
unknown committed
414
impossible position";
415
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
416 417
    goto err;
  }
418

419 420
  /*
    We need to start a packet with something other than 255
421
    to distinguish it from error
422
  */
423
  packet->set("\0", 1, &my_charset_bin); /* This is the start of a new packet */
424

425
  /*
426
    Tell the client about the log name with a fake Rotate event;
427 428 429 430 431 432 433 434 435
    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
436
    already knows the binlog's name.
437 438 439 440 441 442 443
    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).
444
    Always calling fake_rotate_event makes sending of normal
445 446 447 448 449 450 451
    (=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.
452
  */
453
  if (fake_rotate_event(net, packet, log_file_name, pos, &errmsg))
unknown's avatar
unknown committed
454
  {
455 456 457 458
    /*
       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.
459
    */
460 461
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
    goto err;
unknown's avatar
unknown committed
462
  }
unknown's avatar
unknown committed
463
  packet->set("\0", 1, &my_charset_bin);
464 465 466 467 468
  /*
    Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become
    this larger than the corresponding packet (query) sent 
    from client to master.
  */
469
  thd->variables.max_allowed_packet+= MAX_LOG_EVENT_HEADER;
unknown's avatar
unknown committed
470

471
  /*
472 473 474
    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).
475
  */
476
  log_lock = mysql_bin_log.get_log_lock();
477
  if (pos > BIN_LOG_HEADER_SIZE)
478 479 480 481 482
  {
     /*
       Try to find a Format_description_log_event at the beginning of
       the binlog
     */
483 484 485
     if (!(error = Log_event::read_log_event(&log, packet, log_lock)))
     {
       /*
486 487
         The packet has offsets equal to the normal offsets in a binlog
         event +1 (the first character is \0).
488 489 490 491 492 493
       */
       DBUG_PRINT("info",
                  ("Looked for a Format_description_log_event, found event type %d",
                   (*packet)[EVENT_TYPE_OFFSET+1]));
       if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
       {
494 495
         binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
                                       LOG_EVENT_BINLOG_IN_USE_F);
unknown's avatar
unknown committed
496
         (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
497 498 499 500 501
         /*
           mark that this event with "log_pos=0", so the slave
           should not increment master's binlog position
           (rli->group_master_log_pos)
         */
502
         int4store((char*) packet->ptr()+LOG_POS_OFFSET+1, 0);
503 504 505 506 507 508
         /*
           if reconnect master sends FD event with `created' as 0
           to avoid destroying temp tables.
          */
         int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+
                   ST_CREATED_OFFSET+1, (ulong) 0);
509
         /* send it */
510
         if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
511 512 513 514 515
         {
           errmsg = "Failed on my_net_write()";
           my_errno= ER_UNKNOWN_ERROR;
           goto err;
         }
unknown's avatar
unknown committed
516

517
         /*
518 519 520 521
           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.
522 523 524 525
         */
       }
     }
     else
526
     {
527
       if (test_for_non_eof_log_read_errors(error, &errmsg))
528
         goto err;
529 530 531 532 533
       /*
         It's EOF, nothing to do, go on reading next events, the
         Format_description_log_event will be found naturally if it is written.
       */
     }
534 535
     /* reset the packet as we wrote to it in any case */
     packet->set("\0", 1, &my_charset_bin);
536 537 538 539 540
  } /* end of if (pos > BIN_LOG_HEADER_SIZE); */
  else
  {
    /* The Format_description_log_event event will be found naturally. */
  }
541

542 543 544
  /* 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
545
  while (!net->error && net->vio != 0 && !thd->killed)
546 547 548
  {
    while (!(error = Log_event::read_log_event(&log, packet, log_lock)))
    {
549
#ifndef DBUG_OFF
550
      if (max_binlog_dump_events && !left_events--)
551 552 553
      {
	net_flush(net);
	errmsg = "Debugging binlog dump abort";
554
	my_errno= ER_UNKNOWN_ERROR;
555 556
	goto err;
      }
557
#endif
558 559

      if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
unknown's avatar
unknown committed
560
      {
561 562
        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
                                      LOG_EVENT_BINLOG_IN_USE_F);
unknown's avatar
unknown committed
563 564
        (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
      }
565 566 567
      else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT)
        binlog_can_be_corrupted= FALSE;

568
      if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
569 570
      {
	errmsg = "Failed on my_net_write()";
571
	my_errno= ER_UNKNOWN_ERROR;
572 573
	goto err;
      }
unknown's avatar
unknown committed
574

575 576
      DBUG_PRINT("info", ("log event code %d",
			  (*packet)[LOG_EVENT_OFFSET+1] ));
unknown's avatar
unknown committed
577
      if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
578
      {
579
	if (send_file(thd))
580
	{
581
	  errmsg = "failed in send_file()";
582
	  my_errno= ER_UNKNOWN_ERROR;
583 584
	  goto err;
	}
585
      }
586
      packet->set("\0", 1, &my_charset_bin);
587
    }
588 589 590 591 592 593 594

    /*
      here we were reading binlog that was not closed properly (as a result
      of a crash ?). treat any corruption as EOF
    */
    if (binlog_can_be_corrupted && error != LOG_READ_MEM)
      error=LOG_READ_EOF;
595 596
    /*
      TODO: now that we are logging the offset, check to make sure
597
      the recorded offset and the actual match.
598 599 600
      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.
601
    */
602
    if (test_for_non_eof_log_read_errors(error, &errmsg))
603
      goto err;
604

605
    if (!(flags & BINLOG_DUMP_NON_BLOCK) &&
606
        mysql_bin_log.is_active(log_file_name))
607
    {
608 609 610
      /*
	Block until there is more data in the log
      */
611
      if (net_flush(net))
612 613
      {
	errmsg = "failed on net_flush()";
614
	my_errno= ER_UNKNOWN_ERROR;
615 616
	goto err;
      }
617

618 619 620 621 622 623
      /*
	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.
      */
624 625 626 627
      {
	log.error=0;
	bool read_packet = 0, fatal_error = 0;

628
#ifndef DBUG_OFF
629
	if (max_binlog_dump_events && !left_events--)
630 631
	{
	  errmsg = "Debugging binlog dump abort";
632
	  my_errno= ER_UNKNOWN_ERROR;
633 634
	  goto err;
	}
635
#endif
636

637 638 639 640
	/*
	  No one will update the log while we are reading
	  now, but we'll be quick and just read one record

641
	  TODO:
642 643 644
          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.
645
	*/
646

unknown's avatar
unknown committed
647
	pthread_mutex_lock(log_lock);
648
	switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*)0)) {
649
	case 0:
650
	  /* we read successfully, so we'll need to send it to the slave */
651
	  pthread_mutex_unlock(log_lock);
652 653
	  read_packet = 1;
	  break;
654

655
	case LOG_READ_EOF:
656
	  DBUG_PRINT("wait",("waiting for data in binary log"));
unknown's avatar
unknown committed
657
	  if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0)
unknown's avatar
unknown committed
658 659 660 661
	  {
	    pthread_mutex_unlock(log_lock);
	    goto end;
	  }
unknown's avatar
unknown committed
662
	  if (!thd->killed)
663 664
	  {
	    /* Note that the following call unlocks lock_log */
665
	    mysql_bin_log.wait_for_update(thd, 0);
666 667 668
	  }
	  else
	    pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
669
	  DBUG_PRINT("wait",("binary log received update"));
670 671 672
	  break;

	default:
673
	  pthread_mutex_unlock(log_lock);
674 675
	  fatal_error = 1;
	  break;
676
	}
677

678
	if (read_packet)
679
	{
680
	  thd_proc_info(thd, "Sending binlog event to slave");
681
	  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) )
682 683
	  {
	    errmsg = "Failed on my_net_write()";
684
	    my_errno= ER_UNKNOWN_ERROR;
685 686
	    goto err;
	  }
unknown's avatar
unknown committed
687

688
	  if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
689
	  {
690
	    if (send_file(thd))
unknown's avatar
unknown committed
691
	    {
692
	      errmsg = "failed in send_file()";
693
	      my_errno= ER_UNKNOWN_ERROR;
unknown's avatar
unknown committed
694 695
	      goto err;
	    }
696
	  }
697
	  packet->set("\0", 1, &my_charset_bin);
698 699 700 701
	  /*
	    No need to net_flush because we will get to flush later when
	    we hit EOF pretty quick
	  */
702
	}
703

704
	if (fatal_error)
705 706
	{
	  errmsg = "error reading log entry";
707
          my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
708
	  goto err;
709
	}
710 711
	log.error=0;
      }
712
    }
713 714 715
    else
    {
      bool loop_breaker = 0;
716
      /* need this to break out of the for loop from switch */
717

718
      thd_proc_info(thd, "Finished reading one binlog; switching to next binlog");
719
      switch (mysql_bin_log.find_next_log(&linfo, 1)) {
720 721 722 723 724 725 726
      case LOG_INFO_EOF:
	loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
	break;
      case 0:
	break;
      default:
	errmsg = "could not find next log";
727
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
728 729 730
	goto err;
      }

731 732
      if (loop_breaker)
        break;
unknown's avatar
foo1  
unknown committed
733

734 735
      end_io_cache(&log);
      (void) my_close(file, MYF(MY_WME));
736

737
      /*
738 739 740 741 742 743 744
        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.
745
      */
746
      if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||
747 748
	  fake_rotate_event(net, packet, log_file_name, BIN_LOG_HEADER_SIZE,
                            &errmsg))
749 750
      {
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
unknown's avatar
unknown committed
751
	goto err;
752
      }
unknown's avatar
unknown committed
753

unknown's avatar
unknown committed
754
      packet->length(0);
755
      packet->append('\0');
756
    }
757
  }
758

unknown's avatar
unknown committed
759
end:
760 761
  end_io_cache(&log);
  (void)my_close(file, MYF(MY_WME));
762

763
  my_eof(thd);
764
  thd_proc_info(thd, "Waiting to finalize termination");
unknown's avatar
unknown committed
765 766 767
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
768
  DBUG_VOID_RETURN;
769

unknown's avatar
unknown committed
770
err:
771
  thd_proc_info(thd, "Waiting to finalize termination");
772
  end_io_cache(&log);
unknown's avatar
unknown committed
773 774 775 776 777 778 779
  /*
    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
  */
780
  pthread_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
781 782
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
783 784
  if (file >= 0)
    (void) my_close(file, MYF(MY_WME));
785

unknown's avatar
unknown committed
786
  my_message(my_errno, errmsg, MYF(0));
787 788 789
  DBUG_VOID_RETURN;
}

790 791 792 793 794 795 796 797 798 799 800 801 802 803

/**
  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.

  @param net_report If true, saves the exit status into thd->main_da.

  @retval 0 success
  @retval 1 error
*/
804
int start_slave(THD* thd , Master_info* mi,  bool net_report)
805
{
806
  int slave_errno= 0;
807
  int thread_mask;
808
  DBUG_ENTER("start_slave");
809

810
  if (check_access(thd, SUPER_ACL, any_db,0,0,0,0))
811
    DBUG_RETURN(1);
812
  lock_slave_threads(mi);  // this allows us to cleanly read slave_running
813
  // Get a mask of _stopped_ threads
814
  init_thread_mask(&thread_mask,mi,1 /* inverse */);
815
  /*
816 817 818 819
    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
820
  */
unknown's avatar
unknown committed
821
  if (thd->lex->slave_thd_opt)
822
    thread_mask&= thd->lex->slave_thd_opt;
823
  if (thread_mask) //some threads are stopped, start them
824
  {
825 826
    if (init_master_info(mi,master_info_file,relay_log_info_file, 0,
			 thread_mask))
827 828
      slave_errno=ER_MASTER_INFO;
    else if (server_id_supplied && *mi->host)
829
    {
830
      /*
831 832 833
        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.
834 835 836 837 838
      */
      if (thread_mask & SLAVE_SQL)
      {
        pthread_mutex_lock(&mi->rli.data_lock);

unknown's avatar
unknown committed
839
        if (thd->lex->mi.pos)
840
        {
841
          mi->rli.until_condition= Relay_log_info::UNTIL_MASTER_POS;
unknown's avatar
unknown committed
842
          mi->rli.until_log_pos= thd->lex->mi.pos;
843 844
          /*
             We don't check thd->lex->mi.log_file_name for NULL here
845 846
             since it is checked in sql_yacc.yy
          */
unknown's avatar
unknown committed
847
          strmake(mi->rli.until_log_name, thd->lex->mi.log_file_name,
848
                  sizeof(mi->rli.until_log_name)-1);
849
        }
unknown's avatar
unknown committed
850
        else if (thd->lex->mi.relay_log_pos)
851
        {
852
          mi->rli.until_condition= Relay_log_info::UNTIL_RELAY_POS;
unknown's avatar
unknown committed
853 854
          mi->rli.until_log_pos= thd->lex->mi.relay_log_pos;
          strmake(mi->rli.until_log_name, thd->lex->mi.relay_log_name,
855
                  sizeof(mi->rli.until_log_name)-1);
856 857
        }
        else
858
          mi->rli.clear_until_condition();
859

860
        if (mi->rli.until_condition != Relay_log_info::UNTIL_NONE)
861 862 863 864 865 866 867 868 869 870 871 872
        {
          /* 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.
873
            */
unknown's avatar
unknown committed
874
            if (p_end==p || *p_end)
875 876 877 878
              slave_errno=ER_BAD_SLAVE_UNTIL_COND;
          }
          else
            slave_errno=ER_BAD_SLAVE_UNTIL_COND;
879

880
          /* mark the cached result of the UNTIL comparison as "undefined" */
881
          mi->rli.until_log_names_cmp_result=
882
            Relay_log_info::UNTIL_LOG_NAMES_CMP_UNKNOWN;
883 884 885

          /* Issuing warning then started without --skip-slave-start */
          if (!opt_skip_slave_start)
886
            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
unknown's avatar
foo1  
unknown committed
887
                         ER_MISSING_SKIP_SLAVE,
888 889
                         ER(ER_MISSING_SKIP_SLAVE));
        }
890

891 892
        pthread_mutex_unlock(&mi->rli.data_lock);
      }
unknown's avatar
unknown committed
893
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
894
        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
895
                     ER(ER_UNTIL_COND_IGNORED));
896

unknown's avatar
unknown committed
897
      if (!slave_errno)
898
        slave_errno = start_slave_threads(0 /*no mutex */,
899 900 901 902
					1 /* wait for start */,
					mi,
					master_info_file,relay_log_info_file,
					thread_mask);
903
    }
904 905 906
    else
      slave_errno = ER_BAD_SLAVE;
  }
907
  else
908 909
  {
    /* no error if all threads are already started, only a warning */
910 911
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_RUNNING,
                 ER(ER_SLAVE_WAS_RUNNING));
912
  }
unknown's avatar
foo1  
unknown committed
913

914
  unlock_slave_threads(mi);
915

916 917
  if (slave_errno)
  {
918
    if (net_report)
unknown's avatar
unknown committed
919
      my_message(slave_errno, ER(slave_errno), MYF(0));
920
    DBUG_RETURN(1);
921 922
  }
  else if (net_report)
923
    my_ok(thd);
924

925
  DBUG_RETURN(0);
926 927
}

928

929 930 931 932 933 934 935 936 937 938 939 940 941
/**
  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.

  @param net_report If true, saves the exit status into thd->main_da.

  @retval 0 success
  @retval 1 error
*/
942
int stop_slave(THD* thd, Master_info* mi, bool net_report )
943
{
944 945
  DBUG_ENTER("stop_slave");
  
946
  int slave_errno;
947 948
  if (!thd)
    thd = current_thd;
949

950
  if (check_access(thd, SUPER_ACL, any_db,0,0,0,0))
951
    DBUG_RETURN(1);
952
  thd_proc_info(thd, "Killing slave");
953 954
  int thread_mask;
  lock_slave_threads(mi);
955
  // Get a mask of _running_ threads
956
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
957 958 959 960 961 962
  /*
    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
963 964
  if (thd->lex->slave_thd_opt)
    thread_mask &= thd->lex->slave_thd_opt;
965 966 967 968 969 970 971 972 973 974 975 976 977

  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));
  }
978
  unlock_slave_threads(mi);
979
  thd_proc_info(thd, 0);
980

981 982 983
  if (slave_errno)
  {
    if (net_report)
unknown's avatar
unknown committed
984
      my_message(slave_errno, ER(slave_errno), MYF(0));
985
    DBUG_RETURN(1);
986 987
  }
  else if (net_report)
988
    my_ok(thd);
989

990
  DBUG_RETURN(0);
991 992
}

993

994 995
/**
  Execute a RESET SLAVE statement.
996

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

1000
  @param mi Pointer to Master_info object for the slave.
1001

1002 1003 1004
  @retval 0 success
  @retval 1 error
*/
1005
int reset_slave(THD *thd, Master_info* mi)
1006 1007 1008
{
  MY_STAT stat_area;
  char fname[FN_REFLEN];
1009 1010
  int thread_mask= 0, error= 0;
  uint sql_errno=0;
1011
  const char* errmsg=0;
unknown's avatar
unknown committed
1012 1013
  DBUG_ENTER("reset_slave");

1014
  lock_slave_threads(mi);
1015 1016 1017 1018 1019 1020 1021
  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
1022 1023 1024

  ha_reset_slave(thd);

1025
  // delete relay logs, clear relay log coordinates
1026 1027 1028
  if ((error= purge_relay_logs(&mi->rli, thd,
			       1 /* just reset */,
			       &errmsg)))
1029 1030
  {
    sql_errno= ER_RELAY_LOG_FAIL;
1031
    goto err;
1032
  }
1033

unknown's avatar
unknown committed
1034 1035 1036 1037 1038 1039 1040 1041
  /*
    Clear master's log coordinates and reset host/user/etc to the values
    specified in mysqld's options (only for good display of SHOW SLAVE STATUS;
    next init_master_info() (in start_slave() for example) would have set them
    the same way; but here this is for the case where the user does SHOW SLAVE
    STATUS; before doing START SLAVE;
  */
  init_master_info_with_options(mi);
1042
  /*
1043
     Reset errors (the idea is that we forget about the
unknown's avatar
unknown committed
1044 1045
     old master).
  */
1046
  mi->clear_error();
1047
  mi->rli.clear_error();
1048
  mi->rli.clear_until_condition();
1049

1050
  // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
1051
  end_master_info(mi);
1052
  // and delete these two files
1053 1054
  fn_format(fname, master_info_file, mysql_data_home, "", 4+32);
  if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME)))
1055
  {
1056 1057
    error=1;
    goto err;
1058
  }
1059
  // delete relay_log_info_file
1060
  fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
1061
  if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME)))
1062 1063 1064 1065
  {
    error=1;
    goto err;
  }
1066

1067 1068
err:
  unlock_slave_threads(mi);
unknown's avatar
unknown committed
1069
  if (error)
1070
    my_error(sql_errno, MYF(0), errmsg);
unknown's avatar
unknown committed
1071
  DBUG_RETURN(error);
1072 1073
}

1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091
/*

  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
  Binlog_dump thread is waiting (pthread_cond_wait) for binlog update, then it
  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
1092

1093

1094 1095 1096
void kill_zombie_dump_threads(uint32 slave_server_id)
{
  pthread_mutex_lock(&LOCK_thread_count);
1097 1098 1099
  I_List_iterator<THD> it(threads);
  THD *tmp;

1100 1101 1102 1103
  while ((tmp=it++))
  {
    if (tmp->command == COM_BINLOG_DUMP &&
       tmp->server_id == slave_server_id)
1104
    {
1105
      pthread_mutex_lock(&tmp->LOCK_thd_data);	// Lock from delete
1106
      break;
1107 1108
    }
  }
1109
  pthread_mutex_unlock(&LOCK_thread_count);
1110 1111 1112 1113 1114 1115 1116
  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
1117
    tmp->awake(THD::KILL_QUERY);
1118
    pthread_mutex_unlock(&tmp->LOCK_thd_data);
1119
  }
1120 1121
}

1122

1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134
/**
  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
*/
1135
bool change_master(THD* thd, Master_info* mi)
1136
{
1137
  int thread_mask;
unknown's avatar
unknown committed
1138 1139
  const char* errmsg= 0;
  bool need_relay_log_purge= 1;
1140 1141
  DBUG_ENTER("change_master");

1142
  lock_slave_threads(mi);
1143 1144
  init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
  if (thread_mask) // We refuse if any slave thread is running
1145
  {
unknown's avatar
unknown committed
1146
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
1147
    unlock_slave_threads(mi);
unknown's avatar
unknown committed
1148
    DBUG_RETURN(TRUE);
1149
  }
1150

1151
  thd_proc_info(thd, "Changing master");
1152
  LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
1153
  // TODO: see if needs re-write
1154 1155
  if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
		       thread_mask))
1156
  {
unknown's avatar
unknown committed
1157
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
1158
    unlock_slave_threads(mi);
unknown's avatar
unknown committed
1159
    DBUG_RETURN(TRUE);
1160 1161
  }

1162 1163 1164 1165
  /*
    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
1166
  */
1167 1168

  /*
unknown's avatar
foo1  
unknown committed
1169
    If the user specified host or port without binlog or position,
1170
    reset binlog's name to FIRST and position to 4.
unknown's avatar
foo1  
unknown committed
1171
  */
1172

1173 1174
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
  {
1175
    mi->master_log_name[0] = 0;
1176
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
1177
  }
1178

1179
  if (lex_mi->log_file_name)
1180
    strmake(mi->master_log_name, lex_mi->log_file_name,
1181
	    sizeof(mi->master_log_name)-1);
1182
  if (lex_mi->pos)
unknown's avatar
unknown committed
1183
  {
1184
    mi->master_log_pos= lex_mi->pos;
unknown's avatar
unknown committed
1185
  }
unknown's avatar
unknown committed
1186
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
1187 1188

  if (lex_mi->host)
1189
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
1190
  if (lex_mi->user)
1191
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
1192
  if (lex_mi->password)
1193
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
1194
  if (lex_mi->port)
1195
    mi->port = lex_mi->port;
1196
  if (lex_mi->connect_retry)
1197
    mi->connect_retry = lex_mi->connect_retry;
unknown's avatar
foo1  
unknown committed
1198

unknown's avatar
unknown committed
1199 1200
  if (lex_mi->ssl != LEX_MASTER_INFO::SSL_UNCHANGED)
    mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::SSL_ENABLE);
1201 1202 1203 1204 1205

  if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::SSL_UNCHANGED)
    mi->ssl_verify_server_cert=
      (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::SSL_ENABLE);

unknown's avatar
unknown committed
1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217
  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 ||
1218 1219
      lex_mi->ssl_cert || lex_mi->ssl_cipher || lex_mi->ssl_key ||
      lex_mi->ssl_verify_server_cert )
unknown's avatar
foo1  
unknown committed
1220
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
unknown's avatar
unknown committed
1221 1222
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
#endif
1223

1224 1225
  if (lex_mi->relay_log_name)
  {
1226
    need_relay_log_purge= 0;
1227 1228 1229 1230
    strmake(mi->rli.group_relay_log_name,lex_mi->relay_log_name,
	    sizeof(mi->rli.group_relay_log_name)-1);
    strmake(mi->rli.event_relay_log_name,lex_mi->relay_log_name,
	    sizeof(mi->rli.event_relay_log_name)-1);
1231 1232 1233 1234
  }

  if (lex_mi->relay_log_pos)
  {
1235
    need_relay_log_purge= 0;
1236
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
1237 1238
  }

unknown's avatar
unknown committed
1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263
  /*
    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
1264 1265 1266
     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
1267 1268
             sizeof(mi->master_log_name)-1);
  }
unknown's avatar
unknown committed
1269 1270 1271 1272
  /*
    Relay log's IO_CACHE may not be inited, if rli->inited==0 (server was never
    a slave before).
  */
1273 1274 1275 1276 1277 1278
  if (flush_master_info(mi, 0))
  {
    my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
    unlock_slave_threads(mi);
    DBUG_RETURN(TRUE);
  }
1279
  if (need_relay_log_purge)
1280
  {
1281
    relay_log_purge= 1;
1282
    thd_proc_info(thd, "Purging old relay logs");
1283 1284
    if (purge_relay_logs(&mi->rli, thd,
			 0 /* not only reset, but also reinit */,
1285 1286
			 &errmsg))
    {
1287
      my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
1288
      unlock_slave_threads(mi);
unknown's avatar
unknown committed
1289
      DBUG_RETURN(TRUE);
1290 1291 1292 1293 1294
    }
  }
  else
  {
    const char* msg;
1295
    relay_log_purge= 0;
1296 1297
    /* Relay log is already initialized */
    if (init_relay_log_pos(&mi->rli,
1298 1299
			   mi->rli.group_relay_log_name,
			   mi->rli.group_relay_log_pos,
1300
			   0 /*no data lock*/,
1301
			   &msg, 0))
1302
    {
1303
      my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
1304
      unlock_slave_threads(mi);
unknown's avatar
unknown committed
1305
      DBUG_RETURN(TRUE);
1306
    }
1307
  }
unknown's avatar
unknown committed
1308 1309 1310 1311 1312 1313 1314 1315 1316 1317
  /*
    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
1318
  mi->rli.group_master_log_pos= mi->master_log_pos;
1319
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
1320 1321
  strmake(mi->rli.group_master_log_name,mi->master_log_name,
	  sizeof(mi->rli.group_master_log_name)-1);
unknown's avatar
unknown committed
1322

1323 1324
  if (!mi->rli.group_master_log_name[0]) // uninitialized case
    mi->rli.group_master_log_pos=0;
1325 1326

  pthread_mutex_lock(&mi->rli.data_lock);
1327
  mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
1328
  /* Clear the errors, for a clean start */
1329
  mi->rli.clear_error();
1330
  mi->rli.clear_until_condition();
unknown's avatar
unknown committed
1331 1332 1333 1334 1335 1336
  /*
    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).
1337
  */
unknown's avatar
unknown committed
1338
  flush_relay_log_info(&mi->rli);
1339
  pthread_cond_broadcast(&mi->data_cond);
1340
  pthread_mutex_unlock(&mi->rli.data_lock);
1341

1342
  unlock_slave_threads(mi);
1343
  thd_proc_info(thd, 0);
1344
  my_ok(thd);
unknown's avatar
unknown committed
1345
  DBUG_RETURN(FALSE);
1346 1347
}

1348 1349 1350 1351 1352 1353 1354 1355 1356 1357

/**
  Execute a RESET MASTER statement.

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

  @retval 0 success
  @retval 1 error
*/
1358
int reset_master(THD* thd)
1359
{
1360
  if (!mysql_bin_log.is_open())
1361
  {
unknown's avatar
unknown committed
1362 1363
    my_message(ER_FLUSH_MASTER_BINLOG_CLOSED,
               ER(ER_FLUSH_MASTER_BINLOG_CLOSED), MYF(ME_BELL+ME_WAITTANG));
1364
    return 1;
1365
  }
1366
  return mysql_bin_log.reset_logs(thd);
1367 1368
}

unknown's avatar
unknown committed
1369 1370 1371 1372
int cmp_master_pos(const char* log_file_name1, ulonglong log_pos1,
		   const char* log_file_name2, ulonglong log_pos2)
{
  int res;
1373 1374
  size_t log_file_name1_len=  strlen(log_file_name1);
  size_t log_file_name2_len=  strlen(log_file_name2);
unknown's avatar
unknown committed
1375

1376 1377 1378 1379 1380 1381 1382 1383
  //  We assume that both log names match up to '.'
  if (log_file_name1_len == log_file_name2_len)
  {
    if ((res= strcmp(log_file_name1, log_file_name2)))
      return res;
    return (log_pos1 < log_pos2) ? -1 : (log_pos1 == log_pos2) ? 0 : 1;
  }
  return ((log_file_name1_len < log_file_name2_len) ? -1 : 1);
unknown's avatar
unknown committed
1384 1385
}

unknown's avatar
unknown committed
1386

1387 1388 1389 1390 1391 1392 1393 1394 1395
/**
  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
*/
1396
bool mysql_show_binlog_events(THD* thd)
unknown's avatar
unknown committed
1397
{
1398
  Protocol *protocol= thd->protocol;
unknown's avatar
unknown committed
1399
  List<Item> field_list;
1400
  const char *errmsg = 0;
1401
  bool ret = TRUE;
unknown's avatar
unknown committed
1402 1403
  IO_CACHE log;
  File file = -1;
1404
  MYSQL_BIN_LOG *binary_log= NULL;
1405
  DBUG_ENTER("mysql_show_binlog_events");
1406 1407

  Log_event::init_show_field_list(&field_list);
1408 1409
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1410
    DBUG_RETURN(TRUE);
1411

1412 1413 1414
  Format_description_log_event *description_event= new
    Format_description_log_event(3); /* MySQL 4.0 by default */

1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436
  DBUG_ASSERT(thd->lex->sql_command == SQLCOM_SHOW_BINLOG_EVENTS ||
              thd->lex->sql_command == SQLCOM_SHOW_RELAYLOG_EVENTS);

  /* 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
1437

1438
  if (binary_log->is_open())
unknown's avatar
unknown committed
1439
  {
1440
    LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
1441
    SELECT_LEX_UNIT *unit= &thd->lex->unit;
1442
    ha_rows event_count, limit_start, limit_end;
1443
    my_off_t pos = max(BIN_LOG_HEADER_SIZE, lex_mi->pos); // user-friendly
1444 1445
    char search_file_name[FN_REFLEN], *name;
    const char *log_file_name = lex_mi->log_file_name;
1446
    pthread_mutex_t *log_lock = binary_log->get_log_lock();
1447 1448
    LOG_INFO linfo;
    Log_event* ev;
1449

1450 1451 1452
    unit->set_limit(thd->lex->current_select);
    limit_start= unit->offset_limit_cnt;
    limit_end= unit->select_limit_cnt;
unknown's avatar
unknown committed
1453

1454
    name= search_file_name;
unknown's avatar
unknown committed
1455
    if (log_file_name)
1456
      binary_log->make_log_name(search_file_name, log_file_name);
unknown's avatar
unknown committed
1457
    else
1458
      name=0;					// Find first log
unknown's avatar
unknown committed
1459 1460

    linfo.index_file_offset = 0;
1461

1462
    if (binary_log->find_log_pos(&linfo, name, 1))
unknown's avatar
unknown committed
1463 1464 1465 1466 1467
    {
      errmsg = "Could not find target log";
      goto err;
    }

1468 1469 1470 1471
    pthread_mutex_lock(&LOCK_thread_count);
    thd->current_linfo = &linfo;
    pthread_mutex_unlock(&LOCK_thread_count);

unknown's avatar
unknown committed
1472 1473 1474
    if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0)
      goto err;

1475 1476 1477 1478 1479
    /*
      to account binlog event header size
    */
    thd->variables.max_allowed_packet += MAX_LOG_EVENT_HEADER;

1480
    pthread_mutex_lock(log_lock);
1481

1482
    /*
1483 1484 1485 1486 1487 1488
      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).
1489 1490 1491 1492 1493
    */
    ev = Log_event::read_log_event(&log,(pthread_mutex_t*)0,description_event);
    if (ev)
    {
      if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
1494 1495
      {
        delete description_event;
1496
        description_event= (Format_description_log_event*) ev;
1497
      }
1498 1499 1500 1501
      else
        delete ev;
    }

unknown's avatar
unknown committed
1502 1503
    my_b_seek(&log, pos);

1504 1505 1506 1507 1508
    if (!description_event->is_valid())
    {
      errmsg="Invalid Format_description event; could be out of memory";
      goto err;
    }
1509

1510
    for (event_count = 0;
1511 1512
	 (ev = Log_event::read_log_event(&log,(pthread_mutex_t*) 0,
                                         description_event)); )
unknown's avatar
unknown committed
1513 1514
    {
      if (event_count >= limit_start &&
1515
	  ev->net_send(protocol, linfo.log_file_name, pos))
1516 1517 1518
      {
	errmsg = "Net error";
	delete ev;
1519
	pthread_mutex_unlock(log_lock);
1520 1521 1522 1523
	goto err;
      }

      pos = my_b_tell(&log);
unknown's avatar
unknown committed
1524 1525 1526 1527 1528 1529 1530 1531 1532
      delete ev;

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

    if (event_count < limit_end && log.error)
    {
      errmsg = "Wrong offset or I/O error";
1533
      pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1534 1535
      goto err;
    }
1536

1537
    pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1538 1539
  }

1540 1541
  ret= FALSE;

unknown's avatar
unknown committed
1542
err:
1543
  delete description_event;
unknown's avatar
unknown committed
1544 1545 1546 1547 1548
  if (file >= 0)
  {
    end_io_cache(&log);
    (void) my_close(file, MYF(MY_WME));
  }
1549

unknown's avatar
unknown committed
1550
  if (errmsg)
1551 1552
    my_error(ER_ERROR_WHEN_EXECUTING_COMMAND, MYF(0),
             "SHOW BINLOG EVENTS", errmsg);
1553
  else
1554
    my_eof(thd);
unknown's avatar
unknown committed
1555

1556 1557 1558
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
1559
  DBUG_RETURN(ret);
unknown's avatar
unknown committed
1560 1561
}

unknown's avatar
unknown committed
1562

1563 1564 1565 1566 1567 1568 1569 1570 1571
/**
  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
1572
bool show_binlog_info(THD* thd)
1573
{
1574
  Protocol *protocol= thd->protocol;
1575 1576 1577
  DBUG_ENTER("show_binlog_info");
  List<Item> field_list;
  field_list.push_back(new Item_empty_string("File", FN_REFLEN));
1578 1579
  field_list.push_back(new Item_return_int("Position",20,
					   MYSQL_TYPE_LONGLONG));
1580 1581
  field_list.push_back(new Item_empty_string("Binlog_Do_DB",255));
  field_list.push_back(new Item_empty_string("Binlog_Ignore_DB",255));
1582

1583 1584
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1585
    DBUG_RETURN(TRUE);
1586 1587
  protocol->prepare_for_resend();

1588 1589 1590 1591 1592
  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);
1593
    protocol->store(li.log_file_name + dir_len, &my_charset_bin);
1594
    protocol->store((ulonglong) li.pos);
1595 1596
    protocol->store(binlog_filter->get_do_db());
    protocol->store(binlog_filter->get_ignore_db());
1597
    if (protocol->write())
unknown's avatar
unknown committed
1598
      DBUG_RETURN(TRUE);
1599
  }
1600
  my_eof(thd);
unknown's avatar
unknown committed
1601
  DBUG_RETURN(FALSE);
1602
}
unknown's avatar
unknown committed
1603

1604

1605 1606
/**
  Execute a SHOW BINARY LOGS statement.
1607

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

1611 1612
  @retval FALSE success
  @retval TRUE failure
1613
*/
unknown's avatar
unknown committed
1614
bool show_binlogs(THD* thd)
unknown's avatar
unknown committed
1615
{
1616
  IO_CACHE *index_file;
1617 1618
  LOG_INFO cur;
  File file;
unknown's avatar
unknown committed
1619 1620
  char fname[FN_REFLEN];
  List<Item> field_list;
1621
  uint length;
unknown's avatar
unknown committed
1622
  int cur_dir_len;
1623 1624
  Protocol *protocol= thd->protocol;
  DBUG_ENTER("show_binlogs");
1625 1626

  if (!mysql_bin_log.is_open())
1627
  {
unknown's avatar
unknown committed
1628
    my_message(ER_NO_BINARY_LOGGING, ER(ER_NO_BINARY_LOGGING), MYF(0));
1629
    return 1;
1630
  }
unknown's avatar
unknown committed
1631

1632
  field_list.push_back(new Item_empty_string("Log_name", 255));
unknown's avatar
foo1  
unknown committed
1633
  field_list.push_back(new Item_return_int("File_size", 20,
1634
                                           MYSQL_TYPE_LONGLONG));
1635 1636
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1637
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1638 1639
  
  pthread_mutex_lock(mysql_bin_log.get_log_lock());
unknown's avatar
unknown committed
1640
  mysql_bin_log.lock_index();
1641
  index_file=mysql_bin_log.get_index_file();
unknown's avatar
unknown committed
1642 1643 1644 1645
  
  mysql_bin_log.raw_get_current_log(&cur); // dont take mutex
  pthread_mutex_unlock(mysql_bin_log.get_log_lock()); // lockdep, OK
  
unknown's avatar
unknown committed
1646
  cur_dir_len= dirname_length(cur.log_file_name);
1647

1648 1649 1650 1651
  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)
1652
  {
unknown's avatar
unknown committed
1653 1654 1655
    int dir_len;
    ulonglong file_length= 0;                   // Length if open fails
    fname[--length] = '\0';                     // remove the newline
1656

1657
    protocol->prepare_for_resend();
unknown's avatar
unknown committed
1658 1659 1660 1661 1662 1663 1664
    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
1665 1666
    {
      /* this is an old log, open it and find the size */
1667
      if ((file= my_open(fname, O_RDONLY | O_SHARE | O_BINARY,
unknown's avatar
unknown committed
1668
                         MYF(0))) >= 0)
1669
      {
unknown's avatar
unknown committed
1670
        file_length= (ulonglong) my_seek(file, 0L, MY_SEEK_END, MYF(0));
1671 1672 1673
        my_close(file, MYF(0));
      }
    }
unknown's avatar
unknown committed
1674
    protocol->store(file_length);
1675
    if (protocol->write())
1676
      goto err;
1677
  }
unknown's avatar
unknown committed
1678
  mysql_bin_log.unlock_index();
1679
  my_eof(thd);
unknown's avatar
unknown committed
1680
  DBUG_RETURN(FALSE);
1681

1682 1683
err:
  mysql_bin_log.unlock_index();
unknown's avatar
unknown committed
1684
  DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1685 1686
}

1687 1688 1689 1690 1691 1692 1693 1694 1695
/**
   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
   @return 0
*/
1696 1697
int log_loaded_block(IO_CACHE* file)
{
1698
  DBUG_ENTER("log_loaded_block");
1699
  LOAD_FILE_INFO *lf_info;
1700 1701
  uint block_len;
  /* buffer contains position where we started last read */
1702
  uchar* buffer= (uchar*) my_b_get_buffer_start(file);
1703 1704
  uint max_event_size= current_thd->variables.max_allowed_packet;
  lf_info= (LOAD_FILE_INFO*) file->arg;
1705
  if (lf_info->thd->current_stmt_binlog_row_based)
1706
    DBUG_RETURN(0);
1707
  if (lf_info->last_pos_in_file != HA_POS_ERROR &&
1708
      lf_info->last_pos_in_file >= my_b_get_pos_in_file(file))
1709
    DBUG_RETURN(0);
1710
  
1711
  for (block_len= (uint) (my_b_get_bytes_in_buffer(file)); block_len > 0;
1712 1713
       buffer += min(block_len, max_event_size),
       block_len -= min(block_len, max_event_size))
1714
  {
1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732
    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);
      mysql_bin_log.write(&a);
    }
    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);
      mysql_bin_log.write(&b);
      lf_info->wrote_create_file= 1;
      DBUG_SYNC_POINT("debug_lock.created_file_event",10);
    }
1733
  }
1734
  DBUG_RETURN(0);
1735
}
1736

unknown's avatar
unknown committed
1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757
/*
  Replication System Variables
*/

class sys_var_slave_skip_counter :public sys_var
{
public:
  sys_var_slave_skip_counter(sys_var_chain *chain, const char *name_arg)
    :sys_var(name_arg)
  { chain_sys_var(chain); }
  bool check(THD *thd, set_var *var);
  bool update(THD *thd, set_var *var);
  bool check_type(enum_var_type type) { return type != OPT_GLOBAL; }
  /*
    We can't retrieve the value of this, so we don't have to define
    type() or value_ptr()
  */
};

static sys_var_chain vars = { NULL, NULL };

1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769
static sys_var_const    sys_log_slave_updates(&vars, "log_slave_updates",
                                              OPT_GLOBAL, SHOW_MY_BOOL,
                                              (uchar*) &opt_log_slave_updates);
static sys_var_const    sys_relay_log(&vars, "relay_log",
                                      OPT_GLOBAL, SHOW_CHAR_PTR,
                                      (uchar*) &opt_relay_logname);
static sys_var_const    sys_relay_log_index(&vars, "relay_log_index",
                                      OPT_GLOBAL, SHOW_CHAR_PTR,
                                      (uchar*) &opt_relaylog_index_name);
static sys_var_const    sys_relay_log_info_file(&vars, "relay_log_info_file",
                                      OPT_GLOBAL, SHOW_CHAR_PTR,
                                      (uchar*) &relay_log_info_file);
unknown's avatar
unknown committed
1770 1771
static sys_var_bool_ptr	sys_relay_log_purge(&vars, "relay_log_purge",
					    &relay_log_purge);
1772 1773 1774 1775 1776 1777 1778 1779
static sys_var_const    sys_relay_log_space_limit(&vars,
                                                  "relay_log_space_limit",
                                                  OPT_GLOBAL, SHOW_LONGLONG,
                                                  (uchar*)
                                                  &relay_log_space_limit);
static sys_var_const    sys_slave_load_tmpdir(&vars, "slave_load_tmpdir",
                                              OPT_GLOBAL, SHOW_CHAR_PTR,
                                              (uchar*) &slave_load_tmpdir);
unknown's avatar
unknown committed
1780 1781
static sys_var_long_ptr	sys_slave_net_timeout(&vars, "slave_net_timeout",
					      &slave_net_timeout);
1782 1783 1784
static sys_var_const    sys_slave_skip_errors(&vars, "slave_skip_errors",
                                              OPT_GLOBAL, SHOW_CHAR,
                                              (uchar*) slave_skip_error_names);
unknown's avatar
unknown committed
1785 1786
static sys_var_long_ptr	sys_slave_trans_retries(&vars, "slave_transaction_retries",
						&slave_trans_retries);
1787 1788
static sys_var_int_ptr sys_sync_binlog_period(&vars, "sync_binlog", &sync_binlog_period);
static sys_var_int_ptr sys_sync_relaylog_period(&vars, "sync_relay_log", &sync_relaylog_period);
unknown's avatar
unknown committed
1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840
static sys_var_slave_skip_counter sys_slave_skip_counter(&vars, "sql_slave_skip_counter");


bool sys_var_slave_skip_counter::check(THD *thd, set_var *var)
{
  int result= 0;
  pthread_mutex_lock(&LOCK_active_mi);
  pthread_mutex_lock(&active_mi->rli.run_lock);
  if (active_mi->rli.slave_running)
  {
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
    result=1;
  }
  pthread_mutex_unlock(&active_mi->rli.run_lock);
  pthread_mutex_unlock(&LOCK_active_mi);
  var->save_result.ulong_value= (ulong) var->value->val_int();
  return result;
}


bool sys_var_slave_skip_counter::update(THD *thd, set_var *var)
{
  pthread_mutex_lock(&LOCK_active_mi);
  pthread_mutex_lock(&active_mi->rli.run_lock);
  /*
    The following test should normally never be true as we test this
    in the check function;  To be safe against multiple
    SQL_SLAVE_SKIP_COUNTER request, we do the check anyway
  */
  if (!active_mi->rli.slave_running)
  {
    pthread_mutex_lock(&active_mi->rli.data_lock);
    active_mi->rli.slave_skip_counter= var->save_result.ulong_value;
    pthread_mutex_unlock(&active_mi->rli.data_lock);
  }
  pthread_mutex_unlock(&active_mi->rli.run_lock);
  pthread_mutex_unlock(&LOCK_active_mi);
  return 0;
}


int init_replication_sys_vars()
{
  if (mysql_add_sys_var_chain(vars.first, my_long_options))
  {
    /* should not happen */
    fprintf(stderr, "failed to initialize replication system variables");
    unireg_abort(1);
  }
  return 0;
}

unknown's avatar
SCRUM  
unknown committed
1841
#endif /* HAVE_REPLICATION */
1842 1843