darwin,unix,win: allow stopping read in TCP/UDP from alloc callback

Fixes: https://github.com/libuv/libuv/issues/2399
Signed-off-by: Juan José Arboleda <soyjuanarbol@gmail.com>

CMakeLists.txt

@@ -649,6 +649,7 @@ if(LIBUV_BUILD_TESTS)
        test/test-tcp-flags.c
        test/test-tcp-oob.c
        test/test-tcp-open.c
+       test/test-tcp-read-stop-from-alloc.c
        test/test-tcp-read-stop.c
        test/test-tcp-reuseport.c
        test/test-tcp-read-stop-start.c
@@ -679,6 +680,7 @@ if(LIBUV_BUILD_TESTS)
        test/test-tty-escape-sequence-processing.c
        test/test-tty.c
        test/test-udp-alloc-cb-fail.c
+       test/test-udp-alloc-recv-stop.c
        test/test-udp-bind.c
        test/test-udp-connect.c
        test/test-udp-connect6.c

Makefile.am

@@ -275,6 +275,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \
                          test/test-tcp-connect6-error.c \
                          test/test-tcp-flags.c \
                          test/test-tcp-open.c \
+                         test/test-tcp-read-stop-from-alloc.c \
                          test/test-tcp-read-stop.c \
                          test/test-tcp-reuseport.c \
                          test/test-tcp-read-stop-start.c \
@@ -306,6 +307,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \
                          test/test-tty-escape-sequence-processing.c \
                          test/test-tty.c \
                          test/test-udp-alloc-cb-fail.c \
+                         test/test-udp-alloc-recv-stop.c \
                          test/test-udp-bind.c \
                          test/test-udp-connect.c \
                          test/test-udp-connect6.c \

docs/src/stream.rst

@@ -156,6 +156,9 @@ API
     may be pending on the next input event on that TTY on Windows, and does not
     indicate failure.
 
+    .. versionchanged:: 1.51.0 If :c:func:`uv_read_stop` is called from a
+        `uv_alloc_cb`, all allocated resources must be freed manually.
+
 .. c:function:: int uv_write(uv_write_t* req, uv_stream_t* handle, const uv_buf_t bufs[], unsigned int nbufs, uv_write_cb cb)
 
     Write data to stream. Buffers are written in order. Example:

docs/src/udp.rst

@@ -486,6 +486,9 @@ API
 
     :returns: 0 on success, or an error code < 0 on failure.
 
+    .. versionchanged:: 1.51.0 If :c:func:`uv_udp_recv_stop` is called from a
+        `uv_alloc_cb`, all allocated resources must be freed manually.
+
 .. c:function:: size_t uv_udp_get_send_queue_size(const uv_udp_t* handle)
 
     Returns `handle->send_queue_size`.

src/unix/stream.c

@@ -1051,8 +1051,10 @@ static void uv__read(uv_stream_t* stream) {
     buf = uv_buf_init(NULL, 0);
     stream->alloc_cb((uv_handle_t*)stream, 64 * 1024, &buf);
     if (buf.base == NULL || buf.len == 0) {
-      /* User indicates it can't or won't handle the read. */
-      stream->read_cb(stream, UV_ENOBUFS, &buf);
+      /* maybe uv_read_stop() was called in alloc cb */
+      if (stream->read_cb != NULL)
+        /* User indicates it can't or won't handle the read. */
+        stream->read_cb(stream, UV_ENOBUFS, &buf);
       return;
     }
 

src/unix/udp.c

@@ -240,7 +240,9 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
     buf = uv_buf_init(NULL, 0);
     handle->alloc_cb((uv_handle_t*) handle, UV__UDP_DGRAM_MAXSIZE, &buf);
     if (buf.base == NULL || buf.len == 0) {
-      handle->recv_cb(handle, UV_ENOBUFS, &buf, NULL, 0);
+      /* uv_udp_recv_stop may have been called in alloc_cb */
+      if (handle->recv_cb != NULL)
+        handle->recv_cb(handle, UV_ENOBUFS, &buf, NULL, 0);
       return;
     }
     assert(buf.base != NULL);

src/win/tcp.c

@@ -1043,7 +1043,9 @@ void uv__process_tcp_read_req(uv_loop_t* loop, uv_tcp_t* handle,
       buf = uv_buf_init(NULL, 0);
       handle->alloc_cb((uv_handle_t*) handle, 65536, &buf);
       if (buf.base == NULL || buf.len == 0) {
-        handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf);
+        /* maybe uv_read_stop() was called in alloc cb */
+        if (handle->read_cb != NULL)
+          handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf);
         break;
       }
       assert(buf.base != NULL);

src/win/udp.c

@@ -458,7 +458,9 @@ void uv__process_udp_recv_req(uv_loop_t* loop, uv_udp_t* handle,
       buf = uv_buf_init(NULL, 0);
       handle->alloc_cb((uv_handle_t*) handle, UV__UDP_DGRAM_MAXSIZE, &buf);
       if (buf.base == NULL || buf.len == 0) {
-        handle->recv_cb(handle, UV_ENOBUFS, &buf, NULL, 0);
+        /* uv_udp_recv_stop may have been called in alloc_cb */
+        if (handle->recv_cb != NULL)
+          handle->recv_cb(handle, UV_ENOBUFS, &buf, NULL, 0);
         goto done;
       }
 

test/test-list.h

@@ -153,6 +153,7 @@ TEST_DECLARE   (tcp_flags)
 TEST_DECLARE   (tcp_write_to_half_open_connection)
 TEST_DECLARE   (tcp_unexpected_read)
 TEST_DECLARE   (tcp_read_stop)
+TEST_DECLARE   (tcp_read_stop_from_alloc)
 TEST_DECLARE   (tcp_read_stop_start)
 TEST_DECLARE   (tcp_reuseport)
 TEST_DECLARE   (tcp_rst)
@@ -163,6 +164,7 @@ TEST_DECLARE   (tcp_bind6_error_inval)
 TEST_DECLARE   (tcp_bind6_localhost_ok)
 TEST_DECLARE   (tcp_write_ready)
 TEST_DECLARE   (udp_alloc_cb_fail)
+TEST_DECLARE   (udp_alloc_cb_stop_recv)
 TEST_DECLARE   (udp_bind)
 TEST_DECLARE   (udp_bind_reuseaddr)
 TEST_DECLARE   (udp_connect)
@@ -778,6 +780,9 @@ TASK_LIST_START
   TEST_ENTRY  (tcp_read_stop)
   TEST_HELPER (tcp_read_stop, tcp4_echo_server)
 
+  TEST_ENTRY  (tcp_read_stop_from_alloc)
+  TEST_HELPER (tcp_read_stop_from_alloc, tcp4_echo_server)
+
   TEST_ENTRY  (tcp_read_stop_start)
 
   TEST_ENTRY  (tcp_reuseport)
@@ -792,6 +797,7 @@ TASK_LIST_START
   TEST_ENTRY  (tcp_bind6_localhost_ok)
 
   TEST_ENTRY  (udp_alloc_cb_fail)
+  TEST_ENTRY  (udp_alloc_cb_stop_recv)
   TEST_ENTRY  (udp_bind)
   TEST_ENTRY  (udp_bind_reuseaddr)
   TEST_ENTRY  (udp_connect)

test/test-tcp-read-stop-from-alloc.c

@@ -0,0 +1,80 @@
+/* Copyright 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_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*) &tcp_handle, NULL);
+}
+
+
+static void alloc_cb(uv_handle_t *handle, size_t sugg, uv_buf_t *buffer) {
+  buffer->base = NULL;
+  buffer->len = 0;
+  ASSERT_OK(uv_read_stop((uv_stream_t*) handle));
+}
+
+
+static void connect_cb(uv_connect_t* req, int status) {
+  uv_buf_t buf;
+
+  ASSERT_OK(status);
+
+  /* Write something to the echo server so we get something */
+  buf = uv_buf_init("hola", 4);
+  ASSERT_OK(uv_write(&write_req,
+                     (uv_stream_t*) &tcp_handle,
+                     &buf,
+                     1,
+                     write_cb));
+
+  ASSERT_OK(uv_read_start((uv_stream_t*) &tcp_handle,
+                          alloc_cb,
+                          (uv_read_cb) fail_cb));
+}
+
+
+TEST_IMPL(tcp_read_stop_from_alloc) {
+  uv_connect_t connect_req;
+  struct sockaddr_in addr;
+
+  ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
+  ASSERT_OK(uv_timer_init(uv_default_loop(), &timer_handle));
+  ASSERT_OK(uv_tcp_init(uv_default_loop(), &tcp_handle));
+  ASSERT_OK(uv_tcp_connect(&connect_req,
+                           &tcp_handle,
+                           (const struct sockaddr*) &addr,
+                           connect_cb));
+  ASSERT_OK(uv_run(uv_default_loop(), UV_RUN_DEFAULT));
+  MAKE_VALGRIND_HAPPY(uv_default_loop());
+
+  return 0;
+}

test/test-udp-alloc-recv-stop.c

@@ -0,0 +1,81 @@
+/* Copyright 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 "task.h"
+#include "uv.h"
+
+static uv_udp_t server;
+static uv_udp_t client;
+
+
+static void fail_cb(uv_udp_t* handle,
+                    ssize_t nread,
+                    const uv_buf_t* buf,
+                    const struct sockaddr* addr,
+                    unsigned flags) {
+/* win will always call this cb so run the assertion in other platforms */
+#ifndef _WIN32
+  ASSERT(0 && "fail_cb called");
+#endif
+}
+
+
+static void alloc_cb(uv_handle_t* handle, size_t sugg_size, uv_buf_t* buf) {
+  ASSERT_OK(uv_udp_recv_stop((uv_udp_t*) handle));
+}
+
+
+TEST_IMPL(udp_alloc_cb_stop_recv) {
+  struct sockaddr_in addr;
+  uv_udp_send_t req;
+  uv_buf_t buf;
+  int r;
+
+  ASSERT_OK(uv_ip4_addr("0.0.0.0", TEST_PORT, &addr));
+
+  r = uv_udp_init(uv_default_loop(), &server);
+  ASSERT_OK(r);
+
+  r = uv_udp_bind(&server, (const struct sockaddr*) &addr, 0);
+  ASSERT_OK(r);
+
+  r = uv_udp_recv_start(&server, alloc_cb, (uv_udp_recv_cb) fail_cb);
+  ASSERT_OK(r);
+
+  ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
+
+  r = uv_udp_init(uv_default_loop(), &client);
+  ASSERT_OK(r);
+
+  buf = uv_buf_init("PING", 4);
+  r = uv_udp_send(&req,
+                  &client,
+                  &buf,
+                  1,
+                  (const struct sockaddr*) &addr,
+                  NULL);
+  ASSERT_OK(r);
+
+  uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+
+  MAKE_VALGRIND_HAPPY(uv_default_loop());
+  return 0;
+}