[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/crc32.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)
{
	uint32_t crc;

	cmd->cmd_crc32 = le32_to_cpu(cmd->cmd_crc32);
	cmd->pdu_crc32 = le32_to_cpu(cmd->pdu_crc32);

	crc = crc32(cmd, sizeof(*cmd) - 2 * sizeof(uint32_t));
	if (crc != cmd->cmd_crc32) {
		log_err("fio: server bad crc on command (got %x, wanted %x)\n",
				cmd->cmd_crc32, 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;
	uint32_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 = crc32(ret->payload, ret->pdu_len);
	if (crc != ret->pdu_crc32) {
		log_err("fio: server bad crc on payload (got %x, wanted %x)\n",
				ret->pdu_crc32, crc);
		free(ret);
		return NULL;
	}

	return ret;
}

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

	cmd->cmd_crc32 = cpu_to_le32(crc32(cmd,
					sizeof(*cmd) - 2 * sizeof(uint32_t)));

	pdu_len = le32_to_cpu(cmd->pdu_len);
	if (pdu_len)
		cmd->pdu_crc32 = cpu_to_le32(crc32(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_le16(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)
{
	return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial);
}

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