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