MySQL的commit是如何commit源码解析

int ha_commit_trans(THD *thd, bool all, bool ignore_global_read_lock) {
  /*
    Save transaction owned gtid into table before transaction prepare
    if binlog is disabled, or binlog is enabled and log_replica_updates
    is disabled with slave SQL thread or slave worker thread.
  */
  std::tie(error, need_clear_owned_gtid) = commit_owned_gtids(thd, all);
...
  // Prepare 阶段
  if (!trn_ctx->no_2pc(trx_scope) && (trn_ctx->rw_ha_count(trx_scope) > 1))
    error = tc_log->prepare(thd, all);
...
  // Commit 阶段
 if (error || (error = tc_log->commit(thd, all))) {
    ha_rollback_trans(thd, all);
    error = 1;
    goto end;
  }
}

|mysql_execute_command
|--trans_commit
|----ha_commit_trans
|------MYSQL_BIN_LOG::prepare
// 开启 binlog prepare 和 innodb prepare
|--------ha_prepare_low       
// Binlog prepare：获取上一个事务最大的 sequence number 时间戳
|----------binlog_prepare   
// innodb prepare
|----------innobase_xa_prepare            　　　　
|------------trx_prepare_for_mysql
// 1. 调用 trx_prepare_low 
// 2. 事务状态设置为Prepared 
// 3. 释放 RC 及以下隔离级别的 GAP Lock 
// 4. 刷盘 Redo（已推迟到 Commit 阶段的 Flush stage）
|--------------trx_prepare                        
|----------------trx_prepare_low
// 1. 将 undo log segment 的状态从 TRX_UNDO_ACTIVE 修改为 TRX_UNDO_PREPARED 
// 2. undo log 写入事务 XID
|------------------trx_undo_set_state_at_prepare

int MYSQL_BIN_LOG::ordered_commit(THD *thd, bool all, bool skip_commit) {
  /*
    Stage #0: 保证从实例的 SQL 线程按照 Relay log 的事务顺序进行提交
  */
  if (Commit_order_manager::wait_for_its_turn_before_flush_stage(thd) ||
      ending_trans(thd, all) ||
      Commit_order_manager::get_rollback_status(thd)) {
    if (Commit_order_manager::wait(thd)) {
      return thd->commit_error;
    }
  }
  /*
    Stage #1: flushing transactions to binary log
    While flushing, we allow new threads to enter and will process
    them in due time. Once the queue was empty, we cannot reap
    anything more since it is possible that a thread entered and
    appointed itself leader for the flush phase.
  */
  if (change_stage(thd, Commit_stage_manager::BINLOG_FLUSH_STAGE, thd, nullptr,
                   &LOCK_log)) {
    DBUG_PRINT("return", ("Thread ID: %u, commit_error: %d", thd->thread_id(),
                          thd->commit_error));
    return finish_commit(thd);
  }
  THD *wait_queue = nullptr, *final_queue = nullptr;
  mysql_mutex_t *leave_mutex_before_commit_stage = nullptr;
  my_off_t flush_end_pos = 0;
  bool update_binlog_end_pos_after_sync;
  // Flush 阶段主要的处理逻辑
  flush_error =
      process_flush_stage_queue(&total_bytes, &do_rotate, &wait_queue);
  if (flush_error == 0 && total_bytes > 0)
    /*
      flush binlog cache到file cache
    */
    flush_error = flush_cache_to_file(&flush_end_pos);
  // 后面根据 sync_binlog 参数决定更新 binlog pos 的位置并广播 Binlog 更新信号
  update_binlog_end_pos_after_sync = (get_sync_period() == 1);
  /*
    If the flush finished successfully, we can call the after_flush
    hook. Being invoked here, we have the guarantee that the hook is
    executed before the before/after_send_hooks on the dump thread
    preventing race conditions among these plug-ins.
  */
  if (flush_error == 0) {
    const char *file_name_ptr = log_file_name + dirname_length(log_file_name);
    assert(flush_end_pos != 0);
    /*
      插桩调用 after_flush，将已经 flush 的 binlog file 和 position 注册到半同步复制插件中，
      用于后续对比 slave 应答接受到的 binlog position。
    */
    if (RUN_HOOK(binlog_storage, after_flush,
                 (thd, file_name_ptr, flush_end_pos))) {
      LogErr(ERROR_LEVEL, ER_BINLOG_FAILED_TO_RUN_AFTER_FLUSH_HOOK);
      flush_error = ER_ERROR_ON_WRITE;
    }
    // 如果 sync_binlog!=1，在 flush stage 更新 binlog 位点并广播 update 信号，从库的 Dump 线程可以由此感知 Binlog 的更新
    if (!update_binlog_end_pos_after_sync) update_binlog_end_pos();
  }

int MYSQL_BIN_LOG::process_flush_stage_queue(my_off_t *total_bytes_var,
                                             bool *rotate_var,
                                             THD **out_queue_var) {
  int no_flushes = 0;
  my_off_t total_bytes = 0;
  mysql_mutex_assert_owner(&LOCK_log);
  // 根据 innodb_flush_log_at_trx_commit 参数进行 redo log 的刷盘操作
  THD *first_seen = fetch_and_process_flush_stage_queue();
  // 调用 get_server_sidno() 和 Gtid_state::get_automatic_gno 生成 GTID
  assign_automatic_gtids_to_flush_group(first_seen);
  /* Flush thread caches to binary log. */
  for (THD *head = first_seen; head; head = head->next_to_commit) {
    Thd_backup_and_restore switch_thd(current_thd, head);
    /*
      flush binlog_cache_mngr 的 stmt_cache和trx_cache。
      flush trx_cache：
        - 生成 last_committed 和 sequence_number
        - flush GTID log event
        - 将 trx_cache 中的数据 flush 到 binlog cache 中
        - 准备提交事务后的 Binlog pos
        - 递增 prepread XID
    */
    std::pair<int, my_off_t> result = flush_thread_caches(head);
    total_bytes += result.second;
    if (flush_error == 1) flush_error = result.first;
#ifndef NDEBUG
    no_flushes++;
#endif
  }
  *out_queue_var = first_seen;
  *total_bytes_var = total_bytes;
  if (total_bytes > 0 &&
      (m_binlog_file->get_real_file_size() >= (my_off_t)max_size ||
       DBUG_EVALUATE_IF("simulate_max_binlog_size", true, false)))
    *rotate_var = true;
#ifndef NDEBUG
  DBUG_PRINT("info", ("no_flushes:= %d", no_flushes));
  no_flushes = 0;
#endif
  return flush_error;
}

// 获取并清空 BINLOG_FLUSH_STAGE 和 COMMIT_ORDER_FLUSH_STAGE 队列，flush 事务到磁盘；不再阻塞 slave 的 preservecommitorder 的执行
|fetch_and_process_flush_stage_queue  
// 存储引擎层将 prepare 状态的 redolog 根据 innodb_flush_log_at_trx_commit 参数刷盘
|--ha_flush_logs                      
|----innobase_flush_logs
|------log_buffer_flush_to_disk

/*
  Stage #2: Syncing binary log file to disk
*/
if (change_stage(thd, Commit_stage_manager::SYNC_STAGE, wait_queue, &LOCK_log,
                 &LOCK_sync)) {
  DBUG_PRINT("return", ("Thread ID: %u, commit_error: %d", thd->thread_id(),
                        thd->commit_error));
  return finish_commit(thd);
}
/*
  - sync_counter：commit group的数量
  - get_sync_period()：获取sync_binlog参数的值
  - 如果sync stage队列中的commit group大于等于sync_binlog的值，当前leader就调用fsync()进行刷盘操作（sync_binlog_file(false)），
    在sync之前可能会进行等待，等待更多的commit group入队，等待的时间为binlog_group_commit_sync_no_delay_count或binlog_group_commit_sync_delay，默认都为0。
  - 如果sync stage队列中的commit group小于sync_binlog的值，当前leader不会调用fsync()进行刷盘也不会等待
  - 如果sync_binlog为0，每个commit group都会触发等待动作，但是不会sync
  - 如果sync_binlog为1，每个commit group都会触发等待动作，且会sync
*/
if (!flush_error && (sync_counter + 1 >= get_sync_period()))
  Commit_stage_manager::get_instance().wait_count_or_timeout(
      opt_binlog_group_commit_sync_no_delay_count,
      opt_binlog_group_commit_sync_delay, Commit_stage_manager::SYNC_STAGE);
final_queue = Commit_stage_manager::get_instance().fetch_queue_acquire_lock(
    Commit_stage_manager::SYNC_STAGE);
if (flush_error == 0 && total_bytes > 0) {
  DEBUG_SYNC(thd, "before_sync_binlog_file");
  std::pair<bool, bool> result = sync_binlog_file(false);
  sync_error = result.first;
}
/*
 如果sync_binlog==1,在sync stage阶段更新binog位点，并广播update信号，从库的Dump线程可以由此感知Binlog的更新
 （位点在flush stage中的process_flush_stage_queue()
                       |--flush_thread_caches()
                       |-----set_trans_pos()函数中设置）
*/
if (update_binlog_end_pos_after_sync && flush_error == 0 && sync_error == 0) {
  THD *tmp_thd = final_queue;
  const char *binlog_file = nullptr;
  my_off_t pos = 0;
  while (tmp_thd != nullptr) {
    if (tmp_thd->commit_error == THD::CE_NONE) {
      tmp_thd->get_trans_fixed_pos(&binlog_file, &pos);
    }
    tmp_thd = tmp_thd->next_to_commit;
  }
  if (binlog_file != nullptr && pos > 0) {
    update_binlog_end_pos(binlog_file, pos);
  }
}
DEBUG_SYNC(thd, "bgc_after_sync_stage_before_commit_stage");
leave_mutex_before_commit_stage = &LOCK_sync;

  /*
    Stage #3: Commit all transactions in order.
  */
commit_stage:
  /* binlog_order_commits：是否进行 order commit，即保持 redo 和 binlog 的提交顺序一致 */
  if ((opt_binlog_order_commits || Clone_handler::need_commit_order()) &&
      (sync_error == 0 || binlog_error_action != ABORT_SERVER)) {
    if (change_stage(thd, Commit_stage_manager::COMMIT_STAGE, final_queue,
                     leave_mutex_before_commit_stage, &LOCK_commit)) {
      DBUG_PRINT("return", ("Thread ID: %u, commit_error: %d", thd->thread_id(),
                            thd->commit_error));
      return finish_commit(thd);
    }
    THD *commit_queue =
        Commit_stage_manager::get_instance().fetch_queue_acquire_lock(
            Commit_stage_manager::COMMIT_STAGE);
    DBUG_EXECUTE_IF("semi_sync_3-way_deadlock",
                    DEBUG_SYNC(thd, "before_process_commit_stage_queue"););
    if (flush_error == 0 && sync_error == 0)
      /* after_sync hook */
      sync_error = call_after_sync_hook(commit_queue);
    /*
      Commit 阶段的主要处理逻辑
    */
    process_commit_stage_queue(thd, commit_queue);
    /**
     * After commit stage
     */
    if (change_stage(thd, Commit_stage_manager::AFTER_COMMIT_STAGE,
                     commit_queue, &LOCK_commit, &LOCK_after_commit)) {
      DBUG_PRINT("return", ("Thread ID: %u, commit_error: %d", thd->thread_id(),
                            thd->commit_error));
      return finish_commit(thd);
    }
    THD *after_commit_queue =
        Commit_stage_manager::get_instance().fetch_queue_acquire_lock(
            Commit_stage_manager::AFTER_COMMIT_STAGE);
    /* after_commit hook */
    process_after_commit_stage_queue(thd, after_commit_queue);
    final_queue = after_commit_queue;
    mysql_mutex_unlock(&LOCK_after_commit);
  } else {
    if (leave_mutex_before_commit_stage)
      mysql_mutex_unlock(leave_mutex_before_commit_stage);
    if (flush_error == 0 && sync_error == 0)
      sync_error = call_after_sync_hook(final_queue);
  }
  /* 广播 m_stage_cond_binlog 信号变量，唤醒挂起的 follower */
  Commit_stage_manager::get_instance().signal_done(final_queue);
  DBUG_EXECUTE_IF("block_leader_after_delete", {
    const char action[] = "now SIGNAL leader_proceed";
    assert(!debug_sync_set_action(thd, STRING_WITH_LEN(action)));
  };);
  /*
    Finish the commit before executing a rotate, or run the risk of a
    deadlock. We don't need the return value here since it is in
    thd->commit_error, which is returned below.
  */
  (void)finish_commit(thd);
  DEBUG_SYNC(thd, "bgc_after_commit_stage_before_rotation");
  return thd->commit_error == THD::CE_COMMIT_ERROR;
}

void MYSQL_BIN_LOG::process_commit_stage_queue(THD *thd, THD *first) {
  mysql_mutex_assert_owner(&LOCK_commit);
#ifndef NDEBUG
  thd->get_transaction()->m_flags.ready_preempt =
      true;  // formality by the leader
#endif
  for (THD *head = first; head; head = head->next_to_commit) {
    DBUG_PRINT("debug", ("Thread ID: %u, commit_error: %d, commit_pending: %s",
                         head->thread_id(), head->commit_error,
                         YESNO(head->tx_commit_pending)));
    DBUG_EXECUTE_IF(
        "block_leader_after_delete",
        if (thd != head) { DBUG_SET("+d,after_delete_wait"); };);
    /*
      If flushing failed, set commit_error for the session, skip the
      transaction and proceed with the next transaction instead. This
      will mark all threads as failed, since the flush failed.
      If flush succeeded, attach to the session and commit it in the
      engines.
    */
#ifndef NDEBUG
    Commit_stage_manager::get_instance().clear_preempt_status(head);
#endif
    /*
      更新全局的 m_max_committed_transaction（用作后续事务的 last_committed），
      并初始本事务上下文的 sequence number
    */
    if (head->get_transaction()->sequence_number != SEQ_UNINIT) {
      mysql_mutex_lock(&LOCK_replica_trans_dep_tracker);
      m_dependency_tracker.update_max_committed(head);
      mysql_mutex_unlock(&LOCK_replica_trans_dep_tracker);
    }
    /*
      Flush/Sync error should be ignored and continue
      to commit phase. And thd->commit_error cannot be
      COMMIT_ERROR at this moment.
    */
    assert(head->commit_error != THD::CE_COMMIT_ERROR);
    Thd_backup_and_restore switch_thd(thd, head);
    bool all = head->get_transaction()->m_flags.real_commit;
    assert(!head->get_transaction()->m_flags.commit_low ||
           head->get_transaction()->m_flags.ready_preempt);<br>  // Binlog Commit、Innodb Commit
    ::finish_transaction_in_engines(head, all, false);
    DBUG_PRINT("debug", ("commit_error: %d, commit_pending: %s",
                         head->commit_error, YESNO(head->tx_commit_pending)));
  }
  /*
    锁定 sidno，更新整组事务 的executed_gtid
    - 如果没开启 binlog，@@GLOBAL.GTID_PURGED 的值是从 executed_gtid 获取的，
      此时 @@GLOBAL.GTID_PURGED 的值和 @@GLOBAL.GTID_EXECUTED 永远是一致的，
      就不需要在记录 lost_gtids
    - 如果开启了 binlog，但是未开启 log_replica_updates，slave 的 SQL 线程或 slave worker 线程
      将自身的 GTID 更新到 executed_gtids、lost_gtids
  */
  gtid_state->update_commit_group(first);
  for (THD *head = first; head; head = head->next_to_commit) {
    Thd_backup_and_restore switch_thd(thd, head);
    auto all = head->get_transaction()->m_flags.real_commit;
    // 只针对外部 XA 事务，在存储引擎层将事务标记为 Prepared
    trx_coordinator::set_prepared_in_tc_in_engines(head, all);
    /*
      在存储引擎层提交之后，递减 Prepared 状态下的 XID 计数器
    */
    if (head->get_transaction()->m_flags.xid_written) dec_prep_xids(head);
  }
}

|::finish_transaction_in_engines
|--trx_coordinator::commit_in_engines
|----ha_commit_low
// Binlog 层提交什么也不做（空函数）
|------binlog_commit
// 存储引擎层提交
|------innobase_commit                                
|--------innobase_commit_low
|----------trx_commit_for_mysql
// 为持久化 GTID 提前分配 update undo segment
|------------trx_undo_gtid_add_update_undo  
// 更新数据字典中被修改表的 update_time 时间
|------------trx_update_mod_tables_timestamp     
// 分配 Mini-transaction handle 和 buffer
|------------trx_commit          
// 提交 mini-transaction
|--------------trx_commit_low                         
|----------------trx_write_serialisation_history
// 更新 undo 状态：
// 对于 insert 状态从 TRX_UNDO_ACTIVE 修改为 TRX_UNDO_TO_FREE
// update 修改为 TRX_UNDO_TO_PURGE
// 如果事务为 update 还需要将 rollback segments 分配 trx no，并将其添加到 purge 队列中
|------------------trx_undo_set_state_at_finish      
//将 update undo log header 添加到 history list 开头释放一些内存对象;
|------------------trx_undo_update_cleanup  
 // 在系统事务表记录 binlog 位点
|------------------trx_sys_update_mysql_binlog_offset 
|----------------trx_commit_in_memory
//- 关闭 mvcc read view
//- 持久化 GTID
//- 释放 insert undo log
//- 唤醒后台线程开始干活，如：master thread、purge thread、page_cleaner

bool Commit_stage_manager::enroll_for(StageID stage, THD *thd,
                                     mysql_mutex_t *stage_mutex,
                                     mysql_mutex_t *enter_mutex) {
 // 如果队列为空，线程就是 leader
 thd->rpl_thd_ctx.binlog_group_commit_ctx().assign_ticket();
 bool leader = this->append_to(stage, thd);
 /*
  如果 FLUSH stage 队列（(BINLOG_FLUSH_STAGE 或 COMMIT_ORDER_FLUSH_STAGE）不为空，此线程就不能成为 leader。leader
  需要获取 enter_mutex
 */
 if (leader) {
   if (stage == COMMIT_ORDER_FLUSH_STAGE) {
     leader = m_queue[BINLOG_FLUSH_STAGE].is_empty();
   /*
     leader 转换的逻辑。
     session 的队列有5种：
       - Binlog flush queue: flush redo 并写 Binlog File
       - Commit order flush queue: 针对 commit order 的事务，但是会参与 group commit 的开头部分，直到引擎层的 flush。
       - Sync queue: sync transaction
       - Commit queue: 提交事务
       - After commit queue: 调用事务的 after_commit hook
    */
   } else if (stage == BINLOG_FLUSH_STAGE &&  // 当前线程是 BINLOG_FLUSH_STAGE 中的第一个线程；但是 COMMIT_ORDER_FLUSH_STAGE
                                              // 已经有了 leader，此时当前线程会挂起，等待 COMMIT_ORDER_FLUSH_STAGE 的 leader 的信号唤醒
              !m_queue[COMMIT_ORDER_FLUSH_STAGE].is_empty()) {
     /*
       当前事务是 binlog queue 中的第一个线程，但是在 commit order queue 中已经有了一个 leader。
       此时当前线程会作为 leader，而 commit order leader 会转变为 follower。
       改变 leader 的原因是 commit order leader 不能作为 binlog 线程的 leader，因为 commit order threads
       必须在 binlog threads 操作完之前离开 commit group。
       转变 leader 为 followers 的步骤如下：
       1. commit order thread 首先进入 flush stage，并成为 commit order leader。
       2. commit order leader 尝试获取 stage mutex，这可能会需要一些时间，比如 mutex 已经被上一个
       commit group的leader获取。
       3. 在此期间，一个 binlog 线程进入了 flush stage。它需要等待来自 commit order leader 的信号。
       4. commit order leader 获取了 stage mutex，然后它会检查是否有  binlog thread进入了 flush stage，
       如果发现了就转变 leader。
       5. commit order leader 给  binlog leader发送一个信号，并成为 follower，等待 commit 的完成
       （和其他 follower 的行为一致）。
       6. binlog leader 被 commit order leader 的信号唤醒并执行 group commit。
     */
     CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP("before_binlog_leader_wait");
     while (thd->tx_commit_pending)
       mysql_cond_wait(&m_stage_cond_leader,
                       &m_queue_lock[BINLOG_FLUSH_STAGE]);
   }
 }
 unlock_queue(stage);
 /*
   通知下一个组提交事务进入队列
 */
 if (stage == BINLOG_FLUSH_STAGE) {
   Commit_order_manager::finish_one(thd);
   CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP("after_binlog_leader_wait");
 } else if (stage == COMMIT_ORDER_FLUSH_STAGE) {
   Commit_order_manager::finish_one(thd);
 }
 /*
   当进入第一个 stage 时，可以不用获取 stage mutex
 */
 if (stage_mutex && need_unlock_stage_mutex) mysql_mutex_unlock(stage_mutex);
 /*
   如果队列非空，当前线程作为 follower 等待 leader 处理队列
 */
 if (!leader) {
   CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP("before_follower_wait");
   mysql_mutex_lock(&m_lock_done);
#ifndef NDEBUG
   thd->get_transaction()->m_flags.ready_preempt = true;
   if (leader_await_preempt_status) mysql_cond_signal(&m_cond_preempt);
#endif
   // tx_commit_pending：还有事务 commit 操作未完成
   while (thd->tx_commit_pending) {
     if (stage == COMMIT_ORDER_FLUSH_STAGE) {
       mysql_cond_wait(&m_stage_cond_commit_order, &m_lock_done);
     } else {
       // follower 线程在此处挂起，等待 leader 提交事务完成后被唤醒
       mysql_cond_wait(&m_stage_cond_binlog, &m_lock_done);
     }
   }
   mysql_mutex_unlock(&m_lock_done);
   return false;
 }
#ifndef NDEBUG
 if (stage == Commit_stage_manager::SYNC_STAGE)
   DEBUG_SYNC(thd, "bgc_between_flush_and_sync");
#endif
 bool need_lock_enter_mutex = false;
 if (leader && enter_mutex != nullptr) {
   /*
     如果由于在轮替 Binlog 时已经获取了 LOCK_log，就不在需要获取 enter_mutex。
   */
   need_lock_enter_mutex = !(mysql_bin_log.is_rotating_caused_by_incident &&
                             enter_mutex == mysql_bin_log.get_log_lock());
   if (need_lock_enter_mutex)
     mysql_mutex_lock(enter_mutex);
   else
     mysql_mutex_assert_owner(enter_mutex);
 }
 // leader 转换的逻辑
 if (stage == COMMIT_ORDER_FLUSH_STAGE) {
   CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP(
       "after_commit_order_thread_becomes_leader");
   lock_queue(stage);
   if (!m_queue[BINLOG_FLUSH_STAGE].is_empty()) {
     if (need_lock_enter_mutex) mysql_mutex_unlock(enter_mutex);
     THD *binlog_leader = m_queue[BINLOG_FLUSH_STAGE].get_leader();
     binlog_leader->tx_commit_pending = false;
     mysql_cond_signal(&m_stage_cond_leader);
     unlock_queue(stage);
     mysql_mutex_lock(&m_lock_done);
     /* wait for signal from binlog leader */
     CONDITIONAL_SYNC_POINT_FOR_TIMESTAMP(
         "before_commit_order_leader_waits_for_binlog_leader");
     while (thd->tx_commit_pending)
       mysql_cond_wait(&m_stage_cond_commit_order, &m_lock_done);
     mysql_mutex_unlock(&m_lock_done);
     leader = false;
     return leader;
   }
 }
 return leader;