X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=engines%2Fnet.c;h=02314515d4f70866c886bc792a54fdd54fc3b522;hp=43026e5f747469c50093587d234732f874e801fe;hb=a2e1b08afa641078832d69871ae5e404db3e4fc1;hpb=cec6b55da1c282b5b91ad346c7804171fccf151e diff --git a/engines/net.c b/engines/net.c index 43026e5f..02314515 100644 --- a/engines/net.c +++ b/engines/net.c @@ -1,6 +1,5 @@ /* - * Transfer data over the net. Pretty basic setup, will only support - * 1 file per thread/job. + * Transfer data over the net. */ #include #include @@ -10,6 +9,7 @@ #include #include #include +#include #include "../fio.h" #include "../os.h" @@ -48,18 +48,16 @@ static int fio_netio_prep(struct thread_data *td, struct io_u *io_u) struct net_data *nd = td->io_ops->data; struct fio_file *f = io_u->file; - if (nd->send_to_net) { - if (io_u->ddir == DDIR_READ) { - td_verror(td, EINVAL); - return 1; - } - } else { - if (io_u->ddir == DDIR_WRITE) { - td_verror(td, EINVAL); - return 1; - } + /* + * Make sure we don't see spurious reads to a receiver, and vice versa + */ + if ((nd->send_to_net && io_u->ddir == DDIR_READ) || + (!nd->send_to_net && io_u->ddir == DDIR_WRITE)) { + printf("boo!\n"); + td_verror(td, EINVAL); + return 1; } - + if (io_u->ddir == DDIR_SYNC) return 0; if (io_u->offset == f->last_completed_pos) @@ -77,12 +75,22 @@ static int fio_netio_queue(struct thread_data *td, struct io_u *io_u) { struct net_data *nd = td->io_ops->data; struct fio_file *f = io_u->file; - int ret = 0; - - if (io_u->ddir == DDIR_WRITE) - ret = write(f->fd, io_u->xfer_buf, io_u->xfer_buflen); - else if (io_u->ddir == DDIR_READ) - ret = read(f->fd, io_u->xfer_buf, io_u->xfer_buflen); + int ret, flags = 0; + + 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); + } else if (io_u->ddir == DDIR_READ) { + flags = MSG_WAITALL; + ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags); + } else + ret = 0; /* must be a SYNC */ if (ret != (int) io_u->xfer_buflen) { if (ret > 0) { @@ -95,6 +103,8 @@ static int fio_netio_queue(struct thread_data *td, struct io_u *io_u) if (!io_u->error) nd->last_io_u = io_u; + else + td_verror(td, io_u->error); return io_u->error; } @@ -111,10 +121,11 @@ static int fio_netio_setup_connect(struct thread_data *td, const char *host, addr.sin_port = htons(port); if (inet_aton(host, &addr.sin_addr) != 1) { - struct hostent *hent = gethostbyname(host); + struct hostent *hent; + hent = gethostbyname(host); if (!hent) { - td_vmsg(td, errno, "gethostbyname"); + td_verror(td, errno); return 1; } @@ -122,14 +133,14 @@ static int fio_netio_setup_connect(struct thread_data *td, const char *host, } for_each_file(td, f, i) { - f->fd = socket(AF_INET, SOCK_STREAM, 0); + f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if (f->fd < 0) { - td_vmsg(td, errno, "socket"); + td_verror(td, errno); return 1; } if (connect(f->fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) { - td_vmsg(td, errno, "connect"); + td_verror(td, errno); return 1; } } @@ -138,24 +149,81 @@ static int fio_netio_setup_connect(struct thread_data *td, const char *host, } +static int fio_netio_accept_connections(struct thread_data *td, int fd, + struct sockaddr_in *addr) +{ + socklen_t socklen = sizeof(*addr); + unsigned int accepts = 0; + struct pollfd pfd; + + fprintf(f_out, "fio: waiting for %u connections\n", td->nr_files); + + /* + * Accept loop. poll for incoming events, accept them. Repeat until we + * have all connections. + */ + while (!td->terminate && accepts < td->nr_files) { + struct fio_file *f; + int ret, i; + + pfd.fd = fd; + pfd.events = POLLIN; + + ret = poll(&pfd, 1, -1); + if (ret < 0) { + if (errno == EINTR) + continue; + + td_verror(td, errno); + break; + } else if (!ret) + continue; + + /* + * should be impossible + */ + if (!(pfd.revents & POLLIN)) + continue; + + for_each_file(td, f, i) { + if (f->fd != -1) + continue; + + f->fd = accept(fd, (struct sockaddr *) addr, &socklen); + if (f->fd < 0) { + td_verror(td, errno); + return 1; + } + accepts++; + break; + } + } + + return 0; +} + static int fio_netio_setup_listen(struct thread_data *td, unsigned short port) { struct sockaddr_in addr; - socklen_t socklen; - struct fio_file *f; - int fd, opt, i; + int fd, opt; - fd = socket(AF_INET, SOCK_STREAM, 0); + fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if (fd < 0) { - td_vmsg(td, errno, "socket"); + td_verror(td, errno); return 1; } opt = 1; if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) { - td_vmsg(td, errno, "setsockopt"); + td_verror(td, errno); return 1; } +#ifdef SO_REUSEPORT + if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) { + td_verror(td, errno); + return 1; + } +#endif memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; @@ -163,26 +231,15 @@ static int fio_netio_setup_listen(struct thread_data *td, unsigned short port) addr.sin_port = htons(port); if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) { - td_vmsg(td, errno, "bind"); + td_verror(td, errno); return 1; } if (listen(fd, 1) < 0) { - td_vmsg(td, errno, "listen"); + td_verror(td, errno); return 1; } - fprintf(f_out, "fio: waiting for %u connections\n", td->nr_files); - - socklen = sizeof(addr); - for_each_file(td, f, i) { - f->fd = accept(fd, (struct sockaddr *) &addr, &socklen); - if (f->fd < 0) { - td_vmsg(td, errno, "accept"); - return 1; - } - } - - return 0; + return fio_netio_accept_connections(td, fd, &addr); } static int fio_netio_setup(struct thread_data *td) @@ -194,6 +251,11 @@ static int fio_netio_setup(struct thread_data *td) char *sep; int ret, i; + if (!td->total_file_size) { + log_err("fio: need size= set\n"); + return 1; + } + /* * work around for late init call */ @@ -220,7 +282,7 @@ static int fio_netio_setup(struct thread_data *td) strcpy(host, buf); port = atoi(sep); - if (td->ddir == READ) { + if (td->ddir == DDIR_READ) { nd->send_to_net = 0; ret = fio_netio_setup_listen(td, port); } else {