1 /* SPDX-License-Identifier: MIT */
2 /*
3 * Simple test case showing using sendmsg and recvmsg through io_uring
4 */
5 #include <stdio.h>
6 #include <stdlib.h>
7 #include <string.h>
8 #include <unistd.h>
9 #include <errno.h>
10 #include <arpa/inet.h>
11 #include <sys/types.h>
12 #include <sys/socket.h>
13 #include <pthread.h>
14 #include <assert.h>
15
16 #include "liburing.h"
17
18 static char str[] = "This is a test of sendmsg and recvmsg over io_uring!";
19
20 static int ud;
21
22 #define MAX_MSG 128
23
24 #define PORT 10203
25 #define HOST "127.0.0.1"
26
27 #define BUF_BGID 10
28 #define BUF_BID 89
29
30 #define MAX_IOV_COUNT 10
31
32 static int no_pbuf_ring;
33
recv_prep(struct io_uring * ring,int * sockfd,struct iovec iov[],int iov_count,int bgid,int async)34 static int recv_prep(struct io_uring *ring, int *sockfd, struct iovec iov[],
35 int iov_count, int bgid, int async)
36 {
37 struct sockaddr_in saddr;
38 struct msghdr msg;
39 struct io_uring_sqe *sqe;
40 int ret, val = 1;
41
42 memset(&saddr, 0, sizeof(saddr));
43 saddr.sin_family = AF_INET;
44 saddr.sin_addr.s_addr = htonl(INADDR_ANY);
45 saddr.sin_port = htons(PORT);
46
47 *sockfd = socket(AF_INET, SOCK_DGRAM, 0);
48 if (*sockfd < 0) {
49 perror("socket");
50 return 1;
51 }
52
53 val = 1;
54 setsockopt(*sockfd, SOL_SOCKET, SO_REUSEPORT, &val, sizeof(val));
55 setsockopt(*sockfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
56
57 ret = bind(*sockfd, (struct sockaddr *)&saddr, sizeof(saddr));
58 if (ret < 0) {
59 perror("bind");
60 goto err;
61 }
62
63 sqe = io_uring_get_sqe(ring);
64 if (!sqe) {
65 fprintf(stderr, "io_uring_get_sqe failed\n");
66 return 1;
67 }
68
69 io_uring_prep_recvmsg(sqe, *sockfd, &msg, 0);
70 if (bgid) {
71 iov->iov_base = NULL;
72 sqe->flags |= IOSQE_BUFFER_SELECT;
73 sqe->buf_group = bgid;
74 iov_count = 1;
75 }
76 sqe->user_data = ++ud;
77 if (async)
78 sqe->flags |= IOSQE_ASYNC;
79 memset(&msg, 0, sizeof(msg));
80 msg.msg_namelen = sizeof(struct sockaddr_in);
81 msg.msg_iov = iov;
82 msg.msg_iovlen = iov_count;
83
84 ret = io_uring_submit(ring);
85 if (ret <= 0) {
86 fprintf(stderr, "submit failed: %d\n", ret);
87 goto err;
88 }
89
90 return 0;
91 err:
92 close(*sockfd);
93 return 1;
94 }
95
96 struct recv_data {
97 pthread_mutex_t *mutex;
98 int buf_select;
99 int buf_ring;
100 int no_buf_add;
101 int iov_count;
102 int async;
103 };
104
do_recvmsg(struct io_uring * ring,char buf[MAX_MSG+1],struct recv_data * rd)105 static int do_recvmsg(struct io_uring *ring, char buf[MAX_MSG + 1],
106 struct recv_data *rd)
107 {
108 struct io_uring_cqe *cqe;
109 int ret;
110
111 ret = io_uring_wait_cqe(ring, &cqe);
112 if (ret) {
113 fprintf(stdout, "wait_cqe: %d\n", ret);
114 goto err;
115 }
116 if (cqe->res < 0) {
117 if (rd->no_buf_add && (rd->buf_select || rd->buf_ring))
118 return 0;
119 fprintf(stderr, "%s: failed cqe: %d\n", __FUNCTION__, cqe->res);
120 goto err;
121 }
122 if (cqe->flags & IORING_CQE_F_BUFFER) {
123 int bid = cqe->flags >> 16;
124 if (bid != BUF_BID)
125 fprintf(stderr, "Buffer ID mismatch %d\n", bid);
126 }
127
128 if (rd->no_buf_add && (rd->buf_ring || rd->buf_select)) {
129 fprintf(stderr, "Expected -ENOBUFS: %d\n", cqe->res);
130 goto err;
131 }
132
133 if (cqe->res -1 != strlen(str)) {
134 fprintf(stderr, "got wrong length: %d/%d\n", cqe->res,
135 (int) strlen(str) + 1);
136 goto err;
137 }
138
139 if (strncmp(str, buf, MAX_MSG + 1)) {
140 fprintf(stderr, "string mismatch\n");
141 goto err;
142 }
143
144 return 0;
145 err:
146 return 1;
147 }
148
init_iov(struct iovec iov[MAX_IOV_COUNT],int iov_to_use,char buf[MAX_MSG+1])149 static void init_iov(struct iovec iov[MAX_IOV_COUNT], int iov_to_use,
150 char buf[MAX_MSG + 1])
151 {
152 int i, last_idx = iov_to_use - 1;
153
154 assert(0 < iov_to_use && iov_to_use <= MAX_IOV_COUNT);
155 for (i = 0; i < last_idx; ++i) {
156 iov[i].iov_base = buf + i;
157 iov[i].iov_len = 1;
158 }
159
160 iov[last_idx].iov_base = buf + last_idx;
161 iov[last_idx].iov_len = MAX_MSG - last_idx;
162 }
163
recv_fn(void * data)164 static void *recv_fn(void *data)
165 {
166 struct recv_data *rd = data;
167 pthread_mutex_t *mutex = rd->mutex;
168 struct io_uring_buf_ring *br = NULL;
169 char buf[MAX_MSG + 1];
170 struct iovec iov[MAX_IOV_COUNT];
171 struct io_uring ring;
172 int ret, sockfd;
173
174 if (rd->buf_ring && no_pbuf_ring)
175 goto out_no_ring;
176
177 init_iov(iov, rd->iov_count, buf);
178
179 ret = io_uring_queue_init(1, &ring, 0);
180 if (ret) {
181 fprintf(stderr, "queue init failed: %d\n", ret);
182 goto err;
183 }
184
185 if ((rd->buf_ring || rd->buf_select) && !rd->no_buf_add) {
186 if (rd->buf_ring) {
187 struct io_uring_buf_reg reg = { };
188 void *ptr;
189
190 if (posix_memalign(&ptr, 4096, 4096))
191 goto err;
192
193 reg.ring_addr = (unsigned long) ptr;
194 reg.ring_entries = 1;
195 reg.bgid = BUF_BGID;
196 if (io_uring_register_buf_ring(&ring, ®, 0)) {
197 no_pbuf_ring = 1;
198 goto out;
199 }
200
201 br = ptr;
202 io_uring_buf_ring_init(br);
203 io_uring_buf_ring_add(br, buf, sizeof(buf), BUF_BID,
204 io_uring_buf_ring_mask(1), 0);
205 io_uring_buf_ring_advance(br, 1);
206 } else {
207 struct io_uring_sqe *sqe;
208 struct io_uring_cqe *cqe;
209
210 sqe = io_uring_get_sqe(&ring);
211 io_uring_prep_provide_buffers(sqe, buf, sizeof(buf) -1,
212 1, BUF_BGID, BUF_BID);
213 sqe->user_data = ++ud;
214 ret = io_uring_submit(&ring);
215 if (ret != 1) {
216 fprintf(stderr, "submit ret=%d\n", ret);
217 goto err;
218 }
219
220 ret = io_uring_wait_cqe(&ring, &cqe);
221 if (ret) {
222 fprintf(stderr, "wait_cqe=%d\n", ret);
223 goto err;
224 }
225 ret = cqe->res;
226 io_uring_cqe_seen(&ring, cqe);
227 if (ret == -EINVAL) {
228 fprintf(stdout, "PROVIDE_BUFFERS not supported, skip\n");
229 goto out;
230 } else if (ret < 0) {
231 fprintf(stderr, "PROVIDER_BUFFERS %d\n", ret);
232 goto err;
233 }
234 }
235 }
236
237 ret = recv_prep(&ring, &sockfd, iov, rd->iov_count,
238 (rd->buf_ring || rd->buf_select) ? BUF_BGID : 0,
239 rd->async);
240 if (ret) {
241 fprintf(stderr, "recv_prep failed: %d\n", ret);
242 goto err;
243 }
244
245 pthread_mutex_unlock(mutex);
246 ret = do_recvmsg(&ring, buf, rd);
247 close(sockfd);
248
249 io_uring_queue_exit(&ring);
250 if (br)
251 free(br);
252 err:
253 return (void *)(intptr_t)ret;
254 out:
255 io_uring_queue_exit(&ring);
256 out_no_ring:
257 pthread_mutex_unlock(mutex);
258 if (br)
259 free(br);
260 return NULL;
261 }
262
do_sendmsg(void)263 static int do_sendmsg(void)
264 {
265 struct sockaddr_in saddr;
266 struct iovec iov = {
267 .iov_base = str,
268 .iov_len = sizeof(str),
269 };
270 struct msghdr msg;
271 struct io_uring ring;
272 struct io_uring_cqe *cqe;
273 struct io_uring_sqe *sqe;
274 int sockfd, ret;
275
276 ret = io_uring_queue_init(1, &ring, 0);
277 if (ret) {
278 fprintf(stderr, "queue init failed: %d\n", ret);
279 return 1;
280 }
281
282 memset(&saddr, 0, sizeof(saddr));
283 saddr.sin_family = AF_INET;
284 saddr.sin_port = htons(PORT);
285 inet_pton(AF_INET, HOST, &saddr.sin_addr);
286
287 memset(&msg, 0, sizeof(msg));
288 msg.msg_name = &saddr;
289 msg.msg_namelen = sizeof(struct sockaddr_in);
290 msg.msg_iov = &iov;
291 msg.msg_iovlen = 1;
292
293 sockfd = socket(AF_INET, SOCK_DGRAM, 0);
294 if (sockfd < 0) {
295 perror("socket");
296 return 1;
297 }
298
299 usleep(10000);
300
301 sqe = io_uring_get_sqe(&ring);
302 io_uring_prep_sendmsg(sqe, sockfd, &msg, 0);
303 sqe->user_data = ++ud;
304
305 ret = io_uring_submit(&ring);
306 if (ret <= 0) {
307 fprintf(stderr, "submit failed: %d\n", ret);
308 goto err;
309 }
310
311 ret = io_uring_wait_cqe(&ring, &cqe);
312 if (cqe->res < 0) {
313 fprintf(stderr, "%s: failed cqe: %d\n", __FUNCTION__, cqe->res);
314 goto err;
315 }
316
317 close(sockfd);
318 return 0;
319 err:
320 close(sockfd);
321 return 1;
322 }
323
test(int buf_select,int buf_ring,int no_buf_add,int iov_count,int async)324 static int test(int buf_select, int buf_ring, int no_buf_add, int iov_count,
325 int async)
326 {
327 struct recv_data rd;
328 pthread_mutexattr_t attr;
329 pthread_t recv_thread;
330 pthread_mutex_t mutex;
331 int ret;
332 void *retval;
333
334 if (buf_select || buf_ring)
335 assert(iov_count == 1);
336
337 pthread_mutexattr_init(&attr);
338 pthread_mutexattr_setpshared(&attr, 1);
339 pthread_mutex_init(&mutex, &attr);
340 pthread_mutex_lock(&mutex);
341
342 rd.mutex = &mutex;
343 rd.buf_select = buf_select;
344 rd.buf_ring = buf_ring;
345 rd.no_buf_add = no_buf_add;
346 rd.iov_count = iov_count;
347 rd.async = async;
348 ret = pthread_create(&recv_thread, NULL, recv_fn, &rd);
349 if (ret) {
350 pthread_mutex_unlock(&mutex);
351 fprintf(stderr, "Thread create failed\n");
352 return 1;
353 }
354
355 pthread_mutex_lock(&mutex);
356 do_sendmsg();
357 pthread_join(recv_thread, &retval);
358 ret = (intptr_t)retval;
359
360 return ret;
361 }
362
main(int argc,char * argv[])363 int main(int argc, char *argv[])
364 {
365 int ret;
366
367 if (argc > 1)
368 return 0;
369
370 ret = test(0, 0, 0, 1, 0);
371 if (ret) {
372 fprintf(stderr, "send_recvmsg 0 0 0 1 0 failed\n");
373 return 1;
374 }
375
376 ret = test(0, 0, 0, 10, 0);
377 if (ret) {
378 fprintf(stderr, "send_recvmsg multi iov failed\n");
379 return 1;
380 }
381
382 ret = test(1, 0, 0, 1, 0);
383 if (ret) {
384 fprintf(stderr, "send_recvmsg 1 0 0 1 0 failed\n");
385 return 1;
386 }
387
388 ret = test(1, 0, 1, 1, 0);
389 if (ret) {
390 fprintf(stderr, "send_recvmsg 1 0 1 1 0 failed\n");
391 return 1;
392 }
393
394 ret = test(0, 1, 0, 1, 0);
395 if (ret) {
396 fprintf(stderr, "send_recvmsg 0 1 0 1 0 failed\n");
397 return 1;
398 }
399
400 ret = test(1, 1, 0, 1, 0);
401 if (ret) {
402 fprintf(stderr, "send_recvmsg 1 1 0 1 0 failed\n");
403 return 1;
404 }
405
406 ret = test(1, 1, 1, 1, 0);
407 if (ret) {
408 fprintf(stderr, "send_recvmsg 1 1 1 1 0 failed\n");
409 return 1;
410 }
411
412 ret = test(0, 0, 0, 1, 1);
413 if (ret) {
414 fprintf(stderr, "send_recvmsg async 0 0 0 1 1 failed\n");
415 return 1;
416 }
417
418 ret = test(0, 0, 0, 10, 1);
419 if (ret) {
420 fprintf(stderr, "send_recvmsg async multi iov failed\n");
421 return 1;
422 }
423
424 ret = test(1, 0, 0, 1, 1);
425 if (ret) {
426 fprintf(stderr, "send_recvmsg async 1 0 0 1 1 failed\n");
427 return 1;
428 }
429
430 ret = test(1, 0, 1, 1, 1);
431 if (ret) {
432 fprintf(stderr, "send_recvmsg async 1 0 1 1 1 failed\n");
433 return 1;
434 }
435
436 ret = test(0, 1, 0, 1, 1);
437 if (ret) {
438 fprintf(stderr, "send_recvmsg async 0 1 0 1 1 failed\n");
439 return 1;
440 }
441
442 ret = test(1, 1, 0, 1, 1);
443 if (ret) {
444 fprintf(stderr, "send_recvmsg async 1 1 0 1 1 failed\n");
445 return 1;
446 }
447
448 ret = test(1, 1, 1, 1, 1);
449 if (ret) {
450 fprintf(stderr, "send_recvmsg async 1 1 1 1 1 failed\n");
451 return 1;
452 }
453
454 return 0;
455 }
456