refs codership/mysql-wsrep#90 - fixed race in OSU

Make sure that thd uses the same method to begin and end
OSU operation.

sql/handler.cc

@@ -6233,7 +6233,7 @@ int ha_wsrep_abort_transaction(THD *bf_thd, THD *victim_thd, my_bool signal)
 {
   DBUG_ENTER("ha_wsrep_abort_transaction");
   if (!WSREP(bf_thd) &&  
-      !(wsrep_OSU_method_options == WSREP_OSU_RSU &&
+      !(bf_thd->wsrep_OSU_method == WSREP_OSU_RSU &&
         bf_thd->wsrep_exec_mode  == TOTAL_ORDER)) {
     DBUG_RETURN(0);
   }

sql/sql_class.cc

@@ -1203,6 +1203,7 @@ THD::THD()
   wsrep_consistency_check = NO_CONSISTENCY_CHECK;
   wsrep_status_vars       = 0;
   wsrep_mysql_replicated  = 0;
+  wsrep_OSU_method        = WSREP_OSU_NONE;
   wsrep_TOI_pre_query     = NULL;
   wsrep_TOI_pre_query_len = 0;
 #endif
@@ -1609,7 +1610,7 @@ void THD::init(void)
   wsrep_PA_safe= true;
   wsrep_consistency_check = NO_CONSISTENCY_CHECK;
   wsrep_mysql_replicated  = 0;
-
+  wsrep_OSU_method        = WSREP_OSU_NONE;
   wsrep_TOI_pre_query     = NULL;
   wsrep_TOI_pre_query_len = 0;
 

sql/sql_class.h

@@ -2792,6 +2792,7 @@ class THD :public Statement,
                             wsrep_consistency_check;
   wsrep_stats_var*          wsrep_status_vars;
   int                       wsrep_mysql_replicated;
+  ulong                     wsrep_OSU_method;
   const char*               wsrep_TOI_pre_query; /* a query to apply before
                                                     the actual TOI query */
   size_t                    wsrep_TOI_pre_query_len;

sql/wsrep_mysqld.cc

@@ -1392,6 +1392,12 @@ int wsrep_to_isolation_begin(THD *thd, char *db_, char *table_,
                 thd->query(), thd->thread_id);
   }
 
+  /*
+    Save current global value into thd to persist the used method
+    even if global wsrep_OSU_method_options changes during isolation
+   */
+  thd->wsrep_OSU_method= wsrep_OSU_method_options;
+
   /*
     It makes sense to set auto_increment_* to defaults in TOI operations.
     Must be done before wsrep_TOI_begin() since Query_log_event encapsulating
@@ -1407,10 +1413,14 @@ int wsrep_to_isolation_begin(THD *thd, char *db_, char *table_,
 
   if (thd->variables.wsrep_on && thd->wsrep_exec_mode==LOCAL_STATE)
   {
-    switch (wsrep_OSU_method_options) {
+    switch (thd->wsrep_OSU_method) {
     case WSREP_OSU_TOI: ret =  wsrep_TOI_begin(thd, db_, table_,
                                                table_list); break;
     case WSREP_OSU_RSU: ret =  wsrep_RSU_begin(thd, db_, table_); break;
+    default:
+      WSREP_ERROR("Unsupported OSU method: %lu", thd->wsrep_OSU_method);
+      ret= -1;
+      break;
     }
     switch (ret) {
     case 0:  thd->wsrep_exec_mode= TOTAL_ORDER; break;
@@ -1430,13 +1440,18 @@ void wsrep_to_isolation_end(THD *thd)
 {
   if (thd->wsrep_exec_mode == TOTAL_ORDER)
   {
-    switch(wsrep_OSU_method_options)
+    switch(thd->wsrep_OSU_method)
     {
     case WSREP_OSU_TOI: wsrep_TOI_end(thd); break;
     case WSREP_OSU_RSU: wsrep_RSU_end(thd); break;
+    default:
+      WSREP_WARN("Unsupported wsrep OSU method at isolation end: %lu",
+                 thd->wsrep_OSU_method);
+      break;
     }
     wsrep_cleanup_transaction(thd);
   }
+  thd->wsrep_OSU_method= WSREP_OSU_NONE;
 }
 
 #define WSREP_MDL_LOG(severity, msg, req, gra)	                               \

sql/wsrep_mysqld.h

@@ -109,7 +109,12 @@ extern my_bool     wsrep_slave_FK_checks;
 extern my_bool     wsrep_slave_UK_checks;
 extern bool        wsrep_new_cluster;           // bootstrap the cluster ?
 
-enum enum_wsrep_OSU_method { WSREP_OSU_TOI, WSREP_OSU_RSU };
+enum enum_wsrep_OSU_method {
+    WSREP_OSU_TOI,
+    WSREP_OSU_RSU,
+    WSREP_OSU_NONE,
+};
+
 enum enum_wsrep_sync_wait {
     WSREP_SYNC_WAIT_NONE = 0x0,
     // show, select, begin

sql/wsrep_thd.cc

@@ -575,7 +575,7 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
   DBUG_ENTER("wsrep_abort_thd");
 
   if ( (WSREP(bf_thd) ||
-         ( (WSREP_ON || wsrep_OSU_method_options == WSREP_OSU_RSU) &&
+         ( (WSREP_ON || bf_thd->wsrep_OSU_method == WSREP_OSU_RSU) &&
            bf_thd->wsrep_exec_mode == TOTAL_ORDER) )               &&
        victim_thd)
   {