X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=server.c;h=5e0d322baeae1c8c4664f57f68ec9cbb23ff6d59;hp=524c8146670849c58735fc4faa041d0a6dc08095;hb=8c95307bd5f72720460ee4a46519b9a1b474e7cd;hpb=bebe639808147d587bbe776566d390b9ff98773f diff --git a/server.c b/server.c index 524c8146..5e0d322b 100644 --- a/server.c +++ b/server.c @@ -16,13 +16,16 @@ #include #include #include +#include #include "fio.h" #include "server.h" #include "crc/crc16.h" -#include "ieee754.h" +#include "lib/ieee754.h" -int fio_net_port = 8765; +#include "fio_version.h" + +int fio_net_port = FIO_NET_PORT; int exit_backend = 0; @@ -30,32 +33,115 @@ static int server_fd = -1; static char *fio_server_arg; static char *bind_sock; static struct sockaddr_in saddr_in; +static struct sockaddr_in6 saddr_in6; +static int use_ipv6; + +struct fio_fork_item { + struct flist_head list; + int exitval; + int signal; + int exited; + pid_t pid; +}; + +/* Created on fork on new connection */ +static FLIST_HEAD(conn_list); + +/* Created on job fork from connection */ +static FLIST_HEAD(job_list); + +static const char *fio_server_ops[FIO_NET_CMD_NR] = { + "", + "QUIT", + "EXIT", + "JOB", + "JOBLINE", + "TEXT", + "TS", + "GS", + "SEND_ETA", + "ETA", + "PROBE", + "START", + "STOP", + "DISK_UTIL", + "SERVER_START", + "ADD_JOB", + "CMD_RUN" + "CMD_IOLOG", +}; + +const char *fio_server_op(unsigned int op) +{ + static char buf[32]; -int fio_send_data(int sk, const void *p, unsigned int len) + if (op < FIO_NET_CMD_NR) + return fio_server_ops[op]; + + sprintf(buf, "UNKNOWN/%d", op); + return buf; +} + +static ssize_t iov_total_len(const struct iovec *iov, int count) { - assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_PDU); + ssize_t ret = 0; - do { - int ret = send(sk, p, len, 0); + while (count--) { + ret += iov->iov_len; + iov++; + } + + return ret; +} +static int fio_sendv_data(int sk, struct iovec *iov, int count) +{ + ssize_t total_len = iov_total_len(iov, count); + ssize_t ret; + + do { + ret = writev(sk, iov, count); if (ret > 0) { - len -= ret; - if (!len) + total_len -= ret; + if (!total_len) break; - p += ret; - continue; + + while (ret) { + if (ret >= iov->iov_len) { + ret -= iov->iov_len; + iov++; + continue; + } + iov->iov_base += ret; + iov->iov_len -= ret; + ret = 0; + } } else if (!ret) break; else if (errno == EAGAIN || errno == EINTR) continue; + else + break; } while (!exit_backend); - if (!len) + if (!total_len) return 0; + if (errno) + return -errno; + return 1; } +int fio_send_data(int sk, const void *p, unsigned int len) +{ + struct iovec iov = { .iov_base = (void *) p, .iov_len = len }; + + assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_FRAGMENT_PDU); + + return fio_sendv_data(sk, &iov, 1); +} + int fio_recv_data(int sk, void *p, unsigned int len) { do { @@ -71,6 +157,8 @@ int fio_recv_data(int sk, void *p, unsigned int len) break; else if (errno == EAGAIN || errno == EINTR) continue; + else + break; } while (!exit_backend); if (!len) @@ -86,7 +174,7 @@ static int verify_convert_cmd(struct fio_net_cmd *cmd) cmd->cmd_crc16 = le16_to_cpu(cmd->cmd_crc16); cmd->pdu_crc16 = le16_to_cpu(cmd->pdu_crc16); - crc = crc16(cmd, FIO_NET_CMD_CRC_SZ); + crc = fio_crc16(cmd, FIO_NET_CMD_CRC_SZ); if (crc != cmd->cmd_crc16) { log_err("fio: server bad crc on command (got %x, wanted %x)\n", cmd->cmd_crc16, crc); @@ -96,18 +184,18 @@ static int verify_convert_cmd(struct fio_net_cmd *cmd) cmd->version = le16_to_cpu(cmd->version); cmd->opcode = le16_to_cpu(cmd->opcode); cmd->flags = le32_to_cpu(cmd->flags); - cmd->serial = le64_to_cpu(cmd->serial); + cmd->tag = le64_to_cpu(cmd->tag); cmd->pdu_len = le32_to_cpu(cmd->pdu_len); switch (cmd->version) { - case FIO_SERVER_VER2: + case FIO_SERVER_VER: break; default: log_err("fio: bad server cmd version %d\n", cmd->version); return 1; } - if (cmd->pdu_len > FIO_SERVER_MAX_PDU) { + if (cmd->pdu_len > FIO_SERVER_MAX_FRAGMENT_PDU) { log_err("fio: command payload too large: %u\n", cmd->pdu_len); return 1; } @@ -147,8 +235,12 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk) if (first) memcpy(cmdret, &cmd, sizeof(cmd)); - else - assert(cmdret->opcode == cmd.opcode); + else if (cmdret->opcode != cmd.opcode) { + log_err("fio: fragment opcode mismatch (%d != %d)\n", + cmdret->opcode, cmd.opcode); + ret = 1; + break; + } if (!cmd.pdu_len) break; @@ -160,7 +252,7 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk) break; /* Verify payload crc */ - crc = crc16(pdu, cmd.pdu_len); + crc = fio_crc16(pdu, cmd.pdu_len); if (crc != cmd.pdu_crc16) { log_err("fio: server bad crc on payload "); log_err("(got %x, wanted %x)\n", cmd.pdu_crc16, crc); @@ -179,12 +271,21 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk) cmdret = NULL; } else if (cmdret) { /* zero-terminate text input */ - if (cmdret->pdu_len && (cmdret->opcode == FIO_NET_CMD_TEXT || - cmdret->opcode == FIO_NET_CMD_JOB)) { - char *buf = (char *) cmdret->payload; - - buf[cmdret->pdu_len ] = '\0'; + if (cmdret->pdu_len) { + if (cmdret->opcode == FIO_NET_CMD_TEXT) { + struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmdret->payload; + char *buf = (char *) pdu->buf; + + 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; + int len = le32_to_cpu(pdu->buf_len); + + buf[len] = '\0'; + } } + /* frag flag is internal */ cmdret->flags &= ~FIO_NET_CMD_F_MORE; } @@ -192,31 +293,42 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk) return cmdret; } -void fio_net_cmd_crc(struct fio_net_cmd *cmd) +void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, const void *pdu) { uint32_t pdu_len; - cmd->cmd_crc16 = __cpu_to_le16(crc16(cmd, FIO_NET_CMD_CRC_SZ)); + cmd->cmd_crc16 = __cpu_to_le16(fio_crc16(cmd, FIO_NET_CMD_CRC_SZ)); pdu_len = le32_to_cpu(cmd->pdu_len); - if (pdu_len) - cmd->pdu_crc16 = __cpu_to_le16(crc16(cmd->payload, pdu_len)); + cmd->pdu_crc16 = __cpu_to_le16(fio_crc16(pdu, pdu_len)); } -int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size) +void fio_net_cmd_crc(struct fio_net_cmd *cmd) +{ + fio_net_cmd_crc_pdu(cmd, cmd->payload); +} + +int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size, + uint64_t tag) { - struct fio_net_cmd *cmd; - size_t this_len; + struct fio_net_cmd *cmd = NULL; + size_t this_len, cur_len = 0; int ret; do { this_len = size; - if (this_len > FIO_SERVER_MAX_PDU) - this_len = FIO_SERVER_MAX_PDU; + if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU) + this_len = FIO_SERVER_MAX_FRAGMENT_PDU; - cmd = malloc(sizeof(*cmd) + this_len); + if (!cmd || cur_len < sizeof(*cmd) + this_len) { + if (cmd) + free(cmd); - fio_init_net_cmd(cmd, opcode, buf, this_len); + cur_len = sizeof(*cmd) + this_len; + cmd = malloc(cur_len); + } + + fio_init_net_cmd(cmd, opcode, buf, this_len, tag); if (this_len < size) cmd->flags = __cpu_to_le32(FIO_NET_CMD_F_MORE); @@ -224,80 +336,243 @@ int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size) fio_net_cmd_crc(cmd); ret = fio_send_data(fd, cmd, sizeof(*cmd) + this_len); - free(cmd); size -= this_len; buf += this_len; } while (!ret && size); + if (cmd) + free(cmd); + return ret; } -int fio_net_send_simple_cmd(int sk, uint16_t opcode, uint64_t serial) +static int fio_net_send_simple_stack_cmd(int sk, uint16_t opcode, uint64_t tag) { struct fio_net_cmd cmd; - fio_init_net_cmd(&cmd, opcode, NULL, 0); + fio_init_net_cmd(&cmd, opcode, NULL, 0, tag); fio_net_cmd_crc(&cmd); return fio_send_data(sk, &cmd, sizeof(cmd)); } -static int fio_server_send_quit_cmd(void) +/* + * If 'list' is non-NULL, then allocate and store the sent command for + * later verification. + */ +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)); + + 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)); + if (ret) { + free(cmd); + return ret; + } + + flist_add_tail(&cmd->list, list); + return 0; +} + +int fio_net_send_quit(int sk) { dprint(FD_NET, "server: sending quit\n"); - return fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_QUIT, 0); + + return fio_net_send_simple_cmd(sk, FIO_NET_CMD_QUIT, 0, NULL); } -static int handle_job_cmd(struct fio_net_cmd *cmd) +int fio_net_send_stop(int sk, int error, int signal) +{ + struct cmd_end_pdu epdu; + + dprint(FD_NET, "server: sending stop (%d, %d)\n", error, signal); + + 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); +} + +static void fio_server_add_fork_item(pid_t pid, struct flist_head *list) +{ + struct fio_fork_item *ffi; + + ffi = malloc(sizeof(*ffi)); + ffi->exitval = 0; + ffi->signal = 0; + ffi->exited = 0; + ffi->pid = pid; + flist_add_tail(&ffi->list, list); +} + +static void fio_server_add_conn_pid(pid_t pid) +{ + dprint(FD_NET, "server: forked off connection job (pid=%u)\n", pid); + fio_server_add_fork_item(pid, &conn_list); +} + +static void fio_server_add_job_pid(pid_t pid) +{ + dprint(FD_NET, "server: forked off job job (pid=%u)\n", pid); + fio_server_add_fork_item(pid, &job_list); +} + +static void fio_server_check_fork_item(struct fio_fork_item *ffi) +{ + int ret, status; + + ret = waitpid(ffi->pid, &status, WNOHANG); + if (ret < 0) { + if (errno == ECHILD) { + log_err("fio: connection pid %u disappeared\n", ffi->pid); + ffi->exited = 1; + } else + log_err("fio: waitpid: %s\n", strerror(errno)); + } else if (ret == ffi->pid) { + if (WIFSIGNALED(status)) { + ffi->signal = WTERMSIG(status); + ffi->exited = 1; + } + if (WIFEXITED(status)) { + if (WEXITSTATUS(status)) + ffi->exitval = WEXITSTATUS(status); + ffi->exited = 1; + } + } +} + +static void fio_server_fork_item_done(struct fio_fork_item *ffi) +{ + dprint(FD_NET, "pid %u exited, sig=%u, exitval=%d\n", ffi->pid, ffi->signal, ffi->exitval); + + /* + * Fold STOP and QUIT... + */ + fio_net_send_stop(server_fd, ffi->exitval, ffi->signal); + fio_net_send_quit(server_fd); + flist_del(&ffi->list); + free(ffi); +} + +static void fio_server_check_fork_items(struct flist_head *list, int bla) +{ + struct flist_head *entry, *tmp; + struct fio_fork_item *ffi; + + flist_for_each_safe(entry, tmp, list) { + ffi = flist_entry(entry, struct fio_fork_item, list); + + fio_server_check_fork_item(ffi); + + if (ffi->exited) + fio_server_fork_item_done(ffi); + } +} + +static void fio_server_check_jobs(void) +{ + fio_server_check_fork_items(&job_list, 0); +} + +static void fio_server_check_conns(void) +{ + fio_server_check_fork_items(&conn_list, 1); +} + +static int handle_run_cmd(struct fio_net_cmd *cmd) { - char *buf = (char *) cmd->payload; + pid_t pid; int ret; - if (parse_jobs_ini(buf, 1, 0)) { - fio_server_send_quit_cmd(); - return -1; + set_genesis_time(); + + pid = fork(); + if (pid) { + fio_server_add_job_pid(pid); + return 0; } - fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_START, 0); + ret = fio_backend(); + _exit(ret); +} - ret = exec_run(); - fio_server_send_quit_cmd(); - reset_fio_state(); - return ret; +static int handle_job_cmd(struct fio_net_cmd *cmd) +{ + struct cmd_job_pdu *pdu = (struct cmd_job_pdu *) cmd->payload; + void *buf = pdu->buf; + struct cmd_start_pdu spdu; + + pdu->buf_len = le32_to_cpu(pdu->buf_len); + pdu->client_type = le32_to_cpu(pdu->client_type); + + if (parse_jobs_ini(buf, 1, 0, pdu->client_type)) { + fio_net_send_quit(server_fd); + return -1; + } + + spdu.jobs = cpu_to_le32(thread_number); + fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0); + return 0; } static int handle_jobline_cmd(struct fio_net_cmd *cmd) { - struct cmd_line_pdu *pdu = (struct cmd_line_pdu *) cmd->payload; - char *argv[FIO_NET_CMD_JOBLINE_ARGV]; - int ret, i; + void *pdu = cmd->payload; + struct cmd_single_line_pdu *cslp; + struct cmd_line_pdu *clp; + unsigned long offset; + struct cmd_start_pdu spdu; + char **argv; + int i; + + clp = pdu; + clp->lines = le16_to_cpu(clp->lines); + clp->client_type = le16_to_cpu(clp->client_type); + argv = malloc(clp->lines * sizeof(char *)); + offset = sizeof(*clp); - pdu->argc = le16_to_cpu(pdu->argc); + dprint(FD_NET, "server: %d command line args\n", clp->lines); - dprint(FD_NET, "server: %d command line args\n", pdu->argc); + for (i = 0; i < clp->lines; i++) { + cslp = pdu + offset; + argv[i] = (char *) cslp->text; - for (i = 0; i < pdu->argc; i++) { - argv[i] = (char *) pdu->argv[i]; + offset += sizeof(*cslp) + le16_to_cpu(cslp->len); dprint(FD_NET, "server: %d: %s\n", i, argv[i]); } - if (parse_cmd_line(pdu->argc, argv)) { - fio_server_send_quit_cmd(); + if (parse_cmd_line(clp->lines, argv, clp->client_type)) { + fio_net_send_quit(server_fd); + free(argv); return -1; } - fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_START, 0); + free(argv); - ret = exec_run(); - fio_server_send_quit_cmd(); - reset_fio_state(); - return ret; + spdu.jobs = cpu_to_le32(thread_number); + fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0); + return 0; } static int handle_probe_cmd(struct fio_net_cmd *cmd) { struct cmd_probe_pdu probe; + dprint(FD_NET, "server: sending probe reply\n"); + memset(&probe, 0, sizeof(probe)); gethostname((char *) probe.hostname, sizeof(probe.hostname)); #ifdef FIO_BIG_ENDIAN @@ -307,14 +582,63 @@ static int handle_probe_cmd(struct fio_net_cmd *cmd) probe.fio_minor = FIO_MINOR; probe.fio_patch = FIO_PATCH; - return fio_net_send_cmd(server_fd, FIO_NET_CMD_PROBE, &probe, sizeof(probe)); + probe.os = FIO_OS; + probe.arch = FIO_ARCH; + + probe.bpp = sizeof(void *); + + return fio_net_send_cmd(server_fd, FIO_NET_CMD_PROBE, &probe, sizeof(probe), cmd->tag); +} + +static int handle_send_eta_cmd(struct fio_net_cmd *cmd) +{ + struct jobs_eta *je; + size_t size; + int i; + + if (!thread_number) + return 0; + + size = sizeof(*je) + thread_number * sizeof(char) + 1; + je = malloc(size); + memset(je, 0, size); + + if (!calc_thread_status(je, 1)) { + free(je); + return 0; + } + + dprint(FD_NET, "server sending status\n"); + + 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->files_open = cpu_to_le32(je->files_open); + + for (i = 0; i < 2; 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); + + fio_net_send_cmd(server_fd, FIO_NET_CMD_ETA, je, size, cmd->tag); + free(je); + return 0; } static int handle_command(struct fio_net_cmd *cmd) { int ret; - dprint(FD_NET, "server: got opcode %d\n", cmd->opcode); + dprint(FD_NET, "server: got op [%s], pdu=%u, tag=%lx\n", + fio_server_op(cmd->opcode), cmd->pdu_len, cmd->tag); switch (cmd->opcode) { case FIO_NET_CMD_QUIT: @@ -332,19 +656,29 @@ static int handle_command(struct fio_net_cmd *cmd) 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(cmd); + break; default: - log_err("fio: unknown opcode: %d\n", cmd->opcode); + log_err("fio: unknown opcode: %s\n",fio_server_op(cmd->opcode)); ret = 1; } return ret; } -static int handle_connection(int sk, int block) +static int handle_connection(int sk) { struct fio_net_cmd *cmd = NULL; int ret = 0; + reset_fio_state(); + INIT_FLIST_HEAD(&job_list); + server_fd = sk; + /* read forever */ while (!exit_backend) { struct pollfd pfd = { @@ -361,8 +695,7 @@ static int handle_connection(int sk, int block) log_err("fio: poll: %s\n", strerror(errno)); break; } else if (!ret) { - if (!block) - return 0; + fio_server_check_jobs(); continue; } @@ -374,6 +707,8 @@ static int handle_connection(int sk, int block) } } while (!exit_backend); + fio_server_check_jobs(); + if (ret < 0) break; @@ -394,13 +729,8 @@ static int handle_connection(int sk, int block) if (cmd) free(cmd); - return ret; -} - -void fio_server_idle_loop(void) -{ - if (server_fd != -1) - handle_connection(server_fd, 0); + close(sk); + _exit(ret); } static int accept_loop(int listen_sk) @@ -408,61 +738,86 @@ static int accept_loop(int listen_sk) struct sockaddr_in addr; fio_socklen_t len = sizeof(addr); struct pollfd pfd; - int ret, sk, flags, exitval = 0; + int ret = 0, sk, flags, exitval = 0; dprint(FD_NET, "server enter accept loop\n"); flags = fcntl(listen_sk, F_GETFL); flags |= O_NONBLOCK; fcntl(listen_sk, F_SETFL, flags); -again: - pfd.fd = listen_sk; - pfd.events = POLLIN; - do { - ret = poll(&pfd, 1, 100); - if (ret < 0) { - if (errno == EINTR) - break; - log_err("fio: poll: %s\n", strerror(errno)); - goto out; - } else if (!ret) - continue; - if (pfd.revents & POLLIN) - break; - } while (!exit_backend); + while (!exit_backend) { + pid_t pid; - if (exit_backend) - goto out; + pfd.fd = listen_sk; + pfd.events = POLLIN; + do { + ret = poll(&pfd, 1, 100); + if (ret < 0) { + if (errno == EINTR) + break; + log_err("fio: poll: %s\n", strerror(errno)); + break; + } else if (!ret) { + fio_server_check_conns(); + continue; + } - sk = accept(listen_sk, (struct sockaddr *) &addr, &len); - if (sk < 0) { - log_err("fio: accept: %s\n", strerror(errno)); - return -1; - } + if (pfd.revents & POLLIN) + break; + } while (!exit_backend); - dprint(FD_NET, "server: connect from %s\n", inet_ntoa(addr.sin_addr)); + fio_server_check_conns(); - server_fd = sk; + if (exit_backend || ret < 0) + break; - exitval = handle_connection(sk, 1); + sk = accept(listen_sk, (struct sockaddr *) &addr, &len); + if (sk < 0) { + log_err("fio: accept: %s\n", strerror(errno)); + return -1; + } - server_fd = -1; - close(sk); + dprint(FD_NET, "server: connect from %s\n", inet_ntoa(addr.sin_addr)); - if (!exit_backend) - goto again; + pid = fork(); + if (pid) { + close(sk); + fio_server_add_conn_pid(pid); + continue; + } + + /* exits */ + handle_connection(sk); + } -out: return exitval; } -int fio_server_text_output(const char *buf, unsigned int len) +int fio_server_text_output(int level, const char *buf, size_t len) { - if (server_fd != -1) - return fio_net_send_cmd(server_fd, FIO_NET_CMD_TEXT, buf, len); + struct cmd_text_pdu *pdu; + unsigned int tlen; + struct timeval tv; + + if (server_fd == -1) + return log_local_buf(buf, len); + + tlen = sizeof(*pdu) + len; + pdu = malloc(tlen); + + pdu->level = __cpu_to_le32(level); + pdu->buf_len = __cpu_to_le32(len); + + gettimeofday(&tv, NULL); + pdu->log_sec = __cpu_to_le64(tv.tv_sec); + pdu->log_usec = __cpu_to_le64(tv.tv_usec); - return fwrite(buf, len, 1, f_err); + memcpy(pdu->buf, buf, len); + + fio_net_send_cmd(server_fd, FIO_NET_CMD_TEXT, pdu, tlen, 0); + free(pdu); + return len; } static void convert_io_stat(struct io_stat *dst, struct io_stat *src) @@ -512,10 +867,11 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs) strcpy(p.ts.verror, ts->verror); strcpy(p.ts.description, ts->description); - p.ts.error = cpu_to_le32(ts->error); - 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.error = cpu_to_le32(ts->error); + p.ts.thread_number = cpu_to_le32(ts->thread_number); + p.ts.groupid = cpu_to_le32(ts->groupid); + p.ts.pid = cpu_to_le32(ts->pid); + p.ts.members = cpu_to_le32(ts->members); for (i = 0; i < 2; i++) { convert_io_stat(&p.ts.clat_stat[i], &ts->clat_stat[i]); @@ -532,9 +888,10 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs) p.ts.clat_percentiles = cpu_to_le64(ts->clat_percentiles); for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) { - fio_fp64_t *fp = &p.ts.percentile_list[i]; + fio_fp64_t *src = &ts->percentile_list[i]; + fio_fp64_t *dst = &p.ts.percentile_list[i]; - fp->u.i = __cpu_to_le64(fio_double_to_uint64(fp->u.f)); + dst->u.i = __cpu_to_le64(fio_double_to_uint64(src->u.f)); } for (i = 0; i < FIO_IO_U_MAP_NR; i++) { @@ -573,7 +930,7 @@ void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs) convert_gs(&p.rs, rs); - fio_net_send_cmd(server_fd, FIO_NET_CMD_TS, &p, sizeof(p)); + fio_net_send_cmd(server_fd, FIO_NET_CMD_TS, &p, sizeof(p), 0); } void fio_server_send_gs(struct group_run_stats *rs) @@ -583,69 +940,192 @@ 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)); + fio_net_send_cmd(server_fd, FIO_NET_CMD_GS, &gs, sizeof(gs), 0); } -void fio_server_send_status(void) +static void convert_agg(struct disk_util_agg *dst, struct disk_util_agg *src) { - struct jobs_eta *je; - size_t size; - void *buf; int i; - size = sizeof(*je) + thread_number * sizeof(char); - buf = malloc(size); - memset(buf, 0, size); - je = buf; - - if (!calc_thread_status(je)) { - free(je); - return; + for (i = 0; i < 2; i++) { + dst->ios[i] = cpu_to_le32(src->ios[i]); + dst->merges[i] = cpu_to_le32(src->merges[i]); + dst->sectors[i] = cpu_to_le64(src->sectors[i]); + dst->ticks[i] = cpu_to_le32(src->ticks[i]); } - dprint(FD_NET, "server sending status\n"); + dst->io_ticks = cpu_to_le32(src->io_ticks); + dst->time_in_queue = cpu_to_le32(src->time_in_queue); + dst->slavecount = cpu_to_le32(src->slavecount); + dst->max_util.u.i = __cpu_to_le64(fio_double_to_uint64(src->max_util.u.f)); +} - 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->files_open = cpu_to_le32(je->files_open); - je->m_rate = cpu_to_le32(je->m_rate); - je->t_rate = cpu_to_le32(je->t_rate); - je->m_iops = cpu_to_le32(je->m_iops); - je->t_iops = cpu_to_le32(je->t_iops); +static void convert_dus(struct disk_util_stat *dst, struct disk_util_stat *src) +{ + int i; + + strcpy((char *) dst->name, (char *) src->name); for (i = 0; i < 2; i++) { - je->rate[i] = cpu_to_le32(je->rate[i]); - je->iops[i] = cpu_to_le32(je->iops[i]); + dst->ios[i] = cpu_to_le32(src->ios[i]); + dst->merges[i] = cpu_to_le32(src->merges[i]); + dst->sectors[i] = cpu_to_le64(src->sectors[i]); + dst->ticks[i] = cpu_to_le32(src->ticks[i]); } - je->elapsed_sec = cpu_to_le32(je->nr_running); - je->eta_sec = cpu_to_le64(je->eta_sec); + dst->io_ticks = cpu_to_le32(src->io_ticks); + dst->time_in_queue = cpu_to_le32(src->time_in_queue); + dst->msec = cpu_to_le64(src->msec); +} - fio_net_send_cmd(server_fd, FIO_NET_CMD_ETA, buf, size); - free(je); +void fio_server_send_du(void) +{ + struct disk_util *du; + struct flist_head *entry; + struct cmd_du_pdu pdu; + + dprint(FD_NET, "server: sending disk_util %d\n", !flist_empty(&disk_list)); + + memset(&pdu, 0, sizeof(pdu)); + + flist_for_each(entry, &disk_list) { + du = flist_entry(entry, struct disk_util, list); + + 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); + } } -int fio_server_log(const char *format, ...) +/* + * 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) { - char buffer[1024]; - va_list args; - size_t len; + struct fio_net_cmd cmd; + struct iovec iov[2]; - dprint(FD_NET, "server log\n"); + iov[0].iov_base = &cmd; + iov[0].iov_len = sizeof(cmd); + iov[1].iov_base = (void *) buf; + iov[1].iov_len = size; - va_start(args, format); - len = vsnprintf(buffer, sizeof(buffer), format, args); - va_end(args); + __fio_init_net_cmd(&cmd, opcode, size, tag); + cmd.flags = __cpu_to_le32(flags); + fio_net_cmd_crc_pdu(&cmd, buf); - return fio_server_text_output(buffer, len); + return fio_sendv_data(sk, iov, 2); +} + +int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name) +{ + struct cmd_iolog_pdu pdu; + z_stream stream; + void *out_pdu; + int i, ret = 0; + + pdu.thread_number = cpu_to_le32(td->thread_number); + pdu.nr_samples = __cpu_to_le32(log->nr_samples); + pdu.log_type = cpu_to_le32(log->log_type); + strcpy((char *) pdu.name, name); + + for (i = 0; i < log->nr_samples; i++) { + struct io_sample *s = &log->log[i]; + + s->time = cpu_to_le64(s->time); + s->val = cpu_to_le64(s->val); + s->ddir = cpu_to_le32(s->ddir); + s->bs = cpu_to_le32(s->bs); + } + + /* + * Dirty - since the log is potentially huge, compress it into + * FIO_SERVER_MAX_FRAGMENT_PDU chunks and let the receiving + * side defragment it. + */ + out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU); + + stream.zalloc = Z_NULL; + stream.zfree = Z_NULL; + stream.opaque = Z_NULL; + + if (deflateInit(&stream, Z_DEFAULT_COMPRESSION) != Z_OK) { + ret = 1; + goto err; + } + + /* + * Send header first, it's not compressed. + */ + ret = fio_send_cmd_ext_pdu(server_fd, FIO_NET_CMD_IOLOG, &pdu, + sizeof(pdu), 0, FIO_NET_CMD_F_MORE); + if (ret) + goto err_zlib; + + stream.next_in = (void *) log->log; + stream.avail_in = log->nr_samples * sizeof(struct io_sample); + + do { + unsigned int this_len, flags = 0; + int ret; + + stream.avail_out = FIO_SERVER_MAX_FRAGMENT_PDU; + stream.next_out = out_pdu; + ret = deflate(&stream, Z_FINISH); + /* may be Z_OK, or Z_STREAM_END */ + if (ret < 0) + goto err_zlib; + + this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream.avail_out; + + if (stream.avail_in) + flags = FIO_NET_CMD_F_MORE; + + ret = fio_send_cmd_ext_pdu(server_fd, FIO_NET_CMD_IOLOG, + out_pdu, this_len, 0, flags); + if (ret) + goto err_zlib; + } while (stream.avail_in); + +err_zlib: + deflateEnd(&stream); +err: + free(out_pdu); + return ret; +} + +void fio_server_send_add_job(struct thread_data *td) +{ + struct cmd_add_job_pdu pdu; + + memset(&pdu, 0, sizeof(pdu)); + pdu.thread_number = cpu_to_le32(td->thread_number); + 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); +} + +void fio_server_send_start(struct thread_data *td) +{ + assert(server_fd != -1); + + fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_SERVER_START, 0, NULL); } static int fio_init_server_ip(void) { + struct sockaddr *addr; + fio_socklen_t socklen; int sk, opt; - sk = socket(AF_INET, SOCK_STREAM, 0); + if (use_ipv6) + sk = socket(AF_INET6, SOCK_STREAM, 0); + else + sk = socket(AF_INET, SOCK_STREAM, 0); + if (sk < 0) { log_err("fio: socket: %s\n", strerror(errno)); return -1; @@ -665,9 +1145,17 @@ static int fio_init_server_ip(void) } #endif - saddr_in.sin_family = AF_INET; + if (use_ipv6) { + addr = (struct sockaddr *) &saddr_in6; + socklen = sizeof(saddr_in6); + saddr_in6.sin6_family = AF_INET6; + } else { + addr = (struct sockaddr *) &saddr_in; + socklen = sizeof(saddr_in); + saddr_in.sin_family = AF_INET; + } - if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) { + if (bind(sk, addr, socklen) < 0) { log_err("fio: bind: %s\n", strerror(errno)); close(sk); return -1; @@ -723,14 +1211,32 @@ static int fio_init_server_connection(void) if (sk < 0) return sk; - if (!bind_sock) - sprintf(bind_str, "%s:%u", inet_ntoa(saddr_in.sin_addr), fio_net_port); - else + if (!bind_sock) { + char *p, port[16]; + const void *src; + int af; + + if (use_ipv6) { + af = AF_INET6; + src = &saddr_in6.sin6_addr; + } else { + af = AF_INET; + src = &saddr_in.sin_addr; + } + + p = (char *) inet_ntop(af, src, bind_str, sizeof(bind_str)); + + sprintf(port, ",%u", fio_net_port); + if (p) + strcat(p, port); + else + strcpy(bind_str, port); + } else strcpy(bind_str, bind_sock); log_info("fio: server listening on %s\n", bind_str); - if (listen(sk, 1) < 0) { + if (listen(sk, 0) < 0) { log_err("fio: listen: %s\n", strerror(errno)); return -1; } @@ -738,73 +1244,129 @@ static int fio_init_server_connection(void) return sk; } +int fio_server_parse_host(const char *host, int *ipv6, struct in_addr *inp, + struct in6_addr *inp6) + +{ + int ret = 0; + + if (*ipv6) + ret = inet_pton(AF_INET6, host, inp6); + else + ret = inet_pton(AF_INET, host, inp); + + if (ret != 1) { + struct hostent *hent; + + hent = gethostbyname(host); + if (!hent) { + log_err("fio: failed to resolve <%s>\n", host); + return 0; + } + + if (*ipv6) { + if (hent->h_addrtype != AF_INET6) { + log_info("fio: falling back to IPv4\n"); + *ipv6 = 0; + } else + memcpy(inp6, hent->h_addr_list[0], 16); + } + if (!*ipv6) { + if (hent->h_addrtype != AF_INET) { + log_err("fio: lookup type mismatch\n"); + return 0; + } + memcpy(inp, hent->h_addr_list[0], 4); + } + ret = 1; + } + + return !(ret == 1); +} + +/* + * Parse a host/ip/port string. Reads from 'str'. + * + * Outputs: + * + * For IPv4: + * *ptr is the host, *port is the port, inp is the destination. + * For IPv6: + * *ptr is the host, *port is the port, inp6 is the dest, and *ipv6 is 1. + * For local domain sockets: + * *ptr is the filename, *is_sock is 1. + */ int fio_server_parse_string(const char *str, char **ptr, int *is_sock, - int *port, struct in_addr *inp) + int *port, struct in_addr *inp, + struct in6_addr *inp6, int *ipv6) { + const char *host = str; + char *portp; + int lport = 0; + *ptr = NULL; *is_sock = 0; - *port = 0; + *port = fio_net_port; + *ipv6 = 0; if (!strncmp(str, "sock:", 5)) { *ptr = strdup(str + 5); *is_sock = 1; - } else { - const char *host = str; - char *portp; - int lport = 0; - - /* - * Is it ip::port - */ - if (!strncmp(host, "ip:", 3)) - host += 3; - else if (host[0] == ':') { - /* String is :port */ - host++; - lport = atoi(host); + + return 0; + } + + /* + * Is it ip::port + */ + if (!strncmp(host, "ip:", 3)) + host += 3; + else if (!strncmp(host, "ip4:", 4)) + host += 4; + else if (!strncmp(host, "ip6:", 4)) { + host += 4; + *ipv6 = 1; + } else if (host[0] == ':') { + /* String is :port */ + host++; + lport = atoi(host); + if (!lport || lport > 65535) { + log_err("fio: bad server port %u\n", port); + return 1; + } + /* no hostname given, we are done */ + *port = lport; + return 0; + } + + /* + * If no port seen yet, check if there's a last ':' at the end + */ + if (!lport) { + portp = strchr(host, ','); + if (portp) { + *portp = '\0'; + portp++; + lport = atoi(portp); if (!lport || lport > 65535) { log_err("fio: bad server port %u\n", port); return 1; } - /* no hostname given, we are done */ - *port = lport; - return 0; } + } - /* - * If no port seen yet, check if there's a last ':' at the end - */ - if (!lport) { - portp = strchr(host, ':'); - if (portp) { - *portp = '\0'; - portp++; - lport = atoi(portp); - if (!lport || lport > 65535) { - log_err("fio: bad server port %u\n", port); - return 1; - } - } - } - - if (lport) - *port = lport; - - *ptr = strdup(host); + if (lport) + *port = lport; - if (inet_aton(host, inp) != 1) { - struct hostent *hent; + if (!strlen(host)) + return 0; - hent = gethostbyname(host); - if (!hent) { - printf("FAIL\n"); - free(*ptr); - *ptr = NULL; - return 1; - } + *ptr = strdup(host); - memcpy(inp, hent->h_addr, 4); - } + if (fio_server_parse_host(*ptr, ipv6, inp, inp6)) { + free(*ptr); + *ptr = NULL; + return 1; } if (*port == 0) @@ -827,16 +1389,28 @@ int fio_server_parse_string(const char *str, char **ptr, int *is_sock, */ static int fio_handle_server_arg(void) { - int unused; + int port = fio_net_port; + int is_sock, ret = 0; saddr_in.sin_addr.s_addr = htonl(INADDR_ANY); - saddr_in.sin_port = htons(fio_net_port); if (!fio_server_arg) - return 0; + goto out; - return fio_server_parse_string(fio_server_arg, &bind_sock, &unused, - &fio_net_port, &saddr_in.sin_addr); + ret = fio_server_parse_string(fio_server_arg, &bind_sock, &is_sock, + &port, &saddr_in.sin_addr, + &saddr_in6.sin6_addr, &use_ipv6); + + if (!is_sock && bind_sock) { + free(bind_sock); + bind_sock = NULL; + } + +out: + fio_net_port = port; + saddr_in.sin_port = htons(port); + saddr_in6.sin6_port = htons(port); + return ret; } static int fio_server(void) @@ -866,52 +1440,106 @@ static int fio_server(void) return ret; } -static void sig_int(int sig) +void fio_server_got_signal(int signal) { - fio_terminate_threads(TERMINATE_ALL); - exit_backend = 1; + if (signal == SIGPIPE) + server_fd = -1; + else { + log_info("\nfio: terminating on signal %d\n", signal); + exit_backend = 1; + } } -static void server_signal_handler(void) +static int check_existing_pidfile(const char *pidfile) { - struct sigaction act; + struct stat sb; + char buf[16]; + pid_t pid; + FILE *f; + + if (stat(pidfile, &sb)) + return 0; + + f = fopen(pidfile, "r"); + if (!f) + return 0; + + if (fread(buf, sb.st_size, 1, f) <= 0) { + fclose(f); + return 1; + } + fclose(f); - memset(&act, 0, sizeof(act)); - act.sa_handler = sig_int; - act.sa_flags = SA_RESTART; - sigaction(SIGINT, &act, NULL); + pid = atoi(buf); + if (kill(pid, SIGCONT) < 0) + return errno != ESRCH; - memset(&act, 0, sizeof(act)); - act.sa_handler = sig_int; - act.sa_flags = SA_RESTART; - sigaction(SIGTERM, &act, NULL); + return 1; } -int fio_start_server(int daemonize) +static int write_pid(pid_t pid, const char *pidfile) +{ + FILE *fpid; + + fpid = fopen(pidfile, "w"); + if (!fpid) { + log_err("fio: failed opening pid file %s\n", pidfile); + return 1; + } + + fprintf(fpid, "%u\n", (unsigned int) pid); + fclose(fpid); + return 0; +} + +/* + * If pidfile is specified, background us. + */ +int fio_start_server(char *pidfile) { pid_t pid; + int ret; - server_signal_handler(); +#if defined(WIN32) + WSADATA wsd; + WSAStartup(MAKEWORD(2,2), &wsd); +#endif - if (!daemonize) + if (!pidfile) return fio_server(); - openlog("fio", LOG_NDELAY|LOG_NOWAIT|LOG_PID, LOG_USER); + if (check_existing_pidfile(pidfile)) { + log_err("fio: pidfile %s exists and server appears alive\n", + pidfile); + return -1; + } + pid = fork(); if (pid < 0) { - syslog(LOG_ERR, "failed server fork"); + log_err("fio: failed server fork: %s", strerror(errno)); + free(pidfile); return -1; - } else if (pid) - exit(0); + } else if (pid) { + int ret = write_pid(pid, pidfile); + + exit(ret); + } setsid(); + openlog("fio", LOG_NDELAY|LOG_NOWAIT|LOG_PID, LOG_USER); + log_syslog = 1; close(STDIN_FILENO); close(STDOUT_FILENO); close(STDERR_FILENO); f_out = NULL; f_err = NULL; - log_syslog = 1; - return fio_server(); + + ret = fio_server(); + + closelog(); + unlink(pidfile); + free(pidfile); + return ret; } void fio_server_set_arg(const char *arg)