sql_repl.cc 46.4 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
  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 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

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);
unknown's avatar
unknown committed
434

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

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

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

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

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

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

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

605
	  TODO:
606 607 608
          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.
609
	*/
610

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

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

	default:
637
	  pthread_mutex_unlock(log_lock);
638 639
	  fatal_error = 1;
	  break;
640
	}
641

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

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

668
	if (fatal_error)
669 670
	{
	  errmsg = "error reading log entry";
671
          my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
672
	  goto err;
673
	}
674 675
	log.error=0;
      }
676
    }
677 678 679
    else
    {
      bool loop_breaker = 0;
680
      /* need this to break out of the for loop from switch */
681

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

695 696 697 698 699
      if (loop_breaker)
        break;
      
      end_io_cache(&log);
      (void) my_close(file, MYF(MY_WME));
700

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

unknown's avatar
unknown committed
718
      packet->length(0);
719
      packet->append('\0');
720
    }
721
  }
722

unknown's avatar
unknown committed
723
end:
724 725
  end_io_cache(&log);
  (void)my_close(file, MYF(MY_WME));
726

727
  send_eof(thd);
728
  thd->proc_info = "Waiting to finalize termination";
unknown's avatar
unknown committed
729 730 731
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
732
  DBUG_VOID_RETURN;
733

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

unknown's avatar
unknown committed
750
  my_message(my_errno, errmsg, MYF(0));
751 752 753
  DBUG_VOID_RETURN;
}

754
int start_slave(THD* thd , MASTER_INFO* mi,  bool net_report)
755
{
756
  int slave_errno= 0;
757
  int thread_mask;
758
  DBUG_ENTER("start_slave");
759

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

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

830
          /* mark the cached result of the UNTIL comparison as "undefined" */
831
          mi->rli.until_log_names_cmp_result=
832 833 834 835
            RELAY_LOG_INFO::UNTIL_LOG_NAMES_CMP_UNKNOWN;

          /* Issuing warning then started without --skip-slave-start */
          if (!opt_skip_slave_start)
836 837
            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
                         ER_MISSING_SKIP_SLAVE, 
838 839
                         ER(ER_MISSING_SKIP_SLAVE));
        }
840

841 842
        pthread_mutex_unlock(&mi->rli.data_lock);
      }
unknown's avatar
unknown committed
843
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
844
        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
845
                     ER(ER_UNTIL_COND_IGNORED));
846

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

866 867
  if (slave_errno)
  {
868
    if (net_report)
unknown's avatar
unknown committed
869
      my_message(slave_errno, ER(slave_errno), MYF(0));
870
    DBUG_RETURN(1);
871 872
  }
  else if (net_report)
873
    send_ok(thd);
874

875
  DBUG_RETURN(0);
876 877
}

878

879
int stop_slave(THD* thd, MASTER_INFO* mi, bool net_report )
880
{
881
  int slave_errno;
882 883
  if (!thd)
    thd = current_thd;
884

885
  if (check_access(thd, SUPER_ACL, any_db,0,0,0,0))
886
    return 1;
887 888 889
  thd->proc_info = "Killing slave";
  int thread_mask;
  lock_slave_threads(mi);
890
  // Get a mask of _running_ threads
891
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
892 893 894 895 896 897
  /*
    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
898 899
  if (thd->lex->slave_thd_opt)
    thread_mask &= thd->lex->slave_thd_opt;
900 901 902 903 904 905 906 907 908 909 910 911 912

  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));
  }
913
  unlock_slave_threads(mi);
914 915
  thd->proc_info = 0;

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

  return 0;
}

928 929 930 931 932 933 934 935 936 937 938 939 940 941 942

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


943
int reset_slave(THD *thd, MASTER_INFO* mi)
944 945 946
{
  MY_STAT stat_area;
  char fname[FN_REFLEN];
947 948
  int thread_mask= 0, error= 0;
  uint sql_errno=0;
949
  const char* errmsg=0;
unknown's avatar
unknown committed
950 951
  DBUG_ENTER("reset_slave");

952
  lock_slave_threads(mi);
953 954 955 956 957 958 959
  init_thread_mask(&thread_mask,mi,0 /* not inverse */);
  if (thread_mask) // We refuse if any slave thread is running
  {
    sql_errno= ER_SLAVE_MUST_STOP;
    error=1;
    goto err;
  }
unknown's avatar
unknown committed
960 961 962

  ha_reset_slave(thd);

963
  // delete relay logs, clear relay log coordinates
964 965 966
  if ((error= purge_relay_logs(&mi->rli, thd,
			       1 /* just reset */,
			       &errmsg)))
967
    goto err;
968

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

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

1001 1002
err:
  unlock_slave_threads(mi);
unknown's avatar
unknown committed
1003
  if (error)
1004
    my_error(sql_errno, MYF(0), errmsg);
unknown's avatar
unknown committed
1005
  DBUG_RETURN(error);
1006 1007
}

1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026
/*

  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

*/
  
1027

1028 1029 1030
void kill_zombie_dump_threads(uint32 slave_server_id)
{
  pthread_mutex_lock(&LOCK_thread_count);
1031 1032 1033
  I_List_iterator<THD> it(threads);
  THD *tmp;

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

1056

unknown's avatar
unknown committed
1057
bool change_master(THD* thd, MASTER_INFO* mi)
1058
{
1059
  int thread_mask;
unknown's avatar
unknown committed
1060 1061
  const char* errmsg= 0;
  bool need_relay_log_purge= 1;
1062 1063
  DBUG_ENTER("change_master");

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

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

1084 1085 1086 1087
  /*
    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
1088
  */
1089 1090 1091 1092 1093 1094

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

1095 1096
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
  {
1097
    mi->master_log_name[0] = 0;
1098
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
1099
  }
1100

1101
  if (lex_mi->log_file_name)
1102
    strmake(mi->master_log_name, lex_mi->log_file_name,
1103
	    sizeof(mi->master_log_name)-1);
1104
  if (lex_mi->pos)
unknown's avatar
unknown committed
1105
  {
1106
    mi->master_log_pos= lex_mi->pos;
unknown's avatar
unknown committed
1107
  }
1108
  DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos));
1109 1110

  if (lex_mi->host)
1111
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
1112
  if (lex_mi->user)
1113
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
1114
  if (lex_mi->password)
1115
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
1116
  if (lex_mi->port)
1117
    mi->port = lex_mi->port;
1118
  if (lex_mi->connect_retry)
1119
    mi->connect_retry = lex_mi->connect_retry;
unknown's avatar
unknown committed
1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138
 
  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
1139

1140 1141
  if (lex_mi->relay_log_name)
  {
1142
    need_relay_log_purge= 0;
1143 1144 1145 1146
    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);
1147 1148 1149 1150
  }

  if (lex_mi->relay_log_pos)
  {
1151
    need_relay_log_purge= 0;
1152
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
1153 1154
  }

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

  /*
    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
1237
  mi->rli.group_master_log_pos= mi->master_log_pos;
1238 1239
  strmake(mi->rli.group_master_log_name,mi->master_log_name,
	  sizeof(mi->rli.group_master_log_name)-1);
unknown's avatar
unknown committed
1240

1241 1242
  if (!mi->rli.group_master_log_name[0]) // uninitialized case
    mi->rli.group_master_log_pos=0;
1243 1244

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

1260
  unlock_slave_threads(mi);
1261
  thd->proc_info = 0;
1262
  send_ok(thd);
unknown's avatar
unknown committed
1263
  DBUG_RETURN(FALSE);
1264 1265
}

1266
int reset_master(THD* thd)
1267
{
1268
  if (!mysql_bin_log.is_open())
1269
  {
unknown's avatar
unknown committed
1270 1271
    my_message(ER_FLUSH_MASTER_BINLOG_CLOSED,
               ER(ER_FLUSH_MASTER_BINLOG_CLOSED), MYF(ME_BELL+ME_WAITTANG));
1272
    return 1;
1273
  }
1274
  return mysql_bin_log.reset_logs(thd);
1275 1276
}

unknown's avatar
unknown committed
1277 1278 1279 1280
int cmp_master_pos(const char* log_file_name1, ulonglong log_pos1,
		   const char* log_file_name2, ulonglong log_pos2)
{
  int res;
1281 1282
  uint log_file_name1_len=  strlen(log_file_name1);
  uint log_file_name2_len=  strlen(log_file_name2);
unknown's avatar
unknown committed
1283

1284 1285 1286 1287 1288 1289 1290 1291
  //  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
1292 1293
}

unknown's avatar
unknown committed
1294

1295
bool mysql_show_binlog_events(THD* thd)
unknown's avatar
unknown committed
1296
{
1297
  Protocol *protocol= thd->protocol;
1298
  DBUG_ENTER("mysql_show_binlog_events");
unknown's avatar
unknown committed
1299
  List<Item> field_list;
1300
  const char *errmsg = 0;
1301
  bool ret = TRUE;
unknown's avatar
unknown committed
1302 1303
  IO_CACHE log;
  File file = -1;
1304 1305

  Log_event::init_show_field_list(&field_list);
1306 1307
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1308
    DBUG_RETURN(TRUE);
1309

1310 1311 1312
  Format_description_log_event *description_event= new
    Format_description_log_event(3); /* MySQL 4.0 by default */

unknown's avatar
unknown committed
1313 1314 1315 1316 1317 1318 1319
  /*
    Wait for handlers to insert any pending information
    into the binlog.  For e.g. ndb which updates the binlog asynchronously
    this is needed so that the uses sees all its own commands in the binlog
  */
  ha_binlog_wait(thd);

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

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

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

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

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

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

1354
    pthread_mutex_lock(log_lock);
1355

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

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

unknown's avatar
unknown committed
1377 1378
    my_b_seek(&log, pos);

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

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

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

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

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

1411
    pthread_mutex_unlock(log_lock);
unknown's avatar
unknown committed
1412 1413
  }

1414 1415
  ret= FALSE;

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

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

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

unknown's avatar
unknown committed
1438

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

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

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

1471

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

  SYNOPSIS
    show_binlogs()
    thd		Thread specific variable

  RETURN VALUES
unknown's avatar
unknown committed
1480 1481
    FALSE OK
    TRUE  error
1482 1483
*/

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

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

1502
  field_list.push_back(new Item_empty_string("Log_name", 255));
1503 1504
  field_list.push_back(new Item_return_int("File_size", 20, 
                                           MYSQL_TYPE_LONGLONG));
1505 1506
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
unknown's avatar
unknown committed
1507
    DBUG_RETURN(TRUE);
unknown's avatar
unknown committed
1508
  mysql_bin_log.lock_index();
1509
  index_file=mysql_bin_log.get_index_file();
1510 1511

  mysql_bin_log.get_current_log(&cur);
unknown's avatar
unknown committed
1512
  cur_dir_len= dirname_length(cur.log_file_name);
1513

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

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

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

1553

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

  /* file->request_pos contains position where we started last read */
  char* buffer = (char*) file->request_pos;
unknown's avatar
unknown committed
1561
  if (!(block_len = (char*) file->read_end - (char*) buffer))
1562
    return 0;
unknown's avatar
unknown committed
1563
  lf_info = (LOAD_FILE_INFO*) file->arg;
1564 1565
  if (lf_info->thd->current_stmt_binlog_row_based)
    return 0;
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