technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Listening socket creation and binding operations are passed

Browse Source 
to the master process.
This commit is contained in:
Igor Sysoev
2017-08-02 21:11:20 +03:00
parent b150f07e80
commit a94178a05e
5 changed files with 410 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
src/nxt_master_process.c
View File

@@ -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;
}

11
src/nxt_master_process.h
View File

@@ -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);

2
src/nxt_port.h
View File

@@ -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,

263
src/nxt_router.c
View File

@@ -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;
if (nxt_fast_path(ret == NXT_OK)) {
nxt_router_conf_apply(task, tmcf, NULL);
} else {
nxt_router_conf_error(task, tmcf);
}
nxt_log(task, NXT_LOG_CRIT, "failed to apply new conf");
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;
}
skcf->listen.socket = s;
rtsk = nxt_malloc(sizeof(nxt_router_socket_t));
if (nxt_slow_path(rtsk == NULL)) {
goto fail;
}
rtsk->count = 0;
rtsk->fd = skcf->listen.socket;
skcf->socket = rtsk;
nqlk = nxt_queue_next(qlk);
nxt_queue_remove(qlk);
nxt_queue_insert_tail(&tmcf->creating, qlk);
ret = nxt_socket_nonblocking(task, s);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
return NXT_OK;
#ifdef TCP_DEFER_ACCEPT
/* Defer Linux accept() up to for 1 second. */
(void) nxt_socket_setsockopt(task, s, IPPROTO_TCP, TCP_DEFER_ACCEPT, 1);
#endif
ret = nxt_listen_socket(task, s, NXT_LISTEN_BACKLOG);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
rtsk = nxt_malloc(sizeof(nxt_router_socket_t));
if (nxt_slow_path(rtsk == NULL)) {
goto fail;
}
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);
}

4
src/nxt_worker_process.c
View File

@@ -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,
};