Request-app link introduced to prevent mp destroy for penging requests.

nxt_req_conn_link_t still used for lookup connection by request id.
New nxt_req_app_link_t (ra) allocated from conn->mem_pool using mp_retain().
ra stored in app->requests if there is no free worker to process request.

4 files changed, 302 insertions, 125 deletions

diff --git a/src/nxt_conn.h b/src/nxt_conn.h
index c08ed376..f08eafdd 100644
--- a/src/nxt_conn.h
+++ b/src/nxt_conn.h
@@ -189,11 +189,8 @@ typedef struct {
     nxt_req_id_t         req_id;
     nxt_conn_t           *conn;
     nxt_port_t           *app_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     app_link; /* for nxt_app_t.requests */
 } nxt_req_conn_link_t;
 
 
diff --git a/src/nxt_port.c b/src/nxt_port.c
index 3f3fdd19..3c6b60b7 100644
--- a/src/nxt_port.c
+++ b/src/nxt_port.c
@@ -41,6 +41,8 @@ nxt_port_new(nxt_port_id_t id, nxt_pid_t pid, nxt_process_type_t type)
         nxt_mp_destroy(mp);
     }
 
+    nxt_thread_log_debug("port %p %d:%d new, type %d", port, pid, id, type);
+
     return port;
 }
 
@@ -48,6 +50,9 @@ 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_thread_log_debug("port %p %d:%d release, type %d", port, port->pid,
+                         port->id, port->type);
+
     if (port->pair[0] != -1) {
         nxt_fd_close(port->pair[0]);
         port->pair[0] = -1;
diff --git a/src/nxt_port.h b/src/nxt_port.h
index d6156015..cd28d4f3 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -74,6 +74,7 @@ struct nxt_port_s {
     uint32_t            max_size;
     /* Maximum interleave of message parts. */
     uint32_t            max_share;
+    uint32_t            app_req_id;
 
     nxt_port_handler_t  handler;
     nxt_port_handler_t  *data;
diff --git a/src/nxt_router.c b/src/nxt_router.c
index 083a093d..fbd3462e 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -20,18 +20,33 @@ typedef struct {
 } nxt_router_listener_conf_t;
 
 
+typedef struct nxt_req_app_link_s nxt_req_app_link_t;
 typedef struct nxt_start_worker_s nxt_start_worker_t;
 
 struct nxt_start_worker_s {
     uint32_t               stream;
     nxt_app_t              *app;
-    nxt_req_conn_link_t    *rc;
+    nxt_req_app_link_t     *ra;
     nxt_mp_t               *mem_pool;
 
     nxt_work_t             work;
 };
 
 
+struct nxt_req_app_link_s {
+    nxt_req_id_t         req_id;
+    nxt_port_t           *app_port;
+    nxt_port_t           *reply_port;
+    nxt_app_parse_ctx_t  *ap;
+    nxt_req_conn_link_t  *rc;
+
+    nxt_queue_link_t     link; /* for nxt_app_t.requests */
+
+    nxt_mp_t             *mem_pool;
+    nxt_work_t           work;
+};
+
+
 static nxt_router_temp_conf_t *nxt_router_temp_conf(nxt_task_t *task);
 static nxt_int_t nxt_router_conf_new(nxt_task_t *task,
     nxt_router_temp_conf_t *tmcf, u_char *start, u_char *end);
@@ -98,8 +113,10 @@ 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_send_sw_request(nxt_task_t *task, void *obj,
+    void *data);
+static nxt_bool_t nxt_router_app_free(nxt_task_t *task, nxt_app_t *app);
+static nxt_port_t * nxt_router_app_get_port(nxt_app_t *app, uint32_t req_id);
 static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
     void *data);
 
@@ -114,7 +131,7 @@ static void nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj,
 static void nxt_router_process_http_request(nxt_task_t *task,
     nxt_conn_t *c, nxt_app_parse_ctx_t *ap);
 static void nxt_router_process_http_request_mp(nxt_task_t *task,
-    nxt_req_conn_link_t *rc, nxt_mp_t *mp);
+    nxt_req_app_link_t *ra, nxt_port_t *port);
 static void nxt_router_conn_ready(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_conn_close(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_conn_free(nxt_task_t *task, void *obj, void *data);
@@ -156,12 +173,74 @@ nxt_router_start(nxt_task_t *task, void *data)
 }
 
 
+static nxt_start_worker_t *
+nxt_router_sw_create(nxt_task_t *task, nxt_mp_t *mp, nxt_app_t *app,
+    nxt_req_app_link_t *ra)
+{
+    nxt_port_t          *master_port;
+    nxt_runtime_t       *rt;
+    nxt_start_worker_t  *sw;
+
+    sw = nxt_mp_retain(mp, sizeof(nxt_start_worker_t));
+
+    if (nxt_slow_path(sw == NULL)) {
+        return NULL;
+    }
+
+    nxt_memzero(sw, sizeof(nxt_start_worker_t));
+
+    sw->stream = nxt_random(&task->thread->random);
+    sw->mem_pool = mp;
+
+    sw->app = app;
+    sw->ra = ra;
+
+    nxt_debug(task, "sw #%uxD create, request #%uD, app '%V' %p", sw->stream,
+                    ra->req_id, &app->name, app);
+
+    rt = task->thread->runtime;
+    master_port = rt->port_by_type[NXT_PROCESS_MASTER];
+
+    sw->work.handler = nxt_router_send_sw_request;
+    sw->work.task = &master_port->engine->task;
+    sw->work.obj = sw;
+    sw->work.data = task->thread->engine;
+    sw->work.next = NULL;
+
+    if (task->thread->engine != master_port->engine) {
+        nxt_debug(task, "post send sw %uxD to master engine %p", sw->stream,
+                  master_port->engine);
+
+        nxt_event_engine_post(master_port->engine, &sw->work);
+
+    } else {
+        nxt_router_send_sw_request(task, sw, sw->work.data);
+    }
+
+    return sw;
+}
+
+
 static void
 nxt_router_sw_release(nxt_task_t *task, void *obj, void *data)
 {
+    nxt_event_engine_t  *engine;
     nxt_start_worker_t  *sw;
 
     sw = obj;
+    engine = data;
+
+    if (task->thread->engine != engine) {
+        sw->work.handler = nxt_router_sw_release;
+        sw->work.task = &engine->task;
+        sw->work.next = NULL;
+
+        nxt_debug(task, "sw #%uxD post release to %p", sw->stream, engine);
+
+        nxt_event_engine_post(engine, &sw->work);
+
+        return;
+    }
 
     nxt_debug(task, "sw #%uxD release", sw->stream);
 
@@ -169,11 +248,80 @@ nxt_router_sw_release(nxt_task_t *task, void *obj, void *data)
 }
 
 
+static nxt_req_app_link_t *
+nxt_router_ra_create(nxt_task_t *task, nxt_req_conn_link_t *rc)
+{
+    nxt_mp_t            *mp;
+    nxt_req_app_link_t  *ra;
+
+    mp = rc->conn->mem_pool;
+
+    ra = nxt_mp_retain(mp, sizeof(nxt_req_app_link_t));
+
+    if (nxt_slow_path(ra == NULL)) {
+        return NULL;
+    }
+
+    nxt_debug(task, "ra #%uxD create", ra->req_id);
+
+    nxt_memzero(ra, sizeof(nxt_req_app_link_t));
+
+    ra->req_id = rc->req_id;
+    ra->app_port = NULL;
+    ra->rc = rc;
+
+    ra->mem_pool = mp;
+
+    ra->work.handler = NULL;
+    ra->work.task = &task->thread->engine->task;
+    ra->work.obj = ra;
+    ra->work.data = task->thread->engine;
+
+    return ra;
+}
+
+
+static void
+nxt_router_ra_release(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_req_app_link_t  *ra;
+    nxt_event_engine_t  *engine;
+
+    ra = obj;
+    engine = data;
+
+    if (task->thread->engine != engine) {
+        ra->work.handler = nxt_router_ra_release;
+        ra->work.task = &engine->task;
+        ra->work.next = NULL;
+
+        nxt_debug(task, "ra #%uxD post release to %p", ra->req_id, engine);
+
+        nxt_event_engine_post(engine, &ra->work);
+
+        return;
+    }
+
+    nxt_debug(task, "ra #%uxD release", ra->req_id);
+
+    if (ra->app_port != NULL) {
+
+        if (ra->rc->conn != NULL) {
+            ra->rc->app_port = ra->app_port;
+
+        } else {
+            nxt_router_app_release_port(task, ra->app_port, ra->app_port->app);
+        }
+    }
+
+    nxt_mp_release(ra->mem_pool, ra);
+}
+
+
 void
 nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
     nxt_start_worker_t  *sw;
-    nxt_event_engine_t  *engine;
 
     nxt_port_new_port_handler(task, msg);
 
@@ -186,23 +334,16 @@ 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--;
+        sw->app->workers++;
 
-        nxt_router_app_release_port(task, msg->new_port, sw->app);
-
-        engine = sw->work.data;
-
-        sw->work.handler = nxt_router_sw_release;
-        sw->work.task = &engine->task;
+        nxt_debug(task, "sw #%uxD got port %p", sw->stream, msg->new_port);
 
-        nxt_debug(task, "post sw #%uxD release to %p", sw->stream,
-                  sw->work.data);
+        nxt_router_app_release_port(task, msg->new_port, sw->app);
 
-        nxt_event_engine_post(engine, &sw->work);
+        nxt_router_sw_release(task, sw, sw->work.data);
     }
 }
 
@@ -1185,32 +1326,39 @@ 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_port_t  *port;
+    nxt_app_t    *app;
+    nxt_port_t   *port;
 
     nxt_queue_each(app, &router->apps, nxt_app_t, link) {
 
         nxt_queue_remove(&app->link);
 
+        nxt_thread_log_debug("about to remove app '%V' %p", &app->name, app);
+
         app->live = 0;
 
-        if (nxt_router_app_free(app) != 0) {
+        if (nxt_router_app_free(NULL, app) != 0) {
             continue;
         }
 
-        if (nxt_queue_is_empty(&app->requests)) {
+        if (!nxt_queue_is_empty(&app->requests)) {
 
-            do {
-                port = nxt_router_app_get_port(app);
-                if (port == NULL) {
-                    break;
-                }
+            nxt_thread_log_debug("app '%V' %p pending requests found",
+                                 &app->name, app);
+            continue;
+        }
+
+        do {
+            port = nxt_router_app_get_port(app, 0);
+            if (port == NULL) {
+                break;
+            }
 
-                nxt_port_socket_write(&port->engine->task, port,
-                                      NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
-            } while (1);
+            nxt_thread_log_debug("port %p send quit", port);
 
-        }
+            nxt_port_socket_write(&port->engine->task, port,
+                                  NXT_PORT_MSG_QUIT, -1, 0, 0, NULL);
+        } while (1);
 
     } nxt_queue_loop;
 
@@ -1677,6 +1825,12 @@ nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
         }
 
         nxt_buf_chain_add(&b, last);
+
+        if (rc->app_port != NULL) {
+            nxt_router_app_release_port(task, rc->app_port, rc->app_port->app);
+
+            rc->app_port = NULL;
+        }
     }
 
     if (b == NULL) {
@@ -1781,31 +1935,28 @@ nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
     nxt_port_t          *port;
     nxt_runtime_t       *rt;
     nxt_start_worker_t  *sw;
-    nxt_event_engine_t  *engine;
 
     sw = obj;
     app = sw->app;
 
-    if (app->workers + app->pending_workers >= app->max_workers) {
-        engine = sw->work.data;
-
-        sw->work.handler = nxt_router_sw_release;
-        sw->work.task = &engine->task;
+    nxt_queue_insert_tail(&app->requests, &sw->ra->link);
 
-        nxt_debug(task, "%uD/%uD running/penging workers, post sw #%uxD "
-                  "release to %p", sw->stream, sw->work.data);
+    if (app->workers + app->pending_workers >= app->max_workers) {
+        nxt_debug(task, "app '%V' %p %uD/%uD running/penging workers, "
+                  "post sw #%uxD release to %p", &app->name, app,
+                   app->workers, app->pending_workers, sw->stream,
+                   sw->work.data);
 
-        nxt_event_engine_post(engine, &sw->work);
+        nxt_router_sw_release(task, sw, sw->work.data);
 
         return;
     }
 
     app->pending_workers++;
 
-    nxt_debug(task, "send sw #%uD", sw->stream);
+    nxt_debug(task, "sw #%uxD send", sw->stream);
 
     nxt_router_sw_add(task, nxt_router, sw);
-    nxt_queue_insert_tail(&app->requests, &sw->rc->app_link);
 
     rt = task->thread->runtime;
     port = rt->port_by_type[NXT_PROCESS_MASTER];
@@ -1819,8 +1970,17 @@ 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)
+nxt_router_app_free(nxt_task_t *task, nxt_app_t *app)
 {
+    nxt_port_t           *master_port;
+    nxt_runtime_t        *rt;
+    nxt_queue_link_t     *lnk;
+    nxt_req_app_link_t   *ra;
+
+    nxt_thread_log_debug("app '%V' %p state: %d/%uD/%uD/%d", &app->name, app,
+                         app->live, app->workers, app->pending_workers,
+                         nxt_queue_is_empty(&app->requests));
+
     if (app->live == 0 && app->workers == 0 &&
         app->pending_workers == 0 &&
         nxt_queue_is_empty(&app->requests)) {
@@ -1831,12 +1991,26 @@ nxt_router_app_free(nxt_app_t *app)
         return 1;
     }
 
+    if (app->live == 1 && nxt_queue_is_empty(&app->requests) == 0 &&
+       (app->workers + app->pending_workers < app->max_workers)) {
+
+        lnk = nxt_queue_first(&app->requests);
+        nxt_queue_remove(lnk);
+
+        ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
+
+        rt = task->thread->runtime;
+        master_port = rt->port_by_type[NXT_PROCESS_MASTER];
+
+        nxt_router_sw_create(task, master_port->mem_pool, app, ra);
+    }
+
     return 0;
 }
 
 
 static nxt_port_t *
-nxt_router_app_get_port(nxt_app_t *app)
+nxt_router_app_get_port(nxt_app_t *app, uint32_t req_id)
 {
     nxt_port_t        *port;
     nxt_queue_link_t  *lnk;
@@ -1852,6 +2026,8 @@ nxt_router_app_get_port(nxt_app_t *app)
         lnk->next = NULL;
 
         port = nxt_queue_link_data(lnk, nxt_port_t, app_link);
+
+        port->app_req_id = req_id;
     }
 
     nxt_thread_mutex_unlock(&app->mutex);
@@ -1867,7 +2043,7 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
     nxt_port_t           *port;
     nxt_work_t           *work;
     nxt_queue_link_t     *lnk;
-    nxt_req_conn_link_t  *rc;
+    nxt_req_app_link_t   *ra;
 
     port = obj;
     app = data;
@@ -1897,24 +2073,28 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
         lnk = nxt_queue_first(&app->requests);
         nxt_queue_remove(lnk);
 
-        rc = nxt_queue_link_data(lnk, nxt_req_conn_link_t, app_link);
+        ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
+
+        nxt_debug(task, "app '%V' %p process next request #%uxD",
+                  &app->name, app, ra->req_id);
 
-        nxt_debug(task, "app '%V' process next request #%uxD",
-                  &app->name, rc->req_id);
+        ra->app_port = port;
 
-        rc->app_port = port;
+        nxt_router_process_http_request_mp(task, ra, port);
 
-        nxt_router_process_http_request_mp(task, rc, rc->app_port->mem_pool);
+        nxt_router_ra_release(task, ra, ra->work.data);
 
         return;
     }
 
+    port->app_req_id = 0;
+
     if (port->pair[1] == -1) {
-        nxt_debug(task, "app '%V' port already closed (pid %PI dead?)",
-                  &app->name, port->pid);
+        nxt_debug(task, "app '%V' %p port already closed (pid %PI dead?)",
+                  &app->name, app, port->pid);
 
         app->workers--;
-        nxt_router_app_free(app);
+        nxt_router_app_free(task, app);
 
         port->app = NULL;
 
@@ -1924,8 +2104,8 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
     }
 
     if (!app->live) {
-        nxt_debug(task, "app '%V' is not alive, send QUIT to port",
-                  &app->name);
+        nxt_debug(task, "app '%V' %p is not alive, send QUIT to port",
+                  &app->name, app);
 
         nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
                               -1, 0, 0, NULL);
@@ -1933,8 +2113,8 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
         return;
     }
 
-    nxt_debug(task, "app '%V' requests queue is empty, keep the port",
-              &app->name);
+    nxt_debug(task, "app '%V' %p requests queue is empty, keep the port",
+              &app->name, app);
 
     nxt_thread_mutex_lock(&app->mutex);
 
@@ -1951,9 +2131,11 @@ nxt_router_app_remove_port(nxt_port_t *port)
     nxt_bool_t  busy;
 
     app = port->app;
-    busy = 1;
+    busy = port->app_req_id != 0;
 
     if (app == NULL) {
+        nxt_thread_log_debug("port %p app remove, no app", port);
+
         nxt_assert(port->app_link.next == NULL);
 
         return 1;
@@ -1966,84 +2148,66 @@ nxt_router_app_remove_port(nxt_port_t *port)
         nxt_queue_remove(&port->app_link);
         port->app_link.next = NULL;
 
-        busy = 0;
     }
 
     nxt_thread_mutex_unlock(&app->mutex);
 
     if (busy == 0) {
+        nxt_thread_log_debug("port %p app remove, free, app '%V' %p", port,
+                             &app->name, app);
 
         app->workers--;
-        nxt_router_app_free(app);
+        nxt_router_app_free(&port->engine->task, app);
 
         return 1;
     }
 
+    nxt_thread_log_debug("port %p app remove, busy, app '%V' %p, req #%uxD",
+                         port, &app->name, app, port->app_req_id);
+
     return 0;
 }
 
 
-nxt_inline nxt_int_t
-nxt_router_app_port(nxt_task_t *task, nxt_req_conn_link_t *rc)
+static nxt_int_t
+nxt_router_app_port(nxt_task_t *task, nxt_req_app_link_t *ra)
 {
     nxt_app_t                *app;
     nxt_conn_t               *c;
-    nxt_port_t               *port, *master_port;
-    nxt_runtime_t            *rt;
+    nxt_port_t               *port;
     nxt_start_worker_t       *sw;
     nxt_socket_conf_joint_t  *joint;
 
     port = NULL;
-    c = rc->conn;
+    c = ra->rc->conn;
 
     joint = c->listen->socket.data;
     app = joint->socket_conf->application;
 
-
     if (app == NULL) {
-        nxt_router_gen_error(task, rc->conn, 500,
+        nxt_router_gen_error(task, c, 500,
                              "Application is NULL in socket_conf");
         return NXT_ERROR;
     }
 
 
-    port = nxt_router_app_get_port(app);
+    port = nxt_router_app_get_port(app, ra->req_id);
 
     if (port != NULL) {
         nxt_debug(task, "already have port for app '%V'", &app->name);
 
-        rc->app_port = port;
+        ra->app_port = port;
         return NXT_OK;
     }
 
-    sw = nxt_mp_retain(c->mem_pool, sizeof(nxt_start_worker_t));
+    sw = nxt_router_sw_create(task, c->mem_pool, app, ra);
 
     if (nxt_slow_path(sw == NULL)) {
-        nxt_router_gen_error(task, rc->conn, 500,
+        nxt_router_gen_error(task, c, 500,
                              "Failed to allocate start worker struct");
         return NXT_ERROR;
     }
 
-    nxt_memzero(sw, sizeof(nxt_start_worker_t));
-
-    sw->stream = nxt_random(&task->thread->random);
-    sw->app = app;
-    sw->rc = rc;
-    sw->mem_pool = c->mem_pool;
-
-    rt = task->thread->runtime;
-    master_port = rt->port_by_type[NXT_PROCESS_MASTER];
-
-    sw->work.handler = nxt_router_send_sw_request;
-    sw->work.task = &master_port->engine->task;
-    sw->work.obj = sw;
-    sw->work.data = task->thread->engine;
-
-    nxt_debug(task, "post send sw %uxD to master engine %p", sw->stream,
-              master_port->engine);
-
-    nxt_event_engine_post(master_port->engine, &sw->work);
-
     return NXT_AGAIN;
 }
 
@@ -2164,9 +2328,12 @@ static void
 nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     nxt_app_parse_ctx_t *ap)
 {
+    nxt_mp_t             *port_mp;
     nxt_int_t            res;
+    nxt_port_t           *port;
     nxt_req_id_t         req_id;
     nxt_event_engine_t   *engine;
+    nxt_req_app_link_t   *ra;
     nxt_req_conn_link_t  *rc;
 
     engine = task->thread->engine;
@@ -2184,47 +2351,55 @@ 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",
               req_id, c, engine);
 
-    rc->reply_port = engine->port;
 
-    res = nxt_router_app_port(task, rc);
+    ra = nxt_router_ra_create(task, rc);
+
+    ra->ap = ap;
+    ra->reply_port = engine->port;
+
+    res = nxt_router_app_port(task, ra);
 
     if (res != NXT_OK) {
         return;
     }
 
-    nxt_router_process_http_request_mp(task, rc, c->mem_pool);
+    port = ra->app_port;
+
+    if (nxt_slow_path(port == NULL)) {
+        nxt_router_gen_error(task, rc->conn, 500, "Application port not found");
+        return;
+    }
+
+    port_mp = port->mem_pool;
+    port->mem_pool = c->mem_pool;
+
+    nxt_router_process_http_request_mp(task, ra, port);
+
+    port->mem_pool = port_mp;
+
+
+    nxt_router_ra_release(task, ra, ra->work.data);
 }
 
 
 static void
-nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_conn_link_t *rc,
-    nxt_mp_t *mp)
+nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
+    nxt_port_t *port)
 {
-    nxt_mp_t             *port_mp;
     nxt_int_t            res;
-    nxt_port_t           *port, *c_port, *reply_port;
+    nxt_port_t           *c_port, *reply_port;
+    nxt_conn_t           *c;
     nxt_app_wmsg_t       wmsg;
     nxt_app_parse_ctx_t  *ap;
 
-    port = rc->app_port;
-
-    if (nxt_slow_path(port == NULL)) {
-        nxt_router_gen_error(task, rc->conn, 500, "Application port not found");
-        return;
-    }
-
-    reply_port = rc->reply_port;
-    ap = rc->ap;
-
-    port_mp = port->mem_pool;
-    port->mem_pool = mp;
+    reply_port = ra->reply_port;
+    ap = ra->ap;
+    c = ra->rc->conn;
 
     c_port = nxt_process_connected_port_find(port->process, reply_port->pid,
                                              reply_port->id);
@@ -2232,9 +2407,9 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_conn_link_t *rc,
         res = nxt_port_send_port(task, port, reply_port, 0);
 
         if (nxt_slow_path(res != NXT_OK)) {
-            nxt_router_gen_error(task, rc->conn, 500,
+            nxt_router_gen_error(task, c, 500,
                                  "Failed to send reply port to application");
-            goto fail;
+            return;
         }
 
         nxt_process_connected_port_add(port->process, reply_port);
@@ -2243,14 +2418,14 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_conn_link_t *rc,
     wmsg.port = port;
     wmsg.write = NULL;
     wmsg.buf = &wmsg.write;
-    wmsg.stream = rc->req_id;
+    wmsg.stream = ra->req_id;
 
-    res = rc->app_port->app->module->prepare_msg(task, &ap->r, &wmsg);
+    res = port->app->module->prepare_msg(task, &ap->r, &wmsg);
 
     if (nxt_slow_path(res != NXT_OK)) {
-        nxt_router_gen_error(task, rc->conn, 500,
+        nxt_router_gen_error(task, c, 500,
                              "Failed to prepare message for application");
-        goto fail;
+        return;
     }
 
     nxt_debug(task, "about to send %d bytes buffer to worker port %d",
@@ -2258,16 +2433,13 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_conn_link_t *rc,
                     wmsg.port->socket.fd);
 
     res = nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
-                                 -1, rc->req_id, reply_port->id, wmsg.write);
+                                 -1, ra->req_id, reply_port->id, wmsg.write);
 
     if (nxt_slow_path(res != NXT_OK)) {
-        nxt_router_gen_error(task, rc->conn, 500,
+        nxt_router_gen_error(task, c, 500,
                              "Failed to send message to application");
-        goto fail;
+        return;
     }
-
-fail:
-    port->mem_pool = port_mp;
 }
 
 
@@ -2378,6 +2550,8 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
             rc->app_port = NULL;
         }
 
+        rc->conn = NULL;
+
         nxt_event_engine_request_remove(task->thread->engine, rc);
 
     } nxt_queue_loop;