Fixed bug #51855. Race condition in XA START. If several threads

concurrently execute the statement XA START 'x', then mysqld
server could crash.

sql/sql_class.cc:
  xid_cache_insert: added checking for element in cache before
  insert it, return TRUE if such element already exists.
sql/sql_parse.cc:
  mysql_execute_command modified:
  * sequence of calls to xid_cache_search(..)/xid_cache_insert(...)
  replaced by call to xid_cache_insert(...) in alternative
  'case SQLCOM_XA_START:'
  * added comment to alternative 'case SQLCOM_XA_COMMIT:'.

sql/sql_class.cc

@@ -3365,9 +3365,13 @@ bool xid_cache_insert(XID *xid, enum xa_states xa_state)
 bool xid_cache_insert(XID_STATE *xid_state)
 {
   pthread_mutex_lock(&LOCK_xid_cache);
-  DBUG_ASSERT(hash_search(&xid_cache, xid_state->xid.key(),
-                          xid_state->xid.key_length())==0);
-  my_bool res=my_hash_insert(&xid_cache, (uchar*)xid_state);
+  if (hash_search(&xid_cache, xid_state->xid.key(), xid_state->xid.key_length()))
+  {
+    pthread_mutex_unlock(&LOCK_xid_cache);
+    my_error(ER_XAER_DUPID, MYF(0));
+    return TRUE;
+  }
+  my_bool res= my_hash_insert(&xid_cache, (uchar*)xid_state);
   pthread_mutex_unlock(&LOCK_xid_cache);
   return res;
 }

sql/sql_parse.cc

@@ -4730,7 +4730,7 @@ create_sp_error:
         my_error(ER_XAER_NOTA, MYF(0));
         break;
       }
-      thd->transaction.xid_state.xa_state=XA_ACTIVE;
+      thd->transaction.xid_state.xa_state= XA_ACTIVE;
       my_ok(thd);
       break;
     }
@@ -4750,16 +4750,16 @@ create_sp_error:
       my_error(ER_XAER_OUTSIDE, MYF(0));
       break;
     }
-    if (xid_cache_search(thd->lex->xid))
-    {
-      my_error(ER_XAER_DUPID, MYF(0));
-      break;
-    }
     DBUG_ASSERT(thd->transaction.xid_state.xid.is_null());
-    thd->transaction.xid_state.xa_state=XA_ACTIVE;
+    thd->transaction.xid_state.xa_state= XA_ACTIVE;
     thd->transaction.xid_state.rm_error= 0;
     thd->transaction.xid_state.xid.set(thd->lex->xid);
-    xid_cache_insert(&thd->transaction.xid_state);
+    if (xid_cache_insert(&thd->transaction.xid_state))
+    {
+      thd->transaction.xid_state.xa_state= XA_NOTR;
+      thd->transaction.xid_state.xid.null();
+      break;
+    }
     thd->transaction.all.modified_non_trans_table= FALSE;
     thd->options= ((thd->options & ~(OPTION_KEEP_LOG)) | OPTION_BEGIN);
     thd->server_status|= SERVER_STATUS_IN_TRANS;
@@ -4813,6 +4813,16 @@ create_sp_error:
   case SQLCOM_XA_COMMIT:
     if (!thd->transaction.xid_state.xid.eq(thd->lex->xid))
     {
+      /*
+        xid_state.in_thd is always true beside of xa recovery
+        procedure. Note, that there is no race condition here
+        between xid_cache_search and xid_cache_delete, since we're always
+        deleting our own XID (thd->lex->xid == thd->transaction.xid_state.xid).
+        The only case when thd->lex->xid != thd->transaction.xid_state.xid
+        and xid_state->in_thd == 0 is in ha_recover() functionality,
+        which is called before starting client connections, and thus is
+        always single-threaded.
+      */
       XID_STATE *xs=xid_cache_search(thd->lex->xid);
       if (!xs || xs->in_thd)
         my_error(ER_XAER_NOTA, MYF(0));