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

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

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

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;

	fio_ro_check(td, io_u);

	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) {
		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);
		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;
		nd->pipes[0] = nd->pipes[1] = -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 | FIO_UNIDIR |
			  FIO_SIGQUIT,
};

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

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