technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
ed2492a66afdf578d1e4f99dc098ab685607b3ba
nginx-unit/src/nxt_main_process.c
Tiago Natel ed2492a66a Moved credential-related code to nxt_credential.c. 
This is required to avoid include cycles, as some nxt_clone_* functions
depend on the credential structures, but nxt_process depends on clone
structures.
2019-12-06 13:28:05 +00:00

1711 lines
42 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>
#if (NXT_TLS)
#include <nxt_cert.h>
#endif
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;
extern nxt_port_handlers_t nxt_controller_process_port_handlers;
extern nxt_port_handlers_t nxt_router_process_port_handlers;
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 nxt_int_t nxt_main_start_controller_process(nxt_task_t *task,
nxt_runtime_t *rt);
static nxt_int_t nxt_main_create_controller_process(nxt_task_t *task,
nxt_runtime_t *rt, nxt_process_init_t *init);
static nxt_int_t nxt_main_create_router_process(nxt_task_t *task, nxt_runtime_t *rt,
nxt_process_init_t *init);
static nxt_int_t nxt_main_start_router_process(nxt_task_t *task,
nxt_runtime_t *rt);
static nxt_int_t nxt_main_start_discovery_process(nxt_task_t *task,
nxt_runtime_t *rt);
static nxt_int_t nxt_main_start_worker_process(nxt_task_t *task,
nxt_runtime_t *rt, nxt_common_app_conf_t *app_conf, uint32_t stream);
static nxt_int_t nxt_main_create_worker_process(nxt_task_t *task,
nxt_runtime_t *rt, nxt_process_init_t *init);
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_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid);
static void nxt_main_stop_worker_processes(nxt_task_t *task, nxt_runtime_t *rt);
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_port_conf_store_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static void nxt_main_port_access_log_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static nxt_process_init_t *nxt_process_init_create(nxt_task_t *task,
nxt_process_type_t type, const nxt_str_t *name);
static nxt_int_t nxt_process_init_name_set(nxt_process_init_t *init,
nxt_process_type_t type, const nxt_str_t *name);
static nxt_int_t nxt_process_init_creds_set(nxt_task_t *task,
nxt_process_init_t *init, nxt_str_t *user, nxt_str_t *group);
static nxt_int_t nxt_init_set_isolation(nxt_task_t *task,
nxt_process_init_t *init, nxt_conf_value_t *isolation);
static nxt_int_t nxt_init_set_ns(nxt_task_t *task, nxt_process_init_t *init,
nxt_conf_value_t *ns);
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,
};
static const nxt_port_handlers_t nxt_app_process_port_handlers = {
.new_port = nxt_port_new_port_handler,
.change_file = nxt_port_change_log_file_handler,
.mmap = nxt_port_mmap_handler,
.remove_pid = nxt_port_remove_pid_handler,
};
static const nxt_port_handlers_t nxt_discovery_process_port_handlers = {
.quit = nxt_worker_process_quit_handler,
.new_port = nxt_port_new_port_handler,
.change_file = nxt_port_change_log_file_handler,
.mmap = nxt_port_mmap_handler,
.data = nxt_port_data_handler,
.remove_pid = nxt_port_remove_pid_handler,
.rpc_ready = nxt_port_rpc_handler,
.rpc_error = nxt_port_rpc_handler,
};
static const nxt_port_handlers_t *nxt_process_port_handlers[NXT_PROCESS_MAX] =
{
NULL,
&nxt_discovery_process_port_handlers,
&nxt_controller_process_port_handlers,
&nxt_router_process_port_handlers,
&nxt_app_process_port_handlers
};
static const nxt_process_start_t nxt_process_starts[NXT_PROCESS_MAX] = {
NULL,
nxt_discovery_start,
nxt_controller_start,
nxt_router_start,
nxt_app_start
};
static const nxt_process_restart_t nxt_process_restarts[NXT_PROCESS_MAX] = {
NULL,
NULL,
&nxt_main_create_controller_process,
&nxt_main_create_router_process,
NULL
};
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 dicsovery process will send a message processed by
* nxt_main_port_modules_handler() which starts the controller
* and router processes.
*/
return nxt_main_start_discovery_process(task, rt);
}
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),
}
};
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_STR,
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),
},
};
static nxt_conf_map_t nxt_php_app_conf[] = {
{
nxt_string("root"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.php.root),
},
{
nxt_string("script"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.php.script),
},
{
nxt_string("index"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.php.index),
},
{
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),
},
};
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),
},
};
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),
},
};
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_port_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_port_main_start_worker_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *start, ch;
size_t type_len;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_app_type_t idx;
nxt_conf_value_t *conf;
nxt_common_app_conf_t app_conf;
ret = NXT_ERROR;
mp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(mp == NULL)) {
return;
}
b = nxt_buf_chk_make_plain(mp, msg->buf, msg->size);
if (b == NULL) {
return;
}
nxt_debug(task, "main start worker: %*s", b->mem.free - b->mem.pos,
b->mem.pos);
nxt_memzero(&app_conf, sizeof(nxt_common_app_conf_t));
start = b->mem.pos;
app_conf.name.start = start;
app_conf.name.length = nxt_strlen(start);
start += app_conf.name.length + 1;
conf = nxt_conf_json_parse(mp, 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(mp, 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(mp, 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;
}
ret = nxt_main_start_worker_process(task, task->thread->runtime,
&app_conf, msg->port_msg.stream);
failed:
if (ret == NXT_ERROR) {
port = nxt_runtime_port_find(task->thread->runtime, 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);
}
}
nxt_mp_destroy(mp);
}
static nxt_port_handlers_t nxt_main_process_port_handlers = {
.data = nxt_port_main_data_handler,
.process_ready = nxt_port_process_ready_handler,
.start_worker = nxt_port_main_start_worker_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 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)) {
return ret;
}
/*
* A main process port. A write port is not closed
* since it should be inherited by worker processes.
*/
nxt_port_enable(task, port, &nxt_main_process_port_handlers);
process->ready = 1;
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 nxt_int_t
nxt_main_start_controller_process(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_process_init_t *init;
static const nxt_str_t name = nxt_string("controller");
init = nxt_process_init_create(task, NXT_PROCESS_CONTROLLER, &name);
if (nxt_slow_path(init == NULL)) {
return NXT_ERROR;
}
return nxt_main_create_controller_process(task, rt, init);;
}
static nxt_int_t
nxt_main_create_controller_process(nxt_task_t *task, nxt_runtime_t *rt,
nxt_process_init_t *init)
{
ssize_t n;
nxt_int_t ret;
nxt_str_t *conf;
nxt_file_t file;
nxt_file_info_t fi;
nxt_controller_init_t ctrl_init;
nxt_memzero(&ctrl_init, sizeof(nxt_controller_init_t));
conf = &ctrl_init.conf;
nxt_memzero(&file, sizeof(nxt_file_t));
file.name = (nxt_file_name_t *) rt->conf;
ret = nxt_file_open(task, &file, NXT_FILE_RDONLY, NXT_FILE_OPEN, 0);
if (ret == NXT_OK) {
ret = nxt_file_info(&file, &fi);
if (nxt_fast_path(ret == NXT_OK && nxt_is_file(&fi))) {
conf->length = nxt_file_size(&fi);
conf->start = nxt_malloc(conf->length);
if (nxt_slow_path(conf->start == NULL)) {
nxt_file_close(task, &file);
return NXT_ERROR;
}
n = nxt_file_read(&file, conf->start, conf->length, 0);
if (nxt_slow_path(n != (ssize_t) conf->length)) {
nxt_free(conf->start);
conf->start = NULL;
nxt_alert(task, "failed to restore previous configuration: "
"cannot read the file");
}
}
nxt_file_close(task, &file);
}
#if (NXT_TLS)
ctrl_init.certs = nxt_cert_store_load(task);
#endif
init->data = &ctrl_init;
ret = nxt_main_create_worker_process(task, rt, init);
if (ret == NXT_OK) {
if (conf->start != NULL) {
nxt_free(conf->start);
}
#if (NXT_TLS)
if (ctrl_init.certs != NULL) {
nxt_cert_store_release(ctrl_init.certs);
}
#endif
}
return ret;
}
static nxt_int_t
nxt_main_start_discovery_process(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_process_init_t *init;
static const nxt_str_t name = nxt_string("discovery");
init = nxt_process_init_create(task, NXT_PROCESS_DISCOVERY, &name);
if (nxt_slow_path(init == NULL)) {
return NXT_ERROR;
}
return nxt_main_create_worker_process(task, rt, init);
}
static nxt_int_t
nxt_main_start_router_process(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_process_init_t *init;
static const nxt_str_t name = nxt_string("router");
init = nxt_process_init_create(task, NXT_PROCESS_ROUTER, &name);
if (nxt_slow_path(init == NULL)) {
return NXT_ERROR;
}
return nxt_main_create_router_process(task, rt, init);
}
static nxt_int_t
nxt_main_create_router_process(nxt_task_t *task, nxt_runtime_t *rt,
nxt_process_init_t *init)
{
nxt_main_stop_worker_processes(task, rt);
return nxt_main_create_worker_process(task, rt, init);
}
static nxt_int_t
nxt_main_start_worker_process(nxt_task_t *task, nxt_runtime_t *rt,
nxt_common_app_conf_t *app_conf, uint32_t stream)
{
nxt_int_t ret;
nxt_process_init_t *init;
init = nxt_process_init_create(task, NXT_PROCESS_WORKER, &app_conf->name);
if (nxt_slow_path(init == NULL)) {
return NXT_ERROR;
}
if (rt->capabilities.setid) {
ret = nxt_process_init_creds_set(task, init, &app_conf->user,
&app_conf->group);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
} else {
if (!nxt_str_eq(&app_conf->user, (u_char *) rt->user_cred.user,
nxt_strlen(rt->user_cred.user)))
{
nxt_alert(task, "cannot set user \"%V\" for app \"%V\": "
"missing capabilities", &app_conf->user, &app_conf->name);
goto fail;
}
if (app_conf->group.length > 0
&& !nxt_str_eq(&app_conf->group, (u_char *) rt->group,
nxt_strlen(rt->group)))
{
nxt_alert(task, "cannot set group \"%V\" for app \"%V\": "
"missing capabilities", &app_conf->group,
&app_conf->name);
goto fail;
}
}
init->data = app_conf;
init->stream = stream;
ret = nxt_init_set_isolation(task, init, app_conf->isolation);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
return nxt_main_create_worker_process(task, rt, init);
fail:
nxt_mp_destroy(init->mem_pool);
return NXT_ERROR;
}
nxt_int_t
nxt_main_create_worker_process(nxt_task_t *task, nxt_runtime_t *rt,
nxt_process_init_t *init)
{
nxt_int_t ret;
nxt_pid_t pid;
nxt_port_t *port;
nxt_process_t *process;
/*
* TODO: remove process, init, ports from array on memory and fork failures.
*/
process = nxt_runtime_process_new(rt);
if (nxt_slow_path(process == NULL)) {
nxt_mp_destroy(init->mem_pool);
return NXT_ERROR;
}
process->init = init;
port = nxt_port_new(task, 0, 0, init->type);
if (nxt_slow_path(port == NULL)) {
nxt_process_use(task, process, -1);
return NXT_ERROR;
}
nxt_process_port_add(task, process, port);
nxt_process_use(task, process, -1);
ret = nxt_port_socket_init(task, port, 0);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_use(task, port, -1);
return ret;
}
pid = nxt_process_create(task, process);
switch (pid) {
case -1:
nxt_port_close(task, port);
nxt_port_use(task, port, -1);
return NXT_ERROR;
case 0:
/* A worker process, return to the event engine work queue loop. */
nxt_port_use(task, port, -1);
return NXT_AGAIN;
default:
/* The main process created a new process. */
nxt_port_read_close(port);
nxt_port_write_enable(task, port);
nxt_port_use(task, port, -1);
return NXT_OK;
}
}
void
nxt_main_stop_all_processes(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_port_t *port;
nxt_process_t *process;
nxt_runtime_process_each(rt, process) {
if (nxt_pid != process->pid) {
nxt_process_port_each(process, port) {
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
} nxt_process_port_loop;
}
} nxt_runtime_process_loop;
}
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_err_t err;
nxt_pid_t pid;
nxt_debug(task, "sigchld handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
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));
}
nxt_main_cleanup_worker_process(task, pid);
}
}
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_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid)
{
nxt_buf_t *buf;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_process_type_t ptype;
nxt_process_init_t *init;
nxt_process_restart_t restart;
rt = task->thread->runtime;
process = nxt_runtime_process_find(rt, pid);
if (process) {
init = process->init;
process->init = NULL;
ptype = nxt_process_type(process);
restart = nxt_process_restarts[ptype];
if (process->ready) {
init->stream = 0;
}
nxt_process_close_ports(task, process);
if (nxt_exiting) {
nxt_mp_destroy(init->mem_pool);
if (rt->nprocesses <= 2) {
nxt_runtime_quit(task, 0);
}
return;
}
nxt_runtime_process_each(rt, process) {
if (process->pid == nxt_pid
|| process->pid == pid
|| nxt_queue_is_empty(&process->ports))
{
continue;
}
port = nxt_process_port_first(process);
if (nxt_proc_remove_notify_matrix[ptype][port->type] == 0) {
continue;
}
buf = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
sizeof(pid));
if (nxt_slow_path(buf == NULL)) {
continue;
}
buf->mem.free = nxt_cpymem(buf->mem.free, &pid, sizeof(pid));
nxt_port_socket_write(task, port, NXT_PORT_MSG_REMOVE_PID,
-1, init->stream, 0, buf);
} nxt_runtime_process_loop;
if (restart != NULL) {
restart(task, rt, init);
} else {
nxt_mp_destroy(init->mem_pool);
}
}
}
static void
nxt_main_stop_worker_processes(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_port_t *port;
nxt_process_t *process;
nxt_runtime_process_each(rt, process) {
nxt_process_port_each(process, port) {
if (port->type == NXT_PROCESS_WORKER) {
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
}
} nxt_process_port_loop;
} nxt_runtime_process_loop;
}
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];
b = msg->buf;
sa = (nxt_sockaddr_t *) b->mem.pos;
/* TODO check b size and make plain */
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\"",
(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);
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_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_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 void
nxt_main_port_modules_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
uint32_t index;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_conf_value_t *conf, *root, *value;
nxt_app_lang_module_t *lang;
static nxt_str_t root_path = nxt_string("/");
rt = task->thread->runtime;
if (msg->port_msg.pid != rt->port_by_type[NXT_PROCESS_DISCOVERY]->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_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;
}
nxt_debug(task, "lang %d %s \"%s\"",
lang->type, lang->version, lang->file);
}
qsort(rt->languages->elts, rt->languages->nelts,
sizeof(nxt_app_lang_module_t), nxt_app_lang_compare);
fail:
nxt_mp_destroy(mp);
ret = nxt_main_start_controller_process(task, rt);
if (ret == NXT_OK) {
(void) nxt_main_start_router_process(task, rt);
}
}
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)
{
ssize_t n, size, offset;
nxt_buf_t *b;
nxt_int_t ret;
nxt_file_t file;
nxt_runtime_t *rt;
nxt_memzero(&file, sizeof(nxt_file_t));
rt = task->thread->runtime;
file.name = (nxt_file_name_t *) rt->conf_tmp;
if (nxt_slow_path(nxt_file_open(task, &file, NXT_FILE_WRONLY,
NXT_FILE_TRUNCATE, NXT_FILE_OWNER_ACCESS)
!= NXT_OK))
{
goto error;
}
offset = 0;
for (b = msg->buf; b != NULL; b = b->next) {
size = nxt_buf_mem_used_size(&b->mem);
n = nxt_file_write(&file, b->mem.pos, size, offset);
if (nxt_slow_path(n != size)) {
nxt_file_close(task, &file);
(void) nxt_file_delete(file.name);
goto error;
}
offset += n;
}
nxt_file_close(task, &file);
ret = nxt_file_rename(file.name, (nxt_file_name_t *) rt->conf);
if (nxt_fast_path(ret == NXT_OK)) {
return;
}
error:
nxt_alert(task, "failed to store current configuration");
}
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);
}
}
static nxt_int_t
nxt_init_set_isolation(nxt_task_t *task, nxt_process_init_t *init,
nxt_conf_value_t *isolation)
{
nxt_int_t ret;
nxt_conf_value_t *object;
static nxt_str_t nsname = nxt_string("namespaces");
static nxt_str_t uidname = nxt_string("uidmap");
static nxt_str_t gidname = nxt_string("gidmap");
if (isolation == NULL) {
return NXT_OK;
}
object = nxt_conf_get_object_member(isolation, &nsname, NULL);
if (object != NULL) {
ret = nxt_init_set_ns(task, init, object);
if (ret != NXT_OK) {
return ret;
}
}
object = nxt_conf_get_object_member(isolation, &uidname, NULL);
if (object != NULL) {
init->isolation.clone.uidmap = object;
}
object = nxt_conf_get_object_member(isolation, &gidname, NULL);
if (object != NULL) {
init->isolation.clone.gidmap = object;
}
return NXT_OK;
}
static nxt_int_t
nxt_init_set_ns(nxt_task_t *task, nxt_process_init_t *init,
nxt_conf_value_t *namespaces)
{
uint32_t index;
nxt_str_t name;
nxt_int_t flag;
nxt_conf_value_t *value;
index = 0;
for ( ;; ) {
value = nxt_conf_next_object_member(namespaces, &name, &index);
if (value == NULL) {
break;
}
flag = 0;
#if (NXT_HAVE_CLONE_NEWUSER)
if (nxt_str_eq(&name, "credential", 10)) {
flag = CLONE_NEWUSER;
}
#endif
#if (NXT_HAVE_CLONE_NEWPID)
if (nxt_str_eq(&name, "pid", 3)) {
flag = CLONE_NEWPID;
}
#endif
#if (NXT_HAVE_CLONE_NEWNET)
if (nxt_str_eq(&name, "network", 7)) {
flag = CLONE_NEWNET;
}
#endif
#if (NXT_HAVE_CLONE_NEWUTS)
if (nxt_str_eq(&name, "uname", 5)) {
flag = CLONE_NEWUTS;
}
#endif
#if (NXT_HAVE_CLONE_NEWNS)
if (nxt_str_eq(&name, "mount", 5)) {
flag = CLONE_NEWNS;
}
#endif
#if (NXT_HAVE_CLONE_NEWCGROUP)
if (nxt_str_eq(&name, "cgroup", 6)) {
flag = CLONE_NEWCGROUP;
}
#endif
if (!flag) {
nxt_alert(task, "unknown namespace flag: \"%V\"", &name);
return NXT_ERROR;
}
if (nxt_conf_get_boolean(value)) {
init->isolation.clone.flags |= flag;
}
}
return NXT_OK;
}
static nxt_process_init_t *
nxt_process_init_create(nxt_task_t *task, nxt_process_type_t type,
const nxt_str_t *name)
{
nxt_mp_t *mp;
nxt_int_t ret;
nxt_runtime_t *rt;
nxt_process_init_t *init;
mp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(mp == NULL)) {
return NULL;
}
init = nxt_mp_zalloc(mp, sizeof(nxt_process_init_t));
if (nxt_slow_path(init == NULL)) {
goto fail;
}
init->mem_pool = mp;
ret = nxt_process_init_name_set(init, type, name);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
rt = task->thread->runtime;
init->type = type;
init->start = nxt_process_starts[type];
init->port_handlers = nxt_process_port_handlers[type];
init->signals = nxt_worker_process_signals;
init->user_cred = &rt->user_cred;
init->data = &rt;
return init;
fail:
nxt_mp_destroy(mp);
return NULL;
}
static nxt_int_t
nxt_process_init_name_set(nxt_process_init_t *init, nxt_process_type_t type,
const nxt_str_t *name)
{
u_char *str, *end;
size_t size;
const char *fmt;
size = name->length + 1;
if (type == NXT_PROCESS_WORKER) {
size += nxt_length("\"\" application");
fmt = "\"%V\" application%Z";
} else {
fmt = "%V%Z";
}
str = nxt_mp_alloc(init->mem_pool, size);
if (nxt_slow_path(str == NULL)) {
return NXT_ERROR;
}
end = str + size;
nxt_sprintf(str, end, fmt, name);
init->name = (char *) str;
return NXT_OK;
}
static nxt_int_t
nxt_process_init_creds_set(nxt_task_t *task, nxt_process_init_t *init,
nxt_str_t *user, nxt_str_t *group)
{
char *str;
init->user_cred = nxt_mp_zalloc(init->mem_pool, sizeof(nxt_credential_t));
if (nxt_slow_path(init->user_cred == NULL)) {
return NXT_ERROR;
}
str = nxt_mp_zalloc(init->mem_pool, user->length + 1);
if (nxt_slow_path(str == NULL)) {
return NXT_ERROR;
}
nxt_memcpy(str, user->start, user->length);
str[user->length] = '\0';
init->user_cred->user = str;
if (group->start != NULL) {
str = nxt_mp_zalloc(init->mem_pool, group->length + 1);
if (nxt_slow_path(str == NULL)) {
return NXT_ERROR;
}
nxt_memcpy(str, group->start, group->length);
str[group->length] = '\0';
} else {
str = NULL;
}
return nxt_credential_get(task, init->mem_pool, init->user_cred, str);
}
Reference in New Issue View Git Blame Copy Permalink