← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to server/sql_repl.cc

  • Committer: Mark Atwood
  • Date: 2008-07-12 11:37:59 UTC
  • mto: This revision was merged to the branch mainline in revision 139.
  • Revision ID: me@mark.atwood.name-20080712113759-nrjn1bq1e0shuory
Add hello_world() UDF

Show diffs side-by-side

added added

removed removed

Lines of Context:
server/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 "mysql_priv.h"
 
17
#ifdef HAVE_REPLICATION
 
18
 
 
19
#include "rpl_mi.h"
 
20
#include "sql_repl.h"
 
21
#include "log_event.h"
 
22
#include "rpl_filter.h"
 
23
#include <my_dir.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
                             ulonglong 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
  ulong 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 = "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 = "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 = "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 = "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("Failed in send_file() %s", 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 = !bcmp((uchar*) log_name, (uchar*) linfo->log_file_name,
 
214
                     log_name_len);
 
215
      pthread_mutex_unlock(&linfo->lock);
 
216
      if (result)
 
217
        break;
 
218
    }
 
219
  }
 
220
 
 
221
  pthread_mutex_unlock(&LOCK_thread_count);
 
222
  return result;
 
223
}
 
224
 
 
225
bool purge_error_message(THD* thd, int res)
 
226
{
 
227
  uint errmsg= 0;
 
228
 
 
229
  switch (res)  {
 
230
  case 0: break;
 
231
  case LOG_INFO_EOF:    errmsg= ER_UNKNOWN_TARGET_BINLOG; break;
 
232
  case LOG_INFO_IO:     errmsg= ER_IO_ERR_LOG_INDEX_READ; break;
 
233
  case LOG_INFO_INVALID:errmsg= ER_BINLOG_PURGE_PROHIBITED; break;
 
234
  case LOG_INFO_SEEK:   errmsg= ER_FSEEK_FAIL; break;
 
235
  case LOG_INFO_MEM:    errmsg= ER_OUT_OF_RESOURCES; break;
 
236
  case LOG_INFO_FATAL:  errmsg= ER_BINLOG_PURGE_FATAL_ERR; break;
 
237
  case LOG_INFO_IN_USE: errmsg= ER_LOG_IN_USE; break;
 
238
  case LOG_INFO_EMFILE: errmsg= ER_BINLOG_PURGE_EMFILE; break;
 
239
  default:              errmsg= ER_LOG_PURGE_UNKNOWN_ERR; break;
 
240
  }
 
241
 
 
242
  if (errmsg)
 
243
  {
 
244
    my_message(errmsg, ER(errmsg), MYF(0));
 
245
    return true;
 
246
  }
 
247
  my_ok(thd);
 
248
  return false;
 
249
}
 
250
 
 
251
 
 
252
bool purge_master_logs(THD* thd, const char* to_log)
 
253
{
 
254
  char search_file_name[FN_REFLEN];
 
255
  if (!mysql_bin_log.is_open())
 
256
  {
 
257
    my_ok(thd);
 
258
    return false;
 
259
  }
 
260
 
 
261
  mysql_bin_log.make_log_name(search_file_name, to_log);
 
262
  return purge_error_message(thd,
 
263
                             mysql_bin_log.purge_logs(search_file_name, 0, 1,
 
264
                                                      1, NULL));
 
265
}
 
266
 
 
267
 
 
268
bool purge_master_logs_before_date(THD* thd, time_t purge_time)
 
269
{
 
270
  if (!mysql_bin_log.is_open())
 
271
  {
 
272
    my_ok(thd);
 
273
    return 0;
 
274
  }
 
275
  return purge_error_message(thd,
 
276
                             mysql_bin_log.purge_logs_before_date(purge_time));
 
277
}
 
278
 
 
279
int test_for_non_eof_log_read_errors(int error, const char **errmsg)
 
280
{
 
281
  if (error == LOG_READ_EOF)
 
282
    return 0;
 
283
  my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
284
  switch (error) {
 
285
  case LOG_READ_BOGUS:
 
286
    *errmsg = "bogus data in log event";
 
287
    break;
 
288
  case LOG_READ_TOO_LARGE:
 
289
    *errmsg = "log event entry exceeded max_allowed_packet; \
 
290
Increase max_allowed_packet on master";
 
291
    break;
 
292
  case LOG_READ_IO:
 
293
    *errmsg = "I/O error reading log event";
 
294
    break;
 
295
  case LOG_READ_MEM:
 
296
    *errmsg = "memory allocation failed reading log event";
 
297
    break;
 
298
  case LOG_READ_TRUNC:
 
299
    *errmsg = "binlog truncated in the middle of event";
 
300
    break;
 
301
  default:
 
302
    *errmsg = "unknown error reading log event on the master";
 
303
    break;
 
304
  }
 
305
  return error;
 
306
}
 
307
 
 
308
 
 
309
/**
 
310
  An auxiliary function for calling in mysql_binlog_send
 
311
  to initialize the heartbeat timeout in waiting for a binlogged event.
 
312
 
 
313
  @param[in]    thd  THD to access a user variable
 
314
 
 
315
  @return        heartbeat period an ulonglong of nanoseconds
 
316
                 or zero if heartbeat was not demanded by slave
 
317
*/ 
 
318
static ulonglong get_heartbeat_period(THD * thd)
 
319
{
 
320
  my_bool null_value;
 
321
  LEX_STRING name=  { C_STRING_WITH_LEN("master_heartbeat_period")};
 
322
  user_var_entry *entry= 
 
323
    (user_var_entry*) hash_search(&thd->user_vars, (uchar*) name.str,
 
324
                                  name.length);
 
325
  return entry? entry->val_int(&null_value) : 0;
 
326
}
 
327
 
 
328
/*
 
329
  Function prepares and sends repliation heartbeat event.
 
330
 
 
331
  @param net                net object of THD
 
332
  @param packet             buffer to store the heartbeat instance
 
333
  @param event_coordinates  binlog file name and position of the last
 
334
                            real event master sent from binlog
 
335
 
 
336
  @note 
 
337
    Among three essential pieces of heartbeat data Log_event::when
 
338
    is computed locally.
 
339
    The  error to send is serious and should force terminating
 
340
    the dump thread.
 
341
*/
 
342
static int send_heartbeat_event(NET* net, String* packet,
 
343
                                const struct event_coordinates *coord)
 
344
{
 
345
  char header[LOG_EVENT_HEADER_LEN];
 
346
  /*
 
347
    'when' (the timestamp) is set to 0 so that slave could distinguish between
 
348
    real and fake Rotate events (if necessary)
 
349
  */
 
350
  memset(header, 0, 4);  // when
 
351
 
 
352
  header[EVENT_TYPE_OFFSET] = HEARTBEAT_LOG_EVENT;
 
353
 
 
354
  char* p= coord->file_name + dirname_length(coord->file_name);
 
355
 
 
356
  uint ident_len = strlen(p);
 
357
  ulong event_len = ident_len + LOG_EVENT_HEADER_LEN;
 
358
  int4store(header + SERVER_ID_OFFSET, server_id);
 
359
  int4store(header + EVENT_LEN_OFFSET, event_len);
 
360
  int2store(header + FLAGS_OFFSET, 0);
 
361
 
 
362
  int4store(header + LOG_POS_OFFSET, coord->pos);  // log_pos
 
363
 
 
364
  packet->append(header, sizeof(header));
 
365
  packet->append(p, ident_len);             // log_file_name
 
366
 
 
367
  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) ||
 
368
      net_flush(net))
 
369
  {
 
370
    return(-1);
 
371
  }
 
372
  packet->set("\0", 1, &my_charset_bin);
 
373
  return(0);
 
374
}
 
375
 
 
376
/*
 
377
  TODO: Clean up loop to only have one call to send_file()
 
378
*/
 
379
 
 
380
void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
 
381
                       ushort flags)
 
382
{
 
383
  LOG_INFO linfo;
 
384
  char *log_file_name = linfo.log_file_name;
 
385
  char search_file_name[FN_REFLEN], *name;
 
386
  IO_CACHE log;
 
387
  File file = -1;
 
388
  String* packet = &thd->packet;
 
389
  int error;
 
390
  const char *errmsg = "Unknown error";
 
391
  NET* net = &thd->net;
 
392
  pthread_mutex_t *log_lock;
 
393
  bool binlog_can_be_corrupted= false;
 
394
 
 
395
  bzero((char*) &log,sizeof(log));
 
396
  /* 
 
397
     heartbeat_period from @master_heartbeat_period user variable
 
398
  */
 
399
  ulonglong heartbeat_period= get_heartbeat_period(thd);
 
400
  struct timespec heartbeat_buf;
 
401
  struct event_coordinates coord_buf;
 
402
  struct timespec *heartbeat_ts= NULL;
 
403
  struct event_coordinates *coord= NULL;
 
404
  if (heartbeat_period != 0LL)
 
405
  {
 
406
    heartbeat_ts= &heartbeat_buf;
 
407
    set_timespec_nsec(*heartbeat_ts, 0);
 
408
    coord= &coord_buf;
 
409
    coord->file_name= log_file_name; // initialization basing on what slave remembers
 
410
    coord->pos= pos;
 
411
  }
 
412
 
 
413
  if (!mysql_bin_log.is_open())
 
414
  {
 
415
    errmsg = "Binary log is not open";
 
416
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
417
    goto err;
 
418
  }
 
419
  if (!server_id_supplied)
 
420
  {
 
421
    errmsg = "Misconfigured master - server id was not set";
 
422
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
423
    goto err;
 
424
  }
 
425
 
 
426
  name=search_file_name;
 
427
  if (log_ident[0])
 
428
    mysql_bin_log.make_log_name(search_file_name, log_ident);
 
429
  else
 
430
    name=0;                                     // Find first log
 
431
 
 
432
  linfo.index_file_offset = 0;
 
433
 
 
434
  if (mysql_bin_log.find_log_pos(&linfo, name, 1))
 
435
  {
 
436
    errmsg = "Could not find first log file name in binary log index file";
 
437
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
438
    goto err;
 
439
  }
 
440
 
 
441
  pthread_mutex_lock(&LOCK_thread_count);
 
442
  thd->current_linfo = &linfo;
 
443
  pthread_mutex_unlock(&LOCK_thread_count);
 
444
 
 
445
  if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)
 
446
  {
 
447
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
448
    goto err;
 
449
  }
 
450
  if (pos < BIN_LOG_HEADER_SIZE || pos > my_b_filelength(&log))
 
451
  {
 
452
    errmsg= "Client requested master to start replication from \
 
453
impossible position";
 
454
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
455
    goto err;
 
456
  }
 
457
 
 
458
  /*
 
459
    We need to start a packet with something other than 255
 
460
    to distinguish it from error
 
461
  */
 
462
  packet->set("\0", 1, &my_charset_bin); /* This is the start of a new packet */
 
463
 
 
464
  /*
 
465
    Tell the client about the log name with a fake Rotate event;
 
466
    this is needed even if we also send a Format_description_log_event
 
467
    just after, because that event does not contain the binlog's name.
 
468
    Note that as this Rotate event is sent before
 
469
    Format_description_log_event, the slave cannot have any info to
 
470
    understand this event's format, so the header len of
 
471
    Rotate_log_event is FROZEN (so in 5.0 it will have a header shorter
 
472
    than other events except FORMAT_DESCRIPTION_EVENT).
 
473
    Before 4.0.14 we called fake_rotate_event below only if (pos ==
 
474
    BIN_LOG_HEADER_SIZE), because if this is false then the slave
 
475
    already knows the binlog's name.
 
476
    Since, we always call fake_rotate_event; if the slave already knew
 
477
    the log's name (ex: CHANGE MASTER TO MASTER_LOG_FILE=...) this is
 
478
    useless but does not harm much. It is nice for 3.23 (>=.58) slaves
 
479
    which test Rotate events to see if the master is 4.0 (then they
 
480
    choose to stop because they can't replicate 4.0); by always calling
 
481
    fake_rotate_event we are sure that 3.23.58 and newer will detect the
 
482
    problem as soon as replication starts (BUG#198).
 
483
    Always calling fake_rotate_event makes sending of normal
 
484
    (=from-binlog) Rotate events a priori unneeded, but it is not so
 
485
    simple: the 2 Rotate events are not equivalent, the normal one is
 
486
    before the Stop event, the fake one is after. If we don't send the
 
487
    normal one, then the Stop event will be interpreted (by existing 4.0
 
488
    slaves) as "the master stopped", which is wrong. So for safety,
 
489
    given that we want minimum modification of 4.0, we send the normal
 
490
    and fake Rotates.
 
491
  */
 
492
  if (fake_rotate_event(net, packet, log_file_name, pos, &errmsg))
 
493
  {
 
494
    /*
 
495
       This error code is not perfect, as fake_rotate_event() does not
 
496
       read anything from the binlog; if it fails it's because of an
 
497
       error in my_net_write(), fortunately it will say so in errmsg.
 
498
    */
 
499
    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
500
    goto err;
 
501
  }
 
502
  packet->set("\0", 1, &my_charset_bin);
 
503
  /*
 
504
    Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become
 
505
    this larger than the corresponding packet (query) sent 
 
506
    from client to master.
 
507
  */
 
508
  thd->variables.max_allowed_packet+= MAX_LOG_EVENT_HEADER;
 
509
 
 
510
  /*
 
511
    We can set log_lock now, it does not move (it's a member of
 
512
    mysql_bin_log, and it's already inited, and it will be destroyed
 
513
    only at shutdown).
 
514
  */
 
515
  log_lock = mysql_bin_log.get_log_lock();
 
516
  if (pos > BIN_LOG_HEADER_SIZE)
 
517
  {
 
518
     /*
 
519
       Try to find a Format_description_log_event at the beginning of
 
520
       the binlog
 
521
     */
 
522
     if (!(error = Log_event::read_log_event(&log, packet, log_lock)))
 
523
     {
 
524
       /*
 
525
         The packet has offsets equal to the normal offsets in a binlog
 
526
         event +1 (the first character is \0).
 
527
       */
 
528
       if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
 
529
       {
 
530
         binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
 
531
                                       LOG_EVENT_BINLOG_IN_USE_F);
 
532
         (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
 
533
         /*
 
534
           mark that this event with "log_pos=0", so the slave
 
535
           should not increment master's binlog position
 
536
           (rli->group_master_log_pos)
 
537
         */
 
538
         int4store((char*) packet->ptr()+LOG_POS_OFFSET+1, 0);
 
539
         /*
 
540
           if reconnect master sends FD event with `created' as 0
 
541
           to avoid destroying temp tables.
 
542
          */
 
543
         int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+
 
544
                   ST_CREATED_OFFSET+1, (ulong) 0);
 
545
         /* send it */
 
546
         if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
 
547
         {
 
548
           errmsg = "Failed on my_net_write()";
 
549
           my_errno= ER_UNKNOWN_ERROR;
 
550
           goto err;
 
551
         }
 
552
 
 
553
         /*
 
554
           No need to save this event. We are only doing simple reads
 
555
           (no real parsing of the events) so we don't need it. And so
 
556
           we don't need the artificial Format_description_log_event of
 
557
           3.23&4.x.
 
558
         */
 
559
       }
 
560
     }
 
561
     else
 
562
     {
 
563
       if (test_for_non_eof_log_read_errors(error, &errmsg))
 
564
         goto err;
 
565
       /*
 
566
         It's EOF, nothing to do, go on reading next events, the
 
567
         Format_description_log_event will be found naturally if it is written.
 
568
       */
 
569
     }
 
570
     /* reset the packet as we wrote to it in any case */
 
571
     packet->set("\0", 1, &my_charset_bin);
 
572
  } /* end of if (pos > BIN_LOG_HEADER_SIZE); */
 
573
  else
 
574
  {
 
575
    /* The Format_description_log_event event will be found naturally. */
 
576
  }
 
577
 
 
578
  /* seek to the requested position, to start the requested dump */
 
579
  my_b_seek(&log, pos);                 // Seek will done on next read
 
580
 
 
581
  while (!net->error && net->vio != 0 && !thd->killed)
 
582
  {
 
583
    while (!(error = Log_event::read_log_event(&log, packet, log_lock)))
 
584
    {
 
585
      /*
 
586
        log's filename does not change while it's active
 
587
      */
 
588
      if (coord)
 
589
        coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET);
 
590
 
 
591
      if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)
 
592
      {
 
593
        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &
 
594
                                      LOG_EVENT_BINLOG_IN_USE_F);
 
595
        (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;
 
596
      }
 
597
      else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT)
 
598
        binlog_can_be_corrupted= false;
 
599
 
 
600
      if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
 
601
      {
 
602
        errmsg = "Failed on my_net_write()";
 
603
        my_errno= ER_UNKNOWN_ERROR;
 
604
        goto err;
 
605
      }
 
606
 
 
607
      if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
 
608
      {
 
609
        if (send_file(thd))
 
610
        {
 
611
          errmsg = "failed in send_file()";
 
612
          my_errno= ER_UNKNOWN_ERROR;
 
613
          goto err;
 
614
        }
 
615
      }
 
616
      packet->set("\0", 1, &my_charset_bin);
 
617
    }
 
618
 
 
619
    /*
 
620
      here we were reading binlog that was not closed properly (as a result
 
621
      of a crash ?). treat any corruption as EOF
 
622
    */
 
623
    if (binlog_can_be_corrupted && error != LOG_READ_MEM)
 
624
      error=LOG_READ_EOF;
 
625
    /*
 
626
      TODO: now that we are logging the offset, check to make sure
 
627
      the recorded offset and the actual match.
 
628
      Guilhem 2003-06: this is not true if this master is a slave
 
629
      <4.0.15 running with --log-slave-updates, because then log_pos may
 
630
      be the offset in the-master-of-this-master's binlog.
 
631
    */
 
632
    if (test_for_non_eof_log_read_errors(error, &errmsg))
 
633
      goto err;
 
634
 
 
635
    if (!(flags & BINLOG_DUMP_NON_BLOCK) &&
 
636
        mysql_bin_log.is_active(log_file_name))
 
637
    {
 
638
      /*
 
639
        Block until there is more data in the log
 
640
      */
 
641
      if (net_flush(net))
 
642
      {
 
643
        errmsg = "failed on net_flush()";
 
644
        my_errno= ER_UNKNOWN_ERROR;
 
645
        goto err;
 
646
      }
 
647
 
 
648
      /*
 
649
        We may have missed the update broadcast from the log
 
650
        that has just happened, let's try to catch it if it did.
 
651
        If we did not miss anything, we just wait for other threads
 
652
        to signal us.
 
653
      */
 
654
      {
 
655
        log.error=0;
 
656
        bool read_packet = 0, fatal_error = 0;
 
657
 
 
658
        /*
 
659
          No one will update the log while we are reading
 
660
          now, but we'll be quick and just read one record
 
661
 
 
662
          TODO:
 
663
          Add an counter that is incremented for each time we update the
 
664
          binary log.  We can avoid the following read if the counter
 
665
          has not been updated since last read.
 
666
        */
 
667
 
 
668
        pthread_mutex_lock(log_lock);
 
669
        switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*)0)) {
 
670
        case 0:
 
671
          /* we read successfully, so we'll need to send it to the slave */
 
672
          pthread_mutex_unlock(log_lock);
 
673
          read_packet = 1;
 
674
          if (coord)
 
675
            coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET);
 
676
          break;
 
677
 
 
678
        case LOG_READ_EOF:
 
679
        {
 
680
          int ret;
 
681
          if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0)
 
682
          {
 
683
            pthread_mutex_unlock(log_lock);
 
684
            goto end;
 
685
          }
 
686
 
 
687
          do 
 
688
          {
 
689
            if (coord)
 
690
            {
 
691
              assert(heartbeat_ts && heartbeat_period != 0LL);
 
692
              set_timespec_nsec(*heartbeat_ts, heartbeat_period);
 
693
            }
 
694
            ret= mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts);
 
695
            assert(ret == 0 || heartbeat_period != 0LL && coord != NULL);
 
696
            if (ret == ETIMEDOUT || ret == ETIME)
 
697
            {
 
698
              if (send_heartbeat_event(net, packet, coord))
 
699
              {
 
700
                errmsg = "Failed on my_net_write()";
 
701
                my_errno= ER_UNKNOWN_ERROR;
 
702
                pthread_mutex_unlock(log_lock);
 
703
                goto err;
 
704
              }
 
705
            }
 
706
            else
 
707
            {
 
708
              assert(ret == 0);
 
709
            }
 
710
          } while (ret != 0 && coord != NULL && !thd->killed);
 
711
          pthread_mutex_unlock(log_lock);
 
712
        }    
 
713
        break;
 
714
            
 
715
        default:
 
716
          pthread_mutex_unlock(log_lock);
 
717
          fatal_error = 1;
 
718
          break;
 
719
        }
 
720
 
 
721
        if (read_packet)
 
722
        {
 
723
          thd_proc_info(thd, "Sending binlog event to slave");
 
724
          if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) )
 
725
          {
 
726
            errmsg = "Failed on my_net_write()";
 
727
            my_errno= ER_UNKNOWN_ERROR;
 
728
            goto err;
 
729
          }
 
730
 
 
731
          if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)
 
732
          {
 
733
            if (send_file(thd))
 
734
            {
 
735
              errmsg = "failed in send_file()";
 
736
              my_errno= ER_UNKNOWN_ERROR;
 
737
              goto err;
 
738
            }
 
739
          }
 
740
          packet->set("\0", 1, &my_charset_bin);
 
741
          /*
 
742
            No need to net_flush because we will get to flush later when
 
743
            we hit EOF pretty quick
 
744
          */
 
745
        }
 
746
 
 
747
        if (fatal_error)
 
748
        {
 
749
          errmsg = "error reading log entry";
 
750
          my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
751
          goto err;
 
752
        }
 
753
        log.error=0;
 
754
      }
 
755
    }
 
756
    else
 
757
    {
 
758
      bool loop_breaker = 0;
 
759
      /* need this to break out of the for loop from switch */
 
760
 
 
761
      thd_proc_info(thd, "Finished reading one binlog; switching to next binlog");
 
762
      switch (mysql_bin_log.find_next_log(&linfo, 1)) {
 
763
      case LOG_INFO_EOF:
 
764
        loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);
 
765
        break;
 
766
      case 0:
 
767
        break;
 
768
      default:
 
769
        errmsg = "could not find next log";
 
770
        my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
771
        goto err;
 
772
      }
 
773
 
 
774
      if (loop_breaker)
 
775
        break;
 
776
 
 
777
      end_io_cache(&log);
 
778
      (void) my_close(file, MYF(MY_WME));
 
779
 
 
780
      /*
 
781
        Call fake_rotate_event() in case the previous log (the one which
 
782
        we have just finished reading) did not contain a Rotate event
 
783
        (for example (I don't know any other example) the previous log
 
784
        was the last one before the master was shutdown & restarted).
 
785
        This way we tell the slave about the new log's name and
 
786
        position.  If the binlog is 5.0, the next event we are going to
 
787
        read and send is Format_description_log_event.
 
788
      */
 
789
      if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||
 
790
          fake_rotate_event(net, packet, log_file_name, BIN_LOG_HEADER_SIZE,
 
791
                            &errmsg))
 
792
      {
 
793
        my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;
 
794
        goto err;
 
795
      }
 
796
 
 
797
      packet->length(0);
 
798
      packet->append('\0');
 
799
      if (coord)
 
800
        coord->file_name= log_file_name; // reset to the next
 
801
    }
 
802
  }
 
803
 
 
804
end:
 
805
  end_io_cache(&log);
 
806
  (void)my_close(file, MYF(MY_WME));
 
807
 
 
808
  my_eof(thd);
 
809
  thd_proc_info(thd, "Waiting to finalize termination");
 
810
  pthread_mutex_lock(&LOCK_thread_count);
 
811
  thd->current_linfo = 0;
 
812
  pthread_mutex_unlock(&LOCK_thread_count);
 
813
  return;
 
814
 
 
815
err:
 
816
  thd_proc_info(thd, "Waiting to finalize termination");
 
817
  end_io_cache(&log);
 
818
  /*
 
819
    Exclude  iteration through thread list
 
820
    this is needed for purge_logs() - it will iterate through
 
821
    thread list and update thd->current_linfo->index_file_offset
 
822
    this mutex will make sure that it never tried to update our linfo
 
823
    after we return from this stack frame
 
824
  */
 
825
  pthread_mutex_lock(&LOCK_thread_count);
 
826
  thd->current_linfo = 0;
 
827
  pthread_mutex_unlock(&LOCK_thread_count);
 
828
  if (file >= 0)
 
829
    (void) my_close(file, MYF(MY_WME));
 
830
 
 
831
  my_message(my_errno, errmsg, MYF(0));
 
832
  return;
 
833
}
 
834
 
 
835
int start_slave(THD* thd , Master_info* mi,  bool net_report)
 
836
{
 
837
  int slave_errno= 0;
 
838
  int thread_mask;
 
839
 
 
840
  lock_slave_threads(mi);  // this allows us to cleanly read slave_running
 
841
  // Get a mask of _stopped_ threads
 
842
  init_thread_mask(&thread_mask,mi,1 /* inverse */);
 
843
  /*
 
844
    Below we will start all stopped threads.  But if the user wants to
 
845
    start only one thread, do as if the other thread was running (as we
 
846
    don't wan't to touch the other thread), so set the bit to 0 for the
 
847
    other thread
 
848
  */
 
849
  if (thd->lex->slave_thd_opt)
 
850
    thread_mask&= thd->lex->slave_thd_opt;
 
851
  if (thread_mask) //some threads are stopped, start them
 
852
  {
 
853
    if (init_master_info(mi,master_info_file,relay_log_info_file, 0,
 
854
                         thread_mask))
 
855
      slave_errno=ER_MASTER_INFO;
 
856
    else if (server_id_supplied && *mi->host)
 
857
    {
 
858
      /*
 
859
        If we will start SQL thread we will care about UNTIL options If
 
860
        not and they are specified we will ignore them and warn user
 
861
        about this fact.
 
862
      */
 
863
      if (thread_mask & SLAVE_SQL)
 
864
      {
 
865
        pthread_mutex_lock(&mi->rli.data_lock);
 
866
 
 
867
        if (thd->lex->mi.pos)
 
868
        {
 
869
          mi->rli.until_condition= Relay_log_info::UNTIL_MASTER_POS;
 
870
          mi->rli.until_log_pos= thd->lex->mi.pos;
 
871
          /*
 
872
             We don't check thd->lex->mi.log_file_name for NULL here
 
873
             since it is checked in sql_yacc.yy
 
874
          */
 
875
          strmake(mi->rli.until_log_name, thd->lex->mi.log_file_name,
 
876
                  sizeof(mi->rli.until_log_name)-1);
 
877
        }
 
878
        else if (thd->lex->mi.relay_log_pos)
 
879
        {
 
880
          mi->rli.until_condition= Relay_log_info::UNTIL_RELAY_POS;
 
881
          mi->rli.until_log_pos= thd->lex->mi.relay_log_pos;
 
882
          strmake(mi->rli.until_log_name, thd->lex->mi.relay_log_name,
 
883
                  sizeof(mi->rli.until_log_name)-1);
 
884
        }
 
885
        else
 
886
          mi->rli.clear_until_condition();
 
887
 
 
888
        if (mi->rli.until_condition != Relay_log_info::UNTIL_NONE)
 
889
        {
 
890
          /* Preparing members for effective until condition checking */
 
891
          const char *p= fn_ext(mi->rli.until_log_name);
 
892
          char *p_end;
 
893
          if (*p)
 
894
          {
 
895
            //p points to '.'
 
896
            mi->rli.until_log_name_extension= strtoul(++p,&p_end, 10);
 
897
            /*
 
898
              p_end points to the first invalid character. If it equals
 
899
              to p, no digits were found, error. If it contains '\0' it
 
900
              means  conversion went ok.
 
901
            */
 
902
            if (p_end==p || *p_end)
 
903
              slave_errno=ER_BAD_SLAVE_UNTIL_COND;
 
904
          }
 
905
          else
 
906
            slave_errno=ER_BAD_SLAVE_UNTIL_COND;
 
907
 
 
908
          /* mark the cached result of the UNTIL comparison as "undefined" */
 
909
          mi->rli.until_log_names_cmp_result=
 
910
            Relay_log_info::UNTIL_LOG_NAMES_CMP_UNKNOWN;
 
911
 
 
912
          /* Issuing warning then started without --skip-slave-start */
 
913
          if (!opt_skip_slave_start)
 
914
            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 
915
                         ER_MISSING_SKIP_SLAVE,
 
916
                         ER(ER_MISSING_SKIP_SLAVE));
 
917
        }
 
918
 
 
919
        pthread_mutex_unlock(&mi->rli.data_lock);
 
920
      }
 
921
      else if (thd->lex->mi.pos || thd->lex->mi.relay_log_pos)
 
922
        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_UNTIL_COND_IGNORED,
 
923
                     ER(ER_UNTIL_COND_IGNORED));
 
924
 
 
925
      if (!slave_errno)
 
926
        slave_errno = start_slave_threads(0 /*no mutex */,
 
927
                                        1 /* wait for start */,
 
928
                                        mi,
 
929
                                        master_info_file,relay_log_info_file,
 
930
                                        thread_mask);
 
931
    }
 
932
    else
 
933
      slave_errno = ER_BAD_SLAVE;
 
934
  }
 
935
  else
 
936
  {
 
937
    /* no error if all threads are already started, only a warning */
 
938
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_RUNNING,
 
939
                 ER(ER_SLAVE_WAS_RUNNING));
 
940
  }
 
941
 
 
942
  unlock_slave_threads(mi);
 
943
 
 
944
  if (slave_errno)
 
945
  {
 
946
    if (net_report)
 
947
      my_message(slave_errno, ER(slave_errno), MYF(0));
 
948
    return(1);
 
949
  }
 
950
  else if (net_report)
 
951
    my_ok(thd);
 
952
 
 
953
  return(0);
 
954
}
 
955
 
 
956
 
 
957
int stop_slave(THD* thd, Master_info* mi, bool net_report )
 
958
{
 
959
  int slave_errno;
 
960
  if (!thd)
 
961
    thd = current_thd;
 
962
 
 
963
  thd_proc_info(thd, "Killing slave");
 
964
  int thread_mask;
 
965
  lock_slave_threads(mi);
 
966
  // Get a mask of _running_ threads
 
967
  init_thread_mask(&thread_mask,mi,0 /* not inverse*/);
 
968
  /*
 
969
    Below we will stop all running threads.
 
970
    But if the user wants to stop only one thread, do as if the other thread
 
971
    was stopped (as we don't wan't to touch the other thread), so set the
 
972
    bit to 0 for the other thread
 
973
  */
 
974
  if (thd->lex->slave_thd_opt)
 
975
    thread_mask &= thd->lex->slave_thd_opt;
 
976
 
 
977
  if (thread_mask)
 
978
  {
 
979
    slave_errno= terminate_slave_threads(mi,thread_mask,
 
980
                                         1 /*skip lock */);
 
981
  }
 
982
  else
 
983
  {
 
984
    //no error if both threads are already stopped, only a warning
 
985
    slave_errno= 0;
 
986
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE, ER_SLAVE_WAS_NOT_RUNNING,
 
987
                 ER(ER_SLAVE_WAS_NOT_RUNNING));
 
988
  }
 
989
  unlock_slave_threads(mi);
 
990
  thd_proc_info(thd, 0);
 
991
 
 
992
  if (slave_errno)
 
993
  {
 
994
    if (net_report)
 
995
      my_message(slave_errno, ER(slave_errno), MYF(0));
 
996
    return(1);
 
997
  }
 
998
  else if (net_report)
 
999
    my_ok(thd);
 
1000
 
 
1001
  return(0);
 
1002
}
 
1003
 
 
1004
 
 
1005
/*
 
1006
  Remove all relay logs and start replication from the start
 
1007
 
 
1008
  SYNOPSIS
 
1009
    reset_slave()
 
1010
    thd                 Thread handler
 
1011
    mi                  Master info for the slave
 
1012
 
 
1013
  RETURN
 
1014
    0   ok
 
1015
    1   error
 
1016
*/
 
1017
 
 
1018
 
 
1019
int reset_slave(THD *thd, Master_info* mi)
 
1020
{
 
1021
  struct stat stat_area;
 
1022
  char fname[FN_REFLEN];
 
1023
  int thread_mask= 0, error= 0;
 
1024
  uint sql_errno=0;
 
1025
  const char* errmsg=0;
 
1026
 
 
1027
  lock_slave_threads(mi);
 
1028
  init_thread_mask(&thread_mask,mi,0 /* not inverse */);
 
1029
  if (thread_mask) // We refuse if any slave thread is running
 
1030
  {
 
1031
    sql_errno= ER_SLAVE_MUST_STOP;
 
1032
    error=1;
 
1033
    goto err;
 
1034
  }
 
1035
 
 
1036
  ha_reset_slave(thd);
 
1037
 
 
1038
  // delete relay logs, clear relay log coordinates
 
1039
  if ((error= purge_relay_logs(&mi->rli, thd,
 
1040
                               1 /* just reset */,
 
1041
                               &errmsg)))
 
1042
    goto err;
 
1043
 
 
1044
  /* Clear master's log coordinates */
 
1045
  init_master_log_pos(mi);
 
1046
  /*
 
1047
     Reset errors (the idea is that we forget about the
 
1048
     old master).
 
1049
  */
 
1050
  mi->rli.clear_error();
 
1051
  mi->rli.clear_until_condition();
 
1052
 
 
1053
  // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
 
1054
  end_master_info(mi);
 
1055
  // and delete these two files
 
1056
  fn_format(fname, master_info_file, mysql_data_home, "", 4+32);
 
1057
  if (!stat(fname, &stat_area) && my_delete(fname, MYF(MY_WME)))
 
1058
  {
 
1059
    error=1;
 
1060
    goto err;
 
1061
  }
 
1062
  // delete relay_log_info_file
 
1063
  fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
 
1064
  if (!stat(fname, &stat_area) && my_delete(fname, MYF(MY_WME)))
 
1065
  {
 
1066
    error=1;
 
1067
    goto err;
 
1068
  }
 
1069
 
 
1070
err:
 
1071
  unlock_slave_threads(mi);
 
1072
  if (error)
 
1073
    my_error(sql_errno, MYF(0), errmsg);
 
1074
  return(error);
 
1075
}
 
1076
 
 
1077
/*
 
1078
 
 
1079
  Kill all Binlog_dump threads which previously talked to the same slave
 
1080
  ("same" means with the same server id). Indeed, if the slave stops, if the
 
1081
  Binlog_dump thread is waiting (pthread_cond_wait) for binlog update, then it
 
1082
  will keep existing until a query is written to the binlog. If the master is
 
1083
  idle, then this could last long, and if the slave reconnects, we could have 2
 
1084
  Binlog_dump threads in SHOW PROCESSLIST, until a query is written to the
 
1085
  binlog. To avoid this, when the slave reconnects and sends COM_BINLOG_DUMP,
 
1086
  the master kills any existing thread with the slave's server id (if this id is
 
1087
  not zero; it will be true for real slaves, but false for mysqlbinlog when it
 
1088
  sends COM_BINLOG_DUMP to get a remote binlog dump).
 
1089
 
 
1090
  SYNOPSIS
 
1091
    kill_zombie_dump_threads()
 
1092
    slave_server_id     the slave's server id
 
1093
 
 
1094
*/
 
1095
 
 
1096
 
 
1097
void kill_zombie_dump_threads(uint32 slave_server_id)
 
1098
{
 
1099
  pthread_mutex_lock(&LOCK_thread_count);
 
1100
  I_List_iterator<THD> it(threads);
 
1101
  THD *tmp;
 
1102
 
 
1103
  while ((tmp=it++))
 
1104
  {
 
1105
    if (tmp->command == COM_BINLOG_DUMP &&
 
1106
       tmp->server_id == slave_server_id)
 
1107
    {
 
1108
      pthread_mutex_lock(&tmp->LOCK_delete);    // Lock from delete
 
1109
      break;
 
1110
    }
 
1111
  }
 
1112
  pthread_mutex_unlock(&LOCK_thread_count);
 
1113
  if (tmp)
 
1114
  {
 
1115
    /*
 
1116
      Here we do not call kill_one_thread() as
 
1117
      it will be slow because it will iterate through the list
 
1118
      again. We just to do kill the thread ourselves.
 
1119
    */
 
1120
    tmp->awake(THD::KILL_QUERY);
 
1121
    pthread_mutex_unlock(&tmp->LOCK_delete);
 
1122
  }
 
1123
}
 
1124
 
 
1125
 
 
1126
bool change_master(THD* thd, Master_info* mi)
 
1127
{
 
1128
  int thread_mask;
 
1129
  const char* errmsg= 0;
 
1130
  bool need_relay_log_purge= 1;
 
1131
 
 
1132
  lock_slave_threads(mi);
 
1133
  init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
 
1134
  if (thread_mask) // We refuse if any slave thread is running
 
1135
  {
 
1136
    my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
 
1137
    unlock_slave_threads(mi);
 
1138
    return(true);
 
1139
  }
 
1140
 
 
1141
  thd_proc_info(thd, "Changing master");
 
1142
  LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
 
1143
  // TODO: see if needs re-write
 
1144
  if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
 
1145
                       thread_mask))
 
1146
  {
 
1147
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
 
1148
    unlock_slave_threads(mi);
 
1149
    return(true);
 
1150
  }
 
1151
 
 
1152
  /*
 
1153
    Data lock not needed since we have already stopped the running threads,
 
1154
    and we have the hold on the run locks which will keep all threads that
 
1155
    could possibly modify the data structures from running
 
1156
  */
 
1157
 
 
1158
  /*
 
1159
    If the user specified host or port without binlog or position,
 
1160
    reset binlog's name to FIRST and position to 4.
 
1161
  */
 
1162
 
 
1163
  if ((lex_mi->host || lex_mi->port) && !lex_mi->log_file_name && !lex_mi->pos)
 
1164
  {
 
1165
    mi->master_log_name[0] = 0;
 
1166
    mi->master_log_pos= BIN_LOG_HEADER_SIZE;
 
1167
  }
 
1168
 
 
1169
  if (lex_mi->log_file_name)
 
1170
    strmake(mi->master_log_name, lex_mi->log_file_name,
 
1171
            sizeof(mi->master_log_name)-1);
 
1172
  if (lex_mi->pos)
 
1173
  {
 
1174
    mi->master_log_pos= lex_mi->pos;
 
1175
  }
 
1176
 
 
1177
  if (lex_mi->host)
 
1178
    strmake(mi->host, lex_mi->host, sizeof(mi->host)-1);
 
1179
  if (lex_mi->user)
 
1180
    strmake(mi->user, lex_mi->user, sizeof(mi->user)-1);
 
1181
  if (lex_mi->password)
 
1182
    strmake(mi->password, lex_mi->password, sizeof(mi->password)-1);
 
1183
  if (lex_mi->port)
 
1184
    mi->port = lex_mi->port;
 
1185
  if (lex_mi->connect_retry)
 
1186
    mi->connect_retry = lex_mi->connect_retry;
 
1187
  if (lex_mi->heartbeat_opt != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
 
1188
    mi->heartbeat_period = lex_mi->heartbeat_period;
 
1189
  else
 
1190
    mi->heartbeat_period= (float) min(SLAVE_MAX_HEARTBEAT_PERIOD,
 
1191
                                      (slave_net_timeout/2.0));
 
1192
  mi->received_heartbeats= 0LL; // counter lives until master is CHANGEd
 
1193
  if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
 
1194
    mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE);
 
1195
 
 
1196
  if (lex_mi->ssl_verify_server_cert != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
 
1197
    mi->ssl_verify_server_cert=
 
1198
      (lex_mi->ssl_verify_server_cert == LEX_MASTER_INFO::LEX_MI_ENABLE);
 
1199
 
 
1200
  if (lex_mi->ssl_ca)
 
1201
    strmake(mi->ssl_ca, lex_mi->ssl_ca, sizeof(mi->ssl_ca)-1);
 
1202
  if (lex_mi->ssl_capath)
 
1203
    strmake(mi->ssl_capath, lex_mi->ssl_capath, sizeof(mi->ssl_capath)-1);
 
1204
  if (lex_mi->ssl_cert)
 
1205
    strmake(mi->ssl_cert, lex_mi->ssl_cert, sizeof(mi->ssl_cert)-1);
 
1206
  if (lex_mi->ssl_cipher)
 
1207
    strmake(mi->ssl_cipher, lex_mi->ssl_cipher, sizeof(mi->ssl_cipher)-1);
 
1208
  if (lex_mi->ssl_key)
 
1209
    strmake(mi->ssl_key, lex_mi->ssl_key, sizeof(mi->ssl_key)-1);
 
1210
  if (lex_mi->ssl || lex_mi->ssl_ca || lex_mi->ssl_capath ||
 
1211
      lex_mi->ssl_cert || lex_mi->ssl_cipher || lex_mi->ssl_key ||
 
1212
      lex_mi->ssl_verify_server_cert )
 
1213
    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 
1214
                 ER_SLAVE_IGNORED_SSL_PARAMS, ER(ER_SLAVE_IGNORED_SSL_PARAMS));
 
1215
 
 
1216
  if (lex_mi->relay_log_name)
 
1217
  {
 
1218
    need_relay_log_purge= 0;
 
1219
    strmake(mi->rli.group_relay_log_name,lex_mi->relay_log_name,
 
1220
            sizeof(mi->rli.group_relay_log_name)-1);
 
1221
    strmake(mi->rli.event_relay_log_name,lex_mi->relay_log_name,
 
1222
            sizeof(mi->rli.event_relay_log_name)-1);
 
1223
  }
 
1224
 
 
1225
  if (lex_mi->relay_log_pos)
 
1226
  {
 
1227
    need_relay_log_purge= 0;
 
1228
    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
 
1229
  }
 
1230
 
 
1231
  /*
 
1232
    If user did specify neither host nor port nor any log name nor any log
 
1233
    pos, i.e. he specified only user/password/master_connect_retry, he probably
 
1234
    wants replication to resume from where it had left, i.e. from the
 
1235
    coordinates of the **SQL** thread (imagine the case where the I/O is ahead
 
1236
    of the SQL; restarting from the coordinates of the I/O would lose some
 
1237
    events which is probably unwanted when you are just doing minor changes
 
1238
    like changing master_connect_retry).
 
1239
    A side-effect is that if only the I/O thread was started, this thread may
 
1240
    restart from ''/4 after the CHANGE MASTER. That's a minor problem (it is a
 
1241
    much more unlikely situation than the one we are fixing here).
 
1242
    Note: coordinates of the SQL thread must be read here, before the
 
1243
    'if (need_relay_log_purge)' block which resets them.
 
1244
  */
 
1245
  if (!lex_mi->host && !lex_mi->port &&
 
1246
      !lex_mi->log_file_name && !lex_mi->pos &&
 
1247
      need_relay_log_purge)
 
1248
   {
 
1249
     /*
 
1250
       Sometimes mi->rli.master_log_pos == 0 (it happens when the SQL thread is
 
1251
       not initialized), so we use a max().
 
1252
       What happens to mi->rli.master_log_pos during the initialization stages
 
1253
       of replication is not 100% clear, so we guard against problems using
 
1254
       max().
 
1255
      */
 
1256
     mi->master_log_pos = max(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, ulonglong log_pos1,
 
1351
                   const char* log_file_name2, ulonglong 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(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
                                           MYSQL_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((ulonglong) 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
                                           MYSQL_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
    ulonglong 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= (ulonglong) 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, MYSQL_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= strmov(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= (ulong) var->save_result.ulonglong_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 */
 
1854
 
 
1855