[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 "../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;

	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_setup_listen(struct thread_data *td, unsigned short port)
{
	struct sockaddr_in addr;
	socklen_t socklen;
	struct fio_file *f;
	int fd, opt, i;

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

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

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

	return 0;
}

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