+ if (p->value)
+ pos = p->value;
+
+ json_object_add_value_string(job_opt_object, p->name, pos);
+ } else if (client->opt_lists) {
+ struct flist_head *opt_list = &client->opt_lists[pdu->groupid];
+
+ flist_add_tail(&p->list, opt_list);
+ }
+}
+
+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] = le64_to_cpu(agg->ios[i]);
+ agg->merges[i] = le64_to_cpu(agg->merges[i]);
+ agg->sectors[i] = le64_to_cpu(agg->sectors[i]);
+ agg->ticks[i] = le64_to_cpu(agg->ticks[i]);
+ }
+
+ agg->io_ticks = le64_to_cpu(agg->io_ticks);
+ agg->time_in_queue = le64_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->s.ios[i] = le64_to_cpu(dus->s.ios[i]);
+ dus->s.merges[i] = le64_to_cpu(dus->s.merges[i]);
+ dus->s.sectors[i] = le64_to_cpu(dus->s.sectors[i]);
+ dus->s.ticks[i] = le64_to_cpu(dus->s.ticks[i]);
+ }
+
+ dus->s.io_ticks = le64_to_cpu(dus->s.io_ticks);
+ dus->s.time_in_queue = le64_to_cpu(dus->s.time_in_queue);
+ dus->s.msec = le64_to_cpu(dus->s.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");
+ }
+
+ if (output_format & FIO_OUTPUT_JSON) {
+ struct json_object *duobj;
+ json_array_add_disk_util(&du->dus, &du->agg, du_array);
+ duobj = json_array_last_value_object(du_array);
+ json_object_add_client_info(duobj, client);
+ }
+ if (output_format & FIO_OUTPUT_TERSE)
+ print_disk_util(&du->dus, &du->agg, 1, NULL);
+ if (output_format & FIO_OUTPUT_NORMAL)
+ print_disk_util(&du->dus, &du->agg, 0, NULL);
+}
+
+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->nr_setting_up = le32_to_cpu(je->nr_setting_up);
+ je->files_open = le32_to_cpu(je->files_open);
+
+ for (i = 0; i < DDIR_RWDIR_CNT; 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);
+ je->is_pow2 = le32_to_cpu(je->is_pow2);
+ je->unit_base = le32_to_cpu(je->unit_base);
+}
+
+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->nr_setting_up += je->nr_setting_up;
+ dst->files_open += je->files_open;
+
+ for (i = 0; i < DDIR_RWDIR_CNT; 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;
+
+ /*
+ * This wont be correct for multiple strings, but at least it
+ * works for the basic cases.
+ */
+ strcpy((char *) dst->run_str, (char *) je->run_str);
+}
+
+static void remove_reply_cmd(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct fio_net_cmd_reply *reply = NULL;
+ struct flist_head *entry;
+
+ flist_for_each(entry, &client->cmd_list) {
+ reply = flist_entry(entry, struct fio_net_cmd_reply, list);
+
+ if (cmd->tag == (uintptr_t) reply)
+ break;
+
+ reply = NULL;
+ }
+
+ if (!reply) {
+ log_err("fio: client: unable to find matching tag (%llx)\n", (unsigned long long) cmd->tag);
+ return;
+ }
+
+ flist_del(&reply->list);
+ cmd->tag = reply->saved_tag;
+ free(reply);
+}
+
+int fio_client_wait_for_reply(struct fio_client *client, uint64_t tag)
+{
+ do {
+ struct fio_net_cmd_reply *reply = NULL;
+ struct flist_head *entry;
+
+ flist_for_each(entry, &client->cmd_list) {
+ reply = flist_entry(entry, struct fio_net_cmd_reply, list);
+
+ if (tag == (uintptr_t) reply)
+ break;
+
+ reply = NULL;
+ }
+
+ if (!reply)
+ break;
+
+ usleep(1000);
+ } while (1);
+
+ return 0;
+}
+
+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);
+ client->eta_timeouts = 0;
+
+ 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 int fio_client_handle_iolog(struct fio_client *client,
+ struct fio_net_cmd *cmd)
+{
+ struct cmd_iolog_pdu *pdu;
+ bool store_direct;
+
+ pdu = convert_iolog(cmd, &store_direct);
+ if (!pdu) {
+ log_err("fio: failed converting IO log\n");
+ return 1;
+ }
+
+ if (store_direct) {
+ ssize_t ret;
+ size_t sz;
+ int fd;
+
+ fd = open((const char *) pdu->name,
+ O_WRONLY | O_CREAT | O_TRUNC, 0644);
+ if (fd < 0) {
+ log_err("fio: open log: %s\n", strerror(errno));
+ return 1;
+ }
+
+ sz = cmd->pdu_len - sizeof(*pdu);
+ ret = write(fd, pdu->samples, sz);
+ close(fd);
+
+ if (ret != sz) {
+ log_err("fio: short write on compressed log\n");
+ return 1;
+ }
+
+ return 0;
+ } else {
+ FILE *f;
+
+ f = fopen((const char *) pdu->name, "w");
+ if (!f) {
+ log_err("fio: fopen log: %s\n", strerror(errno));
+ return 1;
+ }
+
+ flush_samples(f, pdu->samples,
+ pdu->nr_samples * sizeof(struct io_sample));
+ fclose(f);
+ return 0;
+ }
+}
+
+static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct cmd_probe_reply_pdu *probe = (struct cmd_probe_reply_pdu *) cmd->payload;
+ const char *os, *arch;
+ char bit[16];