References lp:1051808 - merged with lp:codership-mysql 5.5.27 based trunk

patched with: bzr diff lp:codership-mysql/5.5 -r3790..3793
2025-07-30 16:24:05 +03:00 · 2012-09-18 22:49:13 +03:00
parent 076f2f322e
commit 42b69efd5d
5 changed files with 197 additions and 136 deletions
--- a/scripts/wsrep_sst_common.sh
+++ b/scripts/wsrep_sst_common.sh
@ -91,7 +91,7 @@ wsrep_log()
    # echo everything to stderr so that it gets into common error log
    # deliberately made to look different from the rest of the log
    local readonly tst="$(date +%Y%m%d\ %H:%M:%S.%N | cut -b -21)"
-    echo "WSREP_SST: $* ($tst)" >/dev/stderr
+    echo "WSREP_SST: $* ($tst)" >>/dev/stderr
 }

 wsrep_log_error()
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@ -5078,9 +5078,6 @@ restart:
    }
  }
 #ifdef WITH_WSREP
-#define WSREP_TO_ISOLATION_BEGIN(db_, table_, table_list_)                   \
-  if (WSREP(thd) && wsrep_to_isolation_begin(thd, db_, table_, table_list_)) goto err;
-
  if ((thd->lex->sql_command== SQLCOM_INSERT         ||
       thd->lex->sql_command== SQLCOM_INSERT_SELECT  ||
       thd->lex->sql_command== SQLCOM_REPLACE        ||
@ -5094,6 +5091,7 @@ restart:
    {
      WSREP_TO_ISOLATION_BEGIN(NULL, NULL, (*start));
    }
+ error:
 #endif

 err:
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@ -4090,7 +4090,7 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
  if (WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open())
 #else
  if (mysql_bin_log.is_open())
-#endif
+#endif /* WITH_WSREP */
  {
    int errcode= query_error_code(thd, thd->killed == NOT_KILLED);
    result= thd->binlog_query(THD::STMT_QUERY_TYPE,
@ -4100,7 +4100,23 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
                              /* suppress_use */ FALSE,
                              errcode);
  }
+#ifdef WITH_WSREP
+  const CSET_STRING query_save = thd->query_string;
+  thd->set_query_inner((char*)query.ptr(), query.length(), system_charset_info);
+
+  WSREP_TO_ISOLATION_BEGIN((*tables)->s->db.str, (*tables)->s->table_name.str, NULL);
+  WSREP_TO_ISOLATION_END;
+
+  thd_binlog_trx_reset(thd);
+  thd->query_string     = query_save;
+  thd->wsrep_exec_mode  = LOCAL_STATE;
+#endif /* WITH_WSREP */
  return result;
+#ifdef WITH_WSREP
+ error:
+  WSREP_WARN("TO isolation failed: %s", thd->query());
+  return 0;
+#endif
 }

 void select_create::store_values(List<Item> &values)
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@ -110,16 +110,8 @@ static void wsrep_client_rollback(THD *thd);
 extern Format_description_log_event *wsrep_format_desc;
 #define WSREP_MYSQL_DB (char *)"mysql"

-#define WSREP_TO_ISOLATION_BEGIN(db_, table_, table_list_)                   \
-  if (WSREP(thd) && wsrep_to_isolation_begin(thd, db_, table_, table_list_)) goto error;
-
-#define WSREP_TO_ISOLATION_END                                              \
-  if (WSREP(thd) || (thd && thd->wsrep_exec_mode==TOTAL_ORDER))             \
-    wsrep_to_isolation_end(thd);
-
-#else
-#define WSREP_TO_ISOLATION_BEGIN(db_, table_, table_list_)
-#define WSREP_TO_ISOLATION_END 
+static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
+			      Parser_state *parser_state);
 #endif /* WITH_WSREP */
 /**
  @defgroup Runtime_Environment Runtime Environment
@ -905,7 +897,7 @@ bool do_command(THD *thd)
 				     thd->wsrep_retry_query_len);
    }
  }
-  if (thd->wsrep_retry_query)
+  if (thd->wsrep_retry_query && thd->wsrep_conflict_state != REPLAYING)
  {
    my_free(thd->wsrep_retry_query);
    thd->wsrep_retry_query      = NULL;
@ -1009,8 +1001,7 @@ static void wsrep_copy_query(THD *thd)
  thd->wsrep_retry_query_len = thd->query_length();
  thd->wsrep_retry_query     = (char *)my_malloc(
                                 thd->wsrep_retry_query_len + 1, MYF(0));
-  thd->wsrep_retry_command   = thd->command;
-  strcpy(thd->wsrep_retry_query, thd->query());
+  strncpy(thd->wsrep_retry_query, thd->query(), thd->wsrep_retry_query_len);
  thd->wsrep_retry_query[thd->wsrep_retry_query_len] = '\0';
 }
 #endif /* WITH_WSREP */
@ -1044,8 +1035,6 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
  DBUG_PRINT("info", ("command: %d", command));

 #ifdef WITH_WSREP
-  bool is_autocommit= false;
-
  if (WSREP(thd)) {
    if (!thd->in_multi_stmt_transaction_mode())
    {
@ -1058,12 +1047,6 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
    {
      thd->wsrep_conflict_state= NO_CONFLICT;
    }
-
-    is_autocommit= !thd->in_multi_stmt_transaction_mode()     &&
-      thd->wsrep_conflict_state == NO_CONFLICT                &&
-      !thd->wsrep_applier                                     &&
-      wsrep_read_only_option(thd, thd->lex->query_tables);
-
    if (thd->wsrep_conflict_state== MUST_ABORT)
    {
      wsrep_client_rollback(thd);
@ -1237,7 +1220,11 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
    if (parser_state.init(thd, thd->query(), thd->query_length()))
      break;

+#ifdef WITH_WSREP
+    wsrep_mysql_parse(thd, thd->query(), thd->query_length(), &parser_state);
+#else
    mysql_parse(thd, thd->query(), thd->query_length(), &parser_state);
+#endif

    while (!thd->killed && (parser_state.m_lip.found_semicolon != NULL) &&
           ! thd->is_error())
@ -1296,7 +1283,11 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 #endif
      parser_state.reset(beginning_of_next_stmt, length);
      /* TODO: set thd->lex->sql_command to SQLCOM_END here */
+#ifdef WITH_WSREP
+      wsrep_mysql_parse(thd, beginning_of_next_stmt, length, &parser_state);
+#else
      mysql_parse(thd, beginning_of_next_stmt, length, &parser_state);
+#endif
    }

    DBUG_PRINT("info",("query ready"));
@ -1617,116 +1608,9 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
  if (WSREP(thd)) {
    /* wsrep BF abort in query exec phase */
    mysql_mutex_lock(&thd->LOCK_wsrep_thd);
-    if (thd->wsrep_conflict_state == MUST_ABORT) {
-      wsrep_client_rollback(thd);
-
-      WSREP_DEBUG("abort in exec query state, avoiding autocommit");
-    }
-
-    /* checking if BF trx must be replayed */
-    if (thd->wsrep_conflict_state== MUST_REPLAY) {
-      if (thd->wsrep_exec_mode!= REPL_RECV) {
-        if (thd->stmt_da->is_sent) {
-          WSREP_ERROR("replay issue, thd has reported status already");
-        }
-        thd->stmt_da->reset_diagnostics_area();
-
-        thd->wsrep_conflict_state= REPLAYING;
-        mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
-
-        mysql_reset_thd_for_next_command(thd, opt_userstat_running);
-        thd->killed= NOT_KILLED;
-        close_thread_tables(thd);
-        if (thd->locked_tables_mode && thd->lock)
-        {
-	  WSREP_DEBUG("releasing table lock for replaying (%ld)", thd->thread_id);
-          thd->locked_tables_list.unlock_locked_tables(thd);
-	  thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
-        }
-	thd->mdl_context.release_transactional_locks();
-
-        thd_proc_info(thd, "wsrep replaying trx");
-        WSREP_DEBUG("replay trx: %s %lld", 
-                    thd->query() ? thd->query() : "void", 
-                    (long long)thd->wsrep_trx_seqno);
-        struct wsrep_thd_shadow shadow;
-        wsrep_prepare_bf_thd(thd, &shadow);
-        int rcode = wsrep->replay_trx(wsrep,
-                                      &thd->wsrep_trx_handle,
-                                      (void *)thd);
-
-        wsrep_return_from_bf_mode(thd, &shadow);
-        if (thd->wsrep_conflict_state!= REPLAYING)
-          WSREP_WARN("lost replaying mode: %d", thd->wsrep_conflict_state );
-
-        mysql_mutex_lock(&thd->LOCK_wsrep_thd);
-	  
-        switch (rcode) {
-        case WSREP_OK:
-          thd->wsrep_conflict_state= NO_CONFLICT;
-          wsrep->post_commit(wsrep, &thd->wsrep_trx_handle);
-          WSREP_DEBUG("trx_replay successful for: %ld %llu", 
-		      thd->thread_id, (long long)thd->real_id);
-          break;
-        case WSREP_TRX_FAIL:
-          if (thd->stmt_da->is_sent) {
-            WSREP_ERROR("replay failed, thd has reported status");
-          }
-          else
-          {
-            WSREP_DEBUG("replay failed, rolling back");
-            my_error(ER_LOCK_DEADLOCK, MYF(0), "wsrep aborted transaction");
-          }
-          thd->wsrep_conflict_state= ABORTED;
-          wsrep->post_rollback(wsrep, &thd->wsrep_trx_handle);
-          break;
-        default:
-          WSREP_ERROR("trx_replay failed for: %d, query: %s", 
-		      rcode, thd->query() ? thd->query() : "void");
-          /* we're now in inconsistent state, must abort */
-          unireg_abort(1);
-          break;
-        }
-        mysql_mutex_lock(&LOCK_wsrep_replaying);
-        wsrep_replaying--;
-        WSREP_DEBUG("replaying decreased: %d, thd: %lu", 
-                    wsrep_replaying, thd->thread_id);
-        mysql_cond_broadcast(&COND_wsrep_replaying);
-        mysql_mutex_unlock(&LOCK_wsrep_replaying);
-      }
-    }
-    /* setting error code for BF aborted trxs */
-    if (thd->wsrep_conflict_state == ABORTED)
-    {
-      mysql_reset_thd_for_next_command(thd, opt_userstat_running);
-      thd->killed= NOT_KILLED;
-      if (is_autocommit &&
-	  (thd->wsrep_retry_counter < thd->variables.wsrep_retry_autocommit))
-      {
-        WSREP_DEBUG("wsrep retrying AC query: %s", 
-                    (thd->query()) ? thd->query() : "void");
-        thd->wsrep_conflict_state= RETRY_AUTOCOMMIT;
-        thd->wsrep_retry_counter++;            // grow
-        wsrep_copy_query(thd);
-      }
-      else
-      {
-        WSREP_DEBUG("BF aborted, thd: %lu is_AC: %d, retry: %lu - %lu SQL: %s", 
-                    thd->thread_id, is_autocommit, thd->wsrep_retry_counter, 
-		    thd->variables.wsrep_retry_autocommit, thd->query());
-        my_error(ER_LOCK_DEADLOCK, MYF(0), "wsrep aborted transaction");
-        thd->killed= NOT_KILLED;
-        thd->wsrep_conflict_state= NO_CONFLICT;
-        thd->wsrep_retry_counter= 0;             //  reset
-      }
-    }
-    else
-    {
-      set_if_smaller(thd->wsrep_retry_counter, 0); // reset; eventually ok
-    }
    if ((thd->wsrep_conflict_state != REPLAYING) &&
-        (thd->wsrep_conflict_state != RETRY_AUTOCOMMIT)) {
-      
+        (thd->wsrep_conflict_state != RETRY_AUTOCOMMIT))
+    {
      thd->update_server_status();
      thd->protocol->end_statement();
      query_cache_end_of_result(thd);
@ -1734,7 +1618,6 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
    mysql_mutex_unlock(&thd->LOCK_wsrep_thd);

  } else { /* if (WSREP(thd))... */
-
 #endif /* WITH_WSREP */
  DBUG_ASSERT(thd->derived_tables == NULL &&
              (thd->open_tables == NULL ||
@ -6142,6 +6025,155 @@ void mysql_init_multi_delete(LEX *lex)
  lex->query_tables_last= &lex->query_tables;
 }

+#ifdef WITH_WSREP
+static void wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
+                 Parser_state *parser_state)
+{
+  bool is_autocommit= 
+    !thd->in_multi_stmt_transaction_mode()                  &&
+    thd->wsrep_conflict_state == NO_CONFLICT                &&
+    !thd->wsrep_applier                                     &&
+    wsrep_read_only_option(thd, thd->lex->query_tables);
+
+  do
+  {
+    if (thd->wsrep_conflict_state== RETRY_AUTOCOMMIT)
+    {
+      thd->wsrep_conflict_state= NO_CONFLICT;
+    }
+    mysql_parse(thd, rawbuf, length, parser_state);
+
+    if (WSREP(thd)) {
+      /* wsrep BF abort in query exec phase */
+      mysql_mutex_lock(&thd->LOCK_wsrep_thd);
+      if (thd->wsrep_conflict_state == MUST_ABORT) {
+	wsrep_client_rollback(thd);
+
+	WSREP_DEBUG("abort in exec query state, avoiding autocommit");
+      }
+
+      /* checking if BF trx must be replayed */
+      if (thd->wsrep_conflict_state== MUST_REPLAY) {
+	if (thd->wsrep_exec_mode!= REPL_RECV) {
+	  if (thd->stmt_da->is_sent) {
+	    WSREP_ERROR("replay issue, thd has reported status already");
+	  }
+	  thd->stmt_da->reset_diagnostics_area();
+
+	  thd->wsrep_conflict_state= REPLAYING;
+	  mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
+
+	  mysql_reset_thd_for_next_command(thd, opt_userstat_running);
+	  thd->killed= NOT_KILLED;
+	  close_thread_tables(thd);
+	  if (thd->locked_tables_mode && thd->lock)
+	  {
+	    WSREP_DEBUG("releasing table lock for replaying (%ld)", thd->thread_id);
+	    thd->locked_tables_list.unlock_locked_tables(thd);
+	    thd->variables.option_bits&= ~(OPTION_TABLE_LOCK);
+	  }
+	  thd->mdl_context.release_transactional_locks();
+
+	  thd_proc_info(thd, "wsrep replaying trx");
+	  WSREP_DEBUG("replay trx: %s %lld", 
+		      thd->query() ? thd->query() : "void", 
+		      (long long)thd->wsrep_trx_seqno);
+	  struct wsrep_thd_shadow shadow;
+	  wsrep_prepare_bf_thd(thd, &shadow);
+	  int rcode = wsrep->replay_trx(wsrep,
+					&thd->wsrep_trx_handle,
+					(void *)thd);
+
+	  wsrep_return_from_bf_mode(thd, &shadow);
+	  if (thd->wsrep_conflict_state!= REPLAYING)
+	    WSREP_WARN("lost replaying mode: %d", thd->wsrep_conflict_state );
+
+	  mysql_mutex_lock(&thd->LOCK_wsrep_thd);
+	  
+	  switch (rcode) {
+	  case WSREP_OK:
+	    thd->wsrep_conflict_state= NO_CONFLICT;
+	    wsrep->post_commit(wsrep, &thd->wsrep_trx_handle);
+	    WSREP_DEBUG("trx_replay successful for: %ld %llu", 
+			thd->thread_id, (long long)thd->real_id);
+	    break;
+	  case WSREP_TRX_FAIL:
+	    if (thd->stmt_da->is_sent) {
+	      WSREP_ERROR("replay failed, thd has reported status");
+	    }
+	    else
+	    {
+	      WSREP_DEBUG("replay failed, rolling back");
+	      my_error(ER_LOCK_DEADLOCK, MYF(0), "wsrep aborted transaction");
+	    }
+	    thd->wsrep_conflict_state= ABORTED;
+	    wsrep->post_rollback(wsrep, &thd->wsrep_trx_handle);
+	    break;
+	  default:
+	    WSREP_ERROR("trx_replay failed for: %d, query: %s", 
+			rcode, thd->query() ? thd->query() : "void");
+	    /* we're now in inconsistent state, must abort */
+	    unireg_abort(1);
+	    break;
+	  }
+	  mysql_mutex_lock(&LOCK_wsrep_replaying);
+	  wsrep_replaying--;
+	  WSREP_DEBUG("replaying decreased: %d, thd: %lu", 
+		      wsrep_replaying, thd->thread_id);
+	  mysql_cond_broadcast(&COND_wsrep_replaying);
+	  mysql_mutex_unlock(&LOCK_wsrep_replaying);
+        }
+      }
+      /* setting error code for BF aborted trxs */
+      if (thd->wsrep_conflict_state == ABORTED)
+      {
+        mysql_reset_thd_for_next_command(thd, opt_userstat_running);
+        thd->killed= NOT_KILLED;
+        if (is_autocommit &&
+           (thd->wsrep_retry_counter < thd->variables.wsrep_retry_autocommit))
+        {
+          WSREP_DEBUG("wsrep retrying AC query: %s", 
+                      (thd->query()) ? thd->query() : "void");
+
+
+	  close_thread_tables(thd);
+
+          thd->wsrep_conflict_state= RETRY_AUTOCOMMIT;
+          thd->wsrep_retry_counter++;            // grow
+	  wsrep_copy_query(thd);
+          thd->set_time();
+          parser_state->reset(rawbuf, length);
+	}
+	else
+	{
+	  WSREP_DEBUG("BF aborted, thd: %lu is_AC: %d, retry: %lu - %lu SQL: %s", 
+		      thd->thread_id, is_autocommit, thd->wsrep_retry_counter, 
+		      thd->variables.wsrep_retry_autocommit, thd->query());
+	  my_error(ER_LOCK_DEADLOCK, MYF(0), "wsrep aborted transaction");
+	  thd->killed= NOT_KILLED;
+	  thd->wsrep_conflict_state= NO_CONFLICT;
+          if (thd->wsrep_conflict_state != REPLAYING)
+	    thd->wsrep_retry_counter= 0;             //  reset
+	}
+      }
+      else
+      {
+	set_if_smaller(thd->wsrep_retry_counter, 0); // reset; eventually ok
+      }
+      mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
+    }
+  }  while (thd->wsrep_conflict_state== RETRY_AUTOCOMMIT);
+
+  if (thd->wsrep_retry_query)
+  {
+    WSREP_DEBUG("releasing retry_query: %s", thd->wsrep_retry_query);
+    my_free(thd->wsrep_retry_query);
+    thd->wsrep_retry_query      = NULL;
+    thd->wsrep_retry_query_len  = 0;
+    thd->wsrep_retry_command    = COM_CONNECT;
+  }
+}
+#endif /* WITH_WSREP */

 /*
  When you modify mysql_parse(), you may need to mofify
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@ -202,6 +202,21 @@ inline bool is_supported_parser_charset(CHARSET_INFO *cs)
 {
  return test(cs->mbminlen == 1);
 }
+#ifdef WITH_WSREP
+
+#define WSREP_TO_ISOLATION_BEGIN(db_, table_, table_list_)                   \
+  if (WSREP(thd) && wsrep_to_isolation_begin(thd, db_, table_, table_list_)) goto error;
+
+#define WSREP_TO_ISOLATION_END                                              \
+  if (WSREP(thd) || (thd && thd->wsrep_exec_mode==TOTAL_ORDER))             \
+    wsrep_to_isolation_end(thd);
+
+#else
+
+#define WSREP_TO_ISOLATION_BEGIN(db_, table_, table_list_)
+#define WSREP_TO_ISOLATION_END 
+
+#endif /* WITH_WSREP */


 #endif /* SQL_PARSE_INCLUDED */