Issue time fixup for guasi/libaio
[fio.git] / engines / net.c
CommitLineData
ed92ac0c 1/*
da751ca9
JA
2 * net engine
3 *
4 * IO engine that reads/writes to/from sockets.
5 *
ed92ac0c
JA
6 */
7#include <stdio.h>
8#include <stdlib.h>
9#include <unistd.h>
10#include <errno.h>
11#include <assert.h>
12#include <netinet/in.h>
13#include <arpa/inet.h>
14#include <netdb.h>
5fdd124a 15#include <sys/poll.h>
7292056a
JA
16#include <sys/types.h>
17#include <sys/socket.h>
ed92ac0c
JA
18
19#include "../fio.h"
ed92ac0c 20
b5af8293
JA
21struct netio_data {
22 int listenfd;
23 int send_to_net;
9cce02e8
JA
24 int use_splice;
25 int pipes[2];
b5af8293
JA
26 char host[64];
27 struct sockaddr_in addr;
28};
ed92ac0c
JA
29
30static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
31{
b5af8293 32 struct netio_data *nd = td->io_ops->data;
ed92ac0c 33
7a6499da
JA
34 /*
35 * Make sure we don't see spurious reads to a receiver, and vice versa
36 */
b5af8293
JA
37 if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
38 (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
e1161c32 39 td_verror(td, EINVAL, "bad direction");
7a6499da 40 return 1;
ed92ac0c 41 }
7a6499da 42
f85ac25a 43 return 0;
ed92ac0c
JA
44}
45
5921e80c 46#ifdef FIO_HAVE_SPLICE
cd963e18 47static int splice_io_u(int fdin, int fdout, unsigned int len)
ed92ac0c 48{
9cce02e8 49 int bytes = 0;
7a6499da 50
9cce02e8 51 while (len) {
cd963e18 52 int ret = splice(fdin, NULL, fdout, NULL, len, 0);
9cce02e8
JA
53
54 if (ret < 0) {
55 if (!bytes)
56 bytes = ret;
57
58 break;
59 } else if (!ret)
60 break;
61
62 bytes += ret;
f657a2fb 63 len -= ret;
9cce02e8
JA
64 }
65
66 return bytes;
67}
68
69/*
cd963e18 70 * Receive bytes from a socket and fill them into the internal pipe
9cce02e8 71 */
cd963e18 72static int splice_in(struct thread_data *td, struct io_u *io_u)
9cce02e8
JA
73{
74 struct netio_data *nd = td->io_ops->data;
9cce02e8 75
cd963e18 76 return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
9cce02e8
JA
77}
78
79/*
cd963e18 80 * Transmit 'len' bytes from the internal pipe
9cce02e8 81 */
cd963e18
JA
82static int splice_out(struct thread_data *td, struct io_u *io_u,
83 unsigned int len)
9cce02e8
JA
84{
85 struct netio_data *nd = td->io_ops->data;
cd963e18
JA
86
87 return splice_io_u(nd->pipes[0], io_u->file->fd, len);
88}
89
90static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
91{
9cce02e8
JA
92 struct iovec iov = {
93 .iov_base = io_u->xfer_buf,
94 .iov_len = len,
95 };
96 int bytes = 0;
97
98 while (iov.iov_len) {
cd963e18 99 int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
9cce02e8
JA
100
101 if (ret < 0) {
102 if (!bytes)
103 bytes = ret;
104 break;
105 } else if (!ret)
106 break;
107
108 iov.iov_len -= ret;
cd963e18 109 iov.iov_base += ret;
f657a2fb 110 bytes += ret;
9cce02e8
JA
111 }
112
113 return bytes;
cd963e18 114
9cce02e8
JA
115}
116
117/*
cd963e18 118 * vmsplice() pipe to io_u buffer
9cce02e8 119 */
cd963e18
JA
120static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u,
121 unsigned int len)
9cce02e8
JA
122{
123 struct netio_data *nd = td->io_ops->data;
9cce02e8 124
cd963e18
JA
125 return vmsplice_io_u(io_u, nd->pipes[0], len);
126}
9cce02e8 127
cd963e18
JA
128/*
129 * vmsplice() io_u to pipe
130 */
131static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u)
132{
133 struct netio_data *nd = td->io_ops->data;
ed92ac0c 134
cd963e18 135 return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
9cce02e8
JA
136}
137
cd963e18
JA
138/*
139 * splice receive - transfer socket data into a pipe using splice, then map
140 * that pipe data into the io_u using vmsplice.
141 */
9cce02e8
JA
142static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
143{
144 int ret;
145
146 ret = splice_in(td, io_u);
cd963e18
JA
147 if (ret > 0)
148 return vmsplice_io_u_out(td, io_u, ret);
9cce02e8 149
cd963e18 150 return ret;
9cce02e8
JA
151}
152
cd963e18
JA
153/*
154 * splice transmit - map data from the io_u into a pipe by using vmsplice,
155 * then transfer that pipe to a socket using splice.
156 */
9cce02e8
JA
157static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
158{
159 int ret;
160
161 ret = vmsplice_io_u_in(td, io_u);
cd963e18
JA
162 if (ret > 0)
163 return splice_out(td, io_u, ret);
9cce02e8 164
cd963e18 165 return ret;
9cce02e8 166}
5921e80c
JA
167#else
168static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
169{
af8771b9 170 errno = EOPNOTSUPP;
5921e80c
JA
171 return -1;
172}
173
174static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
175{
af8771b9 176 errno = EOPNOTSUPP;
5921e80c
JA
177 return -1;
178}
179#endif
9cce02e8
JA
180
181static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
182{
183 int flags = 0;
184
185 /*
186 * if we are going to write more, set MSG_MORE
187 */
5921e80c 188#ifdef MSG_MORE
9cce02e8
JA
189 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < td->o.size)
190 flags = MSG_MORE;
5921e80c 191#endif
9cce02e8
JA
192
193 return send(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
194}
195
196static int fio_netio_recv(struct io_u *io_u)
197{
198 int flags = MSG_WAITALL;
199
200 return recv(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
201}
202
203static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
204{
205 struct netio_data *nd = td->io_ops->data;
206 int ret;
207
7101d9c2
JA
208 fio_ro_check(td, io_u);
209
9cce02e8
JA
210 if (io_u->ddir == DDIR_WRITE) {
211 if (nd->use_splice)
212 ret = fio_netio_splice_out(td, io_u);
213 else
214 ret = fio_netio_send(td, io_u);
d4f12dd0 215 } else if (io_u->ddir == DDIR_READ) {
9cce02e8
JA
216 if (nd->use_splice)
217 ret = fio_netio_splice_in(td, io_u);
218 else
219 ret = fio_netio_recv(io_u);
d4f12dd0 220 } else
7a6499da 221 ret = 0; /* must be a SYNC */
ed92ac0c 222
cec6b55d 223 if (ret != (int) io_u->xfer_buflen) {
22819ec2 224 if (ret >= 0) {
cec6b55d
JA
225 io_u->resid = io_u->xfer_buflen - ret;
226 io_u->error = 0;
36167d82 227 return FIO_Q_COMPLETED;
ed92ac0c
JA
228 } else
229 io_u->error = errno;
230 }
231
36167d82 232 if (io_u->error)
e1161c32 233 td_verror(td, io_u->error, "xfer");
ed92ac0c 234
36167d82 235 return FIO_Q_COMPLETED;
ed92ac0c
JA
236}
237
b5af8293 238static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
ed92ac0c 239{
b5af8293 240 struct netio_data *nd = td->io_ops->data;
ed92ac0c 241
b5af8293
JA
242 f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
243 if (f->fd < 0) {
244 td_verror(td, errno, "socket");
245 return 1;
ed92ac0c
JA
246 }
247
b5af8293
JA
248 if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
249 td_verror(td, errno, "connect");
250 return 1;
ed92ac0c
JA
251 }
252
253 return 0;
ed92ac0c
JA
254}
255
b5af8293 256static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
5fdd124a 257{
b5af8293
JA
258 struct netio_data *nd = td->io_ops->data;
259 socklen_t socklen = sizeof(nd->addr);
5fdd124a 260 struct pollfd pfd;
b5af8293 261 int ret;
5fdd124a 262
6d86144d 263 log_info("fio: waiting for connection\n");
5fdd124a
JA
264
265 /*
266 * Accept loop. poll for incoming events, accept them. Repeat until we
267 * have all connections.
268 */
b5af8293
JA
269 while (!td->terminate) {
270 pfd.fd = nd->listenfd;
5fdd124a
JA
271 pfd.events = POLLIN;
272
273 ret = poll(&pfd, 1, -1);
274 if (ret < 0) {
275 if (errno == EINTR)
276 continue;
277
e1161c32 278 td_verror(td, errno, "poll");
5fdd124a
JA
279 break;
280 } else if (!ret)
281 continue;
282
0c09442b
JA
283 /*
284 * should be impossible
285 */
286 if (!(pfd.revents & POLLIN))
287 continue;
288
b5af8293
JA
289 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
290 if (f->fd < 0) {
291 td_verror(td, errno, "accept");
292 return 1;
293 }
294 break;
295 }
5fdd124a 296
b5af8293
JA
297 return 0;
298}
299
b5af8293
JA
300static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
301{
302 if (td_read(td))
303 return fio_netio_accept(td, f);
304 else
305 return fio_netio_connect(td, f);
306}
307
308static int fio_netio_setup_connect(struct thread_data *td, const char *host,
309 unsigned short port)
310{
311 struct netio_data *nd = td->io_ops->data;
312
313 nd->addr.sin_family = AF_INET;
314 nd->addr.sin_port = htons(port);
315
316 if (inet_aton(host, &nd->addr.sin_addr) != 1) {
317 struct hostent *hent;
318
319 hent = gethostbyname(host);
320 if (!hent) {
321 td_verror(td, errno, "gethostbyname");
322 return 1;
5fdd124a 323 }
b5af8293
JA
324
325 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
5fdd124a
JA
326 }
327
328 return 0;
329}
330
b5af8293 331static int fio_netio_setup_listen(struct thread_data *td, short port)
ed92ac0c 332{
b5af8293 333 struct netio_data *nd = td->io_ops->data;
5fdd124a 334 int fd, opt;
ed92ac0c 335
6bedbfaf 336 fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
ed92ac0c 337 if (fd < 0) {
e1161c32 338 td_verror(td, errno, "socket");
ed92ac0c
JA
339 return 1;
340 }
341
342 opt = 1;
343 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
e1161c32 344 td_verror(td, errno, "setsockopt");
ed92ac0c
JA
345 return 1;
346 }
6bedbfaf
JA
347#ifdef SO_REUSEPORT
348 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
e1161c32 349 td_verror(td, errno, "setsockopt");
6bedbfaf
JA
350 return 1;
351 }
352#endif
ed92ac0c 353
b5af8293
JA
354 nd->addr.sin_family = AF_INET;
355 nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
356 nd->addr.sin_port = htons(port);
ed92ac0c 357
b5af8293 358 if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
e1161c32 359 td_verror(td, errno, "bind");
ed92ac0c
JA
360 return 1;
361 }
362 if (listen(fd, 1) < 0) {
e1161c32 363 td_verror(td, errno, "listen");
ed92ac0c
JA
364 return 1;
365 }
366
b5af8293
JA
367 nd->listenfd = fd;
368 return 0;
ed92ac0c
JA
369}
370
9bec88e1 371static int fio_netio_init(struct thread_data *td)
ed92ac0c 372{
b5af8293 373 struct netio_data *nd = td->io_ops->data;
443662ef 374 unsigned int port;
b5af8293 375 char host[64], buf[128];
ed92ac0c 376 char *sep;
af52b345 377 int ret;
ed92ac0c 378
413dd459 379 if (td_rw(td)) {
ed92ac0c
JA
380 log_err("fio: network connections must be read OR write\n");
381 return 1;
382 }
16d55aae
JA
383 if (td_random(td)) {
384 log_err("fio: network IO can't be random\n");
385 return 1;
386 }
ed92ac0c 387
2dc1bbeb 388 strcpy(buf, td->o.filename);
ed92ac0c 389
9f9214f2 390 sep = strchr(buf, '/');
443662ef
JA
391 if (!sep)
392 goto bad_host;
ed92ac0c
JA
393
394 *sep = '\0';
395 sep++;
396 strcpy(host, buf);
443662ef
JA
397 if (!strlen(host))
398 goto bad_host;
399
400 port = strtol(sep, NULL, 10);
401 if (!port || port > 65535)
402 goto bad_host;
ed92ac0c 403
413dd459 404 if (td_read(td)) {
b5af8293 405 nd->send_to_net = 0;
ed92ac0c
JA
406 ret = fio_netio_setup_listen(td, port);
407 } else {
b5af8293 408 nd->send_to_net = 1;
ed92ac0c
JA
409 ret = fio_netio_setup_connect(td, host, port);
410 }
411
7bb48f84 412 return ret;
443662ef
JA
413bad_host:
414 log_err("fio: bad network host/port: %s\n", td->o.filename);
415 return 1;
ed92ac0c
JA
416}
417
b5af8293 418static void fio_netio_cleanup(struct thread_data *td)
9bec88e1 419{
b5af8293
JA
420 struct netio_data *nd = td->io_ops->data;
421
422 if (nd) {
64b24cd8
JA
423 if (nd->listenfd != -1)
424 close(nd->listenfd);
425 if (nd->pipes[0] != -1)
426 close(nd->pipes[0]);
427 if (nd->pipes[1] != -1)
428 close(nd->pipes[1]);
429
b5af8293 430 free(nd);
b5af8293
JA
431 }
432}
433
434static int fio_netio_setup(struct thread_data *td)
435{
7bb48f84 436 struct netio_data *nd;
7bb48f84
JA
437
438 if (!td->io_ops->data) {
439 nd = malloc(sizeof(*nd));;
440
441 memset(nd, 0, sizeof(*nd));
442 nd->listenfd = -1;
64b24cd8 443 nd->pipes[0] = nd->pipes[1] = -1;
7bb48f84 444 td->io_ops->data = nd;
7bb48f84 445 }
b5af8293 446
9bec88e1
JA
447 return 0;
448}
449
5921e80c 450#ifdef FIO_HAVE_SPLICE
9cce02e8
JA
451static int fio_netio_setup_splice(struct thread_data *td)
452{
453 struct netio_data *nd;
454
455 fio_netio_setup(td);
456
457 nd = td->io_ops->data;
458 if (nd) {
459 if (pipe(nd->pipes) < 0)
460 return 1;
461
462 nd->use_splice = 1;
463 return 0;
464 }
465
466 return 1;
467}
468
5921e80c
JA
469static struct ioengine_ops ioengine_splice = {
470 .name = "netsplice",
ed92ac0c 471 .version = FIO_IOOPS_VERSION,
ed92ac0c
JA
472 .prep = fio_netio_prep,
473 .queue = fio_netio_queue,
5921e80c 474 .setup = fio_netio_setup_splice,
9bec88e1 475 .init = fio_netio_init,
b5af8293
JA
476 .cleanup = fio_netio_cleanup,
477 .open_file = fio_netio_open_file,
478 .close_file = generic_close_file,
ad830ed7
JA
479 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
480 FIO_SIGQUIT,
ed92ac0c 481};
5921e80c 482#endif
ed92ac0c 483
5921e80c
JA
484static struct ioengine_ops ioengine_rw = {
485 .name = "net",
9cce02e8
JA
486 .version = FIO_IOOPS_VERSION,
487 .prep = fio_netio_prep,
488 .queue = fio_netio_queue,
5921e80c 489 .setup = fio_netio_setup,
9cce02e8
JA
490 .init = fio_netio_init,
491 .cleanup = fio_netio_cleanup,
492 .open_file = fio_netio_open_file,
493 .close_file = generic_close_file,
ad830ed7
JA
494 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
495 FIO_SIGQUIT,
9cce02e8
JA
496};
497
ed92ac0c
JA
498static void fio_init fio_netio_register(void)
499{
9cce02e8 500 register_ioengine(&ioengine_rw);
5921e80c 501#ifdef FIO_HAVE_SPLICE
9cce02e8 502 register_ioengine(&ioengine_splice);
5921e80c 503#endif
ed92ac0c
JA
504}
505
506static void fio_exit fio_netio_unregister(void)
507{
9cce02e8 508 unregister_ioengine(&ioengine_rw);
5921e80c 509#ifdef FIO_HAVE_SPLICE
9cce02e8 510 unregister_ioengine(&ioengine_splice);
5921e80c 511#endif
ed92ac0c 512}