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

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 flags = 0;

	/*
	 * 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 flags = MSG_WAITALL;

	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);
	struct pollfd pfd;
	int ret;

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

	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;

			td_verror(td, errno, "poll");
			break;
		} else if (!ret)
			continue;

		/*
		 * should be impossible
		 */
		if (!(pfd.revents & POLLIN))
			continue;

		f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
		if (f->fd < 0) {
			td_verror(td, errno, "accept");
			return 1;
		}
		break;
	}

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