Port allocation and destroy changed. Worker process stop introduced.

src/nxt_application.h

@@ -100,11 +100,13 @@ typedef struct {
 } nxt_app_request_t;
 
 
-typedef struct {
+typedef struct nxt_app_parse_ctx_s nxt_app_parse_ctx_t;
+
+struct nxt_app_parse_ctx_s {
     nxt_app_request_t         r;
     nxt_http_request_parse_t  parser;
     nxt_mp_t                  *mem_pool;
-} nxt_app_parse_ctx_t;
+};
 
 
 nxt_int_t nxt_app_http_req_init(nxt_task_t *task, nxt_app_parse_ctx_t *ctx);

src/nxt_conn.h

@@ -183,15 +183,17 @@ struct nxt_conn_s {
 
 
 typedef uint32_t nxt_req_id_t;
+typedef struct nxt_app_parse_ctx_s nxt_app_parse_ctx_t;
 
 typedef struct {
-    nxt_req_id_t      req_id;
+    nxt_req_id_t         req_id;
-    nxt_conn_t        *conn;
+    nxt_conn_t           *conn;
-    nxt_port_t        *app_port;
+    nxt_port_t           *app_port;
-    nxt_port_t        *reply_port;
+    nxt_port_t           *reply_port;
+    nxt_app_parse_ctx_t  *ap;
 
-    nxt_queue_link_t  link;     /* for nxt_conn_t.requests */
+    nxt_queue_link_t     link;     /* for nxt_conn_t.requests */
-    nxt_queue_link_t  app_link; /* for nxt_app_t.requests */
+    nxt_queue_link_t     app_link; /* for nxt_app_t.requests */
 } nxt_req_conn_link_t;
 
 

src/nxt_event_engine.c

@@ -38,6 +38,8 @@ nxt_event_engine_create(nxt_task_t *task,
         return NULL;
     }
 
+    nxt_debug(task, "create engine %p", engine);
+
     thread = task->thread;
 
     engine->task.thread = thread;
@@ -440,6 +442,8 @@ nxt_event_engine_change(nxt_event_engine_t *engine,
 void
 nxt_event_engine_free(nxt_event_engine_t *engine)
 {
+    nxt_thread_log_debug("free engine %p", engine);
+
     nxt_event_engine_signal_pipe_free(engine);
     nxt_free(engine->signals);
 

src/nxt_master_process.c

@@ -207,12 +207,13 @@ nxt_master_process_port_create(nxt_task_t *task, nxt_runtime_t *rt)
         return NXT_ERROR;
     }
 
-    port = nxt_process_port_new(rt, process, nxt_port_get_next_id(),
+    port = nxt_port_new(0, nxt_pid, NXT_PROCESS_MASTER);
-                                NXT_PROCESS_MASTER);
     if (nxt_slow_path(port == NULL)) {
         return NXT_ERROR;
     }
 
+    nxt_process_port_add(process, port);
+
     ret = nxt_port_socket_init(task, port, 0);
     if (nxt_slow_path(ret != NXT_OK)) {
         return ret;
@@ -373,15 +374,17 @@ nxt_master_create_worker_process(nxt_task_t *task, nxt_runtime_t *rt,
 
     process->init = init;
 
-    port = nxt_process_port_new(rt, process, 0, init->type);
+    port = nxt_port_new(0, 0, init->type);
     if (nxt_slow_path(port == NULL)) {
         nxt_runtime_process_destroy(rt, process);
         return NXT_ERROR;
     }
 
+    nxt_process_port_add(process, port);
+
     ret = nxt_port_socket_init(task, port, 0);
     if (nxt_slow_path(ret != NXT_OK)) {
-        nxt_mp_free(rt->mem_pool, port);
+        nxt_mp_release(port->mem_pool, port);
         nxt_runtime_process_destroy(rt, process);
         return ret;
     }

src/nxt_port.c

@@ -7,11 +7,72 @@
 #include <nxt_main.h>
 #include <nxt_runtime.h>
 #include <nxt_port.h>
+#include <nxt_router.h>
 
 
 static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 
-static nxt_atomic_uint_t nxt_port_last_id;
+static nxt_atomic_uint_t nxt_port_last_id = 1;
+
+
+nxt_port_t *
+nxt_port_new(nxt_port_id_t id, nxt_pid_t pid, nxt_process_type_t type)
+{
+    nxt_mp_t    *mp;
+    nxt_port_t  *port;
+
+    mp = nxt_mp_create(1024, 128, 256, 32);
+
+    if (nxt_slow_path(mp == NULL)) {
+        return NULL;
+    }
+
+    port = nxt_mp_zalloc(mp, sizeof(nxt_port_t));
+
+    if (nxt_fast_path(port != NULL)) {
+        port->id = id;
+        port->pid = pid;
+        port->type = type;
+        port->mem_pool = mp;
+
+        nxt_queue_init(&port->messages);
+
+    } else {
+        nxt_mp_destroy(mp);
+    }
+
+    return port;
+}
+
+
+nxt_bool_t
+nxt_port_release(nxt_port_t *port)
+{
+    if (port->pair[0] != -1) {
+        nxt_fd_close(port->pair[0]);
+        port->pair[0] = -1;
+    }
+
+    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->link.next != NULL) {
+        nxt_process_port_remove(port);
+    }
+
+    nxt_mp_release(port->mem_pool, port);
+
+    return 1;
+}
+
 
 nxt_port_id_t
 nxt_port_get_next_id()
@@ -32,7 +93,6 @@ nxt_port_enable(nxt_task_t *task, nxt_port_t *port,
     nxt_port_handler_t *handlers)
 {
     port->pid = nxt_pid;
-    port->engine = task->thread->engine;
     port->handler = nxt_port_handler;
     port->data = handlers;
 
@@ -151,7 +211,6 @@ nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port,
 void
 nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
-    nxt_mp_t                 *mp;
     nxt_port_t               *port;
     nxt_process_t            *process;
     nxt_runtime_t            *rt;
@@ -180,26 +239,19 @@ nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
         return;
     }
 
-    port = nxt_process_port_new(rt, process, new_port_msg->id,
+    port = nxt_port_new(new_port_msg->id, new_port_msg->pid,
-                                new_port_msg->type);
+                        new_port_msg->type);
     if (nxt_slow_path(port == NULL)) {
         return;
     }
 
-    mp = nxt_mp_create(1024, 128, 256, 32);
+    nxt_process_port_add(process, port);
-    if (nxt_slow_path(mp == NULL)) {
-        return;
-    }
-
-    port->mem_pool = mp;
 
     port->pair[0] = -1;
     port->pair[1] = msg->fd;
     port->max_size = new_port_msg->max_size;
     port->max_share = new_port_msg->max_share;
 
-    nxt_queue_init(&port->messages);
-
     port->socket.task = task;
 
     nxt_runtime_port_add(rt, port);

src/nxt_port.h

@@ -88,6 +88,7 @@ struct nxt_port_s {
     nxt_pid_t           pid;
 
     nxt_process_type_t  type;
+    nxt_work_t          work;
 };
 
 
@@ -110,6 +111,10 @@ typedef union {
 } nxt_port_data_t;
 
 
+nxt_port_t *nxt_port_new(nxt_port_id_t id, nxt_pid_t pid,
+    nxt_process_type_t type);
+nxt_bool_t nxt_port_release(nxt_port_t *port);
+
 nxt_port_id_t nxt_port_get_next_id(void);
 void nxt_port_reset_next_id(void);
 

src/nxt_port_socket.c

@@ -19,7 +19,6 @@ static void nxt_port_error_handler(nxt_task_t *task, void *obj, void *data);
 nxt_int_t
 nxt_port_socket_init(nxt_task_t *task, nxt_port_t *port, size_t max_size)
 {
-    nxt_mp_t      *mp;
     nxt_int_t     sndbuf, rcvbuf, size;
     nxt_socket_t  snd, rcv;
 
@@ -28,15 +27,6 @@ nxt_port_socket_init(nxt_task_t *task, nxt_port_t *port, size_t max_size)
     port->pair[0] = -1;
     port->pair[1] = -1;
 
-    nxt_queue_init(&port->messages);
-
-    mp = nxt_mp_create(1024, 128, 256, 32);
-    if (nxt_slow_path(mp == NULL)) {
-        return NXT_ERROR;
-    }
-
-    port->mem_pool = mp;
-
     if (nxt_slow_path(nxt_socketpair_create(task, port->pair) != NXT_OK)) {
         goto socketpair_fail;
     }
@@ -100,8 +90,6 @@ getsockopt_fail:
 
 socketpair_fail:
 
-    nxt_mp_destroy(port->mem_pool);
-
     return NXT_ERROR;
 }
 
@@ -172,14 +160,9 @@ nxt_int_t
 nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
     nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port, nxt_buf_t *b)
 {
-    nxt_queue_link_t     *link;
     nxt_port_send_msg_t  *msg;
 
-    for (link = nxt_queue_first(&port->messages);
+    nxt_queue_each(msg, &port->messages, nxt_port_send_msg_t, link) {
-         link != nxt_queue_tail(&port->messages);
-         link = nxt_queue_next(link))
-    {
-        msg = (nxt_port_send_msg_t *) link;
 
         if  (msg->port_msg.stream == stream && 
              msg->port_msg.reply_port == reply_port) {
@@ -191,7 +174,8 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
 
             return NXT_OK;
         }
-    }
+
+    } nxt_queue_loop;
 
     msg = nxt_mp_retain(port->mem_pool, sizeof(nxt_port_send_msg_t));
     if (nxt_slow_path(msg == NULL)) {
@@ -256,7 +240,7 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
             return;
         }
 
-        msg = (nxt_port_send_msg_t *) link;
+        msg = nxt_queue_link_data(link, nxt_port_send_msg_t, link);
 
         iov[0].iov_base = &msg->port_msg;
         iov[0].iov_len = sizeof(nxt_port_msg_t);

src/nxt_process.c

@@ -563,30 +563,11 @@ nxt_user_cred_set(nxt_task_t *task, nxt_user_cred_t *uc)
 }
 
 
-nxt_port_t *
+void
-nxt_process_port_new(nxt_runtime_t *rt, nxt_process_t *process,
+nxt_process_port_add(nxt_process_t *process, nxt_port_t *port)
-    nxt_port_id_t id, nxt_process_type_t type)
 {
-    size_t      size;
+    port->process = process;
-    nxt_port_t  *port;
+    nxt_queue_insert_tail(&process->ports, &port->link);
-
-    size = sizeof(nxt_port_t);
-    if (size == NXT_PROCESS_WORKER) {
-        size += sizeof(nxt_work_t);
-    }
-
-    port = nxt_mp_zalloc(rt->mem_pool, size);
-
-    if (nxt_fast_path(port != NULL)) {
-        port->id = id;
-        port->pid = process->pid;
-        port->process = process;
-        port->type = type;
-
-        nxt_process_port_add(process, port);
-    }
-
-    return port;
 }
 
 

src/nxt_process.h

@@ -82,17 +82,13 @@ NXT_EXPORT void nxt_nanosleep(nxt_nsec_t ns);
 NXT_EXPORT void nxt_process_arguments(nxt_task_t *task, char **orig_argv,
     char ***orig_envp);
 
-NXT_EXPORT nxt_port_t * nxt_process_port_new(nxt_runtime_t *rt,
-    nxt_process_t *process, nxt_port_id_t id, nxt_process_type_t type);
-
 #define nxt_process_port_remove(port)                                         \
     nxt_queue_remove(&port->link)
 
 #define nxt_process_port_first(process)                                       \
     nxt_queue_link_data(nxt_queue_first(&process->ports), nxt_port_t, link)
 
-#define nxt_process_port_add(process, port)                                   \
+NXT_EXPORT void nxt_process_port_add(nxt_process_t *process, nxt_port_t *port);
-    nxt_queue_insert_tail(&process->ports, &port->link)
 
 #define nxt_process_port_each(process, port)                                   \
     nxt_queue_each(port, &process->ports, nxt_port_t, link)

src/nxt_router.c

@@ -99,6 +99,7 @@ static void nxt_router_thread_exit_handler(nxt_task_t *task, void *obj,
 static void nxt_router_conf_release(nxt_task_t *task,
     nxt_socket_conf_joint_t *joint);
 
+static nxt_bool_t nxt_router_app_free(nxt_app_t *app);
 static nxt_port_t * nxt_router_app_get_port(nxt_app_t *app);
 static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
     void *data);
@@ -189,6 +190,12 @@ nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
     if (nxt_fast_path(sw != NULL)) {
         msg->new_port->app = sw->app;
 
+        sw->app->workers++;
+
+        nxt_assert(sw->app->pending_workers != 0);
+
+        sw->app->pending_workers--;
+
         nxt_router_app_release_port(task, msg->new_port, sw->app);
 
         sw->work.handler = nxt_router_sw_release;
@@ -754,25 +761,25 @@ nxt_router_listener_application(nxt_router_temp_conf_t *tmcf, nxt_str_t *name)
 static nxt_socket_conf_t *
 nxt_router_socket_conf(nxt_task_t *task, nxt_mp_t *mp, nxt_sockaddr_t *sa)
 {
-    nxt_socket_conf_t  *conf;
+    nxt_socket_conf_t  *skcf;
 
-    conf = nxt_mp_zget(mp, sizeof(nxt_socket_conf_t));
+    skcf = nxt_mp_zget(mp, sizeof(nxt_socket_conf_t));
-    if (nxt_slow_path(conf == NULL)) {
+    if (nxt_slow_path(skcf == NULL)) {
         return NULL;
     }
 
-    conf->sockaddr = sa;
+    skcf->sockaddr = sa;
 
-    conf->listen.sockaddr = sa;
+    skcf->listen.sockaddr = sa;
-    conf->listen.socklen = sa->socklen;
+    skcf->listen.socklen = sa->socklen;
-    conf->listen.address_length = sa->length;
+    skcf->listen.address_length = sa->length;
 
-    conf->listen.socket = -1;
+    skcf->listen.socket = -1;
-    conf->listen.backlog = NXT_LISTEN_BACKLOG;
+    skcf->listen.backlog = NXT_LISTEN_BACKLOG;
-    conf->listen.flags = NXT_NONBLOCK;
+    skcf->listen.flags = NXT_NONBLOCK;
-    conf->listen.read_after_accept = 1;
+    skcf->listen.read_after_accept = 1;
 
-    return conf;
+    return skcf;
 }
 
 
@@ -1179,17 +1186,33 @@ nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
 static void
 nxt_router_apps_sort(nxt_router_t *router, nxt_router_temp_conf_t *tmcf)
 {
-    nxt_app_t  *app;
+    nxt_app_t   *app;
+    nxt_port_t  *port;
 
     nxt_queue_each(app, &router->apps, nxt_app_t, link) {
 
         nxt_queue_remove(&app->link);
 
-        // TODO RELEASE APP
+        app->live = 0;
-#if 0
+
-        nxt_thread_mutex_destroy(&app->mutex);
+        if (nxt_router_app_free(app) != 0) {
-        nxt_free(app);
+            continue;
-#endif
+        }
+
+        if (nxt_queue_is_empty(&app->requests)) {
+
+            do {
+                port = nxt_router_app_get_port(app);
+                if (port == NULL) {
+                    break;
+                }
+
+                nxt_port_socket_write(&port->engine->task, port,
+                                      NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
+            } while (1);
+
+        }
+
     } nxt_queue_loop;
 
     nxt_queue_add(&router->apps, &tmcf->previous);
@@ -1266,21 +1289,17 @@ nxt_router_thread_start(void *data)
 
     engine->mem_pool = nxt_mp_create(4096, 128, 1024, 64);
 
-    port = nxt_mp_zalloc(engine->mem_pool, sizeof(nxt_port_t));
+    port = nxt_port_new(nxt_port_get_next_id(), nxt_pid, NXT_PROCESS_ROUTER);
     if (nxt_slow_path(port == NULL)) {
         return;
     }
 
-    port->id = nxt_port_get_next_id();
-    port->pid = nxt_pid;
-
     ret = nxt_port_socket_init(task, port, 0);
     if (nxt_slow_path(ret != NXT_OK)) {
+        nxt_mp_release(port->mem_pool, port);
         return;
     }
 
-    port->type = NXT_PROCESS_ROUTER;
-
     engine->port = port;
 
     nxt_port_enable(task, port, nxt_router_app_port_handlers);
@@ -1391,6 +1410,9 @@ nxt_router_listen_socket_delete(nxt_task_t *task, void *obj, void *data)
 
     nxt_fd_event_delete(engine, &listen->socket);
 
+    nxt_debug(task, "engine %p: listen socket delete: %d", engine,
+              listen->socket.fd);
+
     listen->timer.handler = nxt_router_listen_socket_close;
     listen->timer.work_queue = &engine->fast_work_queue;
 
@@ -1414,6 +1436,9 @@ nxt_router_listen_socket_close(nxt_task_t *task, void *obj, void *data)
     listen = nxt_timer_data(timer, nxt_listen_event_t, timer);
     joint = listen->socket.data;
 
+    nxt_debug(task, "engine %p: listen socket close: %d", task->thread->engine,
+              listen->socket.fd);
+
     nxt_queue_remove(&listen->link);
 
     /* 'task' refers to listen->task and we cannot use after nxt_free() */
@@ -1439,6 +1464,9 @@ nxt_router_listen_socket_release(nxt_task_t *task,
 
     nxt_thread_spin_lock(lock);
 
+    nxt_debug(task, "engine %p: listen socket release: rtsk->count %D",
+              task->thread->engine, rtsk->count);
+
     if (--rtsk->count != 0) {
         rtsk = NULL;
     }
@@ -1463,7 +1491,7 @@ nxt_router_conf_release(nxt_task_t *task, nxt_socket_conf_joint_t *joint)
     nxt_router_conf_t      *rtcf;
     nxt_thread_spinlock_t  *lock;
 
-    nxt_debug(task, "conf joint count: %D", joint->count);
+    nxt_debug(task, "conf joint %p count: %D", joint, joint->count);
 
     if (--joint->count != 0) {
         return;
@@ -1477,6 +1505,9 @@ nxt_router_conf_release(nxt_task_t *task, nxt_socket_conf_joint_t *joint)
 
     nxt_thread_spin_lock(lock);
 
+    nxt_debug(task, "conf skcf %p: %D, rtcf %p: %D", skcf, skcf->count,
+              rtcf, rtcf->count);
+
     if (--skcf->count != 0) {
         rtcf = NULL;
 
@@ -1531,18 +1562,10 @@ nxt_router_thread_exit_handler(nxt_task_t *task, void *obj, void *data)
 
     // TODO notify all apps
 
-    if (port->pair[0] != -1) {
+    nxt_mp_thread_adopt(port->mem_pool);
-        nxt_fd_close(port->pair[0]);
+    nxt_port_release(port);
-    }
-
-    if (port->pair[1] != -1) {
-        nxt_fd_close(port->pair[1]);
-    }
-
-    if (port->mem_pool) {
-        nxt_mp_destroy(port->mem_pool);
-    }
 
+    nxt_mp_thread_adopt(engine->mem_pool);
     nxt_mp_destroy(engine->mem_pool);
 
     nxt_event_engine_free(engine);
@@ -1683,17 +1706,17 @@ nxt_router_text_by_code(int code)
     }
 }
 
-static void
-nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
-    const char* fmt, ...)
-{
-    va_list             args;
-    nxt_buf_t           *b, *last;
-    const char          *msg;
 
-    b = nxt_buf_mem_alloc(c->mem_pool, 16384, 0);
+static nxt_buf_t *
+nxt_router_get_error_buf(nxt_task_t *task, nxt_mp_t *mp, int code,
+    const char* fmt, va_list args)
+{
+    nxt_buf_t   *b, *last;
+    const char  *msg;
+
+    b = nxt_buf_mem_ts_alloc(task, mp, 16384);
     if (nxt_slow_path(b == NULL)) {
-        /* TODO pogorevaTb */
+        return NULL;
     }
 
     b->mem.free = nxt_sprintf(b->mem.free, b->mem.end,
@@ -1704,19 +1727,38 @@ nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
 
     msg = (const char *) b->mem.free;
 
-    va_start(args, fmt);
     b->mem.free = nxt_vsprintf(b->mem.free, b->mem.end, fmt, args);
-    va_end(args);
 
     nxt_log_alert(task->log, "error %d: %s", code, msg);
 
-    last = nxt_buf_sync_alloc(c->mem_pool, NXT_BUF_SYNC_LAST);
+    last = nxt_buf_mem_ts_alloc(task, mp, 0);
+
     if (nxt_slow_path(last == NULL)) {
-        /* TODO pogorevaTb */
+        nxt_mp_release(mp, b);
+        return NULL;
     }
 
+    nxt_buf_set_sync(last);
+    nxt_buf_set_last(last);
+
     nxt_buf_chain_add(&b, last);
 
+    return b;
+}
+
+
+
+static void
+nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
+    const char* fmt, ...)
+{
+    va_list    args;
+    nxt_buf_t  *b;
+
+    va_start(args, fmt);
+    b = nxt_router_get_error_buf(task, c->mem_pool, code, fmt, args);
+    va_end(args);
+
     if (c->write == NULL) {
         c->write = b;
         c->write_state = &nxt_router_conn_write_state;
@@ -1742,6 +1784,19 @@ nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
     sw = obj;
     app = sw->app;
 
+    if (app->workers + app->pending_workers >= app->max_workers) {
+        sw->work.handler = nxt_router_sw_release;
+
+        nxt_debug(task, "%uD/%uD running/penging workers, post sw #%uxD "
+                  "release to %p", sw->stream, sw->work.data);
+
+        nxt_event_engine_post(sw->work.data, &sw->work);
+
+        return;
+    }
+
+    app->pending_workers++;
+
     nxt_debug(task, "send sw #%uD", sw->stream);
 
     nxt_router_sw_add(task, nxt_router, sw);
@@ -1758,6 +1813,23 @@ nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
 }
 
 
+static nxt_bool_t
+nxt_router_app_free(nxt_app_t *app)
+{
+    if (app->live == 0 && app->workers == 0 &&
+        app->pending_workers == 0 &&
+        nxt_queue_is_empty(&app->requests)) {
+
+        nxt_thread_mutex_destroy(&app->mutex);
+        nxt_free(app);
+
+        return 1;
+    }
+
+    return 0;
+}
+
+
 static nxt_port_t *
 nxt_router_app_get_port(nxt_app_t *app)
 {
@@ -1789,6 +1861,7 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
     nxt_app_t            *app;
     nxt_port_t           *port;
     nxt_work_t           *work;
+    nxt_process_t        *process;
     nxt_queue_link_t     *lnk;
     nxt_req_conn_link_t  *rc;
 
@@ -1801,7 +1874,7 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
 
 
     if (task->thread->engine != port->engine) {
-        work = (nxt_work_t *) (port + 1);
+        work = &port->work;
 
         nxt_debug(task, "post release port to engine %p", port->engine);
 
@@ -1822,7 +1895,8 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
 
         rc = nxt_queue_link_data(lnk, nxt_req_conn_link_t, app_link);
 
-        nxt_debug(task, "process request #%uxD", rc->req_id);
+        nxt_debug(task, "app '%V' process next request #%uxD",
+                  &app->name, rc->req_id);
 
         rc->app_port = port;
 
@@ -1831,7 +1905,37 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
         return;
     }
 
-    nxt_debug(task, "app requests queue is empty");
+    if (port->pair[1] == -1) {
+        nxt_debug(task, "app '%V' port already closed (pid %PI dead?)",
+                  &app->name, port->pid);
+
+        app->workers--;
+        nxt_router_app_free(app);
+
+        port->app = NULL;
+        process = port->process;
+
+        nxt_port_release(port);
+
+        if (nxt_queue_is_empty(&process->ports)) {
+            nxt_runtime_process_destroy(task->thread->runtime, process);
+        }
+
+        return;
+    }
+
+    if (!app->live) {
+        nxt_debug(task, "app '%V' is not alive, send QUIT to port",
+                  &app->name);
+
+        nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
+                              -1, 0, 0, NULL);
+
+        return;
+    }
+
+    nxt_debug(task, "app '%V' requests queue is empty, keep the port",
+              &app->name);
 
     nxt_thread_mutex_lock(&app->mutex);
 
@@ -1841,29 +1945,42 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
 }
 
 
-void
+nxt_bool_t
 nxt_router_app_remove_port(nxt_port_t *port)
 {
-    nxt_app_t  *app;
+    nxt_app_t   *app;
-
+    nxt_bool_t  busy;
-    if (port->app_link.next == NULL) {
-        return;
-    }
 
     app = port->app;
+    busy = 1;
 
-#if (NXT_DEBUG)
+    if (app == NULL) {
-    if (nxt_slow_path(app == NULL)) {
+        nxt_assert(port->app_link.next == NULL);
-        nxt_abort();
+
+        return 1;
     }
-#endif
 
     nxt_thread_mutex_lock(&app->mutex);
 
-    nxt_queue_remove(&port->app_link);
+    if (port->app_link.next != NULL) {
-    port->app_link.next = NULL;
+
+        nxt_queue_remove(&port->app_link);
+        port->app_link.next = NULL;
+
+        busy = 0;
+    }
 
     nxt_thread_mutex_unlock(&app->mutex);
+
+    if (busy == 0) {
+
+        app->workers--;
+        nxt_router_app_free(app);
+
+        return 1;
+    }
+
+    return 0;
 }
 
 
@@ -1894,11 +2011,12 @@ nxt_router_app_port(nxt_task_t *task, nxt_req_conn_link_t *rc)
     port = nxt_router_app_get_port(app);
 
     if (port != NULL) {
+        nxt_debug(task, "already have port for app '%V'", &app->name);
+
         rc->app_port = port;
         return NXT_OK;
     }
 
-
     sw = nxt_mp_retain(c->mem_pool, sizeof(nxt_start_worker_t));
 
     if (nxt_slow_path(sw == NULL)) {
@@ -2069,6 +2187,8 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
         return;
     }
 
+    rc->ap = ap;
+
     nxt_event_engine_request_add(engine, rc);
 
     nxt_debug(task, "req_id %uxD linked to conn %p at engine %p",
@@ -2104,7 +2224,7 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_conn_link_t *rc,
     }
 
     reply_port = rc->reply_port;
-    ap = rc->conn->socket.data;
+    ap = rc->ap;
 
     port_mp = port->mem_pool;
     port->mem_pool = mp;

src/nxt_router.h

@@ -69,11 +69,13 @@ typedef struct {
 
 
 struct nxt_app_s {
-    nxt_thread_mutex_t     mutex;
+    nxt_thread_mutex_t     mutex;    /* Protects ports queue. */
-    nxt_queue_t            ports;
+    nxt_queue_t            ports;    /* of nxt_port_t.app_link */
+
     nxt_queue_t            requests; /* of nxt_req_conn_link_t */
     nxt_str_t              name;
 
+    uint32_t               pending_workers;
     uint32_t               workers;
     uint32_t               max_workers;
 
@@ -123,6 +125,6 @@ typedef struct {
 void nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 
-void nxt_router_app_remove_port(nxt_port_t *port);
+nxt_bool_t nxt_router_app_remove_port(nxt_port_t *port);
 
 #endif  /* _NXT_ROUTER_H_INCLUDED_ */

src/nxt_runtime.c

@@ -1542,7 +1542,6 @@ nxt_runtime_process_find(nxt_runtime_t *rt, nxt_pid_t pid)
     lhq.proto = &lvlhsh_processes_proto;
 
     if (nxt_lvlhsh_find(&rt->processes, &lhq) == NXT_OK) {
-        nxt_thread_log_debug("process %PI found", pid);
         return lhq.value;
     }
 
@@ -1662,9 +1661,13 @@ nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
 
             nxt_runtime_port_remove(rt, port);
 
+            nxt_port_release(port);
+
         } nxt_process_port_loop;
 
-        nxt_runtime_process_destroy(rt, process);
+        if (nxt_queue_is_empty(&process->ports)) {
+            nxt_runtime_process_destroy(rt, process);
+        }
 
         break;
 
@@ -1709,24 +1712,6 @@ nxt_runtime_port_remove(nxt_runtime_t *rt, nxt_port_t *port)
     if (rt->port_by_type[port->type] == port) {
         rt->port_by_type[port->type] = NULL;
     }
-
-    if (port->pair[0] != -1) {
-        nxt_fd_close(port->pair[0]);
-    }
-
-    if (port->pair[1] != -1) {
-        nxt_fd_close(port->pair[1]);
-    }
-
-    if (port->type == NXT_PROCESS_WORKER) {
-        nxt_router_app_remove_port(port);
-    }
-
-    if (port->mem_pool) {
-        nxt_mp_destroy(port->mem_pool);
-    }
-
-    nxt_mp_free(rt->mem_pool, port);
 }