开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
slave.cc
资源名称：mysql-3.23.35.tar.gz　[点击查看]
上传用户：tsgydb
上传日期：2007-04-14
资源大小：10674k
文件大小：39k
源码类别：

MySQL数据库

开发平台：

Visual C++

slave.cc：源码内容
  1. /* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
  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; either version 2 of the License, or
  6.    (at your option) any later version.
  7.    
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  12.    
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  18. #include "mysql_priv.h"
  19. #include <mysql.h>
  20. #include <myisam.h>
  21. #include "mini_client.h"
  22. #include "slave.h"
  23. #include <thr_alarm.h>
  24. #include <my_dir.h>
  26. #define RPL_LOG_NAME (glob_mi.log_file_name[0] ? glob_mi.log_file_name :
  27.  "FIRST")
  29. bool slave_running = 0;
  30. pthread_t slave_real_id;
  31. MASTER_INFO glob_mi;
  32. HASH replicate_do_table, replicate_ignore_table;
  33. DYNAMIC_ARRAY replicate_wild_do_table, replicate_wild_ignore_table;
  34. bool do_table_inited = 0, ignore_table_inited = 0;
  35. bool wild_do_table_inited = 0, wild_ignore_table_inited = 0;
  36. bool table_rules_on = 0;
  37. uint32 slave_skip_counter = 0; 
  38. static TABLE* save_temporary_tables = 0;
  39. THD* slave_thd = 0;
  40. // when slave thread exits, we need to remember the temporary tables so we
  41. // can re-use them on slave start
  43. static int last_slave_errno = 0;
  44. static char last_slave_error[1024] = "";
  45. #ifndef DBUG_OFF
  46. int disconnect_slave_event_count = 0, abort_slave_event_count = 0;
  47. static int events_till_disconnect = -1, events_till_abort = -1;
  48. static int stuck_count = 0;
  49. #endif
  52. inline void skip_load_data_infile(NET* net);
  53. inline bool slave_killed(THD* thd);
  54. static int init_slave_thread(THD* thd);
  55. static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi);
  56. static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi);
  57. static int safe_sleep(THD* thd, int sec);
  58. static int request_table_dump(MYSQL* mysql, char* db, char* table);
  59. static int create_table_from_dump(THD* thd, NET* net, const char* db,
  60.   const char* table_name);
  61. inline char* rewrite_db(char* db);
  62. static void free_table_ent(TABLE_RULE_ENT* e)
  63. {
  64.   my_free((gptr) e, MYF(0));
  65. }
  67. static byte* get_table_key(TABLE_RULE_ENT* e, uint* len,
  68.    my_bool not_used __attribute__((unused)))
  69. {
  70.   *len = e->key_len;
  71.   return (byte*)e->db;
  72. }
  75. void init_table_rule_hash(HASH* h, bool* h_inited)
  76. {
  77.   hash_init(h, TABLE_RULE_HASH_SIZE,0,0,
  78.     (hash_get_key) get_table_key,
  79.     (void (*)(void*)) free_table_ent, 0);
  80.   *h_inited = 1;
  81. }
  83. void init_table_rule_array(DYNAMIC_ARRAY* a, bool* a_inited)
  84. {
  85.   init_dynamic_array(a, sizeof(TABLE_RULE_ENT*), TABLE_RULE_ARR_SIZE,
  86.      TABLE_RULE_ARR_SIZE);
  87.   *a_inited = 1;
  88. }
  90. static TABLE_RULE_ENT* find_wild(DYNAMIC_ARRAY *a, const char* key, int len)
  91. {
  92.   uint i;
  93.   const char* key_end = key + len;
  94.   
  95.   for(i = 0; i < a->elements; i++)
  96.     {
  97.       TABLE_RULE_ENT* e ;
  98.       get_dynamic(a, (gptr)&e, i);
  99.       if(!wild_case_compare(key, key_end, (const char*)e->db,
  100.     (const char*)(e->db + e->key_len),'\'))
  101. return e;
  102.     }
  103.   
  104.   return 0;
  105. }
  107. int tables_ok(THD* thd, TABLE_LIST* tables)
  108. {
  109.   for (; tables; tables = tables->next)
  110.   {
  111.     if (!tables->updating) 
  112.       continue;
  113.     char hash_key[2*NAME_LEN+2];
  114.     char* p;
  115.     p = strmov(hash_key, tables->db ? tables->db : thd->db);
  116.     *p++ = '.';
  117.     uint len = strmov(p, tables->real_name) - hash_key ;
  118.     if (do_table_inited) // if there are any do's
  119.     {
  120.       if (hash_search(&replicate_do_table, (byte*) hash_key, len))
  121. return 1;
  122.     }
  123.     if (ignore_table_inited) // if there are any do's
  124.     {
  125.       if (hash_search(&replicate_ignore_table, (byte*) hash_key, len))
  126. return 0; 
  127.     }
  128.     if (wild_do_table_inited && find_wild(&replicate_wild_do_table,
  129.   hash_key, len))
  130.       return 1;
  131.     if (wild_ignore_table_inited && find_wild(&replicate_wild_ignore_table,
  132.       hash_key, len))
  133.       return 0;
  134.   }
  136.   // if no explicit rule found
  137.   // and there was a do list, do not replicate. If there was
  138.   // no do list, go ahead
  139.   return !do_table_inited && !wild_do_table_inited;
  140. }
  143. int add_table_rule(HASH* h, const char* table_spec)
  144. {
  145.   const char* dot = strchr(table_spec, '.');
  146.   if(!dot) return 1;
  147.   // len is always > 0 because we know the there exists a '.'
  148.   uint len = (uint)strlen(table_spec);
  149.   TABLE_RULE_ENT* e = (TABLE_RULE_ENT*)my_malloc(sizeof(TABLE_RULE_ENT)
  150.  + len, MYF(MY_WME));
  151.   if(!e) return 1;
  152.   e->db = (char*)e + sizeof(TABLE_RULE_ENT);
  153.   e->tbl_name = e->db + (dot - table_spec) + 1;
  154.   e->key_len = len;
  155.   memcpy(e->db, table_spec, len);
  156.   (void)hash_insert(h, (byte*)e);
  157.   return 0;
  158. }
  160. int add_wild_table_rule(DYNAMIC_ARRAY* a, const char* table_spec)
  161. {
  162.   const char* dot = strchr(table_spec, '.');
  163.   if(!dot) return 1;
  164.   uint len = (uint)strlen(table_spec);
  165.   TABLE_RULE_ENT* e = (TABLE_RULE_ENT*)my_malloc(sizeof(TABLE_RULE_ENT)
  166.  + len, MYF(MY_WME));
  167.   if(!e) return 1;
  168.   e->db = (char*)e + sizeof(TABLE_RULE_ENT);
  169.   e->tbl_name = e->db + (dot - table_spec) + 1;
  170.   e->key_len = len;
  171.   memcpy(e->db, table_spec, len);
  172.   insert_dynamic(a, (gptr)&e);
  173.   return 0;
  174. }
  176. static void free_string_array(DYNAMIC_ARRAY *a)
  177. {
  178.   uint i;
  179.   for(i = 0; i < a->elements; i++)
  180.     {
  181.       char* p;
  182.       get_dynamic(a, (gptr) &p, i);
  183.       my_free(p, MYF(MY_WME));
  184.     }
  185.   delete_dynamic(a);
  186. }
  188. void end_slave()
  189. {
  190.   end_master_info(&glob_mi);
  191.   if(do_table_inited)
  192.     hash_free(&replicate_do_table);
  193.   if(ignore_table_inited)
  194.     hash_free(&replicate_ignore_table);
  195.   if(wild_do_table_inited)
  196.     free_string_array(&replicate_wild_do_table);
  197.   if(wild_ignore_table_inited)
  198.     free_string_array(&replicate_wild_ignore_table);
  199. }
  201. inline bool slave_killed(THD* thd)
  202. {
  203.   return abort_slave || abort_loop || thd->killed;
  204. }
  206. inline void skip_load_data_infile(NET* net)
  207. {
  208.   (void)my_net_write(net, "xfb/dev/null", 10);
  209.   (void)net_flush(net);
  210.   (void)my_net_read(net); // discard response
  211.   send_ok(net); // the master expects it
  212. }
  214. inline char* rewrite_db(char* db)
  215. {
  216.   if(replicate_rewrite_db.is_empty() || !db) return db;
  217.   I_List_iterator<i_string_pair> it(replicate_rewrite_db);
  218.   i_string_pair* tmp;
  220.   while((tmp=it++))
  221.     {
  222.       if(!strcmp(tmp->key, db))
  223. return tmp->val;
  224.     }
  226.   return db;
  227. }
  229. int db_ok(const char* db, I_List<i_string> &do_list,
  230.   I_List<i_string> &ignore_list )
  231. {
  232.   if(do_list.is_empty() && ignore_list.is_empty())
  233.     return 1; // ok to replicate if the user puts no constraints
  235.   // if the user has specified restrictions on which databases to replicate
  236.   // and db was not selected, do not replicate
  237.   if(!db)
  238.     return 0;
  240.   if(!do_list.is_empty()) // if the do's are not empty
  241.     {
  242.       I_List_iterator<i_string> it(do_list);
  243.       i_string* tmp;
  245.       while((tmp=it++))
  246. {
  247.   if(!strcmp(tmp->ptr, db))
  248.     return 1; // match
  249. }
  250.       return 0;
  251.     }
  252.   else // there are some elements in the don't, otherwise we cannot get here
  253.     {
  254.       I_List_iterator<i_string> it(ignore_list);
  255.       i_string* tmp;
  257.       while((tmp=it++))
  258. {
  259.   if(!strcmp(tmp->ptr, db))
  260.     return 0; // match
  261. }
  262.       
  263.       return 1;
  264.     }
  265. }
  267. static int init_strvar_from_file(char* var, int max_size, IO_CACHE* f,
  268.        char* default_val)
  269. {
  270.   uint length;
  271.   if ((length=my_b_gets(f,var, max_size)))
  272.   {
  273.     char* last_p = var + length -1;
  274.     if (*last_p == 'n')
  275.       *last_p = 0; // if we stopped on newline, kill it
  276.     else
  277.     {
  278.       // if we truncated a line or stopped on last char, remove all chars
  279.       // up to and including newline
  280.       int c;
  281.       while( ((c=my_b_get(f)) != 'n' && c != my_b_EOF));
  282.     }
  283.     return 0;
  284.   }
  285.   else if (default_val)
  286.   {
  287.     strmake(var,  default_val, max_size);
  288.     return 0;
  289.   }
  290.   return 1;
  291. }
  293. static int init_intvar_from_file(int* var, IO_CACHE* f, int default_val)
  294. {
  295.   char buf[32];
  296.   
  297.   if (my_b_gets(f, buf, sizeof(buf))) 
  298.   {
  299.     *var = atoi(buf);
  300.     return 0;
  301.   }
  302.   else if(default_val)
  303.   {
  304.     *var = default_val;
  305.     return 0;
  306.   }
  307.   return 1;
  308. }
  311. static int create_table_from_dump(THD* thd, NET* net, const char* db,
  312.   const char* table_name)
  313. {
  314.   uint packet_len = my_net_read(net); // read create table statement
  315.   TABLE_LIST tables;
  316.   int error = 0;
  317.   
  318.   if(packet_len == packet_error)
  319.     {
  320.       send_error(&thd->net, ER_MASTER_NET_READ);
  321.       return 1;
  322.     }
  323.   if(net->read_pos[0] == 255) // error from master
  324.     {
  325.       net->read_pos[packet_len] = 0;
  326.       net_printf(&thd->net, ER_MASTER, net->read_pos + 3);
  327.       return 1;
  328.     }
  329.   thd->command = COM_TABLE_DUMP;
  330.   thd->query = sql_alloc(packet_len + 1);
  331.   if(!thd->query)
  332.     {
  333.       sql_print_error("create_table_from_dump: out of memory");
  334.       net_printf(&thd->net, ER_GET_ERRNO, "Out of memory");
  335.       return 1;
  336.     }
  337.   memcpy(thd->query, net->read_pos, packet_len);
  338.   thd->query[packet_len] = 0;
  339.   thd->current_tablenr = 0;
  340.   thd->query_error = 0;
  341.   thd->net.no_send_ok = 1;
  342.   thd->proc_info = "Creating table from master dump";
  343.   // save old db in case we are creating in a different database
  344.   char* save_db = thd->db;
  345.   thd->db = thd->last_nx_db;
  346.   mysql_parse(thd, thd->query, packet_len); // run create table
  347.   thd->db = save_db; // leave things the way the were before
  348.   
  349.   if(thd->query_error)
  350.   {
  351.     close_thread_tables(thd); // mysql_parse takes care of the error send
  352.     return 1;
  353.   }
  355.   bzero((char*) &tables,sizeof(tables));
  356.   tables.db = (char*)db;
  357.   tables.name = tables.real_name = (char*)table_name;
  358.   tables.lock_type = TL_WRITE;
  359.   thd->proc_info = "Opening master dump table";
  360.   if (!open_ltable(thd, &tables, TL_WRITE))
  361.   {
  362.     // open tables will send the error
  363.     sql_print_error("create_table_from_dump: could not open created table");
  364.     close_thread_tables(thd);
  365.     return 1;
  366.   }
  367.   
  368.   handler *file = tables.table->file;
  369.   thd->proc_info = "Reading master dump table data";
  370.   if (file->net_read_dump(net))
  371.   {
  372.     net_printf(&thd->net, ER_MASTER_NET_READ);
  373.     sql_print_error("create_table_from_dump::failed in
  374.  handler::net_read_dump()");
  375.     close_thread_tables(thd);
  376.     return 1;
  377.   }
  379.   HA_CHECK_OPT check_opt;
  380.   check_opt.init();
  381.   check_opt.flags|= T_VERY_SILENT;
  382.   check_opt.quick = 1;
  383.   thd->proc_info = "Rebuilding the index on master dump table";
  384.   Vio* save_vio = thd->net.vio;
  385.   // we do not want repair() to spam us with messages
  386.   // just send them to the error log, and report the failure in case of
  387.   // problems
  388.   thd->net.vio = 0;
  389.   if (file->repair(thd,&check_opt ))
  390.   {
  391.       net_printf(&thd->net, ER_INDEX_REBUILD,tables.table->real_name );
  392.       error = 1;
  393.   }
  394.   thd->net.vio = save_vio;
  395.   close_thread_tables(thd);
  396.   
  397.   thd->net.no_send_ok = 0;
  398.   return error; 
  399. }
  401. int fetch_nx_table(THD* thd, MASTER_INFO* mi)
  402. {
  403.   MYSQL* mysql = mc_mysql_init(NULL);
  404.   int error = 1;
  405.   int nx_errno = 0;
  406.   if(!mysql)
  407.     {
  408.       sql_print_error("fetch_nx_table: Error in mysql_init()");
  409.       nx_errno = ER_GET_ERRNO;
  410.       goto err;
  411.     }
  413.   safe_connect(thd, mysql, mi);
  414.   if(slave_killed(thd))
  415.     goto err;
  417.   if(request_table_dump(mysql, thd->last_nx_db, thd->last_nx_table))
  418.     {
  419.       nx_errno = ER_GET_ERRNO;
  420.       sql_print_error("fetch_nx_table: failed on table dump request ");
  421.       goto err;
  422.     }
  424.   if(create_table_from_dump(thd, &mysql->net, thd->last_nx_db,
  425.     thd->last_nx_table))
  426.     {
  427.       // create_table_from_dump will have sent the error alread
  428.       sql_print_error("fetch_nx_table: failed on create table ");
  429.       goto err;
  430.     }
  431.   
  432.   error = 0;
  434.  err:
  435.   if (mysql)
  436.     mc_mysql_close(mysql);
  437.   if (nx_errno && thd->net.vio)
  438.     send_error(&thd->net, nx_errno, "Error in fetch_nx_table");
  439.   return error;
  440. }
  442. void end_master_info(MASTER_INFO* mi)
  443. {
  444.   if(mi->fd >= 0)
  445.     {
  446.       end_io_cache(&mi->file);
  447.       (void)my_close(mi->fd, MYF(MY_WME));
  448.       mi->fd = -1;
  449.     }
  450.   mi->inited = 0;
  451. }
  453. int init_master_info(MASTER_INFO* mi)
  454. {
  455.   if (mi->inited)
  456.     return 0;
  457.   int fd,length,error;
  458.   MY_STAT stat_area;
  459.   char fname[FN_REFLEN+128];
  460.   const char *msg;
  461.   fn_format(fname, master_info_file, mysql_data_home, "", 4+16+32);
  463.   // we need a mutex while we are changing master info parameters to
  464.   // keep other threads from reading bogus info
  466.   pthread_mutex_lock(&mi->lock);
  467.   mi->pending = 0;
  468.   fd = mi->fd;
  469.   
  470.   // we do not want any messages if the file does not exist
  471.   if (!my_stat(fname, &stat_area, MYF(0)))
  472.   {
  473.     // if someone removed the file from underneath our feet, just close
  474.     // the old descriptor and re-create the old file
  475.     if (fd >= 0)
  476.       my_close(fd, MYF(MY_WME));
  477.     if ((fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0
  478. || init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,0,
  479.  MYF(MY_WME)))
  480.     {
  481.       if(fd >= 0)
  482. my_close(fd, MYF(0));
  483.       pthread_mutex_unlock(&mi->lock);
  484.       return 1;
  485.     }
  486.     mi->log_file_name[0] = 0;
  487.     mi->pos = 4; // skip magic number
  488.     mi->fd = fd;
  489.       
  490.     if (master_host)
  491.       strmake(mi->host, master_host, sizeof(mi->host) - 1);
  492.     if (master_user)
  493.       strmake(mi->user, master_user, sizeof(mi->user) - 1);
  494.     if (master_password)
  495.       strmake(mi->password, master_password, sizeof(mi->password) - 1);
  496.     mi->port = master_port;
  497.     mi->connect_retry = master_connect_retry;
  498.   }
  499.   else // file exists
  500.   {
  501.     if(fd >= 0)
  502.       reinit_io_cache(&mi->file, READ_CACHE, 0L,0,0);
  503.     else if((fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0
  504.     || init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,
  505.      0, MYF(MY_WME)))
  506.     {
  507.       if(fd >= 0)
  508. my_close(fd, MYF(0));
  509.       pthread_mutex_unlock(&mi->lock);
  510.       return 1;
  511.     }
  512.       
  513.     if ((length=my_b_gets(&mi->file, mi->log_file_name,
  514.    sizeof(mi->log_file_name))) < 1)
  515.     {
  516.       msg="Error reading log file name from master info file ";
  517.       goto error;
  518.     }
  520.     mi->log_file_name[length-1]= 0; // kill n
  521.     char buf[FN_REFLEN];
  522.     if(!my_b_gets(&mi->file, buf, sizeof(buf)))
  523.     {
  524.       msg="Error reading log file position from master info file";
  525.       goto error;
  526.     }
  528.     mi->pos = strtoull(buf,(char**) 0, 10);
  529.     mi->fd = fd;
  530.     if(init_strvar_from_file(mi->host, sizeof(mi->host), &mi->file,
  531.      master_host) ||
  532.        init_strvar_from_file(mi->user, sizeof(mi->user), &mi->file,
  533.      master_user) || 
  534.        init_strvar_from_file(mi->password, sizeof(mi->password), &mi->file,
  535.      master_password) ||
  536.        init_intvar_from_file((int*)&mi->port, &mi->file, master_port) ||
  537.        init_intvar_from_file((int*)&mi->connect_retry, &mi->file,
  538.      master_connect_retry))
  539.     {
  540.       msg="Error reading master configuration";
  541.       goto error;
  542.     }
  543.   }
  544.   
  545.   mi->inited = 1;
  546.   // now change the cache from READ to WRITE - must do this
  547.   // before flush_master_info
  548.   reinit_io_cache(&mi->file, WRITE_CACHE, 0L,0,1);
  549.   error=test(flush_master_info(mi));
  550.   pthread_mutex_unlock(&mi->lock);
  551.   return error;
  553. error:
  554.   sql_print_error(msg);
  555.   end_io_cache(&mi->file);
  556.   my_close(fd, MYF(0));
  557.   pthread_mutex_unlock(&mi->lock);
  558.   return 1;
  559. }
  561. int show_master_info(THD* thd)
  562. {
  563.   DBUG_ENTER("show_master_info");
  564.   List<Item> field_list;
  565.   field_list.push_back(new Item_empty_string("Master_Host",
  566.      sizeof(glob_mi.host)));
  567.   field_list.push_back(new Item_empty_string("Master_User",
  568.      sizeof(glob_mi.user)));
  569.   field_list.push_back(new Item_empty_string("Master_Port", 6));
  570.   field_list.push_back(new Item_empty_string("Connect_retry", 6));
  571.   field_list.push_back( new Item_empty_string("Log_File",
  572.      FN_REFLEN));
  573.   field_list.push_back(new Item_empty_string("Pos", 12));
  574.   field_list.push_back(new Item_empty_string("Slave_Running", 3));
  575.   field_list.push_back(new Item_empty_string("Replicate_do_db", 20));
  576.   field_list.push_back(new Item_empty_string("Replicate_ignore_db", 20));
  577.   field_list.push_back(new Item_empty_string("Last_errno", 4));
  578.   field_list.push_back(new Item_empty_string("Last_error", 20));
  579.   field_list.push_back(new Item_empty_string("Skip_counter", 12));
  580.   if(send_fields(thd, field_list, 1))
  581.     DBUG_RETURN(-1);
  583.   String* packet = &thd->packet;
  584.   packet->length(0);
  585.   
  586.   pthread_mutex_lock(&glob_mi.lock);
  587.   net_store_data(packet, glob_mi.host);
  588.   net_store_data(packet, glob_mi.user);
  589.   net_store_data(packet, (uint32) glob_mi.port);
  590.   net_store_data(packet, (uint32) glob_mi.connect_retry);
  591.   net_store_data(packet, glob_mi.log_file_name);
  592.   net_store_data(packet, (uint32) glob_mi.pos); // QQ: Should be fixed
  593.   pthread_mutex_unlock(&glob_mi.lock);
  594.   pthread_mutex_lock(&LOCK_slave);
  595.   net_store_data(packet, slave_running ? "Yes":"No");
  596.   pthread_mutex_unlock(&LOCK_slave);
  597.   net_store_data(packet, &replicate_do_db);
  598.   net_store_data(packet, &replicate_ignore_db);
  599.   net_store_data(packet, (uint32)last_slave_errno);
  600.   net_store_data(packet, last_slave_error);
  601.   net_store_data(packet, slave_skip_counter);
  602.   
  603.   if (my_net_write(&thd->net, (char*)thd->packet.ptr(), packet->length()))
  604.     DBUG_RETURN(-1);
  606.   send_eof(&thd->net);
  607.   DBUG_RETURN(0);
  608. }
  610. int flush_master_info(MASTER_INFO* mi)
  611. {
  612.   IO_CACHE* file = &mi->file;
  613.   char lbuf[22];
  614.   
  615.   my_b_seek(file, 0L);
  616.   my_b_printf(file, "%sn%sn%sn%sn%sn%dn%dn",
  617.       mi->log_file_name, llstr(mi->pos, lbuf), mi->host, mi->user,
  618.       mi->password, mi->port, mi->connect_retry);
  619.   flush_io_cache(file);
  620.   return 0;
  621. }
  623. int st_master_info::wait_for_pos(THD* thd, String* log_name, ulonglong log_pos)
  624. {
  625.   if (!inited) return -1;
  626.   bool pos_reached;
  627.   int event_count = 0;
  628.   pthread_mutex_lock(&lock);
  629.   while(!thd->killed)
  630.   {
  631.     int cmp_result;
  632.     if (*log_file_name)
  633.     {
  634.       /*
  635. We should use dirname_length() here when we have a version of
  636. this that doesn't modify the argument */
  637.       char *basename = strrchr(log_file_name, FN_LIBCHAR);
  638.       if (basename)
  639. ++basename;
  640.       else
  641. basename = log_file_name;
  642.       cmp_result =  strncmp(basename, log_name->ptr(),
  643.     log_name->length());
  644.     }
  645.     else
  646.       cmp_result = 0;
  647.       
  648.     pos_reached = ((!cmp_result && pos >= log_pos) || cmp_result > 0);
  649.     if (pos_reached || thd->killed)
  650.       break;
  651.     
  652.     const char* msg = thd->enter_cond(&cond, &lock,
  653.       "Waiting for master update");
  654.     pthread_cond_wait(&cond, &lock);
  655.     thd->exit_cond(msg);
  656.     event_count++;
  657.   }
  658.   pthread_mutex_unlock(&lock);
  659.   return thd->killed ? -1 : event_count;
  660. }
  663. static int init_slave_thread(THD* thd)
  664. {
  665.   DBUG_ENTER("init_slave_thread");
  666.   thd->system_thread = thd->bootstrap = 1;
  667.   thd->client_capabilities = 0;
  668.   my_net_init(&thd->net, 0);
  669.   thd->max_packet_length=thd->net.max_packet;
  670.   thd->master_access= ~0;
  671.   thd->priv_user = 0;
  672.   thd->slave_thread = 1;
  673.   thd->options = (((opt_log_slave_updates) ? OPTION_BIN_LOG:0) | OPTION_AUTO_IS_NULL) ;
  674.   thd->system_thread = 1;
  675.   thd->client_capabilities = CLIENT_LOCAL_FILES;
  676.   slave_real_id=thd->real_id=pthread_self();
  677.   pthread_mutex_lock(&LOCK_thread_count);
  678.   thd->thread_id = thread_id++;
  679.   pthread_mutex_unlock(&LOCK_thread_count);
  681.   if (init_thr_lock() ||
  682.       my_pthread_setspecific_ptr(THR_THD,  thd) ||
  683.       my_pthread_setspecific_ptr(THR_MALLOC, &thd->mem_root) ||
  684.       my_pthread_setspecific_ptr(THR_NET,  &thd->net))
  685.   {
  686.     close_connection(&thd->net,ER_OUT_OF_RESOURCES); // is this needed?
  687.     end_thread(thd,0);
  688.     DBUG_RETURN(-1);
  689.   }
  691.   thd->mysys_var=my_thread_var;
  692.   thd->dbug_thread_id=my_thread_id();
  693. #ifndef __WIN__
  694.   sigset_t set;
  695.   VOID(sigemptyset(&set)); // Get mask in use
  696.   VOID(pthread_sigmask(SIG_UNBLOCK,&set,&thd->block_signals));
  697. #endif
  699.   thd->mem_root.free=thd->mem_root.used=0; // Probably not needed
  700.   if (thd->max_join_size == (ulong) ~0L)
  701.     thd->options |= OPTION_BIG_SELECTS;
  703.   thd->proc_info="Waiting for master update";
  704.   thd->version=refresh_version;
  705.   thd->set_time();
  707.   DBUG_RETURN(0);
  708. }
  710. static int safe_sleep(THD* thd, int sec)
  711. {
  712.   thr_alarm_t alarmed;
  713.   thr_alarm_init(&alarmed);
  714.   time_t start_time= time((time_t*) 0);
  715.   time_t end_time= start_time+sec;
  716.   ALARM  alarm_buff;
  718.   while (start_time < end_time)
  719.   {
  720.     int nap_time = (int) (end_time - start_time);
  721.     /*
  722.       the only reason we are asking for alarm is so that
  723.       we will be woken up in case of murder, so if we do not get killed,
  724.       set the alarm so it goes off after we wake up naturally
  725.     */
  726.     thr_alarm(&alarmed, 2 * nap_time,&alarm_buff);
  727.     sleep(nap_time);
  728.     // if we wake up before the alarm goes off, hit the button
  729.     // so it will not wake up the wife and kids :-)
  730.     if (thr_alarm_in_use(&alarmed))
  731.       thr_end_alarm(&alarmed);
  732.     
  733.     if (slave_killed(thd))
  734.       return 1;
  735.     start_time=time((time_t*) 0);
  736.   }
  737.   return 0;
  738. }
  741. static int request_dump(MYSQL* mysql, MASTER_INFO* mi)
  742. {
  743.   char buf[FN_REFLEN + 10];
  744.   int len;
  745.   int binlog_flags = 0; // for now
  746.   char* logname = mi->log_file_name;
  747.   int4store(buf, mi->pos);
  748.   int2store(buf + 4, binlog_flags);
  749.   int4store(buf + 6, server_id);
  750.   len = (uint) strlen(logname);
  751.   memcpy(buf + 10, logname,len);
  752.   if (mc_simple_command(mysql, COM_BINLOG_DUMP, buf, len + 10, 1))
  753.   {
  754.     // something went wrong, so we will just reconnect and retry later
  755.     // in the future, we should do a better error analysis, but for
  756.     // now we just fill up the error log :-)
  757.     sql_print_error("Error on COM_BINLOG_DUMP: %s, will retry in %d secs",
  758.     mc_mysql_error(mysql), master_connect_retry);
  759.     return 1;
  760.   }
  762.   return 0;
  763. }
  765. static int request_table_dump(MYSQL* mysql, char* db, char* table)
  766. {
  767.   char buf[1024];
  768.   char * p = buf;
  769.   uint table_len = (uint) strlen(table);
  770.   uint db_len = (uint) strlen(db);
  771.   if(table_len + db_len > sizeof(buf) - 2)
  772.     {
  773.       sql_print_error("request_table_dump: Buffer overrun");
  774.       return 1;
  775.     } 
  776.   
  777.   *p++ = db_len;
  778.   memcpy(p, db, db_len);
  779.   p += db_len;
  780.   *p++ = table_len;
  781.   memcpy(p, table, table_len);
  782.   
  783.   if (mc_simple_command(mysql, COM_TABLE_DUMP, buf, p - buf + table_len, 1))
  784.   {
  785.     sql_print_error("request_table_dump: Error sending the table dump 
  786. command");
  787.     return 1;
  788.   }
  790.   return 0;
  791. }
  794. static uint read_event(MYSQL* mysql, MASTER_INFO *mi)
  795. {
  796.   uint len = packet_error;
  797.   // for convinience lets think we start by
  798.   // being in the interrupted state :-)
  799.   int read_errno = EINTR;
  801.   // my_real_read() will time us out
  802.   // we check if we were told to die, and if not, try reading again
  803. #ifndef DBUG_OFF
  804.   if (disconnect_slave_event_count && !(events_till_disconnect--))
  805.     return packet_error;      
  806. #endif
  807.   
  808.   while (!abort_loop && !abort_slave && len == packet_error &&
  809.  read_errno == EINTR )
  810.   {
  811.     len = mc_net_safe_read(mysql);
  812.     read_errno = errno;
  813.   }
  814.   if (abort_loop || abort_slave)
  815.     return packet_error;
  816.   if (len == packet_error || (int) len < 1)
  817.   {
  818.     sql_print_error("Error reading packet from server: %s (read_errno %d,
  819. server_errno=%d)",
  820.     mc_mysql_error(mysql), read_errno, mc_mysql_errno(mysql));
  821.     return packet_error;
  822.   }
  824.   if (len == 1)
  825.   {
  826.      sql_print_error("Slave: received 0 length packet from server, apparent
  827.  master shutdown: %s (%d)",
  828.      mc_mysql_error(mysql), read_errno);
  829.      return packet_error;
  830.   }
  831.   
  832.   DBUG_PRINT("info",( "len=%u, net->read_pos[4] = %dn",
  833.       len, mysql->net.read_pos[4]));
  834.   return len - 1;   
  835. }
  838. static int exec_event(THD* thd, NET* net, MASTER_INFO* mi, int event_len)
  839. {
  840.   Log_event * ev = Log_event::read_log_event((const char*)net->read_pos + 1,
  841.      event_len);
  842.   char llbuff[22];
  843.   
  844.   if (ev)
  845.   {
  846.     int type_code = ev->get_type_code();
  847.     if (ev->server_id == ::server_id || slave_skip_counter)
  848.     {
  849.       if(type_code == LOAD_EVENT)
  850. skip_load_data_infile(net);
  852.       mi->inc_pos(event_len);
  853.       flush_master_info(mi);
  854.       if(slave_skip_counter)
  855.         --slave_skip_counter;
  856.       delete ev;     
  857.       return 0; // avoid infinite update loops
  858.     }
  859.   
  860.     thd->server_id = ev->server_id; // use the original server id for logging
  861.     thd->set_time(); // time the query
  862.     if(!ev->when)
  863.       ev->when = time(NULL);
  864.     
  865.     switch(type_code) {
  866.     case QUERY_EVENT:
  867.     {
  868.       Query_log_event* qev = (Query_log_event*)ev;
  869.       int q_len = qev->q_len;
  870.       int expected_error,actual_error = 0;
  871.       init_sql_alloc(&thd->mem_root, 8192,0);
  872.       thd->db = rewrite_db((char*)qev->db);
  873.       if (db_ok(thd->db, replicate_do_db, replicate_ignore_db))
  874.       {
  875. thd->query = (char*)qev->query;
  876. thd->set_time((time_t)qev->when);
  877. thd->current_tablenr = 0;
  878. VOID(pthread_mutex_lock(&LOCK_thread_count));
  879. thd->query_id = query_id++;
  880. VOID(pthread_mutex_unlock(&LOCK_thread_count));
  881. thd->last_nx_table = thd->last_nx_db = 0;
  882. thd->query_error = 0; // clear error
  883. thd->net.last_errno = 0;
  884. thd->net.last_error[0] = 0;
  885. thd->slave_proxy_id = qev->thread_id; // for temp tables
  886. mysql_parse(thd, thd->query, q_len);
  887. if ((expected_error = qev->error_code) !=
  888.     (actual_error = thd->net.last_errno) && expected_error)
  889. {
  890.   const char* errmsg = "Slave: did not get the expected error
  891.  running query from master - expected: '%s', got '%s'"; 
  892.   sql_print_error(errmsg, ER(expected_error),
  893.   actual_error ? thd->net.last_error:"no error"
  894.   );
  895.   thd->query_error = 1;
  896. }
  897. else if (expected_error == actual_error)
  898.   {
  899.     thd->query_error = 0;
  900.     *last_slave_error = 0;
  901.     last_slave_errno = 0;
  902.   }
  903.       }
  904.       thd->db = 0; // prevent db from being freed
  905.       thd->query = 0; // just to be sure
  906.       // assume no convert for next query unless set explictly
  907.       thd->convert_set = 0;
  908.       close_thread_tables(thd);
  909.       
  910.       if (thd->query_error || thd->fatal_error)
  911.       {
  912. sql_print_error("Slave:  error running query '%s' ",
  913. qev->query);
  914. last_slave_errno = actual_error ? actual_error : -1;
  915. my_snprintf(last_slave_error, sizeof(last_slave_error),
  916.     "error '%s' on query '%s'",
  917.     actual_error ? thd->net.last_error :
  918.     "unexpected success or fatal error",
  919.     qev->query
  920.     );
  921.         free_root(&thd->mem_root,0);
  922. delete ev;
  923. return 1;
  924.       }
  925.       free_root(&thd->mem_root,0);
  926.       delete ev;
  928.       mi->inc_pos(event_len);
  929.       flush_master_info(mi);
  930.       break;
  931.     }
  932.   
  933.     case LOAD_EVENT:
  934.     {
  935.       Load_log_event* lev = (Load_log_event*)ev;
  936.       init_sql_alloc(&thd->mem_root, 8192,0);
  937.       thd->db = rewrite_db((char*)lev->db);
  938.       thd->query = 0;
  939.       thd->query_error = 0;
  940.     
  941.       if(db_ok(thd->db, replicate_do_db, replicate_ignore_db))
  942.       {
  943. thd->set_time((time_t)lev->when);
  944. thd->current_tablenr = 0;
  945. VOID(pthread_mutex_lock(&LOCK_thread_count));
  946. thd->query_id = query_id++;
  947. VOID(pthread_mutex_unlock(&LOCK_thread_count));
  949. TABLE_LIST tables;
  950. bzero((char*) &tables,sizeof(tables));
  951. tables.db = thd->db;
  952. tables.name = tables.real_name = (char*)lev->table_name;
  953. tables.lock_type = TL_WRITE;
  954. // the table will be opened in mysql_load    
  955.         if(table_rules_on && !tables_ok(thd, &tables))
  956. {
  957.   skip_load_data_infile(net);
  958. }
  959. else
  960. {
  961.   enum enum_duplicates handle_dup = DUP_IGNORE;
  962.   if(lev->sql_ex.opt_flags && REPLACE_FLAG)
  963.     handle_dup = DUP_REPLACE;
  964.   sql_exchange ex((char*)lev->fname, lev->sql_ex.opt_flags &&
  965.   DUMPFILE_FLAG );
  966.   String field_term(&lev->sql_ex.field_term, 1),
  967.     enclosed(&lev->sql_ex.enclosed, 1),
  968.     line_term(&lev->sql_ex.line_term,1),
  969.     escaped(&lev->sql_ex.escaped, 1),
  970.     line_start(&lev->sql_ex.line_start, 1);
  971.     
  972.   ex.field_term = &field_term;
  973.   if(lev->sql_ex.empty_flags & FIELD_TERM_EMPTY)
  974.     ex.field_term->length(0);
  975.     
  976.   ex.enclosed = &enclosed;
  977.   if(lev->sql_ex.empty_flags & ENCLOSED_EMPTY)
  978.     ex.enclosed->length(0);
  980.   ex.line_term = &line_term;
  981.   if(lev->sql_ex.empty_flags & LINE_TERM_EMPTY)
  982.     ex.line_term->length(0);
  984.   ex.line_start = &line_start;
  985.   if(lev->sql_ex.empty_flags & LINE_START_EMPTY)
  986.     ex.line_start->length(0);
  988.   ex.escaped = &escaped;
  989.   if(lev->sql_ex.empty_flags & ESCAPED_EMPTY)
  990.     ex.escaped->length(0);
  992.   ex.opt_enclosed = (lev->sql_ex.opt_flags & OPT_ENCLOSED_FLAG);
  993.   if(lev->sql_ex.empty_flags & FIELD_TERM_EMPTY)
  994.     ex.field_term->length(0);
  995.     
  996.   ex.skip_lines = lev->skip_lines;
  997.     
  999.   List<Item> fields;
  1000.   lev->set_fields(fields);
  1001.   thd->slave_proxy_id = thd->thread_id;
  1002.   thd->net.vio = net->vio;
  1003.   // mysql_load will use thd->net to read the file
  1004.   thd->net.pkt_nr = net->pkt_nr;
  1005.   // make sure the client does get confused
  1006.   // about the packet sequence
  1007.   if(mysql_load(thd, &ex, &tables, fields, handle_dup, 1,
  1008. TL_WRITE))
  1009.     thd->query_error = 1;
  1010.   if(thd->cuted_fields)
  1011.     sql_print_error("Slave: load data infile at position %s in log 
  1012. '%s' produced %d warning(s)", llstr(glob_mi.pos,llbuff), RPL_LOG_NAME,
  1013.     thd->cuted_fields );
  1014.   net->pkt_nr = thd->net.pkt_nr;
  1015. }
  1016.       }
  1017.       else
  1018.       {
  1019. // we will just ask the master to send us /dev/null if we do not
  1020. // want to load the data :-)
  1021. skip_load_data_infile(net);
  1022.       }
  1023.     
  1024.       thd->net.vio = 0; 
  1025.       thd->db = 0;// prevent db from being freed
  1026.       close_thread_tables(thd);
  1027.       if(thd->query_error)
  1028.       {
  1029. int sql_error = thd->net.last_errno;
  1030. if(!sql_error)
  1031.   sql_error = ER_UNKNOWN_ERROR;
  1033. sql_print_error("Slave: Error '%s' running load data infile ",
  1034. ER(sql_error));
  1035. delete ev;
  1036.         free_root(&thd->mem_root,0);
  1037. return 1;
  1038.       }
  1039.       
  1040.       delete ev;
  1041.       free_root(&thd->mem_root,0);
  1042.     
  1043.       if(thd->fatal_error)
  1044.       {
  1045. sql_print_error("Slave: Fatal error running query '%s' ",
  1046. thd->query);
  1047. return 1;
  1048.       }
  1050.       mi->inc_pos(event_len);
  1051.       flush_master_info(mi);
  1052.       break;
  1053.     }
  1055.     case START_EVENT:
  1056.       close_temporary_tables(thd);
  1057.       mi->inc_pos(event_len);
  1058.       flush_master_info(mi);
  1059.       delete ev;
  1060.       break;
  1061.                   
  1062.     case STOP_EVENT:
  1063.       if(mi->pos > 4) // stop event should be ignored after rotate event
  1064. {
  1065.           close_temporary_tables(thd);
  1066.           mi->inc_pos(event_len);
  1067.           flush_master_info(mi);
  1068. }
  1069.       delete ev;
  1070.       break;
  1071.     case ROTATE_EVENT:
  1072.     {
  1073.       Rotate_log_event* rev = (Rotate_log_event*)ev;
  1074.       int ident_len = rev->ident_len;
  1075.       pthread_mutex_lock(&mi->lock);
  1076.       memcpy(mi->log_file_name, rev->new_log_ident,ident_len );
  1077.       mi->log_file_name[ident_len] = 0;
  1078.       mi->pos = 4; // skip magic number
  1079.       pthread_cond_broadcast(&mi->cond);
  1080.       pthread_mutex_unlock(&mi->lock);
  1081.       flush_master_info(mi);
  1082. #ifndef DBUG_OFF
  1083.       if(abort_slave_event_count)
  1084. ++events_till_abort;
  1085. #endif      
  1086.       delete ev;
  1087.       break;
  1088.     }
  1090.     case INTVAR_EVENT:
  1091.     {
  1092.       Intvar_log_event* iev = (Intvar_log_event*)ev;
  1093.       switch(iev->type)
  1094.       {
  1095.       case LAST_INSERT_ID_EVENT:
  1096. thd->last_insert_id_used = 1;
  1097. thd->last_insert_id = iev->val;
  1098. break;
  1099.       case INSERT_ID_EVENT:
  1100. thd->next_insert_id = iev->val;
  1101. break;
  1103.       }
  1104.       mi->inc_pending(event_len);
  1105.       delete ev;
  1106.       break;
  1107.     }
  1108.     }
  1109.   }
  1110.   else
  1111.   {
  1112.     sql_print_error("
  1113. Could not parse log event entry, check the master for binlog corruptionn
  1114. This may also be a network problem, or just a bug in the master or slave code.
  1115. ");
  1116.     return 1;
  1117.   }
  1118.   return 0;   
  1119. }
  1120.       
  1121. // slave thread
  1123. pthread_handler_decl(handle_slave,arg __attribute__((unused)))
  1124. {
  1125. #ifndef DBUG_OFF
  1126.  slave_begin:  
  1127. #endif  
  1128.   THD *thd; // needs to be first for thread_stack
  1129.   MYSQL *mysql = NULL ;
  1130.   char llbuff[22];
  1132.   pthread_mutex_lock(&LOCK_slave);
  1133.   if (!server_id)
  1134.   {
  1135.     pthread_cond_broadcast(&COND_slave_start);
  1136.     pthread_mutex_unlock(&LOCK_slave);
  1137.     sql_print_error("Server id not set, will not start slave");
  1138.     pthread_exit((void*)1);
  1139.   }
  1140.   
  1141.   if(slave_running)
  1142.     {
  1143.       pthread_cond_broadcast(&COND_slave_start);
  1144.       pthread_mutex_unlock(&LOCK_slave);
  1145.       pthread_exit((void*)1);  // safety just in case
  1146.     }
  1147.   slave_running = 1;
  1148.   abort_slave = 0;
  1149. #ifndef DBUG_OFF  
  1150.   events_till_abort = abort_slave_event_count;
  1151. #endif  
  1152.   pthread_cond_broadcast(&COND_slave_start);
  1153.   pthread_mutex_unlock(&LOCK_slave);
  1154.   
  1155.   int error = 1;
  1156.   bool retried_once = 0;
  1157.   ulonglong last_failed_pos = 0;
  1158.   
  1159.   // needs to call my_thread_init(), otherwise we get a coredump in DBUG_ stuff
  1160.   my_thread_init();
  1161.   slave_thd = thd = new THD; // note that contructor of THD uses DBUG_ !
  1162.   thd->set_time();
  1163.   DBUG_ENTER("handle_slave");
  1165.   pthread_detach_this_thread();
  1166.   if (init_slave_thread(thd) || init_master_info(&glob_mi))
  1167.     {
  1168.       sql_print_error("Failed during slave thread initialization");
  1169.       goto err;
  1170.     }
  1171.   thd->thread_stack = (char*)&thd; // remember where our stack is
  1172.   thd->temporary_tables = save_temporary_tables; // restore temp tables
  1173.   threads.append(thd);
  1174.   
  1175.   DBUG_PRINT("info",("master info: log_file_name=%s, position=%s",
  1176.      glob_mi.log_file_name, llstr(glob_mi.pos,llbuff)));
  1178.   
  1179.   if (!(mysql = mc_mysql_init(NULL)))
  1180.   {
  1181.     sql_print_error("Slave thread: error in mc_mysql_init()");
  1182.     goto err;
  1183.   }
  1184.   
  1185.   thd->proc_info = "connecting to master";
  1186. #ifndef DBUG_OFF  
  1187.   sql_print_error("Slave thread initialized");
  1188. #endif
  1189.   // we can get killed during safe_connect
  1190.   if (!safe_connect(thd, mysql, &glob_mi))
  1191.    sql_print_error("Slave: connected to master '%s@%s:%d',
  1192.   replication started in log '%s' at position %s", glob_mi.user,
  1193.    glob_mi.host, glob_mi.port,
  1194.    RPL_LOG_NAME,
  1195.    llstr(glob_mi.pos,llbuff));
  1196.   else
  1197.   {
  1198.     sql_print_error("Slave thread killed while connecting to master");
  1199.     goto err;
  1200.   }
  1201.   
  1202.   while (!slave_killed(thd))
  1203.   {
  1204.       thd->proc_info = "Requesting binlog dump";
  1205.       if(request_dump(mysql, &glob_mi))
  1206. {
  1207.   sql_print_error("Failed on request_dump()");
  1208.   if(slave_killed(thd))
  1209.     {
  1210.       sql_print_error("Slave thread killed while requesting master 
  1211. dump");
  1212.               goto err;
  1213.     }
  1214.   
  1215.   thd->proc_info = "Waiiting to reconnect after a failed dump request";
  1216.   if(mysql->net.vio)
  1217.     vio_close(mysql->net.vio);
  1218.   // first time retry immediately, assuming that we can recover
  1219.   // right away - if first time fails, sleep between re-tries
  1220.   // hopefuly the admin can fix the problem sometime
  1221.   if(retried_once)
  1222.     safe_sleep(thd, glob_mi.connect_retry);
  1223.   else
  1224.     retried_once = 1;
  1225.   
  1226.   if(slave_killed(thd))
  1227.     {
  1228.       sql_print_error("Slave thread killed while retrying master 
  1229. dump");
  1230.       goto err;
  1231.     }
  1233.   thd->proc_info = "Reconnecting after a failed dump request";
  1234.   last_failed_pos=glob_mi.pos;
  1235.           sql_print_error("Slave: failed dump request, reconnecting to 
  1236. try again, log '%s' at postion %s", RPL_LOG_NAME,
  1237.   llstr(last_failed_pos,llbuff));
  1238.   if(safe_reconnect(thd, mysql, &glob_mi) || slave_killed(thd))
  1239.     {
  1240.       sql_print_error("Slave thread killed during or after reconnect");
  1241.       goto err;
  1242.     }
  1244.   continue;
  1245. }
  1248.       while(!slave_killed(thd))
  1249. {
  1250.   thd->proc_info = "Reading master update";
  1251.   uint event_len = read_event(mysql, &glob_mi);
  1252.   if(slave_killed(thd))
  1253.     {
  1254.       sql_print_error("Slave thread killed while reading event");
  1255.       goto err;
  1256.     }
  1257.   
  1258.   if (event_len == packet_error)
  1259.   {
  1260.     thd->proc_info = "Waiting to reconnect after a failed read";
  1261.     if(mysql->net.vio)
  1262.         vio_close(mysql->net.vio);
  1263.     if(retried_once) // punish repeat offender with sleep
  1264.       safe_sleep(thd, glob_mi.connect_retry);
  1265.     else
  1266.       retried_once = 1; 
  1267.     
  1268.     if(slave_killed(thd))
  1269.       {
  1270. sql_print_error("Slave thread killed while waiting to 
  1271. reconnect after a failed read");
  1272.         goto err;
  1273.       }
  1274.     thd->proc_info = "Reconnecting after a failed read";
  1275.     last_failed_pos= glob_mi.pos;
  1276.     sql_print_error("Slave: Failed reading log event, 
  1277. reconnecting to retry, log '%s' position %s", RPL_LOG_NAME,
  1278.     llstr(last_failed_pos, llbuff));
  1279.     if(safe_reconnect(thd, mysql, &glob_mi) || slave_killed(thd))
  1280.       {
  1281. sql_print_error("Slave thread killed during or after a 
  1282. reconnect done to recover from failed read");
  1283.         goto err;
  1284.       }
  1285.     break;
  1286.   }
  1287.   
  1288.   thd->proc_info = "Processing master log event"; 
  1289.   if(exec_event(thd, &mysql->net, &glob_mi, event_len))
  1290.     {
  1291.       sql_print_error("
  1292. Error running query, slave aborted. Fix the problem, and re-start 
  1293. the slave thread with "mysqladmin start-slave". We stopped at log 
  1294. '%s' position %s",
  1295.       RPL_LOG_NAME, llstr(glob_mi.pos, llbuff));
  1296.       goto err;
  1297.       // there was an error running the query
  1298.       // abort the slave thread, when the problem is fixed, the user
  1299.       // should restart the slave with mysqladmin start-slave
  1300.     }
  1301. #ifndef DBUG_OFF
  1302.   if(abort_slave_event_count && !--events_till_abort)
  1303.     {
  1304.       sql_print_error("Slave: debugging abort");
  1305.       goto err;
  1306.     }
  1307. #endif   
  1308.   
  1309.   // successful exec with offset advance,
  1310.   // the slave repents and his sins are forgiven!
  1311.   if(glob_mi.pos > last_failed_pos)
  1312.     {
  1313.      retried_once = 0;
  1314. #ifndef DBUG_OFF
  1315.      stuck_count = 0;
  1316. #endif
  1317.     }
  1318. #ifndef DBUG_OFF
  1319.   else
  1320.     {
  1321.       // show a little mercy, allow slave to read one more event
  1322.       // before cutting him off - otherwise he gets stuck
  1323.       // on Invar events, since they do not advance the offset
  1324.       // immediately
  1325.       if (++stuck_count > 2)
  1326.         events_till_disconnect++;
  1327.     }
  1328. #endif   
  1330. }
  1331.     }
  1333.   error = 0;
  1334.  err:
  1335.   // print the current replication position 
  1336.   sql_print_error("Slave thread exiting, replication stopped in log '%s' at 
  1337. position %s",
  1338.   RPL_LOG_NAME, llstr(glob_mi.pos,llbuff));
  1339.   thd->query = thd->db = 0; // extra safety
  1340.   if(mysql)
  1341.       mc_mysql_close(mysql);
  1342.   thd->proc_info = "Waiting for slave mutex on exit";
  1343.   pthread_mutex_lock(&LOCK_slave);
  1344.   slave_running = 0;
  1345.   abort_slave = 0;
  1346.   save_temporary_tables = thd->temporary_tables;
  1347.   thd->temporary_tables = 0; // remove tempation from destructor to close them
  1348.   pthread_cond_broadcast(&COND_slave_stopped); // tell the world we are done
  1349.   pthread_mutex_unlock(&LOCK_slave);
  1350.   net_end(&thd->net); // destructor will not free it, because we are weird
  1351.   slave_thd = 0;
  1352.   delete thd;
  1353.   my_thread_end();
  1354. #ifndef DBUG_OFF
  1355.   if(abort_slave_event_count && !events_till_abort)
  1356.     goto slave_begin;
  1357. #endif  
  1358.   pthread_exit(0);
  1359.   DBUG_RETURN(0); // Can't return anything here
  1360. }
  1363. /* try to connect until successful or slave killed */
  1365. static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi)
  1366. {
  1367.   int slave_was_killed;
  1368. #ifndef DBUG_OFF
  1369.   events_till_disconnect = disconnect_slave_event_count;
  1370. #endif  
  1371.   while(!(slave_was_killed = slave_killed(thd)) &&
  1372. !mc_mysql_connect(mysql, mi->host, mi->user, mi->password, 0,
  1373.   mi->port, 0, 0))
  1374.   {
  1375.     sql_print_error("Slave thread: error connecting to master:%s(%d),
  1376.  retry in %d sec", mc_mysql_error(mysql), errno, mi->connect_retry);
  1377.     safe_sleep(thd, mi->connect_retry);
  1378.   }
  1379.   
  1380.   if(!slave_was_killed)
  1381.     {
  1382.       mysql_log.write(thd, COM_CONNECT_OUT, "%s@%s:%d",
  1383.   mi->user, mi->host, mi->port);
  1384. #ifdef SIGNAL_WITH_VIO_CLOSE
  1385.       thd->set_active_vio(mysql->net.vio);
  1386. #endif      
  1387.     }
  1388.   
  1389.   return slave_was_killed;
  1390. }
  1392. /* try to connect until successful or slave killed */
  1394. static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi)
  1395. {
  1396.   int slave_was_killed;
  1397.   char llbuff[22];
  1399.  // if we lost connection after reading a state set event
  1400.   // we will be re-reading it, so pending needs to be cleared
  1401.   mi->pending = 0;
  1402. #ifndef DBUG_OFF
  1403.   events_till_disconnect = disconnect_slave_event_count;
  1404. #endif
  1405.   while(!(slave_was_killed = slave_killed(thd)) && mc_mysql_reconnect(mysql))
  1406.   {
  1407.     sql_print_error("Slave thread: error re-connecting to master:
  1408. %s, last_errno=%d, retry in %d sec",
  1409.     mc_mysql_error(mysql), errno, mi->connect_retry);
  1410.      safe_sleep(thd, mi->connect_retry);
  1411.   }
  1413.   if(!slave_was_killed)
  1414.     {
  1415.      sql_print_error("Slave: reconnected to master '%s@%s:%d',
  1416. replication resumed in log '%s' at position %s", glob_mi.user,
  1417.     glob_mi.host, glob_mi.port,
  1418.     RPL_LOG_NAME,
  1419.     llstr(glob_mi.pos,llbuff));
  1420. #ifdef SIGNAL_WITH_VIO_CLOSE
  1421.       thd->set_active_vio(mysql->net.vio);
  1422. #endif      
  1423.     }
  1425.   return slave_was_killed;
  1426. }
  1428. #ifdef __GNUC__
  1429. template class I_List_iterator<i_string>;
  1430. template class I_List_iterator<i_string_pair>;
  1431. #endif