~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2006 MySQL AB & Sasha

   This program is free software; you can redistribute it and/or modify

   it under the terms of the GNU General Public License as published by

   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,

   but WITHOUT ANY WARRANTY; without even the implied warranty of

   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License

   along with this program; if not, write to the Free Software

   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include <drizzled/server_includes.h>

#ifdef HAVE_REPLICATION

#include "rpl_mi.h"

#include "sql_repl.h"

#include "log_event.h"

#include "rpl_filter.h"

#include <drizzled/drizzled_error_messages.h>

int max_binlog_dump_events = 0; // unlimited

/*

    fake_rotate_event() builds a fake (=which does not exist physically in any

    binlog) Rotate event, which contains the name of the binlog we are going to

    send to the slave (because the slave may not know it if it just asked for

    MASTER_LOG_FILE='', MASTER_LOG_POS=4).

    < 4.0.14, fake_rotate_event() was called only if the requested pos was 4.

    After this version we always call it, so that a 3.23.58 slave can rely on

    it to detect if the master is 4.0 (and stop) (the _fake_ Rotate event has

    zeros in the good positions which, by chance, make it possible for the 3.23

    slave to detect that this event is unexpected) (this is luck which happens

    because the master and slave disagree on the size of the header of

    Log_event).

    Relying on the event length of the Rotate event instead of these

    well-placed zeros was not possible as Rotate events have a variable-length

    part.

*/

static int fake_rotate_event(NET* net, String* packet, char* log_file_name,

                             uint64_t position, const char** errmsg)

{

  char header[LOG_EVENT_HEADER_LEN], buf[ROTATE_HEADER_LEN+100];

  /*

    'when' (the timestamp) is set to 0 so that slave could distinguish between

    real and fake Rotate events (if necessary)

  */

  memset(header, 0, 4);

  header[EVENT_TYPE_OFFSET] = ROTATE_EVENT;

  char* p = log_file_name+dirname_length(log_file_name);

  uint ident_len = (uint) strlen(p);

  ulong event_len = ident_len + LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN;

  int4store(header + SERVER_ID_OFFSET, server_id);

  int4store(header + EVENT_LEN_OFFSET, event_len);

  int2store(header + FLAGS_OFFSET, 0);

  // TODO: check what problems this may cause and fix them

  int4store(header + LOG_POS_OFFSET, 0);

  packet->append(header, sizeof(header));

  int8store(buf+R_POS_OFFSET,position);

  packet->append(buf, ROTATE_HEADER_LEN);

  packet->append(p,ident_len);

  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))

  {

    *errmsg = "failed on my_net_write()";

    return(-1);

  }

  return(0);

}

static int send_file(THD *thd)

{

  NET* net = &thd->net;

  int fd = -1, error = 1;

  size_t bytes;

  char fname[FN_REFLEN+1];

  const char *errmsg = 0;

  int old_timeout;

  unsigned long packet_len;

  uchar buf[IO_SIZE];				// It's safe to alloc this

  /*

    The client might be slow loading the data, give him wait_timeout to do

    the job

  */

  old_timeout= net->read_timeout;

  my_net_set_read_timeout(net, thd->variables.net_wait_timeout);

  /*

    We need net_flush here because the client will not know it needs to send

    us the file name until it has processed the load event entry

  */

  if (net_flush(net) || (packet_len = my_net_read(net)) == packet_error)

  {

    errmsg = _("Failed in send_file() while reading file name");

    goto err;

  }

  // terminate with \0 for fn_format

  *((char*)net->read_pos +  packet_len) = 0;

  fn_format(fname, (char*) net->read_pos + 1, "", "", 4);

  // this is needed to make replicate-ignore-db

  if (!strcmp(fname,"/dev/null"))

    goto end;

  if ((fd = my_open(fname, O_RDONLY, MYF(0))) < 0)

  {

    errmsg = _("Failed in send_file() on open of file");

    goto err;

  }

  while ((long) (bytes= my_read(fd, buf, IO_SIZE, MYF(0))) > 0)

  {

    if (my_net_write(net, buf, bytes))

    {

      errmsg = _("Failed in send_file() while writing data to client");

      goto err;

    }

  }

 end:

  if (my_net_write(net, (uchar*) "", 0) || net_flush(net) ||

      (my_net_read(net) == packet_error))

  {

    errmsg = _("Failed in send_file() while negotiating file transfer close");

    goto err;

  }

  error = 0;

 err:

  my_net_set_read_timeout(net, old_timeout);

  if (fd >= 0)

    (void) my_close(fd, MYF(0));

  if (errmsg)

  {

    sql_print_error(errmsg);

  }

  return(error);

}

/*

  Adjust the position pointer in the binary log file for all running slaves

  SYNOPSIS

    adjust_linfo_offsets()

    purge_offset	Number of bytes removed from start of log index file

  NOTES

    - This is called when doing a PURGE when we delete lines from the

      index log file

  REQUIREMENTS

    - Before calling this function, we have to ensure that no threads are

      using any binary log file before purge_offset.a

  TODO

    - Inform the slave threads that they should sync the position

      in the binary log file with flush_relay_log_info.

      Now they sync is done for next read.

*/

void adjust_linfo_offsets(my_off_t purge_offset)

{

  THD *tmp;

  pthread_mutex_lock(&LOCK_thread_count);

  I_List_iterator<THD> it(threads);

  while ((tmp=it++))

  {

    LOG_INFO* linfo;

    if ((linfo = tmp->current_linfo))

    {

      pthread_mutex_lock(&linfo->lock);

      /*

	Index file offset can be less that purge offset only if

	we just started reading the index file. In that case

	we have nothing to adjust

      */

      if (linfo->index_file_offset < purge_offset)

	linfo->fatal = (linfo->index_file_offset != 0);

      else

	linfo->index_file_offset -= purge_offset;

      pthread_mutex_unlock(&linfo->lock);

    }

  }

  pthread_mutex_unlock(&LOCK_thread_count);

}

bool log_in_use(const char* log_name)

{

  int log_name_len = strlen(log_name) + 1;

  THD *tmp;

  bool result = 0;

  pthread_mutex_lock(&LOCK_thread_count);

  I_List_iterator<THD> it(threads);

  while ((tmp=it++))

  {

    LOG_INFO* linfo;

    if ((linfo = tmp->current_linfo))

    {

      pthread_mutex_lock(&linfo->lock);

      result = !memcmp(log_name, linfo->log_file_name, log_name_len);

      pthread_mutex_unlock(&linfo->lock);

      if (result)

	break;

    }

  }

  pthread_mutex_unlock(&LOCK_thread_count);

  return result;

}

bool purge_error_message(THD* thd, int res)

{

  uint errmsg= 0;

  switch (res)  {

  case 0: break;

  case LOG_INFO_EOF:	errmsg= ER_UNKNOWN_TARGET_BINLOG; break;

  case LOG_INFO_IO:	errmsg= ER_IO_ERR_LOG_INDEX_READ; break;

  case LOG_INFO_INVALID:errmsg= ER_BINLOG_PURGE_PROHIBITED; break;

  case LOG_INFO_SEEK:	errmsg= ER_FSEEK_FAIL; break;

  case LOG_INFO_MEM:	errmsg= ER_OUT_OF_RESOURCES; break;

  case LOG_INFO_FATAL:	errmsg= ER_BINLOG_PURGE_FATAL_ERR; break;

  case LOG_INFO_IN_USE: errmsg= ER_LOG_IN_USE; break;

  case LOG_INFO_EMFILE: errmsg= ER_BINLOG_PURGE_EMFILE; break;

  default:		errmsg= ER_LOG_PURGE_UNKNOWN_ERR; break;

  }

  if (errmsg)

  {

    my_message(errmsg, ER(errmsg), MYF(0));

    return true;

  }

  my_ok(thd);

  return false;

}

bool purge_master_logs(THD* thd, const char* to_log)

{

  char search_file_name[FN_REFLEN];

  if (!mysql_bin_log.is_open())

  {

    my_ok(thd);

    return false;

  }

  mysql_bin_log.make_log_name(search_file_name, to_log);

  return purge_error_message(thd,

			     mysql_bin_log.purge_logs(search_file_name, 0, 1,

						      1, NULL));

}

bool purge_master_logs_before_date(THD* thd, time_t purge_time)

{

  if (!mysql_bin_log.is_open())

  {

    my_ok(thd);

    return 0;

  }

  return purge_error_message(thd,

                             mysql_bin_log.purge_logs_before_date(purge_time));

}

int test_for_non_eof_log_read_errors(int error, const char **errmsg)

{

  if (error == LOG_READ_EOF)

    return 0;

  my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

  switch (error) {

  case LOG_READ_BOGUS:

    *errmsg = "bogus data in log event";

    break;

  case LOG_READ_TOO_LARGE:

    *errmsg = "log event entry exceeded max_allowed_packet; \

Increase max_allowed_packet on master";

    break;

  case LOG_READ_IO:

    *errmsg = "I/O error reading log event";

    break;

  case LOG_READ_MEM:

    *errmsg = "memory allocation failed reading log event";

    break;

  case LOG_READ_TRUNC:

    *errmsg = "binlog truncated in the middle of event";

    break;

  default:

    *errmsg = "unknown error reading log event on the master";

    break;

  }

  return error;

}

/**

  An auxiliary function for calling in mysql_binlog_send

  to initialize the heartbeat timeout in waiting for a binlogged event.

  @param[in]    thd  THD to access a user variable

  @return        heartbeat period an uint64_t of nanoseconds

                 or zero if heartbeat was not demanded by slave

*/ 

static uint64_t get_heartbeat_period(THD * thd)

{

  my_bool null_value;

  LEX_STRING name=  { C_STRING_WITH_LEN("master_heartbeat_period")};

  user_var_entry *entry= 

    (user_var_entry*) hash_search(&thd->user_vars, (uchar*) name.str,

                                  name.length);

  return entry? entry->val_int(&null_value) : 0;

}

/*

  Function prepares and sends repliation heartbeat event.

  @param net                net object of THD

  @param packet             buffer to store the heartbeat instance

  @param event_coordinates  binlog file name and position of the last

                            real event master sent from binlog

  @note 

    Among three essential pieces of heartbeat data Log_event::when

    is computed locally.

    The  error to send is serious and should force terminating

    the dump thread.

*/

static int send_heartbeat_event(NET* net, String* packet,

                                const struct event_coordinates *coord)

{

  char header[LOG_EVENT_HEADER_LEN];

  /*

    'when' (the timestamp) is set to 0 so that slave could distinguish between

    real and fake Rotate events (if necessary)

  */

  memset(header, 0, 4);  // when

  header[EVENT_TYPE_OFFSET] = HEARTBEAT_LOG_EVENT;

  char* p= coord->file_name + dirname_length(coord->file_name);

  uint ident_len = strlen(p);

  ulong event_len = ident_len + LOG_EVENT_HEADER_LEN;

  int4store(header + SERVER_ID_OFFSET, server_id);

  int4store(header + EVENT_LEN_OFFSET, event_len);

  int2store(header + FLAGS_OFFSET, 0);

  int4store(header + LOG_POS_OFFSET, coord->pos);  // log_pos

  packet->append(header, sizeof(header));

  packet->append(p, ident_len);             // log_file_name

  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) ||

      net_flush(net))

  {

    return(-1);

  }

  packet->set("\0", 1, &my_charset_bin);

  return(0);

}

/*

  TODO: Clean up loop to only have one call to send_file()

*/

void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,

		       ushort flags)

{

  LOG_INFO linfo;

  char *log_file_name = linfo.log_file_name;

  char search_file_name[FN_REFLEN], *name;

  IO_CACHE log;

  File file = -1;

  String* packet = &thd->packet;

  int error;

  const char *errmsg = "Unknown error";

  NET* net = &thd->net;

  pthread_mutex_t *log_lock;

  bool binlog_can_be_corrupted= false;

  memset(&log, 0, sizeof(log));

  /* 

     heartbeat_period from @master_heartbeat_period user variable

  */

  uint64_t heartbeat_period= get_heartbeat_period(thd);

  struct timespec heartbeat_buf;

  struct event_coordinates coord_buf;

  struct timespec *heartbeat_ts= NULL;

  struct event_coordinates *coord= NULL;

  if (heartbeat_period != 0LL)

  {

    heartbeat_ts= &heartbeat_buf;

    set_timespec_nsec(*heartbeat_ts, 0);

    coord= &coord_buf;

    coord->file_name= log_file_name; // initialization basing on what slave remembers

    coord->pos= pos;

  }

  if (!mysql_bin_log.is_open())

  {

    errmsg = "Binary log is not open";

    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

    goto err;

  }

  if (!server_id_supplied)

  {

    errmsg = "Misconfigured master - server id was not set";

    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

    goto err;

  }

  name=search_file_name;

  if (log_ident[0])

    mysql_bin_log.make_log_name(search_file_name, log_ident);

  else

    name=0;					// Find first log

  linfo.index_file_offset = 0;

  if (mysql_bin_log.find_log_pos(&linfo, name, 1))

  {

    errmsg = "Could not find first log file name in binary log index file";

    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

    goto err;

  }

  pthread_mutex_lock(&LOCK_thread_count);

  thd->current_linfo = &linfo;

  pthread_mutex_unlock(&LOCK_thread_count);

  if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0)

  {

    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

    goto err;

  }

  if (pos < BIN_LOG_HEADER_SIZE || pos > my_b_filelength(&log))

  {

    errmsg= "Client requested master to start replication from \

impossible position";

    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

    goto err;

  }

  /*

    We need to start a packet with something other than 255

    to distinguish it from error

  */

  packet->set("\0", 1, &my_charset_bin); /* This is the start of a new packet */

  /*

    Tell the client about the log name with a fake Rotate event;

    this is needed even if we also send a Format_description_log_event

    just after, because that event does not contain the binlog's name.

    Note that as this Rotate event is sent before

    Format_description_log_event, the slave cannot have any info to

    understand this event's format, so the header len of

    Rotate_log_event is FROZEN (so in 5.0 it will have a header shorter

    than other events except FORMAT_DESCRIPTION_EVENT).

    Before 4.0.14 we called fake_rotate_event below only if (pos ==

    BIN_LOG_HEADER_SIZE), because if this is false then the slave

    already knows the binlog's name.

    Since, we always call fake_rotate_event; if the slave already knew

    the log's name (ex: CHANGE MASTER TO MASTER_LOG_FILE=...) this is

    useless but does not harm much. It is nice for 3.23 (>=.58) slaves

    which test Rotate events to see if the master is 4.0 (then they

    choose to stop because they can't replicate 4.0); by always calling

    fake_rotate_event we are sure that 3.23.58 and newer will detect the

    problem as soon as replication starts (BUG#198).

    Always calling fake_rotate_event makes sending of normal

    (=from-binlog) Rotate events a priori unneeded, but it is not so

    simple: the 2 Rotate events are not equivalent, the normal one is

    before the Stop event, the fake one is after. If we don't send the

    normal one, then the Stop event will be interpreted (by existing 4.0

    slaves) as "the master stopped", which is wrong. So for safety,

    given that we want minimum modification of 4.0, we send the normal

    and fake Rotates.

  */

  if (fake_rotate_event(net, packet, log_file_name, pos, &errmsg))

  {

    /*

       This error code is not perfect, as fake_rotate_event() does not

       read anything from the binlog; if it fails it's because of an

       error in my_net_write(), fortunately it will say so in errmsg.

    */

    my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

    goto err;

  }

  packet->set("\0", 1, &my_charset_bin);

  /*

    Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become

    this larger than the corresponding packet (query) sent 

    from client to master.

  */

  thd->variables.max_allowed_packet+= MAX_LOG_EVENT_HEADER;

  /*

    We can set log_lock now, it does not move (it's a member of

    mysql_bin_log, and it's already inited, and it will be destroyed

    only at shutdown).

  */

  log_lock = mysql_bin_log.get_log_lock();

  if (pos > BIN_LOG_HEADER_SIZE)

  {

     /*

       Try to find a Format_description_log_event at the beginning of

       the binlog

     */

     if (!(error = Log_event::read_log_event(&log, packet, log_lock)))

     {

       /*

         The packet has offsets equal to the normal offsets in a binlog

         event +1 (the first character is \0).

       */

       if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)

       {

         binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &

                                       LOG_EVENT_BINLOG_IN_USE_F);

         (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;

         /*

           mark that this event with "log_pos=0", so the slave

           should not increment master's binlog position

           (rli->group_master_log_pos)

         */

         int4store((char*) packet->ptr()+LOG_POS_OFFSET+1, 0);

         /*

           if reconnect master sends FD event with `created' as 0

           to avoid destroying temp tables.

          */

         int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+

                   ST_CREATED_OFFSET+1, (ulong) 0);

         /* send it */

         if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))

         {

           errmsg = "Failed on my_net_write()";

           my_errno= ER_UNKNOWN_ERROR;

           goto err;

         }

         /*

           No need to save this event. We are only doing simple reads

           (no real parsing of the events) so we don't need it. And so

           we don't need the artificial Format_description_log_event of

           3.23&4.x.

         */

       }

     }

     else

     {

       if (test_for_non_eof_log_read_errors(error, &errmsg))

         goto err;

       /*

         It's EOF, nothing to do, go on reading next events, the

         Format_description_log_event will be found naturally if it is written.

       */

     }

     /* reset the packet as we wrote to it in any case */

     packet->set("\0", 1, &my_charset_bin);

  } /* end of if (pos > BIN_LOG_HEADER_SIZE); */

  else

  {

    /* The Format_description_log_event event will be found naturally. */

  }

  /* seek to the requested position, to start the requested dump */

  my_b_seek(&log, pos);			// Seek will done on next read

  while (!net->error && net->vio != 0 && !thd->killed)

  {

    while (!(error = Log_event::read_log_event(&log, packet, log_lock)))

    {

      /*

        log's filename does not change while it's active

      */

      if (coord)

        coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET);

      if ((*packet)[EVENT_TYPE_OFFSET+1] == FORMAT_DESCRIPTION_EVENT)

      {

        binlog_can_be_corrupted= test((*packet)[FLAGS_OFFSET+1] &

                                      LOG_EVENT_BINLOG_IN_USE_F);

        (*packet)[FLAGS_OFFSET+1] &= ~LOG_EVENT_BINLOG_IN_USE_F;

      }

      else if ((*packet)[EVENT_TYPE_OFFSET+1] == STOP_EVENT)

        binlog_can_be_corrupted= false;

      if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))

      {

	errmsg = "Failed on my_net_write()";

	my_errno= ER_UNKNOWN_ERROR;

	goto err;

      }

      if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)

      {

	if (send_file(thd))

	{

	  errmsg = "failed in send_file()";

	  my_errno= ER_UNKNOWN_ERROR;

	  goto err;

	}

      }

      packet->set("\0", 1, &my_charset_bin);

    }

    /*

      here we were reading binlog that was not closed properly (as a result

      of a crash ?). treat any corruption as EOF

    */

    if (binlog_can_be_corrupted && error != LOG_READ_MEM)

      error=LOG_READ_EOF;

    /*

      TODO: now that we are logging the offset, check to make sure

      the recorded offset and the actual match.

      Guilhem 2003-06: this is not true if this master is a slave

      <4.0.15 running with --log-slave-updates, because then log_pos may

      be the offset in the-master-of-this-master's binlog.

    */

    if (test_for_non_eof_log_read_errors(error, &errmsg))

      goto err;

    if (!(flags & BINLOG_DUMP_NON_BLOCK) &&

        mysql_bin_log.is_active(log_file_name))

    {

      /*

	Block until there is more data in the log

      */

      if (net_flush(net))

      {

	errmsg = "failed on net_flush()";

	my_errno= ER_UNKNOWN_ERROR;

	goto err;

      }

      /*

	We may have missed the update broadcast from the log

	that has just happened, let's try to catch it if it did.

	If we did not miss anything, we just wait for other threads

	to signal us.

      */

      {

	log.error=0;

	bool read_packet = 0, fatal_error = 0;

	/*

	  No one will update the log while we are reading

	  now, but we'll be quick and just read one record

	  TODO:

          Add an counter that is incremented for each time we update the

          binary log.  We can avoid the following read if the counter

          has not been updated since last read.

	*/

	pthread_mutex_lock(log_lock);

	switch (Log_event::read_log_event(&log, packet, (pthread_mutex_t*)0)) {

	case 0:

	  /* we read successfully, so we'll need to send it to the slave */

	  pthread_mutex_unlock(log_lock);

	  read_packet = 1;

          if (coord)

            coord->pos= uint4korr(packet->ptr() + 1 + LOG_POS_OFFSET);

	  break;

	case LOG_READ_EOF:

        {

          int ret;

	  if (thd->server_id==0) // for mysqlbinlog (mysqlbinlog.server_id==0)

	  {

	    pthread_mutex_unlock(log_lock);

	    goto end;

	  }

          do 

          {

            if (coord)

            {

              assert(heartbeat_ts && heartbeat_period != 0LL);

              set_timespec_nsec(*heartbeat_ts, heartbeat_period);

            }

            ret= mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts);

            assert(ret == 0 || (heartbeat_period != 0LL && coord != NULL));

            if (ret == ETIMEDOUT || ret == ETIME)

            {

              if (send_heartbeat_event(net, packet, coord))

              {

                errmsg = "Failed on my_net_write()";

                my_errno= ER_UNKNOWN_ERROR;

                pthread_mutex_unlock(log_lock);

                goto err;

              }

            }

            else

            {

              assert(ret == 0);

            }

          } while (ret != 0 && coord != NULL && !thd->killed);

          pthread_mutex_unlock(log_lock);

        }    

        break;

        default:

	  pthread_mutex_unlock(log_lock);

	  fatal_error = 1;

	  break;

	}

	if (read_packet)

	{

	  thd_proc_info(thd, "Sending binlog event to slave");

	  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) )

	  {

	    errmsg = "Failed on my_net_write()";

	    my_errno= ER_UNKNOWN_ERROR;

	    goto err;

	  }

	  if ((*packet)[LOG_EVENT_OFFSET+1] == LOAD_EVENT)

	  {

	    if (send_file(thd))

	    {

	      errmsg = "failed in send_file()";

	      my_errno= ER_UNKNOWN_ERROR;

	      goto err;

	    }

	  }

	  packet->set("\0", 1, &my_charset_bin);

	  /*

	    No need to net_flush because we will get to flush later when

	    we hit EOF pretty quick

	  */

	}

	if (fatal_error)

	{

	  errmsg = "error reading log entry";

          my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

	  goto err;

	}

	log.error=0;

      }

    }

    else

    {

      bool loop_breaker = 0;

      /* need this to break out of the for loop from switch */

      thd_proc_info(thd, "Finished reading one binlog; switching to next binlog");

      switch (mysql_bin_log.find_next_log(&linfo, 1)) {

      case LOG_INFO_EOF:

	loop_breaker = (flags & BINLOG_DUMP_NON_BLOCK);

	break;

      case 0:

	break;

      default:

	errmsg = "could not find next log";

	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

	goto err;

      }

      if (loop_breaker)

        break;

      end_io_cache(&log);

      (void) my_close(file, MYF(MY_WME));

      /*

        Call fake_rotate_event() in case the previous log (the one which

        we have just finished reading) did not contain a Rotate event

        (for example (I don't know any other example) the previous log

        was the last one before the master was shutdown & restarted).

        This way we tell the slave about the new log's name and

        position.  If the binlog is 5.0, the next event we are going to

        read and send is Format_description_log_event.

      */

      if ((file=open_binlog(&log, log_file_name, &errmsg)) < 0 ||

	  fake_rotate_event(net, packet, log_file_name, BIN_LOG_HEADER_SIZE,

                            &errmsg))

      {

	my_errno= ER_MASTER_FATAL_ERROR_READING_BINLOG;

	goto err;

      }

      packet->length(0);

      packet->append('\0');

      if (coord)

        coord->file_name= log_file_name; // reset to the next

    }

  }

end:

  end_io_cache(&log);

  (void)my_close(file, MYF(MY_WME));

  my_eof(thd);

  thd_proc_info(thd, "Waiting to finalize termination");

  pthread_mutex_lock(&LOCK_thread_count);

  thd->current_linfo = 0;

  pthread_mutex_unlock(&LOCK_thread_count);