Moved message size to nxt_port_recv_msg_t for convenience.

4 files changed, 17 insertions, 35 deletions

diff --git a/src/nxt_port.h b/src/nxt_port.h
index c69f1ba9..458aca17 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -47,6 +47,7 @@ struct nxt_port_recv_msg_s {
     nxt_buf_t           *buf;
     nxt_port_t          *port;
     nxt_port_msg_t      port_msg;
+    size_t              size;
 };
 
 
diff --git a/src/nxt_port_memory.c b/src/nxt_port_memory.c
index 5832ca25..eabc2e2f 100644
--- a/src/nxt_port_memory.c
+++ b/src/nxt_port_memory.c
@@ -594,7 +594,7 @@ nxt_port_mmap_write(nxt_task_t *task, nxt_port_t *port,
 
 void
 nxt_port_mmap_read(nxt_task_t *task, nxt_port_t *port,
-    nxt_port_recv_msg_t *msg, size_t size)
+    nxt_port_recv_msg_t *msg)
 {
     nxt_buf_t            *b, **pb;
     nxt_port_mmap_msg_t  *end, *mmap_msg;
@@ -605,6 +605,7 @@ nxt_port_mmap_read(nxt_task_t *task, nxt_port_t *port,
     end = (nxt_port_mmap_msg_t *) b->mem.free;
 
     pb = &msg->buf;
+    msg->size = 0;
 
     while (mmap_msg < end) {
         nxt_debug(task, "mmap_msg={%D, %D, %D} from %PI",
@@ -619,6 +620,7 @@ nxt_port_mmap_read(nxt_task_t *task, nxt_port_t *port,
             break;
         }
 
+        msg->size += mmap_msg->size;
         pb = &(*pb)->next;
         mmap_msg++;
     }
diff --git a/src/nxt_port_memory.h b/src/nxt_port_memory.h
index 379201d0..332bd856 100644
--- a/src/nxt_port_memory.h
+++ b/src/nxt_port_memory.h
@@ -43,7 +43,7 @@ nxt_port_mmap_completion(nxt_task_t *task, nxt_work_queue_t *wq, nxt_buf_t *b) {
 
 void
 nxt_port_mmap_read(nxt_task_t *task, nxt_port_t *port,
-    nxt_port_recv_msg_t *msg, size_t size);
+    nxt_port_recv_msg_t *msg);
 
 enum nxt_port_method_e {
     NXT_PORT_METHOD_ANY = 0,
diff --git a/src/nxt_port_socket.c b/src/nxt_port_socket.c
index 4de528fd..71ee315e 100644
--- a/src/nxt_port_socket.c
+++ b/src/nxt_port_socket.c
@@ -10,7 +10,7 @@
 static void nxt_port_write_handler(nxt_task_t *task, void *obj, void *data);
 static void nxt_port_read_handler(nxt_task_t *task, void *obj, void *data);
 static void nxt_port_read_msg_process(nxt_task_t *task, nxt_port_t *port,
-    nxt_port_recv_msg_t *msg, size_t size);
+    nxt_port_recv_msg_t *msg);
 static nxt_buf_t *nxt_port_buf_alloc(nxt_port_t *port);
 static void nxt_port_buf_free(nxt_port_t *port, nxt_buf_t *b);
 static void nxt_port_error_handler(nxt_task_t *task, void *obj, void *data);
@@ -365,8 +365,9 @@ nxt_port_read_handler(nxt_task_t *task, void *obj, void *data)
         if (n > 0) {
 
             msg.buf = b;
+            msg.size = n;
 
-            nxt_port_read_msg_process(task, port, &msg, n);
+            nxt_port_read_msg_process(task, port, &msg);
 
             if (b->mem.pos == b->mem.free) {
                 nxt_port_buf_free(port, b);
@@ -397,53 +398,32 @@ nxt_port_read_handler(nxt_task_t *task, void *obj, void *data)
 
 static void
 nxt_port_read_msg_process(nxt_task_t *task, nxt_port_t *port,
-    nxt_port_recv_msg_t *msg, size_t size)
+    nxt_port_recv_msg_t *msg)
 {
     nxt_buf_t  *b;
     nxt_buf_t  *orig_b;
-    nxt_buf_t  **last_next;
 
-    if (nxt_slow_path(size < sizeof(nxt_port_msg_t))) {
-        nxt_log(port->socket.task, NXT_LOG_CRIT,
-                "port %d: too small message:%uz", port->socket.fd, size);
+    if (nxt_slow_path(msg->size < sizeof(nxt_port_msg_t))) {
+        nxt_log(task, NXT_LOG_CRIT,
+                "port %d: too small message:%uz", port->socket.fd, msg->size);
         goto fail;
     }
 
     /* adjust size to actual buffer used size */
-    size -= sizeof(nxt_port_msg_t);
+    msg->size -= sizeof(nxt_port_msg_t);
 
     b = orig_b = msg->buf;
-    b->mem.free += size;
+    b->mem.free += msg->size;
 
     if (msg->port_msg.mmap) {
-        nxt_port_mmap_read(task, port, msg, size);
+        nxt_port_mmap_read(task, port, msg);
         b = msg->buf;
     }
 
-    last_next = &b->next;
-
-    if (msg->port_msg.last) {
-        /* find reference to last next, the NULL one */
-        while (*last_next) {
-            last_next = &(*last_next)->next;
-        }
-
-        *last_next = nxt_buf_sync_alloc(port->mem_pool, NXT_BUF_SYNC_LAST);
-        if (nxt_slow_path(*last_next == NULL)) {
-            goto fail;
-        }
-    }
-
     port->handler(task, msg);
 
-    if (*last_next != NULL) {
-        /* A sync buffer */
-        nxt_buf_free(port->mem_pool, *last_next);
-        *last_next = NULL;
-    }
-
-    if (orig_b != b) {
-        /* complete mmap buffers */
+    if (msg->port_msg.mmap && orig_b != b) {
+        /* complete used mmap buffers */
         for (; b && nxt_buf_used_size(b) == 0;
             b = b->next) {
             nxt_debug(task, "complete buffer %p", b);
@@ -475,7 +455,6 @@ nxt_port_buf_alloc(nxt_port_t *port)
         b->mem.pos = b->mem.start;
         b->mem.free = b->mem.start;
         b->next = NULL;
-
     } else {
         b = nxt_buf_mem_alloc(port->mem_pool, port->max_size, 0);
         if (nxt_slow_path(b == NULL)) {