diff options
Diffstat (limited to '')
| -rw-r--r-- | src/nxt_master_process.c | 6 | ||||
| -rw-r--r-- | src/nxt_port.c | 34 | ||||
| -rw-r--r-- | src/nxt_port.h | 2 | ||||
| -rw-r--r-- | src/nxt_port_memory.c | 2 | ||||
| -rw-r--r-- | src/nxt_process.c | 2 | ||||
| -rw-r--r-- | src/nxt_process.h | 1 | ||||
| -rw-r--r-- | src/nxt_runtime.c | 124 | ||||
| -rw-r--r-- | src/nxt_runtime.h | 3 |
8 files changed, 106 insertions, 68 deletions
diff --git a/src/nxt_master_process.c b/src/nxt_master_process.c index 2bc97c1c..41e659a4 100644 --- a/src/nxt_master_process.c +++ b/src/nxt_master_process.c @@ -406,7 +406,7 @@ nxt_master_create_worker_process(nxt_task_t *task, nxt_runtime_t *rt, port = nxt_port_new(0, 0, init->type); if (nxt_slow_path(port == NULL)) { - nxt_runtime_process_destroy(rt, process); + nxt_runtime_process_remove(rt, process); return NXT_ERROR; } @@ -658,7 +658,9 @@ nxt_master_cleanup_worker_process(nxt_task_t *task, nxt_pid_t pid) if (!nxt_exiting) { nxt_runtime_process_each(rt, process) { - if (process->pid == nxt_pid) { + if (process->pid == nxt_pid || + process->pid == pid || + nxt_queue_is_empty(&process->ports)) { continue; } diff --git a/src/nxt_port.c b/src/nxt_port.c index 18dc4121..2e5b229b 100644 --- a/src/nxt_port.c +++ b/src/nxt_port.c @@ -106,28 +106,6 @@ nxt_port_enable(nxt_task_t *task, nxt_port_t *port, } -void -nxt_port_write(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t type, - nxt_fd_t fd, uint32_t stream, nxt_buf_t *b) -{ - 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, type, - fd, stream, 0, b); - - } nxt_process_port_loop; - } - } - nxt_runtime_process_loop; -} - - static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) { @@ -276,6 +254,8 @@ nxt_port_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) rt = task->thread->runtime; + nxt_assert(nxt_runtime_is_master(rt)); + process = nxt_runtime_process_get(rt, msg->port_msg.pid); if (nxt_slow_path(process == NULL)) { return; @@ -290,9 +270,7 @@ nxt_port_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) nxt_debug(task, "process %PI ready", msg->port_msg.pid); - if (nxt_runtime_is_master(rt)) { - nxt_port_send_new_port(task, rt, port, msg->port_msg.stream); - } + nxt_port_send_new_port(task, rt, port, msg->port_msg.stream); } @@ -310,7 +288,7 @@ nxt_port_mmap_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) return; } - process = nxt_runtime_process_get(rt, msg->port_msg.pid); + process = nxt_runtime_process_find(rt, msg->port_msg.pid); if (nxt_slow_path(process == NULL)) { nxt_log(task, NXT_LOG_WARN, "failed to get process #%PI", msg->port_msg.pid); @@ -415,14 +393,14 @@ nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) nxt_runtime_t *rt; nxt_process_t *process; - nxt_debug(task, "port remove pid handler"); - buf = msg->buf; nxt_assert(nxt_buf_used_size(buf) == sizeof(pid)); nxt_memcpy(&pid, buf->mem.pos, sizeof(pid)); + nxt_debug(task, "port remove pid %PI handler", pid); + rt = task->thread->runtime; nxt_port_rpc_remove_peer(task, msg->port, pid); diff --git a/src/nxt_port.h b/src/nxt_port.h index 8a5c3390..a7f78b4c 100644 --- a/src/nxt_port.h +++ b/src/nxt_port.h @@ -160,8 +160,6 @@ nxt_int_t nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, void nxt_port_enable(nxt_task_t *task, nxt_port_t *port, nxt_port_handler_t *handlers); -void nxt_port_write(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t type, - nxt_fd_t fd, uint32_t stream, nxt_buf_t *b); void nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt, nxt_port_t *port, uint32_t stream); nxt_int_t nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, diff --git a/src/nxt_port_memory.c b/src/nxt_port_memory.c index d21d263a..ec3227b1 100644 --- a/src/nxt_port_memory.c +++ b/src/nxt_port_memory.c @@ -379,7 +379,7 @@ nxt_port_get_port_incoming_mmap(nxt_task_t *task, nxt_pid_t spid, uint32_t id) nxt_port_mmap_t *port_mmap; nxt_port_mmap_header_t *hdr; - process = nxt_runtime_process_get(task->thread->runtime, spid); + process = nxt_runtime_process_find(task->thread->runtime, spid); if (nxt_slow_path(process == NULL)) { return NULL; } diff --git a/src/nxt_process.c b/src/nxt_process.c index 9ab3eaf3..e6a6fa0c 100644 --- a/src/nxt_process.c +++ b/src/nxt_process.c @@ -577,7 +577,7 @@ nxt_process_port_mp_cleanup(nxt_task_t *task, void *obj, void *data) process->port_cleanups--; if (process->port_cleanups == 0) { - nxt_runtime_process_destroy(rt, process); + nxt_runtime_process_remove(rt, process); } } diff --git a/src/nxt_process.h b/src/nxt_process.h index aa3aa7a5..b76c4f1a 100644 --- a/src/nxt_process.h +++ b/src/nxt_process.h @@ -57,6 +57,7 @@ typedef struct { nxt_pid_t pid; nxt_queue_t ports; /* of nxt_port_t */ nxt_bool_t ready; + nxt_bool_t registered; nxt_uint_t port_cleanups; nxt_process_init_t *init; diff --git a/src/nxt_runtime.c b/src/nxt_runtime.c index 31829225..5f4ec6f4 100644 --- a/src/nxt_runtime.c +++ b/src/nxt_runtime.c @@ -48,6 +48,11 @@ static void nxt_runtime_thread_pool_destroy(nxt_task_t *task, nxt_runtime_t *rt, nxt_runtime_cont_t cont); #endif +static void nxt_runtime_process_destroy(nxt_runtime_t *rt, + nxt_process_t *process); +static nxt_process_t *nxt_runtime_process_remove_pid(nxt_runtime_t *rt, + nxt_pid_t pid); + nxt_int_t nxt_runtime_create(nxt_task_t *task) @@ -71,6 +76,8 @@ nxt_runtime_create(nxt_task_t *task) task->thread->runtime = rt; rt->mem_pool = mp; + nxt_thread_mutex_create(&rt->processes_mutex); + rt->prefix = nxt_current_directory(mp); if (nxt_slow_path(rt->prefix == NULL)) { goto fail; @@ -557,6 +564,8 @@ nxt_runtime_exit(nxt_task_t *task, void *obj, void *data) } nxt_runtime_process_loop; + nxt_thread_mutex_destroy(&rt->processes_mutex); + nxt_mp_destroy(rt->mem_pool); nxt_debug(task, "exit"); @@ -1488,10 +1497,11 @@ nxt_runtime_process_new(nxt_runtime_t *rt) } -void +static void nxt_runtime_process_destroy(nxt_runtime_t *rt, nxt_process_t *process) { nxt_assert(process->port_cleanups == 0); + nxt_assert(process->registered == 0); nxt_port_mmaps_destroy(process->incoming, 1); nxt_port_mmaps_destroy(process->outgoing, 1); @@ -1533,23 +1543,38 @@ static const nxt_lvlhsh_proto_t lvlhsh_processes_proto nxt_aligned(64) = { }; +nxt_inline void +nxt_runtime_process_lhq_pid(nxt_lvlhsh_query_t *lhq, nxt_pid_t *pid) +{ + lhq->key_hash = nxt_murmur_hash2(pid, sizeof(*pid)); + lhq->key.length = sizeof(*pid); + lhq->key.start = (u_char *) pid; + lhq->proto = &lvlhsh_processes_proto; +} + + nxt_process_t * nxt_runtime_process_find(nxt_runtime_t *rt, nxt_pid_t pid) { + nxt_process_t *process; nxt_lvlhsh_query_t lhq; - lhq.key_hash = nxt_murmur_hash2(&pid, sizeof(pid)); - lhq.key.length = sizeof(pid); - lhq.key.start = (u_char *) &pid; - lhq.proto = &lvlhsh_processes_proto; + process = NULL; + + nxt_runtime_process_lhq_pid(&lhq, &pid); + + nxt_thread_mutex_lock(&rt->processes_mutex); if (nxt_lvlhsh_find(&rt->processes, &lhq) == NXT_OK) { - return lhq.value; + process = lhq.value; + + } else { + nxt_thread_log_debug("process %PI not found", pid); } - nxt_thread_log_debug("process %PI not found", pid); + nxt_thread_mutex_unlock(&rt->processes_mutex); - return NULL; + return process; } @@ -1559,13 +1584,14 @@ nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid) nxt_process_t *process; nxt_lvlhsh_query_t lhq; - lhq.key_hash = nxt_murmur_hash2(&pid, sizeof(pid)); - lhq.key.length = sizeof(pid); - lhq.key.start = (u_char *) &pid; - lhq.proto = &lvlhsh_processes_proto; + nxt_runtime_process_lhq_pid(&lhq, &pid); + + nxt_thread_mutex_lock(&rt->processes_mutex); if (nxt_lvlhsh_find(&rt->processes, &lhq) == NXT_OK) { nxt_thread_log_debug("process %PI found", pid); + + nxt_thread_mutex_unlock(&rt->processes_mutex); return lhq.value; } @@ -1589,6 +1615,8 @@ nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid) rt->nprocesses++; + process->registered = 1; + nxt_thread_log_debug("process %PI insert", pid); break; @@ -1597,6 +1625,8 @@ nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid) break; } + nxt_thread_mutex_unlock(&rt->processes_mutex); + return process; } @@ -1607,14 +1637,16 @@ nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process) nxt_port_t *port; nxt_lvlhsh_query_t lhq; - lhq.key_hash = nxt_murmur_hash2(&process->pid, sizeof(process->pid)); - lhq.key.length = sizeof(process->pid); - lhq.key.start = (u_char *) &process->pid; - lhq.proto = &lvlhsh_processes_proto; + nxt_assert(process->registered == 0); + + nxt_runtime_process_lhq_pid(&lhq, &process->pid); + lhq.replace = 0; lhq.value = process; lhq.pool = rt->mem_pool; + nxt_thread_mutex_lock(&rt->processes_mutex); + switch (nxt_lvlhsh_insert(&rt->processes, &lhq)) { case NXT_OK: @@ -1632,37 +1664,70 @@ nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process) } nxt_process_port_loop; + process->registered = 1; + + nxt_thread_log_debug("process %PI added", process->pid); break; default: + nxt_thread_log_debug("process %PI failed to add", process->pid); break; } + + nxt_thread_mutex_unlock(&rt->processes_mutex); } -void -nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process) +static nxt_process_t * +nxt_runtime_process_remove_pid(nxt_runtime_t *rt, nxt_pid_t pid) { - nxt_port_t *port; + nxt_process_t *process; nxt_lvlhsh_query_t lhq; - lhq.key_hash = nxt_murmur_hash2(&process->pid, sizeof(process->pid)); - lhq.key.length = sizeof(process->pid); - lhq.key.start = (u_char *) &process->pid; - lhq.proto = &lvlhsh_processes_proto; - lhq.replace = 0; - lhq.value = process; + process = NULL; + + nxt_runtime_process_lhq_pid(&lhq, &pid); + lhq.pool = rt->mem_pool; + nxt_thread_mutex_lock(&rt->processes_mutex); + switch (nxt_lvlhsh_delete(&rt->processes, &lhq)) { case NXT_OK: rt->nprocesses--; - if (process->port_cleanups == 0) { - nxt_runtime_process_destroy(rt, process); + process = lhq.value; + + process->registered = 0; + + nxt_thread_log_debug("process %PI removed", pid); + break; + + default: + nxt_thread_log_debug("process %PI remove failed", pid); + break; + } + + nxt_thread_mutex_unlock(&rt->processes_mutex); + + return process; +} + + +void +nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process) +{ + nxt_port_t *port; + + if (process->port_cleanups == 0) { + if (process->registered == 1) { + nxt_runtime_process_remove_pid(rt, process->pid); } + nxt_runtime_process_destroy(rt, process); + + } else { nxt_process_port_each(process, port) { nxt_runtime_port_remove(rt, port); @@ -1670,11 +1735,6 @@ nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process) nxt_port_release(port); } nxt_process_port_loop; - - break; - - default: - break; } } diff --git a/src/nxt_runtime.h b/src/nxt_runtime.h index 8ee8560b..7922c2ef 100644 --- a/src/nxt_runtime.h +++ b/src/nxt_runtime.h @@ -37,6 +37,7 @@ struct nxt_runtime_s { nxt_process_t *mprocess; size_t nprocesses; + nxt_thread_mutex_t processes_mutex; nxt_lvlhsh_t processes; /* of nxt_process_t */ nxt_port_t *port_by_type[NXT_PROCESS_MAX]; @@ -101,8 +102,6 @@ nxt_runtime_is_master(nxt_runtime_t *rt) nxt_process_t *nxt_runtime_process_new(nxt_runtime_t *rt); -void nxt_runtime_process_destroy(nxt_runtime_t *rt, nxt_process_t *process); - nxt_process_t *nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid); void nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process); |