stream: squelch ECONNRESET error if already closed

Add new UV__POLLRDHUP event to be emitted when EPOLLRDHUP(in Linux) or
EV_EOF(in BSD / OSX) is detected and only if UV_READABLE is set.

When a read returns ECONNRESET after a UV__POLLRDHUP event, emit EOF instead
of the error.

Add tcp-squelch-connreset test. Not to be run on Windows as it returns
ECONNRESET error.

Fixes in test-poll and test-tcp-open so they pass after these changes.

PR-URL: https://github.com/libuv/libuv/pull/403
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>
Reviewed-By: Saúl Ibarra Corretgé <saghul@gmail.com>

Makefile.am

@@ -224,6 +224,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \
                          test/test-tcp-open.c \
                          test/test-tcp-read-stop.c \
                          test/test-tcp-shutdown-after-write.c \
+                         test/test-tcp-squelch-connreset.c \
                          test/test-tcp-unexpected-read.c \
                          test/test-tcp-oob.c \
                          test/test-tcp-write-to-half-open-connection.c \

src/unix/internal.h

@@ -89,10 +89,11 @@
 #endif
 
 #if defined(__linux__)
-# define UV__POLLIN   UV__EPOLLIN
+# define UV__POLLIN     UV__EPOLLIN
-# define UV__POLLOUT  UV__EPOLLOUT
+# define UV__POLLOUT    UV__EPOLLOUT
-# define UV__POLLERR  UV__EPOLLERR
+# define UV__POLLERR    UV__EPOLLERR
-# define UV__POLLHUP  UV__EPOLLHUP
+# define UV__POLLHUP    UV__EPOLLHUP
+# define UV__POLLRDHUP  UV__EPOLLRDHUP
 #endif
 
 #if defined(__sun) || defined(_AIX)
@@ -118,6 +119,10 @@
 # define UV__POLLHUP  8
 #endif
 
+#ifndef UV__POLLRDHUP
+# define UV__POLLRDHUP  0x200
+#endif
+
 #if !defined(O_CLOEXEC) && defined(__FreeBSD__)
 /*
  * It may be that we are just missing `__POSIX_VISIBLE >= 200809`.
@@ -143,6 +148,7 @@ enum {
   UV_TCP_NODELAY          = 0x400,  /* Disable Nagle. */
   UV_TCP_KEEPALIVE        = 0x800,  /* Turn on keep-alive. */
   UV_TCP_SINGLE_ACCEPT    = 0x1000, /* Only accept() when idle. */
+  UV_STREAM_DISCONNECT    = 0x2000, /* Remote end is forcibly closed */
   UV_HANDLE_IPV6          = 0x10000, /* Handle is bound to a IPv6 socket. */
   UV_UDP_PROCESSING       = 0x20000  /* Handle is running the send callback queue. */
 };

src/unix/kqueue.c

@@ -241,6 +241,9 @@ void uv__io_poll(uv_loop_t* loop, int timeout) {
       if (ev->flags & EV_ERROR)
         revents |= UV__POLLERR;
 
+      if ((w->pevents & UV__POLLIN) && (ev->flags & EV_EOF))
+        revents |= UV__POLLRDHUP;
+
       if (revents == 0)
         continue;
 

src/unix/linux-core.c

@@ -184,6 +184,8 @@ void uv__io_poll(uv_loop_t* loop, int timeout) {
     assert(w->fd < (int) loop->nwatchers);
 
     e.events = w->pevents;
+    if (w->pevents & UV__POLLIN)
+      e.events |= UV__POLLRDHUP;
     e.data = w->fd;
 
     if (w->events == 0)
@@ -321,7 +323,7 @@ void uv__io_poll(uv_loop_t* loop, int timeout) {
        * the current watcher. Also, filters out events that users has not
        * requested us to watch.
        */
-      pe->events &= w->pevents | UV__POLLERR | UV__POLLHUP;
+      pe->events &= w->pevents | UV__POLLERR | UV__POLLHUP | UV__POLLRDHUP;
 
       /* Work around an epoll quirk where it sometimes reports just the
        * EPOLLERR or EPOLLHUP event.  In order to force the event loop to

src/unix/linux-syscalls.h

@@ -76,6 +76,7 @@
 #define UV__EPOLLOUT          4
 #define UV__EPOLLERR          8
 #define UV__EPOLLHUP          16
+#define UV__EPOLLRDHUP        0x2000
 #define UV__EPOLLONESHOT      0x40000000
 #define UV__EPOLLET           0x80000000
 

src/unix/poll.c

@@ -41,7 +41,7 @@ static void uv__poll_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
   }
 
   pevents = 0;
-  if (events & UV__POLLIN)
+  if (events & (UV__POLLIN | UV__POLLRDHUP))
     pevents |= UV_READABLE;
   if (events & UV__POLLOUT)
     pevents |= UV_WRITABLE;

src/unix/stream.c

@@ -1142,6 +1142,8 @@ static void uv__read(uv_stream_t* stream) {
           uv__stream_osx_interrupt_select(stream);
         }
         stream->read_cb(stream, 0, &buf);
+      } else if (errno == ECONNRESET && (stream->flags & UV_STREAM_DISCONNECT)) {
+        uv__stream_eof(stream, &buf);
       } else {
         /* Error. User should call uv_close(). */
         stream->read_cb(stream, -errno, &buf);
@@ -1230,8 +1232,11 @@ static void uv__stream_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
   assert(uv__stream_fd(stream) >= 0);
 
   /* Ignore POLLHUP here. Even it it's set, there may still be data to read. */
-  if (events & (UV__POLLIN | UV__POLLERR | UV__POLLHUP))
+  if (events & (UV__POLLIN | UV__POLLERR | UV__POLLHUP | UV__POLLRDHUP)) {
+    if (events & UV__POLLRDHUP)
+      stream->flags |= UV_STREAM_DISCONNECT;
     uv__read(stream);
+  }
 
   if (uv__stream_fd(stream) == -1)
     return;  /* read_cb closed stream. */

test/test-list.h

@@ -79,6 +79,9 @@ TEST_DECLARE   (tcp_bind_invalid_flags)
 TEST_DECLARE   (tcp_listen_without_bind)
 TEST_DECLARE   (tcp_connect_error_fault)
 TEST_DECLARE   (tcp_connect_timeout)
+#ifndef _WIN32
+TEST_DECLARE   (tcp_squelch_connreset)
+#endif
 TEST_DECLARE   (tcp_close_while_connecting)
 TEST_DECLARE   (tcp_close)
 TEST_DECLARE   (tcp_create_early)
@@ -420,6 +423,9 @@ TASK_LIST_START
   TEST_ENTRY  (tcp_listen_without_bind)
   TEST_ENTRY  (tcp_connect_error_fault)
   TEST_ENTRY  (tcp_connect_timeout)
+#ifndef _WIN32
+  TEST_ENTRY  (tcp_squelch_connreset)
+#endif
   TEST_ENTRY  (tcp_close_while_connecting)
   TEST_ENTRY  (tcp_close)
   TEST_ENTRY  (tcp_create_early)

test/test-poll.c

@@ -204,14 +204,15 @@ static void connection_poll_cb(uv_poll_t* handle, int status, int events) {
         /* Read a couple of bytes. */
         static char buffer[74];
         r = recv(context->sock, buffer, sizeof buffer, 0);
-        ASSERT(r >= 0);
 
         if (r > 0) {
           context->read += r;
-        } else {
+        } else if (r == 0) {
           /* Got FIN. */
           context->got_fin = 1;
           new_events &= ~UV_READABLE;
+        } else {
+          ASSERT(got_eagain());
         }
 
         break;
@@ -222,7 +223,6 @@ static void connection_poll_cb(uv_poll_t* handle, int status, int events) {
         /* Read until EAGAIN. */
         static char buffer[931];
         r = recv(context->sock, buffer, sizeof buffer, 0);
-        ASSERT(r >= 0);
 
         while (r > 0) {
           context->read += r;

test/test-tcp-open.c

@@ -38,6 +38,7 @@ static uv_connect_t connect_req;
 static uv_shutdown_t shutdown_req;
 static uv_write_t write_req;
 
+static int read_bytes = 0;
 
 static void startup(void) {
 #ifdef _WIN32
@@ -111,11 +112,15 @@ static void read_cb(uv_stream_t* tcp, ssize_t nread, const uv_buf_t* buf) {
   ASSERT(tcp != NULL);
 
   if (nread >= 0) {
-    ASSERT(nread == 4);
+    read_bytes += nread;
-    ASSERT(memcmp("PING", buf->base, nread) == 0);
+    if (nread > 0) {
+      ASSERT(nread == 4);
+      ASSERT(memcmp("PING", buf->base, nread) == 0);
+    }
   }
   else {
     ASSERT(nread == UV_EOF);
+    ASSERT(read_bytes == 4);
     printf("GOT EOF\n");
     uv_close((uv_handle_t*)tcp, close_cb);
   }

test/test-tcp-squelch-connreset.c

@@ -0,0 +1,119 @@
+/* Copyright (c) 2015, Santiago Gimeno <santiago.gimeno@gmail.com>
+ *
+ * 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"
+#include <stdio.h>
+#include <stdlib.h>
+
+
+static uv_tcp_t tcp_server;
+static uv_tcp_t tcp_client;
+static uv_tcp_t tcp_server_client;
+static uv_connect_t connect_req;
+static uv_write_t write_req;
+
+static void alloc_cb(uv_handle_t* handle,
+                     size_t size,
+                     uv_buf_t* buf) {
+  buf->base = malloc(size);
+  buf->len = size;
+}
+
+static void read_cb(uv_stream_t* tcp, ssize_t nread, const uv_buf_t* buf) {
+  free(buf->base);
+  ASSERT(nread == UV_EOF);
+}
+
+static void on_connect(uv_connect_t* req, int status) {
+  int r;
+  uv_buf_t outbuf;
+
+  ASSERT(req != NULL);
+  ASSERT(status == 0);
+
+  outbuf = uv_buf_init("ping", 4);
+  r = uv_write(&write_req, (uv_stream_t*) req->handle, &outbuf, 1, NULL);
+  ASSERT(r == 0);
+
+  r = uv_read_start((uv_stream_t*) req->handle, alloc_cb, read_cb);
+  ASSERT(r == 0);
+}
+
+static void on_connection(uv_stream_t* server, int status) {
+  int r;
+
+  ASSERT(status == 0);
+
+  r = uv_tcp_init(uv_default_loop(), &tcp_server_client);
+  ASSERT(r == 0);
+
+  r = uv_accept(server, (uv_stream_t*) &tcp_server_client);
+  ASSERT(r == 0);
+
+  uv_close((uv_handle_t*) &tcp_server_client, NULL);
+  uv_close((uv_handle_t*) &tcp_server, NULL);
+}
+
+static void start_server(void) {
+  struct sockaddr_in addr;
+  int r;
+
+  ASSERT(0 == uv_ip4_addr("0.0.0.0", TEST_PORT, &addr));
+
+  r = uv_tcp_init(uv_default_loop(), &tcp_server);
+  ASSERT(r == 0);
+
+  r = uv_tcp_bind(&tcp_server, (const struct sockaddr*) &addr, 0);
+  ASSERT(r == 0);
+
+  r = uv_listen((uv_stream_t*) &tcp_server, SOMAXCONN, on_connection);
+  ASSERT(r == 0);
+}
+
+static void start_client(void) {
+  struct sockaddr_in addr;
+  int r;
+
+  ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
+
+  r = uv_tcp_init(uv_default_loop(), &tcp_client);
+  ASSERT(r == 0);
+
+  r = uv_tcp_connect(&connect_req,
+                     &tcp_client,
+                     (const struct sockaddr*) &addr,
+                     on_connect);
+  ASSERT(r == 0);
+}
+
+
+TEST_IMPL(tcp_squelch_connreset) {
+
+  start_server();
+
+  start_client();
+
+  uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+
+  MAKE_VALGRIND_HAPPY();
+  return 0;
+}

uv.gyp

@@ -364,6 +364,7 @@
         'test/test-tcp-connect-timeout.c',
         'test/test-tcp-connect6-error.c',
         'test/test-tcp-open.c',
+        'test/test-tcp-squelch-connreset.c',
         'test/test-tcp-write-to-half-open-connection.c',
         'test/test-tcp-write-after-connect.c',
         'test/test-tcp-writealot.c',