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;
}