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 #define MAX_MSG 128
21
22 #define PORT 10200
23 #define HOST "127.0.0.1"
24
25 #define BUF_BGID 10
26 #define BUF_BID 89
27
28 #define MAX_IOV_COUNT 10
29
recv_prep(struct io_uring * ring,struct iovec iov[],int iov_count,int bgid)30 static int recv_prep(struct io_uring *ring, struct iovec iov[], int iov_count,
31 int bgid)
32 {
33 struct sockaddr_in saddr;
34 struct msghdr msg;
35 struct io_uring_sqe *sqe;
36 int sockfd, ret;
37 int val = 1;
38
39 memset(&saddr, 0, sizeof(saddr));
40 saddr.sin_family = AF_INET;
41 saddr.sin_addr.s_addr = htonl(INADDR_ANY);
42 saddr.sin_port = htons(PORT);
43
44 sockfd = socket(AF_INET, SOCK_DGRAM, 0);
45 if (sockfd < 0) {
46 perror("socket");
47 return 1;
48 }
49
50 val = 1;
51 setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &val, sizeof(val));
52 setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
53
54 ret = bind(sockfd, (struct sockaddr *)&saddr, sizeof(saddr));
55 if (ret < 0) {
56 perror("bind");
57 goto err;
58 }
59
60 sqe = io_uring_get_sqe(ring);
61 if (!sqe) {
62 fprintf(stderr, "io_uring_get_sqe failed\n");
63 return 1;
64 }
65
66 io_uring_prep_recvmsg(sqe, sockfd, &msg, 0);
67 if (bgid) {
68 iov->iov_base = NULL;
69 sqe->flags |= IOSQE_BUFFER_SELECT;
70 sqe->buf_group = bgid;
71 iov_count = 1;
72 }
73 memset(&msg, 0, sizeof(msg));
74 msg.msg_namelen = sizeof(struct sockaddr_in);
75 msg.msg_iov = iov;
76 msg.msg_iovlen = iov_count;
77
78 ret = io_uring_submit(ring);
79 if (ret <= 0) {
80 fprintf(stderr, "submit failed: %d\n", ret);
81 goto err;
82 }
83
84 close(sockfd);
85 return 0;
86 err:
87 close(sockfd);
88 return 1;
89 }
90
91 struct recv_data {
92 pthread_mutex_t *mutex;
93 int buf_select;
94 int no_buf_add;
95 int iov_count;
96 };
97
do_recvmsg(struct io_uring * ring,char buf[MAX_MSG+1],struct recv_data * rd)98 static int do_recvmsg(struct io_uring *ring, char buf[MAX_MSG + 1],
99 struct recv_data *rd)
100 {
101 struct io_uring_cqe *cqe;
102 int ret;
103
104 ret = io_uring_wait_cqe(ring, &cqe);
105 if (ret) {
106 fprintf(stdout, "wait_cqe: %d\n", ret);
107 goto err;
108 }
109 if (cqe->res < 0) {
110 if (rd->no_buf_add && rd->buf_select)
111 return 0;
112 fprintf(stderr, "%s: failed cqe: %d\n", __FUNCTION__, cqe->res);
113 goto err;
114 }
115 if (cqe->flags) {
116 int bid = cqe->flags >> 16;
117 if (bid != BUF_BID)
118 fprintf(stderr, "Buffer ID mismatch %d\n", bid);
119 }
120
121 if (rd->no_buf_add && rd->buf_select) {
122 fprintf(stderr, "Expected -ENOBUFS: %d\n", cqe->res);
123 goto err;
124 }
125
126 if (cqe->res -1 != strlen(str)) {
127 fprintf(stderr, "got wrong length: %d/%d\n", cqe->res,
128 (int) strlen(str) + 1);
129 goto err;
130 }
131
132 if (strncmp(str, buf, MAX_MSG + 1)) {
133 fprintf(stderr, "string mismatch\n");
134 goto err;
135 }
136
137 return 0;
138 err:
139 return 1;
140 }
141
init_iov(struct iovec iov[MAX_IOV_COUNT],int iov_to_use,char buf[MAX_MSG+1])142 static void init_iov(struct iovec iov[MAX_IOV_COUNT], int iov_to_use,
143 char buf[MAX_MSG + 1])
144 {
145 int i, last_idx = iov_to_use - 1;
146
147 assert(0 < iov_to_use && iov_to_use <= MAX_IOV_COUNT);
148 for (i = 0; i < last_idx; ++i) {
149 iov[i].iov_base = buf + i;
150 iov[i].iov_len = 1;
151 }
152
153 iov[last_idx].iov_base = buf + last_idx;
154 iov[last_idx].iov_len = MAX_MSG - last_idx;
155 }
156
recv_fn(void * data)157 static void *recv_fn(void *data)
158 {
159 struct recv_data *rd = data;
160 pthread_mutex_t *mutex = rd->mutex;
161 char buf[MAX_MSG + 1];
162 struct iovec iov[MAX_IOV_COUNT];
163 struct io_uring_sqe *sqe;
164 struct io_uring_cqe *cqe;
165 struct io_uring ring;
166 int ret;
167
168 init_iov(iov, rd->iov_count, buf);
169
170 ret = io_uring_queue_init(1, &ring, 0);
171 if (ret) {
172 fprintf(stderr, "queue init failed: %d\n", ret);
173 goto err;
174 }
175
176 if (rd->buf_select && !rd->no_buf_add) {
177 sqe = io_uring_get_sqe(&ring);
178 io_uring_prep_provide_buffers(sqe, buf, sizeof(buf) -1, 1,
179 BUF_BGID, BUF_BID);
180 ret = io_uring_submit(&ring);
181 if (ret != 1) {
182 fprintf(stderr, "submit ret=%d\n", ret);
183 goto err;
184 }
185
186 ret = io_uring_wait_cqe(&ring, &cqe);
187 if (ret) {
188 fprintf(stderr, "wait_cqe=%d\n", ret);
189 goto err;
190 }
191 ret = cqe->res;
192 io_uring_cqe_seen(&ring, cqe);
193 if (ret == -EINVAL) {
194 fprintf(stdout, "PROVIDE_BUFFERS not supported, skip\n");
195 goto out;
196 goto err;
197 } else if (ret < 0) {
198 fprintf(stderr, "PROVIDER_BUFFERS %d\n", ret);
199 goto err;
200 }
201 }
202
203 ret = recv_prep(&ring, iov, rd->iov_count, rd->buf_select ? BUF_BGID : 0);
204 if (ret) {
205 fprintf(stderr, "recv_prep failed: %d\n", ret);
206 goto err;
207 }
208
209 pthread_mutex_unlock(mutex);
210 ret = do_recvmsg(&ring, buf, rd);
211
212 io_uring_queue_exit(&ring);
213
214 err:
215 return (void *)(intptr_t)ret;
216 out:
217 pthread_mutex_unlock(mutex);
218 io_uring_queue_exit(&ring);
219 return NULL;
220 }
221
do_sendmsg(void)222 static int do_sendmsg(void)
223 {
224 struct sockaddr_in saddr;
225 struct iovec iov = {
226 .iov_base = str,
227 .iov_len = sizeof(str),
228 };
229 struct msghdr msg;
230 struct io_uring ring;
231 struct io_uring_cqe *cqe;
232 struct io_uring_sqe *sqe;
233 int sockfd, ret;
234
235 ret = io_uring_queue_init(1, &ring, 0);
236 if (ret) {
237 fprintf(stderr, "queue init failed: %d\n", ret);
238 return 1;
239 }
240
241 memset(&saddr, 0, sizeof(saddr));
242 saddr.sin_family = AF_INET;
243 saddr.sin_port = htons(PORT);
244 inet_pton(AF_INET, HOST, &saddr.sin_addr);
245
246 memset(&msg, 0, sizeof(msg));
247 msg.msg_name = &saddr;
248 msg.msg_namelen = sizeof(struct sockaddr_in);
249 msg.msg_iov = &iov;
250 msg.msg_iovlen = 1;
251
252 sockfd = socket(AF_INET, SOCK_DGRAM, 0);
253 if (sockfd < 0) {
254 perror("socket");
255 return 1;
256 }
257
258 sqe = io_uring_get_sqe(&ring);
259 io_uring_prep_sendmsg(sqe, sockfd, &msg, 0);
260
261 ret = io_uring_submit(&ring);
262 if (ret <= 0) {
263 fprintf(stderr, "submit failed: %d\n", ret);
264 goto err;
265 }
266
267 ret = io_uring_wait_cqe(&ring, &cqe);
268 if (cqe->res < 0) {
269 fprintf(stderr, "%s: failed cqe: %d\n", __FUNCTION__, cqe->res);
270 goto err;
271 }
272
273 close(sockfd);
274 return 0;
275 err:
276 close(sockfd);
277 return 1;
278 }
279
test(int buf_select,int no_buf_add,int iov_count)280 static int test(int buf_select, int no_buf_add, int iov_count)
281 {
282 struct recv_data rd;
283 pthread_mutexattr_t attr;
284 pthread_t recv_thread;
285 pthread_mutex_t mutex;
286 int ret;
287 void *retval;
288
289 pthread_mutexattr_init(&attr);
290 pthread_mutexattr_setpshared(&attr, 1);
291 pthread_mutex_init(&mutex, &attr);
292 pthread_mutex_lock(&mutex);
293
294 rd.mutex = &mutex;
295 rd.buf_select = buf_select;
296 rd.no_buf_add = no_buf_add;
297 rd.iov_count = iov_count;
298 ret = pthread_create(&recv_thread, NULL, recv_fn, &rd);
299 if (ret) {
300 pthread_mutex_unlock(&mutex);
301 fprintf(stderr, "Thread create failed\n");
302 return 1;
303 }
304
305 pthread_mutex_lock(&mutex);
306 do_sendmsg();
307 pthread_join(recv_thread, &retval);
308 ret = (int)(intptr_t)retval;
309
310 return ret;
311 }
312
main(int argc,char * argv[])313 int main(int argc, char *argv[])
314 {
315 int ret;
316
317 if (argc > 1)
318 return 0;
319
320 ret = test(0, 0, 1);
321 if (ret) {
322 fprintf(stderr, "send_recvmsg 0 failed\n");
323 return 1;
324 }
325
326 ret = test(0, 0, 10);
327 if (ret) {
328 fprintf(stderr, "send_recvmsg multi iov failed\n");
329 return 1;
330 }
331
332 ret = test(1, 0, 1);
333 if (ret) {
334 fprintf(stderr, "send_recvmsg 1 0 failed\n");
335 return 1;
336 }
337
338 ret = test(1, 1, 1);
339 if (ret) {
340 fprintf(stderr, "send_recvmsg 1 1 failed\n");
341 return 1;
342 }
343
344 return 0;
345 }
346