nxt_port_buf_completion() and nxt_sendbuf_completion().

nxt_sendbuf_completion() has been renamed to nxt_port_buf_completion()
and moved to src/nxt_port_socket.c.  nxt_sendbuf_completion0() has been
renamed to nxt_sendbuf_completion().

1 files changed, 64 insertions, 1 deletions

diff --git a/src/nxt_port_socket.c b/src/nxt_port_socket.c
index 05a4845f..e688530e 100644
--- a/src/nxt_port_socket.c
+++ b/src/nxt_port_socket.c
@@ -8,6 +8,8 @@
 
 
 static void nxt_port_write_handler(nxt_task_t *task, void *obj, void *data);
+static nxt_buf_t *nxt_port_buf_completion(nxt_task_t *task,
+    nxt_work_queue_t *wq, nxt_buf_t *b, size_t sent, nxt_bool_t mmap_mode);
 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);
@@ -379,7 +381,7 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
                 msg->fd = -1;
             }
 
-            msg->buf = nxt_sendbuf_completion(task, wq, msg->buf, plain_size,
+            msg->buf = nxt_port_buf_completion(task, wq, msg->buf, plain_size,
                                               m == NXT_PORT_METHOD_MMAP);
 
             if (msg->buf != NULL) {
@@ -462,6 +464,67 @@ unlock_mutex:
 }
 
 
+static nxt_buf_t *
+nxt_port_buf_completion(nxt_task_t *task, nxt_work_queue_t *wq, nxt_buf_t *b,
+    size_t sent, nxt_bool_t mmap_mode)
+{
+    size_t  size;
+
+    while (b != NULL) {
+
+        nxt_prefetch(b->next);
+
+        if (!nxt_buf_is_sync(b)) {
+
+            size = nxt_buf_used_size(b);
+
+            if (size != 0) {
+
+                if (sent == 0) {
+                    break;
+                }
+
+                if (nxt_buf_is_port_mmap(b) && mmap_mode) {
+                    /*
+                     * buffer has been sent to other side which is now
+                     * responsible for shared memory bucket release
+                     */
+                    b->is_port_mmap_sent = 1;
+                }
+
+                if (sent < size) {
+
+                    if (nxt_buf_is_mem(b)) {
+                        b->mem.pos += sent;
+                    }
+
+                    if (nxt_buf_is_file(b)) {
+                        b->file_pos += sent;
+                    }
+
+                    break;
+                }
+
+                /* b->mem.free is NULL in file-only buffer. */
+                b->mem.pos = b->mem.free;
+
+                if (nxt_buf_is_file(b)) {
+                    b->file_pos = b->file_end;
+                }
+
+                sent -= size;
+            }
+        }
+
+        nxt_work_queue_add(wq, b->completion_handler, task, b, b->parent);
+
+        b = b->next;
+    }
+
+    return b;
+}
+
+
 void
 nxt_port_read_enable(nxt_task_t *task, nxt_port_t *port)
 {