[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);
	cmd->opcode	= cpu_to_le16(FIO_NET_CMD_TEXT);
	cmd->pdu_len	= cpu_to_le32(len);
	memcpy(&cmd->payload, 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;

	va_start(args, format);
	sprintf(buffer, format, args);
	va_end(args);

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