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 = 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 &= ~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 = MSG_DONTWAIT;
 280
 281         do {
 282                 if (nd->net_protocol == IPPROTO_UDP) {
 283                         socklen_t len = sizeof(nd->addr);
 284                         struct sockaddr *from = (struct sockaddr *) &nd->addr;
 285
 286                         ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
 287                                         io_u->xfer_buflen, flags, from, &len);
 288                         if (is_udp_close(io_u, ret)) {
 289                                 td->done = 1;
 290                                 return 0;
 291                         }
 292                 } else {
 293                         ret = recv(io_u->file->fd, io_u->xfer_buf,
 294                                         io_u->xfer_buflen, flags);
 295                 }
 296                 if (ret > 0)
 297                         break;
 298
 299                 ret = poll_wait(td, io_u->file->fd, POLLIN);
 300                 if (ret <= 0)
 301                         break;
 302                 flags &= ~MSG_DONTWAIT;
 303                 flags |= MSG_WAITALL;
 304         } while (1);
 305
 306         return ret;
 307 }
 308
 309 static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
 310 {
 311         struct netio_data *nd = td->io_ops->data;
 312         int ret;
 313
 314         fio_ro_check(td, io_u);
 315
 316         if (io_u->ddir == DDIR_WRITE) {
 317                 if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP)
 318                         ret = fio_netio_send(td, io_u);
 319                 else
 320                         ret = fio_netio_splice_out(td, io_u);
 321         } else if (io_u->ddir == DDIR_READ) {
 322                 if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP)
 323                         ret = fio_netio_recv(td, io_u);
 324                 else
 325                         ret = fio_netio_splice_in(td, io_u);
 326         } else
 327                 ret = 0;        /* must be a SYNC */
 328
 329         if (ret != (int) io_u->xfer_buflen) {
 330                 if (ret >= 0) {
 331                         io_u->resid = io_u->xfer_buflen - ret;
 332                         io_u->error = 0;
 333                         return FIO_Q_COMPLETED;
 334                 } else {
 335                         int err = errno;
 336
 337                         if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
 338                                 return FIO_Q_BUSY;
 339
 340                         io_u->error = err;
 341                 }
 342         }
 343
 344         if (io_u->error)
 345                 td_verror(td, io_u->error, "xfer");
 346
 347         return FIO_Q_COMPLETED;
 348 }
 349
 350 static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
 351 {
 352         struct netio_data *nd = td->io_ops->data;
 353         int type;
 354
 355         if (nd->net_protocol == IPPROTO_TCP)
 356                 type = SOCK_STREAM;
 357         else
 358                 type = SOCK_DGRAM;
 359
 360         f->fd = socket(AF_INET, type, nd->net_protocol);
 361         if (f->fd < 0) {
 362                 td_verror(td, errno, "socket");
 363                 return 1;
 364         }
 365
 366         if (nd->net_protocol == IPPROTO_UDP)
 367                 return 0;
 368
 369         if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
 370                 td_verror(td, errno, "connect");
 371                 return 1;
 372         }
 373
 374         return 0;
 375 }
 376
 377 static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
 378 {
 379         struct netio_data *nd = td->io_ops->data;
 380         socklen_t socklen = sizeof(nd->addr);
 381
 382         if (nd->net_protocol == IPPROTO_UDP) {
 383                 f->fd = nd->listenfd;
 384                 return 0;
 385         }
 386
 387         log_info("fio: waiting for connection\n");
 388
 389         if (poll_wait(td, nd->listenfd, POLLIN) < 0)
 390                 return 1;
 391
 392         f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
 393         if (f->fd < 0) {
 394                 td_verror(td, errno, "accept");
 395                 return 1;
 396         }
 397
 398         return 0;
 399 }
 400
 401 static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
 402 {
 403         if (td_read(td))
 404                 return fio_netio_accept(td, f);
 405         else
 406                 return fio_netio_connect(td, f);
 407 }
 408
 409 static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
 410 {
 411         struct netio_data *nd = td->io_ops->data;
 412         struct udp_close_msg msg;
 413         struct sockaddr *to = (struct sockaddr *) &nd->addr;
 414         int ret;
 415
 416         msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
 417         msg.cmd = htonl(FIO_LINK_CLOSE);
 418
 419         ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
 420                         sizeof(nd->addr));
 421         if (ret < 0)
 422                 td_verror(td, errno, "sendto udp link close");
 423 }
 424
 425 static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
 426 {
 427         struct netio_data *nd = td->io_ops->data;
 428
 429         /*
 430          * If this is an UDP connection, notify the receiver that we are
 431          * closing down the link
 432          */
 433         if (nd->net_protocol == IPPROTO_UDP)
 434                 fio_netio_udp_close(td, f);
 435
 436         return generic_close_file(td, f);
 437 }
 438
 439 static int fio_netio_setup_connect(struct thread_data *td, const char *host,
 440                                    unsigned short port)
 441 {
 442         struct netio_data *nd = td->io_ops->data;
 443
 444         nd->addr.sin_family = AF_INET;
 445         nd->addr.sin_port = htons(port);
 446
 447         if (inet_aton(host, &nd->addr.sin_addr) != 1) {
 448                 struct hostent *hent;
 449
 450                 hent = gethostbyname(host);
 451                 if (!hent) {
 452                         td_verror(td, errno, "gethostbyname");
 453                         return 1;
 454                 }
 455
 456                 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
 457         }
 458
 459         return 0;
 460 }
 461
 462 static int fio_netio_setup_listen(struct thread_data *td, short port)
 463 {
 464         struct netio_data *nd = td->io_ops->data;
 465         int fd, opt, type;
 466
 467         if (nd->net_protocol == IPPROTO_TCP)
 468                 type = SOCK_STREAM;
 469         else
 470                 type = SOCK_DGRAM;
 471
 472         fd = socket(AF_INET, type, nd->net_protocol);
 473         if (fd < 0) {
 474                 td_verror(td, errno, "socket");
 475                 return 1;
 476         }
 477
 478         opt = 1;
 479         if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
 480                 td_verror(td, errno, "setsockopt");
 481                 return 1;
 482         }
 483 #ifdef SO_REUSEPORT
 484         if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
 485                 td_verror(td, errno, "setsockopt");
 486                 return 1;
 487         }
 488 #endif
 489
 490         nd->addr.sin_family = AF_INET;
 491         nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
 492         nd->addr.sin_port = htons(port);
 493
 494         if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
 495                 td_verror(td, errno, "bind");
 496                 return 1;
 497         }
 498         if (nd->net_protocol == IPPROTO_TCP && listen(fd, 1) < 0) {
 499                 td_verror(td, errno, "listen");
 500                 return 1;
 501         }
 502
 503         nd->listenfd = fd;
 504         return 0;
 505 }
 506
 507 static int fio_netio_init(struct thread_data *td)
 508 {
 509         struct netio_data *nd = td->io_ops->data;
 510         unsigned int port;
 511         char host[64], buf[128];
 512         char *sep, *portp, *modep;
 513         int ret;
 514
 515         if (td_rw(td)) {
 516                 log_err("fio: network connections must be read OR write\n");
 517                 return 1;
 518         }
 519         if (td_random(td)) {
 520                 log_err("fio: network IO can't be random\n");
 521                 return 1;
 522         }
 523
 524         strcpy(buf, td->o.filename);
 525
 526         sep = strchr(buf, '/');
 527         if (!sep)
 528                 goto bad_host;
 529
 530         *sep = '\0';
 531         sep++;
 532         strcpy(host, buf);
 533         if (!strlen(host))
 534                 goto bad_host;
 535
 536         modep = NULL;
 537         portp = sep;
 538         sep = strchr(portp, '/');
 539         if (sep) {
 540                 *sep = '\0';
 541                 modep = sep + 1;
 542         }
 543
 544         port = strtol(portp, NULL, 10);
 545         if (!port || port > 65535)
 546                 goto bad_host;
 547
 548         if (modep) {
 549                 if (!strncmp("tcp", modep, strlen(modep)) ||
 550                     !strncmp("TCP", modep, strlen(modep)))
 551                         nd->net_protocol = IPPROTO_TCP;
 552                 else if (!strncmp("udp", modep, strlen(modep)) ||
 553                          !strncmp("UDP", modep, strlen(modep)))
 554                         nd->net_protocol = IPPROTO_UDP;
 555                 else
 556                         goto bad_host;
 557         } else
 558                 nd->net_protocol = IPPROTO_TCP;
 559
 560         if (td_read(td)) {
 561                 nd->send_to_net = 0;
 562                 ret = fio_netio_setup_listen(td, port);
 563         } else {
 564                 nd->send_to_net = 1;
 565                 ret = fio_netio_setup_connect(td, host, port);
 566         }
 567
 568         return ret;
 569 bad_host:
 570         log_err("fio: bad network host/port/protocol: %s\n", td->o.filename);
 571         return 1;
 572 }
 573
 574 static void fio_netio_cleanup(struct thread_data *td)
 575 {
 576         struct netio_data *nd = td->io_ops->data;
 577
 578         if (nd) {
 579                 if (nd->listenfd != -1)
 580                         close(nd->listenfd);
 581                 if (nd->pipes[0] != -1)
 582                         close(nd->pipes[0]);
 583                 if (nd->pipes[1] != -1)
 584                         close(nd->pipes[1]);
 585
 586                 free(nd);
 587         }
 588 }
 589
 590 static int fio_netio_setup(struct thread_data *td)
 591 {
 592         struct netio_data *nd;
 593
 594         if (!td->io_ops->data) {
 595                 nd = malloc(sizeof(*nd));;
 596
 597                 memset(nd, 0, sizeof(*nd));
 598                 nd->listenfd = -1;
 599                 nd->pipes[0] = nd->pipes[1] = -1;
 600                 td->io_ops->data = nd;
 601         }
 602
 603         return 0;
 604 }
 605
 606 #ifdef FIO_HAVE_SPLICE
 607 static int fio_netio_setup_splice(struct thread_data *td)
 608 {
 609         struct netio_data *nd;
 610
 611         fio_netio_setup(td);
 612
 613         nd = td->io_ops->data;
 614         if (nd) {
 615                 if (pipe(nd->pipes) < 0)
 616                         return 1;
 617
 618                 nd->use_splice = 1;
 619                 return 0;
 620         }
 621
 622         return 1;
 623 }
 624
 625 static struct ioengine_ops ioengine_splice = {
 626         .name           = "netsplice",
 627         .version        = FIO_IOOPS_VERSION,
 628         .prep           = fio_netio_prep,
 629         .queue          = fio_netio_queue,
 630         .setup          = fio_netio_setup_splice,
 631         .init           = fio_netio_init,
 632         .cleanup        = fio_netio_cleanup,
 633         .open_file      = fio_netio_open_file,
 634         .close_file     = generic_close_file,
 635         .flags          = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
 636                           FIO_SIGQUIT,
 637 };
 638 #endif
 639
 640 static struct ioengine_ops ioengine_rw = {
 641         .name           = "net",
 642         .version        = FIO_IOOPS_VERSION,
 643         .prep           = fio_netio_prep,
 644         .queue          = fio_netio_queue,
 645         .setup          = fio_netio_setup,
 646         .init           = fio_netio_init,
 647         .cleanup        = fio_netio_cleanup,
 648         .open_file      = fio_netio_open_file,
 649         .close_file     = fio_netio_close_file,
 650         .flags          = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
 651                           FIO_SIGQUIT,
 652 };
 653
 654 static void fio_init fio_netio_register(void)
 655 {
 656         register_ioengine(&ioengine_rw);
 657 #ifdef FIO_HAVE_SPLICE
 658         register_ioengine(&ioengine_splice);
 659 #endif
 660 }
 661
 662 static void fio_exit fio_netio_unregister(void)
 663 {
 664         unregister_ioengine(&ioengine_rw);
 665 #ifdef FIO_HAVE_SPLICE
 666         unregister_ioengine(&ioengine_splice);
 667 #endif
 668 }