~mysql/mysql-server/mysql-5.1

/* Copyright (C) 2000-2003 MySQL AB

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

/**

  @addtogroup Replication

  @{

  @file

  @brief Code to run the io thread and the sql thread on the

  replication slave.

*/

#include "mysql_priv.h"

#include <mysql.h>

#include <myisam.h>

#include "slave.h"

#include "rpl_mi.h"

#include "rpl_rli.h"

#include "sql_repl.h"

#include "rpl_filter.h"

#include "repl_failsafe.h"

#include <thr_alarm.h>

#include <my_dir.h>

#include <sql_common.h>

#include <errmsg.h>

#include <mysys_err.h>

#ifdef HAVE_REPLICATION

#include "rpl_tblmap.h"

#define FLAGSTR(V,F) ((V)&(F)?#F" ":"")

#define MAX_SLAVE_RETRY_PAUSE 5

bool use_slave_mask = 0;

MY_BITMAP slave_error_mask;

typedef bool (*CHECK_KILLED_FUNC)(THD*,void*);

char* slave_load_tmpdir = 0;

Master_info *active_mi= 0;

my_bool replicate_same_server_id;

ulonglong relay_log_space_limit = 0;

/*

  When slave thread exits, we need to remember the temporary tables so we

  can re-use them on slave start.

  TODO: move the vars below under Master_info

*/

int disconnect_slave_event_count = 0, abort_slave_event_count = 0;

int events_till_abort = -1;

enum enum_slave_reconnect_actions

{

  SLAVE_RECON_ACT_REG= 0,

  SLAVE_RECON_ACT_DUMP= 1,

  SLAVE_RECON_ACT_EVENT= 2,

  SLAVE_RECON_ACT_MAX

};

enum enum_slave_reconnect_messages

{

  SLAVE_RECON_MSG_WAIT= 0,

  SLAVE_RECON_MSG_KILLED_WAITING= 1,

  SLAVE_RECON_MSG_AFTER= 2,

  SLAVE_RECON_MSG_FAILED= 3,

  SLAVE_RECON_MSG_COMMAND= 4,

  SLAVE_RECON_MSG_KILLED_AFTER= 5,

  SLAVE_RECON_MSG_MAX

};

static const char *reconnect_messages[SLAVE_RECON_ACT_MAX][SLAVE_RECON_MSG_MAX]=

{

  {

    "Waiting to reconnect after a failed registration on master",

    "Slave I/O thread killed while waitnig to reconnect after a failed \

registration on master",

    "Reconnecting after a failed registration on master",

    "failed registering on master, reconnecting to try again, \

log '%s' at postion %s",

    "COM_REGISTER_SLAVE",

    "Slave I/O thread killed during or after reconnect"

  },

  {

    "Waiting to reconnect after a failed binlog dump request",

    "Slave I/O thread killed while retrying master dump",

    "Reconnecting after a failed binlog dump request",

    "failed dump request, reconnecting to try again, log '%s' at postion %s",

    "COM_BINLOG_DUMP",

    "Slave I/O thread killed during or after reconnect"

  },

  {

    "Waiting to reconnect after a failed master event read",

    "Slave I/O thread killed while waiting to reconnect after a failed read",

    "Reconnecting after a failed master event read",

    "Slave I/O thread: Failed reading log event, reconnecting to retry, \

log '%s' at postion %s",

    "",

    "Slave I/O thread killed during or after a reconnect done to recover from \

failed read"

  }

};

typedef enum { SLAVE_THD_IO, SLAVE_THD_SQL} SLAVE_THD_TYPE;

static int process_io_rotate(Master_info* mi, Rotate_log_event* rev);

static int process_io_create_file(Master_info* mi, Create_file_log_event* cev);

static bool wait_for_relay_log_space(Relay_log_info* rli);

static inline bool io_slave_killed(THD* thd,Master_info* mi);

static inline bool sql_slave_killed(THD* thd,Relay_log_info* rli);

static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type);

static int safe_connect(THD* thd, MYSQL* mysql, Master_info* mi);

static int safe_reconnect(THD* thd, MYSQL* mysql, Master_info* mi,

                          bool suppress_warnings);

static int connect_to_master(THD* thd, MYSQL* mysql, Master_info* mi,

                             bool reconnect, bool suppress_warnings);

static int safe_sleep(THD* thd, int sec, CHECK_KILLED_FUNC thread_killed,

                      void* thread_killed_arg);

static int request_table_dump(MYSQL* mysql, const char* db, const char* table);

static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db,

                                  const char* table_name, bool overwrite);

static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi);

static Log_event* next_event(Relay_log_info* rli);

static int queue_event(Master_info* mi,const char* buf,ulong event_len);

static int terminate_slave_thread(THD *thd,

                                  pthread_mutex_t* term_lock,

                                  pthread_cond_t* term_cond,

                                  volatile uint *slave_running,

                                  bool skip_lock);

static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info);

/*

  Find out which replications threads are running

  SYNOPSIS

    init_thread_mask()

    mask                Return value here

    mi                  master_info for slave

    inverse             If set, returns which threads are not running

  IMPLEMENTATION

    Get a bit mask for which threads are running so that we can later restart

    these threads.

  RETURN

    mask        If inverse == 0, running threads

                If inverse == 1, stopped threads

*/

void init_thread_mask(int* mask,Master_info* mi,bool inverse)

{

  bool set_io = mi->slave_running, set_sql = mi->rli.slave_running;

  register int tmp_mask=0;

  DBUG_ENTER("init_thread_mask");

  if (set_io)

    tmp_mask |= SLAVE_IO;

  if (set_sql)

    tmp_mask |= SLAVE_SQL;

  if (inverse)

    tmp_mask^= (SLAVE_IO | SLAVE_SQL);

  *mask = tmp_mask;

  DBUG_VOID_RETURN;

}

/*

  lock_slave_threads()

*/

void lock_slave_threads(Master_info* mi)

{

  DBUG_ENTER("lock_slave_threads");

  //TODO: see if we can do this without dual mutex

  pthread_mutex_lock(&mi->run_lock);

  pthread_mutex_lock(&mi->rli.run_lock);

  DBUG_VOID_RETURN;

}

/*

  unlock_slave_threads()

*/

void unlock_slave_threads(Master_info* mi)

{

  DBUG_ENTER("unlock_slave_threads");

  //TODO: see if we can do this without dual mutex

  pthread_mutex_unlock(&mi->rli.run_lock);

  pthread_mutex_unlock(&mi->run_lock);

  DBUG_VOID_RETURN;

}

/* Initialize slave structures */

int init_slave()

{

  DBUG_ENTER("init_slave");

  /*

    This is called when mysqld starts. Before client connections are

    accepted. However bootstrap may conflict with us if it does START SLAVE.

    So it's safer to take the lock.

  */

  pthread_mutex_lock(&LOCK_active_mi);

  /*

    TODO: re-write this to interate through the list of files

    for multi-master

  */

  active_mi= new Master_info;

  /*

    If master_host is not specified, try to read it from the master_info file.

    If master_host is specified, create the master_info file if it doesn't

    exists.

  */

  if (!active_mi)

  {

    sql_print_error("Failed to allocate memory for the master info structure");

    goto err;

  }

  if (init_master_info(active_mi,master_info_file,relay_log_info_file,

                       !master_host, (SLAVE_IO | SLAVE_SQL)))

  {

    sql_print_error("Failed to initialize the master info structure");

    goto err;

  }

  if (server_id && !master_host && active_mi->host[0])

    master_host= active_mi->host;

  /* If server id is not set, start_slave_thread() will say it */

  if (master_host && !opt_skip_slave_start)

  {

    if (start_slave_threads(1 /* need mutex */,

                            0 /* no wait for start*/,

                            active_mi,

                            master_info_file,

                            relay_log_info_file,

                            SLAVE_IO | SLAVE_SQL))

    {

      sql_print_error("Failed to create slave threads");

      goto err;

    }

  }

  pthread_mutex_unlock(&LOCK_active_mi);

  DBUG_RETURN(0);

err:

  pthread_mutex_unlock(&LOCK_active_mi);

  DBUG_RETURN(1);

}

/*

  Init function to set up array for errors that should be skipped for slave

  SYNOPSIS

    init_slave_skip_errors()

    arg         List of errors numbers to skip, separated with ','

  NOTES

    Called from get_options() in mysqld.cc on start-up

*/

void init_slave_skip_errors(const char* arg)

{

  const char *p;

  DBUG_ENTER("init_slave_skip_errors");

  if (bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0))

  {

    fprintf(stderr, "Badly out of memory, please check your system status\n");

    exit(1);

  }

  use_slave_mask = 1;

  for (;my_isspace(system_charset_info,*arg);++arg)

    /* empty */;

  if (!my_strnncoll(system_charset_info,(uchar*)arg,4,(const uchar*)"all",4))

  {

    bitmap_set_all(&slave_error_mask);

    DBUG_VOID_RETURN;

  }

  for (p= arg ; *p; )

  {

    long err_code;

    if (!(p= str2int(p, 10, 0, LONG_MAX, &err_code)))

      break;

    if (err_code < MAX_SLAVE_ERROR)

       bitmap_set_bit(&slave_error_mask,(uint)err_code);

    while (!my_isdigit(system_charset_info,*p) && *p)

      p++;

  }

  DBUG_VOID_RETURN;

}

int terminate_slave_threads(Master_info* mi,int thread_mask,bool skip_lock)

{

  DBUG_ENTER("terminate_slave_threads");

  if (!mi->inited)

    DBUG_RETURN(0); /* successfully do nothing */

  int error,force_all = (thread_mask & SLAVE_FORCE_ALL);

  pthread_mutex_t *sql_lock = &mi->rli.run_lock, *io_lock = &mi->run_lock;

  if ((thread_mask & (SLAVE_IO|SLAVE_FORCE_ALL)))

  {

    DBUG_PRINT("info",("Terminating IO thread"));

    mi->abort_slave=1;

    if ((error=terminate_slave_thread(mi->io_thd,io_lock,

                                      &mi->stop_cond,

                                      &mi->slave_running,

                                      skip_lock)) &&

        !force_all)

      DBUG_RETURN(error);

  }

  if ((thread_mask & (SLAVE_SQL|SLAVE_FORCE_ALL)))

  {

    DBUG_PRINT("info",("Terminating SQL thread"));

    mi->rli.abort_slave=1;

    if ((error=terminate_slave_thread(mi->rli.sql_thd,sql_lock,

                                      &mi->rli.stop_cond,

                                      &mi->rli.slave_running,

                                      skip_lock)) &&

        !force_all)

      DBUG_RETURN(error);

  }

  DBUG_RETURN(0);

}

/**

   Wait for a slave thread to terminate.

   This function is called after requesting the thread to terminate

   (by setting @c abort_slave member of @c Relay_log_info or @c

   Master_info structure to 1). Termination of the thread is

   controlled with the the predicate <code>*slave_running</code>.

   Function will acquire @c term_lock before waiting on the condition

   unless @c skip_lock is true in which case the mutex should be owned

   by the caller of this function and will remain acquired after

   return from the function.

   @param term_lock

          Associated lock to use when waiting for @c term_cond

   @param term_cond

          Condition that is signalled when the thread has terminated

   @param slave_running

          Pointer to predicate to check for slave thread termination

   @param skip_lock

          If @c true the lock will not be acquired before waiting on

          the condition. In this case, it is assumed that the calling

          function acquires the lock before calling this function.

   @retval 0 All OK

 */

static int

terminate_slave_thread(THD *thd,

                       pthread_mutex_t* term_lock,

                       pthread_cond_t* term_cond,

                       volatile uint *slave_running,

                       bool skip_lock)

{

  int error;

  DBUG_ENTER("terminate_slave_thread");

  if (!skip_lock)

    pthread_mutex_lock(term_lock);

  safe_mutex_assert_owner(term_lock);

  if (!*slave_running)

  {

    if (!skip_lock)

      pthread_mutex_unlock(term_lock);

    DBUG_RETURN(ER_SLAVE_NOT_RUNNING);

  }

  DBUG_ASSERT(thd != 0);

  THD_CHECK_SENTRY(thd);

  /*

    Is is critical to test if the slave is running. Otherwise, we might

    be referening freed memory trying to kick it

  */

  while (*slave_running)                        // Should always be true

  {

    DBUG_PRINT("loop", ("killing slave thread"));

    pthread_mutex_lock(&thd->LOCK_delete);

#ifndef DONT_USE_THR_ALARM

    /*

      Error codes from pthread_kill are:

      EINVAL: invalid signal number (can't happen)

      ESRCH: thread already killed (can happen, should be ignored)

    */

    IF_DBUG(int err= ) pthread_kill(thd->real_id, thr_client_alarm);

    DBUG_ASSERT(err != EINVAL);

#endif

    thd->awake(THD::NOT_KILLED);

    pthread_mutex_unlock(&thd->LOCK_delete);

    /*

      There is a small chance that slave thread might miss the first

      alarm. To protect againts it, resend the signal until it reacts

    */

    struct timespec abstime;

    set_timespec(abstime,2);

    error= pthread_cond_timedwait(term_cond, term_lock, &abstime);

    DBUG_ASSERT(error == ETIMEDOUT || error == 0);

  }

  DBUG_ASSERT(*slave_running == 0);

  if (!skip_lock)

    pthread_mutex_unlock(term_lock);

  DBUG_RETURN(0);

}

int start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,

                       pthread_mutex_t *cond_lock,

                       pthread_cond_t *start_cond,

                       volatile uint *slave_running,

                       volatile ulong *slave_run_id,

                       Master_info* mi,

                       bool high_priority)

{

  pthread_t th;

  ulong start_id;

  DBUG_ENTER("start_slave_thread");

  DBUG_ASSERT(mi->inited);

  if (start_lock)

    pthread_mutex_lock(start_lock);

  if (!server_id)

  {

    if (start_cond)

      pthread_cond_broadcast(start_cond);

    if (start_lock)

      pthread_mutex_unlock(start_lock);

    sql_print_error("Server id not set, will not start slave");

    DBUG_RETURN(ER_BAD_SLAVE);

  }

  if (*slave_running)

  {

    if (start_cond)

      pthread_cond_broadcast(start_cond);

    if (start_lock)

      pthread_mutex_unlock(start_lock);

    DBUG_RETURN(ER_SLAVE_MUST_STOP);

  }

  start_id= *slave_run_id;

  DBUG_PRINT("info",("Creating new slave thread"));

  if (high_priority)

    my_pthread_attr_setprio(&connection_attrib,CONNECT_PRIOR);

  if (pthread_create(&th, &connection_attrib, h_func, (void*)mi))

  {

    if (start_lock)

      pthread_mutex_unlock(start_lock);

    DBUG_RETURN(ER_SLAVE_THREAD);

  }

  if (start_cond && cond_lock) // caller has cond_lock

  {

    THD* thd = current_thd;

    while (start_id == *slave_run_id)

    {

      DBUG_PRINT("sleep",("Waiting for slave thread to start"));

      const char* old_msg = thd->enter_cond(start_cond,cond_lock,

                                            "Waiting for slave thread to start");

      pthread_cond_wait(start_cond,cond_lock);

      thd->exit_cond(old_msg);

      pthread_mutex_lock(cond_lock); // re-acquire it as exit_cond() released

      if (thd->killed)

        DBUG_RETURN(thd->killed_errno());

    }

  }

  if (start_lock)

    pthread_mutex_unlock(start_lock);

  DBUG_RETURN(0);

}

/*

  start_slave_threads()

  NOTES

    SLAVE_FORCE_ALL is not implemented here on purpose since it does not make

    sense to do that for starting a slave--we always care if it actually

    started the threads that were not previously running

*/

int start_slave_threads(bool need_slave_mutex, bool wait_for_start,

                        Master_info* mi, const char* master_info_fname,

                        const char* slave_info_fname, int thread_mask)

{

  pthread_mutex_t *lock_io=0,*lock_sql=0,*lock_cond_io=0,*lock_cond_sql=0;

  pthread_cond_t* cond_io=0,*cond_sql=0;

  int error=0;

  DBUG_ENTER("start_slave_threads");

  if (need_slave_mutex)

  {

    lock_io = &mi->run_lock;

    lock_sql = &mi->rli.run_lock;

  }

  if (wait_for_start)

  {

    cond_io = &mi->start_cond;

    cond_sql = &mi->rli.start_cond;

    lock_cond_io = &mi->run_lock;

    lock_cond_sql = &mi->rli.run_lock;

  }

  if (thread_mask & SLAVE_IO)

    error=start_slave_thread(handle_slave_io,lock_io,lock_cond_io,

                             cond_io,

                             &mi->slave_running, &mi->slave_run_id,

                             mi, 1); //high priority, to read the most possible

  if (!error && (thread_mask & SLAVE_SQL))

  {

    error=start_slave_thread(handle_slave_sql,lock_sql,lock_cond_sql,

                             cond_sql,

                             &mi->rli.slave_running, &mi->rli.slave_run_id,

                             mi, 0);

    if (error)

      terminate_slave_threads(mi, thread_mask & SLAVE_IO, 0);

  }

  DBUG_RETURN(error);

}

#ifdef NOT_USED_YET

static int end_slave_on_walk(Master_info* mi, uchar* /*unused*/)

{

  DBUG_ENTER("end_slave_on_walk");

  end_master_info(mi);

  DBUG_RETURN(0);

}

#endif

/*

  Free all resources used by slave

  SYNOPSIS

    end_slave()

*/

void end_slave()

{

  DBUG_ENTER("end_slave");

  /*

    This is called when the server terminates, in close_connections().

    It terminates slave threads. However, some CHANGE MASTER etc may still be

    running presently. If a START SLAVE was in progress, the mutex lock below

    will make us wait until slave threads have started, and START SLAVE

    returns, then we terminate them here.

  */

  pthread_mutex_lock(&LOCK_active_mi);

  if (active_mi)

  {

    /*

      TODO: replace the line below with

      list_walk(&master_list, (list_walk_action)end_slave_on_walk,0);

      once multi-master code is ready.

    */

    terminate_slave_threads(active_mi,SLAVE_FORCE_ALL);

    end_master_info(active_mi);

    delete active_mi;

    active_mi= 0;

  }

  pthread_mutex_unlock(&LOCK_active_mi);

  DBUG_VOID_RETURN;

}

static bool io_slave_killed(THD* thd, Master_info* mi)

{

  DBUG_ENTER("io_slave_killed");

  DBUG_ASSERT(mi->io_thd == thd);

  DBUG_ASSERT(mi->slave_running); // tracking buffer overrun

  DBUG_RETURN(mi->abort_slave || abort_loop || thd->killed);

}

static bool sql_slave_killed(THD* thd, Relay_log_info* rli)

{

  DBUG_ENTER("sql_slave_killed");

  DBUG_ASSERT(rli->sql_thd == thd);

  DBUG_ASSERT(rli->slave_running == 1);// tracking buffer overrun

  if (abort_loop || thd->killed || rli->abort_slave)

  {

    /*

      If we are in an unsafe situation (stopping could corrupt replication),

      we give one minute to the slave SQL thread of grace before really

      terminating, in the hope that it will be able to read more events and

      the unsafe situation will soon be left. Note that this one minute starts

      from the last time anything happened in the slave SQL thread. So it's

      really one minute of idleness, we don't timeout if the slave SQL thread

      is actively working.

    */

    if (rli->last_event_start_time == 0)

      DBUG_RETURN(1);

    DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving "

                        "it some grace period"));

    if (difftime(time(0), rli->last_event_start_time) > 60)

    {

      rli->report(ERROR_LEVEL, 0,

                  "SQL thread had to stop in an unsafe situation, in "

                  "the middle of applying updates to a "

                  "non-transactional table without any primary key. "

                  "There is a risk of duplicate updates when the slave "

                  "SQL thread is restarted. Please check your tables' "

                  "contents after restart.");

      DBUG_RETURN(1);

    }

  }

  DBUG_RETURN(0);

}

/*

  skip_load_data_infile()

  NOTES

    This is used to tell a 3.23 master to break send_file()

*/

void skip_load_data_infile(NET *net)

{

  DBUG_ENTER("skip_load_data_infile");

  (void)net_request_file(net, "/dev/null");

  (void)my_net_read(net);                               // discard response

  (void)net_write_command(net, 0, (uchar*) "", 0, (uchar*) "", 0); // ok

  DBUG_VOID_RETURN;

}

bool net_request_file(NET* net, const char* fname)

{

  DBUG_ENTER("net_request_file");

  DBUG_RETURN(net_write_command(net, 251, (uchar*) fname, strlen(fname),

                                (uchar*) "", 0));

}

/*

  From other comments and tests in code, it looks like

  sometimes Query_log_event and Load_log_event can have db == 0