X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=engines%2Fnet.c;h=0adc00647ed3fd1dea23875b3fbbdc6e27784707;hp=51bc671fc3a1ba23a0e60ce91f80cc375db0990d;hb=dc372f07b450e3b4c758d7a7fdaaca6c21843398;hpb=da751ca9665bcdeca56d2eec5b629a0953c07662 diff --git a/engines/net.c b/engines/net.c index 51bc671f..0adc0064 100644 --- a/engines/net.c +++ b/engines/net.c @@ -15,11 +15,12 @@ #include #include "../fio.h" -#include "../os.h" struct netio_data { int listenfd; int send_to_net; + int use_splice; + int pipes[2]; char host[64]; struct sockaddr_in addr; }; @@ -51,23 +52,162 @@ static int fio_netio_prep(struct thread_data *td, struct io_u *io_u) return 1; } +static int splice_io_u(int fdin, int fdout, unsigned int len) +{ + int bytes = 0; + + while (len) { + int ret = splice(fdin, NULL, fdout, NULL, len, 0); + + if (ret < 0) { + if (!bytes) + bytes = ret; + + break; + } else if (!ret) + break; + + bytes += ret; + len -= ret; + } + + return bytes; +} + +/* + * Receive bytes from a socket and fill them into the internal pipe + */ +static int splice_in(struct thread_data *td, struct io_u *io_u) +{ + struct netio_data *nd = td->io_ops->data; + + return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen); +} + +/* + * Transmit 'len' bytes from the internal pipe + */ +static int splice_out(struct thread_data *td, struct io_u *io_u, + unsigned int len) +{ + struct netio_data *nd = td->io_ops->data; + + return splice_io_u(nd->pipes[0], io_u->file->fd, len); +} + +static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len) +{ + struct iovec iov = { + .iov_base = io_u->xfer_buf, + .iov_len = len, + }; + int bytes = 0; + + while (iov.iov_len) { + int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE); + + if (ret < 0) { + if (!bytes) + bytes = ret; + break; + } else if (!ret) + break; + + iov.iov_len -= ret; + iov.iov_base += ret; + bytes += ret; + } + + return bytes; + +} + +/* + * vmsplice() pipe to io_u buffer + */ +static int vmsplice_io_u_out(struct thread_data *td, struct io_u *io_u, + unsigned int len) +{ + struct netio_data *nd = td->io_ops->data; + + return vmsplice_io_u(io_u, nd->pipes[0], len); +} + +/* + * vmsplice() io_u to pipe + */ +static int vmsplice_io_u_in(struct thread_data *td, struct io_u *io_u) +{ + struct netio_data *nd = td->io_ops->data; + + return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen); +} + +/* + * splice receive - transfer socket data into a pipe using splice, then map + * that pipe data into the io_u using vmsplice. + */ +static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u) +{ + int ret; + + ret = splice_in(td, io_u); + if (ret > 0) + return vmsplice_io_u_out(td, io_u, ret); + + return ret; +} + +/* + * splice transmit - map data from the io_u into a pipe by using vmsplice, + * then transfer that pipe to a socket using splice. + */ +static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u) +{ + int ret; + + ret = vmsplice_io_u_in(td, io_u); + if (ret > 0) + return splice_out(td, io_u, ret); + + return ret; +} + +static int fio_netio_send(struct thread_data *td, struct io_u *io_u) +{ + int flags = 0; + + /* + * if we are going to write more, set MSG_MORE + */ + if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < td->o.size) + flags = MSG_MORE; + + 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) +{ + int flags = MSG_WAITALL; + + 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) { - struct fio_file *f = io_u->file; - int ret, flags = 0; + struct netio_data *nd = td->io_ops->data; + int ret; if (io_u->ddir == DDIR_WRITE) { - /* - * if we are going to write more, set MSG_MORE - */ - if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < - td->io_size) - flags = MSG_MORE; - - ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags); + if (nd->use_splice) + ret = fio_netio_splice_out(td, io_u); + else + ret = fio_netio_send(td, io_u); } else if (io_u->ddir == DDIR_READ) { - flags = MSG_WAITALL; - ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags); + if (nd->use_splice) + ret = fio_netio_splice_in(td, io_u); + else + ret = fio_netio_recv(io_u); } else ret = 0; /* must be a SYNC */ @@ -111,7 +251,7 @@ static int fio_netio_accept(struct thread_data *td, struct fio_file *f) struct pollfd pfd; int ret; - fprintf(f_out, "fio: waiting for connection\n"); + log_info("fio: waiting for connection\n"); /* * Accept loop. poll for incoming events, accept them. Repeat until we @@ -148,7 +288,6 @@ static int fio_netio_accept(struct thread_data *td, struct fio_file *f) return 0; } - static int fio_netio_open_file(struct thread_data *td, struct fio_file *f) { if (td_read(td)) @@ -224,27 +363,24 @@ static int fio_netio_init(struct thread_data *td) { struct netio_data *nd = td->io_ops->data; unsigned short port; - struct fio_file *f; char host[64], buf[128]; - unsigned int i; char *sep; int ret; - if (!td->total_file_size) { - log_err("fio: need size= set\n"); - return 1; - } - if (td_rw(td)) { log_err("fio: network connections must be read OR write\n"); return 1; } + if (td_random(td)) { + log_err("fio: network IO can't be random\n"); + return 1; + } - strcpy(buf, td->filename); + strcpy(buf, td->o.filename); sep = strchr(buf, '/'); if (!sep) { - log_err("fio: bad network host/port <<%s>>\n", td->filename); + log_err("fio: bad network host/port <<%s>>\n", td->o.filename); return 1; } @@ -261,18 +397,7 @@ static int fio_netio_init(struct thread_data *td) ret = fio_netio_setup_connect(td, host, port); } - if (ret) - return ret; - - td->io_size = td->total_file_size; - td->total_io_size = td->io_size; - - for_each_file(td, f, i) { - f->file_size = td->total_file_size / td->nr_files; - f->real_file_size = f->file_size; - } - - return 0; + return ret; } static void fio_netio_cleanup(struct thread_data *td) @@ -280,6 +405,13 @@ static void fio_netio_cleanup(struct thread_data *td) struct netio_data *nd = td->io_ops->data; if (nd) { + if (nd->listenfd != -1) + close(nd->listenfd); + if (nd->pipes[0] != -1) + close(nd->pipes[0]); + if (nd->pipes[1] != -1) + close(nd->pipes[1]); + free(nd); td->io_ops->data = NULL; } @@ -287,15 +419,39 @@ static void fio_netio_cleanup(struct thread_data *td) static int fio_netio_setup(struct thread_data *td) { - struct netio_data *nd = malloc(sizeof(*nd)); + struct netio_data *nd; + + if (!td->io_ops->data) { + nd = malloc(sizeof(*nd));; + + memset(nd, 0, sizeof(*nd)); + nd->listenfd = -1; + nd->pipes[0] = nd->pipes[1] = -1; + td->io_ops->data = nd; + } - memset(nd, 0, sizeof(*nd)); - nd->listenfd = -1; - td->io_ops->data = nd; return 0; } -static struct ioengine_ops ioengine = { +static int fio_netio_setup_splice(struct thread_data *td) +{ + struct netio_data *nd; + + fio_netio_setup(td); + + nd = td->io_ops->data; + if (nd) { + if (pipe(nd->pipes) < 0) + return 1; + + nd->use_splice = 1; + return 0; + } + + return 1; +} + +static struct ioengine_ops ioengine_rw = { .name = "net", .version = FIO_IOOPS_VERSION, .prep = fio_netio_prep, @@ -305,15 +461,30 @@ static struct ioengine_ops ioengine = { .cleanup = fio_netio_cleanup, .open_file = fio_netio_open_file, .close_file = generic_close_file, - .flags = FIO_SYNCIO | FIO_DISKLESSIO, + .flags = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR, +}; + +static struct ioengine_ops ioengine_splice = { + .name = "netsplice", + .version = FIO_IOOPS_VERSION, + .prep = fio_netio_prep, + .queue = fio_netio_queue, + .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, }; static void fio_init fio_netio_register(void) { - register_ioengine(&ioengine); + register_ioengine(&ioengine_rw); + register_ioengine(&ioengine_splice); } static void fio_exit fio_netio_unregister(void) { - unregister_ioengine(&ioengine); + unregister_ioengine(&ioengine_rw); + unregister_ioengine(&ioengine_splice); }