X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=client.c;h=c40511ebe553b3b0f5dd8079bdbb4c072a2c4140;hp=76935ca597b2fd6b17ec2aaa304ccaaa596771a6;hb=37f0c1ae23ad1716403d3d113c3dfdf41c47e329;hpb=af9c9fb34e420fc4d9cf317aa0f3cf6795a5a07f diff --git a/client.c b/client.c index 76935ca5..c40511eb 100644 --- a/client.c +++ b/client.c @@ -20,9 +20,15 @@ #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; @@ -35,7 +41,11 @@ struct fio_client { int skip_newline; int is_sock; - int waiting_eta; + + struct flist_head eta_list; + struct client_eta *eta_in_flight; + + struct flist_head cmd_list; uint16_t argc; char **argv; @@ -51,12 +61,15 @@ enum { Client_exited = 4, }; -struct client_eta { - struct jobs_eta eta; - unsigned int pending; -}; - static FLIST_HEAD(client_list); +static FLIST_HEAD(eta_list); + +static FLIST_HEAD(arg_list); + +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) @@ -64,6 +77,7 @@ static FLIST_HEAD(client_list); 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) { @@ -110,7 +124,10 @@ static void remove_client(struct fio_client *client) fio_client_remove_hash(client); - /* FIXME: check ->waiting_eta and handle it */ + 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) @@ -136,22 +153,49 @@ static void __fio_client_add_cmd_option(struct fio_client *client, void fio_client_add_cmd_option(void *cookie, const char *opt) { struct fio_client *client = cookie; + struct flist_head *entry; 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); + } } int fio_client_add(const char *hostname, void **cookie) { + struct fio_client *existing = *cookie; struct fio_client *client; + 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)); 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, @@ -230,6 +274,8 @@ static int fio_client_connect(struct fio_client *client) else fd = fio_client_connect_ip(client); + dprint(FD_NET, "client: %s connected %d\n", client->hostname, fd); + if (fd < 0) return 1; @@ -249,7 +295,7 @@ 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); } } @@ -278,8 +324,7 @@ 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); + 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) @@ -287,17 +332,22 @@ static int send_client_cmd_line(struct fio_client *client) 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 %d\n", client->argc); + lens = malloc(client->argc * sizeof(unsigned int)); + /* * Find out how much mem we need */ - for (i = 0, mem = 0; i < client->argc; i++) - mem += strlen(client->argv[i]) + 1; + 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 @@ -309,7 +359,7 @@ static int send_client_cmd_line(struct fio_client *client) offset = sizeof(*clp); for (i = 0; i < client->argc; i++) { - uint16_t arg_len = strlen(client->argv[i]) + 1; + uint16_t arg_len = lens[i]; cslp = pdu + offset; strcpy((char *) cslp->text, client->argv[i]); @@ -317,6 +367,7 @@ static int send_client_cmd_line(struct fio_client *client) offset += sizeof(*cslp) + arg_len; } + 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); @@ -525,6 +576,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) @@ -581,20 +646,53 @@ static void sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je) dst->eta_sec = je->eta_sec; } -static void handle_eta(struct fio_net_cmd *cmd) +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 == (uint64_t) icmd) + break; + + icmd = NULL; + } + + if (!icmd) { + log_err("fio: client: unable to find matching tag\n"); + return; + } + + flist_del(&icmd->list); + cmd->tag = icmd->saved_tag; + free(icmd); +} + +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 *) 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); - - if (!--eta->pending) { - display_thread_status(&eta->eta); - free(eta); - } + dec_jobs_eta(eta); } static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd) @@ -628,8 +726,8 @@ static int handle_client(struct fio_client *client) if (!cmd) return 0; - dprint(FD_NET, "client: got cmd op %d from %s\n", - cmd->opcode, client->hostname); + 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: @@ -660,10 +758,12 @@ static int handle_client(struct fio_client *client) free(cmd); break; case FIO_NET_CMD_ETA: - handle_eta(cmd); + 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; @@ -676,7 +776,7 @@ static int handle_client(struct fio_client *client) free(cmd); break; default: - log_err("fio: unknown client op: %d\n", cmd->opcode); + log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode)); free(cmd); break; } @@ -689,15 +789,10 @@ 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); - /* - * We need to do something more clever about checking status - * of command being send, client haven't sent previous ETA - * already, etc. - */ - eta = malloc(sizeof(*eta)); memset(&eta->eta, 0, sizeof(eta->eta)); eta->pending = nr_clients; @@ -705,14 +800,73 @@ static void request_client_etas(void) flist_for_each(entry, &client_list) { client = flist_entry(entry, struct fio_client, list); - client->waiting_eta = 1; + if (!flist_empty(&client->eta_list)) { + skipped++; + continue; + } + + 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, - (uint64_t) eta); + (uint64_t) eta, &client->cmd_list); } + 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) { struct fio_client *client; @@ -724,6 +878,10 @@ int fio_handle_clients(void) 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) { @@ -743,6 +901,9 @@ int fio_handle_clients(void) 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);