unix: fix event loop stall

Fix a rather obscure bug where the event loop stalls when an I/O watcher is
stopped while an artificial event, generated with uv__io_feed(), is pending.

src/unix/core.c

@@ -641,9 +641,6 @@ void uv__io_stop(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
   w->pevents &= ~events;
 
   if (w->pevents == 0) {
-    ngx_queue_remove(&w->pending_queue);
-    ngx_queue_init(&w->pending_queue);
-
     ngx_queue_remove(&w->watcher_queue);
     ngx_queue_init(&w->watcher_queue);
 
@@ -660,6 +657,12 @@ void uv__io_stop(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
 }
 
 
+void uv__io_close(uv_loop_t* loop, uv__io_t* w) {
+  uv__io_stop(loop, w, UV__POLLIN | UV__POLLOUT);
+  ngx_queue_remove(&w->pending_queue);
+}
+
+
 void uv__io_feed(uv_loop_t* loop, uv__io_t* w) {
   if (ngx_queue_empty(&w->pending_queue))
     ngx_queue_insert_tail(&loop->pending_queue, &w->pending_queue);

src/unix/internal.h

@@ -130,6 +130,7 @@ void uv__make_close_pending(uv_handle_t* handle);
 void uv__io_init(uv__io_t* w, uv__io_cb cb, int fd);
 void uv__io_start(uv_loop_t* loop, uv__io_t* w, unsigned int events);
 void uv__io_stop(uv_loop_t* loop, uv__io_t* w, unsigned int events);
+void uv__io_close(uv_loop_t* loop, uv__io_t* w);
 void uv__io_feed(uv_loop_t* loop, uv__io_t* w);
 int uv__io_active(const uv__io_t* w, unsigned int events);
 void uv__io_poll(uv_loop_t* loop, int timeout); /* in milliseconds or -1 */

src/unix/stream.c

@@ -1335,7 +1335,7 @@ void uv__stream_close(uv_stream_t* handle) {
 #endif /* defined(__APPLE__) */
 
   uv_read_stop(handle);
-  uv__io_stop(handle->loop, &handle->io_watcher, UV__POLLOUT);
+  uv__io_close(handle->loop, &handle->io_watcher);
 
   close(handle->io_watcher.fd);
   handle->io_watcher.fd = -1;

src/unix/udp.c

@@ -40,7 +40,7 @@ static int uv__udp_send(uv_udp_send_t* req, uv_udp_t* handle, uv_buf_t bufs[],
 
 
 void uv__udp_close(uv_udp_t* handle) {
-  uv__io_stop(handle->loop, &handle->io_watcher, UV__POLLIN | UV__POLLOUT);
+  uv__io_close(handle->loop, &handle->io_watcher);
   uv__handle_stop(handle);
   close(handle->io_watcher.fd);
   handle->io_watcher.fd = -1;

test/test-list.h

@@ -67,6 +67,7 @@ TEST_DECLARE   (tcp_flags)
 TEST_DECLARE   (tcp_write_error)
 TEST_DECLARE   (tcp_write_to_half_open_connection)
 TEST_DECLARE   (tcp_unexpected_read)
+TEST_DECLARE   (tcp_read_stop)
 TEST_DECLARE   (tcp_bind6_error_addrinuse)
 TEST_DECLARE   (tcp_bind6_error_addrnotavail)
 TEST_DECLARE   (tcp_bind6_error_fault)
@@ -289,6 +290,9 @@ TASK_LIST_START
   TEST_ENTRY  (tcp_write_to_half_open_connection)
   TEST_ENTRY  (tcp_unexpected_read)
 
+  TEST_ENTRY  (tcp_read_stop)
+  TEST_HELPER (tcp_read_stop, tcp4_echo_server)
+
   TEST_ENTRY  (tcp_bind6_error_addrinuse)
   TEST_ENTRY  (tcp_bind6_error_addrnotavail)
   TEST_ENTRY  (tcp_bind6_error_fault)

test/test-tcp-read-stop.c

@@ -0,0 +1,73 @@
+/* Copyright Joyent, Inc. and other Node 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"
+
+static uv_timer_t timer_handle;
+static uv_tcp_t tcp_handle;
+static uv_write_t write_req;
+
+
+static void fail_cb(void) {
+  ASSERT(0 && "fail_cb called");
+}
+
+
+static void write_cb(uv_write_t* req, int status) {
+  uv_close((uv_handle_t*) &timer_handle, NULL);
+  uv_close((uv_handle_t*) &tcp_handle, NULL);
+}
+
+
+static void timer_cb(uv_timer_t* handle, int status) {
+  uv_buf_t buf = uv_buf_init("PING", 4);
+  ASSERT(0 == uv_write(&write_req,
+                       (uv_stream_t*) &tcp_handle,
+                       &buf,
+                       1,
+                       write_cb));
+  ASSERT(0 == uv_read_stop((uv_stream_t*) &tcp_handle));
+}
+
+
+static void connect_cb(uv_connect_t* req, int status) {
+  ASSERT(0 == status);
+  ASSERT(0 == uv_timer_start(&timer_handle, timer_cb, 50, 0));
+  ASSERT(0 == uv_read_start((uv_stream_t*) &tcp_handle,
+                            (uv_alloc_cb) fail_cb,
+                            (uv_read_cb) fail_cb));
+}
+
+
+TEST_IMPL(tcp_read_stop) {
+  uv_connect_t connect_req;
+  struct sockaddr_in addr;
+
+  addr = uv_ip4_addr("127.0.0.1", TEST_PORT);
+  ASSERT(0 == uv_timer_init(uv_default_loop(), &timer_handle));
+  ASSERT(0 == uv_tcp_init(uv_default_loop(), &tcp_handle));
+  ASSERT(0 == uv_tcp_connect(&connect_req, &tcp_handle, addr, connect_cb));
+  ASSERT(0 == uv_run(uv_default_loop()));
+  MAKE_VALGRIND_HAPPY();
+
+  return 0;
+}

uv.gyp

@@ -299,6 +299,7 @@
         'test/test-tcp-write-to-half-open-connection.c',
         'test/test-tcp-writealot.c',
         'test/test-tcp-unexpected-read.c',
+        'test/test-tcp-read-stop.c',
         'test/test-threadpool.c',
         'test/test-threadpool-cancel.c',
         'test/test-mutexes.c',