← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/sql_repl.cc

Fixed bad include

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/sql_repl.cc
 
1
/* Copyright (C) 2000-2006 MySQL AB & Sasha
 
2
 
 
3
   This program is free software; you can redistribute it and/or modify
 
4
   it under the terms of the GNU General Public License as published by
 
5
   the Free Software Foundation; version 2 of the License.
 
6
 
 
7
   This program is distributed in the hope that it will be useful,
 
8
   but WITHOUT ANY WARRANTY; without even the implied warranty of
 
9
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
10
   GNU General Public License for more details.
 
11
 
 
12
   You should have received a copy of the GNU General Public License
 
13
   along with this program; if not, write to the Free Software
 
14
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
15
 
 
16
#include <drizzled/server_includes.h>
 
17
 
 
18
#ifdef HAVE_REPLICATION
 
19
#include "rpl_mi.h"
 
20
#include "sql_repl.h"
 
21
#include "log_event.h"
 
22
#include "rpl_filter.h"
 
23
#include <drizzled/drizzled_error_messages.h>
 
24
 
 
25
int max_binlog_dump_events = 0; // unlimited
 
26
 
 
27
/*
 
28
    fake_rotate_event() builds a fake (=which does not exist physically in any
 
29
    binlog) Rotate event, which contains the name of the binlog we are going to
 
30
    send to the slave (because the slave may not know it if it just asked for
 
31
    MASTER_LOG_FILE='', MASTER_LOG_POS=4).
 
32
    < 4.0.14, fake_rotate_event() was called only if the requested pos was 4.
 
33
    After this version we always call it, so that a 3.23.58 slave can rely on
 
34
    it to detect if the master is 4.0 (and stop) (the _fake_ Rotate event has
 
35
    zeros in the good positions which, by chance, make it possible for the 3.23
 
36
    slave to detect that this event is unexpected) (this is luck which happens
 
37
    because the master and slave disagree on the size of the header of
 
38
    Log_event).
 
39
 
 
40
    Relying on the event length of the Rotate event instead of these
 
41
    well-placed zeros was not possible as Rotate events have a variable-length
 
42
    part.
 
43
*/
 
44
static int fake_rotate_event(NET* net, String* packet, char* log_file_name,
 
45
                             uint64_t position, const char** errmsg)
 
46
{
 
47
  char header[LOG_EVENT_HEADER_LEN], buf[ROTATE_HEADER_LEN+100];
 
48
  /*
 
49
    'when' (the timestamp) is set to 0 so that slave could distinguish between
 
50
    real and fake Rotate events (if necessary)
 
51
  */
 
52
  memset(header, 0, 4);
 
53
  header[EVENT_TYPE_OFFSET] = ROTATE_EVENT;
 
54
 
 
55
  char* p = log_file_name+dirname_length(log_file_name);
 
56
  uint ident_len = (uint) strlen(p);
 
57
  uint32_t event_len = ident_len + LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN;
 
58
  int4store(header + SERVER_ID_OFFSET, server_id);
 
59
  int4store(header + EVENT_LEN_OFFSET, event_len);
 
60
  int2store(header + FLAGS_OFFSET, 0);
 
61
 
 
62
  // TODO: check what problems this may cause and fix them
 
63
  int4store(header + LOG_POS_OFFSET, 0);
 
64
 
 
65
  packet->append(header, sizeof(header));
 
66
  int8store(buf+R_POS_OFFSET,position);
 
67
  packet->append(buf, ROTATE_HEADER_LEN);
 
68
  packet->append(p,ident_len);
 
69
  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
 
70
  {
 
71
    *errmsg = "failed on my_net_write()";
 
72
    return(-1);
 
73
  }
 
74
  return(0);
 
75
}
 
76
 
 
77
static int send_file(THD *thd)
 
78
{
 
79
  NET* net = &thd->net;
 
80
  int fd = -1, error = 1;
 
81
  size_t bytes;
 
82
  char fname[FN_REFLEN+1];
 
83
  const char *errmsg = 0;
 
84
  int old_timeout;
 
85
  unsigned long packet_len;
 
86
  uchar buf[IO_SIZE];                           // It's safe to alloc this
 
87
 
 
88
  /*
 
89
    The client might be slow loading the data, give him wait_timeout to do
 
90
    the job
 
91
  */
 
92
  old_timeout= net->read_timeout;
 
93
  my_net_set_read_timeout(net, thd->variables.net_wait_timeout);
 
94
 
 
95
  /*
 
96
    We need net_flush here because the client will not know it needs to send
 
97
    us the file name until it has processed the load event entry
 
98
  */
 
99
  if (net_flush(net) || (packet_len = my_net_read(net)) == packet_error)
 
100
  {
 
101
    errmsg = _("Failed in send_file() while reading file name");
 
102
    goto err;
 
103
  }
 
104
 
 
105
  // terminate with \0 for fn_format
 
106
  *((char*)net->read_pos +  packet_len) = 0;
 
107
  fn_format(fname, (char*) net->read_pos + 1, "", "", 4);
 
108
  // this is needed to make replicate-ignore-db
 
109
  if (!strcmp(fname,"/dev/null"))
 
110
    goto end;
 
111
 
 
112
  if ((fd = my_open(fname, O_RDONLY, MYF(0))) < 0)
 
113
  {
 
114
    errmsg = _("Failed in send_file() on open of file");
 
115
    goto err;
 
116
  }
 
117
 
 
118
  while ((long) (bytes= my_read(fd, buf, IO_SIZE, MYF(0))) > 0)
 
119
  {
 
120
    if (my_net_write(net, buf, bytes))
 
121
    {
 
122
      errmsg = _("Failed in send_file() while writing data to client");
 
123
      goto err;
 
124
    }
 
125
  }
 
126
 
 
127
 end:
 
128
  if (my_net_write(net, (uchar*) "", 0) || net_flush(net) ||
 
129
      (my_net_read(net) == packet_error))
 
130
  {
 
131
    errmsg = _("Failed in send_file() while negotiating file transfer close");
 
132
    goto err;
 
133
  }
 
134
  error = 0;
 
135
 
 
136
 err:
 
137
  my_net_set_read_timeout(net, old_timeout);
 
138
  if (fd >= 0)
 
139
    (void) my_close(fd, MYF(0));
 
140
  if (errmsg)
 
141
  {
 
142
    sql_print_error(errmsg);
 
143
  }
 
144
  return(error);
 
145
}
 
146
 
 
147
 
 
148
/*
 
149
  Adjust the position pointer in the binary log file for all running slaves
 
150
 
 
151
  SYNOPSIS
 
152
    adjust_linfo_offsets()
 
153
    purge_offset        Number of bytes removed from start of log index file
 
154
 
 
155
  NOTES
 
156
    - This is called when doing a PURGE when we delete lines from the
 
157
      index log file
 
158
 
 
159
  REQUIREMENTS
 
160
    - Before calling this function, we have to ensure that no threads are
 
161
      using any binary log file before purge_offset.a
 
162
 
 
163
  TODO
 
164
    - Inform the slave threads that they should sync the position
 
165
      in the binary log file with flush_relay_log_info.
 
166
      Now they sync is done for next read.
 
167
*/
 
168
 
 
169
void adjust_linfo_offsets(my_off_t purge_offset)
 
170
{
 
171
  THD *tmp;
 
172
 
 
173
  pthread_mutex_lock(&LOCK_thread_count);
 
174
  I_List_iterator<THD> it(threads);
 
175
 
 
176
  while ((tmp=it++))
 
177
  {
 
178
    LOG_INFO* linfo;
 
179
    if ((linfo = tmp->current_linfo))
 
180
    {
 
181
      pthread_mutex_lock(&linfo->lock);
 
182
      /*
 
183
        Index file offset can be less that purge offset only if
 
184
        we just started reading the index file. In that case
 
185
        we have nothing to adjust
 
186
      */
 
187
      if (linfo->index_file_offset < purge_offset)
 
188
        linfo->fatal = (linfo->index_file_offset != 0);
 
189
      else
 
190
        linfo->index_file_offset -= purge_offset;
 
191
      pthread_mutex_unlock(&linfo->lock);
 
192
    }
 
193
  }
 
194
  pthread_mutex_unlock(&LOCK_thread_count);
 
195
}
 
196
 
 
197
 
 
198
bool log_in_use(const char* log_name)
 
199
{
 
200
  int log_name_len = strlen(log_name) + 1;
 
201
  THD *tmp;
 
202
  bool result = 0;
 
203
 
 
204
  pthread_mutex_lock(&LOCK_thread_count);
 
205
  I_List_iterator<THD> it(threads);
 
206
 
 
207
  while ((tmp=it++))
 
208
  {
 
209
    LOG_INFO* linfo;
 
210
    if ((linfo = tmp->current_linfo))
 
211
    {
 
212
      pthread_mutex_lock(&linfo->lock);
 
213
      result = !memcmp(log_name, linfo->log_file_name, log_name_len);
 
214
      pthread_mutex_unlock(&linfo->lock);
 
215
      if (result)
 
216
        break;
 
217
    }
 
218
  }
 
219
 
 
220
  pthread_mutex_unlock(&LOCK_thread_count);
 
221
  return result;
 
222
}
 
223
 
 
224
bool purge_error_message(THD* thd, int res)
 
225
{
 
226
  uint errmsg= 0;
 
227
 
 
228
  switch (res)  {
 
229
  case 0: break;
 
230
  case LOG_INFO_EOF:    errmsg= ER_UNKNOWN_TARGET_BINLOG; break;
 
231
  case LOG_INFO_IO:     errmsg= ER_IO_ERR_LOG_INDEX_READ; break;
 
232
  case LOG_INFO_INVALID:errmsg= ER_BINLOG_PURGE_PROHIBITED; break;
 
233
  case LOG_INFO_SEEK:   errmsg= ER_FSEEK_FAIL; break;
 
234
  case LOG_INFO_MEM:    errmsg= ER_OUT_OF_RESOURCES; break;
 
235
  case LOG_INFO_FATAL:  errmsg= ER_BINLOG_PURGE_FATAL_ERR; break;
 
236
  case LOG_INFO_IN_USE: errmsg= ER_LOG_IN_USE; break;
 
237
  case LOG_INFO_EMFILE: errmsg= ER_BINLOG_PURGE_EMFILE; break;
 
238
  default:              errmsg= ER_LOG_PURGE_UNKNOWN_ERR; break;
 
239
  }
 
240
 
 
241
  if (errmsg)
 
242
  {
 
243
    my_message(errmsg, ER(errmsg), MYF(0));
 
244
    return true;
 
245
  }
 
246
  my_ok(thd);
 
247
  return false;
 
248
}
 
249
 
 
250
 
 
251
bool purge_master_logs(THD* thd, const char* to_log)
 
252
{
 
253
  char search_file_name[FN_REFLEN];
 
254
  if (!mysql_bin_log.is_open())
 
255
  {
 
256
    my_ok(thd);
 
257
    return false;
 
258
  }
 
259
 
 
260
  mysql_bin_log.make_log_name(search_file_name, to_log);
 
261
  return purge_error_message(thd,
 
262
                             mysql_bin_log.purge_logs(search_file_name, 0, 1,
 
263
                                                      1, NULL));
 
264
}
 
265
 
 
266
 
 
267
bool purge_master_logs_before_date(THD* thd, time_t purge_time)
 
268
{
 
269
  if (!mysql_bin_log.is_open())
 
270
  {
 
271
    my_ok(thd);
 
272
    return 0;
 
273
  }
 
274
  return purge_error_message(thd,
 
275
                             mysql_bin_log.purge_logs_before_date(purge_time));
 
276
}
 
277
 
 
278
int test_for_non_eof_log_read_errors(int error, const char **errmsg)
 
279
{
 
280
  if (error == LOG_READ_EOF)
 
281
    return 0;
 
282
  my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
283
  switch (error) {
 
284
  case LOG_READ_BOGUS:
 
285
    *errmsg = "bogus data in log event";
 
286
    break;
 
287
  case LOG_READ_TOO_LARGE:
 
288
    *errmsg = "log event entry exceeded max_allowed_packet; \
 
289
Increase max_allowed_packet on master";
 
290
    break;
 
291
  case LOG_READ_IO:
 
292
    *errmsg = "I/O error reading log event";
 
293
    break;
 
294
  case LOG_READ_MEM:
 
295
    *errmsg = "memory allocation failed reading log event";
 
296
    break;
 
297
  case LOG_READ_TRUNC:
 
298
    *errmsg = "binlog truncated in the middle of event";
 
299
    break;
 
300
  default:
 
301
    *errmsg = "unknown error reading log event on the master";
 
302
    break;
 
303
  }
 
304
  return error;
 
305
}
 
306
 
 
307
 
 
308
/**
 
309
  An auxiliary function for calling in mysql_binlog_send
 
310
  to initialize the heartbeat timeout in waiting for a binlogged event.
 
311
 
 
312
  @param[in]    thd  THD to access a user variable
 
313
 
 
314
  @return        heartbeat period an uint64_t of nanoseconds
 
315
                 or zero if heartbeat was not demanded by slave
 
316
*/ 
 
317
static uint64_t get_heartbeat_period(THD * thd)
 
318
{
 
319
  bool null_value;
 
320
  LEX_STRING name=  { C_STRING_WITH_LEN("master_heartbeat_period")};
 
321
  user_var_entry *entry= 
 
322
    (user_var_entry*) hash_search(&thd->user_vars, (uchar*) name.str,
 
323
                                  name.length);
 
324
  return entry? entry->val_int(&null_value) : 0;
 
325
}
 
326
 
 
327
/*
 
328
  Function prepares and sends repliation heartbeat event.
 
329
 
 
330
  @param net                net object of THD
 
331
  @param packet             buffer to store the heartbeat instance
 
332
  @param event_coordinates  binlog file name and position of the last
 
333
                            real event master sent from binlog
 
334
 
 
335
  @note 
 
336
    Among three essential pieces of heartbeat data Log_event::when
 
337
    is computed locally.
 
338
    The  error to send is serious and should force terminating
 
339
    the dump thread.
 
340
*/
 
341
static int send_heartbeat_event(NET* net, String* packet,
 
342
                                const struct event_coordinates *coord)
 
343
{
 
344
  char header[LOG_EVENT_HEADER_LEN];
 
345
  /*
 
346
    'when' (the timestamp) is set to 0 so that slave could distinguish between
 
347
    real and fake Rotate events (if necessary)
 
348
  */
 
349
  memset(header, 0, 4);  // when
 
350
 
 
351
  header[EVENT_TYPE_OFFSET] = HEARTBEAT_LOG_EVENT;
 
352
 
 
353
  char* p= coord->file_name + dirname_length(coord->file_name);
 
354
 
 
355
  uint ident_len = strlen(p);
 
356
  uint32_t event_len = ident_len + LOG_EVENT_HEADER_LEN;
 
357
  int4store(header + SERVER_ID_OFFSET, server_id);
 
358
  int4store(header + EVENT_LEN_OFFSET, event_len);
 
359
  int2store(header + FLAGS_OFFSET, 0);
 
360
 
 
361
  int4store(header + LOG_POS_OFFSET, coord->pos);  // log_pos
 
362
 
 
363
  packet->append(header, sizeof(header));
 
364
  packet->append(p, ident_len);             // log_file_name
 
365
 
 
366
  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) ||
 
367
      net_flush(net))
 
368
  {
 
369
    return(-1);
 
370
  }
 
371
  packet->set("\0", 1, &my_charset_bin);
 
372
  return(0);
 
373
}
 
374
 
 
375
/*
 
376
  TODO: Clean up loop to only have one call to send_file()
 
377
*/
 
378
 
 
379
void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
 
380
                       ushort flags)
 
381
{
 
382
  LOG_INFO linfo;
 
383
  char *log_file_name = linfo.log_file_name;
 
384
  char search_file_name[FN_REFLEN], *name;
 
385
  IO_CACHE log;
 
386
  File file = -1;
 
387
  String* packet = &thd->packet;
 
388
  int error;
 
389
  const char *errmsg = "Unknown error";
 
390
  NET* net = &thd->net;
 
391
  pthread_mutex_t *log_lock;
 
392
  bool binlog_can_be_corrupted= false;
 
393
 
 
394
  memset(&log, 0, sizeof(log));
 
395
  /* 
 
396
     heartbeat_period from @master_heartbeat_period user variable
 
397
  */
 
398
  uint64_t heartbeat_period= get_heartbeat_period(thd);
 
399
  struct timespec heartbeat_buf;
 
400
  struct event_coordinates coord_buf;
 
401
  struct timespec *heartbeat_ts= NULL;
 
402
  struct event_coordinates *coord= NULL;
 
403
  if (heartbeat_period != 0LL)
 
404
  {
 
405
    heartbeat_ts= &heartbeat_buf;
 
406
    set_timespec_nsec(*heartbeat_ts, 0);
 
407
    coord= &coord_buf;
 
408
    coord->file_name= log_file_name; // initialization basing on what slave remembers
 
409
    coord->pos= pos;
 
410
  }
 
411
 
 
412
  if (!mysql_bin_log.is_open())
 
413
  {
 
414
    errmsg = "Binary log is not open";
 
415
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
416
    goto err;
 
417
  }
 
418
  if (!server_id_supplied)
 
419
  {
 
420
    errmsg = "Misconfigured master - server id was not set";
 
421
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
422
    goto err;
 
423
  }
 
424
 
 
425
  name=search_file_name;
 
426
  if (log_ident[0])
 
427
    mysql_bin_log.make_log_name(search_file_name, log_ident);
 
428
  else
 
429
    name=0;                                     // Find first log
 
430
 
 
431
  linfo.index_file_offset = 0;
 
432
 
 
433
  if (mysql_bin_log.find_log_pos(&linfo, name, 1))
 
434
  {
 
435
    errmsg = "Could not find first log file name in binary log index file";
 
436
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
437
    goto err;
 
438
  }
 
439
 
 
440
  pthread_mutex_lock(&LOCK_thread_count);
 
441
  thd->current_linfo = &linfo;
 
442
  pthread_mutex_unlock(&LOCK_thread_count);
 
443
 
 
444
  if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)
 
445
  {
 
446
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
447
    goto err;
 
448
  }
 
449
  if (pos < BIN_LOG_HEADER_SIZE || pos > my_b_filelength(&log))
 
450
  {
 
451
    errmsg= "Client requested master to start replication from \
 
452
impossible position";
 
453
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
454
    goto err;
 
455
  }
 
456
 
 
457
  /*
 
458
    We need to start a packet with something other than 255
 
459
    to distinguish it from error
 
460
  */
 
461
  packet->set("\0", 1, &my_charset_bin); /* This is the start of a new packet */
 
462
 
 
463
  /*
 
464
    Tell the client about the log name with a fake Rotate event;
 
465
    this is needed even if we also send a Format_description_log_event
 
466
    just after, because that event does not contain the binlog's name.
 
467
    Note that as this Rotate event is sent before
 
468
    Format_description_log_event, the slave cannot have any info to
 
469
    understand this event's format, so the header len of
 
470
    Rotate_log_event is FROZEN (so in 5.0 it will have a header shorter
 
471
    than other events except FORMAT_DESCRIPTION_EVENT).
 
472
    Before 4.0.14 we called fake_rotate_event below only if (pos ==
 
473
    BIN_LOG_HEADER_SIZE), because if this is false then the slave
 
474
    already knows the binlog's name.
 
475
    Since, we always call fake_rotate_event; if the slave already knew
 
476
    the log's name (ex: CHANGE MASTER TO MASTER_LOG_FILE=...) this is
 
477
    useless but does not harm much. It is nice for 3.23 (>=.58) slaves
 
478
    which test Rotate events to see if the master is 4.0 (then they
 
479
    choose to stop because they can't replicate 4.0); by always calling
 
480
    fake_rotate_event we are sure that 3.23.58 and newer will detect the
 
481
    problem as soon as replication starts (BUG#198).
 
482
    Always calling fake_rotate_event makes sending of normal
 
483
    (=from-binlog) Rotate events a priori unneeded, but it is not so
 
484
    simple: the 2 Rotate events are not equivalent, the normal one is
 
485
    before the Stop event, the fake one is after. If we don't send the
 
486
    normal one, then the Stop event will be interpreted (by existing 4.0
 
487
    slaves) as "the master stopped", which is wrong. So for safety,
 
488
    given that we want minimum modification of 4.0, we send the normal
 
489
    and fake Rotates.
 
490
  */
 
491
  if (fake_rotate_event(net, packet, log_file_name, pos, &errmsg))
 
492
  {
 
493
    /*
 
494
       This error code is not perfect, as fake_rotate_event() does not
 
495
       read anything from the binlog; if it fails it's because of an
 
496
       error in my_net_write(), fortunately it will say so in errmsg.
 
497
    */
 
498
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
499
    goto err;
 
500
  }
 
501
  packet->set("\0", 1, &my_charset_bin);
 
502
  /*
 
503
    Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become
 
504
    this larger than the corresponding packet (query) sent 
 
505
    from client to master.
 
506
  */
 
507
  thd->variables.max_allowed_packet+= MAX_LOG_EVENT_HEADER;
 
508
 
 
509
  /*
 
510
    We can set log_lock now, it does not move (it's a member of
 
511
    mysql_bin_log, and it's already inited, and it will be destroyed
 
512
    only at shutdown).
 
513
  */
 
514
  log_lock = mysql_bin_log.get_log_lock();
 
515
  if (pos > BIN_LOG_HEADER_SIZE)
 
516
  {
 
517
     /*
 
518
       Try to find a Format_description_log_event at the beginning of
 
519
       the binlog
 
520
     */
 
521
     if (!(error = Log_event::read_log_event(&log, packet, log_lock)))
 
522
     {
 
523
       /*
 
524
         The packet has offsets equal to the normal offsets in a binlog
 
525
         event +1 (the first character is \0).
 
526
       */
 
527
       if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
 
528
       {
 
529
         binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
 
530
                                       LOG_EVENT_BINLOG_IN_USE_F);
 
531
         (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
 
532
         /*
 
533
           mark that this event with "log_pos=0", so the slave
 
534
           should not increment master's binlog position
 
535
           (rli->group_master_log_pos)
 
536
         */
 
537
         int4store((char*) packet->ptr()+LOG_POS_OFFSET+1, 0);
 
538
         /*
 
539
           if reconnect master sends FD event with `created' as 0
 
540
           to avoid destroying temp tables.
 
541
          */
 
542
         int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+
 
543
                   ST_CREATED_OFFSET+1, (uint32_t) 0);
 
544
         /* send it */
 
545
         if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
 
546
         {
 
547
           errmsg = "Failed on my_net_write()";
 
548
           my_errno= ER_UNKNOWN_ERROR;
 
549
           goto err;
 
550
         }
 
551
 
 
552
         /*
 
553
           No need to save this event. We are only doing simple reads
 
554
           (no real parsing of the events) so we don't need it. And so
 
555
           we don't need the artificial Format_description_log_event of
 
556
           3.23&4.x.
 
557
         */
 
558
       }
 
559
     }
 
560
     else
 
561
     {
 
562
       if (test_for_non_eof_log_read_errors(error, &errmsg))
 
563
         goto err;
 
564
       /*
 
565
         It's EOF, nothing to do, go on reading next events, the
 
566
         Format_description_log_event will be found naturally if it is written.
 
567
       */
 
568
     }
 
569
     /* reset the packet as we wrote to it in any case */
 
570
     packet->set("\0", 1, &my_charset_bin);
 
571
  } /* end of if (pos > BIN_LOG_HEADER_SIZE); */
 
572
  else
 
573
  {
 
574
    /* The Format_description_log_event event will be found naturally. */
 
575
  }
 
576
 
 
577
  /* seek to the requested position, to start the requested dump */
 
578
  my_b_seek(&log, pos);                 // Seek will done on next read
 
579
 
 
580
  while (!net->error && net->vio != 0 && !thd->killed)
 
581
  {
 
582
    while (!(error = Log_event::read_log_event(&log, packet, log_lock)))
 
583
    {
 
584
      /*
 
585
        log's filename does not change while it's active
 
586
      */
 
587
      if (coord)
 
588
        coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET);
 
589
 
 
590
      if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
 
591
      {
 
592
        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
 
593
                                      LOG_EVENT_BINLOG_IN_USE_F);
 
594
        (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
 
595
      }
 
596
      else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT)
 
597
        binlog_can_be_corrupted= false;
 
598
 
 
599
      if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
 
600
      {
 
601
        errmsg = "Failed on my_net_write()";
 
602
        my_errno= ER_UNKNOWN_ERROR;
 
603
        goto err;
 
604
      }
 
605
 
 
606
      if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
 
607
      {
 
608
        if (send_file(thd))
 
609
        {
 
610
          errmsg = "failed in send_file()";
 
611
          my_errno= ER_UNKNOWN_ERROR;
 
612
          goto err;
 
613
        }
 
614
      }
 
615
      packet->set("\0", 1, &my_charset_bin);
 
616
    }
 
617
 
 
618
    /*
 
619
      here we were reading binlog that was not closed properly (as a result
 
620
      of a crash ?). treat any corruption as EOF
 
621
    */
 
622
    if (binlog_can_be_corrupted && error != LOG_READ_MEM)
 
623
      error=LOG_READ_EOF;
 
624
    /*
 
625
      TODO: now that we are logging the offset, check to make sure
 
626
      the recorded offset and the actual match.
 
627
      Guilhem 2003-06: this is not true if this master is a slave
 
628
      <4.0.15 running with --log-slave-updates, because then log_pos may
 
629
      be the offset in the-master-of-this-master's binlog.
 
630
    */
 
631
    if (test_for_non_eof_log_read_errors(error, &errmsg))
 
632
      goto err;
 
633
 
 
634
    if (!(flags & BINLOG_DUMP_NON_BLOCK) &&
 
635
        mysql_bin_log.is_active(log_file_name))
 
636
    {
 
637
      /*
 
638
        Block until there is more data in the log
 
639
      */
 
640
      if (net_flush(net))
 
641
      {
 
642
        errmsg = "failed on net_flush()";
 
643
        my_errno= ER_UNKNOWN_ERROR;
 
644
        goto err;
 
645
      }
 
646
 
 
647
      /*
 
648
        We may have missed the update broadcast from the log
 
649
        that has just happened, let's try to catch it if it did.
 
650
        If we did not miss anything, we just wait for other threads
 
651
        to signal us.
 
652
      */
 
653
      {
 
654
        log.error=0;
 
655
        bool read_packet = 0, fatal_error = 0;
 
656
 
 
657
        /*
 
658
          No one will update the log while we are reading
 
659
          now, but we'll be quick and just read one record
 
660
 
 
661
          TODO:
 
662
          Add an counter that is incremented for each time we update the
 
663
          binary log.  We can avoid the following read if the counter
 
664
          has not been updated since last read.
 
665
        */
 
666
 
 
667
        pthread_mutex_lock(log_lock);
 
668
        switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*)0)) {
 
669
        case 0:
 
670
          /* we read successfully, so we'll need to send it to the slave */
 
671
          pthread_mutex_unlock(log_lock);
 
672
          read_packet = 1;
 
673
          if (coord)
 
674
            coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET);
 
675
          break;
 
676
 
 
677
        case LOG_READ_EOF:
 
678
        {
 
679
          int ret;
 
680
          if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0)
 
681
          {
 
682
            pthread_mutex_unlock(log_lock);
 
683
            goto end;
 
684
          }
 
685
 
 
686
          do 
 
687
          {
 
688
            if (coord)
 
689
            {
 
690
              assert(heartbeat_ts && heartbeat_period != 0LL);
 
691
              set_timespec_nsec(*heartbeat_ts, heartbeat_period);
 
692
            }
 
693
            ret= mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts);
 
694
            assert(ret == 0 || (heartbeat_period != 0LL && coord != NULL));
 
695
            if (ret == ETIMEDOUT || ret == ETIME)
 
696
            {
 
697
              if (send_heartbeat_event(net, packet, coord))
 
698
              {
 
699
                errmsg = "Failed on my_net_write()";
 
700
                my_errno= ER_UNKNOWN_ERROR;
 
701
                pthread_mutex_unlock(log_lock);
 
702
                goto err;
 
703
              }
 
704
            }
 
705
            else
 
706
            {
 
707
              assert(ret == 0);
 
708
            }
 
709
          } while (ret != 0 && coord != NULL && !thd->killed);
 
710
          pthread_mutex_unlock(log_lock);
 
711
        }    
 
712
        break;
 
713
            
 
714
        default:
 
715
          pthread_mutex_unlock(log_lock);
 
716
          fatal_error = 1;
 
717
          break;
 
718
        }
 
719
 
 
720
        if (read_packet)
 
721
        {
 
722
          thd_proc_info(thd, "Sending binlog event to slave");
 
723
          if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) )
 
724
          {
 
725
            errmsg = "Failed on my_net_write()";
 
726
            my_errno= ER_UNKNOWN_ERROR;
 
727
            goto err;
 
728
          }
 
729
 
 
730
          if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
 
731
          {
 
732
            if (send_file(thd))
 
733
            {
 
734
              errmsg = "failed in send_file()";
 
735
              my_errno= ER_UNKNOWN_ERROR;
 
736
              goto err;
 
737
            }
 
738
          }
 
739
          packet->set("\0", 1, &my_charset_bin);
 
740
          /*
 
741
            No need to net_flush because we will get to flush later when
 
742
            we hit EOF pretty quick
 
743
          */
 
744
        }
 
745
 
 
746
        if (fatal_error)
 
747
        {
 
748
          errmsg = "error reading log entry";
 
749
          my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
750
          goto err;
 
751
        }
 
752
        log.error=0;
 
753
      }
 
754
    }
 
755
    else
 
756
    {
 
757
      bool loop_breaker = 0;
 
758
      /* need this to break out of the for loop from switch */
 
759
 
 
760
      thd_proc_info(thd, "Finished reading one binlog; switching to next binlog");
 
761
      switch (mysql_bin_log.find_next_log(&linfo, 1)) {
 
762
      case LOG_INFO_EOF:
 
763
        loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
 
764
        break;
 
765
      case 0:
 
766
        break;
 
767
      default:
 
768
        errmsg = "could not find next log";
 
769
        my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
770
        goto err;
 
771
      }
 
772
 
 
773
      if (loop_breaker)
 
774
        break;
 
775
 
 
776
      end_io_cache(&log);
 
777
      (void) my_close(file, MYF(MY_WME));
 
778
 
 
779
      /*
 
780
        Call fake_rotate_event() in case the previous log (the one which
 
781
        we have just finished reading) did not contain a Rotate event
 
782
        (for example (I don't know any other example) the previous log
 
783
        was the last one before the master was shutdown & restarted).
 
784
        This way we tell the slave about the new log's name and
 
785
        position.  If the binlog is 5.0, the next event we are going to
 
786
        read and send is Format_description_log_event.
 
787
      */
 
788
      if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||
 
789
          fake_rotate_event(net, packet, log_file_name, BIN_LOG_HEADER_SIZE,
 
790
                            &errmsg))
 
791
      {
 
792
        my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
793
        goto err;
 
794
      }
 
795
 
 
796
      packet->length(0);
 
797
      packet->append('\0');
 
798
      if (coord)
 
799
        coord->file_name= log_file_name; // reset to the next
 
800
    }
 
801
  }
 
802
 
 
803
end:
 
804
  end_io_cache(&log);
 
805
  (void)my_close(file, MYF(MY_WME));
 
806
 
 
807
  my_eof(thd);
 
808
  thd_proc_info(thd, "Waiting to finalize termination");
 
809
  pthread_mutex_lock(&LOCK_thread_count);
 
810
  thd->current_linfo = 0;
 
811
  pthread_mutex_unlock(&LOCK_thread_count);
 
812
  return;
 
813
 
 
814
err:
 
815
  thd_proc_info(thd, "Waiting to finalize termination");
 
816
  end_io_cache(&log);
 
817
  /*
 
818
    Exclude  iteration through thread list
 
819
    this is needed for purge_logs() - it will iterate through
 
820
    thread list and update thd->current_linfo->index_file_offset
 
821
    this mutex will make sure that it never tried to update our linfo
 
822
    after we return from this stack frame
 
823
  */
 
824
  pthread_mutex_lock(&LOCK_thread_count);
 
825
  thd->current_linfo = 0;
 
826
  pthread_mutex_unlock(&LOCK_thread_count);
 
827
  if (file >= 0)
 
828
    (void) my_close(file, MYF(MY_WME));
 
829
 
 
830
  my_message(my_errno, errmsg, MYF(0));
 
831
  return;
 
832
}
 
833
 
 
834
int start_slave(THD* thd , Master_info* mi,  bool net_report)
 
835
{
 
836
  int slave_errno= 0;
 
837
  int thread_mask;
 
838
 
 
839
  lock_slave_threads(mi);  // this allows us to cleanly read slave_running
 
840
  // Get a mask of _stopped_ threads
 
841
  init_thread_mask(&thread_mask,mi,1 /* inverse */);
 
842
  /*
 
843
    Below we will start all stopped threads.  But if the user wants to
 
844
    start only one thread, do as if the other thread was running (as we
 
845
    don't wan't to touch the other thread), so set the bit to 0 for the
 
846
    other thread
 
847
  */
 
848
  if (thd->lex->slave_thd_opt)
 
849
    thread_mask&= thd->lex->slave_thd_opt;
 
850
  if (thread_mask) //some threads are stopped, start them
 
851
  {
 
852
    if (init_master_info(mi,master_info_file,relay_log_info_file, 0,
 
853
                         thread_mask))
 
854
      slave_errno=ER_MASTER_INFO;
 
855
    else if (server_id_supplied && *mi->host)
 
856
    {
 
857
      /*
 
858
        If we will start SQL thread we will care about UNTIL options If
 
859
        not and they are specified we will ignore them and warn user
 
860
        about this fact.
 
861
      */
 
862
      if (thread_mask & SLAVE_SQL)
 
863
      {
 
864
        pthread_mutex_lock(&mi->rli.data_lock);
 
865
 
 
866
        if (thd->lex->mi.pos)
 
867
        {
 
868
          mi->rli.until_condition= Relay_log_info::UNTIL_MASTER_POS;
 
869
          mi->rli.until_log_pos= thd->lex->mi.pos;
 
870
          /*
 
871
             We don't check thd->lex->mi.log_file_name for NULL here
 
872
             since it is checked in sql_yacc.yy
 
873
          */
 
874
          strmake(mi->rli.until_log_name, thd->lex->mi.log_file_name,
 
875
                  sizeof(mi->rli.until_log_name)-1);
 
876
        }
 
877
        else if (thd->lex->mi.relay_log_pos)
 
878
        {
 
879
          mi->rli.until_condition= Relay_log_info::UNTIL_RELAY_POS;
 
880
          mi->rli.until_log_pos= thd->lex->mi.relay_log_pos;
 
881
          strmake(mi->rli.until_log_name, thd->lex->mi.relay_log_name,
 
882
                  sizeof(mi->rli.until_log_name)-1);
 
883
        }
 
884
        else
 
885
          mi->rli.clear_until_condition();
 
886
 
 
887
        if (mi->rli.until_condition != Relay_log_info::UNTIL_NONE)
 
888
        {
 
889
          /* Preparing members for effective until condition checking */
 
890
          const char *p= fn_ext(mi->rli.until_log_name);
 
891
          char *p_end;
 
892
          if (*p)
 
893
          {
 
894
            //p points to '.'
 
895
            mi->rli.until_log_name_extension= strtoul(++p,&p_end, 10);
 
896
            /*
 
897
              p_end points to the first invalid character. If it equals
 
898
              to p, no digits were found, error. If it contains '\0' it
 
899
              means  conversion went ok.
 
900
            */
 
901
            if (p_end==p || *p_end)
 
902
              slave_errno=ER_BAD_SLAVE_UNTIL_COND;
 
903
          }
 
904
          else
 
905
            slave_errno=ER_BAD_SLAVE_UNTIL_COND;
 
906
 
 
907
          /* mark the cached result of the UNTIL comparison as "undefined" */
 
908
          mi->rli.until_log_names_cmp_result=
 
909
            Relay_log_info::UNTIL_LOG_NAMES_CMP_UNKNOWN;
 
910
 
 
911
          /* Issuing warning then started without --skip-slave-start */
 
912
          if (!opt_skip_slave_start)
 
913
            push_warning(thd, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
 
914
                         ER_MISSING_SKIP_SLAVE,
 
915
                         ER(ER_MISSING_SKIP_SLAVE));
 
916
        }
 
917
 
 
918
        pthread_mutex_unlock(&mi->rli.data_lock);
 
919
      }
 
920
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
 
921
        push_warning(thd, DRIZZLE_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
 
922
                     ER(ER_UNTIL_COND_IGNORED));
 
923
 
 
924
      if (!slave_errno)
 
925
        slave_errno = start_slave_threads(0 /*no mutex */,
 
926
                                        1 /* wait for start */,
 
927
                                        mi,
 
928
                                        master_info_file,relay_log_info_file,
 
929
                                        thread_mask);
 
930
    }
 
931
    else
 
932
      slave_errno = ER_BAD_SLAVE;
 
933
  }
 
934
  else
 
935
  {
 
936
    /* no error if all threads are already started, only a warning */
 
937
    push_warning(thd, DRIZZLE_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_RUNNING,
 
938
                 ER(ER_SLAVE_WAS_RUNNING));
 
939
  }
 
940
 
 
941
  unlock_slave_threads(mi);
 
942
 
 
943
  if (slave_errno)
 
944
  {
 
945
    if (net_report)
 
946
      my_message(slave_errno, ER(slave_errno), MYF(0));
 
947
    return(1);
 
948
  }
 
949
  else if (net_report)
 
950
    my_ok(thd);
 
951
 
 
952
  return(0);
 
953
}
 
954
 
 
955
 
 
956
int stop_slave(THD* thd, Master_info* mi, bool net_report )
 
957
{
 
958
  int slave_errno;
 
959
  if (!thd)
 
960
    thd = current_thd;
 
961
 
 
962
  thd_proc_info(thd, "Killing slave");
 
963
  int thread_mask;
 
964
  lock_slave_threads(mi);
 
965
  // Get a mask of _running_ threads
 
966
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
 
967
  /*
 
968
    Below we will stop all running threads.
 
969
    But if the user wants to stop only one thread, do as if the other thread
 
970
    was stopped (as we don't wan't to touch the other thread), so set the
 
971
    bit to 0 for the other thread
 
972
  */
 
973
  if (thd->lex->slave_thd_opt)
 
974
    thread_mask &= thd->lex->slave_thd_opt;
 
975
 
 
976
  if (thread_mask)
 
977
  {
 
978
    slave_errno= terminate_slave_threads(mi,thread_mask,
 
979
                                         1 /*skip lock */);
 
980
  }
 
981
  else
 
982
  {
 
983
    //no error if both threads are already stopped, only a warning
 
984
    slave_errno= 0;
 
985
    push_warning(thd, DRIZZLE_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_NOT_RUNNING,
 
986
                 ER(ER_SLAVE_WAS_NOT_RUNNING));
 
987
  }
 
988
  unlock_slave_threads(mi);
 
989
  thd_proc_info(thd, 0);
 
990
 
 
991
  if (slave_errno)
 
992
  {
 
993
    if (net_report)
 
994
      my_message(slave_errno, ER(slave_errno), MYF(0));
 
995
    return(1);
 
996
  }
 
997
  else if (net_report)
 
998
    my_ok(thd);
 
999
 
 
1000
  return(0);
 
1001
}
 
1002
 
 
1003
 
 
1004
/*
 
1005
  Remove all relay logs and start replication from the start
 
1006
 
 
1007
  SYNOPSIS
 
1008
    reset_slave()
 
1009
    thd                 Thread handler
 
1010
    mi                  Master info for the slave
 
1011
 
 
1012
  RETURN
 
1013
    0   ok
 
1014
    1   error
 
1015
*/
 
1016
 
 
1017
 
 
1018
int reset_slave(THD *thd, Master_info* mi)
 
1019
{
 
1020
  struct stat stat_area;
 
1021
  char fname[FN_REFLEN];
 
1022
  int thread_mask= 0, error= 0;
 
1023
  uint sql_errno=0;
 
1024
  const char* errmsg=0;
 
1025
 
 
1026
  lock_slave_threads(mi);
 
1027
  init_thread_mask(&thread_mask,mi,0 /* not inverse */);
 
1028
  if (thread_mask) // We refuse if any slave thread is running
 
1029
  {
 
1030
    sql_errno= ER_SLAVE_MUST_STOP;
 
1031
    error=1;
 
1032
    goto err;
 
1033
  }
 
1034
 
 
1035
  ha_reset_slave(thd);
 
1036
 
 
1037
  // delete relay logs, clear relay log coordinates
 
1038
  if ((error= purge_relay_logs(&mi->rli, thd,
 
1039
                               1 /* just reset */,
 
1040
                               &errmsg)))
 
1041
    goto err;
 
1042
 
 
1043
  /* Clear master's log coordinates */
 
1044
  init_master_log_pos(mi);
 
1045
  /*
 
1046
     Reset errors (the idea is that we forget about the
 
1047
     old master).
 
1048
  */
 
1049
  mi->rli.clear_error();
 
1050
  mi->rli.clear_until_condition();
 
1051
 
 
1052
  // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
 
1053
  end_master_info(mi);
 
1054
  // and delete these two files
 
1055
  fn_format(fname, master_info_file, mysql_data_home, "", 4+32);
 
1056
  if (!stat(fname, &stat_area) && my_delete(fname, MYF(MY_WME)))
 
1057
  {
 
1058
    error=1;
 
1059
    goto err;
 
1060
  }
 
1061
  // delete relay_log_info_file
 
1062
  fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
 
1063
  if (!stat(fname, &stat_area) && my_delete(fname, MYF(MY_WME)))
 
1064
  {
 
1065
    error=1;
 
1066
    goto err;
 
1067
  }
 
1068
 
 
1069
err:
 
1070
  unlock_slave_threads(mi);
 
1071
  if (error)
 
1072
    my_error(sql_errno, MYF(0), errmsg);
 
1073
  return(error);
 
1074
}
 
1075
 
 
1076
/*
 
1077
 
 
1078
  Kill all Binlog_dump threads which previously talked to the same slave
 
1079
  ("same" means with the same server id). Indeed, if the slave stops, if the
 
1080
  Binlog_dump thread is waiting (pthread_cond_wait) for binlog update, then it
 
1081
  will keep existing until a query is written to the binlog. If the master is
 
1082
  idle, then this could last long, and if the slave reconnects, we could have 2
 
1083
  Binlog_dump threads in SHOW PROCESSLIST, until a query is written to the
 
1084
  binlog. To avoid this, when the slave reconnects and sends COM_BINLOG_DUMP,
 
1085
  the master kills any existing thread with the slave's server id (if this id is
 
1086
  not zero; it will be true for real slaves, but false for mysqlbinlog when it
 
1087
  sends COM_BINLOG_DUMP to get a remote binlog dump).
 
1088
 
 
1089
  SYNOPSIS
 
1090
    kill_zombie_dump_threads()
 
1091
    slave_server_id     the slave's server id
 
1092
 
 
1093
*/
 
1094
 
 
1095
 
 
1096
void kill_zombie_dump_threads(uint32_t slave_server_id)
 
1097
{
 
1098
  pthread_mutex_lock(&LOCK_thread_count);
 
1099
  I_List_iterator<THD> it(threads);
 
1100
  THD *tmp;
 
1101
 
 
1102
  while ((tmp=it++))
 
1103
  {
 
1104
    if (tmp->command == COM_BINLOG_DUMP &&
 
1105
       tmp->server_id == slave_server_id)
 
1106
    {
 
1107
      pthread_mutex_lock(&tmp->LOCK_delete);    // Lock from delete
 
1108
      break;
 
1109
    }
 
1110
  }
 
1111
  pthread_mutex_unlock(&LOCK_thread_count);
 
1112
  if (tmp)
 
1113
  {
 
1114
    /*
 
1115
      Here we do not call kill_one_thread() as
 
1116
      it will be slow because it will iterate through the list
 
1117
      again. We just to do kill the thread ourselves.
 
1118
    */
 
1119
    tmp->awake(THD::KILL_QUERY);
 
1120
    pthread_mutex_unlock(&tmp->LOCK_delete);
 
1121
  }
 
1122
}
 
1123
 
 
1124
 
 
1125
bool change_master(THD* thd, Master_info* mi)
 
1126
{
 
1127
  int thread_mask;
 
1128
  const char* errmsg= 0;
 
1129
  bool need_relay_log_purge= 1;
 
1130
 
 
1131
  lock_slave_threads(mi);
 
1132
  init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
 
1133
  if (thread_mask) // We refuse if any slave thread is running
 
1134
  {
 
1135
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
 
1136
    unlock_slave_threads(mi);
 
1137
    return(true);
 
1138
  }
 
1139
 
 
1140
  thd_proc_info(thd, "Changing master");
 
1141
  LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
 
1142
  // TODO: see if needs re-write
 
1143
  if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
 
1144
                       thread_mask))
 
1145
  {
 
1146
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
 
1147
    unlock_slave_threads(mi);
 
1148
    return(true);
 
1149
  }
 
1150
 
 
1151
  /*
 
1152
    Data lock not needed since we have already stopped the running threads,
 
1153
    and we have the hold on the run locks which will keep all threads that
 
1154
    could possibly modify the data structures from running
 
1155
  */
 
1156
 
 
1157
  /*
 
1158
    If the user specified host or port without binlog or position,
 
1159
    reset binlog's name to FIRST and position to 4.
 
1160
  */
 
1161
 
 
1162
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
 
1163
  {
 
1164
    mi->master_log_name[0] = 0;
 
1165
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
 
1166
  }
 
1167
 
 
1168
  if (lex_mi->log_file_name)
 
1169
    strmake(mi->master_log_name, lex_mi->log_file_name,
 
1170
            sizeof(mi->master_log_name)-1);
 
1171
  if (lex_mi->pos)
 
1172
  {
 
1173
    mi->master_log_pos= lex_mi->pos;
 
1174
  }
 
1175
 
 
1176
  if (lex_mi->host)
 
1177
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
 
1178
  if (lex_mi->user)
 
1179
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
 
1180
  if (lex_mi->password)
 
1181
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
 
1182
  if (lex_mi->port)
 
1183
    mi->port = lex_mi->port;
 
1184
  if (lex_mi->connect_retry)
 
1185
    mi->connect_retry = lex_mi->connect_retry;
 
1186
  if (lex_mi->heartbeat_opt != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
 
1187
    mi->heartbeat_period = lex_mi->heartbeat_period;
 
1188
  else
 
1189
    mi->heartbeat_period= (float) min((double)SLAVE_MAX_HEARTBEAT_PERIOD,
 
1190
                                      (slave_net_timeout/2.0));
 
1191
  mi->received_heartbeats= 0LL; // counter lives until master is CHANGEd
 
1192
  if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
 
1193
    mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE);
 
1194
 
 
1195
  if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
 
1196
    mi->ssl_verify_server_cert=
 
1197
      (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::LEX_MI_ENABLE);
 
1198
 
 
1199
  if (lex_mi->ssl_ca)
 
1200
    strmake(mi->ssl_ca, lex_mi->ssl_ca, sizeof(mi->ssl_ca)-1);
 
1201
  if (lex_mi->ssl_capath)
 
1202
    strmake(mi->ssl_capath, lex_mi->ssl_capath, sizeof(mi->ssl_capath)-1);
 
1203
  if (lex_mi->ssl_cert)
 
1204
    strmake(mi->ssl_cert, lex_mi->ssl_cert, sizeof(mi->ssl_cert)-1);
 
1205
  if (lex_mi->ssl_cipher)
 
1206
    strmake(mi->ssl_cipher, lex_mi->ssl_cipher, sizeof(mi->ssl_cipher)-1);
 
1207
  if (lex_mi->ssl_key)
 
1208
    strmake(mi->ssl_key, lex_mi->ssl_key, sizeof(mi->ssl_key)-1);
 
1209
  if (lex_mi->ssl || lex_mi->ssl_ca || lex_mi->ssl_capath ||
 
1210
      lex_mi->ssl_cert || lex_mi->ssl_cipher || lex_mi->ssl_key ||
 
1211
      lex_mi->ssl_verify_server_cert )
 
1212
    push_warning(thd, DRIZZLE_ERROR::WARN_LEVEL_NOTE,
 
1213
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
 
1214
 
 
1215
  if (lex_mi->relay_log_name)
 
1216
  {
 
1217
    need_relay_log_purge= 0;
 
1218
    strmake(mi->rli.group_relay_log_name,lex_mi->relay_log_name,
 
1219
            sizeof(mi->rli.group_relay_log_name)-1);
 
1220
    strmake(mi->rli.event_relay_log_name,lex_mi->relay_log_name,
 
1221
            sizeof(mi->rli.event_relay_log_name)-1);
 
1222
  }
 
1223
 
 
1224
  if (lex_mi->relay_log_pos)
 
1225
  {
 
1226
    need_relay_log_purge= 0;
 
1227
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
 
1228
  }
 
1229
 
 
1230
  /*
 
1231
    If user did specify neither host nor port nor any log name nor any log
 
1232
    pos, i.e. he specified only user/password/master_connect_retry, he probably
 
1233
    wants replication to resume from where it had left, i.e. from the
 
1234
    coordinates of the **SQL** thread (imagine the case where the I/O is ahead
 
1235
    of the SQL; restarting from the coordinates of the I/O would lose some
 
1236
    events which is probably unwanted when you are just doing minor changes
 
1237
    like changing master_connect_retry).
 
1238
    A side-effect is that if only the I/O thread was started, this thread may
 
1239
    restart from ''/4 after the CHANGE MASTER. That's a minor problem (it is a
 
1240
    much more unlikely situation than the one we are fixing here).
 
1241
    Note: coordinates of the SQL thread must be read here, before the
 
1242
    'if (need_relay_log_purge)' block which resets them.
 
1243
  */
 
1244
  if (!lex_mi->host && !lex_mi->port &&
 
1245
      !lex_mi->log_file_name && !lex_mi->pos &&
 
1246
      need_relay_log_purge)
 
1247
   {
 
1248
     /*
 
1249
       Sometimes mi->rli.master_log_pos == 0 (it happens when the SQL thread is
 
1250
       not initialized), so we use a max().
 
1251
       What happens to mi->rli.master_log_pos during the initialization stages
 
1252
       of replication is not 100% clear, so we guard against problems using
 
1253
       max().
 
1254
      */
 
1255
     mi->master_log_pos = ((BIN_LOG_HEADER_SIZE > mi->rli.group_master_log_pos)
 
1256
                           ? BIN_LOG_HEADER_SIZE
 
1257
                           : mi->rli.group_master_log_pos);
 
1258
     strmake(mi->master_log_name, mi->rli.group_master_log_name,
 
1259
             sizeof(mi->master_log_name)-1);
 
1260
  }
 
1261
  /*
 
1262
    Relay log's IO_CACHE may not be inited, if rli->inited==0 (server was never
 
1263
    a slave before).
 
1264
  */
 
1265
  if (flush_master_info(mi, 0))
 
1266
  {
 
1267
    my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
 
1268
    unlock_slave_threads(mi);
 
1269
    return(true);
 
1270
  }
 
1271
  if (need_relay_log_purge)
 
1272
  {
 
1273
    relay_log_purge= 1;
 
1274
    thd_proc_info(thd, "Purging old relay logs");
 
1275
    if (purge_relay_logs(&mi->rli, thd,
 
1276
                         0 /* not only reset, but also reinit */,
 
1277
                         &errmsg))
 
1278
    {
 
1279
      my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
 
1280
      unlock_slave_threads(mi);
 
1281
      return(true);
 
1282
    }
 
1283
  }
 
1284
  else
 
1285
  {
 
1286
    const char* msg;
 
1287
    relay_log_purge= 0;
 
1288
    /* Relay log is already initialized */
 
1289
    if (init_relay_log_pos(&mi->rli,
 
1290
                           mi->rli.group_relay_log_name,
 
1291
                           mi->rli.group_relay_log_pos,
 
1292
                           0 /*no data lock*/,
 
1293
                           &msg, 0))
 
1294
    {
 
1295
      my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
 
1296
      unlock_slave_threads(mi);
 
1297
      return(true);
 
1298
    }
 
1299
  }
 
1300
  /*
 
1301
    Coordinates in rli were spoilt by the 'if (need_relay_log_purge)' block,
 
1302
    so restore them to good values. If we left them to ''/0, that would work;
 
1303
    but that would fail in the case of 2 successive CHANGE MASTER (without a
 
1304
    START SLAVE in between): because first one would set the coords in mi to
 
1305
    the good values of those in rli, the set those in rli to ''/0, then
 
1306
    second CHANGE MASTER would set the coords in mi to those of rli, i.e. to
 
1307
    ''/0: we have lost all copies of the original good coordinates.
 
1308
    That's why we always save good coords in rli.
 
1309
  */
 
1310
  mi->rli.group_master_log_pos= mi->master_log_pos;
 
1311
  strmake(mi->rli.group_master_log_name,mi->master_log_name,
 
1312
          sizeof(mi->rli.group_master_log_name)-1);
 
1313
 
 
1314
  if (!mi->rli.group_master_log_name[0]) // uninitialized case
 
1315
    mi->rli.group_master_log_pos=0;
 
1316
 
 
1317
  pthread_mutex_lock(&mi->rli.data_lock);
 
1318
  mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
 
1319
  /* Clear the errors, for a clean start */
 
1320
  mi->rli.clear_error();
 
1321
  mi->rli.clear_until_condition();
 
1322
  /*
 
1323
    If we don't write new coordinates to disk now, then old will remain in
 
1324
    relay-log.info until START SLAVE is issued; but if mysqld is shutdown
 
1325
    before START SLAVE, then old will remain in relay-log.info, and will be the
 
1326
    in-memory value at restart (thus causing errors, as the old relay log does
 
1327
    not exist anymore).
 
1328
  */
 
1329
  flush_relay_log_info(&mi->rli);
 
1330
  pthread_cond_broadcast(&mi->data_cond);
 
1331
  pthread_mutex_unlock(&mi->rli.data_lock);
 
1332
 
 
1333
  unlock_slave_threads(mi);
 
1334
  thd_proc_info(thd, 0);
 
1335
  my_ok(thd);
 
1336
  return(false);
 
1337
}
 
1338
 
 
1339
int reset_master(THD* thd)
 
1340
{
 
1341
  if (!mysql_bin_log.is_open())
 
1342
  {
 
1343
    my_message(ER_FLUSH_MASTER_BINLOG_CLOSED,
 
1344
               ER(ER_FLUSH_MASTER_BINLOG_CLOSED), MYF(ME_BELL+ME_WAITTANG));
 
1345
    return 1;
 
1346
  }
 
1347
  return mysql_bin_log.reset_logs(thd);
 
1348
}
 
1349
 
 
1350
int cmp_master_pos(const char* log_file_name1, uint64_t log_pos1,
 
1351
                   const char* log_file_name2, uint64_t log_pos2)
 
1352
{
 
1353
  int res;
 
1354
  uint log_file_name1_len=  strlen(log_file_name1);
 
1355
  uint log_file_name2_len=  strlen(log_file_name2);
 
1356
 
 
1357
  //  We assume that both log names match up to '.'
 
1358
  if (log_file_name1_len == log_file_name2_len)
 
1359
  {
 
1360
    if ((res= strcmp(log_file_name1, log_file_name2)))
 
1361
      return res;
 
1362
    return (log_pos1 < log_pos2) ? -1 : (log_pos1 == log_pos2) ? 0 : 1;
 
1363
  }
 
1364
  return ((log_file_name1_len < log_file_name2_len) ? -1 : 1);
 
1365
}
 
1366
 
 
1367
 
 
1368
bool mysql_show_binlog_events(THD* thd)
 
1369
{
 
1370
  Protocol *protocol= thd->protocol;
 
1371
  List<Item> field_list;
 
1372
  const char *errmsg= 0;
 
1373
  bool ret= true;
 
1374
  IO_CACHE log;
 
1375
  File file= -1;
 
1376
 
 
1377
  Log_event::init_show_field_list(&field_list);
 
1378
  if (protocol->send_fields(&field_list,
 
1379
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
 
1380
    return(true);
 
1381
 
 
1382
  Format_description_log_event *description_event= new
 
1383
    Format_description_log_event(3); /* MySQL 4.0 by default */
 
1384
 
 
1385
  /*
 
1386
    Wait for handlers to insert any pending information
 
1387
    into the binlog.  For e.g. ndb which updates the binlog asynchronously
 
1388
    this is needed so that the uses sees all its own commands in the binlog
 
1389
  */
 
1390
  ha_binlog_wait(thd);
 
1391
 
 
1392
  if (mysql_bin_log.is_open())
 
1393
  {
 
1394
    LEX_MASTER_INFO *lex_mi= &thd->lex->mi;
 
1395
    SELECT_LEX_UNIT *unit= &thd->lex->unit;
 
1396
    ha_rows event_count, limit_start, limit_end;
 
1397
    my_off_t pos = max((uint64_t)BIN_LOG_HEADER_SIZE, lex_mi->pos); // user-friendly
 
1398
    char search_file_name[FN_REFLEN], *name;
 
1399
    const char *log_file_name = lex_mi->log_file_name;
 
1400
    pthread_mutex_t *log_lock = mysql_bin_log.get_log_lock();
 
1401
    LOG_INFO linfo;
 
1402
    Log_event* ev;
 
1403
 
 
1404
    unit->set_limit(thd->lex->current_select);
 
1405
    limit_start= unit->offset_limit_cnt;
 
1406
    limit_end= unit->select_limit_cnt;
 
1407
 
 
1408
    name= search_file_name;
 
1409
    if (log_file_name)
 
1410
      mysql_bin_log.make_log_name(search_file_name, log_file_name);
 
1411
    else
 
1412
      name=0;                                   // Find first log
 
1413
 
 
1414
    linfo.index_file_offset = 0;
 
1415
 
 
1416
    if (mysql_bin_log.find_log_pos(&linfo, name, 1))
 
1417
    {
 
1418
      errmsg = "Could not find target log";
 
1419
      goto err;
 
1420
    }
 
1421
 
 
1422
    pthread_mutex_lock(&LOCK_thread_count);
 
1423
    thd->current_linfo = &linfo;
 
1424
    pthread_mutex_unlock(&LOCK_thread_count);
 
1425
 
 
1426
    if ((file=open_binlog(&log, linfo.log_file_name, &errmsg)) < 0)
 
1427
      goto err;
 
1428
 
 
1429
    /*
 
1430
      to account binlog event header size
 
1431
    */
 
1432
    thd->variables.max_allowed_packet += MAX_LOG_EVENT_HEADER;
 
1433
 
 
1434
    pthread_mutex_lock(log_lock);
 
1435
 
 
1436
    /*
 
1437
      open_binlog() sought to position 4.
 
1438
      Read the first event in case it's a Format_description_log_event, to
 
1439
      know the format. If there's no such event, we are 3.23 or 4.x. This
 
1440
      code, like before, can't read 3.23 binlogs.
 
1441
      This code will fail on a mixed relay log (one which has Format_desc then
 
1442
      Rotate then Format_desc).
 
1443
    */
 
1444
    ev = Log_event::read_log_event(&log,(pthread_mutex_t*)0,description_event);
 
1445
    if (ev)
 
1446
    {
 
1447
      if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
 
1448
      {
 
1449
        delete description_event;
 
1450
        description_event= (Format_description_log_event*) ev;
 
1451
      }
 
1452
      else
 
1453
        delete ev;
 
1454
    }
 
1455
 
 
1456
    my_b_seek(&log, pos);
 
1457
 
 
1458
    if (!description_event->is_valid())
 
1459
    {
 
1460
      errmsg="Invalid Format_description event; could be out of memory";
 
1461
      goto err;
 
1462
    }
 
1463
 
 
1464
    for (event_count = 0;
 
1465
         (ev = Log_event::read_log_event(&log,(pthread_mutex_t*) 0,
 
1466
                                         description_event)); )
 
1467
    {
 
1468
      if (event_count >= limit_start &&
 
1469
          ev->net_send(protocol, linfo.log_file_name, pos))
 
1470
      {
 
1471
        errmsg = "Net error";
 
1472
        delete ev;
 
1473
        pthread_mutex_unlock(log_lock);
 
1474
        goto err;
 
1475
      }
 
1476
 
 
1477
      pos = my_b_tell(&log);
 
1478
      delete ev;
 
1479
 
 
1480
      if (++event_count >= limit_end)
 
1481
        break;
 
1482
    }
 
1483
 
 
1484
    if (event_count < limit_end && log.error)
 
1485
    {
 
1486
      errmsg = "Wrong offset or I/O error";
 
1487
      pthread_mutex_unlock(log_lock);
 
1488
      goto err;
 
1489
    }
 
1490
 
 
1491
    pthread_mutex_unlock(log_lock);
 
1492
  }
 
1493
 
 
1494
  ret= false;
 
1495
 
 
1496
err:
 
1497
  delete description_event;
 
1498
  if (file >= 0)
 
1499
  {
 
1500
    end_io_cache(&log);
 
1501
    (void) my_close(file, MYF(MY_WME));
 
1502
  }
 
1503
 
 
1504
  if (errmsg)
 
1505
    my_error(ER_ERROR_WHEN_EXECUTING_COMMAND, MYF(0),
 
1506
             "SHOW BINLOG EVENTS", errmsg);
 
1507
  else
 
1508
    my_eof(thd);
 
1509
 
 
1510
  pthread_mutex_lock(&LOCK_thread_count);
 
1511
  thd->current_linfo = 0;
 
1512
  pthread_mutex_unlock(&LOCK_thread_count);
 
1513
  return(ret);
 
1514
}
 
1515
 
 
1516
 
 
1517
bool show_binlog_info(THD* thd)
 
1518
{
 
1519
  Protocol *protocol= thd->protocol;
 
1520
  List<Item> field_list;
 
1521
  field_list.push_back(new Item_empty_string("File", FN_REFLEN));
 
1522
  field_list.push_back(new Item_return_int("Position",20,
 
1523
                                           DRIZZLE_TYPE_LONGLONG));
 
1524
  field_list.push_back(new Item_empty_string("Binlog_Do_DB",255));
 
1525
  field_list.push_back(new Item_empty_string("Binlog_Ignore_DB",255));
 
1526
 
 
1527
  if (protocol->send_fields(&field_list,
 
1528
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
 
1529
    return(true);
 
1530
  protocol->prepare_for_resend();
 
1531
 
 
1532
  if (mysql_bin_log.is_open())
 
1533
  {
 
1534
    LOG_INFO li;
 
1535
    mysql_bin_log.get_current_log(&li);
 
1536
    int dir_len = dirname_length(li.log_file_name);
 
1537
    protocol->store(li.log_file_name + dir_len, &my_charset_bin);
 
1538
    protocol->store((uint64_t) li.pos);
 
1539
    protocol->store(binlog_filter->get_do_db());
 
1540
    protocol->store(binlog_filter->get_ignore_db());
 
1541
    if (protocol->write())
 
1542
      return(true);
 
1543
  }
 
1544
  my_eof(thd);
 
1545
  return(false);
 
1546
}
 
1547
 
 
1548
 
 
1549
/*
 
1550
  Send a list of all binary logs to client
 
1551
 
 
1552
  SYNOPSIS
 
1553
    show_binlogs()
 
1554
    thd         Thread specific variable
 
1555
 
 
1556
  RETURN VALUES
 
1557
    false OK
 
1558
    true  error
 
1559
*/
 
1560
 
 
1561
bool show_binlogs(THD* thd)
 
1562
{
 
1563
  IO_CACHE *index_file;
 
1564
  LOG_INFO cur;
 
1565
  File file;
 
1566
  char fname[FN_REFLEN];
 
1567
  List<Item> field_list;
 
1568
  uint length;
 
1569
  int cur_dir_len;
 
1570
  Protocol *protocol= thd->protocol;
 
1571
 
 
1572
  if (!mysql_bin_log.is_open())
 
1573
  {
 
1574
    my_message(ER_NO_BINARY_LOGGING, ER(ER_NO_BINARY_LOGGING), MYF(0));
 
1575
    return 1;
 
1576
  }
 
1577
 
 
1578
  field_list.push_back(new Item_empty_string("Log_name", 255));
 
1579
  field_list.push_back(new Item_return_int("File_size", 20,
 
1580
                                           DRIZZLE_TYPE_LONGLONG));
 
1581
  if (protocol->send_fields(&field_list,
 
1582
                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
 
1583
    return(true);
 
1584
  
 
1585
  pthread_mutex_lock(mysql_bin_log.get_log_lock());
 
1586
  mysql_bin_log.lock_index();
 
1587
  index_file=mysql_bin_log.get_index_file();
 
1588
  
 
1589
  mysql_bin_log.raw_get_current_log(&cur); // dont take mutex
 
1590
  pthread_mutex_unlock(mysql_bin_log.get_log_lock()); // lockdep, OK
 
1591
  
 
1592
  cur_dir_len= dirname_length(cur.log_file_name);
 
1593
 
 
1594
  reinit_io_cache(index_file, READ_CACHE, (my_off_t) 0, 0, 0);
 
1595
 
 
1596
  /* The file ends with EOF or empty line */
 
1597
  while ((length=my_b_gets(index_file, fname, sizeof(fname))) > 1)
 
1598
  {
 
1599
    int dir_len;
 
1600
    uint64_t file_length= 0;                   // Length if open fails
 
1601
    fname[--length] = '\0';                     // remove the newline
 
1602
 
 
1603
    protocol->prepare_for_resend();
 
1604
    dir_len= dirname_length(fname);
 
1605
    length-= dir_len;
 
1606
    protocol->store(fname + dir_len, length, &my_charset_bin);
 
1607
 
 
1608
    if (!(strncmp(fname+dir_len, cur.log_file_name+cur_dir_len, length)))
 
1609
      file_length= cur.pos;  /* The active log, use the active position */
 
1610
    else
 
1611
    {
 
1612
      /* this is an old log, open it and find the size */
 
1613
      if ((file= my_open(fname, O_RDONLY | O_SHARE | O_BINARY,
 
1614
                         MYF(0))) >= 0)
 
1615
      {
 
1616
        file_length= (uint64_t) my_seek(file, 0L, MY_SEEK_END, MYF(0));
 
1617
        my_close(file, MYF(0));
 
1618
      }
 
1619
    }
 
1620
    protocol->store(file_length);
 
1621
    if (protocol->write())
 
1622
      goto err;
 
1623
  }
 
1624
  mysql_bin_log.unlock_index();
 
1625
  my_eof(thd);
 
1626
  return(false);
 
1627
 
 
1628
err:
 
1629
  mysql_bin_log.unlock_index();
 
1630
  return(true);
 
1631
}
 
1632
 
 
1633
/**
 
1634
   Load data's io cache specific hook to be executed
 
1635
   before a chunk of data is being read into the cache's buffer
 
1636
   The fuction instantianates and writes into the binlog
 
1637
   replication events along LOAD DATA processing.
 
1638
   
 
1639
   @param file  pointer to io-cache
 
1640
   @return 0
 
1641
*/
 
1642
int log_loaded_block(IO_CACHE* file)
 
1643
{
 
1644
  LOAD_FILE_INFO *lf_info;
 
1645
  uint block_len;
 
1646
  /* buffer contains position where we started last read */
 
1647
  uchar* buffer= (uchar*) my_b_get_buffer_start(file);
 
1648
  uint max_event_size= current_thd->variables.max_allowed_packet;
 
1649
  lf_info= (LOAD_FILE_INFO*) file->arg;
 
1650
  if (lf_info->thd->current_stmt_binlog_row_based)
 
1651
    return(0);
 
1652
  if (lf_info->last_pos_in_file != HA_POS_ERROR &&
 
1653
      lf_info->last_pos_in_file >= my_b_get_pos_in_file(file))
 
1654
    return(0);
 
1655
  
 
1656
  for (block_len= my_b_get_bytes_in_buffer(file); block_len > 0;
 
1657
       buffer += min(block_len, max_event_size),
 
1658
       block_len -= min(block_len, max_event_size))
 
1659
  {
 
1660
    lf_info->last_pos_in_file= my_b_get_pos_in_file(file);
 
1661
    if (lf_info->wrote_create_file)
 
1662
    {
 
1663
      Append_block_log_event a(lf_info->thd, lf_info->thd->db, buffer,
 
1664
                               min(block_len, max_event_size),
 
1665
                               lf_info->log_delayed);
 
1666
      mysql_bin_log.write(&a);
 
1667
    }
 
1668
    else
 
1669
    {
 
1670
      Begin_load_query_log_event b(lf_info->thd, lf_info->thd->db,
 
1671
                                   buffer,
 
1672
                                   min(block_len, max_event_size),
 
1673
                                   lf_info->log_delayed);
 
1674
      mysql_bin_log.write(&b);
 
1675
      lf_info->wrote_create_file= 1;
 
1676
    }
 
1677
  }
 
1678
  return(0);
 
1679
}
 
1680
 
 
1681
/*
 
1682
  Replication System Variables
 
1683
*/
 
1684
 
 
1685
class sys_var_slave_skip_counter :public sys_var
 
1686
{
 
1687
public:
 
1688
  sys_var_slave_skip_counter(sys_var_chain *chain, const char *name_arg)
 
1689
    :sys_var(name_arg)
 
1690
  { chain_sys_var(chain); }
 
1691
  bool check(THD *thd, set_var *var);
 
1692
  bool update(THD *thd, set_var *var);
 
1693
  bool check_type(enum_var_type type) { return type != OPT_GLOBAL; }
 
1694
  /*
 
1695
    We can't retrieve the value of this, so we don't have to define
 
1696
    type() or value_ptr()
 
1697
  */
 
1698
};
 
1699
 
 
1700
class sys_var_sync_binlog_period :public sys_var_long_ptr
 
1701
{
 
1702
public:
 
1703
  sys_var_sync_binlog_period(sys_var_chain *chain, const char *name_arg,
 
1704
                             ulong *value_ptr)
 
1705
    :sys_var_long_ptr(chain, name_arg,value_ptr) {}
 
1706
  bool update(THD *thd, set_var *var);
 
1707
};
 
1708
 
 
1709
static void fix_slave_net_timeout(THD *thd,
 
1710
                                  enum_var_type type __attribute__((unused)))
 
1711
{
 
1712
#ifdef HAVE_REPLICATION
 
1713
  pthread_mutex_lock(&LOCK_active_mi);
 
1714
  if (active_mi && slave_net_timeout < active_mi->heartbeat_period)
 
1715
    push_warning_printf(thd, DRIZZLE_ERROR::WARN_LEVEL_WARN,
 
1716
                        ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE,
 
1717
                        "The currect value for master_heartbeat_period"
 
1718
                        " exceeds the new value of `slave_net_timeout' sec."
 
1719
                        " A sensible value for the period should be"
 
1720
                        " less than the timeout.");
 
1721
  pthread_mutex_unlock(&LOCK_active_mi);
 
1722
#endif
 
1723
  return;
 
1724
}
 
1725
 
 
1726
static sys_var_chain vars = { NULL, NULL };
 
1727
 
 
1728
static sys_var_bool_ptr sys_relay_log_purge(&vars, "relay_log_purge",
 
1729
                                            &relay_log_purge);
 
1730
static sys_var_long_ptr sys_slave_net_timeout(&vars, "slave_net_timeout",
 
1731
                                              &slave_net_timeout,
 
1732
                                              fix_slave_net_timeout);
 
1733
static sys_var_long_ptr sys_slave_trans_retries(&vars, "slave_transaction_retries",
 
1734
                                                &slave_trans_retries);
 
1735
static sys_var_sync_binlog_period sys_sync_binlog_period(&vars, "sync_binlog", &sync_binlog_period);
 
1736
static sys_var_slave_skip_counter sys_slave_skip_counter(&vars, "sql_slave_skip_counter");
 
1737
 
 
1738
static int show_slave_skip_errors(THD *thd, SHOW_VAR *var, char *buff);
 
1739
 
 
1740
 
 
1741
static int show_slave_skip_errors(THD *thd __attribute__((unused)),
 
1742
                                  SHOW_VAR *var, char *buff)
 
1743
{
 
1744
  var->type=SHOW_CHAR;
 
1745
  var->value= buff;
 
1746
  if (!use_slave_mask || bitmap_is_clear_all(&slave_error_mask))
 
1747
  {
 
1748
    var->value= const_cast<char *>("OFF");
 
1749
  }
 
1750
  else if (bitmap_is_set_all(&slave_error_mask))
 
1751
  {
 
1752
    var->value= const_cast<char *>("ALL");
 
1753
  }
 
1754
  else
 
1755
  {
 
1756
    /* 10 is enough assuming errors are max 4 digits */
 
1757
    int i;
 
1758
    var->value= buff;
 
1759
    for (i= 1;
 
1760
         i < MAX_SLAVE_ERROR &&
 
1761
         (buff - var->value) < SHOW_VAR_FUNC_BUFF_SIZE;
 
1762
         i++)
 
1763
    {
 
1764
      if (bitmap_is_set(&slave_error_mask, i))
 
1765
      {
 
1766
        buff= int10_to_str(i, buff, 10);
 
1767
        *buff++= ',';
 
1768
      }
 
1769
    }
 
1770
    if (var->value != buff)
 
1771
      buff--;                           // Remove last ','
 
1772
    if (i < MAX_SLAVE_ERROR)
 
1773
      buff= stpcpy(buff, "...");  // Couldn't show all errors
 
1774
    *buff=0;
 
1775
  }
 
1776
  return 0;
 
1777
}
 
1778
 
 
1779
static st_show_var_func_container
 
1780
show_slave_skip_errors_cont = { &show_slave_skip_errors };
 
1781
 
 
1782
 
 
1783
static SHOW_VAR fixed_vars[]= {
 
1784
  {"log_slave_updates",       (char*) &opt_log_slave_updates,       SHOW_MY_BOOL},
 
1785
  {"relay_log" , (char*) &opt_relay_logname, SHOW_CHAR_PTR},
 
1786
  {"relay_log_index", (char*) &opt_relaylog_index_name, SHOW_CHAR_PTR},
 
1787
  {"relay_log_info_file", (char*) &relay_log_info_file, SHOW_CHAR_PTR},
 
1788
  {"relay_log_space_limit",   (char*) &relay_log_space_limit,       SHOW_LONGLONG},
 
1789
  {"slave_load_tmpdir",       (char*) &slave_load_tmpdir,           SHOW_CHAR_PTR},
 
1790
  {"slave_skip_errors",       (char*) &show_slave_skip_errors_cont,      SHOW_FUNC},
 
1791
};
 
1792
 
 
1793
bool sys_var_slave_skip_counter::check(THD *thd __attribute__((unused)),
 
1794
                                       set_var *var)
 
1795
{
 
1796
  int result= 0;
 
1797
  pthread_mutex_lock(&LOCK_active_mi);
 
1798
  pthread_mutex_lock(&active_mi->rli.run_lock);
 
1799
  if (active_mi->rli.slave_running)
 
1800
  {
 
1801
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
 
1802
    result=1;
 
1803
  }
 
1804
  pthread_mutex_unlock(&active_mi->rli.run_lock);
 
1805
  pthread_mutex_unlock(&LOCK_active_mi);
 
1806
  var->save_result.ulong_value= (ulong) var->value->val_int();
 
1807
  return result;
 
1808
}
 
1809
 
 
1810
 
 
1811
bool sys_var_slave_skip_counter::update(THD *thd __attribute__((unused)),
 
1812
                                        set_var *var)
 
1813
{
 
1814
  pthread_mutex_lock(&LOCK_active_mi);
 
1815
  pthread_mutex_lock(&active_mi->rli.run_lock);
 
1816
  /*
 
1817
    The following test should normally never be true as we test this
 
1818
    in the check function;  To be safe against multiple
 
1819
    SQL_SLAVE_SKIP_COUNTER request, we do the check anyway
 
1820
  */
 
1821
  if (!active_mi->rli.slave_running)
 
1822
  {
 
1823
    pthread_mutex_lock(&active_mi->rli.data_lock);
 
1824
    active_mi->rli.slave_skip_counter= var->save_result.ulong_value;
 
1825
    pthread_mutex_unlock(&active_mi->rli.data_lock);
 
1826
  }
 
1827
  pthread_mutex_unlock(&active_mi->rli.run_lock);
 
1828
  pthread_mutex_unlock(&LOCK_active_mi);
 
1829
  return 0;
 
1830
}
 
1831
 
 
1832
 
 
1833
bool sys_var_sync_binlog_period::update(THD *thd __attribute__((unused)),
 
1834
                                        set_var *var)
 
1835
{
 
1836
  sync_binlog_period= (uint32_t) var->save_result.uint64_t_value;
 
1837
  return 0;
 
1838
}
 
1839
 
 
1840
int init_replication_sys_vars()
 
1841
{
 
1842
  mysql_append_static_vars(fixed_vars, sizeof(fixed_vars) / sizeof(SHOW_VAR));
 
1843
 
 
1844
  if (mysql_add_sys_var_chain(vars.first, my_long_options))
 
1845
  {
 
1846
    /* should not happen */
 
1847
    fprintf(stderr, "failed to initialize replication system variables");
 
1848
    unireg_abort(1);
 
1849
  }
 
1850
  return 0;
 
1851
}
 
1852
 
 
1853
#endif /* HAVE_REPLICATION */