[fio.git] / engines / net.c

/*
 * Transfer data over the net.
 */
#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];
	char *sep;
	int ret, i;

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