linux: add MSG_ERRQUEUE ipv4/ipv6 udp support (#4902)

Fixes: https://github.com/libuv/libuv/issues/4447
Signed-off-by: Juan José Arboleda <soyjuanarbol@gmail.com>
Co-authored-by: theanarkh <theratliter@gmail.com>

CMakeLists.txt

@@ -700,6 +700,7 @@ if(LIBUV_BUILD_TESTS)
        test/test-udp-send-immediate.c
        test/test-udp-sendmmsg-error.c
        test/test-udp-send-unreachable.c
+       test/test-udp-recvmsg-unreachable-error.c
        test/test-udp-try-send.c
        test/test-udp-recv-in-a-row.c
        test/test-udp-reuseport.c

Makefile.am

@@ -325,6 +325,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \
                          test/test-udp-send-immediate.c \
                          test/test-udp-sendmmsg-error.c \
                          test/test-udp-send-unreachable.c \
+                         test/test-udp-recvmsg-unreachable-error.c \
                          test/test-udp-try-send.c \
                          test/test-udp-recv-in-a-row.c \
                          test/test-udp-reuseport.c \

src/unix/udp.c

@@ -32,6 +32,10 @@
 #endif
 #include <sys/un.h>
 
+#if defined(__linux__)
+#include <linux/errqueue.h>
+#endif
+
 #if defined(IPV6_JOIN_GROUP) && !defined(IPV6_ADD_MEMBERSHIP)
 # define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
 #endif
@@ -41,7 +45,7 @@
 #endif
 
 static void uv__udp_run_completed(uv_udp_t* handle);
-static void uv__udp_recvmsg(uv_udp_t* handle);
+static void uv__udp_recvmsg(uv_udp_t* handle, int flag);
 static void uv__udp_sendmsg(uv_udp_t* handle);
 static int uv__udp_maybe_deferred_bind(uv_udp_t* handle,
                                        int domain,
@@ -135,6 +139,39 @@ static void uv__udp_run_completed(uv_udp_t* handle) {
 }
 
 
+#if defined(__linux__)
+static int uv__udp_recvmsg_errqueue(uv_udp_t* handle,
+                                    struct msghdr* h,
+                                    uv_buf_t* buf,
+                                    const struct sockaddr* peer,
+                                    int flags) {
+  struct cmsghdr* cmsg;
+  struct sock_extended_err* serr;
+  struct sockaddr* offender;
+
+  if (!(h->msg_flags & MSG_ERRQUEUE))
+    return 0;
+
+  flags |= UV_UDP_LINUX_RECVERR;
+  for (cmsg = CMSG_FIRSTHDR(h); cmsg != NULL; cmsg = CMSG_NXTHDR(h, cmsg)) {
+    if ((cmsg->cmsg_level == SOL_IP   && cmsg->cmsg_type == IP_RECVERR) ||
+        (cmsg->cmsg_level == SOL_IPV6 && cmsg->cmsg_type == IPV6_RECVERR)) {
+      serr = (struct sock_extended_err*) CMSG_DATA(cmsg);
+
+      offender = SO_EE_OFFENDER(serr);
+      handle->recv_cb(handle,
+                      UV__ERR(serr->ee_errno),
+                      buf,
+                      offender,
+                      flags);
+      return 1; /* handled */
+    }
+  }
+  return 0;
+}
+#endif
+
+
 void uv__udp_io(uv_loop_t* loop, uv__io_t* w, unsigned int revents) {
   uv_udp_t* handle;
 
@@ -142,7 +179,13 @@ void uv__udp_io(uv_loop_t* loop, uv__io_t* w, unsigned int revents) {
   assert(handle->type == UV_UDP);
 
   if (revents & POLLIN)
-    uv__udp_recvmsg(handle);
+    uv__udp_recvmsg(handle, 0);
+
+  /* Just Linux support for now. */
+#if defined(__linux__)
+  if (revents & POLLERR)
+    uv__udp_recvmsg(handle, MSG_ERRQUEUE);
+#endif
 
   if (revents & POLLOUT && !uv__is_closing(handle)) {
     uv__udp_sendmsg(handle);
@@ -150,7 +193,7 @@ void uv__udp_io(uv_loop_t* loop, uv__io_t* w, unsigned int revents) {
   }
 }
 
-static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
+static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf, int flag) {
 #if defined(__linux__) || defined(__FreeBSD__) || defined(__APPLE__)
   struct sockaddr_in6 peers[20];
   struct iovec iov[ARRAY_SIZE(peers)];
@@ -160,6 +203,9 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
   size_t chunks;
   int flags;
   size_t k;
+#if defined(__linux__)
+  char control[ARRAY_SIZE(peers)][64];
+#endif
 
   /* prepare structures for recvmmsg */
   chunks = buf->len / UV__UDP_DGRAM_MAXSIZE;
@@ -177,6 +223,12 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
     msgs[k].msg_hdr.msg_controllen = 0;
     msgs[k].msg_hdr.msg_flags = 0;
     msgs[k].msg_len = 0;
+#if defined(__linux__)
+    if (flag & MSG_ERRQUEUE) {
+      msgs[k].msg_hdr.msg_control = control[k];
+      msgs[k].msg_hdr.msg_controllen = sizeof(control[k]);
+    }
+#endif
   }
 
 #if defined(__APPLE__)
@@ -185,7 +237,7 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
   while (nread == -1 && errno == EINTR);
 #else
   do
-    nread = recvmmsg(handle->io_watcher.fd, msgs, chunks, 0, NULL);
+    nread = recvmmsg(handle->io_watcher.fd, msgs, chunks, flag, NULL);
   while (nread == -1 && errno == EINTR);
 #endif
 
@@ -202,6 +254,13 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
         flags |= UV_UDP_PARTIAL;
 
       chunk_buf = uv_buf_init(iov[k].iov_base, iov[k].iov_len);
+#if defined(__linux__)
+      if ((flag & MSG_ERRQUEUE) &&
+          uv__udp_recvmsg_errqueue(handle, &msgs[k].msg_hdr, &chunk_buf,
+                                   (const struct sockaddr*) &peers[k], flags)) {
+        continue;
+      }
+#endif
       handle->recv_cb(handle,
                       msgs[k].msg_len,
                       &chunk_buf,
@@ -219,13 +278,16 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
 #endif  /* __linux__ || ____FreeBSD__ || __APPLE__ */
 }
 
-static void uv__udp_recvmsg(uv_udp_t* handle) {
+static void uv__udp_recvmsg(uv_udp_t* handle, int flag) {
   struct sockaddr_storage peer;
   struct msghdr h;
   ssize_t nread;
   uv_buf_t buf;
   int flags;
   int count;
+#if defined(__linux__)
+  char control[256];
+#endif
 
   assert(handle->recv_cb != NULL);
   assert(handle->alloc_cb != NULL);
@@ -245,7 +307,7 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
     assert(buf.base != NULL);
 
     if (uv_udp_using_recvmmsg(handle)) {
-      nread = uv__udp_recvmmsg(handle, &buf);
+      nread = uv__udp_recvmmsg(handle, &buf, flag);
       if (nread > 0)
         count -= nread;
       continue;
@@ -257,13 +319,26 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
     h.msg_namelen = sizeof(peer);
     h.msg_iov = (void*) &buf;
     h.msg_iovlen = 1;
+#if defined(__linux__)
+    if (flag & MSG_ERRQUEUE) {
+      h.msg_control = control;
+      h.msg_controllen = sizeof(control);
+    }
+#endif
 
     do {
-      nread = recvmsg(handle->io_watcher.fd, &h, 0);
+      nread = recvmsg(handle->io_watcher.fd, &h, flag);
     }
     while (nread == -1 && errno == EINTR);
 
     if (nread == -1) {
+#if defined(__linux__)
+      if ((flag & MSG_ERRQUEUE) &&
+          uv__udp_recvmsg_errqueue(handle, &h, &buf,
+                                   (const struct sockaddr*) &peer, flags)) {
+        goto out;
+      }
+#endif
       if (errno == EAGAIN || errno == EWOULDBLOCK)
         handle->recv_cb(handle, 0, &buf, NULL, 0);
       else
@@ -274,8 +349,16 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
       if (h.msg_flags & MSG_TRUNC)
         flags |= UV_UDP_PARTIAL;
 
+#if defined(__linux__)
+      if ((flag & MSG_ERRQUEUE) &&
+          uv__udp_recvmsg_errqueue(handle, &h, &buf,
+                                   (const struct sockaddr*) &peer, flags)) {
+        goto out;
+      }
+#endif
       handle->recv_cb(handle, nread, &buf, (const struct sockaddr*) &peer, flags);
     }
+out:
     count--;
   }
   /* recv_cb callback may decide to pause or close the handle */

test/test-list.h

@@ -175,6 +175,8 @@ TEST_DECLARE   (udp_send_and_recv)
 TEST_DECLARE   (udp_send_hang_loop)
 TEST_DECLARE   (udp_send_immediate)
 TEST_DECLARE   (udp_send_unreachable)
+TEST_DECLARE   (udp_recvmsg_unreachable_error)
+TEST_DECLARE   (udp_recvmsg_unreachable_error6)
 TEST_DECLARE   (udp_mmsg)
 TEST_DECLARE   (udp_multicast_join)
 TEST_DECLARE   (udp_multicast_join6)
@@ -805,6 +807,8 @@ TASK_LIST_START
   TEST_ENTRY  (udp_send_hang_loop)
   TEST_ENTRY  (udp_send_immediate)
   TEST_ENTRY  (udp_send_unreachable)
+  TEST_ENTRY  (udp_recvmsg_unreachable_error)
+  TEST_ENTRY  (udp_recvmsg_unreachable_error6)
   TEST_ENTRY  (udp_dgram_too_big)
   TEST_ENTRY  (udp_dual_stack)
   TEST_ENTRY  (udp_ipv6_only)

test/test-udp-recvmsg-unreachable-error.c

@@ -0,0 +1,125 @@
+/* Copyright libuv project and 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"
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#define CLIENT_TEST_PORT 9123
+#define SERVER_TEST_PORT 9124
+#define RECV_CB_MAX_CALL 3 /* ECONNREFUSED, EAGAIN/EWOULDBLOCK, ICMP delivery */
+
+static int recv_cb_called = 0;
+
+static void udp_send_cb(uv_udp_send_t* req, int status) {
+  ASSERT_EQ(status, 0);
+}
+
+static void alloc_cb(uv_handle_t* handle,
+                     size_t suggested_size,
+                     uv_buf_t* buf) {
+  static char storage[4]; /* "PING" */
+  buf->base = storage;
+  buf->len = sizeof(storage);
+}
+
+static void read_cb(uv_udp_t* handle,
+                    ssize_t nread,
+                    const uv_buf_t* buf,
+                    const struct sockaddr* addr,
+                    unsigned flags) {
+  ASSERT(flags == 0 || (flags & UV_UDP_LINUX_RECVERR));
+  recv_cb_called++;
+}
+
+static void timer_cb(uv_timer_t* handle) {
+  uv_udp_t* udp = handle->data;
+  uv_close((uv_handle_t*) udp, NULL);
+}
+
+TEST_IMPL(udp_recvmsg_unreachable_error) {
+#if !defined(__linux__)
+  RETURN_SKIP("This test is Linux-specific");
+#endif
+  struct sockaddr_in server_addr, client_addr;
+  uv_udp_t client;
+  uv_timer_t timer;
+  uv_udp_send_t send_req;
+  uv_buf_t buf = uv_buf_init("PING", 4);
+  ASSERT_OK(uv_ip4_addr("127.0.0.1", CLIENT_TEST_PORT, &client_addr));
+  ASSERT_OK(uv_ip4_addr("127.0.0.1", SERVER_TEST_PORT, &server_addr));
+  ASSERT_OK(uv_udp_init(uv_default_loop(), &client));
+  ASSERT_OK(uv_timer_init(uv_default_loop(), &timer));
+  ASSERT_OK(uv_udp_bind(&client,
+                        (const struct sockaddr*) &client_addr,
+                        UV_UDP_LINUX_RECVERR));
+  ASSERT_OK(uv_udp_recv_start(&client, alloc_cb, read_cb));
+  timer.data = &client;
+  ASSERT_OK(uv_timer_start(&timer, timer_cb, 3000, 0));
+  ASSERT_OK(uv_udp_send(&send_req,
+                        &client,
+                        &buf,
+                        1,
+                        (const struct sockaddr*) &server_addr,
+                        udp_send_cb));
+  uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+  ASSERT_EQ(recv_cb_called, RECV_CB_MAX_CALL);
+  return 0;
+}
+
+TEST_IMPL(udp_recvmsg_unreachable_error6) {
+#if !defined(__linux__)
+  RETURN_SKIP("This test is Linux-specific");
+#endif
+  struct sockaddr_in6 server_addr, client_addr;
+  uv_udp_t client;
+  uv_timer_t timer;
+  uv_udp_send_t send_req;
+  uv_buf_t buf = uv_buf_init("PING", 4);
+
+  ASSERT_OK(uv_ip6_addr("::1", CLIENT_TEST_PORT, &client_addr));
+  ASSERT_OK(uv_ip6_addr("::1", SERVER_TEST_PORT, &server_addr));
+
+  ASSERT_OK(uv_udp_init(uv_default_loop(), &client));
+  ASSERT_OK(uv_timer_init(uv_default_loop(), &timer));
+
+  ASSERT_OK(uv_udp_bind(&client,
+                        (const struct sockaddr*) &client_addr,
+                        UV_UDP_LINUX_RECVERR));
+  ASSERT_OK(uv_udp_recv_start(&client, alloc_cb, read_cb));
+
+  timer.data = &client;
+  ASSERT_OK(uv_timer_start(&timer, timer_cb, 3000, 0));
+
+  ASSERT_OK(uv_udp_send(&send_req,
+                        &client,
+                        &buf,
+                        1,
+                        (const struct sockaddr*) &server_addr,
+                        udp_send_cb));
+
+  uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+
+  ASSERT_EQ(recv_cb_called, RECV_CB_MAX_CALL);
+  return 0;
+}

test/test-udp-send-unreachable.c

@@ -83,6 +83,9 @@ static void recv_cb(uv_udp_t* handle,
   recv_cb_called++;
 
   if (nread < 0) {
+    if (flags && can_recverr)
+      ASSERT(flags & UV_UDP_LINUX_RECVERR);
+    else
       ASSERT(0 && "unexpected error");
   } else if (nread == 0) {
     /* Returning unused buffer */