nginx源码分析线程池详解

  Syntax: thread_pool name threads=number [max_queue=number];  Default: thread_pool default threads=32 max_queue=65536;  Context: main

  /******************* nginx/src/core/ngx_thread_pool.c ************************/  //创建线程池所需的基础结构  static void * ngx_thread_pool_create_conf(ngx_cycle_t *cycle)  {    ngx_thread_pool_conf_t *tcf;     //从cycle->pool指向的内存池中申请一块内存    tcf = ngx_pcalloc(cycle->pool, sizeof(ngx_thread_pool_conf_t));    if (tcf == NULL) {      return NULL;    }          //先申请包含4个ngx_thread_pool_t指针类型元素的数组     //ngx_thread_pool_t结构体中保存了一个线程池相关的信息    if (ngx_array_init(&tcf->pools, cycle->pool, 4,              sizeof(ngx_thread_pool_t *))      != NGX_OK)    {      return NULL;    }       return tcf;  }     //解析处理配置文件中thread_pool的配置，并将相关信息保存的ngx_thread_pool_t中  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;       value = cf->args->elts;       //根据thread_pool配置中的name作为线程池的唯一标识（如果重名，只有第一个有效）    //申请ngx_thread_pool_t结构保存线程池的相关信息    //由此可见，nginx支持配置多个name不同的线程池    tp = ngx_thread_pool_add(cf, &value[1]);    .......    //处理thread_pool配置行的所有元素    for (i = 2; i < cf->args->nelts; i++) {      //检查配置的线程数      if (ngx_strncmp(value[i].data, "threads=", 8) == 0) {       .......      }             //检查配置的最大队列长度      if (ngx_strncmp(value[i].data, "max_queue=", 10) == 0) {       .......      }    }    ......  }     //判断包含多个线程池的数组中的各个线程池的配置是否正确  static char * ngx_thread_pool_init_conf(ngx_cycle_t *cycle, void *conf)  {    ....    ngx_thread_pool_t **tpp;       tpp = tcf->pools.elts;    //遍历数组中所有的线程池配置，并检查其正确性    for (i = 0; i < tcf->pools.nelts; i++) {      .....    }       return NGX_CONF_OK;  }  

  /******************* nginx/src/core/ngx_thread_pool.c ************************/  //创建线程池所需的基础结构  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或者只有一个worker就不起用线程池    if (ngx_process != NGX_PROCESS_WORKER      && ngx_process != NGX_PROCESS_SINGLE)    {      return NGX_OK;    }         //初始化任务队列    ngx_thread_pool_queue_init(&ngx_thread_pool_done);       tpp = tcf->pools.elts;    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)  {    .....    //初始化任务队列    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;    }    ......    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;      }    }    ......  }     //线程池中线程处理主函数  static void *ngx_thread_pool_cycle(void *data)  {     ......     for ( ;; ) {      //阻塞的方式获取线程锁      if (ngx_thread_mutex_lock(&tp->mtx, tp->log) != NGX_OK) {        return NULL;      }         /* the number may become negative */      tp->waiting--;         //如果任务队列为空，就cond_wait阻塞等待有新任务时调用cond_signal/broadcast触发      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，并将其从队列中移除      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的处理函数      task->handler(task->ctx, tp->log);      .....         ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048);         //将经过预处理的任务添加到done队列中等待调用event的回调函数继续处理      *ngx_thread_pool_done.last = task;      ngx_thread_pool_done.last = &task->next;             //防止编译器优化，保证解锁操作是在上述语句执行完毕后再去执行的      ngx_memory_barrier();         ngx_unlock(&ngx_thread_pool_done_lock);             (void) ngx_notify(ngx_thread_pool_handler);    }  }     //处理pool_done队列上task中包含的每个event事件  static void ngx_thread_pool_handler(ngx_event_t *ev)  {    .....    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_memory_barrier();       ngx_unlock(&ngx_thread_pool_done_lock);       while (task) {      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对应的处理函数有针对性的进行处理      event->handler(event);    }  }     

  /*********************** nginx/src/os/unix/ngx_files.c **********************/  //file_cache模块的处理函数（涉及到了线程池）  static ssize_t ngx_http_file_cache_aio_read(ngx_http_request_t *r, ngx_http_cache_t *c)  {    .......  #if (NGX_THREADS)       if (clcf->aio == NGX_HTTP_AIO_THREADS) {      c->file.thread_task = c->thread_task;      //这里注册的函数在下面语句中的ngx_thread_read函数中被调用      c->file.thread_handler = ngx_http_cache_thread_handler;      c->file.thread_ctx = r;      //根据任务的属性，选择正确的线程池，并初始化task结构体中的各个成员          n = ngx_thread_read(&c->file, c->buf->pos, c->body_start, 0, r->pool);         c->thread_task = c->file.thread_task;      c->reading = (n == NGX_AGAIN);         return n;    }  #endif       return ngx_read_file(&c->file, c->buf->pos, c->body_start, 0);  }        //task任务的处理函数  static ngx_int_t ngx_http_cache_thread_handler(ngx_thread_task_t *task, ngx_file_t *file)  {    .......    tp = clcf->thread_pool;    .......         task->event.data = r;    //注册thread_event_handler函数，该函数在处理pool_done队列中event事件时被调用    task->event.handler = ngx_http_cache_thread_event_handler;       //将任务放到线程池的任务队列中    if (ngx_thread_task_post(tp, task) != NGX_OK) {      return NGX_ERROR;    }    ......  }     /*********************** nginx/src/core/ngx_thread_pool.c **********************/  //添加任务到队列中  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;         //通知阻塞的线程有新事件加入，可以解除阻塞    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 "%V"",            task->id, &tp->name);       return NGX_OK;  }