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 br = io_uring_setup_buf_ring(&ring, 1, BUF_BGID, 0, &ret);
188 if (!br) {
189 no_pbuf_ring = 1;
190 goto out;
191 }
192 io_uring_buf_ring_add(br, buf, sizeof(buf), BUF_BID,
193 io_uring_buf_ring_mask(1), 0);
194 io_uring_buf_ring_advance(br, 1);
195 } else {
196 struct io_uring_sqe *sqe;
197 struct io_uring_cqe *cqe;
198
199 sqe = io_uring_get_sqe(&ring);
200 io_uring_prep_provide_buffers(sqe, buf, sizeof(buf) -1,
201 1, BUF_BGID, BUF_BID);
202 sqe->user_data = ++ud;
203 ret = io_uring_submit(&ring);
204 if (ret != 1) {
205 fprintf(stderr, "submit ret=%d\n", ret);
206 goto err;
207 }
208
209 ret = io_uring_wait_cqe(&ring, &cqe);
210 if (ret) {
211 fprintf(stderr, "wait_cqe=%d\n", ret);
212 goto err;
213 }
214 ret = cqe->res;
215 io_uring_cqe_seen(&ring, cqe);
216 if (ret == -EINVAL) {
217 fprintf(stdout, "PROVIDE_BUFFERS not supported, skip\n");
218 goto out;
219 } else if (ret < 0) {
220 fprintf(stderr, "PROVIDER_BUFFERS %d\n", ret);
221 goto err;
222 }
223 }
224 }
225
226 ret = recv_prep(&ring, &sockfd, iov, rd->iov_count,
227 (rd->buf_ring || rd->buf_select) ? BUF_BGID : 0,
228 rd->async);
229 if (ret) {
230 fprintf(stderr, "recv_prep failed: %d\n", ret);
231 goto err;
232 }
233
234 pthread_mutex_unlock(mutex);
235 ret = do_recvmsg(&ring, buf, rd);
236 close(sockfd);
237
238 if (br)
239 io_uring_free_buf_ring(&ring, br, 1, BUF_BGID);
240 io_uring_queue_exit(&ring);
241 err:
242 return (void *)(intptr_t)ret;
243 out:
244 io_uring_queue_exit(&ring);
245 out_no_ring:
246 pthread_mutex_unlock(mutex);
247 if (br)
248 io_uring_free_buf_ring(&ring, br, 1, BUF_BGID);
249 return NULL;
250 }
251
do_sendmsg(void)252 static int do_sendmsg(void)
253 {
254 struct sockaddr_in saddr;
255 struct iovec iov = {
256 .iov_base = str,
257 .iov_len = sizeof(str),
258 };
259 struct msghdr msg;
260 struct io_uring ring;
261 struct io_uring_cqe *cqe;
262 struct io_uring_sqe *sqe;
263 int sockfd, ret;
264
265 ret = io_uring_queue_init(1, &ring, 0);
266 if (ret) {
267 fprintf(stderr, "queue init failed: %d\n", ret);
268 return 1;
269 }
270
271 memset(&saddr, 0, sizeof(saddr));
272 saddr.sin_family = AF_INET;
273 saddr.sin_port = htons(PORT);
274 inet_pton(AF_INET, HOST, &saddr.sin_addr);
275
276 memset(&msg, 0, sizeof(msg));
277 msg.msg_name = &saddr;
278 msg.msg_namelen = sizeof(struct sockaddr_in);
279 msg.msg_iov = &iov;
280 msg.msg_iovlen = 1;
281
282 sockfd = socket(AF_INET, SOCK_DGRAM, 0);
283 if (sockfd < 0) {
284 perror("socket");
285 return 1;
286 }
287
288 usleep(10000);
289
290 sqe = io_uring_get_sqe(&ring);
291 io_uring_prep_sendmsg(sqe, sockfd, &msg, 0);
292 sqe->user_data = ++ud;
293
294 ret = io_uring_submit(&ring);
295 if (ret <= 0) {
296 fprintf(stderr, "submit failed: %d\n", ret);
297 goto err;
298 }
299
300 ret = io_uring_wait_cqe(&ring, &cqe);
301 if (cqe->res < 0) {
302 fprintf(stderr, "%s: failed cqe: %d\n", __FUNCTION__, cqe->res);
303 goto err;
304 }
305
306 close(sockfd);
307 return 0;
308 err:
309 close(sockfd);
310 return 1;
311 }
312
test(int buf_select,int buf_ring,int no_buf_add,int iov_count,int async)313 static int test(int buf_select, int buf_ring, int no_buf_add, int iov_count,
314 int async)
315 {
316 struct recv_data rd;
317 pthread_mutexattr_t attr;
318 pthread_t recv_thread;
319 pthread_mutex_t mutex;
320 int ret;
321 void *retval;
322
323 if (buf_select || buf_ring)
324 assert(iov_count == 1);
325
326 pthread_mutexattr_init(&attr);
327 pthread_mutexattr_setpshared(&attr, 1);
328 pthread_mutex_init(&mutex, &attr);
329 pthread_mutex_lock(&mutex);
330
331 rd.mutex = &mutex;
332 rd.buf_select = buf_select;
333 rd.buf_ring = buf_ring;
334 rd.no_buf_add = no_buf_add;
335 rd.iov_count = iov_count;
336 rd.async = async;
337 ret = pthread_create(&recv_thread, NULL, recv_fn, &rd);
338 if (ret) {
339 pthread_mutex_unlock(&mutex);
340 fprintf(stderr, "Thread create failed\n");
341 return 1;
342 }
343
344 pthread_mutex_lock(&mutex);
345 do_sendmsg();
346 pthread_join(recv_thread, &retval);
347 ret = (intptr_t)retval;
348
349 return ret;
350 }
351
main(int argc,char * argv[])352 int main(int argc, char *argv[])
353 {
354 int ret;
355
356 if (argc > 1)
357 return 0;
358
359 ret = test(0, 0, 0, 1, 0);
360 if (ret) {
361 fprintf(stderr, "send_recvmsg 0 0 0 1 0 failed\n");
362 return 1;
363 }
364
365 ret = test(0, 0, 0, 10, 0);
366 if (ret) {
367 fprintf(stderr, "send_recvmsg multi iov failed\n");
368 return 1;
369 }
370
371 ret = test(1, 0, 0, 1, 0);
372 if (ret) {
373 fprintf(stderr, "send_recvmsg 1 0 0 1 0 failed\n");
374 return 1;
375 }
376
377 ret = test(1, 0, 1, 1, 0);
378 if (ret) {
379 fprintf(stderr, "send_recvmsg 1 0 1 1 0 failed\n");
380 return 1;
381 }
382
383 ret = test(0, 1, 0, 1, 0);
384 if (ret) {
385 fprintf(stderr, "send_recvmsg 0 1 0 1 0 failed\n");
386 return 1;
387 }
388
389 ret = test(1, 1, 0, 1, 0);
390 if (ret) {
391 fprintf(stderr, "send_recvmsg 1 1 0 1 0 failed\n");
392 return 1;
393 }
394
395 ret = test(1, 1, 1, 1, 0);
396 if (ret) {
397 fprintf(stderr, "send_recvmsg 1 1 1 1 0 failed\n");
398 return 1;
399 }
400
401 ret = test(0, 0, 0, 1, 1);
402 if (ret) {
403 fprintf(stderr, "send_recvmsg async 0 0 0 1 1 failed\n");
404 return 1;
405 }
406
407 ret = test(0, 0, 0, 10, 1);
408 if (ret) {
409 fprintf(stderr, "send_recvmsg async multi iov failed\n");
410 return 1;
411 }
412
413 ret = test(1, 0, 0, 1, 1);
414 if (ret) {
415 fprintf(stderr, "send_recvmsg async 1 0 0 1 1 failed\n");
416 return 1;
417 }
418
419 ret = test(1, 0, 1, 1, 1);
420 if (ret) {
421 fprintf(stderr, "send_recvmsg async 1 0 1 1 1 failed\n");
422 return 1;
423 }
424
425 ret = test(0, 1, 0, 1, 1);
426 if (ret) {
427 fprintf(stderr, "send_recvmsg async 0 1 0 1 1 failed\n");
428 return 1;
429 }
430
431 ret = test(1, 1, 0, 1, 1);
432 if (ret) {
433 fprintf(stderr, "send_recvmsg async 1 1 0 1 1 failed\n");
434 return 1;
435 }
436
437 ret = test(1, 1, 1, 1, 1);
438 if (ret) {
439 fprintf(stderr, "send_recvmsg async 1 1 1 1 1 failed\n");
440 return 1;
441 }
442
443 return 0;
444 }
445