sql_repl.cc 45.5 KB
Newer Older
1
/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB & Sasha
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

   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
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   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
18 19
#ifdef HAVE_REPLICATION

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;
27
static int binlog_dump_count = 0;
unknown's avatar
unknown committed
28

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

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

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

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

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

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

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

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

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

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

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

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

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

 err:
unknown's avatar
unknown committed
141
  thd->net.read_timeout = old_timeout;
142 143
  if (fd >= 0)
    (void) my_close(fd, MYF(0));
144 145
  if (errmsg)
  {
146
    sql_print_error("Failed in send_file() %s", errmsg);
147 148 149 150 151
    DBUG_PRINT("error", (errmsg));
  }
  DBUG_RETURN(error);
}

152

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

unknown's avatar
unknown committed
178 179 180
  pthread_mutex_lock(&LOCK_thread_count);
  I_List_iterator<THD> it(threads);

181 182 183 184 185 186
  while ((tmp=it++))
  {
    LOG_INFO* linfo;
    if ((linfo = tmp->current_linfo))
    {
      pthread_mutex_lock(&linfo->lock);
187 188 189 190
      /*
	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
191
      */
192 193 194 195 196 197 198
      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
199 200 201
  pthread_mutex_unlock(&LOCK_thread_count);
}

unknown's avatar
merge  
unknown committed
202

unknown's avatar
unknown committed
203 204 205 206 207
bool log_in_use(const char* log_name)
{
  int log_name_len = strlen(log_name) + 1;
  THD *tmp;
  bool result = 0;
208

unknown's avatar
unknown committed
209 210
  pthread_mutex_lock(&LOCK_thread_count);
  I_List_iterator<THD> it(threads);
211 212 213 214 215

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

  pthread_mutex_unlock(&LOCK_thread_count);
  return result;
}

unknown's avatar
unknown committed
229
bool purge_error_message(THD* thd, int res)
unknown's avatar
unknown committed
230
{
unknown's avatar
unknown committed
231
  uint errmsg= 0;
232

unknown's avatar
unknown committed
233
  switch (res)  {
234
  case 0: break;
unknown's avatar
unknown committed
235 236 237 238 239 240 241 242
  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;
  default:		errmsg= ER_LOG_PURGE_UNKNOWN_ERR; break;
243 244 245 246
  }

  if (errmsg)
  {
unknown's avatar
unknown committed
247
    my_message(errmsg, ER(errmsg), MYF(0));
unknown's avatar
unknown committed
248
    return TRUE;
249
  }
unknown's avatar
unknown committed
250
  send_ok(thd);
unknown's avatar
unknown committed
251
  return FALSE;
unknown's avatar
unknown committed
252 253
}

unknown's avatar
unknown committed
254

unknown's avatar
unknown committed
255
bool purge_master_logs(THD* thd, const char* to_log)
256 257
{
  char search_file_name[FN_REFLEN];
unknown's avatar
unknown committed
258 259
  if (!mysql_bin_log.is_open())
  {
unknown's avatar
unknown committed
260
    send_ok(thd);
unknown's avatar
unknown committed
261
    return FALSE;
unknown's avatar
unknown committed
262
  }
263 264

  mysql_bin_log.make_log_name(search_file_name, to_log);
unknown's avatar
unknown committed
265 266
  return purge_error_message(thd,
			     mysql_bin_log.purge_logs(search_file_name, 0, 1,
unknown's avatar
unknown committed
267
						      1, NULL));
268 269 270
}


unknown's avatar
unknown committed
271
bool purge_master_logs_before_date(THD* thd, time_t purge_time)
272
{
273 274
  if (!mysql_bin_log.is_open())
  {
unknown's avatar
unknown committed
275
    send_ok(thd);
276 277 278 279
    return 0;
  }
  return purge_error_message(thd,
                             mysql_bin_log.purge_logs_before_date(purge_time));
280 281
}

282
int test_for_non_eof_log_read_errors(int error, const char **errmsg)
283
{
284
  if (error == LOG_READ_EOF)
285 286 287 288
    return 0;
  my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
  switch (error) {
  case LOG_READ_BOGUS:
289
    *errmsg = "bogus data in log event";
290 291
    break;
  case LOG_READ_TOO_LARGE:
292
    *errmsg = "log event entry exceeded max_allowed_packet; \
293 294 295
Increase max_allowed_packet on master";
    break;
  case LOG_READ_IO:
296
    *errmsg = "I/O error reading log event";
297 298
    break;
  case LOG_READ_MEM:
299
    *errmsg = "memory allocation failed reading log event";
300 301
    break;
  case LOG_READ_TRUNC:
302
    *errmsg = "binlog truncated in the middle of event";
303 304
    break;
  default:
305
    *errmsg = "unknown error reading log event on the master";
306 307 308 309 310 311
    break;
  }
  return error;
}


312 313 314
/*
  TODO: Clean up loop to only have one call to send_file()
*/
unknown's avatar
merge  
unknown committed
315

316 317
void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
		       ushort flags)
318 319 320
{
  LOG_INFO linfo;
  char *log_file_name = linfo.log_file_name;
321
  char search_file_name[FN_REFLEN], *name;
322 323
  IO_CACHE log;
  File file = -1;
324 325 326 327
  String* packet = &thd->packet;
  int error;
  const char *errmsg = "Unknown error";
  NET* net = &thd->net;
328
  pthread_mutex_t *log_lock;
329
  bool binlog_can_be_corrupted= FALSE;
330 331
#ifndef DBUG_OFF
  int left_events = max_binlog_dump_events;
332
#endif
333
  DBUG_ENTER("mysql_binlog_send");
334 335
  DBUG_PRINT("enter",("log_ident: '%s'  pos: %ld", log_ident, (long) pos));

336
  bzero((char*) &log,sizeof(log));
337

unknown's avatar
unknown committed
338
#ifndef DBUG_OFF
339 340 341
  if (opt_sporadic_binlog_dump_fail && (binlog_dump_count++ % 2))
  {
    errmsg = "Master failed COM_BINLOG_DUMP to test if slave can recover";
342
    my_errno= ER_UNKNOWN_ERROR;
343 344
    goto err;
  }
345
#endif
346

347
  if (!mysql_bin_log.is_open())
348 349
  {
    errmsg = "Binary log is not open";
350
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
351 352
    goto err;
  }
353 354 355
  if (!server_id_supplied)
  {
    errmsg = "Misconfigured master - server id was not set";
356
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
357 358 359
    goto err;
  }

360
  name=search_file_name;
361
  if (log_ident[0])
362 363
    mysql_bin_log.make_log_name(search_file_name, log_ident);
  else
364
    name=0;					// Find first log
365

unknown's avatar
unknown committed
366 367
  linfo.index_file_offset = 0;
  thd->current_linfo = &linfo;
368

369
  if (mysql_bin_log.find_log_pos(&linfo, name, 1))
370
  {
371
    errmsg = "Could not find first log file name in binary log index file";
372
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
373 374
    goto err;
  }
375

376
  if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)
377 378
  {
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
379
    goto err;
380 381
  }
  if (pos < BIN_LOG_HEADER_SIZE || pos > my_b_filelength(&log))
382
  {
383
    errmsg= "Client requested master to start replication from \
unknown's avatar
unknown committed
384
impossible position";
385
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
386 387
    goto err;
  }
388

unknown's avatar
unknown committed
389 390 391
  if (thd->variables.sync_replication)
    ha_repl_report_sent_binlog(thd, log_file_name, pos);

392 393
  /*
    We need to start a packet with something other than 255
394
    to distinguish it from error
395
  */
396
  packet->set("\0", 1, &my_charset_bin); /* This is the start of a new packet */
397

398
  /*
399
    Tell the client about the log name with a fake Rotate event;
400 401 402 403 404 405 406 407 408
    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
409
    already knows the binlog's name.
410 411 412 413 414 415 416
    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).
417
    Always calling fake_rotate_event makes sending of normal
418 419 420 421 422 423 424
    (=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.
425
  */
426
  if (fake_rotate_event(net, packet, log_file_name, pos, &errmsg))
unknown's avatar
unknown committed
427
  {
428 429 430 431
    /*
       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.
432
    */
433 434
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
    goto err;
unknown's avatar
unknown committed
435
  }
unknown's avatar
unknown committed
436
  packet->set("\0", 1, &my_charset_bin);
unknown's avatar
unknown committed
437

438
  /*
439 440 441
    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).
442
  */
443
  log_lock = mysql_bin_log.get_log_lock();
444
  if (pos > BIN_LOG_HEADER_SIZE)
445 446 447 448 449
  {
     /*
       Try to find a Format_description_log_event at the beginning of
       the binlog
     */
450 451 452
     if (!(error = Log_event::read_log_event(&log, packet, log_lock)))
     {
       /*
453 454
         The packet has offsets equal to the normal offsets in a binlog
         event +1 (the first character is \0).
455 456 457 458 459 460
       */
       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)
       {
461 462
         binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
                                       LOG_EVENT_BINLOG_IN_USE_F);
unknown's avatar
unknown committed
463
         (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
464 465 466 467 468
         /*
           mark that this event with "log_pos=0", so the slave
           should not increment master's binlog position
           (rli->group_master_log_pos)
         */
469
         int4store((char*) packet->ptr()+LOG_POS_OFFSET+1, 0);
470 471 472 473 474 475 476
         /* send it */
         if (my_net_write(net, (char*)packet->ptr(), packet->length()))
         {
           errmsg = "Failed on my_net_write()";
           my_errno= ER_UNKNOWN_ERROR;
           goto err;
         }
unknown's avatar
unknown committed
477 478 479 480

         if (thd->variables.sync_replication)
           ha_repl_report_sent_binlog(thd, log_file_name, my_b_tell(&log));

481
         /*
482 483 484 485
           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.
486 487 488 489
         */
       }
     }
     else
490
     {
491
       if (test_for_non_eof_log_read_errors(error, &errmsg))
492
         goto err;
493 494 495 496 497
       /*
         It's EOF, nothing to do, go on reading next events, the
         Format_description_log_event will be found naturally if it is written.
       */
     }
498 499
     /* reset the packet as we wrote to it in any case */
     packet->set("\0", 1, &my_charset_bin);
500 501 502 503 504
  } /* end of if (pos > BIN_LOG_HEADER_SIZE); */
  else
  {
    /* The Format_description_log_event event will be found naturally. */
  }
505

506 507 508
  /* 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
509
  while (!net->error && net->vio != 0 && !thd->killed)
510 511 512
  {
    while (!(error = Log_event::read_log_event(&log, packet, log_lock)))
    {
513
#ifndef DBUG_OFF
514
      if (max_binlog_dump_events && !left_events--)
515 516 517
      {
	net_flush(net);
	errmsg = "Debugging binlog dump abort";
518
	my_errno= ER_UNKNOWN_ERROR;
519 520
	goto err;
      }
521
#endif
522 523

      if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
unknown's avatar
unknown committed
524
      {
525 526
        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
                                      LOG_EVENT_BINLOG_IN_USE_F);
unknown's avatar
unknown committed
527 528
        (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
      }
529 530 531
      else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT)
        binlog_can_be_corrupted= FALSE;

532
      if (my_net_write(net, (char*)packet->ptr(), packet->length()))
533 534
      {
	errmsg = "Failed on my_net_write()";
535
	my_errno= ER_UNKNOWN_ERROR;
536 537
	goto err;
      }
unknown's avatar
unknown committed
538 539 540 541

      if (thd->variables.sync_replication)
        ha_repl_report_sent_binlog(thd, log_file_name, my_b_tell(&log));

542 543
      DBUG_PRINT("info", ("log event code %d",
			  (*packet)[LOG_EVENT_OFFSET+1] ));
unknown's avatar
unknown committed
544
      if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
545
      {
546
	if (send_file(thd))
547
	{
548
	  errmsg = "failed in send_file()";
549
	  my_errno= ER_UNKNOWN_ERROR;
550 551
	  goto err;
	}
552
      }
553
      packet->set("\0", 1, &my_charset_bin);
554
    }
555 556 557 558 559 560 561

    /*
      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;
562 563
    /*
      TODO: now that we are logging the offset, check to make sure
564
      the recorded offset and the actual match.
565 566 567
      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.
568
    */
569
    if (test_for_non_eof_log_read_errors(error, &errmsg))
570
      goto err;
571

572
    if (!(flags & BINLOG_DUMP_NON_BLOCK) &&
573
        mysql_bin_log.is_active(log_file_name))
574
    {
575 576 577
      /*
	Block until there is more data in the log
      */
578
      if (net_flush(net))
579 580
      {
	errmsg = "failed on net_flush()";
581
	my_errno= ER_UNKNOWN_ERROR;
582 583
	goto err;
      }
584

585 586 587 588 589 590
      /*
	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.
      */
591 592 593 594
      {
	log.error=0;
	bool read_packet = 0, fatal_error = 0;

595
#ifndef DBUG_OFF
596
	if (max_binlog_dump_events && !left_events--)
597 598
	{
	  errmsg = "Debugging binlog dump abort";
599
	  my_errno= ER_UNKNOWN_ERROR;
600 601
	  goto err;
	}
602
#endif
603

604 605 606 607
	/*
	  No one will update the log while we are reading
	  now, but we'll be quick and just read one record

608
	  TODO:
609 610 611
          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.
612
	*/
613

unknown's avatar
unknown committed
614
	pthread_mutex_lock(log_lock);
615
	switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*)0)) {
616
	case 0:
617
	  /* we read successfully, so we'll need to send it to the slave */
618
	  pthread_mutex_unlock(log_lock);
619 620
	  read_packet = 1;
	  break;
621

622
	case LOG_READ_EOF:
623
	  DBUG_PRINT("wait",("waiting for data in binary log"));
unknown's avatar
unknown committed
624
	  if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0)
unknown's avatar
unknown committed
625 626 627 628
	  {
	    pthread_mutex_unlock(log_lock);
	    goto end;
	  }
unknown's avatar
unknown committed
629
	  if (!thd->killed)
630 631
	  {
	    /* Note that the following call unlocks lock_log */
632
	    mysql_bin_log.wait_for_update(thd, 0);
633 634 635
	  }
	  else
	    pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
636
	  DBUG_PRINT("wait",("binary log received update"));
637 638 639
	  break;

	default:
640
	  pthread_mutex_unlock(log_lock);
641 642
	  fatal_error = 1;
	  break;
643
	}
644

645
	if (read_packet)
646
	{
647
	  thd->proc_info = "Sending binlog event to slave";
648
	  if (my_net_write(net, (char*)packet->ptr(), packet->length()) )
649 650
	  {
	    errmsg = "Failed on my_net_write()";
651
	    my_errno= ER_UNKNOWN_ERROR;
652 653
	    goto err;
	  }
unknown's avatar
unknown committed
654

unknown's avatar
unknown committed
655 656 657
          if (thd->variables.sync_replication)
            ha_repl_report_sent_binlog(thd, log_file_name, my_b_tell(&log));

658
	  if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
659
	  {
660
	    if (send_file(thd))
unknown's avatar
unknown committed
661
	    {
662
	      errmsg = "failed in send_file()";
663
	      my_errno= ER_UNKNOWN_ERROR;
unknown's avatar
unknown committed
664 665
	      goto err;
	    }
666
	  }
667
	  packet->set("\0", 1, &my_charset_bin);
668 669 670 671
	  /*
	    No need to net_flush because we will get to flush later when
	    we hit EOF pretty quick
	  */
672
	}
673

674
	if (fatal_error)
675 676
	{
	  errmsg = "error reading log entry";
677
          my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
678
	  goto err;
679
	}
680 681
	log.error=0;
      }
682
    }
683 684 685 686
    else
    {
      bool loop_breaker = 0;
      // need this to break out of the for loop from switch
687
      thd->proc_info = "Finished reading one binlog; switching to next binlog";
688
      switch (mysql_bin_log.find_next_log(&linfo, 1)) {
689 690 691 692 693 694 695
      case LOG_INFO_EOF:
	loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
	break;
      case 0:
	break;
      default:
	errmsg = "could not find next log";
696
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
697 698 699
	goto err;
      }

700
      if (loop_breaker)
701 702 703 704
	break;

      end_io_cache(&log);
      (void) my_close(file, MYF(MY_WME));
705

706
      /*
707 708 709 710 711 712 713
        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.
714
      */
715
      if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||
716 717
	  fake_rotate_event(net, packet, log_file_name, BIN_LOG_HEADER_SIZE,
                            &errmsg))
718 719
      {
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
unknown's avatar
unknown committed
720
	goto err;
721
      }
unknown's avatar
unknown committed
722 723 724 725

      if (thd->variables.sync_replication)
        ha_repl_report_sent_binlog(thd, log_file_name, 0);

unknown's avatar
unknown committed
726
      packet->length(0);
727
      packet->append('\0');
728
    }
729
  }
730

unknown's avatar
unknown committed
731
end:
unknown's avatar
unknown committed
732 733 734
  if (thd->variables.sync_replication)
    ha_repl_report_replication_stop(thd);

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

738
  send_eof(thd);
739
  thd->proc_info = "Waiting to finalize termination";
unknown's avatar
unknown committed
740 741 742
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
743
  DBUG_VOID_RETURN;
744

unknown's avatar
unknown committed
745
err:
unknown's avatar
unknown committed
746 747 748
  if (thd->variables.sync_replication)
    ha_repl_report_replication_stop(thd);

749
  thd->proc_info = "Waiting to finalize termination";
750
  end_io_cache(&log);
unknown's avatar
unknown committed
751 752 753 754 755 756 757
  /*
    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
  */
758
  pthread_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
759 760
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
761 762
  if (file >= 0)
    (void) my_close(file, MYF(MY_WME));
unknown's avatar
unknown committed
763
  my_message(my_errno, errmsg, MYF(0));
764 765 766
  DBUG_VOID_RETURN;
}

767
int start_slave(THD* thd , MASTER_INFO* mi,  bool net_report)
768
{
769
  int slave_errno= 0;
770
  int thread_mask;
771
  DBUG_ENTER("start_slave");
772

unknown's avatar
unknown committed
773
  if (check_access(thd, SUPER_ACL, any_db,0,0,0))
774
    DBUG_RETURN(1);
775
  lock_slave_threads(mi);  // this allows us to cleanly read slave_running
776
  // Get a mask of _stopped_ threads
777
  init_thread_mask(&thread_mask,mi,1 /* inverse */);
778
  /*
779 780 781 782
    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
783
  */
unknown's avatar
unknown committed
784
  if (thd->lex->slave_thd_opt)
785
    thread_mask&= thd->lex->slave_thd_opt;
786
  if (thread_mask) //some threads are stopped, start them
787
  {
788 789
    if (init_master_info(mi,master_info_file,relay_log_info_file, 0,
			 thread_mask))
790 791
      slave_errno=ER_MASTER_INFO;
    else if (server_id_supplied && *mi->host)
792
    {
793
      /*
794 795 796
        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.
797 798 799 800 801
      */
      if (thread_mask & SLAVE_SQL)
      {
        pthread_mutex_lock(&mi->rli.data_lock);

unknown's avatar
unknown committed
802
        if (thd->lex->mi.pos)
803 804
        {
          mi->rli.until_condition= RELAY_LOG_INFO::UNTIL_MASTER_POS;
unknown's avatar
unknown committed
805
          mi->rli.until_log_pos= thd->lex->mi.pos;
806 807
          /*
             We don't check thd->lex->mi.log_file_name for NULL here
808 809
             since it is checked in sql_yacc.yy
          */
unknown's avatar
unknown committed
810
          strmake(mi->rli.until_log_name, thd->lex->mi.log_file_name,
811
                  sizeof(mi->rli.until_log_name)-1);
812
        }
unknown's avatar
unknown committed
813
        else if (thd->lex->mi.relay_log_pos)
814 815
        {
          mi->rli.until_condition= RELAY_LOG_INFO::UNTIL_RELAY_POS;
unknown's avatar
unknown committed
816 817
          mi->rli.until_log_pos= thd->lex->mi.relay_log_pos;
          strmake(mi->rli.until_log_name, thd->lex->mi.relay_log_name,
818
                  sizeof(mi->rli.until_log_name)-1);
819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835
        }
        else
          clear_until_condition(&mi->rli);

        if (mi->rli.until_condition != RELAY_LOG_INFO::UNTIL_NONE)
        {
          /* 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.
836
            */
unknown's avatar
unknown committed
837
            if (p_end==p || *p_end)
838 839 840 841
              slave_errno=ER_BAD_SLAVE_UNTIL_COND;
          }
          else
            slave_errno=ER_BAD_SLAVE_UNTIL_COND;
842

843
          /* mark the cached result of the UNTIL comparison as "undefined" */
844
          mi->rli.until_log_names_cmp_result=
845 846 847 848
            RELAY_LOG_INFO::UNTIL_LOG_NAMES_CMP_UNKNOWN;

          /* Issuing warning then started without --skip-slave-start */
          if (!opt_skip_slave_start)
849 850
            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
                         ER_MISSING_SKIP_SLAVE, 
851 852
                         ER(ER_MISSING_SKIP_SLAVE));
        }
853

854 855
        pthread_mutex_unlock(&mi->rli.data_lock);
      }
unknown's avatar
unknown committed
856
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
857
        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
858
                     ER(ER_UNTIL_COND_IGNORED));
859

unknown's avatar
unknown committed
860
      if (!slave_errno)
861
        slave_errno = start_slave_threads(0 /*no mutex */,
862 863 864 865
					1 /* wait for start */,
					mi,
					master_info_file,relay_log_info_file,
					thread_mask);
866
    }
867 868 869
    else
      slave_errno = ER_BAD_SLAVE;
  }
870
  else
871 872
  {
    /* no error if all threads are already started, only a warning */
873 874
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_RUNNING,
                 ER(ER_SLAVE_WAS_RUNNING));
875 876
  }
  
877
  unlock_slave_threads(mi);
878

879 880
  if (slave_errno)
  {
881
    if (net_report)
unknown's avatar
unknown committed
882
      my_message(slave_errno, ER(slave_errno), MYF(0));
883
    DBUG_RETURN(1);
884 885
  }
  else if (net_report)
886
    send_ok(thd);
887

888
  DBUG_RETURN(0);
889 890
}

891

892
int stop_slave(THD* thd, MASTER_INFO* mi, bool net_report )
893
{
894
  int slave_errno;
895 896
  if (!thd)
    thd = current_thd;
897

unknown's avatar
unknown committed
898
  if (check_access(thd, SUPER_ACL, any_db,0,0,0))
899
    return 1;
900 901 902
  thd->proc_info = "Killing slave";
  int thread_mask;
  lock_slave_threads(mi);
903
  // Get a mask of _running_ threads
904
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
905 906 907 908 909 910
  /*
    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
911 912
  if (thd->lex->slave_thd_opt)
    thread_mask &= thd->lex->slave_thd_opt;
913 914 915 916 917 918 919 920 921 922 923 924 925

  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));
  }
926
  unlock_slave_threads(mi);
927 928
  thd->proc_info = 0;

929 930 931
  if (slave_errno)
  {
    if (net_report)
unknown's avatar
unknown committed
932
      my_message(slave_errno, ER(slave_errno), MYF(0));
933 934 935
    return 1;
  }
  else if (net_report)
936
    send_ok(thd);
937 938 939 940

  return 0;
}

941 942 943 944 945 946 947 948 949 950 951 952 953 954 955

/*
  Remove all relay logs and start replication from the start

  SYNOPSIS
    reset_slave()
    thd			Thread handler
    mi			Master info for the slave

  RETURN
    0	ok
    1	error
*/


956
int reset_slave(THD *thd, MASTER_INFO* mi)
957 958 959
{
  MY_STAT stat_area;
  char fname[FN_REFLEN];
960 961
  int thread_mask= 0, error= 0;
  uint sql_errno=0;
962
  const char* errmsg=0;
unknown's avatar
unknown committed
963 964
  DBUG_ENTER("reset_slave");

965
  lock_slave_threads(mi);
966 967 968 969 970 971 972
  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;
  }
973
  // delete relay logs, clear relay log coordinates
974 975 976
  if ((error= purge_relay_logs(&mi->rli, thd,
			       1 /* just reset */,
			       &errmsg)))
977
    goto err;
978

unknown's avatar
unknown committed
979 980 981 982 983 984 985 986
  /*
    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);
987
  /*
988
     Reset errors (the idea is that we forget about the
unknown's avatar
unknown committed
989 990
     old master).
  */
991
  clear_slave_error(&mi->rli);
992
  clear_until_condition(&mi->rli);
993

994
  // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
995
  end_master_info(mi);
996
  // and delete these two files
997 998
  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)))
999
  {
1000 1001
    error=1;
    goto err;
1002
  }
1003
  // delete relay_log_info_file
1004
  fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
1005
  if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME)))
1006 1007 1008 1009
  {
    error=1;
    goto err;
  }
1010

1011 1012
err:
  unlock_slave_threads(mi);
unknown's avatar
unknown committed
1013
  if (error)
1014
    my_error(sql_errno, MYF(0), errmsg);
unknown's avatar
unknown committed
1015
  DBUG_RETURN(error);
1016 1017
}

1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036
/*

  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

*/
  
1037

1038 1039 1040
void kill_zombie_dump_threads(uint32 slave_server_id)
{
  pthread_mutex_lock(&LOCK_thread_count);
1041 1042 1043
  I_List_iterator<THD> it(threads);
  THD *tmp;

1044 1045 1046 1047
  while ((tmp=it++))
  {
    if (tmp->command == COM_BINLOG_DUMP &&
       tmp->server_id == slave_server_id)
1048
    {
1049 1050
      pthread_mutex_lock(&tmp->LOCK_delete);	// Lock from delete
      break;
1051 1052
    }
  }
1053
  pthread_mutex_unlock(&LOCK_thread_count);
1054 1055 1056 1057 1058 1059 1060
  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
1061
    tmp->awake(THD::KILL_QUERY);
1062 1063
    pthread_mutex_unlock(&tmp->LOCK_delete);
  }
1064 1065
}

1066

unknown's avatar
unknown committed
1067
bool change_master(THD* thd, MASTER_INFO* mi)
1068
{
1069
  int thread_mask;
unknown's avatar
unknown committed
1070 1071
  const char* errmsg= 0;
  bool need_relay_log_purge= 1;
1072 1073
  DBUG_ENTER("change_master");

1074
  lock_slave_threads(mi);
1075 1076
  init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
  if (thread_mask) // We refuse if any slave thread is running
1077
  {
unknown's avatar
unknown committed
1078
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
1079
    unlock_slave_threads(mi);
unknown's avatar
unknown committed
1080
    DBUG_RETURN(TRUE);
1081
  }
1082

1083
  thd->proc_info = "Changing master";
1084
  LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
1085
  // TODO: see if needs re-write
1086 1087
  if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
		       thread_mask))
1088
  {
unknown's avatar
unknown committed
1089
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
1090
    unlock_slave_threads(mi);
unknown's avatar
unknown committed
1091
    DBUG_RETURN(TRUE);
1092 1093
  }

1094 1095 1096 1097
  /*
    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
1098
  */
1099 1100 1101 1102 1103 1104

  /*
    If the user specified host or port without binlog or position, 
    reset binlog's name to FIRST and position to 4.
  */ 

1105 1106
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
  {
1107
    mi->master_log_name[0] = 0;
1108
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
1109
  }
1110

1111
  if (lex_mi->log_file_name)
1112
    strmake(mi->master_log_name, lex_mi->log_file_name,
1113
	    sizeof(mi->master_log_name)-1);
1114
  if (lex_mi->pos)
unknown's avatar
unknown committed
1115
  {
1116
    mi->master_log_pos= lex_mi->pos;
unknown's avatar
unknown committed
1117
  }
1118
  DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos));
1119 1120

  if (lex_mi->host)
1121
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
1122
  if (lex_mi->user)
1123
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
1124
  if (lex_mi->password)
1125
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
1126
  if (lex_mi->port)
1127
    mi->port = lex_mi->port;
1128
  if (lex_mi->connect_retry)
1129
    mi->connect_retry = lex_mi->connect_retry;
unknown's avatar
unknown committed
1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148
 
  if (lex_mi->ssl != LEX_MASTER_INFO::SSL_UNCHANGED)
    mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::SSL_ENABLE);
  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 ||
      lex_mi->ssl_cert || lex_mi->ssl_cipher || lex_mi->ssl_key )
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, 
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
#endif
1149

1150 1151
  if (lex_mi->relay_log_name)
  {
1152
    need_relay_log_purge= 0;
1153 1154 1155 1156
    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);
1157 1158 1159 1160
  }

  if (lex_mi->relay_log_pos)
  {
1161
    need_relay_log_purge= 0;
1162
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
1163 1164
  }

unknown's avatar
unknown committed
1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189
  /*
    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
1190 1191 1192
     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
1193 1194
             sizeof(mi->master_log_name)-1);
  }
unknown's avatar
unknown committed
1195 1196 1197 1198 1199
  /*
    Relay log's IO_CACHE may not be inited, if rli->inited==0 (server was never
    a slave before).
  */
  flush_master_info(mi, 0);
1200
  if (need_relay_log_purge)
1201
  {
1202
    relay_log_purge= 1;
1203
    thd->proc_info="Purging old relay logs";
1204 1205
    if (purge_relay_logs(&mi->rli, thd,
			 0 /* not only reset, but also reinit */,
1206 1207
			 &errmsg))
    {
1208
      my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
1209
      unlock_slave_threads(mi);
unknown's avatar
unknown committed
1210
      DBUG_RETURN(TRUE);
1211 1212 1213 1214 1215
    }
  }
  else
  {
    const char* msg;
1216
    relay_log_purge= 0;
1217 1218
    /* Relay log is already initialized */
    if (init_relay_log_pos(&mi->rli,
1219 1220
			   mi->rli.group_relay_log_name,
			   mi->rli.group_relay_log_pos,
1221
			   0 /*no data lock*/,
1222
			   &msg, 0))
1223
    {
1224
      my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
1225
      unlock_slave_threads(mi);
unknown's avatar
unknown committed
1226
      DBUG_RETURN(TRUE);
1227
    }
1228
  }
1229
  mi->rli.group_master_log_pos = mi->master_log_pos;
1230
  DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos));
unknown's avatar
unknown committed
1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241

  /*
    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
1242
  mi->rli.group_master_log_pos= mi->master_log_pos;
1243 1244
  strmake(mi->rli.group_master_log_name,mi->master_log_name,
	  sizeof(mi->rli.group_master_log_name)-1);
unknown's avatar
unknown committed
1245

1246 1247
  if (!mi->rli.group_master_log_name[0]) // uninitialized case
    mi->rli.group_master_log_pos=0;
1248 1249

  pthread_mutex_lock(&mi->rli.data_lock);
1250
  mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
1251 1252
  /* Clear the errors, for a clean start */
  clear_slave_error(&mi->rli);
1253
  clear_until_condition(&mi->rli);
unknown's avatar
unknown committed
1254 1255 1256 1257 1258 1259
  /*
    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).
1260
  */
unknown's avatar
unknown committed
1261
  flush_relay_log_info(&mi->rli);
1262
  pthread_cond_broadcast(&mi->data_cond);
1263
  pthread_mutex_unlock(&mi->rli.data_lock);
1264

1265
  unlock_slave_threads(mi);
1266
  thd->proc_info = 0;
1267
  send_ok(thd);
unknown's avatar
unknown committed
1268
  DBUG_RETURN(FALSE);
1269 1270
}

1271
int reset_master(THD* thd)
1272
{
1273
  if (!mysql_bin_log.is_open())
1274
  {
unknown's avatar
unknown committed
1275 1276
    my_message(ER_FLUSH_MASTER_BINLOG_CLOSED,
               ER(ER_FLUSH_MASTER_BINLOG_CLOSED), MYF(ME_BELL+ME_WAITTANG));
1277
    return 1;
1278
  }
1279
  return mysql_bin_log.reset_logs(thd);
1280 1281
}

unknown's avatar
unknown committed
1282 1283 1284 1285
int cmp_master_pos(const char* log_file_name1, ulonglong log_pos1,
		   const char* log_file_name2, ulonglong log_pos2)
{
  int res;
1286 1287
  uint log_file_name1_len=  strlen(log_file_name1);
  uint log_file_name2_len=  strlen(log_file_name2);
unknown's avatar
unknown committed
1288

1289 1290 1291 1292 1293 1294 1295 1296
  //  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
1297 1298
}

unknown's avatar
unknown committed
1299

1300
bool mysql_show_binlog_events(THD* thd)
unknown's avatar
unknown committed
1301
{
1302
  Protocol *protocol= thd->protocol;
1303
  DBUG_ENTER("mysql_show_binlog_events");
unknown's avatar
unknown committed
1304
  List<Item> field_list;
1305
  const char *errmsg = 0;
1306
  bool ret = TRUE;
unknown's avatar
unknown committed
1307 1308
  IO_CACHE log;
  File file = -1;
1309
  Format_description_log_event *description_event= new
1310
    Format_description_log_event(3); /* MySQL 4.0 by default */
1311 1312

  Log_event::init_show_field_list(&field_list);
1313 1314
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1315
    DBUG_RETURN(TRUE);
1316

unknown's avatar
unknown committed
1317 1318
  if (mysql_bin_log.is_open())
  {
1319
    LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
1320
    ha_rows event_count, limit_start, limit_end;
1321
    my_off_t pos = max(BIN_LOG_HEADER_SIZE, lex_mi->pos); // user-friendly
1322 1323
    char search_file_name[FN_REFLEN], *name;
    const char *log_file_name = lex_mi->log_file_name;
1324
    pthread_mutex_t *log_lock = mysql_bin_log.get_log_lock();
1325 1326
    LOG_INFO linfo;
    Log_event* ev;
1327

1328 1329
    limit_start= thd->lex->current_select->offset_limit;
    limit_end= thd->lex->current_select->select_limit + limit_start;
unknown's avatar
unknown committed
1330

1331
    name= search_file_name;
unknown's avatar
unknown committed
1332 1333 1334
    if (log_file_name)
      mysql_bin_log.make_log_name(search_file_name, log_file_name);
    else
1335
      name=0;					// Find first log
unknown's avatar
unknown committed
1336 1337 1338

    linfo.index_file_offset = 0;
    thd->current_linfo = &linfo;
1339

1340
    if (mysql_bin_log.find_log_pos(&linfo, name, 1))
unknown's avatar
unknown committed
1341 1342 1343 1344 1345 1346 1347 1348
    {
      errmsg = "Could not find target log";
      goto err;
    }

    if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0)
      goto err;

1349
    pthread_mutex_lock(log_lock);
1350

1351
    /*
1352
       open_binlog() sought to position 4.
1353 1354 1355
       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.
1356 1357 1358
       This code will fail on a mixed relay log (one which has Format_desc then
       Rotate then Format_desc).
    */
1359

1360 1361 1362 1363
    ev = Log_event::read_log_event(&log,(pthread_mutex_t*)0,description_event);
    if (ev)
    {
      if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
1364 1365
      {
        delete description_event;
1366
        description_event= (Format_description_log_event*) ev;
1367
      }
1368 1369 1370 1371
      else
        delete ev;
    }

unknown's avatar
unknown committed
1372 1373
    my_b_seek(&log, pos);

1374 1375 1376 1377 1378
    if (!description_event->is_valid())
    {
      errmsg="Invalid Format_description event; could be out of memory";
      goto err;
    }
1379

1380
    for (event_count = 0;
1381
	 (ev = Log_event::read_log_event(&log,(pthread_mutex_t*)0,description_event)); )
unknown's avatar
unknown committed
1382 1383
    {
      if (event_count >= limit_start &&
1384
	  ev->net_send(protocol, linfo.log_file_name, pos))
1385 1386 1387
      {
	errmsg = "Net error";
	delete ev;
1388
	pthread_mutex_unlock(log_lock);
1389 1390 1391 1392
	goto err;
      }

      pos = my_b_tell(&log);
unknown's avatar
unknown committed
1393 1394 1395 1396 1397 1398 1399 1400 1401
      delete ev;

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

    if (event_count < limit_end && log.error)
    {
      errmsg = "Wrong offset or I/O error";
1402
      pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1403 1404
      goto err;
    }
1405

1406
    pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1407 1408
  }

1409 1410
  ret= FALSE;

unknown's avatar
unknown committed
1411
err:
1412
  delete description_event;
unknown's avatar
unknown committed
1413 1414 1415 1416 1417
  if (file >= 0)
  {
    end_io_cache(&log);
    (void) my_close(file, MYF(MY_WME));
  }
1418

unknown's avatar
unknown committed
1419 1420
  if (errmsg)
  {
1421 1422
    my_error(ER_ERROR_WHEN_EXECUTING_COMMAND, MYF(0),
             "SHOW BINLOG EVENTS", errmsg);
unknown's avatar
unknown committed
1423
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1424 1425
  }

1426
  send_eof(thd);
1427 1428 1429
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
1430
  DBUG_RETURN(ret);
unknown's avatar
unknown committed
1431 1432
}

unknown's avatar
unknown committed
1433

unknown's avatar
unknown committed
1434
bool show_binlog_info(THD* thd)
1435
{
1436
  Protocol *protocol= thd->protocol;
1437 1438 1439
  DBUG_ENTER("show_binlog_info");
  List<Item> field_list;
  field_list.push_back(new Item_empty_string("File", FN_REFLEN));
1440 1441
  field_list.push_back(new Item_return_int("Position",20,
					   MYSQL_TYPE_LONGLONG));
1442 1443
  field_list.push_back(new Item_empty_string("Binlog_Do_DB",255));
  field_list.push_back(new Item_empty_string("Binlog_Ignore_DB",255));
1444

1445 1446
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1447
    DBUG_RETURN(TRUE);
1448 1449
  protocol->prepare_for_resend();

1450 1451 1452 1453 1454
  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);
1455
    protocol->store(li.log_file_name + dir_len, &my_charset_bin);
1456
    protocol->store((ulonglong) li.pos);
1457 1458
    protocol->store(binlog_filter->get_do_db());
    protocol->store(binlog_filter->get_ignore_db());
1459
    if (protocol->write())
unknown's avatar
unknown committed
1460
      DBUG_RETURN(TRUE);
1461
  }
1462
  send_eof(thd);
unknown's avatar
unknown committed
1463
  DBUG_RETURN(FALSE);
1464
}
unknown's avatar
unknown committed
1465

1466

1467
/*
unknown's avatar
unknown committed
1468
  Send a list of all binary logs to client
1469 1470 1471 1472 1473 1474

  SYNOPSIS
    show_binlogs()
    thd		Thread specific variable

  RETURN VALUES
unknown's avatar
unknown committed
1475 1476
    FALSE OK
    TRUE  error
1477 1478
*/

unknown's avatar
unknown committed
1479
bool show_binlogs(THD* thd)
unknown's avatar
unknown committed
1480
{
1481
  IO_CACHE *index_file;
unknown's avatar
unknown committed
1482 1483
  char fname[FN_REFLEN];
  List<Item> field_list;
1484
  uint length;
1485 1486
  Protocol *protocol= thd->protocol;
  DBUG_ENTER("show_binlogs");
1487 1488

  if (!mysql_bin_log.is_open())
1489
  {
unknown's avatar
unknown committed
1490
    my_message(ER_NO_BINARY_LOGGING, ER(ER_NO_BINARY_LOGGING), MYF(0));
1491
    return 1;
1492
  }
unknown's avatar
unknown committed
1493

1494
  field_list.push_back(new Item_empty_string("Log_name", 255));
1495 1496
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1497
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1498
  mysql_bin_log.lock_index();
1499 1500 1501 1502 1503 1504
  index_file=mysql_bin_log.get_index_file();
  
  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)
1505
  {
1506
    protocol->prepare_for_resend();
1507
    int dir_len = dirname_length(fname);
1508
    /* The -1 is for removing newline from fname */
1509
    protocol->store(fname + dir_len, length-1-dir_len, &my_charset_bin);
1510
    if (protocol->write())
1511
      goto err;
1512
  }
unknown's avatar
unknown committed
1513
  mysql_bin_log.unlock_index();
1514
  send_eof(thd);
unknown's avatar
unknown committed
1515
  DBUG_RETURN(FALSE);
1516

1517 1518
err:
  mysql_bin_log.unlock_index();
unknown's avatar
unknown committed
1519
  DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1520 1521
}

1522

1523 1524
int log_loaded_block(IO_CACHE* file)
{
1525
  LOAD_FILE_INFO *lf_info;
1526
  uint block_len ;
unknown's avatar
unknown committed
1527 1528 1529

  /* file->request_pos contains position where we started last read */
  char* buffer = (char*) file->request_pos;
unknown's avatar
unknown committed
1530
  if (!(block_len = (char*) file->read_end - (char*) buffer))
1531
    return 0;
unknown's avatar
unknown committed
1532
  lf_info = (LOAD_FILE_INFO*) file->arg;
1533 1534 1535 1536 1537 1538
  if (lf_info->last_pos_in_file != HA_POS_ERROR &&
      lf_info->last_pos_in_file >= file->pos_in_file)
    return 0;
  lf_info->last_pos_in_file = file->pos_in_file;
  if (lf_info->wrote_create_file)
  {
unknown's avatar
unknown committed
1539 1540
    Append_block_log_event a(lf_info->thd, lf_info->thd->db, buffer,
                             block_len, lf_info->log_delayed);
1541 1542 1543 1544
    mysql_bin_log.write(&a);
  }
  else
  {
unknown's avatar
unknown committed
1545 1546 1547 1548
    Begin_load_query_log_event b(lf_info->thd, lf_info->thd->db,
                                 buffer, block_len,
                                 lf_info->log_delayed);
    mysql_bin_log.write(&b);
1549
    lf_info->wrote_create_file = 1;
1550
    DBUG_SYNC_POINT("debug_lock.created_file_event",10);
1551 1552 1553
  }
  return 0;
}
1554

unknown's avatar
SCRUM  
unknown committed
1555
#endif /* HAVE_REPLICATION */
1556 1557