X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=server.c;h=1d374fc69f5c7b37d500cb53e65057dd94a21880;hp=27ff2854be4bac9045b884f9db365bb061229db3;hb=de564c4d8d7f436d5b34656e72167142709c2a99;hpb=9c0037824e313eb9bd7dbc53cf302a42ad011fa8

diff --git a/server.c b/server.c
index 27ff2854..1d374fc6 100644
--- a/server.c
+++ b/server.c
@@ -37,6 +37,7 @@ enum {
 	SK_F_COPY	= 2,
 	SK_F_SIMPLE	= 4,
 	SK_F_VEC	= 8,
+	SK_F_INLINE	= 16,
 };
 
 struct sk_entry {
@@ -54,9 +55,10 @@ struct sk_out {
 				 * protected by below ->lock */
 
 	int sk;			/* socket fd to talk to client */
-	struct fio_mutex *lock;	/* protects ref and below list */
+	struct fio_mutex lock;	/* protects ref and below list */
 	struct flist_head list;	/* list of pending transmit work */
-	struct fio_mutex *wait;	/* wake backend when items added to list */
+	struct fio_mutex wait;	/* wake backend when items added to list */
+	struct fio_mutex xmit;	/* held while sending data */
 };
 
 static char *fio_server_arg;
@@ -116,12 +118,12 @@ static const char *fio_server_ops[FIO_NET_CMD_NR] = {
 
 static void sk_lock(struct sk_out *sk_out)
 {
-	fio_mutex_down(sk_out->lock);
+	fio_mutex_down(&sk_out->lock);
 }
 
 static void sk_unlock(struct sk_out *sk_out)
 {
-	fio_mutex_up(sk_out->lock);
+	fio_mutex_up(&sk_out->lock);
 }
 
 void sk_out_assign(struct sk_out *sk_out)
@@ -137,8 +139,9 @@ void sk_out_assign(struct sk_out *sk_out)
 
 static void sk_out_free(struct sk_out *sk_out)
 {
-	fio_mutex_remove(sk_out->lock);
-	fio_mutex_remove(sk_out->wait);
+	__fio_mutex_remove(&sk_out->lock);
+	__fio_mutex_remove(&sk_out->wait);
+	__fio_mutex_remove(&sk_out->xmit);
 	sfree(sk_out);
 }
 
@@ -517,8 +520,9 @@ int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size,
 	return ret;
 }
 
-static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf, off_t size,
-					 uint64_t *tagptr, int flags)
+static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf,
+					 size_t size, uint64_t *tagptr,
+					 int flags)
 {
 	struct sk_entry *entry;
 
@@ -530,22 +534,28 @@ static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf, off_t size,
 		memcpy(entry->buf, buf, size);
 	} else
 		entry->buf = buf;
+
 	entry->size = size;
 	entry->tagptr = tagptr;
 	entry->flags = flags;
-
 	return entry;
 }
 
+static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry);
+
 static void fio_net_queue_entry(struct sk_entry *entry)
 {
 	struct sk_out *sk_out = pthread_getspecific(sk_out_key);
 
-	sk_lock(sk_out);
-	flist_add_tail(&entry->list, &sk_out->list);
-	sk_unlock(sk_out);
+	if (entry->flags & SK_F_INLINE)
+		handle_sk_entry(sk_out, entry);
+	else {
+		sk_lock(sk_out);
+		flist_add_tail(&entry->list, &sk_out->list);
+		sk_unlock(sk_out);
 
-	fio_mutex_up(sk_out->wait);
+		fio_mutex_up(&sk_out->wait);
+	}
 }
 
 static int fio_net_queue_cmd(uint16_t opcode, void *buf, off_t size,
@@ -1102,9 +1112,11 @@ static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry)
 {
 	int ret;
 
+	fio_mutex_down(&sk_out->xmit);
+
 	if (entry->flags & SK_F_VEC)
 		ret = send_vec_entry(sk_out, entry);
-	if (entry->flags & SK_F_SIMPLE) {
+	else if (entry->flags & SK_F_SIMPLE) {
 		uint64_t tag = 0;
 
 		if (entry->tagptr)
@@ -1114,6 +1126,8 @@ static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry)
 	} else
 		ret = fio_net_send_cmd(sk_out->sk, entry->opcode, entry->buf, entry->size, entry->tagptr, NULL);
 
+	fio_mutex_up(&sk_out->xmit);
+
 	if (ret)
 		log_err("fio: failed handling cmd %s\n", fio_server_op(entry->opcode));
 
@@ -1177,7 +1191,7 @@ static int handle_connection(struct sk_out *sk_out)
 				break;
 			} else if (!ret) {
 				fio_server_check_jobs(&job_list);
-				fio_mutex_down_timeout(sk_out->wait, timeout);
+				fio_mutex_down_timeout(&sk_out->wait, timeout);
 				continue;
 			}
 
@@ -1323,8 +1337,9 @@ static int accept_loop(int listen_sk)
 		sk_out = smalloc(sizeof(*sk_out));
 		sk_out->sk = sk;
 		INIT_FLIST_HEAD(&sk_out->list);
-		sk_out->lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
-		sk_out->wait = fio_mutex_init(FIO_MUTEX_LOCKED);
+		__fio_mutex_init(&sk_out->lock, FIO_MUTEX_UNLOCKED);
+		__fio_mutex_init(&sk_out->wait, FIO_MUTEX_LOCKED);
+		__fio_mutex_init(&sk_out->xmit, FIO_MUTEX_UNLOCKED);
 
 		pid = fork();
 		if (pid) {
@@ -1512,6 +1527,47 @@ void fio_server_send_gs(struct group_run_stats *rs)
 	fio_net_queue_cmd(FIO_NET_CMD_GS, &gs, sizeof(gs), NULL, SK_F_COPY);
 }
 
+void fio_server_send_job_options(struct flist_head *opt_list,
+				 unsigned int groupid)
+{
+	struct cmd_job_option pdu;
+	struct flist_head *entry;
+
+	if (flist_empty(opt_list))
+		return;
+
+	flist_for_each(entry, opt_list) {
+		struct print_option *p;
+		size_t len;
+
+		p = flist_entry(entry, struct print_option, list);
+		memset(&pdu, 0, sizeof(pdu));
+
+		if (groupid == -1U) {
+			pdu.global = __cpu_to_le16(1);
+			pdu.groupid = 0;
+		} else {
+			pdu.global = 0;
+			pdu.groupid = cpu_to_le32(groupid);
+		}
+		len = strlen(p->name);
+		if (len >= sizeof(pdu.name)) {
+			len = sizeof(pdu.name) - 1;
+			pdu.truncated = __cpu_to_le16(1);
+		}
+		memcpy(pdu.name, p->name, len);
+		if (p->value) {
+			len = strlen(p->value);
+			if (len >= sizeof(pdu.value)) {
+				len = sizeof(pdu.value) - 1;
+				pdu.truncated = __cpu_to_le16(1);
+			}
+			memcpy(pdu.value, p->value, len);
+		}
+		fio_net_queue_cmd(FIO_NET_CMD_JOB_OPT, &pdu, sizeof(pdu), NULL, SK_F_COPY);
+	}
+}
+
 static void convert_agg(struct disk_util_agg *dst, struct disk_util_agg *src)
 {
 	int i;
@@ -1608,7 +1664,8 @@ static int fio_send_iolog_gz(struct sk_entry *first, struct io_log *log)
 		this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream.avail_out;
 
 		entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, out_pdu, this_len,
-						NULL, SK_F_FREE | SK_F_VEC);
+						NULL, SK_F_VEC | SK_F_INLINE | SK_F_FREE);
+		out_pdu = NULL;
 		flist_add_tail(&entry->list, &first->next);
 	} while (stream.avail_in);
 
@@ -1620,6 +1677,25 @@ err:
 	return ret;
 }
 
+static int fio_send_gz_chunks(struct sk_entry *first, struct io_log *log)
+{
+	struct sk_entry *entry;
+	struct flist_head *node;
+
+	pthread_mutex_lock(&log->chunk_lock);
+	flist_for_each(node, &log->chunk_list) {
+		struct iolog_compress *c;
+
+		c = flist_entry(node, struct iolog_compress, list);
+		entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, c->buf, c->len,
+						NULL, SK_F_VEC | SK_F_INLINE);
+		flist_add_tail(&entry->list, &first->next);
+	}
+	pthread_mutex_unlock(&log->chunk_lock);
+
+	return 0;
+}
+
 int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
 {
 	struct cmd_iolog_pdu pdu;
@@ -1629,11 +1705,21 @@ int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
 	pdu.nr_samples = cpu_to_le64(log->nr_samples);
 	pdu.thread_number = cpu_to_le32(td->thread_number);
 	pdu.log_type = cpu_to_le32(log->log_type);
-	pdu.compressed = cpu_to_le32(use_zlib);
+
+	if (!flist_empty(&log->chunk_list))
+		pdu.compressed = __cpu_to_le32(STORE_COMPRESSED);
+	else if (use_zlib)
+		pdu.compressed = __cpu_to_le32(XMIT_COMPRESSED);
+	else
+		pdu.compressed = 0;
 
 	strncpy((char *) pdu.name, name, FIO_NET_NAME_MAX);
 	pdu.name[FIO_NET_NAME_MAX - 1] = '\0';
 
+	/*
+	 * We can't do this for a pre-compressed log, but for that case,
+	 * log->nr_samples is zero anyway.
+	 */
 	for (i = 0; i < log->nr_samples; i++) {
 		struct io_sample *s = get_sample(log, i);
 
@@ -1652,23 +1738,26 @@ int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
 	/*
 	 * Assemble header entry first
 	 */
-	first = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, &pdu, sizeof(pdu), NULL, SK_F_COPY | SK_F_VEC);
+	first = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, &pdu, sizeof(pdu), NULL, SK_F_VEC | SK_F_INLINE | SK_F_COPY);
 
 	/*
 	 * Now append actual log entries. Compress if we can, otherwise just
 	 * plain text output.
 	 */
-	if (use_zlib)
+	if (!flist_empty(&log->chunk_list))
+		ret = fio_send_gz_chunks(first, log);
+	else if (use_zlib)
 		ret = fio_send_iolog_gz(first, log);
 	else {
 		struct sk_entry *entry;
+		size_t size = log->nr_samples * log_entry_sz(log);
 
-		entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, log->log,
-					log->nr_samples * log_entry_sz(log),
-					NULL, SK_F_FREE | SK_F_VEC);
+		entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, log->log, size,
+						NULL, SK_F_VEC | SK_F_INLINE);
 		flist_add_tail(&entry->list, &first->next);
 	}
 
+	fio_net_queue_entry(first);
 	return ret;
 }
 
@@ -1889,7 +1978,7 @@ static int fio_init_server_connection(void)
 
 	log_info("fio: server listening on %s\n", bind_str);
 
-	if (listen(sk, 0) < 0) {
+	if (listen(sk, 4) < 0) {
 		log_err("fio: listen: %s\n", strerror(errno));
 		close(sk);
 		return -1;