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