Merge 6d0459bccf into fc90bb9843

CMakeLists.txt

@@ -596,6 +596,7 @@ if(LIBUV_BUILD_TESTS)
        test/test-pass-always.c
        test/test-ping-pong.c
        test/test-pipe-bind-error.c
+       test/test-pipe-blocking-subprocess.c
        test/test-pipe-close-stdout-read-stdin.c
        test/test-pipe-connect-error.c
        test/test-pipe-connect-multiple.c

Makefile.am

@@ -221,6 +221,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \
                          test/test-pass-always.c \
                          test/test-ping-pong.c \
                          test/test-pipe-bind-error.c \
+                         test/test-pipe-blocking-subprocess.c \
                          test/test-pipe-connect-error.c \
                          test/test-pipe-connect-multiple.c \
                          test/test-pipe-connect-prepare.c \

docs/src/loop.rst

@@ -89,10 +89,18 @@ API
     - UV_LOOP_USE_IO_URING_SQPOLL: Enable SQPOLL io_uring instance to handle
       asynchronous file system operations.
 
+    - UV_LOOP_CANCEL_SIGNAL: Allow blocking IO to be cancelled.  The argument is
+      the signal number to use for cancellation.
+
+      If this is not configured, a :c:type:`uv_stream_t` in blocking mode can
+      cause a thread in the thread pool to become permanently blocked.
+
     .. versionchanged:: 1.39.0 added the UV_METRICS_IDLE_TIME option.
 
     .. versionchanged:: 1.49.0 added the UV_LOOP_USE_IO_URING_SQPOLL option.
 
+    .. versionchanged:: 1.52.0 added the UV_LOOP_CANCEL_SIGNAL option.
+
 .. c:function:: int uv_loop_close(uv_loop_t* loop)
 
     Releases all internal loop resources. Call this function only when the loop

include/uv.h

@@ -261,8 +261,9 @@ typedef struct uv_metrics_s uv_metrics_t;
 typedef enum {
   UV_LOOP_BLOCK_SIGNAL = 0,
   UV_METRICS_IDLE_TIME,
-  UV_LOOP_USE_IO_URING_SQPOLL
+  UV_LOOP_USE_IO_URING_SQPOLL,
 #define UV_LOOP_USE_IO_URING_SQPOLL UV_LOOP_USE_IO_URING_SQPOLL
+  UV_LOOP_CANCEL_SIGNAL
 } uv_loop_option;
 
 typedef enum {

include/uv/threadpool.h

@@ -27,11 +27,23 @@
 #ifndef UV_THREADPOOL_H_
 #define UV_THREADPOOL_H_
 
+enum {
+  UV__WORK_BUSY = 0,
+  UV__WORK_CANCELLABLE,
+  UV__WORK_CANCEL_PENDING,
+  UV__WORK_CANCELLED,
+  UV__WORK_DONE
+};
+
 struct uv__work {
   void (*work)(struct uv__work *w);
   void (*done)(struct uv__work *w, int status);
   struct uv_loop_s* loop;
   struct uv__queue wq;
+#ifndef _WIN32
+  pthread_t thread;
+  _Atomic char state;
+#endif
 };
 
 #endif /* UV_THREADPOOL_H_ */

include/uv/unix.h

@@ -245,6 +245,7 @@ typedef struct {
   uv__io_t signal_io_watcher;                                                 \
   uv_signal_t child_watcher;                                                  \
   int emfile_fd;                                                              \
+  int cancel_signum;                                                          \
   UV_PLATFORM_LOOP_FIELDS                                                     \
 
 #define UV_REQ_TYPE_PRIVATE /* empty */
@@ -258,7 +259,7 @@ typedef struct {
   unsigned int write_index;                                                   \
   uv_buf_t* bufs;                                                             \
   unsigned int nbufs;                                                         \
-  int error;                                                                  \
+  ssize_t result;                                                             \
   uv_buf_t bufsml[4];                                                         \
 
 #define UV_CONNECT_PRIVATE_FIELDS                                             \
@@ -292,6 +293,7 @@ typedef struct {
   int delayed_error;                                                          \
   int accepted_fd;                                                            \
   void* queued_fds;                                                           \
+  struct uv__work blocked_write;                                              \
   UV_STREAM_PRIVATE_PLATFORM_FIELDS                                           \
 
 #define UV_TCP_PRIVATE_FIELDS /* empty */

src/threadpool.c

@@ -58,10 +58,17 @@ static void worker(void* arg) {
   struct uv__work* w;
   struct uv__queue* q;
   int is_slow_work;
+#ifndef _WIN32
+  pthread_t self;
+  char expected;
+#endif
 
   uv_thread_setname("libuv-worker");
   uv_sem_post((uv_sem_t*) arg);
   arg = NULL;
+#ifndef _WIN32
+  self = pthread_self();
+#endif
 
   uv_mutex_lock(&mutex);
   for (;;) {
@@ -117,14 +124,31 @@ static void worker(void* arg) {
       }
     }
 
+    w = uv__queue_data(q, struct uv__work, wq);
+#ifndef _WIN32
+    w->thread = self;
+#endif
+
     uv_mutex_unlock(&mutex);
 
-    w = uv__queue_data(q, struct uv__work, wq);
     w->work(w);
 
     uv_mutex_lock(&w->loop->wq_mutex);
     w->work = NULL;  /* Signal uv_cancel() that the work req is done
                         executing. */
+#ifndef _WIN32
+    expected = UV__WORK_CANCELLABLE;
+    if (!atomic_compare_exchange_strong_explicit(&w->state,
+                                                 &expected,
+                                                 UV__WORK_DONE,
+                                                 memory_order_relaxed,
+                                                 memory_order_relaxed)) {
+      if (expected == UV__WORK_CANCEL_PENDING)
+        atomic_store_explicit(&w->state,
+                              UV__WORK_CANCELLED,
+                              memory_order_relaxed);
+    }
+#endif
     uv__queue_insert_tail(&w->loop->wq, &w->wq);
     uv_async_send(&w->loop->wq_async);
     uv_mutex_unlock(&w->loop->wq_mutex);
@@ -273,15 +297,31 @@ static void init_once(void) {
 }
 
 
+#ifndef _WIN32
+void uv__cancel_signal_handler(int signo) {
+    /* We just want to trigger EINTR. */
+}
+#endif
+
+
 void uv__work_submit(uv_loop_t* loop,
                      struct uv__work* w,
                      enum uv__work_kind kind,
                      void (*work)(struct uv__work* w),
                      void (*done)(struct uv__work* w, int status)) {
+#ifndef _WIN32
+  char state;
+#endif
+
   uv_once(&once, init_once);
   w->loop = loop;
   w->work = work;
   w->done = done;
+#ifndef _WIN32
+  state = kind == UV__WORK_FAST_IO_CANCELLABLE ? UV__WORK_CANCELLABLE
+                                               : UV__WORK_BUSY;
+  atomic_store_explicit(&w->state, state, memory_order_relaxed);
+#endif
   post(&w->wq, kind);
 }
 
@@ -289,8 +329,13 @@ void uv__work_submit(uv_loop_t* loop,
 /* TODO(bnoordhuis) teach libuv how to cancel file operations
  * that go through io_uring instead of the thread pool.
  */
-static int uv__work_cancel(uv_loop_t* loop, uv_req_t* req, struct uv__work* w) {
+int uv__work_cancel(uv_loop_t* loop, struct uv__work* w) {
   int cancelled;
+#ifndef _WIN32
+  char expected;
+  pthread_t thread;
+  int i;
+#endif
 
   uv_once(&once, init_once);  /* Ensure |mutex| is initialized. */
   uv_mutex_lock(&mutex);
@@ -303,8 +348,39 @@ static int uv__work_cancel(uv_loop_t* loop, uv_req_t* req, struct uv__work* w) {
   uv_mutex_unlock(&w->loop->wq_mutex);
   uv_mutex_unlock(&mutex);
 
+#ifdef _WIN32
   if (!cancelled)
     return UV_EBUSY;
+#else
+  if (!cancelled) {
+    if (atomic_load_explicit(&w->state, memory_order_relaxed) == UV__WORK_BUSY)
+      return UV_EBUSY;
+
+    if (loop->cancel_signum == -1)
+      return UV_EBUSY;
+
+    expected = UV__WORK_CANCELLABLE;
+    if (atomic_compare_exchange_strong_explicit(&w->state,
+                                                &expected,
+                                                UV__WORK_CANCEL_PENDING,
+                                                memory_order_relaxed,
+                                                memory_order_relaxed)) {
+      thread = w->thread;
+      i = 0;
+      do {
+        if (i >= 10)
+          return UV_EBUSY;
+        if (i > 0)
+          uv_sleep(1 << i++);
+        if (pthread_kill(thread, loop->cancel_signum) != 0)
+          abort();
+      } while (atomic_load_explicit(&w->state, memory_order_relaxed) ==
+             UV__WORK_CANCEL_PENDING);
+    }
+
+    return 0;
+  }
+#endif
 
   w->work = uv__cancelled;
   uv_mutex_lock(&loop->wq_mutex);
@@ -337,6 +413,10 @@ void uv__work_done(uv_async_t* handle) {
 
     w = container_of(q, struct uv__work, wq);
     err = (w->work == uv__cancelled) ? UV_ECANCELED : 0;
+#ifndef _WIN32
+    if (atomic_load_explicit(&w->state, memory_order_relaxed) == UV__WORK_CANCELLED)
+      err = UV_ECANCELED;
+#endif
     w->done(w, err);
     nevents++;
   }
@@ -425,5 +505,5 @@ int uv_cancel(uv_req_t* req) {
     return UV_EINVAL;
   }
 
-  return uv__work_cancel(loop, req, wreq);
+  return uv__work_cancel(loop, wreq);
 }

src/unix/core.c

@@ -344,7 +344,12 @@ static void uv__finish_close(uv_handle_t* handle) {
     case UV_NAMED_PIPE:
     case UV_TCP:
     case UV_TTY:
-      uv__stream_destroy((uv_stream_t*)handle);
+      if (handle->flags & UV_HANDLE_WRITE_PENDING) {
+        handle->flags ^= UV_HANDLE_CLOSED;
+        uv__make_close_pending(handle); /* Back into the queue. */
+        return;
+      }
+      uv__stream_destroy((uv_stream_t *)handle);
       break;
 
     case UV_UDP:

src/unix/internal.h

@@ -591,4 +591,16 @@ int uv__get_constrained_cpu(long long* quota);
 #define UV__KQUEUE_EVFILT_USER 0
 #endif
 
+static inline int uv__work_check_cancelled(struct uv__work* w) {
+  if (w == NULL)
+    return 0;
+
+  if (atomic_load_explicit(&w->state, memory_order_relaxed) !=
+      UV__WORK_CANCEL_PENDING)
+    return 0;
+
+  atomic_store_explicit(&w->state, UV__WORK_CANCELLED, memory_order_relaxed);
+  return 1;
+}
+
 #endif /* UV_UNIX_INTERNAL_H_ */

src/unix/loop.c

@@ -23,6 +23,7 @@
 #include "uv/tree.h"
 #include "internal.h"
 #include "heap-inl.h"
+#include <signal.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
@@ -72,6 +73,7 @@ int uv_loop_init(uv_loop_t* loop) {
   loop->signal_pipefd[1] = -1;
   loop->backend_fd = -1;
   loop->emfile_fd = -1;
+  loop->cancel_signum = -1;
 
   loop->timer_counter = 0;
   loop->stop_flag = 0;
@@ -213,6 +215,7 @@ void uv__loop_close(uv_loop_t* loop) {
 
 int uv__loop_configure(uv_loop_t* loop, uv_loop_option option, va_list ap) {
   uv__loop_internal_fields_t* lfields;
+  struct sigaction sa;
 
   lfields = uv__get_internal_fields(loop);
   if (option == UV_METRICS_IDLE_TIME) {
@@ -227,13 +230,23 @@ int uv__loop_configure(uv_loop_t* loop, uv_loop_option option, va_list ap) {
   }
 #endif
 
+  if (option == UV_LOOP_BLOCK_SIGNAL) {
+    if (va_arg(ap, int) != SIGPROF)
+      return UV_EINVAL;
+    loop->flags |= UV_LOOP_BLOCK_SIGPROF;
+    return 0;
+  }
 
-  if (option != UV_LOOP_BLOCK_SIGNAL)
+  if (option == UV_LOOP_CANCEL_SIGNAL) {
-    return UV_ENOSYS;
+    loop->cancel_signum = va_arg(ap, int);
+    memset(&sa, 0, sizeof sa);
+    sa.sa_handler = &uv__cancel_signal_handler;
+    if (sigaction(loop->cancel_signum, &sa, NULL) == -1) {
+      loop->cancel_signum = -1;
+      return UV__ERR(errno);
+    }
+    return 0;
+  }
 
-  if (va_arg(ap, int) != SIGPROF)
+  return UV_ENOSYS;
-    return UV_EINVAL;
-
-  loop->flags |= UV_LOOP_BLOCK_SIGPROF;
-  return 0;
 }

src/unix/process.c

@@ -216,16 +216,24 @@ static int uv__process_init_stdio(uv_stdio_container_t* container, int fds[2]) {
     return ret;
 
   case UV_INHERIT_FD:
+    fd = container->data.fd;
+    if (fd == -1)
+      return UV_EINVAL;
+    fds[1] = fd;
+    return 0;
+
   case UV_INHERIT_STREAM:
-    if (container->flags & UV_INHERIT_FD)
+    fd = uv__stream_fd(container->data.stream);
-      fd = container->data.fd;
-    else
-      fd = uv__stream_fd(container->data.stream);
 
     if (fd == -1)
       return UV_EINVAL;
-
     fds[1] = fd;
+
+    ret = uv__nonblock(fd, 0);
+    if (ret != 0)
+      return ret;
+    container->data.stream->flags |= UV_HANDLE_BLOCKING_WRITES;
+
     return 0;
 
   default:
@@ -372,9 +380,6 @@ static void uv__process_child_init(const uv_process_options_t* options,
     if (fd == -1)
       uv__write_errno(error_fd);
 
-    if (fd <= 2 && close_fd == -1)
-      uv__nonblock_fcntl(fd, 0);
-
     if (close_fd >= stdio_count)
       uv__close(close_fd);
   }
@@ -625,10 +630,6 @@ static int uv__spawn_set_posix_spawn_file_actions(
     assert(err != ENOSYS);
     if (err != 0)
       goto error;
-
-    /* Make sure the fd is marked as non-blocking (state shared between child
-     * and parent). */
-    uv__nonblock_fcntl(use_fd, 0);
   }
 
   /* Finally, close all the superfluous descriptors */

src/unix/stream.c

@@ -444,7 +444,7 @@ void uv__stream_flush_write_queue(uv_stream_t* stream, int error) {
     uv__queue_remove(q);
 
     req = uv__queue_data(q, uv_write_t, queue);
-    req->error = error;
+    req->result = error;
 
     uv__queue_insert_tail(&stream->write_completed_queue, &req->queue);
   }
@@ -455,6 +455,13 @@ void uv__stream_destroy(uv_stream_t* stream) {
   assert(!uv__io_active(&stream->io_watcher, POLLIN | POLLOUT));
   assert(stream->flags & UV_HANDLE_CLOSED);
 
+  if (stream->io_watcher.fd != -1) {
+    /* Don't close stdio file descriptors.  Nothing good comes from it. */
+    if (stream->io_watcher.fd > STDERR_FILENO)
+      uv__close(stream->io_watcher.fd);
+    stream->io_watcher.fd = -1;
+  }
+
   if (stream->connect_req) {
     uv__req_unregister(stream->loop);
     stream->connect_req->cb(stream->connect_req, UV_ECANCELED);
@@ -722,17 +729,19 @@ static void uv__write_req_finish(uv_write_t* req) {
    * they should stop writing - which they should if we got an error. Something
    * to revisit in future revisions of the libuv API.
    */
-  if (req->error == 0) {
+  if (req->result >= 0) {
     if (req->bufs != req->bufsml)
       uv__free(req->bufs);
     req->bufs = NULL;
+    req->result = 0;
   }
 
   /* Add it to the write_completed_queue where it will have its
    * callback called in the near future.
    */
   uv__queue_insert_tail(&stream->write_completed_queue, &req->queue);
-  uv__io_feed(stream->loop, &stream->io_watcher);
+  if (!(stream->flags & UV_HANDLE_CLOSING))
+    uv__io_feed(stream->loop, &stream->io_watcher);
 }
 
 
@@ -750,10 +759,12 @@ static int uv__handle_fd(uv_handle_t* handle) {
   }
 }
 
+
 static int uv__try_write(uv_stream_t* stream,
                          const uv_buf_t bufs[],
                          unsigned int nbufs,
-                         uv_stream_t* send_handle) {
+                         uv_stream_t* send_handle,
+                         struct uv__work* w) {
   struct iovec* iov;
   int iovmax;
   int iovcnt;
@@ -765,6 +776,7 @@ static int uv__try_write(uv_stream_t* stream,
    */
   iov = (struct iovec*) bufs;
   iovcnt = nbufs;
+  n = 0;
 
   iovmax = uv__getiovmax();
 
@@ -804,13 +816,17 @@ static int uv__try_write(uv_stream_t* stream,
     cmsg.hdr.cmsg_len = CMSG_LEN(sizeof(fd_to_send));
     memcpy(CMSG_DATA(&cmsg.hdr), &fd_to_send, sizeof(fd_to_send));
 
-    do
+    do {
+      if (uv__work_check_cancelled(w))
+        break;
       n = sendmsg(uv__stream_fd(stream), &msg, 0);
-    while (n == -1 && errno == EINTR);
+    } while (n == -1 && errno == EINTR);
   } else {
-    do
+    do {
+      if (uv__work_check_cancelled(w))
+        break;
       n = uv__writev(uv__stream_fd(stream), iov, iovcnt);
-    while (n == -1 && errno == EINTR);
+    } while (n == -1 && errno == EINTR);
   }
 
   if (n >= 0)
@@ -836,6 +852,71 @@ static int uv__try_write(uv_stream_t* stream,
   return UV__ERR(errno);
 }
 
+
+/* A note about blocking writes: The UV_HANDLE_WRITE_PENDING flag is toggled
+ * only on the loop thread (either by uv__write or uv__write_done).  While we're
+ * doing work from the thread pool, we touch only the result field of
+ * uv_write_t; we'll never read it from the loop thread while a blocked write is
+ * pending.
+ */
+static void uv__write_work(struct uv__work* w) {
+  uv_stream_t* stream;
+  struct uv__queue* q;
+  uv_write_t *req;
+
+  stream = container_of(w, uv_stream_t, blocked_write);
+
+  assert(!uv__queue_empty(&stream->write_queue));
+  q = uv__queue_head(&stream->write_queue);
+  req = uv__queue_data(q, uv_write_t, queue);
+  assert(req->handle == stream);
+
+  req->result = uv__try_write(stream,
+                              &(req->bufs[req->write_index]),
+                              req->nbufs - req->write_index,
+                              req->send_handle,
+                              w);
+}
+
+
+static void uv__write_done(struct uv__work* w, int status) {
+  uv_stream_t* stream;
+  struct uv__queue* q;
+  uv_write_t *req;
+
+  stream = container_of(w, uv_stream_t, blocked_write);
+  stream->flags &= ~UV_HANDLE_WRITE_PENDING;
+
+  assert(!uv__queue_empty(&stream->write_queue));
+  q = uv__queue_head(&stream->write_queue);
+  req = uv__queue_data(q, uv_write_t, queue);
+
+  /* This happens when we're cancelled in the thread pool work queue. */
+  if (status != 0) {
+    req->result = status;
+    goto error;
+  }
+
+  if (req->result >= 0) {
+    if (uv__write_req_update(stream, req, req->result))
+      uv__write_req_finish(req);
+  } else if (req->result != UV_EAGAIN)
+    goto error;
+
+  if (!uv__queue_empty(&stream->write_queue)) {
+    uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
+    uv__stream_osx_interrupt_select(stream);
+  }
+
+  return;
+
+error:
+  uv__write_req_finish(req);
+  uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
+  uv__stream_osx_interrupt_select(stream);
+}
+
+
 static void uv__write(uv_stream_t* stream) {
   struct uv__queue* q;
   uv_write_t* req;
@@ -858,10 +939,23 @@ static void uv__write(uv_stream_t* stream) {
     req = uv__queue_data(q, uv_write_t, queue);
     assert(req->handle == stream);
 
-    n = uv__try_write(stream,
+    if (!(stream->flags & UV_HANDLE_BLOCKING_WRITES)) {
-                      &(req->bufs[req->write_index]),
+      n = uv__try_write(stream,
-                      req->nbufs - req->write_index,
+                        &(req->bufs[req->write_index]),
-                      req->send_handle);
+                        req->nbufs - req->write_index,
+                        req->send_handle,
+                        NULL);
+    } else {
+      n = UV_EAGAIN;
+      if (!(stream->flags & UV_HANDLE_WRITE_PENDING)) {
+        stream->flags |= UV_HANDLE_WRITE_PENDING;
+        uv__work_submit(stream->loop,
+                        &stream->blocked_write,
+                        UV__WORK_FAST_IO_CANCELLABLE,
+                        uv__write_work,
+                        uv__write_done);
+      }
+    }
 
     /* Ensure the handle isn't sent again in case this is a partial write. */
     if (n >= 0) {
@@ -876,10 +970,6 @@ static void uv__write(uv_stream_t* stream) {
     } else if (n != UV_EAGAIN)
       goto error;
 
-    /* If this is a blocking stream, try again. */
-    if (stream->flags & UV_HANDLE_BLOCKING_WRITES)
-      continue;
-
     /* We're not done. */
     uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
 
@@ -890,7 +980,7 @@ static void uv__write(uv_stream_t* stream) {
   }
 
 error:
-  req->error = n;
+  req->result = n;
   uv__write_req_finish(req);
   uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
   uv__stream_osx_interrupt_select(stream);
@@ -923,7 +1013,7 @@ static void uv__write_callbacks(uv_stream_t* stream) {
 
     /* NOTE: call callback AFTER freeing the request data. */
     if (req->cb)
-      req->cb(req, req->error);
+      req->cb(req, req->result);
   }
 }
 
@@ -1363,7 +1453,7 @@ int uv_write2(uv_write_t* req,
   uv__req_init(stream->loop, req, UV_WRITE);
   req->cb = cb;
   req->handle = stream;
-  req->error = 0;
+  req->result = 0;
   req->send_handle = send_handle;
   uv__queue_init(&req->queue);
 
@@ -1393,12 +1483,6 @@ int uv_write2(uv_write_t* req,
     uv__write(stream);
   }
   else {
-    /*
-     * blocking streams should never have anything in the queue.
-     * if this assert fires then somehow the blocking stream isn't being
-     * sufficiently flushed in uv__write.
-     */
-    assert(!(stream->flags & UV_HANDLE_BLOCKING_WRITES));
     uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
     uv__stream_osx_interrupt_select(stream);
   }
@@ -1440,7 +1524,7 @@ int uv_try_write2(uv_stream_t* stream,
   if (err < 0)
     return err;
 
-  return uv__try_write(stream, bufs, nbufs, send_handle);
+  return uv__try_write(stream, bufs, nbufs, send_handle, NULL);
 }
 
 
@@ -1542,13 +1626,6 @@ void uv__stream_close(uv_stream_t* handle) {
   uv__handle_stop(handle);
   handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
 
-  if (handle->io_watcher.fd != -1) {
-    /* Don't close stdio file descriptors.  Nothing good comes from it. */
-    if (handle->io_watcher.fd > STDERR_FILENO)
-      uv__close(handle->io_watcher.fd);
-    handle->io_watcher.fd = -1;
-  }
-
   if (handle->accepted_fd != -1) {
     uv__close(handle->accepted_fd);
     handle->accepted_fd = -1;
@@ -1563,6 +1640,9 @@ void uv__stream_close(uv_stream_t* handle) {
     handle->queued_fds = NULL;
   }
 
+  if (handle->flags & UV_HANDLE_WRITE_PENDING)
+    uv__work_cancel(handle->loop, &handle->blocked_write);
+
   assert(!uv__io_active(&handle->io_watcher, POLLIN | POLLOUT));
 }
 

src/uv-common.h

@@ -100,6 +100,7 @@ enum {
   UV_HANDLE_READ_PENDING                = 0x00010000,
   UV_HANDLE_SYNC_BYPASS_IOCP            = 0x00020000,
   UV_HANDLE_ZERO_READ                   = 0x00040000,
+  UV_HANDLE_WRITE_PENDING               = 0x00080000, /* UNIX only */
   UV_HANDLE_EMULATE_IOCP                = 0x00080000,
   UV_HANDLE_BLOCKING_WRITES             = 0x00100000,
   UV_HANDLE_CANCELLATION_PENDING        = 0x00200000,
@@ -213,15 +214,20 @@ int uv__getaddrinfo_translate_error(int sys_err);    /* EAI_* error. */
 enum uv__work_kind {
   UV__WORK_CPU,
   UV__WORK_FAST_IO,
+  UV__WORK_FAST_IO_CANCELLABLE,
   UV__WORK_SLOW_IO
 };
 
+void uv__cancel_signal_handler(int signo);
+
 void uv__work_submit(uv_loop_t* loop,
                      struct uv__work *w,
                      enum uv__work_kind kind,
                      void (*work)(struct uv__work *w),
                      void (*done)(struct uv__work *w, int status));
 
+int uv__work_cancel(uv_loop_t* loop, struct uv__work* w);
+
 void uv__work_done(uv_async_t* handle);
 
 size_t uv__count_bufs(const uv_buf_t bufs[], unsigned int nbufs);

test/test-list.h

@@ -129,6 +129,8 @@ TEST_DECLARE   (tcp_bind_invalid_flags)
 TEST_DECLARE   (tcp_bind_writable_flags)
 TEST_DECLARE   (tcp_bind_or_listen_error_after_close)
 TEST_DECLARE   (tcp_listen_without_bind)
+TEST_DECLARE   (pipe_blocking_cancel)
+TEST_DECLARE   (pipe_blocking_subprocess)
 TEST_DECLARE   (tcp_connect_error_fault)
 TEST_DECLARE   (tcp_connect6_error_fault)
 TEST_DECLARE   (tcp_connect6_link_local)
@@ -756,6 +758,8 @@ TASK_LIST_START
   TEST_ENTRY  (tcp_bind_writable_flags)
   TEST_ENTRY  (tcp_bind_or_listen_error_after_close)
   TEST_ENTRY  (tcp_listen_without_bind)
+  TEST_ENTRY_CUSTOM (pipe_blocking_cancel, 0, 0, 20000)
+  TEST_ENTRY_CUSTOM (pipe_blocking_subprocess, 0, 0, 20000)
   TEST_ENTRY  (tcp_connect_error_fault)
   TEST_ENTRY  (tcp_connect6_error_fault)
   TEST_ENTRY  (tcp_connect6_link_local)

test/test-pipe-blocking-subprocess.c

@@ -0,0 +1,188 @@
+/* Copyright libuv contributors. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to
+ * deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+ * sell copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+ * IN THE SOFTWARE.
+ */
+
+#include "uv.h"
+#include "task.h"
+
+#define FILL_PIPE_NUM 0x40000
+
+static uv_loop_t* loop;
+static char exepath[1024];
+static size_t exepath_size = sizeof exepath;
+static char* args[3];
+static uv_process_options_t options;
+static uv_process_t process;
+static uv_stdio_container_t stdios[3];
+static uv_pipe_t pipe_in;
+static uv_pipe_t pipe_out;
+static uv_file fds[2];
+static size_t total_read;
+static int write_complete;
+static uv_write_t req;
+static uv_buf_t buf;
+static uv_timer_t timer;
+static int closed_streams;
+
+static void close_cb(uv_handle_t *handle) {
+  ++closed_streams;
+}
+
+static void write_cb_ok(uv_write_t* req, int status) {
+  ASSERT_OK(status);
+  ++write_complete;
+
+  if (write_complete == 1) {
+    uv_close((uv_handle_t*)&pipe_in, close_cb);
+    uv_close((uv_handle_t*)&pipe_out, close_cb);
+  }
+}
+
+static void read_cb(uv_stream_t* stream,
+                    ssize_t nread,
+                    const uv_buf_t* buf) {
+  ASSERT_GE(nread, 0);
+  total_read += nread;
+  free(buf->base);
+  if (total_read == 12 + FILL_PIPE_NUM)
+    uv_read_stop(stream);
+}
+
+static void alloc_cb(uv_handle_t* handle, size_t size, uv_buf_t* buf) {
+  buf->base = malloc(size);
+  buf->len = size;
+}
+
+static void init_common(void) {
+  loop = uv_default_loop();
+#ifndef _WIN32
+  uv_loop_configure(loop, UV_LOOP_CANCEL_SIGNAL, SIGUSR1);
+#endif
+
+  ASSERT_OK(uv_pipe_init(loop, &pipe_in, 0));
+  ASSERT_OK(uv_pipe_init(loop, &pipe_out, 0));
+  ASSERT_OK(uv_pipe(fds, 0, 0));
+  ASSERT_OK(uv_pipe_open(&pipe_out, fds[0]));
+  ASSERT_OK(uv_pipe_open(&pipe_in, fds[1]));
+
+  ASSERT_OK(uv_exepath(exepath, &exepath_size));
+  exepath[exepath_size] = '\0';
+  args[0] = exepath;
+  args[1] = "spawn_helper2";
+  args[2] = NULL;
+  options.file = exepath;
+  options.args = args;
+  options.flags = 0;
+  options.stdio_count = ARRAY_SIZE(stdios);
+  options.stdio = stdios;
+  stdios[0].flags = UV_IGNORE;
+  stdios[1].flags = UV_INHERIT_STREAM;
+  stdios[1].data.stream = (uv_stream_t*)&pipe_in;
+  stdios[2].flags = UV_IGNORE;
+}
+
+/* After the subprocess exits, fill the pipe buffer. */
+static void exit_cb_write(uv_process_t* process,
+                          int64_t exit_status,
+                          int term_signal) {
+  ASSERT_EQ(1, exit_status);
+  ASSERT_OK(term_signal);
+  uv_close((uv_handle_t*) process, NULL);
+  ASSERT_OK(uv_write(&req, (uv_stream_t*)&pipe_in, &buf, 1, write_cb_ok));
+}
+
+TEST_IMPL(pipe_blocking_subprocess) {
+#ifdef _WIN32
+  RETURN_SKIP("Unix only test");
+#endif
+
+  init_common();
+  options.exit_cb = exit_cb_write;
+
+  ASSERT_OK(uv_read_start((uv_stream_t*)&pipe_out, alloc_cb, read_cb));
+
+  /* Write enough that the pipe buffer fills. */
+  buf.len = FILL_PIPE_NUM;
+  buf.base = malloc(buf.len);
+  memset(buf.base, 'A', buf.len);
+
+  /* The subprocess forces fds[0] into blocking mode and writes 12 bytes. */
+  ASSERT_OK(uv_spawn(loop, &process, &options));
+
+  ASSERT_OK(uv_run(loop, UV_RUN_DEFAULT));
+  ASSERT_EQ(write_complete, 1);
+  ASSERT_EQ(total_read, buf.len + 12);
+  ASSERT_EQ(closed_streams, 2);
+
+  free(buf.base);
+
+  return 0;
+}
+
+static void timer_cb(uv_timer_t *handle) {
+  /* The write has begun by now.  We'll close the write side of the pipe first,
+   * since closing the read side will trigger SIGPIPE. */
+  uv_close((uv_handle_t*) &pipe_in, close_cb);
+  uv_close((uv_handle_t*) &timer, NULL);
+}
+
+static void write_cb_cancel(uv_write_t* req, int status) {
+  ASSERT(status == UV_ECANCELED || status == UV_EPIPE);
+  ++write_complete;
+  uv_close((uv_handle_t*) &pipe_out, close_cb);
+}
+
+/* After the subprocess exits, fill the pipe buffer and to cancel the write. */
+static void exit_cb_cancel(uv_process_t* process,
+                           int64_t exit_status,
+                           int term_signal) {
+  ASSERT_EQ(1, exit_status);
+  ASSERT_OK(term_signal);
+  uv_close((uv_handle_t*) process, NULL);
+  ASSERT_OK(uv_write(&req, (uv_stream_t*)&pipe_in, &buf, 1, write_cb_cancel));
+  ASSERT_OK(uv_timer_start(&timer, timer_cb, 100, 0));
+}
+
+TEST_IMPL(pipe_blocking_cancel) {
+#ifdef _WIN32
+  RETURN_SKIP("Unix only test");
+#endif
+
+  init_common();
+  options.exit_cb = exit_cb_cancel;
+
+  uv_timer_init(loop, &timer);
+
+  /* Write enough that the pipe buffer fills. */
+  buf.len = FILL_PIPE_NUM;
+  buf.base = malloc(buf.len);
+  memset(buf.base, 'A', buf.len);
+
+  /* The subprocess forces fds[0] into blocking mode and writes 12 bytes. */
+  ASSERT_OK(uv_spawn(loop, &process, &options));
+
+  ASSERT_OK(uv_run(loop, UV_RUN_DEFAULT));
+  ASSERT_EQ(write_complete, 1);
+  ASSERT_EQ(closed_streams, 2);
+
+  free(buf.base);
+
+  return 0;
+}