database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-07-29 05:21:33 +03:00
Code Activity

MDEV-8260 : Issues related to concurrent CTAS

Browse Source 
* Wait for aborted thd (victim) to release MDL locks
* Skip aborting an already aborted thd
* Defer setting OK status in case of CTAS
* Minor cosmetic changes
* Added a test case
This commit is contained in:
Nirbhay Choubey
2015-06-19 19:25:15 -04:00
parent 6050ab6586
commit fc716dc5a5
7 changed files with 142 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
mysql-test/suite/galera/r/galera_concurrent_ctas.result Normal file
View File

@ -0,0 +1 @@
# End of test

57
mysql-test/suite/galera/t/galera_concurrent_ctas.test Normal file
View File

@ -0,0 +1,57 @@
--source include/galera_cluster.inc
--source include/have_innodb.inc
--write_file $MYSQLTEST_VARDIR/tmp/galera_concurrent.sql
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
DROP table t1;
CREATE table t1 as SELECT SLEEP(0);
CREATE table t2 as SELECT SLEEP(0);
CREATE table t3 as SELECT SLEEP(0);
CREATE table t4 as SELECT SLEEP(0);
CREATE table t5 as SELECT SLEEP(0);
CREATE table t6 as SELECT SLEEP(0);
CREATE table t7 as SELECT SLEEP(0);
CREATE table t8 as SELECT SLEEP(0);
CREATE table t9 as SELECT SLEEP(0);
DROP table t1;
DROP table t2;
DROP table t3;
DROP table t4;
DROP table t5;
DROP table t6;
DROP table t7;
DROP table t8;
DROP table t9;
EOF
let $run=10;
while($run)
{
--error 0,1
exec $MYSQL --user=root --host=127.0.0.1 --port=$NODE_MYPORT_1 test
< $MYSQLTEST_VARDIR/tmp/galera_concurrent.sql &
$MYSQL --user=root --host=127.0.0.1 --port=$NODE_MYPORT_2 test
< $MYSQLTEST_VARDIR/tmp/galera_concurrent.sql;
dec $run;
}
--remove_file $MYSQLTEST_VARDIR/tmp/galera_concurrent.sql
--source include/galera_end.inc
--echo # End of test

12
sql/mdl.cc
View File

@ -1199,6 +1199,13 @@ MDL_wait::timed_wait(THD *thd, struct timespec *abs_timeout,
while (!m_wait_status && !thd->killed &&
wait_result != ETIMEDOUT && wait_result != ETIME)
{
#ifdef WITH_WSREP
if (wsrep_thd_is_BF(thd, true))
{
wait_result= mysql_cond_wait(&m_COND_wait_status, &m_LOCK_wait_status);
}
else
#endif
wait_result= mysql_cond_timedwait(&m_COND_wait_status, &m_LOCK_wait_status,
abs_timeout);
}
@ -1283,12 +1290,15 @@ void MDL_lock::Ticket_list::add_ticket(MDL_ticket *ticket)
WSREP_DEBUG("MDL add_ticket inserted before: %lu %s",
wsrep_thd_thread_id(waiting->get_ctx()->get_thd()),
wsrep_thd_query(waiting->get_ctx()->get_thd()));
/* Insert the ticket before the first non-BF waiting thd. */
m_list.insert_after(prev, ticket);
added= true;
}
prev= waiting;
}
if (!added) m_list.push_back(ticket);
/* Otherwise, insert the ticket at the back of the waiting list. */
if (!added) m_list.push_back(ticket);
while ((granted= itg++))
{

2
sql/sql_class.h
View File

@ -3486,6 +3486,8 @@ class select_insert :public select_result_interceptor {
virtual void store_values(List<Item> &values);
virtual bool can_rollback_data() { return 0; }
void send_error(uint errcode,const char *err);
bool prepare_eof();
bool send_ok_packet();
bool send_eof();
virtual void abort_result_set();
/* not implemented: select_insert is never re-used in prepared statements */

71
sql/sql_insert.cc
View File

@ -3650,14 +3650,14 @@ void select_insert::send_error(uint errcode,const char *err)
}
bool select_insert::send_eof()
bool select_insert::prepare_eof()
{
int error;
bool const trans_table= table->file->has_transactions();
ulonglong id, row_count;
bool changed;
killed_state killed_status= thd->killed;
DBUG_ENTER("select_insert::send_eof");
DBUG_ENTER("select_insert::prepare_eof");
DBUG_PRINT("enter", ("trans_table=%d, table_type='%s'",
trans_table, table->file->table_type()));
@ -3699,11 +3699,10 @@ bool select_insert::send_eof()
*/
#ifdef WITH_WSREP
if ((WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open()) &&
(!error || thd->transaction.stmt.modified_non_trans_table))
#else
if (mysql_bin_log.is_open() &&
(!error || thd->transaction.stmt.modified_non_trans_table))
#endif
(!error || thd->transaction.stmt.modified_non_trans_table))
{
int errcode= 0;
if (!error)
@ -3715,7 +3714,7 @@ bool select_insert::send_eof()
trans_table, FALSE, FALSE, errcode))
{
table->file->ha_release_auto_increment();
DBUG_RETURN(1);
DBUG_RETURN(true);
}
}
table->file->ha_release_auto_increment();
@ -3723,27 +3722,49 @@ bool select_insert::send_eof()
if (error)
{
table->file->print_error(error,MYF(0));
DBUG_RETURN(1);
DBUG_RETURN(true);
}
char buff[160];
DBUG_RETURN(false);
}
bool select_insert::send_ok_packet() {
char message[160]; /* status message */
ulong row_count; /* rows affected */
ulong id; /* last insert-id */
DBUG_ENTER("select_insert::send_ok_packet");
if (info.ignore)
sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
(ulong) (info.records - info.copied),
(ulong) thd->warning_info->statement_warn_count());
my_snprintf(message, sizeof(message), ER(ER_INSERT_INFO),
(ulong) info.records, (ulong) (info.records - info.copied),
(ulong) thd->warning_info->statement_warn_count());
else
sprintf(buff, ER(ER_INSERT_INFO), (ulong) info.records,
(ulong) (info.deleted+info.updated),
(ulong) thd->warning_info->statement_warn_count());
my_snprintf(message, sizeof(message), ER(ER_INSERT_INFO),
(ulong) info.records, (ulong) (info.deleted + info.updated),
(ulong) thd->warning_info->statement_warn_count());
row_count= info.copied + info.deleted +
((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
info.touched : info.updated);
((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
info.touched : info.updated);
id= (thd->first_successful_insert_id_in_cur_stmt > 0) ?
thd->first_successful_insert_id_in_cur_stmt :
(thd->arg_of_last_insert_id_function ?
thd->first_successful_insert_id_in_prev_stmt :
(info.copied ? autoinc_value_of_last_inserted_row : 0));
::my_ok(thd, row_count, id, buff);
DBUG_RETURN(0);
::my_ok(thd, row_count, id, message);
DBUG_RETURN(false);
}
bool select_insert::send_eof()
{
bool res;
DBUG_ENTER("select_insert::send_eof");
res= (prepare_eof() || send_ok_packet());
DBUG_RETURN(res);
}
void select_insert::abort_result_set() {
@ -4240,9 +4261,12 @@ void select_create::send_error(uint errcode,const char *err)
bool select_create::send_eof()
{
bool tmp=select_insert::send_eof();
if (tmp)
DBUG_ENTER("select_create::send_eof");
if (prepare_eof())
{
abort_result_set();
DBUG_RETURN(true);
}
else
{
/*
@ -4262,7 +4286,7 @@ bool select_create::send_eof()
thd->thread_id, thd->wsrep_conflict_state, thd->query());
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
abort_result_set();
return TRUE;
DBUG_RETURN(true);
}
mysql_mutex_unlock(&thd->LOCK_wsrep_thd);
#endif /* WITH_WSREP */
@ -4270,6 +4294,9 @@ bool select_create::send_eof()
table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
send_ok_packet();
if (m_plock)
{
mysql_unlock_tables(thd, *m_plock);
@ -4277,7 +4304,7 @@ bool select_create::send_eof()
m_plock= NULL;
}
}
return tmp;
DBUG_RETURN(false);
}

11
sql/wsrep_mysqld.cc
View File

@ -1476,10 +1476,21 @@ void wsrep_to_isolation_end(THD *thd)
gra->wsrep_exec_mode, gra->wsrep_query_state, gra->wsrep_conflict_state, \
gra->command, gra->lex->sql_command, gra->query());
/**
Check if request for the metadata lock should be granted to the requester.
@param requestor_ctx The MDL context of the requestor
@param ticket MDL ticket for the requested lock
@retval TRUE Lock request can be granted
@retval FALSE Lock request cannot be granted
*/
bool
wsrep_grant_mdl_exception(MDL_context *requestor_ctx,
MDL_ticket *ticket
) {
/* Fallback to the non-wsrep behaviour */
if (!WSREP_ON) return FALSE;
THD *request_thd = requestor_ctx->get_thd();

11
sql/wsrep_thd.cc
View File

@ -495,6 +495,17 @@ int wsrep_abort_thd(void *bf_thd_ptr, void *victim_thd_ptr, my_bool signal)
bf_thd->wsrep_exec_mode == TOTAL_ORDER) ) &&
victim_thd)
{
if ((victim_thd->wsrep_conflict_state == MUST_ABORT) ||
(victim_thd->wsrep_conflict_state == ABORTED) ||
(victim_thd->wsrep_conflict_state == ABORTING))
{
WSREP_DEBUG("wsrep_abort_thd called by %llu with victim %llu already "
"aborted. Ignoring.",
(bf_thd) ? (long long)bf_thd->real_id : 0,
(long long)victim_thd->real_id);
DBUG_RETURN(1);
}
WSREP_DEBUG("wsrep_abort_thd, by: %llu, victim: %llu", (bf_thd) ?
(long long)bf_thd->real_id : 0, (long long)victim_thd->real_id);
ha_wsrep_abort_transaction(bf_thd, victim_thd, signal);