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.

1 files changed, 116 insertions, 13 deletions

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);
+    }
+}