前言
上周出現了幾次連接配接逾時、連接配接池滿還有dbc連接配接事務模闆失敗的問題。是以有必要深入了解下MySQL的連接配接過程。
正好，上周研究了怎麼用
Clion調試MySQL源碼，接下來通過調試來研究一下吧。
服務端
啟動
sql/main.cc
extern int mysqld_main(int argc, char **argv);
int main(int argc, char **argv) { return mysqld_main(argc, argv); }
- main：入口檔案，僅調用了mysqld_main函數
sql/mysqld.cc
int mysqld_main(int argc, char **argv)
#endif
{
if (my_init()) // init my_sys library & pthreads
{
...
}
...
if (load_defaults(MYSQL_CONFIG_NAME, load_default_groups, &argc, &argv,
&argv_alloc)) {
...
}
mysqld_socket_acceptor->connection_event_loop();
mysqld_exit(signal_hand_thr_exit_code);
} - mysql_main：MySQL服務端啟動邏輯的主要處理函數
- my_init：系統庫和線程初始化
- load_defaults：加載my.cnf各參數
- connection_event_loop：循環監聽套接字。
sql/conn_handler/connection_acceptor.h
/**
Connection acceptor loop to accept connections from clients.
*/
void connection_event_loop() {
Connection_handler_manager *mgr =
Connection_handler_manager::get_instance();
while (!connection_events_loop_aborted()) {
Channel_info *channel_info = m_listener->listen_for_connection_event();
if (channel_info != NULL) mgr->process_new_connection(channel_info);
}
} - connection_event_loop：通過socket_connection.cc::listen_for_connection_event循環監聽，直到有新的連接配接，開始connection_handler_manager.cc::process_new_connection新連接配接的處理過程。
新連接配接
服務端一直處于監聽狀态，當有新連接配接請求時，調用process_new_connection處理新連接配接。
sql/conn_handler/connection_handler_manager.cc
void Connection_handler_manager::process_new_connection(
Channel_info *channel_info) {
if (connection_events_loop_aborted() ||
!check_and_incr_conn_count(channel_info->is_admin_connection())) {
channel_info->send_error_and_close_channel(ER_CON_COUNT_ERROR, 0, true);
delete channel_info;
return;
}
if (m_connection_handler->add_connection(channel_info)) {
inc_aborted_connects();
delete channel_info;
}
} - connection_events_loop_aborted：先判斷是否已取消監聽
- check_and_incr_conn_count：再判斷（會加鎖）是否現有連接配接數是否大于連接配接最大值（連接配接池滿），未滿，則将線程數加一，滿了則拒絕連接配接。（注意，這裡的判斷邏輯使MySQL的實際最大連接配接數是max_connections + 1）
- add_connection：調用add_connection添加連接配接
sql/conn_handler/connection_handler_pre_thread.cc
bool Per_thread_connection_handler::add_connection(Channel_info *channel_info) {
int error = 0;
my_thread_handle id;
DBUG_ENTER("Per_thread_connection_handler::add_connection");
// Simulate thread creation for test case before we check thread cache
DBUG_EXECUTE_IF("fail_thread_create", error = 1; goto handle_error;);
if (!check_idle_thread_and_enqueue_connection(channel_info))
DBUG_RETURN(false);
/*
There are no idle threads avaliable to take up the new
connection. Create a new thread to handle the connection
*/
channel_info->set_prior_thr_create_utime();
error =
mysql_thread_create(key_thread_one_connection, &id, &connection_attrib,
handle_connection, (void *)channel_info);
#ifndef DBUG_OFF
handle_error:
#endif // !DBUG_OFF
if (error) {
...
//錯誤處理，略
}
Global_THD_manager::get_instance()->inc_thread_created();
DBUG_PRINT("info", ("Thread created"));
DBUG_RETURN(false);
} - 調用check_idle_thread_and_enqueue_connection檢視是否有空閑的線程，有則将本次連接配接資訊加入等待隊列，并給空閑線程發送喚醒信号；否則建立線程處理本次連接配接
- 在新線程中，調用handle_connection函數開始進行邏輯處理。
static void *handle_connection(void *arg) {
Global_THD_manager *thd_manager = Global_THD_manager::get_instance();
Connection_handler_manager *handler_manager =
Connection_handler_manager::get_instance();
Channel_info *channel_info = static_cast<Channel_info *>(arg);
bool pthread_reused MY_ATTRIBUTE((unused)) = false;
if (my_thread_init()) {
...
//錯誤處理，略
}
for (;;) {
THD *thd = init_new_thd(channel_info);
if (thd == NULL) {
...
//錯誤處理，略
}
#ifdef HAVE_PSI_THREAD_INTERFACE
if (pthread_reused) {
...
//錯誤處理，略
}
#endif
#ifdef HAVE_PSI_THREAD_INTERFACE
/* Find the instrumented thread */
PSI_thread *psi = PSI_THREAD_CALL(get_thread)();
/* Save it within THD, so it can be inspected */
thd->set_psi(psi);
#endif /* HAVE_PSI_THREAD_INTERFACE */
mysql_thread_set_psi_id(thd->thread_id());
mysql_thread_set_psi_THD(thd);
mysql_socket_set_thread_owner(
thd->get_protocol_classic()->get_vio()->mysql_socket);
thd_manager->add_thd(thd);
if (thd_prepare_connection(thd))
handler_manager->inc_aborted_connects();
else {
while (thd_connection_alive(thd)) {
if (do_command(thd)) break;
}
end_connection(thd);
}
close_connection(thd, 0, false, false);
thd->get_stmt_da()->reset_diagnostics_area();
thd->release_resources();
// Clean up errors now, before possibly waiting for a new connection.
#ifndef HAVE_WOLFSSL
#if OPENSSL_VERSION_NUMBER < 0x10100000L
ERR_remove_thread_state(0);
#endif /* OPENSSL_VERSION_NUMBER < 0x10100000L */
#endif
thd_manager->remove_thd(thd);
Connection_handler_manager::dec_connection_count();
#ifdef HAVE_PSI_THREAD_INTERFACE
/*
Delete the instrumentation for the job that just completed.
*/
thd->set_psi(NULL);
PSI_THREAD_CALL(delete_current_thread)();
#endif /* HAVE_PSI_THREAD_INTERFACE */
delete thd;
// Server is shutting down so end the pthread.
if (connection_events_loop_aborted()) break;
channel_info = Per_thread_connection_handler::block_until_new_connection();
if (channel_info == NULL) break;
pthread_reused = true;
if (connection_events_loop_aborted()) {
...
//錯誤處理，略
}
}
my_thread_end();
my_thread_exit(0);
return NULL;
} - 會對連接配接進行thd_prepare_connection預處理操作，沒問題後繼續下面的邏輯。
- 當連接配接未被關閉，就會一直do_command處理請求。
- 當連接配接關閉，則走下面關閉邏輯
執行
sql/sql_parse.cc
bool do_command(THD *thd) {
bool return_value;
int rc;
NET *net = NULL;
enum enum_server_command command;
COM_DATA com_data;
DBUG_ENTER("do_command");
DBUG_ASSERT(thd->is_classic_protocol());
/*
indicator of uninitialized lex => normal flow of errors handling
(see my_message_sql)
*/
thd->lex->set_current_select(0);
/*
XXX: this code is here only to clear possible errors of init_connect.
Consider moving to prepare_new_connection_state() instead.
That requires making sure the DA is cleared before non-parsing statements
such as COM_QUIT.
*/
thd->clear_error(); // Clear error message
thd->get_stmt_da()->reset_diagnostics_area();
/*
This thread will do a blocking read from the client which
will be interrupted when the next command is received from
the client, the connection is closed or "net_wait_timeout"
number of seconds has passed.
*/
net = thd->get_protocol_classic()->get_net();
my_net_set_read_timeout(net, thd->variables.net_wait_timeout);
net_new_transaction(net);
/*
Synchronization point for testing of KILL_CONNECTION.
This sync point can wait here, to simulate slow code execution
between the last test of thd->killed and blocking in read().
The goal of this test is to verify that a connection does not
hang, if it is killed at this point of execution.
(Bug#37780 - main.kill fails randomly)
Note that the sync point wait itself will be terminated by a
kill. In this case it consumes a condition broadcast, but does
not change anything else. The consumed broadcast should not
matter here, because the read/recv() below doesn't use it.
*/
DEBUG_SYNC(thd, "before_do_command_net_read");
/*
Because of networking layer callbacks in place,
this call will maintain the following instrumentation:
- IDLE events
- SOCKET events
- STATEMENT events
- STAGE events
when reading a new network packet.
In particular, a new instrumented statement is started.
See init_net_server_extension()
*/
thd->m_server_idle = true;
rc = thd->get_protocol()->get_command(&com_data, &command);
thd->m_server_idle = false;
if (rc) {
...
//錯誤處理，略
}
char desc[VIO_DESCRIPTION_SIZE];
vio_description(net->vio, desc);
DBUG_PRINT("info", ("Command on %s = %d (%s)", desc, command,
command_name[command].str));
DBUG_PRINT("info", ("packet: '%*.s'; command: %d",
thd->get_protocol_classic()->get_packet_length(),
thd->get_protocol_classic()->get_raw_packet(), command));
if (thd->get_protocol_classic()->bad_packet)
DBUG_ASSERT(0); // Should be caught earlier
// Reclaim some memory
thd->get_protocol_classic()->get_output_packet()->shrink(
thd->variables.net_buffer_length);
/* Restore read timeout value */
my_net_set_read_timeout(net, thd->variables.net_read_timeout);
return_value = dispatch_command(thd, &com_data, command);
thd->get_protocol_classic()->get_output_packet()->shrink(
thd->variables.net_buffer_length);
out:
/* The statement instrumentation must be closed in all cases. */
DBUG_ASSERT(thd->m_digest == NULL);
DBUG_ASSERT(thd->m_statement_psi == NULL);
DBUG_RETURN(return_value);
}
- 主要的處理邏輯為dispatch_command，根據不同的command類型進行分發。
/**
Perform one connection-level (COM_XXXX) command.
@param thd connection handle
@param command type of command to perform
@param com_data com_data union to store the generated command
@todo
set thd->lex->sql_command to SQLCOM_END here.
@todo
The following has to be changed to an 8 byte integer
@retval
0 ok
@retval
1 request of thread shutdown, i. e. if command is
COM_QUIT
*/
bool dispatch_command(THD *thd, const COM_DATA *com_data,
enum enum_server_command command) {
... //太長不看
switch (command) {
case ... //太長不看
case COM_QUERY: {
... //太長不看
mysql_parse(thd, &parser_state);
... //太長不看
DBUG_PRINT("info", ("query ready"));
break;
}
case ... //太長不看
default:
my_error(ER_UNKNOWN_COM_ERROR, MYF(0));
break;
}
} - 主要看COM_QUERY這個邏輯，我們要用到的DDL、DML都會走這個流程，這個流程中主要是調用mysql_parse方法
/**
Parse a query.
@param thd Current session.
@param parser_state Parser state.
*/
void mysql_parse(THD *thd, Parser_state *parser_state) {
... //太長不看
mysql_reset_thd_for_next_command(thd);
if (!err) {
err = parse_sql(thd, parser_state, NULL);
... //太長不看
}
if (!err) {
mysql_rewrite_query(thd);
... //太長不看
}
if (!err) {
...
error = mysql_execute_command(thd, true);
...
}
} - 主要是SQL文法解析和執行
- mysql_reset_thd_for_next_command是對下一次執行做準備，重置線程各變量
- mysql_rewrite_query看着像是SQL優化？待定 還沒追進去，記個TODO
- 詞法解析前不應該有緩存嗎？沒有找到緩存的邏輯，記個TODO（後續：原來MySQL8.0取消了query cache，詳見： https://mysqlserverteam.com/mysql-8-0-retiring-support-for-the-query-cache/ ）
關閉連接配接
Channel_info *Per_thread_connection_handler::block_until_new_connection() {
Channel_info *new_conn = NULL;
mysql_mutex_lock(&LOCK_thread_cache);
if (blocked_pthread_count < max_blocked_pthreads && !shrink_cache) {
/* Don't kill the pthread, just block it for reuse */
DBUG_PRINT("info", ("Blocking pthread for reuse"));
/*
mysys_var is bound to the physical thread,
so make sure mysys_var->dbug is reset to a clean state
before picking another session in the thread cache.
*/
DBUG_POP();
DBUG_ASSERT(!_db_is_pushed_());
// Block pthread
blocked_pthread_count++;
while (!connection_events_loop_aborted() && !wake_pthread && !shrink_cache)
mysql_cond_wait(&COND_thread_cache, &LOCK_thread_cache);
blocked_pthread_count--;
if (shrink_cache && blocked_pthread_count <= max_blocked_pthreads) {
mysql_cond_signal(&COND_flush_thread_cache);
}
if (wake_pthread) {
wake_pthread--;
if (!waiting_channel_info_list->empty()) {
new_conn = waiting_channel_info_list->front();
waiting_channel_info_list->pop_front();
DBUG_PRINT("info", ("waiting_channel_info_list->pop %p", new_conn));
} else {
DBUG_ASSERT(0); // We should not get here.
}
}
}
mysql_mutex_unlock(&LOCK_thread_cache);
return new_conn;
} - 如果阻塞的線程數小于最大阻塞線程數，則此線程不回收，而是進入阻塞狀态（等待），等待新連接配接來的時候重複使用。
- 否則關閉線程。
用戶端
【從入門到放棄-MySQL】資料庫連接配接過程分析-用戶端參考文獻：
https://www.cnblogs.com/FateTHarlaown/p/8676166.html![【從入門到放棄-MySQL】資料庫連接配接過程分析-用戶端[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)