Skeleton of router configuration and request processing.

1 files changed, 1039 insertions, 37 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index caef8503..1f96a1df 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -5,81 +5,1083 @@
  * Copyright (C) NGINX, Inc.
  */
 
-#include <nxt_main.h>
-#include <nxt_runtime.h>
-#include <nxt_master_process.h>
+#include <nxt_router.h>
 
 
-static nxt_int_t nxt_router_listen_socket(nxt_task_t *task, nxt_runtime_t *rt);
+static nxt_router_temp_conf_t *nxt_router_temp_conf(nxt_task_t *task,
+    nxt_router_t *router);
+static void nxt_router_listen_sockets_sort(nxt_router_t *router,
+    nxt_router_temp_conf_t *tmcf);
+
+static nxt_int_t nxt_router_stub_conf(nxt_task_t *task,
+    nxt_router_temp_conf_t *tmcf);
+static nxt_int_t nxt_router_listen_sockets_stub_create(nxt_task_t *task,
+    nxt_router_temp_conf_t *tmcf);
+static nxt_socket_conf_t *nxt_router_socket_conf(nxt_task_t *task,
+    nxt_mem_pool_t *mp, nxt_sockaddr_t *sa);
+static nxt_sockaddr_t *nxt_router_listen_sockaddr_stub(nxt_task_t *task,
+    nxt_mem_pool_t *mp, uint32_t port);
+
+static nxt_int_t nxt_router_engines_create(nxt_task_t *task,
+    nxt_router_t *router, nxt_router_temp_conf_t *tmcf,
+    const nxt_event_interface_t *interface);
+static nxt_int_t nxt_router_engine_conf_create(nxt_task_t *task,
+    nxt_mem_pool_t *mp, nxt_router_temp_conf_t *tmcf,
+    nxt_router_engine_conf_t *recf);
+static nxt_int_t nxt_router_engine_conf_update(nxt_task_t *task,
+    nxt_mem_pool_t *mp, nxt_router_temp_conf_t *tmcf,
+    nxt_router_engine_conf_t *recf);
+static nxt_int_t nxt_router_engine_conf_delete(nxt_task_t *task,
+    nxt_mem_pool_t *mp, nxt_router_temp_conf_t *tmcf,
+    nxt_router_engine_conf_t *recf);
+static nxt_int_t nxt_router_engine_joints_create(nxt_task_t *task,
+    nxt_mem_pool_t *mp, nxt_router_engine_conf_t *recf, nxt_queue_t *sockets,
+    nxt_array_t *array, nxt_work_handler_t handler);
+static nxt_int_t nxt_router_engine_joints_delete(nxt_task_t *task,
+    nxt_router_engine_conf_t *recf, nxt_queue_t *sockets, nxt_array_t *array);
+
+static nxt_int_t nxt_router_threads_create(nxt_task_t *task, nxt_runtime_t *rt,
+    nxt_router_temp_conf_t *tmcf);
+static nxt_int_t nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
+    nxt_event_engine_t *engine);
+
+static void nxt_router_engines_post(nxt_router_temp_conf_t *tmcf);
+static void nxt_router_engine_post(nxt_router_engine_conf_t *recf);
+
+static void nxt_router_thread_start(void *data);
+static void nxt_router_listen_socket_create(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_listen_socket_update(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_listen_socket_delete(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_listen_socket_close(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_listen_socket_release(nxt_task_t *task,
+    nxt_socket_conf_joint_t *joint);
+static void nxt_router_thread_exit_handler(nxt_task_t *task, void *obj,
+    void *data);
+static void nxt_router_conf_release(nxt_task_t *task,
+    nxt_socket_conf_joint_t *joint);
+
+static void nxt_router_conn_init(nxt_task_t *task, void *obj, void *data);
+static void nxt_router_conn_http_header_parse(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);
+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 nxt_msec_t nxt_router_conn_timeout_value(nxt_event_conn_t *c,
+    uintptr_t data);
 
 
 nxt_int_t
 nxt_router_start(nxt_task_t *task, nxt_runtime_t *rt)
 {
-    if (nxt_router_listen_socket(task, rt) != NXT_OK) {
+    nxt_int_t                    ret;
+    nxt_router_t                 *router;
+    nxt_router_temp_conf_t       *tmcf;
+    const nxt_event_interface_t  *interface;
+
+    router = nxt_zalloc(sizeof(nxt_router_t));
+    if (nxt_slow_path(router == NULL)) {
+        return NXT_ERROR;
+    }
+
+    nxt_queue_init(&router->engines);
+    nxt_queue_init(&router->sockets);
+
+    /**/
+
+    tmcf = nxt_router_temp_conf(task, router);
+    if (nxt_slow_path(tmcf == NULL)) {
         return NXT_ERROR;
     }
 
+    ret = nxt_router_stub_conf(task, tmcf);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
+
+    nxt_router_listen_sockets_sort(router, tmcf);
+
+    ret = nxt_router_listen_sockets_stub_create(task, tmcf);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
+
+    interface = nxt_service_get(rt->services, "engine", NULL);
+
+    ret = nxt_router_engines_create(task, router, tmcf, interface);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
+
+    ret = nxt_router_threads_create(task, rt, tmcf);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
+
+    nxt_router_engines_post(tmcf);
+
+    nxt_queue_add(&router->sockets, &tmcf->updating);
+    nxt_queue_add(&router->sockets, &tmcf->creating);
+
     return NXT_OK;
 }
 
 
+static nxt_router_temp_conf_t *
+nxt_router_temp_conf(nxt_task_t *task, nxt_router_t *router)
+{
+    nxt_mem_pool_t          *mp, *tmp;
+    nxt_router_conf_t       *rtcf;
+    nxt_router_temp_conf_t  *tmcf;
+
+    mp = nxt_mem_pool_create(1024);
+    if (nxt_slow_path(mp == NULL)) {
+        return NULL;
+    }
+
+    rtcf = nxt_mem_zalloc(mp, sizeof(nxt_router_conf_t));
+    if (nxt_slow_path(rtcf == NULL)) {
+        goto fail;
+    }
+
+    rtcf->mem_pool = mp;
+    rtcf->router = router;
+    rtcf->count = 1;
+
+    tmp = nxt_mem_pool_create(1024);
+    if (nxt_slow_path(tmp == NULL)) {
+        goto fail;
+    }
+
+    tmcf = nxt_mem_zalloc(tmp, sizeof(nxt_router_temp_conf_t));
+    if (nxt_slow_path(tmcf == NULL)) {
+        goto temp_fail;
+    }
+
+    tmcf->mem_pool = tmp;
+    tmcf->conf = rtcf;
+
+    tmcf->engines = nxt_array_create(tmcf->mem_pool, 4,
+                                     sizeof(nxt_router_engine_conf_t));
+    if (nxt_slow_path(tmcf->engines == NULL)) {
+        goto temp_fail;
+    }
+
+    nxt_queue_init(&tmcf->deleting);
+    nxt_queue_init(&tmcf->keeping);
+    nxt_queue_init(&tmcf->updating);
+    nxt_queue_init(&tmcf->pending);
+    nxt_queue_init(&tmcf->creating);
+
+    return tmcf;
+
+temp_fail:
+
+    nxt_mem_pool_destroy(tmp);
+
+fail:
+
+    nxt_mem_pool_destroy(mp);
+
+    return NULL;
+}
+
+
 static nxt_int_t
-nxt_router_listen_socket(nxt_task_t *task, nxt_runtime_t *rt)
+nxt_router_stub_conf(nxt_task_t *task, nxt_router_temp_conf_t *tmcf)
 {
-    nxt_sockaddr_t       *sa;
-    nxt_listen_socket_t  *ls;
+    nxt_sockaddr_t     *sa;
+    nxt_mem_pool_t     *mp;
+    nxt_socket_conf_t  *skcf;
 
-    sa = nxt_sockaddr_alloc(rt->mem_pool, sizeof(struct sockaddr_in),
-                            NXT_INET_ADDR_STR_LEN);
-    if (sa == NULL) {
-        return NXT_ERROR;
+    tmcf->conf->threads = 1;
+
+    mp = tmcf->conf->mem_pool;
+
+    sa = nxt_router_listen_sockaddr_stub(task, mp, 8000);
+    skcf = nxt_router_socket_conf(task, mp, sa);
+
+    skcf->listen.handler = nxt_router_conn_init;
+    skcf->listen.mem_pool_size = nxt_listen_socket_pool_min_size(&skcf->listen)
+                        + sizeof(nxt_event_conn_proxy_t)
+                        + sizeof(nxt_event_conn_t)
+                        + 4 * sizeof(nxt_buf_t);
+
+    skcf->header_buffer_size = 2048;
+    skcf->large_header_buffer_size = 8192;
+    skcf->header_read_timeout = 5000;
+
+    nxt_queue_insert_tail(&tmcf->pending, &skcf->link);
+
+    sa = nxt_router_listen_sockaddr_stub(task, mp, 8001);
+    skcf = nxt_router_socket_conf(task, mp, sa);
+
+    skcf->listen.handler = nxt_stream_connection_init;
+    skcf->listen.mem_pool_size = nxt_listen_socket_pool_min_size(&skcf->listen)
+                        + sizeof(nxt_event_conn_proxy_t)
+                        + sizeof(nxt_event_conn_t)
+                        + 4 * sizeof(nxt_buf_t);
+
+    skcf->header_read_timeout = 5000;
+
+    nxt_queue_insert_tail(&tmcf->pending, &skcf->link);
+
+    return NXT_OK;
+}
+
+
+static nxt_socket_conf_t *
+nxt_router_socket_conf(nxt_task_t *task, nxt_mem_pool_t *mp, nxt_sockaddr_t *sa)
+{
+    nxt_socket_conf_t  *conf;
+
+    conf = nxt_mem_zalloc(mp, sizeof(nxt_socket_conf_t));
+    if (nxt_slow_path(conf == NULL)) {
+        return NULL;
+    }
+
+    conf->listen.sockaddr = sa;
+
+    conf->listen.socket = -1;
+    conf->listen.backlog = NXT_LISTEN_BACKLOG;
+    conf->listen.flags = NXT_NONBLOCK;
+    conf->listen.read_after_accept = 1;
+
+    return conf;
+}
+
+
+static nxt_sockaddr_t *
+nxt_router_listen_sockaddr_stub(nxt_task_t *task, nxt_mem_pool_t *mp,
+    uint32_t port)
+{
+    nxt_sockaddr_t      *sa;
+    struct sockaddr_in  sin;
+
+    nxt_memzero(&sin, sizeof(struct sockaddr_in));
+
+    sin.sin_family = AF_INET;
+    sin.sin_port = htons(port);
+
+    sa = nxt_sockaddr_create(mp, (struct sockaddr *) &sin,
+                             sizeof(struct sockaddr_in), NXT_INET_ADDR_STR_LEN);
+    if (nxt_slow_path(sa == NULL)) {
+        return NULL;
     }
 
     sa->type = SOCK_STREAM;
-    sa->u.sockaddr_in.sin_family = AF_INET;
-    sa->u.sockaddr_in.sin_port = htons(8000);
 
     nxt_sockaddr_text(sa);
 
-    ls = nxt_runtime_listen_socket_add(rt, sa);
-    if (ls == NULL) {
+    return sa;
+}
+
+
+static void
+nxt_router_listen_sockets_sort(nxt_router_t *router,
+    nxt_router_temp_conf_t *tmcf)
+{
+    nxt_queue_link_t   *nqlk, *oqlk, *next;
+    nxt_socket_conf_t  *nskcf, *oskcf;
+
+    for (nqlk = nxt_queue_first(&tmcf->pending);
+         nqlk != nxt_queue_tail(&tmcf->pending);
+         nqlk = next)
+    {
+        next = nxt_queue_next(nqlk);
+        nskcf = nxt_queue_link_data(nqlk, nxt_socket_conf_t, link);
+
+        for (oqlk = nxt_queue_first(&router->sockets);
+             oqlk != nxt_queue_tail(&router->sockets);
+             oqlk = nxt_queue_next(oqlk))
+        {
+            oskcf = nxt_queue_link_data(oqlk, nxt_socket_conf_t, link);
+
+            if (nxt_sockaddr_cmp(nskcf->listen.sockaddr,
+                                 oskcf->listen.sockaddr))
+            {
+                nxt_queue_remove(oqlk);
+                nxt_queue_insert_tail(&tmcf->keeping, oqlk);
+
+                nxt_queue_remove(nqlk);
+                nxt_queue_insert_tail(&tmcf->updating, nqlk);
+
+                break;
+            }
+        }
+    }
+
+    nxt_queue_add(&tmcf->deleting, &router->sockets);
+}
+
+
+static nxt_int_t
+nxt_router_listen_sockets_stub_create(nxt_task_t *task,
+    nxt_router_temp_conf_t *tmcf)
+{
+    nxt_queue_link_t   *qlk, *nqlk;
+    nxt_socket_conf_t  *skcf;
+
+    for (qlk = nxt_queue_first(&tmcf->pending);
+         qlk != nxt_queue_tail(&tmcf->pending);
+         qlk = nqlk)
+    {
+        skcf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
+
+        if (nxt_listen_socket_create(task, &skcf->listen, 0) != NXT_OK) {
+            return NXT_ERROR;
+        }
+
+        nqlk = nxt_queue_next(qlk);
+        nxt_queue_remove(qlk);
+        nxt_queue_insert_tail(&tmcf->creating, qlk);
+    }
+
+    return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_router_engines_create(nxt_task_t *task, nxt_router_t *router,
+    nxt_router_temp_conf_t *tmcf, const nxt_event_interface_t *interface)
+{
+    nxt_int_t                 ret;
+    nxt_uint_t                n, threads;
+    nxt_mem_pool_t            *mp;
+    nxt_queue_link_t          *qlk;
+    nxt_router_engine_conf_t  *recf;
+
+    mp = tmcf->conf->mem_pool;
+    threads = tmcf->conf->threads;
+
+    tmcf->engines = nxt_array_create(tmcf->mem_pool, threads,
+                                     sizeof(nxt_router_engine_conf_t));
+    if (nxt_slow_path(tmcf->engines == NULL)) {
         return NXT_ERROR;
     }
 
-    ls->read_after_accept = 1;
+    n = 0;
 
-    ls->flags = NXT_NONBLOCK;
+    for (qlk = nxt_queue_first(&router->engines);
+         qlk != nxt_queue_tail(&router->engines);
+         qlk = nxt_queue_next(qlk))
+    {
+        recf = nxt_array_zero_add(tmcf->engines);
+        if (nxt_slow_path(recf == NULL)) {
+            return NXT_ERROR;
+        }
 
-#if 0
-    /* STUB */
-    wq = nxt_mem_zalloc(cf->mem_pool, sizeof(nxt_work_queue_t));
-    if (wq == NULL) {
+        recf->engine = nxt_queue_link_data(qlk, nxt_event_engine_t, link);
+        // STUB
+        recf->task = recf->engine->task;
+
+        if (n < threads) {
+            ret = nxt_router_engine_conf_update(task, mp, tmcf, recf);
+
+        } else {
+            ret = nxt_router_engine_conf_delete(task, mp, tmcf, recf);
+        }
+
+        if (nxt_slow_path(ret != NXT_OK)) {
+            return ret;
+        }
+
+        n++;
+    }
+
+    tmcf->new_threads = n;
+
+    while (n < threads) {
+        recf = nxt_array_zero_add(tmcf->engines);
+        if (nxt_slow_path(recf == NULL)) {
+            return NXT_ERROR;
+        }
+
+        recf->engine = nxt_event_engine_create(task, interface, NULL, 0, 0);
+        if (nxt_slow_path(recf->engine == NULL)) {
+            return NXT_ERROR;
+        }
+        // STUB
+        recf->task = recf->engine->task;
+
+        ret = nxt_router_engine_conf_create(task, mp, tmcf, recf);
+        if (nxt_slow_path(ret != NXT_OK)) {
+            return ret;
+        }
+
+        n++;
+    }
+
+    return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_router_engine_conf_create(nxt_task_t *task, nxt_mem_pool_t *mp,
+    nxt_router_temp_conf_t *tmcf, nxt_router_engine_conf_t *recf)
+{
+    nxt_int_t  ret;
+
+    recf->creating = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t));
+    if (nxt_slow_path(recf->creating == NULL)) {
         return NXT_ERROR;
     }
-    nxt_work_queue_name(wq, "listen");
-    /**/
 
-    ls->work_queue = wq;
-#endif
-    ls->handler = nxt_stream_connection_init;
+    ret = nxt_router_engine_joints_create(task, mp, recf, &tmcf->creating,
+                            recf->creating, nxt_router_listen_socket_create);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
 
-    /*
-     * Connection memory pool chunk size is tunned to
-     * allocate the most data in one mem_pool chunk.
-     */
-    ls->mem_pool_size = nxt_listen_socket_pool_min_size(ls)
-                        + sizeof(nxt_event_conn_proxy_t)
-                        + sizeof(nxt_event_conn_t)
-                        + 4 * sizeof(nxt_buf_t);
+    return nxt_router_engine_joints_create(task, mp, recf, &tmcf->updating,
+                            recf->creating, nxt_router_listen_socket_create);
+}
+
+
+static nxt_int_t
+nxt_router_engine_conf_update(nxt_task_t *task, nxt_mem_pool_t *mp,
+    nxt_router_temp_conf_t *tmcf, nxt_router_engine_conf_t *recf)
+{
+    nxt_int_t  ret;
+
+    recf->creating = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t));
+    if (nxt_slow_path(recf->creating == NULL)) {
+        return NXT_ERROR;
+    }
+
+    ret = nxt_router_engine_joints_create(task, mp, recf, &tmcf->creating,
+                            recf->creating, nxt_router_listen_socket_create);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
+
+    recf->updating = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t));
+    if (nxt_slow_path(recf->updating == NULL)) {
+        return NXT_ERROR;
+    }
+
+    ret = nxt_router_engine_joints_create(task, mp, recf, &tmcf->updating,
+                            recf->updating, nxt_router_listen_socket_update);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
 
-    if (nxt_listen_socket_create(task, ls, 0) != NXT_OK) {
+    recf->deleting = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t));
+    if (nxt_slow_path(recf->deleting == NULL)) {
         return NXT_ERROR;
     }
 
-    if (nxt_event_conn_listen(task, ls) != NXT_OK) {
+    return nxt_router_engine_joints_delete(task, recf, &tmcf->deleting,
+                                           recf->deleting);
+}
+
+
+static nxt_int_t
+nxt_router_engine_conf_delete(nxt_task_t *task, nxt_mem_pool_t *mp,
+    nxt_router_temp_conf_t *tmcf, nxt_router_engine_conf_t *recf)
+{
+    nxt_int_t  ret;
+
+    recf->deleting = nxt_array_create(tmcf->mem_pool, 4, sizeof(nxt_work_t));
+    if (nxt_slow_path(recf->deleting == NULL)) {
         return NXT_ERROR;
     }
 
+    ret = nxt_router_engine_joints_delete(task, recf, &tmcf->updating,
+                                          recf->deleting);
+    if (nxt_slow_path(ret != NXT_OK)) {
+        return ret;
+    }
+
+    return nxt_router_engine_joints_delete(task, recf, &tmcf->deleting,
+                                           recf->deleting);
+}
+
+
+static nxt_int_t
+nxt_router_engine_joints_create(nxt_task_t *task, nxt_mem_pool_t *mp,
+    nxt_router_engine_conf_t *recf, nxt_queue_t *sockets, nxt_array_t *array,
+    nxt_work_handler_t handler)
+{
+    nxt_work_t               *work;
+    nxt_queue_link_t         *qlk;
+    nxt_socket_conf_joint_t  *joint;
+
+    for (qlk = nxt_queue_first(sockets);
+         qlk != nxt_queue_tail(sockets);
+         qlk = nxt_queue_next(qlk))
+    {
+        work = nxt_array_add(array);
+        if (nxt_slow_path(work == NULL)) {
+            return NXT_ERROR;
+        }
+
+        work->next = NULL;
+        work->handler = handler;
+        work->task = &recf->task;
+        work->obj = recf->engine;
+
+        joint = nxt_mem_alloc(mp, sizeof(nxt_socket_conf_joint_t));
+        if (nxt_slow_path(joint == NULL)) {
+            return NXT_ERROR;
+        }
+
+        work->data = joint;
+
+        joint->count = 1;
+        joint->socket_conf = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
+    }
+
+    return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_router_engine_joints_delete(nxt_task_t *task,
+    nxt_router_engine_conf_t *recf, nxt_queue_t *sockets, nxt_array_t *array)
+{
+    nxt_work_t        *work;
+    nxt_queue_link_t  *qlk;
+
+    for (qlk = nxt_queue_first(sockets);
+         qlk != nxt_queue_tail(sockets);
+         qlk = nxt_queue_next(qlk))
+    {
+        work = nxt_array_add(array);
+        if (nxt_slow_path(work == NULL)) {
+            return NXT_ERROR;
+        }
+
+        work->next = NULL;
+        work->handler = nxt_router_listen_socket_delete;
+        work->task = &recf->task;
+        work->obj = recf->engine;
+        work->data = nxt_queue_link_data(qlk, nxt_socket_conf_t, link);
+    }
+
     return NXT_OK;
 }
+
+
+static nxt_int_t
+nxt_router_threads_create(nxt_task_t *task, nxt_runtime_t *rt,
+    nxt_router_temp_conf_t *tmcf)
+{
+    nxt_int_t                 ret;
+    nxt_uint_t                i, threads;
+    nxt_router_engine_conf_t  *recf;
+
+    recf = tmcf->engines->elts;
+    threads = tmcf->conf->threads;
+
+    for (i = tmcf->new_threads; i < threads; i++) {
+        ret = nxt_router_thread_create(task, rt, recf[i].engine);
+        if (nxt_slow_path(ret != NXT_OK)) {
+            return ret;
+        }
+    }
+
+    return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
+    nxt_event_engine_t *engine)
+{
+    nxt_int_t            ret;
+    nxt_thread_link_t    *link;
+    nxt_thread_handle_t  handle;
+
+    link = nxt_zalloc(sizeof(nxt_thread_link_t));
+
+    if (nxt_slow_path(link == NULL)) {
+        return NXT_ERROR;
+    }
+
+    link->start = nxt_router_thread_start;
+    link->engine = engine;
+    link->work.handler = nxt_router_thread_exit_handler;
+    link->work.task = task;
+    link->work.data = link;
+
+    nxt_queue_insert_tail(&rt->engines, &engine->link);
+
+    ret = nxt_thread_create(&handle, link);
+
+    if (nxt_slow_path(ret != NXT_OK)) {
+        nxt_queue_remove(&engine->link);
+    }
+
+    return ret;
+}
+
+
+static void
+nxt_router_engines_post(nxt_router_temp_conf_t *tmcf)
+{
+    nxt_uint_t                n;
+    nxt_router_engine_conf_t  *recf;
+
+    recf = tmcf->engines->elts;
+
+    for (n = tmcf->engines->nelts; n != 0; n--) {
+        nxt_router_engine_post(recf);
+        recf++;
+    }
+}
+
+
+static void
+nxt_router_engine_post(nxt_router_engine_conf_t *recf)
+{
+    nxt_uint_t  n;
+    nxt_work_t  *work;
+
+    work = recf->creating->elts;
+
+    for (n = recf->creating->nelts; n != 0; n--) {
+        nxt_event_engine_post(recf->engine, work);
+        work++;
+    }
+}
+
+
+static void
+nxt_router_thread_start(void *data)
+{
+    nxt_thread_t        *thread;
+    nxt_thread_link_t   *link;
+    nxt_event_engine_t  *engine;
+
+    link = data;
+    engine = link->engine;
+
+    thread = nxt_thread();
+
+    /* STUB */
+    thread->runtime = engine->task.thread->runtime;
+
+    engine->task.thread = thread;
+    engine->task.log = thread->log;
+    thread->engine = engine;
+    thread->fiber = &engine->fibers->fiber;
+
+    engine->mem_pool = nxt_mem_cache_pool_create(4096, 1024, 1024, 64);
+
+    nxt_event_engine_start(engine);
+}
+
+
+static void
+nxt_router_listen_socket_create(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_listen_event_t       *listen;
+    nxt_event_engine_t       *engine;
+    nxt_listen_socket_t      *ls;
+    nxt_socket_conf_joint_t  *joint;
+
+    engine = obj;
+    joint = data;
+
+    ls = &joint->socket_conf->listen;
+
+    listen = nxt_listen_event(task, ls);
+    if (nxt_slow_path(listen == NULL)) {
+        nxt_router_listen_socket_release(task, joint);
+        return;
+    }
+
+    listen->socket.data = joint;
+}
+
+
+nxt_inline nxt_listen_event_t *
+nxt_router_listen_event(nxt_queue_t *listen_connections,
+    nxt_socket_conf_t *skcf)
+{
+    nxt_socket_t        socket;
+    nxt_queue_link_t    *link;
+    nxt_listen_event_t  *listen;
+
+    socket = skcf->listen.socket;
+
+    for (link = nxt_queue_first(listen_connections);
+         link != nxt_queue_tail(listen_connections);
+         link = nxt_queue_next(link))
+    {
+        listen = nxt_queue_link_data(link, nxt_listen_event_t, link);
+
+        if (socket == listen->socket.fd) {
+            return listen;
+        }
+    }
+
+    return NULL;
+}
+
+
+static void
+nxt_router_listen_socket_update(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_event_engine_t       *engine;
+    nxt_listen_event_t       *listen;
+    nxt_socket_conf_joint_t  *joint, *old;
+
+    engine = obj;
+    joint = data;
+
+    listen = nxt_router_listen_event(&engine->listen_connections,
+                                     joint->socket_conf);
+
+    old = listen->socket.data;
+    listen->socket.data = joint;
+
+    nxt_router_conf_release(task, old);
+}
+
+
+static void
+nxt_router_listen_socket_delete(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_socket_conf_t   *skcf;
+    nxt_listen_event_t  *listen;
+    nxt_event_engine_t  *engine;
+
+    engine = obj;
+    skcf = data;
+
+    listen = nxt_router_listen_event(&engine->listen_connections, skcf);
+
+    nxt_fd_event_delete(engine, &listen->socket);
+
+    listen->timer.handler = nxt_router_listen_socket_close;
+    listen->timer.work_queue = &engine->fast_work_queue;
+
+    nxt_timer_add(engine, &listen->timer, 0);
+}
+
+
+static void
+nxt_router_listen_socket_close(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_timer_t              *timer;
+    nxt_listen_event_t       *listen;
+    nxt_socket_conf_joint_t  *joint;
+
+    timer = obj;
+    listen = nxt_timer_data(timer, nxt_listen_event_t, timer);
+    joint = listen->socket.data;
+
+    nxt_queue_remove(&listen->link);
+    nxt_free(listen);
+
+    nxt_router_listen_socket_release(task, joint);
+}
+
+
+static void
+nxt_router_listen_socket_release(nxt_task_t *task,
+    nxt_socket_conf_joint_t *joint)
+{
+    nxt_socket_t           s;
+    nxt_listen_socket_t    *ls;
+    nxt_thread_spinlock_t  *lock;
+
+    s = -1;
+    ls = &joint->socket_conf->listen;
+    lock = &joint->socket_conf->router_conf->router->lock;
+
+    nxt_thread_spin_lock(lock);
+
+    if (--ls->count == 0) {
+        s = ls->socket;
+        ls->socket = -1;
+    }
+
+    nxt_thread_spin_unlock(lock);
+
+    if (s != -1) {
+        nxt_socket_close(task, s);
+    }
+
+    nxt_router_conf_release(task, joint);
+}
+
+
+static void
+nxt_router_conf_release(nxt_task_t *task, nxt_socket_conf_joint_t *joint)
+{
+    nxt_socket_conf_t      *skcf;
+    nxt_router_conf_t      *rtcf;
+    nxt_thread_spinlock_t  *lock;
+
+    nxt_debug(task, "conf joint count: %D", joint->count);
+
+    if (--joint->count != 0) {
+        return;
+    }
+
+    nxt_queue_remove(&joint->link);
+
+    skcf = joint->socket_conf;
+    rtcf = skcf->router_conf;
+    lock = &rtcf->router->lock;
+
+    nxt_thread_spin_lock(lock);
+
+    if (--skcf->count != 0) {
+        rtcf = NULL;
+
+    } else {
+        nxt_queue_remove(&skcf->link);
+
+        if (--rtcf->count != 0) {
+            rtcf = NULL;
+        }
+    }
+
+    nxt_thread_spin_unlock(lock);
+
+    if (rtcf != NULL) {
+        nxt_mem_pool_destroy(rtcf->mem_pool);
+    }
+
+    if (nxt_queue_is_empty(&joint->engine->joints)) {
+        nxt_thread_exit(task->thread);
+    }
+}
+
+
+static void
+nxt_router_thread_exit_handler(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_thread_link_t    *link;
+    nxt_event_engine_t   *engine;
+    nxt_thread_handle_t  handle;
+
+    handle = obj;
+    link = data;
+
+    nxt_thread_wait(handle);
+
+    engine = link->engine;
+
+    nxt_queue_remove(&engine->link);
+
+    nxt_mem_cache_pool_destroy(engine->mem_pool);
+
+    nxt_event_engine_free(engine);
+
+    nxt_free(link);
+
+    // TODO: free port
+}
+
+
+static const nxt_event_conn_state_t  nxt_router_conn_read_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_router_conn_http_header_parse,
+    .close_handler = nxt_router_conn_close,
+    .error_handler = nxt_router_conn_error,
+
+    .timer_handler = nxt_router_conn_timeout,
+    .timer_value = nxt_router_conn_timeout_value,
+    .timer_data = offsetof(nxt_socket_conf_t, header_read_timeout),
+};
+
+
+static void
+nxt_router_conn_init(nxt_task_t *task, void *obj, void *data)
+{
+    size_t                   size;
+    nxt_event_conn_t         *c;
+    nxt_event_engine_t       *engine;
+    nxt_socket_conf_joint_t  *joint;
+
+    c = obj;
+    joint = data;
+
+    nxt_debug(task, "router conn init");
+
+    joint->count++;
+
+    size = joint->socket_conf->header_buffer_size;
+    c->read = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+
+    c->socket.data = NULL;
+
+    engine = task->thread->engine;
+    c->read_work_queue = &engine->fast_work_queue;
+    c->write_work_queue = &engine->fast_work_queue;
+
+    c->read_state = &nxt_router_conn_read_state;
+
+    nxt_event_conn_read(engine, c);
+}
+
+
+static const nxt_event_conn_state_t  nxt_router_conn_write_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_router_conn_close,
+    .close_handler = nxt_router_conn_close,
+    .error_handler = nxt_router_conn_error,
+};
+
+
+static void
+nxt_router_conn_http_header_parse(nxt_task_t *task, void *obj, void *data)
+{
+    size_t                    size;
+    nxt_int_t                 ret;
+    nxt_buf_t                 *b;
+    nxt_event_conn_t          *c;
+    nxt_socket_conf_joint_t   *joint;
+    nxt_http_request_parse_t  *rp;
+
+    c = obj;
+    rp = data;
+
+    nxt_debug(task, "router conn http header parse");
+
+    if (rp == NULL) {
+        rp = nxt_mem_zalloc(c->mem_pool, sizeof(nxt_http_request_parse_t));
+        if (nxt_slow_path(rp == NULL)) {
+            nxt_router_conn_close(task, c, data);
+            return;
+        }
+
+        c->socket.data = rp;
+    }
+
+    ret = nxt_http_parse_request(rp, &c->read->mem);
+
+    nxt_debug(task, "http parse request: %d", ret);
+
+    switch (nxt_expect(NXT_DONE, ret)) {
+
+    case NXT_DONE:
+        break;
+
+    case NXT_ERROR:
+        nxt_router_conn_close(task, c, data);
+        return;
+
+    default:  /* NXT_AGAIN */
+
+        if (c->read->mem.free == c->read->mem.end) {
+            joint = c->listen->socket.data;
+            size = joint->socket_conf->large_header_buffer_size,
+
+            b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+            if (nxt_slow_path(b == NULL)) {
+                nxt_router_conn_close(task, c, data);
+                return;
+            }
+
+            size = c->read->mem.free - c->read->mem.pos;
+            nxt_memcpy(b->mem.pos, c->read->mem.pos, size);
+
+            b->mem.free += size;
+            c->read = b;
+        }
+
+        nxt_event_conn_read(task->thread->engine, c);
+        return;
+    }
+
+    c->write = c->read;
+    c->write->mem.pos = c->write->mem.start;
+    c->write_state = &nxt_router_conn_write_state;
+
+    nxt_event_conn_write(task->thread->engine, c);
+}
+
+
+static const nxt_event_conn_state_t  nxt_router_conn_close_state
+    nxt_aligned(64) =
+{
+    .ready_handler = nxt_router_conn_free,
+};
+
+
+static void
+nxt_router_conn_close(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_event_conn_t  *c;
+
+    c = obj;
+
+    nxt_debug(task, "router conn close");
+
+    c->write_state = &nxt_router_conn_close_state;
+
+    nxt_event_conn_close(task->thread->engine, c);
+}
+
+
+static void
+nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_event_conn_t         *c;
+    nxt_socket_conf_joint_t  *joint;
+
+    c = obj;
+
+    nxt_debug(task, "router conn close done");
+
+    joint = c->listen->socket.data;
+    nxt_router_conf_release(task, joint);
+
+    nxt_mem_pool_destroy(c->mem_pool);
+}
+
+
+static void
+nxt_router_conn_error(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_event_conn_t  *c;
+
+    c = obj;
+
+    nxt_debug(task, "router conn error");
+
+    c->write_state = &nxt_router_conn_close_state;
+
+    nxt_event_conn_close(task->thread->engine, c);
+}
+
+
+static void
+nxt_router_conn_timeout(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_timer_t       *timer;
+    nxt_event_conn_t  *c;
+
+    timer = obj;
+
+    nxt_debug(task, "router conn timeout");
+
+    c = nxt_event_read_timer_conn(timer);
+
+    c->write_state = &nxt_router_conn_close_state;
+
+    nxt_event_conn_close(task->thread->engine, c);
+}
+
+
+static nxt_msec_t
+nxt_router_conn_timeout_value(nxt_event_conn_t *c, uintptr_t data)
+{
+    nxt_socket_conf_joint_t  *joint;
+
+    joint = c->listen->socket.data;
+
+    return nxt_value_at(nxt_msec_t, joint->socket_conf, data);
+}