[fio.git] / engines / net.c

/*
 * Transfer data over the net. Pretty basic setup, will only support
 * 1 file per thread/job.
 */
#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;

	if (nd->send_to_net) {
		if (io_u->ddir == DDIR_READ) {
			td_verror(td, EINVAL);
			return 1;
		}
	} else {
		if (io_u->ddir == DDIR_WRITE) {
			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 = 0;

	if (io_u->ddir == DDIR_WRITE)
		ret = write(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
	else if (io_u->ddir == DDIR_READ)
		ret = read(f->fd, io_u->xfer_buf, io_u->xfer_buflen);

	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 = gethostbyname(host);

		if (!hent) {
			td_vmsg(td, errno, "gethostbyname");
			return 1;
		}

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

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

		if (connect(f->fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
			td_vmsg(td, errno, "connect");
			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, 0);
	if (fd < 0) {
		td_vmsg(td, errno, "socket");
		return 1;
	}

	opt = 1;
	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
		td_vmsg(td, errno, "setsockopt");
		return 1;
	}

	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_vmsg(td, errno, "bind");
		return 1;
	}
	if (listen(fd, 1) < 0) {
		td_vmsg(td, errno, "listen");
		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_vmsg(td, errno, "accept");
			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;

	/*
	 * 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 == 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 JA	1	/*
	2	* Transfer data over the net. Pretty basic setup, will only support
	3	* 1 file per thread/job.
	4	*/
	5	#include <stdio.h>
	6	#include <stdlib.h>
	7	#include <unistd.h>
	8	#include <errno.h>
	9	#include <assert.h>
	10	#include <netinet/in.h>
	11	#include <arpa/inet.h>
	12	#include <netdb.h>
	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
	51	if (nd->send_to_net) {
	52	if (io_u->ddir == DDIR_READ) {
	53	td_verror(td, EINVAL);
	54	return 1;
	55	}
	56	} else {
	57	if (io_u->ddir == DDIR_WRITE) {
	58	td_verror(td, EINVAL);
	59	return 1;
	60	}
	61	}
	62
	63	if (io_u->ddir == DDIR_SYNC)
	64	return 0;
65	if (io_u->offset == f->last_completed_pos)
66	return 0;
67
e01547d2 JA	68	/*
	69	* If offset is different from last end position, it's a seek.
	70	* As network io is purely sequential, we don't allow seeks.
	71	*/
ed92ac0c JA	72	td_verror(td, EINVAL);
	73	return 1;
	74	}
	75
	76	static int fio_netio_queue(struct thread_data td, struct io_u io_u)
	77	{
	78	struct net_data *nd = td->io_ops->data;
	79	struct fio_file *f = io_u->file;
cec6b55d	80	int ret = 0;
ed92ac0c JA	81
ed92ac0c JA	82	if (io_u->ddir == DDIR_WRITE)
cec6b55d	83	ret = write(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
ed92ac0c	84	else if (io_u->ddir == DDIR_READ)
cec6b55d	85	ret = read(f->fd, io_u->xfer_buf, io_u->xfer_buflen);
ed92ac0c	86
cec6b55d	87	if (ret != (int) io_u->xfer_buflen) {
ed92ac0c	88	if (ret > 0) {
cec6b55d JA	89	io_u->resid = io_u->xfer_buflen - ret;
	90	io_u->error = 0;
	91	return ret;
ed92ac0c JA	92	} else
	93	io_u->error = errno;
	94	}
	95
	96	if (!io_u->error)
	97	nd->last_io_u = io_u;
	98
	99	return io_u->error;
	100	}
	101
	102	static int fio_netio_setup_connect(struct thread_data td, const char host,
e01547d2	103	unsigned short port)
ed92ac0c JA	104	{
	105	struct sockaddr_in addr;
	106	struct fio_file *f;
2fc26981	107	int i;
ed92ac0c JA	108
	109	memset(&addr, 0, sizeof(addr));
	110	addr.sin_family = AF_INET;
e01547d2	111	addr.sin_port = htons(port);
ed92ac0c JA	112
	113	if (inet_aton(host, &addr.sin_addr) != 1) {
	114	struct hostent *hent = gethostbyname(host);
	115
	116	if (!hent) {
	117	td_vmsg(td, errno, "gethostbyname");
	118	return 1;
	119	}
	120
	121	memcpy(&addr.sin_addr, hent->h_addr, 4);
	122	}
	123
2fc26981 JA	124	for_each_file(td, f, i) {
	125	f->fd = socket(AF_INET, SOCK_STREAM, 0);
	126	if (f->fd < 0) {
	127	td_vmsg(td, errno, "socket");
	128	return 1;
	129	}
ed92ac0c	130
2fc26981 JA	131	if (connect(f->fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
	132	td_vmsg(td, errno, "connect");
	133	return 1;
	134	}
ed92ac0c JA	135	}
	136
	137	return 0;
	138
	139	}
	140
e01547d2	141	static int fio_netio_setup_listen(struct thread_data *td, unsigned short port)
ed92ac0c JA	142	{
	143	struct sockaddr_in addr;
	144	socklen_t socklen;
2fc26981 JA	145	struct fio_file *f;
2fc26981 JA	146	int fd, opt, i;
ed92ac0c JA	147
	148	fd = socket(AF_INET, SOCK_STREAM, 0);
	149	if (fd < 0) {
	150	td_vmsg(td, errno, "socket");
	151	return 1;
	152	}
	153
	154	opt = 1;
	155	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
	156	td_vmsg(td, errno, "setsockopt");
	157	return 1;
	158	}
	159
	160	memset(&addr, 0, sizeof(addr));
	161	addr.sin_family = AF_INET;
	162	addr.sin_addr.s_addr = htonl(INADDR_ANY);
e01547d2	163	addr.sin_port = htons(port);
ed92ac0c JA	164
	165	if (bind(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) {
	166	td_vmsg(td, errno, "bind");
	167	return 1;
	168	}
	169	if (listen(fd, 1) < 0) {
	170	td_vmsg(td, errno, "listen");
	171	return 1;
	172	}
	173
2fc26981 JA	174	fprintf(f_out, "fio: waiting for %u connections\n", td->nr_files);
2fc26981 JA	175
ed92ac0c	176	socklen = sizeof(addr);
2fc26981 JA	177	for_each_file(td, f, i) {
	178	f->fd = accept(fd, (struct sockaddr *) &addr, &socklen);
	179	if (f->fd < 0) {
	180	td_vmsg(td, errno, "accept");
	181	return 1;
	182	}
ed92ac0c JA	183	}
	184
	185	return 0;
	186	}
	187
	188	static int fio_netio_setup(struct thread_data *td)
	189	{
e01547d2	190	char host[64], buf[128];
ed92ac0c	191	struct net_data *nd;
e01547d2	192	unsigned short port;
2fc26981	193	struct fio_file *f;
ed92ac0c	194	char *sep;
2fc26981	195	int ret, i;
ed92ac0c JA	196
	197	/*
	198	* work around for late init call
	199	*/
	200	if (td->io_ops->init(td))
	201	return 1;
	202
	203	nd = td->io_ops->data;
	204
	205	if (td->iomix) {
	206	log_err("fio: network connections must be read OR write\n");
	207	return 1;
	208	}
ed92ac0c JA	209
	210	strcpy(buf, td->filename);
	211
	212	sep = strchr(buf, ':');
	213	if (!sep) {
	214	log_err("fio: bad network host:port <<%s>>\n", td->filename);
	215	return 1;
	216	}
	217
	218	*sep = '\0';
	219	sep++;
	220	strcpy(host, buf);
e01547d2	221	port = atoi(sep);
ed92ac0c JA	222
	223	if (td->ddir == READ) {
	224	nd->send_to_net = 0;
	225	ret = fio_netio_setup_listen(td, port);
	226	} else {
	227	nd->send_to_net = 1;
	228	ret = fio_netio_setup_connect(td, host, port);
	229	}
	230
2fc26981 JA	231	if (ret)
	232	return ret;
	233
	234	td->io_size = td->total_file_size;
	235	td->total_io_size = td->io_size;
	236
	237	for_each_file(td, f, i) {
	238	f->file_size = td->total_file_size / td->nr_files;
	239	f->real_file_size = f->file_size;
ed92ac0c JA	240	}
ed92ac0c JA	241
2fc26981	242	return 0;
ed92ac0c JA	243	}
	244
	245	static void fio_netio_cleanup(struct thread_data *td)
	246	{
	247	if (td->io_ops->data) {
	248	free(td->io_ops->data);
	249	td->io_ops->data = NULL;
	250	}
	251	}
	252
	253	static int fio_netio_init(struct thread_data *td)
	254	{
	255	struct net_data *nd;
	256
e01547d2 JA	257	/*
	258	* Hack to work-around the ->setup() function calling init on its
	259	* own, since it needs ->io_ops->data to be set up.
	260	*/
ed92ac0c JA	261	if (td->io_ops->data)
	262	return 0;
	263
	264	nd = malloc(sizeof(*nd));
	265	nd->last_io_u = NULL;
	266	td->io_ops->data = nd;
	267	return 0;
	268	}
	269
	270	static struct ioengine_ops ioengine = {
	271	.name = "net",
	272	.version = FIO_IOOPS_VERSION,
	273	.init = fio_netio_init,
	274	.prep = fio_netio_prep,
	275	.queue = fio_netio_queue,
	276	.getevents = fio_netio_getevents,
	277	.event = fio_netio_event,
	278	.cleanup = fio_netio_cleanup,
	279	.setup = fio_netio_setup,
	280	.flags = FIO_SYNCIO \| FIO_NETIO,
	281	};
	282
	283	static void fio_init fio_netio_register(void)
	284	{
	285	register_ioengine(&ioengine);
	286	}
	287
	288	static void fio_exit fio_netio_unregister(void)
	289	{
	290	unregister_ioengine(&ioengine);
	291	}