technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
c6f9ca79e6a8517544a0995414de8421a9983687
nginx-unit/src/nxt_router.c
Max Romanov c6f9ca79e6 Fixing body fd access racing condition. 
To avoid closing the body fd prematurely, the fd value is moved from
the request struct to the app link.  The body fd should not be closed
immediately after the request is sent to the application due to possible
request rescheduling.
2020-03-17 14:44:11 +03:00

5392 lines
140 KiB
C
Raw Blame History

/*
* Copyright (C) Igor Sysoev
* Copyright (C) Valentin V. Bartenev
* Copyright (C) NGINX, Inc.
*/
#include <nxt_router.h>
#include <nxt_conf.h>
#if (NXT_TLS)
#include <nxt_cert.h>
#endif
#include <nxt_http.h>
#include <nxt_port_memory_int.h>
#include <nxt_unit_request.h>
#include <nxt_unit_response.h>
#include <nxt_router_request.h>
typedef struct {
nxt_str_t type;
uint32_t processes;
uint32_t max_processes;
uint32_t spare_processes;
nxt_msec_t timeout;
nxt_msec_t res_timeout;
nxt_msec_t idle_timeout;
uint32_t requests;
nxt_conf_value_t *limits_value;
nxt_conf_value_t *processes_value;
} nxt_router_app_conf_t;
typedef struct {
nxt_str_t pass;
nxt_str_t application;
} nxt_router_listener_conf_t;
#if (NXT_TLS)
typedef struct {
nxt_str_t name;
nxt_socket_conf_t *conf;
nxt_queue_link_t link; /* for nxt_socket_conf_t.tls */
} nxt_router_tlssock_t;
#endif
typedef struct {
nxt_socket_conf_t *socket_conf;
nxt_router_temp_conf_t *temp_conf;
} nxt_socket_rpc_t;
typedef struct {
nxt_app_t *app;
nxt_router_temp_conf_t *temp_conf;
} nxt_app_rpc_t;
struct nxt_port_select_state_s {
nxt_app_t *app;
nxt_request_app_link_t *req_app_link;
nxt_port_t *failed_port;
int failed_port_use_delta;
uint8_t start_process; /* 1 bit */
nxt_request_app_link_t *shared_ra;
nxt_port_t *port;
};
typedef struct nxt_port_select_state_s nxt_port_select_state_t;
static void nxt_router_greet_controller(nxt_task_t *task,
nxt_port_t *controller_port);
static void nxt_router_port_select(nxt_task_t *task,
nxt_port_select_state_t *state);
static nxt_int_t nxt_router_port_post_select(nxt_task_t *task,
nxt_port_select_state_t *state);
static nxt_int_t nxt_router_start_app_process(nxt_task_t *task, nxt_app_t *app);
static void nxt_request_app_link_update_peer(nxt_task_t *task,
nxt_request_app_link_t *req_app_link);
nxt_inline void
nxt_request_app_link_inc_use(nxt_request_app_link_t *req_app_link)
{
nxt_atomic_fetch_add(&req_app_link->use_count, 1);
}
nxt_inline void
nxt_request_app_link_chk_use(nxt_request_app_link_t *req_app_link, int i)
{
#if (NXT_DEBUG)
int c;
c = nxt_atomic_fetch_add(&req_app_link->use_count, i);
nxt_assert((c + i) > 0);
#else
(void) nxt_atomic_fetch_add(&req_app_link->use_count, i);
#endif
}
static void nxt_request_app_link_use(nxt_task_t *task,
nxt_request_app_link_t *req_app_link, int i);
static nxt_router_temp_conf_t *nxt_router_temp_conf(nxt_task_t *task);
static void nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data);
static void nxt_router_conf_ready(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf);
static void nxt_router_conf_error(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf);
static void nxt_router_conf_send(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_port_msg_type_t type);
static nxt_int_t nxt_router_conf_create(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, u_char *start, u_char *end);
static nxt_int_t nxt_router_conf_process_static(nxt_task_t *task,
nxt_router_conf_t *rtcf, nxt_conf_value_t *conf);
static nxt_app_t *nxt_router_app_find(nxt_queue_t *queue, nxt_str_t *name);
static void nxt_router_listen_socket_rpc_create(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_socket_conf_t *skcf);
static void nxt_router_listen_socket_ready(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_listen_socket_error(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
#if (NXT_TLS)
static void nxt_router_tls_rpc_create(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_router_tlssock_t *tls);
static void nxt_router_tls_rpc_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
#endif
static void nxt_router_app_rpc_create(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_app_t *app);
static void nxt_router_app_prefork_ready(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_app_prefork_error(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static nxt_socket_conf_t *nxt_router_socket_conf(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_str_t *name);
static nxt_int_t nxt_router_listen_socket_find(nxt_router_temp_conf_t *tmcf,
nxt_socket_conf_t *nskcf, nxt_sockaddr_t *sa);
static nxt_int_t nxt_router_engines_create(nxt_task_t *task,
nxt_router_t *router, nxt_router_temp_conf_t *tmcf,
const nxt_event_interface_t *interface);
static nxt_int_t nxt_router_engine_conf_create(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf);
static nxt_int_t nxt_router_engine_conf_update(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf);
static nxt_int_t nxt_router_engine_conf_delete(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf);
static nxt_int_t nxt_router_engine_joints_create(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf, nxt_queue_t *sockets,
nxt_work_handler_t handler);
static nxt_int_t nxt_router_engine_quit(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf);
static nxt_int_t nxt_router_engine_joints_delete(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf, nxt_queue_t *sockets);
static nxt_int_t nxt_router_threads_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_router_temp_conf_t *tmcf);
static nxt_int_t nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_event_engine_t *engine);
static void nxt_router_apps_sort(nxt_task_t *task, nxt_router_t *router,
nxt_router_temp_conf_t *tmcf);
static void nxt_router_engines_post(nxt_router_t *router,
nxt_router_temp_conf_t *tmcf);
static void nxt_router_engine_post(nxt_event_engine_t *engine,
nxt_work_t *jobs);
static void nxt_router_thread_start(void *data);
static void nxt_router_listen_socket_create(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_listen_socket_update(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_listen_socket_delete(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_worker_thread_quit(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_listen_socket_close(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_thread_exit_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_listen_socket_release(nxt_task_t *task,
nxt_socket_conf_t *skcf);
static void nxt_router_access_log_writer(nxt_task_t *task,
nxt_http_request_t *r, nxt_router_access_log_t *access_log);
static u_char *nxt_router_access_log_date(u_char *buf, nxt_realtime_t *now,
struct tm *tm, size_t size, const char *format);
static void nxt_router_access_log_open(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf);
static void nxt_router_access_log_ready(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_access_log_error(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_access_log_release(nxt_task_t *task,
nxt_thread_spinlock_t *lock, nxt_router_access_log_t *access_log);
static void nxt_router_access_log_reopen_completion(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_access_log_reopen_ready(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_access_log_reopen_error(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_app_port_ready(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_app_port_error(nxt_task_t *task,
nxt_port_recv_msg_t *msg, void *data);
static void nxt_router_app_unlink(nxt_task_t *task, nxt_app_t *app);
static void nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
nxt_apr_action_t action);
static nxt_int_t nxt_router_app_port(nxt_task_t *task, nxt_app_t *app,
nxt_request_app_link_t *req_app_link);
static void nxt_router_app_prepare_request(nxt_task_t *task,
nxt_request_app_link_t *req_app_link);
static nxt_buf_t *nxt_router_prepare_msg(nxt_task_t *task,
nxt_http_request_t *r, nxt_port_t *port, const nxt_str_t *prefix);
static void nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data);
static void nxt_router_adjust_idle_timer(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_app_idle_timeout(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_app_joint_release_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_free_app(nxt_task_t *task, void *obj, void *data);
static const nxt_http_request_state_t nxt_http_request_send_state;
static void nxt_http_request_send_body(nxt_task_t *task, void *obj, void *data);
static void nxt_router_app_joint_use(nxt_task_t *task,
nxt_app_joint_t *app_joint, int i);
static nxt_int_t nxt_router_http_request_done(nxt_task_t *task,
nxt_http_request_t *r);
static void nxt_router_http_request_release(nxt_task_t *task, void *obj,
void *data);
static void nxt_router_oosm_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
extern const nxt_http_request_state_t nxt_http_websocket;
static nxt_router_t *nxt_router;
static const nxt_str_t http_prefix = nxt_string("HTTP_");
static const nxt_str_t empty_prefix = nxt_string("");
static const nxt_str_t *nxt_app_msg_prefix[] = {
&empty_prefix,
&http_prefix,
&http_prefix,
&http_prefix,
&http_prefix,
&empty_prefix,
};
nxt_port_handlers_t nxt_router_process_port_handlers = {
.quit = nxt_worker_process_quit_handler,
.new_port = nxt_router_new_port_handler,
.change_file = nxt_port_change_log_file_handler,
.mmap = nxt_port_mmap_handler,
.data = nxt_router_conf_data_handler,
.remove_pid = nxt_router_remove_pid_handler,
.access_log = nxt_router_access_log_reopen_handler,
.rpc_ready = nxt_port_rpc_handler,
.rpc_error = nxt_port_rpc_handler,
.oosm = nxt_router_oosm_handler,
};
nxt_int_t
nxt_router_start(nxt_task_t *task, void *data)
{
nxt_int_t ret;
nxt_port_t *controller_port;
nxt_router_t *router;
nxt_runtime_t *rt;
rt = task->thread->runtime;
#if (NXT_TLS)
rt->tls = nxt_service_get(rt->services, "SSL/TLS", "OpenSSL");
if (nxt_slow_path(rt->tls == NULL)) {
return NXT_ERROR;
}
ret = rt->tls->library_init(task);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
#endif
ret = nxt_http_init(task, rt);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
router = nxt_zalloc(sizeof(nxt_router_t));
if (nxt_slow_path(router == NULL)) {
return NXT_ERROR;
}
nxt_queue_init(&router->engines);
nxt_queue_init(&router->sockets);
nxt_queue_init(&router->apps);
nxt_router = router;
controller_port = rt->port_by_type[NXT_PROCESS_CONTROLLER];
if (controller_port != NULL) {
nxt_router_greet_controller(task, controller_port);
}
return NXT_OK;
}
static void
nxt_router_greet_controller(nxt_task_t *task, nxt_port_t *controller_port)
{
nxt_port_socket_write(task, controller_port, NXT_PORT_MSG_PROCESS_READY,
-1, 0, 0, NULL);
}
static void
nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port,
void *data)
{
size_t size;
uint32_t stream;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_app_t *app;
nxt_buf_t *b;
nxt_port_t *main_port;
nxt_runtime_t *rt;
app = data;
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
nxt_debug(task, "app '%V' %p start process", &app->name, app);
size = app->name.length + 1 + app->conf.length;
b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool, size);
if (nxt_slow_path(b == NULL)) {
goto failed;
}
nxt_buf_cpystr(b, &app->name);
*b->mem.free++ = '\0';
nxt_buf_cpystr(b, &app->conf);
nxt_router_app_joint_use(task, app->joint, 1);
stream = nxt_port_rpc_register_handler(task, port,
nxt_router_app_port_ready,
nxt_router_app_port_error,
-1, app->joint);
if (nxt_slow_path(stream == 0)) {
nxt_router_app_joint_use(task, app->joint, -1);
goto failed;
}
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_WORKER, -1,
stream, port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, port, stream);
nxt_router_app_joint_use(task, app->joint, -1);
goto failed;
}
nxt_router_app_use(task, app, -1);
return;
failed:
if (b != NULL) {
mp = b->data;
nxt_mp_free(mp, b);
nxt_mp_release(mp);
}
nxt_thread_mutex_lock(&app->mutex);
app->pending_processes--;
nxt_thread_mutex_unlock(&app->mutex);
nxt_router_app_use(task, app, -1);
}
static void
nxt_router_app_joint_use(nxt_task_t *task, nxt_app_joint_t *app_joint, int i)
{
app_joint->use_count += i;
if (app_joint->use_count == 0) {
nxt_assert(app_joint->app == NULL);
nxt_free(app_joint);
}
}
static nxt_int_t
nxt_router_start_app_process(nxt_task_t *task, nxt_app_t *app)
{
nxt_int_t res;
nxt_port_t *router_port;
nxt_runtime_t *rt;
rt = task->thread->runtime;
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
nxt_router_app_use(task, app, 1);
res = nxt_port_post(task, router_port, nxt_router_start_app_process_handler,
app);
if (res == NXT_OK) {
return res;
}
nxt_thread_mutex_lock(&app->mutex);
app->pending_processes--;
nxt_thread_mutex_unlock(&app->mutex);
nxt_router_app_use(task, app, -1);
return NXT_ERROR;
}
nxt_inline void
nxt_request_app_link_init(nxt_task_t *task,
nxt_request_app_link_t *req_app_link, nxt_request_rpc_data_t *req_rpc_data)
{
nxt_buf_t *body;
nxt_event_engine_t *engine;
engine = task->thread->engine;
nxt_memzero(req_app_link, sizeof(nxt_request_app_link_t));
req_app_link->stream = req_rpc_data->stream;
req_app_link->use_count = 1;
req_app_link->req_rpc_data = req_rpc_data;
req_rpc_data->req_app_link = req_app_link;
req_app_link->reply_port = engine->port;
req_app_link->request = req_rpc_data->request;
req_app_link->apr_action = NXT_APR_GOT_RESPONSE;
req_app_link->work.handler = NULL;
req_app_link->work.task = &engine->task;
req_app_link->work.obj = req_app_link;
req_app_link->work.data = engine;
body = req_rpc_data->request->body;
if (body != NULL && nxt_buf_is_file(body)) {
req_app_link->body_fd = body->file->fd;
body->file->fd = -1;
} else {
req_app_link->body_fd = -1;
}
}
nxt_inline nxt_request_app_link_t *
nxt_request_app_link_alloc(nxt_task_t *task,
nxt_request_app_link_t *ra_src, nxt_request_rpc_data_t *req_rpc_data)
{
nxt_mp_t *mp;
nxt_request_app_link_t *req_app_link;
if (ra_src != NULL && ra_src->mem_pool != NULL) {
return ra_src;
}
mp = req_rpc_data->request->mem_pool;
req_app_link = nxt_mp_alloc(mp, sizeof(nxt_request_app_link_t));
if (nxt_slow_path(req_app_link == NULL)) {
req_rpc_data->req_app_link = NULL;
if (ra_src != NULL) {
ra_src->req_rpc_data = NULL;
}
return NULL;
}
nxt_mp_retain(mp);
nxt_request_app_link_init(task, req_app_link, req_rpc_data);
if (ra_src != NULL) {
req_app_link->body_fd = ra_src->body_fd;
}
req_app_link->mem_pool = mp;
return req_app_link;
}
nxt_inline nxt_bool_t
nxt_router_msg_cancel(nxt_task_t *task, nxt_msg_info_t *msg_info,
uint32_t stream)
{
nxt_buf_t *b, *next;
nxt_bool_t cancelled;
if (msg_info->buf == NULL) {
return 0;
}
cancelled = nxt_port_mmap_tracking_cancel(task, &msg_info->tracking,
stream);
if (cancelled) {
nxt_debug(task, "stream #%uD: cancelled by router", stream);
}
for (b = msg_info->buf; b != NULL; b = next) {
next = b->next;
b->next = NULL;
b->completion_handler = msg_info->completion_handler;
if (b->is_port_mmap_sent) {
b->is_port_mmap_sent = cancelled == 0;
b->completion_handler(task, b, b->parent);
}
}
msg_info->buf = NULL;
return cancelled;
}
static void
nxt_request_app_link_update_peer_handler(nxt_task_t *task, void *obj,
void *data)
{
nxt_request_app_link_t *req_app_link;
req_app_link = obj;
nxt_request_app_link_update_peer(task, req_app_link);
nxt_request_app_link_use(task, req_app_link, -1);
}
static void
nxt_request_app_link_update_peer(nxt_task_t *task,
nxt_request_app_link_t *req_app_link)
{
nxt_event_engine_t *engine;
nxt_request_rpc_data_t *req_rpc_data;
engine = req_app_link->work.data;
if (task->thread->engine != engine) {
nxt_request_app_link_inc_use(req_app_link);
req_app_link->work.handler = nxt_request_app_link_update_peer_handler;
req_app_link->work.task = &engine->task;
req_app_link->work.next = NULL;
nxt_debug(task, "req_app_link stream #%uD post update peer to %p",
req_app_link->stream, engine);
nxt_event_engine_post(engine, &req_app_link->work);
return;
}
nxt_debug(task, "req_app_link stream #%uD update peer",
req_app_link->stream);
req_rpc_data = req_app_link->req_rpc_data;
if (req_rpc_data != NULL && req_app_link->app_port != NULL) {
nxt_port_rpc_ex_set_peer(task, engine->port, req_rpc_data,
req_app_link->app_port->pid);
}
}
static void
nxt_request_app_link_release(nxt_task_t *task,
nxt_request_app_link_t *req_app_link)
{
nxt_mp_t *mp;
nxt_http_request_t *r;
nxt_request_rpc_data_t *req_rpc_data;
nxt_assert(task->thread->engine == req_app_link->work.data);
nxt_assert(req_app_link->use_count == 0);
nxt_debug(task, "req_app_link stream #%uD release", req_app_link->stream);
req_rpc_data = req_app_link->req_rpc_data;
if (req_rpc_data != NULL) {
if (nxt_slow_path(req_app_link->err_code != 0)) {
nxt_http_request_error(task, req_rpc_data->request,
req_app_link->err_code);
} else {
req_rpc_data->app_port = req_app_link->app_port;
req_rpc_data->apr_action = req_app_link->apr_action;
req_rpc_data->msg_info = req_app_link->msg_info;
if (req_rpc_data->app->timeout != 0) {
r = req_rpc_data->request;
r->timer.handler = nxt_router_app_timeout;
r->timer_data = req_rpc_data;
nxt_timer_add(task->thread->engine, &r->timer,
req_rpc_data->app->timeout);
}
req_app_link->app_port = NULL;
req_app_link->msg_info.buf = NULL;
}
req_rpc_data->req_app_link = NULL;
req_app_link->req_rpc_data = NULL;
}
if (req_app_link->app_port != NULL) {
nxt_router_app_port_release(task, req_app_link->app_port,
req_app_link->apr_action);
req_app_link->app_port = NULL;
}
if (req_app_link->body_fd != -1) {
nxt_fd_close(req_app_link->body_fd);
req_app_link->body_fd = -1;
}
nxt_router_msg_cancel(task, &req_app_link->msg_info, req_app_link->stream);
mp = req_app_link->mem_pool;
if (mp != NULL) {
nxt_mp_free(mp, req_app_link);
nxt_mp_release(mp);
}
}
static void
nxt_request_app_link_release_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_request_app_link_t *req_app_link;
req_app_link = obj;
nxt_assert(req_app_link->work.data == data);
nxt_atomic_fetch_add(&req_app_link->use_count, -1);
nxt_request_app_link_release(task, req_app_link);
}
static void
nxt_request_app_link_use(nxt_task_t *task, nxt_request_app_link_t *req_app_link,
int i)
{
int c;
nxt_event_engine_t *engine;
c = nxt_atomic_fetch_add(&req_app_link->use_count, i);
if (i < 0 && c == -i) {
engine = req_app_link->work.data;
if (task->thread->engine == engine) {
nxt_request_app_link_release(task, req_app_link);
return;
}
nxt_request_app_link_inc_use(req_app_link);
req_app_link->work.handler = nxt_request_app_link_release_handler;
req_app_link->work.task = &engine->task;
req_app_link->work.next = NULL;
nxt_debug(task, "req_app_link stream #%uD post release to %p",
req_app_link->stream, engine);
nxt_event_engine_post(engine, &req_app_link->work);
}
}
nxt_inline void
nxt_request_app_link_error(nxt_request_app_link_t *req_app_link, int code,
const char *str)
{
req_app_link->app_port = NULL;
req_app_link->err_code = code;
req_app_link->err_str = str;
}
nxt_inline void
nxt_request_app_link_pending(nxt_task_t *task, nxt_app_t *app,
nxt_request_app_link_t *req_app_link)
{
nxt_queue_insert_tail(&req_app_link->app_port->pending_requests,
&req_app_link->link_port_pending);
nxt_queue_insert_tail(&app->pending, &req_app_link->link_app_pending);
nxt_request_app_link_inc_use(req_app_link);
req_app_link->res_time = nxt_thread_monotonic_time(task->thread)
+ app->res_timeout;
nxt_debug(task, "req_app_link stream #%uD enqueue to pending_requests",
req_app_link->stream);
}
nxt_inline nxt_bool_t
nxt_queue_chk_remove(nxt_queue_link_t *lnk)
{
if (lnk->next != NULL) {
nxt_queue_remove(lnk);
lnk->next = NULL;
return 1;
}
return 0;
}
nxt_inline void
nxt_request_rpc_data_unlink(nxt_task_t *task,
nxt_request_rpc_data_t *req_rpc_data)
{
int ra_use_delta;
nxt_request_app_link_t *req_app_link;
if (req_rpc_data->app_port != NULL) {
nxt_router_app_port_release(task, req_rpc_data->app_port,
req_rpc_data->apr_action);
req_rpc_data->app_port = NULL;
}
nxt_router_msg_cancel(task, &req_rpc_data->msg_info, req_rpc_data->stream);
req_app_link = req_rpc_data->req_app_link;
if (req_app_link != NULL) {
req_rpc_data->req_app_link = NULL;
req_app_link->req_rpc_data = NULL;
ra_use_delta = 0;
nxt_thread_mutex_lock(&req_rpc_data->app->mutex);
if (req_app_link->link_app_requests.next == NULL
&& req_app_link->link_port_pending.next == NULL
&& req_app_link->link_app_pending.next == NULL
&& req_app_link->link_port_websockets.next == NULL)
{
req_app_link = NULL;
} else {
ra_use_delta -=
nxt_queue_chk_remove(&req_app_link->link_app_requests)
+ nxt_queue_chk_remove(&req_app_link->link_port_pending)
+ nxt_queue_chk_remove(&req_app_link->link_port_websockets);
nxt_queue_chk_remove(&req_app_link->link_app_pending);
}
nxt_thread_mutex_unlock(&req_rpc_data->app->mutex);
if (req_app_link != NULL) {
nxt_request_app_link_use(task, req_app_link, ra_use_delta);
}
}
if (req_rpc_data->app != NULL) {
nxt_router_app_use(task, req_rpc_data->app, -1);
req_rpc_data->app = NULL;
}
if (req_rpc_data->request != NULL) {
req_rpc_data->request->timer_data = NULL;
nxt_router_http_request_done(task, req_rpc_data->request);
req_rpc_data->request->req_rpc_data = NULL;
req_rpc_data->request = NULL;
}
}
void
nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_port_new_port_handler(task, msg);
if (msg->u.new_port != NULL
&& msg->u.new_port->type == NXT_PROCESS_CONTROLLER)
{
nxt_router_greet_controller(task, msg->u.new_port);
}
if (msg->port_msg.stream == 0) {
return;
}
if (msg->u.new_port == NULL
|| msg->u.new_port->type != NXT_PROCESS_WORKER)
{
msg->port_msg.type = _NXT_PORT_MSG_RPC_ERROR;
}
nxt_port_rpc_handler(task, msg);
}
void
nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_int_t ret;
nxt_buf_t *b;
nxt_router_temp_conf_t *tmcf;
tmcf = nxt_router_temp_conf(task);
if (nxt_slow_path(tmcf == NULL)) {
return;
}
nxt_debug(task, "nxt_router_conf_data_handler(%O): %*s",
nxt_buf_used_size(msg->buf),
(size_t) nxt_buf_used_size(msg->buf), msg->buf->mem.pos);
tmcf->router_conf->router = nxt_router;
tmcf->stream = msg->port_msg.stream;
tmcf->port = nxt_runtime_port_find(task->thread->runtime,
msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_slow_path(tmcf->port == NULL)) {
nxt_alert(task, "reply port not found");
return;
}
nxt_port_use(task, tmcf->port, 1);
b = nxt_buf_chk_make_plain(tmcf->router_conf->mem_pool,
msg->buf, msg->size);
if (nxt_slow_path(b == NULL)) {
nxt_router_conf_error(task, tmcf);
return;
}
ret = nxt_router_conf_create(task, tmcf, b->mem.pos, b->mem.free);
if (nxt_fast_path(ret == NXT_OK)) {
nxt_router_conf_apply(task, tmcf, NULL);
} else {
nxt_router_conf_error(task, tmcf);
}
}
static void
nxt_router_app_process_remove_pid(nxt_task_t *task, nxt_port_t *port,
void *data)
{
union {
nxt_pid_t removed_pid;
void *data;
} u;
u.data = data;
nxt_port_rpc_remove_peer(task, port, u.removed_pid);
}
void
nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_event_engine_t *engine;
nxt_port_remove_pid_handler(task, msg);
nxt_queue_each(engine, &nxt_router->engines, nxt_event_engine_t, link0)
{
nxt_port_post(task, engine->port, nxt_router_app_process_remove_pid,
msg->u.data);
}
nxt_queue_loop;
if (msg->port_msg.stream == 0) {
return;
}
msg->port_msg.type = _NXT_PORT_MSG_RPC_ERROR;
nxt_port_rpc_handler(task, msg);
}
static nxt_router_temp_conf_t *
nxt_router_temp_conf(nxt_task_t *task)
{
nxt_mp_t *mp, *tmp;
nxt_router_conf_t *rtcf;
nxt_router_temp_conf_t *tmcf;
mp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(mp == NULL)) {
return NULL;
}
rtcf = nxt_mp_zget(mp, sizeof(nxt_router_conf_t));
if (nxt_slow_path(rtcf == NULL)) {
goto fail;
}
rtcf->mem_pool = mp;
tmp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(tmp == NULL)) {
goto fail;
}
tmcf = nxt_mp_zget(tmp, sizeof(nxt_router_temp_conf_t));
if (nxt_slow_path(tmcf == NULL)) {
goto temp_fail;
}
tmcf->mem_pool = tmp;
tmcf->router_conf = rtcf;
tmcf->count = 1;
tmcf->engine = task->thread->engine;
tmcf->engines = nxt_array_create(tmcf->mem_pool, 4,
sizeof(nxt_router_engine_conf_t));
if (nxt_slow_path(tmcf->engines == NULL)) {
goto temp_fail;
}
nxt_queue_init(&tmcf->deleting);
nxt_queue_init(&tmcf->keeping);
nxt_queue_init(&tmcf->updating);
nxt_queue_init(&tmcf->pending);
nxt_queue_init(&tmcf->creating);
#if (NXT_TLS)
nxt_queue_init(&tmcf->tls);
#endif
nxt_queue_init(&tmcf->apps);
nxt_queue_init(&tmcf->previous);
return tmcf;
temp_fail:
nxt_mp_destroy(tmp);
fail:
nxt_mp_destroy(mp);
return NULL;
}
nxt_inline nxt_bool_t
nxt_router_app_can_start(nxt_app_t *app)
{
return app->processes + app->pending_processes < app->max_processes
&& app->pending_processes < app->max_pending_processes;
}
nxt_inline nxt_bool_t
nxt_router_app_need_start(nxt_app_t *app)
{
return app->idle_processes + app->pending_processes
< app->spare_processes;
}
static void
nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data)
{
nxt_int_t ret;
nxt_app_t *app;
nxt_router_t *router;
nxt_runtime_t *rt;
nxt_queue_link_t *qlk;
nxt_socket_conf_t *skcf;
nxt_router_conf_t *rtcf;
nxt_router_temp_conf_t *tmcf;
const nxt_event_interface_t *interface;
#if (NXT_TLS)
nxt_router_tlssock_t *tls;
#endif
tmcf = obj;
qlk = nxt_queue_first(&tmcf->pending);
if (qlk != nxt_queue_tail(&tmcf->pending)) {
nxt_queue_remove(qlk);
nxt_queue_insert_tail(&tmcf->creating, qlk);
skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
nxt_router_listen_socket_rpc_create(task, tmcf, skcf);
return;
}
#if (NXT_TLS)
qlk = nxt_queue_first(&tmcf->tls);
if (qlk != nxt_queue_tail(&tmcf->tls)) {
nxt_queue_remove(qlk);
tls = nxt_queue_link_data(qlk, nxt_router_tlssock_t, link);
nxt_router_tls_rpc_create(task, tmcf, tls);
return;
}
#endif
nxt_queue_each(app, &tmcf->apps, nxt_app_t, link) {
if (nxt_router_app_need_start(app)) {
nxt_router_app_rpc_create(task, tmcf, app);
return;
}
} nxt_queue_loop;
rtcf = tmcf->router_conf;
if (rtcf->access_log != NULL && rtcf->access_log->fd == -1) {
nxt_router_access_log_open(task, tmcf);
return;
}
rt = task->thread->runtime;
interface = nxt_service_get(rt->services, "engine", NULL);
router = rtcf->router;
ret = nxt_router_engines_create(task, router, tmcf, interface);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
ret = nxt_router_threads_create(task, rt, tmcf);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
nxt_router_apps_sort(task, router, tmcf);
nxt_router_engines_post(router, tmcf);
nxt_queue_add(&router->sockets, &tmcf->updating);
nxt_queue_add(&router->sockets, &tmcf->creating);
router->access_log = rtcf->access_log;
nxt_router_conf_ready(task, tmcf);
return;
fail:
nxt_router_conf_error(task, tmcf);
return;
}
static void
nxt_router_conf_wait(nxt_task_t *task, void *obj, void *data)
{
nxt_joint_job_t *job;
job = obj;
nxt_router_conf_ready(task, job->tmcf);
}
static void
nxt_router_conf_ready(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
{
nxt_debug(task, "temp conf count:%D", tmcf->count);
if (--tmcf->count == 0) {
nxt_router_conf_send(task, tmcf, NXT_PORT_MSG_RPC_READY_LAST);
}
}
static void
nxt_router_conf_error(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
{
nxt_app_t *app;
nxt_queue_t new_socket_confs;
nxt_socket_t s;
nxt_router_t *router;
nxt_queue_link_t *qlk;
nxt_socket_conf_t *skcf;
nxt_router_conf_t *rtcf;
nxt_alert(task, "failed to apply new conf");
for (qlk = nxt_queue_first(&tmcf->creating);
qlk != nxt_queue_tail(&tmcf->creating);
qlk = nxt_queue_next(qlk))
{
skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
s = skcf->listen->socket;
if (s != -1) {
nxt_socket_close(task, s);
}
nxt_free(skcf->listen);
}
nxt_queue_init(&new_socket_confs);
nxt_queue_add(&new_socket_confs, &tmcf->updating);
nxt_queue_add(&new_socket_confs, &tmcf->pending);
nxt_queue_add(&new_socket_confs, &tmcf->creating);
rtcf = tmcf->router_conf;
nxt_http_routes_cleanup(task, rtcf->routes);
nxt_queue_each(skcf, &new_socket_confs, nxt_socket_conf_t, link) {
if (skcf->action != NULL) {
nxt_http_action_cleanup(task, skcf->action);
}
} nxt_queue_loop;
nxt_queue_each(app, &tmcf->apps, nxt_app_t, link) {
nxt_router_app_unlink(task, app);
} nxt_queue_loop;
router = rtcf->router;
nxt_queue_add(&router->sockets, &tmcf->keeping);
nxt_queue_add(&router->sockets, &tmcf->deleting);
nxt_queue_add(&router->apps, &tmcf->previous);
// TODO: new engines and threads
nxt_router_access_log_release(task, &router->lock, rtcf->access_log);
nxt_mp_destroy(rtcf->mem_pool);
nxt_router_conf_send(task, tmcf, NXT_PORT_MSG_RPC_ERROR);
}
static void
nxt_router_conf_send(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
nxt_port_msg_type_t type)
{
nxt_port_socket_write(task, tmcf->port, type, -1, tmcf->stream, 0, NULL);
nxt_port_use(task, tmcf->port, -1);
tmcf->port = NULL;
}
static nxt_conf_map_t nxt_router_conf[] = {
{
nxt_string("listeners_threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_router_conf_t, threads),
},
};
static nxt_conf_map_t nxt_router_app_conf[] = {
{
nxt_string("type"),
NXT_CONF_MAP_STR,
offsetof(nxt_router_app_conf_t, type),
},
{
nxt_string("limits"),
NXT_CONF_MAP_PTR,
offsetof(nxt_router_app_conf_t, limits_value),
},
{
nxt_string("processes"),
NXT_CONF_MAP_INT32,
offsetof(nxt_router_app_conf_t, processes),
},
{
nxt_string("processes"),
NXT_CONF_MAP_PTR,
offsetof(nxt_router_app_conf_t, processes_value),
},
};
static nxt_conf_map_t nxt_router_app_limits_conf[] = {
{
nxt_string("timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_router_app_conf_t, timeout),
},
{
nxt_string("reschedule_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_router_app_conf_t, res_timeout),
},
{
nxt_string("requests"),
NXT_CONF_MAP_INT32,
offsetof(nxt_router_app_conf_t, requests),
},
};
static nxt_conf_map_t nxt_router_app_processes_conf[] = {
{
nxt_string("spare"),
NXT_CONF_MAP_INT32,
offsetof(nxt_router_app_conf_t, spare_processes),
},
{
nxt_string("max"),
NXT_CONF_MAP_INT32,
offsetof(nxt_router_app_conf_t, max_processes),
},
{
nxt_string("idle_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_router_app_conf_t, idle_timeout),
},
};
static nxt_conf_map_t nxt_router_listener_conf[] = {
{
nxt_string("pass"),
NXT_CONF_MAP_STR_COPY,
offsetof(nxt_router_listener_conf_t, pass),
},
{
nxt_string("application"),
NXT_CONF_MAP_STR_COPY,
offsetof(nxt_router_listener_conf_t, application),
},
};
static nxt_conf_map_t nxt_router_http_conf[] = {
{
nxt_string("header_buffer_size"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_socket_conf_t, header_buffer_size),
},
{
nxt_string("large_header_buffer_size"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_socket_conf_t, large_header_buffer_size),
},
{
nxt_string("large_header_buffers"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_socket_conf_t, large_header_buffers),
},
{
nxt_string("body_buffer_size"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_socket_conf_t, body_buffer_size),
},
{
nxt_string("max_body_size"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_socket_conf_t, max_body_size),
},
{
nxt_string("idle_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_socket_conf_t, idle_timeout),
},
{
nxt_string("header_read_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_socket_conf_t, header_read_timeout),
},
{
nxt_string("body_read_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_socket_conf_t, body_read_timeout),
},
{
nxt_string("send_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_socket_conf_t, send_timeout),
},
{
nxt_string("body_temp_path"),
NXT_CONF_MAP_STR,
offsetof(nxt_socket_conf_t, body_temp_path),
},
};
static nxt_conf_map_t nxt_router_websocket_conf[] = {
{
nxt_string("max_frame_size"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_websocket_conf_t, max_frame_size),
},
{
nxt_string("read_timeout"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_websocket_conf_t, read_timeout),
},
{
nxt_string("keepalive_interval"),
NXT_CONF_MAP_MSEC,
offsetof(nxt_websocket_conf_t, keepalive_interval),
},
};
static nxt_int_t
nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
u_char *start, u_char *end)
{
u_char *p;
size_t size;
nxt_mp_t *mp;
uint32_t next;
nxt_int_t ret;
nxt_str_t name, path;
nxt_app_t *app, *prev;
nxt_str_t *t;
nxt_router_t *router;
nxt_app_joint_t *app_joint;
nxt_conf_value_t *conf, *http, *value, *websocket;
nxt_conf_value_t *applications, *application;
nxt_conf_value_t *listeners, *listener;
nxt_conf_value_t *routes_conf, *static_conf;
nxt_socket_conf_t *skcf;
nxt_http_routes_t *routes;
nxt_event_engine_t *engine;
nxt_app_lang_module_t *lang;
nxt_router_app_conf_t apcf;
nxt_router_access_log_t *access_log;
nxt_router_listener_conf_t lscf;
#if (NXT_TLS)
nxt_router_tlssock_t *tls;
#endif
static nxt_str_t http_path = nxt_string("/settings/http");
static nxt_str_t applications_path = nxt_string("/applications");
static nxt_str_t listeners_path = nxt_string("/listeners");
static nxt_str_t routes_path = nxt_string("/routes");
static nxt_str_t access_log_path = nxt_string("/access_log");
#if (NXT_TLS)
static nxt_str_t certificate_path = nxt_string("/tls/certificate");
#endif
static nxt_str_t static_path = nxt_string("/settings/http/static");
static nxt_str_t websocket_path = nxt_string("/settings/http/websocket");
conf = nxt_conf_json_parse(tmcf->mem_pool, start, end, NULL);
if (conf == NULL) {
nxt_alert(task, "configuration parsing error");
return NXT_ERROR;
}
mp = tmcf->router_conf->mem_pool;
ret = nxt_conf_map_object(mp, conf, nxt_router_conf,
nxt_nitems(nxt_router_conf), tmcf->router_conf);
if (ret != NXT_OK) {
nxt_alert(task, "root map error");
return NXT_ERROR;
}
if (tmcf->router_conf->threads == 0) {
tmcf->router_conf->threads = nxt_ncpu;
}
static_conf = nxt_conf_get_path(conf, &static_path);
ret = nxt_router_conf_process_static(task, tmcf->router_conf, static_conf);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
router = tmcf->router_conf->router;
applications = nxt_conf_get_path(conf, &applications_path);
if (applications != NULL) {
next = 0;
for ( ;; ) {
application = nxt_conf_next_object_member(applications,
&name, &next);
if (application == NULL) {
break;
}
nxt_debug(task, "application \"%V\"", &name);
size = nxt_conf_json_length(application, NULL);
app = nxt_malloc(sizeof(nxt_app_t) + name.length + size);
if (app == NULL) {
goto fail;
}
nxt_memzero(app, sizeof(nxt_app_t));
app->name.start = nxt_pointer_to(app, sizeof(nxt_app_t));
app->conf.start = nxt_pointer_to(app, sizeof(nxt_app_t)
+ name.length);
p = nxt_conf_json_print(app->conf.start, application, NULL);
app->conf.length = p - app->conf.start;
nxt_assert(app->conf.length <= size);
nxt_debug(task, "application conf \"%V\"", &app->conf);
prev = nxt_router_app_find(&router->apps, &name);
if (prev != NULL && nxt_strstr_eq(&app->conf, &prev->conf)) {
nxt_free(app);
nxt_queue_remove(&prev->link);
nxt_queue_insert_tail(&tmcf->previous, &prev->link);
continue;
}
apcf.processes = 1;
apcf.max_processes = 1;
apcf.spare_processes = 0;
apcf.timeout = 0;
apcf.res_timeout = 1000;
apcf.idle_timeout = 15000;
apcf.requests = 0;
apcf.limits_value = NULL;
apcf.processes_value = NULL;
app_joint = nxt_malloc(sizeof(nxt_app_joint_t));
if (nxt_slow_path(app_joint == NULL)) {
goto app_fail;
}
nxt_memzero(app_joint, sizeof(nxt_app_joint_t));
ret = nxt_conf_map_object(mp, application, nxt_router_app_conf,
nxt_nitems(nxt_router_app_conf), &apcf);
if (ret != NXT_OK) {
nxt_alert(task, "application map error");
goto app_fail;
}
if (apcf.limits_value != NULL) {
if (nxt_conf_type(apcf.limits_value) != NXT_CONF_OBJECT) {
nxt_alert(task, "application limits is not object");
goto app_fail;
}
ret = nxt_conf_map_object(mp, apcf.limits_value,
nxt_router_app_limits_conf,
nxt_nitems(nxt_router_app_limits_conf),
&apcf);
if (ret != NXT_OK) {
nxt_alert(task, "application limits map error");
goto app_fail;
}
}
if (apcf.processes_value != NULL
&& nxt_conf_type(apcf.processes_value) == NXT_CONF_OBJECT)
{
ret = nxt_conf_map_object(mp, apcf.processes_value,
nxt_router_app_processes_conf,
nxt_nitems(nxt_router_app_processes_conf),
&apcf);
if (ret != NXT_OK) {
nxt_alert(task, "application processes map error");
goto app_fail;
}
} else {
apcf.max_processes = apcf.processes;
apcf.spare_processes = apcf.processes;
}
nxt_debug(task, "application type: %V", &apcf.type);
nxt_debug(task, "application processes: %D", apcf.processes);
nxt_debug(task, "application request timeout: %M", apcf.timeout);
nxt_debug(task, "application reschedule timeout: %M",
apcf.res_timeout);
nxt_debug(task, "application requests: %D", apcf.requests);
lang = nxt_app_lang_module(task->thread->runtime, &apcf.type);
if (lang == NULL) {
nxt_alert(task, "unknown application type: \"%V\"", &apcf.type);
goto app_fail;
}
nxt_debug(task, "application language module: \"%s\"", lang->file);
ret = nxt_thread_mutex_create(&app->mutex);
if (ret != NXT_OK) {
goto app_fail;
}
nxt_queue_init(&app->ports);
nxt_queue_init(&app->spare_ports);
nxt_queue_init(&app->idle_ports);
nxt_queue_init(&app->requests);
nxt_queue_init(&app->pending);
app->name.length = name.length;
nxt_memcpy(app->name.start, name.start, name.length);
app->type = lang->type;
app->max_processes = apcf.max_processes;
app->spare_processes = apcf.spare_processes;
app->max_pending_processes = apcf.spare_processes
? apcf.spare_processes : 1;
app->timeout = apcf.timeout;
app->res_timeout = apcf.res_timeout * 1000000;
app->idle_timeout = apcf.idle_timeout;
app->max_pending_responses = 2;
app->max_requests = apcf.requests;
engine = task->thread->engine;
app->engine = engine;
app->adjust_idle_work.handler = nxt_router_adjust_idle_timer;
app->adjust_idle_work.task = &engine->task;
app->adjust_idle_work.obj = app;
nxt_queue_insert_tail(&tmcf->apps, &app->link);
nxt_router_app_use(task, app, 1);
app->joint = app_joint;
app_joint->use_count = 1;
app_joint->app = app;
app_joint->idle_timer.bias = NXT_TIMER_DEFAULT_BIAS;
app_joint->idle_timer.work_queue = &engine->fast_work_queue;
app_joint->idle_timer.handler = nxt_router_app_idle_timeout;
app_joint->idle_timer.task = &engine->task;
app_joint->idle_timer.log = app_joint->idle_timer.task->log;
app_joint->free_app_work.handler = nxt_router_free_app;
app_joint->free_app_work.task = &engine->task;
app_joint->free_app_work.obj = app_joint;
}
}
routes_conf = nxt_conf_get_path(conf, &routes_path);
if (nxt_fast_path(routes_conf != NULL)) {
routes = nxt_http_routes_create(task, tmcf, routes_conf);
if (nxt_slow_path(routes == NULL)) {
return NXT_ERROR;
}
tmcf->router_conf->routes = routes;
}
ret = nxt_upstreams_create(task, tmcf, conf);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
http = nxt_conf_get_path(conf, &http_path);
#if 0
if (http == NULL) {
nxt_alert(task, "no \"http\" block");
return NXT_ERROR;
}
#endif
websocket = nxt_conf_get_path(conf, &websocket_path);
listeners = nxt_conf_get_path(conf, &listeners_path);
if (listeners != NULL) {
next = 0;
for ( ;; ) {
listener = nxt_conf_next_object_member(listeners, &name, &next);
if (listener == NULL) {
break;
}
skcf = nxt_router_socket_conf(task, tmcf, &name);
if (skcf == NULL) {
goto fail;
}
nxt_memzero(&lscf, sizeof(lscf));
ret = nxt_conf_map_object(mp, listener, nxt_router_listener_conf,
nxt_nitems(nxt_router_listener_conf),
&lscf);
if (ret != NXT_OK) {
nxt_alert(task, "listener map error");
goto fail;
}
nxt_debug(task, "application: %V", &lscf.application);
// STUB, default values if http block is not defined.
skcf->header_buffer_size = 2048;
skcf->large_header_buffer_size = 8192;
skcf->large_header_buffers = 4;
skcf->body_buffer_size = 16 * 1024;
skcf->max_body_size = 8 * 1024 * 1024;
skcf->proxy_header_buffer_size = 64 * 1024;
skcf->proxy_buffer_size = 4096;
skcf->proxy_buffers = 256;
skcf->idle_timeout = 180 * 1000;
skcf->header_read_timeout = 30 * 1000;
skcf->body_read_timeout = 30 * 1000;
skcf->send_timeout = 30 * 1000;
skcf->proxy_timeout = 60 * 1000;
skcf->proxy_send_timeout = 30 * 1000;
skcf->proxy_read_timeout = 30 * 1000;
skcf->websocket_conf.max_frame_size = 1024 * 1024;
skcf->websocket_conf.read_timeout = 60 * 1000;
skcf->websocket_conf.keepalive_interval = 30 * 1000;
nxt_str_null(&skcf->body_temp_path);
if (http != NULL) {
ret = nxt_conf_map_object(mp, http, nxt_router_http_conf,
nxt_nitems(nxt_router_http_conf),
skcf);
if (ret != NXT_OK) {
nxt_alert(task, "http map error");
goto fail;
}
}
if (websocket != NULL) {
ret = nxt_conf_map_object(mp, websocket,
nxt_router_websocket_conf,
nxt_nitems(nxt_router_websocket_conf),
&skcf->websocket_conf);
if (ret != NXT_OK) {
nxt_alert(task, "websocket map error");
goto fail;
}
}
t = &skcf->body_temp_path;
if (t->length == 0) {
t->start = (u_char *) task->thread->runtime->tmp;
t->length = nxt_strlen(t->start);
}
#if (NXT_TLS)
value = nxt_conf_get_path(listener, &certificate_path);
if (value != NULL) {
nxt_conf_get_string(value, &name);
tls = nxt_mp_get(mp, sizeof(nxt_router_tlssock_t));
if (nxt_slow_path(tls == NULL)) {
goto fail;
}
tls->name = name;
tls->conf = skcf;
nxt_queue_insert_tail(&tmcf->tls, &tls->link);
}
#endif
skcf->listen->handler = nxt_http_conn_init;
skcf->router_conf = tmcf->router_conf;
skcf->router_conf->count++;
if (lscf.pass.length != 0) {
skcf->action = nxt_http_action_create(task, tmcf, &lscf.pass);
/* COMPATIBILITY: listener application. */
} else if (lscf.application.length > 0) {
skcf->action = nxt_http_pass_application(task, tmcf,
&lscf.application);
}
}
}
value = nxt_conf_get_path(conf, &access_log_path);
if (value != NULL) {
nxt_conf_get_string(value, &path);
access_log = router->access_log;
if (access_log != NULL && nxt_strstr_eq(&path, &access_log->path)) {
nxt_thread_spin_lock(&router->lock);
access_log->count++;
nxt_thread_spin_unlock(&router->lock);
} else {
access_log = nxt_malloc(sizeof(nxt_router_access_log_t)
+ path.length);
if (access_log == NULL) {
nxt_alert(task, "failed to allocate access log structure");
goto fail;
}
access_log->fd = -1;
access_log->handler = &nxt_router_access_log_writer;
access_log->count = 1;
access_log->path.length = path.length;
access_log->path.start = (u_char *) access_log
+ sizeof(nxt_router_access_log_t);
nxt_memcpy(access_log->path.start, path.start, path.length);
}
tmcf->router_conf->access_log = access_log;
}
nxt_http_routes_resolve(task, tmcf);
nxt_queue_add(&tmcf->deleting, &router->sockets);
nxt_queue_init(&router->sockets);
return NXT_OK;
app_fail:
nxt_free(app);
fail:
nxt_queue_each(app, &tmcf->apps, nxt_app_t, link) {
nxt_queue_remove(&app->link);
nxt_thread_mutex_destroy(&app->mutex);
nxt_free(app);
} nxt_queue_loop;
return NXT_ERROR;
}
static nxt_int_t
nxt_router_conf_process_static(nxt_task_t *task, nxt_router_conf_t *rtcf,
nxt_conf_value_t *conf)
{
uint32_t next, i;
nxt_mp_t *mp;
nxt_str_t *type, extension, str;
nxt_int_t ret;
nxt_uint_t exts;
nxt_conf_value_t *mtypes_conf, *ext_conf, *value;
static nxt_str_t mtypes_path = nxt_string("/mime_types");
mp = rtcf->mem_pool;
ret = nxt_http_static_mtypes_init(mp, &rtcf->mtypes_hash);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
if (conf == NULL) {
return NXT_OK;
}
mtypes_conf = nxt_conf_get_path(conf, &mtypes_path);
if (mtypes_conf != NULL) {
next = 0;
for ( ;; ) {
ext_conf = nxt_conf_next_object_member(mtypes_conf, &str, &next);
if (ext_conf == NULL) {
break;
}
type = nxt_str_dup(mp, NULL, &str);
if (nxt_slow_path(type == NULL)) {
return NXT_ERROR;
}
if (nxt_conf_type(ext_conf) == NXT_CONF_STRING) {
nxt_conf_get_string(ext_conf, &str);
if (nxt_slow_path(nxt_str_dup(mp, &extension, &str) == NULL)) {
return NXT_ERROR;
}
ret = nxt_http_static_mtypes_hash_add(mp, &rtcf->mtypes_hash,
&extension, type);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
continue;
}
exts = nxt_conf_array_elements_count(ext_conf);
for (i = 0; i < exts; i++) {
value = nxt_conf_get_array_element(ext_conf, i);
nxt_conf_get_string(value, &str);
if (nxt_slow_path(nxt_str_dup(mp, &extension, &str) == NULL)) {
return NXT_ERROR;
}
ret = nxt_http_static_mtypes_hash_add(mp, &rtcf->mtypes_hash,
&extension, type);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
}
}
}
return NXT_OK;
}
static nxt_app_t *
nxt_router_app_find(nxt_queue_t *queue, nxt_str_t *name)
{
nxt_app_t *app;
nxt_queue_each(app, queue, nxt_app_t, link) {
if (nxt_strstr_eq(name, &app->name)) {
return app;
}
} nxt_queue_loop;
return NULL;
}
void
nxt_router_listener_application(nxt_router_temp_conf_t *tmcf, nxt_str_t *name,
nxt_http_action_t *action)
{
nxt_app_t *app;
app = nxt_router_app_find(&tmcf->apps, name);
if (app == NULL) {
app = nxt_router_app_find(&tmcf->previous, name);
}
action->u.application = app;
action->handler = nxt_http_application_handler;
}
static nxt_socket_conf_t *
nxt_router_socket_conf(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
nxt_str_t *name)
{
size_t size;
nxt_int_t ret;
nxt_bool_t wildcard;
nxt_sockaddr_t *sa;
nxt_socket_conf_t *skcf;
nxt_listen_socket_t *ls;
sa = nxt_sockaddr_parse(tmcf->mem_pool, name);
if (nxt_slow_path(sa == NULL)) {
nxt_alert(task, "invalid listener \"%V\"", name);
return NULL;
}
sa->type = SOCK_STREAM;
nxt_debug(task, "router listener: \"%*s\"",
(size_t) sa->length, nxt_sockaddr_start(sa));
skcf = nxt_mp_zget(tmcf->router_conf->mem_pool, sizeof(nxt_socket_conf_t));
if (nxt_slow_path(skcf == NULL)) {
return NULL;
}
size = nxt_sockaddr_size(sa);
ret = nxt_router_listen_socket_find(tmcf, skcf, sa);
if (ret != NXT_OK) {
ls = nxt_zalloc(sizeof(nxt_listen_socket_t) + size);
if (nxt_slow_path(ls == NULL)) {
return NULL;
}
skcf->listen = ls;
ls->sockaddr = nxt_pointer_to(ls, sizeof(nxt_listen_socket_t));
nxt_memcpy(ls->sockaddr, sa, size);
nxt_listen_socket_remote_size(ls);
ls->socket = -1;
ls->backlog = NXT_LISTEN_BACKLOG;
ls->flags = NXT_NONBLOCK;
ls->read_after_accept = 1;
}
switch (sa->u.sockaddr.sa_family) {
#if (NXT_HAVE_UNIX_DOMAIN)
case AF_UNIX:
wildcard = 0;
break;
#endif
#if (NXT_INET6)
case AF_INET6:
wildcard = IN6_IS_ADDR_UNSPECIFIED(&sa->u.sockaddr_in6.sin6_addr);
break;
#endif
case AF_INET:
default:
wildcard = (sa->u.sockaddr_in.sin_addr.s_addr == INADDR_ANY);
break;
}
if (!wildcard) {
skcf->sockaddr = nxt_mp_zget(tmcf->router_conf->mem_pool, size);
if (nxt_slow_path(skcf->sockaddr == NULL)) {
return NULL;
}
nxt_memcpy(skcf->sockaddr, sa, size);
}
return skcf;
}
static nxt_int_t
nxt_router_listen_socket_find(nxt_router_temp_conf_t *tmcf,
nxt_socket_conf_t *nskcf, nxt_sockaddr_t *sa)
{
nxt_router_t *router;
nxt_queue_link_t *qlk;
nxt_socket_conf_t *skcf;
router = tmcf->router_conf->router;
for (qlk = nxt_queue_first(&router->sockets);
qlk != nxt_queue_tail(&router->sockets);
qlk = nxt_queue_next(qlk))
{
skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
if (nxt_sockaddr_cmp(skcf->listen->sockaddr, sa)) {
nskcf->listen = skcf->listen;
nxt_queue_remove(qlk);
nxt_queue_insert_tail(&tmcf->keeping, qlk);
nxt_queue_insert_tail(&tmcf->updating, &nskcf->link);
return NXT_OK;
}
}
nxt_queue_insert_tail(&tmcf->pending, &nskcf->link);
return NXT_DECLINED;
}
static void
nxt_router_listen_socket_rpc_create(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_socket_conf_t *skcf)
{
size_t size;
uint32_t stream;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *main_port, *router_port;
nxt_runtime_t *rt;
nxt_socket_rpc_t *rpc;
rpc = nxt_mp_alloc(tmcf->mem_pool, sizeof(nxt_socket_rpc_t));
if (rpc == NULL) {
goto fail;
}
rpc->socket_conf = skcf;
rpc->temp_conf = tmcf;
size = nxt_sockaddr_size(skcf->listen->sockaddr);
b = nxt_buf_mem_alloc(tmcf->mem_pool, size, 0);
if (b == NULL) {
goto fail;
}
b->mem.free = nxt_cpymem(b->mem.free, skcf->listen->sockaddr, size);
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
stream = nxt_port_rpc_register_handler(task, router_port,
nxt_router_listen_socket_ready,
nxt_router_listen_socket_error,
main_port->pid, rpc);
if (nxt_slow_path(stream == 0)) {
goto fail;
}
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_SOCKET, -1,
stream, router_port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, router_port, stream);
goto fail;
}
return;
fail:
nxt_router_conf_error(task, tmcf);
}
static void
nxt_router_listen_socket_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_int_t ret;
nxt_socket_t s;
nxt_socket_rpc_t *rpc;
rpc = data;
s = msg->fd;
ret = nxt_socket_nonblocking(task, s);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
nxt_socket_defer_accept(task, s, rpc->socket_conf->listen->sockaddr);
ret = nxt_listen_socket(task, s, NXT_LISTEN_BACKLOG);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
rpc->socket_conf->listen->socket = s;
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_router_conf_apply, task, rpc->temp_conf, NULL);
return;
fail:
nxt_socket_close(task, s);
nxt_router_conf_error(task, rpc->temp_conf);
}
static void
nxt_router_listen_socket_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_socket_rpc_t *rpc;
nxt_router_temp_conf_t *tmcf;
rpc = data;
tmcf = rpc->temp_conf;
#if 0
u_char *p;
size_t size;
uint8_t error;
nxt_buf_t *in, *out;
nxt_sockaddr_t *sa;
static nxt_str_t socket_errors[] = {
nxt_string("ListenerSystem"),
nxt_string("ListenerNoIPv6"),
nxt_string("ListenerPort"),
nxt_string("ListenerInUse"),
nxt_string("ListenerNoAddress"),
nxt_string("ListenerNoAccess"),
nxt_string("ListenerPath"),
};
sa = rpc->socket_conf->listen->sockaddr;
in = nxt_buf_chk_make_plain(tmcf->mem_pool, msg->buf, msg->size);
if (nxt_slow_path(in == NULL)) {
return;
}
p = in->mem.pos;
error = *p++;
size = nxt_length("listen socket error: ")
+ nxt_length("{listener: \"\", code:\"\", message: \"\"}")
+ sa->length + socket_errors[error].length + (in->mem.free - p);
out = nxt_buf_mem_alloc(tmcf->mem_pool, size, 0);
if (nxt_slow_path(out == NULL)) {
return;
}
out->mem.free = nxt_sprintf(out->mem.free, out->mem.end,
"listen socket error: "
"{listener: \"%*s\", code:\"%V\", message: \"%*s\"}",
(size_t) sa->length, nxt_sockaddr_start(sa),
&socket_errors[error], in->mem.free - p, p);
nxt_debug(task, "%*s", out->mem.free - out->mem.pos, out->mem.pos);
#endif
nxt_router_conf_error(task, tmcf);
}
#if (NXT_TLS)
static void
nxt_router_tls_rpc_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
nxt_router_tlssock_t *tls)
{
nxt_socket_rpc_t *rpc;
rpc = nxt_mp_alloc(tmcf->mem_pool, sizeof(nxt_socket_rpc_t));
if (rpc == NULL) {
nxt_router_conf_error(task, tmcf);
return;
}
rpc->socket_conf = tls->conf;
rpc->temp_conf = tmcf;
nxt_cert_store_get(task, &tls->name, tmcf->mem_pool,
nxt_router_tls_rpc_handler, rpc);
}
static void
nxt_router_tls_rpc_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_mp_t *mp;
nxt_int_t ret;
nxt_tls_conf_t *tlscf;
nxt_socket_rpc_t *rpc;
nxt_router_temp_conf_t *tmcf;
nxt_debug(task, "tls rpc handler");
rpc = data;
tmcf = rpc->temp_conf;
if (msg == NULL || msg->port_msg.type == _NXT_PORT_MSG_RPC_ERROR) {
goto fail;
}
mp = tmcf->router_conf->mem_pool;
tlscf = nxt_mp_zget(mp, sizeof(nxt_tls_conf_t));
if (nxt_slow_path(tlscf == NULL)) {
goto fail;
}
tlscf->chain_file = msg->fd;
ret = task->thread->runtime->tls->server_init(task, tlscf);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
rpc->socket_conf->tls = tlscf;
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_router_conf_apply, task, tmcf, NULL);
return;
fail:
nxt_router_conf_error(task, tmcf);
}
#endif
static void
nxt_router_app_rpc_create(nxt_task_t *task,
nxt_router_temp_conf_t *tmcf, nxt_app_t *app)
{
size_t size;
uint32_t stream;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *main_port, *router_port;
nxt_runtime_t *rt;
nxt_app_rpc_t *rpc;
rpc = nxt_mp_alloc(tmcf->mem_pool, sizeof(nxt_app_rpc_t));
if (rpc == NULL) {
goto fail;
}
rpc->app = app;
rpc->temp_conf = tmcf;
nxt_debug(task, "app '%V' prefork", &app->name);
size = app->name.length + 1 + app->conf.length;
b = nxt_buf_mem_alloc(tmcf->mem_pool, size, 0);
if (nxt_slow_path(b == NULL)) {
goto fail;
}
nxt_buf_cpystr(b, &app->name);
*b->mem.free++ = '\0';
nxt_buf_cpystr(b, &app->conf);
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
stream = nxt_port_rpc_register_handler(task, router_port,
nxt_router_app_prefork_ready,
nxt_router_app_prefork_error,
-1, rpc);
if (nxt_slow_path(stream == 0)) {
goto fail;
}
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_WORKER, -1,
stream, router_port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, router_port, stream);
goto fail;
}
app->pending_processes++;
return;
fail:
nxt_router_conf_error(task, tmcf);
}
static void
nxt_router_app_prefork_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_app_t *app;
nxt_port_t *port;
nxt_app_rpc_t *rpc;
nxt_event_engine_t *engine;
rpc = data;
app = rpc->app;
port = msg->u.new_port;
port->app = app;
app->pending_processes--;
app->processes++;
app->idle_processes++;
engine = task->thread->engine;
nxt_queue_insert_tail(&app->ports, &port->app_link);
nxt_queue_insert_tail(&app->spare_ports, &port->idle_link);
port->idle_start = 0;
nxt_port_inc_use(port);
nxt_work_queue_add(&engine->fast_work_queue,
nxt_router_conf_apply, task, rpc->temp_conf, NULL);
}
static void
nxt_router_app_prefork_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_app_t *app;
nxt_app_rpc_t *rpc;
nxt_router_temp_conf_t *tmcf;
rpc = data;
app = rpc->app;
tmcf = rpc->temp_conf;
nxt_log(task, NXT_LOG_WARN, "failed to start application \"%V\"",
&app->name);
app->pending_processes--;
nxt_router_conf_error(task, tmcf);
}
static nxt_int_t
nxt_router_engines_create(nxt_task_t *task, nxt_router_t *router,
nxt_router_temp_conf_t *tmcf, const nxt_event_interface_t *interface)
{
nxt_int_t ret;
nxt_uint_t n, threads;
nxt_queue_link_t *qlk;
nxt_router_engine_conf_t *recf;
threads = tmcf->router_conf->threads;
tmcf->engines = nxt_array_create(tmcf->mem_pool, threads,
sizeof(nxt_router_engine_conf_t));
if (nxt_slow_path(tmcf->engines == NULL)) {
return NXT_ERROR;
}
n = 0;
for (qlk = nxt_queue_first(&router->engines);
qlk != nxt_queue_tail(&router->engines);
qlk = nxt_queue_next(qlk))
{
recf = nxt_array_zero_add(tmcf->engines);
if (nxt_slow_path(recf == NULL)) {
return NXT_ERROR;
}
recf->engine = nxt_queue_link_data(qlk, nxt_event_engine_t, link0);
if (n < threads) {
recf->action = NXT_ROUTER_ENGINE_KEEP;
ret = nxt_router_engine_conf_update(tmcf, recf);
} else {
recf->action = NXT_ROUTER_ENGINE_DELETE;
ret = nxt_router_engine_conf_delete(tmcf, recf);
}
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
n++;
}
tmcf->new_threads = n;
while (n < threads) {
recf = nxt_array_zero_add(tmcf->engines);
if (nxt_slow_path(recf == NULL)) {
return NXT_ERROR;
}
recf->action = NXT_ROUTER_ENGINE_ADD;
recf->engine = nxt_event_engine_create(task, interface, NULL, 0, 0);
if (nxt_slow_path(recf->engine == NULL)) {
return NXT_ERROR;
}
ret = nxt_router_engine_conf_create(tmcf, recf);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
n++;
}
return NXT_OK;
}
static nxt_int_t
nxt_router_engine_conf_create(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf)
{
nxt_int_t ret;
ret = nxt_router_engine_joints_create(tmcf, recf, &tmcf->creating,
nxt_router_listen_socket_create);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
ret = nxt_router_engine_joints_create(tmcf, recf, &tmcf->updating,
nxt_router_listen_socket_create);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
return ret;
}
static nxt_int_t
nxt_router_engine_conf_update(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf)
{
nxt_int_t ret;
ret = nxt_router_engine_joints_create(tmcf, recf, &tmcf->creating,
nxt_router_listen_socket_create);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
ret = nxt_router_engine_joints_create(tmcf, recf, &tmcf->updating,
nxt_router_listen_socket_update);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
ret = nxt_router_engine_joints_delete(tmcf, recf, &tmcf->deleting);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
return ret;
}
static nxt_int_t
nxt_router_engine_conf_delete(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf)
{
nxt_int_t ret;
ret = nxt_router_engine_quit(tmcf, recf);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
ret = nxt_router_engine_joints_delete(tmcf, recf, &tmcf->updating);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
return nxt_router_engine_joints_delete(tmcf, recf, &tmcf->deleting);
}
static nxt_int_t
nxt_router_engine_joints_create(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf, nxt_queue_t *sockets,
nxt_work_handler_t handler)
{
nxt_int_t ret;
nxt_joint_job_t *job;
nxt_queue_link_t *qlk;
nxt_socket_conf_t *skcf;
nxt_socket_conf_joint_t *joint;
for (qlk = nxt_queue_first(sockets);
qlk != nxt_queue_tail(sockets);
qlk = nxt_queue_next(qlk))
{
job = nxt_mp_get(tmcf->mem_pool, sizeof(nxt_joint_job_t));
if (nxt_slow_path(job == NULL)) {
return NXT_ERROR;
}
job->work.next = recf->jobs;
recf->jobs = &job->work;
job->task = tmcf->engine->task;
job->work.handler = handler;
job->work.task = &job->task;
job->work.obj = job;
job->tmcf = tmcf;
tmcf->count++;
joint = nxt_mp_alloc(tmcf->router_conf->mem_pool,
sizeof(nxt_socket_conf_joint_t));
if (nxt_slow_path(joint == NULL)) {
return NXT_ERROR;
}
job->work.data = joint;
ret = nxt_upstreams_joint_create(tmcf, &joint->upstreams);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
joint->count = 1;
skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
skcf->count++;
joint->socket_conf = skcf;
joint->engine = recf->engine;
}
return NXT_OK;
}
static nxt_int_t
nxt_router_engine_quit(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf)
{
nxt_joint_job_t *job;
job = nxt_mp_get(tmcf->mem_pool, sizeof(nxt_joint_job_t));
if (nxt_slow_path(job == NULL)) {
return NXT_ERROR;
}
job->work.next = recf->jobs;
recf->jobs = &job->work;
job->task = tmcf->engine->task;
job->work.handler = nxt_router_worker_thread_quit;
job->work.task = &job->task;
job->work.obj = NULL;
job->work.data = NULL;
job->tmcf = NULL;
return NXT_OK;
}
static nxt_int_t
nxt_router_engine_joints_delete(nxt_router_temp_conf_t *tmcf,
nxt_router_engine_conf_t *recf, nxt_queue_t *sockets)
{
nxt_joint_job_t *job;
nxt_queue_link_t *qlk;
for (qlk = nxt_queue_first(sockets);
qlk != nxt_queue_tail(sockets);
qlk = nxt_queue_next(qlk))
{
job = nxt_mp_get(tmcf->mem_pool, sizeof(nxt_joint_job_t));
if (nxt_slow_path(job == NULL)) {
return NXT_ERROR;
}
job->work.next = recf->jobs;
recf->jobs = &job->work;
job->task = tmcf->engine->task;
job->work.handler = nxt_router_listen_socket_delete;
job->work.task = &job->task;
job->work.obj = job;
job->work.data = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
job->tmcf = tmcf;
tmcf->count++;
}
return NXT_OK;
}
static nxt_int_t
nxt_router_threads_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_router_temp_conf_t *tmcf)
{
nxt_int_t ret;
nxt_uint_t i, threads;
nxt_router_engine_conf_t *recf;
recf = tmcf->engines->elts;
threads = tmcf->router_conf->threads;
for (i = tmcf->new_threads; i < threads; i++) {
ret = nxt_router_thread_create(task, rt, recf[i].engine);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
}
return NXT_OK;
}
static nxt_int_t
nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_event_engine_t *engine)
{
nxt_int_t ret;
nxt_thread_link_t *link;
nxt_thread_handle_t handle;
link = nxt_zalloc(sizeof(nxt_thread_link_t));
if (nxt_slow_path(link == NULL)) {
return NXT_ERROR;
}
link->start = nxt_router_thread_start;
link->engine = engine;
link->work.handler = nxt_router_thread_exit_handler;
link->work.task = task;
link->work.data = link;
nxt_queue_insert_tail(&rt->engines, &engine->link);
ret = nxt_thread_create(&handle, link);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_queue_remove(&engine->link);
}
return ret;
}
static void
nxt_router_apps_sort(nxt_task_t *task, nxt_router_t *router,
nxt_router_temp_conf_t *tmcf)
{
nxt_app_t *app;
nxt_queue_each(app, &router->apps, nxt_app_t, link) {
nxt_router_app_unlink(task, app);
} nxt_queue_loop;
nxt_queue_add(&router->apps, &tmcf->previous);
nxt_queue_add(&router->apps, &tmcf->apps);
}
static void
nxt_router_engines_post(nxt_router_t *router, nxt_router_temp_conf_t *tmcf)
{
nxt_uint_t n;
nxt_event_engine_t *engine;
nxt_router_engine_conf_t *recf;
recf = tmcf->engines->elts;
for (n = tmcf->engines->nelts; n != 0; n--) {
engine = recf->engine;
switch (recf->action) {
case NXT_ROUTER_ENGINE_KEEP:
break;
case NXT_ROUTER_ENGINE_ADD:
nxt_queue_insert_tail(&router->engines, &engine->link0);
break;
case NXT_ROUTER_ENGINE_DELETE:
nxt_queue_remove(&engine->link0);
break;
}
nxt_router_engine_post(engine, recf->jobs);
recf++;
}
}
static void
nxt_router_engine_post(nxt_event_engine_t *engine, nxt_work_t *jobs)
{
nxt_work_t *work, *next;
for (work = jobs; work != NULL; work = next) {
next = work->next;
work->next = NULL;
nxt_event_engine_post(engine, work);
}
}
static nxt_port_handlers_t nxt_router_app_port_handlers = {
.rpc_error = nxt_port_rpc_handler,
.mmap = nxt_port_mmap_handler,
.data = nxt_port_rpc_handler,
.oosm = nxt_router_oosm_handler,
};
static void
nxt_router_thread_start(void *data)
{
nxt_int_t ret;
nxt_port_t *port;
nxt_task_t *task;
nxt_thread_t *thread;
nxt_thread_link_t *link;
nxt_event_engine_t *engine;
link = data;
engine = link->engine;
task = &engine->task;
thread = nxt_thread();
nxt_event_engine_thread_adopt(engine);
/* STUB */
thread->runtime = engine->task.thread->runtime;
engine->task.thread = thread;
engine->task.log = thread->log;
thread->engine = engine;
thread->task = &engine->task;
#if 0
thread->fiber = &engine->fibers->fiber;
#endif
engine->mem_pool = nxt_mp_create(4096, 128, 1024, 64);
if (nxt_slow_path(engine->mem_pool == NULL)) {
return;
}
port = nxt_port_new(task, nxt_port_get_next_id(), nxt_pid,
NXT_PROCESS_ROUTER);
if (nxt_slow_path(port == NULL)) {
return;
}
ret = nxt_port_socket_init(task, port, 0);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_use(task, port, -1);
return;
}
engine->port = port;
nxt_port_enable(task, port, &nxt_router_app_port_handlers);
nxt_event_engine_start(engine);
}
static void
nxt_router_listen_socket_create(nxt_task_t *task, void *obj, void *data)
{
nxt_joint_job_t *job;
nxt_socket_conf_t *skcf;
nxt_listen_event_t *lev;
nxt_listen_socket_t *ls;
nxt_thread_spinlock_t *lock;
nxt_socket_conf_joint_t *joint;
job = obj;
joint = data;
nxt_queue_insert_tail(&task->thread->engine->joints, &joint->link);
skcf = joint->socket_conf;
ls = skcf->listen;
lev = nxt_listen_event(task, ls);
if (nxt_slow_path(lev == NULL)) {
nxt_router_listen_socket_release(task, skcf);
return;
}
lev->socket.data = joint;
lock = &skcf->router_conf->router->lock;
nxt_thread_spin_lock(lock);
ls->count++;
nxt_thread_spin_unlock(lock);
job->work.next = NULL;
job->work.handler = nxt_router_conf_wait;
nxt_event_engine_post(job->tmcf->engine, &job->work);
}
nxt_inline nxt_listen_event_t *
nxt_router_listen_event(nxt_queue_t *listen_connections,
nxt_socket_conf_t *skcf)
{
nxt_socket_t fd;
nxt_queue_link_t *qlk;
nxt_listen_event_t *lev;
fd = skcf->listen->socket;
for (qlk = nxt_queue_first(listen_connections);
qlk != nxt_queue_tail(listen_connections);
qlk = nxt_queue_next(qlk))
{
lev = nxt_queue_link_data(qlk, nxt_listen_event_t, link);
if (fd == lev->socket.fd) {
return lev;
}
}
return NULL;
}
static void
nxt_router_listen_socket_update(nxt_task_t *task, void *obj, void *data)
{
nxt_joint_job_t *job;
nxt_event_engine_t *engine;
nxt_listen_event_t *lev;
nxt_socket_conf_joint_t *joint, *old;
job = obj;
joint = data;
engine = task->thread->engine;
nxt_queue_insert_tail(&engine->joints, &joint->link);
lev = nxt_router_listen_event(&engine->listen_connections,
joint->socket_conf);
old = lev->socket.data;
lev->socket.data = joint;
lev->listen = joint->socket_conf->listen;
job->work.next = NULL;
job->work.handler = nxt_router_conf_wait;
nxt_event_engine_post(job->tmcf->engine, &job->work);
/*
* The task is allocated from configuration temporary
* memory pool so it can be freed after engine post operation.
*/
nxt_router_conf_release(&engine->task, old);
}
static void
nxt_router_listen_socket_delete(nxt_task_t *task, void *obj, void *data)
{
nxt_joint_job_t *job;
nxt_socket_conf_t *skcf;
nxt_listen_event_t *lev;
nxt_event_engine_t *engine;
job = obj;
skcf = data;
engine = task->thread->engine;
lev = nxt_router_listen_event(&engine->listen_connections, skcf);
nxt_fd_event_delete(engine, &lev->socket);
nxt_debug(task, "engine %p: listen socket delete: %d", engine,
lev->socket.fd);
lev->timer.handler = nxt_router_listen_socket_close;
lev->timer.work_queue = &engine->fast_work_queue;
nxt_timer_add(engine, &lev->timer, 0);
job->work.next = NULL;
job->work.handler = nxt_router_conf_wait;
nxt_event_engine_post(job->tmcf->engine, &job->work);
}
static void
nxt_router_worker_thread_quit(nxt_task_t *task, void *obj, void *data)
{
nxt_event_engine_t *engine;
nxt_debug(task, "router worker thread quit");
engine = task->thread->engine;
engine->shutdown = 1;
if (nxt_queue_is_empty(&engine->joints)) {
nxt_thread_exit(task->thread);
}
}
static void
nxt_router_listen_socket_close(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *timer;
nxt_listen_event_t *lev;
nxt_socket_conf_joint_t *joint;
timer = obj;
lev = nxt_timer_data(timer, nxt_listen_event_t, timer);
nxt_debug(task, "engine %p: listen socket close: %d", task->thread->engine,
lev->socket.fd);
nxt_queue_remove(&lev->link);
joint = lev->socket.data;
lev->socket.data = NULL;
/* 'task' refers to lev->task and we cannot use after nxt_free() */
task = &task->thread->engine->task;
nxt_router_listen_socket_release(task, joint->socket_conf);
nxt_router_listen_event_release(task, lev, joint);
}
static void
nxt_router_listen_socket_release(nxt_task_t *task, nxt_socket_conf_t *skcf)
{
nxt_listen_socket_t *ls;
nxt_thread_spinlock_t *lock;
ls = skcf->listen;
lock = &skcf->router_conf->router->lock;
nxt_thread_spin_lock(lock);
nxt_debug(task, "engine %p: listen socket release: ls->count %D",
task->thread->engine, ls->count);
if (--ls->count != 0) {
ls = NULL;
}
nxt_thread_spin_unlock(lock);
if (ls != NULL) {
nxt_socket_close(task, ls->socket);
nxt_free(ls);
}
}
void
nxt_router_listen_event_release(nxt_task_t *task, nxt_listen_event_t *lev,
nxt_socket_conf_joint_t *joint)
{
nxt_event_engine_t *engine;
nxt_debug(task, "listen event count: %D", lev->count);
if (--lev->count == 0) {
nxt_free(lev);
}
if (joint != NULL) {
nxt_router_conf_release(task, joint);
}
engine = task->thread->engine;
if (engine->shutdown && nxt_queue_is_empty(&engine->joints)) {
nxt_thread_exit(task->thread);
}
}
void
nxt_router_conf_release(nxt_task_t *task, nxt_socket_conf_joint_t *joint)
{
nxt_socket_conf_t *skcf;
nxt_router_conf_t *rtcf;
nxt_thread_spinlock_t *lock;
nxt_debug(task, "conf joint %p count: %D", joint, joint->count);
if (--joint->count != 0) {
return;
}
nxt_queue_remove(&joint->link);
/*
* The joint content can not be safely used after the critical
* section protected by the spinlock because its memory pool may
* be already destroyed by another thread.
*/
skcf = joint->socket_conf;
rtcf = skcf->router_conf;
lock = &rtcf->router->lock;
nxt_thread_spin_lock(lock);
nxt_debug(task, "conf skcf %p: %D, rtcf %p: %D", skcf, skcf->count,
rtcf, rtcf->count);
if (--skcf->count != 0) {
skcf = NULL;
rtcf = NULL;
} else {
nxt_queue_remove(&skcf->link);
if (--rtcf->count != 0) {
rtcf = NULL;
}
}
nxt_thread_spin_unlock(lock);
if (skcf != NULL) {
if (skcf->action != NULL) {
nxt_http_action_cleanup(task, skcf->action);
}
#if (NXT_TLS)
if (skcf->tls != NULL) {
task->thread->runtime->tls->server_free(task, skcf->tls);
}
#endif
}
/* TODO remove engine->port */
/* TODO excude from connected ports */
if (rtcf != NULL) {
nxt_debug(task, "old router conf is destroyed");
nxt_http_routes_cleanup(task, rtcf->routes);
nxt_router_access_log_release(task, lock, rtcf->access_log);
nxt_mp_thread_adopt(rtcf->mem_pool);
nxt_mp_destroy(rtcf->mem_pool);
}
}
static void
nxt_router_access_log_writer(nxt_task_t *task, nxt_http_request_t *r,
nxt_router_access_log_t *access_log)
{
size_t size;
u_char *buf, *p;
nxt_off_t bytes;
static nxt_time_string_t date_cache = {
(nxt_atomic_uint_t) -1,
nxt_router_access_log_date,
"%02d/%s/%4d:%02d:%02d:%02d %c%02d%02d",
nxt_length("31/Dec/1986:19:40:00 +0300"),
NXT_THREAD_TIME_LOCAL,
NXT_THREAD_TIME_SEC,
};
size = r->remote->address_length
+ 6 /* ' - - [' */
+ date_cache.size
+ 3 /* '] "' */
+ r->method->length
+ 1 /* space */
+ r->target.length
+ 1 /* space */
+ r->version.length
+ 2 /* '" ' */
+ 3 /* status */
+ 1 /* space */
+ NXT_OFF_T_LEN
+ 2 /* ' "' */
+ (r->referer != NULL ? r->referer->value_length : 1)
+ 3 /* '" "' */
+ (r->user_agent != NULL ? r->user_agent->value_length : 1)
+ 2 /* '"\n' */
;
buf = nxt_mp_nget(r->mem_pool, size);
if (nxt_slow_path(buf == NULL)) {
return;
}
p = nxt_cpymem(buf, nxt_sockaddr_address(r->remote),
r->remote->address_length);
p = nxt_cpymem(p, " - - [", 6);
p = nxt_thread_time_string(task->thread, &date_cache, p);
p = nxt_cpymem(p, "] \"", 3);
if (r->method->length != 0) {
p = nxt_cpymem(p, r->method->start, r->method->length);
if (r->target.length != 0) {
*p++ = ' ';
p = nxt_cpymem(p, r->target.start, r->target.length);
if (r->version.length != 0) {
*p++ = ' ';
p = nxt_cpymem(p, r->version.start, r->version.length);
}
}
} else {
*p++ = '-';
}
p = nxt_cpymem(p, "\" ", 2);
p = nxt_sprintf(p, p + 3, "%03d", r->status);
*p++ = ' ';
bytes = nxt_http_proto[r->protocol].body_bytes_sent(task, r->proto);
p = nxt_sprintf(p, p + NXT_OFF_T_LEN, "%O", bytes);
p = nxt_cpymem(p, " \"", 2);
if (r->referer != NULL) {
p = nxt_cpymem(p, r->referer->value, r->referer->value_length);
} else {
*p++ = '-';
}
p = nxt_cpymem(p, "\" \"", 3);
if (r->user_agent != NULL) {
p = nxt_cpymem(p, r->user_agent->value, r->user_agent->value_length);
} else {
*p++ = '-';
}
p = nxt_cpymem(p, "\"\n", 2);
nxt_fd_write(access_log->fd, buf, p - buf);
}
static u_char *
nxt_router_access_log_date(u_char *buf, nxt_realtime_t *now, struct tm *tm,
size_t size, const char *format)
{
u_char sign;
time_t gmtoff;
static const char *month[] = { "Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec" };
gmtoff = nxt_timezone(tm) / 60;
if (gmtoff < 0) {
gmtoff = -gmtoff;
sign = '-';
} else {
sign = '+';
}
return nxt_sprintf(buf, buf + size, format,
tm->tm_mday, month[tm->tm_mon], tm->tm_year + 1900,
tm->tm_hour, tm->tm_min, tm->tm_sec,
sign, gmtoff / 60, gmtoff % 60);
}
static void
nxt_router_access_log_open(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
{
uint32_t stream;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *main_port, *router_port;
nxt_runtime_t *rt;
nxt_router_access_log_t *access_log;
access_log = tmcf->router_conf->access_log;
b = nxt_buf_mem_alloc(tmcf->mem_pool, access_log->path.length + 1, 0);
if (nxt_slow_path(b == NULL)) {
goto fail;
}
nxt_buf_cpystr(b, &access_log->path);
*b->mem.free++ = '\0';
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
stream = nxt_port_rpc_register_handler(task, router_port,
nxt_router_access_log_ready,
nxt_router_access_log_error,
-1, tmcf);
if (nxt_slow_path(stream == 0)) {
goto fail;
}
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_ACCESS_LOG, -1,
stream, router_port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, router_port, stream);
goto fail;
}
return;
fail:
nxt_router_conf_error(task, tmcf);
}
static void
nxt_router_access_log_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_router_temp_conf_t *tmcf;
nxt_router_access_log_t *access_log;
tmcf = data;
access_log = tmcf->router_conf->access_log;
access_log->fd = msg->fd;
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_router_conf_apply, task, tmcf, NULL);
}
static void
nxt_router_access_log_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_router_temp_conf_t *tmcf;
tmcf = data;
nxt_router_conf_error(task, tmcf);
}
static void
nxt_router_access_log_release(nxt_task_t *task, nxt_thread_spinlock_t *lock,
nxt_router_access_log_t *access_log)
{
if (access_log == NULL) {
return;
}
nxt_thread_spin_lock(lock);
if (--access_log->count != 0) {
access_log = NULL;
}
nxt_thread_spin_unlock(lock);
if (access_log != NULL) {
if (access_log->fd != -1) {
nxt_fd_close(access_log->fd);
}
nxt_free(access_log);
}
}
typedef struct {
nxt_mp_t *mem_pool;
nxt_router_access_log_t *access_log;
} nxt_router_access_log_reopen_t;
void
nxt_router_access_log_reopen_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_mp_t *mp;
uint32_t stream;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *main_port, *router_port;
nxt_runtime_t *rt;
nxt_router_access_log_t *access_log;
nxt_router_access_log_reopen_t *reopen;
access_log = nxt_router->access_log;
if (access_log == NULL) {
return;
}
mp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(mp == NULL)) {
return;
}
reopen = nxt_mp_get(mp, sizeof(nxt_router_access_log_reopen_t));
if (nxt_slow_path(reopen == NULL)) {
goto fail;
}
reopen->mem_pool = mp;
reopen->access_log = access_log;
b = nxt_buf_mem_alloc(mp, access_log->path.length + 1, 0);
if (nxt_slow_path(b == NULL)) {
goto fail;
}
b->completion_handler = nxt_router_access_log_reopen_completion;
nxt_buf_cpystr(b, &access_log->path);
*b->mem.free++ = '\0';
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
stream = nxt_port_rpc_register_handler(task, router_port,
nxt_router_access_log_reopen_ready,
nxt_router_access_log_reopen_error,
-1, reopen);
if (nxt_slow_path(stream == 0)) {
goto fail;
}
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_ACCESS_LOG, -1,
stream, router_port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, router_port, stream);
goto fail;
}
nxt_mp_retain(mp);
return;
fail:
nxt_mp_destroy(mp);
}
static void
nxt_router_access_log_reopen_completion(nxt_task_t *task, void *obj, void *data)
{
nxt_mp_t *mp;
nxt_buf_t *b;
b = obj;
mp = b->data;
nxt_mp_release(mp);
}
static void
nxt_router_access_log_reopen_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_router_access_log_t *access_log;
nxt_router_access_log_reopen_t *reopen;
reopen = data;
access_log = reopen->access_log;
if (access_log == nxt_router->access_log) {
if (nxt_slow_path(dup2(msg->fd, access_log->fd) == -1)) {
nxt_alert(task, "dup2(%FD, %FD) failed %E",
msg->fd, access_log->fd, nxt_errno);
}
}
nxt_fd_close(msg->fd);
nxt_mp_release(reopen->mem_pool);
}
static void
nxt_router_access_log_reopen_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_router_access_log_reopen_t *reopen;
reopen = data;
nxt_mp_release(reopen->mem_pool);
}
static void
nxt_router_thread_exit_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_port_t *port;
nxt_thread_link_t *link;
nxt_event_engine_t *engine;
nxt_thread_handle_t handle;
handle = (nxt_thread_handle_t) obj;
link = data;
nxt_thread_wait(handle);
engine = link->engine;
nxt_queue_remove(&engine->link);
port = engine->port;
// TODO notify all apps
port->engine = task->thread->engine;
nxt_mp_thread_adopt(port->mem_pool);
nxt_port_use(task, port, -1);
nxt_mp_thread_adopt(engine->mem_pool);
nxt_mp_destroy(engine->mem_pool);
nxt_event_engine_free(engine);
nxt_free(link);
}
static void
nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_int_t ret;
nxt_buf_t *b, *next;
nxt_port_t *app_port;
nxt_unit_field_t *f;
nxt_http_field_t *field;
nxt_http_request_t *r;
nxt_unit_response_t *resp;
nxt_request_app_link_t *req_app_link;
nxt_request_rpc_data_t *req_rpc_data;
b = msg->buf;
req_rpc_data = data;
if (msg->size == 0) {
b = NULL;
}
r = req_rpc_data->request;
if (nxt_slow_path(r == NULL)) {
return;
}
if (r->error) {
nxt_request_rpc_data_unlink(task, req_rpc_data);
return;
}
if (msg->port_msg.last != 0) {
nxt_debug(task, "router data create last buf");
nxt_buf_chain_add(&b, nxt_http_buf_last(r));
nxt_request_rpc_data_unlink(task, req_rpc_data);
} else {
if (req_rpc_data->app != NULL && req_rpc_data->app->timeout != 0) {
r->timer.handler = nxt_router_app_timeout;
r->timer_data = req_rpc_data;
nxt_timer_add(task->thread->engine, &r->timer,
req_rpc_data->app->timeout);
}
}
if (b == NULL) {
return;
}
if (msg->buf == b) {
/* Disable instant buffer completion/re-using by port. */
msg->buf = NULL;
}
if (r->header_sent) {
nxt_buf_chain_add(&r->out, b);
nxt_http_request_send_body(task, r, NULL);
} else {
size_t b_size = nxt_buf_mem_used_size(&b->mem);
if (nxt_slow_path(b_size < sizeof(*resp))) {
goto fail;
}
resp = (void *) b->mem.pos;
if (nxt_slow_path(b_size < sizeof(*resp)
+ resp->fields_count * sizeof(nxt_unit_field_t))) {
goto fail;
}
field = NULL;
for (f = resp->fields; f < resp->fields + resp->fields_count; f++) {
if (f->skip) {
continue;
}
field = nxt_list_add(r->resp.fields);
if (nxt_slow_path(field == NULL)) {
goto fail;
}
field->hash = f->hash;
field->skip = 0;
field->hopbyhop = 0;
field->name_length = f->name_length;
field->value_length = f->value_length;
field->name = nxt_unit_sptr_get(&f->name);
field->value = nxt_unit_sptr_get(&f->value);
ret = nxt_http_field_process(field, &nxt_response_fields_hash, r);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
nxt_debug(task, "header%s: %*s: %*s",
(field->skip ? " skipped" : ""),
(size_t) field->name_length, field->name,
(size_t) field->value_length, field->value);
if (field->skip) {
r->resp.fields->last->nelts--;
}
}
r->status = resp->status;
if (resp->piggyback_content_length != 0) {
b->mem.pos = nxt_unit_sptr_get(&resp->piggyback_content);
b->mem.free = b->mem.pos + resp->piggyback_content_length;
} else {
b->mem.pos = b->mem.free;
}
if (nxt_buf_mem_used_size(&b->mem) == 0) {
next = b->next;
b->next = NULL;
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
b->completion_handler, task, b, b->parent);
b = next;
}
if (b != NULL) {
nxt_buf_chain_add(&r->out, b);
}
nxt_http_request_header_send(task, r, nxt_http_request_send_body, NULL);
if (r->websocket_handshake
&& r->status == NXT_HTTP_SWITCHING_PROTOCOLS)
{
req_app_link = nxt_request_app_link_alloc(task,
req_rpc_data->req_app_link,
req_rpc_data);
if (nxt_slow_path(req_app_link == NULL)) {
goto fail;
}
app_port = req_app_link->app_port;
if (app_port == NULL && req_rpc_data->app_port != NULL) {
req_app_link->app_port = req_rpc_data->app_port;
app_port = req_app_link->app_port;
req_app_link->apr_action = req_rpc_data->apr_action;
req_rpc_data->app_port = NULL;
}
if (nxt_slow_path(app_port == NULL)) {
goto fail;
}
nxt_thread_mutex_lock(&req_rpc_data->app->mutex);
nxt_queue_insert_tail(&app_port->active_websockets,
&req_app_link->link_port_websockets);
nxt_thread_mutex_unlock(&req_rpc_data->app->mutex);
nxt_router_app_port_release(task, app_port, NXT_APR_UPGRADE);
req_app_link->apr_action = NXT_APR_CLOSE;
nxt_debug(task, "req_app_link stream #%uD upgrade",
req_app_link->stream);
r->state = &nxt_http_websocket;
} else {
r->state = &nxt_http_request_send_state;
}
}
return;
fail:
nxt_http_request_error(task, r, NXT_HTTP_SERVICE_UNAVAILABLE);
nxt_request_rpc_data_unlink(task, req_rpc_data);
}
static const nxt_http_request_state_t nxt_http_request_send_state
nxt_aligned(64) =
{
.error_handler = nxt_http_request_error_handler,
};
static void
nxt_http_request_send_body(nxt_task_t *task, void *obj, void *data)
{
nxt_buf_t *out;
nxt_http_request_t *r;
r = obj;
out = r->out;
if (out != NULL) {
r->out = NULL;
nxt_http_request_send(task, r, out);
}
}
static void
nxt_router_response_error_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_int_t res;
nxt_port_t *port;
nxt_bool_t cancelled;
nxt_request_app_link_t *req_app_link;
nxt_request_rpc_data_t *req_rpc_data;
req_rpc_data = data;
req_app_link = req_rpc_data->req_app_link;
if (req_app_link != NULL) {
cancelled = nxt_router_msg_cancel(task, &req_app_link->msg_info,
req_app_link->stream);
if (cancelled) {
res = nxt_router_app_port(task, req_rpc_data->app, req_app_link);
if (res == NXT_OK) {
port = req_app_link->app_port;
if (nxt_slow_path(port == NULL)) {
nxt_log(task, NXT_LOG_ERR,
"port is NULL in cancelled req_app_link");
return;
}
nxt_port_rpc_ex_set_peer(task, task->thread->engine->port,
req_rpc_data, port->pid);
nxt_router_app_prepare_request(task, req_app_link);
}
msg->port_msg.last = 0;
return;
}
}
if (req_rpc_data->request != NULL) {
nxt_http_request_error(task, req_rpc_data->request,
NXT_HTTP_SERVICE_UNAVAILABLE);
}
nxt_request_rpc_data_unlink(task, req_rpc_data);
}
static void
nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_app_t *app;
nxt_port_t *port;
nxt_app_joint_t *app_joint;
app_joint = data;
port = msg->u.new_port;
nxt_assert(app_joint != NULL);
nxt_assert(port != NULL);
app = app_joint->app;
nxt_router_app_joint_use(task, app_joint, -1);
if (nxt_slow_path(app == NULL)) {
nxt_debug(task, "new port ready for released app, send QUIT");
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
return;
}
port->app = app;
nxt_thread_mutex_lock(&app->mutex);
nxt_assert(app->pending_processes != 0);
app->pending_processes--;
app->processes++;
nxt_thread_mutex_unlock(&app->mutex);
nxt_debug(task, "app '%V' new port ready, pid %PI, %d/%d",
&app->name, port->pid, app->processes, app->pending_processes);
nxt_router_app_port_release(task, port, NXT_APR_NEW_PORT);
}
static void
nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
nxt_app_t *app;
nxt_app_joint_t *app_joint;
nxt_queue_link_t *lnk;
nxt_request_app_link_t *req_app_link;
app_joint = data;
nxt_assert(app_joint != NULL);
app = app_joint->app;
nxt_router_app_joint_use(task, app_joint, -1);
if (nxt_slow_path(app == NULL)) {
nxt_debug(task, "start error for released app");
return;
}
nxt_debug(task, "app '%V' %p start error", &app->name, app);
nxt_thread_mutex_lock(&app->mutex);
nxt_assert(app->pending_processes != 0);
app->pending_processes--;
if (!nxt_queue_is_empty(&app->requests)) {
lnk = nxt_queue_last(&app->requests);
nxt_queue_remove(lnk);
lnk->next = NULL;
req_app_link = nxt_queue_link_data(lnk, nxt_request_app_link_t,
link_app_requests);
} else {
req_app_link = NULL;
}
nxt_thread_mutex_unlock(&app->mutex);
if (req_app_link != NULL) {
nxt_debug(task, "app '%V' %p abort next stream #%uD",
&app->name, app, req_app_link->stream);
nxt_request_app_link_error(req_app_link, 500,
"Failed to start application process");
nxt_request_app_link_use(task, req_app_link, -1);
}
}
nxt_inline nxt_port_t *
nxt_router_app_get_port_for_quit(nxt_app_t *app);
void
nxt_router_app_use(nxt_task_t *task, nxt_app_t *app, int i)
{
int c;
c = nxt_atomic_fetch_add(&app->use_count, i);
if (i < 0 && c == -i) {
if (task->thread->engine != app->engine) {
nxt_event_engine_post(app->engine, &app->joint->free_app_work);
} else {
nxt_router_free_app(task, app->joint, NULL);
}
}
}
nxt_inline nxt_bool_t
nxt_router_app_first_port_busy(nxt_app_t *app)
{
nxt_port_t *port;
nxt_queue_link_t *lnk;
lnk = nxt_queue_first(&app->ports);
port = nxt_queue_link_data(lnk, nxt_port_t, app_link);
return port->app_pending_responses > 0;
}
nxt_inline nxt_port_t *
nxt_router_pop_first_port(nxt_app_t *app)
{
nxt_port_t *port;
nxt_queue_link_t *lnk;
lnk = nxt_queue_first(&app->ports);
nxt_queue_remove(lnk);
port = nxt_queue_link_data(lnk, nxt_port_t, app_link);
port->app_pending_responses++;
if (nxt_queue_chk_remove(&port->idle_link)) {
app->idle_processes--;
if (port->idle_start == 0) {
nxt_assert(app->idle_processes < app->spare_processes);
} else {
nxt_assert(app->idle_processes >= app->spare_processes);
port->idle_start = 0;
}
}
if ((app->max_pending_responses == 0
|| port->app_pending_responses < app->max_pending_responses)
&& (app->max_requests == 0
|| port->app_responses + port->app_pending_responses
< app->max_requests))
{
nxt_queue_insert_tail(&app->ports, lnk);
nxt_port_inc_use(port);
} else {
lnk->next = NULL;
}
return port;
}
nxt_inline nxt_port_t *
nxt_router_app_get_port_for_quit(nxt_app_t *app)
{
nxt_port_t *port;
port = NULL;
nxt_thread_mutex_lock(&app->mutex);
nxt_queue_each(port, &app->ports, nxt_port_t, app_link) {
if (port->app_pending_responses > 0) {
port = NULL;
continue;
}
/* Caller is responsible to decrease port use count. */
nxt_queue_chk_remove(&port->app_link);
if (nxt_queue_chk_remove(&port->idle_link)) {
app->idle_processes--;
}
port->app = NULL;
app->processes--;
break;
} nxt_queue_loop;
nxt_thread_mutex_unlock(&app->mutex);
return port;
}
static void
nxt_router_app_unlink(nxt_task_t *task, nxt_app_t *app)
{
nxt_debug(task, "app '%V' %p unlink", &app->name, app);
nxt_queue_remove(&app->link);
nxt_router_app_use(task, app, -1);
}
static void
nxt_router_app_process_request(nxt_task_t *task, void *obj, void *data)
{
nxt_request_app_link_t *req_app_link;
req_app_link = data;
#if (NXT_DEBUG)
{
nxt_app_t *app;
app = obj;
nxt_assert(app != NULL);
nxt_assert(req_app_link != NULL);
nxt_assert(req_app_link->app_port != NULL);
nxt_debug(task, "app '%V' %p process next stream #%uD",
&app->name, app, req_app_link->stream);
}
#endif
nxt_router_app_prepare_request(task, req_app_link);
nxt_request_app_link_use(task, req_app_link, -1);
}
static void
nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port,
nxt_apr_action_t action)
{
int inc_use;
uint32_t dec_pending, got_response;
nxt_app_t *app;
nxt_bool_t port_unchained;
nxt_bool_t send_quit, cancelled, adjust_idle_timer;
nxt_queue_link_t *lnk;
nxt_request_app_link_t *req_app_link, *pending_ra, *re_ra;
nxt_port_select_state_t state;
nxt_assert(port != NULL);
nxt_assert(port->app != NULL);
req_app_link = NULL;
app = port->app;
inc_use = 0;
dec_pending = 0;
got_response = 0;
switch (action) {
case NXT_APR_NEW_PORT:
break;
case NXT_APR_REQUEST_FAILED:
dec_pending = 1;
inc_use = -1;
break;
case NXT_APR_GOT_RESPONSE:
dec_pending = 1;
got_response = 1;
inc_use = -1;
break;
case NXT_APR_UPGRADE:
dec_pending = 1;
got_response = 1;
break;
case NXT_APR_CLOSE:
inc_use = -1;
break;
}
nxt_thread_mutex_lock(&app->mutex);
port->app_pending_responses -= dec_pending;
port->app_responses += got_response;
if (port->pair[1] != -1
&& (app->max_pending_responses == 0
|| port->app_pending_responses < app->max_pending_responses)
&& (app->max_requests == 0
|| port->app_responses + port->app_pending_responses
< app->max_requests))
{
if (port->app_link.next == NULL) {
if (port->app_pending_responses > 0) {
nxt_queue_insert_tail(&app->ports, &port->app_link);
} else {
nxt_queue_insert_head(&app->ports, &port->app_link);
}
nxt_port_inc_use(port);
} else {
if (port->app_pending_responses == 0
&& nxt_queue_first(&app->ports) != &port->app_link)
{
nxt_queue_remove(&port->app_link);
nxt_queue_insert_head(&app->ports, &port->app_link);
}
}
}
if (!nxt_queue_is_empty(&app->ports)
&& !nxt_queue_is_empty(&app->requests))
{
lnk = nxt_queue_first(&app->requests);
nxt_queue_remove(lnk);
lnk->next = NULL;
req_app_link = nxt_queue_link_data(lnk, nxt_request_app_link_t,
link_app_requests);
req_app_link->app_port = nxt_router_pop_first_port(app);
if (req_app_link->app_port->app_pending_responses > 1) {
nxt_request_app_link_pending(task, app, req_app_link);
}
}
/* Pop first pending request for this port. */
if (dec_pending > 0
&& !nxt_queue_is_empty(&port->pending_requests))
{
lnk = nxt_queue_first(&port->pending_requests);
nxt_queue_remove(lnk);
lnk->next = NULL;
pending_ra = nxt_queue_link_data(lnk, nxt_request_app_link_t,
link_port_pending);
nxt_assert(pending_ra->link_app_pending.next != NULL);
nxt_queue_remove(&pending_ra->link_app_pending);
pending_ra->link_app_pending.next = NULL;
} else {
pending_ra = NULL;
}
/* Try to cancel and re-schedule first stalled request for this app. */
if (got_response > 0 && !nxt_queue_is_empty(&app->pending)) {
lnk = nxt_queue_first(&app->pending);
re_ra = nxt_queue_link_data(lnk, nxt_request_app_link_t,
link_app_pending);
if (re_ra->res_time <= nxt_thread_monotonic_time(task->thread)) {
nxt_debug(task, "app '%V' stalled request #%uD detected",
&app->name, re_ra->stream);
cancelled = nxt_router_msg_cancel(task, &re_ra->msg_info,
re_ra->stream);
if (cancelled) {
state.req_app_link = re_ra;
state.app = app;
/*
* Need to increment use count "in advance" because
* nxt_router_port_select() will remove re_ra from lists
* and decrement use count.
*/
nxt_request_app_link_inc_use(re_ra);
nxt_router_port_select(task, &state);
goto re_ra_cancelled;
}
}
}
re_ra = NULL;
re_ra_cancelled:
send_quit = (app->max_requests > 0
&& port->app_pending_responses == 0
&& port->app_responses >= app->max_requests);
if (send_quit) {
port_unchained = nxt_queue_chk_remove(&port->app_link);
port->app = NULL;
app->processes--;
} else {
port_unchained = 0;
}
adjust_idle_timer = 0;
if (port->pair[1] != -1 && !send_quit && port->app_pending_responses == 0
&& nxt_queue_is_empty(&port->active_websockets)
&& port->idle_link.next == NULL)
{
if (app->idle_processes == app->spare_processes
&& app->adjust_idle_work.data == NULL)
{
adjust_idle_timer = 1;
app->adjust_idle_work.data = app;
app->adjust_idle_work.next = NULL;
}
if (app->idle_processes < app->spare_processes) {
nxt_queue_insert_tail(&app->spare_ports, &port->idle_link);
} else {
nxt_queue_insert_tail(&app->idle_ports, &port->idle_link);
port->idle_start = task->thread->engine->timers.now;
}
app->idle_processes++;
}
nxt_thread_mutex_unlock(&app->mutex);
if (adjust_idle_timer) {
nxt_router_app_use(task, app, 1);
nxt_event_engine_post(app->engine, &app->adjust_idle_work);
}
if (pending_ra != NULL) {
nxt_request_app_link_use(task, pending_ra, -1);
}
if (re_ra != NULL) {
if (nxt_router_port_post_select(task, &state) == NXT_OK) {
/*
* Reference counter already incremented above, this will
* keep re_ra while nxt_router_app_process_request()
* task is in queue. Reference counter decreased in
* nxt_router_app_process_request() after processing.
*/
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_router_app_process_request,
&task->thread->engine->task, app, re_ra);
} else {
nxt_request_app_link_use(task, re_ra, -1);
}
}
if (req_app_link != NULL) {
/*
* There should be call nxt_request_app_link_inc_use(req_app_link),
* because of one more link in the queue. But one link was
* recently removed from app->requests linked list.
* Corresponding decrement is in nxt_router_app_process_request().
*/
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_router_app_process_request,
&task->thread->engine->task, app, req_app_link);
goto adjust_use;
}
/* ? */
if (port->pair[1] == -1) {
nxt_debug(task, "app '%V' %p port %p already closed (pid %PI dead?)",
&app->name, app, port, port->pid);
goto adjust_use;
}
if (send_quit) {
nxt_debug(task, "app '%V' %p send QUIT to port",
&app->name, app);
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
if (port_unchained) {
nxt_port_use(task, port, -1);
}
goto adjust_use;
}
nxt_debug(task, "app '%V' %p requests queue is empty, keep the port",
&app->name, app);
adjust_use:
nxt_port_use(task, port, inc_use);
}
void
nxt_router_app_port_close(nxt_task_t *task, nxt_port_t *port)
{
nxt_app_t *app;
nxt_bool_t unchain, start_process;
nxt_port_t *idle_port;
nxt_queue_link_t *idle_lnk;
app = port->app;
nxt_assert(app != NULL);
nxt_thread_mutex_lock(&app->mutex);
unchain = nxt_queue_chk_remove(&port->app_link);
if (nxt_queue_chk_remove(&port->idle_link)) {
app->idle_processes--;
if (port->idle_start == 0
&& app->idle_processes >= app->spare_processes)
{
nxt_assert(!nxt_queue_is_empty(&app->idle_ports));
idle_lnk = nxt_queue_last(&app->idle_ports);
idle_port = nxt_queue_link_data(idle_lnk, nxt_port_t, idle_link);
nxt_queue_remove(idle_lnk);
nxt_queue_insert_tail(&app->spare_ports, idle_lnk);
idle_port->idle_start = 0;
}
}
app->processes--;
start_process = !task->thread->engine->shutdown
&& nxt_router_app_can_start(app)
&& (!nxt_queue_is_empty(&app->requests)
|| nxt_router_app_need_start(app));
if (start_process) {
app->pending_processes++;
}
nxt_thread_mutex_unlock(&app->mutex);
nxt_debug(task, "app '%V' pid %PI closed", &app->name, port->pid);
if (unchain) {
nxt_port_use(task, port, -1);
}
if (start_process) {
nxt_router_start_app_process(task, app);
}
}
static void
nxt_router_adjust_idle_timer(nxt_task_t *task, void *obj, void *data)
{
nxt_app_t *app;
nxt_bool_t queued;
nxt_port_t *port;
nxt_msec_t timeout, threshold;
nxt_queue_link_t *lnk;
nxt_event_engine_t *engine;
app = obj;
queued = (data == app);
nxt_debug(task, "nxt_router_adjust_idle_timer: app \"%V\", queued %b",
&app->name, queued);
engine = task->thread->engine;
nxt_assert(app->engine == engine);
threshold = engine->timers.now + app->joint->idle_timer.bias;
timeout = 0;
nxt_thread_mutex_lock(&app->mutex);
if (queued) {
app->adjust_idle_work.data = NULL;
}
while (app->idle_processes > app->spare_processes) {
nxt_assert(!nxt_queue_is_empty(&app->idle_ports));
lnk = nxt_queue_first(&app->idle_ports);
port = nxt_queue_link_data(lnk, nxt_port_t, idle_link);
timeout = port->idle_start + app->idle_timeout;
if (timeout > threshold) {
break;
}
nxt_queue_remove(lnk);
lnk->next = NULL;
nxt_queue_chk_remove(&port->app_link);
app->idle_processes--;
app->processes--;
port->app = NULL;
nxt_thread_mutex_unlock(&app->mutex);
nxt_debug(task, "app '%V' send QUIT to idle port %PI",
&app->name, port->pid);
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
nxt_port_use(task, port, -1);
nxt_thread_mutex_lock(&app->mutex);
}
nxt_thread_mutex_unlock(&app->mutex);
if (timeout > threshold) {
nxt_timer_add(engine, &app->joint->idle_timer, timeout - threshold);
} else {
nxt_timer_disable(engine, &app->joint->idle_timer);
}
if (queued) {
nxt_router_app_use(task, app, -1);
}
}
static void
nxt_router_app_idle_timeout(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *timer;
nxt_app_joint_t *app_joint;
timer = obj;
app_joint = nxt_container_of(timer, nxt_app_joint_t, idle_timer);
if (nxt_fast_path(app_joint->app != NULL)) {
nxt_router_adjust_idle_timer(task, app_joint->app, NULL);
}
}
static void
nxt_router_app_joint_release_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_timer_t *timer;
nxt_app_joint_t *app_joint;
timer = obj;
app_joint = nxt_container_of(timer, nxt_app_joint_t, idle_timer);
nxt_router_app_joint_use(task, app_joint, -1);
}
static void
nxt_router_free_app(nxt_task_t *task, void *obj, void *data)
{
nxt_app_t *app;
nxt_port_t *port;
nxt_app_joint_t *app_joint;
app_joint = obj;
app = app_joint->app;
for ( ;; ) {
port = nxt_router_app_get_port_for_quit(app);
if (port == NULL) {
break;
}
nxt_debug(task, "send QUIT to app '%V' pid %PI", &app->name, port->pid);
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
nxt_port_use(task, port, -1);
}
nxt_assert(app->processes == 0);
nxt_assert(app->idle_processes == 0);
nxt_assert(nxt_queue_is_empty(&app->requests));
nxt_assert(nxt_queue_is_empty(&app->ports));
nxt_assert(nxt_queue_is_empty(&app->spare_ports));
nxt_assert(nxt_queue_is_empty(&app->idle_ports));
nxt_thread_mutex_destroy(&app->mutex);
nxt_free(app);
app_joint->app = NULL;
if (nxt_timer_delete(task->thread->engine, &app_joint->idle_timer)) {
app_joint->idle_timer.handler = nxt_router_app_joint_release_handler;
nxt_timer_add(task->thread->engine, &app_joint->idle_timer, 0);
} else {
nxt_router_app_joint_use(task, app_joint, -1);
}
}
static void
nxt_router_port_select(nxt_task_t *task, nxt_port_select_state_t *state)
{
int ra_use_delta;
nxt_app_t *app;
nxt_bool_t can_start_process;
nxt_request_app_link_t *req_app_link;
req_app_link = state->req_app_link;
app = state->app;
state->failed_port_use_delta = 0;
ra_use_delta = -nxt_queue_chk_remove(&req_app_link->link_app_requests);
if (nxt_queue_chk_remove(&req_app_link->link_port_pending))
{
nxt_assert(req_app_link->link_app_pending.next != NULL);
nxt_queue_remove(&req_app_link->link_app_pending);
req_app_link->link_app_pending.next = NULL;
ra_use_delta--;
}
state->failed_port = req_app_link->app_port;
if (req_app_link->app_port != NULL) {
state->failed_port_use_delta--;
state->failed_port->app_pending_responses--;
if (nxt_queue_chk_remove(&state->failed_port->app_link)) {
state->failed_port_use_delta--;
}
req_app_link->app_port = NULL;
}
can_start_process = nxt_router_app_can_start(app);
state->port = NULL;
state->start_process = 0;
if (nxt_queue_is_empty(&app->ports)
|| (can_start_process && nxt_router_app_first_port_busy(app)) )
{
req_app_link = nxt_request_app_link_alloc(task, req_app_link,
req_app_link->req_rpc_data);
if (nxt_slow_path(req_app_link == NULL)) {
goto fail;
}
if (nxt_slow_path(state->failed_port != NULL)) {
nxt_queue_insert_head(&app->requests,
&req_app_link->link_app_requests);
} else {
nxt_queue_insert_tail(&app->requests,
&req_app_link->link_app_requests);
}
ra_use_delta++;
nxt_debug(task, "req_app_link stream #%uD enqueue to app->requests",
req_app_link->stream);
if (can_start_process) {
app->pending_processes++;
state->start_process = 1;
}
} else {
state->port = nxt_router_pop_first_port(app);
if (state->port->app_pending_responses > 1) {
req_app_link = nxt_request_app_link_alloc(task, req_app_link,
req_app_link->req_rpc_data);
if (nxt_slow_path(req_app_link == NULL)) {
goto fail;
}
req_app_link->app_port = state->port;
nxt_request_app_link_pending(task, app, req_app_link);
}
if (can_start_process && nxt_router_app_need_start(app)) {
app->pending_processes++;
state->start_process = 1;
}
}
nxt_request_app_link_chk_use(req_app_link, ra_use_delta);
fail:
state->shared_ra = req_app_link;
}
static nxt_int_t
nxt_router_port_post_select(nxt_task_t *task, nxt_port_select_state_t *state)
{
nxt_int_t res;
nxt_app_t *app;
nxt_request_app_link_t *req_app_link;
req_app_link = state->shared_ra;
app = state->app;
if (state->failed_port_use_delta != 0) {
nxt_port_use(task, state->failed_port, state->failed_port_use_delta);
}
if (nxt_slow_path(req_app_link == NULL)) {
if (state->port != NULL) {
nxt_port_use(task, state->port, -1);
}
nxt_request_app_link_error(state->req_app_link, 500,
"Failed to allocate shared req<->app link");
return NXT_ERROR;
}
if (state->port != NULL) {
nxt_debug(task, "already have port for app '%V' %p ", &app->name, app);
req_app_link->app_port = state->port;
if (state->start_process) {
nxt_router_start_app_process(task, app);
}
return NXT_OK;
}
if (!state->start_process) {
nxt_debug(task, "app '%V' %p too many running or pending processes",
&app->name, app);
return NXT_AGAIN;
}
res = nxt_router_start_app_process(task, app);
if (nxt_slow_path(res != NXT_OK)) {
nxt_request_app_link_error(req_app_link, 500,
"Failed to start app process");
return NXT_ERROR;
}
return NXT_AGAIN;
}
static nxt_int_t
nxt_router_app_port(nxt_task_t *task, nxt_app_t *app,
nxt_request_app_link_t *req_app_link)
{
nxt_port_select_state_t state;
state.req_app_link = req_app_link;
state.app = app;
nxt_thread_mutex_lock(&app->mutex);
nxt_router_port_select(task, &state);
nxt_thread_mutex_unlock(&app->mutex);
return nxt_router_port_post_select(task, &state);
}
void
nxt_router_process_http_request(nxt_task_t *task, nxt_http_request_t *r,
nxt_app_t *app)
{
nxt_int_t res;
nxt_port_t *port;
nxt_event_engine_t *engine;
nxt_request_app_link_t ra_local, *req_app_link;
nxt_request_rpc_data_t *req_rpc_data;
engine = task->thread->engine;
req_rpc_data = nxt_port_rpc_register_handler_ex(task, engine->port,
nxt_router_response_ready_handler,
nxt_router_response_error_handler,
sizeof(nxt_request_rpc_data_t));
if (nxt_slow_path(req_rpc_data == NULL)) {
nxt_http_request_error(task, r, NXT_HTTP_INTERNAL_SERVER_ERROR);
return;
}
/*
* At this point we have request req_rpc_data allocated and registered
* in port handlers. Need to fixup request memory pool. Counterpart
* release will be called via following call chain:
* nxt_request_rpc_data_unlink() ->
* nxt_router_http_request_done() ->
* nxt_router_http_request_release()
*/
nxt_mp_retain(r->mem_pool);
r->timer.task = &engine->task;
r->timer.work_queue = &engine->fast_work_queue;
r->timer.log = engine->task.log;
r->timer.bias = NXT_TIMER_DEFAULT_BIAS;
req_rpc_data->stream = nxt_port_rpc_ex_stream(req_rpc_data);
req_rpc_data->app = app;
nxt_router_app_use(task, app, 1);
req_rpc_data->request = r;
r->req_rpc_data = req_rpc_data;
req_app_link = &ra_local;
nxt_request_app_link_init(task, req_app_link, req_rpc_data);
res = nxt_router_app_port(task, app, req_app_link);
req_app_link = req_rpc_data->req_app_link;
if (res == NXT_OK) {
port = req_app_link->app_port;
nxt_assert(port != NULL);
nxt_port_rpc_ex_set_peer(task, engine->port, req_rpc_data, port->pid);
nxt_router_app_prepare_request(task, req_app_link);
}
nxt_request_app_link_use(task, req_app_link, -1);
}
static void
nxt_router_dummy_buf_completion(nxt_task_t *task, void *obj, void *data)
{
}
static void
nxt_router_app_prepare_request(nxt_task_t *task,
nxt_request_app_link_t *req_app_link)
{
nxt_buf_t *buf;
nxt_int_t res;
nxt_port_t *port, *c_port, *reply_port;
nxt_apr_action_t apr_action;
nxt_assert(req_app_link->app_port != NULL);
port = req_app_link->app_port;
reply_port = req_app_link->reply_port;
apr_action = NXT_APR_REQUEST_FAILED;
c_port = nxt_process_connected_port_find(port->process, reply_port->pid,
reply_port->id);
if (nxt_slow_path(c_port != reply_port)) {
res = nxt_port_send_port(task, port, reply_port, 0);
if (nxt_slow_path(res != NXT_OK)) {
nxt_request_app_link_error(req_app_link, 500,
"Failed to send reply port to application");
goto release_port;
}
nxt_process_connected_port_add(port->process, reply_port);
}
buf = nxt_router_prepare_msg(task, req_app_link->request, port,
nxt_app_msg_prefix[port->app->type]);
if (nxt_slow_path(buf == NULL)) {
nxt_request_app_link_error(req_app_link, 500,
"Failed to prepare message for application");
goto release_port;
}
nxt_debug(task, "about to send %O bytes buffer to app process port %d",
nxt_buf_used_size(buf),
port->socket.fd);
apr_action = NXT_APR_NEW_PORT;
req_app_link->msg_info.buf = buf;
req_app_link->msg_info.completion_handler = buf->completion_handler;
for (; buf; buf = buf->next) {
buf->completion_handler = nxt_router_dummy_buf_completion;
}
buf = req_app_link->msg_info.buf;
res = nxt_port_mmap_get_tracking(task, port,
&req_app_link->msg_info.tracking,
req_app_link->stream);
if (nxt_slow_path(res != NXT_OK)) {
nxt_request_app_link_error(req_app_link, 500,
"Failed to get tracking area");
goto release_port;
}
if (req_app_link->body_fd != -1) {
nxt_debug(task, "stream #%uD: send body fd %d", req_app_link->stream,
req_app_link->body_fd);
lseek(req_app_link->body_fd, 0, SEEK_SET);
}
res = nxt_port_socket_twrite(task, port, NXT_PORT_MSG_REQ_HEADERS,
req_app_link->body_fd,
req_app_link->stream, reply_port->id, buf,
&req_app_link->msg_info.tracking);
if (nxt_slow_path(res != NXT_OK)) {
nxt_request_app_link_error(req_app_link, 500,
"Failed to send message to application");
goto release_port;
}
release_port:
nxt_router_app_port_release(task, port, apr_action);
nxt_request_app_link_update_peer(task, req_app_link);
}
struct nxt_fields_iter_s {
nxt_list_part_t *part;
nxt_http_field_t *field;
};
typedef struct nxt_fields_iter_s nxt_fields_iter_t;
static nxt_http_field_t *
nxt_fields_part_first(nxt_list_part_t *part, nxt_fields_iter_t *i)
{
if (part == NULL) {
return NULL;
}
while (part->nelts == 0) {
part = part->next;
if (part == NULL) {
return NULL;
}
}
i->part = part;
i->field = nxt_list_data(i->part);
return i->field;
}
static nxt_http_field_t *
nxt_fields_first(nxt_list_t *fields, nxt_fields_iter_t *i)
{
return nxt_fields_part_first(nxt_list_part(fields), i);
}
static nxt_http_field_t *
nxt_fields_next(nxt_fields_iter_t *i)
{
nxt_http_field_t *end = nxt_list_data(i->part);
end += i->part->nelts;
i->field++;
if (i->field < end) {
return i->field;
}
return nxt_fields_part_first(i->part->next, i);
}
static nxt_buf_t *
nxt_router_prepare_msg(nxt_task_t *task, nxt_http_request_t *r,
nxt_port_t *port, const nxt_str_t *prefix)
{
void *target_pos, *query_pos;
u_char *pos, *end, *p, c;
size_t fields_count, req_size, size, free_size;
size_t copy_size;
nxt_off_t content_length;
nxt_buf_t *b, *buf, *out, **tail;
nxt_http_field_t *field, *dup;
nxt_unit_field_t *dst_field;
nxt_fields_iter_t iter, dup_iter;
nxt_unit_request_t *req;
req_size = sizeof(nxt_unit_request_t)
+ r->method->length + 1
+ r->version.length + 1
+ r->remote->length + 1
+ r->local->length + 1
+ r->server_name.length + 1
+ r->target.length + 1
+ (r->path->start != r->target.start ? r->path->length + 1 : 0);
content_length = r->content_length_n < 0 ? 0 : r->content_length_n;
fields_count = 0;
nxt_list_each(field, r->fields) {
fields_count++;
req_size += field->name_length + prefix->length + 1
+ field->value_length + 1;
} nxt_list_loop;
req_size += fields_count * sizeof(nxt_unit_field_t);
if (nxt_slow_path(req_size > PORT_MMAP_DATA_SIZE)) {
nxt_alert(task, "headers to big to fit in shared memory (%d)",
(int) req_size);
return NULL;
}
out = nxt_port_mmap_get_buf(task, port,
nxt_min(req_size + content_length, PORT_MMAP_DATA_SIZE));
if (nxt_slow_path(out == NULL)) {
return NULL;
}
req = (nxt_unit_request_t *) out->mem.free;
out->mem.free += req_size;
req->content_length = content_length;
p = (u_char *) (req->fields + fields_count);
nxt_debug(task, "fields_count=%d", (int) fields_count);
req->method_length = r->method->length;
nxt_unit_sptr_set(&req->method, p);
p = nxt_cpymem(p, r->method->start, r->method->length);
*p++ = '\0';
req->version_length = r->version.length;
nxt_unit_sptr_set(&req->version, p);
p = nxt_cpymem(p, r->version.start, r->version.length);
*p++ = '\0';
req->remote_length = r->remote->address_length;
nxt_unit_sptr_set(&req->remote, p);
p = nxt_cpymem(p, nxt_sockaddr_address(r->remote),
r->remote->address_length);
*p++ = '\0';
req->local_length = r->local->address_length;
nxt_unit_sptr_set(&req->local, p);
p = nxt_cpymem(p, nxt_sockaddr_address(r->local), r->local->address_length);
*p++ = '\0';
req->tls = (r->tls != NULL);
req->websocket_handshake = r->websocket_handshake;
req->server_name_length = r->server_name.length;
nxt_unit_sptr_set(&req->server_name, p);
p = nxt_cpymem(p, r->server_name.start, r->server_name.length);
*p++ = '\0';
target_pos = p;
req->target_length = (uint32_t) r->target.length;
nxt_unit_sptr_set(&req->target, p);
p = nxt_cpymem(p, r->target.start, r->target.length);
*p++ = '\0';
req->path_length = (uint32_t) r->path->length;
if (r->path->start == r->target.start) {
nxt_unit_sptr_set(&req->path, target_pos);
} else {
nxt_unit_sptr_set(&req->path, p);
p = nxt_cpymem(p, r->path->start, r->path->length);
*p++ = '\0';
}
req->query_length = r->args != NULL ? (uint32_t) r->args->length : 0;
if (r->args != NULL && r->args->start != NULL) {
query_pos = nxt_pointer_to(target_pos,
r->args->start - r->target.start);
nxt_unit_sptr_set(&req->query, query_pos);
} else {
req->query.offset = 0;
}
req->content_length_field = NXT_UNIT_NONE_FIELD;
req->content_type_field = NXT_UNIT_NONE_FIELD;
req->cookie_field = NXT_UNIT_NONE_FIELD;
dst_field = req->fields;
for (field = nxt_fields_first(r->fields, &iter);
field != NULL;
field = nxt_fields_next(&iter))
{
if (field->skip) {
continue;
}
dst_field->hash = field->hash;
dst_field->skip = 0;
dst_field->name_length = field->name_length + prefix->length;
dst_field->value_length = field->value_length;
if (field == r->content_length) {
req->content_length_field = dst_field - req->fields;
} else if (field == r->content_type) {
req->content_type_field = dst_field - req->fields;
} else if (field == r->cookie) {
req->cookie_field = dst_field - req->fields;
}
nxt_debug(task, "add field 0x%04Xd, %d, %d, %p : %d %p",
(int) field->hash, (int) field->skip,
(int) field->name_length, field->name,
(int) field->value_length, field->value);
if (prefix->length != 0) {
nxt_unit_sptr_set(&dst_field->name, p);
p = nxt_cpymem(p, prefix->start, prefix->length);
end = field->name + field->name_length;
for (pos = field->name; pos < end; pos++) {
c = *pos;
if (c >= 'a' && c <= 'z') {
*p++ = (c & ~0x20);
continue;
}
if (c == '-') {
*p++ = '_';
continue;
}
*p++ = c;
}
} else {
nxt_unit_sptr_set(&dst_field->name, p);
p = nxt_cpymem(p, field->name, field->name_length);
}
*p++ = '\0';
nxt_unit_sptr_set(&dst_field->value, p);
p = nxt_cpymem(p, field->value, field->value_length);
if (prefix->length != 0) {
dup_iter = iter;
for (dup = nxt_fields_next(&dup_iter);
dup != NULL;
dup = nxt_fields_next(&dup_iter))
{
if (dup->name_length != field->name_length
|| dup->skip
|| dup->hash != field->hash
|| nxt_memcasecmp(dup->name, field->name, dup->name_length))
{
continue;
}
p = nxt_cpymem(p, ", ", 2);
p = nxt_cpymem(p, dup->value, dup->value_length);
dst_field->value_length += 2 + dup->value_length;
dup->skip = 1;
}
}
*p++ = '\0';
dst_field++;
}
req->fields_count = (uint32_t) (dst_field - req->fields);
nxt_unit_sptr_set(&req->preread_content, out->mem.free);
buf = out;
tail = &buf->next;
for (b = r->body; b != NULL; b = b->next) {
size = nxt_buf_mem_used_size(&b->mem);
pos = b->mem.pos;
while (size > 0) {
if (buf == NULL) {
free_size = nxt_min(size, PORT_MMAP_DATA_SIZE);
buf = nxt_port_mmap_get_buf(task, port, free_size);
if (nxt_slow_path(buf == NULL)) {
while (out != NULL) {
buf = out->next;
out->next = NULL;
out->completion_handler(task, out, out->parent);
out = buf;
}
return NULL;
}
*tail = buf;
tail = &buf->next;
} else {
free_size = nxt_buf_mem_free_size(&buf->mem);
if (free_size < size
&& nxt_port_mmap_increase_buf(task, buf, size, 1)
== NXT_OK)
{
free_size = nxt_buf_mem_free_size(&buf->mem);
}
}
if (free_size > 0) {
copy_size = nxt_min(free_size, size);
buf->mem.free = nxt_cpymem(buf->mem.free, pos, copy_size);
size -= copy_size;
pos += copy_size;
if (size == 0) {
break;
}
}
buf = NULL;
}
}
return out;
}
static void
nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data)
{
nxt_app_t *app;
nxt_bool_t cancelled, unlinked;
nxt_port_t *port;
nxt_timer_t *timer;
nxt_queue_link_t *lnk;
nxt_http_request_t *r;
nxt_request_app_link_t *pending_ra;
nxt_request_rpc_data_t *req_rpc_data;
nxt_port_select_state_t state;
timer = obj;
nxt_debug(task, "router app timeout");
r = nxt_timer_data(timer, nxt_http_request_t, timer);
req_rpc_data = r->timer_data;
app = req_rpc_data->app;
if (app == NULL) {
goto generate_error;
}
port = NULL;
pending_ra = NULL;
if (req_rpc_data->app_port != NULL) {
port = req_rpc_data->app_port;
req_rpc_data->app_port = NULL;
}
if (port == NULL && req_rpc_data->req_app_link != NULL
&& req_rpc_data->req_app_link->app_port != NULL)
{
port = req_rpc_data->req_app_link->app_port;
req_rpc_data->req_app_link->app_port = NULL;
}
if (port == NULL) {
goto generate_error;
}
nxt_thread_mutex_lock(&app->mutex);
unlinked = nxt_queue_chk_remove(&port->app_link);
if (!nxt_queue_is_empty(&port->pending_requests)) {
lnk = nxt_queue_first(&port->pending_requests);
pending_ra = nxt_queue_link_data(lnk, nxt_request_app_link_t,
link_port_pending);
nxt_assert(pending_ra->link_app_pending.next != NULL);
nxt_debug(task, "app '%V' pending request #%uD found",
&app->name, pending_ra->stream);
cancelled = nxt_router_msg_cancel(task, &pending_ra->msg_info,
pending_ra->stream);
if (cancelled) {
state.req_app_link = pending_ra;
state.app = app;
/*
* Need to increment use count "in advance" because
* nxt_router_port_select() will remove pending_ra from lists
* and decrement use count.
*/
nxt_request_app_link_inc_use(pending_ra);
nxt_router_port_select(task, &state);
} else {
pending_ra = NULL;
}
}
nxt_thread_mutex_unlock(&app->mutex);
if (pending_ra != NULL) {
if (nxt_router_port_post_select(task, &state) == NXT_OK) {
/*
* Reference counter already incremented above, this will
* keep pending_ra while nxt_router_app_process_request()
* task is in queue. Reference counter decreased in
* nxt_router_app_process_request() after processing.
*/
nxt_work_queue_add(&task->thread->engine->fast_work_queue,
nxt_router_app_process_request,
&task->thread->engine->task, app, pending_ra);
} else {
nxt_request_app_link_use(task, pending_ra, -1);
}
}
nxt_debug(task, "send quit to app '%V' pid %PI", &app->name, port->pid);
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
nxt_port_use(task, port, unlinked ? -2 : -1);
generate_error:
nxt_http_request_error(task, r, NXT_HTTP_SERVICE_UNAVAILABLE);
nxt_request_rpc_data_unlink(task, req_rpc_data);
}
static nxt_int_t
nxt_router_http_request_done(nxt_task_t *task, nxt_http_request_t *r)
{
r->timer.handler = nxt_router_http_request_release;
nxt_timer_add(task->thread->engine, &r->timer, 0);
return NXT_OK;
}
static void
nxt_router_http_request_release(nxt_task_t *task, void *obj, void *data)
{
nxt_http_request_t *r;
nxt_debug(task, "http app release");
r = nxt_timer_data(obj, nxt_http_request_t, timer);
nxt_mp_release(r->mem_pool);
}
static void
nxt_router_oosm_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
size_t mi;
uint32_t i;
nxt_bool_t ack;
nxt_process_t *process;
nxt_free_map_t *m;
nxt_port_mmap_header_t *hdr;
nxt_debug(task, "oosm in %PI", msg->port_msg.pid);
process = nxt_runtime_process_find(task->thread->runtime,
msg->port_msg.pid);
if (nxt_slow_path(process == NULL)) {
return;
}
ack = 0;
/*
* To mitigate possible racing condition (when OOSM message received
* after some of the memory was already freed), need to try to find
* first free segment in shared memory and send ACK if found.
*/
nxt_thread_mutex_lock(&process->incoming.mutex);
for (i = 0; i < process->incoming.size; i++) {
hdr = process->incoming.elts[i].mmap_handler->hdr;
m = hdr->free_map;
for (mi = 0; mi < MAX_FREE_IDX; mi++) {
if (m[mi] != 0) {
ack = 1;
nxt_debug(task, "oosm: already free #%uD %uz = 0x%08xA",
i, mi, m[mi]);
break;
}
}
}
nxt_thread_mutex_unlock(&process->incoming.mutex);
if (ack) {
(void) nxt_port_socket_write(task, msg->port, NXT_PORT_MSG_SHM_ACK,
-1, 0, 0, NULL);
}
}
Reference in New Issue View Git Blame Copy Permalink