Fio 1.26.2
[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 24 int use_splice;
414c2a3e 25 int net_protocol;
9cce02e8 26 int pipes[2];
b5af8293
JA
27 char host[64];
28 struct sockaddr_in addr;
29};
ed92ac0c 30
664fb3bd
JA
31struct udp_close_msg {
32 uint32_t magic;
33 uint32_t cmd;
34};
35
36enum {
37 FIO_LINK_CLOSE = 0x89,
38 FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
39};
40
371d456c
JA
41/*
42 * Return -1 for error and 'nr events' for a positive number
43 * of events
44 */
45static int poll_wait(struct thread_data *td, int fd, short events)
46{
47 struct pollfd pfd;
48 int ret;
49
50 while (!td->terminate) {
51 pfd.fd = fd;
52 pfd.events = events;
53 ret = poll(&pfd, 1, -1);
54 if (ret < 0) {
55 if (errno == EINTR)
d5b388a5 56 break;
371d456c
JA
57
58 td_verror(td, errno, "poll");
59 return -1;
60 } else if (!ret)
61 continue;
62
63 break;
64 }
65
66 if (pfd.revents & events)
67 return 1;
371d456c
JA
68
69 return -1;
70}
71
ed92ac0c
JA
72static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
73{
b5af8293 74 struct netio_data *nd = td->io_ops->data;
ed92ac0c 75
7a6499da
JA
76 /*
77 * Make sure we don't see spurious reads to a receiver, and vice versa
78 */
b5af8293
JA
79 if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
80 (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
e1161c32 81 td_verror(td, EINVAL, "bad direction");
7a6499da 82 return 1;
ed92ac0c 83 }
7a6499da 84
f85ac25a 85 return 0;
ed92ac0c
JA
86}
87
5921e80c 88#ifdef FIO_HAVE_SPLICE
cd963e18 89static int splice_io_u(int fdin, int fdout, unsigned int len)
ed92ac0c 90{
9cce02e8 91 int bytes = 0;
7a6499da 92
9cce02e8 93 while (len) {
cd963e18 94 int ret = splice(fdin, NULL, fdout, NULL, len, 0);
9cce02e8
JA
95
96 if (ret < 0) {
97 if (!bytes)
98 bytes = ret;
99
100 break;
101 } else if (!ret)
102 break;
103
104 bytes += ret;
f657a2fb 105 len -= ret;
9cce02e8
JA
106 }
107
108 return bytes;
109}
110
111/*
cd963e18 112 * Receive bytes from a socket and fill them into the internal pipe
9cce02e8 113 */
cd963e18 114static int splice_in(struct thread_data *td, struct io_u *io_u)
9cce02e8
JA
115{
116 struct netio_data *nd = td->io_ops->data;
9cce02e8 117
cd963e18 118 return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
9cce02e8
JA
119}
120
121/*
cd963e18 122 * Transmit 'len' bytes from the internal pipe
9cce02e8 123 */
cd963e18
JA
124static int splice_out(struct thread_data *td, struct io_u *io_u,
125 unsigned int len)
9cce02e8
JA
126{
127 struct netio_data *nd = td->io_ops->data;
cd963e18
JA
128
129 return splice_io_u(nd->pipes[0], io_u->file->fd, len);
130}
131
132static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
133{
9cce02e8
JA
134 struct iovec iov = {
135 .iov_base = io_u->xfer_buf,
136 .iov_len = len,
137 };
138 int bytes = 0;
139
140 while (iov.iov_len) {
cd963e18 141 int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
9cce02e8
JA
142
143 if (ret < 0) {
144 if (!bytes)
145 bytes = ret;
146 break;
147 } else if (!ret)
148 break;
149
150 iov.iov_len -= ret;
cd963e18 151 iov.iov_base += ret;
f657a2fb 152 bytes += ret;
9cce02e8
JA
153 }
154
155 return bytes;
cd963e18 156
9cce02e8
JA
157}
158
159/*
cd963e18 160 * vmsplice() pipe to io_u buffer
9cce02e8 161 */
cd963e18
JA
162static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u,
163 unsigned int len)
9cce02e8
JA
164{
165 struct netio_data *nd = td->io_ops->data;
9cce02e8 166
cd963e18
JA
167 return vmsplice_io_u(io_u, nd->pipes[0], len);
168}
9cce02e8 169
cd963e18
JA
170/*
171 * vmsplice() io_u to pipe
172 */
173static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u)
174{
175 struct netio_data *nd = td->io_ops->data;
ed92ac0c 176
cd963e18 177 return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
9cce02e8
JA
178}
179
cd963e18
JA
180/*
181 * splice receive - transfer socket data into a pipe using splice, then map
182 * that pipe data into the io_u using vmsplice.
183 */
9cce02e8
JA
184static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
185{
186 int ret;
187
188 ret = splice_in(td, io_u);
cd963e18
JA
189 if (ret > 0)
190 return vmsplice_io_u_out(td, io_u, ret);
9cce02e8 191
cd963e18 192 return ret;
9cce02e8
JA
193}
194
cd963e18
JA
195/*
196 * splice transmit - map data from the io_u into a pipe by using vmsplice,
197 * then transfer that pipe to a socket using splice.
198 */
9cce02e8
JA
199static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
200{
201 int ret;
202
203 ret = vmsplice_io_u_in(td, io_u);
cd963e18
JA
204 if (ret > 0)
205 return splice_out(td, io_u, ret);
9cce02e8 206
cd963e18 207 return ret;
9cce02e8 208}
5921e80c
JA
209#else
210static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
211{
af8771b9 212 errno = EOPNOTSUPP;
5921e80c
JA
213 return -1;
214}
215
216static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
217{
af8771b9 218 errno = EOPNOTSUPP;
5921e80c
JA
219 return -1;
220}
221#endif
9cce02e8
JA
222
223static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
224{
414c2a3e 225 struct netio_data *nd = td->io_ops->data;
664fb3bd 226 int ret, flags = MSG_DONTWAIT;
371d456c 227
664fb3bd
JA
228 do {
229 if (nd->net_protocol == IPPROTO_UDP) {
230 ret = sendto(io_u->file->fd, io_u->xfer_buf,
231 io_u->xfer_buflen, flags, &nd->addr,
232 sizeof(nd->addr));
233 } else {
234 /*
235 * if we are going to write more, set MSG_MORE
236 */
5921e80c 237#ifdef MSG_MORE
664fb3bd
JA
238 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
239 td->o.size)
240 flags |= MSG_MORE;
5921e80c 241#endif
664fb3bd
JA
242 ret = send(io_u->file->fd, io_u->xfer_buf,
243 io_u->xfer_buflen, flags);
244 }
245 if (ret > 0)
246 break;
9cce02e8 247
664fb3bd
JA
248 ret = poll_wait(td, io_u->file->fd, POLLOUT);
249 if (ret <= 0)
250 break;
251
252 flags &= ~MSG_DONTWAIT;
253 } while (1);
254
255 return ret;
256}
257
258static int is_udp_close(struct io_u *io_u, int len)
259{
260 struct udp_close_msg *msg;
261
262 if (len != sizeof(struct udp_close_msg))
263 return 0;
264
265 msg = io_u->xfer_buf;
266 if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
267 return 0;
268 if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
269 return 0;
270
271 return 1;
9cce02e8
JA
272}
273
414c2a3e 274static int fio_netio_recv(struct thread_data *td, struct io_u *io_u)
9cce02e8 275{
414c2a3e 276 struct netio_data *nd = td->io_ops->data;
664fb3bd
JA
277 int ret, flags = MSG_DONTWAIT;
278
279 do {
280 if (nd->net_protocol == IPPROTO_UDP) {
281 socklen_t len = sizeof(nd->addr);
282
283 ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
284 io_u->xfer_buflen, flags, &nd->addr,
285 &len);
286 if (is_udp_close(io_u, ret)) {
287 td->done = 1;
288 return 0;
289 }
290 } else {
291 ret = recv(io_u->file->fd, io_u->xfer_buf,
292 io_u->xfer_buflen, flags);
293 }
294 if (ret > 0)
295 break;
9cce02e8 296
664fb3bd
JA
297 ret = poll_wait(td, io_u->file->fd, POLLIN);
298 if (ret <= 0)
299 break;
300 flags &= ~MSG_DONTWAIT;
301 flags |= MSG_WAITALL;
302 } while (1);
414c2a3e 303
664fb3bd 304 return ret;
9cce02e8
JA
305}
306
307static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
308{
309 struct netio_data *nd = td->io_ops->data;
310 int ret;
311
7101d9c2
JA
312 fio_ro_check(td, io_u);
313
9cce02e8 314 if (io_u->ddir == DDIR_WRITE) {
414c2a3e 315 if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP)
9cce02e8 316 ret = fio_netio_send(td, io_u);
414c2a3e
JA
317 else
318 ret = fio_netio_splice_out(td, io_u);
d4f12dd0 319 } else if (io_u->ddir == DDIR_READ) {
414c2a3e
JA
320 if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP)
321 ret = fio_netio_recv(td, io_u);
9cce02e8 322 else
414c2a3e 323 ret = fio_netio_splice_in(td, io_u);
d4f12dd0 324 } else
7a6499da 325 ret = 0; /* must be a SYNC */
ed92ac0c 326
cec6b55d 327 if (ret != (int) io_u->xfer_buflen) {
22819ec2 328 if (ret >= 0) {
cec6b55d
JA
329 io_u->resid = io_u->xfer_buflen - ret;
330 io_u->error = 0;
36167d82 331 return FIO_Q_COMPLETED;
414c2a3e
JA
332 } else {
333 int err = errno;
334
335 if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
336 return FIO_Q_BUSY;
337
338 io_u->error = err;
339 }
ed92ac0c
JA
340 }
341
36167d82 342 if (io_u->error)
e1161c32 343 td_verror(td, io_u->error, "xfer");
ed92ac0c 344
36167d82 345 return FIO_Q_COMPLETED;
ed92ac0c
JA
346}
347
b5af8293 348static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
ed92ac0c 349{
b5af8293 350 struct netio_data *nd = td->io_ops->data;
414c2a3e
JA
351 int type;
352
353 if (nd->net_protocol == IPPROTO_TCP)
354 type = SOCK_STREAM;
355 else
356 type = SOCK_DGRAM;
ed92ac0c 357
414c2a3e 358 f->fd = socket(AF_INET, type, nd->net_protocol);
b5af8293
JA
359 if (f->fd < 0) {
360 td_verror(td, errno, "socket");
361 return 1;
ed92ac0c
JA
362 }
363
414c2a3e
JA
364 if (nd->net_protocol == IPPROTO_UDP)
365 return 0;
366
b5af8293
JA
367 if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
368 td_verror(td, errno, "connect");
369 return 1;
ed92ac0c
JA
370 }
371
372 return 0;
ed92ac0c
JA
373}
374
b5af8293 375static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
5fdd124a 376{
b5af8293
JA
377 struct netio_data *nd = td->io_ops->data;
378 socklen_t socklen = sizeof(nd->addr);
5fdd124a 379
414c2a3e
JA
380 if (nd->net_protocol == IPPROTO_UDP) {
381 f->fd = nd->listenfd;
382 return 0;
383 }
384
6d86144d 385 log_info("fio: waiting for connection\n");
5fdd124a 386
371d456c
JA
387 if (poll_wait(td, nd->listenfd, POLLIN) < 0)
388 return 1;
0c09442b 389
371d456c
JA
390 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
391 if (f->fd < 0) {
392 td_verror(td, errno, "accept");
393 return 1;
b5af8293 394 }
5fdd124a 395
b5af8293
JA
396 return 0;
397}
398
b5af8293
JA
399static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
400{
401 if (td_read(td))
402 return fio_netio_accept(td, f);
403 else
404 return fio_netio_connect(td, f);
405}
406
664fb3bd
JA
407static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
408{
409 struct netio_data *nd = td->io_ops->data;
410 struct udp_close_msg msg;
411 int ret;
412
413 msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
414 msg.cmd = htonl(FIO_LINK_CLOSE);
415
416 ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, &nd->addr,
417 sizeof(nd->addr));
418 if (ret < 0)
419 td_verror(td, errno, "sendto udp link close");
420}
421
422static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
423{
424 struct netio_data *nd = td->io_ops->data;
425
426 /*
427 * If this is an UDP connection, notify the receiver that we are
428 * closing down the link
429 */
430 if (nd->net_protocol == IPPROTO_UDP)
431 fio_netio_udp_close(td, f);
432
433 return generic_close_file(td, f);
434}
435
b5af8293
JA
436static int fio_netio_setup_connect(struct thread_data *td, const char *host,
437 unsigned short port)
438{
439 struct netio_data *nd = td->io_ops->data;
440
441 nd->addr.sin_family = AF_INET;
442 nd->addr.sin_port = htons(port);
443
444 if (inet_aton(host, &nd->addr.sin_addr) != 1) {
445 struct hostent *hent;
446
447 hent = gethostbyname(host);
448 if (!hent) {
449 td_verror(td, errno, "gethostbyname");
450 return 1;
5fdd124a 451 }
b5af8293
JA
452
453 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
5fdd124a
JA
454 }
455
456 return 0;
457}
458
b5af8293 459static int fio_netio_setup_listen(struct thread_data *td, short port)
ed92ac0c 460{
b5af8293 461 struct netio_data *nd = td->io_ops->data;
414c2a3e 462 int fd, opt, type;
ed92ac0c 463
414c2a3e
JA
464 if (nd->net_protocol == IPPROTO_TCP)
465 type = SOCK_STREAM;
466 else
467 type = SOCK_DGRAM;
468
469 fd = socket(AF_INET, type, nd->net_protocol);
ed92ac0c 470 if (fd < 0) {
e1161c32 471 td_verror(td, errno, "socket");
ed92ac0c
JA
472 return 1;
473 }
474
475 opt = 1;
476 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
e1161c32 477 td_verror(td, errno, "setsockopt");
ed92ac0c
JA
478 return 1;
479 }
6bedbfaf
JA
480#ifdef SO_REUSEPORT
481 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
e1161c32 482 td_verror(td, errno, "setsockopt");
6bedbfaf
JA
483 return 1;
484 }
485#endif
ed92ac0c 486
b5af8293
JA
487 nd->addr.sin_family = AF_INET;
488 nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
489 nd->addr.sin_port = htons(port);
ed92ac0c 490
b5af8293 491 if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
e1161c32 492 td_verror(td, errno, "bind");
ed92ac0c
JA
493 return 1;
494 }
414c2a3e 495 if (nd->net_protocol == IPPROTO_TCP && listen(fd, 1) < 0) {
e1161c32 496 td_verror(td, errno, "listen");
ed92ac0c
JA
497 return 1;
498 }
499
b5af8293
JA
500 nd->listenfd = fd;
501 return 0;
ed92ac0c
JA
502}
503
9bec88e1 504static int fio_netio_init(struct thread_data *td)
ed92ac0c 505{
b5af8293 506 struct netio_data *nd = td->io_ops->data;
443662ef 507 unsigned int port;
b5af8293 508 char host[64], buf[128];
414c2a3e 509 char *sep, *portp, *modep;
af52b345 510 int ret;
ed92ac0c 511
413dd459 512 if (td_rw(td)) {
ed92ac0c
JA
513 log_err("fio: network connections must be read OR write\n");
514 return 1;
515 }
16d55aae
JA
516 if (td_random(td)) {
517 log_err("fio: network IO can't be random\n");
518 return 1;
519 }
ed92ac0c 520
2dc1bbeb 521 strcpy(buf, td->o.filename);
ed92ac0c 522
9f9214f2 523 sep = strchr(buf, '/');
443662ef
JA
524 if (!sep)
525 goto bad_host;
ed92ac0c
JA
526
527 *sep = '\0';
528 sep++;
529 strcpy(host, buf);
443662ef
JA
530 if (!strlen(host))
531 goto bad_host;
532
414c2a3e
JA
533 modep = NULL;
534 portp = sep;
535 sep = strchr(portp, '/');
536 if (sep) {
537 *sep = '\0';
538 modep = sep + 1;
539 }
540
541 port = strtol(portp, NULL, 10);
443662ef
JA
542 if (!port || port > 65535)
543 goto bad_host;
ed92ac0c 544
414c2a3e 545 if (modep) {
3f8fc5ad
JA
546 if (!strncmp("tcp", modep, strlen(modep)) ||
547 !strncmp("TCP", modep, strlen(modep)))
414c2a3e 548 nd->net_protocol = IPPROTO_TCP;
3f8fc5ad
JA
549 else if (!strncmp("udp", modep, strlen(modep)) ||
550 !strncmp("UDP", modep, strlen(modep)))
414c2a3e
JA
551 nd->net_protocol = IPPROTO_UDP;
552 else
553 goto bad_host;
554 } else
555 nd->net_protocol = IPPROTO_TCP;
556
413dd459 557 if (td_read(td)) {
b5af8293 558 nd->send_to_net = 0;
ed92ac0c
JA
559 ret = fio_netio_setup_listen(td, port);
560 } else {
b5af8293 561 nd->send_to_net = 1;
ed92ac0c
JA
562 ret = fio_netio_setup_connect(td, host, port);
563 }
564
7bb48f84 565 return ret;
443662ef 566bad_host:
414c2a3e 567 log_err("fio: bad network host/port/protocol: %s\n", td->o.filename);
443662ef 568 return 1;
ed92ac0c
JA
569}
570
b5af8293 571static void fio_netio_cleanup(struct thread_data *td)
9bec88e1 572{
b5af8293
JA
573 struct netio_data *nd = td->io_ops->data;
574
575 if (nd) {
64b24cd8
JA
576 if (nd->listenfd != -1)
577 close(nd->listenfd);
578 if (nd->pipes[0] != -1)
579 close(nd->pipes[0]);
580 if (nd->pipes[1] != -1)
581 close(nd->pipes[1]);
582
b5af8293 583 free(nd);
b5af8293
JA
584 }
585}
586
587static int fio_netio_setup(struct thread_data *td)
588{
7bb48f84 589 struct netio_data *nd;
7bb48f84
JA
590
591 if (!td->io_ops->data) {
592 nd = malloc(sizeof(*nd));;
593
594 memset(nd, 0, sizeof(*nd));
595 nd->listenfd = -1;
64b24cd8 596 nd->pipes[0] = nd->pipes[1] = -1;
7bb48f84 597 td->io_ops->data = nd;
7bb48f84 598 }
b5af8293 599
9bec88e1
JA
600 return 0;
601}
602
5921e80c 603#ifdef FIO_HAVE_SPLICE
9cce02e8
JA
604static int fio_netio_setup_splice(struct thread_data *td)
605{
606 struct netio_data *nd;
607
608 fio_netio_setup(td);
609
610 nd = td->io_ops->data;
611 if (nd) {
612 if (pipe(nd->pipes) < 0)
613 return 1;
614
615 nd->use_splice = 1;
616 return 0;
617 }
618
619 return 1;
620}
621
5921e80c
JA
622static struct ioengine_ops ioengine_splice = {
623 .name = "netsplice",
ed92ac0c 624 .version = FIO_IOOPS_VERSION,
ed92ac0c
JA
625 .prep = fio_netio_prep,
626 .queue = fio_netio_queue,
5921e80c 627 .setup = fio_netio_setup_splice,
9bec88e1 628 .init = fio_netio_init,
b5af8293
JA
629 .cleanup = fio_netio_cleanup,
630 .open_file = fio_netio_open_file,
631 .close_file = generic_close_file,
ad830ed7
JA
632 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
633 FIO_SIGQUIT,
ed92ac0c 634};
5921e80c 635#endif
ed92ac0c 636
5921e80c
JA
637static struct ioengine_ops ioengine_rw = {
638 .name = "net",
9cce02e8
JA
639 .version = FIO_IOOPS_VERSION,
640 .prep = fio_netio_prep,
641 .queue = fio_netio_queue,
5921e80c 642 .setup = fio_netio_setup,
9cce02e8
JA
643 .init = fio_netio_init,
644 .cleanup = fio_netio_cleanup,
645 .open_file = fio_netio_open_file,
664fb3bd 646 .close_file = fio_netio_close_file,
ad830ed7
JA
647 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
648 FIO_SIGQUIT,
9cce02e8
JA
649};
650
ed92ac0c
JA
651static void fio_init fio_netio_register(void)
652{
9cce02e8 653 register_ioengine(&ioengine_rw);
5921e80c 654#ifdef FIO_HAVE_SPLICE
9cce02e8 655 register_ioengine(&ioengine_splice);
5921e80c 656#endif
ed92ac0c
JA
657}
658
659static void fio_exit fio_netio_unregister(void)
660{
9cce02e8 661 unregister_ioengine(&ioengine_rw);
5921e80c 662#ifdef FIO_HAVE_SPLICE
9cce02e8 663 unregister_ioengine(&ioengine_splice);
5921e80c 664#endif
ed92ac0c 665}