I need someone to review my below fix for MDEV-7818. This involves the FLUSH TABLES WITH READ LOCK command and associated metadata locking, something that I know little about. Monty suggested the basic semantics - that all currently running parallel replication transactions would be allowed to complete before granting the global read lock. But it really needs a review of the actual implementation, by someone who has an understanding of the FTWRL semantics and implementation. Thanks, - Kristian. Kristian Nielsen <knielsen@knielsen-hq.org> writes:

revision-id: 0e609daea4796fafca94f7adf796addfc506168b parent(s): 22b6c297dc0de6b176238ae820bddb5e70f2c8ab committer: Kristian Nielsen branch nick: mariadb timestamp: 2015-06-09 16:01:53 +0200 message:

MDEV-7818: Deadlock occurring with parallel replication and FTWRL

Problem is that FLUSH TABLES WITH READ LOCK first blocks threads from starting new commits, then waits for running commits to complete. But in-order parallel replication needs commits to happen in a particular order, so this can easily deadlock.

To fix this problem, this patch introduces a way to temporarily pause the parallel replication worker threads. Before starting FTWRL, we let all worker threads complete in-progress transactions, and then wait. Then we proceed to take the global read lock. Once the lock is obtained, we unpause the worker threads. Now commits are blocked from starting by the global read lock, so the deadlock will no longer occur.

--- sql/mysqld.cc | 3 + sql/mysqld.h | 3 + sql/rpl_parallel.cc | 230 ++++++++++++++++++++++++++++++++++++++++++++++++++-- sql/rpl_parallel.h | 27 ++++-- sql/sql_parse.cc | 13 +++ 5 files changed, 262 insertions(+), 14 deletions(-)

diff --git a/sql/mysqld.cc b/sql/mysqld.cc index e05c0b6..af8ae2c 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -9514,6 +9514,9 @@ PSI_stage_info stage_waiting_for_prior_transaction_to_commit= { 0, "Waiting for PSI_stage_info stage_waiting_for_prior_transaction_to_start_commit= { 0, "Waiting for prior transaction to start commit before starting next transaction", 0}; PSI_stage_info stage_waiting_for_room_in_worker_thread= { 0, "Waiting for room in worker thread event queue", 0}; PSI_stage_info stage_waiting_for_workers_idle= { 0, "Waiting for worker threads to be idle", 0}; +PSI_stage_info stage_waiting_for_ftwrl= { 0, "Waiting due to global read lock", 0}; +PSI_stage_info stage_waiting_for_ftwrl_threads_to_pause= { 0, "Waiting for worker threads to pause for global read lock", 0}; +PSI_stage_info stage_waiting_for_rpl_thread_pool= { 0, "Waiting while replication worker thread pool is busy", 0}; PSI_stage_info stage_master_gtid_wait_primary= { 0, "Waiting in MASTER_GTID_WAIT() (primary waiter)", 0}; PSI_stage_info stage_master_gtid_wait= { 0, "Waiting in MASTER_GTID_WAIT()", 0}; PSI_stage_info stage_gtid_wait_other_connection= { 0, "Waiting for other master connection to process GTID received on multiple master connections", 0}; diff --git a/sql/mysqld.h b/sql/mysqld.h index 156e7f9..8c953c1 100644 --- a/sql/mysqld.h +++ b/sql/mysqld.h @@ -454,6 +454,9 @@ extern PSI_stage_info stage_waiting_for_prior_transaction_to_commit; extern PSI_stage_info stage_waiting_for_prior_transaction_to_start_commit; extern PSI_stage_info stage_waiting_for_room_in_worker_thread; extern PSI_stage_info stage_waiting_for_workers_idle; +extern PSI_stage_info stage_waiting_for_ftwrl; +extern PSI_stage_info stage_waiting_for_ftwrl_threads_to_pause; +extern PSI_stage_info stage_waiting_for_rpl_thread_pool; extern PSI_stage_info stage_master_gtid_wait_primary; extern PSI_stage_info stage_master_gtid_wait; extern PSI_stage_info stage_gtid_wait_other_connection; diff --git a/sql/rpl_parallel.cc b/sql/rpl_parallel.cc index 738d0e5..d90507a 100644 --- a/sql/rpl_parallel.cc +++ b/sql/rpl_parallel.cc @@ -280,6 +280,8 @@ do_gco_wait(rpl_group_info *rgi, group_commit_orderer *gco, rpl_parallel_entry *entry= rgi->parallel_entry; uint64 wait_count;

+ mysql_mutex_assert_owner(&entry->LOCK_parallel_entry); + if (!gco->installed) { group_commit_orderer *prev_gco= gco->prev_gco; @@ -336,6 +338,159 @@ do_gco_wait(rpl_group_info *rgi, group_commit_orderer *gco, }

+static void +do_ftwrl_wait(rpl_group_info *rgi, + bool *did_enter_cond, PSI_stage_info *old_stage) +{ + THD *thd= rgi->thd; + rpl_parallel_entry *entry= rgi->parallel_entry; + uint64 sub_id= rgi->gtid_sub_id; + + mysql_mutex_assert_owner(&entry->LOCK_parallel_entry); + + if (unlikely(entry->pause_sub_id > 0) && sub_id > entry->pause_sub_id) + { + thd->ENTER_COND(&entry->COND_parallel_entry, &entry->LOCK_parallel_entry, + &stage_waiting_for_ftwrl, old_stage); + *did_enter_cond= true; + do + { + if (entry->force_abort || rgi->worker_error) + break; + if (thd->check_killed()) + { + thd->send_kill_message(); + slave_output_error_info(rgi, thd); + signal_error_to_sql_driver_thread(thd, rgi, 1); + break; + } + mysql_cond_wait(&entry->COND_parallel_entry, &entry->LOCK_parallel_entry); + } while (entry->pause_sub_id > 0 && sub_id > entry->pause_sub_id); + } + + if (sub_id > entry->largest_started_sub_id) + entry->largest_started_sub_id= sub_id; +} + + +static void +pool_mark_busy(rpl_parallel_thread_pool *pool) +{ + mysql_mutex_assert_owner(&pool->LOCK_rpl_thread_pool); + DBUG_ASSERT(!pool->busy); + pool->busy= true; +} + + +static void +pool_mark_not_busy(rpl_parallel_thread_pool *pool) +{ + mysql_mutex_assert_owner(&pool->LOCK_rpl_thread_pool); + DBUG_ASSERT(pool->busy); + pool->busy= false; + mysql_cond_broadcast(&pool->COND_rpl_thread_pool); +} + + +void +rpl_unpause_for_ftwrl(THD *thd) +{ + uint32 i; + rpl_parallel_thread_pool *pool= &global_rpl_thread_pool; + + DBUG_ASSERT(pool->busy); + + for (i= 0; i < pool->count; ++i) + { + rpl_parallel_entry *e; + rpl_parallel_thread *rpt= pool->threads[i]; + + mysql_mutex_lock(&rpt->LOCK_rpl_thread); + if (!rpt->current_owner) + { + mysql_mutex_unlock(&rpt->LOCK_rpl_thread); + continue; + } + e= rpt->current_entry; + mysql_mutex_lock(&e->LOCK_parallel_entry); + mysql_mutex_unlock(&rpt->LOCK_rpl_thread); + e->pause_sub_id= 0; + mysql_cond_broadcast(&e->COND_parallel_entry); + } + + mysql_mutex_lock(&pool->LOCK_rpl_thread_pool); + pool_mark_not_busy(pool); + mysql_cond_broadcast(&pool->COND_rpl_thread_pool); + mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool); +} + + +/* + . + + Note: in case of error return, unpause_for_ftwrl() must _not_ be called. +*/ +int +rpl_pause_for_ftwrl(THD *thd) +{ + uint32 i; + rpl_parallel_thread_pool *pool= &global_rpl_thread_pool; + int err= 0; + + /* + While the count_pending_pause_for_ftwrl counter is non-zero, the pool + cannot be shutdown/resized, so threads are guaranteed to not disappear. + + This is required to safely be able to access the individual threads below. + (We cannot lock an individual thread while holding LOCK_rpl_thread_pool, + as this can deadlock against release_thread()). + */ + mysql_mutex_lock(&pool->LOCK_rpl_thread_pool); + pool_mark_busy(pool); + mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool); + + for (i= 0; i < pool->count; ++i) + { + PSI_stage_info old_stage; + rpl_parallel_entry *e; + rpl_parallel_thread *rpt= pool->threads[i]; + + mysql_mutex_lock(&rpt->LOCK_rpl_thread); + if (!rpt->current_owner) + { + mysql_mutex_unlock(&rpt->LOCK_rpl_thread); + continue; + } + e= rpt->current_entry; + mysql_mutex_lock(&e->LOCK_parallel_entry); + mysql_mutex_unlock(&rpt->LOCK_rpl_thread); + ++e->need_sub_id_signal; + if (!e->pause_sub_id) + e->pause_sub_id= e->largest_started_sub_id; + thd->ENTER_COND(&e->COND_parallel_entry, &e->LOCK_parallel_entry, + &stage_waiting_for_ftwrl_threads_to_pause, &old_stage); + while (e->last_committed_sub_id < e->pause_sub_id && !err) + { + if (thd->check_killed()) + { + thd->send_kill_message(); + err= 1; + break; + } + mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry); + }; + --e->need_sub_id_signal; + thd->EXIT_COND(&old_stage); + if (err) + break; + } + + if (err) + rpl_unpause_for_ftwrl(thd); + return err; +} + + #ifndef DBUG_OFF static int dbug_simulate_tmp_error(rpl_group_info *rgi, THD *thd) @@ -856,6 +1011,9 @@ handle_rpl_parallel_thread(void *arg)

if (unlikely(entry->stop_on_error_sub_id <= rgi->wait_commit_sub_id)) skip_event_group= true; + if (likely(!skip_event_group)) + do_ftwrl_wait(rgi, &did_enter_cond, &old_stage); + register_wait_for_prior_event_group_commit(rgi, entry);

unlock_or_exit_cond(thd, &entry->LOCK_parallel_entry, @@ -1058,6 +1216,47 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool, rpl_parallel_thread **new_list= NULL; rpl_parallel_thread *new_free_list= NULL; rpl_parallel_thread *rpt_array= NULL; + THD *thd; + PSI_stage_info old_stage; + int res; + + /* + Wait here while the queue is busy. This is done to make FLUSH TABLES WITH + READ LOCK work correctly, without incuring extra locking penalties in + normal operation. FLUSH TABLES WITH READ LOCK needs to lock threads in the + thread pool, and for this we need to make sure the pool will not go away + during the operation. The LOCK_rpl_thread_pool is not suitable for + this. It is taken by release_thread() while holding LOCK_rpl_thread; so it + must be released before locking any LOCK_rpl_thread lock, or a deadlock + can occur. + + So we protect the infrequent operations of FLUSH TABLES WITH READ LOCK and + pool size changes with this condition wait. + */ + thd= current_thd; + res= 0; + mysql_mutex_lock(&pool->LOCK_rpl_thread_pool); + if (thd) + thd->ENTER_COND(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool, + &stage_waiting_for_rpl_thread_pool, &old_stage); + while (pool->busy) + { + if (thd->check_killed()) + { + thd->send_kill_message(); + res= 1; + break; + } + mysql_cond_wait(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool); + } + if (!res) + pool_mark_busy(pool); + if (thd) + thd->EXIT_COND(&old_stage); + else + mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool); + if (res) + return res;

/* Allocate the new list of threads up-front. @@ -1106,7 +1305,14 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool, */ for (i= 0; i < pool->count; ++i) { - rpl_parallel_thread *rpt= pool->get_thread(NULL, NULL); + rpl_parallel_thread *rpt; + + mysql_mutex_lock(&pool->LOCK_rpl_thread_pool); + while ((rpt= pool->free_list) == NULL) + mysql_cond_wait(&pool->COND_rpl_thread_pool, &pool->LOCK_rpl_thread_pool); + pool->free_list= rpt->next; + mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool); + mysql_mutex_lock(&rpt->LOCK_rpl_thread); rpt->stop= true; mysql_cond_signal(&rpt->COND_rpl_thread); mysql_mutex_unlock(&rpt->LOCK_rpl_thread); @@ -1157,7 +1363,7 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool, }

mysql_mutex_lock(&pool->LOCK_rpl_thread_pool); - mysql_cond_broadcast(&pool->COND_rpl_thread_pool); + pool_mark_not_busy(pool); mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool);

return 0; @@ -1182,6 +1388,9 @@ rpl_parallel_change_thread_count(rpl_parallel_thread_pool *pool, } my_free(new_list); } + mysql_mutex_lock(&pool->LOCK_rpl_thread_pool); + pool_mark_not_busy(pool); + mysql_mutex_unlock(&pool->LOCK_rpl_thread_pool); return 1; }

@@ -1445,7 +1654,7 @@ rpl_parallel_thread::loc_free_gco(group_commit_orderer *gco)

rpl_parallel_thread_pool::rpl_parallel_thread_pool() - : count(0), threads(0), free_list(0), inited(false) + : threads(0), free_list(0), count(0), inited(false), busy(false) { }

@@ -1453,9 +1662,10 @@ rpl_parallel_thread_pool::rpl_parallel_thread_pool() int rpl_parallel_thread_pool::init(uint32 size) { - count= 0; threads= NULL; free_list= NULL; + count= 0; + busy= false;

mysql_mutex_init(key_LOCK_rpl_thread_pool, &LOCK_rpl_thread_pool, MY_MUTEX_INIT_SLOW); @@ -1496,8 +1706,14 @@ rpl_parallel_thread_pool::get_thread(rpl_parallel_thread **owner, rpl_parallel_thread *rpt;

mysql_mutex_lock(&LOCK_rpl_thread_pool); - while ((rpt= free_list) == NULL) + for (;;) + { + while (unlikely(busy)) + mysql_cond_wait(&COND_rpl_thread_pool, &LOCK_rpl_thread_pool); + if ((rpt= free_list) != NULL) + break; mysql_cond_wait(&COND_rpl_thread_pool, &LOCK_rpl_thread_pool); + } free_list= rpt->next; mysql_mutex_unlock(&LOCK_rpl_thread_pool); mysql_mutex_lock(&rpt->LOCK_rpl_thread); @@ -1908,7 +2124,7 @@ rpl_parallel::wait_for_workers_idle(THD *thd)

e= (struct rpl_parallel_entry *)my_hash_element(&domain_hash, i); mysql_mutex_lock(&e->LOCK_parallel_entry); - e->need_sub_id_signal= true; + ++e->need_sub_id_signal; thd->ENTER_COND(&e->COND_parallel_entry, &e->LOCK_parallel_entry, &stage_waiting_for_workers_idle, &old_stage); while (e->current_sub_id > e->last_committed_sub_id) @@ -1921,7 +2137,7 @@ rpl_parallel::wait_for_workers_idle(THD *thd) } mysql_cond_wait(&e->COND_parallel_entry, &e->LOCK_parallel_entry); } - e->need_sub_id_signal= false; + --e->need_sub_id_signal; thd->EXIT_COND(&old_stage); if (err) return err; diff --git a/sql/rpl_parallel.h b/sql/rpl_parallel.h index 04f8706..77a4005 100644 --- a/sql/rpl_parallel.h +++ b/sql/rpl_parallel.h @@ -199,12 +199,18 @@ struct rpl_parallel_thread {

struct rpl_parallel_thread_pool { - uint32 count; struct rpl_parallel_thread **threads; struct rpl_parallel_thread *free_list; mysql_mutex_t LOCK_rpl_thread_pool; mysql_cond_t COND_rpl_thread_pool; + uint32 count; bool inited; + /* + While FTWRL runs, this counter is incremented to make SQL thread or + STOP/START slave not try to start new activity while that operation + is in progress. + */ + bool busy;

rpl_parallel_thread_pool(); int init(uint32 size); @@ -219,6 +225,12 @@ struct rpl_parallel_entry { mysql_mutex_t LOCK_parallel_entry; mysql_cond_t COND_parallel_entry; uint32 domain_id; + /* + Incremented by wait_for_workers_idle() and rpl_pause_for_ftwrl() to show + that they are waiting, so that finish_event_group knows to signal them + when last_committed_sub_id is increased. + */ + uint32 need_sub_id_signal; uint64 last_commit_id; bool active; /* @@ -228,12 +240,6 @@ struct rpl_parallel_entry { */ bool force_abort; /* - Set in wait_for_workers_idle() to show that it is waiting, so that - finish_event_group knows to signal it when last_committed_sub_id is - increased. - */ - bool need_sub_id_signal; - /* At STOP SLAVE (force_abort=true), we do not want to process all events in the queue (which could unnecessarily delay stop, if a lot of events happen to be queued). The stop_count provides a safe point at which to stop, so @@ -291,6 +297,11 @@ struct rpl_parallel_entry { The value is ULONGLONG_MAX when no error occured. */ uint64 stop_on_error_sub_id; + /* + During FLUSH TABLES WITH READ LOCK, transactions with sub_id larger than + this value must not start, but wait until the global read lock is released. + */ + uint64 pause_sub_id; /* Total count of event groups queued so far. */ uint64 count_queued_event_groups; /* @@ -331,5 +342,7 @@ extern struct rpl_parallel_thread_pool global_rpl_thread_pool; extern int rpl_parallel_activate_pool(rpl_parallel_thread_pool *pool); extern int rpl_parallel_inactivate_pool(rpl_parallel_thread_pool *pool); extern bool process_gtid_for_restart_pos(Relay_log_info *rli, rpl_gtid *gtid); +extern int rpl_pause_for_ftwrl(THD *thd); +extern void rpl_unpause_for_ftwrl(THD *thd);

#endif /* RPL_PARALLEL_H */ diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc index 5635e9a..ca6b11e 100644 --- a/sql/sql_parse.cc +++ b/sql/sql_parse.cc @@ -4271,6 +4271,17 @@ case SQLCOM_PREPARE: break; }

+ if (lex->type & REFRESH_READ_LOCK) + { + /* + We need to pause any parallel replication slave workers during FLUSH + TABLES WITH READ LOCK. Otherwise we might cause a deadlock, as + worker threads eun run in arbitrary order but need to commit in a + specific given order. + */ + if (rpl_pause_for_ftwrl(thd)) + goto error; + } /* reload_acl_and_cache() will tell us if we are allowed to write to the binlog or not. @@ -4301,6 +4312,8 @@ case SQLCOM_PREPARE: if (!res) my_ok(thd); } + if (lex->type & REFRESH_READ_LOCK) + rpl_unpause_for_ftwrl(thd);

break; } _______________________________________________ commits mailing list commits@mariadb.org https://lists.askmonty.org/cgi-bin/mailman/listinfo/commits