no-image

MySQL核心技術之"執行緒操作"

                                    

預處理

當一個新的連線進來時,MySQL會建立對應的thread(pfs.cc):

/**
  Implementation of the thread instrumentation interface.
  @sa PSI_v1::spawn_thread.
*/
int pfs_spawn_thread_v1(PSI_thread_key key,
                        my_thread_handle *thread, const my_thread_attr_t *attr,
                        void *(*start_routine)(void*), void *arg)
{
  PFS_spawn_thread_arg *psi_arg;
  PFS_thread *parent;

  /* psi_arg can not be global, and can not be a local variable. */
  psi_arg= (PFS_spawn_thread_arg*) my_malloc(PSI_NOT_INSTRUMENTED,
                                             sizeof(PFS_spawn_thread_arg),
                                             MYF(MY_WME));
  if (unlikely(psi_arg == NULL))
    return EAGAIN;

  psi_arg->m_child_key= key;
  psi_arg->m_child_identity= (arg ? arg : thread);
  psi_arg->m_user_start_routine= start_routine;
  psi_arg->m_user_arg= arg;

  parent= my_thread_get_THR_PFS();
  if (parent != NULL)
  {
    /*
      Make a copy of the parent attributes.
      This is required, because instrumentation for this thread (the parent)
      may be destroyed before the child thread instrumentation is created.
    */
    psi_arg->m_thread_internal_id= parent->m_thread_internal_id;

    memcpy(psi_arg->m_username, parent->m_username, sizeof(psi_arg->m_username));
    psi_arg->m_username_length= parent->m_username_length;

    memcpy(psi_arg->m_hostname, parent->m_hostname, sizeof(psi_arg->m_hostname));
    psi_arg->m_hostname_length= parent->m_hostname_length;
  }
  else
  {
    psi_arg->m_thread_internal_id= 0;
    psi_arg->m_username_length= 0;
    psi_arg->m_hostname_length= 0;
  }

  int result= my_thread_create(thread, attr, pfs_spawn_thread, psi_arg);
  if (unlikely(result != 0))
    my_free(psi_arg);
  return result;
}

注意上面程式碼中最後的位置: my_thread_create(thread, attr, pfs_spawn_thread, psi_arg);

主要是對三個變數賦值:m_thread_internal_id

程式碼改動

由於開發原因,我加入了一個boost執行緒池,那麼從原始執行緒產生任務塞入boost執行緒池中的時候,需要做類似的操作,即要傳入一個psi_arg

獲取當前的pfs:

/**
  Implementation of the thread instrumentation interface.
  @sa PSI_v1::get_thread_id.
*/
PSI_thread*
pfs_get_thread_v1(void)
{
  PFS_thread *pfs= my_thread_get_THR_PFS();
  return reinterpret_cast<PSI_thread*> (pfs);
}

注意,不能直接用my_thread_get_THR_PFS()獲取,因為其為static函式,只能用於本檔案使用。同理設定pfs:

/**
  Implementation of the thread instrumentation interface.
  @sa PSI_v1::set_thread.
*/
void pfs_set_thread_v1(PSI_thread* thread)
{
  PFS_thread *pfs= reinterpret_cast<PFS_thread*> (thread);
  my_thread_set_THR_PFS(pfs);
  std::cout << "my_thread_set_THR_PFS in pfs_set_thread_v1, value: "<< pfs << std::endl;
}

附錄:
定義的PSI結構(psi.h):

/**
  Performance Schema Interface, version 1.
  @since PSI_VERSION_1
*/
struct PSI_v1
{
  /** @sa register_mutex_v1_t. */
  register_mutex_v1_t register_mutex;
  /** @sa register_rwlock_v1_t. */
  register_rwlock_v1_t register_rwlock;
  /** @sa register_cond_v1_t. */
  register_cond_v1_t register_cond;
  /** @sa register_thread_v1_t. */
  register_thread_v1_t register_thread;
  /** @sa register_file_v1_t. */
  register_file_v1_t register_file;
  /** @sa register_stage_v1_t. */
  register_stage_v1_t register_stage;
  /** @sa register_statement_v1_t. */
  register_statement_v1_t register_statement;
  /** @sa register_socket_v1_t. */
  register_socket_v1_t register_socket;
  /** @sa init_mutex_v1_t. */
  init_mutex_v1_t init_mutex;
  /** @sa destroy_mutex_v1_t. */
  destroy_mutex_v1_t destroy_mutex;
  /** @sa init_rwlock_v1_t. */
  init_rwlock_v1_t init_rwlock;
  /** @sa destroy_rwlock_v1_t. */
  destroy_rwlock_v1_t destroy_rwlock;
  /** @sa init_cond_v1_t. */
  init_cond_v1_t init_cond;
  /** @sa destroy_cond_v1_t. */
  destroy_cond_v1_t destroy_cond;
  /** @sa init_socket_v1_t. */
  init_socket_v1_t init_socket;
  /** @sa destroy_socket_v1_t. */
  destroy_socket_v1_t destroy_socket;

  /** @sa get_table_share_v1_t. */
  get_table_share_v1_t get_table_share;
  /** @sa release_table_share_v1_t. */
  release_table_share_v1_t release_table_share;
  /** @sa drop_table_share_v1_t. */
  drop_table_share_v1_t drop_table_share;
  /** @sa open_table_v1_t. */
  open_table_v1_t open_table;
  /** @sa unbind_table_v1_t. */
  unbind_table_v1_t unbind_table;
  /** @sa rebind_table_v1_t. */
  rebind_table_v1_t rebind_table;
  /** @sa close_table_v1_t. */
  close_table_v1_t close_table;
  /** @sa create_file_v1_t. */
  create_file_v1_t create_file;
  /** @sa spawn_thread_v1_t. */
  spawn_thread_v1_t spawn_thread;
  /** @sa new_thread_v1_t. */
  new_thread_v1_t new_thread;
  /** @sa set_thread_id_v1_t. */
  set_thread_id_v1_t set_thread_id;
  /** @sa set_thread_THD_v1_t. */
  set_thread_THD_v1_t set_thread_THD;
  /** @sa set_thread_os_id_v1_t. */
  set_thread_os_id_v1_t set_thread_os_id;
  /** @sa get_thread_v1_t. */
  get_thread_v1_t get_thread;
  /** @sa set_thread_user_v1_t. */
  set_thread_user_v1_t set_thread_user;
  /** @sa set_thread_account_v1_t. */
  set_thread_account_v1_t set_thread_account;
  /** @sa set_thread_db_v1_t. */
  set_thread_db_v1_t set_thread_db;
  /** @sa set_thread_command_v1_t. */
  set_thread_command_v1_t set_thread_command;
  /** @sa set_connection_type_v1_t. */
  set_connection_type_v1_t set_connection_type;
  /** @sa set_thread_start_time_v1_t. */
  set_thread_start_time_v1_t set_thread_start_time;
  /** @sa set_thread_state_v1_t. */
  set_thread_state_v1_t set_thread_state;
  /** @sa set_thread_info_v1_t. */
  set_thread_info_v1_t set_thread_info;
  /** @sa set_thread_v1_t. */
  set_thread_v1_t set_thread;
  /** @sa delete_current_thread_v1_t. */
  delete_current_thread_v1_t delete_current_thread;
  /** @sa delete_thread_v1_t. */
  delete_thread_v1_t delete_thread;
  /** @sa get_thread_file_name_locker_v1_t. */
  get_thread_file_name_locker_v1_t get_thread_file_name_locker;
  /** @sa get_thread_file_stream_locker_v1_t. */
  get_thread_file_stream_locker_v1_t get_thread_file_stream_locker;
  /** @sa get_thread_file_descriptor_locker_v1_t. */
  get_thread_file_descriptor_locker_v1_t get_thread_file_descriptor_locker;
  /** @sa unlock_mutex_v1_t. */
  unlock_mutex_v1_t unlock_mutex;
  /** @sa unlock_rwlock_v1_t. */
  unlock_rwlock_v1_t unlock_rwlock;
  /** @sa signal_cond_v1_t. */
  signal_cond_v1_t signal_cond;
  /** @sa broadcast_cond_v1_t. */
  broadcast_cond_v1_t broadcast_cond;
  /** @sa start_idle_wait_v1_t. */
  start_idle_wait_v1_t start_idle_wait;
  /** @sa end_idle_wait_v1_t. */
  end_idle_wait_v1_t end_idle_wait;
  /** @sa start_mutex_wait_v1_t. */
  start_mutex_wait_v1_t start_mutex_wait;
  /** @sa end_mutex_wait_v1_t. */
  end_mutex_wait_v1_t end_mutex_wait;
  /** @sa start_rwlock_rdwait_v1_t. */
  start_rwlock_rdwait_v1_t start_rwlock_rdwait;
  /** @sa end_rwlock_rdwait_v1_t. */
  end_rwlock_rdwait_v1_t end_rwlock_rdwait;
  /** @sa start_rwlock_wrwait_v1_t. */
  start_rwlock_wrwait_v1_t start_rwlock_wrwait;
  /** @sa end_rwlock_wrwait_v1_t. */
  end_rwlock_wrwait_v1_t end_rwlock_wrwait;
  /** @sa start_cond_wait_v1_t. */
  start_cond_wait_v1_t start_cond_wait;
  /** @sa end_cond_wait_v1_t. */
  end_cond_wait_v1_t end_cond_wait;
  /** @sa start_table_io_wait_v1_t. */
  start_table_io_wait_v1_t start_table_io_wait;
  /** @sa end_table_io_wait_v1_t. */
  end_table_io_wait_v1_t end_table_io_wait;
  /** @sa start_table_lock_wait_v1_t. */
  start_table_lock_wait_v1_t start_table_lock_wait;
  /** @sa end_table_lock_wait_v1_t. */
  end_table_lock_wait_v1_t end_table_lock_wait;
  /** @sa start_file_open_wait_v1_t. */
  start_file_open_wait_v1_t start_file_open_wait;
  /** @sa end_file_open_wait_v1_t. */
  end_file_open_wait_v1_t end_file_open_wait;
  /** @sa end_file_open_wait_and_bind_to_descriptor_v1_t. */
  end_file_open_wait_and_bind_to_descriptor_v1_t
    end_file_open_wait_and_bind_to_descriptor;
  /** @sa end_temp_file_open_wait_and_bind_to_descriptor_v1_t. */
  end_temp_file_open_wait_and_bind_to_descriptor_v1_t
    end_temp_file_open_wait_and_bind_to_descriptor;
  /** @sa start_file_wait_v1_t. */
  start_file_wait_v1_t start_file_wait;
  /** @sa end_file_wait_v1_t. */
  end_file_wait_v1_t end_file_wait;
  /** @sa start_file_close_wait_v1_t. */
  start_file_close_wait_v1_t start_file_close_wait;
  /** @sa end_file_close_wait_v1_t. */
  end_file_close_wait_v1_t end_file_close_wait;
  /** @sa start_stage_v1_t. */
  start_stage_v1_t start_stage;
  /** @sa get_current_stage_progress_v1_t. */
  get_current_stage_progress_v1_t get_current_stage_progress;
  /** @sa end_stage_v1_t. */
  end_stage_v1_t end_stage;
  /** @sa get_thread_statement_locker_v1_t. */
  get_thread_statement_locker_v1_t get_thread_statement_locker;
  /** @sa refine_statement_v1_t. */
  refine_statement_v1_t refine_statement;
  /** @sa start_statement_v1_t. */
  start_statement_v1_t start_statement;
  /** @sa set_statement_text_v1_t. */
  set_statement_text_v1_t set_statement_text;
  /** @sa set_statement_lock_time_t. */
  set_statement_lock_time_t set_statement_lock_time;
  /** @sa set_statement_rows_sent_t. */
  set_statement_rows_sent_t set_statement_rows_sent;
  /** @sa set_statement_rows_examined_t. */
  set_statement_rows_examined_t set_statement_rows_examined;
  /** @sa inc_statement_created_tmp_disk_tables. */
  inc_statement_created_tmp_disk_tables_t inc_statement_created_tmp_disk_tables;
  /** @sa inc_statement_created_tmp_tables. */
  inc_statement_created_tmp_tables_t inc_statement_created_tmp_tables;
  /** @sa inc_statement_select_full_join. */
  inc_statement_select_full_join_t inc_statement_select_full_join;
  /** @sa inc_statement_select_full_range_join. */
  inc_statement_select_full_range_join_t inc_statement_select_full_range_join;
  /** @sa inc_statement_select_range. */
  inc_statement_select_range_t inc_statement_select_range;
  /** @sa inc_statement_select_range_check. */
  inc_statement_select_range_check_t inc_statement_select_range_check;
  /** @sa inc_statement_select_scan. */
  inc_statement_select_scan_t inc_statement_select_scan;
  /** @sa inc_statement_sort_merge_passes. */
  inc_statement_sort_merge_passes_t inc_statement_sort_merge_passes;
  /** @sa inc_statement_sort_range. */
  inc_statement_sort_range_t inc_statement_sort_range;
  /** @sa inc_statement_sort_rows. */
  inc_statement_sort_rows_t inc_statement_sort_rows;
  /** @sa inc_statement_sort_scan. */
  inc_statement_sort_scan_t inc_statement_sort_scan;
  /** @sa set_statement_no_index_used. */
  set_statement_no_index_used_t set_statement_no_index_used;
  /** @sa set_statement_no_good_index_used. */
  set_statement_no_good_index_used_t set_statement_no_good_index_used;
  /** @sa end_statement_v1_t. */
  end_statement_v1_t end_statement;
  /** @sa get_thread_transaction_locker_v1_t. */
  get_thread_transaction_locker_v1_t get_thread_transaction_locker;
  /** @sa start_transaction_v1_t. */
  start_transaction_v1_t start_transaction;
  /** @sa set_transaction_xid_v1_t. */
  set_transaction_xid_v1_t set_transaction_xid;
  /** @sa set_transaction_xa_state_v1_t. */
  set_transaction_xa_state_v1_t set_transaction_xa_state;
  /** @sa set_transaction_gtid_v1_t. */
  set_transaction_gtid_v1_t set_transaction_gtid;
  /** @sa set_transaction_trxid_v1_t. */
  set_transaction_trxid_v1_t set_transaction_trxid;
  /** @sa inc_transaction_savepoints_v1_t. */
  inc_transaction_savepoints_v1_t inc_transaction_savepoints;
  /** @sa inc_transaction_rollback_to_savepoint_v1_t. */
  inc_transaction_rollback_to_savepoint_v1_t inc_transaction_rollback_to_savepoint;
  /** @sa inc_transaction_release_savepoint_v1_t. */
  inc_transaction_release_savepoint_v1_t inc_transaction_release_savepoint;
  /** @sa end_transaction_v1_t. */
  end_transaction_v1_t end_transaction;
  /** @sa start_socket_wait_v1_t. */
  start_socket_wait_v1_t start_socket_wait;
  /** @sa end_socket_wait_v1_t. */
  end_socket_wait_v1_t end_socket_wait;
  /** @sa set_socket_state_v1_t. */
  set_socket_state_v1_t set_socket_state;
  /** @sa set_socket_info_v1_t. */
  set_socket_info_v1_t set_socket_info;
  /** @sa set_socket_thread_owner_v1_t. */
  set_socket_thread_owner_v1_t set_socket_thread_owner;
  /** @sa create_prepared_stmt_v1_t. */
  create_prepared_stmt_v1_t create_prepared_stmt;
  /** @sa destroy_prepared_stmt_v1_t. */
  destroy_prepared_stmt_v1_t destroy_prepared_stmt;
  /** @sa reprepare_prepared_stmt_v1_t. */
  reprepare_prepared_stmt_v1_t reprepare_prepared_stmt;
  /** @sa execute_prepared_stmt_v1_t. */
  execute_prepared_stmt_v1_t execute_prepared_stmt;
  /** @sa digest_start_v1_t. */
  digest_start_v1_t digest_start;
  /** @sa digest_end_v1_t. */
  digest_end_v1_t digest_end;
  /** @sa set_thread_connect_attrs_v1_t. */
  set_thread_connect_attrs_v1_t set_thread_connect_attrs;
  /** @sa start_sp_v1_t. */
  start_sp_v1_t start_sp;
  /** @sa start_sp_v1_t. */
  end_sp_v1_t end_sp;
  /** @sa drop_sp_v1_t. */
  drop_sp_v1_t drop_sp;
  /** @sa get_sp_share_v1_t. */
  get_sp_share_v1_t get_sp_share;
  /** @sa release_sp_share_v1_t. */
  release_sp_share_v1_t release_sp_share;
  /** @sa register_memory_v1_t. */
  register_memory_v1_t register_memory;
  /** @sa memory_alloc_v1_t. */
  memory_alloc_v1_t memory_alloc;
  /** @sa memory_realloc_v1_t. */
  memory_realloc_v1_t memory_realloc;
  /** @sa memory_claim_v1_t. */
  memory_claim_v1_t memory_claim;
  /** @sa memory_free_v1_t. */
  memory_free_v1_t memory_free;

  unlock_table_v1_t unlock_table;

  create_metadata_lock_v1_t create_metadata_lock;
  set_metadata_lock_status_v1_t set_metadata_lock_status;
  destroy_metadata_lock_v1_t destroy_metadata_lock;

  start_metadata_wait_v1_t start_metadata_wait;
  end_metadata_wait_v1_t end_metadata_wait;
};

該結構在psi.cc

每一個值都是一個函式。感興趣的可以去看對應實現。

關聯文章