technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
940d695f827eaffd19723429732c07e2dc7fb467
nginx-unit/src/nxt_main_process.c
Max Romanov 2bc9599057 Fixing zombie process appearance and hang up on shutdown. 
After the c8790d2a89bb commit, the SIGCHLD handler may return before processing
all awaiting PIDs.  To avoid zombie processes and ensure successful main
process termination, waitpid() must be called until an error is returned.

This closes #600 issue on GitHub.
2021-11-24 13:11:50 +03:00

1598 lines
40 KiB
C
Raw Blame History

/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#include <nxt_main.h>
#include <nxt_runtime.h>
#include <nxt_port.h>
#include <nxt_main_process.h>
#include <nxt_conf.h>
#include <nxt_router.h>
#include <nxt_port_queue.h>
#if (NXT_TLS)
#include <nxt_cert.h>
#endif
#include <sys/mount.h>
typedef struct {
nxt_socket_t socket;
nxt_socket_error_t error;
u_char *start;
u_char *end;
} nxt_listening_socket_t;
typedef struct {
nxt_uint_t size;
nxt_conf_map_t *map;
} nxt_conf_app_map_t;
static nxt_int_t nxt_main_process_port_create(nxt_task_t *task,
nxt_runtime_t *rt);
static void nxt_main_process_title(nxt_task_t *task);
static void nxt_main_process_sigterm_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_sigquit_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_sigusr1_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_signal_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_cleanup(nxt_task_t *task, nxt_process_t *process);
static void nxt_main_port_socket_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static nxt_int_t nxt_main_listening_socket(nxt_sockaddr_t *sa,
nxt_listening_socket_t *ls);
static void nxt_main_port_modules_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static int nxt_cdecl nxt_app_lang_compare(const void *v1, const void *v2);
static void nxt_main_process_whoami_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static void nxt_main_port_conf_store_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static nxt_int_t nxt_main_file_store(nxt_task_t *task, const char *tmp_name,
const char *name, u_char *buf, size_t size);
static void nxt_main_port_access_log_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
const nxt_sig_event_t nxt_main_process_signals[] = {
nxt_event_signal(SIGHUP, nxt_main_process_signal_handler),
nxt_event_signal(SIGINT, nxt_main_process_sigterm_handler),
nxt_event_signal(SIGQUIT, nxt_main_process_sigquit_handler),
nxt_event_signal(SIGTERM, nxt_main_process_sigterm_handler),
nxt_event_signal(SIGCHLD, nxt_main_process_sigchld_handler),
nxt_event_signal(SIGUSR1, nxt_main_process_sigusr1_handler),
nxt_event_signal_end,
};
nxt_uint_t nxt_conf_ver;
static nxt_bool_t nxt_exiting;
nxt_int_t
nxt_main_process_start(nxt_thread_t *thr, nxt_task_t *task,
nxt_runtime_t *rt)
{
rt->type = NXT_PROCESS_MAIN;
if (nxt_main_process_port_create(task, rt) != NXT_OK) {
return NXT_ERROR;
}
nxt_main_process_title(task);
/*
* The discovery process will send a message processed by
* nxt_main_port_modules_handler() which starts the controller
* and router processes.
*/
return nxt_process_init_start(task, nxt_discovery_process);
}
static nxt_conf_map_t nxt_common_app_conf[] = {
{
nxt_string("type"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, type),
},
{
nxt_string("user"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, user),
},
{
nxt_string("group"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, group),
},
{
nxt_string("working_directory"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, working_directory),
},
{
nxt_string("environment"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, environment),
},
{
nxt_string("isolation"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, isolation),
},
{
nxt_string("limits"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, limits),
},
};
static nxt_conf_map_t nxt_common_app_limits_conf[] = {
{
nxt_string("shm"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_common_app_conf_t, shm_limit),
},
{
nxt_string("requests"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, request_limit),
},
};
static nxt_conf_map_t nxt_external_app_conf[] = {
{
nxt_string("executable"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.external.executable),
},
{
nxt_string("arguments"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.external.arguments),
},
};
static nxt_conf_map_t nxt_python_app_conf[] = {
{
nxt_string("home"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.python.home),
},
{
nxt_string("path"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.python.path),
},
{
nxt_string("module"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.python.module),
},
{
nxt_string("callable"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.python.callable),
},
{
nxt_string("protocol"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.python.protocol),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.python.threads),
},
{
nxt_string("targets"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.python.targets),
},
{
nxt_string("thread_stack_size"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.python.thread_stack_size),
},
};
static nxt_conf_map_t nxt_php_app_conf[] = {
{
nxt_string("targets"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.php.targets),
},
{
nxt_string("options"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.php.options),
},
};
static nxt_conf_map_t nxt_perl_app_conf[] = {
{
nxt_string("script"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.perl.script),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.perl.threads),
},
{
nxt_string("thread_stack_size"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.perl.thread_stack_size),
},
};
static nxt_conf_map_t nxt_ruby_app_conf[] = {
{
nxt_string("script"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.ruby.script),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.ruby.threads),
},
{
nxt_string("hooks"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.ruby.hooks),
}
};
static nxt_conf_map_t nxt_java_app_conf[] = {
{
nxt_string("classpath"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.java.classpath),
},
{
nxt_string("webapp"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.java.webapp),
},
{
nxt_string("options"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.java.options),
},
{
nxt_string("unit_jars"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.java.unit_jars),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.java.threads),
},
{
nxt_string("thread_stack_size"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.java.thread_stack_size),
},
};
static nxt_conf_app_map_t nxt_app_maps[] = {
{ nxt_nitems(nxt_external_app_conf), nxt_external_app_conf },
{ nxt_nitems(nxt_python_app_conf), nxt_python_app_conf },
{ nxt_nitems(nxt_php_app_conf), nxt_php_app_conf },
{ nxt_nitems(nxt_perl_app_conf), nxt_perl_app_conf },
{ nxt_nitems(nxt_ruby_app_conf), nxt_ruby_app_conf },
{ nxt_nitems(nxt_java_app_conf), nxt_java_app_conf },
};
static void
nxt_main_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_debug(task, "main data: %*s",
nxt_buf_mem_used_size(&msg->buf->mem), msg->buf->mem.pos);
}
static void
nxt_main_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
void *mem;
nxt_port_t *port;
nxt_port_new_port_handler(task, msg);
port = msg->u.new_port;
if (port != NULL
&& port->type == NXT_PROCESS_APP
&& msg->fd[1] != -1)
{
mem = nxt_mem_mmap(NULL, sizeof(nxt_port_queue_t),
PROT_READ | PROT_WRITE, MAP_SHARED, msg->fd[1], 0);
if (nxt_fast_path(mem != MAP_FAILED)) {
port->queue = mem;
}
nxt_fd_close(msg->fd[1]);
msg->fd[1] = -1;
}
}
static void
nxt_main_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *start, *p, ch;
size_t type_len;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_app_type_t idx;
nxt_conf_value_t *conf;
nxt_process_init_t *init;
nxt_common_app_conf_t *app_conf;
rt = task->thread->runtime;
port = rt->port_by_type[NXT_PROCESS_ROUTER];
if (nxt_slow_path(port == NULL)) {
nxt_alert(task, "router port not found");
goto close_fds;
}
if (nxt_slow_path(port->pid != nxt_recv_msg_cmsg_pid(msg))) {
nxt_alert(task, "process %PI cannot start processes",
nxt_recv_msg_cmsg_pid(msg));
goto close_fds;
}
process = nxt_process_new(rt);
if (nxt_slow_path(process == NULL)) {
goto close_fds;
}
process->mem_pool = nxt_mp_create(1024, 128, 256, 32);
if (process->mem_pool == NULL) {
nxt_process_use(task, process, -1);
goto close_fds;
}
process->parent_port = rt->port_by_type[NXT_PROCESS_MAIN];
init = nxt_process_init(process);
*init = nxt_proto_process;
b = nxt_buf_chk_make_plain(process->mem_pool, msg->buf, msg->size);
if (b == NULL) {
goto failed;
}
nxt_debug(task, "main start prototype: %*s", b->mem.free - b->mem.pos,
b->mem.pos);
app_conf = nxt_mp_zalloc(process->mem_pool, sizeof(nxt_common_app_conf_t));
if (nxt_slow_path(app_conf == NULL)) {
goto failed;
}
app_conf->shared_port_fd = msg->fd[0];
app_conf->shared_queue_fd = msg->fd[1];
start = b->mem.pos;
app_conf->name.start = start;
app_conf->name.length = nxt_strlen(start);
init->name = (const char *) start;
process->name = nxt_mp_alloc(process->mem_pool, app_conf->name.length
+ sizeof("\"\" prototype") + 1);
if (nxt_slow_path(process->name == NULL)) {
goto failed;
}
p = (u_char *) process->name;
*p++ = '"';
p = nxt_cpymem(p, init->name, app_conf->name.length);
p = nxt_cpymem(p, "\" prototype", 11);
*p = '\0';
app_conf->shm_limit = 100 * 1024 * 1024;
app_conf->request_limit = 0;
start += app_conf->name.length + 1;
conf = nxt_conf_json_parse(process->mem_pool, start, b->mem.free, NULL);
if (conf == NULL) {
nxt_alert(task, "router app configuration parsing error");
goto failed;
}
rt = task->thread->runtime;
app_conf->user.start = (u_char*)rt->user_cred.user;
app_conf->user.length = nxt_strlen(rt->user_cred.user);
ret = nxt_conf_map_object(process->mem_pool, conf, nxt_common_app_conf,
nxt_nitems(nxt_common_app_conf), app_conf);
if (ret != NXT_OK) {
nxt_alert(task, "failed to map common app conf received from router");
goto failed;
}
for (type_len = 0; type_len != app_conf->type.length; type_len++) {
ch = app_conf->type.start[type_len];
if (ch == ' ' || nxt_isdigit(ch)) {
break;
}
}
idx = nxt_app_parse_type(app_conf->type.start, type_len);
if (nxt_slow_path(idx >= nxt_nitems(nxt_app_maps))) {
nxt_alert(task, "invalid app type %d received from router", (int) idx);
goto failed;
}
ret = nxt_conf_map_object(process->mem_pool, conf, nxt_app_maps[idx].map,
nxt_app_maps[idx].size, app_conf);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_alert(task, "failed to map app conf received from router");
goto failed;
}
if (app_conf->limits != NULL) {
ret = nxt_conf_map_object(process->mem_pool, app_conf->limits,
nxt_common_app_limits_conf,
nxt_nitems(nxt_common_app_limits_conf),
app_conf);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_alert(task, "failed to map app limits received from router");
goto failed;
}
}
app_conf->self = conf;
process->stream = msg->port_msg.stream;
process->data.app = app_conf;
ret = nxt_process_start(task, process);
if (nxt_fast_path(ret == NXT_OK || ret == NXT_AGAIN)) {
/* Close shared port fds only in main process. */
if (ret == NXT_OK) {
nxt_fd_close(app_conf->shared_port_fd);
nxt_fd_close(app_conf->shared_queue_fd);
}
/* Avoid fds close in caller. */
msg->fd[0] = -1;
msg->fd[1] = -1;
return;
}
failed:
nxt_process_use(task, process, -1);
port = nxt_runtime_port_find(rt, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_fast_path(port != NULL)) {
nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_ERROR,
-1, msg->port_msg.stream, 0, NULL);
}
close_fds:
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
nxt_fd_close(msg->fd[1]);
msg->fd[1] = -1;
}
static void
nxt_main_process_created_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_port_t *port;
nxt_process_t *process;
nxt_runtime_t *rt;
rt = task->thread->runtime;
port = nxt_runtime_port_find(rt, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_slow_path(port == NULL)) {
return;
}
process = port->process;
nxt_assert(process != NULL);
nxt_assert(process->state == NXT_PROCESS_STATE_CREATING);
#if (NXT_HAVE_CLONE && NXT_HAVE_CLONE_NEWUSER)
if (nxt_is_clone_flag_set(process->isolation.clone.flags, NEWUSER)) {
if (nxt_slow_path(nxt_clone_credential_map(task, process->pid,
process->user_cred,
&process->isolation.clone)
!= NXT_OK))
{
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_ERROR,
-1, msg->port_msg.stream, 0, NULL);
return;
}
}
#endif
process->state = NXT_PROCESS_STATE_CREATED;
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_READY_LAST,
-1, msg->port_msg.stream, 0, NULL);
}
static nxt_port_handlers_t nxt_main_process_port_handlers = {
.data = nxt_main_data_handler,
.new_port = nxt_main_new_port_handler,
.process_created = nxt_main_process_created_handler,
.process_ready = nxt_port_process_ready_handler,
.whoami = nxt_main_process_whoami_handler,
.remove_pid = nxt_port_remove_pid_handler,
.start_process = nxt_main_start_process_handler,
.socket = nxt_main_port_socket_handler,
.modules = nxt_main_port_modules_handler,
.conf_store = nxt_main_port_conf_store_handler,
#if (NXT_TLS)
.cert_get = nxt_cert_store_get_handler,
.cert_delete = nxt_cert_store_delete_handler,
#endif
.access_log = nxt_main_port_access_log_handler,
.rpc_ready = nxt_port_rpc_handler,
.rpc_error = nxt_port_rpc_handler,
};
static void
nxt_main_process_whoami_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_buf_t *buf;
nxt_pid_t pid, ppid;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *pprocess;
nxt_assert(msg->port_msg.reply_port == 0);
if (nxt_slow_path(msg->buf == NULL
|| nxt_buf_used_size(msg->buf) != sizeof(nxt_pid_t)))
{
nxt_alert(task, "whoami: buffer is NULL or unexpected size");
goto fail;
}
nxt_memcpy(&ppid, msg->buf->mem.pos, sizeof(nxt_pid_t));
rt = task->thread->runtime;
pprocess = nxt_runtime_process_find(rt, ppid);
if (nxt_slow_path(pprocess == NULL)) {
nxt_alert(task, "whoami: parent process %PI not found", ppid);
goto fail;
}
pid = nxt_recv_msg_cmsg_pid(msg);
nxt_debug(task, "whoami: from %PI, parent %PI, fd %d", pid, ppid,
msg->fd[0]);
if (msg->fd[0] != -1) {
port = nxt_runtime_process_port_create(task, rt, pid, 0,
NXT_PROCESS_APP);
if (nxt_slow_path(port == NULL)) {
goto fail;
}
nxt_fd_nonblocking(task, msg->fd[0]);
port->pair[0] = -1;
port->pair[1] = msg->fd[0];
msg->fd[0] = -1;
port->max_size = 16 * 1024;
port->max_share = 64 * 1024;
port->socket.task = task;
nxt_port_write_enable(task, port);
} else {
port = nxt_runtime_port_find(rt, pid, 0);
if (nxt_slow_path(port == NULL)) {
goto fail;
}
}
if (ppid != nxt_pid) {
nxt_queue_insert_tail(&pprocess->children, &port->process->link);
}
buf = nxt_buf_mem_alloc(task->thread->engine->mem_pool,
sizeof(nxt_pid_t), 0);
if (nxt_slow_path(buf == NULL)) {
goto fail;
}
buf->mem.free = nxt_cpymem(buf->mem.free, &pid, sizeof(nxt_pid_t));
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_READY_LAST, -1,
msg->port_msg.stream, 0, buf);
fail:
if (msg->fd[0] != -1) {
nxt_fd_close(msg->fd[0]);
}
}
static nxt_int_t
nxt_main_process_port_create(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_int_t ret;
nxt_port_t *port;
nxt_process_t *process;
port = nxt_runtime_process_port_create(task, rt, nxt_pid, 0,
NXT_PROCESS_MAIN);
if (nxt_slow_path(port == NULL)) {
return NXT_ERROR;
}
process = port->process;
ret = nxt_port_socket_init(task, port, 0);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_use(task, port, -1);
return ret;
}
/*
* A main process port. A write port is not closed
* since it should be inherited by processes.
*/
nxt_port_enable(task, port, &nxt_main_process_port_handlers);
process->state = NXT_PROCESS_STATE_READY;
return NXT_OK;
}
static void
nxt_main_process_title(nxt_task_t *task)
{
u_char *p, *end;
nxt_uint_t i;
u_char title[2048];
end = title + sizeof(title) - 1;
p = nxt_sprintf(title, end, "unit: main v" NXT_VERSION " [%s",
nxt_process_argv[0]);
for (i = 1; nxt_process_argv[i] != NULL; i++) {
p = nxt_sprintf(p, end, " %s", nxt_process_argv[i]);
}
if (p < end) {
*p++ = ']';
}
*p = '\0';
nxt_process_title(task, "%s", title);
}
static void
nxt_main_process_sigterm_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_debug(task, "sigterm handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
/* TODO: fast exit. */
nxt_exiting = 1;
nxt_runtime_quit(task, 0);
}
static void
nxt_main_process_sigquit_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_debug(task, "sigquit handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
/* TODO: graceful exit. */
nxt_exiting = 1;
nxt_runtime_quit(task, 0);
}
static void
nxt_main_process_sigusr1_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_mp_t *mp;
nxt_int_t ret;
nxt_uint_t n;
nxt_port_t *port;
nxt_file_t *file, *new_file;
nxt_array_t *new_files;
nxt_runtime_t *rt;
nxt_log(task, NXT_LOG_NOTICE, "signal %d (%s) recevied, %s",
(int) (uintptr_t) obj, data, "log files rotation");
rt = task->thread->runtime;
port = rt->port_by_type[NXT_PROCESS_ROUTER];
if (nxt_fast_path(port != NULL)) {
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_ACCESS_LOG,
-1, 0, 0, NULL);
}
mp = nxt_mp_create(1024, 128, 256, 32);
if (mp == NULL) {
return;
}
n = nxt_list_nelts(rt->log_files);
new_files = nxt_array_create(mp, n, sizeof(nxt_file_t));
if (new_files == NULL) {
nxt_mp_destroy(mp);
return;
}
nxt_list_each(file, rt->log_files) {
/* This allocation cannot fail. */
new_file = nxt_array_add(new_files);
new_file->name = file->name;
new_file->fd = NXT_FILE_INVALID;
new_file->log_level = NXT_LOG_ALERT;
ret = nxt_file_open(task, new_file, O_WRONLY | O_APPEND, O_CREAT,
NXT_FILE_OWNER_ACCESS);
if (ret != NXT_OK) {
goto fail;
}
} nxt_list_loop;
new_file = new_files->elts;
ret = nxt_file_stderr(&new_file[0]);
if (ret == NXT_OK) {
n = 0;
nxt_list_each(file, rt->log_files) {
nxt_port_change_log_file(task, rt, n, new_file[n].fd);
/*
* The old log file descriptor must be closed at the moment
* when no other threads use it. dup2() allows to use the
* old file descriptor for new log file. This change is
* performed atomically in the kernel.
*/
(void) nxt_file_redirect(file, new_file[n].fd);
n++;
} nxt_list_loop;
nxt_mp_destroy(mp);
return;
}
fail:
new_file = new_files->elts;
n = new_files->nelts;
while (n != 0) {
if (new_file->fd != NXT_FILE_INVALID) {
nxt_file_close(task, new_file);
}
new_file++;
n--;
}
nxt_mp_destroy(mp);
}
static void
nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
{
int status;
nxt_int_t ret;
nxt_err_t err;
nxt_pid_t pid;
nxt_port_t *port;
nxt_queue_t children;
nxt_runtime_t *rt;
nxt_process_t *process, *child;
nxt_process_init_t init;
nxt_debug(task, "sigchld handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
rt = task->thread->runtime;
for ( ;; ) {
pid = waitpid(-1, &status, WNOHANG);
if (pid == -1) {
switch (err = nxt_errno) {
case NXT_ECHILD:
return;
case NXT_EINTR:
continue;
default:
nxt_alert(task, "waitpid() failed: %E", err);
return;
}
}
nxt_debug(task, "waitpid(): %PI", pid);
if (pid == 0) {
return;
}
if (WTERMSIG(status)) {
#ifdef WCOREDUMP
nxt_alert(task, "process %PI exited on signal %d%s",
pid, WTERMSIG(status),
WCOREDUMP(status) ? " (core dumped)" : "");
#else
nxt_alert(task, "process %PI exited on signal %d",
pid, WTERMSIG(status));
#endif
} else {
nxt_trace(task, "process %PI exited with code %d",
pid, WEXITSTATUS(status));
}
process = nxt_runtime_process_find(rt, pid);
if (process != NULL) {
nxt_main_process_cleanup(task, process);
if (process->state == NXT_PROCESS_STATE_READY) {
process->stream = 0;
}
nxt_queue_init(&children);
if (!nxt_queue_is_empty(&process->children)) {
nxt_queue_add(&children, &process->children);
nxt_queue_init(&process->children);
nxt_queue_each(child, &children, nxt_process_t, link) {
port = nxt_process_port_first(child);
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
} nxt_queue_loop;
}
if (nxt_exiting) {
nxt_process_close_ports(task, process);
nxt_queue_each(child, &children, nxt_process_t, link) {
nxt_queue_remove(&child->link);
child->link.next = NULL;
nxt_process_close_ports(task, child);
} nxt_queue_loop;
if (rt->nprocesses <= 1) {
nxt_runtime_quit(task, 0);
return;
}
continue;
}
nxt_port_remove_notify_others(task, process);
nxt_queue_each(child, &children, nxt_process_t, link) {
nxt_port_remove_notify_others(task, child);
nxt_queue_remove(&child->link);
child->link.next = NULL;
nxt_process_close_ports(task, child);
} nxt_queue_loop;
init = *(nxt_process_init_t *) nxt_process_init(process);
nxt_process_close_ports(task, process);
if (init.restart) {
ret = nxt_process_init_start(task, init);
if (nxt_slow_path(ret == NXT_ERROR)) {
nxt_alert(task, "failed to restart %s", init.name);
}
}
}
}
}
static void
nxt_main_process_signal_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_trace(task, "signal signo:%d (%s) recevied, ignored",
(int) (uintptr_t) obj, data);
}
static void
nxt_main_process_cleanup(nxt_task_t *task, nxt_process_t *process)
{
if (process->isolation.cleanup != NULL) {
process->isolation.cleanup(task, process);
}
}
static void
nxt_main_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];
port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_slow_path(port == NULL)) {
return;
}
if (nxt_slow_path(port->type != NXT_PROCESS_ROUTER)) {
nxt_alert(task, "process %PI cannot create listener sockets",
msg->port_msg.pid);
return;
}
b = msg->buf;
sa = (nxt_sockaddr_t *) b->mem.pos;
/* TODO check b size and make plain */
ls.socket = -1;
ls.error = NXT_SOCKET_ERROR_SYSTEM;
ls.start = message;
ls.end = message + sizeof(message);
nxt_debug(task, "listening socket \"%*s\"",
(size_t) sa->length, nxt_sockaddr_start(sa));
ret = nxt_main_listening_socket(sa, &ls);
if (ret == NXT_OK) {
nxt_debug(task, "socket(\"%*s\"): %d",
(size_t) sa->length, nxt_sockaddr_start(sa), ls.socket);
out = NULL;
type = NXT_PORT_MSG_RPC_READY_LAST | NXT_PORT_MSG_CLOSE_FD;
} else {
size = ls.end - ls.start;
nxt_alert(task, "%*s", size, ls.start);
out = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
size + 1);
if (nxt_fast_path(out != NULL)) {
*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_main_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);
static const 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",
(size_t) 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",
(size_t) 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",
(size_t) 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;
}
} else
#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",
(size_t) sa->length, nxt_sockaddr_start(sa), err);
goto fail;
}
#if (NXT_HAVE_UNIX_DOMAIN)
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;
}
static nxt_conf_map_t nxt_app_lang_module_map[] = {
{
nxt_string("type"),
NXT_CONF_MAP_INT,
offsetof(nxt_app_lang_module_t, type),
},
{
nxt_string("version"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_app_lang_module_t, version),
},
{
nxt_string("file"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_app_lang_module_t, file),
},
};
static nxt_conf_map_t nxt_app_lang_mounts_map[] = {
{
nxt_string("src"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, src),
},
{
nxt_string("dst"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, dst),
},
{
nxt_string("name"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, name),
},
{
nxt_string("type"),
NXT_CONF_MAP_INT,
offsetof(nxt_fs_mount_t, type),
},
{
nxt_string("flags"),
NXT_CONF_MAP_INT,
offsetof(nxt_fs_mount_t, flags),
},
{
nxt_string("data"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, data),
},
};
static void
nxt_main_port_modules_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
uint32_t index, jindex, nmounts;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_fs_mount_t *mnt;
nxt_conf_value_t *conf, *root, *value, *mounts;
nxt_app_lang_module_t *lang;
static nxt_str_t root_path = nxt_string("/");
static nxt_str_t mounts_name = nxt_string("mounts");
rt = task->thread->runtime;
if (msg->port_msg.pid != rt->port_by_type[NXT_PROCESS_DISCOVERY]->pid) {
nxt_alert(task, "process %PI cannot send modules", msg->port_msg.pid);
return;
}
if (nxt_exiting) {
nxt_debug(task, "ignoring discovered modules, exiting");
return;
}
port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_fast_path(port != NULL)) {
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_ERROR, -1,
msg->port_msg.stream, 0, NULL);
}
b = msg->buf;
if (b == NULL) {
return;
}
mp = nxt_mp_create(1024, 128, 256, 32);
if (mp == NULL) {
return;
}
b = nxt_buf_chk_make_plain(mp, b, msg->size);
if (b == NULL) {
return;
}
nxt_debug(task, "application languages: \"%*s\"",
b->mem.free - b->mem.pos, b->mem.pos);
conf = nxt_conf_json_parse(mp, b->mem.pos, b->mem.free, NULL);
if (conf == NULL) {
goto fail;
}
root = nxt_conf_get_path(conf, &root_path);
if (root == NULL) {
goto fail;
}
for (index = 0; /* void */ ; index++) {
value = nxt_conf_get_array_element(root, index);
if (value == NULL) {
break;
}
lang = nxt_array_zero_add(rt->languages);
if (lang == NULL) {
goto fail;
}
lang->module = NULL;
ret = nxt_conf_map_object(rt->mem_pool, value, nxt_app_lang_module_map,
nxt_nitems(nxt_app_lang_module_map), lang);
if (ret != NXT_OK) {
goto fail;
}
mounts = nxt_conf_get_object_member(value, &mounts_name, NULL);
if (mounts == NULL) {
nxt_alert(task, "missing mounts from discovery message.");
goto fail;
}
if (nxt_conf_type(mounts) != NXT_CONF_ARRAY) {
nxt_alert(task, "invalid mounts type from discovery message.");
goto fail;
}
nmounts = nxt_conf_array_elements_count(mounts);
lang->mounts = nxt_array_create(rt->mem_pool, nmounts,
sizeof(nxt_fs_mount_t));
if (lang->mounts == NULL) {
goto fail;
}
for (jindex = 0; /* */; jindex++) {
value = nxt_conf_get_array_element(mounts, jindex);
if (value == NULL) {
break;
}
mnt = nxt_array_zero_add(lang->mounts);
if (mnt == NULL) {
goto fail;
}
mnt->builtin = 1;
mnt->deps = 1;
ret = nxt_conf_map_object(rt->mem_pool, value,
nxt_app_lang_mounts_map,
nxt_nitems(nxt_app_lang_mounts_map), mnt);
if (ret != NXT_OK) {
goto fail;
}
}
nxt_debug(task, "lang %d %s \"%s\" (%d mounts)",
lang->type, lang->version, lang->file, lang->mounts->nelts);
}
qsort(rt->languages->elts, rt->languages->nelts,
sizeof(nxt_app_lang_module_t), nxt_app_lang_compare);
fail:
nxt_mp_destroy(mp);
ret = nxt_process_init_start(task, nxt_controller_process);
if (ret == NXT_OK) {
ret = nxt_process_init_start(task, nxt_router_process);
}
if (nxt_slow_path(ret == NXT_ERROR)) {
nxt_exiting = 1;
nxt_runtime_quit(task, 1);
}
}
static int nxt_cdecl
nxt_app_lang_compare(const void *v1, const void *v2)
{
int n;
const nxt_app_lang_module_t *lang1, *lang2;
lang1 = v1;
lang2 = v2;
n = lang1->type - lang2->type;
if (n != 0) {
return n;
}
n = nxt_strverscmp(lang1->version, lang2->version);
/* Negate result to move higher versions to the beginning. */
return -n;
}
static void
nxt_main_port_conf_store_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
void *p;
size_t n, size;
nxt_int_t ret;
nxt_port_t *ctl_port;
nxt_runtime_t *rt;
u_char ver[NXT_INT_T_LEN];
rt = task->thread->runtime;
ctl_port = rt->port_by_type[NXT_PROCESS_CONTROLLER];
if (nxt_slow_path(msg->port_msg.pid != ctl_port->pid)) {
nxt_alert(task, "process %PI cannot store conf", msg->port_msg.pid);
return;
}
p = MAP_FAILED;
/*
* Ancient compilers like gcc 4.8.5 on CentOS 7 wants 'size' to be
* initialized in 'cleanup' section.
*/
size = 0;
if (nxt_slow_path(msg->fd[0] == -1)) {
nxt_alert(task, "conf_store_handler: invalid shm fd");
goto error;
}
if (nxt_buf_mem_used_size(&msg->buf->mem) != sizeof(size_t)) {
nxt_alert(task, "conf_store_handler: unexpected buffer size (%d)",
(int) nxt_buf_mem_used_size(&msg->buf->mem));
goto error;
}
nxt_memcpy(&size, msg->buf->mem.pos, sizeof(size_t));
p = nxt_mem_mmap(NULL, size, PROT_READ, MAP_SHARED, msg->fd[0], 0);
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
if (nxt_slow_path(p == MAP_FAILED)) {
goto error;
}
nxt_debug(task, "conf_store_handler(%uz): %*s", size, size, p);
if (nxt_conf_ver != NXT_VERNUM) {
n = nxt_sprintf(ver, ver + NXT_INT_T_LEN, "%d", NXT_VERNUM) - ver;
ret = nxt_main_file_store(task, rt->ver_tmp, rt->ver, ver, n);
if (nxt_slow_path(ret != NXT_OK)) {
goto error;
}
nxt_conf_ver = NXT_VERNUM;
}
ret = nxt_main_file_store(task, rt->conf_tmp, rt->conf, p, size);
if (nxt_fast_path(ret == NXT_OK)) {
goto cleanup;
}
error:
nxt_alert(task, "failed to store current configuration");
cleanup:
if (p != MAP_FAILED) {
nxt_mem_munmap(p, size);
}
if (msg->fd[0] != -1) {
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
}
}
static nxt_int_t
nxt_main_file_store(nxt_task_t *task, const char *tmp_name, const char *name,
u_char *buf, size_t size)
{
ssize_t n;
nxt_int_t ret;
nxt_file_t file;
nxt_memzero(&file, sizeof(nxt_file_t));
file.name = (nxt_file_name_t *) name;
ret = nxt_file_open(task, &file, NXT_FILE_WRONLY, NXT_FILE_TRUNCATE,
NXT_FILE_OWNER_ACCESS);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
n = nxt_file_write(&file, buf, size, 0);
nxt_file_close(task, &file);
if (nxt_slow_path(n != (ssize_t) size)) {
(void) nxt_file_delete(file.name);
return NXT_ERROR;
}
return nxt_file_rename(file.name, (nxt_file_name_t *) name);
}
static void
nxt_main_port_access_log_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *path;
nxt_int_t ret;
nxt_file_t file;
nxt_port_t *port;
nxt_port_msg_type_t type;
nxt_debug(task, "opening access log file");
path = msg->buf->mem.pos;
nxt_memzero(&file, sizeof(nxt_file_t));
file.name = (nxt_file_name_t *) path;
file.log_level = NXT_LOG_ERR;
ret = nxt_file_open(task, &file, O_WRONLY | O_APPEND, O_CREAT,
NXT_FILE_OWNER_ACCESS);
type = (ret == NXT_OK) ? NXT_PORT_MSG_RPC_READY_LAST | NXT_PORT_MSG_CLOSE_FD
: NXT_PORT_MSG_RPC_ERROR;
port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_fast_path(port != NULL)) {
(void) nxt_port_socket_write(task, port, type, file.fd,
msg->port_msg.stream, 0, NULL);
}
}
Reference in New Issue View Git Blame Copy Permalink