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

Process stop notification from master to all other processes.

Browse Source 
New port message type introduced NXT_PORT_MSG_REMOVE_PID. Default handler
removes process description from nxt_runtime_t with all ports, incoming and
outgoing mmaps etc.
This commit is contained in:
Max Romanov
2017-07-07 16:01:34 +03:00
parent 61008a7c0b
commit 74cda90e31
11 changed files with 113 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/nginext/nxt_go_port.c
View File

@@ -146,7 +146,7 @@ nxt_go_port_on_read(void *buf, size_t buf_size, void *oob, size_t oob_size)
nxt_go_debug("using data in shared memory");
}
if (port_msg->type > NXT_PORT_MSG_MAX) {
if (port_msg->type >= NXT_PORT_MSG_MAX) {
nxt_go_warn("unknown message type (%d)", (int)port_msg->type);
goto fail;
}
@@ -181,6 +181,13 @@ nxt_go_port_on_read(void *buf, size_t buf_size, void *oob, size_t oob_size)
return nxt_go_data_handler(port_msg, buf_size);
case NXT_PORT_MSG_REMOVE_PID:
nxt_go_debug("remove pid");
/* TODO remove all ports for this pid in Go */
/* TODO remove incoming & outgoing mmaps for this pid */
break;
default:
goto fail;
}

20
src/nxt_master_process.c
View File

@@ -51,7 +51,7 @@ nxt_master_process_start(nxt_thread_t *thr, nxt_task_t *task,
{
nxt_int_t ret;
rt->type = NXT_PROCESS_MASTER;
rt->types |= (1U << NXT_PROCESS_MASTER);
if (nxt_master_process_port_create(task, rt) != NXT_OK) {
return NXT_ERROR;
@@ -467,6 +467,7 @@ nxt_master_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
static void
nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid)
{
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_process_init_t *init;
@@ -478,10 +479,23 @@ nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid)
if (process) {
init = process->init;
/* TODO: free ports fds. */
nxt_runtime_process_remove(rt, process);
if (!nxt_exiting) {
nxt_runtime_process_each(rt, process)
{
if (process->pid == nxt_pid) {
continue;
}
port = nxt_process_port_first(process);
nxt_port_socket_write(task, port, NXT_PORT_MSG_REMOVE_PID,
-1, pid, 0, NULL);
}
nxt_runtime_process_loop;
}
if (nxt_exiting) {
if (rt->nprocesses == 2) {

22
src/nxt_port.c
View File

@@ -52,7 +52,7 @@ nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_port_handler_t *handlers;
if (nxt_fast_path(msg->port_msg.type <= NXT_PORT_MSG_MAX)) {
if (nxt_fast_path(msg->port_msg.type < NXT_PORT_MSG_MAX)) {
nxt_debug(task, "port %d: message type:%uD",
msg->port->socket.fd, msg->port_msg.type);
@@ -291,6 +291,26 @@ nxt_port_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
}
void
nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_pid_t pid;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_debug(task, "port remove pid handler");
rt = task->thread->runtime;
pid = msg->port_msg.stream;
process = nxt_runtime_process_find(rt, pid);
if (process) {
nxt_runtime_process_remove(rt, process);
}
}
void
nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{

9
src/nxt_port.h
View File

@@ -14,9 +14,10 @@ typedef enum {
NXT_PORT_MSG_CHANGE_FILE,
NXT_PORT_MSG_MMAP,
NXT_PORT_MSG_DATA,
} nxt_port_msg_type_t;
NXT_PORT_MSG_REMOVE_PID,
#define NXT_PORT_MSG_MAX NXT_PORT_MSG_DATA
NXT_PORT_MSG_MAX,
} nxt_port_msg_type_t;
/* Passed as a first iov chunk. */
@@ -56,10 +57,9 @@ struct nxt_port_recv_msg_s {
struct nxt_port_s {
/* Must be the first field. */
nxt_fd_event_t socket;
nxt_queue_link_t link;
nxt_queue_link_t link; /* for nxt_process_t.ports */
nxt_queue_t messages; /* of nxt_port_send_msg_t */
@@ -132,6 +132,7 @@ void nxt_port_change_log_file_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
void nxt_port_mmap_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
void nxt_port_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
void nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
void nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);

9
src/nxt_port_socket.c
View File

@@ -222,7 +222,7 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
nxt_queue_insert_tail(&port->messages, &msg->link);
if (port->socket.write_ready) {
nxt_port_write_handler(task, port, NULL);
nxt_port_write_handler(task, &port->socket, NULL);
}
return NXT_OK;
@@ -236,14 +236,14 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
nxt_port_t *port;
struct iovec iov[NXT_IOBUF_MAX * 10];
nxt_queue_link_t *link;
nxt_port_method_t m;
nxt_port_send_msg_t *msg;
nxt_sendbuf_coalesce_t sb;
nxt_port_method_t m;
size_t plain_size;
nxt_buf_t *plain_buf;
port = obj;
port = nxt_container_of(obj, nxt_port_t, socket);
do {
link = nxt_queue_first(&port->messages);
@@ -389,7 +389,7 @@ nxt_port_read_handler(nxt_task_t *task, void *obj, void *data)
struct iovec iov[2];
nxt_port_recv_msg_t msg;
port = msg.port = obj;
port = msg.port = nxt_container_of(obj, nxt_port_t, socket);
for ( ;; ) {
@@ -522,5 +522,6 @@ nxt_port_buf_free(nxt_port_t *port, nxt_buf_t *b)
static void
nxt_port_error_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_debug(task, "port error handler %p", obj);
/* TODO */
}

4
src/nxt_process.c
View File

@@ -46,6 +46,8 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
process->pid = nxt_pid;
process->init->port->pid = nxt_pid;
rt->types = 0;
nxt_runtime_process_add(rt, process);
nxt_process_start(task, process->init);
@@ -94,7 +96,7 @@ nxt_process_start(nxt_task_t *task, nxt_process_init_t *process)
thread = task->thread;
rt = thread->runtime;
rt->type = process->type;
rt->types |= (1U << process->type);
engine = thread->engine;

5
src/nxt_queue.h
View File

@@ -222,12 +222,13 @@ NXT_EXPORT void nxt_queue_sort(nxt_queue_t *queue,
#define nxt_queue_each(elt, queue, type, link) \
do { \
nxt_queue_link_t *_lnk; \
nxt_queue_link_t *_lnk, *_nxt; \
\
for (_lnk = nxt_queue_first(queue); \
_lnk != nxt_queue_tail(queue); \
_lnk = nxt_queue_next(_lnk)) { \
_lnk = _nxt) { \
\
_nxt = nxt_queue_next(_lnk); \
elt = nxt_queue_link_data(_lnk, type, link); \
#define nxt_queue_loop \

1
src/nxt_router.c
View File

@@ -899,6 +899,7 @@ static nxt_port_handler_t nxt_router_app_port_handlers[] = {
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_router_app_data_handler,
nxt_port_remove_pid_handler,
};

32
src/nxt_runtime.c
View File

@@ -445,7 +445,7 @@ nxt_single_process_start(nxt_thread_t *thr, nxt_task_t *task, nxt_runtime_t *rt)
#endif
rt->type = NXT_PROCESS_SINGLE;
rt->types |= (1U << NXT_PROCESS_SINGLE);
nxt_runtime_listen_sockets_enable(task, rt);
@@ -479,7 +479,7 @@ nxt_runtime_quit(nxt_task_t *task)
#endif
if (rt->type == NXT_PROCESS_MASTER) {
if (nxt_runtime_is_master(rt)) {
nxt_master_stop_worker_processes(task, rt);
done = 0;
}
@@ -523,7 +523,8 @@ nxt_runtime_close_idle_connections(nxt_event_engine_t *engine)
static void
nxt_runtime_exit(nxt_task_t *task, void *obj, void *data)
{
nxt_runtime_t *rt;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_event_engine_t *engine;
rt = obj;
@@ -539,7 +540,7 @@ nxt_runtime_exit(nxt_task_t *task, void *obj, void *data)
#endif
if (rt->type <= NXT_PROCESS_MASTER) {
if (nxt_runtime_is_master(rt)) {
if (rt->pid_file != NULL) {
nxt_file_delete(rt->pid_file);
}
@@ -549,6 +550,14 @@ nxt_runtime_exit(nxt_task_t *task, void *obj, void *data)
nxt_event_engine_signals_stop(engine);
}
nxt_runtime_process_each(rt, process) {
nxt_runtime_process_remove(rt, process);
} nxt_runtime_process_loop;
nxt_mp_destroy(rt->mem_pool);
nxt_debug(task, "exit");
exit(0);
@@ -1642,6 +1651,7 @@ nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
} nxt_process_port_loop;
nxt_mp_free(rt->mem_pool, process);
break;
default:
@@ -1683,6 +1693,20 @@ nxt_runtime_port_remove(nxt_runtime_t *rt, nxt_port_t *port)
/* TODO lock ports */
nxt_port_hash_remove(&rt->ports, rt->mem_pool, port);
if (port->pair[0] != -1) {
nxt_fd_close(port->pair[0]);
}
if (port->pair[1] != -1) {
nxt_fd_close(port->pair[1]);
}
if (port->mem_pool) {
nxt_mp_destroy(port->mem_pool);
}
nxt_mp_free(port->process->mem_pool, port);
}

21
src/nxt_runtime.h
View File

@@ -45,7 +45,7 @@ struct nxt_runtime_s {
uint32_t last_engine_id;
nxt_process_type_t type;
uint32_t types; /* bitset of nxt_process_type_t */
nxt_timer_t timer;
@@ -84,6 +84,20 @@ nxt_int_t nxt_runtime_thread_pool_create(nxt_thread_t *thr, nxt_runtime_t *rt,
#endif
nxt_inline nxt_bool_t
nxt_runtime_is_type(nxt_runtime_t *rt, nxt_process_type_t type)
{
return (rt->types & (1U << type)) != 0;
}
nxt_inline nxt_bool_t
nxt_runtime_is_master(nxt_runtime_t *rt)
{
return nxt_runtime_is_type(rt, NXT_PROCESS_MASTER);
}
nxt_process_t *nxt_runtime_process_new(nxt_runtime_t *rt);
nxt_process_t *nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid);
@@ -141,10 +155,13 @@ void nxt_port_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
#define nxt_runtime_process_each(rt, process) \
do { \
nxt_lvlhsh_each_t _lhe; \
nxt_process_t *_nxt; \
\
for (process = nxt_runtime_process_first(rt, &_lhe); \
process != NULL; \
process = nxt_runtime_process_next(rt, &_lhe)) { \
process = _nxt) { \
\
_nxt = nxt_runtime_process_next(rt, &_lhe); \
#define nxt_runtime_process_loop \
} \

3
src/nxt_worker_process.c
View File

@@ -28,6 +28,7 @@ nxt_port_handler_t nxt_worker_process_port_handlers[] = {
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_port_data_handler,
nxt_port_remove_pid_handler,
};
@@ -37,6 +38,7 @@ nxt_port_handler_t nxt_app_process_port_handlers[] = {
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_port_app_data_handler,
nxt_port_remove_pid_handler,
};
@@ -46,6 +48,7 @@ nxt_port_handler_t nxt_router_process_port_handlers[] = {
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_router_conf_data_handler,
nxt_port_remove_pid_handler,
};