[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 net_data {
	int send_to_net;
	struct io_u *last_io_u;
};

static int fio_netio_getevents(struct thread_data *td, int fio_unused min,
				int max, struct timespec fio_unused *t)
{
	assert(max <= 1);

	/*
	 * we can only have one finished io_u for sync io, since the depth
	 * is always 1
	 */
	if (list_empty(&td->io_u_busylist))
		return 0;

	return 1;
}

static struct io_u *fio_netio_event(struct thread_data *td, int event)
{
	struct net_data *nd = td->io_ops->data;

	assert(event == 0);

	return nd->last_io_u;
}

static int fio_netio_prep(struct thread_data *td, struct io_u *io_u)
{
	struct net_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)) {
		printf("boo!\n");
		td_verror(td, EINVAL);
		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);
	return 1;
}

static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
{
	struct net_data *nd = td->io_ops->data;
	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 ret;
		} else
			io_u->error = errno;
	}

	if (!io_u->error)
		nd->last_io_u = io_u;
	else
		td_verror(td, io_u->error);

	return io_u->error;
}

static int fio_netio_setup_connect(struct thread_data *td, const char *host,
				   unsigned short port)
{
	struct sockaddr_in addr;
	struct fio_file *f;
	int i;

	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);

	if (inet_aton(host, &addr.sin_addr) != 1) {
		struct hostent *hent;

		hent = gethostbyname(host);
		if (!hent) {
			td_verror(td, errno);
			return 1;
		}

		memcpy(&addr.sin_addr, hent->h_addr, 4);
	}

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

		if (connect(f->fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
			td_verror(td, errno);
			return 1;
		}
	}

	return 0;

}

static int fio_netio_accept_connections(struct thread_data *td, int fd,
					struct sockaddr_in *addr)
{
	socklen_t socklen = sizeof(*addr);
	unsigned int accepts = 0;
	struct pollfd pfd;

	fprintf(f_out, "fio: waiting for %u connections\n", td->nr_files);

	/*
	 * Accept loop. poll for incoming events, accept them. Repeat until we
	 * have all connections.
	 */
	while (!td->terminate && accepts < td->nr_files) {
		struct fio_file *f;
		int ret, i;

		pfd.fd = fd;
		pfd.events = POLLIN;

		ret = poll(&pfd, 1, -1);
		if (ret < 0) {
			if (errno == EINTR)
				continue;

			td_verror(td, errno);
			break;
		} else if (!ret)
			continue;

		/*
		 * should be impossible
		 */
		if (!(pfd.revents & POLLIN))
			continue;

		for_each_file(td, f, i) {
			if (f->fd != -1)
				continue;

			f->fd = accept(fd, (struct sockaddr *) addr, &socklen);
			if (f->fd < 0) {
				td_verror(td, errno);
				return 1;
			}
			accepts++;
			break;
		}
	}

	return 0;
}

static int fio_netio_setup_listen(struct thread_data *td, unsigned short port)
{
	struct sockaddr_in addr;
	int fd, opt;

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

	opt = 1;
	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
		td_verror(td, errno);
		return 1;
	}
#ifdef SO_REUSEPORT
	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
		td_verror(td, errno);
		return 1;
	}
#endif

	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY);
	addr.sin_port = htons(port);

	if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
		td_verror(td, errno);
		return 1;
	}
	if (listen(fd, 1) < 0) {
		td_verror(td, errno);
		return 1;
	}

	return fio_netio_accept_connections(td, fd, &addr);
}

static int fio_netio_setup(struct thread_data *td)
{
	char host[64], buf[128];
	struct net_data *nd;
	unsigned short port;
	struct fio_file *f;
	char *sep;
	int ret, i;

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

	/*
	 * work around for late init call
	 */
	if (td->io_ops->init(td))
		return 1;

	nd = td->io_ops->data;

	if (td->iomix) {
		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->ddir == DDIR_READ) {
		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)
{
	if (td->io_ops->data) {
		free(td->io_ops->data);
		td->io_ops->data = NULL;
	}
}

static int fio_netio_init(struct thread_data *td)
{
	struct net_data *nd;

	/*
	 * Hack to work-around the ->setup() function calling init on its
	 * own, since it needs ->io_ops->data to be set up.
	 */
	if (td->io_ops->data)
		return 0;

	nd  = malloc(sizeof(*nd));
	nd->last_io_u = NULL;
	td->io_ops->data = nd;
	return 0;
}

static struct ioengine_ops ioengine = {
	.name		= "net",
	.version	= FIO_IOOPS_VERSION,
	.init		= fio_netio_init,
	.prep		= fio_netio_prep,
	.queue		= fio_netio_queue,
	.getevents	= fio_netio_getevents,
	.event		= fio_netio_event,
	.cleanup	= fio_netio_cleanup,
	.setup		= fio_netio_setup,
	.flags		= FIO_SYNCIO | FIO_NETIO,
};

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