[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(nd->pipes[1], NULL, f->fd, NULL, len, 0);

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

			break;
		} else if (!ret)
			break;

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