重要数据结构

Rpl_info 的基类，保存了一些错误信息，如 IO/SQL thread last error

class Slave_reporting_capability
{

  // 获取last error
  Error const& last_error() const { return m_last_error; }

}


Master_info 、Relay_log_info 的基类，很多重要的锁和信号量都在这里


class Rpl_info : public Slave_reporting_capability
{
  /*
    为了避免死锁，需要按照以下顺序加锁
    run_lock, data_lock, relay_log.LOCK_log, relay_log.LOCK_index
    run_lock, sleep_lock
    run_lock, info_thd_lock
    
    info_thd_lock 保护对 info_thd 的操作
    读操作需要获取 info_thd_lock 或 run_lock
    写操作需要获取 info_thd_lock 和 run_lock
    
    data_lock : 保护对数据的读写
    run_lock  : 保护运行状态，变量 slave_running, slave_run_id
    sleep_lock: 对slave_sleep做互斥，防止同时进入多个slave_sleep
  */
  mysql_mutex_t data_lock, run_lock, sleep_lock, info_thd_lock;

  /*
    data_cond: data_lock 保护的数据被修改时发出此信号楼，只有 Relay_log_info 会使用
    start_cond: sql thread start (start slave 先启动 io thread，后启动 sql thread，sql thread 启动代表全部启动成功)
    stop_cond: sql/io thread stop
    sleep_cond: slave被kill，目前只发现5.6中有应用，5.7 中没找到发出此信号量的代码
  */
  mysql_cond_t data_cond, start_cond, stop_cond, sleep_cond;
  
  /* 
    关联 io/sql/worker thread 的 thd
    Master_info 对应 io thread
    Relay_log_info 对应 sql thread
    Slave_worker 对应 worker thread 
  */
  THD *info_thd;
  
  
  /* 已初始化 */
  bool inited;
  
  /* 已终止slave */
  volatile bool abort_slave;
  
  /*
    当前slave运行状态，io thread 即 mi->slave_running有三种状态
    MYSQL_SLAVE_NOT_RUN         0
	MYSQL_SLAVE_RUN_NOT_CONNECT 1
	MYSQL_SLAVE_RUN_CONNECT     2 
	
	sql thread 即 mi->slave_running 只有 0 1两种状态
  */
  volatile uint slave_running;
  
  /* 用于判断slave thread 是否启动的变量，每次启动时递增1 */
  volatile ulong slave_run_id;

  /* repository's handler */
  Rpl_info_handler *handler;
  
  /* 
    唯一标识一条 info 记录的 id，标识一行或一个文件
    Master_info 和 Relay_log_info 可以通过 channel 标识唯一
    Worker_info 需要通过 {id, channel} 标识唯一
  */
  uint internal_id;

  /* 通道名，多源复制使用 */
  char channel[CHANNEL_NAME_LENGTH+1];
  
  /* 实现了两个虚函数，读写 Rpl_info */
  virtual bool read_info(Rpl_info_handler *from)= 0;
  virtual bool write_info(Rpl_info_handler *to)= 0;
}

/*
  Master_info 用户 IO thread
  主要保存以下信息：
    连接到Master的用户信息
    当前 master log name
    当前 master log offset
    一些其他控制变量
    
  Master_info 读取 master.info repository 初始化，通常是表或文件
  通过函数 mi_init_info() 进行初始化
  
  调用 flush_info() 可以将 master.info 写入磁盘，每次从master读取数据都需要刷盘
*/
class Master_info : public Rpl_info
{
  /* 前面保存了 user、host 等连接信息 */
  /* 下面是连接信息的 set/get 函数 */

  /* 和 master 的连接 */
  MYSQL* mysql;
  
  
  /* 对应的 Relay_log_info */        
  Relay_log_info *rli;
  
  /* IO 线程复制延迟 */
  long clock_diff_with_master;
  
  /* 心跳间隔 */
  float heartbeat_period;         // interface with CHANGE MASTER or master.info
  
  /* 收到心跳次数 */
  ulonglong received_heartbeats;  // counter of received heartbeat events

  /* 上次心跳时间 */
  time_t last_heartbeat;

  /* 上次收到event的时间 */
  time_t last_recv_event_timestamp;

  /* 忽略复制的server_id */
  Server_ids *ignore_server_ids;

  /* master server_id */
  ulong master_id;
  
  /* FORMAT_DESCRIPTION_EVENT前的checksum 算法 */
  binary_log::enum_binlog_checksum_alg checksum_alg_before_fd;
  
  /* 初始化 Master_info, 里面调用read_info() 从 Rpl_info_handler 中读取信息 */
  int mi_init_info();
  
  /* 清理 Master_info，里面会调用 Rpl_info_handler 的 eng_info() */
  void end_info();
  
  /* Master_info 信息落盘，每次从主库读取数据后都会执行 */
  int flush_info(bool force= FALSE);
  
  /*
    从master收到的 Format_description_log_event 写在 relay log 末尾
    
    IO thread 开始时创建，IO thread 结束时销毁
    IO thread 收到一个 Format_description_log_event 时更新
    每次rotate时，IO thread 写Format_description_log_event到新relay log
    每次执行FLUSH LOGS，client 写Format_description_log_event到新relay log
  */
  Format_description_log_event *mi_description_event;

  /* 最近一个GTID，可能是未完成事务的GTID，用于事务结束时写入 Retrieved_Gtid_Set */
  Gtid last_gtid_queued;
  
  /* 用于判断事务边界 */
  Transaction_boundary_parser transaction_parser;
  
  /* 
    channel lock，以下操作需要持有写锁
    START SLAVE;
	STOP SLAVE;
	CHANGE MASTER;
	RESET SLAVE;
	end_slave();
  */
  Checkable_rwlock *m_channel_lock;
	
  /* channel 被引用的次数，只有为0时可以删除channel */
  Atomic_int32 references;
}

/*
  主要保存以下信息:
    当前relay log
    当前relay log offset
    master log name
    与上次更新对应的主库日志序列
    sql thread 其他信息

	初始化过程和 Master_info 类似
	
	以下情况下 relay.info table/file 需要更新：
	1. relay log file rotated
	2. SQL thread stopped
	3. while processing a Xid_log_event
	4. after a Query_log_event(commit or rollback)
	5. after processing any statement written to the binary log without a transaction context.
	
	并行复制相关代码留作以后分析，本次暂不涉及
*/
class Relay_log_info : public Rpl_info
{
  /* 备份状态标志位，用于标志是否在语句中 */
  enum enum_state_flag {
    /* 在语句中 */
    IN_STMT,

    /** Flag counter.  Should always be last */
    STATE_FLAGS_COUNT
  };
  
  /* 是否复制相同server_id的event，一般是false */
  bool replicate_same_server_id;
  
  /* 正在执行或最后一个执行的GTID，用来填充 performance_schema.replication_applier_status_by_worke 的 last_seen_transaction 列
  */
  Gtid_specification currently_executing_gtid;
  
  
  /* 读取下面变量时，必须受data_lock保护 */
  
  /* 当前relay log的文件描述符 */
  File cur_log_fd;
  
  /* reay_log 对象，MYSQL_BIN_LOG类留作以后分析*/
  MYSQL_BIN_LOG relay_log;
  
  /* 主要用于查询log_pos */
  LOG_INFO linfo;
  
  /*
    cur_log 指向 relay_log.get_log_file() 或者 cache_buf
    取决于relay_log_file是热日志，还是需要打开冷日志
    
    cache_buf 在打开冷日志时适用
  */
  IO_CACHE cache_buf,*cur_log;
  
  /* 标识是否正在recovery */
  bool is_relay_log_recovery;
  
  /* 下面的变量可以不加锁读 */
  
  /*
    restart slave 时需要访问临时表。
    这个变量值在 init/end 时修改
    SQL thread 只读
  */
  TABLE *save_temporary_tables;
  
  /* 对应的 Master_info */
  Master_info *mi;
  
  /* 打开临时表的数量 */
  Atomic_int32 channel_open_temp_tables;
  
  /* 保存 relay_log.get_open_count() */
  uint32 cur_log_old_open_count;
  
  /* init_info() 曾经失败，RESET SLAVE 可以修复错误 */
  bool error_on_rli_init_info;
  
  /* 这里跳过一些 Group replication 相关变量 */
  
  /* received gtid set */
  Gtid_set gtid_set;
  
  /* 标识此对象是否属于SQL线程（属于SQL线程为0） */
  bool rli_fake;
  
  /* 标识 retrieved GTID set 是否已被初始化 */
  bool gtid_retrieved_initialized;
  
  /* 上一个错误的GTID */
  Gtid last_sql_error_gtid;
  
  
  /* 日志空间限制，日志空间总量(用于sys_var:relay_log_space，控制relay log空间) */
  ulonglong log_space_limit,log_space_total;
  
  /* 是否忽略日志空间限制 */
  bool ignore_log_space_limit;

  
  /* 需要清理空间时，SQL线程指示IO线程rotate logs */
  bool sql_force_rotate_relay;
  
  /* 上次主库记录binlog的时间 */
  time_t last_master_timestamp;
  
  /* 上次执行event的时间 */
  time_t last_exec_event_timestamp;
  
  /* 跳过error event */
  volatile uint32 slave_skip_counter;
  
  /* 标记是否需要中断pos_wait，change master 和 reset slave时需要中断 */
  volatile ulong abort_pos_wait;
  
  /* log_space 相关信号量*/
  mysql_mutex_t log_space_lock;
  mysql_cond_t log_space_cond;

  /* 这里有一些 START SLAVE UNTIL 相关变量 */
  
  /* 重试事务次数(trans_retries是重试次数上限)，重试事务计数（retried_trans记录重试了多少次） */
  ulong trans_retries, retried_trans;
 
  /* 
    延迟复制时间
    CHANGE MASTER TO MASTER_DELAY=X.
    由data_lock保护， SQL thread 读取
    SQL thread 运行时该变量不可写
  */
  time_t sql_delay;
  
  /* sql_delay 结束时间 */
  time_t sql_delay_end;

  /* enum_state_flag 的标志位 */
  uint32 m_flags;  
}

函数分析

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

  case SQLCOM_CHANGE_MASTER:
  {
    if (check_global_access(thd, SUPER_ACL))
      goto error;
    res= change_master_cmd(thd);
    break;
  }
  
/*
  函数具备以下功能
    更改接收/执行日志的配置/位点
    purge relay log
    删除 worker info（并行复制使用）
*/

/* 分析 change_master 函数会略过一部分逻辑 */
int change_master(THD* thd, Master_info* mi, LEX_MASTER_INFO* lex_mi,
                  bool preserve_logs)
{
  /*
    如果SQL thread 和 IO thread已经停止，并且没有指定 relay_log_pos和relay_log_file
    会purge relay log
  */
  bool need_relay_log_purge= 1;

  /* 为了修改 mysql.slave_master_info，需要无视read_only和super_read_only */
  thd->set_skip_readonly_check();
  
  /* channel 加读写锁，即将对channel做修改，函数结束时才会释放锁 */
  mi->channel_wrlock();
  
  /*
    对 mi->run_lock 和 rli->run_lock 加锁
    防止线程运行状态发生变化
  */
  lock_slave_threads(mi);

  /* 设置thread_mask，用来标识 IO/SQL thread 的运行状态 */
  init_thread_mask(&thread_mask, mi, 0);

  /* 设置auto_position=1需要IO/SQL thread 都不在运行状态，否则报错退出 */
  if (thread_mask)
  {
    if (lex_mi->auto_position != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
    {
      error= ER_SLAVE_CHANNEL_MUST_STOP;
      my_error(ER_SLAVE_CHANNEL_MUST_STOP, MYF(0), mi->get_channel());
      goto err;
    }
    
    /* 如果 SQL thread 和 IO thread 没有全部停止，不能purge relay log */
    need_relay_log_purge= 0;
  }

  
  /*
    下面是一些错误判断，都是很明显的错误
      1. 如果设置了auto_position，同时又指定了复制位点，如 relay_log_pos，报错退出
      2. auto_position 需要 GTID_MODE != OFF
      3. IO thread 运行时不能改变 IO thread 相关配置
      4. SQL thread 运行时不能改变 SQL thread 相关配置
      5. 如果指定了master_host，那么master_host不能是空串  
  */

  /* 记录当前状态 */
  THD_STAGE_INFO(thd, stage_changing_master); 

  /* 标识停止的线程，给load_mi_and_rli_from_repositories()使用 */
  init_thread_mask(&thread_mask_stopped_threads, mi, 1);

  /* 
    从仓库加载 mi 和 rli 的配置
    只有停止状态的线程可以加载配置（SQL thread 对应 rli，IO thread 对应 mi）
  */
  if (load_mi_and_rli_from_repositories(mi, false, thread_mask_stopped_threads))                                                                                                                            
  {
    error= ER_MASTER_INFO;
    my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
    goto err;
  }

  /* 
    修改mi相关配置，并保存老配置
    save_ 变量中保存的老配置用于打印日志
  */
  if (have_receive_option)
  {
    strmake(saved_host, mi->host, HOSTNAME_LENGTH);
    strmake(saved_bind_addr, mi->bind_addr, HOSTNAME_LENGTH);
    saved_port= mi->port;
    strmake(saved_log_name, mi->get_master_log_name(), FN_REFLEN - 1);
    saved_log_pos= mi->get_master_log_pos();

    if ((error= change_receive_options(thd, lex_mi, mi)))                                                                                                                                                   
    {
      goto err;
    }
  }

  /* 打印日志，change master 的源值和目标值 */
  if (have_receive_option)
    sql_print_information("'CHANGE MASTER TO%s executed'. "
      "Previous state master_host='%s', master_port= %u, master_log_file='%s', "
      "master_log_pos= %ld, master_bind='%s'. "
      "New state master_host='%s', master_port= %u, master_log_file='%s', "
      "master_log_pos= %ld, master_bind='%s'.",
      mi->get_for_channel_str(true),
      saved_host, saved_port, saved_log_name, (ulong) saved_log_pos,
      saved_bind_addr, mi->host, mi->port, mi->get_master_log_name(),
      (ulong) mi->get_master_log_pos(), mi->bind_addr);

  /* 修改rli相关配置 */
  if (have_execute_option)
    change_execute_options(lex_mi, mi);
    
  /* 持久化master_info */
  if ((thread_mask & SLAVE_IO) == 0 && flush_master_info(mi, true))
  {
    error= ER_RELAY_LOG_INIT;
    my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
    goto err;
  }
  
  if ((thread_mask & SLAVE_SQL) == 0)
  {

    /* 记录全局变量 relay_log_purge */                                                                                                                                                            
    bool save_relay_log_purge= relay_log_purge;

    if (need_relay_log_purge)
    {
      const char* errmsg= 0;

      /* purge relay logs */
      relay_log_purge= 1;
      THD_STAGE_INFO(thd, stage_purging_old_relay_logs);
      if (mi->rli->purge_relay_logs(thd,
                                    0 /* not only reset, but also reinit */,
                                    &errmsg))
      {
        error= ER_RELAY_LOG_FAIL;
        my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
        goto err;
      }
    }
    else
    {

      const char* msg;
      relay_log_purge= 0;

      DBUG_ASSERT(mi->rli->inited);
      
      /*初始化 relay_log_pos */
      if (mi->rli->init_relay_log_pos(mi->rli->get_group_relay_log_name(),
                                      mi->rli->get_group_relay_log_pos(),
                                      true/*we do need mi->rli->data_lock*/,
                                      &msg, 0))
      {
        error= ER_RELAY_LOG_INIT;
        my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
        goto err;
      }
    }
    
    /* 恢复全局变量 relay_log_purge 的值 */
    relay_log_purge= save_relay_log_purge;

    /* 清理until condition */
    mi->rli->clear_until_condition();

    /* relay_log_info 持久化到磁盘 */
    if (mi->rli->flush_info(true))
    {
      error= ER_RELAY_LOG_INIT;
      my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush relay info file.");
      goto err;
    }

  }

/* 出错后跳到次数，释放之前申请的锁 */
err:

  unlock_slave_threads(mi);
  mi->channel_unlock();
  DBUG_RETURN(error);

}

总结

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

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

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