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

整体的流程图如上，其中最重要的就是最后2步，分别处理data_stream和session

void MySQL_Thread::run() {
    // 死循环
    while (shutdown == 0) {
 
        // processing_idles初始化是false，上次检查时间大于ping的周期
        if (processing_idles == false &&
            (last_processing_idles < curtime - mysql_thread___ping_interval_server_msec * 1000)) {
            // 管理idle的连接池，step1(连接池管理)
                // 如果机器不是online的，这个机器上的连接直接回收
                // 如果服务器上的连接大于最大连接，回收
                // 如果服务器上的连接超过了一定水位，回收
                // 如果连接超时了connection_max_age，回收
            // 管理idle的连接池，step2(多路复用)
                // 检查这些连接与后端的ping的状态，如果正常ping通，并且连接满足复用的条件则保留，否则就是回收连接并返回-1
            run___get_multiple_idle_connections(num_idles);
        }
        // 处理mirror任务
        handle_mirror_queue_mysql_sessions();
 
        // 根据ds的来管理session状态
        // 根据ds的来管理mypolls timeout状态
        // 根据ds的来管理监听events
        ProcessAllMyDS_BeforePoll();
 
        // 如果有新的客户端连接过来，那就增加poll的事件监听
        while ((n = __sync_add_and_fetch(&mypolls.pending_listener_add, 0))) {    // spin here
            poll_listener_add(n);}
 
        // 接受事件
        rc = poll(mypolls.fds, mypolls.len, ttw);
         
        // 处理断开的连接，删除对应的poll的事件监听
        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相等，标记_sess->killed=true;
        // kq.query_ids:线程id相等、_sess->client_myds存在、username相等、_sess->mybe后端存在，标记_sess->killed=true;标记后端_sess->mybe->server_myds->wait_until=curtime;_sess->mybe->server_myds->kill_type=1;
        // 2、将maintenance_loop=true
        handle_kill_queues();
 
        if (maintenance_loop) {
            //根据mirror的任务情况调整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();
        }
 
        // 检查当前线程和全局线程的变量版本差异&更新
        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();
    }
}