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                 /*
 183                  * should be impossible
 184                  */
 185                 if (!(pfd.revents & POLLIN))
 186                         continue;
 187
 188                 for_each_file(td, f, i) {
 189                         if (f->fd != -1)
 190                                 continue;
 191
 192                         f->fd = accept(fd, (struct sockaddr *) addr, &socklen);
 193                         if (f->fd < 0) {
 194                                 td_verror(td, errno);
 195                                 return 1;
 196                         }
 197                         accepts++;
 198                         break;
 199                 }
 200         }
 201
 202         return 0;
 203 }
 204
 205 static int fio_netio_setup_listen(struct thread_data *td, unsigned short port)
 206 {
 207         struct sockaddr_in addr;
 208         int fd, opt;
 209
 210         fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
 211         if (fd < 0) {
 212                 td_verror(td, errno);
 213                 return 1;
 214         }
 215
 216         opt = 1;
 217         if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
 218                 td_verror(td, errno);
 219                 return 1;
 220         }
 221 #ifdef SO_REUSEPORT
 222         if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
 223                 td_verror(td, errno);
 224                 return 1;
 225         }
 226 #endif
 227
 228         memset(&addr, 0, sizeof(addr));
 229         addr.sin_family = AF_INET;
 230         addr.sin_addr.s_addr = htonl(INADDR_ANY);
 231         addr.sin_port = htons(port);
 232
 233         if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
 234                 td_verror(td, errno);
 235                 return 1;
 236         }
 237         if (listen(fd, 1) < 0) {
 238                 td_verror(td, errno);
 239                 return 1;
 240         }
 241
 242         return fio_netio_accept_connections(td, fd, &addr);
 243 }
 244
 245 static int fio_netio_setup(struct thread_data *td)
 246 {
 247         char host[64], buf[128];
 248         struct net_data *nd;
 249         unsigned short port;
 250         struct fio_file *f;
 251         char *sep;
 252         int ret, i;
 253
 254         if (!td->total_file_size) {
 255                 log_err("fio: need size= set\n");
 256                 return 1;
 257         }
 258
 259         /*
 260          * work around for late init call
 261          */
 262         if (td->io_ops->init(td))
 263                 return 1;
 264
 265         nd = td->io_ops->data;
 266
 267         if (td->iomix) {
 268                 log_err("fio: network connections must be read OR write\n");
 269                 return 1;
 270         }
 271
 272         strcpy(buf, td->filename);
 273
 274         sep = strchr(buf, ':');
 275         if (!sep) {
 276                 log_err("fio: bad network host:port <<%s>>\n", td->filename);
 277                 return 1;
 278         }
 279
 280         *sep = '\0';
 281         sep++;
 282         strcpy(host, buf);
 283         port = atoi(sep);
 284
 285         if (td->ddir == DDIR_READ) {
 286                 nd->send_to_net = 0;
 287                 ret = fio_netio_setup_listen(td, port);
 288         } else {
 289                 nd->send_to_net = 1;
 290                 ret = fio_netio_setup_connect(td, host, port);
 291         }
 292
 293         if (ret)
 294                 return ret;
 295
 296         td->io_size = td->total_file_size;
 297         td->total_io_size = td->io_size;
 298
 299         for_each_file(td, f, i) {
 300                 f->file_size = td->total_file_size / td->nr_files;
 301                 f->real_file_size = f->file_size;
 302         }
 303
 304         return 0;
 305 }
 306
 307 static void fio_netio_cleanup(struct thread_data *td)
 308 {
 309         if (td->io_ops->data) {
 310                 free(td->io_ops->data);
 311                 td->io_ops->data = NULL;
 312         }
 313 }
 314
 315 static int fio_netio_init(struct thread_data *td)
 316 {
 317         struct net_data *nd;
 318
 319         /*
 320          * Hack to work-around the ->setup() function calling init on its
 321          * own, since it needs ->io_ops->data to be set up.
 322          */
 323         if (td->io_ops->data)
 324                 return 0;
 325
 326         nd  = malloc(sizeof(*nd));
 327         nd->last_io_u = NULL;
 328         td->io_ops->data = nd;
 329         return 0;
 330 }
 331
 332 static struct ioengine_ops ioengine = {
 333         .name           = "net",
 334         .version        = FIO_IOOPS_VERSION,
 335         .init           = fio_netio_init,
 336         .prep           = fio_netio_prep,
 337         .queue          = fio_netio_queue,
 338         .getevents      = fio_netio_getevents,
 339         .event          = fio_netio_event,
 340         .cleanup        = fio_netio_cleanup,
 341         .setup          = fio_netio_setup,
 342         .flags          = FIO_SYNCIO | FIO_NETIO,
 343 };
 344
 345 static void fio_init fio_netio_register(void)
 346 {
 347         register_ioengine(&ioengine);
 348 }
 349
 350 static void fio_exit fio_netio_unregister(void)
 351 {
 352         unregister_ioengine(&ioengine);
 353 }