我的书签
添加书签
移除书签重要数据结构
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 做修改，需要注意在锁保护下修改变量，同时注意加锁顺序，避免死锁
