unix: sync writes

Fixes #64.

uv-unix.c

@@ -218,6 +218,7 @@ int uv_tcp_init(uv_tcp_t* tcp) {
   tcp->fd = -1;
   tcp->delayed_error = 0;
   ngx_queue_init(&tcp->write_queue);
+  ngx_queue_init(&tcp->write_completed_queue);
   tcp->write_queue_size = 0;
 
   ev_init(&tcp->read_watcher, uv__tcp_io);
@@ -227,6 +228,7 @@ int uv_tcp_init(uv_tcp_t* tcp) {
   tcp->write_watcher.data = tcp;
 
   assert(ngx_queue_empty(&tcp->write_queue));
+  assert(ngx_queue_empty(&tcp->write_completed_queue));
   assert(tcp->write_queue_size == 0);
 
   return 0;
@@ -528,12 +530,14 @@ static void uv__drain(uv_tcp_t* tcp) {
 }
 
 
-void uv__write(uv_tcp_t* tcp) {
+/* On success returns NULL. On error returns a pointer to the write request
+ * which had the error.
+ */
+static uv_req_t* uv__write(uv_tcp_t* tcp) {
   uv_req_t* req;
   struct iovec* iov;
   int iovcnt;
   ssize_t n;
-  uv_write_cb cb;
 
   assert(tcp->fd >= 0);
 
@@ -543,8 +547,7 @@ void uv__write(uv_tcp_t* tcp) {
   req = uv_write_queue_head(tcp);
   if (!req) {
     assert(tcp->write_queue_size == 0);
-    uv__drain(tcp);
-    return;
+    return NULL;
   }
 
   assert(req->handle == (uv_handle_t*)tcp);
@@ -562,21 +565,16 @@ void uv__write(uv_tcp_t* tcp) {
 
   n = writev(tcp->fd, iov, iovcnt);
 
-  cb = (uv_write_cb)req->cb;
-
   if (n < 0) {
     if (errno != EAGAIN) {
-      uv_err_t err = uv_err_new((uv_handle_t*)tcp, errno);
-
-      if (cb) {
-        cb(req, -1);
-      }
-      return;
+      /* Error */
+      uv_err_new((uv_handle_t*)tcp, errno);
+      return req;
     }
   } else {
     /* Successful write */
 
-    /* The loop updates the counters. */
+    /* Update the counters. */
     while (n > 0) {
       uv_buf_t* buf = &(req->bufs[req->write_index]);
       size_t len = buf->len;
@@ -611,19 +609,13 @@ void uv__write(uv_tcp_t* tcp) {
           free(req->bufs); /* FIXME: we should not be allocing for each read */
           req->bufs = NULL;
 
-          /* NOTE: call callback AFTER freeing the request data. */
-          if (cb) {
-            cb(req, 0);
-          }
-
-          if (!ngx_queue_empty(&tcp->write_queue)) {
-            assert(tcp->write_queue_size > 0);
-          } else {
-            /* Write queue drained. */
-            uv__drain(tcp);
-          }
-
-          return;
+          /* Add it to the write_completed_queue where it will have its
+           * callback called in the near future.
+           * TODO: start trying to write the next request.
+           */
+          ngx_queue_insert_tail(&tcp->write_completed_queue, &req->queue);
+          ev_feed_event(EV_DEFAULT_ &tcp->write_watcher, EV_WRITE);
+          return NULL;
         }
       }
     }
@@ -632,9 +624,42 @@ void uv__write(uv_tcp_t* tcp) {
   /* Either we've counted n down to zero or we've got EAGAIN. */
   assert(n == 0 || n == -1);
 
-  /* We're not done yet. */
-  assert(ev_is_active(&tcp->write_watcher));
+  /* We're not done. */
   ev_io_start(EV_DEFAULT_ &tcp->write_watcher);
+
+  return NULL;
+}
+
+
+static void uv__write_callbacks(uv_tcp_t* tcp) {
+  uv_write_cb cb;
+  int callbacks_made = 0;
+  ngx_queue_t* q;
+  uv_req_t* req;
+
+  while (!ngx_queue_empty(&tcp->write_completed_queue)) {
+    /* Pop a req off write_completed_queue. */
+    q = ngx_queue_head(&tcp->write_completed_queue);
+    assert(q);
+    req = ngx_queue_data(q, struct uv_req_s, queue);
+    ngx_queue_remove(q);
+
+    cb = (uv_write_cb) req->cb;
+
+    /* NOTE: call callback AFTER freeing the request data. */
+    if (cb) {
+      cb(req, 0);
+    }
+
+    callbacks_made++;
+  }
+
+  assert(ngx_queue_empty(&tcp->write_completed_queue));
+
+  /* Write queue drained. */
+  if (!uv_write_queue_head(tcp)) {
+    uv__drain(tcp);
+  }
 }
 
 
@@ -727,7 +752,17 @@ void uv__tcp_io(EV_P_ ev_io* watcher, int revents) {
     }
 
     if (revents & EV_WRITE) {
-      uv__write(tcp);
+      uv_req_t* req = uv__write(tcp);
+      if (req) {
+        /* Error. Notify the user. */
+        uv_write_cb cb = (uv_write_cb) req->cb;
+
+        if (cb) {
+          cb(req, -1);
+        }
+      } else {
+        uv__write_callbacks(tcp);
+      }
     }
   }
 }
@@ -872,6 +907,7 @@ static size_t uv__buf_count(uv_buf_t bufs[], int bufcnt) {
  */
 int uv_write(uv_req_t* req, uv_buf_t bufs[], int bufcnt) {
   uv_tcp_t* tcp = (uv_tcp_t*)req->handle;
+  int empty_queue = (tcp->write_queue_size == 0);
   assert(tcp->fd >= 0);
 
   ngx_queue_init(&req->queue);
@@ -893,7 +929,29 @@ int uv_write(uv_req_t* req, uv_buf_t bufs[], int bufcnt) {
   assert(tcp->write_watcher.data == tcp);
   assert(tcp->write_watcher.fd == tcp->fd);
 
-  ev_io_start(EV_DEFAULT_ &tcp->write_watcher);
+  /* If the queue was empty when this function began, we should attempt to
+   * do the write immediately. Otherwise start the write_watcher and wait
+   * for the fd to become writable.
+   */
+  if (empty_queue) {
+    if (uv__write(tcp)) {
+      /* Error. uv_last_error has been set. */
+      return -1;
+    }
+  }
+
+  /* If the queue is now empty - we've flushed the request already. That
+   * means we need to make the callback. The callback can only be done on a
+   * fresh stack so we feed the event loop in order to service it.
+   */
+  if (ngx_queue_empty(&tcp->write_queue)) {
+    ev_feed_event(EV_DEFAULT_ &tcp->write_watcher, EV_WRITE);
+  } else {
+    /* Otherwise there is data to write - so we should wait for the file
+     * descriptor to become writable.
+     */
+    ev_io_start(EV_DEFAULT_ &tcp->write_watcher);
+  }
 
   return 0;
 }

uv-unix.h

@@ -64,7 +64,8 @@ typedef struct {
   uv_req_t *shutdown_req; \
   ev_io read_watcher; \
   ev_io write_watcher; \
-  ngx_queue_t write_queue;
+  ngx_queue_t write_queue; \
+  ngx_queue_t write_completed_queue;
 
 
 /* UV_PREPARE */ \