From 80a5e3ba01cd12c4dababdfac2e2baac2349b710 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juan=20Jos=C3=A9?= Date: Mon, 6 Oct 2025 09:45:58 -0500 Subject: [PATCH] linux: add MSG_ERRQUEUE ipv4/ipv6 udp support (#4902) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes: https://github.com/libuv/libuv/issues/4447 Signed-off-by: Juan José Arboleda Co-authored-by: theanarkh --- CMakeLists.txt | 1 + Makefile.am | 1 + src/unix/udp.c | 97 +++++++++++++++-- test/test-list.h | 4 + test/test-udp-recvmsg-unreachable-error.c | 125 ++++++++++++++++++++++ test/test-udp-send-unreachable.c | 5 +- 6 files changed, 225 insertions(+), 8 deletions(-) create mode 100644 test/test-udp-recvmsg-unreachable-error.c diff --git a/CMakeLists.txt b/CMakeLists.txt index e725bff71..449dc8322 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/Makefile.am b/Makefile.am index 8a41f18ab..797efc83e 100644 --- a/Makefile.am +++ b/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 \ diff --git a/src/unix/udp.c b/src/unix/udp.c index f66b00a72..c96a0e4e6 100644 --- a/src/unix/udp.c +++ b/src/unix/udp.c @@ -32,6 +32,10 @@ #endif #include +#if defined(__linux__) +#include +#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 */ diff --git a/test/test-list.h b/test/test-list.h index 0dea54469..954eff330 100644 --- a/test/test-list.h +++ b/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) diff --git a/test/test-udp-recvmsg-unreachable-error.c b/test/test-udp-recvmsg-unreachable-error.c new file mode 100644 index 000000000..48990cc6d --- /dev/null +++ b/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 +#include + +#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; +} diff --git a/test/test-udp-send-unreachable.c b/test/test-udp-send-unreachable.c index 0a2f4a47a..a480206e0 100644 --- a/test/test-udp-send-unreachable.c +++ b/test/test-udp-send-unreachable.c @@ -83,7 +83,10 @@ static void recv_cb(uv_udp_t* handle, recv_cb_called++; if (nread < 0) { - ASSERT(0 && "unexpected error"); + if (flags && can_recverr) + ASSERT(flags & UV_UDP_LINUX_RECVERR); + else + ASSERT(0 && "unexpected error"); } else if (nread == 0) { /* Returning unused buffer */ ASSERT_NULL(addr);