[fio.git] / engines / net.c

/*
 * net engine
 *
 * IO engine that reads/writes to/from sockets.
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <assert.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/poll.h>
#include <sys/types.h>
#include <sys/socket.h>

#include "../fio.h"

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;
};

struct udp_close_msg {
	uint32_t magic;
	uint32_t cmd;
};

enum {
	FIO_LINK_CLOSE = 0x89,
	FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
};

/*
 * 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;

	/*
	 * 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)) {
		td_verror(td, EINVAL, "bad direction");
		return 1;
	}
		
	return 0;
}

#ifdef FIO_HAVE_SPLICE
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;
}
#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)
{
	struct netio_data *nd = td->io_ops->data;
	int ret, flags = 0;
#ifdef MSG_DONTWAIT
	flags = MSG_DONTWAIT;
#endif

	do {
		if (nd->net_protocol == IPPROTO_UDP) {
			struct sockaddr *to = (struct sockaddr *) &nd->addr;

			ret = sendto(io_u->file->fd, io_u->xfer_buf,
					io_u->xfer_buflen, flags, to,
					sizeof(*to));
		} else {
			/*
			 * 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
			ret = send(io_u->file->fd, io_u->xfer_buf,
					io_u->xfer_buflen, flags);
		}
		if (ret > 0)
			break;

		ret = poll_wait(td, io_u->file->fd, POLLOUT);
		if (ret <= 0)
			break;

#ifdef MSG_DONTWAIT
		flags &= ~MSG_DONTWAIT;
#endif
	} while (1);

	return ret;
}

static int is_udp_close(struct io_u *io_u, int len)
{
	struct udp_close_msg *msg;

	if (len != sizeof(struct udp_close_msg))
		return 0;

	msg = io_u->xfer_buf;
	if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
		return 0;
	if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
		return 0;

	return 1;
}

static int fio_netio_recv(struct thread_data *td, struct io_u *io_u)
{
	struct netio_data *nd = td->io_ops->data;
	int ret, flags = 0;
#ifdef MSG_DONTWAIT
	flags = MSG_DONTWAIT;
#endif

	do {
		if (nd->net_protocol == IPPROTO_UDP) {
			socklen_t len = sizeof(nd->addr);
			struct sockaddr *from = (struct sockaddr *) &nd->addr;

			ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
					io_u->xfer_buflen, flags, from, &len);
			if (is_udp_close(io_u, ret)) {
				td->done = 1;
				return 0;
			}
		} else {
			ret = recv(io_u->file->fd, io_u->xfer_buf,
					io_u->xfer_buflen, flags);
		}
		if (ret > 0)
			break;

		ret = poll_wait(td, io_u->file->fd, POLLIN);
		if (ret <= 0)
			break;
#ifdef MSG_DONTWAIT
		flags &= ~MSG_DONTWAIT;
#endif
		flags |= MSG_WAITALL;
	} while (1);

	return ret;
}

static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
{
	struct netio_data *nd = td->io_ops->data;
	int ret;

	fio_ro_check(td, io_u);

	if (io_u->ddir == DDIR_WRITE) {
		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 || nd->net_protocol == IPPROTO_UDP)
			ret = fio_netio_recv(td, io_u);
		else
			ret = fio_netio_splice_in(td, io_u);
	} else
		ret = 0;	/* must be a SYNC */

	if (ret != (int) io_u->xfer_buflen) {
		if (ret >= 0) {
			io_u->resid = io_u->xfer_buflen - ret;
			io_u->error = 0;
			return FIO_Q_COMPLETED;
		} else {
			int err = errno;

			if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
				return FIO_Q_BUSY;

			io_u->error = err;
		}
	}

	if (io_u->error)
		td_verror(td, io_u->error, "xfer");

	return FIO_Q_COMPLETED;
}

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, 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;
	}

	return 0;
}

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);

	if (nd->net_protocol == IPPROTO_UDP) {
		f->fd = nd->listenfd;
		return 0;
	}

	log_info("fio: waiting for connection\n");

	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;
	}

	return 0;
}

static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
{
	if (td_read(td))
		return fio_netio_accept(td, f);
	else
		return fio_netio_connect(td, f);
}

static void fio_netio_udp_close(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_CLOSE_MAGIC);
	msg.cmd = htonl(FIO_LINK_CLOSE);

	ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
			sizeof(nd->addr));
	if (ret < 0)
		td_verror(td, errno, "sendto udp link close");
}

static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
{
	struct netio_data *nd = td->io_ops->data;

	/*
	 * If this is an UDP connection, notify the receiver that we are
	 * closing down the link
	 */
	if (nd->net_protocol == IPPROTO_UDP)
		fio_netio_udp_close(td, f);

	return generic_close_file(td, f);
}

static int fio_netio_setup_connect(struct thread_data *td, const char *host,
				   unsigned short port)
{
	struct netio_data *nd = td->io_ops->data;

	nd->addr.sin_family = AF_INET;
	nd->addr.sin_port = htons(port);

	if (inet_aton(host, &nd->addr.sin_addr) != 1) {
		struct hostent *hent;

		hent = gethostbyname(host);
		if (!hent) {
			td_verror(td, errno, "gethostbyname");
			return 1;
		}

		memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
	}

	return 0;
}

static int fio_netio_setup_listen(struct thread_data *td, short port)
{
	struct netio_data *nd = td->io_ops->data;
	int fd, opt, type;

	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;
	}

	opt = 1;
	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
		td_verror(td, errno, "setsockopt");
		return 1;
	}
#ifdef SO_REUSEPORT
	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
		td_verror(td, errno, "setsockopt");
		return 1;
	}
#endif

	nd->addr.sin_family = AF_INET;
	nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
	nd->addr.sin_port = htons(port);

	if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
		td_verror(td, errno, "bind");
		return 1;
	}
	if (nd->net_protocol == IPPROTO_TCP && listen(fd, 1) < 0) {
		td_verror(td, errno, "listen");
		return 1;
	}

	nd->listenfd = fd;
	return 0;
}

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, *portp, *modep;
	int ret;

	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->o.filename);

	sep = strchr(buf, '/');
	if (!sep)
		goto bad_host;

	*sep = '\0';
	sep++;
	strcpy(host, buf);
	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;
		ret = fio_netio_setup_listen(td, port);
	} else {
		nd->send_to_net = 1;
		ret = fio_netio_setup_connect(td, host, port);
	}

	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)
{
	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);
	}
}

static int fio_netio_setup(struct thread_data *td)
{
	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;
	}

	return 0;
}

#ifdef FIO_HAVE_SPLICE
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_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 |
			  FIO_SIGQUIT | FIO_PIPEIO,
};
#endif

static struct ioengine_ops ioengine_rw = {
	.name		= "net",
	.version	= FIO_IOOPS_VERSION,
	.prep		= fio_netio_prep,
	.queue		= fio_netio_queue,
	.setup		= fio_netio_setup,
	.init		= fio_netio_init,
	.cleanup	= fio_netio_cleanup,
	.open_file	= fio_netio_open_file,
	.close_file	= fio_netio_close_file,
	.flags		= FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
			  FIO_SIGQUIT | FIO_PIPEIO,
};

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
}
Commit	Line	Data
ed92ac0c	1	/*
da751ca9 JA	2	* net engine
	3	*
	4	* IO engine that reads/writes to/from sockets.
	5	*
ed92ac0c JA	6	*/
	7	#include <stdio.h>
	8	#include <stdlib.h>
	9	#include <unistd.h>
	10	#include <errno.h>
	11	#include <assert.h>
	12	#include <netinet/in.h>
	13	#include <arpa/inet.h>
	14	#include <netdb.h>
5fdd124a	15	#include <sys/poll.h>
7292056a JA	16	#include <sys/types.h>
7292056a JA	17	#include <sys/socket.h>
ed92ac0c JA	18
ed92ac0c JA	19	#include "../fio.h"
ed92ac0c	20
b5af8293 JA	21	struct netio_data {
	22	int listenfd;
	23	int send_to_net;
9cce02e8	24	int use_splice;
414c2a3e	25	int net_protocol;
9cce02e8	26	int pipes[2];
b5af8293 JA	27	char host[64];
	28	struct sockaddr_in addr;
	29	};
ed92ac0c	30
664fb3bd JA	31	struct udp_close_msg {
	32	uint32_t magic;
	33	uint32_t cmd;
	34	};
	35
	36	enum {
	37	FIO_LINK_CLOSE = 0x89,
	38	FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
	39	};
	40
371d456c JA	41	/*
	42	* Return -1 for error and 'nr events' for a positive number
	43	* of events
	44	*/
	45	static int poll_wait(struct thread_data *td, int fd, short events)
	46	{
	47	struct pollfd pfd;
	48	int ret;
	49
	50	while (!td->terminate) {
	51	pfd.fd = fd;
	52	pfd.events = events;
	53	ret = poll(&pfd, 1, -1);
	54	if (ret < 0) {
	55	if (errno == EINTR)
d5b388a5	56	break;
371d456c JA	57
	58	td_verror(td, errno, "poll");
	59	return -1;
	60	} else if (!ret)
	61	continue;
	62
	63	break;
	64	}
	65
	66	if (pfd.revents & events)
	67	return 1;
371d456c JA	68
	69	return -1;
	70	}
	71
ed92ac0c JA	72	static int fio_netio_prep(struct thread_data td, struct io_u io_u)
ed92ac0c JA	73	{
b5af8293	74	struct netio_data *nd = td->io_ops->data;
ed92ac0c	75
7a6499da JA	76	/*
	77	* Make sure we don't see spurious reads to a receiver, and vice versa
	78	*/
b5af8293 JA	79	if ((nd->send_to_net && io_u->ddir == DDIR_READ) \|\|
b5af8293 JA	80	(!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
e1161c32	81	td_verror(td, EINVAL, "bad direction");
7a6499da	82	return 1;
ed92ac0c	83	}
7a6499da	84
f85ac25a	85	return 0;
ed92ac0c JA	86	}
ed92ac0c JA	87
5921e80c	88	#ifdef FIO_HAVE_SPLICE
cd963e18	89	static int splice_io_u(int fdin, int fdout, unsigned int len)
ed92ac0c	90	{
9cce02e8	91	int bytes = 0;
7a6499da	92
9cce02e8	93	while (len) {
cd963e18	94	int ret = splice(fdin, NULL, fdout, NULL, len, 0);
9cce02e8 JA	95
	96	if (ret < 0) {
	97	if (!bytes)
	98	bytes = ret;
	99
	100	break;
	101	} else if (!ret)
	102	break;
	103
	104	bytes += ret;
f657a2fb	105	len -= ret;
9cce02e8 JA	106	}
	107
	108	return bytes;
	109	}
	110
	111	/*
cd963e18	112	* Receive bytes from a socket and fill them into the internal pipe
9cce02e8	113	*/
cd963e18	114	static int splice_in(struct thread_data td, struct io_u io_u)
9cce02e8 JA	115	{
9cce02e8 JA	116	struct netio_data *nd = td->io_ops->data;
9cce02e8	117
cd963e18	118	return splice_io_u(io_u->file->fd, nd->pipes[1], io_u->xfer_buflen);
9cce02e8 JA	119	}
	120
	121	/*
cd963e18	122	* Transmit 'len' bytes from the internal pipe
9cce02e8	123	*/
cd963e18 JA	124	static int splice_out(struct thread_data td, struct io_u io_u,
cd963e18 JA	125	unsigned int len)
9cce02e8 JA	126	{
9cce02e8 JA	127	struct netio_data *nd = td->io_ops->data;
cd963e18 JA	128
	129	return splice_io_u(nd->pipes[0], io_u->file->fd, len);
	130	}
	131
	132	static int vmsplice_io_u(struct io_u *io_u, int fd, unsigned int len)
	133	{
9cce02e8 JA	134	struct iovec iov = {
	135	.iov_base = io_u->xfer_buf,
	136	.iov_len = len,
	137	};
	138	int bytes = 0;
	139
	140	while (iov.iov_len) {
cd963e18	141	int ret = vmsplice(fd, &iov, 1, SPLICE_F_MOVE);
9cce02e8 JA	142
	143	if (ret < 0) {
	144	if (!bytes)
	145	bytes = ret;
	146	break;
	147	} else if (!ret)
	148	break;
	149
	150	iov.iov_len -= ret;
cd963e18	151	iov.iov_base += ret;
f657a2fb	152	bytes += ret;
9cce02e8 JA	153	}
	154
	155	return bytes;
cd963e18	156
9cce02e8 JA	157	}
	158
	159	/*
cd963e18	160	* vmsplice() pipe to io_u buffer
9cce02e8	161	*/
cd963e18 JA	162	static int vmsplice_io_u_out(struct thread_data td, struct io_u io_u,
cd963e18 JA	163	unsigned int len)
9cce02e8 JA	164	{
9cce02e8 JA	165	struct netio_data *nd = td->io_ops->data;
9cce02e8	166
cd963e18 JA	167	return vmsplice_io_u(io_u, nd->pipes[0], len);
cd963e18 JA	168	}
9cce02e8	169
cd963e18 JA	170	/*
	171	* vmsplice() io_u to pipe
	172	*/
	173	static int vmsplice_io_u_in(struct thread_data td, struct io_u io_u)
	174	{
	175	struct netio_data *nd = td->io_ops->data;
ed92ac0c	176
cd963e18	177	return vmsplice_io_u(io_u, nd->pipes[1], io_u->xfer_buflen);
9cce02e8 JA	178	}
9cce02e8 JA	179
cd963e18 JA	180	/*
	181	* splice receive - transfer socket data into a pipe using splice, then map
	182	* that pipe data into the io_u using vmsplice.
	183	*/
9cce02e8 JA	184	static int fio_netio_splice_in(struct thread_data td, struct io_u io_u)
	185	{
	186	int ret;
	187
	188	ret = splice_in(td, io_u);
cd963e18 JA	189	if (ret > 0)
cd963e18 JA	190	return vmsplice_io_u_out(td, io_u, ret);
9cce02e8	191
cd963e18	192	return ret;
9cce02e8 JA	193	}
9cce02e8 JA	194
cd963e18 JA	195	/*
	196	* splice transmit - map data from the io_u into a pipe by using vmsplice,
	197	* then transfer that pipe to a socket using splice.
	198	*/
9cce02e8 JA	199	static int fio_netio_splice_out(struct thread_data td, struct io_u io_u)
	200	{
	201	int ret;
	202
	203	ret = vmsplice_io_u_in(td, io_u);
cd963e18 JA	204	if (ret > 0)
cd963e18 JA	205	return splice_out(td, io_u, ret);
9cce02e8	206
cd963e18	207	return ret;
9cce02e8	208	}
5921e80c JA	209	#else
	210	static int fio_netio_splice_in(struct thread_data td, struct io_u io_u)
	211	{
af8771b9	212	errno = EOPNOTSUPP;
5921e80c JA	213	return -1;
	214	}
	215
	216	static int fio_netio_splice_out(struct thread_data td, struct io_u io_u)
	217	{
af8771b9	218	errno = EOPNOTSUPP;
5921e80c JA	219	return -1;
	220	}
	221	#endif
9cce02e8 JA	222
	223	static int fio_netio_send(struct thread_data td, struct io_u io_u)
	224	{
414c2a3e	225	struct netio_data *nd = td->io_ops->data;
f5886fef CC	226	int ret, flags = 0;
	227	#ifdef MSG_DONTWAIT
	228	flags = MSG_DONTWAIT;
	229	#endif
371d456c	230
664fb3bd JA	231	do {
664fb3bd JA	232	if (nd->net_protocol == IPPROTO_UDP) {
62b38926 JA	233	struct sockaddr to = (struct sockaddr ) &nd->addr;
62b38926 JA	234
664fb3bd	235	ret = sendto(io_u->file->fd, io_u->xfer_buf,
62b38926 JA	236	io_u->xfer_buflen, flags, to,
62b38926 JA	237	sizeof(*to));
664fb3bd JA	238	} else {
	239	/*
	240	* if we are going to write more, set MSG_MORE
	241	*/
5921e80c	242	#ifdef MSG_MORE
664fb3bd JA	243	if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
	244	td->o.size)
	245	flags \|= MSG_MORE;
5921e80c	246	#endif
664fb3bd JA	247	ret = send(io_u->file->fd, io_u->xfer_buf,
	248	io_u->xfer_buflen, flags);
	249	}
	250	if (ret > 0)
	251	break;
9cce02e8	252
664fb3bd JA	253	ret = poll_wait(td, io_u->file->fd, POLLOUT);
	254	if (ret <= 0)
	255	break;
	256
f5886fef	257	#ifdef MSG_DONTWAIT
664fb3bd	258	flags &= ~MSG_DONTWAIT;
f5886fef	259	#endif
664fb3bd JA	260	} while (1);
	261
	262	return ret;
	263	}
	264
	265	static int is_udp_close(struct io_u *io_u, int len)
	266	{
	267	struct udp_close_msg *msg;
	268
	269	if (len != sizeof(struct udp_close_msg))
	270	return 0;
	271
	272	msg = io_u->xfer_buf;
	273	if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
	274	return 0;
	275	if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
	276	return 0;
	277
	278	return 1;
9cce02e8 JA	279	}
9cce02e8 JA	280
414c2a3e	281	static int fio_netio_recv(struct thread_data td, struct io_u io_u)
9cce02e8	282	{
414c2a3e	283	struct netio_data *nd = td->io_ops->data;
f5886fef CC	284	int ret, flags = 0;
	285	#ifdef MSG_DONTWAIT
	286	flags = MSG_DONTWAIT;
	287	#endif
664fb3bd JA	288
	289	do {
	290	if (nd->net_protocol == IPPROTO_UDP) {
	291	socklen_t len = sizeof(nd->addr);
62b38926	292	struct sockaddr from = (struct sockaddr ) &nd->addr;
664fb3bd JA	293
664fb3bd JA	294	ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
62b38926	295	io_u->xfer_buflen, flags, from, &len);
664fb3bd JA	296	if (is_udp_close(io_u, ret)) {
	297	td->done = 1;
	298	return 0;
	299	}
	300	} else {
	301	ret = recv(io_u->file->fd, io_u->xfer_buf,
	302	io_u->xfer_buflen, flags);
	303	}
	304	if (ret > 0)
	305	break;
9cce02e8	306
664fb3bd JA	307	ret = poll_wait(td, io_u->file->fd, POLLIN);
	308	if (ret <= 0)
	309	break;
f5886fef	310	#ifdef MSG_DONTWAIT
664fb3bd	311	flags &= ~MSG_DONTWAIT;
f5886fef	312	#endif
664fb3bd JA	313	flags \|= MSG_WAITALL;
664fb3bd JA	314	} while (1);
414c2a3e	315
664fb3bd	316	return ret;
9cce02e8 JA	317	}
	318
	319	static int fio_netio_queue(struct thread_data td, struct io_u io_u)
	320	{
	321	struct netio_data *nd = td->io_ops->data;
	322	int ret;
	323
7101d9c2 JA	324	fio_ro_check(td, io_u);
7101d9c2 JA	325
9cce02e8	326	if (io_u->ddir == DDIR_WRITE) {
414c2a3e	327	if (!nd->use_splice \|\| nd->net_protocol == IPPROTO_UDP)
9cce02e8	328	ret = fio_netio_send(td, io_u);
414c2a3e JA	329	else
414c2a3e JA	330	ret = fio_netio_splice_out(td, io_u);
d4f12dd0	331	} else if (io_u->ddir == DDIR_READ) {
414c2a3e JA	332	if (!nd->use_splice \|\| nd->net_protocol == IPPROTO_UDP)
414c2a3e JA	333	ret = fio_netio_recv(td, io_u);
9cce02e8	334	else
414c2a3e	335	ret = fio_netio_splice_in(td, io_u);
d4f12dd0	336	} else
7a6499da	337	ret = 0; /* must be a SYNC */
ed92ac0c	338
cec6b55d	339	if (ret != (int) io_u->xfer_buflen) {
22819ec2	340	if (ret >= 0) {
cec6b55d JA	341	io_u->resid = io_u->xfer_buflen - ret;
cec6b55d JA	342	io_u->error = 0;
36167d82	343	return FIO_Q_COMPLETED;
414c2a3e JA	344	} else {
	345	int err = errno;
	346
	347	if (io_u->ddir == DDIR_WRITE && err == EMSGSIZE)
	348	return FIO_Q_BUSY;
	349
	350	io_u->error = err;
	351	}
ed92ac0c JA	352	}
ed92ac0c JA	353
36167d82	354	if (io_u->error)
e1161c32	355	td_verror(td, io_u->error, "xfer");
ed92ac0c	356
36167d82	357	return FIO_Q_COMPLETED;
ed92ac0c JA	358	}
ed92ac0c JA	359
b5af8293	360	static int fio_netio_connect(struct thread_data td, struct fio_file f)
ed92ac0c	361	{
b5af8293	362	struct netio_data *nd = td->io_ops->data;
414c2a3e JA	363	int type;
	364
	365	if (nd->net_protocol == IPPROTO_TCP)
	366	type = SOCK_STREAM;
	367	else
	368	type = SOCK_DGRAM;
ed92ac0c	369
414c2a3e	370	f->fd = socket(AF_INET, type, nd->net_protocol);
b5af8293 JA	371	if (f->fd < 0) {
	372	td_verror(td, errno, "socket");
	373	return 1;
ed92ac0c JA	374	}
ed92ac0c JA	375
414c2a3e JA	376	if (nd->net_protocol == IPPROTO_UDP)
	377	return 0;
	378
b5af8293 JA	379	if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
	380	td_verror(td, errno, "connect");
	381	return 1;
ed92ac0c JA	382	}
	383
	384	return 0;
ed92ac0c JA	385	}
ed92ac0c JA	386
b5af8293	387	static int fio_netio_accept(struct thread_data td, struct fio_file f)
5fdd124a	388	{
b5af8293 JA	389	struct netio_data *nd = td->io_ops->data;
b5af8293 JA	390	socklen_t socklen = sizeof(nd->addr);
5fdd124a	391
414c2a3e JA	392	if (nd->net_protocol == IPPROTO_UDP) {
	393	f->fd = nd->listenfd;
	394	return 0;
	395	}
	396
6d86144d	397	log_info("fio: waiting for connection\n");
5fdd124a	398
371d456c JA	399	if (poll_wait(td, nd->listenfd, POLLIN) < 0)
371d456c JA	400	return 1;
0c09442b	401
371d456c JA	402	f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
	403	if (f->fd < 0) {
	404	td_verror(td, errno, "accept");
	405	return 1;
b5af8293	406	}
5fdd124a	407
b5af8293 JA	408	return 0;
	409	}
	410
b5af8293 JA	411	static int fio_netio_open_file(struct thread_data td, struct fio_file f)
	412	{
	413	if (td_read(td))
	414	return fio_netio_accept(td, f);
	415	else
	416	return fio_netio_connect(td, f);
	417	}
	418
664fb3bd JA	419	static void fio_netio_udp_close(struct thread_data td, struct fio_file f)
	420	{
	421	struct netio_data *nd = td->io_ops->data;
	422	struct udp_close_msg msg;
62b38926	423	struct sockaddr to = (struct sockaddr ) &nd->addr;
664fb3bd JA	424	int ret;
	425
	426	msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
	427	msg.cmd = htonl(FIO_LINK_CLOSE);
	428
62b38926	429	ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
664fb3bd JA	430	sizeof(nd->addr));
	431	if (ret < 0)
	432	td_verror(td, errno, "sendto udp link close");
	433	}
	434
	435	static int fio_netio_close_file(struct thread_data td, struct fio_file f)
	436	{
	437	struct netio_data *nd = td->io_ops->data;
	438
	439	/*
	440	* If this is an UDP connection, notify the receiver that we are
	441	* closing down the link
	442	*/
	443	if (nd->net_protocol == IPPROTO_UDP)
	444	fio_netio_udp_close(td, f);
	445
	446	return generic_close_file(td, f);
	447	}
	448
b5af8293 JA	449	static int fio_netio_setup_connect(struct thread_data td, const char host,
	450	unsigned short port)
	451	{
	452	struct netio_data *nd = td->io_ops->data;
	453
	454	nd->addr.sin_family = AF_INET;
	455	nd->addr.sin_port = htons(port);
	456
	457	if (inet_aton(host, &nd->addr.sin_addr) != 1) {
	458	struct hostent *hent;
	459
	460	hent = gethostbyname(host);
	461	if (!hent) {
	462	td_verror(td, errno, "gethostbyname");
	463	return 1;
5fdd124a	464	}
b5af8293 JA	465
b5af8293 JA	466	memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
5fdd124a JA	467	}
	468
	469	return 0;
	470	}
	471
b5af8293	472	static int fio_netio_setup_listen(struct thread_data *td, short port)
ed92ac0c	473	{
b5af8293	474	struct netio_data *nd = td->io_ops->data;
414c2a3e	475	int fd, opt, type;
ed92ac0c	476
414c2a3e JA	477	if (nd->net_protocol == IPPROTO_TCP)
	478	type = SOCK_STREAM;
	479	else
	480	type = SOCK_DGRAM;
	481
	482	fd = socket(AF_INET, type, nd->net_protocol);
ed92ac0c	483	if (fd < 0) {
e1161c32	484	td_verror(td, errno, "socket");
ed92ac0c JA	485	return 1;
	486	}
	487
	488	opt = 1;
	489	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
e1161c32	490	td_verror(td, errno, "setsockopt");
ed92ac0c JA	491	return 1;
ed92ac0c JA	492	}
6bedbfaf JA	493	#ifdef SO_REUSEPORT
6bedbfaf JA	494	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
e1161c32	495	td_verror(td, errno, "setsockopt");
6bedbfaf JA	496	return 1;
	497	}
	498	#endif
ed92ac0c	499
b5af8293 JA	500	nd->addr.sin_family = AF_INET;
	501	nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
	502	nd->addr.sin_port = htons(port);
ed92ac0c	503
b5af8293	504	if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
e1161c32	505	td_verror(td, errno, "bind");
ed92ac0c JA	506	return 1;
ed92ac0c JA	507	}
414c2a3e	508	if (nd->net_protocol == IPPROTO_TCP && listen(fd, 1) < 0) {
e1161c32	509	td_verror(td, errno, "listen");
ed92ac0c JA	510	return 1;
	511	}
	512
b5af8293 JA	513	nd->listenfd = fd;
b5af8293 JA	514	return 0;
ed92ac0c JA	515	}
ed92ac0c JA	516
9bec88e1	517	static int fio_netio_init(struct thread_data *td)
ed92ac0c	518	{
b5af8293	519	struct netio_data *nd = td->io_ops->data;
443662ef	520	unsigned int port;
b5af8293	521	char host[64], buf[128];
414c2a3e	522	char sep, portp, *modep;
af52b345	523	int ret;
ed92ac0c	524
413dd459	525	if (td_rw(td)) {
ed92ac0c JA	526	log_err("fio: network connections must be read OR write\n");
	527	return 1;
	528	}
16d55aae JA	529	if (td_random(td)) {
	530	log_err("fio: network IO can't be random\n");
	531	return 1;
	532	}
ed92ac0c	533
2dc1bbeb	534	strcpy(buf, td->o.filename);
ed92ac0c	535
9f9214f2	536	sep = strchr(buf, '/');
443662ef JA	537	if (!sep)
443662ef JA	538	goto bad_host;
ed92ac0c JA	539
	540	*sep = '\0';
	541	sep++;
	542	strcpy(host, buf);
443662ef JA	543	if (!strlen(host))
	544	goto bad_host;
	545
414c2a3e JA	546	modep = NULL;
	547	portp = sep;
	548	sep = strchr(portp, '/');
	549	if (sep) {
	550	*sep = '\0';
	551	modep = sep + 1;
	552	}
	553
	554	port = strtol(portp, NULL, 10);
443662ef JA	555	if (!port \|\| port > 65535)
443662ef JA	556	goto bad_host;
ed92ac0c	557
414c2a3e	558	if (modep) {
3f8fc5ad JA	559	if (!strncmp("tcp", modep, strlen(modep)) \|\|
3f8fc5ad JA	560	!strncmp("TCP", modep, strlen(modep)))
414c2a3e	561	nd->net_protocol = IPPROTO_TCP;
3f8fc5ad JA	562	else if (!strncmp("udp", modep, strlen(modep)) \|\|
3f8fc5ad JA	563	!strncmp("UDP", modep, strlen(modep)))
414c2a3e JA	564	nd->net_protocol = IPPROTO_UDP;
	565	else
	566	goto bad_host;
	567	} else
	568	nd->net_protocol = IPPROTO_TCP;
	569
413dd459	570	if (td_read(td)) {
b5af8293	571	nd->send_to_net = 0;
ed92ac0c JA	572	ret = fio_netio_setup_listen(td, port);
ed92ac0c JA	573	} else {
b5af8293	574	nd->send_to_net = 1;
ed92ac0c JA	575	ret = fio_netio_setup_connect(td, host, port);
	576	}
	577
7bb48f84	578	return ret;
443662ef	579	bad_host:
414c2a3e	580	log_err("fio: bad network host/port/protocol: %s\n", td->o.filename);
443662ef	581	return 1;
ed92ac0c JA	582	}
ed92ac0c JA	583
b5af8293	584	static void fio_netio_cleanup(struct thread_data *td)
9bec88e1	585	{
b5af8293 JA	586	struct netio_data *nd = td->io_ops->data;
	587
	588	if (nd) {
64b24cd8 JA	589	if (nd->listenfd != -1)
	590	close(nd->listenfd);
	591	if (nd->pipes[0] != -1)
	592	close(nd->pipes[0]);
	593	if (nd->pipes[1] != -1)
	594	close(nd->pipes[1]);
	595
b5af8293	596	free(nd);
b5af8293 JA	597	}
	598	}
	599
	600	static int fio_netio_setup(struct thread_data *td)
	601	{
7bb48f84	602	struct netio_data *nd;
7bb48f84 JA	603
	604	if (!td->io_ops->data) {
	605	nd = malloc(sizeof(*nd));;
	606
	607	memset(nd, 0, sizeof(*nd));
	608	nd->listenfd = -1;
64b24cd8	609	nd->pipes[0] = nd->pipes[1] = -1;
7bb48f84	610	td->io_ops->data = nd;
7bb48f84	611	}
b5af8293	612
9bec88e1 JA	613	return 0;
	614	}
	615
5921e80c	616	#ifdef FIO_HAVE_SPLICE
9cce02e8 JA	617	static int fio_netio_setup_splice(struct thread_data *td)
	618	{
	619	struct netio_data *nd;
	620
	621	fio_netio_setup(td);
	622
	623	nd = td->io_ops->data;
	624	if (nd) {
	625	if (pipe(nd->pipes) < 0)
	626	return 1;
	627
	628	nd->use_splice = 1;
	629	return 0;
	630	}
	631
	632	return 1;
	633	}
	634
5921e80c JA	635	static struct ioengine_ops ioengine_splice = {
5921e80c JA	636	.name = "netsplice",
ed92ac0c	637	.version = FIO_IOOPS_VERSION,
ed92ac0c JA	638	.prep = fio_netio_prep,
ed92ac0c JA	639	.queue = fio_netio_queue,
5921e80c	640	.setup = fio_netio_setup_splice,
9bec88e1	641	.init = fio_netio_init,
b5af8293 JA	642	.cleanup = fio_netio_cleanup,
	643	.open_file = fio_netio_open_file,
	644	.close_file = generic_close_file,
ad830ed7	645	.flags = FIO_SYNCIO \| FIO_DISKLESSIO \| FIO_UNIDIR \|
9c0d2241	646	FIO_SIGQUIT \| FIO_PIPEIO,
ed92ac0c	647	};
5921e80c	648	#endif
ed92ac0c	649
5921e80c JA	650	static struct ioengine_ops ioengine_rw = {
5921e80c JA	651	.name = "net",
9cce02e8 JA	652	.version = FIO_IOOPS_VERSION,
	653	.prep = fio_netio_prep,
	654	.queue = fio_netio_queue,
5921e80c	655	.setup = fio_netio_setup,
9cce02e8 JA	656	.init = fio_netio_init,
	657	.cleanup = fio_netio_cleanup,
	658	.open_file = fio_netio_open_file,
664fb3bd	659	.close_file = fio_netio_close_file,
ad830ed7	660	.flags = FIO_SYNCIO \| FIO_DISKLESSIO \| FIO_UNIDIR \|
9c0d2241	661	FIO_SIGQUIT \| FIO_PIPEIO,
9cce02e8 JA	662	};
9cce02e8 JA	663
ed92ac0c JA	664	static void fio_init fio_netio_register(void)
ed92ac0c JA	665	{
9cce02e8	666	register_ioengine(&ioengine_rw);
5921e80c	667	#ifdef FIO_HAVE_SPLICE
9cce02e8	668	register_ioengine(&ioengine_splice);
5921e80c	669	#endif
ed92ac0c JA	670	}
	671
	672	static void fio_exit fio_netio_unregister(void)
	673	{
9cce02e8	674	unregister_ioengine(&ioengine_rw);
5921e80c	675	#ifdef FIO_HAVE_SPLICE
9cce02e8	676	unregister_ioengine(&ioengine_splice);
5921e80c	677	#endif
ed92ac0c	678	}