projects / fio.git / blame_incremental
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (non-incremental) | history | HEAD
engine: more checks on IPv6 and multicast
[fio.git] / engines / net.c
... / ...
CommitLineData
1/*
2 * net engine
3 *
4 * IO engine that reads/writes to/from sockets.
5 *
6 */
7#include <stdio.h>
8#include <stdlib.h>
9#include <unistd.h>
10#include <signal.h>
11#include <errno.h>
12#include <assert.h>
13#include <netinet/in.h>
14#include <netinet/tcp.h>
15#include <arpa/inet.h>
16#include <netdb.h>
17#include <sys/poll.h>
18#include <sys/types.h>
19#include <sys/stat.h>
20#include <sys/socket.h>
21#include <sys/un.h>
22
23#include "../fio.h"
24
25struct netio_data {
26 int listenfd;
27 int use_splice;
28 int pipes[2];
29 struct sockaddr_in addr;
30 struct sockaddr_in6 addr6;
31 struct sockaddr_un addr_un;
32};
33
34struct netio_options {
35 struct thread_data *td;
36 unsigned int port;
37 unsigned int proto;
38 unsigned int listen;
39 unsigned int pingpong;
40 unsigned int nodelay;
41 unsigned int ttl;
42 char *intfc;
43};
44
45struct udp_close_msg {
46 uint32_t magic;
47 uint32_t cmd;
48};
49
50enum {
51 FIO_LINK_CLOSE = 0x89,
52 FIO_LINK_OPEN_CLOSE_MAGIC = 0x6c696e6b,
53 FIO_LINK_OPEN = 0x98,
54
55 FIO_TYPE_TCP = 1,
56 FIO_TYPE_UDP = 2,
57 FIO_TYPE_UNIX = 3,
58 FIO_TYPE_TCP_V6 = 4,
59 FIO_TYPE_UDP_V6 = 5,
60};
61
62static int str_hostname_cb(void *data, const char *input);
63static struct fio_option options[] = {
64 {
65 .name = "hostname",
66 .lname = "net engine hostname",
67 .type = FIO_OPT_STR_STORE,
68 .cb = str_hostname_cb,
69 .help = "Hostname for net IO engine",
70 .category = FIO_OPT_C_ENGINE,
71 .group = FIO_OPT_G_NETIO,
72 },
73 {
74 .name = "port",
75 .lname = "net engine port",
76 .type = FIO_OPT_INT,
77 .off1 = offsetof(struct netio_options, port),
78 .minval = 1,
79 .maxval = 65535,
80 .help = "Port to use for TCP or UDP net connections",
81 .category = FIO_OPT_C_ENGINE,
82 .group = FIO_OPT_G_NETIO,
83 },
84 {
85 .name = "protocol",
86 .lname = "net engine protocol",
87 .alias = "proto",
88 .type = FIO_OPT_STR,
89 .off1 = offsetof(struct netio_options, proto),
90 .help = "Network protocol to use",
91 .def = "tcp",
92 .posval = {
93 { .ival = "tcp",
94 .oval = FIO_TYPE_TCP,
95 .help = "Transmission Control Protocol",
96 },
97 { .ival = "tcpv6",
98 .oval = FIO_TYPE_TCP_V6,
99 .help = "Transmission Control Protocol V6",
100 },
101 { .ival = "udp",
102 .oval = FIO_TYPE_UDP,
103 .help = "User Datagram Protocol",
104 },
105 { .ival = "udpv6",
106 .oval = FIO_TYPE_UDP_V6,
107 .help = "User Datagram Protocol V6",
108 },
109 { .ival = "unix",
110 .oval = FIO_TYPE_UNIX,
111 .help = "UNIX domain socket",
112 },
113 },
114 .category = FIO_OPT_C_ENGINE,
115 .group = FIO_OPT_G_NETIO,
116 },
117#ifdef CONFIG_TCP_NODELAY
118 {
119 .name = "nodelay",
120 .type = FIO_OPT_BOOL,
121 .off1 = offsetof(struct netio_options, nodelay),
122 .help = "Use TCP_NODELAY on TCP connections",
123 .category = FIO_OPT_C_ENGINE,
124 .group = FIO_OPT_G_NETIO,
125 },
126#endif
127 {
128 .name = "listen",
129 .lname = "net engine listen",
130 .type = FIO_OPT_STR_SET,
131 .off1 = offsetof(struct netio_options, listen),
132 .help = "Listen for incoming TCP connections",
133 .category = FIO_OPT_C_ENGINE,
134 .group = FIO_OPT_G_NETIO,
135 },
136 {
137 .name = "pingpong",
138 .type = FIO_OPT_STR_SET,
139 .off1 = offsetof(struct netio_options, pingpong),
140 .help = "Ping-pong IO requests",
141 .category = FIO_OPT_C_ENGINE,
142 .group = FIO_OPT_G_NETIO,
143 },
144 {
145 .name = "interface",
146 .lname = "net engine interface",
147 .type = FIO_OPT_STR_STORE,
148 .off1 = offsetof(struct netio_options, intfc),
149 .help = "Network interface to use",
150 .category = FIO_OPT_C_ENGINE,
151 .group = FIO_OPT_G_NETIO,
152 },
153 {
154 .name = "ttl",
155 .lname = "net engine multicast ttl",
156 .type = FIO_OPT_INT,
157 .off1 = offsetof(struct netio_options, ttl),
158 .def = "1",
159 .minval = 0,
160 .help = "Time-to-live value for outgoing UDP multicast packets",
161 .category = FIO_OPT_C_ENGINE,
162 .group = FIO_OPT_G_NETIO,
163 },
164 {
165 .name = NULL,
166 },
167};
168
169static inline int is_udp(struct netio_options *o)
170{
171 return o->proto == FIO_TYPE_UDP || o->proto == FIO_TYPE_UDP_V6;
172}
173
174static inline int is_tcp(struct netio_options *o)
175{
176 return o->proto == FIO_TYPE_TCP || o->proto == FIO_TYPE_TCP_V6;
177}
178
179static inline int is_ipv6(struct netio_options *o)
180{
181 return o->proto == FIO_TYPE_UDP_V6 || o->proto == FIO_TYPE_TCP_V6;
182}
183
184/*
185 * Return -1 for error and 'nr events' for a positive number
186 * of events
187 */
188static int poll_wait(struct thread_data *td, int fd, short events)
189{
190 struct pollfd pfd;
191 int ret;
192
193 while (!td->terminate) {
194 pfd.fd = fd;
195 pfd.events = events;
196 ret = poll(&pfd, 1, -1);
197 if (ret < 0) {
198 if (errno == EINTR)
199 break;
200
201 td_verror(td, errno, "poll");
202 return -1;
203 } else if (!ret)
204 continue;
205
206 break;
207 }
208
209 if (pfd.revents & events)
210 return 1;
211
212 return -1;
213}
214
215static int fio_netio_is_multicast(const char *mcaddr)
216{
217 in_addr_t addr = inet_network(mcaddr);
218 if (addr == -1)
219 return 0;
220
221 if (inet_network("224.0.0.0") <= addr &&
222 inet_network("239.255.255.255") >= addr)
223 return 1;
224
225 return 0;
226}
227
228
229static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
230{
231 struct netio_options *o = td->eo;
232
233 /*
234 * Make sure we don't see spurious reads to a receiver, and vice versa
235 */
236 if (is_tcp(o))
237 return 0;
238
239 if ((o->listen && io_u->ddir == DDIR_WRITE) ||
240 (!o->listen && io_u->ddir == DDIR_READ)) {
241 td_verror(td, EINVAL, "bad direction");
242 return 1;
243 }
244
245 return 0;
246}
247
248#ifdef CONFIG_LINUX_SPLICE
249static int splice_io_u(int fdin, int fdout, unsigned int len)
250{
251 int bytes = 0;
252
253 while (len) {
254 int ret = splice(fdin, NULL, fdout, NULL, len, 0);
255
256 if (ret < 0) {
257 if (!bytes)
258 bytes = ret;
259
260 break;
261 } else if (!ret)
262 break;
263
264 bytes += ret;
265 len -= ret;
266 }
267
268 return bytes;
269}
270
271/*
272 * Receive bytes from a socket and fill them into the internal pipe
273 */
274static int splice_in(struct thread_data *td, struct io_u *io_u)
275{
276 struct netio_data *nd = td->io_ops->data;
277
278 return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
279}
280
281/*
282 * Transmit 'len' bytes from the internal pipe
283 */
284static int splice_out(struct thread_data *td, struct io_u *io_u,
285 unsigned int len)
286{
287 struct netio_data *nd = td->io_ops->data;
288
289 return splice_io_u(nd->pipes[0], io_u->file->fd, len);
290}
291
292static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
293{
294 struct iovec iov = {
295 .iov_base = io_u->xfer_buf,
296 .iov_len = len,
297 };
298 int bytes = 0;
299
300 while (iov.iov_len) {
301 int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
302
303 if (ret < 0) {
304 if (!bytes)
305 bytes = ret;
306 break;
307 } else if (!ret)
308 break;
309
310 iov.iov_len -= ret;
311 iov.iov_base += ret;
312 bytes += ret;
313 }
314
315 return bytes;
316
317}
318
319/*
320 * vmsplice() pipe to io_u buffer
321 */
322static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u,
323 unsigned int len)
324{
325 struct netio_data *nd = td->io_ops->data;
326
327 return vmsplice_io_u(io_u, nd->pipes[0], len);
328}
329
330/*
331 * vmsplice() io_u to pipe
332 */
333static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u)
334{
335 struct netio_data *nd = td->io_ops->data;
336
337 return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
338}
339
340/*
341 * splice receive - transfer socket data into a pipe using splice, then map
342 * that pipe data into the io_u using vmsplice.
343 */
344static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
345{
346 int ret;
347
348 ret = splice_in(td, io_u);
349 if (ret > 0)
350 return vmsplice_io_u_out(td, io_u, ret);
351
352 return ret;
353}
354
355/*
356 * splice transmit - map data from the io_u into a pipe by using vmsplice,
357 * then transfer that pipe to a socket using splice.
358 */
359static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
360{
361 int ret;
362
363 ret = vmsplice_io_u_in(td, io_u);
364 if (ret > 0)
365 return splice_out(td, io_u, ret);
366
367 return ret;
368}
369#else
370static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
371{
372 errno = EOPNOTSUPP;
373 return -1;
374}
375
376static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
377{
378 errno = EOPNOTSUPP;
379 return -1;
380}
381#endif
382
383static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
384{
385 struct netio_data *nd = td->io_ops->data;
386 struct netio_options *o = td->eo;
387 int ret, flags = 0;
388
389 do {
390 if (is_udp(o)) {
391 struct sockaddr *to;
392 socklen_t len;
393
394 if (is_ipv6(o)) {
395 to = (struct sockaddr *) &nd->addr6;
396 len = sizeof(nd->addr6);
397 } else {
398 to = (struct sockaddr *) &nd->addr;
399 len = sizeof(nd->addr);
400 }
401
402 ret = sendto(io_u->file->fd, io_u->xfer_buf,
403 io_u->xfer_buflen, flags, to, len);
404 } else {
405 /*
406 * if we are going to write more, set MSG_MORE
407 */
408#ifdef MSG_MORE
409 if ((td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
410 td->o.size) && !o->pingpong)
411 flags |= MSG_MORE;
412#endif
413 ret = send(io_u->file->fd, io_u->xfer_buf,
414 io_u->xfer_buflen, flags);
415 }
416 if (ret > 0)
417 break;
418
419 ret = poll_wait(td, io_u->file->fd, POLLOUT);
420 if (ret <= 0)
421 break;
422 } while (1);
423
424 return ret;
425}
426
427static int is_udp_close(struct io_u *io_u, int len)
428{
429 struct udp_close_msg *msg;
430
431 if (len != sizeof(struct udp_close_msg))
432 return 0;
433
434 msg = io_u->xfer_buf;
435 if (ntohl(msg->magic) != FIO_LINK_OPEN_CLOSE_MAGIC)
436 return 0;
437 if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
438 return 0;
439
440 return 1;
441}
442
443static int fio_netio_recv(struct thread_data *td, struct io_u *io_u)
444{
445 struct netio_data *nd = td->io_ops->data;
446 struct netio_options *o = td->eo;
447 int ret, flags = 0;
448
449 do {
450 if (is_udp(o)) {
451 struct sockaddr *from;
452 socklen_t l, *len = &l;
453
454 if (o->listen) {
455 if (!is_ipv6(o)) {
456 from = (struct sockaddr *) &nd->addr;
457 *len = sizeof(nd->addr);
458 } else {
459 from = (struct sockaddr *) &nd->addr6;
460 *len = sizeof(nd->addr6);
461 }
462 } else {
463 from = NULL;
464 len = NULL;
465 }
466
467 ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
468 io_u->xfer_buflen, flags, from, len);
469 if (is_udp_close(io_u, ret)) {
470 td->done = 1;
471 return 0;
472 }
473 } else {
474 ret = recv(io_u->file->fd, io_u->xfer_buf,
475 io_u->xfer_buflen, flags);
476 }
477 if (ret > 0)
478 break;
479 else if (!ret && (flags & MSG_WAITALL))
480 break;
481
482 ret = poll_wait(td, io_u->file->fd, POLLIN);
483 if (ret <= 0)
484 break;
485 flags |= MSG_WAITALL;
486 } while (1);
487
488 return ret;
489}
490
491static int __fio_netio_queue(struct thread_data *td, struct io_u *io_u,
492 enum fio_ddir ddir)
493{
494 struct netio_data *nd = td->io_ops->data;
495 struct netio_options *o = td->eo;
496 int ret;
497
498 if (ddir == DDIR_WRITE) {
499 if (!nd->use_splice || is_udp(o) ||
500 o->proto == FIO_TYPE_UNIX)
501 ret = fio_netio_send(td, io_u);
502 else
503 ret = fio_netio_splice_out(td, io_u);
504 } else if (ddir == DDIR_READ) {
505 if (!nd->use_splice || is_udp(o) ||
506 o->proto == FIO_TYPE_UNIX)
507 ret = fio_netio_recv(td, io_u);
508 else
509 ret = fio_netio_splice_in(td, io_u);
510 } else
511 ret = 0; /* must be a SYNC */
512
513 if (ret != (int) io_u->xfer_buflen) {
514 if (ret >= 0) {
515 io_u->resid = io_u->xfer_buflen - ret;
516 io_u->error = 0;
517 return FIO_Q_COMPLETED;
518 } else {
519 int err = errno;
520
521 if (ddir == DDIR_WRITE && err == EMSGSIZE)
522 return FIO_Q_BUSY;
523
524 io_u->error = err;
525 }
526 }
527
528 if (io_u->error)
529 td_verror(td, io_u->error, "xfer");
530
531 return FIO_Q_COMPLETED;
532}
533
534static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
535{
536 struct netio_options *o = td->eo;
537 int ret;
538
539 fio_ro_check(td, io_u);
540
541 ret = __fio_netio_queue(td, io_u, io_u->ddir);
542 if (!o->pingpong || ret != FIO_Q_COMPLETED)
543 return ret;
544
545 /*
546 * For ping-pong mode, receive or send reply as needed
547 */
548 if (td_read(td) && io_u->ddir == DDIR_READ)
549 ret = __fio_netio_queue(td, io_u, DDIR_WRITE);
550 else if (td_write(td) && io_u->ddir == DDIR_WRITE)
551 ret = __fio_netio_queue(td, io_u, DDIR_READ);
552
553 return ret;
554}
555
556static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
557{
558 struct netio_data *nd = td->io_ops->data;
559 struct netio_options *o = td->eo;
560 int type, domain;
561
562 if (o->proto == FIO_TYPE_TCP) {
563 domain = AF_INET;
564 type = SOCK_STREAM;
565 } else if (o->proto == FIO_TYPE_TCP_V6) {
566 domain = AF_INET6;
567 type = SOCK_STREAM;
568 } else if (o->proto == FIO_TYPE_UDP) {
569 domain = AF_INET;
570 type = SOCK_DGRAM;
571 } else if (o->proto == FIO_TYPE_UDP_V6) {
572 domain = AF_INET6;
573 type = SOCK_DGRAM;
574 } else if (o->proto == FIO_TYPE_UNIX) {
575 domain = AF_UNIX;
576 type = SOCK_STREAM;
577 } else {
578 log_err("fio: bad network type %d\n", o->proto);
579 f->fd = -1;
580 return 1;
581 }
582
583 f->fd = socket(domain, type, 0);
584 if (f->fd < 0) {
585 td_verror(td, errno, "socket");
586 return 1;
587 }
588
589#ifdef CONFIG_TCP_NODELAY
590 if (o->nodelay && is_tcp(o)) {
591 int optval = 1;
592
593 if (setsockopt(f->fd, IPPROTO_TCP, TCP_NODELAY, (void *) &optval, sizeof(int)) < 0) {
594 log_err("fio: cannot set TCP_NODELAY option on socket (%s), disable with 'nodelay=0'\n", strerror(errno));
595 return 1;
596 }
597 }
598#endif
599
600 if (is_udp(o)) {
601 if (!fio_netio_is_multicast(td->o.filename))
602 return 0;
603 if (is_ipv6(o)) {
604 log_err("fio: multicast not supported on IPv6\n");
605 close(f->fd);
606 return 1;
607 }
608
609 if (o->intfc) {
610 struct in_addr interface_addr;
611
612 if (inet_aton(o->intfc, &interface_addr) == 0) {
613 log_err("fio: interface not valid interface IP\n");
614 close(f->fd);
615 return 1;
616 }
617 if (setsockopt(f->fd, IPPROTO_IP, IP_MULTICAST_IF, (const char*)&interface_addr, sizeof(interface_addr)) < 0) {
618 td_verror(td, errno, "setsockopt IP_MULTICAST_IF");
619 close(f->fd);
620 return 1;
621 }
622 }
623 if (setsockopt(f->fd, IPPROTO_IP, IP_MULTICAST_TTL, (const char*)&o->ttl, sizeof(o->ttl)) < 0) {
624 td_verror(td, errno, "setsockopt IP_MULTICAST_TTL");
625 close(f->fd);
626 return 1;
627 }
628 return 0;
629 } else if (o->proto == FIO_TYPE_TCP) {
630 socklen_t len = sizeof(nd->addr);
631
632 if (connect(f->fd, (struct sockaddr *) &nd->addr, len) < 0) {
633 td_verror(td, errno, "connect");
634 close(f->fd);
635 return 1;
636 }
637 } else if (o->proto == FIO_TYPE_TCP_V6) {
638 socklen_t len = sizeof(nd->addr6);
639
640 if (connect(f->fd, (struct sockaddr *) &nd->addr6, len) < 0) {
641 td_verror(td, errno, "connect");
642 close(f->fd);
643 return 1;
644 }
645
646 } else {
647 struct sockaddr_un *addr = &nd->addr_un;
648 socklen_t len;
649
650 len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
651
652 if (connect(f->fd, (struct sockaddr *) addr, len) < 0) {
653 td_verror(td, errno, "connect");
654 close(f->fd);
655 return 1;
656 }
657 }
658
659 return 0;
660}
661
662static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
663{
664 struct netio_data *nd = td->io_ops->data;
665 struct netio_options *o = td->eo;
666 socklen_t socklen;
667 int state;
668
669 if (is_udp(o)) {
670 f->fd = nd->listenfd;
671 return 0;
672 }
673
674 state = td->runstate;
675 td_set_runstate(td, TD_SETTING_UP);
676
677 log_info("fio: waiting for connection\n");
678
679 if (poll_wait(td, nd->listenfd, POLLIN) < 0)
680 goto err;
681
682 if (o->proto == FIO_TYPE_TCP) {
683 socklen = sizeof(nd->addr);
684 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
685 } else {
686 socklen = sizeof(nd->addr6);
687 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr6, &socklen);
688 }
689
690 if (f->fd < 0) {
691 td_verror(td, errno, "accept");
692 goto err;
693 }
694
695#ifdef CONFIG_TCP_NODELAY
696 if (o->nodelay && is_tcp(o)) {
697 int optval = 1;
698
699 if (setsockopt(f->fd, IPPROTO_TCP, TCP_NODELAY, (void *) &optval, sizeof(int)) < 0) {
700 log_err("fio: cannot set TCP_NODELAY option on socket (%s), disable with 'nodelay=0'\n", strerror(errno));
701 return 1;
702 }
703 }
704#endif
705
706 reset_all_stats(td);
707 td_set_runstate(td, state);
708 return 0;
709err:
710 td_set_runstate(td, state);
711 return 1;
712}
713
714static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
715{
716 struct netio_data *nd = td->io_ops->data;
717 struct netio_options *o = td->eo;
718 struct udp_close_msg msg;
719 struct sockaddr *to;
720 socklen_t len;
721 int ret;
722
723 if (is_ipv6(o)) {
724 to = (struct sockaddr *) &nd->addr6;
725 len = sizeof(nd->addr6);
726 } else {
727 to = (struct sockaddr *) &nd->addr;
728 len = sizeof(nd->addr);
729 }
730
731 msg.magic = htonl(FIO_LINK_OPEN_CLOSE_MAGIC);
732 msg.cmd = htonl(FIO_LINK_CLOSE);
733
734 ret = sendto(f->fd, (void *) &msg, sizeof(msg), MSG_WAITALL, to, len);
735 if (ret < 0)
736 td_verror(td, errno, "sendto udp link close");
737}
738
739static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
740{
741 struct netio_options *o = td->eo;
742
743 /*
744 * If this is an UDP connection, notify the receiver that we are
745 * closing down the link
746 */
747 if (is_udp(o))
748 fio_netio_udp_close(td, f);
749
750 return generic_close_file(td, f);
751}
752
753static int fio_netio_udp_recv_open(struct thread_data *td, struct fio_file *f)
754{
755 struct netio_data *nd = td->io_ops->data;
756 struct netio_options *o = td->eo;
757 struct udp_close_msg msg;
758 struct sockaddr *to;
759 socklen_t len;
760 int ret;
761
762 if (is_ipv6(o)) {
763 len = sizeof(nd->addr6);
764 to = (struct sockaddr *) &nd->addr6;
765 } else {
766 len = sizeof(nd->addr);
767 to = (struct sockaddr *) &nd->addr;
768 }
769
770 ret = recvfrom(f->fd, (void *) &msg, sizeof(msg), MSG_WAITALL, to, &len);
771 if (ret < 0) {
772 td_verror(td, errno, "recvfrom udp link open");
773 return ret;
774 }
775
776 if (ntohl(msg.magic) != FIO_LINK_OPEN_CLOSE_MAGIC ||
777 ntohl(msg.cmd) != FIO_LINK_OPEN) {
778 log_err("fio: bad udp open magic %x/%x\n", ntohl(msg.magic),
779 ntohl(msg.cmd));
780 return -1;
781 }
782
783 return 0;
784}
785
786static int fio_netio_udp_send_open(struct thread_data *td, struct fio_file *f)
787{
788 struct netio_data *nd = td->io_ops->data;
789 struct netio_options *o = td->eo;
790 struct udp_close_msg msg;
791 struct sockaddr *to;
792 socklen_t len;
793 int ret;
794
795 if (is_ipv6(o)) {
796 len = sizeof(nd->addr6);
797 to = (struct sockaddr *) &nd->addr6;
798 } else {
799 len = sizeof(nd->addr);
800 to = (struct sockaddr *) &nd->addr;
801 }
802
803 msg.magic = htonl(FIO_LINK_OPEN_CLOSE_MAGIC);
804 msg.cmd = htonl(FIO_LINK_OPEN);
805
806 ret = sendto(f->fd, (void *) &msg, sizeof(msg), MSG_WAITALL, to, len);
807 if (ret < 0) {
808 td_verror(td, errno, "sendto udp link open");
809 return ret;
810 }
811
812 return 0;
813}
814
815static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
816{
817 int ret;
818 struct netio_options *o = td->eo;
819
820 if (o->listen)
821 ret = fio_netio_accept(td, f);
822 else
823 ret = fio_netio_connect(td, f);
824
825 if (ret) {
826 f->fd = -1;
827 return ret;
828 }
829
830 if (is_udp(o)) {
831 if (td_write(td))
832 ret = fio_netio_udp_send_open(td, f);
833 else {
834 int state;
835
836 state = td->runstate;
837 td_set_runstate(td, TD_SETTING_UP);
838 ret = fio_netio_udp_recv_open(td, f);
839 td_set_runstate(td, state);
840 }
841 }
842
843 if (ret)
844 fio_netio_close_file(td, f);
845
846 return ret;
847}
848
849static int fio_fill_addr(struct thread_data *td, const char *host, int af,
850 void *dst, struct addrinfo **res)
851{
852 struct netio_options *o = td->eo;
853 struct addrinfo hints;
854 int ret;
855
856 if (inet_pton(af, host, dst))
857 return 0;
858
859 memset(&hints, 0, sizeof(hints));
860
861 if (is_tcp(o))
862 hints.ai_socktype = SOCK_STREAM;
863 else
864 hints.ai_socktype = SOCK_DGRAM;
865
866 if (is_ipv6(o))
867 hints.ai_family = AF_INET6;
868 else
869 hints.ai_family = AF_INET;
870
871 ret = getaddrinfo(host, NULL, &hints, res);
872 if (ret) {
873 int e = EINVAL;
874 char str[128];
875
876 if (ret == EAI_SYSTEM)
877 e = errno;
878
879 snprintf(str, sizeof(str), "getaddrinfo: %s", gai_strerror(ret));
880 td_verror(td, e, str);
881 return 1;
882 }
883
884 return 0;
885}
886
887static int fio_netio_setup_connect_inet(struct thread_data *td,
888 const char *host, unsigned short port)
889{
890 struct netio_data *nd = td->io_ops->data;
891 struct netio_options *o = td->eo;
892 struct addrinfo *res = NULL;
893 void *dst, *src;
894 int af, len;
895
896 if (!host) {
897 log_err("fio: connect with no host to connect to.\n");
898 if (td_read(td))
899 log_err("fio: did you forget to set 'listen'?\n");
900
901 td_verror(td, EINVAL, "no hostname= set");
902 return 1;
903 }
904
905 nd->addr.sin_family = AF_INET;
906 nd->addr.sin_port = htons(port);
907 nd->addr6.sin6_family = AF_INET6;
908 nd->addr6.sin6_port = htons(port);
909
910 if (is_ipv6(o)) {
911 af = AF_INET6;
912 dst = &nd->addr6;
913 } else {
914 af = AF_INET;
915 dst = &nd->addr;
916 }
917
918 if (fio_fill_addr(td, host, af, dst, &res))
919 return 1;
920
921 if (!res)
922 return 0;
923
924 if (is_ipv6(o)) {
925 len = sizeof(nd->addr6.sin6_addr);
926 src = &((struct sockaddr_in6 *) res->ai_addr)->sin6_addr;
927 } else {
928 len = sizeof(nd->addr.sin_addr);
929 src = &((struct sockaddr_in *) res->ai_addr)->sin_addr;
930 }
931
932 memcpy(dst, src, len);
933 freeaddrinfo(res);
934 return 0;
935}
936
937static int fio_netio_setup_connect_unix(struct thread_data *td,
938 const char *path)
939{
940 struct netio_data *nd = td->io_ops->data;
941 struct sockaddr_un *soun = &nd->addr_un;
942
943 soun->sun_family = AF_UNIX;
944 strcpy(soun->sun_path, path);
945 return 0;
946}
947
948static int fio_netio_setup_connect(struct thread_data *td)
949{
950 struct netio_options *o = td->eo;
951
952 if (is_udp(o) || is_tcp(o))
953 return fio_netio_setup_connect_inet(td, td->o.filename,o->port);
954 else
955 return fio_netio_setup_connect_unix(td, td->o.filename);
956}
957
958static int fio_netio_setup_listen_unix(struct thread_data *td, const char *path)
959{
960 struct netio_data *nd = td->io_ops->data;
961 struct sockaddr_un *addr = &nd->addr_un;
962 mode_t mode;
963 int len, fd;
964
965 fd = socket(AF_UNIX, SOCK_STREAM, 0);
966 if (fd < 0) {
967 log_err("fio: socket: %s\n", strerror(errno));
968 return -1;
969 }
970
971 mode = umask(000);
972
973 memset(addr, 0, sizeof(*addr));
974 addr->sun_family = AF_UNIX;
975 strcpy(addr->sun_path, path);
976 unlink(path);
977
978 len = sizeof(addr->sun_family) + strlen(path) + 1;
979
980 if (bind(fd, (struct sockaddr *) addr, len) < 0) {
981 log_err("fio: bind: %s\n", strerror(errno));
982 close(fd);
983 return -1;
984 }
985
986 umask(mode);
987 nd->listenfd = fd;
988 return 0;
989}
990
991static int fio_netio_setup_listen_inet(struct thread_data *td, short port)
992{
993 struct netio_data *nd = td->io_ops->data;
994 struct netio_options *o = td->eo;
995 struct ip_mreq mr;
996 struct sockaddr_in sin;
997 struct sockaddr_in6 sin6;
998 struct sockaddr *saddr;
999 int fd, opt, type, domain;
1000 socklen_t len;
1001
1002 memset(&sin, 0, sizeof(sin));
1003 memset(&sin6, 0, sizeof(sin6));
1004
1005 if (o->proto == FIO_TYPE_TCP) {
1006 type = SOCK_STREAM;
1007 domain = AF_INET;
1008 } else if (o->proto == FIO_TYPE_TCP_V6) {
1009 type = SOCK_STREAM;
1010 domain = AF_INET6;
1011 } else if (o->proto == FIO_TYPE_UDP) {
1012 type = SOCK_DGRAM;
1013 domain = AF_INET;
1014 } else if (o->proto == FIO_TYPE_UDP_V6) {
1015 type = SOCK_DGRAM;
1016 domain = AF_INET6;
1017 } else {
1018 log_err("fio: unknown proto %d\n", o->proto);
1019 return 1;
1020 }
1021
1022 fd = socket(domain, type, 0);
1023 if (fd < 0) {
1024 td_verror(td, errno, "socket");
1025 return 1;
1026 }
1027
1028 opt = 1;
1029 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void *) &opt, sizeof(opt)) < 0) {
1030 td_verror(td, errno, "setsockopt");
1031 close(fd);
1032 return 1;
1033 }
1034#ifdef SO_REUSEPORT
1035 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, (void *) &opt, sizeof(opt)) < 0) {
1036 td_verror(td, errno, "setsockopt");
1037 close(fd);
1038 return 1;
1039 }
1040#endif
1041
1042 if (td->o.filename) {
1043 if (!is_udp(o) || !fio_netio_is_multicast(td->o.filename)) {
1044 log_err("fio: hostname not valid for non-multicast inbound network IO\n");
1045 close(fd);
1046 return 1;
1047 }
1048 if (is_ipv6(o)) {
1049 log_err("fio: IPv6 not supported for multicast network IO");
1050 close(fd);
1051 return 1;
1052 }
1053
1054 inet_aton(td->o.filename, &sin.sin_addr);
1055
1056 mr.imr_multiaddr = sin.sin_addr;
1057 if (o->intfc) {
1058 if (inet_aton(o->intfc, &mr.imr_interface) == 0) {
1059 log_err("fio: interface not valid interface IP\n");
1060 close(fd);
1061 return 1;
1062 }
1063 } else {
1064 mr.imr_interface.s_addr = htonl(INADDR_ANY);
1065 }
1066
1067 if (setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (const char*)&mr, sizeof(mr)) < 0) {
1068 td_verror(td, errno, "setsockopt IP_ADD_MEMBERSHIP");
1069 close(fd);
1070 return 1;
1071 }
1072 }
1073
1074 if (!is_ipv6(o)) {
1075 saddr = (struct sockaddr *) &nd->addr;
1076 len = sizeof(nd->addr);
1077
1078 nd->addr.sin_family = AF_INET;
1079 nd->addr.sin_addr.s_addr = sin.sin_addr.s_addr ? sin.sin_addr.s_addr : htonl(INADDR_ANY);
1080 nd->addr.sin_port = htons(port);
1081 } else {
1082 saddr = (struct sockaddr *) &nd->addr6;
1083 len = sizeof(nd->addr6);
1084
1085 nd->addr6.sin6_family = AF_INET6;
1086 nd->addr6.sin6_addr = sin6.sin6_addr.s6_addr ? sin6.sin6_addr : in6addr_any;
1087 nd->addr6.sin6_port = htons(port);
1088 }
1089
1090 if (bind(fd, saddr, len) < 0) {
1091 td_verror(td, errno, "bind");
1092 return 1;
1093 }
1094
1095 nd->listenfd = fd;
1096 return 0;
1097}
1098
1099static int fio_netio_setup_listen(struct thread_data *td)
1100{
1101 struct netio_data *nd = td->io_ops->data;
1102 struct netio_options *o = td->eo;
1103 int ret;
1104
1105 if (is_udp(o) || is_tcp(o))
1106 ret = fio_netio_setup_listen_inet(td, o->port);
1107 else
1108 ret = fio_netio_setup_listen_unix(td, td->o.filename);
1109
1110 if (ret)
1111 return ret;
1112 if (is_udp(o))
1113 return 0;
1114
1115 if (listen(nd->listenfd, 10) < 0) {
1116 td_verror(td, errno, "listen");
1117 nd->listenfd = -1;
1118 return 1;
1119 }
1120
1121 return 0;
1122}
1123
1124static int fio_netio_init(struct thread_data *td)
1125{
1126 struct netio_options *o = td->eo;
1127 int ret;
1128
1129#ifdef WIN32
1130 WSADATA wsd;
1131 WSAStartup(MAKEWORD(2,2), &wsd);
1132#endif
1133
1134 if (td_random(td)) {
1135 log_err("fio: network IO can't be random\n");
1136 return 1;
1137 }
1138
1139 if (o->proto == FIO_TYPE_UNIX && o->port) {
1140 log_err("fio: network IO port not valid with unix socket\n");
1141 return 1;
1142 } else if (o->proto != FIO_TYPE_UNIX && !o->port) {
1143 log_err("fio: network IO requires port for tcp or udp\n");
1144 return 1;
1145 }
1146
1147 if (!is_tcp(o)) {
1148 if (o->listen) {
1149 log_err("fio: listen only valid for TCP proto IO\n");
1150 return 1;
1151 }
1152 if (td_rw(td)) {
1153 log_err("fio: datagram network connections must be"
1154 " read OR write\n");
1155 return 1;
1156 }
1157 if (o->proto == FIO_TYPE_UNIX && !td->o.filename) {
1158 log_err("fio: UNIX sockets need host/filename\n");
1159 return 1;
1160 }
1161 o->listen = td_read(td);
1162 }
1163
1164 if (o->listen)
1165 ret = fio_netio_setup_listen(td);
1166 else
1167 ret = fio_netio_setup_connect(td);
1168
1169 return ret;
1170}
1171
1172static void fio_netio_cleanup(struct thread_data *td)
1173{
1174 struct netio_data *nd = td->io_ops->data;
1175
1176 if (nd) {
1177 if (nd->listenfd != -1)
1178 close(nd->listenfd);
1179 if (nd->pipes[0] != -1)
1180 close(nd->pipes[0]);
1181 if (nd->pipes[1] != -1)
1182 close(nd->pipes[1]);
1183
1184 free(nd);
1185 }
1186}
1187
1188static int fio_netio_setup(struct thread_data *td)
1189{
1190 struct netio_data *nd;
1191
1192 if (!td->files_index) {
1193 add_file(td, td->o.filename ?: "net");
1194 td->o.nr_files = td->o.nr_files ?: 1;
1195 }
1196
1197 if (!td->io_ops->data) {
1198 nd = malloc(sizeof(*nd));;
1199
1200 memset(nd, 0, sizeof(*nd));
1201 nd->listenfd = -1;
1202 nd->pipes[0] = nd->pipes[1] = -1;
1203 td->io_ops->data = nd;
1204 }
1205
1206 return 0;
1207}
1208
1209static void fio_netio_terminate(struct thread_data *td)
1210{
1211 kill(td->pid, SIGUSR2);
1212}
1213
1214#ifdef CONFIG_LINUX_SPLICE
1215static int fio_netio_setup_splice(struct thread_data *td)
1216{
1217 struct netio_data *nd;
1218
1219 fio_netio_setup(td);
1220
1221 nd = td->io_ops->data;
1222 if (nd) {
1223 if (pipe(nd->pipes) < 0)
1224 return 1;
1225
1226 nd->use_splice = 1;
1227 return 0;
1228 }
1229
1230 return 1;
1231}
1232
1233static struct ioengine_ops ioengine_splice = {
1234 .name = "netsplice",
1235 .version = FIO_IOOPS_VERSION,
1236 .prep = fio_netio_prep,
1237 .queue = fio_netio_queue,
1238 .setup = fio_netio_setup_splice,
1239 .init = fio_netio_init,
1240 .cleanup = fio_netio_cleanup,
1241 .open_file = fio_netio_open_file,
1242 .close_file = fio_netio_close_file,
1243 .terminate = fio_netio_terminate,
1244 .options = options,
1245 .option_struct_size = sizeof(struct netio_options),
1246 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
1247 FIO_PIPEIO,
1248};
1249#endif
1250
1251static struct ioengine_ops ioengine_rw = {
1252 .name = "net",
1253 .version = FIO_IOOPS_VERSION,
1254 .prep = fio_netio_prep,
1255 .queue = fio_netio_queue,
1256 .setup = fio_netio_setup,
1257 .init = fio_netio_init,
1258 .cleanup = fio_netio_cleanup,
1259 .open_file = fio_netio_open_file,
1260 .close_file = fio_netio_close_file,
1261 .terminate = fio_netio_terminate,
1262 .options = options,
1263 .option_struct_size = sizeof(struct netio_options),
1264 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
1265 FIO_PIPEIO | FIO_BIT_BASED,
1266};
1267
1268static int str_hostname_cb(void *data, const char *input)
1269{
1270 struct netio_options *o = data;
1271
1272 if (o->td->o.filename)
1273 free(o->td->o.filename);
1274 o->td->o.filename = strdup(input);
1275 return 0;
1276}
1277
1278static void fio_init fio_netio_register(void)
1279{
1280 register_ioengine(&ioengine_rw);
1281#ifdef CONFIG_LINUX_SPLICE
1282 register_ioengine(&ioengine_splice);
1283#endif
1284}
1285
1286static void fio_exit fio_netio_unregister(void)
1287{
1288 unregister_ioengine(&ioengine_rw);
1289#ifdef CONFIG_LINUX_SPLICE
1290 unregister_ioengine(&ioengine_splice);
1291#endif
1292}
fio - Flexible IO Tester