Introducing application timeout.

1 files changed, 334 insertions, 76 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index ae569267..7357667d 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -10,8 +10,11 @@
 
 
 typedef struct {
-    nxt_str_t  type;
-    uint32_t   workers;
+    nxt_str_t         type;
+    uint32_t          workers;
+    nxt_msec_t        timeout;
+    uint32_t          requests;
+    nxt_conf_value_t  *limits_value;
 } nxt_router_app_conf_t;
 
 
@@ -31,9 +34,20 @@ struct nxt_start_worker_s {
 };
 
 
+typedef struct {
+    uint32_t             stream;
+    nxt_conn_t           *conn;
+    nxt_port_t           *app_port;
+    nxt_req_app_link_t   *ra;
+
+    nxt_queue_link_t     link;     /* for nxt_conn_t.requests */
+} nxt_req_conn_link_t;
+
+
 struct nxt_req_app_link_s {
-    nxt_req_id_t         req_id;
+    uint32_t             stream;
     nxt_port_t           *app_port;
+    nxt_pid_t            app_pid;
     nxt_port_t           *reply_port;
     nxt_app_parse_ctx_t  *ap;
     nxt_req_conn_link_t  *rc;
@@ -51,6 +65,18 @@ typedef struct {
 } nxt_socket_rpc_t;
 
 
+typedef struct {
+    nxt_mp_t             *mem_pool;
+    nxt_port_recv_msg_t  msg;
+    nxt_work_t           work;
+} nxt_remove_pid_msg_t;
+
+
+static void nxt_router_worker_remove_pid_handler(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_worker_remove_pid_done(nxt_task_t *task, void *obj,
+    void *data);
+
 static nxt_router_temp_conf_t *nxt_router_temp_conf(nxt_task_t *task);
 static void nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_conf_ready(nxt_task_t *task,
@@ -105,8 +131,6 @@ static void nxt_router_engines_post(nxt_router_t *router,
     nxt_router_temp_conf_t *tmcf);
 static void nxt_router_engine_post(nxt_event_engine_t *engine,
     nxt_work_t *jobs);
-static void nxt_router_app_data_handler(nxt_task_t *task,
-    nxt_port_recv_msg_t *msg);
 
 static void nxt_router_thread_start(void *data);
 static void nxt_router_listen_socket_create(nxt_task_t *task, void *obj,
@@ -129,7 +153,7 @@ static void nxt_router_conf_release(nxt_task_t *task,
 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 nxt_port_t * nxt_router_app_get_port(nxt_app_t *app, uint32_t stream);
 static void nxt_router_app_release_port(nxt_task_t *task, void *obj,
     void *data);
 
@@ -153,6 +177,7 @@ 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);
 static void nxt_router_conn_error(nxt_task_t *task, void *obj, void *data);
 static void nxt_router_conn_timeout(nxt_task_t *task, void *obj, void *data);
+static void nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data);
 static nxt_msec_t nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data);
 
 static void nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
@@ -213,8 +238,8 @@ nxt_router_sw_create(nxt_task_t *task, nxt_app_t *app, nxt_req_app_link_t *ra)
     sw->app = app;
     sw->ra = ra;
 
-    nxt_debug(task, "sw %p create, request #%uxD, app '%V' %p", sw,
-                    ra->req_id, &app->name, app);
+    nxt_debug(task, "sw %p create, stream #%uD, app '%V' %p", sw,
+                    ra->stream, &app->name, app);
 
     rt = task->thread->runtime;
     main_port = rt->port_by_type[NXT_PROCESS_MAIN];
@@ -248,13 +273,29 @@ nxt_router_sw_release(nxt_task_t *task, nxt_start_worker_t *sw)
 }
 
 
+nxt_inline void
+nxt_router_rc_unlink(nxt_req_conn_link_t *rc)
+{
+    nxt_queue_remove(&rc->link);
+
+    if (rc->ra != NULL) {
+        rc->ra->rc = NULL;
+        rc->ra = NULL;
+    }
+
+    rc->conn = NULL;
+}
+
+
 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_event_engine_t  *engine;
     nxt_req_app_link_t  *ra;
 
     mp = rc->conn->mem_pool;
+    engine = task->thread->engine;
 
     ra = nxt_mp_retain(mp, sizeof(nxt_req_app_link_t));
 
@@ -262,20 +303,22 @@ nxt_router_ra_create(nxt_task_t *task, nxt_req_conn_link_t *rc)
         return NULL;
     }
 
-    nxt_debug(task, "ra #%uxD create", rc->req_id);
+    nxt_debug(task, "ra stream #%uD create", rc->stream);
 
     nxt_memzero(ra, sizeof(nxt_req_app_link_t));
 
-    ra->req_id = rc->req_id;
-    ra->app_port = NULL;
+    ra->stream = rc->stream;
+    ra->app_pid = -1;
     ra->rc = rc;
+    rc->ra = ra;
+    ra->reply_port = engine->port;
 
     ra->mem_pool = mp;
 
     ra->work.handler = NULL;
-    ra->work.task = &task->thread->engine->task;
+    ra->work.task = &engine->task;
     ra->work.obj = ra;
-    ra->work.data = task->thread->engine;
+    ra->work.data = engine;
 
     return ra;
 }
@@ -284,39 +327,87 @@ nxt_router_ra_create(nxt_task_t *task, nxt_req_conn_link_t *rc)
 static void
 nxt_router_ra_release(nxt_task_t *task, void *obj, void *data)
 {
+    nxt_port_t          *app_port;
     nxt_req_app_link_t  *ra;
     nxt_event_engine_t  *engine;
 
     ra = obj;
     engine = data;
 
+    if (ra->app_port != NULL) {
+
+        app_port = ra->app_port;
+        ra->app_port = NULL;
+
+        if (task->thread->engine != engine) {
+            ra->app_pid = app_port->pid;
+        }
+
+        nxt_router_app_release_port(task, app_port, app_port->app);
+
+#if 0
+        /* Uncomment to hold app port until complete response received. */
+        if (ra->rc != NULL) {
+            ra->rc->app_port = ra->app_port;
+
+        } else {
+            nxt_router_app_release_port(task, ra->app_port, ra->app_port->app);
+        }
+#endif
+    }
+
     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_debug(task, "ra stream #%uD post release to %p",
+                  ra->stream, engine);
 
         nxt_event_engine_post(engine, &ra->work);
 
         return;
     }
 
-    nxt_debug(task, "ra #%uxD release", ra->req_id);
+    if (ra->rc != NULL && ra->app_pid != -1) {
+        nxt_port_rpc_ex_set_peer(task, engine->port, ra->rc, ra->app_pid);
+    }
 
-    if (ra->app_port != NULL) {
+    nxt_debug(task, "ra stream #%uD release", ra->stream);
 
-        nxt_router_app_release_port(task, ra->app_port, ra->app_port->app);
+    nxt_mp_release(ra->mem_pool, ra);
+}
 
-#if 0
-        /* Uncomment to hold app port until complete response received. */
-        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);
-        }
-#endif
+static void
+nxt_router_ra_abort(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_conn_t          *c;
+    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_abort;
+        ra->work.task = &engine->task;
+        ra->work.next = NULL;
+
+        nxt_debug(task, "ra stream #%uD post abort to %p", ra->stream, engine);
+
+        nxt_event_engine_post(engine, &ra->work);
+
+        return;
+    }
+
+    nxt_debug(task, "ra stream #%uD abort", ra->stream);
+
+    if (ra->rc != NULL) {
+        c = ra->rc->conn;
+
+        nxt_router_gen_error(task, c, 500,
+                             "Failed to start application worker");
     }
 
     nxt_mp_release(ra->mem_pool, ra);
@@ -384,18 +475,83 @@ nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 void
 nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
+    nxt_mp_t              *mp;
+    nxt_buf_t             *buf;
+    nxt_event_engine_t    *engine;
+    nxt_remove_pid_msg_t  *rp;
+
     nxt_port_remove_pid_handler(task, msg);
 
     if (msg->port_msg.stream == 0) {
         return;
     }
 
+    mp = nxt_mp_create(1024, 128, 256, 32);
+
+    buf = nxt_buf_mem_alloc(mp, nxt_buf_used_size(msg->buf), 0);
+    buf->mem.free = nxt_cpymem(buf->mem.free, msg->buf->mem.pos,
+                               nxt_buf_used_size(msg->buf));
+
+    nxt_queue_each(engine, &nxt_router->engines, nxt_event_engine_t, link0)
+    {
+        rp = nxt_mp_retain(mp, sizeof(nxt_remove_pid_msg_t));
+
+        rp->mem_pool = mp;
+
+        rp->msg.fd = msg->fd;
+        rp->msg.buf = buf;
+        rp->msg.port = engine->port;
+        rp->msg.port_msg = msg->port_msg;
+        rp->msg.size = msg->size;
+        rp->msg.new_port = NULL;
+
+        rp->work.handler = nxt_router_worker_remove_pid_handler;
+        rp->work.task = &engine->task;
+        rp->work.obj = rp;
+        rp->work.data = task->thread->engine;
+        rp->work.next = NULL;
+
+        nxt_event_engine_post(engine, &rp->work);
+    }
+    nxt_queue_loop;
+
     msg->port_msg.type = _NXT_PORT_MSG_RPC_ERROR;
 
     nxt_port_rpc_handler(task, msg);
 }
 
 
+static void
+nxt_router_worker_remove_pid_handler(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_event_engine_t    *engine;
+    nxt_remove_pid_msg_t  *rp;
+
+    rp = obj;
+
+    nxt_port_remove_pid_handler(task, &rp->msg);
+
+    engine = rp->work.data;
+
+    rp->work.handler = nxt_router_worker_remove_pid_done;
+    rp->work.task = &engine->task;
+    rp->work.next = NULL;
+
+    nxt_event_engine_post(engine, &rp->work);
+}
+
+
+static void
+nxt_router_worker_remove_pid_done(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_remove_pid_msg_t  *rp;
+
+    rp = obj;
+
+    nxt_mp_release(rp->mem_pool, rp);
+}
+
+
 static nxt_router_temp_conf_t *
 nxt_router_temp_conf(nxt_task_t *task)
 {
@@ -607,6 +763,27 @@ static nxt_conf_map_t  nxt_router_app_conf[] = {
         NXT_CONF_MAP_INT32,
         offsetof(nxt_router_app_conf_t, workers),
     },
+
+    {
+        nxt_string("limits"),
+        NXT_CONF_MAP_PTR,
+        offsetof(nxt_router_app_conf_t, limits_value),
+    },
+};
+
+
+static nxt_conf_map_t  nxt_router_app_limits_conf[] = {
+    {
+        nxt_string("timeout"),
+        NXT_CONF_MAP_MSEC,
+        offsetof(nxt_router_app_conf_t, timeout),
+    },
+
+    {
+        nxt_string("requests"),
+        NXT_CONF_MAP_INT32,
+        offsetof(nxt_router_app_conf_t, requests),
+    },
 };
 
 
@@ -754,6 +931,9 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
         }
 
         apcf.workers = 1;
+        apcf.timeout = 0;
+        apcf.requests = 0;
+        apcf.limits_value = NULL;
 
         ret = nxt_conf_map_object(mp, application, nxt_router_app_conf,
                                   nxt_nitems(nxt_router_app_conf), &apcf);
@@ -762,8 +942,27 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
             goto app_fail;
         }
 
+        if (apcf.limits_value != NULL) {
+
+            if (nxt_conf_type(apcf.limits_value) != NXT_CONF_OBJECT) {
+                nxt_log(task, NXT_LOG_CRIT, "application limits is not object");
+                goto app_fail;
+            }
+
+            ret = nxt_conf_map_object(mp, apcf.limits_value,
+                                      nxt_router_app_limits_conf,
+                                      nxt_nitems(nxt_router_app_limits_conf),
+                                      &apcf);
+            if (ret != NXT_OK) {
+                nxt_log(task, NXT_LOG_CRIT, "application limits map error");
+                goto app_fail;
+            }
+        }
+
         nxt_debug(task, "application type: %V", &apcf.type);
         nxt_debug(task, "application workers: %D", apcf.workers);
+        nxt_debug(task, "application timeout: %D", apcf.timeout);
+        nxt_debug(task, "application requests: %D", apcf.requests);
 
         lang = nxt_app_lang_module(task->thread->runtime, &apcf.type);
 
@@ -802,6 +1001,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
 
         app->type = type;
         app->max_workers = apcf.workers;
+        app->timeout = apcf.timeout;
         app->live = 1;
         app->prepare_msg = nxt_app_prepare_msg[type];
 
@@ -1589,7 +1789,7 @@ static nxt_port_handler_t  nxt_router_app_port_handlers[] = {
     NULL, /* NXT_PORT_MSG_CHANGE_FILE  */
     /* TODO: remove mmap_handler from app ports */
     nxt_port_mmap_handler, /* NXT_PORT_MSG_MMAP         */
-    nxt_router_app_data_handler,
+    nxt_port_rpc_handler,  /* NXT_PORT_MSG_DATA         */
     NULL, /* NXT_PORT_MSG_REMOVE_PID   */
     NULL, /* NXT_PORT_MSG_READY        */
     NULL, /* NXT_PORT_MSG_START_WORKER */
@@ -2008,23 +2208,16 @@ static const nxt_conn_state_t  nxt_router_conn_write_state
 
 
 static void
-nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
+nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
+    void *data)
 {
     size_t               dump_size;
     nxt_buf_t            *b, *last;
     nxt_conn_t           *c;
     nxt_req_conn_link_t  *rc;
-    nxt_event_engine_t   *engine;
 
     b = msg->buf;
-    engine = task->thread->engine;
-
-    rc = nxt_event_engine_request_find(engine, msg->port_msg.stream);
-    if (nxt_slow_path(rc == NULL)) {
-        nxt_debug(task, "request id %08uxD not found", msg->port_msg.stream);
-
-        return;
-    }
+    rc = data;
 
     c = rc->conn;
 
@@ -2058,7 +2251,7 @@ nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
             rc->app_port = NULL;
         }
 
-        rc->conn = NULL;
+        nxt_router_rc_unlink(rc);
     }
 
     if (b == NULL) {
@@ -2084,6 +2277,21 @@ nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 }
 
 
+static void
+nxt_router_response_error_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
+    void *data)
+{
+    nxt_req_conn_link_t  *rc;
+
+    rc = data;
+
+    nxt_router_gen_error(task, rc->conn, 500,
+                         "Application terminated unexpectedly");
+
+    nxt_router_rc_unlink(rc);
+}
+
+
 nxt_inline const char *
 nxt_router_text_by_code(int code)
 {
@@ -2147,20 +2355,21 @@ nxt_router_gen_error(nxt_task_t *task, nxt_conn_t *c, int code,
     const char* fmt, ...)
 {
     va_list    args;
+    nxt_mp_t   *mp;
     nxt_buf_t  *b;
 
+    /* TODO: fix when called from main thread */
+    /* TODO: fix when called in the middle of response */
+
+    mp = nxt_mp_create(1024, 128, 256, 32);
+
     va_start(args, fmt);
-    b = nxt_router_get_error_buf(task, c->mem_pool, code, fmt, args);
+    b = nxt_router_get_error_buf(task, mp, code, fmt, args);
     va_end(args);
 
-    if (c->socket.data != NULL) {
-        nxt_mp_free(c->mem_pool, c->socket.data);
-        c->socket.data = NULL;
-    }
-
     if (c->socket.fd == -1) {
-        nxt_mp_release(c->mem_pool, b->next);
-        nxt_mp_release(c->mem_pool, b);
+        nxt_mp_release(mp, b->next);
+        nxt_mp_release(mp, b);
         return;
     }
 
@@ -2204,17 +2413,35 @@ nxt_router_sw_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
 static void
 nxt_router_sw_error(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data)
 {
+    nxt_app_t           *app;
+    nxt_queue_link_t    *lnk;
+    nxt_req_app_link_t  *ra;
     nxt_start_worker_t  *sw;
 
     sw = data;
 
     nxt_assert(sw != NULL);
+    nxt_assert(sw->app != NULL);
     nxt_assert(sw->app->pending_workers != 0);
 
+    app = sw->app;
+
     sw->app->pending_workers--;
 
     nxt_debug(task, "sw %p error, failed to start app '%V'",
-              sw, &sw->app->name);
+              sw, &app->name);
+
+    if (!nxt_queue_is_empty(&app->requests)) {
+        lnk = nxt_queue_last(&app->requests);
+        nxt_queue_remove(lnk);
+
+        ra = nxt_queue_link_data(lnk, nxt_req_app_link_t, link);
+
+        nxt_debug(task, "app '%V' %p abort next stream #%uD",
+                  &app->name, app, ra->stream);
+
+        nxt_router_ra_abort(task, ra, ra->work.data);
+    }
 
     nxt_router_sw_release(task, sw);
 }
@@ -2237,11 +2464,11 @@ nxt_router_send_sw_request(nxt_task_t *task, void *obj, void *data)
 
     if (nxt_queue_is_empty(&app->requests)) {
         ra = sw->ra;
-        app_port = nxt_router_app_get_port(app, ra->req_id);
+        app_port = nxt_router_app_get_port(app, ra->stream);
 
         if (app_port != NULL) {
-            nxt_debug(task, "app '%V' %p process request #%uxD",
-                      &app->name, app, ra->req_id);
+            nxt_debug(task, "app '%V' %p process stream #%uD",
+                      &app->name, app, ra->stream);
 
             ra->app_port = app_port;
 
@@ -2330,7 +2557,7 @@ 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)
+nxt_router_app_get_port(nxt_app_t *app, uint32_t stream)
 {
     nxt_port_t        *port;
     nxt_queue_link_t  *lnk;
@@ -2347,7 +2574,7 @@ nxt_router_app_get_port(nxt_app_t *app, uint32_t req_id)
 
         port = nxt_queue_link_data(lnk, nxt_port_t, app_link);
 
-        port->app_req_id = req_id;
+        port->app_stream = stream;
     }
 
     nxt_thread_mutex_unlock(&app->mutex);
@@ -2395,11 +2622,11 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
 
         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' %p process next stream #%uD",
+                  &app->name, app, ra->stream);
 
         ra->app_port = port;
-        port->app_req_id = ra->req_id;
+        port->app_stream = ra->stream;
 
         nxt_router_process_http_request_mp(task, ra, port);
 
@@ -2408,7 +2635,7 @@ nxt_router_app_release_port(nxt_task_t *task, void *obj, void *data)
         return;
     }
 
-    port->app_req_id = 0;
+    port->app_stream = 0;
 
     if (port->pair[1] == -1) {
         nxt_debug(task, "app '%V' %p port already closed (pid %PI dead?)",
@@ -2452,7 +2679,7 @@ nxt_router_app_remove_port(nxt_port_t *port)
     nxt_bool_t  busy;
 
     app = port->app;
-    busy = port->app_req_id != 0;
+    busy = port->app_stream != 0;
 
     if (app == NULL) {
         nxt_thread_log_debug("port %p app remove, no app", port);
@@ -2483,8 +2710,9 @@ nxt_router_app_remove_port(nxt_port_t *port)
         return 1;
     }
 
-    nxt_thread_log_debug("port %p app remove, busy, app '%V' %p, req #%uxD",
-                         port, &app->name, app, port->app_req_id);
+    nxt_thread_log_debug("port %p app remove, busy, app '%V' %p, "
+                         "app stream #%uD", port, &app->name, app,
+                         port->app_stream);
 
     return 0;
 }
@@ -2496,6 +2724,7 @@ 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;
+    nxt_event_engine_t       *engine;
     nxt_start_worker_t       *sw;
     nxt_socket_conf_joint_t  *joint;
 
@@ -2511,8 +2740,16 @@ nxt_router_app_port(nxt_task_t *task, nxt_req_app_link_t *ra)
         return NXT_ERROR;
     }
 
+    engine = task->thread->engine;
 
-    port = nxt_router_app_get_port(app, ra->req_id);
+    nxt_timer_disable(engine, &c->read_timer);
+
+    if (app->timeout != 0) {
+        c->read_timer.handler = nxt_router_app_timeout;
+        nxt_timer_add(engine, &c->read_timer, app->timeout);
+    }
+
+    port = nxt_router_app_get_port(app, ra->stream);
 
     if (port != NULL) {
         nxt_debug(task, "already have port for app '%V'", &app->name);
@@ -2740,18 +2977,16 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     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;
 
-    do {
-        req_id = nxt_random(&task->thread->random);
-    } while (nxt_event_engine_request_find(engine, req_id) != NULL);
-
-    rc = nxt_conn_request_add(c, req_id);
+    rc = nxt_port_rpc_register_handler_ex(task, engine->port,
+                                          nxt_router_response_ready_handler,
+                                          nxt_router_response_error_handler,
+                                          sizeof(nxt_req_conn_link_t));
 
     if (nxt_slow_path(rc == NULL)) {
         nxt_router_gen_error(task, c, 500, "Failed to allocate "
@@ -2760,17 +2995,19 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
         return;
     }
 
-    nxt_event_engine_request_add(engine, rc);
+    rc->stream = nxt_port_rpc_ex_stream(rc);
+    rc->conn = c;
 
-    nxt_debug(task, "req_id %uxD linked to conn %p at engine %p",
-              req_id, c, engine);
+    nxt_queue_insert_tail(&c->requests, &rc->link);
+
+    nxt_debug(task, "stream #%uD linked to conn %p at engine %p",
+              rc->stream, c, engine);
 
     c->socket.data = NULL;
 
     ra = nxt_router_ra_create(task, rc);
 
     ra->ap = ap;
-    ra->reply_port = engine->port;
 
     res = nxt_router_app_port(task, ra);
 
@@ -2781,10 +3018,12 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     port = ra->app_port;
 
     if (nxt_slow_path(port == NULL)) {
-        nxt_router_gen_error(task, rc->conn, 500, "Application port not found");
+        nxt_router_gen_error(task, c, 500, "Application port not found");
         return;
     }
 
+    nxt_port_rpc_ex_set_peer(task, engine->port, rc, port->pid);
+
     port_mp = port->mem_pool;
     port->mem_pool = c->mem_pool;
 
@@ -2792,7 +3031,6 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
 
     port->mem_pool = port_mp;
 
-
     nxt_router_ra_release(task, ra, ra->work.data);
 }
 
@@ -2807,6 +3045,10 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
     nxt_app_wmsg_t       wmsg;
     nxt_app_parse_ctx_t  *ap;
 
+    /* TODO: it is unsafe to use ra->rc and ra->rc->conn in main thread */
+
+    nxt_assert(ra->rc != NULL);
+
     reply_port = ra->reply_port;
     ap = ra->ap;
     c = ra->rc->conn;
@@ -2828,7 +3070,7 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
     wmsg.port = port;
     wmsg.write = NULL;
     wmsg.buf = &wmsg.write;
-    wmsg.stream = ra->req_id;
+    wmsg.stream = ra->stream;
 
     res = port->app->prepare_msg(task, &ap->r, &wmsg);
 
@@ -2843,7 +3085,7 @@ nxt_router_process_http_request_mp(nxt_task_t *task, nxt_req_app_link_t *ra,
                     wmsg.port->socket.fd);
 
     res = nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
-                                 -1, ra->req_id, reply_port->id, wmsg.write);
+                                 -1, ra->stream, reply_port->id, wmsg.write);
 
     if (nxt_slow_path(res != NXT_OK)) {
         nxt_router_gen_error(task, c, 500,
@@ -3217,7 +3459,7 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
 
     nxt_queue_each(rc, &c->requests, nxt_req_conn_link_t, link) {
 
-        nxt_debug(task, "conn %p close, req %uxD", c, rc->req_id);
+        nxt_debug(task, "conn %p close, stream #%uD", c, rc->stream);
 
         if (rc->app_port != NULL) {
             nxt_router_app_release_port(task, rc->app_port, rc->app_port->app);
@@ -3225,9 +3467,9 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
             rc->app_port = NULL;
         }
 
-        rc->conn = NULL;
+        nxt_router_rc_unlink(rc);
 
-        nxt_event_engine_request_remove(task->thread->engine, rc);
+        nxt_port_rpc_cancel(task, task->thread->engine->port, rc->stream);
 
     } nxt_queue_loop;
 
@@ -3281,6 +3523,22 @@ nxt_router_conn_timeout(nxt_task_t *task, void *obj, void *data)
 }
 
 
+static void
+nxt_router_app_timeout(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_conn_t   *c;
+    nxt_timer_t  *timer;
+
+    timer = obj;
+
+    nxt_debug(task, "router app timeout");
+
+    c = nxt_read_timer_conn(timer);
+
+    nxt_router_gen_error(task, c, 408, "Application timeout");
+}
+
+
 static nxt_msec_t
 nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data)
 {