nginx1.17.9源码分析之线程池
我们发现事件驱动的软件都得配一个线程池。libuv和nginx都是。因为事件驱动的软件是单线程。但是有些事情总会引起线程阻塞。所以这个事情就不能放到主线程里做。这就是为什么事件驱动都要配一个线程池。把任务交给线程池中的线程。主线程继续执行。任务完成后通知主线程或者执行回调就行。 我们先看一下nginx线程池的架构。然后开始分析。
在这里插入图片描述 线程池模块在nginx里属于核心模块。在nginx初始化的时候。会初始化一个保存线程池配置的结构体（见图）。nginx默认开启四个线程池。
static void *
ngx_thread_pool_create_conf(ngx_cycle_t *cycle)
{
ngx_thread_pool_conf_t *tcf;
tcf = ngx_pcalloc(cycle->pool, sizeof(ngx_thread_pool_conf_t));
if (tcf == NULL) {
return NULL;
}
if (ngx_array_init(&tcf->pools, cycle->pool, 4,
sizeof(ngx_thread_pool_t *))
!= NGX_OK)
{
return NULL;
}
return tcf;
}
上面的函数就是构造出文章开头的那个图的结构。创建了保存配置的结构，nginx开始解析指令。在分析解析指令前，我们先看一下几个工具函数。
// 根据名字查找池子
ngx_thread_pool_t *
ngx_thread_pool_get(ngx_cycle_t *cycle, ngx_str_t *name)
{
ngx_uint_t i;
ngx_thread_pool_t **tpp;
ngx_thread_pool_conf_t *tcf;
tcf = (ngx_thread_pool_conf_t *) ngx_get_conf(cycle->conf_ctx,
ngx_thread_pool_module);
tpp = tcf->pools.elts;
for (i = 0; i < tcf->pools.nelts; i++) {
if (tpp[i]->name.len == name->len
&& ngx_strncmp(tpp[i]->name.data, name->data, name->len) == 0)
{
return tpp[i];
}
}
return NULL;
}
nginx每个线程池都有一个名字，这个函数就是从图里面的数组中找到名字对应的线程池。
ngx_thread_pool_t *
ngx_thread_pool_add(ngx_conf_t *cf, ngx_str_t *name)
{
ngx_thread_pool_t *tp, **tpp;
ngx_thread_pool_conf_t *tcf;
// 没有名字则取默认值
if (name == NULL) {
name = &ngx_thread_pool_default;
}
// 已存在直接返回
tp = ngx_thread_pool_get(cf->cycle, name);
if (tp) {
return tp;
}
// 分配一个新的池子
tp = ngx_pcalloc(cf->pool, sizeof(ngx_thread_pool_t));
if (tp == NULL) {
return NULL;
}
tp->name = *name;
tp->file = cf->conf_file->file.name.data;
tp->line = cf->conf_file->line;
// 拿到一开始时创建的，用于保存配置的结构体
tcf = (ngx_thread_pool_conf_t *) ngx_get_conf(cf->cycle->conf_ctx,
ngx_thread_pool_module);
// push进数组，数组会自动扩容
tpp = ngx_array_push(&tcf->pools);
if (tpp == NULL) {
return NULL;
}
*tpp = tp;
return tp;
}
上面的函数就是往数组中追加一个元素（表示线程池的结构体）。如果已经存在则报错。 我们看一下，nginx如何解析指令的。配置线程池的指令是
thread_pool name threads=number [max_queue=number]
解析到这个指令的时候，nginx会执行ngx_thread_pool。
static char *
ngx_thread_pool(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
ngx_str_t *value;
ngx_uint_t i;
ngx_thread_pool_t *tp;
// thread_pool指令后的参数
value = cf->args->elts;
// 根据名字（没有则取默认名字）新建一个结构体
tp = ngx_thread_pool_add(cf, &value[1]);
// threads有值说明之前已经配置过这个名字
if (tp->threads) {
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
"duplicate thread pool \"%V\"", &tp->name);
return NGX_CONF_ERROR;
}
tp->max_queue = 65536;
// 解析剩下的参数
for (i = 2; i < cf->args->nelts; i++) {
if (ngx_strncmp(value[i].data, "threads=", 8) == 0) {
// 设置线程数
tp->threads = ngx_atoi(value[i].data + 8, value[i].len - 8);
continue;
}
if (ngx_strncmp(value[i].data, "max_queue=", 10) == 0) {
// 设置任务个数上限
tp->max_queue = ngx_atoi(value[i].data + 10, value[i].len - 10);
continue;
}
}
// 等于0说明指令里没有配置threads参数，报错max_queue可以不配
if (tp->threads == 0) {
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
"\"%V\" must have \"threads\" parameter",
&cmd->name);
return NGX_CONF_ERROR;
}
return NGX_CONF_OK;
}
上面的代码主要构造文章开始那个图中的结构。根据nginx的流程 1 创建保存配置的结构 2 解析配置 3 校验和补偿处理配置 解析完配置后，nginx接着校验和补偿处理。
// 处理完用户的配置后，可能需要做补偿处理
static char *
ngx_thread_pool_init_conf(ngx_cycle_t *cycle, void *conf)
{
ngx_thread_pool_conf_t *tcf = conf;
ngx_uint_t i;
ngx_thread_pool_t **tpp;
tpp = tcf->pools.elts;
// 图中那个数组
for (i = 0; i < tcf->pools.nelts; i++) {
// 用户已经配置了线程数
if (tpp[i]->threads) {
continue;
}
// 没有配置线程数，但是取了默认名字，则其他信息也设置为默认值
if (
tpp[i]->name.len == ngx_thread_pool_default.len
&&
ngx_strncmp(
tpp[i]->name.data,
ngx_thread_pool_default.data,
ngx_thread_pool_default.len)
== 0
)
{
tpp[i]->threads = 32;
tpp[i]->max_queue = 65536;
continue;
}
// 配置了名字但是没有配置线程数，报错
ngx_log_error(NGX_LOG_EMERG, cycle->log, 0,
"unknown thread pool \"%V\" in %s:%ui",
&tpp[i]->name, tpp[i]->file, tpp[i]->line);
return NGX_CONF_ERROR;
}
return NGX_CONF_OK;
}
到此，关于线程池的数据结构已经处理完毕。接下就是创建线程和初始化线程池的数据了。在每个worker初始化的时候，会根据线程池的配置，创建对应的线程。
static ngx_int_t
ngx_thread_pool_init_worker(ngx_cycle_t *cycle)
{
ngx_uint_t i;
ngx_thread_pool_t **tpp;
ngx_thread_pool_conf_t *tcf;
// 线程池只用于worker进程
if (ngx_process != NGX_PROCESS_WORKER
&& ngx_process != NGX_PROCESS_SINGLE)
{
return NGX_OK;
}
tcf = (ngx_thread_pool_conf_t *) ngx_get_conf(cycle->conf_ctx,
ngx_thread_pool_module);
if (tcf == NULL) {
return NGX_OK;
}
// 初始化队列（已完成的任务）
ngx_thread_pool_queue_init(&ngx_thread_pool_done);
// 线程池结构体数组
tpp = tcf->pools.elts;
// 每个worker启动一个或多个线程池
for (i = 0; i < tcf->pools.nelts; i++) {
if (ngx_thread_pool_init(tpp[i], cycle->log, cycle->pool) != NGX_OK) {
return NGX_ERROR;
}
}
return NGX_OK;
}
上面的代码遍历线程池结构体数组。针对每一个线程池结构体创建多个线程。
static ngx_int_t
ngx_thread_pool_init(ngx_thread_pool_t *tp, ngx_log_t *log, ngx_pool_t *pool)
{
int err;
pthread_t tid;
ngx_uint_t n;
pthread_attr_t attr;
ngx_thread_pool_queue_init(&tp->queue);
// 初始化互斥变量
if (ngx_thread_mutex_create(&tp->mtx, log) != NGX_OK) {
return NGX_ERROR;
}
// 初始化条件变量
if (ngx_thread_cond_create(&tp->cond, log) != NGX_OK) {
(void) ngx_thread_mutex_destroy(&tp->mtx, log);
return NGX_ERROR;
}
tp->log = log;
// 初始化线程属性
err = pthread_attr_init(&attr);
// 设置状态为分离，线程退出时资源马上被回收，不需要等待父线程回收
err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
// 创建n个线程,工作函数是ngx_thread_pool_cycle，入参是tp
for (n = 0; n < tp->threads; n++) {
err = pthread_create(&tid, &attr, ngx_thread_pool_cycle, tp);
}
// 用完销毁
(void) pthread_attr_destroy(&attr);
return NGX_OK;
}
这时候，多个线程就被创建了。然后每个线程执行自己的工作函数。
// 处理任务
static void *
ngx_thread_pool_cycle(void *data)
{
ngx_thread_pool_t *tp = data;
int err;
sigset_t set;
ngx_thread_task_t *task;
// 全置1
sigfillset(&set);
// 下面几个信号清零
sigdelset(&set, SIGILL);
sigdelset(&set, SIGFPE);
sigdelset(&set, SIGSEGV);
sigdelset(&set, SIGBUS);
// 屏蔽除了上面几个之外的信号
err = pthread_sigmask(SIG_BLOCK, &set, NULL);
for ( ;; ) {
// 加锁访问队列
if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {
return NULL;
}
// 摘下一个任务
tp->waiting--;
while (tp->queue.first == NULL) {
// 没有任务，等待条件满足时被唤醒
if (ngx_thread_cond_wait(&tp->cond, &tp->mtx, tp->log)
!= NGX_OK)
{
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
return NULL;
}
}
// 摘下第一个任务
task = tp->queue.first;
// 更新头指针
tp->queue.first = task->next;
// 没有任务了，更新尾指针指向头指针的地址，回到初始化状态
if (tp->queue.first == NULL) {
tp->queue.last = &tp->queue.first;
}
// 摘完节点，解锁
if (ngx_thread_mutex_unlock(&tp->mtx, tp->log) != NGX_OK) {
return NULL;
}
// 执行任务
task->handler(task->ctx, tp->log);
task->next = NULL;
ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);
// 执行完插入done队列尾部，done队列是所有线程池公用的，任务队列是每个线程池私有的
*ngx_thread_pool_done.last = task;
// 指向最后一个节点的next域的地址
ngx_thread_pool_done.last = &task->next;
ngx_memory_barrier();
ngx_unlock(&ngx_thread_pool_done_lock);
// 有任务完成，发通知
(void) ngx_notify(ngx_thread_pool_handler);
}
}
线程池维护了一个任务队列，池中的线程互斥访问队列，从中摘下任务执行。任务执行完后把已完成的任务放到完成队列中（所有线程池共享）。并且通知负责处理完成任务节点的函数。
static void
ngx_thread_pool_handler(ngx_event_t *ev)
{
ngx_event_t *event;
ngx_thread_task_t *task;
// 加锁访问队列
ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);
// 指向整个done队列的节点，先保存下来，而不是直接遍历first指针，否则回调里一直加任务导致死循环
task = ngx_thread_pool_done.first;
// 重置头尾指针
ngx_thread_pool_done.first = NULL;
ngx_thread_pool_done.last = &ngx_thread_pool_done.first;
ngx_memory_barrier();
ngx_unlock(&ngx_thread_pool_done_lock);
while (task) {
event = &task->event;
// 指向下一个节点
task = task->next;
// 设置完成标记
event->complete = 1;
event->active = 0;
event->handler(event);
}
}
这就是nginx线程池的原理。和大部分的线程池实现类似，代码看起来很多，但是逻辑还是比较清晰的。
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2020-03-20，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读