10.0-base merge

1 files changed, 145 insertions, 48 deletions

diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index 8429f1c7f58..3f79a0cb528 100644
--- a/sql/rpl_gtid.cc
+++ b/sql/rpl_gtid.cc
@@ -203,18 +203,15 @@ rpl_slave_state::truncate_state_table(THD *thd)
 {
   TABLE_LIST tlist;
   int err= 0;
-  TABLE *table;
 
+  tmp_disable_binlog(thd);
   tlist.init_one_table(STRING_WITH_LEN("mysql"),
                        rpl_gtid_slave_state_table_name.str,
                        rpl_gtid_slave_state_table_name.length,
                        NULL, TL_WRITE);
   if (!(err= open_and_lock_tables(thd, &tlist, FALSE, 0)))
   {
-    table= tlist.table;
-    table->no_replicate= 1;
-    table->s->is_gtid_slave_pos= TRUE; // TEMPORARY CODE
-    err= table->file->ha_truncate();
+    err= tlist.table->file->ha_truncate();
 
     if (err)
     {
@@ -231,6 +228,7 @@ rpl_slave_state::truncate_state_table(THD *thd)
     thd->mdl_context.release_transactional_locks();
   }
 
+  reenable_binlog(thd);
   return err;
 }
 
@@ -350,14 +348,14 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
   if ((err= gtid_check_rpl_slave_state_table(table)))
     goto end;
 
-  table->no_replicate= 1;
-  table->s->is_gtid_slave_pos= TRUE; // TEMPORARY CODE
   if (!in_transaction)
   {
     DBUG_PRINT("info", ("resetting OPTION_BEGIN"));
     thd->variables.option_bits&=
-      ~(ulonglong)(OPTION_NOT_AUTOCOMMIT|OPTION_BEGIN);
+      ~(ulonglong)(OPTION_NOT_AUTOCOMMIT|OPTION_BEGIN|OPTION_BIN_LOG);
   }
+  else
+    thd->variables.option_bits&= ~(ulonglong)OPTION_BIN_LOG;
 
   bitmap_set_all(table->write_set);
 
@@ -840,7 +838,7 @@ rpl_binlog_state::rpl_binlog_state()
 
 
 void
-rpl_binlog_state::reset()
+rpl_binlog_state::reset_nolock()
 {
   uint32 i;
 
@@ -849,12 +847,22 @@ rpl_binlog_state::reset()
   my_hash_reset(&hash);
 }
 
+
+void
+rpl_binlog_state::reset()
+{
+  mysql_mutex_lock(&LOCK_binlog_state);
+  reset_nolock();
+  mysql_mutex_unlock(&LOCK_binlog_state);
+}
+
+
 void rpl_binlog_state::free()
 {
   if (initialized)
   {
     initialized= 0;
-    reset();
+    reset_nolock();
     my_hash_free(&hash);
     mysql_mutex_destroy(&LOCK_binlog_state);
   }
@@ -865,14 +873,20 @@ bool
 rpl_binlog_state::load(struct rpl_gtid *list, uint32 count)
 {
   uint32 i;
+  bool res= false;
 
-  reset();
+  mysql_mutex_lock(&LOCK_binlog_state);
+  reset_nolock();
   for (i= 0; i < count; ++i)
   {
-    if (update(&(list[i]), false))
-      return true;
+    if (update_nolock(&(list[i]), false))
+    {
+      res= true;
+      break;
+    }
   }
-  return false;
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
@@ -891,7 +905,7 @@ rpl_binlog_state::~rpl_binlog_state()
   Returns 0 for ok, 1 for error.
 */
 int
-rpl_binlog_state::update(const struct rpl_gtid *gtid, bool strict)
+rpl_binlog_state::update_nolock(const struct rpl_gtid *gtid, bool strict)
 {
   element *elem;
 
@@ -910,7 +924,7 @@ rpl_binlog_state::update(const struct rpl_gtid *gtid, bool strict)
     if (!elem->update_element(gtid))
       return 0;
   }
-  else if (!alloc_element(gtid))
+  else if (!alloc_element_nolock(gtid))
     return 0;
 
   my_error(ER_OUT_OF_RESOURCES, MYF(0));
@@ -918,6 +932,17 @@ rpl_binlog_state::update(const struct rpl_gtid *gtid, bool strict)
 }
 
 
+int
+rpl_binlog_state::update(const struct rpl_gtid *gtid, bool strict)
+{
+  int res;
+  mysql_mutex_lock(&LOCK_binlog_state);
+  res= update_nolock(gtid, strict);
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
+}
+
+
 /*
   Fill in a new GTID, allocating next sequence number, and update state
   accordingly.
@@ -927,25 +952,30 @@ rpl_binlog_state::update_with_next_gtid(uint32 domain_id, uint32 server_id,
                                         rpl_gtid *gtid)
 {
   element *elem;
+  int res= 0;
 
   gtid->domain_id= domain_id;
   gtid->server_id= server_id;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   if ((elem= (element *)my_hash_search(&hash, (const uchar *)(&domain_id), 0)))
   {
     gtid->seq_no= ++elem->seq_no_counter;
     if (!elem->update_element(gtid))
-      return 0;
+      goto end;
   }
   else
   {
     gtid->seq_no= 1;
-    if (!alloc_element(gtid))
-      return 0;
+    if (!alloc_element_nolock(gtid))
+      goto end;
   }
 
   my_error(ER_OUT_OF_RESOURCES, MYF(0));
-  return 1;
+  res= 1;
+end:
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
@@ -991,7 +1021,7 @@ rpl_binlog_state::element::update_element(const rpl_gtid *gtid)
 
 
 int
-rpl_binlog_state::alloc_element(const rpl_gtid *gtid)
+rpl_binlog_state::alloc_element_nolock(const rpl_gtid *gtid)
 {
   element *elem;
   rpl_gtid *lookup_gtid;
@@ -1035,7 +1065,9 @@ rpl_binlog_state::check_strict_sequence(uint32 domain_id, uint32 server_id,
                                         uint64 seq_no)
 {
   element *elem;
+  bool res= 0;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   if ((elem= (element *)my_hash_search(&hash,
                                        (const uchar *)(&domain_id), 0)) &&
       elem->last_gtid && elem->last_gtid->seq_no >= seq_no)
@@ -1043,9 +1075,10 @@ rpl_binlog_state::check_strict_sequence(uint32 domain_id, uint32 server_id,
     my_error(ER_GTID_STRICT_OUT_OF_ORDER, MYF(0), domain_id, server_id, seq_no,
              elem->last_gtid->domain_id, elem->last_gtid->server_id,
              elem->last_gtid->seq_no);
-    return 1;
+    res= 1;
   }
-  return 0;
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
@@ -1060,17 +1093,23 @@ int
 rpl_binlog_state::bump_seq_no_if_needed(uint32 domain_id, uint64 seq_no)
 {
   element *elem;
+  int res;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   if ((elem= (element *)my_hash_search(&hash, (const uchar *)(&domain_id), 0)))
   {
     if (elem->seq_no_counter < seq_no)
       elem->seq_no_counter= seq_no;
-    return 0;
+    res= 0;
+    goto end;
   }
 
   /* We need to allocate a new, empty element to remember the next seq_no. */
   if (!(elem= (element *)my_malloc(sizeof(*elem), MYF(MY_WME))))
-    return 1;
+  {
+    res= 1;
+    goto end;
+  }
 
   elem->domain_id= domain_id;
   my_hash_init(&elem->hash, &my_charset_bin, 32,
@@ -1079,11 +1118,18 @@ rpl_binlog_state::bump_seq_no_if_needed(uint32 domain_id, uint64 seq_no)
   elem->last_gtid= NULL;
   elem->seq_no_counter= seq_no;
   if (0 == my_hash_insert(&hash, (const uchar *)elem))
-    return 0;
+  {
+    res= 0;
+    goto end;
+  }
 
   my_hash_free(&elem->hash);
   my_free(elem);
-  return 1;
+  res= 1;
+
+end:
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
@@ -1099,7 +1145,9 @@ rpl_binlog_state::write_to_iocache(IO_CACHE *dest)
 {
   ulong i, j;
   char buf[21];
+  int res= 0;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   for (i= 0; i < hash.records; ++i)
   {
     size_t res;
@@ -1124,11 +1172,16 @@ rpl_binlog_state::write_to_iocache(IO_CACHE *dest)
       longlong10_to_str(gtid->seq_no, buf, 10);
       res= my_b_printf(dest, "%u-%u-%s\n", gtid->domain_id, gtid->server_id, buf);
       if (res == (size_t) -1)
-        return 1;
+      {
+        res= 1;
+        goto end;
+      }
     }
   }
 
-  return 0;
+end:
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
@@ -1139,26 +1192,31 @@ rpl_binlog_state::read_from_iocache(IO_CACHE *src)
   char buf[10+1+10+1+20+1+1];
   char *p, *end;
   rpl_gtid gtid;
+  int res= 0;
 
-  reset();
+  mysql_mutex_lock(&LOCK_binlog_state);
+  reset_nolock();
   for (;;)
   {
-    size_t res= my_b_gets(src, buf, sizeof(buf));
-    if (!res)
+    size_t len= my_b_gets(src, buf, sizeof(buf));
+    if (!len)
       break;
     p= buf;
-    end= buf + res;
-    if (gtid_parser_helper(&p, end, &gtid))
-      return 1;
-    if (update(&gtid, false))
-      return 1;
+    end= buf + len;
+    if (gtid_parser_helper(&p, end, &gtid) ||
+        update_nolock(&gtid, false))
+    {
+      res= 1;
+      break;
+    }
   }
-  return 0;
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
 rpl_gtid *
-rpl_binlog_state::find(uint32 domain_id, uint32 server_id)
+rpl_binlog_state::find_nolock(uint32 domain_id, uint32 server_id)
 {
   element *elem;
   if (!(elem= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0)))
@@ -1167,14 +1225,28 @@ rpl_binlog_state::find(uint32 domain_id, uint32 server_id)
 }
 
 rpl_gtid *
+rpl_binlog_state::find(uint32 domain_id, uint32 server_id)
+{
+  rpl_gtid *p;
+  mysql_mutex_lock(&LOCK_binlog_state);
+  p= find_nolock(domain_id, server_id);
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return p;
+}
+
+rpl_gtid *
 rpl_binlog_state::find_most_recent(uint32 domain_id)
 {
   element *elem;
+  rpl_gtid *gtid= NULL;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   elem= (element *)my_hash_search(&hash, (const uchar *)&domain_id, 0);
   if (elem && elem->last_gtid)
-    return elem->last_gtid;
-  return NULL;
+    gtid= elem->last_gtid;
+  mysql_mutex_unlock(&LOCK_binlog_state);
+
+  return gtid;
 }
 
 
@@ -1184,8 +1256,10 @@ rpl_binlog_state::count()
   uint32 c= 0;
   uint32 i;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   for (i= 0; i < hash.records; ++i)
     c+= ((element *)my_hash_element(&hash, i))->hash.records;
+  mysql_mutex_unlock(&LOCK_binlog_state);
 
   return c;
 }
@@ -1195,7 +1269,9 @@ int
 rpl_binlog_state::get_gtid_list(rpl_gtid *gtid_list, uint32 list_size)
 {
   uint32 i, j, pos;
+  int res= 0;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   pos= 0;
   for (i= 0; i < hash.records; ++i)
   {
@@ -1218,12 +1294,17 @@ rpl_binlog_state::get_gtid_list(rpl_gtid *gtid_list, uint32 list_size)
         gtid= e->last_gtid;
 
       if (pos >= list_size)
-        return 1;
+      {
+        res= 1;
+        goto end;
+      }
       memcpy(&gtid_list[pos++], gtid, sizeof(*gtid));
     }
   }
 
-  return 0;
+end:
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }
 
 
@@ -1242,12 +1323,17 @@ rpl_binlog_state::get_most_recent_gtid_list(rpl_gtid **list, uint32 *size)
 {
   uint32 i;
   uint32 alloc_size, out_size;
+  int res= 0;
 
+  out_size= 0;
+  mysql_mutex_lock(&LOCK_binlog_state);
   alloc_size= hash.records;
   if (!(*list= (rpl_gtid *)my_malloc(alloc_size * sizeof(rpl_gtid),
                                      MYF(MY_WME))))
-    return 1;
-  out_size= 0;
+  {
+    res= 1;
+    goto end;
+  }
   for (i= 0; i < alloc_size; ++i)
   {
     element *e= (element *)my_hash_element(&hash, i);
@@ -1256,8 +1342,10 @@ rpl_binlog_state::get_most_recent_gtid_list(rpl_gtid **list, uint32 *size)
     memcpy(&((*list)[out_size++]), e->last_gtid, sizeof(rpl_gtid));
   }
 
+end:
+  mysql_mutex_unlock(&LOCK_binlog_state);
   *size= out_size;
-  return 0;
+  return res;
 }
 
 
@@ -1267,6 +1355,7 @@ rpl_binlog_state::append_pos(String *str)
   uint32 i;
   bool first= true;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   for (i= 0; i < hash.records; ++i)
   {
     element *e= (element *)my_hash_element(&hash, i);
@@ -1274,6 +1363,7 @@ rpl_binlog_state::append_pos(String *str)
         rpl_slave_state_tostring_helper(str, e->last_gtid, &first))
       return true;
   }
+  mysql_mutex_unlock(&LOCK_binlog_state);
 
   return false;
 }
@@ -1284,7 +1374,9 @@ rpl_binlog_state::append_state(String *str)
 {
   uint32 i, j;
   bool first= true;
+  bool res= false;
 
+  mysql_mutex_lock(&LOCK_binlog_state);
   for (i= 0; i < hash.records; ++i)
   {
     element *e= (element *)my_hash_element(&hash, i);
@@ -1306,11 +1398,16 @@ rpl_binlog_state::append_state(String *str)
         gtid= e->last_gtid;
 
       if (rpl_slave_state_tostring_helper(str, gtid, &first))
-        return true;
+      {
+        res= true;
+        goto end;
+      }
     }
   }
 
-  return false;
+end:
+  mysql_mutex_unlock(&LOCK_binlog_state);
+  return res;
 }