[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 <syslog.h>

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

int fio_net_port = 8765;

int exit_backend = 0;

static int server_fd = -1;

int fio_send_data(int sk, const void *p, unsigned int len)
{
	assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_PDU);

	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;
}

/*
 * Read (and defragment, if necessary) incoming commands
 */
struct fio_net_cmd *fio_net_recv_cmd(int sk)
{
	struct fio_net_cmd cmd, *cmdret = NULL;
	size_t cmd_size = 0, pdu_offset = 0;
	uint16_t crc;
	int ret, first = 1;
	void *pdu = NULL;

	do {
		ret = fio_recv_data(sk, &cmd, sizeof(cmd));
		if (ret)
			break;

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

		if (first)
			cmd_size = sizeof(cmd) + cmd.pdu_len;
		else
			cmd_size += cmd.pdu_len;

		cmdret = realloc(cmdret, cmd_size);
		pdu = (void *) cmdret->payload;

		if (first)
			memcpy(cmdret, &cmd, sizeof(cmd));
		else
			assert(cmdret->opcode == cmd.opcode);

		if (!cmd.pdu_len)
			break;

		/* There's payload, get it */
		pdu = (void *) cmdret->payload + pdu_offset;
		ret = fio_recv_data(sk, pdu, cmd.pdu_len);
		if (ret)
			break;

		/* Verify payload crc */
		crc = crc16(pdu, cmd.pdu_len);
		if (crc != cmd.pdu_crc16) {
			log_err("fio: server bad crc on payload ");
			log_err("(got %x, wanted %x)\n", cmd.pdu_crc16, crc);
			ret = 1;
			break;
		}

		pdu_offset += cmd.pdu_len;
		cmdret->pdu_len += cmd.pdu_len;
		first = 0;
	} while (cmd.flags & FIO_NET_CMD_F_MORE);

	if (ret) {
		free(cmdret);
		cmdret = NULL;
	}

	if (cmdret)
		cmdret->flags &= ~FIO_NET_CMD_F_MORE;

	return cmdret;
}

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));
}

int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size)
{
	struct fio_net_cmd *cmd;
	size_t this_len;
	int ret;

	do {
		this_len = size;
		if (this_len > FIO_SERVER_MAX_PDU)
			this_len = FIO_SERVER_MAX_PDU;

		cmd = malloc(sizeof(*cmd) + this_len);

		fio_init_net_cmd(cmd, opcode, buf, this_len);

		if (this_len < size)
			cmd->flags = cpu_to_le32(FIO_NET_CMD_F_MORE);

		fio_net_cmd_crc(cmd);

		ret = fio_send_data(fd, cmd, sizeof(*cmd) + this_len);
		free(cmd);
		size -= this_len;
		buf += this_len;
	} while (!ret && size);

	return ret;
}

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)
{
	dprint(FD_NET, "server: sending quit\n");
	return send_simple_command(server_fd, FIO_NET_CMD_QUIT, 0);
}

static int handle_cur_job(struct fio_net_cmd *cmd)
{
	void *buf = cmd->payload;
	int ret;

	parse_jobs_ini(buf, 1, 0);
	ret = exec_run();
	send_quit_command();
	reset_fio_state();
	return ret;
}

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

	dprint(FD_NET, "server: got opcode %d\n", cmd->opcode);

	switch (cmd->opcode) {
	case FIO_NET_CMD_QUIT:
		return 1;
	case FIO_NET_CMD_EXIT:
		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);
		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_recv_cmd(sk);
		if (!cmd) {
			ret = 1;
			break;
		}

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

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

		free(cmd);
		cmd = NULL;
	}

	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;
	}

	dprint(FD_NET, "server got a connection\n");

	server_fd = sk;

	exitval = handle_connection(sk);

	server_fd = -1;
	close(sk);

	if (!exit_backend)
		goto again;

out:
	return exitval;
}

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

	dprint(FD_NET, "starting server\n");

	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)
{
	if (server_fd != -1)
		return fio_net_send_cmd(server_fd, FIO_NET_CMD_TEXT, buf, len);

	return 0;
}

static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
{
	dst->max_val	= cpu_to_le64(src->max_val);
	dst->min_val	= cpu_to_le64(src->min_val);
	dst->samples	= cpu_to_le64(src->samples);
	/* FIXME */
	dst->mean	= cpu_to_le64(src->mean);
	dst->S		= cpu_to_le64(src->S);
}

static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
{
	int i;

	for (i = 0; i < 2; i++) {
		dst->max_run[i]		= cpu_to_le64(src->max_run[i]);
		dst->min_run[i]		= cpu_to_le64(src->min_run[i]);
		dst->max_bw[i]		= cpu_to_le64(src->max_bw[i]);
		dst->min_bw[i]		= cpu_to_le64(src->min_bw[i]);
		dst->io_kb[i]		= cpu_to_le64(src->io_kb[i]);
		dst->agg[i]		= cpu_to_le64(src->agg[i]);
	}

	dst->kb_base	= cpu_to_le32(src->kb_base);
	dst->groupid	= cpu_to_le32(src->groupid);
}

/*
 * Send a CMD_TS, which packs struct thread_stat and group_run_stats
 * into a single payload.
 */
void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs)
{
	struct cmd_ts_pdu p;
	int i, j;

	strcpy(p.ts.name, ts->name);
	strcpy(p.ts.verror, ts->verror);
	strcpy(p.ts.description, ts->description);

	p.ts.error		= cpu_to_le32(ts->error);
	p.ts.groupid	= cpu_to_le32(ts->groupid);
	p.ts.pid		= cpu_to_le32(ts->pid);
	p.ts.members	= cpu_to_le32(ts->members);

	for (i = 0; i < 2; i++) {
		convert_io_stat(&p.ts.clat_stat[i], &ts->clat_stat[i]);
		convert_io_stat(&p.ts.slat_stat[i], &ts->slat_stat[i]);
		convert_io_stat(&p.ts.lat_stat[i], &ts->lat_stat[i]);
		convert_io_stat(&p.ts.bw_stat[i], &ts->bw_stat[i]);
	}

	p.ts.usr_time		= cpu_to_le64(ts->usr_time);
	p.ts.sys_time		= cpu_to_le64(ts->sys_time);
	p.ts.ctx		= cpu_to_le64(ts->ctx);
	p.ts.minf		= cpu_to_le64(ts->minf);
	p.ts.majf		= cpu_to_le64(ts->majf);
	p.ts.clat_percentiles	= cpu_to_le64(ts->clat_percentiles);
	p.ts.percentile_list	= NULL;

	for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
		p.ts.io_u_map[i]	= cpu_to_le32(ts->io_u_map[i]);
		p.ts.io_u_submit[i]	= cpu_to_le32(ts->io_u_submit[i]);
		p.ts.io_u_complete[i]	= cpu_to_le32(ts->io_u_complete[i]);
	}

	for (i = 0; i < FIO_IO_U_LAT_U_NR; i++) {
		p.ts.io_u_lat_u[i]	= cpu_to_le32(ts->io_u_lat_u[i]);
		p.ts.io_u_lat_m[i]	= cpu_to_le32(ts->io_u_lat_m[i]);
	}

	for (i = 0; i < 2; i++)
		for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
			p.ts.io_u_plat[i][j] = cpu_to_le32(ts->io_u_plat[i][j]);

	for (i = 0; i < 3; i++) {
		p.ts.total_io_u[i]	= cpu_to_le64(ts->total_io_u[i]);
		p.ts.short_io_u[i]	= cpu_to_le64(ts->total_io_u[i]);
	}

	p.ts.total_submit		= cpu_to_le64(ts->total_submit);
	p.ts.total_complete	= cpu_to_le64(ts->total_complete);

	for (i = 0; i < 2; i++) {
		p.ts.io_bytes[i]	= cpu_to_le64(ts->io_bytes[i]);
		p.ts.runtime[i]		= cpu_to_le64(ts->runtime[i]);
	}

	p.ts.total_run_time	= cpu_to_le64(ts->total_run_time);
	p.ts.continue_on_error	= cpu_to_le16(ts->continue_on_error);
	p.ts.total_err_count	= cpu_to_le64(ts->total_err_count);
	p.ts.first_error	= cpu_to_le64(ts->first_error);
	p.ts.kb_base		= cpu_to_le64(ts->kb_base);

	convert_gs(&p.rs, rs);

	fio_net_send_cmd(server_fd, FIO_NET_CMD_TS, &p, sizeof(p));
}

void fio_server_send_gs(struct group_run_stats *rs)
{
	struct group_run_stats gs;

	convert_gs(&gs, rs);
	fio_net_send_cmd(server_fd, FIO_NET_CMD_GS, &gs, sizeof(gs));
}

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);
}

int fio_start_server(int daemonize)
{
	pid_t pid;

	if (!daemonize)
		return fio_server();

	openlog("fio", LOG_NDELAY|LOG_NOWAIT|LOG_PID, LOG_USER);
	pid = fork();
	if (pid < 0) {
		syslog(LOG_ERR, "failed server fork");
		return 1;
	} else if (pid)
		exit(0);

	setsid();
	close(STDIN_FILENO);
	close(STDOUT_FILENO);
	close(STDERR_FILENO);
	f_out = NULL;
	f_err = NULL;
	log_syslog = 1;
	return fio_server();
}
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>
e46d8091	15	#include <syslog.h>
50d16976 JA	16
50d16976 JA	17	#include "fio.h"
132159a5	18	#include "server.h"
fcee5ff6	19	#include "crc/crc16.h"
50d16976	20
132159a5	21	int fio_net_port = 8765;
50d16976	22
009b1be4 JA	23	int exit_backend = 0;
009b1be4 JA	24
46c48f1f	25	static int server_fd = -1;
37db14fe	26
132159a5 JA	27	int fio_send_data(int sk, const void *p, unsigned int len)
132159a5 JA	28	{
794d69ca JA	29	assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_PDU);
794d69ca JA	30
132159a5 JA	31	do {
	32	int ret = send(sk, p, len, 0);
	33
	34	if (ret > 0) {
	35	len -= ret;
	36	if (!len)
	37	break;
	38	p += ret;
	39	continue;
	40	} else if (!ret)
	41	break;
	42	else if (errno == EAGAIN \|\| errno == EINTR)
	43	continue;
	44	} while (!exit_backend);
	45
	46	if (!len)
	47	return 0;
	48
	49	return 1;
	50	}
	51
	52	int fio_recv_data(int sk, void *p, unsigned int len)
	53	{
	54	do {
	55	int ret = recv(sk, p, len, MSG_WAITALL);
	56
	57	if (ret > 0) {
	58	len -= ret;
	59	if (!len)
	60	break;
	61	p += ret;
	62	continue;
	63	} else if (!ret)
	64	break;
	65	else if (errno == EAGAIN \|\| errno == EINTR)
	66	continue;
	67	} while (!exit_backend);
	68
	69	if (!len)
	70	return 0;
	71
	72	return -1;
	73	}
	74
	75	static int verify_convert_cmd(struct fio_net_cmd *cmd)
	76	{
fcee5ff6	77	uint16_t crc;
132159a5	78
fcee5ff6 JA	79	cmd->cmd_crc16 = le16_to_cpu(cmd->cmd_crc16);
fcee5ff6 JA	80	cmd->pdu_crc16 = le16_to_cpu(cmd->pdu_crc16);
132159a5	81
fcee5ff6 JA	82	crc = crc16(cmd, FIO_NET_CMD_CRC_SZ);
fcee5ff6 JA	83	if (crc != cmd->cmd_crc16) {
132159a5	84	log_err("fio: server bad crc on command (got %x, wanted %x)\n",
fcee5ff6	85	cmd->cmd_crc16, crc);
132159a5 JA	86	return 1;
	87	}
	88
	89	cmd->version = le16_to_cpu(cmd->version);
	90	cmd->opcode = le16_to_cpu(cmd->opcode);
	91	cmd->flags = le32_to_cpu(cmd->flags);
	92	cmd->serial = le64_to_cpu(cmd->serial);
	93	cmd->pdu_len = le32_to_cpu(cmd->pdu_len);
	94
	95	switch (cmd->version) {
	96	case FIO_SERVER_VER1:
	97	break;
	98	default:
	99	log_err("fio: bad server cmd version %d\n", cmd->version);
	100	return 1;
	101	}
	102
	103	if (cmd->pdu_len > FIO_SERVER_MAX_PDU) {
	104	log_err("fio: command payload too large: %u\n", cmd->pdu_len);
	105	return 1;
	106	}
	107
	108	return 0;
	109	}
	110
a64e88da JA	111	/*
	112	* Read (and defragment, if necessary) incoming commands
	113	*/
	114	struct fio_net_cmd *fio_net_recv_cmd(int sk)
132159a5	115	{
a64e88da JA	116	struct fio_net_cmd cmd, *cmdret = NULL;
a64e88da JA	117	size_t cmd_size = 0, pdu_offset = 0;
fcee5ff6	118	uint16_t crc;
a64e88da JA	119	int ret, first = 1;
a64e88da JA	120	void *pdu = NULL;
132159a5	121
a64e88da JA	122	do {
	123	ret = fio_recv_data(sk, &cmd, sizeof(cmd));
	124	if (ret)
	125	break;
132159a5	126
a64e88da JA	127	/* We have a command, verify it and swap if need be */
	128	ret = verify_convert_cmd(&cmd);
	129	if (ret)
	130	break;
132159a5	131
a64e88da JA	132	if (first)
	133	cmd_size = sizeof(cmd) + cmd.pdu_len;
	134	else
	135	cmd_size += cmd.pdu_len;
132159a5	136
a64e88da JA	137	cmdret = realloc(cmdret, cmd_size);
a64e88da JA	138	pdu = (void *) cmdret->payload;
132159a5	139
a64e88da JA	140	if (first)
	141	memcpy(cmdret, &cmd, sizeof(cmd));
	142	else
	143	assert(cmdret->opcode == cmd.opcode);
	144
	145	if (!cmd.pdu_len)
	146	break;
	147
	148	/* There's payload, get it */
	149	pdu = (void *) cmdret->payload + pdu_offset;
	150	ret = fio_recv_data(sk, pdu, cmd.pdu_len);
	151	if (ret)
	152	break;
	153
	154	/* Verify payload crc */
	155	crc = crc16(pdu, cmd.pdu_len);
	156	if (crc != cmd.pdu_crc16) {
	157	log_err("fio: server bad crc on payload ");
	158	log_err("(got %x, wanted %x)\n", cmd.pdu_crc16, crc);
	159	ret = 1;
	160	break;
	161	}
	162
	163	pdu_offset += cmd.pdu_len;
	164	cmdret->pdu_len += cmd.pdu_len;
	165	first = 0;
	166	} while (cmd.flags & FIO_NET_CMD_F_MORE);
132159a5	167
a64e88da JA	168	if (ret) {
	169	free(cmdret);
	170	cmdret = NULL;
132159a5 JA	171	}
132159a5 JA	172
a64e88da JA	173	if (cmdret)
	174	cmdret->flags &= ~FIO_NET_CMD_F_MORE;
	175
	176	return cmdret;
132159a5 JA	177	}
	178
	179	void fio_net_cmd_crc(struct fio_net_cmd *cmd)
	180	{
	181	uint32_t pdu_len;
	182
fcee5ff6	183	cmd->cmd_crc16 = cpu_to_le16(crc16(cmd, FIO_NET_CMD_CRC_SZ));
132159a5 JA	184
	185	pdu_len = le32_to_cpu(cmd->pdu_len);
	186	if (pdu_len)
fcee5ff6	187	cmd->pdu_crc16 = cpu_to_le16(crc16(cmd->payload, pdu_len));
132159a5 JA	188	}
132159a5 JA	189
a64e88da	190	int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size)
794d69ca JA	191	{
	192	struct fio_net_cmd *cmd;
	193	size_t this_len;
	194	int ret;
	195
	196	do {
	197	this_len = size;
	198	if (this_len > FIO_SERVER_MAX_PDU)
	199	this_len = FIO_SERVER_MAX_PDU;
	200
	201	cmd = malloc(sizeof(*cmd) + this_len);
	202
	203	fio_init_net_cmd(cmd, opcode, buf, this_len);
	204
	205	if (this_len < size)
d7959186	206	cmd->flags = cpu_to_le32(FIO_NET_CMD_F_MORE);
794d69ca JA	207
	208	fio_net_cmd_crc(cmd);
	209
	210	ret = fio_send_data(fd, cmd, sizeof(*cmd) + this_len);
	211	free(cmd);
	212	size -= this_len;
	213	buf += this_len;
	214	} while (!ret && size);
	215
	216	return ret;
	217	}
	218
132159a5 JA	219	static int send_simple_command(int sk, uint16_t opcode, uint64_t serial)
	220	{
	221	struct fio_net_cmd cmd = {
	222	.version = cpu_to_le16(FIO_SERVER_VER1),
	223	.opcode = cpu_to_le16(opcode),
bdab4441	224	.serial = cpu_to_le64(serial),
132159a5 JA	225	};
	226
	227	fio_net_cmd_crc(&cmd);
	228
	229	return fio_send_data(sk, &cmd, sizeof(cmd));
	230	}
	231
	232	/*
	233	* Send an ack for this command
	234	*/
	235	static int ack_command(int sk, struct fio_net_cmd *cmd)
	236	{
fcee5ff6	237	#if 0
132159a5	238	return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial);
fcee5ff6 JA	239	#else
	240	return 0;
	241	#endif
132159a5 JA	242	}
	243
	244	#if 0
	245	static int nak_command(int sk, struct fio_net_cmd *cmd)
	246	{
	247	return send_simple_command(sk, FIO_NET_CMD_NAK, cmd->serial);
	248	}
	249	#endif
	250
437377e1 JA	251	static int send_quit_command(void)
437377e1 JA	252	{
46c48f1f	253	dprint(FD_NET, "server: sending quit\n");
437377e1 JA	254	return send_simple_command(server_fd, FIO_NET_CMD_QUIT, 0);
	255	}
	256
794d69ca	257	static int handle_cur_job(struct fio_net_cmd *cmd)
132159a5	258	{
a64e88da JA	259	void *buf = cmd->payload;
a64e88da JA	260	int ret;
132159a5	261
a64e88da	262	parse_jobs_ini(buf, 1, 0);
794d69ca JA	263	ret = exec_run();
	264	send_quit_command();
	265	reset_fio_state();
132159a5 JA	266	return ret;
	267	}
	268
	269	static int handle_command(struct fio_net_cmd *cmd)
	270	{
	271	int ret;
	272
46c48f1f JA	273	dprint(FD_NET, "server: got opcode %d\n", cmd->opcode);
46c48f1f JA	274
132159a5 JA	275	switch (cmd->opcode) {
132159a5 JA	276	case FIO_NET_CMD_QUIT:
d7959186 JA	277	return 1;
d7959186 JA	278	case FIO_NET_CMD_EXIT:
132159a5 JA	279	exit_backend = 1;
	280	return 1;
	281	case FIO_NET_CMD_ACK:
	282	return 0;
	283	case FIO_NET_CMD_NAK:
	284	return 1;
	285	case FIO_NET_CMD_JOB:
794d69ca	286	ret = handle_cur_job(cmd);
132159a5 JA	287	break;
	288	default:
	289	log_err("fio: unknown opcode: %d\n", cmd->opcode);
	290	ret = 1;
	291	}
	292
	293	return ret;
	294	}
	295
	296	static int handle_connection(int sk)
	297	{
	298	struct fio_net_cmd *cmd = NULL;
	299	int ret = 0;
	300
	301	/* read forever */
	302	while (!exit_backend) {
a64e88da	303	cmd = fio_net_recv_cmd(sk);
132159a5 JA	304	if (!cmd) {
	305	ret = 1;
	306	break;
	307	}
	308
	309	ret = ack_command(sk, cmd);
	310	if (ret)
	311	break;
	312
	313	ret = handle_command(cmd);
	314	if (ret)
	315	break;
	316
	317	free(cmd);
c77a99e7	318	cmd = NULL;
132159a5 JA	319	}
	320
	321	if (cmd)
	322	free(cmd);
	323
	324	return ret;
	325	}
	326
50d16976 JA	327	static int accept_loop(int listen_sk)
	328	{
	329	struct sockaddr addr;
	330	unsigned int len = sizeof(addr);
009b1be4	331	struct pollfd pfd;
132159a5	332	int ret, sk, flags, exitval = 0;
50d16976	333
009b1be4 JA	334	flags = fcntl(listen_sk, F_GETFL);
	335	flags \|= O_NONBLOCK;
	336	fcntl(listen_sk, F_SETFL, flags);
50d16976	337	again:
009b1be4 JA	338	pfd.fd = listen_sk;
	339	pfd.events = POLLIN;
	340	do {
	341	ret = poll(&pfd, 1, 100);
	342	if (ret < 0) {
	343	if (errno == EINTR)
	344	break;
fcee5ff6	345	log_err("fio: poll: %s\n", strerror(errno));
009b1be4 JA	346	goto out;
	347	} else if (!ret)
	348	continue;
	349
	350	if (pfd.revents & POLLIN)
	351	break;
	352	} while (!exit_backend);
	353
	354	if (exit_backend)
	355	goto out;
	356
50d16976	357	sk = accept(listen_sk, &addr, &len);
50d16976	358	if (sk < 0) {
690e09ae	359	log_err("fio: accept: %s\n", strerror(errno));
50d16976 JA	360	return -1;
	361	}
	362
46c48f1f JA	363	dprint(FD_NET, "server got a connection\n");
46c48f1f JA	364
37db14fe JA	365	server_fd = sk;
37db14fe JA	366
132159a5	367	exitval = handle_connection(sk);
50d16976	368
37db14fe	369	server_fd = -1;
50d16976	370	close(sk);
5c341e9a	371
009b1be4	372	if (!exit_backend)
5c341e9a JA	373	goto again;
5c341e9a JA	374
009b1be4	375	out:
132159a5	376	return exitval;
50d16976 JA	377	}
50d16976 JA	378
e46d8091	379	static int fio_server(void)
50d16976 JA	380	{
	381	struct sockaddr_in saddr_in;
	382	struct sockaddr addr;
	383	unsigned int len;
afcf7758	384	int sk, opt, ret;
50d16976	385
46c48f1f JA	386	dprint(FD_NET, "starting server\n");
46c48f1f JA	387
50d16976 JA	388	sk = socket(AF_INET, SOCK_STREAM, 0);
50d16976 JA	389	if (sk < 0) {
690e09ae	390	log_err("fio: socket: %s\n", strerror(errno));
50d16976 JA	391	return -1;
	392	}
	393
	394	opt = 1;
	395	if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
690e09ae	396	log_err("fio: setsockopt: %s\n", strerror(errno));
50d16976 JA	397	return -1;
	398	}
	399	#ifdef SO_REUSEPORT
	400	if (setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
690e09ae	401	log_err("fio: setsockopt: %s\n", strerror(errno));
50d16976 JA	402	return 1;
	403	}
	404	#endif
	405
	406	saddr_in.sin_family = AF_INET;
	407	saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
132159a5	408	saddr_in.sin_port = htons(fio_net_port);
50d16976 JA	409
50d16976 JA	410	if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) {
690e09ae	411	log_err("fio: bind: %s\n", strerror(errno));
50d16976 JA	412	return -1;
	413	}
	414
	415	if (listen(sk, 1) < 0) {
690e09ae	416	log_err("fio: listen: %s\n", strerror(errno));
50d16976 JA	417	return -1;
	418	}
	419
	420	len = sizeof(addr);
	421	if (getsockname(sk, &addr, &len) < 0) {
690e09ae	422	log_err("fio: getsockname: %s\n", strerror(errno));
50d16976 JA	423	return -1;
	424	}
	425
afcf7758 JA	426	ret = accept_loop(sk);
	427	close(sk);
	428	return ret;
50d16976	429	}
37db14fe	430
142575e6	431	int fio_server_text_output(const char *buf, unsigned int len)
37db14fe	432	{
337d75a8 JA	433	if (server_fd != -1)
	434	return fio_net_send_cmd(server_fd, FIO_NET_CMD_TEXT, buf, len);
	435
	436	return 0;
142575e6 JA	437	}
142575e6 JA	438
a64e88da JA	439	static void convert_io_stat(struct io_stat dst, struct io_stat src)
	440	{
	441	dst->max_val = cpu_to_le64(src->max_val);
	442	dst->min_val = cpu_to_le64(src->min_val);
	443	dst->samples = cpu_to_le64(src->samples);
	444	/* FIXME */
	445	dst->mean = cpu_to_le64(src->mean);
	446	dst->S = cpu_to_le64(src->S);
	447	}
	448
	449	static void convert_gs(struct group_run_stats dst, struct group_run_stats src)
	450	{
	451	int i;
	452
	453	for (i = 0; i < 2; i++) {
	454	dst->max_run[i] = cpu_to_le64(src->max_run[i]);
	455	dst->min_run[i] = cpu_to_le64(src->min_run[i]);
	456	dst->max_bw[i] = cpu_to_le64(src->max_bw[i]);
	457	dst->min_bw[i] = cpu_to_le64(src->min_bw[i]);
	458	dst->io_kb[i] = cpu_to_le64(src->io_kb[i]);
	459	dst->agg[i] = cpu_to_le64(src->agg[i]);
	460	}
	461
	462	dst->kb_base = cpu_to_le32(src->kb_base);
	463	dst->groupid = cpu_to_le32(src->groupid);
	464	}
	465
	466	/*
	467	* Send a CMD_TS, which packs struct thread_stat and group_run_stats
	468	* into a single payload.
	469	*/
	470	void fio_server_send_ts(struct thread_stat ts, struct group_run_stats rs)
	471	{
	472	struct cmd_ts_pdu p;
	473	int i, j;
	474
	475	strcpy(p.ts.name, ts->name);
	476	strcpy(p.ts.verror, ts->verror);
	477	strcpy(p.ts.description, ts->description);
	478
	479	p.ts.error = cpu_to_le32(ts->error);
	480	p.ts.groupid = cpu_to_le32(ts->groupid);
	481	p.ts.pid = cpu_to_le32(ts->pid);
	482	p.ts.members = cpu_to_le32(ts->members);
	483
	484	for (i = 0; i < 2; i++) {
	485	convert_io_stat(&p.ts.clat_stat[i], &ts->clat_stat[i]);
	486	convert_io_stat(&p.ts.slat_stat[i], &ts->slat_stat[i]);
	487	convert_io_stat(&p.ts.lat_stat[i], &ts->lat_stat[i]);
	488	convert_io_stat(&p.ts.bw_stat[i], &ts->bw_stat[i]);
	489	}
	490
	491	p.ts.usr_time = cpu_to_le64(ts->usr_time);
	492	p.ts.sys_time = cpu_to_le64(ts->sys_time);
	493	p.ts.ctx = cpu_to_le64(ts->ctx);
	494	p.ts.minf = cpu_to_le64(ts->minf);
	495	p.ts.majf = cpu_to_le64(ts->majf);
	496	p.ts.clat_percentiles = cpu_to_le64(ts->clat_percentiles);
	497	p.ts.percentile_list = NULL;
	498
	499	for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
	500	p.ts.io_u_map[i] = cpu_to_le32(ts->io_u_map[i]);
	501	p.ts.io_u_submit[i] = cpu_to_le32(ts->io_u_submit[i]);
	502	p.ts.io_u_complete[i] = cpu_to_le32(ts->io_u_complete[i]);
503	}
504
505	for (i = 0; i < FIO_IO_U_LAT_U_NR; i++) {
506	p.ts.io_u_lat_u[i] = cpu_to_le32(ts->io_u_lat_u[i]);
507	p.ts.io_u_lat_m[i] = cpu_to_le32(ts->io_u_lat_m[i]);
508	}
509
510	for (i = 0; i < 2; i++)
511	for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
512	p.ts.io_u_plat[i][j] = cpu_to_le32(ts->io_u_plat[i][j]);
513
514	for (i = 0; i < 3; i++) {
515	p.ts.total_io_u[i] = cpu_to_le64(ts->total_io_u[i]);
516	p.ts.short_io_u[i] = cpu_to_le64(ts->total_io_u[i]);
517	}
518
519	p.ts.total_submit = cpu_to_le64(ts->total_submit);
520	p.ts.total_complete = cpu_to_le64(ts->total_complete);
521
522	for (i = 0; i < 2; i++) {
523	p.ts.io_bytes[i] = cpu_to_le64(ts->io_bytes[i]);
524	p.ts.runtime[i] = cpu_to_le64(ts->runtime[i]);
525	}
526
527	p.ts.total_run_time = cpu_to_le64(ts->total_run_time);
528	p.ts.continue_on_error = cpu_to_le16(ts->continue_on_error);
529	p.ts.total_err_count = cpu_to_le64(ts->total_err_count);
530	p.ts.first_error = cpu_to_le64(ts->first_error);
531	p.ts.kb_base = cpu_to_le64(ts->kb_base);
532
533	convert_gs(&p.rs, rs);
534
535	fio_net_send_cmd(server_fd, FIO_NET_CMD_TS, &p, sizeof(p));
536	}
537
538	void fio_server_send_gs(struct group_run_stats *rs)
539	{
540	struct group_run_stats gs;
541
542	convert_gs(&gs, rs);
543	fio_net_send_cmd(server_fd, FIO_NET_CMD_GS, &gs, sizeof(gs));
544	}
545
142575e6 JA	546	int fio_server_log(const char *format, ...)
	547	{
	548	char buffer[1024];
	549	va_list args;
82fa6b21	550	size_t len;
142575e6 JA	551
142575e6 JA	552	va_start(args, format);
82fa6b21	553	len = vsnprintf(buffer, sizeof(buffer), format, args);
142575e6 JA	554	va_end(args);
142575e6 JA	555
82fa6b21	556	return fio_server_text_output(buffer, len);
37db14fe	557	}
e46d8091 JA	558
	559	int fio_start_server(int daemonize)
	560	{
	561	pid_t pid;
	562
	563	if (!daemonize)
	564	return fio_server();
	565
	566	openlog("fio", LOG_NDELAY\|LOG_NOWAIT\|LOG_PID, LOG_USER);
	567	pid = fork();
	568	if (pid < 0) {
	569	syslog(LOG_ERR, "failed server fork");
	570	return 1;
	571	} else if (pid)
	572	exit(0);
	573
	574	setsid();
	575	close(STDIN_FILENO);
	576	close(STDOUT_FILENO);
	577	close(STDERR_FILENO);
	578	f_out = NULL;
	579	f_err = NULL;
	580	log_syslog = 1;
	581	return fio_server();
	582	}