Private parameters for ioengines
[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 <errno.h>
11#include <assert.h>
12#include <netinet/in.h>
13#include <arpa/inet.h>
14#include <netdb.h>
15#include <sys/poll.h>
16#include <sys/types.h>
17#include <sys/stat.h>
18#include <sys/socket.h>
19#include <sys/un.h>
20
21#include "../fio.h"
22
23struct netio_data {
24 int listenfd;
25 int use_splice;
26 int pipes[2];
27 struct sockaddr_in addr;
28 struct sockaddr_un addr_un;
29};
30
31struct netio_options {
32 struct thread_data *td;
33 unsigned int port;
34 unsigned int proto;
35 unsigned int listen;
36};
37
38struct udp_close_msg {
39 uint32_t magic;
40 uint32_t cmd;
41};
42
43enum {
44 FIO_LINK_CLOSE = 0x89,
45 FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
46
47 FIO_TYPE_TCP = 1,
48 FIO_TYPE_UDP = 2,
49 FIO_TYPE_UNIX = 3,
50};
51
52static int str_hostname_cb(void *data, const char *input);
53static struct fio_option options[] = {
54 {
55 .name = "hostname",
56 .type = FIO_OPT_STR_STORE,
57 .cb = str_hostname_cb,
58 .help = "Hostname for net IO engine",
59 },
60 {
61 .name = "port",
62 .type = FIO_OPT_INT,
63 .off1 = offsetof(struct netio_options, port),
64 .minval = 1,
65 .maxval = 65535,
66 .help = "Port to use for TCP or UDP net connections",
67 },
68 {
69 .name = "protocol",
70 .alias = "proto",
71 .type = FIO_OPT_STR,
72 .off1 = offsetof(struct netio_options, proto),
73 .help = "Network protocol to use",
74 .def = "tcp",
75 .posval = {
76 { .ival = "tcp",
77 .oval = FIO_TYPE_TCP,
78 .help = "Transmission Control Protocol",
79 },
80 { .ival = "udp",
81 .oval = FIO_TYPE_UDP,
82 .help = "Unreliable Datagram Protocol",
83 },
84 { .ival = "unix",
85 .oval = FIO_TYPE_UNIX,
86 .help = "UNIX domain socket",
87 },
88 },
89 },
90 {
91 .name = "listen",
92 .type = FIO_OPT_STR_SET,
93 .off1 = offsetof(struct netio_options, listen),
94 .help = "Listen for incoming TCP connections",
95 },
96 {
97 .name = NULL,
98 },
99};
100
101/*
102 * Return -1 for error and 'nr events' for a positive number
103 * of events
104 */
105static int poll_wait(struct thread_data *td, int fd, short events)
106{
107 struct pollfd pfd;
108 int ret;
109
110 while (!td->terminate) {
111 pfd.fd = fd;
112 pfd.events = events;
113 ret = poll(&pfd, 1, -1);
114 if (ret < 0) {
115 if (errno == EINTR)
116 break;
117
118 td_verror(td, errno, "poll");
119 return -1;
120 } else if (!ret)
121 continue;
122
123 break;
124 }
125
126 if (pfd.revents & events)
127 return 1;
128
129 return -1;
130}
131
132static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
133{
134 struct netio_options *o = td->eo;
135
136 /*
137 * Make sure we don't see spurious reads to a receiver, and vice versa
138 */
139 if (o->proto == FIO_TYPE_TCP)
140 return 0;
141
142 if ((o->listen && io_u->ddir == DDIR_WRITE) ||
143 (!o->listen && io_u->ddir == DDIR_READ)) {
144 td_verror(td, EINVAL, "bad direction");
145 return 1;
146 }
147
148 return 0;
149}
150
151#ifdef FIO_HAVE_SPLICE
152static int splice_io_u(int fdin, int fdout, unsigned int len)
153{
154 int bytes = 0;
155
156 while (len) {
157 int ret = splice(fdin, NULL, fdout, NULL, len, 0);
158
159 if (ret < 0) {
160 if (!bytes)
161 bytes = ret;
162
163 break;
164 } else if (!ret)
165 break;
166
167 bytes += ret;
168 len -= ret;
169 }
170
171 return bytes;
172}
173
174/*
175 * Receive bytes from a socket and fill them into the internal pipe
176 */
177static int splice_in(struct thread_data *td, struct io_u *io_u)
178{
179 struct netio_data *nd = td->io_ops->data;
180
181 return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
182}
183
184/*
185 * Transmit 'len' bytes from the internal pipe
186 */
187static int splice_out(struct thread_data *td, struct io_u *io_u,
188 unsigned int len)
189{
190 struct netio_data *nd = td->io_ops->data;
191
192 return splice_io_u(nd->pipes[0], io_u->file->fd, len);
193}
194
195static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
196{
197 struct iovec iov = {
198 .iov_base = io_u->xfer_buf,
199 .iov_len = len,
200 };
201 int bytes = 0;
202
203 while (iov.iov_len) {
204 int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
205
206 if (ret < 0) {
207 if (!bytes)
208 bytes = ret;
209 break;
210 } else if (!ret)
211 break;
212
213 iov.iov_len -= ret;
214 iov.iov_base += ret;
215 bytes += ret;
216 }
217
218 return bytes;
219
220}
221
222/*
223 * vmsplice() pipe to io_u buffer
224 */
225static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u,
226 unsigned int len)
227{
228 struct netio_data *nd = td->io_ops->data;
229
230 return vmsplice_io_u(io_u, nd->pipes[0], len);
231}
232
233/*
234 * vmsplice() io_u to pipe
235 */
236static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u)
237{
238 struct netio_data *nd = td->io_ops->data;
239
240 return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
241}
242
243/*
244 * splice receive - transfer socket data into a pipe using splice, then map
245 * that pipe data into the io_u using vmsplice.
246 */
247static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
248{
249 int ret;
250
251 ret = splice_in(td, io_u);
252 if (ret > 0)
253 return vmsplice_io_u_out(td, io_u, ret);
254
255 return ret;
256}
257
258/*
259 * splice transmit - map data from the io_u into a pipe by using vmsplice,
260 * then transfer that pipe to a socket using splice.
261 */
262static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
263{
264 int ret;
265
266 ret = vmsplice_io_u_in(td, io_u);
267 if (ret > 0)
268 return splice_out(td, io_u, ret);
269
270 return ret;
271}
272#else
273static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
274{
275 errno = EOPNOTSUPP;
276 return -1;
277}
278
279static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
280{
281 errno = EOPNOTSUPP;
282 return -1;
283}
284#endif
285
286static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
287{
288 struct netio_data *nd = td->io_ops->data;
289 struct netio_options *o = td->eo;
290 int ret, flags = OS_MSG_DONTWAIT;
291
292 do {
293 if (o->proto == FIO_TYPE_UDP) {
294 struct sockaddr *to = (struct sockaddr *) &nd->addr;
295
296 ret = sendto(io_u->file->fd, io_u->xfer_buf,
297 io_u->xfer_buflen, flags, to,
298 sizeof(*to));
299 } else {
300 /*
301 * if we are going to write more, set MSG_MORE
302 */
303#ifdef MSG_MORE
304 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
305 td->o.size)
306 flags |= MSG_MORE;
307#endif
308 ret = send(io_u->file->fd, io_u->xfer_buf,
309 io_u->xfer_buflen, flags);
310 }
311 if (ret > 0)
312 break;
313
314 ret = poll_wait(td, io_u->file->fd, POLLOUT);
315 if (ret <= 0)
316 break;
317
318 flags &= ~OS_MSG_DONTWAIT;
319 } while (1);
320
321 return ret;
322}
323
324static int is_udp_close(struct io_u *io_u, int len)
325{
326 struct udp_close_msg *msg;
327
328 if (len != sizeof(struct udp_close_msg))
329 return 0;
330
331 msg = io_u->xfer_buf;
332 if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
333 return 0;
334 if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
335 return 0;
336
337 return 1;
338}
339
340static int fio_netio_recv(struct thread_data *td, struct io_u *io_u)
341{
342 struct netio_data *nd = td->io_ops->data;
343 struct netio_options *o = td->eo;
344 int ret, flags = OS_MSG_DONTWAIT;
345
346 do {
347 if (o->proto == FIO_TYPE_UDP) {
348 fio_socklen_t len = sizeof(nd->addr);
349 struct sockaddr *from = (struct sockaddr *) &nd->addr;
350
351 ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
352 io_u->xfer_buflen, flags, from, &len);
353 if (is_udp_close(io_u, ret)) {
354 td->done = 1;
355 return 0;
356 }
357 } else {
358 ret = recv(io_u->file->fd, io_u->xfer_buf,
359 io_u->xfer_buflen, flags);
360 }
361 if (ret > 0)
362 break;
363
364 ret = poll_wait(td, io_u->file->fd, POLLIN);
365 if (ret <= 0)
366 break;
367 flags &= ~OS_MSG_DONTWAIT;
368 flags |= MSG_WAITALL;
369 } while (1);
370
371 return ret;
372}
373
374static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
375{
376 struct netio_data *nd = td->io_ops->data;
377 struct netio_options *o = td->eo;
378 int ret;
379
380 fio_ro_check(td, io_u);
381
382 if (io_u->ddir == DDIR_WRITE) {
383 if (!nd->use_splice || o->proto == FIO_TYPE_UDP ||
384 o->proto == FIO_TYPE_UNIX)
385 ret = fio_netio_send(td, io_u);
386 else
387 ret = fio_netio_splice_out(td, io_u);
388 } else if (io_u->ddir == DDIR_READ) {
389 if (!nd->use_splice || o->proto == FIO_TYPE_UDP ||
390 o->proto == FIO_TYPE_UNIX)
391 ret = fio_netio_recv(td, io_u);
392 else
393 ret = fio_netio_splice_in(td, io_u);
394 } else
395 ret = 0; /* must be a SYNC */
396
397 if (ret != (int) io_u->xfer_buflen) {
398 if (ret >= 0) {
399 io_u->resid = io_u->xfer_buflen - ret;
400 io_u->error = 0;
401 return FIO_Q_COMPLETED;
402 } else {
403 int err = errno;
404
405 if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
406 return FIO_Q_BUSY;
407
408 io_u->error = err;
409 }
410 }
411
412 if (io_u->error)
413 td_verror(td, io_u->error, "xfer");
414
415 return FIO_Q_COMPLETED;
416}
417
418static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
419{
420 struct netio_data *nd = td->io_ops->data;
421 struct netio_options *o = td->eo;
422 int type, domain;
423
424 if (o->proto == FIO_TYPE_TCP) {
425 domain = AF_INET;
426 type = SOCK_STREAM;
427 } else if (o->proto == FIO_TYPE_UDP) {
428 domain = AF_INET;
429 type = SOCK_DGRAM;
430 } else if (o->proto == FIO_TYPE_UNIX) {
431 domain = AF_UNIX;
432 type = SOCK_STREAM;
433 } else {
434 log_err("fio: bad network type %d\n", o->proto);
435 f->fd = -1;
436 return 1;
437 }
438
439 f->fd = socket(domain, type, 0);
440 if (f->fd < 0) {
441 td_verror(td, errno, "socket");
442 return 1;
443 }
444
445 if (o->proto == FIO_TYPE_UDP)
446 return 0;
447 else if (o->proto == FIO_TYPE_TCP) {
448 fio_socklen_t len = sizeof(nd->addr);
449
450 if (connect(f->fd, (struct sockaddr *) &nd->addr, len) < 0) {
451 td_verror(td, errno, "connect");
452 close(f->fd);
453 return 1;
454 }
455 } else {
456 struct sockaddr_un *addr = &nd->addr_un;
457 fio_socklen_t len;
458
459 len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
460
461 if (connect(f->fd, (struct sockaddr *) addr, len) < 0) {
462 td_verror(td, errno, "connect");
463 close(f->fd);
464 return 1;
465 }
466 }
467
468 return 0;
469}
470
471static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
472{
473 struct netio_data *nd = td->io_ops->data;
474 struct netio_options *o = td->eo;
475 fio_socklen_t socklen = sizeof(nd->addr);
476
477 if (o->proto == FIO_TYPE_UDP) {
478 f->fd = nd->listenfd;
479 return 0;
480 }
481
482 log_info("fio: waiting for connection\n");
483
484 if (poll_wait(td, nd->listenfd, POLLIN) < 0)
485 return 1;
486
487 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
488 if (f->fd < 0) {
489 td_verror(td, errno, "accept");
490 return 1;
491 }
492
493 return 0;
494}
495
496static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
497{
498 int ret;
499
500 if (td_read(td))
501 ret = fio_netio_accept(td, f);
502 else
503 ret = fio_netio_connect(td, f);
504
505 if (ret)
506 f->fd = -1;
507 return ret;
508}
509
510static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
511{
512 struct netio_data *nd = td->io_ops->data;
513 struct udp_close_msg msg;
514 struct sockaddr *to = (struct sockaddr *) &nd->addr;
515 int ret;
516
517 msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
518 msg.cmd = htonl(FIO_LINK_CLOSE);
519
520 ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
521 sizeof(nd->addr));
522 if (ret < 0)
523 td_verror(td, errno, "sendto udp link close");
524}
525
526static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
527{
528 struct netio_options *o = td->eo;
529
530 /*
531 * If this is an UDP connection, notify the receiver that we are
532 * closing down the link
533 */
534 if (o->proto == FIO_TYPE_UDP)
535 fio_netio_udp_close(td, f);
536
537 return generic_close_file(td, f);
538}
539
540static int fio_netio_setup_connect_inet(struct thread_data *td,
541 const char *host, unsigned short port)
542{
543 struct netio_data *nd = td->io_ops->data;
544
545 nd->addr.sin_family = AF_INET;
546 nd->addr.sin_port = htons(port);
547
548 if (inet_aton(host, &nd->addr.sin_addr) != 1) {
549 struct hostent *hent;
550
551 hent = gethostbyname(host);
552 if (!hent) {
553 td_verror(td, errno, "gethostbyname");
554 return 1;
555 }
556
557 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
558 }
559
560 return 0;
561}
562
563static int fio_netio_setup_connect_unix(struct thread_data *td,
564 const char *path)
565{
566 struct netio_data *nd = td->io_ops->data;
567 struct sockaddr_un *soun = &nd->addr_un;
568
569 soun->sun_family = AF_UNIX;
570 strcpy(soun->sun_path, path);
571 return 0;
572}
573
574static int fio_netio_setup_connect(struct thread_data *td)
575{
576 struct netio_options *o = td->eo;
577
578 if (o->proto == FIO_TYPE_UDP || o->proto == FIO_TYPE_TCP)
579 return fio_netio_setup_connect_inet(td, td->o.filename,o->port);
580 else
581 return fio_netio_setup_connect_unix(td, td->o.filename);
582}
583
584static int fio_netio_setup_listen_unix(struct thread_data *td, const char *path)
585{
586 struct netio_data *nd = td->io_ops->data;
587 struct sockaddr_un *addr = &nd->addr_un;
588 mode_t mode;
589 int len, fd;
590
591 fd = socket(AF_UNIX, SOCK_STREAM, 0);
592 if (fd < 0) {
593 log_err("fio: socket: %s\n", strerror(errno));
594 return -1;
595 }
596
597 mode = umask(000);
598
599 memset(addr, 0, sizeof(*addr));
600 addr->sun_family = AF_UNIX;
601 strcpy(addr->sun_path, path);
602 unlink(path);
603
604 len = sizeof(addr->sun_family) + strlen(path) + 1;
605
606 if (bind(fd, (struct sockaddr *) addr, len) < 0) {
607 log_err("fio: bind: %s\n", strerror(errno));
608 close(fd);
609 return -1;
610 }
611
612 umask(mode);
613 nd->listenfd = fd;
614 return 0;
615}
616
617static int fio_netio_setup_listen_inet(struct thread_data *td, short port)
618{
619 struct netio_data *nd = td->io_ops->data;
620 struct netio_options *o = td->eo;
621 int fd, opt, type;
622
623 if (o->proto == FIO_TYPE_TCP)
624 type = SOCK_STREAM;
625 else
626 type = SOCK_DGRAM;
627
628 fd = socket(AF_INET, type, 0);
629 if (fd < 0) {
630 td_verror(td, errno, "socket");
631 return 1;
632 }
633
634 opt = 1;
635 if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
636 td_verror(td, errno, "setsockopt");
637 return 1;
638 }
639#ifdef SO_REUSEPORT
640 if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
641 td_verror(td, errno, "setsockopt");
642 return 1;
643 }
644#endif
645
646 nd->addr.sin_family = AF_INET;
647 nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
648 nd->addr.sin_port = htons(port);
649
650 if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
651 td_verror(td, errno, "bind");
652 return 1;
653 }
654
655 nd->listenfd = fd;
656 return 0;
657}
658
659static int fio_netio_setup_listen(struct thread_data *td)
660{
661 struct netio_data *nd = td->io_ops->data;
662 struct netio_options *o = td->eo;
663 int ret;
664
665 if (o->proto == FIO_TYPE_UDP || o->proto == FIO_TYPE_TCP)
666 ret = fio_netio_setup_listen_inet(td, o->port);
667 else
668 ret = fio_netio_setup_listen_unix(td, td->o.filename);
669
670 if (ret)
671 return ret;
672 if (o->proto == FIO_TYPE_UDP)
673 return 0;
674
675 if (listen(nd->listenfd, 10) < 0) {
676 td_verror(td, errno, "listen");
677 nd->listenfd = -1;
678 return 1;
679 }
680
681 return 0;
682}
683
684static int fio_netio_init(struct thread_data *td)
685{
686 struct netio_options *o = td->eo;
687 int ret;
688
689 if (td_random(td)) {
690 log_err("fio: network IO can't be random\n");
691 return 1;
692 }
693
694 if (o->proto == FIO_TYPE_UNIX && o->port) {
695 log_err("fio: network IO port not valid with unix socket\n");
696 return 1;
697 } else if (o->proto != FIO_TYPE_UNIX && !o->port) {
698 log_err("fio: network IO requires port for tcp or udp\n");
699 return 1;
700 }
701
702 if (o->proto != FIO_TYPE_TCP) {
703 if (o->listen) {
704 log_err("fio: listen only valid for TCP proto IO\n");
705 return 1;
706 }
707 if (td_rw(td)) {
708 log_err("fio: datagram network connections must be"
709 " read OR write\n");
710 return 1;
711 }
712 o->listen = td_read(td);
713 }
714
715 if (o->proto != FIO_TYPE_UNIX && o->listen && td->o.filename) {
716 log_err("fio: hostname not valid for inbound network IO\n");
717 return 1;
718 }
719
720 if (o->listen)
721 ret = fio_netio_setup_listen(td);
722 else
723 ret = fio_netio_setup_connect(td);
724
725 return ret;
726}
727
728static void fio_netio_cleanup(struct thread_data *td)
729{
730 struct netio_data *nd = td->io_ops->data;
731
732 if (nd) {
733 if (nd->listenfd != -1)
734 close(nd->listenfd);
735 if (nd->pipes[0] != -1)
736 close(nd->pipes[0]);
737 if (nd->pipes[1] != -1)
738 close(nd->pipes[1]);
739
740 free(nd);
741 }
742}
743
744static int fio_netio_setup(struct thread_data *td)
745{
746 struct netio_data *nd;
747
748 if (!td->files_index) {
749 add_file(td, td->o.filename ?: "net");
750 td->o.nr_files = td->o.nr_files ?: 1;
751 }
752
753 if (!td->io_ops->data) {
754 nd = malloc(sizeof(*nd));;
755
756 memset(nd, 0, sizeof(*nd));
757 nd->listenfd = -1;
758 nd->pipes[0] = nd->pipes[1] = -1;
759 td->io_ops->data = nd;
760 }
761
762 return 0;
763}
764
765#ifdef FIO_HAVE_SPLICE
766static int fio_netio_setup_splice(struct thread_data *td)
767{
768 struct netio_data *nd;
769
770 fio_netio_setup(td);
771
772 nd = td->io_ops->data;
773 if (nd) {
774 if (pipe(nd->pipes) < 0)
775 return 1;
776
777 nd->use_splice = 1;
778 return 0;
779 }
780
781 return 1;
782}
783
784static struct ioengine_ops ioengine_splice = {
785 .name = "netsplice",
786 .version = FIO_IOOPS_VERSION,
787 .prep = fio_netio_prep,
788 .queue = fio_netio_queue,
789 .setup = fio_netio_setup_splice,
790 .init = fio_netio_init,
791 .cleanup = fio_netio_cleanup,
792 .open_file = fio_netio_open_file,
793 .close_file = generic_close_file,
794 .options = options,
795 .option_struct_size = sizeof(struct netio_options),
796 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
797 FIO_SIGTERM | FIO_PIPEIO,
798};
799#endif
800
801static struct ioengine_ops ioengine_rw = {
802 .name = "net",
803 .version = FIO_IOOPS_VERSION,
804 .prep = fio_netio_prep,
805 .queue = fio_netio_queue,
806 .setup = fio_netio_setup,
807 .init = fio_netio_init,
808 .cleanup = fio_netio_cleanup,
809 .open_file = fio_netio_open_file,
810 .close_file = fio_netio_close_file,
811 .options = options,
812 .option_struct_size = sizeof(struct netio_options),
813 .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
814 FIO_SIGTERM | FIO_PIPEIO,
815};
816
817static int str_hostname_cb(void *data, const char *input)
818{
819 struct netio_options *o = data;
820
821 if (o->td->o.filename)
822 free(o->td->o.filename);
823 o->td->o.filename = strdup(input);
824 return 0;
825}
826
827static void fio_init fio_netio_register(void)
828{
829 register_ioengine(&ioengine_rw);
830#ifdef FIO_HAVE_SPLICE
831 register_ioengine(&ioengine_splice);
832#endif
833}
834
835static void fio_exit fio_netio_unregister(void)
836{
837 unregister_ioengine(&ioengine_rw);
838#ifdef FIO_HAVE_SPLICE
839 unregister_ioengine(&ioengine_splice);
840#endif
841}