mirror of
https://github.com/nginx/nginx.git
synced 2024-12-15 21:39:00 +08:00
c34368715f
Work around pthread_cond_destroy() and pthread_mutex_destroy() returning EBUSY. A proper solution would be to ensure all threads are terminated.
638 lines
15 KiB
C
638 lines
15 KiB
C
|
|
/*
|
|
* Copyright (C) Nginx, Inc.
|
|
* Copyright (C) Valentin V. Bartenev
|
|
* Copyright (C) Ruslan Ermilov
|
|
*/
|
|
|
|
|
|
#include <ngx_config.h>
|
|
#include <ngx_core.h>
|
|
#include <ngx_thread_pool.h>
|
|
|
|
|
|
typedef struct {
|
|
ngx_array_t pools;
|
|
} ngx_thread_pool_conf_t;
|
|
|
|
|
|
typedef struct {
|
|
ngx_thread_mutex_t mtx;
|
|
ngx_thread_task_t *first;
|
|
ngx_thread_task_t **last;
|
|
} ngx_thread_pool_queue_t;
|
|
|
|
|
|
struct ngx_thread_pool_s {
|
|
ngx_thread_pool_queue_t queue;
|
|
ngx_int_t waiting;
|
|
ngx_thread_cond_t cond;
|
|
|
|
ngx_log_t *log;
|
|
ngx_pool_t *pool;
|
|
|
|
ngx_str_t name;
|
|
ngx_uint_t threads;
|
|
ngx_int_t max_queue;
|
|
|
|
u_char *file;
|
|
ngx_uint_t line;
|
|
};
|
|
|
|
|
|
static ngx_int_t ngx_thread_pool_init(ngx_thread_pool_t *tp, ngx_log_t *log,
|
|
ngx_pool_t *pool);
|
|
static ngx_int_t ngx_thread_pool_queue_init(ngx_thread_pool_queue_t *queue,
|
|
ngx_log_t *log);
|
|
static ngx_int_t ngx_thread_pool_queue_destroy(ngx_thread_pool_queue_t *queue,
|
|
ngx_log_t *log);
|
|
static void ngx_thread_pool_destroy(ngx_thread_pool_t *tp);
|
|
|
|
static void *ngx_thread_pool_cycle(void *data);
|
|
static void ngx_thread_pool_handler(ngx_event_t *ev);
|
|
|
|
static char *ngx_thread_pool(ngx_conf_t *cf, ngx_command_t *cmd, void *conf);
|
|
|
|
static void *ngx_thread_pool_create_conf(ngx_cycle_t *cycle);
|
|
static char *ngx_thread_pool_init_conf(ngx_cycle_t *cycle, void *conf);
|
|
|
|
static ngx_int_t ngx_thread_pool_init_worker(ngx_cycle_t *cycle);
|
|
static void ngx_thread_pool_exit_worker(ngx_cycle_t *cycle);
|
|
|
|
|
|
static ngx_command_t ngx_thread_pool_commands[] = {
|
|
|
|
{ ngx_string("thread_pool"),
|
|
NGX_MAIN_CONF|NGX_DIRECT_CONF|NGX_CONF_TAKE23,
|
|
ngx_thread_pool,
|
|
0,
|
|
0,
|
|
NULL },
|
|
|
|
ngx_null_command
|
|
};
|
|
|
|
|
|
static ngx_core_module_t ngx_thread_pool_module_ctx = {
|
|
ngx_string("thread_pool"),
|
|
ngx_thread_pool_create_conf,
|
|
ngx_thread_pool_init_conf
|
|
};
|
|
|
|
|
|
ngx_module_t ngx_thread_pool_module = {
|
|
NGX_MODULE_V1,
|
|
&ngx_thread_pool_module_ctx, /* module context */
|
|
ngx_thread_pool_commands, /* module directives */
|
|
NGX_CORE_MODULE, /* module type */
|
|
NULL, /* init master */
|
|
NULL, /* init module */
|
|
ngx_thread_pool_init_worker, /* init process */
|
|
NULL, /* init thread */
|
|
NULL, /* exit thread */
|
|
ngx_thread_pool_exit_worker, /* exit process */
|
|
NULL, /* exit master */
|
|
NGX_MODULE_V1_PADDING
|
|
};
|
|
|
|
|
|
static ngx_str_t ngx_thread_pool_default = ngx_string("default");
|
|
|
|
static ngx_uint_t ngx_thread_pool_task_id;
|
|
static ngx_thread_pool_queue_t ngx_thread_pool_done;
|
|
|
|
|
|
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;
|
|
}
|
|
|
|
if (ngx_thread_pool_queue_init(&tp->queue, log) != NGX_OK) {
|
|
return NGX_ERROR;
|
|
}
|
|
|
|
if (ngx_thread_cond_create(&tp->cond, log) != NGX_OK) {
|
|
(void) ngx_thread_pool_queue_destroy(&tp->queue, log);
|
|
return NGX_ERROR;
|
|
}
|
|
|
|
tp->log = log;
|
|
tp->pool = pool;
|
|
|
|
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
|
|
|
|
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;
|
|
}
|
|
|
|
|
|
static ngx_int_t
|
|
ngx_thread_pool_queue_init(ngx_thread_pool_queue_t *queue, ngx_log_t *log)
|
|
{
|
|
queue->first = NULL;
|
|
queue->last = &queue->first;
|
|
|
|
return ngx_thread_mutex_create(&queue->mtx, log);
|
|
}
|
|
|
|
|
|
static ngx_int_t
|
|
ngx_thread_pool_queue_destroy(ngx_thread_pool_queue_t *queue, ngx_log_t *log)
|
|
{
|
|
#if 0
|
|
return ngx_thread_mutex_destroy(&queue->mtx, log);
|
|
#else
|
|
return NGX_OK;
|
|
#endif
|
|
}
|
|
|
|
|
|
static void
|
|
ngx_thread_pool_destroy(ngx_thread_pool_t *tp)
|
|
{
|
|
/* TODO: exit threads */
|
|
|
|
#if 0
|
|
(void) ngx_thread_cond_destroy(&tp->cond, tp->log);
|
|
#endif
|
|
|
|
(void) ngx_thread_pool_queue_destroy(&tp->queue, tp->log);
|
|
}
|
|
|
|
|
|
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_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->queue.mtx, tp->log) != NGX_OK) {
|
|
return NGX_ERROR;
|
|
}
|
|
|
|
if (tp->waiting >= tp->max_queue) {
|
|
(void) ngx_thread_mutex_unlock(&tp->queue.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->queue.mtx, tp->log);
|
|
return NGX_ERROR;
|
|
}
|
|
|
|
*tp->queue.last = task;
|
|
tp->queue.last = &task->next;
|
|
|
|
tp->waiting++;
|
|
|
|
(void) ngx_thread_mutex_unlock(&tp->queue.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;
|
|
}
|
|
|
|
|
|
static void *
|
|
ngx_thread_pool_cycle(void *data)
|
|
{
|
|
ngx_thread_pool_t *tp = data;
|
|
|
|
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->queue.mtx, tp->log) != NGX_OK) {
|
|
return NULL;
|
|
}
|
|
|
|
/* the number may become negative */
|
|
tp->waiting--;
|
|
|
|
while (tp->queue.first == NULL) {
|
|
if (ngx_thread_cond_wait(&tp->cond, &tp->queue.mtx, tp->log)
|
|
!= NGX_OK)
|
|
{
|
|
(void) ngx_thread_mutex_unlock(&tp->queue.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->queue.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 \"%V\"",
|
|
task->id, &tp->name);
|
|
|
|
task->handler(task->ctx, tp->log);
|
|
|
|
ngx_log_debug2(NGX_LOG_DEBUG_CORE, tp->log, 0,
|
|
"complete task #%ui in thread pool \"%V\"",
|
|
task->id, &tp->name);
|
|
|
|
task->next = NULL;
|
|
|
|
if (ngx_thread_mutex_lock(&ngx_thread_pool_done.mtx, tp->log)
|
|
!= NGX_OK)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
*ngx_thread_pool_done.last = task;
|
|
ngx_thread_pool_done.last = &task->next;
|
|
|
|
if (ngx_thread_mutex_unlock(&ngx_thread_pool_done.mtx, tp->log)
|
|
!= NGX_OK)
|
|
{
|
|
return NULL;
|
|
}
|
|
|
|
(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_log_debug0(NGX_LOG_DEBUG_CORE, ev->log, 0, "thread pool handler");
|
|
|
|
if (ngx_thread_mutex_lock(&ngx_thread_pool_done.mtx, ev->log) != NGX_OK) {
|
|
return;
|
|
}
|
|
|
|
task = ngx_thread_pool_done.first;
|
|
ngx_thread_pool_done.first = NULL;
|
|
ngx_thread_pool_done.last = &ngx_thread_pool_done.first;
|
|
|
|
if (ngx_thread_mutex_unlock(&ngx_thread_pool_done.mtx, ev->log) != NGX_OK) {
|
|
return;
|
|
}
|
|
|
|
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->handler(event);
|
|
}
|
|
}
|
|
|
|
|
|
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;
|
|
}
|
|
|
|
|
|
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;
|
|
}
|
|
|
|
|
|
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;
|
|
|
|
tp = ngx_thread_pool_add(cf, &value[1]);
|
|
|
|
if (tp == NULL) {
|
|
return NGX_CONF_ERROR;
|
|
}
|
|
|
|
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);
|
|
|
|
if (tp->threads == (ngx_uint_t) NGX_ERROR || tp->threads == 0) {
|
|
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
|
|
"invalid threads value \"%V\"", &value[i]);
|
|
return NGX_CONF_ERROR;
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
if (ngx_strncmp(value[i].data, "max_queue=", 10) == 0) {
|
|
|
|
tp->max_queue = ngx_atoi(value[i].data + 10, value[i].len - 10);
|
|
|
|
if (tp->max_queue == NGX_ERROR) {
|
|
ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
|
|
"invalid max_queue value \"%V\"", &value[i]);
|
|
return NGX_CONF_ERROR;
|
|
}
|
|
|
|
continue;
|
|
}
|
|
}
|
|
|
|
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;
|
|
}
|
|
|
|
|
|
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);
|
|
|
|
tpp = ngx_array_push(&tcf->pools);
|
|
if (tpp == NULL) {
|
|
return NULL;
|
|
}
|
|
|
|
*tpp = tp;
|
|
|
|
return tp;
|
|
}
|
|
|
|
|
|
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;
|
|
}
|
|
|
|
|
|
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;
|
|
|
|
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;
|
|
}
|
|
|
|
if (ngx_thread_pool_queue_init(&ngx_thread_pool_done, cycle->log)
|
|
!= NGX_OK)
|
|
{
|
|
return NGX_ERROR;
|
|
}
|
|
|
|
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 void
|
|
ngx_thread_pool_exit_worker(ngx_cycle_t *cycle)
|
|
{
|
|
ngx_uint_t i;
|
|
ngx_thread_pool_t **tpp;
|
|
ngx_thread_pool_conf_t *tcf;
|
|
|
|
if (ngx_process != NGX_PROCESS_WORKER
|
|
&& ngx_process != NGX_PROCESS_SINGLE)
|
|
{
|
|
return;
|
|
}
|
|
|
|
tcf = (ngx_thread_pool_conf_t *) ngx_get_conf(cycle->conf_ctx,
|
|
ngx_thread_pool_module);
|
|
|
|
if (tcf == NULL) {
|
|
return;
|
|
}
|
|
|
|
tpp = tcf->pools.elts;
|
|
|
|
for (i = 0; i < tcf->pools.nelts; i++) {
|
|
ngx_thread_pool_destroy(tpp[i]);
|
|
}
|
|
|
|
(void) ngx_thread_pool_queue_destroy(&ngx_thread_pool_done, cycle->log);
|
|
}
|