[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 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();
		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
	192	static int handle_cur_job(struct fio_net_cmd *cmd, int done)
	193	{
	194	unsigned int left = job_max_len - job_cur_len;
	195	int ret = 0;
	196
	197	if (left < cmd->pdu_len) {
	198	job_buf = realloc(job_buf, job_max_len + 2 * cmd->pdu_len);
	199	job_max_len += 2 * cmd->pdu_len;
	200	}
	201
	202	memcpy(job_buf + job_cur_len, cmd->payload, cmd->pdu_len);
	203	job_cur_len += cmd->pdu_len;
	204
	205	if (done) {
	206	parse_jobs_ini(job_buf, 1, 0);
	207	ret = exec_run();
	208	reset_fio_state();
	209	free(job_buf);
	210	job_buf = NULL;
	211	job_cur_len = job_max_len = 0;
	212	}
	213
	214	return ret;
	215	}
	216
	217	static int handle_command(struct fio_net_cmd *cmd)
	218	{
	219	int ret;
	220
	221	switch (cmd->opcode) {
	222	case FIO_NET_CMD_QUIT:
	223	exit_backend = 1;
	224	return 1;
	225	case FIO_NET_CMD_ACK:
	226	return 0;
	227	case FIO_NET_CMD_NAK:
	228	return 1;
	229	case FIO_NET_CMD_JOB:
	230	ret = handle_cur_job(cmd, 0);
	231	break;
	232	case FIO_NET_CMD_JOB_END:
	233	ret = handle_cur_job(cmd, 1);
	234	break;
	235	default:
	236	log_err("fio: unknown opcode: %d\n", cmd->opcode);
	237	ret = 1;
	238	}
	239
	240	return ret;
	241	}
	242
	243	static int handle_connection(int sk)
	244	{
	245	struct fio_net_cmd *cmd = NULL;
	246	int ret = 0;
247
248	/* read forever */
249	while (!exit_backend) {
37db14fe	250	cmd = fio_net_cmd_read(sk);
132159a5 JA	251	if (!cmd) {
	252	ret = 1;
	253	break;
	254	}
	255
	256	ret = ack_command(sk, cmd);
	257	if (ret)
	258	break;
	259
	260	ret = handle_command(cmd);
	261	if (ret)
	262	break;
	263
	264	free(cmd);
	265	}
	266
	267	if (cmd)
	268	free(cmd);
	269
	270	return ret;
	271	}
	272
50d16976 JA	273	static int accept_loop(int listen_sk)
	274	{
	275	struct sockaddr addr;
	276	unsigned int len = sizeof(addr);
009b1be4	277	struct pollfd pfd;
132159a5	278	int ret, sk, flags, exitval = 0;
50d16976	279
009b1be4 JA	280	flags = fcntl(listen_sk, F_GETFL);
	281	flags \|= O_NONBLOCK;
	282	fcntl(listen_sk, F_SETFL, flags);
50d16976	283	again:
009b1be4 JA	284	pfd.fd = listen_sk;
	285	pfd.events = POLLIN;
	286	do {
	287	ret = poll(&pfd, 1, 100);
	288	if (ret < 0) {
	289	if (errno == EINTR)
	290	break;
fcee5ff6	291	log_err("fio: poll: %s\n", strerror(errno));
009b1be4 JA	292	goto out;
	293	} else if (!ret)
	294	continue;
	295
	296	if (pfd.revents & POLLIN)
	297	break;
	298	} while (!exit_backend);
	299
	300	if (exit_backend)
	301	goto out;
	302
50d16976	303	sk = accept(listen_sk, &addr, &len);
50d16976	304	if (sk < 0) {
690e09ae	305	log_err("fio: accept: %s\n", strerror(errno));
50d16976 JA	306	return -1;
	307	}
	308
37db14fe JA	309	server_fd = sk;
37db14fe JA	310
132159a5	311	exitval = handle_connection(sk);
50d16976	312
37db14fe	313	server_fd = -1;
50d16976	314	close(sk);
5c341e9a	315
009b1be4	316	if (!exit_backend)
5c341e9a JA	317	goto again;
5c341e9a JA	318
009b1be4	319	out:
132159a5	320	return exitval;
50d16976 JA	321	}
	322
	323	int fio_server(void)
	324	{
	325	struct sockaddr_in saddr_in;
	326	struct sockaddr addr;
	327	unsigned int len;
afcf7758	328	int sk, opt, ret;
50d16976 JA	329
	330	sk = socket(AF_INET, SOCK_STREAM, 0);
	331	if (sk < 0) {
690e09ae	332	log_err("fio: socket: %s\n", strerror(errno));
50d16976 JA	333	return -1;
	334	}
	335
	336	opt = 1;
	337	if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
690e09ae	338	log_err("fio: setsockopt: %s\n", strerror(errno));
50d16976 JA	339	return -1;
	340	}
	341	#ifdef SO_REUSEPORT
	342	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
690e09ae	343	log_err("fio: setsockopt: %s\n", strerror(errno));
50d16976 JA	344	return 1;
	345	}
	346	#endif
	347
	348	saddr_in.sin_family = AF_INET;
	349	saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
132159a5	350	saddr_in.sin_port = htons(fio_net_port);
50d16976 JA	351
50d16976 JA	352	if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) {
690e09ae	353	log_err("fio: bind: %s\n", strerror(errno));
50d16976 JA	354	return -1;
	355	}
	356
	357	if (listen(sk, 1) < 0) {
690e09ae	358	log_err("fio: listen: %s\n", strerror(errno));
50d16976 JA	359	return -1;
	360	}
	361
	362	len = sizeof(addr);
	363	if (getsockname(sk, &addr, &len) < 0) {
690e09ae	364	log_err("fio: getsockname: %s\n", strerror(errno));
50d16976 JA	365	return -1;
	366	}
	367
afcf7758 JA	368	ret = accept_loop(sk);
	369	close(sk);
	370	return ret;
50d16976	371	}
37db14fe	372
142575e6	373	int fio_server_text_output(const char *buf, unsigned int len)
37db14fe JA	374	{
37db14fe JA	375	struct fio_net_cmd *cmd;
142575e6	376	int size = sizeof(*cmd) + len;
37db14fe	377
142575e6	378	cmd = malloc(size);
4d8f8780	379	fio_init_net_cmd(cmd, FIO_NET_CMD_TEXT, buf, len);
37db14fe JA	380	fio_net_cmd_crc(cmd);
37db14fe JA	381
142575e6	382	fio_send_data(server_fd, cmd, size);
37db14fe	383	free(cmd);
142575e6 JA	384	return size;
	385	}
	386
	387	int fio_server_log(const char *format, ...)
	388	{
	389	char buffer[1024];
	390	va_list args;
82fa6b21	391	size_t len;
142575e6 JA	392
142575e6 JA	393	va_start(args, format);
82fa6b21	394	len = vsnprintf(buffer, sizeof(buffer), format, args);
142575e6 JA	395	va_end(args);
142575e6 JA	396
82fa6b21	397	return fio_server_text_output(buffer, len);
37db14fe	398	}