windows: ref pipe writes to keep the event loop alive

include/uv-private/uv-win.h

@@ -191,10 +191,9 @@ RB_HEAD(uv_timer_tree_s, uv_timer_s);
 
 #define UV_WRITE_PRIVATE_FIELDS           \
   int ipc_header;                         \
-  uv_buf_t* write_buffer;                 \
+  uv_buf_t write_buffer;                  \
   HANDLE event_handle;                    \
-  HANDLE wait_handle;                     \
+  HANDLE wait_handle;
-  uv_write_t* next_non_overlapped_write;
 
 #define UV_CONNECT_PRIVATE_FIELDS         \
   /* empty */

src/win/pipe.c

@@ -726,11 +726,11 @@ static DWORD WINAPI uv_pipe_writefile_thread_proc(void* parameter) {
   assert(req != NULL);
   assert(req->type == UV_WRITE);
   assert(handle->type == UV_NAMED_PIPE);
-  assert(req->write_buffer);
+  assert(req->write_buffer.base);
 
   result = WriteFile(handle->handle,
-                     req->write_buffer->base,
+                     req->write_buffer.base,
-                     req->write_buffer->len,
+                     req->write_buffer.len,
                      &bytes,
                      NULL);
 
@@ -897,14 +897,14 @@ int uv_pipe_read2_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
 
 static void uv_insert_non_overlapped_write_req(uv_pipe_t* handle,
     uv_write_t* req) {
-  req->next_non_overlapped_write = NULL;
+  req->next_req = NULL;
   if (handle->non_overlapped_writes_tail) {
-    req->next_non_overlapped_write =
+    req->next_req =
-      handle->non_overlapped_writes_tail->next_non_overlapped_write;
+      handle->non_overlapped_writes_tail->next_req;
-    handle->non_overlapped_writes_tail->next_non_overlapped_write = req;
+    handle->non_overlapped_writes_tail->next_req = (uv_req_t*)req;
     handle->non_overlapped_writes_tail = req;
   } else {
-    req->next_non_overlapped_write = req;
+    req->next_req = (uv_req_t*)req;
     handle->non_overlapped_writes_tail = req;
   }
 }
@@ -914,13 +914,13 @@ static uv_write_t* uv_remove_non_overlapped_write_req(uv_pipe_t* handle) {
   uv_write_t* req;
 
   if (handle->non_overlapped_writes_tail) {
-    req = handle->non_overlapped_writes_tail->next_non_overlapped_write;
+    req = (uv_write_t*)handle->non_overlapped_writes_tail->next_req;
 
     if (req == handle->non_overlapped_writes_tail) {
       handle->non_overlapped_writes_tail = NULL;
     } else {
-      handle->non_overlapped_writes_tail->next_non_overlapped_write =
+      handle->non_overlapped_writes_tail->next_req =
-        req->next_non_overlapped_write;
+        req->next_req;
     }
 
     return req;
@@ -1054,6 +1054,10 @@ static int uv_pipe_write_impl(uv_loop_t* loop, uv_write_t* req,
       handle->write_queue_size += req->queued_bytes;
     }
 
+    if (handle->write_reqs_pending == 0) {
+      uv_ref(loop);
+    }
+
     handle->reqs_pending++;
     handle->write_reqs_pending++;
 
@@ -1064,7 +1068,7 @@ static int uv_pipe_write_impl(uv_loop_t* loop, uv_write_t* req,
   }
 
   if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
-    req->write_buffer = &bufs[0];
+    req->write_buffer = bufs[0];
     uv_insert_non_overlapped_write_req(handle, req);
     if (handle->write_reqs_pending == 0) {
       uv_queue_non_overlapped_write(handle);
@@ -1108,6 +1112,10 @@ static int uv_pipe_write_impl(uv_loop_t* loop, uv_write_t* req,
     }
   }
 
+  if (handle->write_reqs_pending == 0) {
+    uv_ref(loop);
+  }
+
   handle->reqs_pending++;
   handle->write_reqs_pending++;
 
@@ -1362,6 +1370,10 @@ void uv_process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,
     uv_queue_non_overlapped_write(handle);
   }
 
+  if (handle->write_reqs_pending == 0) {
+    uv_unref(loop);
+  }
+
   if (handle->write_reqs_pending == 0 &&
       handle->flags & UV_HANDLE_SHUTTING) {
     uv_want_endgame(loop, (uv_handle_t*)handle);

test/run-tests.c

@@ -208,9 +208,9 @@ static int stdio_over_pipes_helper() {
   uv_pipe_open(&stdin_pipe, 0);
   uv_pipe_open(&stdout_pipe, 1);
 
-  r = uv_read_start((uv_stream_t*)&stdin_pipe, on_pipe_read_alloc,
+  /* Unref both stdio handles to make sure that all writes complete. */
-    on_pipe_read);
+  uv_unref(loop);
-  ASSERT(r == 0);
+  uv_unref(loop);
 
   for (i = 0; i < COUNTOF(buffers); i++) {
     buf[i] = uv_buf_init((char*)buffers[i], strlen(buffers[i]));
@@ -224,6 +224,19 @@ static int stdio_over_pipes_helper() {
 
   uv_run(loop);
 
+  ASSERT(after_write_called == 7);
+  ASSERT(on_pipe_read_called == 0);
+  ASSERT(close_cb_called == 0);
+
+  uv_ref(loop);
+  uv_ref(loop);
+
+  r = uv_read_start((uv_stream_t*)&stdin_pipe, on_pipe_read_alloc,
+    on_pipe_read);
+  ASSERT(r == 0);
+
+  uv_run(loop);
+
   ASSERT(after_write_called == 7);
   ASSERT(on_pipe_read_called == 1);
   ASSERT(close_cb_called == 2);