mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2026-05-02 21:25:36 +02:00
Code Issues Projects Releases Packages Wiki Activity

MDEV-29293 MariaDB stuck on starting commit state

Browse source 
This is a backport from 10.5.

The problem seems to be a deadlock between KILL command execution
and BF abort issued by an applier, where:
* KILL has locked victim's LOCK_thd_kill and LOCK_thd_data.
* Applier has innodb side global lock mutex and victim trx mutex.
* KILL is calling innobase_kill_query, and is blocked by innodb
  global lock mutex.
* Applier is in wsrep_innobase_kill_one_trx and is blocked by
  victim's LOCK_thd_kill.

The fix in this commit removes the TOI replication of KILL command
and makes KILL execution less intrusive operation. Aborting the
victim happens now by using awake_no_mutex() and ha_abort_transaction().
If the KILL happens when the transaction is committing, the
KILL operation is postponed to happen after the statement
has completed in order to avoid KILL to interrupt commit
processing.

Notable changes in this commit:
* wsrep client connections's error state may remain sticky after
  client connection is closed. This error message will then pop
  up for the next client session issuing first SQL statement.
  This problem raised with test galera.galera_bf_kill.
  The fix is to reset wsrep client error state, before a THD is
  reused for next connetion.
* Release THD locks in wsrep_abort_transaction when locking
  innodb mutexes. This guarantees same locking order as with applier
  BF aborting.
* BF abort from MDL was changed to do BF abort on server/wsrep-lib
  side first, and only then do the BF abort on InnoDB side. This
  removes the need to call back from InnoDB for BF aborts which originate
  from MDL and simplifies the locking.
* Removed wsrep_thd_set_wsrep_aborter() from service_wsrep.h.
  The manipulation of the wsrep_aborter can be done solely on
  server side. Moreover, it is now debug only variable and
  could be excluded from optimized builds.
* Remove LOCK_thd_kill from wsrep_thd_LOCK/UNLOCK to allow more
  fine grained locking for SR BF abort which may require locking
  of victim LOCK_thd_kill. Added explicit call for
  wsrep_thd_kill_LOCK/UNLOCK where appropriate.
* Wsrep-lib was updated to version which allows external
  locking for BF abort calls.

Changes to MTR tests:
* Disable galera_bf_abort_group_commit. This test is going to
  be removed (MDEV-30855).
* Record galera_gcache_recover_manytrx as result file was incomplete.
  Trivial change.
* Make galera_create_table_as_select more deterministic:
  Wait until CTAS execution has reached MDL wait for multi-master
  conflict case. Expected error from multi-master conflict is
  ER_QUERY_INTERRUPTED. This is because CTAS does not yet have open
  wsrep transaction when it is waiting for MDL, query gets interrupted
  instead of BF aborted. This should be addressed in separate task.
* A new test galera_kill_group_commit to verify correct behavior
  when KILL is executed while the transaction is committing.

Co-authored-by: Seppo Jaakola <seppo.jaakola@iki.fi>
Co-authored-by: Jan Lindström <jan.lindstrom@galeracluster.com>
Signed-off-by: Julius Goryavsky <julius.goryavsky@mariadb.com>
This commit is contained in:
Teemu Ollakka 2023-04-17 16:04:01 +03:00 • committed by Julius Goryavsky
commit 6966d7fe4b
25 changed files with 573 additions and 245 deletions
Download patch file Download diff file Expand all files Collapse all files

7
sql/handler.cc
View file

@ -7170,6 +7170,9 @@ Compare_keys handler::compare_key_parts(const Field &old_field,
concurrent accesses. And it's an overkill to take LOCK_plugin and
iterate the whole installed_htons[] array every time.
@note Object victim_thd is not guaranteed to exist after this
function returns.
@param bf_thd brute force THD asking for the abort
@param victim_thd victim THD to be aborted
@ -7183,6 +7186,8 @@ int ha_abort_transaction(THD *bf_thd, THD *victim_thd, my_bool signal)
if (!WSREP(bf_thd) &&
!(bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU &&
wsrep_thd_is_toi(bf_thd))) {
mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
DBUG_RETURN(0);
}
@ -7194,6 +7199,8 @@ int ha_abort_transaction(THD *bf_thd, THD *victim_thd, my_bool signal)
else
{
WSREP_WARN("Cannot abort InnoDB transaction");
mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
}
DBUG_RETURN(0);

38
sql/service_wsrep.cc
View file

@ -29,14 +29,17 @@ extern "C" my_bool wsrep_on(const THD *thd)
extern "C" void wsrep_thd_LOCK(const THD *thd)
{
mysql_mutex_lock(&thd->LOCK_thd_kill);
mysql_mutex_lock(&thd->LOCK_thd_data);
}
extern "C" int wsrep_thd_TRYLOCK(const THD *thd)
{
return mysql_mutex_trylock(&thd->LOCK_thd_data);
}
extern "C" void wsrep_thd_UNLOCK(const THD *thd)
{
mysql_mutex_unlock(&thd->LOCK_thd_data);
mysql_mutex_unlock(&thd->LOCK_thd_kill);
}
extern "C" void wsrep_thd_kill_LOCK(const THD *thd)
@ -249,21 +252,13 @@ extern "C" my_bool wsrep_thd_bf_abort(THD *bf_thd, THD *victim_thd,
if ((ret || !wsrep_on(victim_thd)) && signal)
{
if (victim_thd->wsrep_aborter && victim_thd->wsrep_aborter != bf_thd->thread_id)
{
WSREP_DEBUG("victim is killed already by %llu, skipping awake",
victim_thd->wsrep_aborter);
wsrep_thd_UNLOCK(victim_thd);
return false;
}
victim_thd->wsrep_aborter= bf_thd->thread_id;
victim_thd->awake_no_mutex(KILL_QUERY);
}
else
WSREP_DEBUG("wsrep_thd_bf_abort skipped awake for %llu", thd_get_thread_id(victim_thd));
WSREP_DEBUG("wsrep_thd_bf_abort skipped awake for %llu",
thd_get_thread_id(victim_thd));
wsrep_thd_UNLOCK(victim_thd);
return ret;
}
@ -349,25 +344,6 @@ extern "C" void wsrep_commit_ordered(THD *thd)
}
}
extern "C" bool wsrep_thd_set_wsrep_aborter(THD *bf_thd, THD *victim_thd)
{
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
if (!bf_thd)
{
victim_thd->wsrep_aborter= 0;
WSREP_DEBUG("wsrep_thd_set_wsrep_aborter resetting wsrep_aborter");
return false;
}
if (victim_thd->wsrep_aborter && victim_thd->wsrep_aborter != bf_thd->thread_id)
{
return true;
}
victim_thd->wsrep_aborter= bf_thd->thread_id;
WSREP_DEBUG("wsrep_thd_set_wsrep_aborter setting wsrep_aborter %u",
victim_thd->wsrep_aborter);
return false;
}
extern "C" void wsrep_report_bf_lock_wait(const THD *thd,
unsigned long long trx_id)
{

27
sql/sql_class.cc
View file

@ -1279,6 +1279,11 @@ void THD::init()
m_wsrep_next_trx_id = WSREP_UNDEFINED_TRX_ID;
wsrep_replicate_GTID = false;
wsrep_aborter = 0;
wsrep_abort_by_kill = NOT_KILLED;
wsrep_abort_by_kill_err = 0;
#ifndef DBUG_OFF
wsrep_killed_state = 0;
#endif /* DBUG_OFF */
wsrep_desynced_backup_stage= false;
#endif /* WITH_WSREP */
@ -1637,6 +1642,13 @@ void THD::reset_for_reuse()
#endif
#ifdef WITH_WSREP
wsrep_free_status(this);
wsrep_cs().reset_error();
wsrep_aborter= 0;
wsrep_abort_by_kill= NOT_KILLED;
wsrep_abort_by_kill_err= 0;
#ifndef DBUG_OFF
wsrep_killed_state= 0;
#endif /* DBUG_OFF */
#endif /* WITH_WSREP */
}
@ -1886,7 +1898,9 @@ void THD::awake_no_mutex(killed_state state_to_set)
}
/* Interrupt target waiting inside a storage engine. */
if (state_to_set != NOT_KILLED && !wsrep_is_bf_aborted(this))
if (state_to_set != NOT_KILLED &&
IF_WSREP(!wsrep_is_bf_aborted(this) && wsrep_abort_by_kill == NOT_KILLED,
true))
ha_kill_query(this, thd_kill_level(this));
abort_current_cond_wait(false);
@ -2128,6 +2142,17 @@ void THD::reset_killed()
mysql_mutex_unlock(&LOCK_thd_kill);
}
#ifdef WITH_WSREP
if (WSREP_NNULL(this))
{
if (wsrep_abort_by_kill != NOT_KILLED)
{
mysql_mutex_assert_not_owner(&LOCK_thd_kill);
mysql_mutex_lock(&LOCK_thd_kill);
wsrep_abort_by_kill= NOT_KILLED;
wsrep_abort_by_kill_err= 0;
mysql_mutex_unlock(&LOCK_thd_kill);
}
}
mysql_mutex_assert_not_owner(&LOCK_thd_data);
mysql_mutex_lock(&LOCK_thd_data);
wsrep_aborter= 0;

9
sql/sql_class.h
View file

@ -4982,7 +4982,14 @@ public:
bool wsrep_ignore_table;
/* thread who has started kill for this THD protected by LOCK_thd_data*/
my_thread_id wsrep_aborter;
/* Kill signal used, if thread was killed by manual KILL. Protected by
LOCK_thd_kill. */
std::atomic<killed_state> wsrep_abort_by_kill;
/* */
struct err_info* wsrep_abort_by_kill_err;
#ifndef DBUG_OFF
int wsrep_killed_state;
#endif /* DBUG_OFF */
/* true if BF abort is observed in do_command() right after reading
client's packet, and if the client has sent PS execute command. */
bool wsrep_delayed_BF_abort;

58
sql/sql_parse.cc
View file

@ -7848,7 +7848,7 @@ static bool wsrep_mysql_parse(THD *thd, char *rawbuf, uint length,
thd->wsrep_retry_counter < thd->variables.wsrep_retry_autocommit)
{
#ifdef ENABLED_DEBUG_SYNC
DBUG_EXECUTE_IF("sync.wsrep_retry_autocommit",
DBUG_EXECUTE_IF("sync.wsrep_retry_autocommit",
{
const char act[]=
"now "
@ -9212,21 +9212,15 @@ kill_one_thread(THD *thd, my_thread_id id, killed_state kill_signal, killed_type
thd->security_ctx->user_matches(tmp->security_ctx))
#endif /* WITH_WSREP */
{
{
#ifdef WITH_WSREP
DEBUG_SYNC(thd, "before_awake_no_mutex");
if (tmp->wsrep_aborter && tmp->wsrep_aborter != thd->thread_id)
{
/* victim is in hit list already, bail out */
WSREP_DEBUG("victim %lld has wsrep aborter: %lu, skipping awake()",
id, tmp->wsrep_aborter);
error= 0;
}
else
if (WSREP(tmp))
{
/* Object tmp is not guaranteed to exist after wsrep_kill_thd()
returns, so do early return from this function. */
DBUG_RETURN(wsrep_kill_thd(thd, tmp, kill_signal));
}
#endif /* WITH_WSREP */
{
WSREP_DEBUG("kill_one_thread victim: %lld wsrep_aborter %lu"
" by signal %d",
id, tmp->wsrep_aborter, kill_signal);
tmp->awake_no_mutex(kill_signal);
error= 0;
}
@ -9350,18 +9344,6 @@ static
void sql_kill(THD *thd, my_thread_id 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 (likely(!(error= kill_one_thread(thd, id, state, type))))
{
if (!thd->killed)
@ -9371,13 +9353,6 @@ void sql_kill(THD *thd, my_thread_id 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), (long long) id);
#endif /* WITH_WSREP */
}
@ -9386,18 +9361,6 @@ 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 */
switch (error= kill_threads_for_user(thd, user, state, &rows))
{
case 0:
@ -9413,11 +9376,6 @@ void sql_kill_user(THD *thd, LEX_USER *user, killed_state state)
default:
my_error(error, MYF(0));
}
#ifdef WITH_WSREP
return;
wsrep_error_label:
my_error(ER_CANNOT_USER, MYF(0), user ? user->user.str : "NULL");
#endif /* WITH_WSREP */
}

2
sql/sql_plugin_services.inl
View file

@ -150,6 +150,7 @@ static struct wsrep_service_st wsrep_handler = {
wsrep_on,
wsrep_prepare_key_for_innodb,
wsrep_thd_LOCK,
wsrep_thd_TRYLOCK,
wsrep_thd_UNLOCK,
wsrep_thd_query,
wsrep_thd_retry_counter,
@ -173,7 +174,6 @@ static struct wsrep_service_st wsrep_handler = {
wsrep_get_debug,
wsrep_commit_ordered,
wsrep_thd_is_applying,
wsrep_thd_set_wsrep_aborter,
wsrep_report_bf_lock_wait,
wsrep_thd_kill_LOCK,
wsrep_thd_kill_UNLOCK,

8
sql/wsrep_dummy.cc
View file

@ -56,6 +56,11 @@ my_bool wsrep_on(const THD *)
void wsrep_thd_LOCK(const THD *)
{ }
int wsrep_thd_TRYLOCK(const THD *)
{
return 0;
}
void wsrep_thd_UNLOCK(const THD *)
{ }
@ -141,9 +146,6 @@ void wsrep_log(void (*)(const char *, ...), const char *, ...)
my_bool wsrep_thd_is_applying(const THD*)
{ return 0;}
bool wsrep_thd_set_wsrep_aborter(THD*, THD*)
{ return 0;}
void wsrep_report_bf_lock_wait(const THD*,
unsigned long long)
{}

1
sql/wsrep_high_priority_service.cc
View file

@ -480,6 +480,7 @@ int Wsrep_high_priority_service::log_dummy_write_set(const wsrep::ws_handle& ws_
m_thd->wait_for_prior_commit();
}
WSREP_DEBUG("checkpointing dummy write set %lld", ws_meta.seqno().get());
wsrep_set_SE_checkpoint(ws_meta.gtid());
if (!WSREP_EMULATE_BINLOG(m_thd))

45
sql/wsrep_mysqld.cc
View file

@ -2053,11 +2053,6 @@ static int wsrep_TOI_event_buf(THD* thd, uchar** buf, size_t* buf_len)
case SQLCOM_DROP_TABLE:
err= wsrep_drop_table_query(thd, buf, buf_len);
break;
case SQLCOM_KILL:
WSREP_DEBUG("KILL as TOI: %s", thd->query());
err= wsrep_to_buf_helper(thd, thd->query(), thd->query_length(),
buf, buf_len);
break;
case SQLCOM_CREATE_ROLE:
if (sp_process_definer(thd))
{
@ -2440,8 +2435,15 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
/* Here we will call wsrep_abort_transaction so we should hold
THD::LOCK_thd_data to protect victim from concurrent usage
and THD::LOCK_thd_kill to protect from disconnect or delete. */
wsrep_thd_LOCK(granted_thd);
and THD::LOCK_thd_kill to protect from disconnect or delete.
Note that all calls to wsrep_abort_thd() and ha_abort_transaction()
unlock LOCK_thd_kill for granted_thd, so granted_thd must not be
accessed after any of those calls. Moreover all other if branches
must release those locks.
*/
mysql_mutex_lock(&granted_thd->LOCK_thd_kill);
mysql_mutex_lock(&granted_thd->LOCK_thd_data);
if (wsrep_thd_is_toi(granted_thd) ||
wsrep_thd_is_applying(granted_thd))
@ -2450,22 +2452,22 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
{
WSREP_DEBUG("BF thread waiting for SR in aborting state");
ticket->wsrep_report(wsrep_debug);
wsrep_thd_UNLOCK(granted_thd);
mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
}
else if (wsrep_thd_is_SR(granted_thd) && !wsrep_thd_is_SR(request_thd))
{
WSREP_MDL_LOG(INFO, "MDL conflict, DDL vs SR",
schema, schema_len, request_thd, granted_thd);
wsrep_abort_thd(request_thd, granted_thd, 1);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_data);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_kill);
}
else
{
WSREP_MDL_LOG(INFO, "MDL BF-BF conflict", schema, schema_len,
request_thd, granted_thd);
ticket->wsrep_report(true);
wsrep_thd_UNLOCK(granted_thd);
mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
unireg_abort(1);
}
}
@ -2474,7 +2476,8 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
{
WSREP_DEBUG("BF thread waiting for FLUSH");
ticket->wsrep_report(wsrep_debug);
wsrep_thd_UNLOCK(granted_thd);
mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
}
else if (request_thd->lex->sql_command == SQLCOM_DROP_TABLE)
{
@ -2482,8 +2485,6 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
wsrep_thd_transaction_state_str(granted_thd));
ticket->wsrep_report(wsrep_debug);
wsrep_abort_thd(request_thd, granted_thd, 1);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_data);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_kill);
}
else
{
@ -2493,8 +2494,6 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
if (granted_thd->wsrep_trx().active())
{
wsrep_abort_thd(request_thd, granted_thd, true);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_data);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_kill);
}
else
{
@ -2504,15 +2503,16 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
*/
if (wsrep_thd_is_BF(request_thd, FALSE))
{
granted_thd->awake_no_mutex(KILL_QUERY_HARD);
ha_abort_transaction(request_thd, granted_thd, TRUE);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_data);
mysql_mutex_assert_not_owner(&granted_thd->LOCK_thd_kill);
}
else
{
WSREP_MDL_LOG(INFO, "MDL unknown BF-BF conflict", schema, schema_len,
request_thd, granted_thd);
ticket->wsrep_report(true);
mysql_mutex_unlock(&granted_thd->LOCK_thd_data);
mysql_mutex_unlock(&granted_thd->LOCK_thd_kill);
unireg_abort(1);
}
}
@ -2528,17 +2528,22 @@ void wsrep_handle_mdl_conflict(MDL_context *requestor_ctx,
static bool abort_replicated(THD *thd)
{
bool ret_code= false;
wsrep_thd_kill_LOCK(thd);
wsrep_thd_LOCK(thd);
if (thd->wsrep_trx().state() == wsrep::transaction::s_committing)
{
WSREP_DEBUG("aborting replicated trx: %llu", (ulonglong)(thd->real_id));
(void)wsrep_abort_thd(thd, thd, TRUE);
wsrep_abort_thd(thd, thd, TRUE);
ret_code= true;
}
else
{
/* wsrep_abort_thd() above releases LOCK_thd_data and LOCK_thd_kill, so
must do it here too. */
wsrep_thd_UNLOCK(thd);
wsrep_thd_kill_UNLOCK(thd);
}
return ret_code;
}

8
sql/wsrep_server_service.cc
View file

@ -143,9 +143,13 @@ void Wsrep_server_service::release_high_priority_service(wsrep::high_priority_se
wsrep_delete_threadvars();
}
void Wsrep_server_service::background_rollback(wsrep::client_state& client_state)
void Wsrep_server_service::background_rollback(
wsrep::unique_lock<wsrep::mutex> &lock WSREP_UNUSED,
wsrep::client_state &client_state)
{
Wsrep_client_state& cs= static_cast<Wsrep_client_state&>(client_state);
DBUG_ASSERT(lock.owns_lock());
Wsrep_client_state &cs= static_cast<Wsrep_client_state &>(client_state);
mysql_mutex_assert_owner(&cs.thd()->LOCK_thd_data);
wsrep_fire_rollbacker(cs.thd());
}

3
sql/wsrep_server_service.h
View file

@ -46,7 +46,8 @@ public:
void release_high_priority_service(wsrep::high_priority_service*);
void background_rollback(wsrep::client_state&);
void background_rollback(wsrep::unique_lock<wsrep::mutex> &,
wsrep::client_state &);
void bootstrap();
void log_message(enum wsrep::log::level, const char*);

213
sql/wsrep_thd.cc
View file

@ -307,50 +307,9 @@ void wsrep_fire_rollbacker(THD *thd)
}
}
int wsrep_abort_thd(THD *bf_thd,
THD *victim_thd,
my_bool signal)
static bool wsrep_bf_abort_low(THD *bf_thd, THD *victim_thd)
{
DBUG_ENTER("wsrep_abort_thd");
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
/* Note that when you use RSU node is desynced from cluster, thus WSREP(thd)
might not be true.
*/
if ((WSREP(bf_thd) ||
((WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU) &&
wsrep_thd_is_toi(bf_thd))) &&
!wsrep_thd_is_aborting(victim_thd))
{
WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu",
(long long)bf_thd->real_id, (long long)victim_thd->real_id);
ha_abort_transaction(bf_thd, victim_thd, signal);
}
else
{
WSREP_DEBUG("wsrep_abort_thd not effective: bf %llu victim %llu "
"wsrep %d wsrep_on %d RSU %d TOI %d aborting %d",
(long long)bf_thd->real_id, (long long)victim_thd->real_id,
WSREP_NNULL(bf_thd), WSREP_ON,
bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU,
wsrep_thd_is_toi(bf_thd),
wsrep_thd_is_aborting(victim_thd));
wsrep_thd_UNLOCK(victim_thd);
}
DBUG_RETURN(1);
}
bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
{
WSREP_LOG_THD(bf_thd, "BF aborter before");
WSREP_LOG_THD(victim_thd, "victim before");
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
#ifdef ENABLED_DEBUG_SYNC
DBUG_EXECUTE_IF("sync.wsrep_bf_abort",
@ -364,6 +323,87 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
};);
#endif
wsrep::seqno bf_seqno(bf_thd->wsrep_trx().ws_meta().seqno());
bool ret;
{
/* Adopt the lock, it is being held by the caller. */
Wsrep_mutex wsm{&victim_thd->LOCK_thd_data};
wsrep::unique_lock<wsrep::mutex> lock{wsm, std::adopt_lock};
if (wsrep_thd_is_toi(bf_thd))
{
ret= victim_thd->wsrep_cs().total_order_bf_abort(lock, bf_seqno);
}
else
{
DBUG_ASSERT(WSREP(victim_thd) ? victim_thd->wsrep_trx().active() : 1);
ret= victim_thd->wsrep_cs().bf_abort(lock, bf_seqno);
}
if (ret)
{
/* BF abort should be allowed only once by wsrep-lib.*/
DBUG_ASSERT(victim_thd->wsrep_aborter == 0);
victim_thd->wsrep_aborter= bf_thd->thread_id;
wsrep_bf_aborts_counter++;
}
lock.release(); /* No unlock at the end of the scope. */
}
/* Sanity check for wsrep-lib calls to return with LOCK_thd_data held. */
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
return ret;
}
void wsrep_abort_thd(THD *bf_thd,
THD *victim_thd,
my_bool signal)
{
DBUG_ENTER("wsrep_abort_thd");
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
/* Note that when you use RSU node is desynced from cluster, thus WSREP(thd)
might not be true.
*/
if ((WSREP(bf_thd)
|| ((WSREP_ON || bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU)
&& wsrep_thd_is_toi(bf_thd)))
&& !wsrep_thd_is_aborting(victim_thd) &&
wsrep_bf_abort_low(bf_thd, victim_thd) &&
!victim_thd->wsrep_cs().is_rollbacker_active())
{
WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu",
(long long)bf_thd->real_id, (long long)victim_thd->real_id);
victim_thd->awake_no_mutex(KILL_QUERY_HARD);
ha_abort_transaction(bf_thd, victim_thd, signal);
}
else
{
WSREP_DEBUG("wsrep_abort_thd not effective: bf %llu victim %llu "
"wsrep %d wsrep_on %d RSU %d TOI %d aborting %d",
(long long)bf_thd->real_id, (long long)victim_thd->real_id,
WSREP_NNULL(bf_thd), WSREP_ON,
bf_thd->variables.wsrep_OSU_method == WSREP_OSU_RSU,
wsrep_thd_is_toi(bf_thd),
wsrep_thd_is_aborting(victim_thd));
mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
}
DBUG_VOID_RETURN;
}
bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
{
WSREP_LOG_THD(bf_thd, "BF aborter before");
WSREP_LOG_THD(victim_thd, "victim before");
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
if (WSREP(victim_thd) && !victim_thd->wsrep_trx().active())
{
WSREP_DEBUG("wsrep_bf_abort, BF abort for non active transaction");
@ -385,32 +425,81 @@ bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd)
mysql_mutex_lock(&victim_thd->LOCK_thd_data);
}
bool ret;
wsrep::seqno bf_seqno(bf_thd->wsrep_trx().ws_meta().seqno());
return wsrep_bf_abort_low(bf_thd, victim_thd);
}
if (wsrep_thd_is_toi(bf_thd))
uint wsrep_kill_thd(THD *thd, THD *victim_thd, killed_state kill_signal)
{
DBUG_ENTER("wsrep_kill_thd");
DBUG_ASSERT(WSREP(victim_thd));
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_kill);
mysql_mutex_assert_owner(&victim_thd->LOCK_thd_data);
using trans= wsrep::transaction;
auto trx_state= victim_thd->wsrep_trx().state();
#ifndef DBUG_OFF
victim_thd->wsrep_killed_state= trx_state;
#endif /* DBUG_OFF */
/*
Already killed or in commit codepath. Mark the victim as killed,
the killed status will be restored in wsrep_after_commit() and
will be processed after the commit is over. In case of multiple
KILLs happened on commit codepath, the last one will be effective.
*/
if (victim_thd->wsrep_abort_by_kill ||
trx_state == trans::s_preparing ||
trx_state == trans::s_committing ||
trx_state == trans::s_ordered_commit)
{
/* Here we enter wsrep-lib were LOCK_thd_data will be acquired,
thus we need to release it. However, we can still hold
LOCK_thd_kill to protect from disconnect or delete. */
victim_thd->wsrep_abort_by_kill= kill_signal;
mysql_mutex_unlock(&victim_thd->LOCK_thd_data);
ret= victim_thd->wsrep_cs().total_order_bf_abort(bf_seqno);
mysql_mutex_lock(&victim_thd->LOCK_thd_data);
mysql_mutex_unlock(&victim_thd->LOCK_thd_kill);
DBUG_RETURN(0);
}
else
/*
Mark killed victim_thd with kill_signal so that awake_no_mutex does
not dive into storage engine. We use ha_abort_transaction()
to do the storage engine part for wsrep THDs.
*/
DEBUG_SYNC(thd, "wsrep_kill_before_awake_no_mutex");
victim_thd->wsrep_abort_by_kill= kill_signal;
victim_thd->awake_no_mutex(kill_signal);
/* ha_abort_transaction() releases tmp->LOCK_thd_kill, so tmp
is not safe to access anymore. */
ha_abort_transaction(thd, victim_thd, 1);
DBUG_RETURN(0);
}
void wsrep_backup_kill_for_commit(THD *thd)
{
DBUG_ASSERT(WSREP(thd));
mysql_mutex_assert_owner(&thd->LOCK_thd_kill);
DBUG_ASSERT(thd->killed != NOT_KILLED);
mysql_mutex_lock(&thd->LOCK_thd_data);
/* If the transaction will roll back, keep the killed state.
For must replay, the replay will happen in different THD context
which is high priority and cannot be killed. The owning thread will
pick the killed state in after statement processing. */
if (thd->wsrep_trx().state() != wsrep::transaction::s_cert_failed &&
thd->wsrep_trx().state() != wsrep::transaction::s_must_abort &&
thd->wsrep_trx().state() != wsrep::transaction::s_aborting &&
thd->wsrep_trx().state() != wsrep::transaction::s_must_replay)
{
/* Test: mysql-wsrep-features#165. Here we enter wsrep-lib
were LOCK_thd_data will be acquired and later LOCK_thd_kill
thus we need to release them. */
wsrep_thd_UNLOCK(victim_thd);
ret= victim_thd->wsrep_cs().bf_abort(bf_seqno);
wsrep_thd_LOCK(victim_thd);
thd->wsrep_abort_by_kill= thd->killed;
thd->wsrep_abort_by_kill_err= thd->killed_err;
thd->killed= NOT_KILLED;
thd->killed_err= 0;
}
if (ret)
{
wsrep_bf_aborts_counter++;
}
return ret;
mysql_mutex_unlock(&thd->LOCK_thd_data);
}
void wsrep_restore_kill_after_commit(THD *thd)
{
DBUG_ASSERT(WSREP(thd));
mysql_mutex_assert_owner(&thd->LOCK_thd_kill);
thd->killed= thd->wsrep_abort_by_kill;
thd->killed_err= thd->wsrep_abort_by_kill_err;
thd->wsrep_abort_by_kill= NOT_KILLED;
thd->wsrep_abort_by_kill_err= 0;
}
int wsrep_create_threadvars()

36
sql/wsrep_thd.h
View file

@ -88,10 +88,44 @@ bool wsrep_create_appliers(long threads, bool mutex_protected=false);
void wsrep_create_rollbacker();
bool wsrep_bf_abort(THD* bf_thd, THD* victim_thd);
int wsrep_abort_thd(THD *bf_thd,
/*
Abort transaction for victim_thd. This function is called from
MDL BF abort codepath.
@note This thread unlocks victim_thd->LOCK_thd_kill, so accessing
victim_thd after the function returns is not safe anymore.
*/
void wsrep_abort_thd(THD *bf_thd,
THD *victim_thd,
my_bool signal) __attribute__((nonnull(1,2)));
/**
Kill wsrep connection with kill_signal. Object thd is not
guaranteed to exist anymore when this function returns.
Asserts that the caller holds victim_thd->LOCK_thd_kill,
victim_thd->LOCK_thd_data.
Releases victim_thd->LOCK_thd_kill, victim_thd->LOCK_thd_data.
@param thd THD object for connection that executes the KILL.
@param victim_thd THD object for connection to be killed.
@param kill_signal Kill signal.
@return Zero if the kill was successful, otherwise non-zero error code.
*/
uint wsrep_kill_thd(THD *thd, THD *victim_thd, killed_state kill_signal);
/*
Backup kill status for commit.
*/
void wsrep_backup_kill_for_commit(THD *);
/*
Restore KILL status after commit.
*/
void wsrep_restore_kill_after_commit(THD *);
/*
Helper methods to deal with thread local storage.
The purpose of these methods is to hide the details of thread

26
sql/wsrep_trans_observer.h
View file

@ -252,6 +252,11 @@ static inline int wsrep_before_prepare(THD* thd, bool all)
wsrep_xid_init(&thd->wsrep_xid,
thd->wsrep_trx().ws_meta().gtid());
}
mysql_mutex_lock(&thd->LOCK_thd_kill);
if (thd->killed) wsrep_backup_kill_for_commit(thd);
mysql_mutex_unlock(&thd->LOCK_thd_kill);
DBUG_RETURN(ret);
}
@ -294,6 +299,11 @@ static inline int wsrep_before_commit(THD* thd, bool all)
thd->wsrep_trx().ws_meta().gtid());
wsrep_register_for_group_commit(thd);
}
mysql_mutex_lock(&thd->LOCK_thd_kill);
if (thd->killed) wsrep_backup_kill_for_commit(thd);
mysql_mutex_unlock(&thd->LOCK_thd_kill);
DBUG_RETURN(ret);
}
@ -314,7 +324,8 @@ static inline int wsrep_ordered_commit(THD* thd,
const wsrep_apply_error&)
{
DBUG_ENTER("wsrep_ordered_commit");
WSREP_DEBUG("wsrep_ordered_commit: %d", wsrep_is_real(thd, all));
WSREP_DEBUG("wsrep_ordered_commit: %d %lld", wsrep_is_real(thd, all),
(long long) wsrep_thd_trx_seqno(thd));
DBUG_ASSERT(wsrep_run_commit_hook(thd, all));
DBUG_RETURN(thd->wsrep_cs().ordered_commit());
}
@ -420,9 +431,16 @@ int wsrep_after_statement(THD* thd)
wsrep::to_c_string(thd->wsrep_cs().mode()),
wsrep::to_c_string(thd->wsrep_cs().transaction().state()),
wsrep_thd_query(thd));
DBUG_RETURN((thd->wsrep_cs().state() != wsrep::client_state::s_none &&
thd->wsrep_cs().mode() == Wsrep_client_state::m_local) ?
thd->wsrep_cs().after_statement() : 0);
int ret= ((thd->wsrep_cs().state() != wsrep::client_state::s_none &&
thd->wsrep_cs().mode() == Wsrep_client_state::m_local) ?
thd->wsrep_cs().after_statement() : 0);
if (wsrep_is_active(thd))
{
mysql_mutex_lock(&thd->LOCK_thd_kill);
wsrep_restore_kill_after_commit(thd);
mysql_mutex_unlock(&thd->LOCK_thd_kill);
}
DBUG_RETURN(ret);
}
static inline void wsrep_after_apply(THD* thd)