diff options
| -rw-r--r-- | src/nxt_application.c | 62 | ||||
| -rw-r--r-- | src/nxt_main_process.c | 9 |
2 files changed, 51 insertions, 20 deletions
diff --git a/src/nxt_application.c b/src/nxt_application.c index 7fecd70d..f9ac8a7f 100644 --- a/src/nxt_application.c +++ b/src/nxt_application.c @@ -26,6 +26,10 @@ typedef struct { 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); @@ -46,9 +50,11 @@ static nxt_application_module_t *nxt_app; nxt_int_t nxt_discovery_start(nxt_task_t *task, void *data) { - nxt_buf_t *b; - nxt_port_t *main_port; - nxt_runtime_t *rt; + uint32_t stream; + nxt_buf_t *b; + nxt_int_t ret; + nxt_port_t *main_port, *discovery_port; + nxt_runtime_t *rt; nxt_debug(task, "DISCOVERY"); @@ -56,30 +62,26 @@ nxt_discovery_start(nxt_task_t *task, void *data) b = nxt_discovery_modules(task, rt->modules); if (nxt_slow_path(b == NULL)) { - exit(1); + return NXT_ERROR; } main_port = rt->port_by_type[NXT_PROCESS_MAIN]; + discovery_port = rt->port_by_type[NXT_PROCESS_DISCOVERY]; - nxt_port_socket_write(task, main_port, NXT_PORT_MSG_MODULES, -1, - 0, -1, b); - - return NXT_OK; -} - + stream = nxt_port_rpc_register_handler(task, discovery_port, + nxt_discovery_quit, + nxt_discovery_quit, + main_port->pid, NULL); -static void -nxt_discovery_completion_handler(nxt_task_t *task, void *obj, void *data) -{ - nxt_mp_t *mp; - nxt_buf_t *b; + ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_MODULES, -1, + stream, discovery_port->id, b); - b = obj; - mp = b->data; - - nxt_mp_destroy(mp); + if (nxt_slow_path(ret != NXT_OK)) { + nxt_port_rpc_cancel(task, discovery_port, stream); + return NXT_ERROR; + } - exit(0); + return NXT_OK; } @@ -279,6 +281,26 @@ fail: } +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_worker_process_quit_handler(task, msg); +} + + nxt_int_t nxt_app_start(nxt_task_t *task, void *data) { diff --git a/src/nxt_main_process.c b/src/nxt_main_process.c index 80af51a5..f0b55c1b 100644 --- a/src/nxt_main_process.c +++ b/src/nxt_main_process.c @@ -1112,6 +1112,7 @@ nxt_main_port_modules_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) 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; @@ -1124,6 +1125,14 @@ nxt_main_port_modules_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) 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) { |