Skip to content

Commit 80a5e3b

Browse files
juanarboltheanarkh
andauthored
linux: add MSG_ERRQUEUE ipv4/ipv6 udp support (libuv#4902)
Fixes: libuv#4447 Signed-off-by: Juan José Arboleda <[email protected]> Co-authored-by: theanarkh <[email protected]>
1 parent 87943b0 commit 80a5e3b

File tree

6 files changed

+225
-8
lines changed

6 files changed

+225
-8
lines changed

CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -700,6 +700,7 @@ if(LIBUV_BUILD_TESTS)
700700
test/test-udp-send-immediate.c
701701
test/test-udp-sendmmsg-error.c
702702
test/test-udp-send-unreachable.c
703+
test/test-udp-recvmsg-unreachable-error.c
703704
test/test-udp-try-send.c
704705
test/test-udp-recv-in-a-row.c
705706
test/test-udp-reuseport.c

Makefile.am

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -325,6 +325,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \
325325
test/test-udp-send-immediate.c \
326326
test/test-udp-sendmmsg-error.c \
327327
test/test-udp-send-unreachable.c \
328+
test/test-udp-recvmsg-unreachable-error.c \
328329
test/test-udp-try-send.c \
329330
test/test-udp-recv-in-a-row.c \
330331
test/test-udp-reuseport.c \

src/unix/udp.c

Lines changed: 90 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,10 @@
3232
#endif
3333
#include <sys/un.h>
3434

35+
#if defined(__linux__)
36+
#include <linux/errqueue.h>
37+
#endif
38+
3539
#if defined(IPV6_JOIN_GROUP) && !defined(IPV6_ADD_MEMBERSHIP)
3640
# define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
3741
#endif
@@ -41,7 +45,7 @@
4145
#endif
4246

4347
static void uv__udp_run_completed(uv_udp_t* handle);
44-
static void uv__udp_recvmsg(uv_udp_t* handle);
48+
static void uv__udp_recvmsg(uv_udp_t* handle, int flag);
4549
static void uv__udp_sendmsg(uv_udp_t* handle);
4650
static int uv__udp_maybe_deferred_bind(uv_udp_t* handle,
4751
int domain,
@@ -135,22 +139,61 @@ static void uv__udp_run_completed(uv_udp_t* handle) {
135139
}
136140

137141

142+
#if defined(__linux__)
143+
static int uv__udp_recvmsg_errqueue(uv_udp_t* handle,
144+
struct msghdr* h,
145+
uv_buf_t* buf,
146+
const struct sockaddr* peer,
147+
int flags) {
148+
struct cmsghdr* cmsg;
149+
struct sock_extended_err* serr;
150+
struct sockaddr* offender;
151+
152+
if (!(h->msg_flags & MSG_ERRQUEUE))
153+
return 0;
154+
155+
flags |= UV_UDP_LINUX_RECVERR;
156+
for (cmsg = CMSG_FIRSTHDR(h); cmsg != NULL; cmsg = CMSG_NXTHDR(h, cmsg)) {
157+
if ((cmsg->cmsg_level == SOL_IP && cmsg->cmsg_type == IP_RECVERR) ||
158+
(cmsg->cmsg_level == SOL_IPV6 && cmsg->cmsg_type == IPV6_RECVERR)) {
159+
serr = (struct sock_extended_err*) CMSG_DATA(cmsg);
160+
161+
offender = SO_EE_OFFENDER(serr);
162+
handle->recv_cb(handle,
163+
UV__ERR(serr->ee_errno),
164+
buf,
165+
offender,
166+
flags);
167+
return 1; /* handled */
168+
}
169+
}
170+
return 0;
171+
}
172+
#endif
173+
174+
138175
void uv__udp_io(uv_loop_t* loop, uv__io_t* w, unsigned int revents) {
139176
uv_udp_t* handle;
140177

141178
handle = container_of(w, uv_udp_t, io_watcher);
142179
assert(handle->type == UV_UDP);
143180

144181
if (revents & POLLIN)
145-
uv__udp_recvmsg(handle);
182+
uv__udp_recvmsg(handle, 0);
183+
184+
/* Just Linux support for now. */
185+
#if defined(__linux__)
186+
if (revents & POLLERR)
187+
uv__udp_recvmsg(handle, MSG_ERRQUEUE);
188+
#endif
146189

147190
if (revents & POLLOUT && !uv__is_closing(handle)) {
148191
uv__udp_sendmsg(handle);
149192
uv__udp_run_completed(handle);
150193
}
151194
}
152195

153-
static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
196+
static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf, int flag) {
154197
#if defined(__linux__) || defined(__FreeBSD__) || defined(__APPLE__)
155198
struct sockaddr_in6 peers[20];
156199
struct iovec iov[ARRAY_SIZE(peers)];
@@ -160,6 +203,9 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
160203
size_t chunks;
161204
int flags;
162205
size_t k;
206+
#if defined(__linux__)
207+
char control[ARRAY_SIZE(peers)][64];
208+
#endif
163209

164210
/* prepare structures for recvmmsg */
165211
chunks = buf->len / UV__UDP_DGRAM_MAXSIZE;
@@ -177,6 +223,12 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
177223
msgs[k].msg_hdr.msg_controllen = 0;
178224
msgs[k].msg_hdr.msg_flags = 0;
179225
msgs[k].msg_len = 0;
226+
#if defined(__linux__)
227+
if (flag & MSG_ERRQUEUE) {
228+
msgs[k].msg_hdr.msg_control = control[k];
229+
msgs[k].msg_hdr.msg_controllen = sizeof(control[k]);
230+
}
231+
#endif
180232
}
181233

182234
#if defined(__APPLE__)
@@ -185,7 +237,7 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
185237
while (nread == -1 && errno == EINTR);
186238
#else
187239
do
188-
nread = recvmmsg(handle->io_watcher.fd, msgs, chunks, 0, NULL);
240+
nread = recvmmsg(handle->io_watcher.fd, msgs, chunks, flag, NULL);
189241
while (nread == -1 && errno == EINTR);
190242
#endif
191243

@@ -202,6 +254,13 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
202254
flags |= UV_UDP_PARTIAL;
203255

204256
chunk_buf = uv_buf_init(iov[k].iov_base, iov[k].iov_len);
257+
#if defined(__linux__)
258+
if ((flag & MSG_ERRQUEUE) &&
259+
uv__udp_recvmsg_errqueue(handle, &msgs[k].msg_hdr, &chunk_buf,
260+
(const struct sockaddr*) &peers[k], flags)) {
261+
continue;
262+
}
263+
#endif
205264
handle->recv_cb(handle,
206265
msgs[k].msg_len,
207266
&chunk_buf,
@@ -219,13 +278,16 @@ static int uv__udp_recvmmsg(uv_udp_t* handle, uv_buf_t* buf) {
219278
#endif /* __linux__ || ____FreeBSD__ || __APPLE__ */
220279
}
221280

222-
static void uv__udp_recvmsg(uv_udp_t* handle) {
281+
static void uv__udp_recvmsg(uv_udp_t* handle, int flag) {
223282
struct sockaddr_storage peer;
224283
struct msghdr h;
225284
ssize_t nread;
226285
uv_buf_t buf;
227286
int flags;
228287
int count;
288+
#if defined(__linux__)
289+
char control[256];
290+
#endif
229291

230292
assert(handle->recv_cb != NULL);
231293
assert(handle->alloc_cb != NULL);
@@ -245,7 +307,7 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
245307
assert(buf.base != NULL);
246308

247309
if (uv_udp_using_recvmmsg(handle)) {
248-
nread = uv__udp_recvmmsg(handle, &buf);
310+
nread = uv__udp_recvmmsg(handle, &buf, flag);
249311
if (nread > 0)
250312
count -= nread;
251313
continue;
@@ -257,13 +319,26 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
257319
h.msg_namelen = sizeof(peer);
258320
h.msg_iov = (void*) &buf;
259321
h.msg_iovlen = 1;
322+
#if defined(__linux__)
323+
if (flag & MSG_ERRQUEUE) {
324+
h.msg_control = control;
325+
h.msg_controllen = sizeof(control);
326+
}
327+
#endif
260328

261329
do {
262-
nread = recvmsg(handle->io_watcher.fd, &h, 0);
330+
nread = recvmsg(handle->io_watcher.fd, &h, flag);
263331
}
264332
while (nread == -1 && errno == EINTR);
265333

266334
if (nread == -1) {
335+
#if defined(__linux__)
336+
if ((flag & MSG_ERRQUEUE) &&
337+
uv__udp_recvmsg_errqueue(handle, &h, &buf,
338+
(const struct sockaddr*) &peer, flags)) {
339+
goto out;
340+
}
341+
#endif
267342
if (errno == EAGAIN || errno == EWOULDBLOCK)
268343
handle->recv_cb(handle, 0, &buf, NULL, 0);
269344
else
@@ -274,8 +349,16 @@ static void uv__udp_recvmsg(uv_udp_t* handle) {
274349
if (h.msg_flags & MSG_TRUNC)
275350
flags |= UV_UDP_PARTIAL;
276351

352+
#if defined(__linux__)
353+
if ((flag & MSG_ERRQUEUE) &&
354+
uv__udp_recvmsg_errqueue(handle, &h, &buf,
355+
(const struct sockaddr*) &peer, flags)) {
356+
goto out;
357+
}
358+
#endif
277359
handle->recv_cb(handle, nread, &buf, (const struct sockaddr*) &peer, flags);
278360
}
361+
out:
279362
count--;
280363
}
281364
/* recv_cb callback may decide to pause or close the handle */

test/test-list.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -175,6 +175,8 @@ TEST_DECLARE (udp_send_and_recv)
175175
TEST_DECLARE (udp_send_hang_loop)
176176
TEST_DECLARE (udp_send_immediate)
177177
TEST_DECLARE (udp_send_unreachable)
178+
TEST_DECLARE (udp_recvmsg_unreachable_error)
179+
TEST_DECLARE (udp_recvmsg_unreachable_error6)
178180
TEST_DECLARE (udp_mmsg)
179181
TEST_DECLARE (udp_multicast_join)
180182
TEST_DECLARE (udp_multicast_join6)
@@ -805,6 +807,8 @@ TASK_LIST_START
805807
TEST_ENTRY (udp_send_hang_loop)
806808
TEST_ENTRY (udp_send_immediate)
807809
TEST_ENTRY (udp_send_unreachable)
810+
TEST_ENTRY (udp_recvmsg_unreachable_error)
811+
TEST_ENTRY (udp_recvmsg_unreachable_error6)
808812
TEST_ENTRY (udp_dgram_too_big)
809813
TEST_ENTRY (udp_dual_stack)
810814
TEST_ENTRY (udp_ipv6_only)
Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
/* Copyright libuv project and contributors. All rights reserved.
2+
*
3+
* Permission is hereby granted, free of charge, to any person obtaining a copy
4+
* of this software and associated documentation files (the "Software"), to
5+
* deal in the Software without restriction, including without limitation the
6+
* rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
7+
* sell copies of the Software, and to permit persons to whom the Software is
8+
* furnished to do so, subject to the following conditions:
9+
*
10+
* The above copyright notice and this permission notice shall be included in
11+
* all copies or substantial portions of the Software.
12+
*
13+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14+
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15+
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16+
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17+
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18+
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
19+
* IN THE SOFTWARE.
20+
*/
21+
22+
#include "uv.h"
23+
#include "task.h"
24+
25+
#include <stdio.h>
26+
#include <stdlib.h>
27+
28+
#define CLIENT_TEST_PORT 9123
29+
#define SERVER_TEST_PORT 9124
30+
#define RECV_CB_MAX_CALL 3 /* ECONNREFUSED, EAGAIN/EWOULDBLOCK, ICMP delivery */
31+
32+
static int recv_cb_called = 0;
33+
34+
static void udp_send_cb(uv_udp_send_t* req, int status) {
35+
ASSERT_EQ(status, 0);
36+
}
37+
38+
static void alloc_cb(uv_handle_t* handle,
39+
size_t suggested_size,
40+
uv_buf_t* buf) {
41+
static char storage[4]; /* "PING" */
42+
buf->base = storage;
43+
buf->len = sizeof(storage);
44+
}
45+
46+
static void read_cb(uv_udp_t* handle,
47+
ssize_t nread,
48+
const uv_buf_t* buf,
49+
const struct sockaddr* addr,
50+
unsigned flags) {
51+
ASSERT(flags == 0 || (flags & UV_UDP_LINUX_RECVERR));
52+
recv_cb_called++;
53+
}
54+
55+
static void timer_cb(uv_timer_t* handle) {
56+
uv_udp_t* udp = handle->data;
57+
uv_close((uv_handle_t*) udp, NULL);
58+
}
59+
60+
TEST_IMPL(udp_recvmsg_unreachable_error) {
61+
#if !defined(__linux__)
62+
RETURN_SKIP("This test is Linux-specific");
63+
#endif
64+
struct sockaddr_in server_addr, client_addr;
65+
uv_udp_t client;
66+
uv_timer_t timer;
67+
uv_udp_send_t send_req;
68+
uv_buf_t buf = uv_buf_init("PING", 4);
69+
ASSERT_OK(uv_ip4_addr("127.0.0.1", CLIENT_TEST_PORT, &client_addr));
70+
ASSERT_OK(uv_ip4_addr("127.0.0.1", SERVER_TEST_PORT, &server_addr));
71+
ASSERT_OK(uv_udp_init(uv_default_loop(), &client));
72+
ASSERT_OK(uv_timer_init(uv_default_loop(), &timer));
73+
ASSERT_OK(uv_udp_bind(&client,
74+
(const struct sockaddr*) &client_addr,
75+
UV_UDP_LINUX_RECVERR));
76+
ASSERT_OK(uv_udp_recv_start(&client, alloc_cb, read_cb));
77+
timer.data = &client;
78+
ASSERT_OK(uv_timer_start(&timer, timer_cb, 3000, 0));
79+
ASSERT_OK(uv_udp_send(&send_req,
80+
&client,
81+
&buf,
82+
1,
83+
(const struct sockaddr*) &server_addr,
84+
udp_send_cb));
85+
uv_run(uv_default_loop(), UV_RUN_DEFAULT);
86+
ASSERT_EQ(recv_cb_called, RECV_CB_MAX_CALL);
87+
return 0;
88+
}
89+
90+
TEST_IMPL(udp_recvmsg_unreachable_error6) {
91+
#if !defined(__linux__)
92+
RETURN_SKIP("This test is Linux-specific");
93+
#endif
94+
struct sockaddr_in6 server_addr, client_addr;
95+
uv_udp_t client;
96+
uv_timer_t timer;
97+
uv_udp_send_t send_req;
98+
uv_buf_t buf = uv_buf_init("PING", 4);
99+
100+
ASSERT_OK(uv_ip6_addr("::1", CLIENT_TEST_PORT, &client_addr));
101+
ASSERT_OK(uv_ip6_addr("::1", SERVER_TEST_PORT, &server_addr));
102+
103+
ASSERT_OK(uv_udp_init(uv_default_loop(), &client));
104+
ASSERT_OK(uv_timer_init(uv_default_loop(), &timer));
105+
106+
ASSERT_OK(uv_udp_bind(&client,
107+
(const struct sockaddr*) &client_addr,
108+
UV_UDP_LINUX_RECVERR));
109+
ASSERT_OK(uv_udp_recv_start(&client, alloc_cb, read_cb));
110+
111+
timer.data = &client;
112+
ASSERT_OK(uv_timer_start(&timer, timer_cb, 3000, 0));
113+
114+
ASSERT_OK(uv_udp_send(&send_req,
115+
&client,
116+
&buf,
117+
1,
118+
(const struct sockaddr*) &server_addr,
119+
udp_send_cb));
120+
121+
uv_run(uv_default_loop(), UV_RUN_DEFAULT);
122+
123+
ASSERT_EQ(recv_cb_called, RECV_CB_MAX_CALL);
124+
return 0;
125+
}

test/test-udp-send-unreachable.c

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,10 @@ static void recv_cb(uv_udp_t* handle,
8383
recv_cb_called++;
8484

8585
if (nread < 0) {
86-
ASSERT(0 && "unexpected error");
86+
if (flags && can_recverr)
87+
ASSERT(flags & UV_UDP_LINUX_RECVERR);
88+
else
89+
ASSERT(0 && "unexpected error");
8790
} else if (nread == 0) {
8891
/* Returning unused buffer */
8992
ASSERT_NULL(addr);

0 commit comments

Comments
 (0)