Mercurial > hg > nginx-quic
view src/core/ngx_thread_pool.c @ 7286:d27aa9060c95
Stream: udp streams.
Previously, only one client packet could be processed in a udp stream session
even though multiple response packets were supported. Now multiple packets
coming from the same client address and port are delivered to the same stream
session.
If it's required to maintain a single stream of data, nginx should be
configured in a way that all packets from a client are delivered to the same
worker. On Linux and DragonFly BSD the "reuseport" parameter should be
specified for this. Other systems do not currently provide appropriate
mechanisms. For these systems a single stream of udp packets is only
guaranteed in single-worker configurations.
The proxy_response directive now specifies how many packets are expected in
response to a single client packet.
author | Roman Arutyunyan <arut@nginx.com> |
---|---|
date | Mon, 04 Jun 2018 19:50:00 +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]); } }