[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 "../fio.h"

struct netio_data {
	int listenfd;
	int send_to_net;
	int use_splice;
	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;
	struct fio_file *f = io_u->file;

	/*
	 * 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;
	}
		
	if (io_u->ddir == DDIR_SYNC)
		return 0;
	if (io_u->offset == f->last_completed_pos)
		return 0;

	/*
	 * If offset is different from last end position, it's a seek.
	 * As network io is purely sequential, we don't allow seeks.
	 */
	td_verror(td, EINVAL, "cannot seek");
	return 1;
}

/*
 * 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;
	unsigned int len = io_u->xfer_buflen;
	struct fio_file *f = io_u->file;
	int bytes = 0;

	while (len) {
		int ret = splice(f->fd, NULL, nd->pipes[1], NULL, len, 0);

		if (ret < 0) {
			if (!bytes)
				bytes = ret;

			break;
		} else if (!ret)
			break;

		bytes += ret;
		len -= ret;
	}

	return bytes;
}

/*
 * 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;
	struct fio_file *f = io_u->file;
	int bytes = 0;

	while (len) {
		int ret = splice(nd->pipes[0], NULL, f->fd, NULL, len, 0);

		if (ret < 0) {
			if (!bytes)
				bytes = ret;
			
			break;
		} else if (!ret)
			break;

		bytes += ret;
		len -= 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;
	struct iovec iov = {
		.iov_base = io_u->xfer_buf,
		.iov_len = len,
	};
	int bytes = 0;

	while (iov.iov_len) {
		int ret = vmsplice(nd->pipes[0], &iov, 1, 0);

		if (ret < 0) {
			if (!bytes)
				bytes = ret;
			break;
		} else if (!ret)
			break;

		iov.iov_len -= ret;
		bytes += ret;
		if (iov.iov_len)
			iov.iov_base += ret;
	}

	return bytes;
}

/*
 * 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;
	struct iovec iov = {
		.iov_base = io_u->xfer_buf,
		.iov_len = io_u->xfer_buflen,
	};
	unsigned int bytes = 0;

	while (iov.iov_len) {
		int ret = vmsplice(nd->pipes[1], &iov, 1, 0);

		if (ret < 0)
			return -1;
		else if (!ret)
			return bytes;

		iov.iov_len -= ret;
		bytes += ret;
		if (iov.iov_len)
			iov.iov_base += ret;
	}

	return bytes;
}

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

	return vmsplice_io_u_out(td, io_u, ret);
}

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

	return splice_out(td, io_u, ret);
}

static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
{
	int flags = 0;

	/*
	 * if we are going to write more, set MSG_MORE
	 */
	if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < td->o.size)
		flags = MSG_MORE;

	return send(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
}

static int fio_netio_recv(struct io_u *io_u)
{
	int flags = MSG_WAITALL;

	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;

	if (io_u->ddir == DDIR_WRITE) {
		if (nd->use_splice)
			ret = fio_netio_splice_out(td, io_u);
		else
			ret = fio_netio_send(td, io_u);
	} else if (io_u->ddir == DDIR_READ) {
		if (nd->use_splice)
			ret = fio_netio_splice_in(td, io_u);
		else
			ret = fio_netio_recv(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
			io_u->error = errno;
	}

	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;

	f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (f->fd < 0) {
		td_verror(td, errno, "socket");
		return 1;
	}

	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;

	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;

	fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	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 (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 short port;
	char host[64], buf[128];
	char *sep;
	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) {
		log_err("fio: bad network host/port <<%s>>\n", td->o.filename);
		return 1;
	}

	*sep = '\0';
	sep++;
	strcpy(host, buf);
	port = atoi(sep);

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

static void fio_netio_cleanup(struct thread_data *td)
{
	struct netio_data *nd = td->io_ops->data;

	if (nd) {
		free(nd);
		td->io_ops->data = NULL;
	}
}

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;
		td->io_ops->data = nd;
	}

	return 0;
}

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

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

static void fio_init fio_netio_register(void)
{
	register_ioengine(&ioengine_rw);
	register_ioengine(&ioengine_splice);
}

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