1 /* SPDX-License-Identifier: MIT */
2 /*
3 * Simple test case showing using send and recv through io_uring
4 */
5 #include <errno.h>
6 #include <stdio.h>
7 #include <stdlib.h>
8 #include <string.h>
9 #include <unistd.h>
10 #include <arpa/inet.h>
11 #include <sys/types.h>
12 #include <sys/socket.h>
13 #include <pthread.h>
14
15 #include "liburing.h"
16 #include "helpers.h"
17
18 static char str[] = "This is a test of send and recv over io_uring!";
19
20 #define MAX_MSG 128
21
22 #define PORT 10202
23 #define HOST "127.0.0.1"
24
recv_prep(struct io_uring * ring,struct iovec * iov,int * sock,int registerfiles,int async,int provide)25 static int recv_prep(struct io_uring *ring, struct iovec *iov, int *sock,
26 int registerfiles, int async, int provide)
27 {
28 struct sockaddr_in saddr;
29 struct io_uring_sqe *sqe;
30 int sockfd, ret, val, use_fd;
31
32 memset(&saddr, 0, sizeof(saddr));
33 saddr.sin_family = AF_INET;
34 saddr.sin_addr.s_addr = htonl(INADDR_ANY);
35 saddr.sin_port = htons(PORT);
36
37 sockfd = socket(AF_INET, SOCK_DGRAM, 0);
38 if (sockfd < 0) {
39 perror("socket");
40 return 1;
41 }
42
43 val = 1;
44 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
45
46 ret = bind(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));
47 if (ret < 0) {
48 perror("bind");
49 goto err;
50 }
51
52 if (registerfiles) {
53 ret = io_uring_register_files(ring, &sockfd, 1);
54 if (ret) {
55 fprintf(stderr, "file reg failed\n");
56 goto err;
57 }
58 use_fd = 0;
59 } else {
60 use_fd = sockfd;
61 }
62
63 sqe = io_uring_get_sqe(ring);
64 io_uring_prep_recv(sqe, use_fd, iov->iov_base, iov->iov_len, 0);
65 if (registerfiles)
66 sqe->flags |= IOSQE_FIXED_FILE;
67 if (async)
68 sqe->flags |= IOSQE_ASYNC;
69 if (provide)
70 sqe->flags |= IOSQE_BUFFER_SELECT;
71 sqe->user_data = 2;
72
73 ret = io_uring_submit(ring);
74 if (ret <= 0) {
75 fprintf(stderr, "submit failed: %d\n", ret);
76 goto err;
77 }
78
79 *sock = sockfd;
80 return 0;
81 err:
82 close(sockfd);
83 return 1;
84 }
85
do_recv(struct io_uring * ring,struct iovec * iov,int enobufs)86 static int do_recv(struct io_uring *ring, struct iovec *iov, int enobufs)
87 {
88 struct io_uring_cqe *cqe;
89 int ret;
90
91 ret = io_uring_wait_cqe(ring, &cqe);
92 if (ret) {
93 fprintf(stdout, "wait_cqe: %d\n", ret);
94 return 1;
95 }
96 if (cqe->res == -EINVAL) {
97 fprintf(stdout, "recv not supported, skipping\n");
98 goto out;
99 }
100 if (cqe->res == -ENOBUFS && enobufs) {
101 if (cqe->flags & IORING_CQE_F_SOCK_NONEMPTY) {
102 fprintf(stdout, "NONEMPTY set on -ENOBUFS\n");
103 goto err;
104 }
105 goto out;
106 }
107 if (cqe->res < 0) {
108 fprintf(stderr, "failed cqe: %d\n", cqe->res);
109 goto err;
110 }
111
112 if (cqe->res -1 != strlen(str)) {
113 fprintf(stderr, "got wrong length: %d/%d\n", cqe->res,
114 (int) strlen(str) + 1);
115 goto err;
116 }
117
118 if (strcmp(str, iov->iov_base)) {
119 fprintf(stderr, "string mismatch\n");
120 goto err;
121 }
122
123 out:
124 io_uring_cqe_seen(ring, cqe);
125 return 0;
126 err:
127 io_uring_cqe_seen(ring, cqe);
128 return 1;
129 }
130
131 struct recv_data {
132 pthread_mutex_t mutex;
133 int use_sqthread;
134 int registerfiles;
135 int async;
136 int provide;
137 };
138
recv_fn(void * data)139 static void *recv_fn(void *data)
140 {
141 struct recv_data *rd = data;
142 char buf[MAX_MSG + 1];
143 struct iovec iov = {
144 .iov_base = buf,
145 .iov_len = sizeof(buf) - 1,
146 };
147 struct io_uring_params p = { };
148 struct io_uring ring;
149 int ret, sock;
150
151 if (rd->use_sqthread)
152 p.flags = IORING_SETUP_SQPOLL;
153 ret = t_create_ring_params(1, &ring, &p);
154 if (ret == T_SETUP_SKIP) {
155 pthread_mutex_unlock(&rd->mutex);
156 ret = 0;
157 goto err;
158 } else if (ret < 0) {
159 pthread_mutex_unlock(&rd->mutex);
160 goto err;
161 }
162
163 if (rd->use_sqthread && !rd->registerfiles) {
164 if (!(p.features & IORING_FEAT_SQPOLL_NONFIXED)) {
165 fprintf(stdout, "Non-registered SQPOLL not available, skipping\n");
166 pthread_mutex_unlock(&rd->mutex);
167 goto err;
168 }
169 }
170
171 ret = recv_prep(&ring, &iov, &sock, rd->registerfiles, rd->async,
172 rd->provide);
173 if (ret) {
174 fprintf(stderr, "recv_prep failed: %d\n", ret);
175 goto err;
176 }
177 pthread_mutex_unlock(&rd->mutex);
178 ret = do_recv(&ring, &iov, rd->provide);
179
180 close(sock);
181 io_uring_queue_exit(&ring);
182 err:
183 return (void *)(intptr_t)ret;
184 }
185
do_send(void)186 static int do_send(void)
187 {
188 struct sockaddr_in saddr;
189 struct iovec iov = {
190 .iov_base = str,
191 .iov_len = sizeof(str),
192 };
193 struct io_uring ring;
194 struct io_uring_cqe *cqe;
195 struct io_uring_sqe *sqe;
196 int sockfd, ret;
197
198 ret = io_uring_queue_init(1, &ring, 0);
199 if (ret) {
200 fprintf(stderr, "queue init failed: %d\n", ret);
201 return 1;
202 }
203
204 memset(&saddr, 0, sizeof(saddr));
205 saddr.sin_family = AF_INET;
206 saddr.sin_port = htons(PORT);
207 inet_pton(AF_INET, HOST, &saddr.sin_addr);
208
209 sockfd = socket(AF_INET, SOCK_DGRAM, 0);
210 if (sockfd < 0) {
211 perror("socket");
212 goto err2;
213 }
214
215 ret = connect(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));
216 if (ret < 0) {
217 perror("connect");
218 goto err;
219 }
220
221 sqe = io_uring_get_sqe(&ring);
222 io_uring_prep_send(sqe, sockfd, iov.iov_base, iov.iov_len, 0);
223 sqe->user_data = 1;
224
225 ret = io_uring_submit(&ring);
226 if (ret <= 0) {
227 fprintf(stderr, "submit failed: %d\n", ret);
228 goto err;
229 }
230
231 ret = io_uring_wait_cqe(&ring, &cqe);
232 if (cqe->res == -EINVAL) {
233 fprintf(stdout, "send not supported, skipping\n");
234 goto err;
235 }
236 if (cqe->res != iov.iov_len) {
237 fprintf(stderr, "failed cqe: %d\n", cqe->res);
238 goto err;
239 }
240
241 close(sockfd);
242 io_uring_queue_exit(&ring);
243 return 0;
244
245 err:
246 close(sockfd);
247 err2:
248 io_uring_queue_exit(&ring);
249 return 1;
250 }
251
test(int use_sqthread,int regfiles,int async,int provide)252 static int test(int use_sqthread, int regfiles, int async, int provide)
253 {
254 pthread_mutexattr_t attr;
255 pthread_t recv_thread;
256 struct recv_data rd;
257 int ret;
258 void *retval;
259
260 pthread_mutexattr_init(&attr);
261 pthread_mutexattr_setpshared(&attr, 1);
262 pthread_mutex_init(&rd.mutex, &attr);
263 pthread_mutex_lock(&rd.mutex);
264 rd.use_sqthread = use_sqthread;
265 rd.registerfiles = regfiles;
266 rd.async = async;
267 rd.provide = provide;
268
269 ret = pthread_create(&recv_thread, NULL, recv_fn, &rd);
270 if (ret) {
271 fprintf(stderr, "Thread create failed: %d\n", ret);
272 pthread_mutex_unlock(&rd.mutex);
273 return 1;
274 }
275
276 pthread_mutex_lock(&rd.mutex);
277 do_send();
278 pthread_join(recv_thread, &retval);
279 return (intptr_t)retval;
280 }
281
test_invalid(void)282 static int test_invalid(void)
283 {
284 struct io_uring ring;
285 int ret, i;
286 int fds[2];
287 struct io_uring_cqe *cqe;
288 struct io_uring_sqe *sqe;
289
290 ret = t_create_ring(8, &ring, IORING_SETUP_SUBMIT_ALL);
291 if (ret) {
292 if (ret == -EINVAL)
293 return 0;
294 return ret;
295 }
296
297 ret = t_create_socket_pair(fds, true);
298 if (ret)
299 return ret;
300
301 sqe = io_uring_get_sqe(&ring);
302 io_uring_prep_sendmsg(sqe, fds[0], NULL, MSG_WAITALL);
303 sqe->flags |= IOSQE_ASYNC;
304
305 sqe = io_uring_get_sqe(&ring);
306 io_uring_prep_recvmsg(sqe, fds[1], NULL, 0);
307 sqe->flags |= IOSQE_ASYNC;
308
309 ret = io_uring_submit_and_wait(&ring, 2);
310 if (ret != 2)
311 return ret;
312
313 for (i = 0; i < 2; i++) {
314 ret = io_uring_peek_cqe(&ring, &cqe);
315 if (ret || cqe->res != -EFAULT)
316 return -1;
317 io_uring_cqe_seen(&ring, cqe);
318 }
319
320 io_uring_queue_exit(&ring);
321 close(fds[0]);
322 close(fds[1]);
323 return 0;
324 }
325
main(int argc,char * argv[])326 int main(int argc, char *argv[])
327 {
328 int ret;
329
330 if (argc > 1)
331 return 0;
332
333 ret = test_invalid();
334 if (ret) {
335 fprintf(stderr, "test_invalid failed\n");
336 return ret;
337 }
338
339 ret = test(0, 0, 1, 1);
340 if (ret) {
341 fprintf(stderr, "test sqthread=0 1 1 failed\n");
342 return ret;
343 }
344
345 ret = test(1, 1, 1, 1);
346 if (ret) {
347 fprintf(stderr, "test sqthread=1 reg=1 1 1 failed\n");
348 return ret;
349 }
350
351 ret = test(1, 0, 1, 1);
352 if (ret) {
353 fprintf(stderr, "test sqthread=1 reg=0 1 1 failed\n");
354 return ret;
355 }
356
357 ret = test(0, 0, 0, 1);
358 if (ret) {
359 fprintf(stderr, "test sqthread=0 0 1 failed\n");
360 return ret;
361 }
362
363 ret = test(1, 1, 0, 1);
364 if (ret) {
365 fprintf(stderr, "test sqthread=1 reg=1 0 1 failed\n");
366 return ret;
367 }
368
369 ret = test(1, 0, 0, 1);
370 if (ret) {
371 fprintf(stderr, "test sqthread=1 reg=0 0 1 failed\n");
372 return ret;
373 }
374
375 ret = test(0, 0, 1, 0);
376 if (ret) {
377 fprintf(stderr, "test sqthread=0 0 1 failed\n");
378 return ret;
379 }
380
381 ret = test(1, 1, 1, 0);
382 if (ret) {
383 fprintf(stderr, "test sqthread=1 reg=1 1 0 failed\n");
384 return ret;
385 }
386
387 ret = test(1, 0, 1, 0);
388 if (ret) {
389 fprintf(stderr, "test sqthread=1 reg=0 1 0 failed\n");
390 return ret;
391 }
392
393 ret = test(0, 0, 0, 0);
394 if (ret) {
395 fprintf(stderr, "test sqthread=0 0 0 failed\n");
396 return ret;
397 }
398
399 ret = test(1, 1, 0, 0);
400 if (ret) {
401 fprintf(stderr, "test sqthread=1 reg=1 0 0 failed\n");
402 return ret;
403 }
404
405 ret = test(1, 0, 0, 0);
406 if (ret) {
407 fprintf(stderr, "test sqthread=1 reg=0 0 0 failed\n");
408 return ret;
409 }
410
411 return 0;
412 }
413