X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=server.c;h=2fd9b45afb0e60403333440e26555faf270b8846;hp=a5415bacb0b1c50bc45aa51ec56fe339bde1aacc;hb=9d9937b51dc6a8fa3ee12349ac1ce290e0429b9d;hpb=4d27d42967a582183788580d03d8394e60379207

diff --git a/server.c b/server.c
index a5415bac..2fd9b45a 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 {
@@ -45,17 +46,19 @@ struct sk_entry {
 	int opcode;		/* Actual command fields */
 	void *buf;
 	off_t size;
-	uint64_t *tagptr;
+	uint64_t tag;
 	struct flist_head next;	/* Other sk_entry's, if linked command */
 };
 
 struct sk_out {
-	unsigned int refs;
-
-	int sk;
-	struct fio_mutex *lock;
-	struct flist_head list;
-	struct fio_mutex *wait;
+	unsigned int refs;	/* frees sk_out when it drops to zero.
+				 * protected by below ->lock */
+
+	int sk;			/* socket fd to talk to client */
+	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 xmit;	/* held while sending data */
 };
 
 static char *fio_server_arg;
@@ -111,16 +114,17 @@ static const char *fio_server_ops[FIO_NET_CMD_NR] = {
 	"LOAD_FILE",
 	"VTRIGGER",
 	"SENDFILE",
+	"JOB_OPT",
 };
 
 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)
@@ -134,29 +138,61 @@ void sk_out_assign(struct sk_out *sk_out)
 	pthread_setspecific(sk_out_key, sk_out);
 }
 
-static void __sk_out_drop(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);
 }
 
-void sk_out_drop(void)
+static int __sk_out_drop(struct sk_out *sk_out)
 {
-	struct sk_out *sk_out = pthread_getspecific(sk_out_key);
-
 	if (sk_out) {
 		int refs;
 
 		sk_lock(sk_out);
+		assert(sk_out->refs != 0);
 		refs = --sk_out->refs;
 		sk_unlock(sk_out);
 
-		if (!refs)
-			__sk_out_drop(sk_out);
-
-		pthread_setspecific(sk_out_key, NULL);
+		if (!refs) {
+			sk_out_free(sk_out);
+			pthread_setspecific(sk_out_key, NULL);
+			return 0;
+		}
 	}
+
+	return 1;
+}
+
+void sk_out_drop(void)
+{
+	struct sk_out *sk_out;
+
+	sk_out = pthread_getspecific(sk_out_key);
+	__sk_out_drop(sk_out);
+}
+
+static void __fio_init_net_cmd(struct fio_net_cmd *cmd, uint16_t opcode,
+			       uint32_t pdu_len, uint64_t tag)
+{
+	memset(cmd, 0, sizeof(*cmd));
+
+	cmd->version	= __cpu_to_le16(FIO_SERVER_VER);
+	cmd->opcode	= cpu_to_le16(opcode);
+	cmd->tag	= cpu_to_le64(tag);
+	cmd->pdu_len	= cpu_to_le32(pdu_len);
+}
+
+
+static void fio_init_net_cmd(struct fio_net_cmd *cmd, uint16_t opcode,
+			     const void *pdu, uint32_t pdu_len, uint64_t tag)
+{
+	__fio_init_net_cmd(cmd, opcode, pdu_len, tag);
+
+	if (pdu)
+		memcpy(&cmd->payload, pdu, pdu_len);
 }
 
 const char *fio_server_op(unsigned int op)
@@ -218,7 +254,7 @@ static int fio_sendv_data(int sk, struct iovec *iov, int count)
 	return 1;
 }
 
-int fio_send_data(int sk, const void *p, unsigned int len)
+static int fio_send_data(int sk, const void *p, unsigned int len)
 {
 	struct iovec iov = { .iov_base = (void *) p, .iov_len = len };
 
@@ -227,10 +263,17 @@ int fio_send_data(int sk, const void *p, unsigned int len)
 	return fio_sendv_data(sk, &iov, 1);
 }
 
-int fio_recv_data(int sk, void *p, unsigned int len)
+static int fio_recv_data(int sk, void *p, unsigned int len, bool wait)
 {
+	int flags;
+
+	if (wait)
+		flags = MSG_WAITALL;
+	else
+		flags = OS_MSG_DONTWAIT;
+
 	do {
-		int ret = recv(sk, p, len, MSG_WAITALL);
+		int ret = recv(sk, p, len, flags);
 
 		if (ret > 0) {
 			len -= ret;
@@ -240,9 +283,11 @@ int fio_recv_data(int sk, void *p, unsigned int len)
 			continue;
 		} else if (!ret)
 			break;
-		else if (errno == EAGAIN || errno == EINTR)
-			continue;
-		else
+		else if (errno == EAGAIN || errno == EINTR) {
+			if (wait)
+				continue;
+			break;
+		} else
 			break;
 	} while (!exit_backend);
 
@@ -291,7 +336,7 @@ static int verify_convert_cmd(struct fio_net_cmd *cmd)
 /*
  * Read (and defragment, if necessary) incoming commands
  */
-struct fio_net_cmd *fio_net_recv_cmd(int sk)
+struct fio_net_cmd *fio_net_recv_cmd(int sk, bool wait)
 {
 	struct fio_net_cmd cmd, *tmp, *cmdret = NULL;
 	size_t cmd_size = 0, pdu_offset = 0;
@@ -300,7 +345,7 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk)
 	void *pdu = NULL;
 
 	do {
-		ret = fio_recv_data(sk, &cmd, sizeof(cmd));
+		ret = fio_recv_data(sk, &cmd, sizeof(cmd), wait);
 		if (ret)
 			break;
 
@@ -344,7 +389,7 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk)
 
 		/* There's payload, get it */
 		pdu = (void *) cmdret->payload + pdu_offset;
-		ret = fio_recv_data(sk, pdu, cmd.pdu_len);
+		ret = fio_recv_data(sk, pdu, cmd.pdu_len, wait);
 		if (ret)
 			break;
 
@@ -419,7 +464,7 @@ static void free_reply(uint64_t tag)
 	free(reply);
 }
 
-void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, const void *pdu)
+static void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, const void *pdu)
 {
 	uint32_t pdu_len;
 
@@ -429,7 +474,7 @@ void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, const void *pdu)
 	cmd->pdu_crc16 = __cpu_to_le16(fio_crc16(pdu, pdu_len));
 }
 
-void fio_net_cmd_crc(struct fio_net_cmd *cmd)
+static void fio_net_cmd_crc(struct fio_net_cmd *cmd)
 {
 	fio_net_cmd_crc_pdu(cmd, cmd->payload);
 }
@@ -486,8 +531,9 @@ int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size,
 	return ret;
 }
 
-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;
 
@@ -499,22 +545,31 @@ 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;
+	if (tagptr)
+		entry->tag = *tagptr;
+	else
+		entry->tag = 0;
 	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,
@@ -713,6 +768,8 @@ static int handle_run_cmd(struct sk_out *sk_out, struct flist_head *job_list,
 	pid_t pid;
 	int ret;
 
+	sk_out_assign(sk_out);
+
 	fio_time_init();
 	set_genesis_time();
 
@@ -724,6 +781,7 @@ static int handle_run_cmd(struct sk_out *sk_out, struct flist_head *job_list,
 
 	ret = fio_backend(sk_out);
 	free_threads_shm();
+	sk_out_drop();
 	_exit(ret);
 }
 
@@ -914,9 +972,9 @@ static int handle_trigger_cmd(struct fio_net_cmd *cmd)
 		struct all_io_list state;
 
 		state.threads = cpu_to_le64((uint64_t) 0);
-		fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, &state, sizeof(state), NULL, SK_F_COPY);
+		fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, &state, sizeof(state), NULL, SK_F_COPY | SK_F_INLINE);
 	} else
-		fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, rep, sz, NULL, SK_F_FREE);
+		fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, rep, sz, NULL, SK_F_FREE | SK_F_INLINE);
 
 	exec_trigger(buf);
 	return 0;
@@ -1005,17 +1063,35 @@ static int fio_send_cmd_ext_pdu(int sk, uint16_t opcode, const void *buf,
 {
 	struct fio_net_cmd cmd;
 	struct iovec iov[2];
+	size_t this_len;
+	int ret;
 
 	iov[0].iov_base = (void *) &cmd;
 	iov[0].iov_len = sizeof(cmd);
-	iov[1].iov_base = (void *) buf;
-	iov[1].iov_len = size;
 
-	__fio_init_net_cmd(&cmd, opcode, size, tag);
-	cmd.flags = __cpu_to_le32(flags);
-	fio_net_cmd_crc_pdu(&cmd, buf);
+	do {
+		uint32_t this_flags = flags;
+
+		this_len = size;
+		if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
+			this_len = FIO_SERVER_MAX_FRAGMENT_PDU;
+
+		if (this_len < size)
+			this_flags |= FIO_NET_CMD_F_MORE;
+
+		__fio_init_net_cmd(&cmd, opcode, this_len, tag);
+		cmd.flags = __cpu_to_le32(this_flags);
+		fio_net_cmd_crc_pdu(&cmd, buf);
+
+		iov[1].iov_base = (void *) buf;
+		iov[1].iov_len = this_len;
 
-	return fio_sendv_data(sk, iov, 2);
+		ret = fio_sendv_data(sk, iov, 2);
+		size -= this_len;
+		buf += this_len;
+	} while (!ret && size);
+
+	return ret;
 }
 
 static void finish_entry(struct sk_entry *entry)
@@ -1028,29 +1104,24 @@ static void finish_entry(struct sk_entry *entry)
 	sfree(entry);
 }
 
-static void entry_set_flags_tag(struct sk_entry *entry, struct flist_head *list,
-				unsigned int *flags, uint64_t *tag)
+static void entry_set_flags(struct sk_entry *entry, struct flist_head *list,
+			    unsigned int *flags)
 {
 	if (!flist_empty(list))
 		*flags = FIO_NET_CMD_F_MORE;
 	else
 		*flags = 0;
-
-	if (entry->tagptr)
-		*tag = *entry->tagptr;
-	else
-		*tag = 0;
 }
 
 static int send_vec_entry(struct sk_out *sk_out, struct sk_entry *first)
 {
 	unsigned int flags;
-	uint64_t tag;
 	int ret;
 
-	entry_set_flags_tag(first, &first->next, &flags, &tag);
+	entry_set_flags(first, &first->next, &flags);
 
-	ret = fio_send_cmd_ext_pdu(sk_out->sk, first->opcode, first->buf, first->size, tag, flags);
+	ret = fio_send_cmd_ext_pdu(sk_out->sk, first->opcode, first->buf,
+					first->size, first->tag, flags);
 
 	while (!flist_empty(&first->next)) {
 		struct sk_entry *next;
@@ -1058,9 +1129,10 @@ static int send_vec_entry(struct sk_out *sk_out, struct sk_entry *first)
 		next = flist_first_entry(&first->next, struct sk_entry, list);
 		flist_del_init(&next->list);
 
-		entry_set_flags_tag(next, &first->next, &flags, &tag);
+		entry_set_flags(next, &first->next, &flags);
 
-		ret += fio_send_cmd_ext_pdu(sk_out->sk, next->opcode, next->buf, next->size, tag, flags);
+		ret += fio_send_cmd_ext_pdu(sk_out->sk, next->opcode, next->buf,
+						next->size, next->tag, flags);
 		finish_entry(next);
 	}
 
@@ -1071,17 +1143,19 @@ 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) {
-		uint64_t tag = 0;
-
-		if (entry->tagptr)
-			tag = *entry->tagptr;
+	else if (entry->flags & SK_F_SIMPLE) {
+		ret = fio_net_send_simple_cmd(sk_out->sk, entry->opcode,
+						entry->tag, NULL);
+	} else {
+		ret = fio_net_send_cmd(sk_out->sk, entry->opcode, entry->buf,
+					entry->size, &entry->tag, NULL);
+	}
 
-		ret = fio_net_send_simple_cmd(sk_out->sk, entry->opcode, tag, NULL);
-	} 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));
@@ -1146,7 +1220,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;
 			}
 
@@ -1163,7 +1237,7 @@ static int handle_connection(struct sk_out *sk_out)
 		if (ret < 0)
 			break;
 
-		cmd = fio_net_recv_cmd(sk_out->sk);
+		cmd = fio_net_recv_cmd(sk_out->sk, true);
 		if (!cmd) {
 			ret = -1;
 			break;
@@ -1183,13 +1257,15 @@ static int handle_connection(struct sk_out *sk_out)
 	handle_xmits(sk_out);
 
 	close(sk_out->sk);
+	sk_out->sk = -1;
+	__sk_out_drop(sk_out);
 	_exit(ret);
 }
 
 /* get the address on this host bound by the input socket, 
  * whether it is ipv6 or ipv4 */
 
-int get_my_addr_str(int sk)
+static int get_my_addr_str(int sk)
 {
 	struct sockaddr_in6 myaddr6 = { 0, };
 	struct sockaddr_in myaddr4 = { 0, };
@@ -1223,7 +1299,7 @@ int get_my_addr_str(int sk)
 	return 0;
 }
 
-static int accept_loop(struct sk_out *sk_out, int listen_sk)
+static int accept_loop(int listen_sk)
 {
 	struct sockaddr_in addr;
 	struct sockaddr_in6 addr6;
@@ -1237,6 +1313,7 @@ static int accept_loop(struct sk_out *sk_out, int listen_sk)
 	fio_set_fd_nonblocking(listen_sk, "server");
 
 	while (!exit_backend) {
+		struct sk_out *sk_out;
 		const char *from;
 		char buf[64];
 		pid_t pid;
@@ -1286,18 +1363,28 @@ static int accept_loop(struct sk_out *sk_out, int listen_sk)
 
 		dprint(FD_NET, "server: connect from %s\n", from);
 
+		sk_out = smalloc(sizeof(*sk_out));
 		sk_out->sk = sk;
+		INIT_FLIST_HEAD(&sk_out->list);
+		__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) {
 			close(sk);
 			fio_server_add_conn_pid(&conn_list, pid);
-			pthread_setspecific(sk_out_key, sk_out);
 			continue;
 		}
 
-		/* exits */
-		get_my_addr_str(sk); /* if error, it's already logged, non-fatal */
+		/* if error, it's already logged, non-fatal */
+		get_my_addr_str(sk);
+
+		/*
+		 * Assign sk_out here, it'll be dropped in handle_connection()
+		 * since that function calls _exit() when done
+		 */
+		sk_out_assign(sk_out);
 		handle_connection(sk_out);
 	}
 
@@ -1469,6 +1556,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;
@@ -1525,107 +1653,214 @@ void fio_server_send_du(void)
 	}
 }
 
-static int fio_send_iolog_gz(struct sk_entry *first, struct io_log *log)
-{
-	int ret = 0;
 #ifdef CONFIG_ZLIB
-	struct sk_entry *entry;
-	z_stream stream;
+static int __fio_append_iolog_gz(struct sk_entry *first, struct io_log *log,
+				 struct io_logs *cur_log, z_stream *stream)
+{
+	unsigned int this_len;
 	void *out_pdu;
+	int ret;
 
-	/*
-	 * Dirty - since the log is potentially huge, compress it into
-	 * FIO_SERVER_MAX_FRAGMENT_PDU chunks and let the receiving
-	 * side defragment it.
-	 */
-	out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
+	stream->next_in = (void *) cur_log->log;
+	stream->avail_in = cur_log->nr_samples * log_entry_sz(log);
+
+	do {
+		struct sk_entry *entry;
+
+		/*
+		 * Dirty - since the log is potentially huge, compress it into
+		 * FIO_SERVER_MAX_FRAGMENT_PDU chunks and let the receiving
+		 * side defragment it.
+		 */
+		out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
 
+		stream->avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
+		stream->next_out = out_pdu;
+		ret = deflate(stream, Z_BLOCK);
+		/* may be Z_OK, or Z_STREAM_END */
+		if (ret < 0) {
+			free(out_pdu);
+			return 1;
+		}
+
+		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_VEC | SK_F_INLINE | SK_F_FREE);
+		flist_add_tail(&entry->list, &first->next);
+	} while (stream->avail_in);
+
+	return 0;
+}
+
+static int fio_append_iolog_gz(struct sk_entry *first, struct io_log *log)
+{
+	int ret = 0;
+	z_stream stream;
+
+	memset(&stream, 0, sizeof(stream));
 	stream.zalloc = Z_NULL;
 	stream.zfree = Z_NULL;
 	stream.opaque = Z_NULL;
 
-	if (deflateInit(&stream, Z_DEFAULT_COMPRESSION) != Z_OK) {
-		ret = 1;
-		goto err;
+	if (deflateInit(&stream, Z_DEFAULT_COMPRESSION) != Z_OK)
+		return 1;
+
+	while (!flist_empty(&log->io_logs)) {
+		struct io_logs *cur_log;
+
+		cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
+		flist_del_init(&cur_log->list);
+
+		ret = __fio_append_iolog_gz(first, log, cur_log, &stream);
+		if (ret)
+			break;
 	}
 
-	stream.next_in = (void *) log->log;
-	stream.avail_in = log->nr_samples * log_entry_sz(log);
+	ret = deflate(&stream, Z_FINISH);
 
-	do {
+	while (ret != Z_STREAM_END) {
+		struct sk_entry *entry;
 		unsigned int this_len;
+		void *out_pdu;
 
+		out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
 		stream.avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
 		stream.next_out = out_pdu;
+
 		ret = deflate(&stream, Z_FINISH);
 		/* may be Z_OK, or Z_STREAM_END */
-		if (ret < 0)
-			goto err_zlib;
+		if (ret < 0) {
+			free(out_pdu);
+			break;
+		}
 
 		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);
 		flist_add_tail(&entry->list, &first->next);
-	} while (stream.avail_in);
+	} while (ret != Z_STREAM_END);
 
-err_zlib:
-	deflateEnd(&stream);
-err:
-	free(out_pdu);
+	ret = deflateEnd(&stream);
+	if (ret == Z_OK)
+		return 0;
+
+	return 1;
+}
+#else
+static int fio_append_iolog_gz(struct sk_entry *first, struct io_log *log)
+{
+	return 1;
+}
 #endif
-	return ret;
+
+static int fio_append_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;
+}
+
+static int fio_append_text_log(struct sk_entry *first, struct io_log *log)
+{
+	struct sk_entry *entry;
+
+	while (!flist_empty(&log->io_logs)) {
+		struct io_logs *cur_log;
+		size_t size;
+
+		cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
+		flist_del_init(&cur_log->list);
+
+		size = cur_log->nr_samples * log_entry_sz(log);
+
+		entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, cur_log->log, size,
+						NULL, SK_F_VEC | SK_F_INLINE);
+		flist_add_tail(&entry->list, &first->next);
+	}
+
+	return 0;
 }
 
 int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
 {
 	struct cmd_iolog_pdu pdu;
 	struct sk_entry *first;
-	int i, ret = 0;
+	struct flist_head *entry;
+	int ret = 0;
 
-	pdu.nr_samples = cpu_to_le64(log->nr_samples);
+	pdu.nr_samples = cpu_to_le64(iolog_nr_samples(log));
 	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';
 
-	for (i = 0; i < log->nr_samples; i++) {
-		struct io_sample *s = get_sample(log, i);
+	/*
+	 * We can't do this for a pre-compressed log, but for that case,
+	 * log->nr_samples is zero anyway.
+	 */
+	flist_for_each(entry, &log->io_logs) {
+		struct io_logs *cur_log;
+		int i;
+
+		cur_log = flist_entry(entry, struct io_logs, list);
+
+		for (i = 0; i < cur_log->nr_samples; i++) {
+			struct io_sample *s = get_sample(log, cur_log, i);
 
-		s->time		= cpu_to_le64(s->time);
-		s->val		= cpu_to_le64(s->val);
-		s->__ddir	= cpu_to_le32(s->__ddir);
-		s->bs		= cpu_to_le32(s->bs);
+			s->time		= cpu_to_le64(s->time);
+			s->val		= cpu_to_le64(s->val);
+			s->__ddir	= cpu_to_le32(s->__ddir);
+			s->bs		= cpu_to_le32(s->bs);
 
-		if (log->log_offset) {
-			struct io_sample_offset *so = (void *) s;
+			if (log->log_offset) {
+				struct io_sample_offset *so = (void *) s;
 
-			so->offset = cpu_to_le64(so->offset);
+				so->offset = cpu_to_le64(so->offset);
+			}
 		}
 	}
 
 	/*
 	 * 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.
+	 * Now append actual log entries. If log compression was enabled on
+	 * the job, just send out the compressed chunks directly. If we
+	 * have a plain log, compress if we can, then send. Otherwise, send
+	 * the plain text output.
 	 */
-	if (use_zlib)
-		ret = fio_send_iolog_gz(first, log);
-	else {
-		struct sk_entry *entry;
-
-		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);
-		flist_add_tail(&entry->list, &first->next);
-	}
+	if (!flist_empty(&log->chunk_list))
+		ret = fio_append_gz_chunks(first, log);
+	else if (use_zlib)
+		ret = fio_append_iolog_gz(first, log);
+	else
+		ret = fio_append_text_log(first, log);
 
+	fio_net_queue_entry(first);
 	return ret;
 }
 
@@ -1638,7 +1873,8 @@ void fio_server_send_add_job(struct thread_data *td)
 	pdu.groupid = cpu_to_le32(td->groupid);
 	convert_thread_options_to_net(&pdu.top, &td->o);
 
-	fio_net_queue_cmd(FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), NULL, SK_F_COPY);
+	fio_net_queue_cmd(FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), NULL,
+				SK_F_COPY);
 }
 
 void fio_server_send_start(struct thread_data *td)
@@ -1651,20 +1887,21 @@ void fio_server_send_start(struct thread_data *td)
 }
 
 int fio_server_get_verify_state(const char *name, int threadnumber,
-				void **datap, int *version)
+				void **datap)
 {
 	struct thread_io_list *s;
 	struct cmd_sendfile out;
 	struct cmd_reply *rep;
 	uint64_t tag;
 	void *data;
+	int ret;
 
 	dprint(FD_NET, "server: request verify state\n");
 
 	rep = smalloc(sizeof(*rep));
 	if (!rep) {
 		log_err("fio: smalloc pool too small\n");
-		return 1;
+		return ENOMEM;
 	}
 
 	__fio_mutex_init(&rep->lock, FIO_MUTEX_LOCKED);
@@ -1674,23 +1911,27 @@ int fio_server_get_verify_state(const char *name, int threadnumber,
 	verify_state_gen_name((char *) out.path, sizeof(out.path), name, me,
 				threadnumber);
 	tag = (uint64_t) (uintptr_t) rep;
-	fio_net_queue_cmd(FIO_NET_CMD_SENDFILE, &out, sizeof(out), &tag, SK_F_COPY);
+	fio_net_queue_cmd(FIO_NET_CMD_SENDFILE, &out, sizeof(out), &tag,
+				SK_F_COPY);
 
 	/*
 	 * Wait for the backend to receive the reply
 	 */
 	if (fio_mutex_down_timeout(&rep->lock, 10000)) {
 		log_err("fio: timed out waiting for reply\n");
+		ret = ETIMEDOUT;
 		goto fail;
 	}
 
 	if (rep->error) {
-		log_err("fio: failure on receiving state file: %s\n", strerror(rep->error));
+		log_err("fio: failure on receiving state file %s: %s\n",
+				out.path, strerror(rep->error));
+		ret = rep->error;
 fail:
 		*datap = NULL;
 		sfree(rep);
 		fio_net_queue_quit();
-		return 1;
+		return ret;
 	}
 
 	/*
@@ -1698,12 +1939,15 @@ fail:
 	 * the header, and the thread_io_list checksum
 	 */
 	s = rep->data + sizeof(struct verify_state_hdr);
-	if (verify_state_hdr(rep->data, s, version))
+	if (verify_state_hdr(rep->data, s)) {
+		ret = EILSEQ;
 		goto fail;
+	}
 
 	/*
 	 * Don't need the header from now, copy just the thread_io_list
 	 */
+	ret = 0;
 	rep->size -= sizeof(struct verify_state_hdr);
 	data = malloc(rep->size);
 	memcpy(data, s, rep->size);
@@ -1712,7 +1956,7 @@ fail:
 	sfree(rep->data);
 	__fio_mutex_remove(&rep->lock);
 	sfree(rep);
-	return 0;
+	return ret;
 }
 
 static int fio_init_server_ip(void)
@@ -1740,11 +1984,10 @@ static int fio_init_server_ip(void)
 		return -1;
 	}
 #ifdef SO_REUSEPORT
-	if (setsockopt(sk, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
-		log_err("fio: setsockopt(REUSEPORT): %s\n", strerror(errno));
-		close(sk);
-		return -1;
-	}
+	/*
+	 * Not fatal if fails, so just ignore it if that happens
+	 */
+	setsockopt(sk, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
 #endif
 
 	if (use_ipv6) {
@@ -1846,7 +2089,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;
@@ -2037,9 +2280,24 @@ static void set_sig_handlers(void)
 	sigaction(SIGINT, &act, NULL);
 }
 
+void fio_server_destroy_sk_key(void)
+{
+	pthread_key_delete(sk_out_key);
+}
+
+int fio_server_create_sk_key(void)
+{
+	if (pthread_key_create(&sk_out_key, NULL)) {
+		log_err("fio: can't create sk_out backend key\n");
+		return 1;
+	}
+
+	pthread_setspecific(sk_out_key, NULL);
+	return 0;
+}
+
 static int fio_server(void)
 {
-	struct sk_out *sk_out;
 	int sk, ret;
 
 	dprint(FD_NET, "starting server\n");
@@ -2053,17 +2311,7 @@ static int fio_server(void)
 
 	set_sig_handlers();
 
-	if (pthread_key_create(&sk_out_key, NULL))
-		log_err("fio: can't create sk_out backend key\n");
-
-	sk_out = smalloc(sizeof(*sk_out));
-	INIT_FLIST_HEAD(&sk_out->list);
-	sk_out->lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
-	sk_out->wait = fio_mutex_init(FIO_MUTEX_LOCKED);
-
-	sk_out_assign(sk_out);
-
-	ret = accept_loop(sk_out, sk);
+	ret = accept_loop(sk);
 
 	close(sk);
 
@@ -2074,8 +2322,6 @@ static int fio_server(void)
 	if (bind_sock)
 		free(bind_sock);
 
-	sk_out_drop();
-
 	return ret;
 }