~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2003 DRIZZLE 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 <drizzled/server_includes.h>

#include <storage/myisam/myisam.h>

#include <drizzled/replication/mi.h>

#include <drizzled/replication/rli.h>

#include <drizzled/replication/replication.h>

#include <drizzled/replication/filter.h>

#include <mysys/thr_alarm.h>

#include <libdrizzle/errmsg.h>

#include <mysys/mysys_err.h>

#include <drizzled/error.h>

#include <drizzled/sql_parse.h>

#include <drizzled/gettext.h>

#include <signal.h>

#if TIME_WITH_SYS_TIME

# include <sys/time.h>

# include <time.h>

#else

# if HAVE_SYS_TIME_H

#  include <sys/time.h>

# else

#  include <time.h>

# endif

#endif

#include <drizzled/tztime.h>

#include <drizzled/replication/tblmap.h>

#define MAX_SLAVE_RETRY_PAUSE 5

bool use_slave_mask = 0;

MY_BITMAP slave_error_mask;

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

char* slave_load_tmpdir = 0;

Master_info *active_mi= 0;

bool replicate_same_server_id;

uint64_t 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

*/

int32_t disconnect_slave_event_count = 0, abort_slave_event_count = 0;

int32_t 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]=

{

  {

    N_("Waiting to reconnect after a failed registration on master"),

    N_("Slave I/O thread killed while waitnig to reconnect after a "

                 "failed registration on master"),

    N_("Reconnecting after a failed registration on master"),

    N_("failed registering on master, reconnecting to try again, "

                 "log '%s' at postion %s"),

    "COM_REGISTER_SLAVE",

    N_("Slave I/O thread killed during or after reconnect")

  },

  {

    N_("Waiting to reconnect after a failed binlog dump request"),

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

    N_("Reconnecting after a failed binlog dump request"),

    N_("failed dump request, reconnecting to try again, "

                 "log '%s' at postion %s"),

    "COM_BINLOG_DUMP",

    N_("Slave I/O thread killed during or after reconnect")

  },

  {

    N_("Waiting to reconnect after a failed master event read"),

    N_("Slave I/O thread killed while waiting to reconnect "

                 "after a failed read"),

    N_("Reconnecting after a failed master event read"),

    N_("Slave I/O thread: Failed reading log event, "

                 "reconnecting to retry, log '%s' at postion %s"),

    "",

    N_("Slave I/O thread killed during or after a "

                 "reconnect done to recover from failed read")

  }

};

typedef enum { SLAVE_Session_IO, SLAVE_Session_SQL} SLAVE_Session_TYPE;

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

static int32_t 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(Session* session,Master_info* mi);

static inline bool sql_slave_killed(Session* session,Relay_log_info* rli);

static int32_t init_slave_thread(Session* session, SLAVE_Session_TYPE session_type);

static int32_t safe_connect(Session* session, DRIZZLE *drizzle, Master_info* mi);

static int32_t safe_reconnect(Session* session, DRIZZLE *drizzle, Master_info* mi,

                          bool suppress_warnings);

static int32_t connect_to_master(Session* session, DRIZZLE *drizzle, Master_info* mi,

                             bool reconnect, bool suppress_warnings);

static int32_t safe_sleep(Session* session, int32_t sec, CHECK_KILLED_FUNC thread_killed,

                      void* thread_killed_arg);

static int32_t get_master_version_and_clock(DRIZZLE *drizzle, Master_info* mi);

static Log_event* next_event(Relay_log_info* rli);

static int32_t queue_event(Master_info* mi,const char* buf,uint32_t event_len);

static int32_t terminate_slave_thread(Session *session,

                                  pthread_mutex_t* term_lock,

                                  pthread_cond_t* term_cond,

                                  volatile uint32_t *slave_running,

                                  bool skip_lock);

static bool check_io_slave_killed(Session *session, 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(int32_t* mask,Master_info* mi,bool inverse)

{

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

  register int32_t tmp_mask=0;

  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;

  return;

}

/*

  lock_slave_threads()

*/

void lock_slave_threads(Master_info* mi)

{

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

  pthread_mutex_lock(&mi->run_lock);

  pthread_mutex_lock(&mi->rli.run_lock);

  return;

}

/*

  unlock_slave_threads()

*/

void unlock_slave_threads(Master_info* mi)

{

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

  pthread_mutex_unlock(&mi->rli.run_lock);

  pthread_mutex_unlock(&mi->run_lock);

  return;

}

/* Initialize slave structures */

int32_t 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 (active_mi->init_master_info(master_info_file, relay_log_info_file, (SLAVE_IO | SLAVE_SQL)))

  {

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

    goto err;

  }

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

  if (active_mi->host[0] && !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);

  return(0);

err:

  pthread_mutex_unlock(&LOCK_active_mi);

  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;

  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,(unsigned char*)arg,4,(const unsigned char*)"all",4))

  {

    bitmap_set_all(&slave_error_mask);

    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,(uint32_t)err_code);

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

      p++;

  }

  return;

}

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

{

  if (!mi->inited)

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

  int32_t 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)))

  {

    mi->abort_slave=1;

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

                                      &mi->stop_cond,

                                      &mi->slave_running,

                                      skip_lock)) &&

        !force_all)

      return(error);

  }

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

  {

    mi->rli.abort_slave=1;

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

                                      &mi->rli.stop_cond,

                                      &mi->rli.slave_running,

                                      skip_lock)) &&

        !force_all)

      return(error);

  }

  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 int32_t

terminate_slave_thread(Session *session,

                       pthread_mutex_t* term_lock,

                       pthread_cond_t* term_cond,

                       volatile uint32_t *slave_running,

                       bool skip_lock)

{

  int32_t error;

  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);

    return(ER_SLAVE_NOT_RUNNING);

  }

  assert(session != 0);

  Session_CHECK_SENTRY(session);

  /*

    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

  {

    pthread_mutex_lock(&session->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)

    */

    int32_t err= pthread_kill(session->real_id, thr_client_alarm);

    assert(err != EINVAL);

#endif

    session->awake(Session::NOT_KILLED);

    pthread_mutex_unlock(&session->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);

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

  }

  assert(*slave_running == 0);

  if (!skip_lock)

    pthread_mutex_unlock(term_lock);

  return(0);

}

int32_t start_slave_thread(pthread_handler h_func, pthread_mutex_t *start_lock,

                           pthread_mutex_t *cond_lock,

                           pthread_cond_t *start_cond,

                           volatile uint32_t *slave_running,

                           volatile uint32_t *slave_run_id,

                           Master_info* mi,

                           bool high_priority)

{

  pthread_t th;

  uint32_t start_id;

  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"));

    return(ER_BAD_SLAVE);

  }

  if (*slave_running)

  {

    if (start_cond)

      pthread_cond_broadcast(start_cond);

    if (start_lock)

      pthread_mutex_unlock(start_lock);

    return(ER_SLAVE_MUST_STOP);

  }

  start_id= *slave_run_id;

  if (high_priority)

  {

    struct sched_param tmp_sched_param;

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

    tmp_sched_param.sched_priority= CONNECT_PRIOR;

    (void)pthread_attr_setschedparam(&connection_attrib, &tmp_sched_param);

  }

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

  {

    if (start_lock)

      pthread_mutex_unlock(start_lock);

    return(ER_SLAVE_THREAD);

  }

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

  {

    Session* session = current_session;

    while (start_id == *slave_run_id)

    {

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

                                            "Waiting for slave thread to start");

      pthread_cond_wait(start_cond,cond_lock);

      session->exit_cond(old_msg);

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

      if (session->killed)

        return(session->killed_errno());

    }

  }

  if (start_lock)

    pthread_mutex_unlock(start_lock);

  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

*/

int32_t start_slave_threads(bool need_slave_mutex, bool wait_for_start,

                        Master_info* mi, const char*, const char*,

                        int32_t 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;

  int32_t error=0;

  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);

  }

  return(error);

}

#ifdef NOT_USED_YET

static int32_t end_slave_on_walk(Master_info* mi, unsigned char* /*unused*/)

{

  end_master_info(mi);

  return(0);

}

#endif

/*

  Free all resources used by slave

  SYNOPSIS

    end_slave()

*/

void 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);

    active_mi->end_master_info();

    delete active_mi;

    active_mi= 0;

  }

  pthread_mutex_unlock(&LOCK_active_mi);

  return;

}

static bool io_slave_killed(Session* session, Master_info* mi)

{

  assert(mi->io_session == session);

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

  return(mi->abort_slave || abort_loop || session->killed);

}

static bool sql_slave_killed(Session* session, Relay_log_info* rli)

{

  assert(rli->sql_session == session);

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

  if (abort_loop || session->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)

      return(1);

    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."));

      return(1);

    }

  }

  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)

{

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

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

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

  return;

}

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

{

  return(net_write_command(net, 251, (unsigned char*) fname, strlen(fname),

                                (unsigned char*) "", 0));

}

/*

  From other comments and tests in code, it looks like

  sometimes Query_log_event and Load_log_event can have db == 0

  (see rewrite_db() above for example)

  (cases where this happens are unclear; it may be when the master is 3.23).

*/

const char *print_slave_db_safe(const char* db)

{

  return((db ? db : ""));

}

int32_t init_strvar_from_file(char *var, int32_t max_size, IO_CACHE *f,

                                 const char *default_val)

{

  uint32_t length;

  if ((length=my_b_gets(f,var, max_size)))

  {

    char* last_p = var + length -1;

    if (*last_p == '\n')

      *last_p = 0; // if we stopped on newline, kill it

    else

    {

      /*

        If we truncated a line or stopped on last char, remove all chars

        up to and including newline.

      */

      int32_t c;

      while (((c=my_b_get(f)) != '\n' && c != my_b_EOF)) {};

    }

    return(0);

  }

  else if (default_val)

  {

    strmake(var,  default_val, max_size-1);

    return(0);

  }

  return(1);

}

int32_t init_intvar_from_file(int32_t* var, IO_CACHE* f, int32_t default_val)

{

  char buf[32];

  if (my_b_gets(f, buf, sizeof(buf)))

  {

    *var = atoi(buf);

    return(0);

  }

  else if (default_val)

  {

    *var = default_val;

    return(0);

  }

  return(1);

}

int32_t init_floatvar_from_file(float* var, IO_CACHE* f, float default_val)

{

  char buf[16];

  if (my_b_gets(f, buf, sizeof(buf)))

  {

    if (sscanf(buf, "%f", var) != 1)

      return(1);

    else

      return(0);

  }