engines/net.c

   1 /*
   2  * Transfer data over the net.
   3  */
   4 #include <stdio.h>
   5 #include <stdlib.h>
   6 #include <unistd.h>
   7 #include <errno.h>
   8 #include <assert.h>
   9 #include <netinet/in.h>
  10 #include <arpa/inet.h>
  11 #include <netdb.h>
  12 #include <sys/poll.h>
  13
  14 #include "../fio.h"
  15 #include "../os.h"
  16
  17 struct netio_data {
  18         int listenfd;
  19         int send_to_net;
  20         char host[64];
  21         struct sockaddr_in addr;
  22 };
  23
  24 static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
  25 {
  26         struct netio_data *nd = td->io_ops->data;
  27         struct fio_file *f = io_u->file;
  28
  29         /*
  30          * Make sure we don't see spurious reads to a receiver, and vice versa
  31          */
  32         if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
  33             (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
  34                 td_verror(td, EINVAL, "bad direction");
  35                 return 1;
  36         }
  37
  38         if (io_u->ddir == DDIR_SYNC)
  39                 return 0;
  40         if (io_u->offset == f->last_completed_pos)
  41                 return 0;
  42
  43         /*
  44          * If offset is different from last end position, it's a seek.
  45          * As network io is purely sequential, we don't allow seeks.
  46          */
  47         td_verror(td, EINVAL, "cannot seek");
  48         return 1;
  49 }
  50
  51 static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
  52 {
  53         struct fio_file *f = io_u->file;
  54         int ret, flags = 0;
  55
  56         if (io_u->ddir == DDIR_WRITE) {
  57                 /*
  58                  * if we are going to write more, set MSG_MORE
  59                  */
  60                 if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
  61                     td->io_size)
  62                         flags = MSG_MORE;
  63
  64                 ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
  65         } else if (io_u->ddir == DDIR_READ) {
  66                 flags = MSG_WAITALL;
  67                 ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
  68         } else
  69                 ret = 0;        /* must be a SYNC */
  70
  71         if (ret != (int) io_u->xfer_buflen) {
  72                 if (ret >= 0) {
  73                         io_u->resid = io_u->xfer_buflen - ret;
  74                         io_u->error = 0;
  75                         return FIO_Q_COMPLETED;
  76                 } else
  77                         io_u->error = errno;
  78         }
  79
  80         if (io_u->error)
  81                 td_verror(td, io_u->error, "xfer");
  82
  83         return FIO_Q_COMPLETED;
  84 }
  85
  86 static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
  87 {
  88         struct netio_data *nd = td->io_ops->data;
  89
  90         f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  91         if (f->fd < 0) {
  92                 td_verror(td, errno, "socket");
  93                 return 1;
  94         }
  95
  96         if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
  97                 td_verror(td, errno, "connect");
  98                 return 1;
  99         }
 100
 101         return 0;
 102 }
 103
 104 static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
 105 {
 106         struct netio_data *nd = td->io_ops->data;
 107         socklen_t socklen = sizeof(nd->addr);
 108         struct pollfd pfd;
 109         int ret;
 110
 111         fprintf(f_out, "fio: waiting for connection\n");
 112
 113         /*
 114          * Accept loop. poll for incoming events, accept them. Repeat until we
 115          * have all connections.
 116          */
 117         while (!td->terminate) {
 118                 pfd.fd = nd->listenfd;
 119                 pfd.events = POLLIN;
 120
 121                 ret = poll(&pfd, 1, -1);
 122                 if (ret < 0) {
 123                         if (errno == EINTR)
 124                                 continue;
 125
 126                         td_verror(td, errno, "poll");
 127                         break;
 128                 } else if (!ret)
 129                         continue;
 130
 131                 /*
 132                  * should be impossible
 133                  */
 134                 if (!(pfd.revents & POLLIN))
 135                         continue;
 136
 137                 f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
 138                 if (f->fd < 0) {
 139                         td_verror(td, errno, "accept");
 140                         return 1;
 141                 }
 142                 break;
 143         }
 144
 145         return 0;
 146 }
 147
 148
 149 static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
 150 {
 151         if (td_read(td))
 152                 return fio_netio_accept(td, f);
 153         else
 154                 return fio_netio_connect(td, f);
 155 }
 156
 157 static int fio_netio_setup_connect(struct thread_data *td, const char *host,
 158                                    unsigned short port)
 159 {
 160         struct netio_data *nd = td->io_ops->data;
 161
 162         nd->addr.sin_family = AF_INET;
 163         nd->addr.sin_port = htons(port);
 164
 165         if (inet_aton(host, &nd->addr.sin_addr) != 1) {
 166                 struct hostent *hent;
 167
 168                 hent = gethostbyname(host);
 169                 if (!hent) {
 170                         td_verror(td, errno, "gethostbyname");
 171                         return 1;
 172                 }
 173
 174                 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
 175         }
 176
 177         return 0;
 178 }
 179
 180 static int fio_netio_setup_listen(struct thread_data *td, short port)
 181 {
 182         struct netio_data *nd = td->io_ops->data;
 183         int fd, opt;
 184
 185         fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
 186         if (fd < 0) {
 187                 td_verror(td, errno, "socket");
 188                 return 1;
 189         }
 190
 191         opt = 1;
 192         if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
 193                 td_verror(td, errno, "setsockopt");
 194                 return 1;
 195         }
 196 #ifdef SO_REUSEPORT
 197         if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
 198                 td_verror(td, errno, "setsockopt");
 199                 return 1;
 200         }
 201 #endif
 202
 203         nd->addr.sin_family = AF_INET;
 204         nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
 205         nd->addr.sin_port = htons(port);
 206
 207         if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
 208                 td_verror(td, errno, "bind");
 209                 return 1;
 210         }
 211         if (listen(fd, 1) < 0) {
 212                 td_verror(td, errno, "listen");
 213                 return 1;
 214         }
 215
 216         nd->listenfd = fd;
 217         return 0;
 218 }
 219
 220 static int fio_netio_init(struct thread_data *td)
 221 {
 222         struct netio_data *nd = td->io_ops->data;
 223         unsigned short port;
 224         struct fio_file *f;
 225         char host[64], buf[128];
 226         char *sep;
 227         int ret, i;
 228
 229         if (!td->total_file_size) {
 230                 log_err("fio: need size= set\n");
 231                 return 1;
 232         }
 233
 234         if (td_rw(td)) {
 235                 log_err("fio: network connections must be read OR write\n");
 236                 return 1;
 237         }
 238
 239         strcpy(buf, td->filename);
 240
 241         sep = strchr(buf, ':');
 242         if (!sep) {
 243                 log_err("fio: bad network host:port <<%s>>\n", td->filename);
 244                 return 1;
 245         }
 246
 247         *sep = '\0';
 248         sep++;
 249         strcpy(host, buf);
 250         port = atoi(sep);
 251
 252         if (td_read(td)) {
 253                 nd->send_to_net = 0;
 254                 ret = fio_netio_setup_listen(td, port);
 255         } else {
 256                 nd->send_to_net = 1;
 257                 ret = fio_netio_setup_connect(td, host, port);
 258         }
 259
 260         if (ret)
 261                 return ret;
 262
 263         td->io_size = td->total_file_size;
 264         td->total_io_size = td->io_size;
 265
 266         for_each_file(td, f, i) {
 267                 f->file_size = td->total_file_size / td->nr_files;
 268                 f->real_file_size = f->file_size;
 269         }
 270
 271         td->nr_open_files = td->nr_files;
 272         return 0;
 273 }
 274
 275 static void fio_netio_cleanup(struct thread_data *td)
 276 {
 277         struct netio_data *nd = td->io_ops->data;
 278
 279         if (nd) {
 280                 free(nd);
 281                 td->io_ops->data = NULL;
 282         }
 283 }
 284
 285 static int fio_netio_setup(struct thread_data *td)
 286 {
 287         struct netio_data *nd = malloc(sizeof(*nd));
 288
 289         memset(nd, 0, sizeof(*nd));
 290         nd->listenfd = -1;
 291         td->io_ops->data = nd;
 292         return 0;
 293 }
 294
 295 static struct ioengine_ops ioengine = {
 296         .name           = "net",
 297         .version        = FIO_IOOPS_VERSION,
 298         .prep           = fio_netio_prep,
 299         .queue          = fio_netio_queue,
 300         .setup          = fio_netio_setup,
 301         .init           = fio_netio_init,
 302         .cleanup        = fio_netio_cleanup,
 303         .open_file      = fio_netio_open_file,
 304         .close_file     = generic_close_file,
 305         .flags          = FIO_SYNCIO | FIO_DISKLESSIO,
 306 };
 307
 308 static void fio_init fio_netio_register(void)
 309 {
 310         register_ioengine(&ioengine);
 311 }
 312
 313 static void fio_exit fio_netio_unregister(void)
 314 {
 315         unregister_ioengine(&ioengine);
 316 }