+ probe.cpus = __cpu_to_le32(cpus_online());
+
+ /*
+ * If the client supports compression and we do too, then enable it
+ */
+ if (has_zlib && le64_to_cpu(pdu->flags) & FIO_PROBE_FLAG_ZLIB) {
+ probe.flags = __cpu_to_le64(FIO_PROBE_FLAG_ZLIB);
+ use_zlib = 1;
+ } else {
+ probe.flags = 0;
+ use_zlib = 0;
+ }
+
+ return fio_net_queue_cmd(FIO_NET_CMD_PROBE, &probe, sizeof(probe), &tag, SK_F_COPY);
+}
+
+static int handle_send_eta_cmd(struct fio_net_cmd *cmd)
+{
+ struct jobs_eta *je;
+ uint64_t tag = cmd->tag;
+ size_t size;
+ int i;
+
+ dprint(FD_NET, "server sending status\n");
+
+ /*
+ * Fake ETA return if we don't have a local one, otherwise the client
+ * will end up timing out waiting for a response to the ETA request
+ */
+ je = get_jobs_eta(true, &size);
+ if (!je) {
+ size = sizeof(*je);
+ je = calloc(1, size);
+ } else {
+ je->nr_running = cpu_to_le32(je->nr_running);
+ je->nr_ramp = cpu_to_le32(je->nr_ramp);
+ je->nr_pending = cpu_to_le32(je->nr_pending);
+ je->nr_setting_up = cpu_to_le32(je->nr_setting_up);
+ je->files_open = cpu_to_le32(je->files_open);
+
+ for (i = 0; i < DDIR_RWDIR_CNT; i++) {
+ je->m_rate[i] = cpu_to_le32(je->m_rate[i]);
+ je->t_rate[i] = cpu_to_le32(je->t_rate[i]);
+ je->m_iops[i] = cpu_to_le32(je->m_iops[i]);
+ je->t_iops[i] = cpu_to_le32(je->t_iops[i]);
+ je->rate[i] = cpu_to_le32(je->rate[i]);
+ je->iops[i] = cpu_to_le32(je->iops[i]);
+ }
+
+ je->elapsed_sec = cpu_to_le64(je->elapsed_sec);
+ je->eta_sec = cpu_to_le64(je->eta_sec);
+ je->nr_threads = cpu_to_le32(je->nr_threads);
+ je->is_pow2 = cpu_to_le32(je->is_pow2);
+ je->unit_base = cpu_to_le32(je->unit_base);
+ }
+
+ fio_net_queue_cmd(FIO_NET_CMD_ETA, je, size, &tag, SK_F_FREE);
+ return 0;
+}
+
+static int send_update_job_reply(uint64_t __tag, int error)
+{
+ uint64_t tag = __tag;
+ uint32_t pdu_error;
+
+ pdu_error = __cpu_to_le32(error);
+ return fio_net_queue_cmd(FIO_NET_CMD_UPDATE_JOB, &pdu_error, sizeof(pdu_error), &tag, SK_F_COPY);
+}
+
+static int handle_update_job_cmd(struct fio_net_cmd *cmd)
+{
+ struct cmd_add_job_pdu *pdu = (struct cmd_add_job_pdu *) cmd->payload;
+ struct thread_data *td;
+ uint32_t tnumber;
+
+ tnumber = le32_to_cpu(pdu->thread_number);
+
+ dprint(FD_NET, "server: updating options for job %u\n", tnumber);
+
+ if (!tnumber || tnumber > thread_number) {
+ send_update_job_reply(cmd->tag, ENODEV);
+ return 0;
+ }
+
+ td = &threads[tnumber - 1];
+ convert_thread_options_to_cpu(&td->o, &pdu->top);
+ send_update_job_reply(cmd->tag, 0);
+ return 0;
+}
+
+static int handle_trigger_cmd(struct fio_net_cmd *cmd)
+{
+ struct cmd_vtrigger_pdu *pdu = (struct cmd_vtrigger_pdu *) cmd->payload;
+ char *buf = (char *) pdu->cmd;
+ struct all_io_list *rep;
+ size_t sz;
+
+ pdu->len = le16_to_cpu(pdu->len);
+ buf[pdu->len] = '\0';
+
+ rep = get_all_io_list(IO_LIST_ALL, &sz);
+ if (!rep) {
+ 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);
+ } else
+ fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, rep, sz, NULL, SK_F_FREE);
+
+ exec_trigger(buf);
+ return 0;
+}
+
+static int handle_command(struct sk_out *sk_out, struct flist_head *job_list,
+ struct fio_net_cmd *cmd)
+{
+ int ret;
+
+ dprint(FD_NET, "server: got op [%s], pdu=%u, tag=%llx\n",
+ fio_server_op(cmd->opcode), cmd->pdu_len,
+ (unsigned long long) cmd->tag);
+
+ switch (cmd->opcode) {
+ case FIO_NET_CMD_QUIT:
+ fio_terminate_threads(TERMINATE_ALL);
+ ret = 0;
+ break;
+ case FIO_NET_CMD_EXIT:
+ exit_backend = 1;
+ return -1;
+ case FIO_NET_CMD_LOAD_FILE:
+ ret = handle_load_file_cmd(cmd);
+ break;
+ case FIO_NET_CMD_JOB:
+ ret = handle_job_cmd(cmd);
+ break;
+ case FIO_NET_CMD_JOBLINE:
+ ret = handle_jobline_cmd(cmd);
+ break;
+ case FIO_NET_CMD_PROBE:
+ ret = handle_probe_cmd(cmd);
+ break;
+ case FIO_NET_CMD_SEND_ETA:
+ ret = handle_send_eta_cmd(cmd);
+ break;
+ case FIO_NET_CMD_RUN:
+ ret = handle_run_cmd(sk_out, job_list, cmd);
+ break;
+ case FIO_NET_CMD_UPDATE_JOB:
+ ret = handle_update_job_cmd(cmd);
+ break;
+ case FIO_NET_CMD_VTRIGGER:
+ ret = handle_trigger_cmd(cmd);
+ break;
+ case FIO_NET_CMD_SENDFILE: {
+ struct cmd_sendfile_reply *in;
+ struct cmd_reply *rep;
+
+ rep = (struct cmd_reply *) (uintptr_t) cmd->tag;
+
+ in = (struct cmd_sendfile_reply *) cmd->payload;
+ in->size = le32_to_cpu(in->size);
+ in->error = le32_to_cpu(in->error);
+ if (in->error) {
+ ret = 1;
+ rep->error = in->error;
+ } else {
+ ret = 0;
+ rep->data = smalloc(in->size);
+ if (!rep->data) {
+ ret = 1;
+ rep->error = ENOMEM;
+ } else {
+ rep->size = in->size;
+ memcpy(rep->data, in->data, in->size);
+ }
+ }
+ fio_mutex_up(&rep->lock);
+ break;
+ }
+ default:
+ log_err("fio: unknown opcode: %s\n", fio_server_op(cmd->opcode));
+ ret = 1;
+ }
+
+ return ret;
+}
+
+/*
+ * Send a command with a separate PDU, not inlined in the command
+ */
+static int fio_send_cmd_ext_pdu(int sk, uint16_t opcode, const void *buf,
+ off_t size, uint64_t tag, uint32_t flags)
+{
+ struct fio_net_cmd cmd;
+ struct iovec iov[2];
+
+ 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);
+
+ return fio_sendv_data(sk, iov, 2);
+}
+
+static void finish_entry(struct sk_entry *entry)
+{
+ if (entry->flags & SK_F_FREE)
+ free(entry->buf);
+ else if (entry->flags & SK_F_COPY)
+ sfree(entry->buf);
+
+ sfree(entry);
+}
+
+static void entry_set_flags_tag(struct sk_entry *entry, struct flist_head *list,
+ unsigned int *flags, uint64_t *tag)
+{
+ if (!flist_empty(list))
+ *flags = FIO_NET_CMD_F_MORE;
+ else
+ *flags = 0;