Event engines refactoring.

Author: igorsysoev

auto/sources

@@ -55,10 +55,9 @@ NXT_LIB_DEPS=" \
     src/nxt_service.h \
     src/nxt_fiber.h \
     src/nxt_log_moderation.h \
-    src/nxt_event_set.h \
     src/nxt_event_engine.h \
     src/nxt_timer.h \
-    src/nxt_event_fd.h \
+    src/nxt_fd_event.h \
     src/nxt_event_conn.h \
     src/nxt_event_file.h \
     src/nxt_job.h \
@@ -113,9 +112,9 @@ NXT_LIB_SRCS=" \
     src/nxt_service.c \
     src/nxt_fiber.c \
     src/nxt_log_moderation.c \
-    src/nxt_event_set.c \
     src/nxt_event_engine.c \
     src/nxt_timer.c \
+    src/nxt_fd_event.c \
     src/nxt_event_conn.c \
     src/nxt_event_conn_connect.c \
     src/nxt_event_conn_accept.c \
@@ -163,13 +162,13 @@ NXT_LIB_GNUTLS_SRCS="src/nxt_gnutls.c"
 NXT_LIB_CYASSL_SRCS="src/nxt_cyassl.c"
 NXT_LIB_POLARSSL_SRCS="src/nxt_polarssl.c"
 
-NXT_LIB_EPOLL_SRCS="src/nxt_epoll.c"
-NXT_LIB_KQUEUE_SRCS="src/nxt_kqueue.c"
-NXT_LIB_EVENTPORT_SRCS="src/nxt_eventport.c"
-NXT_LIB_DEVPOLL_SRCS="src/nxt_devpoll.c"
-NXT_LIB_POLLSET_SRCS="src/nxt_pollset.c"
-NXT_LIB_POLL_SRCS="src/nxt_poll.c"
-NXT_LIB_SELECT_SRCS="src/nxt_select.c"
+NXT_LIB_EPOLL_SRCS="src/nxt_epoll_engine.c"
+NXT_LIB_KQUEUE_SRCS="src/nxt_kqueue_engine.c"
+NXT_LIB_EVENTPORT_SRCS="src/nxt_eventport_engine.c"
+NXT_LIB_DEVPOLL_SRCS="src/nxt_devpoll_engine.c"
+NXT_LIB_POLLSET_SRCS="src/nxt_pollset_engine.c"
+NXT_LIB_POLL_SRCS="src/nxt_poll_engine.c"
+NXT_LIB_SELECT_SRCS="src/nxt_select_engine.c"
 
 NXT_LIB_LINUX_SENDFILE_SRCS="src/nxt_linux_sendfile.c"
 NXT_LIB_FREEBSD_SENDFILE_SRCS="src/nxt_freebsd_sendfile.c"

src/nxt_application.c

@@ -516,7 +516,7 @@ nxt_app_conn_update(nxt_thread_t *thr, nxt_event_conn_t *c, nxt_log_t *log)
     c->task.log = &c->log;
     c->task.ident = c->log.ident;
 
-    c->io = thr->engine->event->io;
+    c->io = thr->engine->event.io;
     c->max_chunk = NXT_INT32_T_MAX;
     c->sendfile = NXT_CONN_SENDFILE_UNSET;
 
@@ -831,10 +831,27 @@ nxt_app_delivery_ready(nxt_task_t *task, void *obj, void *data)
 }
 
 
+static const nxt_event_conn_state_t  nxt_app_delivery_close_state
+    nxt_aligned(64) =
+{
+    NXT_EVENT_NO_BUF_PROCESS,
+    NXT_EVENT_TIMER_NO_AUTORESET,
+
+    nxt_app_close_request,
+    NULL,
+    NULL,
+
+    NULL,
+    NULL,
+    0,
+};
+
+
 static void
 nxt_app_delivery_completion(nxt_task_t *task, void *obj, void *data)
 {
     nxt_buf_t          *b, *bn, *free;
+    nxt_event_conn_t   *c;
     nxt_app_request_t  *r;
 
     nxt_debug(task, "app delivery completion");
@@ -857,8 +874,10 @@ nxt_app_delivery_completion(nxt_task_t *task, void *obj, void *data)
         if (nxt_buf_is_last(b)) {
             r = (nxt_app_request_t *) b->parent;
 
-            nxt_work_queue_add(&task->thread->engine->final_work_queue,
-                               nxt_app_close_request, task, r, NULL);
+            c = r->event_conn;
+            c->write_state = &nxt_app_delivery_close_state;
+
+            nxt_event_conn_close(task->thread->engine, c);
         }
     }
 
@@ -940,12 +959,11 @@ nxt_app_close_request(nxt_task_t *task, void *obj, void *data)
     nxt_event_conn_t   *c;
     nxt_app_request_t  *r;
 
-    r = obj;
-    c = r->event_conn;
+    c = obj;
 
     nxt_debug(task, "app close connection");
 
-    nxt_event_conn_close(task, c);
+    r = c->socket.data;
 
     nxt_mem_pool_destroy(c->mem_pool);
     nxt_mem_pool_destroy(r->mem_pool);

src/nxt_cycle.c

@@ -297,8 +297,8 @@ nxt_cycle_systemd_listen_sockets(nxt_thread_t *thr, nxt_cycle_t *cycle)
 static nxt_int_t
 nxt_cycle_event_engines(nxt_thread_t *thr, nxt_cycle_t *cycle)
 {
-    nxt_event_engine_t         *engine, **e, **engines;
-    const nxt_event_set_ops_t  *event_set;
+    nxt_event_engine_t           *engine, **e, **engines;
+    const nxt_event_interface_t  *interface;
 
     cycle->engines = nxt_array_create(cycle->mem_pool, 1,
                                       sizeof(nxt_event_engine_t *));
@@ -318,14 +318,14 @@ nxt_cycle_event_engines(nxt_thread_t *thr, nxt_cycle_t *cycle)
         *e = engines[0];
 
     } else {
-        event_set = nxt_service_get(cycle->services, "engine", NULL);
+        interface = nxt_service_get(cycle->services, "engine", NULL);
 
-        if (nxt_slow_path(event_set == NULL)) {
+        if (nxt_slow_path(interface == NULL)) {
             /* TODO: log */
             return NXT_ERROR;
         }
 
-        engine = nxt_event_engine_create(thr, event_set,
+        engine = nxt_event_engine_create(thr, interface,
                                          nxt_master_process_signals, 0, 0);
 
         if (nxt_slow_path(engine == NULL)) {
@@ -464,9 +464,9 @@ nxt_cycle_start(nxt_task_t *task, void *obj, void *data)
 static void
 nxt_cycle_initial_start(nxt_task_t *task, nxt_cycle_t *cycle)
 {
-    nxt_int_t                  ret;
-    nxt_thread_t               *thr;
-    const nxt_event_set_ops_t  *event_set;
+    nxt_int_t                    ret;
+    nxt_thread_t                 *thr;
+    const nxt_event_interface_t  *interface;
 
     thr = task->thread;
 
@@ -482,12 +482,12 @@ nxt_cycle_initial_start(nxt_task_t *task, nxt_cycle_t *cycle)
          * 1) inherited kqueue descriptor is invalid,
          * 2) the signal thread is not inherited.
          */
-        event_set = nxt_service_get(cycle->services, "engine", cycle->engine);
-        if (event_set == NULL) {
+        interface = nxt_service_get(cycle->services, "engine", cycle->engine);
+        if (interface == NULL) {
             goto fail;
         }
 
-        ret = nxt_event_engine_change(thr, task, event_set, cycle->batch);
+        ret = nxt_event_engine_change(thr, task, interface, cycle->batch);
         if (ret != NXT_OK) {
             goto fail;
         }
@@ -608,7 +608,7 @@ nxt_cycle_close_idle_connections(nxt_thread_t *thr, nxt_task_t *task)
 
         if (!c->socket.read_ready) {
             nxt_queue_remove(link);
-            nxt_event_conn_close(task, c);
+            nxt_event_conn_close(thr->engine, c);
         }
     }
 }
@@ -635,7 +635,7 @@ nxt_cycle_exit(nxt_task_t *task, void *obj, void *data)
         nxt_cycle_pid_file_delete(cycle);
     }
 
-    if (!task->thread->engine->event->signal_support) {
+    if (!task->thread->engine->event.signal_support) {
         nxt_event_engine_signals_stop(task->thread->engine);
     }
 
@@ -650,17 +650,17 @@ static nxt_int_t
 nxt_cycle_event_engine_change(nxt_thread_t *thr, nxt_task_t *task,
     nxt_cycle_t *cycle)
 {
-    const nxt_event_set_ops_t  *event_set;
+    const nxt_event_interface_t  *interface;
 
     if (thr->engine->batch == cycle->batch
-        && nxt_strcmp(thr->engine->event->name, cycle->engine) == 0)
+        && nxt_strcmp(thr->engine->event.name, cycle->engine) == 0)
     {
         return NXT_OK;
     }
 
-    event_set = nxt_service_get(cycle->services, "engine", cycle->engine);
-    if (event_set != NULL) {
-        return nxt_event_engine_change(thr, task, event_set, cycle->batch);
+    interface = nxt_service_get(cycle->services, "engine", cycle->engine);
+    if (interface != NULL) {
+        return nxt_event_engine_change(thr, task, interface, cycle->batch);
     }
 
     return NXT_ERROR;
@@ -789,11 +789,11 @@ nxt_cycle_thread_pool_exit(nxt_task_t *task, void *obj, void *data)
 static nxt_int_t
 nxt_cycle_conf_init(nxt_thread_t *thr, nxt_cycle_t *cycle)
 {
-    nxt_int_t                  ret;
-    nxt_str_t                  *prefix;
-    nxt_file_t                 *file;
-    nxt_file_name_str_t        file_name;
-    const nxt_event_set_ops_t  *event_set;
+    nxt_int_t                    ret;
+    nxt_str_t                    *prefix;
+    nxt_file_t                   *file;
+    nxt_file_name_str_t          file_name;
+    const nxt_event_interface_t  *interface;
 
     cycle->daemon = 1;
     cycle->master_process = 1;
@@ -815,12 +815,12 @@ nxt_cycle_conf_init(nxt_thread_t *thr, nxt_cycle_t *cycle)
 
     /* An engine's parameters. */
 
-    event_set = nxt_service_get(cycle->services, "engine", cycle->engine);
-    if (event_set == NULL) {
+    interface = nxt_service_get(cycle->services, "engine", cycle->engine);
+    if (interface == NULL) {
         return NXT_ERROR;
     }
 
-    cycle->engine = event_set->name;
+    cycle->engine = interface->name;
 
     prefix = nxt_file_name_is_absolute(cycle->pid) ? NULL : cycle->prefix;