ProxySQL源碼分析5-MySQL_Thread::run
2021-09-06 19:09:16

整體的流程圖如上，其中最重要的就是最后2步，分別處理data_stream和session

void MySQL_Thread::run() {
    // 死循環(huán)
    while (shutdown == 0) {
 
        // processing_idles初始化是false，上次檢查時間大于ping的周期
        if (processing_idles == false &&
            (last_processing_idles < curtime - mysql_thread___ping_interval_server_msec * 1000)) {
            // 管理idle的連接池，step1(連接池管理)
                // 如果機(jī)器不是online的，這個機(jī)器上的連接直接回收
                // 如果服務(wù)器上的連接大于最大連接，回收
                // 如果服務(wù)器上的連接超過了一定水位，回收
                // 如果連接超時了connection_max_age，回收
            // 管理idle的連接池，step2(多路復(fù)用)
                // 檢查這些連接與后端的ping的狀態(tài)，如果正常ping通，并且連接滿足復(fù)用的條件則保留，否則就是回收連接并返回-1
            run___get_multiple_idle_connections(num_idles);
        }
        // 處理mirror任務(wù)
        handle_mirror_queue_mysql_sessions();
 
        // 根據(jù)ds的來管理session狀態(tài)
        // 根據(jù)ds的來管理mypolls timeout狀態(tài)
        // 根據(jù)ds的來管理監(jiān)聽events
        ProcessAllMyDS_BeforePoll();
 
        // 如果有新的客戶端連接過來，那就增加poll的事件監(jiān)聽
        while ((n = __sync_add_and_fetch(&mypolls.pending_listener_add, 0))) {    // spin here
            poll_listener_add(n);}
 
        // 接受事件
        rc = poll(mypolls.fds, mypolls.len, ttw);
         
        // 處理斷開的連接，刪除對應(yīng)的poll的事件監(jiān)聽
        while ((n = __sync_add_and_fetch(&mypolls.pending_listener_del, 0))) {    // spin here
            if (static_cast<int>(n) == -1) {
                for (unsigned int i = 0; i < mypolls.len; i++) {
                    if (mypolls.myds[i] && mypolls.myds[i]->myds_type == MYDS_LISTENER) {
                        poll_listener_del(mypolls.myds[i]->fd);
                    }
                }
            } else {
                poll_listener_del(n);
            }
            assert(__sync_bool_compare_and_swap(&mypolls.pending_listener_del, n, 0));
        }
 
        // 1、Scan_Sessions_to_Kill_All:將mysql_sessions、idle_mysql_sessions、resume_mysql_sessions、myexchange.idle_mysql_sessions、myexchange.resume_mysql_sessions全部遍歷kill一遍
        // kq.conn_ids:線程id相等、_sess->client_myds存在、username相等，標(biāo)記_sess->killed=true;
        // kq.query_ids:線程id相等、_sess->client_myds存在、username相等、_sess->mybe后端存在，標(biāo)記_sess->killed=true;標(biāo)記后端_sess->mybe->server_myds->wait_until=curtime;_sess->mybe->server_myds->kill_type=1;
        // 2、將maintenance_loop=true
        handle_kill_queues();
 
        if (maintenance_loop) {
            //根據(jù)mirror的任務(wù)情況調(diào)整GloMTH的status_variables，增加mirror_concurrency
            run___cleanup_mirror_queue();
            // this function is called by each thread to update global query statistics
            GloQPro->update_query_processor_stats();
        }
 
        // 檢查當(dāng)前線程和全局線程的變量版本差異&更新
        if (__sync_add_and_fetch(&__global_MySQL_Thread_Variables_version, 0) >
            __thread_MySQL_Thread_Variables_version) {
            refresh_variables();
        }
 
        // 遍歷一遍mypoll
        // 新建連接 listener_handle_new_connection
        // 處理已有連接 process_data_on_data_stream
        ProcessAllMyDS_AfterPoll();
        //
        process_all_sessions();
        return_local_connections();
    }
}