Merge branch 'master' into gfio
[fio.git] / server.c
index 274497514484f36a4b22b730bc0127316c6dc16e..32c1d7a4b389bd3c3cc50ba4a989dd7fc1a8d50c 100644 (file)
--- a/server.c
+++ b/server.c
@@ -23,8 +23,6 @@
 #include "crc/crc16.h"
 #include "lib/ieee754.h"
 
-#include "fio_version.h"
-
 int fio_net_port = FIO_NET_PORT;
 
 int exit_backend = 0;
@@ -276,7 +274,7 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk)
                                struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmdret->payload;
                                char *buf = (char *) pdu->buf;
 
-                               buf[pdu->buf_len ] = '\0';
+                               buf[pdu->buf_len] = '\0';
                        } else if (cmdret->opcode == FIO_NET_CMD_JOB) {
                                struct cmd_job_pdu *pdu = (struct cmd_job_pdu *) cmdret->payload;
                                char *buf = (char *) pdu->buf;
@@ -293,6 +291,33 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk)
        return cmdret;
 }
 
+static void add_reply(uint64_t tag, struct flist_head *list)
+{
+       struct fio_net_cmd_reply *reply = (struct fio_net_cmd_reply *) tag;
+
+       flist_add_tail(&reply->list, list);
+}
+
+static uint64_t alloc_reply(uint64_t tag, uint16_t opcode)
+{
+       struct fio_net_cmd_reply *reply;
+
+       reply = calloc(1, sizeof(*reply));
+       INIT_FLIST_HEAD(&reply->list);
+       gettimeofday(&reply->tv, NULL);
+       reply->saved_tag = tag;
+       reply->opcode = opcode;
+
+       return (uintptr_t) reply;
+}
+
+static void free_reply(uint64_t tag)
+{
+       struct fio_net_cmd_reply *reply = (struct fio_net_cmd_reply *) tag;
+
+       free(reply);
+}
+
 void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, const void *pdu)
 {
        uint32_t pdu_len;
@@ -309,12 +334,19 @@ void fio_net_cmd_crc(struct fio_net_cmd *cmd)
 }
 
 int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size,
-                    uint64_t tag)
+                    uint64_t *tagptr, struct flist_head *list)
 {
        struct fio_net_cmd *cmd = NULL;
        size_t this_len, cur_len = 0;
+       uint64_t tag;
        int ret;
 
+       if (list) {
+               assert(tagptr);
+               tag = *tagptr = alloc_reply(*tagptr, opcode);
+       } else
+               tag = tagptr ? *tagptr : 0;
+
        do {
                this_len = size;
                if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
@@ -340,6 +372,13 @@ int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size,
                buf += this_len;
        } while (!ret && size);
 
+       if (list) {
+               if (ret)
+                       free_reply(tag);
+               else
+                       add_reply(tag, list);
+       }
+
        if (cmd)
                free(cmd);
 
@@ -363,28 +402,22 @@ static int fio_net_send_simple_stack_cmd(int sk, uint16_t opcode, uint64_t tag)
 int fio_net_send_simple_cmd(int sk, uint16_t opcode, uint64_t tag,
                            struct flist_head *list)
 {
-       struct fio_net_int_cmd *cmd;
        int ret;
 
-       if (!list)
-               return fio_net_send_simple_stack_cmd(sk, opcode, tag);
-
-       cmd = malloc(sizeof(*cmd));
+       if (list)
+               tag = alloc_reply(tag, opcode);
 
-       fio_init_net_cmd(&cmd->cmd, opcode, NULL, 0, (uintptr_t) cmd);
-       fio_net_cmd_crc(&cmd->cmd);
-
-       INIT_FLIST_HEAD(&cmd->list);
-       gettimeofday(&cmd->tv, NULL);
-       cmd->saved_tag = tag;
-
-       ret = fio_send_data(sk, &cmd->cmd, sizeof(cmd->cmd));
+       ret = fio_net_send_simple_stack_cmd(sk, opcode, tag);
        if (ret) {
-               free(cmd);
+               if (list)
+                       free_reply(tag);
+
                return ret;
        }
 
-       flist_add_tail(&cmd->list, list);
+       if (list)
+               add_reply(tag, list);
+
        return 0;
 }
 
@@ -395,15 +428,24 @@ int fio_net_send_quit(int sk)
        return fio_net_send_simple_cmd(sk, FIO_NET_CMD_QUIT, 0, NULL);
 }
 
-int fio_net_send_stop(int sk, int error, int signal)
+static int fio_net_send_ack(int sk, struct fio_net_cmd *cmd, int error,
+                           int signal)
 {
        struct cmd_end_pdu epdu;
+       uint64_t tag = 0;
 
-       dprint(FD_NET, "server: sending stop (%d, %d)\n", error, signal);
+       if (cmd)
+               tag = cmd->tag;
 
        epdu.error = __cpu_to_le32(error);
        epdu.signal = __cpu_to_le32(signal);
-       return fio_net_send_cmd(sk, FIO_NET_CMD_STOP, &epdu, sizeof(epdu), 0);
+       return fio_net_send_cmd(sk, FIO_NET_CMD_STOP, &epdu, sizeof(epdu), &tag, NULL);
+}
+
+int fio_net_send_stop(int sk, int error, int signal)
+{
+       dprint(FD_NET, "server: sending stop (%d, %d)\n", error, signal);
+       return fio_net_send_ack(sk, NULL, error, signal);
 }
 
 static void fio_server_add_fork_item(pid_t pid, struct flist_head *list)
@@ -506,6 +548,7 @@ static int handle_run_cmd(struct fio_net_cmd *cmd)
        }
 
        ret = fio_backend();
+       free_threads_shm();
        _exit(ret);
 }
 
@@ -524,7 +567,8 @@ static int handle_job_cmd(struct fio_net_cmd *cmd)
        }
 
        spdu.jobs = cpu_to_le32(thread_number);
-       fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0);
+       spdu.stat_outputs = cpu_to_le32(stat_number);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), NULL, NULL);
        return 0;
 }
 
@@ -563,37 +607,39 @@ static int handle_jobline_cmd(struct fio_net_cmd *cmd)
        free(argv);
 
        spdu.jobs = cpu_to_le32(thread_number);
-       fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0);
+       spdu.stat_outputs = cpu_to_le32(stat_number);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), NULL, NULL);
        return 0;
 }
 
 static int handle_probe_cmd(struct fio_net_cmd *cmd)
 {
        struct cmd_probe_pdu probe;
+       uint64_t tag = cmd->tag;
 
        dprint(FD_NET, "server: sending probe reply\n");
 
        memset(&probe, 0, sizeof(probe));
        gethostname((char *) probe.hostname, sizeof(probe.hostname));
-#ifdef FIO_BIG_ENDIAN
+#ifdef CONFIG_BIG_ENDIAN
        probe.bigendian = 1;
 #endif
-       probe.fio_major = FIO_MAJOR;
-       probe.fio_minor = FIO_MINOR;
-       probe.fio_patch = FIO_PATCH;
+       strncpy((char *) probe.fio_version, fio_version_string, sizeof(probe.fio_version));
 
        probe.os        = FIO_OS;
        probe.arch      = FIO_ARCH;
-
        probe.bpp       = sizeof(void *);
+       probe.cpus      = __cpu_to_le32(cpus_online());
+       probe.flags     = 0;
 
-       return fio_net_send_cmd(server_fd, FIO_NET_CMD_PROBE, &probe, sizeof(probe), cmd->tag);
+       return fio_net_send_cmd(server_fd, FIO_NET_CMD_PROBE, &probe, sizeof(probe), &tag, NULL);
 }
 
 static int handle_send_eta_cmd(struct fio_net_cmd *cmd)
 {
        struct jobs_eta *je;
        size_t size;
+       uint64_t tag = cmd->tag;
        int i;
 
        if (!thread_number)
@@ -615,30 +661,60 @@ static int handle_send_eta_cmd(struct fio_net_cmd *cmd)
        je->nr_pending          = cpu_to_le32(je->nr_pending);
        je->files_open          = cpu_to_le32(je->files_open);
 
-       for (i = 0; i < 2; i++) {
+       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);
 
-       fio_net_send_cmd(server_fd, FIO_NET_CMD_ETA, je, size, cmd->tag);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_ETA, je, size, &tag, NULL);
        free(je);
        return 0;
 }
 
+static int send_update_job_reply(int fd, uint64_t __tag, int error)
+{
+       uint64_t tag = __tag;
+       uint32_t pdu_error;
+
+       pdu_error = __cpu_to_le32(error);
+       return fio_net_send_cmd(fd, FIO_NET_CMD_UPDATE_JOB, &pdu_error, sizeof(pdu_error), &tag, NULL);
+}
+
+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(server_fd, cmd->tag, ENODEV);
+               return 0;
+       }
+
+       td = &threads[tnumber - 1];
+       convert_thread_options_to_cpu(&td->o, &pdu->top);
+       send_update_job_reply(server_fd, cmd->tag, 0);
+       return 0;
+}
+
 static int handle_command(struct fio_net_cmd *cmd)
 {
        int ret;
 
-       dprint(FD_NET, "server: got op [%s], pdu=%u, tag=%lx\n",
-                       fio_server_op(cmd->opcode), cmd->pdu_len, cmd->tag);
+       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:
@@ -662,8 +738,11 @@ static int handle_command(struct fio_net_cmd *cmd)
        case FIO_NET_CMD_RUN:
                ret = handle_run_cmd(cmd);
                break;
+       case FIO_NET_CMD_UPDATE_JOB:
+               ret = handle_update_job_cmd(cmd);
+               break;
        default:
-               log_err("fio: unknown opcode: %s\n",fio_server_op(cmd->opcode));
+               log_err("fio: unknown opcode: %s\n", fio_server_op(cmd->opcode));
                ret = 1;
        }
 
@@ -692,7 +771,7 @@ static int handle_connection(int sk)
 
                        if (!flist_empty(&job_list))
                                timeout = 100;
-               
+
                        ret = poll(&pfd, 1, timeout);
                        if (ret < 0) {
                                if (errno == EINTR)
@@ -741,7 +820,7 @@ static int handle_connection(int sk)
 static int accept_loop(int listen_sk)
 {
        struct sockaddr_in addr;
-       fio_socklen_t len = sizeof(addr);
+       socklen_t len = sizeof(addr);
        struct pollfd pfd;
        int ret = 0, sk, flags, exitval = 0;
 
@@ -825,7 +904,7 @@ int fio_server_text_output(int level, const char *buf, size_t len)
 
        memcpy(pdu->buf, buf, len);
 
-       fio_net_send_cmd(server_fd, FIO_NET_CMD_TEXT, pdu, tlen, 0);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_TEXT, pdu, tlen, NULL, NULL);
        free(pdu);
        return len;
 }
@@ -847,7 +926,7 @@ static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
 {
        int i;
 
-       for (i = 0; i < 2; i++) {
+       for (i = 0; i < DDIR_RWDIR_CNT; i++) {
                dst->max_run[i]         = cpu_to_le64(src->max_run[i]);
                dst->min_run[i]         = cpu_to_le64(src->min_run[i]);
                dst->max_bw[i]          = cpu_to_le64(src->max_bw[i]);
@@ -857,7 +936,9 @@ static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
        }
 
        dst->kb_base    = cpu_to_le32(src->kb_base);
+       dst->unit_base  = cpu_to_le32(src->unit_base);
        dst->groupid    = cpu_to_le32(src->groupid);
+       dst->unified_rw_rep     = cpu_to_le32(src->unified_rw_rep);
 }
 
 /*
@@ -882,8 +963,9 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs)
        p.ts.groupid            = cpu_to_le32(ts->groupid);
        p.ts.pid                = cpu_to_le32(ts->pid);
        p.ts.members            = cpu_to_le32(ts->members);
+       p.ts.unified_rw_rep     = cpu_to_le32(ts->unified_rw_rep);
 
-       for (i = 0; i < 2; i++) {
+       for (i = 0; i < DDIR_RWDIR_CNT; i++) {
                convert_io_stat(&p.ts.clat_stat[i], &ts->clat_stat[i]);
                convert_io_stat(&p.ts.slat_stat[i], &ts->slat_stat[i]);
                convert_io_stat(&p.ts.lat_stat[i], &ts->lat_stat[i]);
@@ -915,11 +997,11 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs)
                p.ts.io_u_lat_m[i]      = cpu_to_le32(ts->io_u_lat_m[i]);
        }
 
-       for (i = 0; i < 2; i++)
+       for (i = 0; i < DDIR_RWDIR_CNT; i++)
                for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
                        p.ts.io_u_plat[i][j] = cpu_to_le32(ts->io_u_plat[i][j]);
 
-       for (i = 0; i < 3; i++) {
+       for (i = 0; i < DDIR_RWDIR_CNT; i++) {
                p.ts.total_io_u[i]      = cpu_to_le64(ts->total_io_u[i]);
                p.ts.short_io_u[i]      = cpu_to_le64(ts->short_io_u[i]);
        }
@@ -927,7 +1009,7 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs)
        p.ts.total_submit       = cpu_to_le64(ts->total_submit);
        p.ts.total_complete     = cpu_to_le64(ts->total_complete);
 
-       for (i = 0; i < 2; i++) {
+       for (i = 0; i < DDIR_RWDIR_CNT; i++) {
                p.ts.io_bytes[i]        = cpu_to_le64(ts->io_bytes[i]);
                p.ts.runtime[i]         = cpu_to_le64(ts->runtime[i]);
        }
@@ -937,10 +1019,11 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs)
        p.ts.total_err_count    = cpu_to_le64(ts->total_err_count);
        p.ts.first_error        = cpu_to_le32(ts->first_error);
        p.ts.kb_base            = cpu_to_le32(ts->kb_base);
+       p.ts.unit_base          = cpu_to_le32(ts->unit_base);
 
        convert_gs(&p.rs, rs);
 
-       fio_net_send_cmd(server_fd, FIO_NET_CMD_TS, &p, sizeof(p), 0);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_TS, &p, sizeof(p), NULL, NULL);
 }
 
 void fio_server_send_gs(struct group_run_stats *rs)
@@ -950,7 +1033,7 @@ void fio_server_send_gs(struct group_run_stats *rs)
        dprint(FD_NET, "server sending group run stats\n");
 
        convert_gs(&gs, rs);
-       fio_net_send_cmd(server_fd, FIO_NET_CMD_GS, &gs, sizeof(gs), 0);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_GS, &gs, sizeof(gs), NULL, NULL);
 }
 
 static void convert_agg(struct disk_util_agg *dst, struct disk_util_agg *src)
@@ -1004,7 +1087,7 @@ void fio_server_send_du(void)
                convert_dus(&pdu.dus, &du->dus);
                convert_agg(&pdu.agg, &du->agg);
 
-               fio_net_send_cmd(server_fd, FIO_NET_CMD_DU, &pdu, sizeof(pdu), 0);
+               fio_net_send_cmd(server_fd, FIO_NET_CMD_DU, &pdu, sizeof(pdu), NULL, NULL);
        }
 }
 
@@ -1115,7 +1198,7 @@ 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_send_cmd(server_fd, FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), 0);
+       fio_net_send_cmd(server_fd, FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), NULL, NULL);
 }
 
 void fio_server_send_start(struct thread_data *td)
@@ -1128,7 +1211,7 @@ void fio_server_send_start(struct thread_data *td)
 static int fio_init_server_ip(void)
 {
        struct sockaddr *addr;
-       fio_socklen_t socklen;
+       socklen_t socklen;
        int sk, opt;
 
        if (use_ipv6)
@@ -1142,7 +1225,7 @@ static int fio_init_server_ip(void)
        }
 
        opt = 1;
-       if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
+       if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, (void *)&opt, sizeof(opt)) < 0) {
                log_err("fio: setsockopt: %s\n", strerror(errno));
                close(sk);
                return -1;
@@ -1177,7 +1260,7 @@ static int fio_init_server_ip(void)
 static int fio_init_server_sock(void)
 {
        struct sockaddr_un addr;
-       fio_socklen_t len;
+       socklen_t len;
        mode_t mode;
        int sk;
 
@@ -1512,7 +1595,7 @@ int fio_start_server(char *pidfile)
 
 #if defined(WIN32)
        WSADATA wsd;
-       WSAStartup(MAKEWORD(2,2), &wsd);
+       WSAStartup(MAKEWORD(2, 2), &wsd);
 #endif
 
        if (!pidfile)