From 6a64533fa3b96bb64bfde4615e40376d65a292cb Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Wed, 4 Oct 2017 14:58:47 +0300 Subject: Introducing use counters for port and app. Thread safe port write. Use counter helps to simplify logic around port and application free. Port 'post' function introduced to simplify post execution of particular function to original port engine's thread. Write message queue is protected by mutex which makes port write operation thread safe. --- src/nxt_port.c | 129 +++++++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 116 insertions(+), 13 deletions(-) (limited to 'src/nxt_port.c') diff --git a/src/nxt_port.c b/src/nxt_port.c index d7f42012..3f7dc411 100644 --- a/src/nxt_port.c +++ b/src/nxt_port.c @@ -27,13 +27,15 @@ nxt_port_mp_cleanup(nxt_task_t *task, void *obj, void *data) nxt_assert(port->pair[0] == -1); nxt_assert(port->pair[1] == -1); - nxt_assert(port->app_stream == 0); + nxt_assert(port->use_count == 0); nxt_assert(port->app_link.next == NULL); nxt_assert(nxt_queue_is_empty(&port->messages)); nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_streams)); nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_peers)); + nxt_thread_mutex_destroy(&port->write_mutex); + nxt_mp_free(mp, port); } @@ -58,10 +60,12 @@ nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid, port->pid = pid; port->type = type; port->mem_pool = mp; + port->use_count = 1; nxt_mp_cleanup(mp, nxt_port_mp_cleanup, task, port, mp); nxt_queue_init(&port->messages); + nxt_thread_mutex_create(&port->write_mutex); } else { nxt_mp_destroy(mp); @@ -73,11 +77,11 @@ nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid, } -nxt_bool_t -nxt_port_release(nxt_port_t *port) +void +nxt_port_close(nxt_task_t *task, nxt_port_t *port) { - nxt_thread_log_debug("port %p %d:%d release, type %d", port, port->pid, - port->id, port->type); + nxt_debug(task, "port %p %d:%d close, type %d", port, port->pid, + port->id, port->type); if (port->pair[0] != -1) { nxt_fd_close(port->pair[0]); @@ -87,21 +91,31 @@ nxt_port_release(nxt_port_t *port) if (port->pair[1] != -1) { nxt_fd_close(port->pair[1]); port->pair[1] = -1; - } - if (port->type == NXT_PROCESS_WORKER) { - if (nxt_router_app_remove_port(port) == 0) { - return 0; + if (port->app != NULL) { + nxt_router_app_port_close(task, port); } } +} + + +static void +nxt_port_release(nxt_task_t *task, nxt_port_t *port) +{ + nxt_debug(task, "port %p %d:%d release, type %d", port, port->pid, + port->id, port->type); + + if (port->app != NULL) { + nxt_router_app_use(task, port->app, -1); + + port->app = NULL; + } if (port->link.next != NULL) { nxt_process_port_remove(port); } nxt_mp_release(port->mem_pool, NULL); - - return 1; } @@ -263,7 +277,9 @@ nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) port->socket.task = task; - nxt_runtime_port_add(rt, port); + nxt_runtime_port_add(task, port); + + nxt_port_use(task, port, -1); nxt_port_write_enable(task, port); @@ -434,7 +450,7 @@ nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) process = nxt_runtime_process_find(rt, pid); if (process) { - nxt_runtime_process_remove(rt, process); + nxt_runtime_process_remove(task, process); } } @@ -444,3 +460,90 @@ nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) { nxt_debug(task, "port empty handler"); } + + +typedef struct { + nxt_work_t work; + nxt_port_t *port; + nxt_port_post_handler_t handler; +} nxt_port_work_t; + + +static void +nxt_port_post_handler(nxt_task_t *task, void *obj, void *data) +{ + nxt_port_t *port; + nxt_port_work_t *pw; + nxt_port_post_handler_t handler; + + pw = obj; + port = pw->port; + handler = pw->handler; + + nxt_free(pw); + + handler(task, port, data); + + nxt_port_use(task, port, -1); +} + + +nxt_int_t +nxt_port_post(nxt_task_t *task, nxt_port_t *port, + nxt_port_post_handler_t handler, void *data) +{ + nxt_port_work_t *pw; + + if (task->thread->engine == port->engine) { + handler(task, port, data); + + return NXT_OK; + } + + pw = nxt_zalloc(sizeof(nxt_port_work_t)); + + if (nxt_slow_path(pw == NULL)) { + return NXT_ERROR; + } + + nxt_atomic_fetch_add(&port->use_count, 1); + + pw->work.handler = nxt_port_post_handler; + pw->work.task = &port->engine->task; + pw->work.obj = pw; + pw->work.data = data; + + pw->port = port; + pw->handler = handler; + + nxt_event_engine_post(port->engine, &pw->work); + + return NXT_OK; +} + + +static void +nxt_port_release_handler(nxt_task_t *task, nxt_port_t *port, void *data) +{ + /* no op */ +} + + +void +nxt_port_use(nxt_task_t *task, nxt_port_t *port, int i) +{ + int c; + + c = nxt_atomic_fetch_add(&port->use_count, i); + + if (i < 0 && c == -i) { + + if (task->thread->engine == port->engine) { + nxt_port_release(task, port); + + return; + } + + nxt_port_post(task, port, nxt_port_release_handler, NULL); + } +} -- cgit