diff options
| -rw-r--r-- | src/nxt_master_process.c | 197 | ||||
| -rw-r--r-- | src/nxt_master_process.h | 11 | ||||
| -rw-r--r-- | src/nxt_port.h | 2 | ||||
| -rw-r--r-- | src/nxt_router.c | 255 | ||||
| -rw-r--r-- | src/nxt_worker_process.c | 4 |
5 files changed, 406 insertions, 63 deletions
diff --git a/src/nxt_master_process.c b/src/nxt_master_process.c index 313456ac..e4b95fe8 100644 --- a/src/nxt_master_process.c +++ b/src/nxt_master_process.c @@ -12,6 +12,14 @@ #include <nxt_application.h> +typedef struct { + nxt_socket_t socket; + nxt_socket_error_t error; + u_char *start; + u_char *end; +} nxt_listening_socket_t; + + static nxt_int_t nxt_master_process_port_create(nxt_task_t *task, nxt_runtime_t *rt); static void nxt_master_process_title(nxt_task_t *task); @@ -32,6 +40,10 @@ static void nxt_master_process_sigusr1_handler(nxt_task_t *task, void *obj, static void nxt_master_process_sigchld_handler(nxt_task_t *task, void *obj, void *data); static void nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid); +static void nxt_master_port_socket_handler(nxt_task_t *task, + nxt_port_recv_msg_t *msg); +static nxt_int_t nxt_master_listening_socket(nxt_sockaddr_t *sa, + nxt_listening_socket_t *ls); const nxt_sig_event_t nxt_master_process_signals[] = { @@ -206,6 +218,7 @@ static nxt_port_handler_t nxt_master_process_port_handlers[] = { NULL, /* NXT_PORT_MSG_REMOVE_PID */ nxt_port_ready_handler, nxt_port_master_start_worker_handler, + nxt_master_port_socket_handler, nxt_port_rpc_handler, nxt_port_rpc_handler, }; @@ -691,3 +704,187 @@ nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid) } } } + + +static void +nxt_master_port_socket_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) +{ + size_t size; + nxt_int_t ret; + nxt_buf_t *b, *out; + nxt_port_t *port; + nxt_sockaddr_t *sa; + nxt_port_msg_type_t type; + nxt_listening_socket_t ls; + u_char message[2048]; + + b = msg->buf; + sa = (nxt_sockaddr_t *) b->mem.pos; + + out = NULL; + + ls.socket = -1; + ls.error = NXT_SOCKET_ERROR_SYSTEM; + ls.start = message; + ls.end = message + sizeof(message); + + port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid, + msg->port_msg.reply_port); + + nxt_debug(task, "listening socket \"%*s\"", + sa->length, nxt_sockaddr_start(sa)); + + ret = nxt_master_listening_socket(sa, &ls); + + if (ret == NXT_OK) { + nxt_debug(task, "socket(\"%*s\"): %d", + sa->length, nxt_sockaddr_start(sa), ls.socket); + + type = NXT_PORT_MSG_RPC_READY_LAST | NXT_PORT_MSG_CLOSE_FD; + + } else { + size = ls.end - ls.start; + + nxt_log(task, NXT_LOG_CRIT, "%*s", size, ls.start); + + out = nxt_buf_mem_alloc(port->mem_pool, size + 1, 0); + if (nxt_slow_path(out == NULL)) { + return; + } + + *out->mem.free++ = (uint8_t) ls.error; + + out->mem.free = nxt_cpymem(out->mem.free, ls.start, size); + + type = NXT_PORT_MSG_RPC_ERROR; + } + + nxt_port_socket_write(task, port, type, ls.socket, msg->port_msg.stream, + 0, out); +} + + +static nxt_int_t +nxt_master_listening_socket(nxt_sockaddr_t *sa, nxt_listening_socket_t *ls) +{ + nxt_err_t err; + nxt_socket_t s; + + const socklen_t length = sizeof(int); + const static int enable = 1; + + s = socket(sa->u.sockaddr.sa_family, sa->type, 0); + + if (nxt_slow_path(s == -1)) { + err = nxt_errno; + +#if (NXT_INET6) + + if (err == EAFNOSUPPORT && sa->u.sockaddr.sa_family == AF_INET6) { + ls->error = NXT_SOCKET_ERROR_NOINET6; + } + +#endif + + ls->end = nxt_sprintf(ls->start, ls->end, + "socket(\\\"%*s\\\") failed %E", + sa->length, nxt_sockaddr_start(sa), err); + + return NXT_ERROR; + } + + if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &enable, length) != 0) { + ls->end = nxt_sprintf(ls->start, ls->end, + "setsockopt(\\\"%*s\\\", SO_REUSEADDR) failed %E", + sa->length, nxt_sockaddr_start(sa), nxt_errno); + goto fail; + } + +#if (NXT_INET6) + + if (sa->u.sockaddr.sa_family == AF_INET6) { + + if (setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY, &enable, length) != 0) { + ls->end = nxt_sprintf(ls->start, ls->end, + "setsockopt(\\\"%*s\\\", IPV6_V6ONLY) failed %E", + sa->length, nxt_sockaddr_start(sa), nxt_errno); + goto fail; + } + } + +#endif + + if (bind(s, &sa->u.sockaddr, sa->socklen) != 0) { + err = nxt_errno; + +#if (NXT_HAVE_UNIX_DOMAIN) + + if (sa->u.sockaddr.sa_family == AF_UNIX) { + switch (err) { + + case EACCES: + ls->error = NXT_SOCKET_ERROR_ACCESS; + break; + + case ENOENT: + case ENOTDIR: + ls->error = NXT_SOCKET_ERROR_PATH; + break; + } + + goto next; + } + +#endif + + switch (err) { + + case EACCES: + ls->error = NXT_SOCKET_ERROR_PORT; + break; + + case EADDRINUSE: + ls->error = NXT_SOCKET_ERROR_INUSE; + break; + + case EADDRNOTAVAIL: + ls->error = NXT_SOCKET_ERROR_NOADDR; + break; + } + + ls->end = nxt_sprintf(ls->start, ls->end, "bind(\\\"%*s\\\") failed %E", + sa->length, nxt_sockaddr_start(sa), err); + goto fail; + } + +#if (NXT_HAVE_UNIX_DOMAIN) + +next: + + if (sa->u.sockaddr.sa_family == AF_UNIX) { + char *filename; + mode_t access; + + filename = sa->u.sockaddr_un.sun_path; + access = (S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH); + + if (chmod(filename, access) != 0) { + ls->end = nxt_sprintf(ls->start, ls->end, + "chmod(\\\"%*s\\\") failed %E", + filename, nxt_errno); + goto fail; + } + } + +#endif + + ls->socket = s; + + return NXT_OK; + +fail: + + (void) close(s); + + return NXT_ERROR; +} diff --git a/src/nxt_master_process.h b/src/nxt_master_process.h index bcdf645d..3a7e4988 100644 --- a/src/nxt_master_process.h +++ b/src/nxt_master_process.h @@ -8,6 +8,17 @@ #define _NXT_MASTER_PROCESS_H_INCLUDED_ +typedef enum { + NXT_SOCKET_ERROR_SYSTEM = 0, + NXT_SOCKET_ERROR_NOINET6, + NXT_SOCKET_ERROR_PORT, + NXT_SOCKET_ERROR_INUSE, + NXT_SOCKET_ERROR_NOADDR, + NXT_SOCKET_ERROR_ACCESS, + NXT_SOCKET_ERROR_PATH, +} nxt_socket_error_t; + + nxt_int_t nxt_master_process_start(nxt_thread_t *thr, nxt_task_t *task, nxt_runtime_t *runtime); void nxt_master_stop_worker_processes(nxt_task_t *task, nxt_runtime_t *runtime); diff --git a/src/nxt_port.h b/src/nxt_port.h index 1e39c732..90e88a68 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -22,6 +22,7 @@ typedef enum { _NXT_PORT_MSG_REMOVE_PID, _NXT_PORT_MSG_READY, _NXT_PORT_MSG_START_WORKER, + _NXT_PORT_MSG_SOCKET, _NXT_PORT_MSG_RPC_READY, _NXT_PORT_MSG_RPC_ERROR, @@ -38,6 +39,7 @@ typedef enum { NXT_PORT_MSG_READY = _NXT_PORT_MSG_READY | NXT_PORT_MSG_LAST, NXT_PORT_MSG_START_WORKER = _NXT_PORT_MSG_START_WORKER | NXT_PORT_MSG_LAST, + NXT_PORT_MSG_SOCKET = _NXT_PORT_MSG_SOCKET | NXT_PORT_MSG_LAST, NXT_PORT_MSG_RPC_READY = _NXT_PORT_MSG_RPC_READY, NXT_PORT_MSG_RPC_READY_LAST = _NXT_PORT_MSG_RPC_READY | NXT_PORT_MSG_LAST, NXT_PORT_MSG_RPC_ERROR = _NXT_PORT_MSG_RPC_ERROR | NXT_PORT_MSG_LAST, diff --git a/src/nxt_router.c b/src/nxt_router.c index 5f654050..5beceb38 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -45,10 +45,15 @@ struct nxt_req_app_link_s { }; +typedef struct { + nxt_socket_conf_t *socket_conf; + nxt_router_temp_conf_t *temp_conf; +} nxt_socket_rpc_t; + + static nxt_router_temp_conf_t *nxt_router_temp_conf(nxt_task_t *task); -static nxt_int_t nxt_router_conf_new(nxt_task_t *task, - nxt_router_temp_conf_t *tmcf, u_char *start, u_char *end); -static void nxt_router_conf_success(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); @@ -62,8 +67,12 @@ static nxt_int_t nxt_router_conf_create(nxt_task_t *task, static nxt_app_t *nxt_router_app_find(nxt_queue_t *queue, nxt_str_t *name); static nxt_app_t *nxt_router_listener_application(nxt_router_temp_conf_t *tmcf, nxt_str_t *name); -static nxt_int_t nxt_router_listen_sockets_stub_create(nxt_task_t *task, - nxt_router_temp_conf_t *tmcf); +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); static nxt_socket_conf_t *nxt_router_socket_conf(nxt_task_t *task, nxt_mp_t *mp, nxt_sockaddr_t *sa); @@ -306,8 +315,8 @@ void nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) { size_t dump_size; - nxt_buf_t *b; nxt_int_t ret; + nxt_buf_t *b; nxt_router_temp_conf_t *tmcf; b = msg->buf; @@ -329,18 +338,17 @@ nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) tmcf->conf->router = nxt_router; tmcf->stream = msg->port_msg.stream; tmcf->port = nxt_runtime_port_find(task->thread->runtime, - msg->port_msg.pid, 0); + msg->port_msg.pid, + msg->port_msg.reply_port); - ret = nxt_router_conf_new(task, tmcf, b->mem.pos, b->mem.free); + ret = nxt_router_conf_create(task, tmcf, b->mem.pos, b->mem.free); - if (ret == NXT_OK) { - nxt_router_conf_success(task, tmcf); - return; - } - - nxt_log(task, NXT_LOG_CRIT, "failed to apply new conf"); + if (nxt_fast_path(ret == NXT_OK)) { + nxt_router_conf_apply(task, tmcf, NULL); - nxt_router_conf_error(task, tmcf); + } else { + nxt_router_conf_error(task, tmcf); + } } @@ -421,41 +429,46 @@ fail: } -static nxt_int_t -nxt_router_conf_new(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, - u_char *start, u_char *end) +static void +nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data) { nxt_int_t ret; nxt_router_t *router; nxt_runtime_t *rt; + nxt_queue_link_t *qlk; + nxt_socket_conf_t *skcf; + nxt_router_temp_conf_t *tmcf; const nxt_event_interface_t *interface; - ret = nxt_router_conf_create(task, tmcf, start, end); - if (nxt_slow_path(ret != NXT_OK)) { - return ret; - } + tmcf = obj; - router = tmcf->conf->router; + qlk = nxt_queue_first(&tmcf->pending); - nxt_router_listen_sockets_sort(router, tmcf); + if (qlk != nxt_queue_tail(&tmcf->pending)) { + nxt_queue_remove(qlk); + nxt_queue_insert_tail(&tmcf->creating, qlk); - ret = nxt_router_listen_sockets_stub_create(task, tmcf); - if (nxt_slow_path(ret != NXT_OK)) { - return ret; + skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link); + + nxt_router_listen_socket_rpc_create(task, tmcf, skcf); + + return; } rt = task->thread->runtime; interface = nxt_service_get(rt->services, "engine", NULL); + router = tmcf->conf->router; + ret = nxt_router_engines_create(task, router, tmcf, interface); if (nxt_slow_path(ret != NXT_OK)) { - return ret; + goto fail; } ret = nxt_router_threads_create(task, rt, tmcf); if (nxt_slow_path(ret != NXT_OK)) { - return ret; + goto fail; } nxt_router_apps_sort(router, tmcf); @@ -465,7 +478,15 @@ nxt_router_conf_new(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_queue_add(&router->sockets, &tmcf->updating); nxt_queue_add(&router->sockets, &tmcf->creating); - return NXT_OK; + nxt_router_conf_ready(task, tmcf); + + return; + +fail: + + nxt_router_conf_error(task, tmcf); + + return; } @@ -476,12 +497,12 @@ nxt_router_conf_wait(nxt_task_t *task, void *obj, void *data) job = obj; - nxt_router_conf_success(task, job->tmcf); + nxt_router_conf_ready(task, job->tmcf); } static void -nxt_router_conf_success(nxt_task_t *task, nxt_router_temp_conf_t *tmcf) +nxt_router_conf_ready(nxt_task_t *task, nxt_router_temp_conf_t *tmcf) { nxt_debug(task, "temp conf count:%D", tmcf->count); @@ -499,6 +520,8 @@ nxt_router_conf_error(nxt_task_t *task, nxt_router_temp_conf_t *tmcf) nxt_queue_link_t *qlk; nxt_socket_conf_t *skcf; + nxt_log(task, NXT_LOG_CRIT, "failed to apply new conf"); + for (qlk = nxt_queue_first(&tmcf->creating); qlk != nxt_queue_tail(&tmcf->creating); qlk = nxt_queue_next(qlk)) @@ -789,6 +812,8 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_queue_insert_tail(&tmcf->pending, &skcf->link); } + nxt_router_listen_sockets_sort(tmcf->conf->router, tmcf); + return NXT_OK; app_fail: @@ -906,55 +931,159 @@ nxt_router_listen_sockets_sort(nxt_router_t *router, } -static nxt_int_t -nxt_router_listen_sockets_stub_create(nxt_task_t *task, - nxt_router_temp_conf_t *tmcf) +static void +nxt_router_listen_socket_rpc_create(nxt_task_t *task, + nxt_router_temp_conf_t *tmcf, nxt_socket_conf_t *skcf) +{ + uint32_t stream; + 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; + + b = nxt_buf_mem_alloc(tmcf->mem_pool, skcf->sockaddr->sockaddr_size, 0); + if (b == NULL) { + goto fail; + } + + b->mem.free = nxt_cpymem(b->mem.free, skcf->sockaddr, + skcf->sockaddr->sockaddr_size); + + rt = task->thread->runtime; + main_port = rt->port_by_type[NXT_PROCESS_MASTER]; + 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 (stream == 0) { + goto fail; + } + + nxt_port_socket_write(task, main_port, NXT_PORT_MSG_SOCKET, -1, + stream, router_port->id, b); + + 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_queue_link_t *qlk, *nqlk; - nxt_socket_conf_t *skcf; + nxt_socket_rpc_t *rpc; nxt_router_socket_t *rtsk; - for (qlk = nxt_queue_first(&tmcf->pending); - qlk != nxt_queue_tail(&tmcf->pending); - qlk = nqlk) - { - skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link); + rpc = data; - s = nxt_listen_socket_create0(task, skcf->sockaddr, NXT_NONBLOCK); - if (nxt_slow_path(s == -1)) { - return NXT_ERROR; - } + s = msg->fd; - ret = nxt_listen_socket(task, s, NXT_LISTEN_BACKLOG); - if (nxt_slow_path(ret != NXT_OK)) { - goto fail; - } + ret = nxt_socket_nonblocking(task, s); + if (nxt_slow_path(ret != NXT_OK)) { + goto fail; + } - skcf->listen.socket = s; +#ifdef TCP_DEFER_ACCEPT - rtsk = nxt_malloc(sizeof(nxt_router_socket_t)); - if (nxt_slow_path(rtsk == NULL)) { - goto fail; - } + /* Defer Linux accept() up to for 1 second. */ + (void) nxt_socket_setsockopt(task, s, IPPROTO_TCP, TCP_DEFER_ACCEPT, 1); + +#endif - rtsk->count = 0; - rtsk->fd = skcf->listen.socket; - skcf->socket = rtsk; + ret = nxt_listen_socket(task, s, NXT_LISTEN_BACKLOG); + if (nxt_slow_path(ret != NXT_OK)) { + goto fail; + } - nqlk = nxt_queue_next(qlk); - nxt_queue_remove(qlk); - nxt_queue_insert_tail(&tmcf->creating, qlk); + rtsk = nxt_malloc(sizeof(nxt_router_socket_t)); + if (nxt_slow_path(rtsk == NULL)) { + goto fail; } - return NXT_OK; + rtsk->count = 0; + rtsk->fd = s; + + rpc->socket_conf->listen.socket = s; + rpc->socket_conf->socket = rtsk; + + 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); - return NXT_ERROR; + 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) +{ + u_char *p; + size_t size; + uint8_t error; + nxt_buf_t *in, *out; + nxt_sockaddr_t *sa; + nxt_socket_rpc_t *rpc; + nxt_router_temp_conf_t *tmcf; + + 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"), + }; + + rpc = data; + sa = rpc->socket_conf->sockaddr; + + in = msg->buf; + p = in->mem.pos; + + error = *p++; + + size = sizeof("listen socket error: ") - 1 + + sizeof("{listener: \"\", code:\"\", message: \"\"}") - 1 + + sa->length + socket_errors[error].length + (in->mem.free - p); + + tmcf = rpc->temp_conf; + + 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\"}", + 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); + + nxt_router_conf_error(task, tmcf); } diff --git a/src/nxt_worker_process.c b/src/nxt_worker_process.c index fd705f97..bd6674d5 100644 --- a/src/nxt_worker_process.c +++ b/src/nxt_worker_process.c @@ -31,6 +31,7 @@ nxt_port_handler_t nxt_controller_process_port_handlers[] = { nxt_port_remove_pid_handler, NULL, /* NXT_PORT_MSG_READY */ NULL, /* NXT_PORT_MSG_START_WORKER */ + NULL, /* NXT_PORT_MSG_SOCKET */ nxt_port_rpc_handler, nxt_port_rpc_handler, }; @@ -45,6 +46,7 @@ nxt_port_handler_t nxt_worker_process_port_handlers[] = { nxt_port_remove_pid_handler, NULL, /* NXT_PORT_MSG_READY */ NULL, /* NXT_PORT_MSG_START_WORKER */ + NULL, /* NXT_PORT_MSG_SOCKET */ nxt_port_rpc_handler, nxt_port_rpc_handler, }; @@ -59,6 +61,7 @@ nxt_port_handler_t nxt_app_process_port_handlers[] = { nxt_port_remove_pid_handler, NULL, /* NXT_PORT_MSG_READY */ NULL, /* NXT_PORT_MSG_START_WORKER */ + NULL, /* NXT_PORT_MSG_SOCKET */ nxt_port_rpc_handler, nxt_port_rpc_handler, }; @@ -73,6 +76,7 @@ nxt_port_handler_t nxt_router_process_port_handlers[] = { nxt_router_remove_pid_handler, NULL, /* NXT_PORT_MSG_READY */ NULL, /* NXT_PORT_MSG_START_WORKER */ + NULL, /* NXT_PORT_MSG_SOCKET */ nxt_port_rpc_handler, nxt_port_rpc_handler, }; |