pre_read fixes
[fio.git] / 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 #include <sys/types.h>
17 #include <sys/socket.h>
18
19 #include "../fio.h"
20
21 struct netio_data {
22         int listenfd;
23         int send_to_net;
24         int use_splice;
25         int net_protocol;
26         int pipes[2];
27         char host[64];
28         struct sockaddr_in addr;
29 };
30
31 struct udp_close_msg {
32         uint32_t magic;
33         uint32_t cmd;
34 };
35
36 enum {
37         FIO_LINK_CLOSE = 0x89,
38         FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
39 };
40
41 /*
42  * Return -1 for error and 'nr events' for a positive number
43  * of events
44  */
45 static int poll_wait(struct thread_data *td, int fd, short events)
46 {
47         struct pollfd pfd;
48         int ret;
49
50         while (!td->terminate) {
51                 pfd.fd = fd;
52                 pfd.events = events;
53                 ret = poll(&pfd, 1, -1);
54                 if (ret < 0) {
55                         if (errno == EINTR)
56                                 break;
57
58                         td_verror(td, errno, "poll");
59                         return -1;
60                 } else if (!ret)
61                         continue;
62
63                 break;
64         }
65
66         if (pfd.revents & events)
67                 return 1;
68
69         return -1;
70 }
71
72 static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
73 {
74         struct netio_data *nd = td->io_ops->data;
75
76         /*
77          * Make sure we don't see spurious reads to a receiver, and vice versa
78          */
79         if ((nd->send_to_net && io_u->ddir == DDIR_READ) ||
80             (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
81                 td_verror(td, EINVAL, "bad direction");
82                 return 1;
83         }
84                 
85         return 0;
86 }
87
88 #ifdef FIO_HAVE_SPLICE
89 static int splice_io_u(int fdin, int fdout, unsigned int len)
90 {
91         int bytes = 0;
92
93         while (len) {
94                 int ret = splice(fdin, NULL, fdout, NULL, len, 0);
95
96                 if (ret < 0) {
97                         if (!bytes)
98                                 bytes = ret;
99
100                         break;
101                 } else if (!ret)
102                         break;
103
104                 bytes += ret;
105                 len -= ret;
106         }
107
108         return bytes;
109 }
110
111 /*
112  * Receive bytes from a socket and fill them into the internal pipe
113  */
114 static int splice_in(struct thread_data *td, struct io_u *io_u)
115 {
116         struct netio_data *nd = td->io_ops->data;
117
118         return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
119 }
120
121 /*
122  * Transmit 'len' bytes from the internal pipe
123  */
124 static int splice_out(struct thread_data *td, struct io_u *io_u,
125                       unsigned int len)
126 {
127         struct netio_data *nd = td->io_ops->data;
128
129         return splice_io_u(nd->pipes[0], io_u->file->fd, len);
130 }
131
132 static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
133 {
134         struct iovec iov = {
135                 .iov_base = io_u->xfer_buf,
136                 .iov_len = len,
137         };
138         int bytes = 0;
139
140         while (iov.iov_len) {
141                 int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
142
143                 if (ret < 0) {
144                         if (!bytes)
145                                 bytes = ret;
146                         break;
147                 } else if (!ret)
148                         break;
149
150                 iov.iov_len -= ret;
151                 iov.iov_base += ret;
152                 bytes += ret;
153         }
154
155         return bytes;
156
157 }
158
159 /*
160  * vmsplice() pipe to io_u buffer
161  */
162 static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u,
163                              unsigned int len)
164 {
165         struct netio_data *nd = td->io_ops->data;
166
167         return vmsplice_io_u(io_u, nd->pipes[0], len);
168 }
169
170 /*
171  * vmsplice() io_u to pipe
172  */
173 static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u)
174 {
175         struct netio_data *nd = td->io_ops->data;
176
177         return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
178 }
179
180 /*
181  * splice receive - transfer socket data into a pipe using splice, then map
182  * that pipe data into the io_u using vmsplice.
183  */
184 static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
185 {
186         int ret;
187
188         ret = splice_in(td, io_u);
189         if (ret > 0)
190                 return vmsplice_io_u_out(td, io_u, ret);
191
192         return ret;
193 }
194
195 /*
196  * splice transmit - map data from the io_u into a pipe by using vmsplice,
197  * then transfer that pipe to a socket using splice.
198  */
199 static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
200 {
201         int ret;
202
203         ret = vmsplice_io_u_in(td, io_u);
204         if (ret > 0)
205                 return splice_out(td, io_u, ret);
206
207         return ret;
208 }
209 #else
210 static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u)
211 {
212         errno = EOPNOTSUPP;
213         return -1;
214 }
215
216 static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
217 {
218         errno = EOPNOTSUPP;
219         return -1;
220 }
221 #endif
222
223 static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
224 {
225         struct netio_data *nd = td->io_ops->data;
226         int ret, flags = MSG_DONTWAIT;
227
228         do {
229                 if (nd->net_protocol == IPPROTO_UDP) {
230                         struct sockaddr *to = (struct sockaddr *) &nd->addr;
231
232                         ret = sendto(io_u->file->fd, io_u->xfer_buf,
233                                         io_u->xfer_buflen, flags, to,
234                                         sizeof(*to));
235                 } else {
236                         /*
237                          * if we are going to write more, set MSG_MORE
238                          */
239 #ifdef MSG_MORE
240                         if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
241                             td->o.size)
242                                 flags |= MSG_MORE;
243 #endif
244                         ret = send(io_u->file->fd, io_u->xfer_buf,
245                                         io_u->xfer_buflen, flags);
246                 }
247                 if (ret > 0)
248                         break;
249
250                 ret = poll_wait(td, io_u->file->fd, POLLOUT);
251                 if (ret <= 0)
252                         break;
253
254                 flags &= ~MSG_DONTWAIT;
255         } while (1);
256
257         return ret;
258 }
259
260 static int is_udp_close(struct io_u *io_u, int len)
261 {
262         struct udp_close_msg *msg;
263
264         if (len != sizeof(struct udp_close_msg))
265                 return 0;
266
267         msg = io_u->xfer_buf;
268         if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
269                 return 0;
270         if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
271                 return 0;
272
273         return 1;
274 }
275
276 static int fio_netio_recv(struct thread_data *td, struct io_u *io_u)
277 {
278         struct netio_data *nd = td->io_ops->data;
279         int ret, flags = MSG_DONTWAIT;
280
281         do {
282                 if (nd->net_protocol == IPPROTO_UDP) {
283                         socklen_t len = sizeof(nd->addr);
284                         struct sockaddr *from = (struct sockaddr *) &nd->addr;
285
286                         ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
287                                         io_u->xfer_buflen, flags, from, &len);
288                         if (is_udp_close(io_u, ret)) {
289                                 td->done = 1;
290                                 return 0;
291                         }
292                 } else {
293                         ret = recv(io_u->file->fd, io_u->xfer_buf,
294                                         io_u->xfer_buflen, flags);
295                 }
296                 if (ret > 0)
297                         break;
298
299                 ret = poll_wait(td, io_u->file->fd, POLLIN);
300                 if (ret <= 0)
301                         break;
302                 flags &= ~MSG_DONTWAIT;
303                 flags |= MSG_WAITALL;
304         } while (1);
305
306         return ret;
307 }
308
309 static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
310 {
311         struct netio_data *nd = td->io_ops->data;
312         int ret;
313
314         fio_ro_check(td, io_u);
315
316         if (io_u->ddir == DDIR_WRITE) {
317                 if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP)
318                         ret = fio_netio_send(td, io_u);
319                 else
320                         ret = fio_netio_splice_out(td, io_u);
321         } else if (io_u->ddir == DDIR_READ) {
322                 if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP)
323                         ret = fio_netio_recv(td, io_u);
324                 else
325                         ret = fio_netio_splice_in(td, io_u);
326         } else
327                 ret = 0;        /* must be a SYNC */
328
329         if (ret != (int) io_u->xfer_buflen) {
330                 if (ret >= 0) {
331                         io_u->resid = io_u->xfer_buflen - ret;
332                         io_u->error = 0;
333                         return FIO_Q_COMPLETED;
334                 } else {
335                         int err = errno;
336
337                         if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
338                                 return FIO_Q_BUSY;
339
340                         io_u->error = err;
341                 }
342         }
343
344         if (io_u->error)
345                 td_verror(td, io_u->error, "xfer");
346
347         return FIO_Q_COMPLETED;
348 }
349
350 static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
351 {
352         struct netio_data *nd = td->io_ops->data;
353         int type;
354
355         if (nd->net_protocol == IPPROTO_TCP)
356                 type = SOCK_STREAM;
357         else
358                 type = SOCK_DGRAM;
359
360         f->fd = socket(AF_INET, type, nd->net_protocol);
361         if (f->fd < 0) {
362                 td_verror(td, errno, "socket");
363                 return 1;
364         }
365
366         if (nd->net_protocol == IPPROTO_UDP)
367                 return 0;
368
369         if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
370                 td_verror(td, errno, "connect");
371                 return 1;
372         }
373
374         return 0;
375 }
376
377 static int fio_netio_accept(struct thread_data *td, struct fio_file *f)
378 {
379         struct netio_data *nd = td->io_ops->data;
380         socklen_t socklen = sizeof(nd->addr);
381
382         if (nd->net_protocol == IPPROTO_UDP) {
383                 f->fd = nd->listenfd;
384                 return 0;
385         }
386
387         log_info("fio: waiting for connection\n");
388
389         if (poll_wait(td, nd->listenfd, POLLIN) < 0)
390                 return 1;
391
392         f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
393         if (f->fd < 0) {
394                 td_verror(td, errno, "accept");
395                 return 1;
396         }
397
398         return 0;
399 }
400
401 static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
402 {
403         if (td_read(td))
404                 return fio_netio_accept(td, f);
405         else
406                 return fio_netio_connect(td, f);
407 }
408
409 static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
410 {
411         struct netio_data *nd = td->io_ops->data;
412         struct udp_close_msg msg;
413         struct sockaddr *to = (struct sockaddr *) &nd->addr;
414         int ret;
415
416         msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
417         msg.cmd = htonl(FIO_LINK_CLOSE);
418
419         ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
420                         sizeof(nd->addr));
421         if (ret < 0)
422                 td_verror(td, errno, "sendto udp link close");
423 }
424
425 static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
426 {
427         struct netio_data *nd = td->io_ops->data;
428
429         /*
430          * If this is an UDP connection, notify the receiver that we are
431          * closing down the link
432          */
433         if (nd->net_protocol == IPPROTO_UDP)
434                 fio_netio_udp_close(td, f);
435
436         return generic_close_file(td, f);
437 }
438
439 static int fio_netio_setup_connect(struct thread_data *td, const char *host,
440                                    unsigned short port)
441 {
442         struct netio_data *nd = td->io_ops->data;
443
444         nd->addr.sin_family = AF_INET;
445         nd->addr.sin_port = htons(port);
446
447         if (inet_aton(host, &nd->addr.sin_addr) != 1) {
448                 struct hostent *hent;
449
450                 hent = gethostbyname(host);
451                 if (!hent) {
452                         td_verror(td, errno, "gethostbyname");
453                         return 1;
454                 }
455
456                 memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
457         }
458
459         return 0;
460 }
461
462 static int fio_netio_setup_listen(struct thread_data *td, short port)
463 {
464         struct netio_data *nd = td->io_ops->data;
465         int fd, opt, type;
466
467         if (nd->net_protocol == IPPROTO_TCP)
468                 type = SOCK_STREAM;
469         else
470                 type = SOCK_DGRAM;
471
472         fd = socket(AF_INET, type, nd->net_protocol);
473         if (fd < 0) {
474                 td_verror(td, errno, "socket");
475                 return 1;
476         }
477
478         opt = 1;
479         if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
480                 td_verror(td, errno, "setsockopt");
481                 return 1;
482         }
483 #ifdef SO_REUSEPORT
484         if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
485                 td_verror(td, errno, "setsockopt");
486                 return 1;
487         }
488 #endif
489
490         nd->addr.sin_family = AF_INET;
491         nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
492         nd->addr.sin_port = htons(port);
493
494         if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
495                 td_verror(td, errno, "bind");
496                 return 1;
497         }
498         if (nd->net_protocol == IPPROTO_TCP && listen(fd, 1) < 0) {
499                 td_verror(td, errno, "listen");
500                 return 1;
501         }
502
503         nd->listenfd = fd;
504         return 0;
505 }
506
507 static int fio_netio_init(struct thread_data *td)
508 {
509         struct netio_data *nd = td->io_ops->data;
510         unsigned int port;
511         char host[64], buf[128];
512         char *sep, *portp, *modep;
513         int ret;
514
515         if (td_rw(td)) {
516                 log_err("fio: network connections must be read OR write\n");
517                 return 1;
518         }
519         if (td_random(td)) {
520                 log_err("fio: network IO can't be random\n");
521                 return 1;
522         }
523
524         strcpy(buf, td->o.filename);
525
526         sep = strchr(buf, '/');
527         if (!sep)
528                 goto bad_host;
529
530         *sep = '\0';
531         sep++;
532         strcpy(host, buf);
533         if (!strlen(host))
534                 goto bad_host;
535
536         modep = NULL;
537         portp = sep;
538         sep = strchr(portp, '/');
539         if (sep) {
540                 *sep = '\0';
541                 modep = sep + 1;
542         }
543                 
544         port = strtol(portp, NULL, 10);
545         if (!port || port > 65535)
546                 goto bad_host;
547
548         if (modep) {
549                 if (!strncmp("tcp", modep, strlen(modep)) ||
550                     !strncmp("TCP", modep, strlen(modep)))
551                         nd->net_protocol = IPPROTO_TCP;
552                 else if (!strncmp("udp", modep, strlen(modep)) ||
553                          !strncmp("UDP", modep, strlen(modep)))
554                         nd->net_protocol = IPPROTO_UDP;
555                 else
556                         goto bad_host;
557         } else
558                 nd->net_protocol = IPPROTO_TCP;
559
560         if (td_read(td)) {
561                 nd->send_to_net = 0;
562                 ret = fio_netio_setup_listen(td, port);
563         } else {
564                 nd->send_to_net = 1;
565                 ret = fio_netio_setup_connect(td, host, port);
566         }
567
568         return ret;
569 bad_host:
570         log_err("fio: bad network host/port/protocol: %s\n", td->o.filename);
571         return 1;
572 }
573
574 static void fio_netio_cleanup(struct thread_data *td)
575 {
576         struct netio_data *nd = td->io_ops->data;
577
578         if (nd) {
579                 if (nd->listenfd != -1)
580                         close(nd->listenfd);
581                 if (nd->pipes[0] != -1)
582                         close(nd->pipes[0]);
583                 if (nd->pipes[1] != -1)
584                         close(nd->pipes[1]);
585
586                 free(nd);
587         }
588 }
589
590 static int fio_netio_setup(struct thread_data *td)
591 {
592         struct netio_data *nd;
593
594         if (!td->io_ops->data) {
595                 nd = malloc(sizeof(*nd));;
596
597                 memset(nd, 0, sizeof(*nd));
598                 nd->listenfd = -1;
599                 nd->pipes[0] = nd->pipes[1] = -1;
600                 td->io_ops->data = nd;
601         }
602
603         return 0;
604 }
605
606 #ifdef FIO_HAVE_SPLICE
607 static int fio_netio_setup_splice(struct thread_data *td)
608 {
609         struct netio_data *nd;
610
611         fio_netio_setup(td);
612
613         nd = td->io_ops->data;
614         if (nd) {
615                 if (pipe(nd->pipes) < 0)
616                         return 1;
617
618                 nd->use_splice = 1;
619                 return 0;
620         }
621
622         return 1;
623 }
624
625 static struct ioengine_ops ioengine_splice = {
626         .name           = "netsplice",
627         .version        = FIO_IOOPS_VERSION,
628         .prep           = fio_netio_prep,
629         .queue          = fio_netio_queue,
630         .setup          = fio_netio_setup_splice,
631         .init           = fio_netio_init,
632         .cleanup        = fio_netio_cleanup,
633         .open_file      = fio_netio_open_file,
634         .close_file     = generic_close_file,
635         .flags          = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
636                           FIO_SIGQUIT,
637 };
638 #endif
639
640 static struct ioengine_ops ioengine_rw = {
641         .name           = "net",
642         .version        = FIO_IOOPS_VERSION,
643         .prep           = fio_netio_prep,
644         .queue          = fio_netio_queue,
645         .setup          = fio_netio_setup,
646         .init           = fio_netio_init,
647         .cleanup        = fio_netio_cleanup,
648         .open_file      = fio_netio_open_file,
649         .close_file     = fio_netio_close_file,
650         .flags          = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
651                           FIO_SIGQUIT,
652 };
653
654 static void fio_init fio_netio_register(void)
655 {
656         register_ioengine(&ioengine_rw);
657 #ifdef FIO_HAVE_SPLICE
658         register_ioengine(&ioengine_splice);
659 #endif
660 }
661
662 static void fio_exit fio_netio_unregister(void)
663 {
664         unregister_ioengine(&ioengine_rw);
665 #ifdef FIO_HAVE_SPLICE
666         unregister_ioengine(&ioengine_splice);
667 #endif
668 }