sql_repl.cc 46.4 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 20
#include "sql_repl.h"
#include "log_event.h"
unknown's avatar
unknown committed
21
#include <my_dir.h>
22

23
int max_binlog_dump_events = 0; // unlimited
24
my_bool opt_sporadic_binlog_dump_fail = 0;
25
#ifndef DBUG_OFF
26
static int binlog_dump_count = 0;
27
#endif
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
  int old_timeout;
unknown's avatar
unknown committed
88
  unsigned long 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
  /*
    The client might be slow loading the data, give him wait_timeout to do
    the job
  */
96 97
  old_timeout= net->read_timeout;
  net_set_read_timeout(net, thd->variables.net_wait_timeout);
unknown's avatar
unknown committed
98 99 100 101 102

  /*
    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:
141
  net_set_read_timeout(net, 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

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

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

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

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

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

      if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
unknown's avatar
unknown committed
530
      {
531 532
        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
                                      LOG_EVENT_BINLOG_IN_USE_F);
unknown's avatar
unknown committed
533 534
        (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
      }
535 536 537
      else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT)
        binlog_can_be_corrupted= FALSE;

538
      if (my_net_write(net, (char*)packet->ptr(), packet->length()))
539 540
      {
	errmsg = "Failed on my_net_write()";
541
	my_errno= ER_UNKNOWN_ERROR;
542 543
	goto err;
      }
unknown's avatar
unknown committed
544

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

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

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

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

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

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

611
	  TODO:
612 613 614
          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.
615
	*/
616

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

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

	default:
643
	  pthread_mutex_unlock(log_lock);
644 645
	  fatal_error = 1;
	  break;
646
	}
647

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

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
    else
    {
      bool loop_breaker = 0;
686
      /* need this to break out of the for loop from switch */
687

688
      thd->proc_info = "Finished reading one binlog; switching to next binlog";
689
      switch (mysql_bin_log.find_next_log(&linfo, 1)) {
690 691 692 693 694 695 696
      case LOG_INFO_EOF:
	loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
	break;
      case 0:
	break;
      default:
	errmsg = "could not find next log";
697
	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
698 699 700
	goto err;
      }

701 702
      if (loop_breaker)
        break;
unknown's avatar
foo1  
unknown committed
703

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

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

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

unknown's avatar
unknown committed
729
end:
730 731
  end_io_cache(&log);
  (void)my_close(file, MYF(MY_WME));
732

733
  send_eof(thd);
734
  thd->proc_info = "Waiting to finalize termination";
unknown's avatar
unknown committed
735 736 737
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
738
  DBUG_VOID_RETURN;
739

unknown's avatar
unknown committed
740
err:
741
  thd->proc_info = "Waiting to finalize termination";
742
  end_io_cache(&log);
unknown's avatar
unknown committed
743 744 745 746 747 748 749
  /*
    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
  */
750
  pthread_mutex_lock(&LOCK_thread_count);
unknown's avatar
unknown committed
751 752
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
753 754
  if (file >= 0)
    (void) my_close(file, MYF(MY_WME));
755

unknown's avatar
unknown committed
756
  my_message(my_errno, errmsg, MYF(0));
757 758 759
  DBUG_VOID_RETURN;
}

760
int start_slave(THD* thd , MASTER_INFO* mi,  bool net_report)
761
{
762
  int slave_errno= 0;
763
  int thread_mask;
764
  DBUG_ENTER("start_slave");
765

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

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

836
          /* mark the cached result of the UNTIL comparison as "undefined" */
837
          mi->rli.until_log_names_cmp_result=
838 839 840 841
            RELAY_LOG_INFO::UNTIL_LOG_NAMES_CMP_UNKNOWN;

          /* Issuing warning then started without --skip-slave-start */
          if (!opt_skip_slave_start)
842
            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
unknown's avatar
foo1  
unknown committed
843
                         ER_MISSING_SKIP_SLAVE,
844 845
                         ER(ER_MISSING_SKIP_SLAVE));
        }
846

847 848
        pthread_mutex_unlock(&mi->rli.data_lock);
      }
unknown's avatar
unknown committed
849
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
850
        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
851
                     ER(ER_UNTIL_COND_IGNORED));
852

unknown's avatar
unknown committed
853
      if (!slave_errno)
854
        slave_errno = start_slave_threads(0 /*no mutex */,
855 856 857 858
					1 /* wait for start */,
					mi,
					master_info_file,relay_log_info_file,
					thread_mask);
859
    }
860 861 862
    else
      slave_errno = ER_BAD_SLAVE;
  }
863
  else
864 865
  {
    /* no error if all threads are already started, only a warning */
866 867
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_RUNNING,
                 ER(ER_SLAVE_WAS_RUNNING));
868
  }
unknown's avatar
foo1  
unknown committed
869

870
  unlock_slave_threads(mi);
871

872 873
  if (slave_errno)
  {
874
    if (net_report)
unknown's avatar
unknown committed
875
      my_message(slave_errno, ER(slave_errno), MYF(0));
876
    DBUG_RETURN(1);
877 878
  }
  else if (net_report)
879
    send_ok(thd);
880

881
  DBUG_RETURN(0);
882 883
}

884

885
int stop_slave(THD* thd, MASTER_INFO* mi, bool net_report )
886
{
887 888
  DBUG_ENTER("stop_slave");
  
889
  int slave_errno;
890 891
  if (!thd)
    thd = current_thd;
892

893
  if (check_access(thd, SUPER_ACL, any_db,0,0,0,0))
894
    DBUG_RETURN(1);
895 896 897
  thd->proc_info = "Killing slave";
  int thread_mask;
  lock_slave_threads(mi);
898
  // Get a mask of _running_ threads
899
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
900 901 902 903 904 905
  /*
    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
906 907
  if (thd->lex->slave_thd_opt)
    thread_mask &= thd->lex->slave_thd_opt;
908 909 910 911 912 913 914 915 916 917 918 919 920

  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));
  }
921
  unlock_slave_threads(mi);
922 923
  thd->proc_info = 0;

924 925 926
  if (slave_errno)
  {
    if (net_report)
unknown's avatar
unknown committed
927
      my_message(slave_errno, ER(slave_errno), MYF(0));
928
    DBUG_RETURN(1);
929 930
  }
  else if (net_report)
931
    send_ok(thd);
932

933
  DBUG_RETURN(0);
934 935
}

936 937 938 939 940 941 942 943 944 945 946 947 948 949 950

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


951
int reset_slave(THD *thd, MASTER_INFO* mi)
952 953 954
{
  MY_STAT stat_area;
  char fname[FN_REFLEN];
955 956
  int thread_mask= 0, error= 0;
  uint sql_errno=0;
957
  const char* errmsg=0;
unknown's avatar
unknown committed
958 959
  DBUG_ENTER("reset_slave");

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

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

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

1006 1007
err:
  unlock_slave_threads(mi);
unknown's avatar
unknown committed
1008
  if (error)
1009
    my_error(sql_errno, MYF(0), errmsg);
unknown's avatar
unknown committed
1010
  DBUG_RETURN(error);
1011 1012
}

1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030
/*

  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
1031

1032

1033 1034 1035
void kill_zombie_dump_threads(uint32 slave_server_id)
{
  pthread_mutex_lock(&LOCK_thread_count);
1036 1037 1038
  I_List_iterator<THD> it(threads);
  THD *tmp;

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

1061

unknown's avatar
unknown committed
1062
bool change_master(THD* thd, MASTER_INFO* mi)
1063
{
1064
  int thread_mask;
unknown's avatar
unknown committed
1065 1066
  const char* errmsg= 0;
  bool need_relay_log_purge= 1;
1067 1068
  DBUG_ENTER("change_master");

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

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

1089 1090 1091 1092
  /*
    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
1093
  */
1094 1095

  /*
unknown's avatar
foo1  
unknown committed
1096
    If the user specified host or port without binlog or position,
1097
    reset binlog's name to FIRST and position to 4.
unknown's avatar
foo1  
unknown committed
1098
  */
1099

1100 1101
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
  {
1102
    mi->master_log_name[0] = 0;
1103
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
1104
  }
1105

1106
  if (lex_mi->log_file_name)
1107
    strmake(mi->master_log_name, lex_mi->log_file_name,
1108
	    sizeof(mi->master_log_name)-1);
1109
  if (lex_mi->pos)
unknown's avatar
unknown committed
1110
  {
1111
    mi->master_log_pos= lex_mi->pos;
unknown's avatar
unknown committed
1112
  }
unknown's avatar
unknown committed
1113
  DBUG_PRINT("info", ("master_log_pos: %lu", (ulong) mi->master_log_pos));
1114 1115

  if (lex_mi->host)
1116
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
1117
  if (lex_mi->user)
1118
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
1119
  if (lex_mi->password)
1120
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
1121
  if (lex_mi->port)
1122
    mi->port = lex_mi->port;
1123
  if (lex_mi->connect_retry)
1124
    mi->connect_retry = lex_mi->connect_retry;
unknown's avatar
foo1  
unknown committed
1125

unknown's avatar
unknown committed
1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140
  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 )
unknown's avatar
foo1  
unknown committed
1141
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
unknown's avatar
unknown committed
1142 1143
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
#endif
1144

1145 1146
  if (lex_mi->relay_log_name)
  {
1147
    need_relay_log_purge= 0;
1148 1149 1150 1151
    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);
1152 1153 1154 1155
  }

  if (lex_mi->relay_log_pos)
  {
1156
    need_relay_log_purge= 0;
1157
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
1158 1159
  }

unknown's avatar
unknown committed
1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184
  /*
    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
1185 1186 1187
     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
1188 1189
             sizeof(mi->master_log_name)-1);
  }
unknown's avatar
unknown committed
1190 1191 1192 1193
  /*
    Relay log's IO_CACHE may not be inited, if rli->inited==0 (server was never
    a slave before).
  */
1194 1195 1196 1197 1198 1199
  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);
  }
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;
unknown's avatar
unknown committed
1230
  DBUG_PRINT("info", ("master_log_pos: %lu", (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 1310

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

1315 1316 1317
  Format_description_log_event *description_event= new
    Format_description_log_event(3); /* MySQL 4.0 by default */

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

1330 1331 1332
    unit->set_limit(thd->lex->current_select);
    limit_start= unit->offset_limit_cnt;
    limit_end= unit->select_limit_cnt;
unknown's avatar
unknown committed
1333

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

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

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

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

1352
    pthread_mutex_lock(log_lock);
1353

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

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

unknown's avatar
unknown committed
1375 1376
    my_b_seek(&log, pos);

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

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

      pos = my_b_tell(&log);
unknown's avatar
unknown committed
1396 1397 1398 1399 1400 1401 1402 1403 1404
      delete ev;

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

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

1409
    pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1410 1411
  }

1412 1413
  ret= FALSE;

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

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

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

unknown's avatar
unknown committed
1436

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

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

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

1469

1470
/*
unknown's avatar
unknown committed
1471
  Send a list of all binary logs to client
1472 1473 1474 1475 1476 1477

  SYNOPSIS
    show_binlogs()
    thd		Thread specific variable

  RETURN VALUES
unknown's avatar
unknown committed
1478 1479
    FALSE OK
    TRUE  error
1480 1481
*/

unknown's avatar
unknown committed
1482
bool show_binlogs(THD* thd)
unknown's avatar
unknown committed
1483
{
1484
  IO_CACHE *index_file;
1485 1486
  LOG_INFO cur;
  File file;
unknown's avatar
unknown committed
1487 1488
  char fname[FN_REFLEN];
  List<Item> field_list;
1489
  uint length;
unknown's avatar
unknown committed
1490
  int cur_dir_len;
1491 1492
  Protocol *protocol= thd->protocol;
  DBUG_ENTER("show_binlogs");
1493 1494

  if (!mysql_bin_log.is_open())
1495
  {
unknown's avatar
unknown committed
1496
    my_message(ER_NO_BINARY_LOGGING, ER(ER_NO_BINARY_LOGGING), MYF(0));
1497
    return 1;
1498
  }
unknown's avatar
unknown committed
1499

1500
  field_list.push_back(new Item_empty_string("Log_name", 255));
unknown's avatar
foo1  
unknown committed
1501
  field_list.push_back(new Item_return_int("File_size", 20,
1502
                                           MYSQL_TYPE_LONGLONG));
1503 1504
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1505
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1506 1507
  
  pthread_mutex_lock(mysql_bin_log.get_log_lock());
unknown's avatar
unknown committed
1508
  mysql_bin_log.lock_index();
1509
  index_file=mysql_bin_log.get_index_file();
unknown's avatar
unknown committed
1510 1511 1512 1513
  
  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
1514
  cur_dir_len= dirname_length(cur.log_file_name);
1515

1516 1517 1518 1519
  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)
1520
  {
unknown's avatar
unknown committed
1521 1522 1523
    int dir_len;
    ulonglong file_length= 0;                   // Length if open fails
    fname[--length] = '\0';                     // remove the newline
1524

1525
    protocol->prepare_for_resend();
unknown's avatar
unknown committed
1526 1527 1528 1529 1530 1531 1532
    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
1533 1534
    {
      /* this is an old log, open it and find the size */
1535
      if ((file= my_open(fname, O_RDONLY | O_SHARE | O_BINARY,
unknown's avatar
unknown committed
1536
                         MYF(0))) >= 0)
1537
      {
unknown's avatar
unknown committed
1538
        file_length= (ulonglong) my_seek(file, 0L, MY_SEEK_END, MYF(0));
1539 1540 1541
        my_close(file, MYF(0));
      }
    }
unknown's avatar
unknown committed
1542
    protocol->store(file_length);
1543
    if (protocol->write())
1544
      goto err;
1545
  }
unknown's avatar
unknown committed
1546
  mysql_bin_log.unlock_index();
1547
  send_eof(thd);
unknown's avatar
unknown committed
1548
  DBUG_RETURN(FALSE);
1549

1550 1551
err:
  mysql_bin_log.unlock_index();
unknown's avatar
unknown committed
1552
  DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1553 1554
}

1555

1556 1557
int log_loaded_block(IO_CACHE* file)
{
1558
  LOAD_FILE_INFO *lf_info;
1559
  uint block_len ;
unknown's avatar
unknown committed
1560 1561 1562

  /* file->request_pos contains position where we started last read */
  char* buffer = (char*) file->request_pos;
unknown's avatar
unknown committed
1563
  if (!(block_len = (char*) file->read_end - (char*) buffer))
1564
    return 0;
unknown's avatar
unknown committed
1565
  lf_info = (LOAD_FILE_INFO*) file->arg;
1566 1567 1568 1569 1570 1571
  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
1572 1573
    Append_block_log_event a(lf_info->thd, lf_info->thd->db, buffer,
                             block_len, lf_info->log_delayed);
1574 1575 1576 1577
    mysql_bin_log.write(&a);
  }
  else
  {
unknown's avatar
unknown committed
1578 1579 1580 1581
    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);
1582
    lf_info->wrote_create_file = 1;
1583
    DBUG_SYNC_POINT("debug_lock.created_file_event",10);
1584 1585 1586
  }
  return 0;
}
1587

unknown's avatar
SCRUM  
unknown committed
1588
#endif /* HAVE_REPLICATION */
1589 1590