Using event engine memory buffers in HTTP/1 layer.

4 files changed, 18 insertions, 15 deletions

diff --git a/src/nxt_conn.h b/src/nxt_conn.h
index 7284808b..2c1d49a0 100644
--- a/src/nxt_conn.h
+++ b/src/nxt_conn.h
@@ -8,7 +8,7 @@
 #define _NXT_CONN_H_INCLUDED_
 
 
-typedef ssize_t (*nxt_conn_io_read_t)(nxt_conn_t *c);
+typedef ssize_t (*nxt_conn_io_read_t)(nxt_task_t *task, nxt_conn_t *c);
 typedef nxt_msec_t (*nxt_conn_timer_value_t)(nxt_conn_t *c, uintptr_t data);
 
 
diff --git a/src/nxt_conn_read.c b/src/nxt_conn_read.c
index 83969b31..3285abcd 100644
--- a/src/nxt_conn_read.c
+++ b/src/nxt_conn_read.c
@@ -69,7 +69,7 @@ nxt_conn_io_read(nxt_task_t *task, void *obj, void *data)
             n = c->io->recvbuf(c, c->read);
 
         } else {
-            n = state->io_read_handler(c);
+            n = state->io_read_handler(task, c);
             /* The state can be changed by io_read_handler. */
             state = c->read_state;
         }
diff --git a/src/nxt_h1proto.c b/src/nxt_h1proto.c
index 6d24bade..0e70a670 100644
--- a/src/nxt_h1proto.c
+++ b/src/nxt_h1proto.c
@@ -18,10 +18,10 @@
  */
 
 #if (NXT_TLS)
-static ssize_t nxt_http_idle_io_read_handler(nxt_conn_t *c);
+static ssize_t nxt_http_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
 static void nxt_http_conn_test(nxt_task_t *task, void *obj, void *data);
 #endif
-static ssize_t nxt_h1p_idle_io_read_handler(nxt_conn_t *c);
+static ssize_t nxt_h1p_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
 static void nxt_h1p_conn_proto_init(nxt_task_t *task, void *obj, void *data);
 static void nxt_h1p_conn_request_init(nxt_task_t *task, void *obj, void *data);
 static void nxt_h1p_conn_request_header_parse(nxt_task_t *task, void *obj,
@@ -196,7 +196,7 @@ static const nxt_conn_state_t  nxt_http_idle_state
 
 
 static ssize_t
-nxt_http_idle_io_read_handler(nxt_conn_t *c)
+nxt_http_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
 {
     size_t                   size;
     ssize_t                  n;
@@ -216,7 +216,7 @@ nxt_http_idle_io_read_handler(nxt_conn_t *c)
 
     size = joint->socket_conf->header_buffer_size;
 
-    b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+    b = nxt_event_engine_buf_mem_alloc(task->thread->engine, size);
     if (nxt_slow_path(b == NULL)) {
         c->socket.error = NXT_ENOMEM;
         return NXT_ERROR;
@@ -234,7 +234,7 @@ nxt_http_idle_io_read_handler(nxt_conn_t *c)
 
     } else {
         c->read = NULL;
-        nxt_mp_free(c->mem_pool, b);
+        nxt_event_engine_buf_mem_free(task->thread->engine, b);
     }
 
     return n;
@@ -248,12 +248,14 @@ nxt_http_conn_test(nxt_task_t *task, void *obj, void *data)
     nxt_buf_t                *b;
     nxt_conn_t               *c;
     nxt_tls_conf_t           *tls;
+    nxt_event_engine_t       *engine;
     nxt_socket_conf_joint_t  *joint;
 
     c = obj;
 
     nxt_debug(task, "h1p conn https test");
 
+    engine = task->thread->engine;
     b = c->read;
     p = b->mem.pos;
 
@@ -262,7 +264,7 @@ nxt_http_conn_test(nxt_task_t *task, void *obj, void *data)
     if (p[0] != 0x16) {
         b->mem.free = b->mem.pos;
 
-        nxt_conn_read(task->thread->engine, c);
+        nxt_conn_read(engine, c);
         return;
     }
 
@@ -292,7 +294,7 @@ nxt_http_conn_test(nxt_task_t *task, void *obj, void *data)
 #endif
 
     c->read = NULL;
-    nxt_mp_free(c->mem_pool, b);
+    nxt_event_engine_buf_mem_free(engine, b);
 
     joint = c->listen->socket.data;
 
@@ -330,7 +332,7 @@ static const nxt_conn_state_t  nxt_h1p_idle_state
 
 
 static ssize_t
-nxt_h1p_idle_io_read_handler(nxt_conn_t *c)
+nxt_h1p_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
 {
     size_t                   size;
     ssize_t                  n;
@@ -353,7 +355,7 @@ nxt_h1p_idle_io_read_handler(nxt_conn_t *c)
     if (b == NULL) {
         size = joint->socket_conf->header_buffer_size;
 
-        b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
+        b = nxt_event_engine_buf_mem_alloc(task->thread->engine, size);
         if (nxt_slow_path(b == NULL)) {
             c->socket.error = NXT_ENOMEM;
             return NXT_ERROR;
@@ -367,7 +369,7 @@ nxt_h1p_idle_io_read_handler(nxt_conn_t *c)
 
     } else {
         c->read = NULL;
-        nxt_mp_free(c->mem_pool, b);
+        nxt_event_engine_buf_mem_free(task->thread->engine, b);
     }
 
     return n;
@@ -1232,7 +1234,8 @@ nxt_h1p_complete_buffers(nxt_task_t *task, nxt_h1proto_t *h1p)
         size = nxt_buf_mem_used_size(&in->mem);
 
         if (size == 0) {
-            nxt_mp_free(c->mem_pool, in);
+            nxt_work_queue_add(&task->thread->engine->fast_work_queue,
+                               in->completion_handler, task, in, in->parent);
 
             c->read = NULL;
         }
diff --git a/src/nxt_h1proto_websocket.c b/src/nxt_h1proto_websocket.c
index 13754be0..cd67a8a6 100644
--- a/src/nxt_h1proto_websocket.c
+++ b/src/nxt_h1proto_websocket.c
@@ -26,7 +26,7 @@ static void nxt_h1p_conn_ws_keepalive_enable(nxt_task_t *task,
 static void nxt_h1p_conn_ws_frame_process(nxt_task_t *task, nxt_conn_t *c,
     nxt_h1proto_t *h1p, nxt_websocket_header_t *wsh);
 static void nxt_h1p_conn_ws_error(nxt_task_t *task, void *obj, void *data);
-static ssize_t nxt_h1p_ws_io_read_handler(nxt_conn_t *c);
+static ssize_t nxt_h1p_ws_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
 static void nxt_h1p_conn_ws_timeout(nxt_task_t *task, void *obj, void *data);
 static void nxt_h1p_conn_ws_frame_payload_read(nxt_task_t *task, void *obj,
     void *data);
@@ -473,7 +473,7 @@ nxt_h1p_conn_ws_error(nxt_task_t *task, void *obj, void *data)
 
 
 static ssize_t
-nxt_h1p_ws_io_read_handler(nxt_conn_t *c)
+nxt_h1p_ws_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
 {
     size_t     size;
     ssize_t    n;