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

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

			td_verror(td, errno, "poll");
			return -1;
		} else if (!ret)
			continue;

		break;
	}

	if (pfd.revents & events)
		return 1;
	else if (td->terminate)
		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;

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

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

	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 thread_data *td, struct io_u *io_u)
{
	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 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)
{
	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 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,
};
#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	= generic_close_file,
	.flags		= FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
			  FIO_SIGQUIT,
};

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