数据库内核月报 － 2018 / 06

class MYSQL_BIN_LOG: public TC_LOG
 {
 ...
 public:
 /* Committed transactions timestamp */
 Logical_clock max_committed_transaction;
 /* "Prepared" transactions timestamp */
 Logical_clock transaction_counter;
 

class MYSQL_BIN_LOG : public TC_LOG {
 public:
  /** Manage the MTS dependency tracking */                                                                                                                                     
  Transaction_dependency_tracker m_dependency_tracker;
  ...
}

-----------------------------------------------------------------------

bool MYSQL_BIN_LOG::write_gtid(THD *thd, binlog_cache_data *cache_data,
                               Binlog_event_writer *writer) {
  ...
  int64 sequence_number, last_committed;

  /* Generate logical timestamps for MTS */
  m_dependency_tracker.get_dependency(thd, sequence_number, last_committed);
  ...
}

/**   
  Get the dependencies in a transaction, the main entry point for the
  dependency tracking work.
*/
void Transaction_dependency_tracker::get_dependency(THD *thd,
                                                    int64 &sequence_number,
                                                    int64 &commit_parent) {
  sequence_number = commit_parent = 0;

  switch (m_opt_tracking_mode) {
    case DEPENDENCY_TRACKING_COMMIT_ORDER:
      
      /* COMMIT_ORDERE 只调用 m_commit_order.get_dependency() */
      m_commit_order.get_dependency(thd, sequence_number, commit_parent);
      break;
    case DEPENDENCY_TRACKING_WRITESET:
      m_commit_order.get_dependency(thd, sequence_number, commit_parent);
      
      /* WRITESET 在 COMMIT_ORDERE 的基础上再调用 m_writeset.get_dependency() */
      m_writeset.get_dependency(thd, sequence_number, commit_parent);
      break;
    case DEPENDENCY_TRACKING_WRITESET_SESSION:
      m_commit_order.get_dependency(thd, sequence_number, commit_parent);
      m_writeset.get_dependency(thd, sequence_number, commit_parent);
      
      /* WRITESET_SESSION 在 WRITESET 的基础上再调用 m_writeset_session.get_dependency */
      m_writeset_session.get_dependency(thd, sequence_number, commit_parent);                                                                                                                               
      break;
    default:
      DBUG_ASSERT(0);  // blow up on debug
      /*
        Fallback to commit order on production builds.
       */
      m_commit_order.get_dependency(thd, sequence_number, commit_parent);
  }
}

/*
  WRITESET 模式下获取事务依赖，获取到的 commit_parent 和 writeset_history 有关
  writeset_history 增长到 binlog_transaction_dependency_history_size 时会立即清空
*/
void Writeset_trx_dependency_tracker::get_dependency(THD *thd,
                                                     int64 &sequence_number,
                                                     int64 &commit_parent) {
  /* 获取当前的 session write_set */
  Rpl_transaction_write_set_ctx *write_set_ctx =
      thd->get_transaction()->get_transaction_write_set_ctx();
  std::vector<uint64> *writeset = write_set_ctx->get_write_set();
  
  /* 
    判断是否可以使用 write_set 
    如下几种情况不可以使用 write_set：
    * DDL
    * session 的 hash 算法和 history 不同
    * 事务更新了被外键关联的字段
  */
  bool can_use_writesets =
      // empty writeset implies DDL or similar, except if there are missing keys
      (writeset->size() != 0 || write_set_ctx->get_has_missing_keys() ||
       /*
         The empty transactions do not need to clear the writeset history, since
         they can be executed in parallel.
       */
       is_empty_transaction_in_binlog_cache(thd)) &&                                                                                                                                                        
      // hashing algorithm for the session must be the same as used by other
      // rows in history
      (global_system_variables.transaction_write_set_extraction ==
       thd->variables.transaction_write_set_extraction) &&
      // must not use foreign keys
      !write_set_ctx->get_has_related_foreign_keys();

  /* write_history 长度是否超过最大值 */
  bool exceeds_capacity = false;

  if (can_use_writesets) {
    /*
     Check if adding this transaction exceeds the capacity of the writeset
     history. If that happens, m_writeset_history will be cleared only after
     using its information for current transaction.
    */
    
    /* 判断 write_history 是否超过最大值 */
    exceeds_capacity =
        m_writeset_history.size() + writeset->size() > m_opt_max_history_size;

    /*
     Compute the greatest sequence_number among all conflicts and add the
     transaction's row hashes to the history.
    */
    /* 
      遍历 session 的 writeset，查找在 writeset_history 中的冲突行
      如果冲突，则更新 last_parent（last_parent 是临时变量，并不是 commit parent
      如果没冲突，write_history 没找错最大值，则插入 write_history
    */
    int64 last_parent = m_writeset_history_start;
    for (std::vector<uint64>::iterator it = writeset->begin();
         it != writeset->end(); ++it) {
      Writeset_history::iterator hst = m_writeset_history.find(*it);
      if (hst != m_writeset_history.end()) {
        if (hst->second > last_parent && hst->second < sequence_number)
          last_parent = hst->second;

        hst->second = sequence_number;
      } else {
        if (!exceeds_capacity)
          m_writeset_history.insert(
              std::pair<uint64, int64>(*it, sequence_number));
      }
    }
    
    /*
      如果更新了没有主键的表，则不能更新 commit_parent
      但是因为要更新 writeset_history 所以不能直接设置 can_use_writesets
    */
    if (!write_set_ctx->get_has_missing_keys()) {
      commit_parent = std::min(last_parent, commit_parent);
    }   
  }

  /* 如果 writeset_history 已满，或者不可以使用 WriteSet，则清空WriteSet */
  if (exceeds_capacity || !can_use_writesets) {
    m_writeset_history_start = sequence_number;
    m_writeset_history.clear();
  }

}

/* WRITESET_SESSION 使用 session parent 更新 commit parent 并记录当前的session parent */
void Writeset_session_trx_dependency_tracker::get_dependency(
    THD *thd, int64 &sequence_number, int64 &commit_parent) {
  int64 session_parent = thd->rpl_thd_ctx.dependency_tracker_ctx()
                             .get_last_session_sequence_number();

  if (session_parent != 0 && session_parent < sequence_number)
    commit_parent = std::max(commit_parent, session_parent);

  thd->rpl_thd_ctx.dependency_tracker_ctx().set_last_session_sequence_number(
      sequence_number);
}