+ 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);
+}
+
+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_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);
+
+ 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_reply_pdu *probe = (struct cmd_probe_reply_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);
+ probe->flags = le64_to_cpu(probe->flags);
+
+ log_info("hostname=%s, be=%u, %s, os=%s, arch=%s, fio=%s, flags=%lx\n",
+ probe->hostname, probe->bigendian, bit, os, arch,
+ probe->fio_version, (unsigned long) probe->flags);
+
+ 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 = le32_to_cpu(pdu->jobs);
+ client->nr_stat = le32_to_cpu(pdu->stat_outputs);
+
+ if (sum_stat_clients > 1)
+ do_output_all_clients = 1;
+
+ sum_stat_clients += client->nr_stat;
+}
+
+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);
+}