gre-ilya/libuv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

linux: add UDP support for MSG_ERRQUEUE on Linux IPv4/IPv6

Browse Source 
Fixes: https://github.com/libuv/libuv/issues/4447
Co-authored-by: theanarkh <theratliter@gmail.com>
Signed-off-by: Juan José Arboleda <soyjuanarbol@gmail.com>
This commit is contained in:
Juan José Arboleda 2025-10-03 14:27:11 -05:00
parent ee67f3db88
commit d984dea65c
6 changed files with 225 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CMakeLists.txt
View File

@ -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

1
Makefile.am
View File

@ -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 \

97
src/unix/udp.c
View File

@ -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 */

4
test/test-list.h
View File

@ -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)

125
test/test-udp-recvmsg-unreachable-error.c Normal file
View File

@ -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;
}

5
test/test-udp-send-unreachable.c
View File

@ -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);