X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=client.c;h=fbeac35bfb30dd46635edfab70a6adb851b81c0e;hp=4d233415b8761aad83cb91a9633045dac46f0b3d;hb=498c92c27b475bcc0cb65e529fb02713bc1fb62c;hpb=60efd14e3b5b5a2adb9f7c9ecfb9bfba38f76ce9 diff --git a/client.c b/client.c index 4d233415..fbeac35b 100644 --- a/client.c +++ b/client.c @@ -8,65 +8,113 @@ #include #include #include -#include +#include +#include #include #include #include +#include #include "fio.h" #include "server.h" -#include "crc/crc32.h" #include "flist.h" +#include "hash.h" + +struct client_eta { + struct jobs_eta eta; + unsigned int pending; +}; struct fio_client { struct flist_head list; + struct flist_head hash_list; + struct flist_head arg_list; struct sockaddr_in addr; + struct sockaddr_un addr_un; char *hostname; + int port; int fd; + char *name; + int state; + int skip_newline; + int is_sock; + int disk_stats_shown; + unsigned int jobs; + int error; + + struct flist_head eta_list; + struct client_eta *eta_in_flight; + + struct flist_head cmd_list; uint16_t argc; char **argv; }; +static struct timeval eta_tv; + enum { Client_created = 0, Client_connected = 1, Client_started = 2, - Client_stopped = 3, - Client_exited = 4, + Client_running = 3, + Client_stopped = 4, + Client_exited = 5, }; static FLIST_HEAD(client_list); +static FLIST_HEAD(eta_list); -static int handle_client(struct fio_client *client, int one); +static FLIST_HEAD(arg_list); -static struct fio_client *find_client_by_fd(int fd) +static struct thread_stat client_ts; +static struct group_run_stats client_gs; +static int sum_stat_clients; +static int sum_stat_nr; + +#define FIO_CLIENT_HASH_BITS 7 +#define FIO_CLIENT_HASH_SZ (1 << FIO_CLIENT_HASH_BITS) +#define FIO_CLIENT_HASH_MASK (FIO_CLIENT_HASH_SZ - 1) +static struct flist_head client_hash[FIO_CLIENT_HASH_SZ]; + +static int handle_client(struct fio_client *client); +static void dec_jobs_eta(struct client_eta *eta); + +static void fio_client_add_hash(struct fio_client *client) { - struct fio_client *client; - struct flist_head *entry; + int bucket = hash_long(client->fd, FIO_CLIENT_HASH_BITS); - flist_for_each(entry, &client_list) { - client = flist_entry(entry, struct fio_client, list); + bucket &= FIO_CLIENT_HASH_MASK; + flist_add(&client->hash_list, &client_hash[bucket]); +} - if (client->fd == fd) - return client; - } +static void fio_client_remove_hash(struct fio_client *client) +{ + if (!flist_empty(&client->hash_list)) + flist_del_init(&client->hash_list); +} - return NULL; +static void fio_init fio_client_hash_init(void) +{ + int i; + + for (i = 0; i < FIO_CLIENT_HASH_SZ; i++) + INIT_FLIST_HEAD(&client_hash[i]); } -static struct fio_client *find_client_by_name(const char *name) +static struct fio_client *find_client_by_fd(int fd) { + int bucket = hash_long(fd, FIO_CLIENT_HASH_BITS) & FIO_CLIENT_HASH_MASK; struct fio_client *client; struct flist_head *entry; - flist_for_each(entry, &client_list) { - client = flist_entry(entry, struct fio_client, list); + flist_for_each(entry, &client_hash[bucket]) { + client = flist_entry(entry, struct fio_client, hash_list); - if (!strcmp(name, client->hostname)) + if (client->fd == fd) return client; } @@ -75,93 +123,169 @@ static struct fio_client *find_client_by_name(const char *name) static void remove_client(struct fio_client *client) { - dprint(FD_NET, "removed client <%s>\n", client->hostname); + dprint(FD_NET, "client: removed <%s>\n", client->hostname); flist_del(&client->list); - nr_clients--; + + fio_client_remove_hash(client); + + if (!flist_empty(&client->eta_list)) { + flist_del_init(&client->eta_list); + dec_jobs_eta(client->eta_in_flight); + } free(client->hostname); if (client->argv) free(client->argv); + if (client->name) + free(client->name); free(client); + nr_clients--; + sum_stat_clients--; } static void __fio_client_add_cmd_option(struct fio_client *client, const char *opt) { - client->argc++; + int index; + + index = client->argc++; client->argv = realloc(client->argv, sizeof(char *) * client->argc); - client->argv[client->argc - 1] = strdup(opt); + client->argv[index] = strdup(opt); + dprint(FD_NET, "client: add cmd %d: %s\n", index, opt); } -void fio_client_add_cmd_option(const char *hostname, const char *opt) +void fio_client_add_cmd_option(void *cookie, const char *opt) { - struct fio_client *client; - - if (!hostname || !opt) - return; + struct fio_client *client = cookie; + struct flist_head *entry; - client = find_client_by_name(hostname); - if (!client) { - log_err("fio: unknown client %s\n", hostname); + if (!client || !opt) return; - } __fio_client_add_cmd_option(client, opt); + + /* + * Duplicate arguments to shared client group + */ + flist_for_each(entry, &arg_list) { + client = flist_entry(entry, struct fio_client, arg_list); + + __fio_client_add_cmd_option(client, opt); + } } -void fio_client_add(const char *hostname) +int fio_client_add(const char *hostname, void **cookie) { + struct fio_client *existing = *cookie; struct fio_client *client; - dprint(FD_NET, "added client <%s>\n", hostname); + if (existing) { + /* + * We always add our "exec" name as the option, hence 1 + * means empty. + */ + if (existing->argc == 1) + flist_add_tail(&existing->arg_list, &arg_list); + else { + while (!flist_empty(&arg_list)) + flist_del_init(arg_list.next); + } + } + client = malloc(sizeof(*client)); memset(client, 0, sizeof(*client)); - client->hostname = strdup(hostname); + INIT_FLIST_HEAD(&client->list); + INIT_FLIST_HEAD(&client->hash_list); + INIT_FLIST_HEAD(&client->arg_list); + INIT_FLIST_HEAD(&client->eta_list); + INIT_FLIST_HEAD(&client->cmd_list); + + if (fio_server_parse_string(hostname, &client->hostname, + &client->is_sock, &client->port, + &client->addr.sin_addr)) + return -1; + client->fd = -1; __fio_client_add_cmd_option(client, "fio"); flist_add(&client->list, &client_list); nr_clients++; + dprint(FD_NET, "client: added <%s>\n", client->hostname); + *cookie = client; + return 0; } -static int fio_client_connect(struct fio_client *client) +static int fio_client_connect_ip(struct fio_client *client) { int fd; - dprint(FD_NET, "connect to host %s\n", client->hostname); - - memset(&client->addr, 0, sizeof(client->addr)); client->addr.sin_family = AF_INET; - client->addr.sin_port = htons(fio_net_port); - - if (inet_aton(client->hostname, &client->addr.sin_addr) != 1) { - struct hostent *hent; - - hent = gethostbyname(client->hostname); - if (!hent) { - log_err("fio: gethostbyname: %s\n", strerror(errno)); - return 1; - } - - memcpy(&client->addr.sin_addr, hent->h_addr, 4); - } + client->addr.sin_port = htons(client->port); fd = socket(AF_INET, SOCK_STREAM, 0); if (fd < 0) { log_err("fio: socket: %s\n", strerror(errno)); - return 1; + return -1; } if (connect(fd, (struct sockaddr *) &client->addr, sizeof(client->addr)) < 0) { log_err("fio: connect: %s\n", strerror(errno)); - log_err("fio: failed to connect to %s\n", client->hostname); - return 1; + log_err("fio: failed to connect to %s:%u\n", client->hostname, + client->port); + close(fd); + return -1; } + return fd; +} + +static int fio_client_connect_sock(struct fio_client *client) +{ + struct sockaddr_un *addr = &client->addr_un; + fio_socklen_t len; + int fd; + + memset(addr, 0, sizeof(*addr)); + addr->sun_family = AF_UNIX; + strcpy(addr->sun_path, client->hostname); + + fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (fd < 0) { + log_err("fio: socket: %s\n", strerror(errno)); + return -1; + } + + len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1; + if (connect(fd, (struct sockaddr *) addr, len) < 0) { + log_err("fio: connect; %s\n", strerror(errno)); + close(fd); + return -1; + } + + return fd; +} + +static int fio_client_connect(struct fio_client *client) +{ + int fd; + + dprint(FD_NET, "client: connect to host %s\n", client->hostname); + + if (client->is_sock) + fd = fio_client_connect_sock(client); + else + fd = fio_client_connect_ip(client); + + dprint(FD_NET, "client: %s connected %d\n", client->hostname, fd); + + if (fd < 0) + return 1; + client->fd = fd; + fio_client_add_hash(client); client->state = Client_connected; return 0; } @@ -176,13 +300,13 @@ void fio_clients_terminate(void) flist_for_each(entry, &client_list) { client = flist_entry(entry, struct fio_client, list); - fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0); + fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0, NULL); } } static void sig_int(int sig) { - dprint(FD_NET, "client: got sign %d\n", sig); + dprint(FD_NET, "client: got signal %d\n", sig); fio_clients_terminate(); } @@ -205,23 +329,52 @@ static void probe_client(struct fio_client *client) { dprint(FD_NET, "client: send probe\n"); - fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0); - handle_client(client, 1); + fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0, &client->cmd_list); } static int send_client_cmd_line(struct fio_client *client) { - struct cmd_line_pdu *pdu; + struct cmd_single_line_pdu *cslp; + struct cmd_line_pdu *clp; + unsigned long offset; + unsigned int *lens; + void *pdu; + size_t mem; int i, ret; - dprint(FD_NET, "client: send cmdline\n"); + dprint(FD_NET, "client: send cmdline %d\n", client->argc); - pdu = malloc(sizeof(*pdu)); - for (i = 0; i < client->argc; i++) - strcpy((char *) pdu->argv[i], client->argv[i]); + lens = malloc(client->argc * sizeof(unsigned int)); + + /* + * Find out how much mem we need + */ + for (i = 0, mem = 0; i < client->argc; i++) { + lens[i] = strlen(client->argv[i]) + 1; + mem += lens[i]; + } + + /* + * We need one cmd_line_pdu, and argc number of cmd_single_line_pdu + */ + mem += sizeof(*clp) + (client->argc * sizeof(*cslp)); + + pdu = malloc(mem); + clp = pdu; + offset = sizeof(*clp); + + for (i = 0; i < client->argc; i++) { + uint16_t arg_len = lens[i]; + + cslp = pdu + offset; + strcpy((char *) cslp->text, client->argv[i]); + cslp->len = cpu_to_le16(arg_len); + offset += sizeof(*cslp) + arg_len; + } - pdu->argc = cpu_to_le16(client->argc); - ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, sizeof(*pdu)); + free(lens); + clp->lines = cpu_to_le16(client->argc); + ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0); free(pdu); return ret; } @@ -269,12 +422,13 @@ static int fio_client_send_ini(struct fio_client *client, const char *filename) fd = open(filename, O_RDONLY); if (fd < 0) { - log_err("fio: job file open: %s\n", strerror(errno)); + log_err("fio: job file <%s> open: %s\n", filename, strerror(errno)); return 1; } if (fstat(fd, &sb) < 0) { log_err("fio: job file stat: %s\n", strerror(errno)); + close(fd); return 1; } @@ -298,11 +452,14 @@ static int fio_client_send_ini(struct fio_client *client, const char *filename) if (len) { log_err("fio: failed reading job file %s\n", filename); + close(fd); + free(buf); return 1; } - ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size); + ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size, 0); free(buf); + close(fd); return ret; } @@ -326,9 +483,12 @@ static void convert_io_stat(struct io_stat *dst, struct io_stat *src) dst->max_val = le64_to_cpu(src->max_val); dst->min_val = le64_to_cpu(src->min_val); dst->samples = le64_to_cpu(src->samples); - /* FIXME */ - dst->mean = __le64_to_cpu(src->mean); - dst->S = __le64_to_cpu(src->S); + + /* + * Floats arrive as IEEE 754 encoded uint64_t, convert back to double + */ + dst->mean.u.f = fio_uint64_to_double(le64_to_cpu(dst->mean.u.i)); + dst->S.u.f = fio_uint64_to_double(le64_to_cpu(dst->S.u.i)); } static void convert_ts(struct thread_stat *dst, struct thread_stat *src) @@ -353,7 +513,13 @@ static void convert_ts(struct thread_stat *dst, struct thread_stat *src) dst->minf = le64_to_cpu(src->minf); dst->majf = le64_to_cpu(src->majf); dst->clat_percentiles = le64_to_cpu(src->clat_percentiles); - dst->percentile_list = NULL; + + for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) { + fio_fp64_t *fps = &src->percentile_list[i]; + fio_fp64_t *fpd = &dst->percentile_list[i]; + + fpd->u.f = fio_uint64_to_double(le64_to_cpu(fps->u.i)); + } for (i = 0; i < FIO_IO_U_MAP_NR; i++) { dst->io_u_map[i] = le32_to_cpu(src->io_u_map[i]); @@ -415,6 +581,20 @@ static void handle_ts(struct fio_net_cmd *cmd) convert_gs(&p->rs, &p->rs); show_thread_status(&p->ts, &p->rs); + + if (sum_stat_clients == 1) + return; + + sum_thread_stats(&client_ts, &p->ts, sum_stat_nr); + sum_group_stats(&client_gs, &p->rs); + + client_ts.members++; + client_ts.groupid = p->ts.groupid; + + if (++sum_stat_nr == sum_stat_clients) { + strcpy(client_ts.name, "All clients"); + show_thread_status(&client_ts, &client_gs); + } } static void handle_gs(struct fio_net_cmd *cmd) @@ -425,9 +605,56 @@ static void handle_gs(struct fio_net_cmd *cmd) show_group_stats(gs); } -static void handle_eta(struct fio_net_cmd *cmd) +static void convert_agg(struct disk_util_agg *agg) +{ + int i; + + for (i = 0; i < 2; i++) { + agg->ios[i] = le32_to_cpu(agg->ios[i]); + agg->merges[i] = le32_to_cpu(agg->merges[i]); + agg->sectors[i] = le64_to_cpu(agg->sectors[i]); + agg->ticks[i] = le32_to_cpu(agg->ticks[i]); + } + + agg->io_ticks = le32_to_cpu(agg->io_ticks); + agg->time_in_queue = le32_to_cpu(agg->time_in_queue); + agg->slavecount = le32_to_cpu(agg->slavecount); + agg->max_util.u.f = __le64_to_cpu(fio_uint64_to_double(agg->max_util.u.i)); +} + +static void convert_dus(struct disk_util_stat *dus) +{ + int i; + + for (i = 0; i < 2; i++) { + dus->ios[i] = le32_to_cpu(dus->ios[i]); + dus->merges[i] = le32_to_cpu(dus->merges[i]); + dus->sectors[i] = le64_to_cpu(dus->sectors[i]); + dus->ticks[i] = le32_to_cpu(dus->ticks[i]); + } + + dus->io_ticks = le32_to_cpu(dus->io_ticks); + dus->time_in_queue = le32_to_cpu(dus->time_in_queue); + dus->msec = le64_to_cpu(dus->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; + + convert_dus(&du->dus); + convert_agg(&du->agg); + + if (!client->disk_stats_shown) { + client->disk_stats_shown = 1; + log_info("\nDisk stats (read/write):\n"); + } + + print_disk_util(&du->dus, &du->agg, terse_output); +} + +static void convert_jobs_eta(struct jobs_eta *je) { - struct jobs_eta *je = (struct jobs_eta *) cmd->payload; int i; je->nr_running = le32_to_cpu(je->nr_running); @@ -444,83 +671,285 @@ static void handle_eta(struct fio_net_cmd *cmd) je->iops[i] = le32_to_cpu(je->iops[i]); } - je->elapsed_sec = le32_to_cpu(je->nr_running); + je->elapsed_sec = le64_to_cpu(je->elapsed_sec); je->eta_sec = le64_to_cpu(je->eta_sec); +} + +static void 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->files_open += je->files_open; + dst->m_rate += je->m_rate; + dst->t_rate += je->t_rate; + dst->m_iops += je->m_iops; + dst->t_iops += je->t_iops; + + for (i = 0; i < 2; 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; +} + +static void dec_jobs_eta(struct client_eta *eta) +{ + if (!--eta->pending) { + display_thread_status(&eta->eta); + free(eta); + } +} + +static void remove_reply_cmd(struct fio_client *client, struct fio_net_cmd *cmd) +{ + struct fio_net_int_cmd *icmd = NULL; + struct flist_head *entry; + + flist_for_each(entry, &client->cmd_list) { + icmd = flist_entry(entry, struct fio_net_int_cmd, list); + + if (cmd->tag == (uintptr_t) icmd) + break; + + icmd = NULL; + } + + if (!icmd) { + log_err("fio: client: unable to find matching tag\n"); + return; + } - display_thread_status(je); + flist_del(&icmd->list); + cmd->tag = icmd->saved_tag; + free(icmd); } -static void handle_probe(struct fio_net_cmd *cmd) +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); + + convert_jobs_eta(je); + sum_jobs_eta(&eta->eta, je); + dec_jobs_eta(eta); +} + +static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd) { struct cmd_probe_pdu *probe = (struct cmd_probe_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"; - log_info("Probe: hostname=%s, fio ver %u.%u.%u\n", probe->hostname, - probe->fio_major, probe->fio_minor, probe->fio_patch); + sprintf(bit, "%d-bit", probe->bpp * 8); + + log_info("hostname=%s, be=%u, %s, os=%s, arch=%s, fio=%u.%u.%u\n", + probe->hostname, probe->bigendian, bit, os, arch, + probe->fio_major, probe->fio_minor, probe->fio_patch); + + if (!client->name) + client->name = strdup((char *) probe->hostname); } -static int handle_client(struct fio_client *client, int one) +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); +} + +static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd) +{ + struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload; + + client->state = Client_stopped; + client->error = le32_to_cpu(pdu->error); + + if (client->error) + log_info("client <%s>: exited with error %d\n", client->hostname, client->error); +} + +static int handle_client(struct fio_client *client) { struct fio_net_cmd *cmd; - int done = 0; dprint(FD_NET, "client: handle %s\n", client->hostname); - while ((cmd = fio_net_recv_cmd(client->fd, 1)) != NULL) { - dprint(FD_NET, "%s: got cmd op %d\n", client->hostname, - cmd->opcode); + cmd = fio_net_recv_cmd(client->fd); + if (!cmd) + return 0; + + dprint(FD_NET, "client: got cmd op %s from %s\n", + fio_server_op(cmd->opcode), client->hostname); + + switch (cmd->opcode) { + case FIO_NET_CMD_QUIT: + remove_client(client); + free(cmd); + break; + case FIO_NET_CMD_TEXT: { + const char *buf = (const char *) cmd->payload; + 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, cmd->pdu_len, 1, f_out); + fflush(f_out); + client->skip_newline = strchr(buf, '\n') == NULL; + free(cmd); + break; + } + case FIO_NET_CMD_DU: + handle_du(client, cmd); + free(cmd); + break; + case FIO_NET_CMD_TS: + handle_ts(cmd); + free(cmd); + break; + case FIO_NET_CMD_GS: + handle_gs(cmd); + free(cmd); + break; + case FIO_NET_CMD_ETA: + remove_reply_cmd(client, cmd); + handle_eta(client, cmd); + free(cmd); + break; + case FIO_NET_CMD_PROBE: + remove_reply_cmd(client, cmd); + handle_probe(client, cmd); + free(cmd); + break; + case FIO_NET_CMD_RUN: + client->state = Client_running; + free(cmd); + break; + case FIO_NET_CMD_START: + handle_start(client, cmd); + free(cmd); + break; + case FIO_NET_CMD_STOP: + handle_stop(client, cmd); + free(cmd); + break; + default: + log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode)); + free(cmd); + break; + } - switch (cmd->opcode) { - case FIO_NET_CMD_QUIT: - remove_client(client); - free(cmd); - done = 1; - break; - case FIO_NET_CMD_TEXT: { - const char *buf = (const char *) cmd->payload; - - if (!client->skip_newline) - fprintf(f_out, "Client <%s>: ", client->hostname); - fwrite(buf, cmd->pdu_len, 1, f_out); - fflush(f_out); - client->skip_newline = strchr(buf, '\n') == NULL; - free(cmd); - break; - } - case FIO_NET_CMD_TS: - handle_ts(cmd); - free(cmd); - break; - case FIO_NET_CMD_GS: - handle_gs(cmd); - free(cmd); - break; - case FIO_NET_CMD_ETA: - handle_eta(cmd); - free(cmd); - break; - case FIO_NET_CMD_PROBE: - handle_probe(cmd); - free(cmd); - break; - case FIO_NET_CMD_START: - client->state = Client_started; - free(cmd); - break; - case FIO_NET_CMD_STOP: - client->state = Client_stopped; - free(cmd); - break; - default: - log_err("fio: unknown client op: %d\n", cmd->opcode); - free(cmd); - break; + return 1; +} + +static void request_client_etas(void) +{ + struct fio_client *client; + struct flist_head *entry; + struct client_eta *eta; + int skipped = 0; + + dprint(FD_NET, "client: request eta (%d)\n", nr_clients); + + eta = malloc(sizeof(*eta)); + memset(&eta->eta, 0, sizeof(eta->eta)); + eta->pending = nr_clients; + + flist_for_each(entry, &client_list) { + client = flist_entry(entry, struct fio_client, list); + + if (!flist_empty(&client->eta_list)) { + skipped++; + continue; } + if (client->state != Client_running) + continue; - if (done || one) - break; + assert(!client->eta_in_flight); + flist_add_tail(&client->eta_list, &eta_list); + client->eta_in_flight = eta; + fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_SEND_ETA, + (uintptr_t) eta, &client->cmd_list); } - return 0; + while (skipped--) + dec_jobs_eta(eta); + + dprint(FD_NET, "client: requested eta tag %p\n", eta); +} + +static int client_check_cmd_timeout(struct fio_client *client, + struct timeval *now) +{ + struct fio_net_int_cmd *cmd; + struct flist_head *entry, *tmp; + int ret = 0; + + flist_for_each_safe(entry, tmp, &client->cmd_list) { + cmd = flist_entry(entry, struct fio_net_int_cmd, list); + + if (mtime_since(&cmd->tv, now) < FIO_NET_CLIENT_TIMEOUT) + continue; + + log_err("fio: client %s, timeout on cmd %s\n", client->hostname, + fio_server_op(cmd->cmd.opcode)); + flist_del(&cmd->list); + free(cmd); + ret = 1; + } + + return flist_empty(&client->cmd_list) && ret; +} + +static int fio_client_timed_out(void) +{ + struct fio_client *client; + struct flist_head *entry, *tmp; + struct timeval tv; + int ret = 0; + + gettimeofday(&tv, NULL); + + flist_for_each_safe(entry, tmp, &client_list) { + client = flist_entry(entry, struct fio_client, list); + + if (flist_empty(&client->cmd_list)) + continue; + + if (!client_check_cmd_timeout(client, &tv)) + continue; + + log_err("fio: client %s timed out\n", client->hostname); + remove_client(client); + ret = 1; + } + + return ret; } int fio_handle_clients(void) @@ -528,10 +957,16 @@ int fio_handle_clients(void) struct fio_client *client; struct flist_head *entry; struct pollfd *pfds; - int i, ret = 0; + int i, ret = 0, retval = 0; + + gettimeofday(&eta_tv, NULL); pfds = malloc(nr_clients * sizeof(struct pollfd)); + sum_stat_clients = nr_clients; + init_thread_stat(&client_ts); + init_group_run_stat(&client_gs); + while (!exit_backend && nr_clients) { i = 0; flist_for_each(entry, &client_list) { @@ -545,6 +980,17 @@ int fio_handle_clients(void) assert(i == nr_clients); do { + struct timeval tv; + + gettimeofday(&tv, NULL); + if (mtime_since(&eta_tv, &tv) >= 900) { + request_client_etas(); + memcpy(&eta_tv, &tv, sizeof(tv)); + + if (fio_client_timed_out()) + break; + } + ret = poll(pfds, nr_clients, 100); if (ret < 0) { if (errno == EINTR) @@ -561,13 +1007,20 @@ int fio_handle_clients(void) client = find_client_by_fd(pfds[i].fd); if (!client) { - log_err("fio: unknown client\n"); + log_err("fio: unknown client fd %d\n", pfds[i].fd); continue; } - handle_client(client, 0); + if (!handle_client(client)) { + log_info("client: host=%s disconnected\n", + client->hostname); + remove_client(client); + retval = 1; + } + if (client->error) + retval = 1; } } free(pfds); - return 0; + return retval; }