MySQL semi-sync 实现原理分析

after_commit--> repl_semi_report_commit--> ReplSemiSyncMaster::commitTrx --> cond_timewait
after_rollback --> repl_semi_report_rollback --> ReplSemiSyncMaster::commitTrx --> cond_timewait

after_flush --> repl_semi_report_binlog_update --> repl_semisync.writeTranxInBinlog --> (active_tranxs_->insert_tranx_node)
after_sync --> repl_semi_report_binlog_sync --> ReplSemiSyncMaster::commitTrx --> cond_timewait

start --> repl_semi_binlog_dump_start --> repl_semisync.add_slave
stop --> repl_semi_binlog_dump_end --> repl_semisync.remove_slave
reserve_header --> repl_semi_reserve_header --> repl_semisync.reserveSyncHeader
before_send_event --> repl_semi_before_send_event --> repl_semisync.updateSyncHeader
after_send_event --> repl_semi_after_send_event --> repl_semisync.readSlaveReply --> ReplSemiSyncMaster.reportReplyBinlog --> cond_broadcast
before_reset_master --> repl_semi_before_reset_master --> repl_semisync.beforeResetMaster
after_reset_master --> repl_semi_reset_master --> repl_semisync.resetMaster

start --> repl_semi_slave_io_start --> repl_semisync.slaveStart
stop --> repl_semi_slave_io_end --> repl_semisync.stopStart
request_transmit --> repl_semi_slave_request_dump
after_read_event --> repl_semi_slave_read_event
after_queue_event --> repl_semi_slave_queue_event
reset --> repl_semi_reset_slave

 
MYSQL_BIN_LOG::ordered_commit
// 进入flush阶段
|--> change_stage(thd, Stage_manager::FLUSH_STAGE, thd, NULL, &LOCK_log)
|--> if follower 
   |--> MYSQL_BIN_LOG::finish_commit 
   |--> RUN_HOOK(transaction, after_commit, (head, all))
   |--> RETURN
 
|--> process_flush_stage_queue
// 执行write操作
|--> flush_cache_to_file
 
// 添加到活跃事务链表中
|--> RUN_HOOK(binlog_storage, after_flush, (thd, file_name_ptr, flush_end_pos))
 
// 通知所有binglog dump线程复制binlog到备库
|--> signal_update
 
// 执行fsync将数据刷盘
|--> sync_binlog_file
 
// 进入sync阶段
|--> change_stage(thd, Stage_manager::SYNC_STAGE, wait_queue, &LOCK_log, &LOCK_sync)
|--> if follower 
   |--> MYSQL_BIN_LOG::finish_commit 
   |--> RUN_HOOK(transaction, after_commit, (head, all))
   |--> RETURN
 
// 进入commit阶段
|--> change_stage(thd, Stage_manager::COMMIT_STAGE, final_queue, &LOCK_sync, &LOCK_commit)
|--> if follower 
   |--> MYSQL_BIN_LOG::finish_commit 
      |--> RUN_HOOK(transaction, after_commit, (head, all))
   |--> RETURN
 
process_after_commit_stage_queue --> RUN_HOOK(transaction, after_commit, (head, all))
 
|--> MYSQL_BIN_LOG::finish_commit 
   // 等待备库ack收到
   |--> RUN_HOOK(transaction, after_commit, (head, all))

mysql_binlog_send
 
// 初始化semi-sync
|--> RUN_HOOK(binlog_transmit, transmit_start, (thd, flags, log_ident, pos))
 
while loop
|--> mysql_bin_log.wait_for_update_bin_log(thd, heartbeat_ts)
|--> reset_transmit_packet --> RUN_HOOK(binlog_transmit, reserve_header, (thd, flags, packet))
// 设置binlog头部，要求备库ack
|--> RUN_HOOK(binlog_transmit, before_send_event, (thd, flags, packet, log_file_name, pos))
// 发送binlog到备库
... my_net_write(net, (uchar*) packet->ptr(), packet->length())
// 等待读取备库ack，唤醒等待ack的用户线程，从活跃事务链表中清除
|--> RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet))
end while
 
// dump停止
|--> RUN_HOOK(binlog_transmit, transmit_stop, (thd, flags))

handle_slave_io
|--> RUN_HOOK(binlog_relay_io, thread_start, (thd, mi))
|--> register_slave_on_master(mysql, mi, &suppress_warnings)
 
while (!io_slave_killed(thd,mi))
|--> request_dump
   |--> RUN_HOOK(binlog_relay_io, before_request_transmit, (thd, mi, binlog_flags))
|--> read_event
// 读取semi-sync头部并进行校验
|--> RUN_HOOK(binlog_relay_io, after_read_event,
                   (thd, mi,(const char*)mysql->net.read_pos + 1,
                    event_len, &event_buf, &event_len))
|--> queue_event
// 回复主库ack
|--> RUN_HOOK(binlog_relay_io, after_queue_event,
                   (thd, mi, event_buf, event_len, synced))
end while
 
RUN_HOOK(binlog_relay_io, thread_stop, (thd, mi))