sql_repl.cc 46.3 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"
hf@deer.mysql.r18.ru's avatar
SCRUM  
hf@deer.mysql.r18.ru committed
18 19
#ifdef HAVE_REPLICATION

20 21
#include "sql_repl.h"
#include "log_event.h"
22
#include <my_dir.h>
23

24
int max_binlog_dump_events = 0; // unlimited
25
my_bool opt_sporadic_binlog_dump_fail = 0;
26
static int binlog_dump_count = 0;
sasha@mysql.sashanet.com's avatar
sasha@mysql.sashanet.com committed
27

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

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

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

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

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

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

80 81 82 83 84 85
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;
86 87
  int old_timeout;
  uint packet_len;
88
  char buf[IO_SIZE];				// It's safe to alloc this
89 90
  DBUG_ENTER("send_file");

91 92 93 94 95 96 97 98 99 100 101
  /*
    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
  */
sasha@mysql.sashanet.com's avatar
sasha@mysql.sashanet.com committed
102
  if (net_flush(net) || (packet_len = my_net_read(net)) == packet_error)
103
  {
104
    errmsg = "while reading file name";
105 106 107
    goto err;
  }

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

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

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

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

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

151

152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
/*
  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.
*/

173 174 175
void adjust_linfo_offsets(my_off_t purge_offset)
{
  THD *tmp;
176

177 178 179
  pthread_mutex_lock(&LOCK_thread_count);
  I_List_iterator<THD> it(threads);

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

monty@narttu.mysql.fi's avatar
merge  
monty@narttu.mysql.fi committed
201

202 203 204 205 206
bool log_in_use(const char* log_name)
{
  int log_name_len = strlen(log_name) + 1;
  THD *tmp;
  bool result = 0;
207

208 209
  pthread_mutex_lock(&LOCK_thread_count);
  I_List_iterator<THD> it(threads);
210 211 212 213 214

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

  pthread_mutex_unlock(&LOCK_thread_count);
  return result;
}

228
bool purge_error_message(THD* thd, int res)
229
{
230
  uint errmsg= 0;
231

monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
232
  switch (res)  {
233
  case 0: break;
234 235 236 237 238 239 240 241
  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;
242 243 244 245
  }

  if (errmsg)
  {
bell@sanja.is.com.ua's avatar
bell@sanja.is.com.ua committed
246
    my_message(errmsg, ER(errmsg), MYF(0));
247
    return TRUE;
248
  }
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
249
  send_ok(thd);
250
  return FALSE;
251 252
}

monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
253

254
bool purge_master_logs(THD* thd, const char* to_log)
255 256
{
  char search_file_name[FN_REFLEN];
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
257 258
  if (!mysql_bin_log.is_open())
  {
ram@gw.mysql.r18.ru's avatar
ram@gw.mysql.r18.ru committed
259
    send_ok(thd);
260
    return FALSE;
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
261
  }
262 263

  mysql_bin_log.make_log_name(search_file_name, to_log);
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
264 265
  return purge_error_message(thd,
			     mysql_bin_log.purge_logs(search_file_name, 0, 1,
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
266
						      1, NULL));
267 268 269
}


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

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


311 312 313
/*
  TODO: Clean up loop to only have one call to send_file()
*/
monty@narttu.mysql.fi's avatar
merge  
monty@narttu.mysql.fi committed
314

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

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

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

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

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

365 366
  linfo.index_file_offset = 0;
  thd->current_linfo = &linfo;
367

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

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

heikki@hundin.mysql.fi's avatar
heikki@hundin.mysql.fi committed
388 389 390
  if (thd->variables.sync_replication)
    ha_repl_report_sent_binlog(thd, log_file_name, pos);

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

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

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

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

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

505 506 507
  /* seek to the requested position, to start the requested dump */
  my_b_seek(&log, pos);			// Seek will done on next read

monty@donna.mysql.com's avatar
monty@donna.mysql.com committed
508
  while (!net->error && net->vio != 0 && !thd->killed)
509 510 511
  {
    while (!(error = Log_event::read_log_event(&log, packet, log_lock)))
    {
512
#ifndef DBUG_OFF
513
      if (max_binlog_dump_events && !left_events--)
514 515 516
      {
	net_flush(net);
	errmsg = "Debugging binlog dump abort";
517
	my_errno= ER_UNKNOWN_ERROR;
518 519
	goto err;
      }
520
#endif
521 522

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

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

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

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

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

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

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

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

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

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

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

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

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

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

heikki@hundin.mysql.fi's avatar
heikki@hundin.mysql.fi committed
654 655 656
          if (thd->variables.sync_replication)
            ha_repl_report_sent_binlog(thd, log_file_name, my_b_tell(&log));

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

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

699
      if (loop_breaker)
700 701 702 703
	break;

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

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

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

725
      packet->length(0);
726
      packet->append('\0');
727
    }
728
  }
729

vva@eagle.mysql.r18.ru's avatar
vva@eagle.mysql.r18.ru committed
730
end:
heikki@hundin.mysql.fi's avatar
heikki@hundin.mysql.fi committed
731 732 733
  if (thd->variables.sync_replication)
    ha_repl_report_replication_stop(thd);

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

737
  send_eof(thd);
738
  thd->proc_info = "Waiting to finalize termination";
739 740 741
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
742
  DBUG_VOID_RETURN;
743

vva@eagle.mysql.r18.ru's avatar
vva@eagle.mysql.r18.ru committed
744
err:
heikki@hundin.mysql.fi's avatar
heikki@hundin.mysql.fi committed
745 746 747
  if (thd->variables.sync_replication)
    ha_repl_report_replication_stop(thd);

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

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

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

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

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

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

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

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

878 879
  if (slave_errno)
  {
880
    if (net_report)
bell@sanja.is.com.ua's avatar
bell@sanja.is.com.ua committed
881
      my_message(slave_errno, ER(slave_errno), MYF(0));
882
    DBUG_RETURN(1);
883 884
  }
  else if (net_report)
885
    send_ok(thd);
886

887
  DBUG_RETURN(0);
888 889
}

890

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

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

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

928 929 930
  if (slave_errno)
  {
    if (net_report)
bell@sanja.is.com.ua's avatar
bell@sanja.is.com.ua committed
931
      my_message(slave_errno, ER(slave_errno), MYF(0));
932 933 934
    return 1;
  }
  else if (net_report)
935
    send_ok(thd);
936 937 938 939

  return 0;
}

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

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


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

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

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

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

1010 1011
err:
  unlock_slave_threads(mi);
bell@sanja.is.com.ua's avatar
bell@sanja.is.com.ua committed
1012
  if (error)
1013
    my_error(sql_errno, MYF(0), errmsg);
1014
  DBUG_RETURN(error);
1015 1016
}

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

  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

*/
  
1036

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

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

1065

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

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

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

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

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

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

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

  if (lex_mi->host)
1120
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
1121
  if (lex_mi->user)
1122
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
1123
  if (lex_mi->password)
1124
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
1125
  if (lex_mi->port)
1126
    mi->port = lex_mi->port;
1127
  if (lex_mi->connect_retry)
1128
    mi->connect_retry = lex_mi->connect_retry;
1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147
 
  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
1148

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

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

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

  /*
    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.
  */
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
1241
  mi->rli.group_master_log_pos= mi->master_log_pos;
1242 1243
  strmake(mi->rli.group_master_log_name,mi->master_log_name,
	  sizeof(mi->rli.group_master_log_name)-1);
guilhem@mysql.com's avatar
guilhem@mysql.com committed
1244

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

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

1264
  unlock_slave_threads(mi);
1265
  thd->proc_info = 0;
1266
  send_ok(thd);
1267
  DBUG_RETURN(FALSE);
1268 1269
}

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

1281 1282 1283 1284
int cmp_master_pos(const char* log_file_name1, ulonglong log_pos1,
		   const char* log_file_name2, ulonglong log_pos2)
{
  int res;
1285 1286
  uint log_file_name1_len=  strlen(log_file_name1);
  uint log_file_name2_len=  strlen(log_file_name2);
monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
1287

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

monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
1298

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

  Log_event::init_show_field_list(&field_list);
1312 1313
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
1314
    DBUG_RETURN(TRUE);
1315

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

1328 1329 1330
    unit->set_limit(thd->lex->current_select);
    limit_start= unit->offset_limit_cnt;
    limit_end= unit->select_limit_cnt;
1331

1332
    name= search_file_name;
1333 1334 1335
    if (log_file_name)
      mysql_bin_log.make_log_name(search_file_name, log_file_name);
    else
1336
      name=0;					// Find first log
1337 1338 1339

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

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

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

1350
    pthread_mutex_lock(log_lock);
1351

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

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

1373 1374
    my_b_seek(&log, pos);

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

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

      pos = my_b_tell(&log);
1394 1395 1396 1397 1398 1399 1400 1401 1402
      delete ev;

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

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

1407
    pthread_mutex_unlock(log_lock);
1408 1409
  }

1410 1411
  ret= FALSE;

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

1420 1421
  if (errmsg)
  {
1422 1423
    my_error(ER_ERROR_WHEN_EXECUTING_COMMAND, MYF(0),
             "SHOW BINLOG EVENTS", errmsg);
1424
    DBUG_RETURN(TRUE);
1425 1426
  }

1427
  send_eof(thd);
1428 1429 1430
  pthread_mutex_lock(&LOCK_thread_count);
  thd->current_linfo = 0;
  pthread_mutex_unlock(&LOCK_thread_count);
1431
  DBUG_RETURN(ret);
1432 1433
}

monty@mashka.mysql.fi's avatar
monty@mashka.mysql.fi committed
1434

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

1446 1447
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
1448
    DBUG_RETURN(TRUE);
1449 1450
  protocol->prepare_for_resend();

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

1467

1468
/*
guilhem@mysql.com's avatar
guilhem@mysql.com committed
1469
  Send a list of all binary logs to client
1470 1471 1472 1473 1474 1475

  SYNOPSIS
    show_binlogs()
    thd		Thread specific variable

  RETURN VALUES
1476 1477
    FALSE OK
    TRUE  error
1478 1479
*/

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

  if (!mysql_bin_log.is_open())
1493
  {
bell@sanja.is.com.ua's avatar
bell@sanja.is.com.ua committed
1494
    my_message(ER_NO_BINARY_LOGGING, ER(ER_NO_BINARY_LOGGING), MYF(0));
1495
    return 1;
1496
  }
1497

1498
  field_list.push_back(new Item_empty_string("Log_name", 255));
1499 1500
  field_list.push_back(new Item_return_int("File_size", 20, 
                                           MYSQL_TYPE_LONGLONG));
1501 1502
  if (protocol->send_fields(&field_list,
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
1503
    DBUG_RETURN(TRUE);
1504
  mysql_bin_log.lock_index();
1505
  index_file=mysql_bin_log.get_index_file();
1506 1507

  mysql_bin_log.get_current_log(&cur);
monty@mysql.com's avatar
monty@mysql.com committed
1508
  cur_dir_len= dirname_length(cur.log_file_name);
1509

1510 1511 1512 1513
  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)
1514
  {
monty@mysql.com's avatar
monty@mysql.com committed
1515 1516 1517
    int dir_len;
    ulonglong file_length= 0;                   // Length if open fails
    fname[--length] = '\0';                     // remove the newline
1518

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

1544 1545
err:
  mysql_bin_log.unlock_index();
1546
  DBUG_RETURN(TRUE);
1547 1548
}

1549

1550 1551
int log_loaded_block(IO_CACHE* file)
{
1552
  LOAD_FILE_INFO *lf_info;
1553
  uint block_len ;
monty@bitch.mysql.fi's avatar
monty@bitch.mysql.fi committed
1554 1555 1556

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

hf@deer.mysql.r18.ru's avatar
SCRUM  
hf@deer.mysql.r18.ru committed
1582
#endif /* HAVE_REPLICATION */
1583 1584