1 /* SPDX-License-Identifier: MIT */
2 /*
3 * Test case for socket read/write through IORING_OP_READV and
4 * IORING_OP_WRITEV, using both TCP and sockets and blocking and
5 * non-blocking IO.
6 *
7 * Heavily based on a test case from Hrvoje Zeba <[email protected]>
8 */
9 #include <stdio.h>
10 #include <stdlib.h>
11 #include <stdint.h>
12 #include <assert.h>
13
14 #include <pthread.h>
15 #include <errno.h>
16 #include <fcntl.h>
17 #include <unistd.h>
18 #include <sys/socket.h>
19 #include <sys/un.h>
20 #include <netinet/tcp.h>
21 #include <netinet/in.h>
22 #include <arpa/inet.h>
23
24 #include "liburing.h"
25
26 #define RECV_BUFF_SIZE 2
27 #define SEND_BUFF_SIZE 3
28
29 #define PORT 0x1234
30
31 struct params {
32 int tcp;
33 int non_blocking;
34 };
35
36 pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
37 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
38 int rcv_ready = 0;
39
set_rcv_ready(void)40 static void set_rcv_ready(void)
41 {
42 pthread_mutex_lock(&mutex);
43
44 rcv_ready = 1;
45 pthread_cond_signal(&cond);
46
47 pthread_mutex_unlock(&mutex);
48 }
49
wait_for_rcv_ready(void)50 static void wait_for_rcv_ready(void)
51 {
52 pthread_mutex_lock(&mutex);
53
54 while (!rcv_ready)
55 pthread_cond_wait(&cond, &mutex);
56
57 pthread_mutex_unlock(&mutex);
58 }
59
rcv(void * arg)60 static void *rcv(void *arg)
61 {
62 struct params *p = arg;
63 int s0;
64 int res;
65
66 if (p->tcp) {
67 int val = 1;
68
69
70 s0 = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, IPPROTO_TCP);
71 res = setsockopt(s0, SOL_SOCKET, SO_REUSEPORT, &val, sizeof(val));
72 assert(res != -1);
73 res = setsockopt(s0, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));
74 assert(res != -1);
75
76 struct sockaddr_in addr;
77
78 addr.sin_family = AF_INET;
79 addr.sin_port = htons(PORT);
80 addr.sin_addr.s_addr = inet_addr("127.0.0.1");
81 res = bind(s0, (struct sockaddr *) &addr, sizeof(addr));
82 assert(res != -1);
83 } else {
84 s0 = socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0);
85 assert(s0 != -1);
86
87 struct sockaddr_un addr;
88 memset(&addr, 0, sizeof(addr));
89
90 addr.sun_family = AF_UNIX;
91 memcpy(addr.sun_path, "\0sock", 6);
92 res = bind(s0, (struct sockaddr *) &addr, sizeof(addr));
93 assert(res != -1);
94 }
95 res = listen(s0, 128);
96 assert(res != -1);
97
98 set_rcv_ready();
99
100 int s1 = accept(s0, NULL, NULL);
101 assert(s1 != -1);
102
103 if (p->non_blocking) {
104 int flags = fcntl(s1, F_GETFL, 0);
105 assert(flags != -1);
106
107 flags |= O_NONBLOCK;
108 res = fcntl(s1, F_SETFL, flags);
109 assert(res != -1);
110 }
111
112 struct io_uring m_io_uring;
113 void *ret = NULL;
114
115 res = io_uring_queue_init(32, &m_io_uring, 0);
116 assert(res >= 0);
117
118 int bytes_read = 0;
119 int expected_byte = 0;
120 int done = 0;
121
122 while (!done && bytes_read != 33) {
123 char buff[RECV_BUFF_SIZE];
124 struct iovec iov;
125
126 iov.iov_base = buff;
127 iov.iov_len = sizeof(buff);
128
129 struct io_uring_sqe *sqe = io_uring_get_sqe(&m_io_uring);
130 assert(sqe != NULL);
131
132 io_uring_prep_readv(sqe, s1, &iov, 1, 0);
133
134 res = io_uring_submit(&m_io_uring);
135 assert(res != -1);
136
137 struct io_uring_cqe *cqe;
138 unsigned head;
139 unsigned count = 0;
140
141 while (!done && count != 1) {
142 io_uring_for_each_cqe(&m_io_uring, head, cqe) {
143 if (cqe->res < 0)
144 assert(cqe->res == -EAGAIN);
145 else {
146 int i;
147
148 for (i = 0; i < cqe->res; i++) {
149 if (buff[i] != expected_byte) {
150 fprintf(stderr,
151 "Received %d, wanted %d\n",
152 buff[i], expected_byte);
153 ret++;
154 done = 1;
155 }
156 expected_byte++;
157 }
158 bytes_read += cqe->res;
159 }
160
161 count++;
162 }
163
164 assert(count <= 1);
165 io_uring_cq_advance(&m_io_uring, count);
166 }
167 }
168
169 shutdown(s1, SHUT_RDWR);
170 close(s1);
171 close(s0);
172 io_uring_queue_exit(&m_io_uring);
173 return ret;
174 }
175
snd(void * arg)176 static void *snd(void *arg)
177 {
178 struct params *p = arg;
179 int s0;
180 int ret;
181
182 wait_for_rcv_ready();
183
184 if (p->tcp) {
185 int val = 1;
186
187 s0 = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, IPPROTO_TCP);
188 ret = setsockopt(s0, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val));
189 assert(ret != -1);
190
191 struct sockaddr_in addr;
192
193 addr.sin_family = AF_INET;
194 addr.sin_port = htons(PORT);
195 addr.sin_addr.s_addr = inet_addr("127.0.0.1");
196 ret = connect(s0, (struct sockaddr*) &addr, sizeof(addr));
197 assert(ret != -1);
198 } else {
199 s0 = socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0);
200 assert(s0 != -1);
201
202 struct sockaddr_un addr;
203 memset(&addr, 0, sizeof(addr));
204
205 addr.sun_family = AF_UNIX;
206 memcpy(addr.sun_path, "\0sock", 6);
207 ret = connect(s0, (struct sockaddr*) &addr, sizeof(addr));
208 assert(ret != -1);
209 }
210
211 if (p->non_blocking) {
212 int flags = fcntl(s0, F_GETFL, 0);
213 assert(flags != -1);
214
215 flags |= O_NONBLOCK;
216 ret = fcntl(s0, F_SETFL, flags);
217 assert(ret != -1);
218 }
219
220 struct io_uring m_io_uring;
221
222 ret = io_uring_queue_init(32, &m_io_uring, 0);
223 assert(ret >= 0);
224
225 int bytes_written = 0;
226 int done = 0;
227
228 while (!done && bytes_written != 33) {
229 char buff[SEND_BUFF_SIZE];
230 int i;
231
232 for (i = 0; i < SEND_BUFF_SIZE; i++)
233 buff[i] = i + bytes_written;
234
235 struct iovec iov;
236
237 iov.iov_base = buff;
238 iov.iov_len = sizeof(buff);
239
240 struct io_uring_sqe *sqe = io_uring_get_sqe(&m_io_uring);
241 assert(sqe != NULL);
242
243 io_uring_prep_writev(sqe, s0, &iov, 1, 0);
244
245 ret = io_uring_submit(&m_io_uring);
246 assert(ret != -1);
247
248 struct io_uring_cqe *cqe;
249 unsigned head;
250 unsigned count = 0;
251
252 while (!done && count != 1) {
253 io_uring_for_each_cqe(&m_io_uring, head, cqe) {
254 if (cqe->res < 0) {
255 if (cqe->res == -EPIPE) {
256 done = 1;
257 break;
258 }
259 assert(cqe->res == -EAGAIN);
260 } else {
261 bytes_written += cqe->res;
262 }
263
264 count++;
265 }
266
267 assert(count <= 1);
268 io_uring_cq_advance(&m_io_uring, count);
269 }
270 usleep(100000);
271 }
272
273 shutdown(s0, SHUT_RDWR);
274 close(s0);
275 io_uring_queue_exit(&m_io_uring);
276 return NULL;
277 }
278
main(int argc,char * argv[])279 int main(int argc, char *argv[])
280 {
281 struct params p;
282 pthread_t t1, t2;
283 void *res1, *res2;
284 int i, exit_val = 0;
285
286 if (argc > 1)
287 return 0;
288
289 for (i = 0; i < 4; i++) {
290 p.tcp = i & 1;
291 p.non_blocking = (i & 2) >> 1;
292
293 rcv_ready = 0;
294
295 pthread_create(&t1, NULL, rcv, &p);
296 pthread_create(&t2, NULL, snd, &p);
297 pthread_join(t1, &res1);
298 pthread_join(t2, &res2);
299 if (res1 || res2) {
300 fprintf(stderr, "Failed tcp=%d, non_blocking=%d\n", p.tcp, p.non_blocking);
301 exit_val = 1;
302 }
303 }
304
305 return exit_val;
306 }
307