I/O operations refactoring.

Author: igorsysoev

auto/sources

@@ -126,6 +126,7 @@ NXT_LIB_SRCS=" \
     src/nxt_job_resolve.c \
     src/nxt_sockaddr.c \
     src/nxt_listen_socket.c \
+    src/nxt_upstream_round_robin.c \
 "
 
 NXT_LIB_SRC0=" \
@@ -190,7 +191,6 @@ NXT_LIB_UNIT_TEST_SRCS=" \
     test/nxt_rbtree_unit_test.c \
     test/nxt_term_parse_unit_test.c \
     test/nxt_msec_diff_unit_test.c \
-    test/nxt_exp_approximation.c \
     test/nxt_mem_cache_pool_unit_test.c \
     test/nxt_mem_zone_unit_test.c \
     test/nxt_lvlhsh_unit_test.c \
@@ -330,6 +330,7 @@ NXT_SRCS=" \
     src/nxt_cycle.c \
     src/nxt_port.c \
     src/nxt_application.c \
+    src/nxt_stream_module.c \
     src/nxt_master_process.c \
     src/nxt_worker_process.c \
 "

src/nxt_aix_send_file.c

@@ -32,7 +32,7 @@ nxt_aix_event_conn_io_send_file(nxt_event_conn_t *c, nxt_buf_t *b, size_t limit)
     sb.size = 0;
     sb.limit = limit;
 
-    nhd = nxt_sendbuf_mem_coalesce(&sb);
+    nhd = nxt_sendbuf_mem_coalesce(c->socket.task, &sb);
 
     if (nhd == 0 && sb.sync) {
         return 0;
@@ -53,7 +53,7 @@ nxt_aix_event_conn_io_send_file(nxt_event_conn_t *c, nxt_buf_t *b, size_t limit)
     sb.iobuf = &tr;
     sb.nmax = 1;
 
-    ntr = nxt_sendbuf_mem_coalesce(&sb);
+    ntr = nxt_sendbuf_mem_coalesce(c->socket.task, &sb);
 
     nxt_memzero(&sfp, sizeof(struct sf_parms));
 
@@ -71,17 +71,16 @@ nxt_aix_event_conn_io_send_file(nxt_event_conn_t *c, nxt_buf_t *b, size_t limit)
         sfp.trailer_length = tr.iov_len;
     }
 
-    nxt_log_debug(c->socket.log, "send_file(%d) fd:%FD @%O:%O hd:%ui tr:%ui",
-                  c->socket.fd, fb->file->fd, fb->file_pos, file_size,
-                  nhd, ntr);
+    nxt_debug(c->socket.task, "send_file(%d) fd:%FD @%O:%O hd:%ui tr:%ui",
+              c->socket.fd, fb->file->fd, fb->file_pos, file_size, nhd, ntr);
 
     n = send_file(&c->socket.fd, &sfp, 0);
 
     err = (n == -1) ? nxt_errno : 0;
     sent = sfp.bytes_sent;
 
-    nxt_log_debug(c->socket.log, "send_file(%d): %d sent:%O",
-                  c->socket.fd, n, sent);
+    nxt_debug(c->socket.task, "send_file(%d): %d sent:%O",
+              c->socket.fd, n, sent);
 
     /*
      * -1  an error has occurred, errno contains the error code;
@@ -102,16 +101,15 @@ nxt_aix_event_conn_io_send_file(nxt_event_conn_t *c, nxt_buf_t *b, size_t limit)
 
         default:
             c->socket.error = err;
-            nxt_log_error(nxt_socket_error_level(err, c->socket.log_error),
-                          c->socket.log, "send_file(%d) failed %E \"%FN\" "
-                          "fd:%FD @%O:%O hd:%ui tr:%ui", c->socket.fd, err,
-                          fb->file->name, fb->file->fd, fb->file_pos,
-                          file_size, nhd, ntr);
+            nxt_log(c->socket.task, nxt_socket_error_level(err),
+                "send_file(%d) failed %E \"%FN\" fd:%FD @%O:%O hd:%ui tr:%ui",
+                c->socket.fd, err, fb->file->name, fb->file->fd, fb->file_pos,
+                file_size, nhd, ntr);
 
             return NXT_ERROR;
         }
 
-        nxt_log_debug(c->socket.log, "sendfile() %E", err);
+        nxt_debug(c->socket.task, "sendfile() %E", err);
 
         return sent;
     }

src/nxt_application.c

@@ -20,7 +20,6 @@ static void nxt_app_conn_update(nxt_thread_t *thr, nxt_event_conn_t *c,
     nxt_log_t *log);
 static nxt_int_t nxt_app_write_finish(nxt_app_request_t *r);
 static void nxt_app_buf_send(nxt_event_conn_t *c, nxt_buf_t *out);
-static void nxt_app_buf_completion(nxt_task_t *task, void *obj, void *data);
 static void nxt_app_delivery_handler(nxt_task_t *task, void *obj, void *data);
 static void nxt_app_delivery_ready(nxt_task_t *task, void *obj, void *data);
 static void nxt_app_delivery_completion(nxt_task_t *task, void *obj,
@@ -683,7 +682,7 @@ nxt_app_write(nxt_app_request_t *r, const u_char *data, size_t length)
 
             nxt_buf_mem_init(b, start, 4096);
 
-            b->completion_handler = nxt_app_buf_completion;
+            b->completion_handler = NULL;
 
             nxt_app_buf_current_number++;
         }
@@ -713,7 +712,7 @@ nxt_app_write_finish(nxt_app_request_t *r)
         return NXT_ERROR;
     }
 
-    b->completion_handler = nxt_app_buf_completion;
+    b->completion_handler = NULL;
     b->parent = (nxt_buf_t *) r;
 
     out = r->output_buf;
@@ -745,20 +744,6 @@ nxt_app_buf_send(nxt_event_conn_t *c, nxt_buf_t *out)
 }
 
 
-static void
-nxt_app_buf_completion(nxt_task_t *task, void *obj, void *data)
-{
-    nxt_buf_t  *b;
-
-    b = obj;
-
-    nxt_debug(task, "app buf completion");
-
-    b->next = nxt_app_buf_done;
-    nxt_app_buf_done = b;
-}
-
-
 static void
 nxt_app_delivery_handler(nxt_task_t *task, void *obj, void *data)
 {
@@ -797,14 +782,14 @@ nxt_app_delivery_handler(nxt_task_t *task, void *obj, void *data)
     c->write = b;
     c->write_state = &nxt_app_delivery_write_state;
 
-    nxt_event_conn_write(task, c);
+    nxt_event_conn_write(task->thread->engine, c);
 }
 
 
 static const nxt_event_conn_state_t  nxt_app_delivery_write_state
     nxt_aligned(64) =
 {
-    NXT_EVENT_BUF_PROCESS,
+    NXT_EVENT_NO_BUF_PROCESS,
     NXT_EVENT_TIMER_AUTORESET,
 
     nxt_app_delivery_ready,
@@ -820,12 +805,26 @@ static const nxt_event_conn_state_t  nxt_app_delivery_write_state
 static void
 nxt_app_delivery_ready(nxt_task_t *task, void *obj, void *data)
 {
+    nxt_buf_t         *b, *next;
     nxt_event_conn_t  *c;
 
     c = obj;
 
     nxt_debug(task, "app delivery ready");
 
+    for (b = c->write; b != NULL; b = next) {
+
+        if (nxt_buf_is_mem(b)) {
+            if (b->mem.pos != b->mem.free) {
+                break;
+            }
+        }
+
+        next = b->next;
+        b->next = nxt_app_buf_done;
+        nxt_app_buf_done = b;
+    }
+
     nxt_work_queue_add(c->write_work_queue,
                        nxt_app_delivery_completion, task, c, NULL);
 }