Re-merge "unix,stream: clear read/write states on close/eof"

This reverts commit 46f36e3df1a666620f6749427f15651cbc4b7001.

PR-URL: https://github.com/libuv/libuv/pull/3006
Refs: https://github.com/libuv/libuv/pull/2967
Refs: https://github.com/libuv/libuv/pull/2409
Refs: https://github.com/libuv/libuv/issues/2943
Refs: https://github.com/libuv/libuv/pull/2968
Refs: https://github.com/nodejs/node/pull/36111
Reviewed-By: Santiago Gimeno <santiago.gimeno@gmail.com>

CMakeLists.txt

@@ -485,6 +485,9 @@ if(LIBUV_BUILD_TESTS)
        test/test-metrics.c
        test/test-multiple-listen.c
        test/test-mutexes.c
+       test/test-not-readable-nor-writable-on-read-error.c
+       test/test-not-readable-on-eof.c
+       test/test-not-writable-after-shutdown.c
        test/test-osx-select.c
        test/test-pass-always.c
        test/test-ping-pong.c

Makefile.am

@@ -206,6 +206,9 @@ test_run_tests_SOURCES = test/blackhole-server.c \
                          test/test-metrics.c \
                          test/test-multiple-listen.c \
                          test/test-mutexes.c \
+                         test/test-not-readable-nor-writable-on-read-error.c \
+                         test/test-not-readable-on-eof.c \
+                         test/test-not-writable-after-shutdown.c \
                          test/test-osx-select.c \
                          test/test-pass-always.c \
                          test/test-ping-pong.c \

src/unix/stream.c

@@ -1011,6 +1011,7 @@ uv_handle_type uv__handle_type(int fd) {
 static void uv__stream_eof(uv_stream_t* stream, const uv_buf_t* buf) {
   stream->flags |= UV_HANDLE_READ_EOF;
   stream->flags &= ~UV_HANDLE_READING;
+  stream->flags &= ~UV_HANDLE_READABLE;
   uv__io_stop(stream->loop, &stream->io_watcher, POLLIN);
   if (!uv__io_active(&stream->io_watcher, POLLOUT))
     uv__handle_stop(stream);
@@ -1198,6 +1199,7 @@ static void uv__read(uv_stream_t* stream) {
 #endif
       } else {
         /* Error. User should call uv_close(). */
+        stream->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
         stream->read_cb(stream, UV__ERR(errno), &buf);
         if (stream->flags & UV_HANDLE_READING) {
           stream->flags &= ~UV_HANDLE_READING;
@@ -1286,6 +1288,7 @@ int uv_shutdown(uv_shutdown_t* req, uv_stream_t* stream, uv_shutdown_cb cb) {
   req->cb = cb;
   stream->shutdown_req = req;
   stream->flags |= UV_HANDLE_SHUTTING;
+  stream->flags &= ~UV_HANDLE_WRITABLE;
 
   uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
   uv__stream_osx_interrupt_select(stream);

src/win/tcp.c

@@ -1024,6 +1024,7 @@ void uv_process_tcp_read_req(uv_loop_t* loop, uv_tcp_t* handle,
          */
         err = WSAECONNRESET;
       }
+      handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
 
       handle->read_cb((uv_stream_t*)handle,
                       uv_translate_sys_error(err),
@@ -1105,6 +1106,7 @@ void uv_process_tcp_read_req(uv_loop_t* loop, uv_tcp_t* handle,
              * Unix. */
             err = WSAECONNRESET;
           }
+          handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
 
           handle->read_cb((uv_stream_t*)handle,
                           uv_translate_sys_error(err),

test/echo-server.c

@@ -69,7 +69,6 @@ static void after_shutdown(uv_shutdown_t* req, int status) {
   free(req);
 }
 
-
 static void after_read(uv_stream_t* handle,
                        ssize_t nread,
                        const uv_buf_t* buf) {
@@ -96,13 +95,20 @@ static void after_read(uv_stream_t* handle,
   /*
    * Scan for the letter Q which signals that we should quit the server.
    * If we get QS it means close the stream.
+   * If we get QSH it means disable linger before close the socket.
    */
   if (!server_closed) {
     for (i = 0; i < nread; i++) {
       if (buf->base[i] == 'Q') {
         if (i + 1 < nread && buf->base[i + 1] == 'S') {
+          int reset = 0;
+          if (i + 2 < nread && buf->base[i + 2] == 'H')
+            reset = 1;
           free(buf->base);
-          uv_close((uv_handle_t*)handle, on_close);
+          if (reset && handle->type == UV_TCP)
+            ASSERT(0 == uv_tcp_close_reset((uv_tcp_t*) handle, on_close));
+          else
+            uv_close((uv_handle_t*) handle, on_close);
           return;
         } else {
           uv_close(server, on_server_close);

test/test-list.h

@@ -504,6 +504,10 @@ TEST_DECLARE   (handle_type_name)
 TEST_DECLARE   (req_type_name)
 TEST_DECLARE   (getters_setters)
 
+TEST_DECLARE   (not_writable_after_shutdown)
+TEST_DECLARE   (not_readable_nor_writable_on_read_error)
+TEST_DECLARE   (not_readable_on_eof)
+
 #ifndef _WIN32
 TEST_DECLARE  (fork_timer)
 TEST_DECLARE  (fork_socketpair)
@@ -1127,6 +1131,13 @@ TASK_LIST_START
   TEST_ENTRY  (idna_toascii)
 #endif
 
+  TEST_ENTRY    (not_writable_after_shutdown)
+  TEST_HELPER   (not_writable_after_shutdown, tcp4_echo_server)
+  TEST_ENTRY    (not_readable_nor_writable_on_read_error)
+  TEST_HELPER   (not_readable_nor_writable_on_read_error, tcp4_echo_server)
+  TEST_ENTRY    (not_readable_on_eof)
+  TEST_HELPER   (not_readable_on_eof, tcp4_echo_server)
+
   TEST_ENTRY  (metrics_idle_time)
   TEST_ENTRY  (metrics_idle_time_thread)
   TEST_ENTRY  (metrics_idle_time_zero)

test/test-not-readable-nor-writable-on-read-error.c

@@ -0,0 +1,104 @@
+/* Copyright the libuv project 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_loop_t loop;
+static uv_tcp_t tcp_client;
+static uv_connect_t connect_req;
+static uv_write_t write_req;
+static char reset_me_cmd[] = {'Q', 'S', 'H'};
+
+static int connect_cb_called;
+static int read_cb_called;
+static int write_cb_called;
+static int close_cb_called;
+
+static void write_cb(uv_write_t* req, int status) {
+  write_cb_called++;
+  ASSERT(status == 0);
+}
+
+static void alloc_cb(uv_handle_t* handle,
+                     size_t suggested_size,
+                     uv_buf_t* buf) {
+  static char slab[64];
+  buf->base = slab;
+  buf->len = sizeof(slab);
+}
+
+static void close_cb(uv_handle_t* handle) {
+  close_cb_called++;
+}
+
+static void read_cb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
+  read_cb_called++;
+
+  ASSERT((nread < 0) && (nread != UV_EOF));
+  ASSERT(0 == uv_is_writable(handle));
+  ASSERT(0 == uv_is_readable(handle));
+
+  uv_close((uv_handle_t*) handle, close_cb);
+}
+
+static void connect_cb(uv_connect_t* req, int status) {
+  int r;
+  uv_buf_t reset_me;
+
+  connect_cb_called++;
+  ASSERT(status == 0);
+
+  r = uv_read_start((uv_stream_t*) &tcp_client, alloc_cb, read_cb);
+  ASSERT(r == 0);
+
+  reset_me = uv_buf_init(reset_me_cmd, sizeof(reset_me_cmd));
+
+  r = uv_write(&write_req,
+               (uv_stream_t*) &tcp_client,
+               &reset_me,
+               1,
+               write_cb);
+
+  ASSERT(r == 0);
+}
+
+TEST_IMPL(not_readable_nor_writable_on_read_error) {
+  struct sockaddr_in sa;
+  ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &sa));
+  ASSERT(0 == uv_loop_init(&loop));
+  ASSERT(0 == uv_tcp_init(&loop, &tcp_client));
+
+  ASSERT(0 == uv_tcp_connect(&connect_req,
+                             &tcp_client,
+                             (const struct sockaddr*) &sa,
+                             connect_cb));
+
+  ASSERT(0 == uv_run(&loop, UV_RUN_DEFAULT));
+
+  ASSERT(connect_cb_called == 1);
+  ASSERT(read_cb_called == 1);
+  ASSERT(write_cb_called == 1);
+  ASSERT(close_cb_called == 1);
+
+  MAKE_VALGRIND_HAPPY();
+  return 0;
+}

test/test-not-readable-on-eof.c

@@ -0,0 +1,103 @@
+/* Copyright the libuv project 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_loop_t loop;
+static uv_tcp_t tcp_client;
+static uv_connect_t connect_req;
+static uv_write_t write_req;
+static char close_me_cmd[] = {'Q', 'S'};
+
+static int connect_cb_called;
+static int read_cb_called;
+static int write_cb_called;
+static int close_cb_called;
+
+static void write_cb(uv_write_t* req, int status) {
+  write_cb_called++;
+  ASSERT(status == 0);
+}
+
+static void alloc_cb(uv_handle_t* handle,
+                     size_t suggested_size,
+                     uv_buf_t* buf) {
+  static char slab[64];
+  buf->base = slab;
+  buf->len = sizeof(slab);
+}
+
+static void close_cb(uv_handle_t* handle) {
+  close_cb_called++;
+}
+
+static void read_cb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
+  read_cb_called++;
+
+  ASSERT(nread == UV_EOF);
+  ASSERT(0 == uv_is_readable(handle));
+
+  uv_close((uv_handle_t*) handle, close_cb);
+}
+
+static void connect_cb(uv_connect_t* req, int status) {
+  int r;
+  uv_buf_t close_me;
+
+  connect_cb_called++;
+  ASSERT(status == 0);
+
+  r = uv_read_start((uv_stream_t*) &tcp_client, alloc_cb, read_cb);
+  ASSERT(r == 0);
+
+  close_me = uv_buf_init(close_me_cmd, sizeof(close_me_cmd));
+
+  r = uv_write(&write_req,
+               (uv_stream_t*) &tcp_client,
+               &close_me,
+               1,
+               write_cb);
+
+  ASSERT(r == 0);
+}
+
+TEST_IMPL(not_readable_on_eof) {
+  struct sockaddr_in sa;
+  ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &sa));
+  ASSERT(0 == uv_loop_init(&loop));
+  ASSERT(0 == uv_tcp_init(&loop, &tcp_client));
+
+  ASSERT(0 == uv_tcp_connect(&connect_req,
+                             &tcp_client,
+                             (const struct sockaddr*) &sa,
+                             connect_cb));
+
+  ASSERT(0 == uv_run(&loop, UV_RUN_DEFAULT));
+
+  ASSERT(connect_cb_called == 1);
+  ASSERT(read_cb_called == 1);
+  ASSERT(write_cb_called == 1);
+  ASSERT(close_cb_called == 1);
+
+  MAKE_VALGRIND_HAPPY();
+  return 0;
+}

test/test-not-writable-after-shutdown.c

@@ -0,0 +1,69 @@
+/* Copyright the libuv project 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_shutdown_t shutdown_req;
+
+static void close_cb(uv_handle_t* handle) {
+
+}
+
+static void shutdown_cb(uv_shutdown_t* req, int status) {
+  uv_close((uv_handle_t*) req->handle, close_cb);
+}
+
+static void connect_cb(uv_connect_t* req, int status) {
+  int r;
+  ASSERT(status == 0);
+
+  r = uv_shutdown(&shutdown_req, req->handle, shutdown_cb);
+  ASSERT(r == 0);
+
+  ASSERT(0 == uv_is_writable(req->handle));
+}
+
+TEST_IMPL(not_writable_after_shutdown) {
+  int r;
+  struct sockaddr_in addr;
+  uv_loop_t* loop;
+  uv_tcp_t socket;
+  uv_connect_t connect_req;
+
+  ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
+  loop = uv_default_loop();
+
+  r = uv_tcp_init(loop, &socket);
+  ASSERT(r == 0);
+
+  r = uv_tcp_connect(&connect_req,
+                     &socket,
+                     (const struct sockaddr*) &addr,
+                     connect_cb);
+  ASSERT(r == 0);
+
+  r = uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+  ASSERT(r == 0);
+
+  MAKE_VALGRIND_HAPPY();
+  return 0;
+}