+static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
+{
+ dst->max_val = le64_to_cpu(src->max_val);
+ dst->min_val = le64_to_cpu(src->min_val);
+ dst->samples = le64_to_cpu(src->samples);
+
+ /*
+ * Floats arrive as IEEE 754 encoded uint64_t, convert back to double
+ */
+ dst->mean.u.f = fio_uint64_to_double(le64_to_cpu(dst->mean.u.i));
+ dst->S.u.f = fio_uint64_to_double(le64_to_cpu(dst->S.u.i));
+}
+
+static void convert_ts(struct thread_stat *dst, struct thread_stat *src)
+{
+ int i, j;
+
+ 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]);
+ convert_io_stat(&dst->slat_stat[i], &src->slat_stat[i]);
+ convert_io_stat(&dst->lat_stat[i], &src->lat_stat[i]);
+ convert_io_stat(&dst->bw_stat[i], &src->bw_stat[i]);
+ }
+
+ dst->usr_time = le64_to_cpu(src->usr_time);
+ dst->sys_time = le64_to_cpu(src->sys_time);
+ dst->ctx = le64_to_cpu(src->ctx);
+ dst->minf = le64_to_cpu(src->minf);
+ dst->majf = le64_to_cpu(src->majf);
+ dst->clat_percentiles = le64_to_cpu(src->clat_percentiles);
+
+ for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) {
+ fio_fp64_t *fps = &src->percentile_list[i];
+ fio_fp64_t *fpd = &dst->percentile_list[i];
+
+ fpd->u.f = fio_uint64_to_double(le64_to_cpu(fps->u.i));
+ }
+
+ for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
+ dst->io_u_map[i] = le32_to_cpu(src->io_u_map[i]);
+ dst->io_u_submit[i] = le32_to_cpu(src->io_u_submit[i]);
+ dst->io_u_complete[i] = le32_to_cpu(src->io_u_complete[i]);
+ }
+
+ for (i = 0; i < FIO_IO_U_LAT_U_NR; i++) {
+ dst->io_u_lat_u[i] = le32_to_cpu(src->io_u_lat_u[i]);
+ dst->io_u_lat_m[i] = le32_to_cpu(src->io_u_lat_m[i]);
+ }
+
+ for (i = 0; i < 2; i++)
+ for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
+ dst->io_u_plat[i][j] = le32_to_cpu(src->io_u_plat[i][j]);
+
+ for (i = 0; i < 3; i++) {
+ dst->total_io_u[i] = le64_to_cpu(src->total_io_u[i]);
+ dst->short_io_u[i] = le64_to_cpu(src->short_io_u[i]);
+ }
+
+ dst->total_submit = le64_to_cpu(src->total_submit);
+ dst->total_complete = le64_to_cpu(src->total_complete);
+
+ for (i = 0; i < 2; i++) {
+ dst->io_bytes[i] = le64_to_cpu(src->io_bytes[i]);
+ dst->runtime[i] = le64_to_cpu(src->runtime[i]);
+ }
+
+ dst->total_run_time = le64_to_cpu(src->total_run_time);
+ dst->continue_on_error = le16_to_cpu(src->continue_on_error);
+ dst->total_err_count = le64_to_cpu(src->total_err_count);
+ dst->first_error = le32_to_cpu(src->first_error);
+ dst->kb_base = le32_to_cpu(src->kb_base);
+}
+
+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] = le64_to_cpu(src->max_run[i]);
+ dst->min_run[i] = le64_to_cpu(src->min_run[i]);
+ dst->max_bw[i] = le64_to_cpu(src->max_bw[i]);
+ dst->min_bw[i] = le64_to_cpu(src->min_bw[i]);
+ dst->io_kb[i] = le64_to_cpu(src->io_kb[i]);
+ dst->agg[i] = le64_to_cpu(src->agg[i]);
+ }
+
+ dst->kb_base = le32_to_cpu(src->kb_base);
+ dst->groupid = le32_to_cpu(src->groupid);
+}
+
+static void handle_ts(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct cmd_ts_pdu *p = (struct cmd_ts_pdu *) cmd->payload;
+
+ show_thread_status(&p->ts, &p->rs);
+
+ if (sum_stat_clients == 1)
+ return;
+
+ sum_thread_stats(&client_ts, &p->ts, sum_stat_nr);
+ 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) {
+ strcpy(client_ts.name, "All clients");
+ show_thread_status(&client_ts, &client_gs);
+ }
+}
+
+static void handle_gs(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct group_run_stats *gs = (struct group_run_stats *) cmd->payload;
+
+ show_group_stats(gs);
+}
+
+static void handle_text(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmd->payload;
+ const char *buf = (const char *) pdu->buf;
+ const char *name;
+ int fio_unused ret;
+
+ name = client->name ? client->name : client->hostname;
+
+ if (!client->skip_newline)
+ fprintf(f_out, "<%s> ", name);
+ ret = fwrite(buf, pdu->buf_len, 1, f_out);
+ fflush(f_out);
+ client->skip_newline = strchr(buf, '\n') == NULL;
+}
+
+static void convert_agg(struct disk_util_agg *agg)
+{
+ int i;
+
+ for (i = 0; i < 2; i++) {
+ agg->ios[i] = le32_to_cpu(agg->ios[i]);
+ agg->merges[i] = le32_to_cpu(agg->merges[i]);
+ agg->sectors[i] = le64_to_cpu(agg->sectors[i]);
+ agg->ticks[i] = le32_to_cpu(agg->ticks[i]);
+ }
+
+ agg->io_ticks = le32_to_cpu(agg->io_ticks);
+ agg->time_in_queue = le32_to_cpu(agg->time_in_queue);
+ agg->slavecount = le32_to_cpu(agg->slavecount);
+ agg->max_util.u.f = fio_uint64_to_double(__le64_to_cpu(agg->max_util.u.i));
+}
+
+static void convert_dus(struct disk_util_stat *dus)
+{
+ int i;
+
+ for (i = 0; i < 2; i++) {
+ dus->ios[i] = le32_to_cpu(dus->ios[i]);
+ dus->merges[i] = le32_to_cpu(dus->merges[i]);
+ dus->sectors[i] = le64_to_cpu(dus->sectors[i]);
+ dus->ticks[i] = le32_to_cpu(dus->ticks[i]);
+ }
+
+ dus->io_ticks = le32_to_cpu(dus->io_ticks);
+ dus->time_in_queue = le32_to_cpu(dus->time_in_queue);
+ dus->msec = le64_to_cpu(dus->msec);
+}
+
+static void handle_du(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct cmd_du_pdu *du = (struct cmd_du_pdu *) cmd->payload;
+
+ if (!client->disk_stats_shown) {
+ client->disk_stats_shown = 1;
+ log_info("\nDisk stats (read/write):\n");
+ }
+
+ print_disk_util(&du->dus, &du->agg, terse_output);
+}
+
+static void convert_jobs_eta(struct jobs_eta *je)
+{
+ int i;
+
+ je->nr_running = le32_to_cpu(je->nr_running);
+ je->nr_ramp = le32_to_cpu(je->nr_ramp);
+ je->nr_pending = le32_to_cpu(je->nr_pending);
+ je->files_open = le32_to_cpu(je->files_open);
+
+ for (i = 0; i < 2; i++) {
+ je->m_rate[i] = le32_to_cpu(je->m_rate[i]);
+ je->t_rate[i] = le32_to_cpu(je->t_rate[i]);
+ je->m_iops[i] = le32_to_cpu(je->m_iops[i]);
+ je->t_iops[i] = le32_to_cpu(je->t_iops[i]);
+ je->rate[i] = le32_to_cpu(je->rate[i]);
+ je->iops[i] = le32_to_cpu(je->iops[i]);
+ }
+
+ je->elapsed_sec = le64_to_cpu(je->elapsed_sec);
+ je->eta_sec = le64_to_cpu(je->eta_sec);
+ je->nr_threads = le32_to_cpu(je->nr_threads);
+}
+
+void fio_client_sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je)
+{
+ int i;
+
+ dst->nr_running += je->nr_running;
+ dst->nr_ramp += je->nr_ramp;
+ dst->nr_pending += je->nr_pending;
+ dst->files_open += je->files_open;
+
+ for (i = 0; i < 2; i++) {
+ dst->m_rate[i] += je->m_rate[i];
+ dst->t_rate[i] += je->t_rate[i];
+ dst->m_iops[i] += je->m_iops[i];
+ dst->t_iops[i] += je->t_iops[i];
+ dst->rate[i] += je->rate[i];
+ dst->iops[i] += je->iops[i];
+ }
+
+ dst->elapsed_sec += je->elapsed_sec;
+
+ if (je->eta_sec > dst->eta_sec)
+ dst->eta_sec = je->eta_sec;
+
+ dst->nr_threads += je->nr_threads;
+ /* we need to handle je->run_str too ... */
+}
+
+void fio_client_dec_jobs_eta(struct client_eta *eta, client_eta_op eta_fn)
+{
+ if (!--eta->pending) {
+ eta_fn(&eta->eta);
+ free(eta);
+ }
+}
+
+static void remove_reply_cmd(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct fio_net_int_cmd *icmd = NULL;
+ struct flist_head *entry;
+
+ flist_for_each(entry, &client->cmd_list) {
+ icmd = flist_entry(entry, struct fio_net_int_cmd, list);
+
+ if (cmd->tag == (uintptr_t) icmd)
+ break;
+
+ icmd = NULL;
+ }
+
+ if (!icmd) {
+ log_err("fio: client: unable to find matching tag\n");
+ return;
+ }
+
+ flist_del(&icmd->list);
+ cmd->tag = icmd->saved_tag;
+ free(icmd);
+}
+
+static void handle_eta(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct jobs_eta *je = (struct jobs_eta *) cmd->payload;
+ struct client_eta *eta = (struct client_eta *) (uintptr_t) cmd->tag;
+
+ dprint(FD_NET, "client: got eta tag %p, %d\n", eta, eta->pending);
+
+ assert(client->eta_in_flight == eta);
+
+ client->eta_in_flight = NULL;
+ flist_del_init(&client->eta_list);
+
+ if (client->ops->jobs_eta)
+ client->ops->jobs_eta(client, je);
+
+ fio_client_sum_jobs_eta(&eta->eta, je);
+ fio_client_dec_jobs_eta(eta, client->ops->eta);
+}
+
+static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct cmd_probe_pdu *probe = (struct cmd_probe_pdu *) cmd->payload;
+ const char *os, *arch;
+ char bit[16];
+
+ os = fio_get_os_string(probe->os);
+ if (!os)
+ os = "unknown";
+
+ arch = fio_get_arch_string(probe->arch);
+ if (!arch)
+ os = "unknown";
+
+ sprintf(bit, "%d-bit", probe->bpp * 8);
+
+ log_info("hostname=%s, be=%u, %s, os=%s, arch=%s, fio=%u.%u.%u\n",
+ probe->hostname, probe->bigendian, bit, os, arch,
+ probe->fio_major, probe->fio_minor, probe->fio_patch);
+
+ if (!client->name)
+ client->name = strdup((char *) probe->hostname);
+}
+
+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 = pdu->jobs;
+}
+
+static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ if (client->error)
+ log_info("client <%s>: exited with error %d\n", client->hostname, client->error);
+}
+
+static void convert_stop(struct fio_net_cmd *cmd)
+{
+ struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload;
+
+ pdu->error = le32_to_cpu(pdu->error);
+}
+
+static void convert_text(struct fio_net_cmd *cmd)
+{
+ struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmd->payload;
+
+ pdu->level = le32_to_cpu(pdu->level);
+ pdu->buf_len = le32_to_cpu(pdu->buf_len);
+ pdu->log_sec = le64_to_cpu(pdu->log_sec);
+ 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)