X-Git-Url: https://git.kernel.dk/?a=blobdiff_plain;f=engines%2Fnet.c;h=700f1626be8640c43f373375af0c54942b63e1e0;hb=d5b388a560783a61af7b424757bc6dead2309c28;hp=45914776d0b9c47a4d5110d7c575c7f4f4ddf2c7;hpb=2992b059b8f54ac91e723a8bde629b4d8fed513e;p=fio.git diff --git a/engines/net.c b/engines/net.c index 45914776..700f1626 100644 --- a/engines/net.c +++ b/engines/net.c @@ -13,6 +13,8 @@ #include #include #include +#include +#include #include "../fio.h" @@ -20,11 +22,43 @@ 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) + break; + + td_verror(td, errno, "poll"); + return -1; + } else if (!ret) + continue; + + break; + } + + if (pfd.revents & events) + 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; @@ -165,20 +199,25 @@ static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u) #else static int fio_netio_splice_in(struct thread_data *td, struct io_u *io_u) { - errno = -EOPNOTSUPP; + errno = EOPNOTSUPP; return -1; } static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u) { - errno = -EOPNOTSUPP; + 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 @@ -188,14 +227,33 @@ static int fio_netio_send(struct thread_data *td, struct io_u *io_u) 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; + + if (nd->net_protocol == IPPROTO_UDP) { + socklen_t len = sizeof(nd->addr); - return recv(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags); + 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) @@ -206,15 +264,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 */ @@ -223,8 +281,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) @@ -236,13 +300,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; - f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (nd->net_protocol == IPPROTO_TCP) + type = SOCK_STREAM; + else + type = SOCK_DGRAM; + + 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; @@ -255,41 +328,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; @@ -329,9 +382,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; @@ -357,7 +415,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; } @@ -371,7 +429,7 @@ static int fio_netio_init(struct thread_data *td) struct netio_data *nd = td->io_ops->data; unsigned int port; char host[64], buf[128]; - char *sep; + char *sep, *portp, *modep; int ret; if (td_rw(td)) { @@ -395,10 +453,30 @@ static int fio_netio_init(struct thread_data *td) if (!strlen(host)) goto bad_host; - port = strtol(sep, NULL, 10); + 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; ret = fio_netio_setup_listen(td, port); @@ -409,7 +487,7 @@ static int fio_netio_init(struct thread_data *td) return ret; bad_host: - log_err("fio: bad network host/port: %s\n", td->o.filename); + log_err("fio: bad network host/port/protocol: %s\n", td->o.filename); return 1; }