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->o.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         char host[64], buf[128];
 228         char *sep;
 229         int ret;
 230
 231         if (td_rw(td)) {
 232                 log_err("fio: network connections must be read OR write\n");
 233                 return 1;
 234         }
 235
 236         strcpy(buf, td->o.filename);
 237
 238         sep = strchr(buf, '/');
 239         if (!sep) {
 240                 log_err("fio: bad network host/port <<%s>>\n", td->o.filename);
 241                 return 1;
 242         }
 243
 244         *sep = '\0';
 245         sep++;
 246         strcpy(host, buf);
 247         port = atoi(sep);
 248
 249         if (td_read(td)) {
 250                 nd->send_to_net = 0;
 251                 ret = fio_netio_setup_listen(td, port);
 252         } else {
 253                 nd->send_to_net = 1;
 254                 ret = fio_netio_setup_connect(td, host, port);
 255         }
 256
 257         return ret;
 258 }
 259
 260 static void fio_netio_cleanup(struct thread_data *td)
 261 {
 262         struct netio_data *nd = td->io_ops->data;
 263
 264         if (nd) {
 265                 free(nd);
 266                 td->io_ops->data = NULL;
 267         }
 268 }
 269
 270 static int fio_netio_setup(struct thread_data *td)
 271 {
 272         struct netio_data *nd;
 273         struct fio_file *f;
 274         unsigned int i;
 275
 276         if (!td->io_ops->data) {
 277                 nd = malloc(sizeof(*nd));;
 278
 279                 memset(nd, 0, sizeof(*nd));
 280                 nd->listenfd = -1;
 281                 td->io_ops->data = nd;
 282
 283                 for_each_file(td, f, i)
 284                         f->real_file_size = -1ULL;
 285         }
 286
 287         return 0;
 288 }
 289
 290 static struct ioengine_ops ioengine = {
 291         .name           = "net",
 292         .version        = FIO_IOOPS_VERSION,
 293         .prep           = fio_netio_prep,
 294         .queue          = fio_netio_queue,
 295         .setup          = fio_netio_setup,
 296         .init           = fio_netio_init,
 297         .cleanup        = fio_netio_cleanup,
 298         .open_file      = fio_netio_open_file,
 299         .close_file     = generic_close_file,
 300         .flags          = FIO_SYNCIO | FIO_DISKLESSIO,
 301 };
 302
 303 static void fio_init fio_netio_register(void)
 304 {
 305         register_ioengine(&ioengine);
 306 }
 307
 308 static void fio_exit fio_netio_unregister(void)
 309 {
 310         unregister_ioengine(&ioengine);
 311 }