重要数据结构

image.png

  2. Rpl_info 的基类，保存了一些错误信息，如 IO/SQL thread last error
  4. class Slave_reporting_capability
  5. {
  7.   // 获取last error
  8.   Error const& last_error() const { return m_last_error; }
  10. }
  13. Master_info 、Relay_log_info 的基类，很多重要的锁和信号量都在这里
  16. class Rpl_info : public Slave_reporting_capability
  17. {
  18.   /*
  19.     为了避免死锁，需要按照以下顺序加锁
  20.     run_lock, data_lock, relay_log.LOCK_log, relay_log.LOCK_index
  21.     run_lock, sleep_lock
  22.     run_lock, info_thd_lock
  24.     info_thd_lock 保护对 info_thd 的操作
  25.     读操作需要获取 info_thd_lock 或 run_lock
  26.     写操作需要获取 info_thd_lock 和 run_lock
  28.     data_lock : 保护对数据的读写
  29.     run_lock  : 保护运行状态，变量 slave_running, slave_run_id
  30.     sleep_lock: 对slave_sleep做互斥，防止同时进入多个slave_sleep
  31.   */
  32.   mysql_mutex_t data_lock, run_lock, sleep_lock, info_thd_lock;
  34.   /*
  35.     data_cond: data_lock 保护的数据被修改时发出此信号楼，只有 Relay_log_info 会使用
  36.     start_cond: sql thread start (start slave 先启动 io thread，后启动 sql thread，sql thread 启动代表全部启动成功)
  37.     stop_cond: sql/io thread stop
  38.     sleep_cond: slave被kill，目前只发现5.6中有应用，5.7 中没找到发出此信号量的代码
  39.   */
  40.   mysql_cond_t data_cond, start_cond, stop_cond, sleep_cond;
  42.   /* 
  43.     关联 io/sql/worker thread 的 thd
  44.     Master_info 对应 io thread
  45.     Relay_log_info 对应 sql thread
  46.     Slave_worker 对应 worker thread 
  47.   */
  48.   THD *info_thd;
  51.   /* 已初始化 */
  52.   bool inited;
  54.   /* 已终止slave */
  55.   volatile bool abort_slave;
  57.   /*
  58.     当前slave运行状态，io thread 即 mi->slave_running有三种状态
  59.     MYSQL_SLAVE_NOT_RUN         0
  60.     MYSQL_SLAVE_RUN_NOT_CONNECT 1
  61.     MYSQL_SLAVE_RUN_CONNECT     2 
  63.     sql thread 即 mi->slave_running 只有 0 1两种状态
  64.   */
  65.   volatile uint slave_running;
  67.   /* 用于判断slave thread 是否启动的变量，每次启动时递增1 */
  68.   volatile ulong slave_run_id;
  70.   /* repository's handler */
  71.   Rpl_info_handler *handler;
  73.   /* 
  74.     唯一标识一条 info 记录的 id，标识一行或一个文件
  75.     Master_info 和 Relay_log_info 可以通过 channel 标识唯一
  76.     Worker_info 需要通过 {id, channel} 标识唯一
  77.   */
  78.   uint internal_id;
  80.   /* 通道名，多源复制使用 */
  81.   char channel[CHANNEL_NAME_LENGTH+1];
  83.   /* 实现了两个虚函数，读写 Rpl_info */
  84.   virtual bool read_info(Rpl_info_handler *from)= 0;
  85.   virtual bool write_info(Rpl_info_handler *to)= 0;
  86. }
  88. /*
  89.   Master_info 用户 IO thread
  90.   主要保存以下信息：
  91.     连接到Master的用户信息
  92.     当前 master log name
  93.     当前 master log offset
  94.     一些其他控制变量
  96.   Master_info 读取 master.info repository 初始化，通常是表或文件
  97.   通过函数 mi_init_info() 进行初始化
  99.   调用 flush_info() 可以将 master.info 写入磁盘，每次从master读取数据都需要刷盘
  100. */
  101. class Master_info : public Rpl_info
  102. {
  103.   /* 前面保存了 user、host 等连接信息 */
  104.   /* 下面是连接信息的 set/get 函数 */
  106.   /* 和 master 的连接 */
  107.   MYSQL* mysql;
  110.   /* 对应的 Relay_log_info */        
  111.   Relay_log_info *rli;
  113.   /* IO 线程复制延迟 */
  114.   long clock_diff_with_master;
  116.   /* 心跳间隔 */
  117.   float heartbeat_period;         // interface with CHANGE MASTER or master.info
  119.   /* 收到心跳次数 */
  120.   ulonglong received_heartbeats;  // counter of received heartbeat events
  122.   /* 上次心跳时间 */
  123.   time_t last_heartbeat;
  125.   /* 上次收到event的时间 */
  126.   time_t last_recv_event_timestamp;
  128.   /* 忽略复制的server_id */
  129.   Server_ids *ignore_server_ids;
  131.   /* master server_id */
  132.   ulong master_id;
  134.   /* FORMAT_DESCRIPTION_EVENT前的checksum 算法 */
  135.   binary_log::enum_binlog_checksum_alg checksum_alg_before_fd;
  137.   /* 初始化 Master_info, 里面调用read_info() 从 Rpl_info_handler 中读取信息 */
  138.   int mi_init_info();
  140.   /* 清理 Master_info，里面会调用 Rpl_info_handler 的 eng_info() */
  141.   void end_info();
  143.   /* Master_info 信息落盘，每次从主库读取数据后都会执行 */
  144.   int flush_info(bool force= FALSE);
  146.   /*
  147.     从master收到的 Format_description_log_event 写在 relay log 末尾
  149.     IO thread 开始时创建，IO thread 结束时销毁
  150.     IO thread 收到一个 Format_description_log_event 时更新
  151.     每次rotate时，IO thread 写Format_description_log_event到新relay log
  152.     每次执行FLUSH LOGS，client 写Format_description_log_event到新relay log
  153.   */
  154.   Format_description_log_event *mi_description_event;
  156.   /* 最近一个GTID，可能是未完成事务的GTID，用于事务结束时写入 Retrieved_Gtid_Set */
  157.   Gtid last_gtid_queued;
  159.   /* 用于判断事务边界 */
  160.   Transaction_boundary_parser transaction_parser;
  162.   /* 
  163.     channel lock，以下操作需要持有写锁
  164.     START SLAVE;
  165.     STOP SLAVE;
  166.     CHANGE MASTER;
  167.     RESET SLAVE;
  168.     end_slave();
  169.   */
  170.   Checkable_rwlock *m_channel_lock;
  172.   /* channel 被引用的次数，只有为0时可以删除channel */
  173.   Atomic_int32 references;
  174. }
  1. /*
  2.   主要保存以下信息:
  3.     当前relay log
  4.     当前relay log offset
  5.     master log name
  6.     与上次更新对应的主库日志序列
  7.     sql thread 其他信息
  9.     初始化过程和 Master_info 类似
  11.     以下情况下 relay.info table/file 需要更新：
  12.     1. relay log file rotated
  13.     2. SQL thread stopped
  14.     3. while processing a Xid_log_event
  15.     4. after a Query_log_event(commit or rollback)
  16.     5. after processing any statement written to the binary log without a transaction context.
  18.     并行复制相关代码留作以后分析，本次暂不涉及
  19. */
  20. class Relay_log_info : public Rpl_info
  21. {
  22.   /* 备份状态标志位，用于标志是否在语句中 */
  23.   enum enum_state_flag {
  24.     /* 在语句中 */
  25.     IN_STMT,
  27.     /** Flag counter.  Should always be last */
  28.     STATE_FLAGS_COUNT
  29.   };
  31.   /* 是否复制相同server_id的event，一般是false */
  32.   bool replicate_same_server_id;
  34.   /* 正在执行或最后一个执行的GTID，用来填充 performance_schema.replication_applier_status_by_worke 的 last_seen_transaction 列
  35.   */
  36.   Gtid_specification currently_executing_gtid;
  39.   /* 读取下面变量时，必须受data_lock保护 */
  41.   /* 当前relay log的文件描述符 */
  42.   File cur_log_fd;
  44.   /* reay_log 对象，MYSQL_BIN_LOG类留作以后分析*/
  45.   MYSQL_BIN_LOG relay_log;
  47.   /* 主要用于查询log_pos */
  48.   LOG_INFO linfo;
  50.   /*
  51.     cur_log 指向 relay_log.get_log_file() 或者 cache_buf
  52.     取决于relay_log_file是热日志，还是需要打开冷日志
  54.     cache_buf 在打开冷日志时适用
  55.   */
  56.   IO_CACHE cache_buf,*cur_log;
  58.   /* 标识是否正在recovery */
  59.   bool is_relay_log_recovery;
  61.   /* 下面的变量可以不加锁读 */
  63.   /*
  64.     restart slave 时需要访问临时表。
  65.     这个变量值在 init/end 时修改
  66.     SQL thread 只读
  67.   */
  68.   TABLE *save_temporary_tables;
  70.   /* 对应的 Master_info */
  71.   Master_info *mi;
  73.   /* 打开临时表的数量 */
  74.   Atomic_int32 channel_open_temp_tables;
  76.   /* 保存 relay_log.get_open_count() */
  77.   uint32 cur_log_old_open_count;
  79.   /* init_info() 曾经失败，RESET SLAVE 可以修复错误 */
  80.   bool error_on_rli_init_info;
  82.   /* 这里跳过一些 Group replication 相关变量 */
  84.   /* received gtid set */
  85.   Gtid_set gtid_set;
  87.   /* 标识此对象是否属于SQL线程（属于SQL线程为0） */
  88.   bool rli_fake;
  90.   /* 标识 retrieved GTID set 是否已被初始化 */
  91.   bool gtid_retrieved_initialized;
  93.   /* 上一个错误的GTID */
  94.   Gtid last_sql_error_gtid;
  97.   /* 日志空间限制，日志空间总量(用于sys_var:relay_log_space，控制relay log空间) */
  98.   ulonglong log_space_limit,log_space_total;
  100.   /* 是否忽略日志空间限制 */
  101.   bool ignore_log_space_limit;
  104.   /* 需要清理空间时，SQL线程指示IO线程rotate logs */
  105.   bool sql_force_rotate_relay;
  107.   /* 上次主库记录binlog的时间 */
  108.   time_t last_master_timestamp;
  110.   /* 上次执行event的时间 */
  111.   time_t last_exec_event_timestamp;
  113.   /* 跳过error event */
  114.   volatile uint32 slave_skip_counter;
  116.   /* 标记是否需要中断pos_wait，change master 和 reset slave时需要中断 */
  117.   volatile ulong abort_pos_wait;
  119.   /* log_space 相关信号量*/
  120.   mysql_mutex_t log_space_lock;
  121.   mysql_cond_t log_space_cond;
  123.   /* 这里有一些 START SLAVE UNTIL 相关变量 */
  125.   /* 重试事务次数(trans_retries是重试次数上限)，重试事务计数（retried_trans记录重试了多少次） */
  126.   ulong trans_retries, retried_trans;
  128.   /* 
  129.     延迟复制时间
  130.     CHANGE MASTER TO MASTER_DELAY=X.
  131.     由data_lock保护， SQL thread 读取
  132.     SQL thread 运行时该变量不可写
  133.   */
  134.   time_t sql_delay;
  136.   /* sql_delay 结束时间 */
  137.   time_t sql_delay_end;
  139.   /* enum_state_flag 的标志位 */
  140.   uint32 m_flags;  
  141. }

函数分析

mysql_execute_command() 当做入口开始分析，可以看出change_master需要SUPER权限

  2.   case SQLCOM_CHANGE_MASTER:
  3.   {
  4.     if (check_global_access(thd, SUPER_ACL))
  5.       goto error;
  6.     res= change_master_cmd(thd);
  7.     break;
  8.   }
  10. /*
  11.   函数具备以下功能
  12.     更改接收/执行日志的配置/位点
  13.     purge relay log
  14.     删除 worker info（并行复制使用）
  15. */
  17. /* 分析 change_master 函数会略过一部分逻辑 */
  18. int change_master(THD* thd, Master_info* mi, LEX_MASTER_INFO* lex_mi,
  19.                   bool preserve_logs)
  20. {
  21.   /*
  22.     如果SQL thread 和 IO thread已经停止，并且没有指定 relay_log_pos和relay_log_file
  23.     会purge relay log
  24.   */
  25.   bool need_relay_log_purge= 1;
  27.   /* 为了修改 mysql.slave_master_info，需要无视read_only和super_read_only */
  28.   thd->set_skip_readonly_check();
  30.   /* channel 加读写锁，即将对channel做修改，函数结束时才会释放锁 */
  31.   mi->channel_wrlock();
  33.   /*
  34.     对 mi->run_lock 和 rli->run_lock 加锁
  35.     防止线程运行状态发生变化
  36.   */
  37.   lock_slave_threads(mi);
  39.   /* 设置thread_mask，用来标识 IO/SQL thread 的运行状态 */
  40.   init_thread_mask(&thread_mask, mi, 0);
  42.   /* 设置auto_position=1需要IO/SQL thread 都不在运行状态，否则报错退出 */
  43.   if (thread_mask)
  44.   {
  45.     if (lex_mi->auto_position != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
  46.     {
  47.       error= ER_SLAVE_CHANNEL_MUST_STOP;
  48.       my_error(ER_SLAVE_CHANNEL_MUST_STOP, MYF(0), mi->get_channel());
  49.       goto err;
  50.     }
  52.     /* 如果 SQL thread 和 IO thread 没有全部停止，不能purge relay log */
  53.     need_relay_log_purge= 0;
  54.   }
  57.   /*
  58.     下面是一些错误判断，都是很明显的错误
  59.       1. 如果设置了auto_position，同时又指定了复制位点，如 relay_log_pos，报错退出
  60.       2. auto_position 需要 GTID_MODE != OFF
  61.       3. IO thread 运行时不能改变 IO thread 相关配置
  62.       4. SQL thread 运行时不能改变 SQL thread 相关配置
  63.       5. 如果指定了master_host，那么master_host不能是空串  
  64.   */
  66.   /* 记录当前状态 */
  67.   THD_STAGE_INFO(thd, stage_changing_master); 
  69.   /* 标识停止的线程，给load_mi_and_rli_from_repositories()使用 */
  70.   init_thread_mask(&thread_mask_stopped_threads, mi, 1);
  72.   /* 
  73.     从仓库加载 mi 和 rli 的配置
  74.     只有停止状态的线程可以加载配置（SQL thread 对应 rli，IO thread 对应 mi）
  75.   */
  76.   if (load_mi_and_rli_from_repositories(mi, false, thread_mask_stopped_threads))                                                                                                                            
  77.   {
  78.     error= ER_MASTER_INFO;
  79.     my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
  80.     goto err;
  81.   }
  83.   /* 
  84.     修改mi相关配置，并保存老配置
  85.     save_ 变量中保存的老配置用于打印日志
  86.   */
  87.   if (have_receive_option)
  88.   {
  89.     strmake(saved_host, mi->host, HOSTNAME_LENGTH);
  90.     strmake(saved_bind_addr, mi->bind_addr, HOSTNAME_LENGTH);
  91.     saved_port= mi->port;
  92.     strmake(saved_log_name, mi->get_master_log_name(), FN_REFLEN - 1);
  93.     saved_log_pos= mi->get_master_log_pos();
  95.     if ((error= change_receive_options(thd, lex_mi, mi)))                                                                                                                                                   
  96.     {
  97.       goto err;
  98.     }
  99.   }
  101.   /* 打印日志，change master 的源值和目标值 */
  102.   if (have_receive_option)
  103.     sql_print_information("'CHANGE MASTER TO%s executed'. "
  104.       "Previous state master_host='%s', master_port= %u, master_log_file='%s', "
  105.       "master_log_pos= %ld, master_bind='%s'. "
  106.       "New state master_host='%s', master_port= %u, master_log_file='%s', "
  107.       "master_log_pos= %ld, master_bind='%s'.",
  108.       mi->get_for_channel_str(true),
  109.       saved_host, saved_port, saved_log_name, (ulong) saved_log_pos,
  110.       saved_bind_addr, mi->host, mi->port, mi->get_master_log_name(),
  111.       (ulong) mi->get_master_log_pos(), mi->bind_addr);
  113.   /* 修改rli相关配置 */
  114.   if (have_execute_option)
  115.     change_execute_options(lex_mi, mi);
  117.   /* 持久化master_info */
  118.   if ((thread_mask & SLAVE_IO) == 0 && flush_master_info(mi, true))
  119.   {
  120.     error= ER_RELAY_LOG_INIT;
  121.     my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
  122.     goto err;
  123.   }
  125.   if ((thread_mask & SLAVE_SQL) == 0)
  126.   {
  128.     /* 记录全局变量 relay_log_purge */                                                                                                                                                            
  129.     bool save_relay_log_purge= relay_log_purge;
  131.     if (need_relay_log_purge)
  132.     {
  133.       const char* errmsg= 0;
  135.       /* purge relay logs */
  136.       relay_log_purge= 1;
  137.       THD_STAGE_INFO(thd, stage_purging_old_relay_logs);
  138.       if (mi->rli->purge_relay_logs(thd,
  139.                                     0 /* not only reset, but also reinit */,
  140.                                     &errmsg))
  141.       {
  142.         error= ER_RELAY_LOG_FAIL;
  143.         my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
  144.         goto err;
  145.       }
  146.     }
  147.     else
  148.     {
  150.       const char* msg;
  151.       relay_log_purge= 0;
  153.       DBUG_ASSERT(mi->rli->inited);
  155.       /*初始化 relay_log_pos */
  156.       if (mi->rli->init_relay_log_pos(mi->rli->get_group_relay_log_name(),
  157.                                       mi->rli->get_group_relay_log_pos(),
  158.                                       true/*we do need mi->rli->data_lock*/,
  159.                                       &msg, 0))
  160.       {
  161.         error= ER_RELAY_LOG_INIT;
  162.         my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
  163.         goto err;
  164.       }
  165.     }
  167.     /* 恢复全局变量 relay_log_purge 的值 */
  168.     relay_log_purge= save_relay_log_purge;
  170.     /* 清理until condition */
  171.     mi->rli->clear_until_condition();
  173.     /* relay_log_info 持久化到磁盘 */
  174.     if (mi->rli->flush_info(true))
  175.     {
  176.       error= ER_RELAY_LOG_INIT;
  177.       my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush relay info file.");
  178.       goto err;
  179.     }
  181.   }
  183. /* 出错后跳到次数，释放之前申请的锁 */
  184. err:
  186.   unlock_slave_threads(mi);
  187.   mi->channel_unlock();
  188.   DBUG_RETURN(error);
  190. }

总结

change master 主要功能是修改 SQL 和 IO 线程的配置信息，执行时可能会purge relay log

没有特殊情况，建议指定auto_position=1，不要自己指定复制位点，避免数据丢失风险

如需对change master 做修改，需要注意在锁保护下修改变量，同时注意加锁顺序，避免死锁