/* * Copyright (C) Max Romanov * Copyright (C) Igor Sysoev * Copyright (C) Valentin V. Bartenev * Copyright (C) NGINX, Inc. */ #include #include #include #include #include #include #include #include #include #include #if (NXT_HAVE_PR_SET_NO_NEW_PRIVS) #include #endif typedef struct { nxt_app_type_t type; nxt_str_t version; nxt_str_t file; nxt_array_t *mounts; } nxt_module_t; static nxt_int_t nxt_discovery_start(nxt_task_t *task, nxt_process_data_t *data); static nxt_buf_t *nxt_discovery_modules(nxt_task_t *task, const char *path); static nxt_int_t nxt_discovery_module(nxt_task_t *task, nxt_mp_t *mp, nxt_array_t *modules, const char *name); static void nxt_discovery_completion_handler(nxt_task_t *task, void *obj, void *data); 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); static uint32_t compat[] = { NXT_VERNUM, NXT_DEBUG, }; 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 = { .quit = nxt_signal_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, }; 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, .rpc_error = nxt_port_rpc_handler, }; const nxt_process_init_t nxt_discovery_process = { .name = "discovery", .type = NXT_PROCESS_DISCOVERY, .prefork = NULL, .restart = 0, .setup = nxt_process_core_setup, .start = nxt_discovery_start, .port_handlers = &nxt_discovery_process_port_handlers, .signals = nxt_process_signals, }; 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, .start = NULL, .prefork = NULL, .restart = 0, .port_handlers = &nxt_app_process_port_handlers, .signals = nxt_process_signals, }; static nxt_int_t nxt_discovery_start(nxt_task_t *task, nxt_process_data_t *data) { uint32_t stream; nxt_buf_t *b; nxt_int_t ret; nxt_port_t *main_port, *discovery_port; nxt_runtime_t *rt; nxt_log(task, NXT_LOG_INFO, "discovery started"); rt = task->thread->runtime; b = nxt_discovery_modules(task, rt->modules); if (nxt_slow_path(b == NULL)) { return NXT_ERROR; } main_port = rt->port_by_type[NXT_PROCESS_MAIN]; discovery_port = rt->port_by_type[NXT_PROCESS_DISCOVERY]; stream = nxt_port_rpc_register_handler(task, discovery_port, nxt_discovery_quit, nxt_discovery_quit, main_port->pid, NULL); if (nxt_slow_path(stream == 0)) { return NXT_ERROR; } ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_MODULES, -1, stream, discovery_port->id, b); if (nxt_slow_path(ret != NXT_OK)) { nxt_port_rpc_cancel(task, discovery_port, stream); return NXT_ERROR; } return NXT_OK; } static nxt_buf_t * nxt_discovery_modules(nxt_task_t *task, const char *path) { char *name; u_char *p, *end; size_t size; glob_t glb; nxt_mp_t *mp; nxt_buf_t *b; nxt_int_t ret; nxt_uint_t i, n, j; nxt_array_t *modules, *mounts; nxt_module_t *module; nxt_fs_mount_t *mnt; b = NULL; mp = nxt_mp_create(1024, 128, 256, 32); if (mp == NULL) { return b; } ret = glob(path, 0, NULL, &glb); n = glb.gl_pathc; if (ret != 0) { nxt_log(task, NXT_LOG_NOTICE, "no modules matching: \"%s\" found", path); n = 0; } modules = nxt_array_create(mp, n, sizeof(nxt_module_t)); if (modules == NULL) { goto fail; } for (i = 0; i < n; i++) { name = glb.gl_pathv[i]; ret = nxt_discovery_module(task, mp, modules, name); if (ret != NXT_OK) { goto fail; } } size = nxt_length("[]"); module = modules->elts; n = modules->nelts; for (i = 0; i < n; i++) { nxt_debug(task, "module: %d %V %V", module[i].type, &module[i].version, &module[i].file); size += nxt_length("{\"type\": ,"); size += nxt_length(" \"version\": \"\","); size += nxt_length(" \"file\": \"\","); size += nxt_length(" \"mounts\": []},"); size += NXT_INT_T_LEN + module[i].version.length + module[i].file.length; mounts = module[i].mounts; size += mounts->nelts * nxt_length("{\"src\": \"\", \"dst\": \"\", " "\"type\": , \"name\": \"\", " "\"flags\": , \"data\": \"\"},"); mnt = mounts->elts; for (j = 0; j < mounts->nelts; j++) { size += nxt_strlen(mnt[j].src) + nxt_strlen(mnt[j].dst) + nxt_strlen(mnt[j].name) + (2 * NXT_INT_T_LEN) + (mnt[j].data == NULL ? 0 : nxt_strlen(mnt[j].data)); } } b = nxt_buf_mem_alloc(mp, size, 0); if (b == NULL) { goto fail; } b->completion_handler = nxt_discovery_completion_handler; p = b->mem.free; end = b->mem.end; *p++ = '['; for (i = 0; i < n; i++) { mounts = module[i].mounts; p = nxt_sprintf(p, end, "{\"type\": %d, \"version\": \"%V\", " "\"file\": \"%V\", \"mounts\": [", module[i].type, &module[i].version, &module[i].file); mnt = mounts->elts; for (j = 0; j < mounts->nelts; j++) { p = nxt_sprintf(p, end, "{\"src\": \"%s\", \"dst\": \"%s\", " "\"name\": \"%s\", \"type\": %d, \"flags\": %d, " "\"data\": \"%s\"},", mnt[j].src, mnt[j].dst, mnt[j].name, mnt[j].type, mnt[j].flags, mnt[j].data == NULL ? (u_char *) "" : mnt[j].data); } *p++ = ']'; *p++ = '}'; *p++ = ','; } *p++ = ']'; if (nxt_slow_path(p > end)) { nxt_alert(task, "discovery write past the buffer"); goto fail; } b->mem.free = p; fail: globfree(&glb); return b; } static nxt_int_t nxt_discovery_module(nxt_task_t *task, nxt_mp_t *mp, nxt_array_t *modules, const char *name) { void *dl; nxt_str_t version; nxt_int_t ret; nxt_uint_t i, j, n; nxt_array_t *mounts; nxt_module_t *module; nxt_app_type_t type; nxt_fs_mount_t *to; nxt_app_module_t *app; const nxt_fs_mount_t *from; /* * Only memory allocation failure should return NXT_ERROR. * Any module processing errors are ignored. */ ret = NXT_ERROR; dl = dlopen(name, RTLD_GLOBAL | RTLD_NOW); if (dl == NULL) { nxt_alert(task, "dlopen(\"%s\"), failed: \"%s\"", name, dlerror()); return NXT_OK; } app = dlsym(dl, "nxt_app_module"); if (app != NULL) { nxt_log(task, NXT_LOG_NOTICE, "module: %V %s \"%s\"", &app->type, app->version, name); if (app->compat_length != sizeof(compat) || nxt_memcmp(app->compat, compat, sizeof(compat)) != 0) { nxt_log(task, NXT_LOG_NOTICE, "incompatible module %s", name); goto done; } type = nxt_app_parse_type(app->type.start, app->type.length); if (type == NXT_APP_UNKNOWN) { nxt_log(task, NXT_LOG_NOTICE, "unknown module type %V", &app->type); goto done; } module = modules->elts; n = modules->nelts; version.start = (u_char *) app->version; version.length = nxt_strlen(app->version); for (i = 0; i < n; i++) { if (type == module[i].type && nxt_strstr_eq(&module[i].version, &version)) { nxt_log(task, NXT_LOG_NOTICE, "ignoring %s module with the same " "application language version %V %V as in %V", name, &app->type, &version, &module[i].file); goto done; } } module = nxt_array_add(modules); if (module == NULL) { goto fail; } module->type = type; nxt_str_dup(mp, &module->version, &version); if (module->version.start == NULL) { goto fail; } module->file.length = nxt_strlen(name); module->file.start = nxt_mp_alloc(mp, module->file.length); if (module->file.start == NULL) { goto fail; } nxt_memcpy(module->file.start, name, module->file.length); module->mounts = nxt_array_create(mp, app->nmounts, sizeof(nxt_fs_mount_t)); if (nxt_slow_path(module->mounts == NULL)) { goto fail; } mounts = module->mounts; for (j = 0; j < app->nmounts; j++) { from = &app->mounts[j]; to = nxt_array_zero_add(mounts); if (nxt_slow_path(to == NULL)) { goto fail; } to->src = nxt_cstr_dup(mp, to->src, from->src); if (nxt_slow_path(to->src == NULL)) { goto fail; } to->dst = nxt_cstr_dup(mp, to->dst, from->dst); if (nxt_slow_path(to->dst == NULL)) { goto fail; } to->name = nxt_cstr_dup(mp, to->name, from->name); if (nxt_slow_path(to->name == NULL)) { goto fail; } to->type = from->type; if (from->data != NULL) { to->data = nxt_cstr_dup(mp, to->data, from->data); if (nxt_slow_path(to->data == NULL)) { goto fail; } } to->flags = from->flags; } } else { nxt_alert(task, "dlsym(\"%s\"), failed: \"%s\"", name, dlerror()); } done: ret = NXT_OK; fail: if (dlclose(dl) != 0) { nxt_alert(task, "dlclose(\"%s\"), failed: \"%s\"", name, dlerror()); } return ret; } static void nxt_discovery_completion_handler(nxt_task_t *task, void *obj, void *data) { nxt_mp_t *mp; nxt_buf_t *b; b = obj; mp = b->data; nxt_mp_destroy(mp); } static void nxt_discovery_quit(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data) { nxt_signal_quit_handler(task, msg); } static nxt_int_t nxt_proto_setup(nxt_task_t *task, nxt_process_t *process) { nxt_int_t ret; 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); return NXT_ERROR; } nxt_app = lang->module; if (nxt_app == NULL) { nxt_debug(task, "application language module: %s \"%s\"", lang->version, lang->file); nxt_app = nxt_app_module_load(task, lang->file); if (nxt_slow_path(nxt_app == NULL)) { return NXT_ERROR; } } if (nxt_slow_path(nxt_app_set_environment(app_conf->environment) != NXT_OK)) { nxt_alert(task, "failed to set environment"); return NXT_ERROR; } if (nxt_app->setup != NULL) { ret = nxt_app->setup(task, process, app_conf); if (nxt_slow_path(ret != NXT_OK)) { return ret; } } #if (NXT_HAVE_ISOLATION_ROOTFS) if (process->isolation.rootfs != NULL) { if (process->isolation.mounts != NULL) { ret = nxt_isolation_prepare_rootfs(task, process); if (nxt_slow_path(ret != NXT_OK)) { return ret; } } ret = nxt_isolation_change_root(task, process); if (nxt_slow_path(ret != NXT_OK)) { return NXT_ERROR; } } #endif if (app_conf->working_directory != NULL && app_conf->working_directory[0] != 0) { ret = chdir(app_conf->working_directory); if (nxt_slow_path(ret != 0)) { nxt_log(task, NXT_LOG_WARN, "chdir(%s) failed %E", app_conf->working_directory, nxt_errno); return NXT_ERROR; } } 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) { char *err; void *dl; nxt_app_module_t *app; dl = dlopen(name, RTLD_GLOBAL | RTLD_LAZY); if (nxt_slow_path(dl == NULL)) { err = dlerror(); nxt_alert(task, "dlopen(\"%s\") failed: \"%s\"", name, err != NULL ? err : "(null)"); return NULL; } app = dlsym(dl, "nxt_app_module"); if (nxt_slow_path(app == NULL)) { err = dlerror(); nxt_alert(task, "dlsym(\"%s\", \"nxt_app_module\") failed: \"%s\"", name, err != NULL ? err : "(null)"); if (dlclose(dl) != 0) { err = dlerror(); nxt_alert(task, "dlclose(\"%s\") failed: \"%s\"", name, err != NULL ? err : "(null)"); } } return app; } static nxt_int_t nxt_app_set_environment(nxt_conf_value_t *environment) { char *env, *p; uint32_t next; nxt_str_t name, value; nxt_conf_value_t *value_obj; if (environment != NULL) { next = 0; for ( ;; ) { value_obj = nxt_conf_next_object_member(environment, &name, &next); if (value_obj == NULL) { break; } nxt_conf_get_string(value_obj, &value); env = nxt_malloc(name.length + value.length + 2); if (nxt_slow_path(env == NULL)) { return NXT_ERROR; } p = nxt_cpymem(env, name.start, name.length); *p++ = '='; p = nxt_cpymem(p, value.start, value.length); *p = '\0'; if (nxt_slow_path(putenv(env) != 0)) { return NXT_ERROR; } } } return NXT_OK; } static u_char * nxt_cstr_dup(nxt_mp_t *mp, u_char *dst, u_char *src) { u_char *p; size_t len; len = nxt_strlen(src); if (dst == NULL) { dst = nxt_mp_alloc(mp, len + 1); if (nxt_slow_path(dst == NULL)) { return NULL; } } p = nxt_cpymem(dst, src, len); *p = '\0'; return dst; } 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) { u_char *p, *end, *version; size_t version_length; nxt_uint_t i, n; nxt_app_type_t type; nxt_app_lang_module_t *lang; end = name->start + name->length; version = end; for (p = name->start; p < end; p++) { if (*p == ' ') { version = p + 1; break; } if (*p >= '0' && *p <= '9') { version = p; break; } } type = nxt_app_parse_type(name->start, p - name->start); if (type == NXT_APP_UNKNOWN) { return NULL; } version_length = end - version; lang = rt->languages->elts; n = rt->languages->nelts; for (i = 0; i < n; i++) { /* * Versions are sorted in descending order * so first match chooses the highest version. */ if (lang[i].type == type && nxt_strvers_match(lang[i].version, version, version_length)) { return &lang[i]; } } return NULL; } nxt_app_type_t nxt_app_parse_type(u_char *p, size_t length) { nxt_str_t str; str.length = length; str.start = p; if (nxt_str_eq(&str, "external", 8) || nxt_str_eq(&str, "go", 2)) { return NXT_APP_EXTERNAL; } else if (nxt_str_eq(&str, "python", 6)) { return NXT_APP_PYTHON; } else if (nxt_str_eq(&str, "php", 3)) { return NXT_APP_PHP; } else if (nxt_str_eq(&str, "perl", 4)) { return NXT_APP_PERL; } else if (nxt_str_eq(&str, "ruby", 4)) { return NXT_APP_RUBY; } else if (nxt_str_eq(&str, "java", 4)) { return NXT_APP_JAVA; } return NXT_APP_UNKNOWN; } 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, *proto_port, *router_port; nxt_runtime_t *rt; nxt_memzero(init, sizeof(nxt_unit_init_t)); rt = task->thread->runtime; proto_port = rt->port_by_type[NXT_PROCESS_PROTOTYPE]; if (nxt_slow_path(proto_port == NULL)) { return NXT_ERROR; } router_port = rt->port_by_type[NXT_PROCESS_ROUTER]; if (nxt_slow_path(router_port == NULL)) { return NXT_ERROR; } my_port = nxt_runtime_port_find(rt, nxt_pid, 0); if (nxt_slow_path(my_port == NULL)) { return NXT_ERROR; } 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 = proto_port->pair[1]; init->ready_stream = my_port->process->stream; init->router_port.id.pid = router_port->pid; init->router_port.id.id = router_port->id; init->router_port.in_fd = -1; init->router_port.out_fd = router_port->pair[1]; init->read_port.id.pid = my_port->pid; init->read_port.id.id = my_port->id; init->read_port.in_fd = my_port->pair[0]; init->read_port.out_fd = my_port->pair[1]; init->shared_port_fd = conf->shared_port_fd; init->shared_queue_fd = conf->shared_queue_fd; init->log_fd = 2; init->shm_limit = conf->shm_limit; init->request_limit = conf->request_limit; 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; }