MySQL relay_log_recovery源码分析

此处所列代码为mysql8.0.23版本，其它版本可能略有差异。

relay_log_recovery按照官方文档描述，在server启动的时候，会创建新的relay_log，并将sql_thread指向新的relay_log坐标，并且引导io_thread将从master读取的event写入新的relay log中。从而跳过旧的relay log的apply，进而避免旧的relay log出现问题导致复制出现问题。

但是在MTS下，在出现复制gap的时候，仍然需要在旧的relay log上将gap补齐，之后，才会指向新relay log。如果当前GTID_MODE=ON && 启用了MASTER_AUTO_POSITION=1，那么不会计算gap，直接初始化并指向新的relay log。

主要的代码出现在三个函数中：
在load_mi_and_rli_from_repositories函数中，调用了

if (((thread_mask & SLAVE_SQL) != 0 || !(mi->rli->inited)) &&
mi->rli->rli_init_info(skip_received_gtid_set_recovery))

其中，在mi->rli->rli_init_info(skip_received_gtid_set_recovery)中的

if (is_relay_log_recovery && init_recovery(mi)) {
error = 1;
goto err;
}

部分，根据relay_log_recovery==True，进而调用了 init_recovery(mi)方法。在 init_recovery(mi)方法中，

根据下面的代码部分，如果当前GTID_MODE=ON && 启用了MASTER_AUTO_POSITION=1的情况下，跳过了gap的计算，直接进入到进入到初始化并指向新relay log的代码。也就是说，在这种情况下，不需要特别的补齐gap的处理，完全依赖slave启动后，通过从库与主库的gtid_executed变量差异，自动拉取缺失的gtid事务。

bool is_gtid_with_autopos_on =
(global_gtid_mode.get() == Gtid_mode::ON) && mi->is_auto_position();

。之后，进入到计算gap的过程，通过调用

error = mts_recovery_groups(rli)

计算了所有worker中是否有gap，以及需要恢复的事务量。

if (rli->mts_recovery_group_cnt) return error;

如果存在需要恢复的事务，则函数直接返回，跳过了后续的将SQL_THREAD和IO_THREAD指向新relay_log的方法调用：

if (run_relay_log_recovery) recover_relay_log(mi);

在 recover_relay_log(mi)方法中，完成将SQL_THREAD和IO_THREAD指向新relay_log。

最后，在load_mi_and_rli_from_repositories函数的末尾，调用了

if (!init_error && mi->rli->is_relay_log_recovery &&
mi->rli->mts_recovery_group_cnt)
init_error = fill_mts_gaps_and_recover(mi);

在fill_mts_gaps_and_recover(mi)中，通过

rli->until_condition = Relay_log_info::UNTIL_SQL_AFTER_MTS_GAPS;

的方式，并且通过

recovery_error = start_slave_thread(
key_thread_slave_sql, handle_slave_sql, &rli->run_lock, &rli->run_lock,
&rli->start_cond, &rli->slave_running, &rli->slave_run_id, mi);

启动SQL_THREAD线程，完成gap的恢复。之后，在fill_mts_gaps_and_recover函数接下来的部分，调用

recover_relay_log(mi);

将SQL_THREAD、IO_THREAD指向新的relay log。

到此，load_mi_and_rli_from_repositories函数就执行完了。后面的流程就是正常的启动slave线程进行复制。将IO_THREAD读取的event写入新的relay log，SQL_THREAD从新的relay log进行读取并apply。

int load_mi_and_rli_from_repositories(Master_info *mi, bool ignore_if_no_info,int thread_mask,bool skip_received_gtid_set_recovery) {DBUG_TRACE;DBUG_ASSERT(mi != nullptr && mi->rli != nullptr);int init_error = 0;enum_return_check check_return = ERROR_CHECKING_REPOSITORY;THD *thd = current_thd;/*We need a mutex while we are changing master info parameters tokeep other threads from reading bogus info*/mysql_mutex_lock(&mi->data_lock);mysql_mutex_lock(&mi->rli->data_lock);/*When info tables are used and autocommit= 0 we force a newtransaction start to avoid table access deadlocks when START SLAVEis executed after RESET SLAVE.*/if (is_autocommit_off_and_infotables(thd)) {if (trans_begin(thd)) {init_error = 1;goto end;}}/*This takes care of the startup dependency between the master_infoand relay_info. It initializes the master info if the SLAVE_IOthread is being started and the relay log info if either theSLAVE_SQL thread is being started or was not initialized as it isrequired by the SLAVE_IO thread.*/check_return = mi->check_info();if (check_return == ERROR_CHECKING_REPOSITORY) {init_error = 1;goto end;}if (!(ignore_if_no_info && check_return == REPOSITORY_DOES_NOT_EXIST)) {if ((thread_mask & SLAVE_IO) != 0 && mi->mi_init_info()) init_error = 1;}check_return = mi->rli->check_info();if (check_return == ERROR_CHECKING_REPOSITORY) {init_error = 1;goto end;}if (!(ignore_if_no_info && check_return == REPOSITORY_DOES_NOT_EXIST)) {if (((thread_mask & SLAVE_SQL) != 0 || !(mi->rli->inited)) &&mi->rli->rli_init_info(skip_received_gtid_set_recovery))init_error = 1;else {/*During rli_init_info() above, the relay log is opened (if rli was notinitialized yet). The function below expects the relay log to be openedto get its coordinates and store as the last flushed relay logcoordinates from I/O thread point of view.*/mi->update_flushed_relay_log_info();}}DBUG_EXECUTE_IF("enable_mts_worker_failure_init",{ DBUG_SET("+d,mts_worker_thread_init_fails"); });
end:/*When info tables are used and autocommit= 0 we force transactioncommit to avoid table access deadlocks when START SLAVE is executedafter RESET SLAVE.*/if (is_autocommit_off_and_infotables(thd))if (trans_commit(thd)) init_error = 1;mysql_mutex_unlock(&mi->rli->data_lock);mysql_mutex_unlock(&mi->data_lock);/*Handling MTS Relay-log recovery after successful initialization of mi andrli objects.MTS Relay-log recovery is handled by SSUG command. In order to start theslave applier thread rli needs to be inited and mi->rli->data_lock shouldbe in released state. Hence we do the MTS recovery at this point of timewhere both conditions are satisfied.*/if (!init_error && mi->rli->is_relay_log_recovery &&mi->rli->mts_recovery_group_cnt)init_error = fill_mts_gaps_and_recover(mi);return init_error;
}

int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {int error = 0;enum_return_check check_return = ERROR_CHECKING_REPOSITORY;const char *msg = nullptr;DBUG_TRACE;mysql_mutex_assert_owner(&data_lock);/*If Relay_log_info is issued again after a failed init_info(), forinstance because of missing relay log files, it will generate newfiles and ignore the previous failure, to avoid that we seterror_on_rli_init_info as true.This a consequence of the behaviour change, in the past server wasstopped when there were replication initialization errors, now it isnot and so init_info() must be aware of previous failures.*/if (error_on_rli_init_info) goto err;if (inited) {return recovery_parallel_workers ? mts_recovery_groups(this) : 0;}slave_skip_counter = 0;abort_pos_wait = 0;log_space_limit = relay_log_space_limit;log_space_total = 0;tables_to_lock = nullptr;tables_to_lock_count = 0;char pattern[FN_REFLEN];(void)my_realpath(pattern, slave_load_tmpdir, 0);/*@TODO:In MSR, sometimes slave fail with the following error:Unable to use slave's temporary directory /tmp -Can't create/write to file'/tmp/SQL_LOAD-92d1eee0-9de4-11e3-8874-68730ad50fcb'    (Errcode: 17 - Fileexists), Error_code: 1*/if (fn_format(pattern, PREFIX_SQL_LOAD, pattern, "",MY_SAFE_PATH | MY_RETURN_REAL_PATH) == NullS) {LogErr(ERROR_LEVEL, ER_SLAVE_CANT_USE_TEMPDIR, slave_load_tmpdir);return 1;}unpack_filename(slave_patternload_file, pattern);slave_patternload_file_size = strlen(slave_patternload_file);/*The relay log will now be opened, as a WRITE_CACHE IO_CACHE.Note that the I/O thread flushes it to disk after writing everyevent, in flush_info within the master info.*//*For the maximum log size, we choose max_relay_log_size if it isnon-zero, max_binlog_size otherwise. If later the user does SETGLOBAL on one of these variables, fix_max_binlog_size andfix_max_relay_log_size will reconsider the choice (for exampleif the user changes max_relay_log_size to zero, we have toswitch to using max_binlog_size for the relay log) and updaterelay_log.max_size (and mysql_bin_log.max_size).*/{/* Reports an error and returns, if the --relay-log's pathis a directory.*/if (opt_relay_logname &&opt_relay_logname[strlen(opt_relay_logname) - 1] == FN_LIBCHAR) {LogErr(ERROR_LEVEL, ER_RPL_RELAY_LOG_NEEDS_FILE_NOT_DIRECTORY,opt_relay_logname);return 1;}/* Reports an error and returns, if the --relay-log-index's pathis a directory.*/if (opt_relaylog_index_name &&opt_relaylog_index_name[strlen(opt_relaylog_index_name) - 1] ==FN_LIBCHAR) {LogErr(ERROR_LEVEL, ER_RPL_RELAY_LOG_INDEX_NEEDS_FILE_NOT_DIRECTORY,opt_relaylog_index_name);return 1;}char buf[FN_REFLEN];/* The base name of the relay log file considering multisource rep */const char *ln;/*relay log name without channel prefix taking into account--relay-log option.*/const char *ln_without_channel_name;static bool name_warning_sent = false;/*Buffer to add channel name suffix when relay-log option is provided.*/char relay_bin_channel[FN_REFLEN];/*Buffer to add channel name suffix when relay-log-index option is provided*/char relay_bin_index_channel[FN_REFLEN];/* name of the index file if opt_relaylog_index_name is set*/const char *log_index_name;ln_without_channel_name =relay_log.generate_name(opt_relay_logname, "-relay-bin", buf);ln = add_channel_to_relay_log_name(relay_bin_channel, FN_REFLEN,ln_without_channel_name);/* We send the warning only at startup, not after every RESET SLAVE */if (!opt_relay_logname_supplied && !opt_relaylog_index_name_supplied &&!name_warning_sent) {/*User didn't give us info to name the relay log index file.Picking `hostname`-relay-bin.index like we do, causes replication tofail if this slave's hostname is changed later. So, we would like toinstead require a name. But as we don't want to break many existingsetups, we only give warning, not error.*/LogErr(WARNING_LEVEL, ER_RPL_PLEASE_USE_OPTION_RELAY_LOG,ln_without_channel_name);name_warning_sent = true;}/*If relay log index option is set, convert into channel specificindex file. If the opt_relaylog_index has an extension, we stripit too. This is inconsistent to relay log names.*/if (opt_relaylog_index_name_supplied) {char index_file_withoutext[FN_REFLEN];relay_log.generate_name(opt_relaylog_index_name, "",index_file_withoutext);log_index_name = add_channel_to_relay_log_name(relay_bin_index_channel, FN_REFLEN, index_file_withoutext);} elselog_index_name = nullptr;if (relay_log.open_index_file(log_index_name, ln, true)) {LogErr(ERROR_LEVEL, ER_RPL_OPEN_INDEX_FILE_FAILED);return 1;}if (!gtid_retrieved_initialized) {/* Store the GTID of a transaction spanned in multiple relay log files */Gtid_monitoring_info *partial_trx = mi->get_gtid_monitoring_info();partial_trx->clear();
#ifndef DBUG_OFFget_sid_lock()->wrlock();gtid_set->dbug_print("set of GTIDs in relay log before initialization");get_sid_lock()->unlock();
#endif/*In the init_gtid_set below we pass the mi->transaction_parser.This will be useful to ensure that we only add a GTID tothe Retrieved_Gtid_Set for fully retrieved transactions. Also, it willbe useful to ensure the Retrieved_Gtid_Set behavior when autopositioning is disabled (we could have transactions spanning multiplerelay log files in this case).We will skip this initialization if relay_log_recovery is set in orderto save time, as neither the GTIDs nor the transaction_parser statewould be useful when the relay log will be cleaned up later when callinginit_recovery.*/if (!is_relay_log_recovery && !gtid_retrieved_initialized &&!skip_received_gtid_set_recovery &&relay_log.init_gtid_sets(gtid_set, nullptr, opt_slave_sql_verify_checksum,true /*true=need lock*/, &mi->transaction_parser, partial_trx)) {LogErr(ERROR_LEVEL, ER_RPL_CANT_INITIALIZE_GTID_SETS_IN_RLI_INIT_INFO);return 1;}gtid_retrieved_initialized = true;
#ifndef DBUG_OFFget_sid_lock()->wrlock();gtid_set->dbug_print("set of GTIDs in relay log after initialization");get_sid_lock()->unlock();
#endif}/*Configures what object is used by the current log to store processedgtid(s). This is necessary in the MYSQL_BIN_LOG::MYSQL_BIN_LOG tocorrectly compute the set of previous gtids.*/relay_log.set_previous_gtid_set_relaylog(gtid_set);/*note, that if open() fails, we'll still have index file openbut a destructor will take care of that*/mysql_mutex_t *log_lock = relay_log.get_log_lock();mysql_mutex_lock(log_lock);if (relay_log.open_binlog(ln, nullptr,(max_relay_log_size ? max_relay_log_size : max_binlog_size), true,true /*need_lock_index=true*/, true /*need_sid_lock=true*/,mi->get_mi_description_event())) {mysql_mutex_unlock(log_lock);LogErr(ERROR_LEVEL, ER_RPL_CANT_OPEN_LOG_IN_RLI_INIT_INFO);return 1;}mysql_mutex_unlock(log_lock);}/*This checks if the repository was created before and thus therewill be values to be read. Please, do not move this call afterthe handler->init_info().*/if ((check_return = check_info()) == ERROR_CHECKING_REPOSITORY) {msg = "Error checking relay log repository";error = 1;goto err;}if (handler->init_info()) {msg = "Error reading relay log configuration";error = 1;goto err;}check_return = check_if_info_was_cleared(check_return);if (check_return & REPOSITORY_EXISTS) {if (read_info(handler)) {msg = "Error reading relay log configuration";error = 1;goto err;}/*Clone required cleanup must be done only once, thence we onlydo it when server is booting.*/if (clone_startup && get_server_state() == SERVER_BOOTING) {char *channel_name =(const_cast<Relay_log_info *>(mi->rli))->get_channel();bool is_group_replication_channel =channel_map.is_group_replication_channel_name(channel_name);if (is_group_replication_channel) {if (clear_info()) {msg ="Error cleaning relay log configuration for group replication ""after clone";error = 1;goto err;}if (Rpl_info_factory::reset_workers(this)) {msg ="Error cleaning relay log worker configuration for group ""replication after clone";error = 1;goto err;}check_return = REPOSITORY_CLEARED;} else {if (!is_relay_log_recovery) {LogErr(WARNING_LEVEL, ER_RPL_RELAY_LOG_RECOVERY_INFO_AFTER_CLONE,channel_name);// After a clone if we detect information is present we always invoke// relay log recovery. Not doing so would probably mean failure at// initialization due to missing relay log files.if (init_recovery(mi)) {msg = "Error on the relay log recovery after a clone operation";error = 1;goto err;}}}}}if (check_return == REPOSITORY_DOES_NOT_EXIST ||  // Hasn't been initializedcheck_return == REPOSITORY_CLEARED  // Was initialized but was RESET) {/* Init relay log with first entry in the relay index file */if (reset_group_relay_log_pos(&msg)) {error = 1;goto err;}group_master_log_name[0] = 0;group_master_log_pos = 0;} else {if (is_relay_log_recovery && init_recovery(mi)) {error = 1;goto err;}if (is_group_relay_log_name_invalid(&msg)) {LogErr(ERROR_LEVEL, ER_RPL_MTS_RECOVERY_CANT_OPEN_RELAY_LOG,group_relay_log_name, std::to_string(group_relay_log_pos).c_str());error = 1;goto err;}}inited = true;error_on_rli_init_info = false;if (flush_info(true)) {msg = "Error reading relay log configuration";error = 1;goto err;}if (count_relay_log_space()) {msg = "Error counting relay log space";error = 1;goto err;}/*In case of MTS the recovery is deferred until the end ofload_mi_and_rli_from_repositories.*/if (!mi->rli->mts_recovery_group_cnt) is_relay_log_recovery = false;return error;err:handler->end_info();inited = false;error_on_rli_init_info = true;if (msg) LogErr(ERROR_LEVEL, ER_RPL_RLI_INIT_INFO_MSG, msg);relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT,true /*need_lock_log=true*/, true /*need_lock_index=true*/);return error;
}

int init_recovery(Master_info *mi) {DBUG_TRACE;int error = 0;Relay_log_info *rli = mi->rli;char *group_master_log_name = nullptr;/* Set the recovery_parallel_workers to 0 if Auto Position is enabled. */bool is_gtid_with_autopos_on =(global_gtid_mode.get() == Gtid_mode::ON) && mi->is_auto_position();if (is_gtid_with_autopos_on) rli->recovery_parallel_workers = 0;if (rli->recovery_parallel_workers) {/*This is not idempotent and a crash after this function and beforethe recovery is actually done may lead the system to an inconsistentstate.This may happen because the gap is not persitent stored anywhereand eventually old relay log files will be removed and furthercalculations on the gaps will be impossible.We need to improve this. /Alfranio.*/error = mts_recovery_groups(rli);if (rli->mts_recovery_group_cnt) return error;}group_master_log_name = const_cast<char *>(rli->get_group_master_log_name());if (!error) {bool run_relay_log_recovery = true;if (!group_master_log_name[0]) {if (rli->replicate_same_server_id) {error = 1;LogErr(ERROR_LEVEL,ER_RPL_RECOVERY_REPLICATE_SAME_SERVER_ID_REQUIRES_POSITION);return error;}error = find_first_relay_log_with_rotate_from_master(rli);if (error == 2) {// No events from the master on relay log - skip relay log recoveryrun_relay_log_recovery = false;error = 0;} else if (error)return error;}if (run_relay_log_recovery) recover_relay_log(mi);}return error;
}

static inline int fill_mts_gaps_and_recover(Master_info *mi) {DBUG_TRACE;Relay_log_info *rli = mi->rli;int recovery_error = 0;rli->is_relay_log_recovery = false;Until_mts_gap *until_mg = new Until_mts_gap(rli);rli->set_until_option(until_mg);rli->until_condition = Relay_log_info::UNTIL_SQL_AFTER_MTS_GAPS;until_mg->init();rli->channel_mts_submode = (mts_parallel_option == MTS_PARALLEL_TYPE_DB_NAME)? MTS_PARALLEL_TYPE_DB_NAME: MTS_PARALLEL_TYPE_LOGICAL_CLOCK;LogErr(INFORMATION_LEVEL, ER_RPL_MTS_RECOVERY_STARTING_COORDINATOR);recovery_error = start_slave_thread(key_thread_slave_sql, handle_slave_sql, &rli->run_lock, &rli->run_lock,&rli->start_cond, &rli->slave_running, &rli->slave_run_id, mi);if (recovery_error) {LogErr(WARNING_LEVEL, ER_RPL_MTS_RECOVERY_FAILED_TO_START_COORDINATOR);goto err;}mysql_mutex_lock(&rli->run_lock);mysql_cond_wait(&rli->stop_cond, &rli->run_lock);mysql_mutex_unlock(&rli->run_lock);if (rli->until_condition != Relay_log_info::UNTIL_DONE) {LogErr(WARNING_LEVEL, ER_RPL_MTS_AUTOMATIC_RECOVERY_FAILED);goto err;}rli->clear_until_option();/*We need a mutex while we are changing master info parameters tokeep other threads from reading bogus info*/mysql_mutex_lock(&mi->data_lock);mysql_mutex_lock(&rli->data_lock);recover_relay_log(mi);if (mi->flush_info(true) || rli->flush_info(true)) {recovery_error = 1;mysql_mutex_unlock(&mi->data_lock);mysql_mutex_unlock(&rli->data_lock);goto err;}rli->inited = true;rli->error_on_rli_init_info = false;mysql_mutex_unlock(&mi->data_lock);mysql_mutex_unlock(&rli->data_lock);LogErr(INFORMATION_LEVEL, ER_RPL_MTS_RECOVERY_SUCCESSFUL);return recovery_error;
err:/*If recovery failed means we failed to initialize rli object in the caseof MTS. We should not allow the START SLAVE command to work as we do inthe case of STS. i.e if init_recovery call fails then we set inited=0.*/rli->end_info();rli->inited = false;rli->error_on_rli_init_info = true;rli->clear_until_option();return recovery_error;
}

MySQL relay_log_recovery源码分析相关推荐

java mysql 源码分析_JAVA JDBC(MySQL)驱动源码分析
JAVA连接数据库是其众多功能中的一部分,主要有两种方式连接DataBase: 一种是采用JDBC-ODBC桥,另一种则是称之为纯驱动连接DataBase,第一种方式在大型项目中基本上不再使用,本系列 ...
mysql bulkupdate_django_bulk_update源码分析
## django_bulk_update源码分析这个第三方插件的体量几乎只相当于工作时两三天的代码量了,是一个比较容易开始进行源代码阅读的模块,阅读完这个代码对自定义的进行django拓展也是一个 ...
java jdbc(mysql)驱动源码分析_JAVA JDBC(MySQL)驱动源码分析(二)
本文系转载,地址:http://blog.csdn.net/brilliancezhou/article/details/5425687 上一篇中分析了Class.forName("com. ...
java jdbc(mysql)驱动源码分析,JAVA JDBC(MySQL)驱动源码分析(四)
connect方法是java.sql.Driver接口中定义的方法,如果连接的数据库不同,那么为不同的数据库编写JDBC驱动将变得很灵活,实现Driver接口即可.连接数据库时首先得装载JDBC驱动, ...
【MySQL】——mysql exporter源码分析
一.前言最近在做mysql innodb cluster的监控,发现mysql innodb cluster的集群状态没有对应的指标能采集到,索性看一波源码,魔改一把吧.源码链接奉上: https: ...
erlang mysql driver_erlang_mysql_driver 源码分析2
pool模型探究erlang_mysql_driver对同一时刻大量请求的支持 mysql:fetch 和 mysql_conn 今天看到网络上的一篇文章,说erlang_mysql_driver的 ...
mysql select源码分析_MYSQL源码分析(三)--Select语句
●MYSQL的查询语句起始于:mysql_execute_command(THD *thd),(sql_http://www.doczj.com/doc/2e2a5f295f0e7cd18425367 ...
pbp 读取 mysql数据_SqlAlchemy 中操作数据库时session和scoped_session的区别(源码分析)...
原生session: from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from sqlalch ...
《MySQL 8.0.22执行器源码分析（3.2）关于HashJoinIterator》
在本文章之前,应该了解的概念: 连接的一些概念.NLJ.BNL.HashJoin算法. 目录关于join连接 probe行保存概念 Hashjoin执行流程(十分重要) HashJoinIterat ...
mysql源码分析——InnoDB引擎启动分析
一.InnoDB启动在MySql中,InnoDB的启动流程其实是很重要的.一些更细节的问题,就藏在了这其中.在前面分析过整个数据库启动的流程,本篇就具体分析一下InnoDB引擎启动所做的各种动作.在 ...

MySQL relay_log_recovery源码分析

MySQL relay_log_recovery源码分析相关推荐

最新文章

热门文章