[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"
#include "../os.h"

struct netio_data {
	int listenfd;
	int send_to_net;
	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;
}

static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
{
	struct fio_file *f = io_u->file;
	int ret, flags = 0;

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

		ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
	} else if (io_u->ddir == DDIR_READ) {
		flags = MSG_WAITALL;
		ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
	} 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;

	fprintf(f_out, "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;
	struct fio_file *f;
	char host[64], buf[128];
	unsigned int i;
	char *sep;
	int ret;

	if (!td->total_file_size) {
		log_err("fio: need size= set\n");
		return 1;
	}

	if (td_rw(td)) {
		log_err("fio: network connections must be read OR write\n");
		return 1;
	}

	strcpy(buf, td->filename);

	sep = strchr(buf, '/');
	if (!sep) {
		log_err("fio: bad network host/port <<%s>>\n", td->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);
	}

	if (ret)
		return ret;

	td->io_size = td->total_file_size;
	td->total_io_size = td->io_size;

	for_each_file(td, f, i) {
		f->file_size = td->total_file_size / td->nr_files;
		f->real_file_size = f->file_size;
	}

	return 0;
}

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 = malloc(sizeof(*nd));

	memset(nd, 0, sizeof(*nd));
	nd->listenfd = -1;
	td->io_ops->data = nd;
	return 0;
}

static struct ioengine_ops ioengine = {
	.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 void fio_init fio_netio_register(void)
{
	register_ioengine(&ioengine);
}

static void fio_exit fio_netio_unregister(void)
{
	unregister_ioengine(&ioengine);
}
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
	17	#include "../fio.h"
	18	#include "../os.h"
	19
b5af8293 JA	20	struct netio_data {
	21	int listenfd;
	22	int send_to_net;
	23	char host[64];
	24	struct sockaddr_in addr;
	25	};
ed92ac0c JA	26
	27	static int fio_netio_prep(struct thread_data td, struct io_u io_u)
	28	{
b5af8293	29	struct netio_data *nd = td->io_ops->data;
ed92ac0c JA	30	struct fio_file *f = io_u->file;
ed92ac0c JA	31
7a6499da JA	32	/*
	33	* Make sure we don't see spurious reads to a receiver, and vice versa
	34	*/
b5af8293 JA	35	if ((nd->send_to_net && io_u->ddir == DDIR_READ) \|\|
b5af8293 JA	36	(!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
e1161c32	37	td_verror(td, EINVAL, "bad direction");
7a6499da	38	return 1;
ed92ac0c	39	}
7a6499da	40
ed92ac0c JA	41	if (io_u->ddir == DDIR_SYNC)
	42	return 0;
	43	if (io_u->offset == f->last_completed_pos)
	44	return 0;
	45
e01547d2 JA	46	/*
	47	* If offset is different from last end position, it's a seek.
	48	* As network io is purely sequential, we don't allow seeks.
	49	*/
e1161c32	50	td_verror(td, EINVAL, "cannot seek");
ed92ac0c JA	51	return 1;
	52	}
	53
	54	static int fio_netio_queue(struct thread_data td, struct io_u io_u)
	55	{
ed92ac0c	56	struct fio_file *f = io_u->file;
d4f12dd0	57	int ret, flags = 0;
7a6499da JA	58
7a6499da JA	59	if (io_u->ddir == DDIR_WRITE) {
7a6499da JA	60	/*
	61	* if we are going to write more, set MSG_MORE
	62	*/
	63	if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
	64	td->io_size)
	65	flags = MSG_MORE;
ed92ac0c	66
7a6499da	67	ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
d4f12dd0 JA	68	} else if (io_u->ddir == DDIR_READ) {
	69	flags = MSG_WAITALL;
	70	ret = recv(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
	71	} else
7a6499da	72	ret = 0; /* must be a SYNC */
ed92ac0c	73
cec6b55d	74	if (ret != (int) io_u->xfer_buflen) {
22819ec2	75	if (ret >= 0) {
cec6b55d JA	76	io_u->resid = io_u->xfer_buflen - ret;
cec6b55d JA	77	io_u->error = 0;
36167d82	78	return FIO_Q_COMPLETED;
ed92ac0c JA	79	} else
	80	io_u->error = errno;
	81	}
	82
36167d82	83	if (io_u->error)
e1161c32	84	td_verror(td, io_u->error, "xfer");
ed92ac0c	85
36167d82	86	return FIO_Q_COMPLETED;
ed92ac0c JA	87	}
ed92ac0c JA	88
b5af8293	89	static int fio_netio_connect(struct thread_data td, struct fio_file f)
ed92ac0c	90	{
b5af8293	91	struct netio_data *nd = td->io_ops->data;
ed92ac0c	92
b5af8293 JA	93	f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	94	if (f->fd < 0) {
	95	td_verror(td, errno, "socket");
	96	return 1;
ed92ac0c JA	97	}
ed92ac0c JA	98
b5af8293 JA	99	if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
	100	td_verror(td, errno, "connect");
	101	return 1;
ed92ac0c JA	102	}
	103
	104	return 0;
ed92ac0c JA	105	}
ed92ac0c JA	106
b5af8293	107	static int fio_netio_accept(struct thread_data td, struct fio_file f)
5fdd124a	108	{
b5af8293 JA	109	struct netio_data *nd = td->io_ops->data;
b5af8293 JA	110	socklen_t socklen = sizeof(nd->addr);
5fdd124a	111	struct pollfd pfd;
b5af8293	112	int ret;
5fdd124a	113
b5af8293	114	fprintf(f_out, "fio: waiting for connection\n");
5fdd124a JA	115
	116	/*
	117	* Accept loop. poll for incoming events, accept them. Repeat until we
	118	* have all connections.
	119	*/
b5af8293 JA	120	while (!td->terminate) {
b5af8293 JA	121	pfd.fd = nd->listenfd;
5fdd124a JA	122	pfd.events = POLLIN;
	123
	124	ret = poll(&pfd, 1, -1);
	125	if (ret < 0) {
	126	if (errno == EINTR)
	127	continue;
	128
e1161c32	129	td_verror(td, errno, "poll");
5fdd124a JA	130	break;
	131	} else if (!ret)
	132	continue;
	133
0c09442b JA	134	/*
	135	* should be impossible
	136	*/
	137	if (!(pfd.revents & POLLIN))
	138	continue;
	139
b5af8293 JA	140	f->fd = accept(nd->listenfd, (struct sockaddr *) &nd->addr, &socklen);
	141	if (f->fd < 0) {
	142	td_verror(td, errno, "accept");
	143	return 1;
	144	}
	145	break;
	146	}
5fdd124a	147
b5af8293 JA	148	return 0;
	149	}
	150
	151
	152	static int fio_netio_open_file(struct thread_data td, struct fio_file f)
	153	{
	154	if (td_read(td))
	155	return fio_netio_accept(td, f);
	156	else
	157	return fio_netio_connect(td, f);
	158	}
	159
	160	static int fio_netio_setup_connect(struct thread_data td, const char host,
	161	unsigned short port)
	162	{
	163	struct netio_data *nd = td->io_ops->data;
	164
	165	nd->addr.sin_family = AF_INET;
	166	nd->addr.sin_port = htons(port);
	167
	168	if (inet_aton(host, &nd->addr.sin_addr) != 1) {
	169	struct hostent *hent;
	170
	171	hent = gethostbyname(host);
	172	if (!hent) {
	173	td_verror(td, errno, "gethostbyname");
	174	return 1;
5fdd124a	175	}
b5af8293 JA	176
b5af8293 JA	177	memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
5fdd124a JA	178	}
	179
	180	return 0;
	181	}
	182
b5af8293	183	static int fio_netio_setup_listen(struct thread_data *td, short port)
ed92ac0c	184	{
b5af8293	185	struct netio_data *nd = td->io_ops->data;
5fdd124a	186	int fd, opt;
ed92ac0c	187
6bedbfaf	188	fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
ed92ac0c	189	if (fd < 0) {
e1161c32	190	td_verror(td, errno, "socket");
ed92ac0c JA	191	return 1;
	192	}
	193
	194	opt = 1;
	195	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
e1161c32	196	td_verror(td, errno, "setsockopt");
ed92ac0c JA	197	return 1;
ed92ac0c JA	198	}
6bedbfaf JA	199	#ifdef SO_REUSEPORT
6bedbfaf JA	200	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
e1161c32	201	td_verror(td, errno, "setsockopt");
6bedbfaf JA	202	return 1;
	203	}
	204	#endif
ed92ac0c	205
b5af8293 JA	206	nd->addr.sin_family = AF_INET;
	207	nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
	208	nd->addr.sin_port = htons(port);
ed92ac0c	209
b5af8293	210	if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
e1161c32	211	td_verror(td, errno, "bind");
ed92ac0c JA	212	return 1;
	213	}
	214	if (listen(fd, 1) < 0) {
e1161c32	215	td_verror(td, errno, "listen");
ed92ac0c JA	216	return 1;
	217	}
	218
b5af8293 JA	219	nd->listenfd = fd;
b5af8293 JA	220	return 0;
ed92ac0c JA	221	}
ed92ac0c JA	222
9bec88e1	223	static int fio_netio_init(struct thread_data *td)
ed92ac0c	224	{
b5af8293	225	struct netio_data *nd = td->io_ops->data;
e01547d2	226	unsigned short port;
2fc26981	227	struct fio_file *f;
b5af8293	228	char host[64], buf[128];
af52b345	229	unsigned int i;
ed92ac0c	230	char *sep;
af52b345	231	int ret;
ed92ac0c	232
7a6499da JA	233	if (!td->total_file_size) {
	234	log_err("fio: need size= set\n");
	235	return 1;
	236	}
	237
413dd459	238	if (td_rw(td)) {
ed92ac0c JA	239	log_err("fio: network connections must be read OR write\n");
	240	return 1;
	241	}
ed92ac0c JA	242
	243	strcpy(buf, td->filename);
	244
9f9214f2	245	sep = strchr(buf, '/');
ed92ac0c	246	if (!sep) {
9f9214f2	247	log_err("fio: bad network host/port <<%s>>\n", td->filename);
ed92ac0c JA	248	return 1;
	249	}
	250
	251	*sep = '\0';
	252	sep++;
	253	strcpy(host, buf);
e01547d2	254	port = atoi(sep);
ed92ac0c	255
413dd459	256	if (td_read(td)) {
b5af8293	257	nd->send_to_net = 0;
ed92ac0c JA	258	ret = fio_netio_setup_listen(td, port);
ed92ac0c JA	259	} else {
b5af8293	260	nd->send_to_net = 1;
ed92ac0c JA	261	ret = fio_netio_setup_connect(td, host, port);
	262	}
	263
2fc26981 JA	264	if (ret)
	265	return ret;
	266
	267	td->io_size = td->total_file_size;
	268	td->total_io_size = td->io_size;
	269
	270	for_each_file(td, f, i) {
	271	f->file_size = td->total_file_size / td->nr_files;
	272	f->real_file_size = f->file_size;
ed92ac0c JA	273	}
ed92ac0c JA	274
2fc26981	275	return 0;
ed92ac0c JA	276	}
ed92ac0c JA	277
b5af8293	278	static void fio_netio_cleanup(struct thread_data *td)
9bec88e1	279	{
b5af8293 JA	280	struct netio_data *nd = td->io_ops->data;
	281
	282	if (nd) {
	283	free(nd);
	284	td->io_ops->data = NULL;
	285	}
	286	}
	287
	288	static int fio_netio_setup(struct thread_data *td)
	289	{
	290	struct netio_data nd = malloc(sizeof(nd));
	291
	292	memset(nd, 0, sizeof(*nd));
	293	nd->listenfd = -1;
	294	td->io_ops->data = nd;
9bec88e1 JA	295	return 0;
	296	}
	297
ed92ac0c JA	298	static struct ioengine_ops ioengine = {
	299	.name = "net",
	300	.version = FIO_IOOPS_VERSION,
ed92ac0c JA	301	.prep = fio_netio_prep,
ed92ac0c JA	302	.queue = fio_netio_queue,
ed92ac0c	303	.setup = fio_netio_setup,
9bec88e1	304	.init = fio_netio_init,
b5af8293 JA	305	.cleanup = fio_netio_cleanup,
	306	.open_file = fio_netio_open_file,
	307	.close_file = generic_close_file,
	308	.flags = FIO_SYNCIO \| FIO_DISKLESSIO,
ed92ac0c JA	309	};
	310
	311	static void fio_init fio_netio_register(void)
	312	{
	313	register_ioengine(&ioengine);
	314	}
	315
	316	static void fio_exit fio_netio_unregister(void)
	317	{
	318	unregister_ioengine(&ioengine);
	319	}