数据库内核月报 － 2020 / 05

//创建一个新线程，处理新连接
handle_connection() in connection_handler_per_thread.cc
|
| //准备工作
->thd_prepare_connection() in sql_connect.cc
  | 
  | //进行登陆操作
  ->login_connection() in sql_connect.cc
    |
    | //对此连接的有效性进行验证
    ->check_connection() in sql_connect.cc
      |
  		| //验证登陆
  		->acl_authenticate() in sql_authentication.cc
  		|
  		| //对登陆连接事件进行处理
      ->mysql_audit_notify() in sql_audit.cc
        |
  			| //对登陆连接事件进行处理，并获得错误码
        ->mysql_audit_notify() in sql_audit.cc
          |
  				| //获取需要处理登陆事件的插件
  				->mysql_audit_acquire_plugins() in sql_audit.cc
  				|
  				| //将连接事件分发，并按照需求是都获取插件处理的返回值
          ->event_class_dispatch_error() in sql_audit.cc
            |
  					| //将连接事件分发
            ->event_class_dispatch() in sql_audit.cc
              |
  						| // 调用插件的相关处理函数处理连接事件
              ->plugins_dispatch() in sql_audit.cc
              	|
  							| //检查当前插件是否需要处理此事件
  							->check_audit_mask（）in sql_audit.cc
  							|
  							| //connection_control处理连接事件
              	->connection_control_notify() in connection_control.cc
              	  |
                  | //依次遍历订阅了连接事件的订阅者处理此事件
              	  ->notify_event() in connection_control_coordinator.cc
              	    |
                  	| //处理连接事件
              	    ->notify_event() in connection_delay.cc

/**
  @brief  Handle a connection event and if requried,
  wait for random amount of time before returning.
  We only care about CONNECT and CHANGE_USER sub events.
  @param [in] thd                THD pointer
  @param [in] coordinator        Connection_event_coordinator
  @param [in] connection_event   Connection event to be handled
  @param [in] error_handler      Error handler object
  @returns status of connection event handling
    @retval false  Successfully handled an event.
    @retval true   Something went wrong.
                   error_buffer may contain details.
*/

bool Connection_delay_action::notify_event(
    MYSQL_THD thd, Connection_event_coordinator_services *coordinator,
    const mysql_event_connection *connection_event,
    Error_handler *error_handler) {
  
  ...

	// 只关注CONNECT与CHANGE_USER事件
  if (subclass != MYSQL_AUDIT_CONNECTION_CONNECT &&
      subclass != MYSQL_AUDIT_CONNECTION_CHANGE_USER)
    DBUG_RETURN(error);

  RD_lock rd_lock(m_lock);

  int64 threshold = this->get_threshold();

  // 拿到当前阈值检查阈值是否有效，DISABLE_THRESHOLD=0
  if (threshold <= DISABLE_THRESHOLD) DBUG_RETURN(error);

  int64 current_count = 0;
  bool user_present = false;
  Sql_string userhost;

  make_hash_key(thd, userhost);

  DBUG_PRINT("info", ("Connection control : Connection event lookup for: %s",
                      userhost.c_str()));

  // 获取到当前失败登陆的次数
  user_present = m_userhost_hash.match_entry(userhost, (void *)&current_count)
                     ? false
                     : true;

  // 如果失败次数超过阈值，无论这次连接成功与否，都需要延迟
  // 同时更新统计信息
  if (current_count >= threshold || current_count < 0) {
    
    ulonglong wait_time = get_wait_time((current_count + 1) - threshold);

    if ((error = coordinator->notify_status_var(
             &self, STAT_CONNECTION_DELAY_TRIGGERED, ACTION_INC))) {
      error_handler->handle_error(
          ER_CONN_CONTROL_STAT_CONN_DELAY_TRIGGERED_UPDATE_FAILED);
    }

    // 在产生延迟时，需要释放读写锁，以减少锁的粒度
    // 防止阻塞对于IS table的数据访问
    rd_lock.unlock();
    conditional_wait(thd, wait_time);
    rd_lock.lock();
  }

  if (connection_event->status) {
    
    // 如果此次登陆失败，那么更新LF Hash
    if (m_userhost_hash.create_or_update_entry(userhost)) {
      error_handler->handle_error(
          ER_CONN_CONTROL_FAILED_TO_UPDATE_CONN_DELAY_HASH, userhost.c_str());
      error = true;
    }
  } else {
    
    // 如果此次登陆成功并且LF Hash中有数据，那么就删除LF Hash中的数据
    if (user_present) {
      (void)m_userhost_hash.remove_entry(userhost);
    }
  }

  DBUG_RETURN(error);
}