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

Introducing application prototype processes.

Browse Source
This commit is contained in:
Tiago Natel de Moura
2021-11-09 15:48:44 +03:00
parent 1de660b6df
commit e207415a78
15 changed files with 1109 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
docs/changes.xml
View File

@@ -95,6 +95,12 @@ request routing by the query string.
</para>
</change>
<change type="feature">
<para>
PHP opcache is shared between application processes.
</para>
</change>
<change type="bugfix">
<para>
fixed building with glibc 2.34, notably Fedora 35.

455
src/nxt_application.c
View File

@@ -42,9 +42,23 @@ static void nxt_discovery_quit(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data);
static nxt_app_module_t *nxt_app_module_load(nxt_task_t *task,
const char *name);
static nxt_int_t nxt_proto_setup(nxt_task_t *task, nxt_process_t *process);
static nxt_int_t nxt_proto_start(nxt_task_t *task, nxt_process_data_t *data);
static nxt_int_t nxt_app_setup(nxt_task_t *task, nxt_process_t *process);
static nxt_int_t nxt_app_set_environment(nxt_conf_value_t *environment);
static void nxt_proto_start_process_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static void nxt_proto_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
static void nxt_proto_process_created_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static void nxt_proto_quit_children(nxt_task_t *task);
static nxt_process_t *nxt_proto_process_find(nxt_task_t *task, nxt_pid_t pid);
static void nxt_proto_process_add(nxt_task_t *task, nxt_process_t *process);
static nxt_process_t *nxt_proto_process_remove(nxt_task_t *task, nxt_pid_t pid);
static u_char *nxt_cstr_dup(nxt_mp_t *mp, u_char *dst, u_char *src);
static void nxt_proto_signal_handler(nxt_task_t *task, void *obj, void *data);
static void nxt_proto_sigterm_handler(nxt_task_t *task, void *obj, void *data);
static void nxt_proto_sigchld_handler(nxt_task_t *task, void *obj, void *data);
nxt_str_t nxt_server = nxt_string(NXT_SERVER);
@@ -55,7 +69,12 @@ static uint32_t compat[] = {
};
static nxt_lvlhsh_t nxt_proto_processes;
static nxt_queue_t nxt_proto_children;
static nxt_bool_t nxt_proto_exiting;
static nxt_app_module_t *nxt_app;
static nxt_common_app_conf_t *nxt_app_conf;
static const nxt_port_handlers_t nxt_discovery_process_port_handlers = {
@@ -70,6 +89,29 @@ static const nxt_port_handlers_t nxt_discovery_process_port_handlers = {
};
const nxt_sig_event_t nxt_prototype_signals[] = {
nxt_event_signal(SIGHUP, nxt_proto_signal_handler),
nxt_event_signal(SIGINT, nxt_proto_sigterm_handler),
nxt_event_signal(SIGQUIT, nxt_proto_sigterm_handler),
nxt_event_signal(SIGTERM, nxt_proto_sigterm_handler),
nxt_event_signal(SIGCHLD, nxt_proto_sigchld_handler),
nxt_event_signal_end,
};
static const nxt_port_handlers_t nxt_proto_process_port_handlers = {
.quit = nxt_proto_quit_handler,
.change_file = nxt_port_change_log_file_handler,
.new_port = nxt_port_new_port_handler,
.process_created = nxt_proto_process_created_handler,
.process_ready = nxt_port_process_ready_handler,
.remove_pid = nxt_port_remove_pid_handler,
.start_process = nxt_proto_start_process_handler,
.rpc_ready = nxt_port_rpc_handler,
.rpc_error = nxt_port_rpc_handler,
};
static const nxt_port_handlers_t nxt_app_process_port_handlers = {
.quit = nxt_signal_quit_handler,
.rpc_ready = nxt_port_rpc_handler,
@@ -89,12 +131,23 @@ const nxt_process_init_t nxt_discovery_process = {
};
const nxt_process_init_t nxt_proto_process = {
.type = NXT_PROCESS_PROTOTYPE,
.prefork = nxt_isolation_main_prefork,
.restart = 0,
.setup = nxt_proto_setup,
.start = nxt_proto_start,
.port_handlers = &nxt_proto_process_port_handlers,
.signals = nxt_prototype_signals,
};
const nxt_process_init_t nxt_app_process = {
.type = NXT_PROCESS_APP,
.setup = nxt_app_setup,
.prefork = nxt_isolation_main_prefork,
.start = NULL,
.prefork = NULL,
.restart = 0,
.start = NULL, /* set to module->start */
.port_handlers = &nxt_app_process_port_handlers,
.signals = nxt_process_signals,
};
@@ -443,15 +496,18 @@ nxt_discovery_quit(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
static nxt_int_t
nxt_app_setup(nxt_task_t *task, nxt_process_t *process)
nxt_proto_setup(nxt_task_t *task, nxt_process_t *process)
{
nxt_int_t ret;
nxt_process_init_t *init;
nxt_app_lang_module_t *lang;
nxt_common_app_conf_t *app_conf;
app_conf = process->data.app;
nxt_queue_init(&nxt_proto_children);
nxt_app_conf = app_conf;
lang = nxt_app_lang_module(task->thread->runtime, &app_conf->type);
if (nxt_slow_path(lang == NULL)) {
nxt_alert(task, "unknown application type: \"%V\"", &app_conf->type);
@@ -479,7 +535,6 @@ nxt_app_setup(nxt_task_t *task, nxt_process_t *process)
if (nxt_app->setup != NULL) {
ret = nxt_app->setup(task, process, app_conf);
if (nxt_slow_path(ret != NXT_OK)) {
return ret;
}
@@ -514,16 +569,254 @@ nxt_app_setup(nxt_task_t *task, nxt_process_t *process)
}
}
init = nxt_process_init(process);
init->start = nxt_app->start;
process->state = NXT_PROCESS_STATE_CREATED;
return NXT_OK;
}
static nxt_int_t
nxt_proto_start(nxt_task_t *task, nxt_process_data_t *data)
{
nxt_debug(task, "prototype waiting for clone messages");
return NXT_OK;
}
static void
nxt_proto_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *p;
nxt_int_t ret;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_process_init_t *init;
rt = task->thread->runtime;
process = nxt_process_new(rt);
if (nxt_slow_path(process == NULL)) {
goto failed;
}
process->mem_pool = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(process->mem_pool == NULL)) {
nxt_process_use(task, process, -1);
goto failed;
}
process->parent_port = rt->port_by_type[NXT_PROCESS_PROTOTYPE];
init = nxt_process_init(process);
*init = nxt_app_process;
process->name = nxt_mp_alloc(process->mem_pool, nxt_app_conf->name.length
+ sizeof("\"\" application") + 1);
if (nxt_slow_path(process->name == NULL)) {
nxt_process_use(task, process, -1);
goto failed;
}
init->start = nxt_app->start;
init->name = (const char *) nxt_app_conf->name.start;
p = (u_char *) process->name;
*p++ = '"';
p = nxt_cpymem(p, nxt_app_conf->name.start, nxt_app_conf->name.length);
p = nxt_cpymem(p, "\" application", 13);
*p = '\0';
process->user_cred = &rt->user_cred;
process->data.app = nxt_app_conf;
process->stream = msg->port_msg.stream;
ret = nxt_process_start(task, process);
if (nxt_slow_path(ret == NXT_ERROR)) {
nxt_process_use(task, process, -1);
goto failed;
}
nxt_proto_process_add(task, process);
return;
failed:
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);
}
}
static void
nxt_proto_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_debug(task, "prototype quit handler");
nxt_proto_quit_children(task);
nxt_proto_exiting = 1;
if (nxt_queue_is_empty(&nxt_proto_children)) {
nxt_process_quit(task, 0);
}
}
static void
nxt_proto_quit_children(nxt_task_t *task)
{
nxt_port_t *port;
nxt_process_t *process;
nxt_queue_each(process, &nxt_proto_children, nxt_process_t, link) {
port = nxt_process_port_first(process);
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
}
nxt_queue_loop;
}
static void
nxt_proto_process_created_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_pid_t isolated_pid, pid;
nxt_process_t *process;
isolated_pid = nxt_recv_msg_cmsg_pid(msg);
process = nxt_proto_process_find(task, isolated_pid);
if (nxt_slow_path(process == NULL)) {
return;
}
process->state = NXT_PROCESS_STATE_CREATED;
pid = msg->port_msg.pid;
if (process->pid != pid) {
nxt_debug(task, "app process %PI (aka %PI) is created", isolated_pid,
pid);
nxt_runtime_process_remove(task->thread->runtime, process);
process->pid = pid;
nxt_runtime_process_add(task, process);
} else {
nxt_debug(task, "app process %PI is created", isolated_pid);
}
}
static void
nxt_proto_signal_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_trace(task, "signal signo:%d (%s) received, ignored",
(int) (uintptr_t) obj, data);
}
static void
nxt_proto_sigterm_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_trace(task, "signal signo:%d (%s) received",
(int) (uintptr_t) obj, data);
nxt_proto_quit_children(task);
nxt_proto_exiting = 1;
if (nxt_queue_is_empty(&nxt_proto_children)) {
nxt_process_quit(task, 0);
}
}
static void
nxt_proto_sigchld_handler(nxt_task_t *task, void *obj, void *data)
{
int status;
nxt_err_t err;
nxt_pid_t pid;
nxt_process_t *process;
nxt_debug(task, "proto 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, "app process (isolated %PI) exited on signal %d%s",
pid, WTERMSIG(status),
WCOREDUMP(status) ? " (core dumped)" : "");
#else
nxt_alert(task, "app process (isolated %PI) exited on signal %d",
pid, WTERMSIG(status));
#endif
} else {
nxt_trace(task, "app process (isolated %PI) exited with code %d",
pid, WEXITSTATUS(status));
}
process = nxt_proto_process_remove(task, pid);
if (process == NULL) {
continue;
}
if (process->state != NXT_PROCESS_STATE_CREATING) {
nxt_port_remove_notify_others(task, process);
}
nxt_process_close_ports(task, process);
if (nxt_proto_exiting && nxt_queue_is_empty(&nxt_proto_children)) {
nxt_process_quit(task, 0);
return;
}
}
}
static nxt_app_module_t *
nxt_app_module_load(nxt_task_t *task, const char *name)
{
@@ -619,6 +912,19 @@ nxt_cstr_dup(nxt_mp_t *mp, u_char *dst, u_char *src)
}
static nxt_int_t
nxt_app_setup(nxt_task_t *task, nxt_process_t *process)
{
nxt_process_init_t *init;
process->state = NXT_PROCESS_STATE_CREATED;
init = nxt_process_init(process);
return init->start(task, &process->data);
}
nxt_app_lang_module_t *
nxt_app_lang_module(nxt_runtime_t *rt, nxt_str_t *name)
{
@@ -707,15 +1013,15 @@ nxt_int_t
nxt_unit_default_init(nxt_task_t *task, nxt_unit_init_t *init,
nxt_common_app_conf_t *conf)
{
nxt_port_t *my_port, *main_port, *router_port;
nxt_port_t *my_port, *proto_port, *router_port;
nxt_runtime_t *rt;
nxt_memzero(init, sizeof(nxt_unit_init_t));
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
if (nxt_slow_path(main_port == NULL)) {
proto_port = rt->port_by_type[NXT_PROCESS_PROTOTYPE];
if (nxt_slow_path(proto_port == NULL)) {
return NXT_ERROR;
}
@@ -729,10 +1035,10 @@ nxt_unit_default_init(nxt_task_t *task, nxt_unit_init_t *init,
return NXT_ERROR;
}
init->ready_port.id.pid = main_port->pid;
init->ready_port.id.id = main_port->id;
init->ready_port.id.pid = proto_port->pid;
init->ready_port.id.id = proto_port->id;
init->ready_port.in_fd = -1;
init->ready_port.out_fd = main_port->pair[1];
init->ready_port.out_fd = proto_port->pair[1];
init->ready_stream = my_port->process->stream;
@@ -753,3 +1059,122 @@ nxt_unit_default_init(nxt_task_t *task, nxt_unit_init_t *init,
return NXT_OK;
}
static nxt_int_t
nxt_proto_lvlhsh_isolated_pid_test(nxt_lvlhsh_query_t *lhq, void *data)
{
nxt_pid_t *qpid;
nxt_process_t *process;
process = data;
qpid = (nxt_pid_t *) lhq->key.start;
if (*qpid == process->isolated_pid) {
return NXT_OK;
}
return NXT_DECLINED;
}
static const nxt_lvlhsh_proto_t lvlhsh_processes_proto nxt_aligned(64) = {
NXT_LVLHSH_DEFAULT,
nxt_proto_lvlhsh_isolated_pid_test,
nxt_lvlhsh_alloc,
nxt_lvlhsh_free,
};
nxt_inline void
nxt_proto_process_lhq_pid(nxt_lvlhsh_query_t *lhq, nxt_pid_t *pid)
{
lhq->key_hash = nxt_murmur_hash2(pid, sizeof(nxt_pid_t));
lhq->key.length = sizeof(nxt_pid_t);
lhq->key.start = (u_char *) pid;
lhq->proto = &lvlhsh_processes_proto;
}
static void
nxt_proto_process_add(nxt_task_t *task, nxt_process_t *process)
{
nxt_runtime_t *rt;
nxt_lvlhsh_query_t lhq;
rt = task->thread->runtime;
nxt_proto_process_lhq_pid(&lhq, &process->isolated_pid);
lhq.replace = 0;
lhq.value = process;
lhq.pool = rt->mem_pool;
switch (nxt_lvlhsh_insert(&nxt_proto_processes, &lhq)) {
case NXT_OK:
nxt_debug(task, "process (isolated %PI) added", process->isolated_pid);
nxt_queue_insert_tail(&nxt_proto_children, &process->link);
break;
default:
nxt_debug(task, "process (isolated %PI) failed to add",
process->isolated_pid);
break;
}
}
static nxt_process_t *
nxt_proto_process_remove(nxt_task_t *task, nxt_pid_t pid)
{
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_lvlhsh_query_t lhq;
nxt_proto_process_lhq_pid(&lhq, &pid);
rt = task->thread->runtime;
lhq.pool = rt->mem_pool;
switch (nxt_lvlhsh_delete(&nxt_proto_processes, &lhq)) {
case NXT_OK:
nxt_debug(task, "process (isolated %PI) removed", pid);
process = lhq.value;
nxt_queue_remove(&process->link);
break;
default:
nxt_debug(task, "process (isolated %PI) remove failed", pid);
process = NULL;
break;
}
return process;
}
static nxt_process_t *
nxt_proto_process_find(nxt_task_t *task, nxt_pid_t pid)
{
nxt_process_t *process;
nxt_lvlhsh_query_t lhq;
nxt_proto_process_lhq_pid(&lhq, &pid);
if (nxt_lvlhsh_find(&nxt_proto_processes, &lhq) == NXT_OK) {
process = lhq.value;
} else {
nxt_debug(task, "process (isolated %PI) not found", pid);
process = NULL;
}
return process;
}

10
src/nxt_external.c
View File

@@ -67,7 +67,7 @@ nxt_external_start(nxt_task_t *task, nxt_process_data_t *data)
nxt_str_t str;
nxt_int_t rc;
nxt_uint_t i, argc;
nxt_port_t *my_port, *main_port, *router_port;
nxt_port_t *my_port, *proto_port, *router_port;
nxt_runtime_t *rt;
nxt_conf_value_t *value;
nxt_common_app_conf_t *conf;
@@ -76,17 +76,17 @@ nxt_external_start(nxt_task_t *task, nxt_process_data_t *data)
rt = task->thread->runtime;
conf = data->app;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
proto_port = rt->port_by_type[NXT_PROCESS_PROTOTYPE];
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
my_port = nxt_runtime_port_find(rt, nxt_pid, 0);
if (nxt_slow_path(main_port == NULL || my_port == NULL
if (nxt_slow_path(proto_port == NULL || my_port == NULL
|| router_port == NULL))
{
return NXT_ERROR;
}
rc = nxt_external_fd_no_cloexec(task, main_port->pair[1]);
rc = nxt_external_fd_no_cloexec(task, proto_port->pair[1]);
if (nxt_slow_path(rc != NXT_OK)) {
return NXT_ERROR;
}
@@ -115,7 +115,7 @@ nxt_external_start(nxt_task_t *task, nxt_process_data_t *data)
"%PI,%ud,%d,%d;"
"%d,%z,%uD,%Z",
NXT_VERSION, my_port->process->stream,
main_port->pid, main_port->id, main_port->pair[1],
proto_port->pid, proto_port->id, proto_port->pair[1],
router_port->pid, router_port->id, router_port->pair[1],
my_port->pid, my_port->id, my_port->pair[0],
my_port->pair[1],

183
src/nxt_main_process.c
View File

@@ -10,6 +10,7 @@
#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
@@ -52,6 +53,8 @@ static nxt_int_t nxt_main_listening_socket(nxt_sockaddr_t *sa,
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,
@@ -326,7 +329,7 @@ static nxt_conf_app_map_t nxt_app_maps[] = {
static void
nxt_port_main_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
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);
@@ -334,7 +337,33 @@ nxt_port_main_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
static void
nxt_port_main_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
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;
@@ -374,16 +403,18 @@ nxt_port_main_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
return;
}
process->parent_port = rt->port_by_type[NXT_PROCESS_MAIN];
init = nxt_process_init(process);
*init = nxt_app_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 process: %*s", b->mem.free - b->mem.pos,
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));
@@ -399,7 +430,7 @@ nxt_port_main_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
init->name = (const char *) start;
process->name = nxt_mp_alloc(process->mem_pool, app_conf->name.length
+ sizeof("\"\" application") + 1);
+ sizeof("\"\" prototype") + 1);
if (nxt_slow_path(process->name == NULL)) {
goto failed;
@@ -408,7 +439,7 @@ nxt_port_main_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
p = (u_char *) process->name;
*p++ = '"';
p = nxt_cpymem(p, init->name, app_conf->name.length);
p = nxt_cpymem(p, "\" application", 13);
p = nxt_cpymem(p, "\" prototype", 11);
*p = '\0';
app_conf->shm_limit = 100 * 1024 * 1024;
@@ -504,21 +535,17 @@ nxt_main_process_created_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
rt = task->thread->runtime;
process = nxt_runtime_process_find(rt, msg->port_msg.pid);
if (nxt_slow_path(process == NULL)) {
return;
}
nxt_assert(process->state == NXT_PROCESS_STATE_CREATING);
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,
@@ -542,10 +569,13 @@ nxt_main_process_created_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
static nxt_port_handlers_t nxt_main_process_port_handlers = {
.data = nxt_port_main_data_handler,
.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,
.start_process = nxt_port_main_start_process_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,
@@ -559,6 +589,88 @@ static nxt_port_handlers_t nxt_main_process_port_handlers = {
};
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)
{
@@ -751,8 +863,10 @@ nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
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;
nxt_process_t *process, *child;
nxt_process_init_t init;
nxt_debug(task, "sigchld handler signo:%d (%s)",
@@ -809,7 +923,31 @@ nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
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);
}
@@ -819,6 +957,15 @@ nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
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);

1
src/nxt_main_process.h
View File

@@ -27,6 +27,7 @@ NXT_EXPORT extern nxt_uint_t nxt_conf_ver;
NXT_EXPORT extern const nxt_process_init_t nxt_discovery_process;
NXT_EXPORT extern const nxt_process_init_t nxt_controller_process;
NXT_EXPORT extern const nxt_process_init_t nxt_router_process;
NXT_EXPORT extern const nxt_process_init_t nxt_proto_process;
NXT_EXPORT extern const nxt_process_init_t nxt_app_process;
extern const nxt_sig_event_t nxt_main_process_signals[];

28
src/nxt_port.c
View File

@@ -12,6 +12,8 @@
#include <nxt_port_queue.h>
static void nxt_port_remove_pid(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_pid_t pid);
static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
static nxt_atomic_uint_t nxt_port_last_id = 1;
@@ -274,6 +276,8 @@ nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_debug(task, "port %PI:%d already exists", new_port_msg->pid,
new_port_msg->id);
msg->u.new_port = port;
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
return;
@@ -384,14 +388,13 @@ nxt_port_change_log_file(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t slot,
port = nxt_process_port_first(process);
b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
sizeof(nxt_port_data_t));
b = nxt_buf_mem_alloc(task->thread->engine->mem_pool,
sizeof(nxt_uint_t), 0);
if (nxt_slow_path(b == NULL)) {
continue;
}
*(nxt_uint_t *) b->mem.pos = slot;
b->mem.free += sizeof(nxt_uint_t);
b->mem.free = nxt_cpymem(b->mem.free, &slot, sizeof(nxt_uint_t));
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_CHANGE_FILE,
fd, 0, 0, b);
@@ -497,16 +500,25 @@ nxt_port_remove_notify_others(nxt_task_t *task, nxt_process_t *process)
void
nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_buf_t *buf;
nxt_pid_t pid;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_buf_t *buf;
buf = msg->buf;
nxt_assert(nxt_buf_used_size(buf) == sizeof(pid));
nxt_memcpy(&pid, buf->mem.pos, sizeof(pid));
nxt_memcpy(&pid, buf->mem.pos, sizeof(nxt_pid_t));
nxt_port_remove_pid(task, msg, pid);
}
static void
nxt_port_remove_pid(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_pid_t pid)
{
nxt_runtime_t *rt;
nxt_process_t *process;
msg->u.removed_pid = pid;

3
src/nxt_port.h
View File

@@ -33,6 +33,7 @@ struct nxt_port_handlers_s {
/* New process */
nxt_port_handler_t process_created;
nxt_port_handler_t process_ready;
nxt_port_handler_t whoami;
/* Process exit/crash notification. */
nxt_port_handler_t remove_pid;
@@ -92,6 +93,7 @@ typedef enum {
_NXT_PORT_MSG_PROCESS_CREATED = nxt_port_handler_idx(process_created),
_NXT_PORT_MSG_PROCESS_READY = nxt_port_handler_idx(process_ready),
_NXT_PORT_MSG_WHOAMI = nxt_port_handler_idx(whoami),
_NXT_PORT_MSG_REMOVE_PID = nxt_port_handler_idx(remove_pid),
_NXT_PORT_MSG_QUIT = nxt_port_handler_idx(quit),
@@ -131,6 +133,7 @@ typedef enum {
NXT_PORT_MSG_PROCESS_CREATED = nxt_msg_last(_NXT_PORT_MSG_PROCESS_CREATED),
NXT_PORT_MSG_PROCESS_READY = nxt_msg_last(_NXT_PORT_MSG_PROCESS_READY),
NXT_PORT_MSG_WHOAMI = nxt_msg_last(_NXT_PORT_MSG_WHOAMI),
NXT_PORT_MSG_QUIT = nxt_msg_last(_NXT_PORT_MSG_QUIT),
NXT_PORT_MSG_REMOVE_PID = nxt_msg_last(_NXT_PORT_MSG_REMOVE_PID),

270
src/nxt_process.c
View File

@@ -5,7 +5,6 @@
*/
#include <nxt_main.h>
#include <nxt_main_process.h>
#if (NXT_HAVE_CLONE)
#include <nxt_clone.h>
@@ -17,10 +16,26 @@
#include <sys/prctl.h>
#endif
#if (NXT_HAVE_CLONE) && (NXT_HAVE_CLONE_NEWPID)
#define nxt_is_pid_isolated(process) \
nxt_is_clone_flag_set(process->isolation.clone.flags, NEWPID)
#else
#define nxt_is_pid_isolated(process) \
(0)
#endif
static nxt_pid_t nxt_process_create(nxt_task_t *task, nxt_process_t *process);
static nxt_int_t nxt_process_do_start(nxt_task_t *task, nxt_process_t *process);
static nxt_int_t nxt_process_whoami(nxt_task_t *task, nxt_process_t *process);
static nxt_int_t nxt_process_setup(nxt_task_t *task, nxt_process_t *process);
static nxt_int_t nxt_process_child_fixup(nxt_task_t *task,
nxt_process_t *process);
static void nxt_process_whoami_ok(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data);
static void nxt_process_whoami_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data);
static nxt_int_t nxt_process_send_created(nxt_task_t *task,
nxt_process_t *process);
static nxt_int_t nxt_process_send_ready(nxt_task_t *task,
@@ -44,19 +59,28 @@ nxt_uid_t nxt_euid;
nxt_gid_t nxt_egid;
nxt_bool_t nxt_proc_conn_matrix[NXT_PROCESS_MAX][NXT_PROCESS_MAX] = {
{ 1, 1, 1, 1, 1 },
{ 1, 0, 0, 0, 0 },
{ 1, 0, 0, 1, 0 },
{ 1, 0, 1, 0, 1 },
{ 1, 0, 0, 1, 0 },
{ 1, 1, 1, 1, 1, 1 },
{ 1, 0, 0, 0, 0, 0 },
{ 1, 0, 0, 1, 0, 0 },
{ 1, 0, 1, 1, 1, 1 },
{ 1, 0, 0, 1, 0, 0 },
{ 1, 0, 0, 1, 0, 0 },
};
nxt_bool_t nxt_proc_remove_notify_matrix[NXT_PROCESS_MAX][NXT_PROCESS_MAX] = {
{ 0, 0, 0, 0, 0 },
{ 0, 0, 0, 0, 0 },
{ 0, 0, 0, 1, 0 },
{ 0, 0, 1, 0, 1 },
{ 0, 0, 0, 1, 0 },
{ 0, 0, 0, 0, 0, 0 },
{ 0, 0, 0, 0, 0, 0 },
{ 0, 0, 0, 1, 0, 0 },
{ 0, 0, 1, 0, 1, 1 },
{ 0, 0, 0, 1, 0, 0 },
{ 1, 0, 0, 1, 0, 0 },
};
static const nxt_port_handlers_t nxt_process_whoami_port_handlers = {
.quit = nxt_signal_quit_handler,
.rpc_ready = nxt_port_rpc_handler,
.rpc_error = nxt_port_rpc_handler,
};
@@ -78,6 +102,8 @@ nxt_process_new(nxt_runtime_t *rt)
process->use_count = 1;
nxt_queue_init(&process->children);
return process;
}
@@ -108,6 +134,8 @@ nxt_process_init_start(nxt_task_t *task, nxt_process_init_t init)
return NXT_ERROR;
}
process->parent_port = rt->port_by_type[rt->type];
process->name = init.name;
process->user_cred = &rt->user_cred;
@@ -177,7 +205,7 @@ nxt_process_start(nxt_task_t *task, nxt_process_t *process)
break;
default:
/* The main process created a new process. */
/* The parent process created a new process. */
nxt_process_use(task, process, -1);
@@ -216,41 +244,16 @@ nxt_process_child_fixup(nxt_task_t *task, nxt_process_t *process)
init = nxt_process_init(process);
nxt_ppid = nxt_pid;
nxt_pid = nxt_getpid();
process->pid = nxt_pid;
process->isolated_pid = nxt_pid;
/* Clean inherited cached thread tid. */
task->thread->tid = 0;
#if (NXT_HAVE_CLONE && NXT_HAVE_CLONE_NEWPID)
if (nxt_is_clone_flag_set(process->isolation.clone.flags, NEWPID)) {
ssize_t pidsz;
char procpid[10];
nxt_debug(task, "%s isolated pid is %d", process->name, nxt_pid);
pidsz = readlink("/proc/self", procpid, sizeof(procpid));
if (nxt_slow_path(pidsz < 0 || pidsz >= (ssize_t) sizeof(procpid))) {
nxt_alert(task, "failed to read real pid from /proc/self");
return NXT_ERROR;
}
procpid[pidsz] = '\0';
nxt_pid = (nxt_pid_t) strtol(procpid, NULL, 10);
nxt_assert(nxt_pid > 0 && nxt_errno != ERANGE);
process->pid = nxt_pid;
task->thread->tid = nxt_pid;
nxt_debug(task, "%s real pid is %d", process->name, nxt_pid);
}
#endif
ptype = init->type;
nxt_port_reset_next_id();
@@ -262,7 +265,9 @@ nxt_process_child_fixup(nxt_task_t *task, nxt_process_t *process)
/* Remove not ready processes. */
nxt_runtime_process_each(rt, p) {
if (nxt_proc_conn_matrix[ptype][nxt_process_type(p)] == 0) {
if (nxt_proc_conn_matrix[ptype][nxt_process_type(p)] == 0
&& p->pid != nxt_ppid) /* Always keep parent's port. */
{
nxt_debug(task, "remove not required process %PI", p->pid);
nxt_process_close_ports(task, p);
@@ -315,9 +320,8 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
return -1;
}
nxt_runtime_process_add(task, process);
if (nxt_slow_path(nxt_process_setup(task, process) != NXT_OK)) {
ret = nxt_process_setup(task, process);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_process_quit(task, 1);
}
@@ -337,6 +341,7 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
#endif
process->pid = pid;
process->isolated_pid = pid;
nxt_runtime_process_add(task, process);
@@ -348,7 +353,6 @@ static nxt_int_t
nxt_process_setup(nxt_task_t *task, nxt_process_t *process)
{
nxt_int_t ret;
nxt_port_t *port, *main_port;
nxt_thread_t *thread;
nxt_runtime_t *rt;
nxt_process_init_t *init;
@@ -388,17 +392,45 @@ nxt_process_setup(nxt_task_t *task, nxt_process_t *process)
return NXT_ERROR;
}
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
nxt_port_read_close(process->parent_port);
nxt_port_write_enable(task, process->parent_port);
nxt_port_read_close(main_port);
nxt_port_write_enable(task, main_port);
/*
* If the parent process is already isolated, rt->pid_isolation is already
* set to 1 at this point.
*/
if (nxt_is_pid_isolated(process)) {
rt->is_pid_isolated = 1;
}
if (rt->is_pid_isolated
|| process->parent_port != rt->port_by_type[NXT_PROCESS_MAIN])
{
ret = nxt_process_whoami(task, process);
} else {
ret = nxt_process_do_start(task, process);
}
return ret;
}
static nxt_int_t
nxt_process_do_start(nxt_task_t *task, nxt_process_t *process)
{
nxt_int_t ret;
nxt_port_t *port;
nxt_process_init_t *init;
nxt_runtime_process_add(task, process);
init = nxt_process_init(process);
port = nxt_process_port_first(process);
nxt_port_enable(task, port, init->port_handlers);
ret = init->setup(task, process);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
@@ -434,6 +466,113 @@ nxt_process_setup(nxt_task_t *task, nxt_process_t *process)
}
static nxt_int_t
nxt_process_whoami(nxt_task_t *task, nxt_process_t *process)
{
uint32_t stream;
nxt_fd_t fd;
nxt_buf_t *buf;
nxt_int_t ret;
nxt_port_t *my_port, *main_port;
nxt_runtime_t *rt;
rt = task->thread->runtime;
my_port = nxt_process_port_first(process);
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
nxt_assert(my_port != NULL && main_port != NULL);
nxt_port_enable(task, my_port, &nxt_process_whoami_port_handlers);
buf = nxt_buf_mem_alloc(main_port->mem_pool, sizeof(nxt_pid_t), 0);
if (nxt_slow_path(buf == NULL)) {
return NXT_ERROR;
}
buf->mem.free = nxt_cpymem(buf->mem.free, &nxt_ppid, sizeof(nxt_pid_t));
stream = nxt_port_rpc_register_handler(task, my_port,
nxt_process_whoami_ok,
nxt_process_whoami_error,
main_port->pid, process);
if (nxt_slow_path(stream == 0)) {
nxt_mp_free(main_port->mem_pool, buf);
return NXT_ERROR;
}
fd = (process->parent_port != main_port) ? my_port->pair[1] : -1;
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_WHOAMI,
fd, stream, my_port->id, buf);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_alert(task, "%s failed to send WHOAMI message", process->name);
nxt_port_rpc_cancel(task, my_port, stream);
nxt_mp_free(main_port->mem_pool, buf);
return NXT_ERROR;
}
return NXT_OK;
}
static void
nxt_process_whoami_ok(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
{
nxt_pid_t pid, isolated_pid;
nxt_buf_t *buf;
nxt_port_t *port;
nxt_process_t *process;
nxt_runtime_t *rt;
process = data;
buf = msg->buf;
nxt_assert(nxt_buf_used_size(buf) == sizeof(nxt_pid_t));
nxt_memcpy(&pid, buf->mem.pos, sizeof(nxt_pid_t));
isolated_pid = nxt_pid;
if (isolated_pid != pid) {
nxt_pid = pid;
process->pid = pid;
nxt_process_port_each(process, port) {
port->pid = pid;
} nxt_process_port_loop;
}
rt = task->thread->runtime;
if (process->parent_port != rt->port_by_type[NXT_PROCESS_MAIN]) {
port = process->parent_port;
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_PROCESS_CREATED,
-1, 0, 0, NULL);
nxt_log(task, NXT_LOG_INFO, "%s started", process->name);
}
if (nxt_slow_path(nxt_process_do_start(task, process) != NXT_OK)) {
nxt_process_quit(task, 1);
}
}
static void
nxt_process_whoami_error(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
{
nxt_alert(task, "WHOAMI error");
nxt_process_quit(task, 1);
}
static nxt_int_t
nxt_process_send_created(nxt_task_t *task, nxt_process_t *process)
{
@@ -483,6 +622,9 @@ nxt_process_created_ok(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
nxt_process_init_t *init;
process = data;
process->state = NXT_PROCESS_STATE_READY;
init = nxt_process_init(process);
ret = nxt_process_apply_creds(task, process);
@@ -492,11 +634,23 @@ nxt_process_created_ok(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
nxt_log(task, NXT_LOG_INFO, "%s started", process->name);
ret = nxt_process_send_ready(task, process);
if (nxt_slow_path(ret != NXT_OK)) {
goto fail;
}
ret = init->start(task, &process->data);
fail:
if (nxt_process_type(process) != NXT_PROCESS_PROTOTYPE) {
nxt_port_write_close(nxt_process_port_first(process));
}
nxt_process_quit(task, ret == NXT_OK ? 0 : 1);
if (nxt_fast_path(ret == NXT_OK)) {
return;
}
fail:
nxt_process_quit(task, 1);
}
@@ -584,7 +738,8 @@ nxt_process_apply_creds(nxt_task_t *task, nxt_process_t *process)
#if (NXT_HAVE_CLONE && NXT_HAVE_CLONE_NEWUSER)
if (!cap_setid
&& nxt_is_clone_flag_set(process->isolation.clone.flags, NEWUSER)) {
&& nxt_is_clone_flag_set(process->isolation.clone.flags, NEWUSER))
{
cap_setid = 1;
}
#endif
@@ -618,16 +773,9 @@ static nxt_int_t
nxt_process_send_ready(nxt_task_t *task, nxt_process_t *process)
{
nxt_int_t ret;
nxt_port_t *main_port;
nxt_runtime_t *rt;
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
nxt_assert(main_port != NULL);
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_PROCESS_READY,
ret = nxt_port_socket_write(task, process->parent_port,
NXT_PORT_MSG_PROCESS_READY,
-1, process->stream, 0, NULL);
if (nxt_slow_path(ret != NXT_OK)) {

11
src/nxt_process.h
View File

@@ -99,19 +99,26 @@ typedef struct {
struct nxt_process_s {
nxt_pid_t pid;
const char *name;
nxt_queue_t ports; /* of nxt_port_t */
nxt_queue_t ports; /* of nxt_port_t.link */
nxt_process_state_t state;
nxt_bool_t registered;
nxt_int_t use_count;
nxt_port_mmaps_t incoming;
nxt_pid_t isolated_pid;
const char *name;
nxt_port_t *parent_port;
uint32_t stream;
nxt_mp_t *mem_pool;
nxt_credential_t *user_cred;
nxt_queue_t children; /* of nxt_process_t.link */
nxt_queue_link_t link; /* for nxt_process_t.children */
nxt_process_data_t data;
nxt_process_isolation_t isolation;

1
src/nxt_process_type.h
View File

@@ -13,6 +13,7 @@ typedef enum {
NXT_PROCESS_DISCOVERY,
NXT_PROCESS_CONTROLLER,
NXT_PROCESS_ROUTER,
NXT_PROCESS_PROTOTYPE,
NXT_PROCESS_APP,
NXT_PROCESS_MAX,

241
src/nxt_router.c
View File

@@ -65,12 +65,14 @@ typedef struct {
typedef struct {
nxt_app_t *app;
nxt_router_temp_conf_t *temp_conf;
uint8_t proto; /* 1 bit */
} nxt_app_rpc_t;
typedef struct {
nxt_app_joint_t *app_joint;
uint32_t generation;
uint8_t proto; /* 1 bit */
} nxt_app_joint_rpc_t;
@@ -392,25 +394,44 @@ nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port,
{
size_t size;
uint32_t stream;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_app_t *app;
nxt_buf_t *b;
nxt_port_t *main_port;
nxt_port_t *dport;
nxt_runtime_t *rt;
nxt_app_joint_rpc_t *app_joint_rpc;
app = data;
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
nxt_thread_mutex_lock(&app->mutex);
dport = app->proto_port;
nxt_thread_mutex_unlock(&app->mutex);
if (dport != NULL) {
nxt_debug(task, "app '%V' %p start process", &app->name, app);
b = NULL;
} else {
if (app->proto_port_requests > 0) {
nxt_debug(task, "app '%V' %p wait for prototype process",
&app->name, app);
app->proto_port_requests++;
goto skip;
}
nxt_debug(task, "app '%V' %p start prototype process", &app->name, app);
rt = task->thread->runtime;
dport = rt->port_by_type[NXT_PROCESS_MAIN];
size = app->name.length + 1 + app->conf.length;
b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool, size);
b = nxt_buf_mem_alloc(task->thread->engine->mem_pool, size, 0);
if (nxt_slow_path(b == NULL)) {
goto failed;
}
@@ -418,6 +439,7 @@ nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port,
nxt_buf_cpystr(b, &app->name);
*b->mem.free++ = '\0';
nxt_buf_cpystr(b, &app->conf);
}
app_joint_rpc = nxt_port_rpc_register_handler_ex(task, port,
nxt_router_app_port_ready,
@@ -429,7 +451,7 @@ nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port,
stream = nxt_port_rpc_ex_stream(app_joint_rpc);
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_PROCESS,
ret = nxt_port_socket_write(task, dport, NXT_PORT_MSG_START_PROCESS,
-1, stream, port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, port, stream);
@@ -439,26 +461,23 @@ nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port,
app_joint_rpc->app_joint = app->joint;
app_joint_rpc->generation = app->generation;
app_joint_rpc->proto = (b != NULL);
if (b != NULL) {
app->proto_port_requests++;
b = NULL;
}
nxt_router_app_joint_use(task, app->joint, 1);
nxt_router_app_use(task, app, -1);
return;
failed:
if (b != NULL) {
mp = b->data;
nxt_mp_free(mp, b);
nxt_mp_release(mp);
nxt_mp_free(b->data, b);
}
nxt_thread_mutex_lock(&app->mutex);
app->pending_processes--;
nxt_thread_mutex_unlock(&app->mutex);
skip:
nxt_router_app_use(task, app, -1);
}
@@ -658,6 +677,12 @@ nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_router_greet_controller(task, msg->u.new_port);
}
if (port != NULL && port->type == NXT_PROCESS_PROTOTYPE) {
nxt_port_rpc_handler(task, msg);
return;
}
if (port == NULL || port->type != NXT_PROCESS_APP) {
if (msg->port_msg.stream == 0) {
@@ -683,6 +708,8 @@ nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
return;
}
nxt_debug(task, "new port id %d (%d)", port->id, port->type);
/*
* Port with "id == 0" is application 'main' port and it always
* should come with non-zero stream.
@@ -819,7 +846,8 @@ nxt_router_app_restart_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_app_t *app;
nxt_int_t ret;
nxt_str_t app_name;
nxt_port_t *port, *reply_port, *shared_port, *old_shared_port;
nxt_port_t *reply_port, *shared_port, *old_shared_port;
nxt_port_t *proto_port;
nxt_port_msg_type_t reply;
reply_port = nxt_runtime_port_find(task->thread->runtime,
@@ -862,12 +890,15 @@ nxt_router_app_restart_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_thread_mutex_lock(&app->mutex);
nxt_queue_each(port, &app->ports, nxt_port_t, app_link) {
proto_port = app->proto_port;
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1,
0, 0, NULL);
if (proto_port != NULL) {
nxt_debug(task, "send QUIT to prototype '%V' pid %PI", &app->name,
proto_port->pid);
} nxt_queue_loop;
app->proto_port = NULL;
proto_port->app = NULL;
}
app->generation++;
@@ -883,6 +914,15 @@ nxt_router_app_restart_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_port_close(task, old_shared_port);
nxt_port_use(task, old_shared_port, -1);
if (proto_port != NULL) {
(void) nxt_port_socket_write(task, proto_port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
nxt_port_close(task, proto_port);
nxt_port_use(task, proto_port, -1);
}
reply = NXT_PORT_MSG_RPC_READY_LAST;
} else {
@@ -2735,19 +2775,16 @@ nxt_router_app_rpc_create(nxt_task_t *task,
uint32_t stream;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *main_port, *router_port;
nxt_port_t *router_port, *dport;
nxt_runtime_t *rt;
nxt_app_rpc_t *rpc;
rpc = nxt_mp_alloc(tmcf->mem_pool, sizeof(nxt_app_rpc_t));
if (rpc == NULL) {
goto fail;
}
rt = task->thread->runtime;
rpc->app = app;
rpc->temp_conf = tmcf;
dport = app->proto_port;
nxt_debug(task, "app '%V' prefork", &app->name);
if (dport == NULL) {
nxt_debug(task, "app '%V' prototype prefork", &app->name);
size = app->name.length + 1 + app->conf.length;
@@ -2762,27 +2799,43 @@ nxt_router_app_rpc_create(nxt_task_t *task,
*b->mem.free++ = '\0';
nxt_buf_cpystr(b, &app->conf);
rt = task->thread->runtime;
main_port = rt->port_by_type[NXT_PROCESS_MAIN];
dport = rt->port_by_type[NXT_PROCESS_MAIN];
} else {
nxt_debug(task, "app '%V' prefork", &app->name);
b = NULL;
}
router_port = rt->port_by_type[NXT_PROCESS_ROUTER];
stream = nxt_port_rpc_register_handler(task, router_port,
rpc = nxt_port_rpc_register_handler_ex(task, router_port,
nxt_router_app_prefork_ready,
nxt_router_app_prefork_error,
-1, rpc);
if (nxt_slow_path(stream == 0)) {
sizeof(nxt_app_rpc_t));
if (nxt_slow_path(rpc == NULL)) {
goto fail;
}
ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_PROCESS,
-1, stream, router_port->id, b);
rpc->app = app;
rpc->temp_conf = tmcf;
rpc->proto = (b != NULL);
stream = nxt_port_rpc_ex_stream(rpc);
ret = nxt_port_socket_write(task, dport,
NXT_PORT_MSG_START_PROCESS,
-1, stream, router_port->id, b);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_rpc_cancel(task, router_port, stream);
goto fail;
}
if (b == NULL) {
nxt_port_rpc_ex_set_peer(task, router_port, rpc, dport->pid);
app->pending_processes++;
}
return;
@@ -2807,9 +2860,24 @@ nxt_router_app_prefork_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
port = msg->u.new_port;
nxt_assert(port != NULL);
nxt_assert(port->type == NXT_PROCESS_APP);
nxt_assert(port->id == 0);
if (rpc->proto) {
nxt_assert(app->proto_port == NULL);
nxt_assert(port->type == NXT_PROCESS_PROTOTYPE);
nxt_port_inc_use(port);
app->proto_port = port;
port->app = app;
nxt_router_app_rpc_create(task, rpc->temp_conf, app);
return;
}
nxt_assert(port->type == NXT_PROCESS_APP);
port->app = app;
port->main_app_port = port;
@@ -2851,10 +2919,16 @@ nxt_router_app_prefork_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
app = rpc->app;
tmcf = rpc->temp_conf;
if (rpc->proto) {
nxt_log(task, NXT_LOG_WARN, "failed to start prototype \"%V\"",
&app->name);
} else {
nxt_log(task, NXT_LOG_WARN, "failed to start application \"%V\"",
&app->name);
app->pending_processes--;
}
nxt_router_conf_error(task, tmcf);
}
@@ -4413,8 +4487,9 @@ static void
nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
void *data)
{
uint32_t n;
nxt_app_t *app;
nxt_bool_t start_process;
nxt_bool_t start_process, restarted;
nxt_port_t *port;
nxt_app_joint_t *app_joint;
nxt_app_joint_rpc_t *app_joint_rpc;
@@ -4427,7 +4502,6 @@ nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_assert(app_joint != NULL);
nxt_assert(port != NULL);
nxt_assert(port->type == NXT_PROCESS_APP);
nxt_assert(port->id == 0);
app = app_joint->app;
@@ -4444,11 +4518,51 @@ nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg,
nxt_thread_mutex_lock(&app->mutex);
restarted = (app->generation != app_joint_rpc->generation);
if (app_joint_rpc->proto) {
nxt_assert(app->proto_port == NULL);
nxt_assert(port->type == NXT_PROCESS_PROTOTYPE);
n = app->proto_port_requests;
app->proto_port_requests = 0;
if (nxt_slow_path(restarted)) {
nxt_thread_mutex_unlock(&app->mutex);
nxt_debug(task, "proto port ready for restarted app, send QUIT");
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0,
NULL);
} else {
port->app = app;
app->proto_port = port;
nxt_thread_mutex_unlock(&app->mutex);
nxt_port_use(task, port, 1);
}
port = task->thread->runtime->port_by_type[NXT_PROCESS_ROUTER];
while (n > 0) {
nxt_router_app_use(task, app, 1);
nxt_router_start_app_process_handler(task, port, app);
n--;
}
return;
}
nxt_assert(port->type == NXT_PROCESS_APP);
nxt_assert(app->pending_processes != 0);
app->pending_processes--;
if (nxt_slow_path(app->generation != app_joint_rpc->generation)) {
if (nxt_slow_path(restarted)) {
nxt_debug(task, "new port ready for restarted app, send QUIT");
start_process = !task->thread->engine->shutdown
@@ -4591,7 +4705,6 @@ nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
}
nxt_inline nxt_port_t *
nxt_router_app_get_port_for_quit(nxt_task_t *task, nxt_app_t *app)
{
@@ -4791,6 +4904,20 @@ nxt_router_app_port_close(nxt_task_t *task, nxt_port_t *port)
nxt_thread_mutex_lock(&app->mutex);
if (port == app->proto_port) {
app->proto_port = NULL;
port->app = NULL;
nxt_thread_mutex_unlock(&app->mutex);
nxt_debug(task, "app '%V' prototype pid %PI closed", &app->name,
port->pid);
nxt_port_use(task, port, -1);
return;
}
nxt_port_hash_remove(&app->port_hash, port);
app->port_hash_count--;
@@ -4979,7 +5106,7 @@ static void
nxt_router_free_app(nxt_task_t *task, void *obj, void *data)
{
nxt_app_t *app;
nxt_port_t *port;
nxt_port_t *port, *proto_port;
nxt_app_joint_t *app_joint;
app_joint = obj;
@@ -4991,10 +5118,6 @@ nxt_router_free_app(nxt_task_t *task, void *obj, void *data)
break;
}
nxt_debug(task, "send QUIT to app '%V' pid %PI", &app->name, port->pid);
nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
nxt_port_use(task, port, -1);
}
@@ -5015,8 +5138,28 @@ nxt_router_free_app(nxt_task_t *task, void *obj, void *data)
nxt_port_use(task, port, -1);
}
proto_port = app->proto_port;
if (proto_port != NULL) {
nxt_debug(task, "send QUIT to prototype '%V' pid %PI", &app->name,
proto_port->pid);
app->proto_port = NULL;
proto_port->app = NULL;
}
nxt_thread_mutex_unlock(&app->mutex);
if (proto_port != NULL) {
nxt_port_socket_write(task, proto_port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
nxt_port_close(task, proto_port);
nxt_port_use(task, proto_port, -1);
}
nxt_assert(app->proto_port == NULL);
nxt_assert(app->processes == 0);
nxt_assert(app->active_requests == 0);
nxt_assert(app->port_hash_count == 0);

2
src/nxt_router.h
View File

@@ -126,6 +126,7 @@ struct nxt_app_s {
uint32_t max_pending_processes;
uint32_t generation;
uint32_t proto_port_requests;
nxt_msec_t timeout;
nxt_msec_t idle_timeout;
@@ -144,6 +145,7 @@ struct nxt_app_s {
nxt_app_joint_t *joint;
nxt_port_t *shared_port;
nxt_port_t *proto_port;
nxt_port_mmaps_t outgoing;
};

19
src/nxt_runtime.c
View File

@@ -40,8 +40,6 @@ static void nxt_runtime_thread_pool_init(void);
static void nxt_runtime_thread_pool_exit(nxt_task_t *task, void *obj,
void *data);
static nxt_process_t *nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid);
static void nxt_runtime_process_remove(nxt_runtime_t *rt,
nxt_process_t *process);
static void nxt_runtime_port_add(nxt_task_t *task, nxt_port_t *port);
@@ -504,7 +502,9 @@ nxt_runtime_stop_app_processes(nxt_task_t *task, nxt_runtime_t *rt)
init = nxt_process_init(process);
if (init->type == NXT_PROCESS_APP) {
if (init->type == NXT_PROCESS_APP
|| init->type == NXT_PROCESS_PROTOTYPE)
{
nxt_process_port_each(process, port) {
@@ -528,6 +528,8 @@ nxt_runtime_stop_all_processes(nxt_task_t *task, nxt_runtime_t *rt)
nxt_process_port_each(process, port) {
nxt_debug(task, "%d sending quit to %PI", rt->type, port->pid);
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0,
0, NULL);
@@ -1389,10 +1391,21 @@ nxt_runtime_pid_file_create(nxt_task_t *task, nxt_file_name_t *pid_file)
void
nxt_runtime_process_release(nxt_runtime_t *rt, nxt_process_t *process)
{
nxt_process_t *child;
if (process->registered == 1) {
nxt_runtime_process_remove(rt, process);
}
if (process->link.next != NULL) {
nxt_queue_remove(&process->link);
}
nxt_queue_each(child, &process->children, nxt_process_t, link) {
nxt_queue_remove(&child->link);
child->link.next = NULL;
} nxt_queue_loop;
nxt_assert(process->use_count == 0);
nxt_assert(process->registered == 0);

2
src/nxt_runtime.h
View File

@@ -54,6 +54,7 @@ struct nxt_runtime_s {
uint8_t daemon;
uint8_t batch;
uint8_t status;
uint8_t is_pid_isolated;
const char *engine;
uint32_t engine_connections;
@@ -95,6 +96,7 @@ nxt_int_t nxt_runtime_thread_pool_create(nxt_thread_t *thr, nxt_runtime_t *rt,
void nxt_runtime_process_add(nxt_task_t *task, nxt_process_t *process);
void nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process);
nxt_process_t *nxt_runtime_process_find(nxt_runtime_t *rt, nxt_pid_t pid);

10
src/nxt_unit.c
View File

@@ -418,6 +418,9 @@ typedef struct {
} nxt_unit_port_hash_id_t;
static pid_t nxt_unit_pid;
nxt_unit_ctx_t *
nxt_unit_init(nxt_unit_init_t *init)
{
@@ -428,6 +431,8 @@ nxt_unit_init(nxt_unit_init_t *init)
nxt_unit_impl_t *lib;
nxt_unit_port_t ready_port, router_port, read_port;
nxt_unit_pid = getpid();
lib = nxt_unit_create(init);
if (nxt_slow_path(lib == NULL)) {
return NULL;
@@ -471,6 +476,7 @@ nxt_unit_init(nxt_unit_init_t *init)
}
lib->pid = read_port.id.pid;
nxt_unit_pid = lib->pid;
ctx = &lib->main_ctx.ctx;
@@ -6571,7 +6577,7 @@ nxt_unit_log(nxt_unit_ctx_t *ctx, int level, const char *fmt, ...)
log_fd = lib->log_fd;
} else {
pid = getpid();
pid = nxt_unit_pid;
log_fd = STDERR_FILENO;
}
@@ -6615,7 +6621,7 @@ nxt_unit_req_log(nxt_unit_request_info_t *req, int level, const char *fmt, ...)
log_fd = lib->log_fd;
} else {
pid = getpid();
pid = nxt_unit_pid;
log_fd = STDERR_FILENO;
}