X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=engines%2Fnet.c;h=afb3265c6ada813179fa12853ec121d6632ab378;hp=552ad0bdb551abebb68f26571df5dc5e2ad20b8f;hb=35651820abe54473964cc1832fdfe23d7af3209a;hpb=7101d9c24abec4be58a086d85d6d92ec6e6492e9 diff --git a/engines/net.c b/engines/net.c index 552ad0bd..afb3265c 100644 --- a/engines/net.c +++ b/engines/net.c @@ -13,6 +13,8 @@ #include #include #include +#include +#include #include "../fio.h" @@ -20,15 +22,48 @@ struct netio_data { int listenfd; int send_to_net; int use_splice; + int net_protocol; int pipes[2]; char host[64]; struct sockaddr_in addr; }; +/* + * Return -1 for error and 'nr events' for a positive number + * of events + */ +static int poll_wait(struct thread_data *td, int fd, short events) +{ + struct pollfd pfd; + int ret; + + while (!td->terminate) { + pfd.fd = fd; + pfd.events = events; + ret = poll(&pfd, 1, -1); + if (ret < 0) { + if (errno == EINTR) + continue; + + td_verror(td, errno, "poll"); + return -1; + } else if (!ret) + continue; + + break; + } + + if (pfd.revents & events) + return 1; + else if (td->terminate) + return 1; + + return -1; +} + static int fio_netio_prep(struct thread_data *td, struct io_u *io_u) { struct netio_data *nd = td->io_ops->data; - struct fio_file *f = io_u->file; /* * Make sure we don't see spurious reads to a receiver, and vice versa @@ -39,19 +74,10 @@ static int fio_netio_prep(struct thread_data *td, struct io_u *io_u) return 1; } - if (io_u->ddir == DDIR_SYNC) - return 0; - if (io_u->offset == f->last_completed_pos) - return 0; - - /* - * If offset is different from last end position, it's a seek. - * As network io is purely sequential, we don't allow seeks. - */ - td_verror(td, EINVAL, "cannot seek"); - return 1; + return 0; } +#ifdef FIO_HAVE_SPLICE static int splice_io_u(int fdin, int fdout, unsigned int len) { int bytes = 0; @@ -172,25 +198,64 @@ static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u) return ret; } +#else +static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u) +{ + errno = EOPNOTSUPP; + return -1; +} + +static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u) +{ + errno = EOPNOTSUPP; + return -1; +} +#endif static int fio_netio_send(struct thread_data *td, struct io_u *io_u) { - int flags = 0; + struct netio_data *nd = td->io_ops->data; + int ret, flags = 0; + + ret = poll_wait(td, io_u->file->fd, POLLOUT); + if (ret <= 0) + return ret; /* * if we are going to write more, set MSG_MORE */ +#ifdef MSG_MORE if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < td->o.size) flags = MSG_MORE; +#endif - return send(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags); + if (nd->net_protocol == IPPROTO_UDP) { + return sendto(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, + 0, &nd->addr, sizeof(nd->addr)); + } else { + return send(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, + flags); + } } -static int fio_netio_recv(struct io_u *io_u) +static int fio_netio_recv(struct thread_data *td, struct io_u *io_u) { - int flags = MSG_WAITALL; + struct netio_data *nd = td->io_ops->data; + int ret, flags = MSG_WAITALL; + + ret = poll_wait(td, io_u->file->fd, POLLIN); + if (ret <= 0) + return ret; - return recv(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags); + if (nd->net_protocol == IPPROTO_UDP) { + socklen_t len = sizeof(nd->addr); + + return recvfrom(io_u->file->fd, io_u->xfer_buf, + io_u->xfer_buflen, 0, &nd->addr, &len); + } else { + return recv(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, + flags); + } } static int fio_netio_queue(struct thread_data *td, struct io_u *io_u) @@ -201,15 +266,15 @@ static int fio_netio_queue(struct thread_data *td, struct io_u *io_u) fio_ro_check(td, io_u); if (io_u->ddir == DDIR_WRITE) { - if (nd->use_splice) - ret = fio_netio_splice_out(td, io_u); - else + if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP) ret = fio_netio_send(td, io_u); + else + ret = fio_netio_splice_out(td, io_u); } else if (io_u->ddir == DDIR_READ) { - if (nd->use_splice) - ret = fio_netio_splice_in(td, io_u); + if (!nd->use_splice || nd->net_protocol == IPPROTO_UDP) + ret = fio_netio_recv(td, io_u); else - ret = fio_netio_recv(io_u); + ret = fio_netio_splice_in(td, io_u); } else ret = 0; /* must be a SYNC */ @@ -218,8 +283,14 @@ static int fio_netio_queue(struct thread_data *td, struct io_u *io_u) io_u->resid = io_u->xfer_buflen - ret; io_u->error = 0; return FIO_Q_COMPLETED; - } else - io_u->error = errno; + } else { + int err = errno; + + if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE) + return FIO_Q_BUSY; + + io_u->error = err; + } } if (io_u->error) @@ -231,13 +302,22 @@ static int fio_netio_queue(struct thread_data *td, struct io_u *io_u) static int fio_netio_connect(struct thread_data *td, struct fio_file *f) { struct netio_data *nd = td->io_ops->data; + int type; + + if (nd->net_protocol == IPPROTO_TCP) + type = SOCK_STREAM; + else + type = SOCK_DGRAM; - f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + f->fd = socket(AF_INET, type, nd->net_protocol); if (f->fd < 0) { td_verror(td, errno, "socket"); return 1; } + if (nd->net_protocol == IPPROTO_UDP) + return 0; + if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) { td_verror(td, errno, "connect"); return 1; @@ -250,41 +330,21 @@ static int fio_netio_accept(struct thread_data *td, struct fio_file *f) { struct netio_data *nd = td->io_ops->data; socklen_t socklen = sizeof(nd->addr); - struct pollfd pfd; - int ret; - - log_info("fio: waiting for connection\n"); - - /* - * Accept loop. poll for incoming events, accept them. Repeat until we - * have all connections. - */ - while (!td->terminate) { - pfd.fd = nd->listenfd; - pfd.events = POLLIN; - ret = poll(&pfd, 1, -1); - if (ret < 0) { - if (errno == EINTR) - continue; + if (nd->net_protocol == IPPROTO_UDP) { + f->fd = nd->listenfd; + return 0; + } - td_verror(td, errno, "poll"); - break; - } else if (!ret) - continue; + log_info("fio: waiting for connection\n"); - /* - * should be impossible - */ - if (!(pfd.revents & POLLIN)) - continue; + if (poll_wait(td, nd->listenfd, POLLIN) < 0) + return 1; - f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen); - if (f->fd < 0) { - td_verror(td, errno, "accept"); - return 1; - } - break; + f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen); + if (f->fd < 0) { + td_verror(td, errno, "accept"); + return 1; } return 0; @@ -324,9 +384,14 @@ static int fio_netio_setup_connect(struct thread_data *td, const char *host, static int fio_netio_setup_listen(struct thread_data *td, short port) { struct netio_data *nd = td->io_ops->data; - int fd, opt; + int fd, opt, type; - fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (nd->net_protocol == IPPROTO_TCP) + type = SOCK_STREAM; + else + type = SOCK_DGRAM; + + fd = socket(AF_INET, type, nd->net_protocol); if (fd < 0) { td_verror(td, errno, "socket"); return 1; @@ -352,7 +417,7 @@ static int fio_netio_setup_listen(struct thread_data *td, short port) td_verror(td, errno, "bind"); return 1; } - if (listen(fd, 1) < 0) { + if (nd->net_protocol == IPPROTO_TCP && listen(fd, 1) < 0) { td_verror(td, errno, "listen"); return 1; } @@ -364,9 +429,9 @@ static int fio_netio_setup_listen(struct thread_data *td, short port) static int fio_netio_init(struct thread_data *td) { struct netio_data *nd = td->io_ops->data; - unsigned short port; + unsigned int port; char host[64], buf[128]; - char *sep; + char *sep, *portp, *modep; int ret; if (td_rw(td)) { @@ -381,15 +446,38 @@ static int fio_netio_init(struct thread_data *td) strcpy(buf, td->o.filename); sep = strchr(buf, '/'); - if (!sep) { - log_err("fio: bad network host/port <<%s>>\n", td->o.filename); - return 1; - } + if (!sep) + goto bad_host; *sep = '\0'; sep++; strcpy(host, buf); - port = atoi(sep); + if (!strlen(host)) + goto bad_host; + + modep = NULL; + portp = sep; + sep = strchr(portp, '/'); + if (sep) { + *sep = '\0'; + modep = sep + 1; + } + + port = strtol(portp, NULL, 10); + if (!port || port > 65535) + goto bad_host; + + if (modep) { + if (!strncmp("tcp", modep, strlen(modep)) || + !strncmp("TCP", modep, strlen(modep))) + nd->net_protocol = IPPROTO_TCP; + else if (!strncmp("udp", modep, strlen(modep)) || + !strncmp("UDP", modep, strlen(modep))) + nd->net_protocol = IPPROTO_UDP; + else + goto bad_host; + } else + nd->net_protocol = IPPROTO_TCP; if (td_read(td)) { nd->send_to_net = 0; @@ -400,6 +488,9 @@ static int fio_netio_init(struct thread_data *td) } return ret; +bad_host: + log_err("fio: bad network host/port/protocol: %s\n", td->o.filename); + return 1; } static void fio_netio_cleanup(struct thread_data *td) @@ -415,7 +506,6 @@ static void fio_netio_cleanup(struct thread_data *td) close(nd->pipes[1]); free(nd); - td->io_ops->data = NULL; } } @@ -435,6 +525,7 @@ static int fio_netio_setup(struct thread_data *td) return 0; } +#ifdef FIO_HAVE_SPLICE static int fio_netio_setup_splice(struct thread_data *td) { struct netio_data *nd; @@ -453,40 +544,47 @@ static int fio_netio_setup_splice(struct thread_data *td) return 1; } -static struct ioengine_ops ioengine_rw = { - .name = "net", +static struct ioengine_ops ioengine_splice = { + .name = "netsplice", .version = FIO_IOOPS_VERSION, .prep = fio_netio_prep, .queue = fio_netio_queue, - .setup = fio_netio_setup, + .setup = fio_netio_setup_splice, .init = fio_netio_init, .cleanup = fio_netio_cleanup, .open_file = fio_netio_open_file, .close_file = generic_close_file, - .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR, + .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR | + FIO_SIGQUIT, }; +#endif -static struct ioengine_ops ioengine_splice = { - .name = "netsplice", +static struct ioengine_ops ioengine_rw = { + .name = "net", .version = FIO_IOOPS_VERSION, .prep = fio_netio_prep, .queue = fio_netio_queue, - .setup = fio_netio_setup_splice, + .setup = fio_netio_setup, .init = fio_netio_init, .cleanup = fio_netio_cleanup, .open_file = fio_netio_open_file, .close_file = generic_close_file, - .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR, + .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR | + FIO_SIGQUIT, }; static void fio_init fio_netio_register(void) { register_ioengine(&ioengine_rw); +#ifdef FIO_HAVE_SPLICE register_ioengine(&ioengine_splice); +#endif } static void fio_exit fio_netio_unregister(void) { unregister_ioengine(&ioengine_rw); +#ifdef FIO_HAVE_SPLICE unregister_ioengine(&ioengine_splice); +#endif }