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