核心数据结构
任务封装
struct ngx_thread_task_s {
ngx_thread_task_t *next; //指向下一个提交的任务
ngx_uint_t id; //任务id 没添加一个任务就自增加
void *ctx; //执行回调函数的参数
void (*handler)(void *data, ngx_log_t *log); //回调函数
ngx_event_t event; //一个任务和一个事件对应
};
任务队列
typedef struct { //见ngx_thread_pool_done
ngx_thread_task_t *first;
/*
*ngx_thread_pool_t->queue.last = task; 新添加的任务通过last连接在一起
ngx_thread_pool_t->queue.last = &task->next; 下次在添加新任务就让task->next指向新任务了
*/
ngx_thread_task_t **last;
} ngx_thread_pool_queue_t; //线程池队列 初始化在ngx_thread_pool_queue_init
线程池管理器
//一个该结构对应一个threads_pool配置
struct ngx_thread_pool_s {//该结构式存放在ngx_thread_pool_conf_t->pool数组中的,见ngx_thread_pool_init_worker
ngx_thread_mutex_t mtx; //线程锁 ngx_thread_pool_init中初始化
//ngx_thread_task_post中添加的任务被添加到该队列中
ngx_thread_pool_queue_t queue;//ngx_thread_pool_init ngx_thread_pool_queue_init中初始化
//在该线程池poll中每添加一个线程,waiting子减,当线程全部正在执行任务后,waiting会恢复到0
//如果所有线程都已经在执行任务(也就是waiting>-0),又来了任务,那么任务就只能等待。所以waiting表示等待被执行的任务数
ngx_int_t waiting;//等待的任务数 ngx_thread_task_post加1 ngx_thread_pool_cycle减1
ngx_thread_cond_t cond;//条件变量 ngx_thread_pool_init中初始化
ngx_log_t *log;//ngx_thread_pool_init中初始化
ngx_str_t name;//thread_pool name threads=number [max_queue=number];中的name ngx_thread_pool
//如果没有配置,在ngx_thread_pool_init_conf默认赋值为32
ngx_uint_t threads;//thread_pool name threads=number [max_queue=number];中的number ngx_thread_pool
//如果没有配置,在ngx_thread_pool_init_conf默认赋值为65535
//指的是线程已经全部用完的情况下,还可以添加多少个任务到等待队列
ngx_int_t max_queue;//thread_pool name threads=number [max_queue=number];中的max_queue ngx_thread_pool
u_char *file;//配置文件名
ngx_uint_t line;//thread_pool配置在配置文件中的行号
};
相关全局变量定义
static ngx_str_t ngx_thread_pool_default = ngx_string("default");
static ngx_uint_t ngx_thread_pool_task_id;//任务编号,全局变量从0开始
static ngx_atomic_t ngx_thread_pool_done_lock;//为0,可以获取锁,不为0,则不能获取到该锁
static ngx_thread_pool_queue_t ngx_thread_pool_done; //所有的
源码剖析
ngx_thread_task_s相关操作
ngx_thread_task_alloc
从内存池中分配一块ngx_thread_task_t
线程任务的内存
ngx_thread_task_t *
ngx_thread_task_alloc(ngx_pool_t *pool, size_t size)
{
ngx_thread_task_t *task;
task = ngx_pcalloc(pool, sizeof(ngx_thread_task_t) + size);
if (task == NULL) {
return NULL;
}
task->ctx = task + 1;
return task;
}
ngx_thread_task_post
任务添加到对应的线程池任务队列中
ngx_int_t
ngx_thread_task_post(ngx_thread_pool_t *tp, ngx_thread_task_t *task)
{
if (task->event.active) {
ngx_log_error(NGX_LOG_ALERT, tp->log, 0,
"task #%ui already active", task->id);
return NGX_ERROR;
}
//上锁
if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {
return NGX_ERROR;
}
//任务队列中的任务数量不能大于最大数量
if (tp->waiting >= tp->max_queue) {
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
ngx_log_error(NGX_LOG_ERR, tp->log, 0,
"thread pool \"%V\" queue overflow: %i tasks waiting",
&tp->name, tp->waiting);
return NGX_ERROR;
}
task->event.active = 1;
task->id = ngx_thread_pool_task_id++;
task->next = NULL;
ngx_log_debugall(tp->log, 0, "ngx add task to thread, task id:%ui", task->id);
//生产者,消费者
//唤醒其他任务消费线程
if (ngx_thread_cond_signal(&tp->cond, tp->log) != NGX_OK) {
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
return NGX_ERROR;
}
//添加到任务队列
*tp->queue.last = task;
tp->queue.last = &task->next;
tp->waiting++;
//解锁
(void) ngx_thread_mutex_unlock(&tp->mtx, tp->log);
ngx_log_debug2(NGX_LOG_DEBUG_CORE, tp->log, 0,
"task #%ui added to thread pool name: \"%V\" complete",
task->id, &tp->name);
return NGX_OK;
}
线程池配置参数
- 默认32个消费者线程
- 默认任务队列最大长度为65536
//生成一个ngx_thread_pool_conf_t线程池参数配置类
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;
}
//如果配置thread_poll default,则指定默认的threads和max_queue
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;
}
线程池管理器初始化ngx_thread_pool_init
#define ngx_thread_pool_queue_init(q) \
(q)->first = NULL; \
(q)->last = &(q)->first
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;
if (ngx_notify == NULL) {
ngx_log_error(NGX_LOG_ALERT, log, 0,
"the configured event method cannot be used with thread pools");
return NGX_ERROR;
}
//初始化任务队列
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);
if (err) {
ngx_log_error(NGX_LOG_ALERT, log, err,
"pthread_attr_init() failed");
return NGX_ERROR;
}
#if 0
err = pthread_attr_setstacksize(&attr, PTHREAD_STACK_MIN);
if (err) {
ngx_log_error(NGX_LOG_ALERT, log, err,
"pthread_attr_setstacksize() failed");
return NGX_ERROR;
}
#endif
//通过设置的threads创建相应数量的消费线程
for (n = 0; n < tp->threads; n++) {
err = pthread_create(&tid, &attr, ngx_thread_pool_cycle, tp);
if (err) {
ngx_log_error(NGX_LOG_ALERT, log, err,
"pthread_create() failed");
return NGX_ERROR;
}
}
(void) pthread_attr_destroy(&attr);
return NGX_OK;
}
销毁线程池ngx_thread_pool_destroy
一般线程池的销毁都是通过设置标志位,传输信号的方式通知销毁,而nginx是通过post一个销毁任务,任务消费线程执行该任务就会销毁,很精妙
static void
ngx_thread_pool_exit_handler(void *data, ngx_log_t *log)
{
ngx_uint_t *lock = data;
*lock = 0;
pthread_exit(0);
}
static void
ngx_thread_pool_destroy(ngx_thread_pool_t *tp)
{
ngx_uint_t n;
ngx_thread_task_t task;
volatile ngx_uint_t lock;
ngx_memzero(&task, sizeof(ngx_thread_task_t));
task.handler = ngx_thread_pool_exit_handler;//任务退出执行函数
task.ctx = (void *) &lock;//指向传入的参数
for (n = 0; n < tp->threads; n++) { //tp中所有的线程池添加该任务
lock = 1;
if (ngx_thread_task_post(tp, &task) != NGX_OK) {
return;
}
while (lock) { //主进程判断如果lock没有改变,就让CPU给其他线程执行,以此等待,相当于pthread_join
ngx_sched_yield();
}
//只有线程池中的一个线程执行了exit_handler后才能会继续for循环
task.event.active = 0;
}
//此时到这边,所有的线程都已经退出 //条件变量销毁 互斥锁
(void) ngx_thread_cond_destroy(&tp->cond, tp->log);
(void) ngx_thread_mutex_destroy(&tp->mtx, tp->log);
}
任务消费者ngx_thread_pool_cycle
static void *
ngx_thread_pool_cycle(void *data)
{
ngx_thread_pool_t *tp = data; //一个该结构对应一个threads_pool配置
int err;
sigset_t set;
ngx_thread_task_t *task;
#if 0
ngx_time_update();
#endif
ngx_log_debug1(NGX_LOG_DEBUG_CORE, tp->log, 0,
"thread in pool \"%V\" started", &tp->name);
sigfillset(&set);
sigdelset(&set, SIGILL);
sigdelset(&set, SIGFPE);
sigdelset(&set, SIGSEGV);
sigdelset(&set, SIGBUS);
err = pthread_sigmask(SIG_BLOCK, &set, NULL); //
if (err) {
ngx_log_error(NGX_LOG_ALERT, tp->log, err, "pthread_sigmask() failed");
return NULL;
}
for ( ;; ) {//一次任务执行完后又会走到这里,循环
//上锁
if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {
return NULL;
}
/* the number may become negative */
tp->waiting--;
//如果队列中有任务,则直接执行任务,不会在while中等待conf signal
while (tp->queue.first == NULL) { //此时任务队列为空,在条件变量上等待 配合ngx_thread_task_post阅读
//在添加任务的时候唤醒ngx_thread_task_post -> ngx_thread_cond_signal
if (ngx_thread_cond_wait(&tp->cond, &tp->mtx, tp->log) //等待ngx_thread_cond_signal后才会返回
!= 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;
}
#if 0
ngx_time_update();
#endif
ngx_log_debug2(NGX_LOG_DEBUG_CORE, tp->log, 0,
"run task #%ui in thread pool name:\"%V\"",
task->id, &tp->name);
task->handler(task->ctx, tp->log); //每个任务有各自的ctx,因此这里不需要加锁
ngx_log_debug2(NGX_LOG_DEBUG_CORE, tp->log, 0,
"complete task #%ui in thread pool name: \"%V\"",
task->id, &tp->name);
task->next = NULL;
ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);
//task添加到队列尾部,同时可以保证多次添加的时候,让新task和以前的task形成一个还,first执行第一个task,last指向最后一个task
*ngx_thread_pool_done.last = task;
ngx_thread_pool_done.last = &task->next;
ngx_unlock(&ngx_thread_pool_done_lock);
//ngx_notify通告主线程,该任务处理完毕,ngx_thread_pool_handler由主线程执行,也就是进程cycle{}通过epoll_wait返回执行,而不是由线程池中的线程执行
(void) ngx_notify(ngx_thread_pool_handler);
}
}
//任务处理完后,epoll的通知读事件会调用该函数
ngx_notify通告主线程,该任务处理完毕,ngx_thread_pool_handler由主线程执行,也就是进程cycle{}通过epoll_wait返回执行,而不是由线程池中的线程执行
static void
ngx_thread_pool_handler(ngx_event_t *ev)
{
ngx_event_t *event;
ngx_thread_task_t *task;
ngx_log_debug0(NGX_LOG_DEBUG_CORE, ev->log, 0, "thread pool handler");
ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);
task = ngx_thread_pool_done.first;
ngx_thread_pool_done.first = NULL;
ngx_thread_pool_done.last = &ngx_thread_pool_done.first;
ngx_unlock(&ngx_thread_pool_done_lock);
while (task) {//遍历执行前面队列ngx_thread_pool_done中的每一个任务
ngx_log_debug1(NGX_LOG_DEBUG_CORE, ev->log, 0,
"run completion handler for task #%ui", task->id);
event = &task->event;
task = task->next;
event->complete = 1;
event->active = 0;
event->handler(event);
}
}