X-Git-Url: https://git.kernel.dk/?a=blobdiff_plain;f=client.c;h=ed8f51e0337c9de5802e320149e8c72d5c7d6846;hb=80102c86786d7d50ba72366cee03061735b3cbac;hp=225f6643e5aaa0ec283f35f8903415789debb3b7;hpb=9988ca70151a191c477179be7999300ddff16230;p=fio.git

diff --git a/client.c b/client.c
index 225f6643..ed8f51e0 100644
--- a/client.c
+++ b/client.c
@@ -14,6 +14,7 @@
 #include <arpa/inet.h>
 #include <netdb.h>
 #include <signal.h>
+#include <zlib.h>
 
 #include "fio.h"
 #include "client.h"
@@ -27,15 +28,19 @@ static void handle_gs(struct fio_client *client, struct fio_net_cmd *cmd);
 static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd);
 static void handle_text(struct fio_client *client, struct fio_net_cmd *cmd);
 static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd);
+static void handle_start(struct fio_client *client, struct fio_net_cmd *cmd);
 
 struct client_ops fio_client_ops = {
-	.text_op	= handle_text,
+	.text		= handle_text,
 	.disk_util	= handle_du,
 	.thread_status	= handle_ts,
 	.group_stats	= handle_gs,
 	.stop		= handle_stop,
+	.start		= handle_start,
 	.eta		= display_thread_status,
 	.probe		= handle_probe,
+	.eta_msec	= FIO_CLIENT_DEF_ETA_MSEC,
+	.client_type	= FIO_CLIENT_TYPE_CLI,
 };
 
 static struct timeval eta_tv;
@@ -96,15 +101,28 @@ static struct fio_client *find_client_by_fd(int fd)
 	return NULL;
 }
 
-static void remove_client(struct fio_client *client)
+void fio_put_client(struct fio_client *client)
 {
-	assert(client->refs);
-
 	if (--client->refs)
 		return;
 
+	free(client->hostname);
+	if (client->argv)
+		free(client->argv);
+	if (client->name)
+		free(client->name);
+
+	free(client);
+}
+
+static void remove_client(struct fio_client *client)
+{
+	assert(client->refs);
+
 	dprint(FD_NET, "client: removed <%s>\n", client->hostname);
-	flist_del(&client->list);
+
+	if (!flist_empty(&client->list))
+		flist_del_init(&client->list);
 
 	fio_client_remove_hash(client);
 
@@ -113,20 +131,22 @@ static void remove_client(struct fio_client *client)
 		fio_client_dec_jobs_eta(client->eta_in_flight, client->ops->eta);
 	}
 
-	free(client->hostname);
-	if (client->argv)
-		free(client->argv);
-	if (client->name)
-		free(client->name);
+	close(client->fd);
+	client->fd = -1;
+
+	if (client->ops->removed)
+		client->ops->removed(client);
 
-	free(client);
 	nr_clients--;
 	sum_stat_clients--;
+
+	fio_put_client(client);
 }
 
-static void put_client(struct fio_client *client)
+struct fio_client *fio_get_client(struct fio_client *client)
 {
-	remove_client(client);
+	client->refs++;
+	return client;
 }
 
 static void __fio_client_add_cmd_option(struct fio_client *client,
@@ -194,6 +214,7 @@ struct fio_client *fio_client_add_explicit(struct client_ops *ops,
 	client->fd = -1;
 	client->ops = ops;
 	client->refs = 1;
+	client->type = ops->client_type;
 
 	__fio_client_add_cmd_option(client, "fio");
 
@@ -243,6 +264,7 @@ int fio_client_add(struct client_ops *ops, const char *hostname, void **cookie)
 	client->fd = -1;
 	client->ops = ops;
 	client->refs = 1;
+	client->type = ops->client_type;
 
 	__fio_client_add_cmd_option(client, "fio");
 
@@ -253,6 +275,13 @@ int fio_client_add(struct client_ops *ops, const char *hostname, void **cookie)
 	return 0;
 }
 
+static void probe_client(struct fio_client *client)
+{
+	dprint(FD_NET, "client: send probe\n");
+
+	fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0, &client->cmd_list);
+}
+
 static int fio_client_connect_ip(struct fio_client *client)
 {
 	struct sockaddr *addr;
@@ -275,16 +304,20 @@ static int fio_client_connect_ip(struct fio_client *client)
 
 	fd = socket(domain, SOCK_STREAM, 0);
 	if (fd < 0) {
+		int ret = -errno;
+
 		log_err("fio: socket: %s\n", strerror(errno));
-		return -1;
+		return ret;
 	}
 
 	if (connect(fd, addr, socklen) < 0) {
+		int ret = -errno;
+
 		log_err("fio: connect: %s\n", strerror(errno));
 		log_err("fio: failed to connect to %s:%u\n", client->hostname,
 								client->port);
 		close(fd);
-		return -1;
+		return ret;
 	}
 
 	return fd;
@@ -302,15 +335,19 @@ static int fio_client_connect_sock(struct fio_client *client)
 
 	fd = socket(AF_UNIX, SOCK_STREAM, 0);
 	if (fd < 0) {
+		int ret = -errno;
+
 		log_err("fio: socket: %s\n", strerror(errno));
-		return -1;
+		return ret;
 	}
 
 	len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
 	if (connect(fd, (struct sockaddr *) addr, len) < 0) {
+		int ret = -errno;
+
 		log_err("fio: connect; %s\n", strerror(errno));
 		close(fd);
-		return -1;
+		return ret;
 	}
 
 	return fd;
@@ -330,17 +367,19 @@ int fio_client_connect(struct fio_client *client)
 	dprint(FD_NET, "client: %s connected %d\n", client->hostname, fd);
 
 	if (fd < 0)
-		return 1;
+		return fd;
 
 	client->fd = fd;
 	fio_client_add_hash(client);
 	client->state = Client_connected;
+
+	probe_client(client);
 	return 0;
 }
 
-void fio_client_terminate(struct fio_client *client)
+int fio_client_terminate(struct fio_client *client)
 {
-	fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0, NULL);
+	return fio_net_send_quit(client->fd);
 }
 
 void fio_clients_terminate(void)
@@ -377,13 +416,6 @@ static void client_signal_handler(void)
 	sigaction(SIGTERM, &act, NULL);
 }
 
-static void probe_client(struct fio_client *client)
-{
-	dprint(FD_NET, "client: send probe\n");
-
-	fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0, &client->cmd_list);
-}
-
 static int send_client_cmd_line(struct fio_client *client)
 {
 	struct cmd_single_line_pdu *cslp;
@@ -426,6 +458,7 @@ static int send_client_cmd_line(struct fio_client *client)
 
 	free(lens);
 	clp->lines = cpu_to_le16(client->argc);
+	clp->client_type = __cpu_to_le16(client->type);
 	ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
 	free(pdu);
 	return ret;
@@ -455,8 +488,6 @@ int fio_clients_connect(void)
 			continue;
 		}
 
-		probe_client(client);
-
 		if (client->argc > 1)
 			send_client_cmd_line(client);
 	}
@@ -497,8 +528,11 @@ int fio_start_all_clients(void)
  */
 static int __fio_client_send_ini(struct fio_client *client, const char *filename)
 {
+	struct cmd_job_pdu *pdu;
+	size_t p_size;
 	struct stat sb;
-	char *p, *buf;
+	char *p;
+	void *buf;
 	off_t len;
 	int fd, ret;
 
@@ -506,17 +540,23 @@ static int __fio_client_send_ini(struct fio_client *client, const char *filename
 
 	fd = open(filename, O_RDONLY);
 	if (fd < 0) {
+		int ret = -errno;
+
 		log_err("fio: job file <%s> open: %s\n", filename, strerror(errno));
-		return 1;
+		return ret;
 	}
 
 	if (fstat(fd, &sb) < 0) {
+		int ret = -errno;
+
 		log_err("fio: job file stat: %s\n", strerror(errno));
 		close(fd);
-		return 1;
+		return ret;
 	}
 
-	buf = malloc(sb.st_size);
+	p_size = sb.st_size + sizeof(*pdu);
+	pdu = malloc(p_size);
+	buf = pdu->buf;
 
 	len = sb.st_size;
 	p = buf;
@@ -541,22 +581,25 @@ static int __fio_client_send_ini(struct fio_client *client, const char *filename
 		return 1;
 	}
 
+	pdu->buf_len = __cpu_to_le32(sb.st_size);
+	pdu->client_type = cpu_to_le32(client->type);
+
 	client->sent_job = 1;
-	ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size, 0);
-	free(buf);
+	ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, pdu, p_size, 0);
+	free(pdu);
 	close(fd);
 	return ret;
 }
 
 int fio_client_send_ini(struct fio_client *client, const char *filename)
 {
-	if (__fio_client_send_ini(client, filename)) {
-		remove_client(client);
-		return 1;
-	}
+	int ret;
 
-	client->sent_job = 1;
-	return 0;
+	ret = __fio_client_send_ini(client, filename);
+	if (!ret)
+		client->sent_job = 1;
+
+	return ret;
 }
 
 int fio_clients_send_ini(const char *filename)
@@ -567,7 +610,8 @@ int fio_clients_send_ini(const char *filename)
 	flist_for_each_safe(entry, tmp, &client_list) {
 		client = flist_entry(entry, struct fio_client, list);
 
-		fio_client_send_ini(client, filename);
+		if (fio_client_send_ini(client, filename))
+			remove_client(client);
 	}
 
 	return !nr_clients;
@@ -590,10 +634,11 @@ static void convert_ts(struct thread_stat *dst, struct thread_stat *src)
 {
 	int i, j;
 
-	dst->error	= le32_to_cpu(src->error);
-	dst->groupid	= le32_to_cpu(src->groupid);
-	dst->pid	= le32_to_cpu(src->pid);
-	dst->members	= le32_to_cpu(src->members);
+	dst->error		= le32_to_cpu(src->error);
+	dst->thread_number	= le32_to_cpu(src->thread_number);
+	dst->groupid		= le32_to_cpu(src->groupid);
+	dst->pid		= le32_to_cpu(src->pid);
+	dst->members		= le32_to_cpu(src->members);
 
 	for (i = 0; i < 2; i++) {
 		convert_io_stat(&dst->clat_stat[i], &src->clat_stat[i]);
@@ -681,6 +726,7 @@ static void handle_ts(struct fio_client *client, struct fio_net_cmd *cmd)
 	sum_group_stats(&client_gs, &p->rs);
 
 	client_ts.members++;
+	client_ts.thread_number = p->ts.thread_number;
 	client_ts.groupid = p->ts.groupid;
 
 	if (++sum_stat_nr == sum_stat_clients) {
@@ -887,7 +933,7 @@ static void handle_start(struct fio_client *client, struct fio_net_cmd *cmd)
 	struct cmd_start_pdu *pdu = (struct cmd_start_pdu *) cmd->payload;
 
 	client->state = Client_started;
-	client->jobs = le32_to_cpu(pdu->jobs);
+	client->jobs = pdu->jobs;
 }
 
 static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd)
@@ -913,6 +959,83 @@ static void convert_text(struct fio_net_cmd *cmd)
 	pdu->log_usec	= le64_to_cpu(pdu->log_usec);
 }
 
+/*
+ * This has been compressed on the server side, since it can be big.
+ * Uncompress here.
+ */
+static struct cmd_iolog_pdu *convert_iolog(struct fio_net_cmd *cmd)
+{
+	struct cmd_iolog_pdu *pdu = (struct cmd_iolog_pdu *) cmd->payload;
+	struct cmd_iolog_pdu *ret;
+	uint32_t nr_samples;
+	unsigned long total;
+	z_stream stream;
+	void *p;
+	int i;
+
+	stream.zalloc = Z_NULL;
+	stream.zfree = Z_NULL;
+	stream.opaque = Z_NULL;
+	stream.avail_in = 0;
+	stream.next_in = Z_NULL;
+
+	if (inflateInit(&stream) != Z_OK)
+		return NULL;
+
+	/*
+	 * Get header first, it's not compressed
+	 */
+	nr_samples = le32_to_cpu(pdu->nr_samples);
+
+	total = nr_samples * sizeof(struct io_sample);
+	ret = malloc(total + sizeof(*pdu));
+	ret->thread_number = le32_to_cpu(pdu->thread_number);
+	ret->nr_samples = nr_samples;
+	ret->log_type = le32_to_cpu(pdu->log_type);
+	strcpy((char *) ret->name, (char *) pdu->name);
+
+	p = (void *) ret + sizeof(*pdu);
+
+	stream.avail_in = cmd->pdu_len - sizeof(*pdu);
+	stream.next_in = (void *) pdu + sizeof(*pdu);
+	while (stream.avail_in) {
+		unsigned int this_chunk = 65536;
+		unsigned int this_len;
+		int err;
+
+		if (this_chunk > total)
+			this_chunk = total;
+
+		stream.avail_out = this_chunk;
+		stream.next_out = p;
+		err = inflate(&stream, Z_NO_FLUSH);
+		/* may be Z_OK, or Z_STREAM_END */
+		if (err < 0) {
+			log_err("fio: inflate error %d\n", err);
+			free(ret);
+			ret = NULL;
+			goto out;
+		}
+
+		this_len = this_chunk - stream.avail_out;
+		p += this_len;
+		total -= this_len;
+	}
+
+	for (i = 0; i < ret->nr_samples; i++) {
+		struct io_sample *s = &ret->samples[i];
+
+		s->time	= le64_to_cpu(s->time);
+		s->val	= le64_to_cpu(s->val);
+		s->ddir	= le32_to_cpu(s->ddir);
+		s->bs	= le32_to_cpu(s->bs);
+	}
+
+out:
+	inflateEnd(&stream);
+	return ret;
+}
+
 int fio_handle_client(struct fio_client *client)
 {
 	struct client_ops *ops = client->ops;
@@ -930,13 +1053,13 @@ int fio_handle_client(struct fio_client *client)
 	switch (cmd->opcode) {
 	case FIO_NET_CMD_QUIT:
 		if (ops->quit)
-			ops->quit(client);
+			ops->quit(client, cmd);
 		remove_client(client);
 		free(cmd);
 		break;
 	case FIO_NET_CMD_TEXT:
 		convert_text(cmd);
-		ops->text_op(client, cmd);
+		ops->text(client, cmd);
 		free(cmd);
 		break;
 	case FIO_NET_CMD_DU: {
@@ -984,18 +1107,25 @@ int fio_handle_client(struct fio_client *client)
 		break;
 	case FIO_NET_CMD_SERVER_START:
 		client->state = Client_running;
+		if (ops->job_start)
+			ops->job_start(client, cmd);
 		free(cmd);
 		break;
-	case FIO_NET_CMD_START:
-		handle_start(client, cmd);
+	case FIO_NET_CMD_START: {
+		struct cmd_start_pdu *pdu = (struct cmd_start_pdu *) cmd->payload;
+
+		pdu->jobs = le32_to_cpu(pdu->jobs);
+		ops->start(client, cmd);
 		free(cmd);
 		break;
+		}
 	case FIO_NET_CMD_STOP: {
 		struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload;
 
 		convert_stop(cmd);
 		client->state = Client_stopped;
-		client->error = pdu->error;
+		client->error = le32_to_cpu(pdu->error);
+		client->signal = le32_to_cpu(pdu->signal);
 		ops->stop(client, cmd);
 		free(cmd);
 		break;
@@ -1005,6 +1135,15 @@ int fio_handle_client(struct fio_client *client)
 			ops->add_job(client, cmd);
 		free(cmd);
 		break;
+	case FIO_NET_CMD_IOLOG:
+		if (ops->iolog) {
+			struct cmd_iolog_pdu *pdu;
+
+			pdu = convert_iolog(cmd);
+			ops->iolog(client, pdu);
+		}
+		free(cmd);
+		break;
 	default:
 		log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode));
 		free(cmd);
@@ -1144,7 +1283,7 @@ int fio_handle_clients(struct client_ops *ops)
 			struct timeval tv;
 
 			gettimeofday(&tv, NULL);
-			if (mtime_since(&eta_tv, &tv) >= 900) {
+			if (mtime_since(&eta_tv, &tv) >= ops->eta_msec) {
 				request_client_etas(ops);
 				memcpy(&eta_tv, &tv, sizeof(tv));
 
@@ -1152,7 +1291,7 @@ int fio_handle_clients(struct client_ops *ops)
 					break;
 			}
 
-			ret = poll(pfds, nr_clients, 100);
+			ret = poll(pfds, nr_clients, ops->eta_msec);
 			if (ret < 0) {
 				if (errno == EINTR)
 					continue;
@@ -1178,7 +1317,7 @@ int fio_handle_clients(struct client_ops *ops)
 				retval = 1;
 			} else if (client->error)
 				retval = 1;
-			put_client(client);
+			fio_put_client(client);
 		}
 	}