[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
	1	/*
	2	* Transfer data over the net.
	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>
	12	#include <sys/poll.h>
	13
	14	#include "../fio.h"
	15	#include "../os.h"
	16
	17	struct netio_data {
	18	int listenfd;
	19	int send_to_net;
	20	char host[64];
	21	struct sockaddr_in addr;
	22	};
	23
	24	static int fio_netio_prep(struct thread_data td, struct io_u io_u)
	25	{
	26	struct netio_data *nd = td->io_ops->data;
	27	struct fio_file *f = io_u->file;
	28
	29	/*
	30	* Make sure we don't see spurious reads to a receiver, and vice versa
	31	*/
	32	if ((nd->send_to_net && io_u->ddir == DDIR_READ) \|\|
	33	(!nd->send_to_net && io_u->ddir == DDIR_WRITE)) {
	34	td_verror(td, EINVAL, "bad direction");
	35	return 1;
	36	}
	37
	38	if (io_u->ddir == DDIR_SYNC)
	39	return 0;
	40	if (io_u->offset == f->last_completed_pos)
	41	return 0;
	42
	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	*/
	47	td_verror(td, EINVAL, "cannot seek");
	48	return 1;
	49	}
	50
	51	static int fio_netio_queue(struct thread_data td, struct io_u io_u)
	52	{
	53	struct fio_file *f = io_u->file;
	54	int ret, flags = 0;
	55
	56	if (io_u->ddir == DDIR_WRITE) {
	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;
	63
	64	ret = send(f->fd, io_u->xfer_buf, io_u->xfer_buflen, flags);
	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
	69	ret = 0; /* must be a SYNC */
	70
	71	if (ret != (int) io_u->xfer_buflen) {
	72	if (ret >= 0) {
	73	io_u->resid = io_u->xfer_buflen - ret;
	74	io_u->error = 0;
	75	return FIO_Q_COMPLETED;
	76	} else
	77	io_u->error = errno;
	78	}
	79
	80	if (io_u->error)
	81	td_verror(td, io_u->error, "xfer");
	82
	83	return FIO_Q_COMPLETED;
	84	}
	85
	86	static int fio_netio_connect(struct thread_data td, struct fio_file f)
	87	{
	88	struct netio_data *nd = td->io_ops->data;
	89
	90	f->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	91	if (f->fd < 0) {
	92	td_verror(td, errno, "socket");
	93	return 1;
	94	}
	95
	96	if (connect(f->fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
	97	td_verror(td, errno, "connect");
	98	return 1;
	99	}
	100
	101	return 0;
	102	}
	103
	104	static int fio_netio_accept(struct thread_data td, struct fio_file f)
	105	{
	106	struct netio_data *nd = td->io_ops->data;
	107	socklen_t socklen = sizeof(nd->addr);
	108	struct pollfd pfd;
	109	int ret;
	110
	111	fprintf(f_out, "fio: waiting for connection\n");
	112
	113	/*
	114	* Accept loop. poll for incoming events, accept them. Repeat until we
	115	* have all connections.
	116	*/
	117	while (!td->terminate) {
	118	pfd.fd = nd->listenfd;
	119	pfd.events = POLLIN;
	120
	121	ret = poll(&pfd, 1, -1);
	122	if (ret < 0) {
	123	if (errno == EINTR)
	124	continue;
	125
	126	td_verror(td, errno, "poll");
	127	break;
	128	} else if (!ret)
	129	continue;
	130
	131	/*
	132	* should be impossible
	133	*/
	134	if (!(pfd.revents & POLLIN))
	135	continue;
	136
	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	}
	144
	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;
	172	}
	173
	174	memcpy(&nd->addr.sin_addr, hent->h_addr, 4);
	175	}
	176
	177	return 0;
	178	}
	179
	180	static int fio_netio_setup_listen(struct thread_data *td, short port)
	181	{
	182	struct netio_data *nd = td->io_ops->data;
	183	int fd, opt;
	184
	185	fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	186	if (fd < 0) {
	187	td_verror(td, errno, "socket");
	188	return 1;
	189	}
	190
	191	opt = 1;
	192	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
	193	td_verror(td, errno, "setsockopt");
	194	return 1;
	195	}
	196	#ifdef SO_REUSEPORT
	197	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
	198	td_verror(td, errno, "setsockopt");
	199	return 1;
	200	}
	201	#endif
	202
	203	nd->addr.sin_family = AF_INET;
	204	nd->addr.sin_addr.s_addr = htonl(INADDR_ANY);
	205	nd->addr.sin_port = htons(port);
	206
	207	if (bind(fd, (struct sockaddr *) &nd->addr, sizeof(nd->addr)) < 0) {
	208	td_verror(td, errno, "bind");
	209	return 1;
	210	}
	211	if (listen(fd, 1) < 0) {
	212	td_verror(td, errno, "listen");
	213	return 1;
	214	}
	215
	216	nd->listenfd = fd;
	217	return 0;
	218	}
	219
	220	static int fio_netio_init(struct thread_data *td)
	221	{
	222	struct netio_data *nd = td->io_ops->data;
	223	unsigned short port;
	224	struct fio_file *f;
	225	char host[64], buf[128];
	226	char *sep;
	227	int ret, i;
	228
	229	if (!td->total_file_size) {
	230	log_err("fio: need size= set\n");
	231	return 1;
	232	}
	233
	234	if (td_rw(td)) {
	235	log_err("fio: network connections must be read OR write\n");
	236	return 1;
	237	}
	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);
	250	port = atoi(sep);
	251
	252	if (td_read(td)) {
	253	nd->send_to_net = 0;
	254	ret = fio_netio_setup_listen(td, port);
	255	} else {
	256	nd->send_to_net = 1;
	257	ret = fio_netio_setup_connect(td, host, port);
	258	}
	259
	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;
	269	}
	270
	271	return 0;
	272	}
	273
	274	static void fio_netio_cleanup(struct thread_data *td)
	275	{
	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;
	291	return 0;
	292	}
	293
	294	static struct ioengine_ops ioengine = {
	295	.name = "net",
	296	.version = FIO_IOOPS_VERSION,
	297	.prep = fio_netio_prep,
	298	.queue = fio_netio_queue,
	299	.setup = fio_netio_setup,
	300	.init = fio_netio_init,
	301	.cleanup = fio_netio_cleanup,
	302	.open_file = fio_netio_open_file,
	303	.close_file = generic_close_file,
	304	.flags = FIO_SYNCIO \| FIO_DISKLESSIO,
	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	}