Merge mariadb-10.2.41 into 10.2

4 files changed, 101 insertions, 23 deletions

diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5ada018e540..2eec056ec9d 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1863,6 +1863,7 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
 
   if (needs_thr_lock_abort)
   {
+    bool mutex_released= false;
     mysql_mutex_lock(&in_use->LOCK_thd_data);
     /* If not already dying */
     if (in_use->killed != KILL_CONNECTION_HARD)
@@ -1879,18 +1880,21 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
           thread can see those instances (e.g. see partitioning code).
         */
         if (!thd_table->needs_reopen())
-        {
           signalled|= mysql_lock_abort_for_thread(this, thd_table);
-          if (WSREP(this) && wsrep_thd_is_BF(this, FALSE))
-          {
-            WSREP_DEBUG("remove_table_from_cache: %llu",
-                        (unsigned long long) this->real_id);
-            wsrep_abort_thd((void *)this, (void *)in_use, FALSE);
-          }
-        }
       }
+#ifdef WITH_WSREP
+      if (WSREP(this) && wsrep_thd_is_BF(this, false))
+      {
+        WSREP_DEBUG("notify_shared_lock: BF thread %llu query %s"
+                    " victim %llu query %s",
+                    this->real_id, wsrep_thd_query(this),
+                    in_use->real_id, wsrep_thd_query(in_use));
+        wsrep_abort_thd((void *)this, (void *)in_use, false);
+        mutex_released= true;
+      }
+#endif /* WITH_WSREP */
     }
-    mysql_mutex_unlock(&in_use->LOCK_thd_data);
+    if (!mutex_released) mysql_mutex_unlock(&in_use->LOCK_thd_data);
   }
   DBUG_RETURN(signalled);
 }
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 3e1f248b082..67e2fcd6764 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2017, Oracle and/or its affiliates.
-   Copyright (c) 2008, 2020, MariaDB
+   Copyright (c) 2008, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -9069,6 +9069,18 @@ static
 void sql_kill(THD *thd, longlong id, killed_state state, killed_type type)
 {
   uint error;
+#ifdef WITH_WSREP
+  if (WSREP(thd))
+  {
+    WSREP_DEBUG("sql_kill called");
+    if (thd->wsrep_applier)
+    {
+      WSREP_DEBUG("KILL in applying, bailing out here");
+      return;
+    }
+    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL)
+  }
+#endif /* WITH_WSREP */
   if (!(error= kill_one_thread(thd, id, state, type)))
   {
     if (!thd->killed)
@@ -9078,6 +9090,13 @@ void sql_kill(THD *thd, longlong id, killed_state state, killed_type type)
   }
   else
     my_error(error, MYF(0), id);
+#ifdef WITH_WSREP
+  return;
+ wsrep_error_label:
+  error= (type == KILL_TYPE_QUERY ? ER_KILL_QUERY_DENIED_ERROR :
+                                    ER_KILL_DENIED_ERROR);
+  my_error(error, MYF(0), id);
+#endif /* WITH_WSREP */
 }
 
 
@@ -9086,6 +9105,18 @@ void sql_kill_user(THD *thd, LEX_USER *user, killed_state state)
 {
   uint error;
   ha_rows rows;
+#ifdef WITH_WSREP
+  if (WSREP(thd))
+  {
+    WSREP_DEBUG("sql_kill_user called");
+    if (thd->wsrep_applier)
+    {
+      WSREP_DEBUG("KILL in applying, bailing out here");
+      return;
+    }
+    WSREP_TO_ISOLATION_BEGIN(WSREP_MYSQL_DB, NULL, NULL)
+  }
+#endif /* WITH_WSREP */
   if (!(error= kill_threads_for_user(thd, user, state, &rows)))
     my_ok(thd, rows);
   else
@@ -9096,6 +9127,11 @@ void sql_kill_user(THD *thd, LEX_USER *user, killed_state state)
     */
     my_error(error, MYF(0), user->host.str, user->user.str);
   }
+#ifdef WITH_WSREP
+  return;
+ wsrep_error_label:
+  my_error(ER_CANNOT_USER, MYF(0), user ? user->user.str : "NULL");
+#endif /* WITH_WSREP */
 }
 
 
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index f22d8bf0f5a..4a99f781fdd 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -1,4 +1,4 @@
-/* Copyright 2008-2015 Codership Oy <http://www.codership.com>
+/* Copyright 2008-2021 Codership Oy <http://www.codership.com>
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -835,13 +835,25 @@ void wsrep_thr_init()
   DBUG_VOID_RETURN;
 }
 
+/* This is wrapper for wsrep_break_lock in thr_lock.c */
+static int wsrep_thr_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
+{
+  THD* victim_thd= (THD *) victim_thd_ptr;
+  /* We need to lock THD::LOCK_thd_data to protect victim
+  from concurrent usage or disconnect or delete. */
+  mysql_mutex_lock(&victim_thd->LOCK_thd_data);
+  int res= wsrep_abort_thd(bf_thd_ptr, victim_thd_ptr, signal);
+  return res;
+}
+
+
 void wsrep_init_startup (bool first)
 {
   if (wsrep_init()) unireg_abort(1);
 
   wsrep_thr_lock_init(
      (wsrep_thd_is_brute_force_fun)wsrep_thd_is_BF,
-     (wsrep_abort_thd_fun)wsrep_abort_thd,
+     (wsrep_abort_thd_fun)wsrep_thr_abort_thd,
      wsrep_debug, wsrep_convert_LOCK_to_trx,
      (wsrep_on_fun)wsrep_on);
 
@@ -1694,6 +1706,11 @@ static int wsrep_TOI_begin(THD *thd, char *db_, char *table_,
   case SQLCOM_DROP_TABLE:
     buf_err= wsrep_drop_table_query(thd, &buf, &buf_len);
     break;
+  case SQLCOM_KILL:
+    WSREP_DEBUG("KILL as TOI: %s", thd->query());
+    buf_err= wsrep_to_buf_helper(thd, thd->query(), thd->query_length(),
+                                 &buf, &buf_len);
+    break;
   case SQLCOM_CREATE_ROLE:
     if (sp_process_definer(thd))
     {
@@ -2058,8 +2075,13 @@ bool wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
         ticket->wsrep_report(true);
       }
 
-      mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
-      wsrep_abort_thd((void *) request_thd, (void *) granted_thd, 1);
+      /* This will call wsrep_abort_transaction so we should hold
+      THD::LOCK_thd_data to protect victim from concurrent usage
+      or disconnect or delete. */
+      if (request_thd->wsrep_exec_mode == REPL_RECV)
+	DEBUG_SYNC(request_thd, "wsrep_after_granted_lock");
+
+      wsrep_abort_thd((void *) request_thd, (void *) granted_thd, true);
       ret= false;
     }
   }
@@ -2241,6 +2263,7 @@ error:
 static bool abort_replicated(THD *thd)
 {
   bool ret_code= false;
+  mysql_mutex_lock(&thd->LOCK_thd_data);
   if (thd->wsrep_query_state== QUERY_COMMITTING)
   {
     WSREP_DEBUG("aborting replicated trx: %llu", (ulonglong)(thd->real_id));
@@ -2248,6 +2271,8 @@ static bool abort_replicated(THD *thd)
     (void)wsrep_abort_thd(thd, thd, TRUE);
     ret_code= true;
   }
+  else
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
   return ret_code;
 }
 
@@ -2294,6 +2319,8 @@ static bool have_client_connections()
                        (longlong) tmp->thread_id));
     if (is_client_connection(tmp) && tmp->killed == KILL_CONNECTION)
     {
+      WSREP_DEBUG("Informing thread %lld that it's time to die",
+                  (longlong)tmp->thread_id);
       (void)abort_replicated(tmp);
       return true;
     }
@@ -2378,6 +2405,8 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
   {
     DBUG_PRINT("quit",("Informing thread %lld that it's time to die",
                        (longlong) tmp->thread_id));
+    WSREP_DEBUG("Informing thread %lld that it's time to die",
+                (longlong)tmp->thread_id);
     /* We skip slave threads & scheduler on this first loop through. */
     if (!is_client_connection(tmp))
       continue;
@@ -2394,15 +2423,18 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
       continue;
     }
 
-    /* replicated transactions must be skipped */
+    /* replicated transactions must be skipped and aborted
+    with wsrep_abort_thd. */
     if (abort_replicated(tmp))
       continue;
 
     WSREP_DEBUG("closing connection %lld", (longlong) tmp->thread_id);
 
     /*
-      instead of wsrep_close_thread() we do now  soft kill by THD::awake
-     */
+      instead of wsrep_close_thread() we do now  soft kill by
+      THD::awake(). Here also victim needs to be protected from
+      concurrent usage or disconnect or delete.
+    */
     mysql_mutex_lock(&tmp->LOCK_thd_data);
 
     tmp->awake(KILL_CONNECTION);
@@ -2423,7 +2455,6 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
   I_List_iterator<THD> it2(threads);
   while ((tmp=it2++))
   {
-#ifndef __bsdi__				// Bug in BSDI kernel
     if (is_client_connection(tmp) &&
         !abort_replicated(tmp)    &&
         !is_replaying_connection(tmp) &&
@@ -2432,7 +2463,6 @@ void wsrep_close_client_connections(my_bool wait_to_end, THD *except_caller_thd)
       WSREP_INFO("killing local connection: %lld", (longlong) tmp->thread_id);
       close_connection(tmp,0);
     }
-#endif
   }
 
   DBUG_PRINT("quit",("Waiting for threads to die (count=%u)",thread_count));
@@ -2621,7 +2651,8 @@ extern "C" void wsrep_thd_set_query_state(
 
 void wsrep_thd_set_conflict_state(THD *thd, enum wsrep_conflict_state state)
 {
-  if (WSREP(thd)) thd->wsrep_conflict_state= state;
+  mysql_mutex_assert_owner(&thd->LOCK_thd_data);
+  thd->wsrep_conflict_state= state;
 }
 
 
@@ -2762,6 +2793,9 @@ extern "C" void wsrep_thd_awake(THD *thd, my_bool signal)
 {
   if (signal)
   {
+    /* Here we should hold THD::LOCK_thd_data to
+    protect from concurrent usage. */
+    mysql_mutex_assert_owner(&thd->LOCK_thd_data);
     thd->awake(KILL_QUERY);
   }
   else
diff --git a/sql/wsrep_thd.cc b/sql/wsrep_thd.cc
index ef8c0e132f7..ae797c4c712 100644
--- a/sql/wsrep_thd.cc
+++ b/sql/wsrep_thd.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2013 Codership Oy <info@codership.com>
+/* Copyright (C) 2013-2021 Codership Oy <info@codership.com>
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -804,10 +804,12 @@ my_bool wsrep_thd_is_local(void *thd_ptr, my_bool sync)
 
 int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
 {
-  THD *victim_thd = (THD *) victim_thd_ptr;
-  THD *bf_thd     = (THD *) bf_thd_ptr;
+  THD *victim_thd= (THD *) victim_thd_ptr;
+  THD *bf_thd= (THD *) bf_thd_ptr;
   DBUG_ENTER("wsrep_abort_thd");
 
+  mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
+
   if ( (WSREP(bf_thd) ||
          ( (WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU) &&
            bf_thd->wsrep_exec_mode == TOTAL_ORDER) )                         &&
@@ -821,6 +823,7 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
                   "aborted. Ignoring.",
                   (bf_thd) ? (long long)bf_thd->real_id : 0,
                   (long long)victim_thd->real_id);
+      mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
       DBUG_RETURN(1);
     }
 
@@ -831,6 +834,7 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
   else
   {
     WSREP_DEBUG("wsrep_abort_thd not effective: %p %p", bf_thd, victim_thd);
+    mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
   }
 
   DBUG_RETURN(1);