[fio.git] / server.c

#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <unistd.h>
#include <limits.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/poll.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/mman.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>

#include "fio.h"
#include "server.h"
#include "crc/crc16.h"

int fio_net_port = 8765;

int exit_backend = 0;

static char *job_buf;
static unsigned int job_cur_len;
static unsigned int job_max_len;

static int server_fd;

int fio_send_data(int sk, const void *p, unsigned int len)
{
	do {
		int ret = send(sk, p, len, 0);

		if (ret > 0) {
			len -= ret;
			if (!len)
				break;
			p += ret;
			continue;
		} else if (!ret)
			break;
		else if (errno == EAGAIN || errno == EINTR)
			continue;
	} while (!exit_backend);

	if (!len)
		return 0;

	return 1;
}

int fio_recv_data(int sk, void *p, unsigned int len)
{
	do {
		int ret = recv(sk, p, len, MSG_WAITALL);

		if (ret > 0) {
			len -= ret;
			if (!len)
				break;
			p += ret;
			continue;
		} else if (!ret)
			break;
		else if (errno == EAGAIN || errno == EINTR)
			continue;
	} while (!exit_backend);

	if (!len)
		return 0;

	return -1;
}

static int verify_convert_cmd(struct fio_net_cmd *cmd)
{
	uint16_t crc;

	cmd->cmd_crc16 = le16_to_cpu(cmd->cmd_crc16);
	cmd->pdu_crc16 = le16_to_cpu(cmd->pdu_crc16);

	crc = crc16(cmd, FIO_NET_CMD_CRC_SZ);
	if (crc != cmd->cmd_crc16) {
		log_err("fio: server bad crc on command (got %x, wanted %x)\n",
				cmd->cmd_crc16, crc);
		return 1;
	}

	cmd->version	= le16_to_cpu(cmd->version);
	cmd->opcode	= le16_to_cpu(cmd->opcode);
	cmd->flags	= le32_to_cpu(cmd->flags);
	cmd->serial	= le64_to_cpu(cmd->serial);
	cmd->pdu_len	= le32_to_cpu(cmd->pdu_len);

	switch (cmd->version) {
	case FIO_SERVER_VER1:
		break;
	default:
		log_err("fio: bad server cmd version %d\n", cmd->version);
		return 1;
	}

	if (cmd->pdu_len > FIO_SERVER_MAX_PDU) {
		log_err("fio: command payload too large: %u\n", cmd->pdu_len);
		return 1;
	}

	return 0;
}

struct fio_net_cmd *fio_net_cmd_read(int sk)
{
	struct fio_net_cmd cmd, *ret = NULL;
	uint16_t crc;

	if (fio_recv_data(sk, &cmd, sizeof(cmd)))
		return NULL;

	/* We have a command, verify it and swap if need be */
	if (verify_convert_cmd(&cmd))
		return NULL;

	/* Command checks out, alloc real command and fill in */
	ret = malloc(sizeof(cmd) + cmd.pdu_len);
	memcpy(ret, &cmd, sizeof(cmd));

	if (!ret->pdu_len)
		return ret;

	/* There's payload, get it */
	if (fio_recv_data(sk, (void *) ret + sizeof(*ret), ret->pdu_len)) {
		free(ret);
		return NULL;
	}

	/* Verify payload crc */
	crc = crc16(ret->payload, ret->pdu_len);
	if (crc != ret->pdu_crc16) {
		log_err("fio: server bad crc on payload (got %x, wanted %x)\n",
				ret->pdu_crc16, crc);
		free(ret);
		return NULL;
	}

	return ret;
}

void fio_net_cmd_crc(struct fio_net_cmd *cmd)
{
	uint32_t pdu_len;

	cmd->cmd_crc16 = cpu_to_le16(crc16(cmd, FIO_NET_CMD_CRC_SZ));

	pdu_len = le32_to_cpu(cmd->pdu_len);
	if (pdu_len)
		cmd->pdu_crc16 = cpu_to_le16(crc16(cmd->payload, pdu_len));
}

static int send_simple_command(int sk, uint16_t opcode, uint64_t serial)
{
	struct fio_net_cmd cmd = {
		.version	= cpu_to_le16(FIO_SERVER_VER1),
		.opcode		= cpu_to_le16(opcode),
		.serial		= cpu_to_le64(serial),
	};

	fio_net_cmd_crc(&cmd);

	return fio_send_data(sk, &cmd, sizeof(cmd));
}

/*
 * Send an ack for this command
 */
static int ack_command(int sk, struct fio_net_cmd *cmd)
{
#if 0
	return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial);
#else
	return 0;
#endif
}

#if 0
static int nak_command(int sk, struct fio_net_cmd *cmd)
{
	return send_simple_command(sk, FIO_NET_CMD_NAK, cmd->serial);
}
#endif

static int send_quit_command(void)
{
	return send_simple_command(server_fd, FIO_NET_CMD_QUIT, 0);
}

static int handle_cur_job(struct fio_net_cmd *cmd, int done)
{
	unsigned int left = job_max_len - job_cur_len;
	int ret = 0;

	if (left < cmd->pdu_len) {
		job_buf = realloc(job_buf, job_max_len + 2 * cmd->pdu_len);
		job_max_len += 2 * cmd->pdu_len;
	}

	memcpy(job_buf + job_cur_len, cmd->payload, cmd->pdu_len);
	job_cur_len += cmd->pdu_len;

	if (done) {
		parse_jobs_ini(job_buf, 1, 0);
		ret = exec_run();
		send_quit_command();
		reset_fio_state();
		free(job_buf);
		job_buf = NULL;
		job_cur_len = job_max_len = 0;
	}

	return ret;
}

static int handle_command(struct fio_net_cmd *cmd)
{
	int ret;

	switch (cmd->opcode) {
	case FIO_NET_CMD_QUIT:
		exit_backend = 1;
		return 1;
	case FIO_NET_CMD_ACK:
		return 0;
	case FIO_NET_CMD_NAK:
		return 1;
	case FIO_NET_CMD_JOB:
		ret = handle_cur_job(cmd, 0);
		break;
	case FIO_NET_CMD_JOB_END:
		ret = handle_cur_job(cmd, 1);
		break;
	default:
		log_err("fio: unknown opcode: %d\n", cmd->opcode);
		ret = 1;
	}

	return ret;
}

static int handle_connection(int sk)
{
	struct fio_net_cmd *cmd = NULL;
	int ret = 0;

	/* read forever */
	while (!exit_backend) {
		cmd = fio_net_cmd_read(sk);
		if (!cmd) {
			ret = 1;
			break;
		}

		ret = ack_command(sk, cmd);
		if (ret)
			break;

		ret = handle_command(cmd);
		if (ret)
			break;

		free(cmd);
	}

	if (cmd)
		free(cmd);

	return ret;
}

static int accept_loop(int listen_sk)
{
	struct sockaddr addr;
	unsigned int len = sizeof(addr);
	struct pollfd pfd;
	int ret, sk, flags, exitval = 0;

	flags = fcntl(listen_sk, F_GETFL);
	flags |= O_NONBLOCK;
	fcntl(listen_sk, F_SETFL, flags);
again:
	pfd.fd = listen_sk;
	pfd.events = POLLIN;
	do {
		ret = poll(&pfd, 1, 100);
		if (ret < 0) {
			if (errno == EINTR)
				break;
			log_err("fio: poll: %s\n", strerror(errno));
			goto out;
		} else if (!ret)
			continue;

		if (pfd.revents & POLLIN)
			break;
	} while (!exit_backend);

	if (exit_backend)
		goto out;

	sk = accept(listen_sk, &addr, &len);
	if (sk < 0) {
		log_err("fio: accept: %s\n", strerror(errno));
		return -1;
	}

	server_fd = sk;

	exitval = handle_connection(sk);

	server_fd = -1;
	close(sk);

	if (!exit_backend)
		goto again;

out:
	return exitval;
}

int fio_server(void)
{
	struct sockaddr_in saddr_in;
	struct sockaddr addr;
	unsigned int len;
	int sk, opt, ret;

	sk = socket(AF_INET, SOCK_STREAM, 0);
	if (sk < 0) {
		log_err("fio: socket: %s\n", strerror(errno));
		return -1;
	}

	opt = 1;
	if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
		log_err("fio: setsockopt: %s\n", strerror(errno));
		return -1;
	}
#ifdef SO_REUSEPORT
	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
		log_err("fio: setsockopt: %s\n", strerror(errno));
		return 1;
	}
#endif

	saddr_in.sin_family = AF_INET;
	saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
	saddr_in.sin_port = htons(fio_net_port);

	if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) {
		log_err("fio: bind: %s\n", strerror(errno));
		return -1;
	}

	if (listen(sk, 1) < 0) {
		log_err("fio: listen: %s\n", strerror(errno));
		return -1;
	}

	len = sizeof(addr);
	if (getsockname(sk, &addr, &len) < 0) {
		log_err("fio: getsockname: %s\n", strerror(errno));
		return -1;
	}

	ret = accept_loop(sk);
	close(sk);
	return ret;
}

int fio_server_text_output(const char *buf, unsigned int len)
{
	struct fio_net_cmd *cmd;
	int size = sizeof(*cmd) + len;

	cmd = malloc(size);
	fio_init_net_cmd(cmd, FIO_NET_CMD_TEXT, buf, len);
	fio_net_cmd_crc(cmd);

	fio_send_data(server_fd, cmd, size);
	free(cmd);
	return size;
}

int fio_server_log(const char *format, ...)
{
	char buffer[1024];
	va_list args;
	size_t len;

	va_start(args, format);
	len = vsnprintf(buffer, sizeof(buffer), format, args);
	va_end(args);

	return fio_server_text_output(buffer, len);
}
Commit	Line	Data
50d16976 JA	1	#include <stdio.h>
50d16976 JA	2	#include <stdlib.h>
142575e6	3	#include <stdarg.h>
50d16976 JA	4	#include <unistd.h>
50d16976 JA	5	#include <limits.h>
50d16976 JA	6	#include <errno.h>
	7	#include <fcntl.h>
	8	#include <sys/poll.h>
50d16976 JA	9	#include <sys/types.h>
50d16976 JA	10	#include <sys/wait.h>
50d16976	11	#include <sys/mman.h>
50d16976 JA	12	#include <netinet/in.h>
	13	#include <arpa/inet.h>
	14	#include <netdb.h>
	15
	16	#include "fio.h"
132159a5	17	#include "server.h"
fcee5ff6	18	#include "crc/crc16.h"
50d16976	19
132159a5	20	int fio_net_port = 8765;
50d16976	21
009b1be4 JA	22	int exit_backend = 0;
009b1be4 JA	23
132159a5 JA	24	static char *job_buf;
	25	static unsigned int job_cur_len;
	26	static unsigned int job_max_len;
	27
37db14fe JA	28	static int server_fd;
37db14fe JA	29
132159a5 JA	30	int fio_send_data(int sk, const void *p, unsigned int len)
	31	{
	32	do {
	33	int ret = send(sk, p, len, 0);
	34
	35	if (ret > 0) {
	36	len -= ret;
	37	if (!len)
	38	break;
	39	p += ret;
	40	continue;
	41	} else if (!ret)
	42	break;
	43	else if (errno == EAGAIN \|\| errno == EINTR)
	44	continue;
	45	} while (!exit_backend);
	46
	47	if (!len)
	48	return 0;
	49
	50	return 1;
	51	}
	52
	53	int fio_recv_data(int sk, void *p, unsigned int len)
	54	{
	55	do {
	56	int ret = recv(sk, p, len, MSG_WAITALL);
	57
	58	if (ret > 0) {
	59	len -= ret;
	60	if (!len)
	61	break;
	62	p += ret;
	63	continue;
	64	} else if (!ret)
	65	break;
	66	else if (errno == EAGAIN \|\| errno == EINTR)
	67	continue;
	68	} while (!exit_backend);
	69
	70	if (!len)
	71	return 0;
	72
	73	return -1;
	74	}
	75
	76	static int verify_convert_cmd(struct fio_net_cmd *cmd)
	77	{
fcee5ff6	78	uint16_t crc;
132159a5	79
fcee5ff6 JA	80	cmd->cmd_crc16 = le16_to_cpu(cmd->cmd_crc16);
fcee5ff6 JA	81	cmd->pdu_crc16 = le16_to_cpu(cmd->pdu_crc16);
132159a5	82
fcee5ff6 JA	83	crc = crc16(cmd, FIO_NET_CMD_CRC_SZ);
fcee5ff6 JA	84	if (crc != cmd->cmd_crc16) {
132159a5	85	log_err("fio: server bad crc on command (got %x, wanted %x)\n",
fcee5ff6	86	cmd->cmd_crc16, crc);
132159a5 JA	87	return 1;
	88	}
	89
	90	cmd->version = le16_to_cpu(cmd->version);
	91	cmd->opcode = le16_to_cpu(cmd->opcode);
	92	cmd->flags = le32_to_cpu(cmd->flags);
	93	cmd->serial = le64_to_cpu(cmd->serial);
	94	cmd->pdu_len = le32_to_cpu(cmd->pdu_len);
	95
	96	switch (cmd->version) {
	97	case FIO_SERVER_VER1:
	98	break;
	99	default:
	100	log_err("fio: bad server cmd version %d\n", cmd->version);
	101	return 1;
	102	}
	103
	104	if (cmd->pdu_len > FIO_SERVER_MAX_PDU) {
	105	log_err("fio: command payload too large: %u\n", cmd->pdu_len);
	106	return 1;
	107	}
	108
	109	return 0;
	110	}
	111
37db14fe	112	struct fio_net_cmd *fio_net_cmd_read(int sk)
132159a5 JA	113	{
132159a5 JA	114	struct fio_net_cmd cmd, *ret = NULL;
fcee5ff6	115	uint16_t crc;
132159a5 JA	116
	117	if (fio_recv_data(sk, &cmd, sizeof(cmd)))
	118	return NULL;
	119
	120	/* We have a command, verify it and swap if need be */
	121	if (verify_convert_cmd(&cmd))
	122	return NULL;
	123
	124	/* Command checks out, alloc real command and fill in */
	125	ret = malloc(sizeof(cmd) + cmd.pdu_len);
	126	memcpy(ret, &cmd, sizeof(cmd));
	127
	128	if (!ret->pdu_len)
	129	return ret;
	130
	131	/* There's payload, get it */
	132	if (fio_recv_data(sk, (void ) ret + sizeof(ret), ret->pdu_len)) {
	133	free(ret);
	134	return NULL;
	135	}
	136
	137	/* Verify payload crc */
fcee5ff6 JA	138	crc = crc16(ret->payload, ret->pdu_len);
fcee5ff6 JA	139	if (crc != ret->pdu_crc16) {
132159a5	140	log_err("fio: server bad crc on payload (got %x, wanted %x)\n",
fcee5ff6	141	ret->pdu_crc16, crc);
132159a5 JA	142	free(ret);
	143	return NULL;
	144	}
	145
132159a5 JA	146	return ret;
	147	}
	148
	149	void fio_net_cmd_crc(struct fio_net_cmd *cmd)
	150	{
	151	uint32_t pdu_len;
	152
fcee5ff6	153	cmd->cmd_crc16 = cpu_to_le16(crc16(cmd, FIO_NET_CMD_CRC_SZ));
132159a5 JA	154
	155	pdu_len = le32_to_cpu(cmd->pdu_len);
	156	if (pdu_len)
fcee5ff6	157	cmd->pdu_crc16 = cpu_to_le16(crc16(cmd->payload, pdu_len));
132159a5 JA	158	}
	159
	160	static int send_simple_command(int sk, uint16_t opcode, uint64_t serial)
	161	{
	162	struct fio_net_cmd cmd = {
	163	.version = cpu_to_le16(FIO_SERVER_VER1),
	164	.opcode = cpu_to_le16(opcode),
bdab4441	165	.serial = cpu_to_le64(serial),
132159a5 JA	166	};
	167
	168	fio_net_cmd_crc(&cmd);
	169
	170	return fio_send_data(sk, &cmd, sizeof(cmd));
	171	}
	172
	173	/*
	174	* Send an ack for this command
	175	*/
	176	static int ack_command(int sk, struct fio_net_cmd *cmd)
	177	{
fcee5ff6	178	#if 0
132159a5	179	return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial);
fcee5ff6 JA	180	#else
	181	return 0;
	182	#endif
132159a5 JA	183	}
	184
	185	#if 0
	186	static int nak_command(int sk, struct fio_net_cmd *cmd)
	187	{
	188	return send_simple_command(sk, FIO_NET_CMD_NAK, cmd->serial);
	189	}
	190	#endif
	191
437377e1 JA	192	static int send_quit_command(void)
	193	{
	194	return send_simple_command(server_fd, FIO_NET_CMD_QUIT, 0);
	195	}
	196
132159a5 JA	197	static int handle_cur_job(struct fio_net_cmd *cmd, int done)
	198	{
	199	unsigned int left = job_max_len - job_cur_len;
	200	int ret = 0;
	201
	202	if (left < cmd->pdu_len) {
	203	job_buf = realloc(job_buf, job_max_len + 2 * cmd->pdu_len);
	204	job_max_len += 2 * cmd->pdu_len;
	205	}
	206
	207	memcpy(job_buf + job_cur_len, cmd->payload, cmd->pdu_len);
	208	job_cur_len += cmd->pdu_len;
	209
	210	if (done) {
	211	parse_jobs_ini(job_buf, 1, 0);
	212	ret = exec_run();
437377e1	213	send_quit_command();
132159a5 JA	214	reset_fio_state();
	215	free(job_buf);
	216	job_buf = NULL;
	217	job_cur_len = job_max_len = 0;
	218	}
	219
	220	return ret;
	221	}
	222
	223	static int handle_command(struct fio_net_cmd *cmd)
	224	{
	225	int ret;
	226
	227	switch (cmd->opcode) {
	228	case FIO_NET_CMD_QUIT:
	229	exit_backend = 1;
	230	return 1;
	231	case FIO_NET_CMD_ACK:
	232	return 0;
	233	case FIO_NET_CMD_NAK:
	234	return 1;
	235	case FIO_NET_CMD_JOB:
	236	ret = handle_cur_job(cmd, 0);
	237	break;
	238	case FIO_NET_CMD_JOB_END:
	239	ret = handle_cur_job(cmd, 1);
	240	break;
	241	default:
	242	log_err("fio: unknown opcode: %d\n", cmd->opcode);
	243	ret = 1;
	244	}
	245
	246	return ret;
	247	}
	248
	249	static int handle_connection(int sk)
	250	{
	251	struct fio_net_cmd *cmd = NULL;
	252	int ret = 0;
	253
	254	/* read forever */
	255	while (!exit_backend) {
37db14fe	256	cmd = fio_net_cmd_read(sk);
132159a5 JA	257	if (!cmd) {
	258	ret = 1;
	259	break;
	260	}
	261
	262	ret = ack_command(sk, cmd);
	263	if (ret)
	264	break;
	265
	266	ret = handle_command(cmd);
	267	if (ret)
	268	break;
	269
	270	free(cmd);
	271	}
	272
	273	if (cmd)
	274	free(cmd);
	275
	276	return ret;
	277	}
	278
50d16976 JA	279	static int accept_loop(int listen_sk)
	280	{
	281	struct sockaddr addr;
	282	unsigned int len = sizeof(addr);
009b1be4	283	struct pollfd pfd;
132159a5	284	int ret, sk, flags, exitval = 0;
50d16976	285
009b1be4 JA	286	flags = fcntl(listen_sk, F_GETFL);
	287	flags \|= O_NONBLOCK;
	288	fcntl(listen_sk, F_SETFL, flags);
50d16976	289	again:
009b1be4 JA	290	pfd.fd = listen_sk;
	291	pfd.events = POLLIN;
	292	do {
	293	ret = poll(&pfd, 1, 100);
	294	if (ret < 0) {
	295	if (errno == EINTR)
	296	break;
fcee5ff6	297	log_err("fio: poll: %s\n", strerror(errno));
009b1be4 JA	298	goto out;
	299	} else if (!ret)
	300	continue;
	301
	302	if (pfd.revents & POLLIN)
	303	break;
	304	} while (!exit_backend);
	305
	306	if (exit_backend)
	307	goto out;
	308
50d16976	309	sk = accept(listen_sk, &addr, &len);
50d16976	310	if (sk < 0) {
690e09ae	311	log_err("fio: accept: %s\n", strerror(errno));
50d16976 JA	312	return -1;
	313	}
	314
37db14fe JA	315	server_fd = sk;
37db14fe JA	316
132159a5	317	exitval = handle_connection(sk);
50d16976	318
37db14fe	319	server_fd = -1;
50d16976	320	close(sk);
5c341e9a	321
009b1be4	322	if (!exit_backend)
5c341e9a JA	323	goto again;
5c341e9a JA	324
009b1be4	325	out:
132159a5	326	return exitval;
50d16976 JA	327	}
	328
	329	int fio_server(void)
	330	{
	331	struct sockaddr_in saddr_in;
	332	struct sockaddr addr;
	333	unsigned int len;
afcf7758	334	int sk, opt, ret;
50d16976 JA	335
	336	sk = socket(AF_INET, SOCK_STREAM, 0);
	337	if (sk < 0) {
690e09ae	338	log_err("fio: socket: %s\n", strerror(errno));
50d16976 JA	339	return -1;
	340	}
	341
	342	opt = 1;
	343	if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
690e09ae	344	log_err("fio: setsockopt: %s\n", strerror(errno));
50d16976 JA	345	return -1;
	346	}
	347	#ifdef SO_REUSEPORT
	348	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
690e09ae	349	log_err("fio: setsockopt: %s\n", strerror(errno));
50d16976 JA	350	return 1;
	351	}
	352	#endif
	353
	354	saddr_in.sin_family = AF_INET;
	355	saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
132159a5	356	saddr_in.sin_port = htons(fio_net_port);
50d16976 JA	357
50d16976 JA	358	if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) {
690e09ae	359	log_err("fio: bind: %s\n", strerror(errno));
50d16976 JA	360	return -1;
	361	}
	362
	363	if (listen(sk, 1) < 0) {
690e09ae	364	log_err("fio: listen: %s\n", strerror(errno));
50d16976 JA	365	return -1;
	366	}
	367
	368	len = sizeof(addr);
	369	if (getsockname(sk, &addr, &len) < 0) {
690e09ae	370	log_err("fio: getsockname: %s\n", strerror(errno));
50d16976 JA	371	return -1;
	372	}
	373
afcf7758 JA	374	ret = accept_loop(sk);
	375	close(sk);
	376	return ret;
50d16976	377	}
37db14fe	378
142575e6	379	int fio_server_text_output(const char *buf, unsigned int len)
37db14fe JA	380	{
37db14fe JA	381	struct fio_net_cmd *cmd;
142575e6	382	int size = sizeof(*cmd) + len;
37db14fe	383
142575e6	384	cmd = malloc(size);
4d8f8780	385	fio_init_net_cmd(cmd, FIO_NET_CMD_TEXT, buf, len);
37db14fe JA	386	fio_net_cmd_crc(cmd);
37db14fe JA	387
142575e6	388	fio_send_data(server_fd, cmd, size);
37db14fe	389	free(cmd);
142575e6 JA	390	return size;
	391	}
	392
	393	int fio_server_log(const char *format, ...)
	394	{
	395	char buffer[1024];
	396	va_list args;
82fa6b21	397	size_t len;
142575e6 JA	398
142575e6 JA	399	va_start(args, format);
82fa6b21	400	len = vsnprintf(buffer, sizeof(buffer), format, args);
142575e6 JA	401	va_end(args);
142575e6 JA	402
82fa6b21	403	return fio_server_text_output(buffer, len);
37db14fe	404	}