1. thread pool 简介

MariaDB 共有三种线程调度方式

  • one-thread-per-connection 每个连接一个线程

  • no-threads 所有连接共用一个线程

  • pool-of-threads 线程池

no-threads 只适用于简单的系统，并发数稍高性能就会严重下降

one-thread-per-connection 在多数情况下性能优良，是个合适的选择，生产系统也常用此配置。但在高并发、短连接的业务场景下，使用 one-thread-per-connection 会频繁得创建和销毁线程，严重影响性能

pool-of-threads 适用于高并发短连接的业务场景，线程复用，避免频繁创建和销毁线程带来的性能损耗

MariaDB 的 thread pool 在 win 和 unix 系统的实现不同，本文分析 unix 系统下的实现

thread pool 由若干个 thread_group 组成，每个 thread_group 有若干个 worker 线程，和 0~1 个 listenr 线程

server 接收到连接请求时，将这个连接分配给一个 group 处理，listener 线程负责监听请求，worker 线程处理请求内容

2.代码概览

struct scheduler_functions 内部声明了连接建立相关的属性和方法，这里使用函数指针实现多态

每种线程调度方式，分别实例化一个struct scheduler_functions，给相关变量和函数指针赋值

thread pool 相关实现在 sql/threadpool_common.cc 和 sql/threadpool_unix.cc 中

  1. sql/scheduler.h
  3. struct scheduler_functions
  4. {
  5.   uint max_threads, *connection_count;
  6.   ulong *max_connections;
  7.   bool (*init)(void);
  8.   bool (*init_new_connection_thread)(void);                                                                                                                                                                 
  9.   void (*add_connection)(THD *thd);
  10.   void (*thd_wait_begin)(THD *thd, int wait_type);
  11.   void (*thd_wait_end)(THD *thd);
  12.   void (*post_kill_notification)(THD *thd);
  13.   bool (*end_thread)(THD *thd, bool cache_thread);
  14.   void (*end)(void);
  15. };
  17. -------------------------------------------------
  19. sql/threadpool_common.cc
  21. static scheduler_functions tp_scheduler_functions=                                                                                                                                                          
  22. {
  23.   0,                                  // max_threads
  24.   NULL,
  25.   NULL,
  26.   tp_init,                            // init
  27.   NULL,                               // init_new_connection_thread
  28.   tp_add_connection,                  // add_connection
  29.   tp_wait_begin,                      // thd_wait_begin
  30.   tp_wait_end,                        // thd_wait_end
  31.   post_kill_notification,             // post_kill_notification
  32.   NULL,                               // end_thread
  33.   tp_end                              // end
  34. };
  • tp_init

初始化 thread_group

  1. main
  3.   mysqld_main
  5.     network_init
  7.       /* 调用相应 thread_scheduler 下的 init 函数，这里是 tp_init */
  8.       MYSQL_CALLBACK_ELSE(thread_scheduler, init, (), 0)
  9.       tp_init
  11.         /* thread_group_t 申请内存 */
  12.         all_groups= (thread_group_t *)
  13.           my_malloc(sizeof(thread_group_t) * threadpool_max_size, MYF(MY_WME|MY_ZEROFILL));
  15.         /* 初始化 thread_group_t 数组 */
  16.         for (uint i= 0; i < threadpool_max_size; i++) 
  17.         {
  18.           thread_group_init(&all_groups[i], get_connection_attrib());  
  19.         }
  21.         /* 设置最大执行时间，worker thread */
  22.         pool_timer.tick_interval= threadpool_stall_limit;
  24.         /* 开启 timer thread*/
  25.         start_timer(&pool_timer);
  • tp_add_connection

创建 connection，将 connection 分配到一个 thread_group，插入 group 队列

唤醒或创建一个 worker thread，如果 group 中没有活跃的 worker thread

  1. main
  3.   mysqld_main
  5.     handle_connections_sockets
  7.       create_new_thread
  9.         /* 调用相应 thread_scheduler 下的 add_connection 函数，这里是 tp_add_connection */
  10.         MYSQL_CALLBACK(thd->scheduler, add_connection, (thd));
  11.         tp_add_connection
  13.           /* 申请一个 connection */
  14.           connection_t *connection= alloc_connection(thd);
  16.           /* round-robin 映射到一个 group */
  17.           thread_group_t *group= 
  18.             &all_groups[thd->thread_id%group_count];
  20.           queue_put(group, connection)
  22.             /* connection 插入队列 */
  23.             thread_group->queue.push_back(connection);
  25.             /*
  26.               如果没有活跃 woker 线程，唤醒一个处理新连接 
  27.               此时连接尚未 login，认证等工作在 worker 线程中完成
  28.             */
  29.             if (thread_group->active_thread_count == 0)
  30.               wake_or_create_thread(thread_group);                                        
  32. wake_or_create_thread
  34.   /* 取出 thread_group->waiting_threads 中的 waiting thread (如有) */
  35.   if (wake_thread(thread_group) == 0)
  36.     DBUG_RETURN(0);
  38.   /* 
  39.     没有活跃线程时，立即创建一个，在其他地方也有这个逻辑
  40.     保证每个 thread_group 只是存在一个活跃线程
  41.   */
  42.   if (thread_group->active_thread_count == 0)
  43.     DBUG_RETURN(create_worker(thread_group));
  45.   /* 
  46.     对创建新的 woker thread 限流，距离上个线程创建间隔一定时间才允许再次创建 worker thread
  47.     这个 group 线程总数越多，要求的间隔越长
  48.     4线程以下间隔为0，大于等于 4/8/16 线程时，要求间隔分别为 50*1000/100*1000/200*1000 微秒
  49.   */  
  50.   if (time_since_last_thread_created >
  51.        microsecond_throttling_interval(thread_group))
  52.   {
  53.     DBUG_RETURN(create_worker(thread_group));
  54.   }
  • worker_main

从 queue 中取出和处理 event

在 group 没有 listener 时会变成 listener

没取到 event 时，进入休眠状态，等待被唤醒

最后进入休眠状态的 worker thread 最先被唤醒

  1. worker_main
  3.   for(;;)
  4.   {                                                                                                                                                              
  5.     connection = get_event(&this_thread, thread_group, &ts);
  7.     /* 没有 event 时，跳出循环，结束 worker thread */
  8.     if (!connection)
  9.       break;
  11.     handle_event(connection);
  12.   }
  15. get_event
  17.   for(;;)
  18.   {
  19.     /* 
  20.       取出 tp_add_connection/listener 插入的 connection
  21.     */
  22.     connection = queue_get(thread_group);
  24.     /* 取到 connection，跳出循环，进入下一步 handle_event */
  25.     if(connection)
  26.       break;
  28.     /* 如果没有 listener thread，这个线程变为 listener */
  29.     if(!thread_group->listener)
  30.     {
  31.       /* listener 变为 worker 处理一个 connection */
  32.       connection = listener(current_thread, thread_group);
  34.       /* listener 已变为 worker */
  35.       thread_group->listener= NULL;
  37.       /* 跳出循环，进入 handle_event 处理 connection */
  38.       break;
  39.     }
  41.     /* 进入休眠状态前，尝试非阻塞方式获取一个 connection */
  42.     if (io_poll_wait(thread_group->pollfd,&nev,1, 0) == 1)
  43.     {
  44.       thread_group->io_event_count++;
  45.       connection = (connection_t *)native_event_get_userdata(&nev);
  46.       break;
  47.     }
  49.     /*
  50.       进入休眠状态
  51.       最后休眠的 worker thread 最先被唤醒，更容易命中 cache
  52.     */
  53.     current_thread->woken = false;
  54.     thread_group->waiting_threads.push_front(current_thread);
  56.     /* 等待被唤醒 */
  57.     mysql_cond_wait
  58.   }
  60.   /* 返回获取到的 connection */
  61.   DBUG_RETURN(connection);
  64. handle_event
  66.   if (!connection->logged_in)
  67.   {
  68.     /* login connection */
  69.     err= threadpool_add_connection(connection->thd);
  71.       login_connection
  73.     connection->logged_in= true;
  74.   }
  75.   else
  76.   {
  77.     /* 处理 connection 请求 */
  78.     err= threadpool_process_request(connection->thd);
  80.       do_command
  81.   }
  83.   /* 
  84.     告诉客户端可以读
  85.     可能会变动 connection 所属的 group 
  86.   */
  87.   err= start_io(connection);
  89.     /* 
  90.       group_count 允许动态改变，所以处理 connection 的 group 可能发生变动
  91.       这里检查 group 是否需要变动
  92.     */
  93.     thread_group_t *group = 
  94.       &all_groups[connection->thd->thread_id%group_count];
  96.     if (group != connection->thread_group)
  97.       change_group(connection, connection->thread_group, group)
  99.     return io_poll_start_read(group->pollfd, fd, connection);
  • listener

listener 线程通过 epoll_wait 监听 group 关联的描述符，epoll使用一个文件描述符管理多个描述符

监听获得的 event 插入队列，如果插入前队列为空，listener 变成 worker 线程处理第一个event，其余插入队列，否则所有 event 插入队列

如果 group 中没有活跃线程，唤醒或者创建一个 worker 线程

这里考虑一种情况，如果监听只获得一个 event 且队列为空，那么这个 event 将会被 listener 处理，队列中不会插入新的 event，此时只需要 listener 变成 worker 线程处理 event，不需要再唤醒其他 worker 线程

  1. listener
  3.   for(;;)
  4.   {
  5.     /* 监听事件 */
  6.     cnt = io_poll_wait(thread_group->pollfd, ev, MAX_EVENTS, -1);
  8.     /* 如果队列为空，listener 处理第一个事件，一定概率（只有一个事件）可以不再唤醒 worker thread */
  9.     bool listener_picks_event= thread_group->queue.is_empty();
  12.     /* 第一个事件留给 listener 处理，其余放入队列，或者全部放入队列 */
  13.     for(int i=(listener_picks_event)?1:0; i < cnt ; i++) 
  14.     {
  15.       connection_t *c= (connection_t *)native_event_get_userdata(&ev[i]);
  16.       thread_group->queue.push_back(c);
  17.     }
  19.     if (listener_picks_event)
  20.     {
  21.       /* Handle the first event. */
  22.       retval= (connection_t *)native_event_get_userdata(&ev[0]);
  23.       mysql_mutex_unlock(&thread_group->mutex);
  24.       break;
  25.     }
  27.     /* 
  28.       队列中已经有一些 event，但是没有活跃线程
  29.       唤醒一个 worker 线程处理队列中 event
  30.       唤醒失败，并且仅有一个线程（仅有listener），创建一个 worker 线程
  31.     */
  32.     if(thread_group->active_thread_count==0)
  33.     {    
  34.       if(wake_thread(thread_group))
  35.       {    
  36.         if(thread_group->thread_count == 1)
  37.           create_worker(thread_group);   
  38.       }    
  39.     }
  41.   }
  43.   DBUG_RETURN(retval);
  • timer_thread

检查 listener、worker thread 的运行情况和 connection 超时情况，每隔 thread_pool_stall_limit 检查一次

如果 listener 不存在，且检查周期内没有监听到新的 event，则创建一个 worker thread（自动变成 listener）

如果没有活跃线程（运行状态的 worker thread），且检查周期内没有处理 event，创建一个 worker thread

关闭长时间空闲的 connection

  1. timer_thread
  3.   for(;;)
  4.   {
  5.     /*
  6.       等待一个 tick_interval，即 thread_pool_stall_limit
  7.       正常情况下回等待至超时，只有 stop_timer 会发出 timer->cond 信号
  8.     */
  9.     set_timespec_nsec(ts,timer->tick_interval*1000000);
  10.     err= mysql_cond_timedwait(&timer->cond, &timer->mutex, &ts);
  12.     /* ETIMEDOUT 等待超时，代表 timer 没有被 shutdown */
  13.     if (err == ETIMEDOUT)
  14.     {
  15.       timer->current_microtime= microsecond_interval_timer();
  17.       /* 遍历 thread_group，查看是否 stall */
  18.       for (i= 0; i < threadpool_max_size; i++)
  19.       {
  20.         if(all_groups[i].connection_count)
  21.            check_stall(&all_groups[i]);
  22.       }
  24.       /* 关闭空闲连接 */
  25.       if (timer->next_timeout_check <= timer->current_microtime)
  26.         timeout_check(timer);
  27.     }
  28.   }
  30. check_stall
  32.   /*
  33.     thread_group->io_event_count 代表插入队列的 event 数量
  34.     thread_group 没有 listener，且没有 event 在队列中，这个检查周期内 group 没有监听连接信息
  35.     唤醒或者创建一个 worker 线程，这个线程会自动成为 listener
  36.   */
  37.   if (!thread_group->listener && !thread_group->io_event_count)                                                                                                                                             
  38.   {
  39.     wake_or_create_thread(thread_group);
  40.     return;
  41.   }
  43.   /* Reset io event count */
  44.   thread_group->io_event_count= 0;
  46.   /*
  47.     thread_group->queue_event_count 代表已经出队的 event 数量
  48.     队列非空，并且出队 event 数量为0，这个检查周期内 worker thread 没有处理 event
  49.     唤醒或创建一个 worker thread, thread_group 标记为 stalled， 下个 event 出队时 reset stalled标记
  50.   */
  51.   if (!thread_group->queue.is_empty() && !thread_group->queue_event_count)
  52.   {
  53.     thread_group->stalled= true;
  54.     wake_or_create_thread(thread_group);
  55.   }
  57.   /* Reset queue event count */
  58.   thread_group->queue_event_count= 0;
  • tp_wait_begin/tp_wait_end

线程进入阻塞状态前/阻塞状态结束后，分别调用者两个函数汇报状态，用于维护 active_thread_count 即活跃线程数量

tp_wait_begin 将活跃线程数 -1，活跃线程为 0 时，唤醒或创建一个线程

tp_wait_eng 将活跃线程数 +1

这里需要注意，不是所有的等待都需要调用 tp_wait_begin，预期内短时间结束的等待，比如 mutex，可以不调用 tp_wait_begin

行锁或者磁盘IO这种长时间的等待，需要调用 tp_wait_begin，汇报这个线程暂时无法处理请求，需要开启新的线程

线程池和连接池

线程池是在 server 端的优化，避免频繁的创建和销毁线程

连接池是在 client 端的优化，减少连接创建时间，节省资源

连接池和线程池是两个不同的概念，可以同时使用