win: Implement non-overlapped pipe cancellation

CMakeLists.txt

@@ -196,7 +196,8 @@ if(WIN32)
        ws2_32
        dbghelp
        ole32
-       shell32)
+       shell32
+       synchronization)
   list(APPEND uv_sources
        src/win/async.c
        src/win/core.c

configure.ac

@@ -74,7 +74,7 @@ AM_CONDITIONAL([OS400],    [AS_CASE([$host_os],[os400],         [true], [false])
 AM_CONDITIONAL([SUNOS],    [AS_CASE([$host_os],[solaris*],      [true], [false])])
 AM_CONDITIONAL([WINNT],    [AS_CASE([$host_os],[mingw*],        [true], [false])])
 AS_CASE([$host_os],[mingw*], [
-    LIBS="$LIBS -lws2_32 -lpsapi -liphlpapi -luserenv -luser32 -ldbghelp -lole32 -lshell32"
+    LIBS="$LIBS -lws2_32 -lpsapi -liphlpapi -luserenv -luser32 -ldbghelp -lole32 -lshell32 -lsynchronization"
 ])
 AS_CASE([$host_os], [solaris2.10], [
     CFLAGS="$CFLAGS -DSUNOS_NO_IFADDRS"

include/uv/win.h

@@ -376,9 +376,11 @@ typedef struct {
       ULONG_PTR result; /* overlapped.Internal is reused to hold the result */\
       HANDLE pipeHandle;                                                      \
       DWORD duplex_flags;                                                     \
-      WCHAR* name;                                                             \
+      WCHAR* name;                                                            \
     } connect;                                                                \
   } u;                                                                        \
+  /* Singly linked list of pending reqs. For non-overlapped pipes, also used  \
+   * to keep track of reqs no yet submitted to the thread pool */             \
   struct uv_req_s* next_req;                                                  \
   union {                                                                     \
     void* reserved2[1];                                                       \
@@ -478,7 +480,10 @@ typedef struct {
 
 #define uv_pipe_connection_fields                                             \
   uv_timer_t* eof_timer;                                                      \
-  uv_write_t dummy; /* TODO: retained for ABI compat; remove this in v2.x. */ \
+  /* TODO: This is here for ABI compat - remove in 2.x. */                    \
+  uintptr_t dummy[sizeof(uv_write_t) / sizeof(uintptr_t) - 2];                \
+  uv_write_t* non_overlapped_write_active;                                    \
+  volatile HANDLE writefile_thread_handle;                                    \
   DWORD ipc_remote_pid;                                                       \
   union {                                                                     \
     uint32_t payload_remaining;                                               \
@@ -487,7 +492,7 @@ typedef struct {
   struct uv__queue ipc_xfer_queue;                                            \
   int ipc_xfer_queue_length;                                                  \
   uv_write_t* non_overlapped_writes_tail;                                     \
-  CRITICAL_SECTION readfile_thread_lock;                                      \
+  CRITICAL_SECTION thread_lock;                                               \
   volatile HANDLE readfile_thread_handle;
 
 #define UV_PIPE_PRIVATE_FIELDS                                                \

src/uv-common.h

@@ -103,7 +103,7 @@ enum {
   UV_HANDLE_ZERO_READ                   = 0x00040000,
   UV_HANDLE_EMULATE_IOCP                = 0x00080000,
   UV_HANDLE_BLOCKING_WRITES             = 0x00100000,
-  UV_HANDLE_CANCELLATION_PENDING        = 0x00200000,
+  UV_HANDLE_READ_CANCELLATION_PENDING   = 0x00200000,
 
   /* Used by uv_tcp_t and uv_udp_t handles */
   UV_HANDLE_IPV6                        = 0x00400000,

src/win/internal.h

@@ -134,6 +134,7 @@ int uv__pipe_write(uv_loop_t* loop,
                    uv_stream_t* send_handle,
                    uv_write_cb cb);
 void uv__pipe_shutdown(uv_loop_t* loop, uv_pipe_t* handle, uv_shutdown_t* req);
+int uv__pipe_write_cancel_non_overlapped(uv_pipe_t* handle, uv_write_t* req);
 
 void uv__process_pipe_read_req(uv_loop_t* loop, uv_pipe_t* handle,
     uv_req_t* req);

src/win/pipe.c

@@ -90,6 +90,13 @@ typedef struct {
 } uv__coalesced_write_t;
 
 
+static uv_write_t* uv__coalesced_write_user_req(uv_write_t* req) {
+  if (req->coalesced)
+    return container_of(req, uv__coalesced_write_t, req)->user_req;
+  return req;
+}
+
+
 static void eof_timer_init(uv_pipe_t* pipe);
 static void eof_timer_start(uv_pipe_t* pipe);
 static void eof_timer_stop(uv_pipe_t* pipe);
@@ -518,8 +525,10 @@ static int uv__set_pipe_handle(uv_loop_t* loop,
       mode_info.Mode & FILE_SYNCHRONOUS_IO_NONALERT) {
     /* Non-overlapped pipe. */
     handle->flags |= UV_HANDLE_NON_OVERLAPPED_PIPE;
-    handle->pipe.conn.readfile_thread_handle = NULL;
+    handle->pipe.conn.non_overlapped_write_active = NULL;
-    InitializeCriticalSection(&handle->pipe.conn.readfile_thread_lock);
+    handle->pipe.conn.readfile_thread_handle = INVALID_HANDLE_VALUE;
+    handle->pipe.conn.writefile_thread_handle = INVALID_HANDLE_VALUE;
+    InitializeCriticalSection(&handle->pipe.conn.thread_lock);
   } else {
     /* Overlapped pipe.  Try to associate with IOCP. */
     if (CreateIoCompletionPort(pipeHandle,
@@ -679,7 +688,7 @@ void uv__pipe_endgame(uv_loop_t* loop, uv_pipe_t* handle) {
     }
 
     if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE)
-      DeleteCriticalSection(&handle->pipe.conn.readfile_thread_lock);
+      DeleteCriticalSection(&handle->pipe.conn.thread_lock);
   }
 
   if (handle->flags & UV_HANDLE_PIPESERVER) {
@@ -996,51 +1005,69 @@ error:
   return 0;
 }
 
+/* Cancel a synchronous I/O operation running in a thread pool thread.
+ * thread_ptr is the volatile handle set by the worker thread, and lock
+ * is held briefly to synchronize the handshake. */
+static void uv__pipe_cancel_synchronous_io(volatile HANDLE* thread_ptr,
+                                            CRITICAL_SECTION* lock) {
+  HANDLE thread;
+  HANDLE expected_invalid = INVALID_HANDLE_VALUE;
+
+  EnterCriticalSection(lock);
+
+  thread = *thread_ptr;
+  if (thread == NULL) {
+    /* The thread pool thread has not yet reached the point of blocking, we
+     * can pre-empt it here. However, we still need to wait for the thread
+     * to acknowledge the interrupt. Otherwise it could race with the next
+     * request. It does this by setting *thread_ptr back to NULL. */
+    *thread_ptr = INVALID_HANDLE_VALUE;
+    do {
+      LeaveCriticalSection(lock);
+      WaitOnAddress(thread_ptr, &expected_invalid, sizeof(*thread_ptr), INFINITE);
+      EnterCriticalSection(lock);
+    } while (*thread_ptr != NULL);
+    /* Finally set this back to INVALID_HANDLE_VALUE to retain the
+     * invariant that it'll be INVALID_HANDLE_VALUE on exit from this function. */
+    *thread_ptr = INVALID_HANDLE_VALUE;
+  } else {
+    /* Spin until the thread has acknowledged (by changing *thread_ptr)
+     * that it is past the point of blocking. */
+    while (thread != INVALID_HANDLE_VALUE) {
+      BOOL r = CancelSynchronousIo(thread);
+      assert(r || GetLastError() == ERROR_NOT_FOUND);
+      LeaveCriticalSection(lock);
+      SwitchToThread();
+      EnterCriticalSection(lock);
+      thread = *thread_ptr;
+    }
+  }
+
+  LeaveCriticalSection(lock);
+}
+
 
 void uv__pipe_interrupt_read(uv_pipe_t* handle) {
-  BOOL r;
-
   if (!(handle->flags & UV_HANDLE_READ_PENDING))
     return; /* No pending reads. */
-  if (handle->flags & UV_HANDLE_CANCELLATION_PENDING)
+  if (handle->flags & UV_HANDLE_READ_CANCELLATION_PENDING)
     return; /* Already cancelled. */
   if (handle->handle == INVALID_HANDLE_VALUE)
     return; /* Pipe handle closed. */
 
   if (!(handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE)) {
     /* Cancel asynchronous read. */
-    r = CancelIoEx(handle->handle, &handle->read_req.u.io.overlapped);
+    BOOL r = CancelIoEx(handle->handle, &handle->read_req.u.io.overlapped);
     assert(r || GetLastError() == ERROR_NOT_FOUND);
     (void) r;
   } else {
-    /* Cancel synchronous read (which is happening in the thread pool). */
+    uv__pipe_cancel_synchronous_io(
-    HANDLE thread;
+        &handle->pipe.conn.readfile_thread_handle,
-    volatile HANDLE* thread_ptr = &handle->pipe.conn.readfile_thread_handle;
+        &handle->pipe.conn.thread_lock);
-
-    EnterCriticalSection(&handle->pipe.conn.readfile_thread_lock);
-
-    thread = *thread_ptr;
-    if (thread == NULL) {
-      /* The thread pool thread has not yet reached the point of blocking, we
-       * can pre-empt it by setting thread_handle to INVALID_HANDLE_VALUE. */
-      *thread_ptr = INVALID_HANDLE_VALUE;
-
-    } else {
-      /* Spin until the thread has acknowledged (by setting the thread to
-       * INVALID_HANDLE_VALUE) that it is past the point of blocking. */
-      while (thread != INVALID_HANDLE_VALUE) {
-        r = CancelSynchronousIo(thread);
-        assert(r || GetLastError() == ERROR_NOT_FOUND);
-        SwitchToThread(); /* Yield thread. */
-        thread = *thread_ptr;
-      }
-    }
-
-    LeaveCriticalSection(&handle->pipe.conn.readfile_thread_lock);
   }
 
   /* Set flag to indicate that read has been cancelled. */
-  handle->flags |= UV_HANDLE_CANCELLATION_PENDING;
+  handle->flags |= UV_HANDLE_READ_CANCELLATION_PENDING;
 }
 
 
@@ -1051,6 +1078,113 @@ void uv__pipe_read_stop(uv_pipe_t* handle) {
 }
 
 
+/* Remove the element after prev from the non-overlapped write queue.
+ * prev must be a node in the queue. Returns the removed element. */
+static uv_write_t* uv__remove_non_overlapped_write_req_after(
+    uv_pipe_t* handle, uv_write_t* prev) {
+  uv_write_t* req;
+
+  req = (uv_write_t*) prev->next_req;
+  if (req == prev) {
+    /* Only element. */
+    handle->pipe.conn.non_overlapped_writes_tail = NULL;
+  } else {
+    prev->next_req = req->next_req;
+    if (req == handle->pipe.conn.non_overlapped_writes_tail)
+      handle->pipe.conn.non_overlapped_writes_tail = prev;
+  }
+  return req;
+}
+
+
+static uv_write_t* uv_remove_non_overlapped_write_req(uv_pipe_t* handle) {
+  if (handle->pipe.conn.non_overlapped_writes_tail == NULL)
+    return NULL;
+
+  return uv__remove_non_overlapped_write_req_after(
+      handle, handle->pipe.conn.non_overlapped_writes_tail);
+}
+
+
+/* Find and remove a specific request from the non-overlapped write queue.
+ * For coalesced writes, match against the user-facing req.
+ * Returns the actual queued req if found and removed, NULL otherwise. */
+static uv_write_t* uv__remove_specific_non_overlapped_write_req(
+    uv_pipe_t* handle, uv_write_t* target) {
+  uv_write_t* tail;
+  uv_write_t* prev;
+  uv_write_t* curr;
+
+  tail = handle->pipe.conn.non_overlapped_writes_tail;
+  if (tail == NULL)
+    return NULL;
+
+  prev = tail;
+  curr = (uv_write_t*) tail->next_req;
+  do {
+    if (uv__coalesced_write_user_req(curr) == target)
+      return uv__remove_non_overlapped_write_req_after(handle, prev);
+
+    prev = curr;
+    curr = (uv_write_t*) curr->next_req;
+  } while (prev != tail);
+
+  return NULL;
+}
+
+
+int uv__pipe_write_cancel_non_overlapped(uv_pipe_t* handle, uv_write_t* req) {
+  uv_write_t* queued_req;
+
+  assert(handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE);
+  assert(!(handle->flags & UV_HANDLE_BLOCKING_WRITES));
+
+  /* On the thread pool - send the cancellation signal. */
+  if (handle->pipe.conn.non_overlapped_write_active != NULL &&
+      uv__coalesced_write_user_req(
+          handle->pipe.conn.non_overlapped_write_active) == req) {
+    /* N.B.: It's possible to end up here multiple times if `req` is cancelled
+     * again after an initial cancellation, but before the completion is processed.
+     * This is harmless - uv__pipe_cancel_synchronous_io will see that the thread
+     * has concluded and do nothing. */
+    uv__pipe_cancel_synchronous_io(
+        &handle->pipe.conn.writefile_thread_handle,
+        &handle->pipe.conn.thread_lock);
+    return 0;
+  }
+
+  /* If it's in the queue, remove and complete it as cancelled. */
+  queued_req = uv__remove_specific_non_overlapped_write_req(handle, req);
+  if (queued_req != NULL) {
+    SET_REQ_ERROR(queued_req, ERROR_OPERATION_ABORTED);
+    SET_REQ_NWRITTEN(queued_req, 0);
+    uv__insert_pending_req(handle->loop, (uv_req_t*) queued_req);
+    return 0;
+  }
+
+  /* Already completed (including previously cancelled). */
+  return 0;
+}
+
+/* Cancel all active and pending non-overlapped writes. */
+static void uv__pipe_flush_non_overlapped_writes(uv_pipe_t* handle) {
+  uv_write_t* req;
+
+  /* Cancel the active write if there is one. */
+  if (handle->pipe.conn.non_overlapped_write_active != NULL)
+    uv__pipe_cancel_synchronous_io(
+        &handle->pipe.conn.writefile_thread_handle,
+        &handle->pipe.conn.thread_lock);
+
+  /* Drain the entire queue. */
+  while ((req = uv_remove_non_overlapped_write_req(handle)) != NULL) {
+    SET_REQ_ERROR(req, ERROR_OPERATION_ABORTED);
+    SET_REQ_NWRITTEN(req, 0);
+    uv__insert_pending_req(handle->loop, (uv_req_t*) req);
+  }
+}
+
+
 /* Cleans up uv_pipe_t (server or connection) and all resources associated with
  * it. */
 void uv__pipe_close(uv_loop_t* loop, uv_pipe_t* handle) {
@@ -1095,6 +1229,15 @@ void uv__pipe_close(uv_loop_t* loop, uv_pipe_t* handle) {
 
   if ((handle->flags & UV_HANDLE_CONNECTION)
       && handle->handle != INVALID_HANDLE_VALUE) {
+    if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
+      /* The non-overlapped writer thread reads handle->handle. To avoid
+       * the possibility of a stale value, we must shoot it down now before we
+       * close the handle. While we're at it, cancel the whole queue, to
+       * avoid putting things on the thread pool only for them to come straight
+       * back with an invalid handle error. */
+      uv__pipe_flush_non_overlapped_writes(handle);
+    }
+
     /* This will eventually destroy the write queue for us too. */
     close_pipe(handle);
   }
@@ -1239,22 +1382,16 @@ int uv__pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {
 }
 
 
-static DWORD WINAPI uv_pipe_zero_readfile_thread_proc(void* arg) {
+/* Register the current thread for cancellation via CancelSynchronousIo.
-  uv_read_t* req = (uv_read_t*) arg;
+ * Returns 0 on success, or a Windows error code on failure (including
-  uv_pipe_t* handle = (uv_pipe_t*) req->data;
+ * ERROR_OPERATION_ABORTED if pre-empted by uv__pipe_cancel_synchronous_io).
-  uv_loop_t* loop = handle->loop;
+ * On success, the caller must call uv__pipe_end_synchronous_io after the
-  volatile HANDLE* thread_ptr = &handle->pipe.conn.readfile_thread_handle;
+ * blocking operation. On failure, the caller must not. */
-  CRITICAL_SECTION* lock = &handle->pipe.conn.readfile_thread_lock;
+static DWORD uv__pipe_begin_synchronous_io(volatile HANDLE* thread_ptr,
+                                            CRITICAL_SECTION* lock,
+                                            HANDLE* thread_out) {
   HANDLE thread;
-  DWORD bytes;
-  DWORD err;
 
-  assert(req->type == UV_READ);
-  assert(handle->type == UV_NAMED_PIPE);
-
-  err = 0;
-
-  /* Create a handle to the current thread. */
   if (!DuplicateHandle(GetCurrentProcess(),
                        GetCurrentThread(),
                        GetCurrentProcess(),
@@ -1262,46 +1399,74 @@ static DWORD WINAPI uv_pipe_zero_readfile_thread_proc(void* arg) {
                        0,
                        FALSE,
                        DUPLICATE_SAME_ACCESS)) {
-    err = GetLastError();
+    /* Even if we failed here, we still need to synchronize */
-    goto out1;
+    DWORD err = GetLastError();
+    EnterCriticalSection(lock);
+    *thread_ptr = *thread_ptr == INVALID_HANDLE_VALUE ? NULL : INVALID_HANDLE_VALUE;
+    LeaveCriticalSection(lock);
+    WakeByAddressSingle((HANDLE*) thread_ptr);
+    return err;
   }
 
-  /* The lock needs to be held when thread handle is modified. */
   EnterCriticalSection(lock);
   if (*thread_ptr == INVALID_HANDLE_VALUE) {
-    /* uv__pipe_interrupt_read() cancelled reading before we got here. */
+    /* Cancelled before we got here. */
-    err = ERROR_OPERATION_ABORTED;
+    /* Acknowledge the cancellation and wake the waiting canceller. */
-  } else {
+    *thread_ptr = NULL;
-    /* Let main thread know which worker thread is doing the blocking read. */
+    LeaveCriticalSection(lock);
-    assert(*thread_ptr == NULL);
+    WakeByAddressSingle((HANDLE*) thread_ptr);
-    *thread_ptr = thread;
+    CloseHandle(thread);
+    return ERROR_OPERATION_ABORTED;
   }
+  assert(*thread_ptr == NULL);
+  *thread_ptr = thread;
   LeaveCriticalSection(lock);
 
-  if (err)
+  *thread_out = thread;
-    goto out2;
+  return 0;
+}
+
+
+/* Deregister the current thread after a blocking operation and synchronize
+ * with any in-progress cancellation. */
+static void uv__pipe_end_synchronous_io(volatile HANDLE* thread_ptr,
+                                         CRITICAL_SECTION* lock,
+                                         HANDLE thread) {
+  /* Acquire the lock, set the thread handle to INVALID_HANDLE_VALUE to signal
+   * completion, then release. This synchronizes with the cancellation spin
+   * loop which releases the lock around SwitchToThread(). */
+  EnterCriticalSection(lock);
+  assert(thread == *thread_ptr);
+  *thread_ptr = INVALID_HANDLE_VALUE;
+  LeaveCriticalSection(lock);
+
+  CloseHandle(thread);
+}
+
+
+static DWORD WINAPI uv_pipe_zero_readfile_thread_proc(void* arg) {
+  uv_read_t* req = (uv_read_t*) arg;
+  uv_pipe_t* handle = (uv_pipe_t*) req->data;
+  uv_loop_t* loop = handle->loop;
+  volatile HANDLE* thread_ptr = &handle->pipe.conn.readfile_thread_handle;
+  CRITICAL_SECTION* lock = &handle->pipe.conn.thread_lock;
+  HANDLE thread;
+  DWORD bytes;
+  DWORD err;
+
+  assert(req->type == UV_READ);
+  assert(handle->type == UV_NAMED_PIPE);
+
+  err = uv__pipe_begin_synchronous_io(thread_ptr, lock, &thread);
+  if (err)
+    goto done;
 
-  /* Block the thread until data is available on the pipe, or the read is
-   * cancelled. */
   if (!ReadFile(handle->handle, &uv_zero_, 0, &bytes, NULL))
     err = GetLastError();
 
-  /* Let the main thread know the worker is past the point of blocking. */
+  uv__pipe_end_synchronous_io(thread_ptr, lock, thread);
-  assert(thread == *thread_ptr);
-  *thread_ptr = INVALID_HANDLE_VALUE;
 
-  /* Briefly acquire the mutex. Since the main thread holds the lock while it
+done:
-   * is spinning trying to cancel this thread's I/O, we will block here until
-   * it stops doing that. */
-  EnterCriticalSection(lock);
-  LeaveCriticalSection(lock);
-
-out2:
-  /* Close the handle to the current thread. */
-  CloseHandle(thread);
-
-out1:
-  /* Set request status and post a completion record to the IOCP. */
   if (err)
     SET_REQ_ERROR(req, err);
   else
@@ -1314,24 +1479,39 @@ out1:
 
 static DWORD WINAPI uv_pipe_writefile_thread_proc(void* parameter) {
   int result;
+  HANDLE thread;
   DWORD bytes;
+  DWORD err;
   uv_write_t* req = (uv_write_t*) parameter;
   uv_pipe_t* handle = (uv_pipe_t*) req->handle;
   uv_loop_t* loop = handle->loop;
+  volatile HANDLE* thread_ptr = &handle->pipe.conn.writefile_thread_handle;
+  CRITICAL_SECTION* lock = &handle->pipe.conn.thread_lock;
 
   assert(req != NULL);
   assert(req->type == UV_WRITE);
   assert(handle->type == UV_NAMED_PIPE);
 
+  bytes = 0;
+
+  err = uv__pipe_begin_synchronous_io(thread_ptr, lock, &thread);
+  if (err)
+    goto done;
+
   result = WriteFile(handle->handle,
                      req->write_buffer.base,
                      req->write_buffer.len,
                      &bytes,
                      NULL);
 
-  if (!result) {
+  if (!result)
-    SET_REQ_ERROR(req, GetLastError());
+    err = GetLastError();
-  }
+
+  uv__pipe_end_synchronous_io(thread_ptr, lock, thread);
+
+done:
+  if (err)
+    SET_REQ_ERROR(req, err);
   SET_REQ_NWRITTEN(req, bytes);
 
   POST_COMPLETION_FOR_REQ(loop, req);
@@ -1389,6 +1569,7 @@ static void uv__pipe_queue_read(uv_loop_t* loop, uv_pipe_t* handle) {
   req = &handle->read_req;
 
   if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
+    assert(handle->pipe.conn.readfile_thread_handle == INVALID_HANDLE_VALUE);
     handle->pipe.conn.readfile_thread_handle = NULL; /* Reset cancellation. */
     if (!QueueUserWorkItem(&uv_pipe_zero_readfile_thread_proc,
                            req,
@@ -1483,30 +1664,13 @@ static void uv__insert_non_overlapped_write_req(uv_pipe_t* handle,
 }
 
 
-static uv_write_t* uv_remove_non_overlapped_write_req(uv_pipe_t* handle) {
-  uv_write_t* req;
-
-  if (handle->pipe.conn.non_overlapped_writes_tail) {
-    req = (uv_write_t*)handle->pipe.conn.non_overlapped_writes_tail->next_req;
-
-    if (req == handle->pipe.conn.non_overlapped_writes_tail) {
-      handle->pipe.conn.non_overlapped_writes_tail = NULL;
-    } else {
-      handle->pipe.conn.non_overlapped_writes_tail->next_req =
-        req->next_req;
-    }
-
-    return req;
-  } else {
-    /* queue empty */
-    return NULL;
-  }
-}
-
-
 static void uv__queue_non_overlapped_write(uv_pipe_t* handle) {
   uv_write_t* req = uv_remove_non_overlapped_write_req(handle);
   if (req) {
+    assert(handle->pipe.conn.non_overlapped_write_active == NULL);
+    assert(handle->pipe.conn.writefile_thread_handle == INVALID_HANDLE_VALUE);
+    handle->pipe.conn.non_overlapped_write_active = req;
+    handle->pipe.conn.writefile_thread_handle = NULL; /* Reset cancellation. */
     if (!QueueUserWorkItem(&uv_pipe_writefile_thread_proc,
                            req,
                            WT_EXECUTELONGFUNCTION)) {
@@ -1643,13 +1807,17 @@ static int uv__pipe_write_data(uv_loop_t* loop,
     REGISTER_HANDLE_REQ(loop, handle);
     handle->reqs_pending++;
     handle->stream.conn.write_reqs_pending++;
-    POST_COMPLETION_FOR_REQ(loop, req);
+    uv__insert_pending_req(loop, (uv_req_t*)req);
     return 0;
   } else if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
     req->write_buffer = write_buf;
     uv__insert_non_overlapped_write_req(handle, req);
     if (handle->stream.conn.write_reqs_pending == 0) {
       uv__queue_non_overlapped_write(handle);
+      /* There shouldn't have been any queued writes before, so we should have
+       * dispatched the request we just added. Sanity check the state
+       * synchronization here. */
+      assert(handle->pipe.conn.non_overlapped_writes_tail == NULL);
     }
 
     /* Request queued by the kernel. */
@@ -2124,7 +2292,7 @@ void uv__process_pipe_read_req(uv_loop_t* loop,
   DWORD bytes_requested;
   assert(handle->type == UV_NAMED_PIPE);
 
-  handle->flags &= ~(UV_HANDLE_READ_PENDING | UV_HANDLE_CANCELLATION_PENDING);
+  handle->flags &= ~(UV_HANDLE_READ_PENDING | UV_HANDLE_READ_CANCELLATION_PENDING);
   DECREASE_PENDING_REQ_COUNT(handle);
   eof_timer_stop(handle);
 
@@ -2202,6 +2370,20 @@ void uv__process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,
 
   err = GET_REQ_ERROR(req);
 
+  /* For non-overlapped pipes, manage the write queue before unwrapping
+   * coalesced writes, since the coalesced wrapper is what's in the queue.
+   *
+   * If this request was the active write (dispatched to the thread pool),
+   * clear the active slot and dispatch the next queued write.  Queue-cancelled
+   * writes (removed by uv__pipe_write_cancel or flush) were never dispatched,
+   * so they must not trigger another dispatch. */
+  if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
+    if (req == handle->pipe.conn.non_overlapped_write_active) {
+      handle->pipe.conn.non_overlapped_write_active = NULL;
+      uv__queue_non_overlapped_write(handle);
+    }
+  }
+
   /* If this was a coalesced write, extract pointer to the user_provided
    * uv_write_t structure so we can pass the expected pointer to the callback,
    * then free the heap-allocated write req. */
@@ -2220,12 +2402,6 @@ void uv__process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,
 
   handle->stream.conn.write_reqs_pending--;
 
-  if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE &&
-      handle->pipe.conn.non_overlapped_writes_tail) {
-    assert(handle->stream.conn.write_reqs_pending > 0);
-    uv__queue_non_overlapped_write(handle);
-  }
-
   if (handle->stream.conn.write_reqs_pending == 0 &&
       uv__is_stream_shutting(handle))
     uv__pipe_shutdown(loop, handle, handle->stream.conn.shutdown_req);

src/win/stream.c

@@ -266,16 +266,14 @@ int uv_write_cancel(uv_write_t* req) {
 
   switch (stream->type) {
     case UV_TCP:
-      handle = (HANDLE)((uv_tcp_t*)stream)->socket;
+      handle = (HANDLE) ((uv_tcp_t*) stream)->socket;
       break;
     case UV_NAMED_PIPE:
-      handle = ((uv_pipe_t*)stream)->handle;
+      handle = ((uv_pipe_t*) stream)->handle;
 
       if ((stream->flags & (UV_HANDLE_BLOCKING_WRITES | UV_HANDLE_NON_OVERLAPPED_PIPE)) ==
           UV_HANDLE_NON_OVERLAPPED_PIPE) {
-        /* Non-overlapped, non-blocking writes run on the threadpool and we
+        return uv__pipe_write_cancel_non_overlapped((uv_pipe_t*) stream, req);
-           do not have a good way to cancel them. */
-        return UV_ENOTSUP;
       }
 
       break;

src/win/tty.c

@@ -997,13 +997,13 @@ void uv_process_tty_read_line_req(uv_loop_t* loop, uv_tty_t* handle,
                       &buf);
     }
   } else {
-    if (!(handle->flags & UV_HANDLE_CANCELLATION_PENDING) &&
+    if (!(handle->flags & UV_HANDLE_READ_CANCELLATION_PENDING) &&
         req->u.io.overlapped.InternalHigh != 0) {
       /* Read successful. TODO: read unicode, convert to utf-8 */
       DWORD bytes = req->u.io.overlapped.InternalHigh;
       handle->read_cb((uv_stream_t*) handle, bytes, &buf);
     }
-    handle->flags &= ~UV_HANDLE_CANCELLATION_PENDING;
+    handle->flags &= ~UV_HANDLE_READ_CANCELLATION_PENDING;
   }
 
   /* Wait for more input events. */
@@ -1086,13 +1086,13 @@ int uv__tty_read_stop(uv_tty_t* handle) {
     if (!WriteConsoleInputW(handle->handle, &record, 1, &written)) {
       return GetLastError();
     }
-  } else if (!(handle->flags & UV_HANDLE_CANCELLATION_PENDING)) {
+  } else if (!(handle->flags & UV_HANDLE_READ_CANCELLATION_PENDING)) {
     /* Cancel line-buffered read if not already pending */
     err = uv__cancel_read_console(handle);
     if (err)
       return err;
 
-    handle->flags |= UV_HANDLE_CANCELLATION_PENDING;
+    handle->flags |= UV_HANDLE_READ_CANCELLATION_PENDING;
   }
 
   return 0;
@@ -1105,7 +1105,7 @@ static int uv__cancel_read_console(uv_tty_t* handle) {
   DWORD err = 0;
   LONG status;
 
-  assert(!(handle->flags & UV_HANDLE_CANCELLATION_PENDING));
+  assert(!(handle->flags & UV_HANDLE_READ_CANCELLATION_PENDING));
 
   /* Hold the output lock during the cancellation, to ensure that further
      writes don't interfere with the screen state. It will be the ReadConsole

test/test-list.h

@@ -112,6 +112,8 @@ TEST_DECLARE   (tcp_write_in_a_row)
 TEST_DECLARE   (tcp_try_write_error)
 TEST_DECLARE   (tcp_write_queue_order)
 TEST_DECLARE   (tcp_write_cancel)
+TEST_DECLARE   (pipe_write_cancel)
+TEST_DECLARE   (pipe_write_cancel_all)
 TEST_DECLARE   (tcp_write_nwritten)
 TEST_DECLARE   (pipe_write_nwritten)
 TEST_DECLARE   (tcp_open)
@@ -727,6 +729,8 @@ TASK_LIST_START
 
   TEST_ENTRY  (tcp_write_queue_order)
   TEST_ENTRY  (tcp_write_cancel)
+  TEST_ENTRY  (pipe_write_cancel)
+  TEST_ENTRY  (pipe_write_cancel_all)
   TEST_ENTRY  (tcp_write_nwritten)
   TEST_ENTRY  (pipe_write_nwritten)
 

test/test-write-cancel.c

@@ -28,22 +28,26 @@
 
 #define REQ_COUNT 100
 
-static uv_tcp_t server;
-static uv_tcp_t client;
-static uv_tcp_t incoming;
 static int close_cb_called;
 static int write_cb_called;
 static int cancelled_count;
 static int connected;
 static int closing;
 
-static uv_write_t write_reqs[REQ_COUNT];
-static char buf_data[16 * 1024];
-
 static void close_cb(uv_handle_t* handle) {
   close_cb_called++;
 }
 
+
+/*
+ * tcp_write_cancel
+ */
+static uv_tcp_t server;
+static uv_tcp_t client;
+static uv_tcp_t incoming;
+static uv_write_t write_reqs[REQ_COUNT];
+static char buf_data[16 * 1024];
+
 static void connection_cb(uv_stream_t* tcp, int status) {
   ASSERT_OK(status);
   ASSERT_OK(uv_tcp_init(tcp->loop, &incoming));
@@ -54,11 +58,12 @@ static void connection_cb(uv_stream_t* tcp, int status) {
 
 static void write_cb(uv_write_t* req, int status) {
   write_cb_called++;
-  if (status == UV_ECANCELED && !closing)
+  if (status == UV_ECANCELED && closing == 0)
     cancelled_count++;
 
-  if (cancelled_count >= 5 && !closing) {
+  if (cancelled_count >= 5 && closing == 0) {
-    closing = 1;
+    closing++;
+    ASSERT_EQ(1, closing);
     uv_close((uv_handle_t*) &client, close_cb);
     uv_close((uv_handle_t*) &server, close_cb);
     if (connected)
@@ -144,7 +149,152 @@ TEST_IMPL(tcp_write_cancel) {
 
 
 /*
- * Test that uv_write_nwritten returns correct byte count on success.
+ * pipe_write_cancel / pipe_write_cancel_all
+ *
+ * uv_pipe() creates non-overlapped handles, so on Windows these exercise
+ * the synchronous (thread pool) write cancellation path. Half the writes
+ * use multiple buffers to exercise the coalesced write path.
+ */
+static uv_pipe_t pipe_cancel_writer;
+static uv_pipe_t pipe_cancel_reader;
+static uv_write_t pipe_cancel_write_reqs[REQ_COUNT];
+static char pipe_cancel_buf_data[64 * 1024];
+static int cancel_target;
+static uv_write_cb pipe_cancel_cb;
+
+static void pipe_cancel_setup(uv_loop_t* loop) {
+  uv_buf_t bufs[4];
+  int fds[2];
+  int r;
+  int i;
+
+  close_cb_called = 0;
+  write_cb_called = 0;
+  cancelled_count = 0;
+  closing = 0;
+  memset(pipe_cancel_buf_data, 'D', sizeof(pipe_cancel_buf_data));
+
+  ASSERT_OK(uv_pipe(fds, 0, 0));
+
+  ASSERT_OK(uv_pipe_init(loop, &pipe_cancel_writer, 0));
+  ASSERT_OK(uv_pipe_init(loop, &pipe_cancel_reader, 0));
+
+  ASSERT_OK(uv_pipe_open(&pipe_cancel_writer, fds[1]));
+  ASSERT_OK(uv_pipe_open(&pipe_cancel_reader, fds[0]));
+
+  bufs[0] = uv_buf_init(pipe_cancel_buf_data,
+                         sizeof(pipe_cancel_buf_data) / 4);
+  bufs[1] = uv_buf_init(pipe_cancel_buf_data +
+                         sizeof(pipe_cancel_buf_data) / 4,
+                         sizeof(pipe_cancel_buf_data) / 4);
+  bufs[2] = uv_buf_init(pipe_cancel_buf_data +
+                         sizeof(pipe_cancel_buf_data) / 2,
+                         sizeof(pipe_cancel_buf_data) / 4);
+  bufs[3] = uv_buf_init(pipe_cancel_buf_data +
+                         3 * sizeof(pipe_cancel_buf_data) / 4,
+                         sizeof(pipe_cancel_buf_data) / 4);
+
+  /* Queue many writes to fill the pipe buffer. Alternate between single-buffer
+   * and multi-buffer writes so the latter exercise the coalesced write path
+   * on Windows. */
+  for (i = 0; i < REQ_COUNT; i++) {
+    if (i % 2 == 0) {
+      r = uv_write(&pipe_cancel_write_reqs[i],
+                   (uv_stream_t*) &pipe_cancel_writer,
+                   &bufs[0],
+                   1,
+                   pipe_cancel_cb);
+    } else {
+      r = uv_write(&pipe_cancel_write_reqs[i],
+                   (uv_stream_t*) &pipe_cancel_writer,
+                   bufs,
+                   4,
+                   pipe_cancel_cb);
+    }
+    ASSERT_OK(r);
+  }
+}
+
+static void pipe_cancel_close(void) {
+  if (closing == 0) {
+    closing++;
+    ASSERT_EQ(1, closing);
+    uv_close((uv_handle_t*) &pipe_cancel_writer, close_cb);
+    uv_close((uv_handle_t*) &pipe_cancel_reader, close_cb);
+  }
+}
+
+static void pipe_cancel_write_cb_tail(uv_write_t* req, int status) {
+  write_cb_called++;
+  if (status == UV_ECANCELED && closing == 0)
+    cancelled_count++;
+
+  if (cancelled_count >= cancel_target && closing == 0)
+    pipe_cancel_close();
+}
+
+TEST_IMPL(pipe_write_cancel) {
+  uv_loop_t* loop;
+  int i;
+
+  loop = uv_default_loop();
+  cancel_target = 5;
+  pipe_cancel_cb = pipe_cancel_write_cb_tail;
+  pipe_cancel_setup(loop);
+
+  /* Cancel the trailing writes which should be queued. */
+  for (i = REQ_COUNT - 5; i < REQ_COUNT; i++)
+    ASSERT_OK(uv_write_cancel(&pipe_cancel_write_reqs[i]));
+
+  ASSERT_OK(uv_run(loop, UV_RUN_DEFAULT));
+
+  ASSERT_EQ(5, cancelled_count);
+  ASSERT_EQ(2, close_cb_called);
+
+  MAKE_VALGRIND_HAPPY(loop);
+  return 0;
+}
+
+static void pipe_cancel_write_cb_all(uv_write_t* req, int status) {
+  write_cb_called++;
+  if (status == UV_ECANCELED)
+    cancelled_count++;
+}
+
+TEST_IMPL(pipe_write_cancel_all) {
+  uv_loop_t* loop;
+  int i;
+
+  loop = uv_default_loop();
+  pipe_cancel_cb = pipe_cancel_write_cb_all;
+  pipe_cancel_setup(loop);
+
+  /* Cancel all writes. The first few may have already completed; one may be
+   * currently blocking in the thread pool (on Windows). Both cases are
+   * exercised by cancelling every request in order. */
+  for (i = 0; i < REQ_COUNT; i++)
+    uv_write_cancel(&pipe_cancel_write_reqs[i]);
+
+  /* Close the handles so the event loop can drain. */
+  uv_close((uv_handle_t*) &pipe_cancel_writer, close_cb);
+  uv_close((uv_handle_t*) &pipe_cancel_reader, close_cb);
+
+  ASSERT_OK(uv_run(loop, UV_RUN_DEFAULT));
+
+  /* Every write must have gotten a callback. */
+  ASSERT_EQ(REQ_COUNT, write_cb_called);
+  /* We don't know exactly how many were cancelled vs already completed,
+   * but at least the tail end should have been cancelled. */
+  ASSERT_GT(cancelled_count, 0);
+  ASSERT_EQ(2, close_cb_called);
+
+  MAKE_VALGRIND_HAPPY(loop);
+  return 0;
+}
+
+
+/*
+ * tcp_write_nwritten
  */
 static uv_tcp_t nwritten_server;
 static uv_tcp_t nwritten_client;
@@ -154,18 +304,14 @@ static int nwritten_cb_called;
 static size_t nwritten_value;
 static char nwritten_buf_data[1024];
 
-static void nwritten_close_cb(uv_handle_t* handle) {
-  close_cb_called++;
-}
-
 static void nwritten_write_cb(uv_write_t* req, int status) {
   ASSERT_OK(status);
   nwritten_cb_called++;
   nwritten_value = uv_write_nwritten(req);
 
-  uv_close((uv_handle_t*) &nwritten_client, nwritten_close_cb);
+  uv_close((uv_handle_t*) &nwritten_client, close_cb);
-  uv_close((uv_handle_t*) &nwritten_server, nwritten_close_cb);
+  uv_close((uv_handle_t*) &nwritten_server, close_cb);
-  uv_close((uv_handle_t*) &nwritten_incoming, nwritten_close_cb);
+  uv_close((uv_handle_t*) &nwritten_incoming, close_cb);
 }
 
 static void nwritten_connection_cb(uv_stream_t* tcp, int status) {
@@ -219,7 +365,7 @@ TEST_IMPL(tcp_write_nwritten) {
 
 
 /*
- * Test that uv_write_nwritten returns correct byte count for pipes.
+ * pipe_write_nwritten
  */
 static uv_pipe_t pipe_client;
 static uv_pipe_t pipe_server;
@@ -228,17 +374,13 @@ static int pipe_cb_called;
 static size_t pipe_nwritten_value;
 static char pipe_buf_data[1024];
 
-static void pipe_close_cb(uv_handle_t* handle) {
-  close_cb_called++;
-}
-
 static void pipe_write_cb(uv_write_t* req, int status) {
   ASSERT_OK(status);
   pipe_cb_called++;
   pipe_nwritten_value = uv_write_nwritten(req);
 
-  uv_close((uv_handle_t*) &pipe_client, pipe_close_cb);
+  uv_close((uv_handle_t*) &pipe_client, close_cb);
-  uv_close((uv_handle_t*) &pipe_server, pipe_close_cb);
+  uv_close((uv_handle_t*) &pipe_server, close_cb);
 }
 
 TEST_IMPL(pipe_write_nwritten) {