Quellcode-Bibliothek rpl_parallel.cc
Sprache: C
#include "mariadb.h" #include "rpl_parallel.h" #include "slave.h" #include "rpl_mi.h" #include "sql_parse.h" #include "debug_sync.h" #include "sql_repl.h" #include "wsrep_mysqld.h" #ifdef WITH_WSREP#include "wsrep_trans_observer.h" #endif /* Code for optional parallel execution of replicated events on the slave . /* Maximum number of queued events to accumulate in a local free list , before moving them to the global free list . There is additional a limit of how much to accumulate based on opt_slave_parallel_max_queued . #define QEV_BATCH_FREE 200 struct rpl_parallel_thread_pool global_rpl_thread_pool;static void signal_error_to_sql_driver_thread(THD *thd, rpl_group_info *rgi,int err);static void static int struct rpl_parallel_thread *rpt)int err;#ifdef WITH_WSREPif (wsrep_before_statement(thd))"Parallel slave failed at wsrep_before_statement() hook" );return (1 );#endif /* WITH_WSREP */ sizeof (rgi->event_relay_log_name_buf),sizeof (rgi->future_event_master_log_name),if (event_can_update_last_master_timestamp(ev))#ifdef WITH_WSREPif (wsrep_after_statement(thd))"Parallel slave failed at wsrep_after_statement() hook" );1 ;#endif /* WITH_WSREP */ /* ToDo: error handling. */ return err;static void int cmp;/* Events that are not part of an event group , such as Format Description , Stop , GTID List and such , are executed directly in the driver SQL thread , to keep the relay log state up - to - date . But the associated position update is done here , in sync with other normal events as they are queued to worker threads . if ((thd->variables.option_bits & OPTION_BEGIN) &&return ;/* Do not update position if an earlier event group caused an error abort. */ if (e->stop_on_error_sub_id < (uint64)ULONGLONG_MAX ||return ;#ifdef ENABLED_DEBUG_SYNC"pause_sql_thread_on_fde" ,"now SIGNAL paused_on_fde WAIT_FOR sql_thread_continue" "now SIGNAL main_sql_thread_continue" #endif if (cmp < 0 )else if (cmp == 0 &&;if (cmp < 0 )sizeof (rli->group_master_log_name),else if (cmp == 0 /* Wait for any pending deadlock kills . Since deadlock kills happen asynchronously , we need to be sure they will be completed before starting a new transaction . Otherwise the new transaction might suffer a spurious kill . void while (rgi->killed_for_retry == rpl_group_info::RETRY_KILL_PENDING)static void if (rgi->get_finish_event_group_called())return ;int err;if (unlikely(rgi->worker_error))/* In case a previous wait was killed , we need to re - register to be able to repeat the wait . And before doing that , we un - register any previous registration ( in case we got an error earlier and skipped waiting ) . /* Remove any left - over registration to wait for a prior commit to complete . Normally , such wait would already have been removed at this point by wait_for_prior_commit ( ) called from within COMMIT processing . However , in case of MyISAM and no binlog , we might not have any commit processing , and so we need to do the wait here , before waking up any subsequent commits , to preserve correct order of event execution . Also , in the error case we might have skipped waiting and thus need to remove it explicitly . Or the wait might have been killed and we need to repeat the registration and the wait . It is important in the non - error case to do a wait , not just an unregister . Because we might be last in a group - commit that is replicated in parallel , and the following event will then wait for us to complete and rely on this also ensuring that any other event in the group has completed . And in the error case , correct GCO lifetime relies on the fact that once the last event group in the GCO has executed wait_for_prior_commit ( ) , all earlier event groups have also committed ; this way no more mark_start_commit ( ) calls can be made and it is safe to de - allocate the GCO . Thus this final wait is done with kill ignored during the wait . This is fine , at this point there is no active query or transaction to abort , and the thread will continue as soon as earlier event groups complete . Note though , that in the non - error case there is no guarantee that finish_event_group ( ) will be run in - order . For example , a successful binlog group commit will wakeup all participating event groups simultaneously so only thread scheduling will decide the order in which finish_event_group ( ) calls acquire LOCK_parallel_entry . false );if (unlikely(err) && !rgi->worker_error)/* Calls to check_duplicate_gtid ( ) must match up with record_and_update_gtid ( ) ( or release_domain_owner ( ) in error case ) . This assertion tries to catch any missing release of the domain . R);/* We need to mark that this event group started its commit phase , in case we missed it before ( otherwise we would deadlock the next event group that is waiting for this ) . In most cases ( normal DML ) , it will be a no - op . false ;if (entry->last_committed_sub_id < sub_id)/* Record that this event group has finished ( eg . transaction is committed , if transactional ) , so other event groups will no longer attempt to wait for us to commit . Once we have increased entry - > last_committed_sub_id , no other threads will execute register_wait_for_prior_commit ( ) against us . Thus , by doing one extra ( usually redundant ) wakeup_subsequent_commits ( ) we can ensure that no register_wait_for_prior_commit ( ) can ever happen without a subsequent wakeup_subsequent_commits ( ) to wake it up . We can race here with the next transactions , but that is fine , as long as we check that we do not decrease last_committed_sub_id . If this commit is done , then any prior commits will also have been done and also no longer need waiting for . if (entry->need_sub_id_signal)/* Now free any GCOs in which all transactions have committed. */ while (tmp_gco &&/* We must not free a GCO before the wait_count of the following GCO has been reached and wakeup has been sent . Otherwise we will lose the wakeup and hang ( there were several such bugs in the past ) . The intention is that this is ensured already since we only free when the last event group in the GCO has committed ( tmp_gco - > last_sub_id < = sub_id ) . However , if we have a bug , we have extra check on next_gco - > wait_count to hopefully avoid hanging ; we have here an assertion in debug builds that this check does not in fact trigger . while (tmp_gco)/* If this event group got error , then any following event groups that have not yet started should just skip their group , preparing for stop of the SQL driver thread . if (unlikely(rgi->worker_error) && entry->stop_on_error_sub_id > sub_id)#ifdef ENABLED_DEBUG_SYNC"hold_worker_on_schedule" , {if (entry->stop_on_error_sub_id < (uint64)ULONGLONG_MAX)"now SIGNAL continue_worker" ));"rpl_parallel_simulate_wait_at_retry" , {if (rgi->current_gtid.seq_no == 1000 ) {"now WAIT_FOR proceed_by_1000" ));"hold_worker2_favor_worker3" , {if (rgi->current_gtid.seq_no == 2001 ) {"now SIGNAL cont_worker3" ));#endif if (rgi->killed_for_retry == rpl_group_info::RETRY_KILL_PENDING)false ;true );static void int err)"hold_worker2_favor_worker3" , {if (rgi->current_gtid.seq_no == 2002 ) {"now WAIT_FOR cont_worker2" ));true );true ;false ;static void bool *did_enter_cond,if (*did_enter_cond)false ;else static void if (rgi->wait_commit_sub_id > entry->last_committed_sub_id)/* Register that the commit of this event group must wait for the commit of the previous event group to complete before it may complete itself , so that we preserve commit order . /* Do not start parallel execution of this event group until all prior groups have reached the commit phase that are not safe to run in parallel with . static void bool *did_enter_cond, PSI_stage_info *old_stage)if (!gco->installed)if (prev_gco)true ;if (wait_count > entry->count_committing_event_groups)"rpl_parallel_start_waiting_for_prior" );true ;do if (!rgi->worker_error && unlikely(thd->check_killed(1 )))"rpl_parallel_start_waiting_for_prior_killed" );1 );/* Even though we were killed , we need to continue waiting for the prior event groups to signal that we can continue . Otherwise we mess up the accounting for ordering . However , now that we have marked the error , events will just be skipped rather than executed , and things will progress quickly towards stop . while (wait_count > entry->count_committing_event_groups);static bool bool should_stop= false ;if (unlikely(entry->force_abort) && rgi->gtid_sub_id > entry->stop_sub_id)/* We are stopping ( STOP SLAVE ) , and this event group need not be applied before we can safely stop . So return a flag that will cause us to skip , rather than execute , the following events . Once all queued events have been skipped , the STOP SLAVE is complete ( for this thread ) . true ;if (unlikely(entry->stop_on_error_sub_id <= rgi->wait_commit_sub_id))1 ;true ;if (likely(!should_stop))/* Since we did not decide to stop , bump the largest_started_sub_id while still holding LOCK_parallel_entry . if (rgi->gtid_sub_id > entry->largest_started_sub_id)return should_stop;static bool bool *did_enter_cond, PSI_stage_info *old_stage)bool aborted= false ;"do_ftwrl_wait" );/* If a FLUSH TABLES WITH READ LOCK ( FTWRL ) is pending , check if this transaction is later than transactions that have priority to complete before FTWRL . If so , wait here so that FTWRL can proceed and complete first . ( entry - > pause_sub_id is ULONGLONG_MAX if no FTWRL is pending , which makes this test false as required ) . if (unlikely(sub_id > entry->pause_sub_id))true ;do if (entry->force_abort || rgi->worker_error)true ;break ;if (unlikely(thd->check_killed()))1 );break ;while (sub_id > entry->pause_sub_id);"delay_ftwrl_wait_gtid_0_x_100" , {if (rgi->current_gtid.domain_id == 0 &&100 ) {/* Simulate delayed wakeup from the mysql_cond_wait ( ) . To do this , we need to have the LOCK_parallel_entry mutex released during the wait . "now SIGNAL pause_wait_started WAIT_FOR pause_wait_continue" ));/* We do not call EXIT_COND ( ) here , as this will be done later by our caller ( since we set * did_enter_cond to true ) . static int int res= 0 ;bool did_enter_cond= false ;/* 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 . "mark_busy_mdev_22370" ,my_sleep(1000000 ););if (pool->busy)if (thd)true ;do if (thd && unlikely(thd->check_killed()))1 ;break ;while (pool->busy);if (!res)true ;if (did_enter_cond)else return res;static void false ;void "rpl_unpause_after_ftwrl" );for (i= 0 ; i < pool->count; ++i)if (!rpt->current_owner)continue ;false ;/* Do not change pause_sub_id if force_abort is set . force_abort is set in case of STOP SLAVE . Reason : If pause_sub_id is not changed and force_abort_is set , any parallel slave thread waiting in do_ftwrl_wait ( ) will on wakeup return from do_ftwrl_wait ( ) with 1 . This will set skip_event_group to 1 in handle_rpl_parallel_thread ( ) and the parallel thread will abort at once . If pause_sub_id is changed , the code in handle_rpl_parallel_thread ( ) would continue to execute the transaction in the queue , which would cause some transactions to be lost . if (!e->force_abort)/* . Note : in case of error return , rpl_unpause_after_ftwrl ( ) must _ not_ be called . int int err;4 , 4 ); // array of replication source mi:s "rpl_pause_for_ftwrl" );/* 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 ( ) ) . if ((err= pool_mark_busy(pool, thd)))for (i= 0 ; i < pool->count; ++i)if (!rpt->current_owner)continue ;/* Setting the rpt - > pause_for_ftwrl flag makes sure that the thread will not de - allocate itself until signalled to do so by rpl_unpause_after_ftwrl ( ) . true ;if (e->pause_sub_id == (uint64)ULONGLONG_MAX)"pause_for_ftwrl_wait" , {"now " "SIGNAL pause_ftwrl_waiting " "WAIT_FOR pause_ftwrl_cont" ));while (e->pause_sub_id < (uint64)ULONGLONG_MAX &&if (unlikely(thd->check_killed()))1 ;break ;if (err)break ;/* Notify any source any domain waiting - for - master Start - Alter to give way . bool found= false ;for (uint i= 0 ; i < mi_arr.elements() && !found; i++)if (!found)while ((info= info_iterator++))true ;if (err)#ifndef DBUG_OFFstatic int if (rgi->current_gtid.domain_id == 0 && rgi->current_gtid.seq_no == 100 &&4 )0 ));return 1 ;return 0 ;#endif /* If we detect a deadlock due to eg . storage engine locks that conflict with the fixed commit order , then the later transaction will be killed asynchroneously to allow the former to complete its commit . In this case , we convert the ' killed ' error into a deadlock error , and retry the later transaction . If we are doing optimistic parallel apply of transactions not known to be safe , we convert any error to a deadlock error , but then at retry we will wait for prior transactions to commit first , so that the retries can be done non - speculative . static void int err_code;if (!thd->get_stmt_da()->is_error())return ;if ((rgi->speculation == rpl_group_info::SPECULATE_OPTIMISTIC &&0 ));/* Check if an event marks the end of an event group . Returns non - zero if so , zero otherwise . In addition , returns 1 if the group is committing , 2 if it is rolling back . static int if (event_type == XID_EVENT || event_type == XA_PREPARE_LOG_EVENT)return 1 ;if (event_type == QUERY_EVENT) // COMMIT/ROLLBACK are never compressed if (qev->is_commit() ||"XA COMMIT" )) ||"XA ROLLBACK" )))return 1 ;if (qev->is_rollback())return 2 ;return 0 ;static int 1 ;const char *errmsg;int err;char log_name[FN_REFLEN];0 ;0 ;0 ;#ifdef WITH_WSREP"sync.wsrep_retry_event_group" , {const char act[]= "now " "SIGNAL sync.wsrep_retry_event_group_reached " "WAIT_FOR signal.wsrep_retry_event_group" ;#endif /* WITH_WSREP */ /* If we already started committing before getting the deadlock ( or other error ) that caused us to need to retry , we have already signalled subsequent transactions that we have started committing . This is potentially a problem , as now we will rollback , and if subsequent transactions would start to execute now , they could see an unexpected state of the database and get eg . key not found or duplicate key error . However , to get a deadlock in the first place , there must have been another earlier transaction that is waiting for us . Thus that other transaction has _ not_ yet started to commit , and any subsequent transactions will still be waiting at this point . So here , we decrement back the count of transactions that started committing ( if we already incremented it ) , undoing the effect of an earlier mark_start_commit ( ) . Then later , when the retry succeeds and we commit again , we can do a new mark_start_commit ( ) and eventually wake up subsequent transactions at the proper time . We need to do the unmark before the rollback , to be sure that the transaction we deadlocked with will not signal that it started to commit until after the unmark . "inject_mdev8302" , { my_sleep(20000 );});"rpl_parallel_retry_after_unmark" );/* We might get the deadlock error that causes the retry during commit , while sitting in wait_for_prior_commit ( ) . If this happens , we will have a pending error in the wait_for_commit object . So clear this by unregistering ( and later re - registering ) the wait . if (thd->wait_for_commit_ptr)"inject_mdev8031" , {/* Simulate that we get deadlock killed at this exact point. */ #ifdef ENABLED_DEBUG_SYNC"rpl_parallel_simulate_wait_at_retry" , {if (rgi->current_gtid.seq_no == 1001 ) {"rpl_parallel_simulate_wait_at_retry WAIT_FOR proceed_by_1001" ));"rpl_parallel_simulate_wait_at_retry" );#endif /* We are still applying the event group , even though we will roll it back and retry it . So for - - gtid - ignore - duplicates , keep ownership of the domain during the retry so another master connection will not try to take over and duplicate apply the same event group ( MDEV - 33475 ) . 1 , 1 /* keep_domain_owner */); #ifdef ENABLED_DEBUG_SYNC"hold_worker2_favor_worker3" , {if (rgi->current_gtid.seq_no == 2003 ) {"now WAIT_FOR cont_worker3" ));#endif /* If we retry due to a deadlock kill that occurred during the commit step , we might have already updated ( but not committed ) an update of table mysql . gtid_slave_pos , and cleared the gtid_pending flag . Now we have rolled back any such update , so we must set the gtid_pending flag back to true so that we will do a new update when / if we succeed with the retry . true ;for (;;)if (rgi->gtid_sub_id < entry->stop_on_error_sub_id ||"simulate_mdev_12746" ))else /* A failure of a preceding " parent " transaction may not be seen by the current one through its own worker_error . Such induced error gets set by ourselves now . 1 ;0 ));goto err;/* Let us wait for all prior transactions to complete before trying again . This way , we avoid repeatedly conflicting with and getting deadlock killed by the same earlier transaction . if (!(err= thd->wait_for_prior_commit()))break ;if (!has_temporary_error(thd))goto err;/* If we get a temporary error such as a deadlock kill , we can safely ignore it , as we already rolled back . But we still want to retry the wait for the prior transaction to complete its commit . if (thd->wait_for_commit_ptr)"inject_mdev8031" , {/* Inject a small sleep to give prior transaction a chance to commit. */ 100000 );/* Let us clear any lingering deadlock kill one more time , here after wait_for_prior_commit ( ) has completed . This should rule out any possibility of an old deadlock kill lingering on beyond this point . #ifdef WITH_WSREPif (WSREP(thd))/* Exec after statement hook to make sure that the failed transaction if (wsrep_after_statement(thd))"Parallel slave worker failed at wsrep_after_statement() hook" );1 ;goto err;"parallel slave retry, after trx start" );#endif /* WITH_WSREP */ if ((fd= open_binlog(&rlog, log_name, &errmsg)) <0 )1 ;goto err;delete description_event;if (!description_event)1 ;goto err;"inject_mdev8031" , {/* Simulate pending KILL caught in read_relay_log_description_event(). */ if (unlikely(thd->check_killed()))1 ;goto err;do /* The loop is here so we can try again the next relay log file on EOF. */ for (;;)int error;if (ev)break ;if (unlikely(error))"slave SQL thread aborted because of I/O error" ;1 ;goto check_retry;if (unlikely(rlog.error > 0 ))"Slave SQL thread: I/O error reading " "event(errno: %d cur_log->error: %d)" ,"Aborting slave SQL thread because of partial event read" ;1 ;goto err;/* EOF. Move to the next relay log. */ 1 ;/* Find the next relay log file. */ if ((err= rli->relay_log.find_log_pos(&linfo, log_name, 1 )) ||1 )))char buff[22 ];"next log error: %d offset: %s log: %s" ,goto err;"inject_retry_event_group_open_binlog_kill" , {if (retries < 2 )/* Simulate that we get deadlock killed during open_binlog(). */ 1 ;1 ;goto check_retry;if ((fd= open_binlog(&rlog, log_name, &errmsg)) <0 )1 ;goto check_retry;/* Loop to try again on the new log file. */ if (event_type == FORMAT_DESCRIPTION_EVENT)delete description_event;continue ;else if (event_type == START_ENCRYPTION_EVENT)delete ev;continue ;else if (!Log_event::is_group_event(event_type))delete ev;continue ;if (!qev)delete ev;0 ));1 ;goto err;if (is_group_ending(ev, event_type) == 1 )"rpl_parallel_simulate_double_temp_err_gtid_0_x_100" ,if (retries == 0 ) err= dbug_simulate_tmp_error(rgi, thd););"rpl_parallel_simulate_infinite_temp_err_gtid_0_x_100" ,if (!err)continue ;if (has_temporary_error(thd))if (retries < slave_trans_retries)if (fd >= 0 )1 ;goto do_retry;"Slave worker thread retried transaction %lu time(s) " "in vain, giving up. Consider raising the value of " "the slave_transaction_retries variable." ,goto err;while (event_count < events_to_execute);if (description_event)delete description_event;if (fd >= 0 )if (errmsg)"Error reading relay log event: %s" , errmsg);return err;void *arg)struct rpl_parallel_thread::queued_event *events;bool group_standalone= true ;bool in_event_group= false ;bool skip_event_group= false ;0 ;int err;struct rpl_parallel_thread *rpt= (struct rpl_parallel_thread *)arg;"rpl_parallel" );new THD(next_thread_id());0 ;1 ;0 ;"Waiting for work from main SQL threads" );/* We need to use ( at least ) REPEATABLE READ isolation level . Otherwise speculative parallel apply can run out - of - order and give wrong results for statement - based replication . while (rpt->delay_start)true ;#ifdef WITH_WSREPif (wsrep_before_command(thd))"Parallel slave failed at wsrep_before_command() hook" );true ;#endif /* WITH_WSREP */ while (!rpt->stop)0 ;/* There are 4 cases that should cause us to wake up : - Events have been queued for us to handle . - We have an owner , but no events and not inside event group - > we need to release ourself to the thread pool - SQL thread is stopping , and we have an owner but no events , and we are inside an event group ; no more events will be queued to us , so we need to abort the group ( force_abort = = 1 ) . - Thread pool shutdown ( rpt - > stop = = 1 ) . while (!( (events= rpt->event_queue) ||if (!wait_count++)for (qev= events; qev; qev= next_qev)bool end_of_group;int group_ending;if (qev->typ == rpl_parallel_thread::queued_event::QUEUED_POS_UPDATE)continue ;else if (qev->typ ==if (in_event_group)/* Master restarted ( crashed ) in the middle of an event group . So we need to roll back and discard that event group . 1 );false ;continue ;/* Handle a new event group, which will be initiated by a GTID event. */ if ((event_type= qev->ev->get_type_code()) == GTID_EVENT)0 ;if (rpt->channel_name_length)bool did_enter_cond= false ;"rpl_parallel_delay_gtid_0_x_100_start" , {if (rgi->current_gtid.domain_id==0 &&100 )10000 );#ifdef ENABLED_DEBUG_SYNC"hold_worker_on_schedule" , {if (rgi->current_gtid.domain_id == 0 &&100 ) {"now SIGNAL reached_pause WAIT_FOR continue_worker" ));"rpl_parallel_scheduled_gtid_0_x_100" , {if (rgi->current_gtid.domain_id == 0 &&100 ) {"now SIGNAL scheduled_gtid_0_x_100" ));#endif if (unlikely(thd->wait_for_commit_ptr) && group_rgi != NULL)/* This indicates that we get a new GTID event in the middle of a not completed event group . This is corrupt binlog ( the master will never write such binlog ) , so it does not happen unless someone tries to inject wrong crafted binlog , but let us still try to handle it somewhat nicely . true );true ;/* If the standalone flag is set , then this event group consists of a single statement ( possibly preceeded by some Intvar_log_event and similar ) , without any terminating COMMIT / ROLLBACK / XID . 0 != (static_cast <Gtid_log_event *>(qev->ev)->flags2 &"gco_wait_delay_gtid_0_x_99" , {if (rgi->current_gtid.domain_id == 0 && rgi->current_gtid.seq_no == 99 ) {"now SIGNAL gco_wait_paused WAIT_FOR gco_wait_cont" ));if (likely(!skip_event_group))/* Register ourself to wait for the previous commit , if we need to do such registration _ and_ that previous commit has not already occurred . if (opt_gtid_ignore_duplicates &&int res=if (res < 0 )/* Error. */ 1 );else if (!res)/* GTID already applied by another master connection, skip. */ true ;else /* We have to apply the event. */ /* If we are optimistically running transactions in parallel , but this particular event group should not run in parallel with what came before , then wait now for the prior transaction to complete its commit . if (rgi->speculation == rpl_group_info::SPECULATE_WAIT &&1 );/* We do not unmark_start_commit ( ) here in case of an explicit ROLLBACK statement . Such events should be very rare , there is no real reason to try to group commit them - on the contrary , it seems best to avoid running them in parallel with following group commits , as with ROLLBACK events we are already deep in dangerous corner cases with mix of transactional and non - transactional tables or the like . And avoiding the mark_start_commit ( ) here allows us to keep an assertion in ha_rollback_trans ( ) that we do not rollback after doing mark_start_commit ( ) . if (group_ending == 1 && likely(!rgi->worker_error))/* Do an extra check for ( deadlock ) kill here . This helps prevent a lingering deadlock kill that occurred during normal DML processing to propagate past the mark_start_commit ( ) . If we detect a deadlock only after mark_start_commit ( ) , we have to unmark , which has at least a theoretical possibility of leaving a window where it looks like all transactions in a GCO have started committing , while in fact one will need to rollback and retry . Normally this will not happen , since the kill is there to resolve a deadlock that is preventing at least one transaction from proceeding . One case it can happen is with InnoDB dict stats update , which can temporarily cause transactions to block each other , but locks are released immediately , they don ' t linger until commit . There could be other similar cases , there were historically a number of subtle bugs in this area . But once we start the commit , we can expect that no new lock conflicts will be introduced . So by handling any lingering deadlock kill at this point just before mark_start_commit ( ) , we should be robust even towards spurious deadlock kills . if (rgi->killed_for_retry != rpl_group_info::RETRY_KILL_NONE)if (!thd->killed)"rpl_parallel_before_mark_start_commit" );if (thd->lex->stmt_accessed_temp_table())/* Temporary tables are special , they require strict single - threaded use as they have no locks protecting concurrent access . Therefore , we cannot safely use the optimization of overlapping the commit of this transaction with the start of the following . So we skip the early mark_start_commit ( ) and also block any wakeup_subsequent_commits ( ) until this event group is fully done , inside finish_event_group ( ) . true ;else "rpl_parallel_after_mark_start_commit" );#ifdef ENABLED_DEBUG_SYNC"halt_past_mark_start_commit" ,"now WAIT_FOR past_mark_continue" )));"-d,halt_past_mark_start_commit" );#endif /* If the SQL thread is stopping , we just skip execution of all the following event groups . We still do all the normal waiting and wakeup processing between the event groups as a simple way to ensure that everything is stopped and cleaned up correctly . if (likely(!rgi->worker_error) && !skip_event_group)#ifndef DBUG_OFF0 ;"rpl_parallel_simulate_temp_err_xid" ,if (event_type == XID_EVENT)0 ));1 ;"rpl_parallel_simulate_temp_err_xid" );if (!err)#endif if (unlikely(thd->check_killed()))1 ;else "rpl_parallel_simulate_temp_err_gtid_0_x_100" ,if (unlikely(err))"rpl_mdev31655_zero_retries" ,if ((rgi->current_gtid.seq_no % 1000 ) == 0 )0 ;if (has_temporary_error(thd) && max_retries > 0 )else delete qev->ev;if (unlikely(err))if (!rgi->worker_error)if (end_of_group)false ;false ;"rpl_parallel_end_of_group" );/* Now that we have the lock , we can move everything from our local free lists to the real free lists that are also accessible from the SQL driver thread . if ((events= rpt->event_queue) != NULL)/* Take next group of events from the replication pool . This is faster than having to wakeup the pool manager thread to give us a new event . goto more_events;if (in_event_group && group_rgi->parallel_entry->force_abort)/* We are asked to abort , without getting the remaining events in the current event group . We have to rollback the current transaction and update the last sub_id value so that SQL thread will know we are done with the half - processed event group . 1 );false ;false ;if (!in_event_group)/* If we are in a FLUSH TABLES FOR READ LOCK, wait for it */ while (rpt->current_entry && rpt->pause_for_ftwrl)/* We are currently in the delicate process of pausing parallel replication while FLUSH TABLES WITH READ LOCK is starting . We must not de - allocate the thread ( setting rpt - > current_owner = NULL ) until rpl_unpause_after_ftwrl ( ) has woken us up . /* Wait for rpl_unpause_after_ftwrl ( ) to wake us up . Note that rpl_pause_for_ftwrl ( ) may wait for ' e - > pause_sub_id ' to change . This should happen eventually in finish_event_group ( ) if (rpt->pause_for_ftwrl)/* Tell wait_for_done() that we are done, if it is waiting. */ if (likely(rpt->current_entry) &&if (!rpt->stop)#ifdef WITH_WSREP#endif /* WITH_WSREP */ 0 ;"Slave worker thread exiting" );0 ;delete thd;false ;return NULL;static void /** Change thread count for global parallel worker threads @ param pool parallel thread pool @ param new_count Number of threads to be in pool . 0 in shutdown @ param force Force thread count to new_count even if slave threads are running By default we don ' t resize pool of there are running threads . However during shutdown we will always do it . This is needed as any_slave_sql_running ( ) returns 1 during shutdown as we don ' t want to access master_info while Master_info_index : : free_connections are running . static int bool force)int res;if ((res= pool_mark_busy(pool, current_thd)))return res;/* Protect against parallel pool resizes */ if (pool->count == new_count)return 0 ;/* If we are about to delete pool , do an extra check that there are no new slave threads running since we marked pool busy if (!new_count && !force)if (any_slave_sql_running(false ))"warning" ,"SQL threads running while trying to reset parallel pool" ));return 0 ; // Ok to not resize pool /* Allocate the new list of threads up - front . That way , if we fail half - way , we only need to free whatever we managed to allocate , and will not be left with a half - functional thread pool . if (new_count &&sizeof (*new_list),sizeof (*rpt_array),0 ), (int (new_count*sizeof (*new_list) +sizeof (*rpt_array))));goto err;for (i= 0 ; i < new_count; ++i)true ;if (mysql_thread_create(key_rpl_parallel_thread, &th, &connection_attrib,0 ));goto err;/* Grab each old thread in turn , and signal it to stop . Note that since we require all replication threads to be stopped before changing the parallel replication worker thread pool , all the threads will be already idle and will terminate immediately . for (i= 0 ; i < pool->count; ++i)while ((rpt= pool->free_list) == NULL)true ;for (i= 0 ; i < pool->count; ++i)while (rpt->running)while (rpt->qev_free_list)while (rpt->rgi_free_list)delete rpt->rgi_free_list;while (rpt->gco_free_list)if (new_count < pool->count)if (new_count > pool->count)for (i= 0 ; i < pool->count; ++i)false ;0 ;0 ;false ;while (!pool->threads[i]->running)return 0 ;if (new_list)while (new_free_list)false ;true ;while (!new_free_list->running)while (new_free_list->running)return 1 ;/* Deactivate the parallel replication thread pool , if there are now no more SQL threads running . int rpl_parallel_resize_pool_if_no_slaves(void )/* master_info_index is set to NULL on shutdown */ if (opt_slave_parallel_threads > 0 && !any_slave_sql_running(false ))return rpl_parallel_inactivate_pool(&global_rpl_thread_pool);return 0 ;/** Pool activation is preceeded by taking a " lock " of pool_mark_busy which guarantees the number of running slaves drops to zero atomicly with the number of pool workers . This resolves race between the function caller thread and one that may be attempting to deactivate the pool . int int rc= 0 ;struct pool_bkp_for_pfs* bkp= &pool->pfs_bkp;if ((rc= pool_mark_busy(pool, current_thd)))return rc; // killed if (!pool->count)0 );if (!rc)if (pool->count)if (bkp->inited)if (bkp->count != pool->count)else false ; // Mark backup as stale during pool init else return rc;int return rpl_parallel_change_thread_count(pool, 0 , 0 );void if (loc_qev_list)/* Signal that our queue can now accept more events. */ 0 ;0 ;if (loc_rgi_list)if (loc_gco_list)void if (ir)0 ;if ((qev= qev_free_list))else if (!(qev= (queued_event *)my_malloc(PSI_INSTRUMENT_ME, sizeof (*qev), MYF(0 ))))0 ), (int )sizeof (*qev));return NULL;return qev;if (!qev)return NULL;sizeof (qev->event_relay_log_name),sizeof (qev->future_event_master_log_name),return qev;const char *relay_log_name,if (!qev)return NULL;sizeof (qev->event_relay_log_name),sizeof (qev->future_event_master_log_name),return qev;void if (ir != last_ir)if (last_ir)if (!loc_qev_list)else /* We want to release to the global free list only occasionally , to avoid having to take the LOCK_rpl_thread muted too many times . However , we do need to release regularly . If we let the unreleased part grow too large , then the SQL driver thread may go to sleep waiting for the queue to drop below opt_slave_parallel_max_queued , and this in turn can stall all other worker threads for more stuff to do . if (++qev_free_pending >= QEV_BATCH_FREE ||3 )void if ((rgi= rgi_free_list))else if (!(rgi= new rpl_group_info(rli)))0 ), (int )sizeof (*rgi));return NULL;true ;if ((rgi->deferred_events_collecting= rli->mi->rpl_filter->is_on()) &&new Deferred_log_events(rli);if (event_group_new_gtid(rgi, gtid_ev))0 ));return NULL;0 ;/* rgi is transaction specific so we need to move this value to rgi */ this ;false ;false ;/* We can remove the reserved_start_alter_thread flag . If we get more concurrent alter handle_split_alter will automatically set this flag again . false ;return rgi;void if (!loc_rgi_list)else void if ((gco= gco_free_list))else if (!(gco= (group_commit_orderer *)my_malloc(PSI_INSTRUMENT_ME, sizeof (*gco), MYF(0 ))))0 ), (int )sizeof (*gco));return NULL;false ;0 ;#ifndef DBUG_OFFfalse ;#endif return gco;void #ifndef DBUG_OFFtrue ;#endif if (!loc_gco_list)else void rpl_group_info::finish_start_alter_event_group()this ->gtid_sub_id, this ->parallel_entry, this );0 ), last_error_number(0 ), last_error_timestamp(0 ),0 ), last_trans_retry_count(0 ), start_time(0 )0 ), free_list(0 ), count(0 ), inited(false ),current_start_alters(0 ), busy(false ),0 , false , false , NULL}int 0 ;false ;true ;/* The pool is initially empty . Threads will be spawned when a slave SQL thread is started . return 0 ;void void if (!inited)return ;this , 0 , 1 );void if (!inited)return ;false ;/* Wait for a worker thread to become idle . When one does , grab the thread for our use and return it . Note that we return with the worker threads ' s LOCK_rpl_thread mutex locked . struct rpl_parallel_thread *0 );while (unlikely(busy) || !(rpt= free_list))return rpt;/* Release a thread to the thread pool . The thread should be locked , and should not have any work queued for it . void if (!list)void if (pfs_bkp.inited)for (uint i=0 ; i<count;i++)if (rpt->channel_name_length)if (rli->err_thread_id && rpt->thd->thread_id == rli->err_thread_id)sizeof (rli->last_error().message));1000000 ;false ;true ;/* START ALTER , COMMIT ALTER / ROLLBACK ALTER scheduling Steps : - 1 . ( For Gtid_log_event SA ) . Get the worker thread which is either e - > rpl_threads [ i ] is NULL means worker from poll has not been assigned yet e - > rpl_threads [ i ] - > current_owner ! = & e - > rpl_threads [ i ] Thread has been released , or about to //same as choose_thread logic ! e - > rpl_threads [ i ] - > current_start_alter_id is 0 , safe to schedule . We dont want to schedule on worker which already have been scheduled SA but CA / RA has not been scheduled yet . current_start_alter_id will indicate this . If we dont do this we will get deadlock . 2 . ( For Gtid_log_event SA ) call choose_thread_internal so that e - > rpl_threads [ idx ] is not null update the current_start_alter_id 3 . ( For Gtid_log_event SA ) update local e - > pending_start_alters ( local ) variable and pool - > current_start_alters ( global ) We need 2 status variable ( global and local ) because we can have slave_domain_parallel_threads ! = pool - > threads . 4 . ( For CA / RA Gtid_log_event ) Update e - > pending_start_alters and pool - > current_start_alters while holding mutex lock on pool ( if SA is not assigned to reserved thread ) @ returns true Worker allocated ( choose_thread_internal called ) false Worker not allocated ( choose_thread_internal not called ) static bool handle_split_alter(rpl_parallel_entry *e, //uint32 *idx, //choose_thread_internal specific bool *did_enter_cond, rpl_group_info* rgi,bool thread_allocated= false ;0 , *idx= &i;//Step 1 if (flags_extra & Gtid_log_event::FL_START_ALTER_E1 ||//This will arrange finding threads for CA/RA as well //as concurrent DDL /* j is needed for round robin scheduling , we will start with rpl_thread_idx go till rpl_thread_max and then start with 0 to rpl_thread_idx auto j= static_cast <uint32>(e->thread_sched_fifo->head() - e->rpl_threads); // formerly e->rpl_thread_idx; for (uint i= 0 ; i < e->rpl_thread_max; i++)if (!e->rpl_threads[j].thr ||//This condition will hit atleast one time no matter what happens "info" , ("Start alter id %d" , j));goto idx_found;//We did not find and idx 0 );return false ;//e->rpl_thread_idx= *idx; /* place the found *idx index into the head */ true ;if (flags_extra & Gtid_log_event::FL_START_ALTER_E1)/* We are locking LOCK_rpl_thread_pool becuase we are going to update current_start_alters if (e->pending_start_alters < e->rpl_thread_max - 1 &&1 )else true ;0 ;0 ;if (flags_extra & (Gtid_log_event::FL_COMMIT_ALTER_E1 |//Free the corresponding rpt current_start_alter_id for (uint i= 0 ; i < e->rpl_thread_max; i++)if (e->rpl_threads[i].thr &&0 ;0 ;"info" , ("Commit/Rollback alter id %d" , i));break ;return thread_allocated;/* Check when we have done a complete round of scheduling for workers 0 , 1 , . . . , ( rpl_thread_max - 1 ) , in this order . This often occurs every rpl_thread_max event group , but XA XID dependency restrictions can cause insertion of extra out - of - order worker scheduling in - between the normal round - robin scheduling . void static_cast <uint32>(cur - rpl_threads);if (idx == current_generation_idx)if (idx >= rpl_thread_max)/* A new generation; all workers have been scheduled at least once. */ 0 ;0 ;while (i < maybe_active_xid.elements)/* Purge no longer active XID from the list : - In generation N , XID might have been scheduled for worker W . - Events in generation ( N + 1 ) might run freely in parallel with W . - Events in generation ( N + 2 ) will have done wait_for_prior_commit for the event group with XID ( or a later one ) , but the XID might still be active for a bit longer after wakeup_prior_commit ( ) . - Events in generation ( N + 3 ) will have done wait_for_prior_commit ( ) for an event in W _ after_ the XID , so are sure not to see the XID active . Therefore , XID can be safely scheduled to a different worker in generation ( N + 3 ) when last prior use was in generation N ( or earlier ) . if (a->generation + 3 <= cur_gen)continue ;if (xid->eq(&a->xid))/* Update the last used generation and return the match. */ return a->thr;/* try to keep allocated memory in the range of [2,10] * initial_chunk_size */ if (maybe_active_xid.elements <= 2 * active_xid_init_alloc() &&10 * active_xid_init_alloc())/* No matching XID conflicts. */ return nullptr;/* Obtain a worker thread that we can queue an event to . Each invocation allocates a new worker thread , to maximise parallelism . However , only up to a maximum of - - slave - domain - parallel - threads workers can be occupied by a single replication domain ; after that point , we start re - using worker threads that are still executing events that were queued earlier for this thread . We never queue more than - - rpl - parallel - wait - queue_max amount of events for one worker , to avoid the SQL driver thread using up all memory with queued events while worker threads are stalling . Note that this function returns with rpl_parallel_thread : : LOCK_rpl_thread locked . Exception is if we were killed , in which case NULL is returned . The * did_enter_cond flag is set true if we had to wait for a worker thread to become free ( with mysql_cond_wait ( ) ) . If so , old_stage will also be set , and the LOCK_rpl_thread must be released with THD : : EXIT_COND ( ) instead of mysql_mutex_unlock . When ` gtid_ev ' is not NULL the last worker thread will be returned again , if it is still available . Otherwise a new worker thread is allocated . A worker for XA transaction is determined through xid hashing which ensure for a XA - complete to be scheduled to the same - xid XA - prepare worker . bool *did_enter_cond,if (gtid_ev)/* New event group; cycle the thread scheduling buckets round-robin. */ //rpl_thread_idx will be updated handle_split_alter if (handle_split_alter(this , gtid_ev, &cur_thr, did_enter_cond, rgi,return cur_thr->thr;if (gtid_ev->flags2 &if ((cur_thr= check_xa_xid_dependency(>id_ev->xid)))/* A previously scheduled event group with the same XID might still be active in a worker , so schedule this event group in the same worker to avoid a conflict . else /* Record this XID now active. */ if (!a)return NULL;else else return choose_thread_internal(cur_thr /*idx*/, did_enter_cond, rgi, old_stage); bool *did_enter_cond,if (thr)false ;for (;;)if (thr->current_owner != &cur_thr->thr)/* The worker thread became idle , and returned to the free list and possibly was allocated to a different request . So we should allocate a new worker thread . break ;else if (thr->queued_size <= opt_slave_parallel_max_queued)/* The thread is ready to queue into. */ break ;else /* We have reached the limit of how much memory we are allowed to use for queuing events , so wait for the thread to consume some of its queue . if (!*did_enter_cond)/* We need to do the debug_sync before ENTER_COND ( ) . Because debug_sync changes the thd - > mysys_var - > current_mutex , and this can cause THD : : awake to use the wrong mutex . #ifdef ENABLED_DEBUG_SYNC"rpl_parallel_wait_queue_max" , {"now SIGNAL wait_queue_ready" ));#endif true ;if (unlikely(rli->sql_driver_thd->check_killed(1 )))0 ));#ifdef ENABLED_DEBUG_SYNC"rpl_parallel_wait_queue_max" , {"now SIGNAL wait_queue_killed" ));#endif return NULL;if (!thr)this );return thr;static void void *element)while (e->current_gco)false )32 ,sizeof (uint32),void false ;struct rpl_parallel_entry *e;if (!(e= (rpl_parallel_entry *)my_hash_search(&domain_hash,const uchar *)&domain_id,sizeof (domain_id))))/* Allocate a new, empty one. */ if (count == 0 || count > opt_slave_parallel_threads)if (!my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME|MY_ZEROFILL),sizeof (*e),sizeof (*p),sizeof (*fifo),0 ), (int )(sizeof (*e)+count*sizeof (*p)));return NULL;/* Initialize a FIFO of scheduled worker threads. */ new (fifo) I_List<rpl_parallel_entry::sched_bucket>;/* ( We cycle the FIFO _ before_ allocating next entry in rpl_parallel_entry : : choose_thread ( ) . So initialize the FIFO with the highest element at the front , just so that the first event group gets scheduled on entry 0 ) . new (p+count-1 ) rpl_parallel_entry::sched_bucket);for (ulong i= 0 ; i < count-1 ; ++i)new (p+i) rpl_parallel_entry::sched_bucket);0 ;0 ;sizeof (rpl_parallel_entry::xid_active_generation),0 , e->active_xid_init_alloc(), 0 , MYF(0 ));0 ;if (my_hash_insert(&domain_hash, (uchar *)e))return NULL;else false ;return e;/** Wait until all sql worker threads has stopped processing This is called when sql thread has been killed / stopped void struct rpl_parallel_entry *e;/* First signal all workers that they must force quit ; no more events will be queued to complete any partial event groups executed . for (i= 0 ; i < domain_hash.records; ++i)struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);/* We want the worker threads to stop as quickly as is safe . If the slave SQL threads are behind , we could have significant amount of events queued for the workers , and we want to stop without waiting for them all to be applied first . But if any event group has already started executing in a worker , we want to be sure that all prior event groups are also executed , so that we stop at a consistent point in the binlog stream ( per replication domain ) . At this point , we are holding LOCK_parallel_entry , and we know that no event group after e - > largest_started_sub_id has started running yet . We record this value in e - > stop_sub_id , and then each event group can check their own sub_id against it . If their sub_id is strictly larger , then that event group will be skipped . If we stop due to reaching the START SLAVE UNTIL condition , then we need to continue executing any queued events up to that point . true ;for (j= 0 ; j < e->rpl_thread_max; ++j)if ((rpt= e->rpl_threads[j].thr))if (rpt->current_owner == &e->rpl_threads[j].thr)#ifdef ENABLED_DEBUG_SYNC"rpl_parallel_wait_for_done_trigger" ,"now SIGNAL wait_for_done_waiting" ));#endif /* Shutdown SA alter threads through marking their execution states to force their early post - SA execution exit . Upon that the affected SA threads change their state to COMPLETED , notify any waiting CA | RA and this thread . true ; // a sign to stop in concurrently coming in new SA:s while ((info= info_iterator++))if (info->state == start_alter_state::COMPLETED)continue ;// Any possible CA that is (will be) waiting will complete this ALTER instance true ;// notify SA:s // await SA in the COMPLETED state while (info->state == start_alter_state::ROLLBACK_ALTER)"rpl_slave_stop_CA_before_binlog" ,"now signal proceed_CA_1" ));for (i= 0 ; i < domain_hash.records; ++i)struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);for (j= 0 ; j < e->rpl_thread_max; ++j)if ((rpt= e->rpl_threads[j].thr))while (rpt->current_owner == &e->rpl_threads[j].thr)// Now that all threads are docked, remained alter states are safe to destroy while ((info= info_iterator++))false ;/* This function handles the case where the SQL driver thread reached the START SLAVE UNTIL position ; we stop queueing more events but continue processing remaining , already queued events ; then use executes manual STOP SLAVE ; then this function signals to worker threads that they should stop the processing of any remaining queued events . void struct rpl_parallel_entry *e;for (i= 0 ; i < domain_hash.records; ++i)struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);if (e->force_abort)bool Relay_log_info::are_sql_threads_caught_up()if (!sql_thread_caught_up)return false ;/* The SQL thread sets @ ref worker_threads_caught_up to ` false ` but not ` true ` . Therefore , this place needs to check if it can now be ` true ` . if (!worker_threads_caught_up && ( // No need to re-check if already `true`. // `nullptr` case true ; // Refresh return worker_threads_caught_up;int /* We only need to queue the server restart if we still have a thread working on a ( potentially partial ) event group . If the last thread we queued for has finished , then it cannot have any partial event group that needs aborting . Thus there is no need for the full complexity of choose_thread ( ) . We only need to check if we have a current worker thread , and queue for it if so . if (!thr)return 0 ;if (thr->current_owner != &cur_thr->thr)/* No active worker thread, so no need to queue the master restart. */ return 0 ;if (!(qev= thr->get_qev(fdev, 0 , rli)))return 1 ;this ;return 0 ;int /* The domain_hash is only accessed by the SQL driver thread , so it is safe to iterate over without a lock . for (i= 0 ; i < max_i; ++i)struct rpl_parallel_entry *e;int err= 0 ;struct rpl_parallel_entry *)my_hash_element(&domain_hash, i);while (e->current_sub_id > e->last_committed_sub_id)if (unlikely(thd->check_killed()))1 ;break ;if (err)return err;return 0 ;/* Handle seeing a GTID during slave restart in GTID mode . If we stopped with different replication domains having reached different positions in the relay log , we need to skip event groups in domains that are further progressed . Updates the state with the seen GTID , and returns true if this GTID should be skipped , false otherwise . bool if (likely(state->count() == 0 ) ||return false ;if (gtid->server_id == gtid_entry->gtid.server_id)if (gtid->seq_no >= seq_no)/* This domain has reached its start position . So remove it , so that further events will be processed normally . return gtid->seq_no <= seq_no;else return true ;/* This is used when we get an error during processing in do_event ( ) ; We will not queue any event to the thread , but we still need to wake it up to be sure that it will be returned to the pool . static void bool *did_enter_cond, PSI_stage_info *old_stage)/* do_event ( ) is executed by the sql_driver_thd thread . It ' s main purpose is to find a thread that can execute the query . @ retval 0 ok , event was accepted @ retval 1 error @ retval - 1 event should be executed serially , in the sql driver thread int enum Log_event_type typ;bool is_group_event;bool did_enter_cond= false ;"slave_crash_if_parallel_apply" , DBUG_SUICIDE(););/* Handle master log name change, seen in Rotate_log_event. */ if (unlikely(typ == ROTATE_EVENT))static_cast <Rotate_log_event *>(ev);/* Sanity check that we do not have a " real " rotate event inside an event group ( or what we think is inside ) . The rotate event is only written to the binlog by the master at the end of a binlog file , and event groups cannot span files . An artificial rotate event generated on - the - fly in the dump thread can still appear inside an event group . And event groups can span relay log file boundaries , so rotate events for the relay log can also appear inside event groups . if ((rev->server_id != global_system_variables.server_id ||1 );/* Execute queries non - parallel if slave_skip_counter is set , as it ' s is easier to skip queries in single threaded mode . if (rli->slave_skip_counter)return -1 ;/* Execute pre-10.0 event, which have no GTID, in single-threaded mode. */ if (unlikely(!current) && typ != GTID_EVENT &&return -1 ;/* Note: rli->data_lock is released by sql_delay_event(). */ if (sql_delay_event(ev, rli->sql_driver_thd, serial_rgi))/* If sql_delay_event ( ) returns non - zero , it means that the wait timed out due to slave stop . We should not queue the event in this case , it must not be applied yet . delete ev;return 1 ;if (unlikely(typ == FORMAT_DESCRIPTION_EVENT))static_cast <Format_description_log_event *>(ev);if (fdev->created)/* This format description event marks a new binlog after a master server restart . We are going to close all temporary tables to clean up any possible left - overs after a prior master crash . Thus we need to wait for all prior events to execute to completion , in case they need access to any of the temporary tables . We also need to notify the worker thread running the prior incomplete event group ( if any ) , as such event group signifies an incompletely written group cut short by a master crash , and must be rolled back . if (current->queue_master_restart(serial_rgi, fdev) ||delete ev;return 1 ;else if (unlikely(typ == GTID_LIST_EVENT))static_cast <Gtid_list_log_event *>(ev);while (count)/* Stop queueing additional event groups once the SQL thread is requested to stop . We have to queue any remaining events of any event group that has already been partially queued , but after that we will just ignore any further events the SQL driver thread may try to queue , and eventually it will stop . if ((typ == GTID_EVENT || !is_group_event) && rli->abort_slave)true ;if (sql_thread_stopping)delete ev;/* Return " no error " ; normal stop is not an error , and otherwise the error has already been recorded . return 0 ;if (unlikely(rli->gtid_skip_flag != GTID_SKIP_NOT) && is_group_event)if (typ == GTID_EVENT)else if (rli->gtid_skip_flag == GTID_SKIP_STANDALONE)if (!Log_event::is_part_of_group(typ))else if (typ == XID_EVENT || typ == XA_PREPARE_LOG_EVENT ||// COMMIT/ROLLBACK are never compressed return 0 ;if (typ == GTID_EVENT)static_cast <Gtid_log_event *>(ev);0 : gtid_ev->domain_id);if (!(e= find(domain_id, rli)))0 ));delete ev;return 1 ;1 );if (process_gtid_for_restart_pos(rli, >id))/* This domain has progressed further into the relay log before the last SQL thread restart . So we need to skip this event group to not doubly apply it . return 0 ;else /* Find a worker thread to queue the event for . Prefer a new thread , so we maximise parallelism ( at least for the group commit ) . But do not exceed a limit of - - slave - domain - parallel - threads ; instead re - use a thread that we queued for previously . if (!cur_thread)/* This means we were killed. The error is already signalled. */ delete ev;return 1 ;if (!(qev= cur_thread->get_qev(ev, event_size, rli)))delete ev;return 1 ;if (typ == GTID_EVENT)bool new_gco;enum rpl_group_info::enum_speculation speculation;if (!(rgi= cur_thread->get_rgi(rli, gtid_ev, e, event_size)))delete ev;return 1 ;/* We queue the event group in a new worker thread , to run in parallel with previous groups . To preserve commit order within the replication domain , we set up rgi - > wait_commit_sub_id to make the new group commit only after the previous group has committed . Event groups that group - committed together on the master can be run in parallel with each other without restrictions . But one batch of group - commits may not start before all groups in the previous batch have initiated their commit phase ; we set up rgi - > gco to ensure that . true ;0 ;if (likely(gco))if (mode <= SLAVE_PARALLEL_MINIMAL ||/* MULTI_BATCH is also set when the current gtid even being a member of a commit group is flagged as DDL which disallows parallel . /* Make sure we do not attempt to run DDL in parallel speculatively. */ if (gtid_flags & Gtid_log_event::FL_DDL)if (!(flags & group_commit_orderer::MULTI_BATCH))/* Still the same batch of event groups that group - committed together on the master , so we can run in parallel . false ;else if ((mode >= SLAVE_PARALLEL_OPTIMISTIC) &&/* In transactional parallel mode , we optimistically attempt to run non - DDL in parallel . In case of conflicts , we catch the conflict as a deadlock or other error , roll back and retry serially . The assumption is that only a few event groups will be non - transactional or otherwise unsuitable for parallel apply . Those transactions are still scheduled in parallel , but we set a flag that will make the worker thread wait for everything before to complete before starting . false ;if (!(gtid_flags & Gtid_log_event::FL_TRANSACTIONAL) ||/* This transaction should not be speculatively run in parallel with what came before , either because it cannot safely be rolled back in case of a conflict , or because it was marked as likely to conflict and require expensive rollback and retry . Here we mark it as such , and then the worker thread will do a wait_for_prior_commit ( ) before starting it . We do not introduce a new group_commit_orderer , since we still want following transactions to run in parallel with transactions prior to this one . else else if (gtid_flags & Gtid_log_event::FL_DDL)if (gtid_flags & Gtid_log_event::FL_GROUP_COMMIT_ID)else 0 ;if (new_gco)/* Do not run this event group in parallel with what came before ; instead wait for everything prior to at least have started its commit phase , to avoid any risk of performing any conflicting action too early . Remember the count that marks the end of the previous batch of event groups that run in parallel , and allocate a new gco . if (!(gco= cur_thread->get_gco(count, gco, e->current_sub_id)))delete ev;return 1 ;else if (!is_group_event)int err;bool tmp;/* Events like ROTATE and FORMAT_DESCRIPTION . Do not run in worker thread . Same for events not preceeded by GTID ( we should not see those normally , but they might be from an old master ) . true ;if (ev->is_relay_log_event())0 ;else if (typ == ROTATE_EVENT)static_cast <Rotate_log_event *>(ev))->pos;else if (err)return 1 ;/* Queue a position update , so that the position will be updated in a reasonable way relative to other events : - If the currently executing events are queued serially for a single thread , the position will only be updated when everything before has completed . - If we are executing multiple independent events in parallel , then at least the position will not be updated until one of them has reached the current point . else /* The original execution time of the event from the master is stored on the serial_rgi , so copy it to our new one for parallel execution . /* Queue the event for processing . return 0 ;
Messung V0.5 in Prozent C=96 H=81 G=88
¤ Dauer der Verarbeitung: 0.137 Sekunden
(vorverarbeitet am 2026-10-08)
¤ *© Formatika GbR, Deutschland
2026-10-09
Neuigkeiten Aktuelles Motto des Tages letze Version des Elbe Quellennavigators
Jenseits des Üblichen ....
Besucher
Statistik #Sources=1126438 #Domains=1867298
