Mercurial > hg > nginx-quic
view src/core/ngx_thread_pool.c @ 8456:c72d8839f427
Mail: fixed SMTP pipelining to send the response immediately.
Previously, if there were some pipelined SMTP data in the buffer when
a proxied connection with the backend was established, nginx called
ngx_mail_proxy_handler() to send these data, and not tried to send the
response to the last command. In most cases, this response was later sent
along with the response to the pipelined command, but if for some reason
client decides to wait for the response before finishing the next command
this might result in a connection hang.
Fix is to always call ngx_mail_proxy_handler() to send the response, and
additionally post an event to send the pipelined data if needed.
author | Maxim Dounin <mdounin@mdounin.ru> |
---|---|
date | Wed, 19 May 2021 03:13:12 +0300 |
parents | 33d075b9097d |
children |
line wrap: on
line source
/* * 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_task_t *first; ngx_thread_task_t **last; } ngx_thread_pool_queue_t; #define ngx_thread_pool_queue_init(q) \ (q)->first = NULL; \ (q)->last = &(q)->first struct ngx_thread_pool_s { ngx_thread_mutex_t mtx; ngx_thread_pool_queue_t queue; ngx_int_t waiting; ngx_thread_cond_t cond; ngx_log_t *log; 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 void ngx_thread_pool_destroy(ngx_thread_pool_t *tp); static void ngx_thread_pool_exit_handler(void *data, ngx_log_t *log); 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_atomic_t ngx_thread_pool_done_lock; 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; } 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; } err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); if (err) { ngx_log_error(NGX_LOG_ALERT, log, err, "pthread_attr_setdetachstate() 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 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++) { lock = 1; if (ngx_thread_task_post(tp, &task) != NGX_OK) { return; } while (lock) { ngx_sched_yield(); } task.event.active = 0; } (void) ngx_thread_cond_destroy(&tp->cond, tp->log); (void) ngx_thread_mutex_destroy(&tp->mtx, tp->log); } static void ngx_thread_pool_exit_handler(void *data, ngx_log_t *log) { ngx_uint_t *lock = data; *lock = 0; pthread_exit(0); } 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->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; } 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->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->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; } #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; ngx_spinlock(&ngx_thread_pool_done_lock, 1, 2048); *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); } } 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_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->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; } 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 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]); } }