engines/net.c

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