unsigned int port;
unsigned int proto;
unsigned int listen;
+ unsigned int pingpong;
};
struct udp_close_msg {
enum {
FIO_LINK_CLOSE = 0x89,
- FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
+ FIO_LINK_OPEN_CLOSE_MAGIC = 0x6c696e6b,
+ FIO_LINK_OPEN = 0x98,
FIO_TYPE_TCP = 1,
FIO_TYPE_UDP = 2,
},
{ .ival = "udp",
.oval = FIO_TYPE_UDP,
- .help = "Unreliable Datagram Protocol",
+ .help = "User Datagram Protocol",
},
{ .ival = "unix",
.oval = FIO_TYPE_UNIX,
.off1 = offsetof(struct netio_options, listen),
.help = "Listen for incoming TCP connections",
},
+ {
+ .name = "pingpong",
+ .type = FIO_OPT_STR_SET,
+ .off1 = offsetof(struct netio_options, pingpong),
+ .help = "Ping-pong IO requests",
+ },
{
.name = NULL,
},
td_verror(td, EINVAL, "bad direction");
return 1;
}
-
+
return 0;
}
{
struct netio_data *nd = td->io_ops->data;
struct netio_options *o = td->eo;
- int ret, flags = OS_MSG_DONTWAIT;
+ int ret, flags = 0;
do {
if (o->proto == FIO_TYPE_UDP) {
* 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)
+ if ((td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
+ td->o.size) && !o->pingpong)
flags |= MSG_MORE;
#endif
ret = send(io_u->file->fd, io_u->xfer_buf,
ret = poll_wait(td, io_u->file->fd, POLLOUT);
if (ret <= 0)
break;
-
- flags &= ~OS_MSG_DONTWAIT;
} while (1);
return ret;
return 0;
msg = io_u->xfer_buf;
- if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
+ if (ntohl(msg->magic) != FIO_LINK_OPEN_CLOSE_MAGIC)
return 0;
if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
return 0;
{
struct netio_data *nd = td->io_ops->data;
struct netio_options *o = td->eo;
- int ret, flags = OS_MSG_DONTWAIT;
+ int ret, flags = 0;
do {
if (o->proto == FIO_TYPE_UDP) {
}
if (ret > 0)
break;
+ else if (!ret && (flags & MSG_WAITALL))
+ break;
ret = poll_wait(td, io_u->file->fd, POLLIN);
if (ret <= 0)
break;
- flags &= ~OS_MSG_DONTWAIT;
flags |= MSG_WAITALL;
} while (1);
return ret;
}
-static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
+static int __fio_netio_queue(struct thread_data *td, struct io_u *io_u,
+ enum fio_ddir ddir)
{
struct netio_data *nd = td->io_ops->data;
struct netio_options *o = td->eo;
int ret;
- fio_ro_check(td, io_u);
-
- if (io_u->ddir == DDIR_WRITE) {
+ if (ddir == DDIR_WRITE) {
if (!nd->use_splice || o->proto == FIO_TYPE_UDP ||
o->proto == FIO_TYPE_UNIX)
ret = fio_netio_send(td, io_u);
else
ret = fio_netio_splice_out(td, io_u);
- } else if (io_u->ddir == DDIR_READ) {
+ } else if (ddir == DDIR_READ) {
if (!nd->use_splice || o->proto == FIO_TYPE_UDP ||
o->proto == FIO_TYPE_UNIX)
ret = fio_netio_recv(td, io_u);
} else {
int err = errno;
- if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
+ if (ddir == DDIR_WRITE && err == EMSGSIZE)
return FIO_Q_BUSY;
io_u->error = err;
return FIO_Q_COMPLETED;
}
+static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
+{
+ struct netio_options *o = td->eo;
+ int ret;
+
+ fio_ro_check(td, io_u);
+
+ ret = __fio_netio_queue(td, io_u, io_u->ddir);
+ if (!o->pingpong || ret != FIO_Q_COMPLETED)
+ return ret;
+
+ /*
+ * For ping-pong mode, receive or send reply as needed
+ */
+ if (td_read(td) && io_u->ddir == DDIR_READ)
+ ret = __fio_netio_queue(td, io_u, DDIR_WRITE);
+ else if (td_write(td) && io_u->ddir == DDIR_WRITE)
+ ret = __fio_netio_queue(td, io_u, DDIR_READ);
+
+ return ret;
+}
+
static int fio_netio_connect(struct thread_data *td, struct fio_file *f)
{
struct netio_data *nd = td->io_ops->data;
struct netio_data *nd = td->io_ops->data;
struct netio_options *o = td->eo;
fio_socklen_t socklen = sizeof(nd->addr);
+ int state;
if (o->proto == FIO_TYPE_UDP) {
f->fd = nd->listenfd;
return 0;
}
+ state = td->runstate;
+ td_set_runstate(td, TD_SETTING_UP);
+
log_info("fio: waiting for connection\n");
if (poll_wait(td, nd->listenfd, POLLIN) < 0)
- return 1;
+ goto err;
f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
if (f->fd < 0) {
td_verror(td, errno, "accept");
- return 1;
+ goto err;
}
+ reset_all_stats(td);
+ td_set_runstate(td, state);
return 0;
-}
-
-static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
-{
- int ret;
- struct netio_options *o = td->eo;
-
- if (o->listen)
- ret = fio_netio_accept(td, f);
- else
- ret = fio_netio_connect(td, f);
-
- if (ret)
- f->fd = -1;
- return ret;
+err:
+ td_set_runstate(td, state);
+ return 1;
}
static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
struct sockaddr *to = (struct sockaddr *) &nd->addr;
int ret;
- msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
+ msg.magic = htonl(FIO_LINK_OPEN_CLOSE_MAGIC);
msg.cmd = htonl(FIO_LINK_CLOSE);
ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
return generic_close_file(td, f);
}
+static int fio_netio_udp_recv_open(struct thread_data *td, struct fio_file *f)
+{
+ struct netio_data *nd = td->io_ops->data;
+ struct udp_close_msg msg;
+ struct sockaddr *to = (struct sockaddr *) &nd->addr;
+ fio_socklen_t len = sizeof(nd->addr);
+ int ret;
+
+ ret = recvfrom(f->fd, &msg, sizeof(msg), MSG_WAITALL, to, &len);
+ if (ret < 0) {
+ td_verror(td, errno, "sendto udp link open");
+ return ret;
+ }
+
+ if (ntohl(msg.magic) != FIO_LINK_OPEN_CLOSE_MAGIC ||
+ ntohl(msg.cmd) != FIO_LINK_OPEN) {
+ log_err("fio: bad udp open magic %x/%x\n", ntohl(msg.magic),
+ ntohl(msg.cmd));
+ return -1;
+ }
+
+ return 0;
+}
+
+static int fio_netio_udp_send_open(struct thread_data *td, struct fio_file *f)
+{
+ struct netio_data *nd = td->io_ops->data;
+ struct udp_close_msg msg;
+ struct sockaddr *to = (struct sockaddr *) &nd->addr;
+ int ret;
+
+ msg.magic = htonl(FIO_LINK_OPEN_CLOSE_MAGIC);
+ msg.cmd = htonl(FIO_LINK_OPEN);
+
+ ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
+ sizeof(nd->addr));
+ if (ret < 0) {
+ td_verror(td, errno, "sendto udp link open");
+ return ret;
+ }
+
+ return 0;
+}
+
+static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
+{
+ int ret;
+ struct netio_options *o = td->eo;
+
+ if (o->listen)
+ ret = fio_netio_accept(td, f);
+ else
+ ret = fio_netio_connect(td, f);
+
+ if (ret) {
+ f->fd = -1;
+ return ret;
+ }
+
+ if (o->proto == FIO_TYPE_UDP) {
+ if (td_write(td))
+ ret = fio_netio_udp_send_open(td, f);
+ else {
+ int state;
+
+ state = td->runstate;
+ td_set_runstate(td, TD_SETTING_UP);
+ ret = fio_netio_udp_recv_open(td, f);
+ td_set_runstate(td, state);
+ }
+ }
+
+ if (ret)
+ fio_netio_close_file(td, f);
+
+ return ret;
+}
+
static int fio_netio_setup_connect_inet(struct thread_data *td,
const char *host, unsigned short port)
{
struct netio_data *nd = td->io_ops->data;
+ if (!host) {
+ log_err("fio: connect with no host to connect to.\n");
+ if (td_read(td))
+ log_err("fio: did you forget to set 'listen'?\n");
+
+ td_verror(td, EINVAL, "no hostname= set");
+ return 1;
+ }
+
nd->addr.sin_family = AF_INET;
nd->addr.sin_port = htons(port);
struct netio_options *o = td->eo;
int ret;
+#ifdef WIN32
+ WSADATA wsd;
+ WSAStartup(MAKEWORD(2,2), &wsd);
+#endif
+
if (td_random(td)) {
log_err("fio: network IO can't be random\n");
return 1;
if (o->proto != FIO_TYPE_TCP) {
if (o->listen) {
- log_err("fio: listen only valid for TCP proto IO\n");
- return 1;
+ log_err("fio: listen only valid for TCP proto IO\n");
+ return 1;
}
if (td_rw(td)) {
- log_err("fio: datagram network connections must be"
+ log_err("fio: datagram network connections must be"
" read OR write\n");
- return 1;
+ return 1;
+ }
+ if (o->proto == FIO_TYPE_UNIX && !td->o.filename) {
+ log_err("fio: UNIX sockets need host/filename\n");
+ return 1;
}
o->listen = td_read(td);
}
projects / fio.git / blobdiff