X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=client.c;h=a84b92c4739ff7f83492b49d8567a70776893226;hp=ace70d357b97a3e882bacfea811e265779a85b44;hb=807f9971e7bfedfc905d2cb2c38a6e558db2f343;hpb=93bcfd20e37cef8cec350fe06d3a086724c9f257 diff --git a/client.c b/client.c index ace70d35..a84b92c4 100644 --- a/client.c +++ b/client.c @@ -16,46 +16,39 @@ #include #include "fio.h" +#include "client.h" #include "server.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; - union { - struct sockaddr_in addr; - struct sockaddr_in6 addr6; - 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; - int ipv6; +static void fio_client_text_op(struct fio_client *client, + FILE *f, __u16 pdu_len, const char *buf) +{ + const char *name; + int fio_unused ret; - struct flist_head eta_list; - struct client_eta *eta_in_flight; + name = client->name ? client->name : client->hostname; - struct flist_head cmd_list; + if (!client->skip_newline) + fprintf(f, "<%s> ", name); + ret = fwrite(buf, pdu_len, 1, f); + fflush(f); + client->skip_newline = strchr(buf, '\n') == NULL; +} - uint16_t argc; - char **argv; +static void handle_du(struct fio_client *client, struct fio_net_cmd *cmd); +static void handle_ts(struct fio_net_cmd *cmd); +static void handle_gs(struct fio_net_cmd *cmd); +static void handle_eta(struct fio_client *client, struct fio_net_cmd *cmd); +static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd); + +struct client_ops fio_client_ops = { + .text_op = fio_client_text_op, + .disk_util = handle_du, + .thread_status = handle_ts, + .group_stats = handle_gs, + .eta = handle_eta, + .probe = handle_probe, }; static struct timeval eta_tv; @@ -84,9 +77,6 @@ static int sum_stat_nr; #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) { int bucket = hash_long(client->fd, FIO_CLIENT_HASH_BITS); @@ -134,7 +124,7 @@ static void remove_client(struct fio_client *client) if (!flist_empty(&client->eta_list)) { flist_del_init(&client->eta_list); - dec_jobs_eta(client->eta_in_flight); + fio_client_dec_jobs_eta(client->eta_in_flight, display_thread_status); } free(client->hostname); @@ -179,6 +169,49 @@ void fio_client_add_cmd_option(void *cookie, const char *opt) } } +struct fio_client *fio_client_add_explicit(const char *hostname, int type, + int port) +{ + struct fio_client *client; + + 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); + + client->hostname = strdup(hostname); + + if (type == Fio_client_socket) + client->is_sock = 1; + else { + int ipv6; + + ipv6 = type == Fio_client_ipv6; + if (fio_server_parse_host(hostname, &ipv6, + &client->addr.sin_addr, + &client->addr6.sin6_addr)) + goto err; + + client->port = port; + } + + 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); + return client; +err: + free(client); + return NULL; +} + int fio_client_add(const char *hostname, void **cookie) { struct fio_client *existing = *cookie; @@ -481,6 +514,7 @@ static int fio_client_send_ini(struct fio_client *client, const char *filename) return 1; } + client->sent_job = 1; ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size, 0); free(buf); close(fd); @@ -497,6 +531,8 @@ int fio_clients_send_ini(const char *filename) if (fio_client_send_ini(client, filename)) remove_client(client); + + client->sent_job = 1; } return !nr_clients; @@ -677,7 +713,7 @@ static void handle_du(struct fio_client *client, struct fio_net_cmd *cmd) print_disk_util(&du->dus, &du->agg, terse_output); } -static void convert_jobs_eta(struct jobs_eta *je) +void fio_client_convert_jobs_eta(struct jobs_eta *je) { int i; @@ -685,12 +721,12 @@ static void convert_jobs_eta(struct jobs_eta *je) je->nr_ramp = le32_to_cpu(je->nr_ramp); je->nr_pending = le32_to_cpu(je->nr_pending); je->files_open = le32_to_cpu(je->files_open); - je->m_rate = le32_to_cpu(je->m_rate); - je->t_rate = le32_to_cpu(je->t_rate); - je->m_iops = le32_to_cpu(je->m_iops); - je->t_iops = le32_to_cpu(je->t_iops); for (i = 0; i < 2; i++) { + je->m_rate[i] = le32_to_cpu(je->m_rate[i]); + je->t_rate[i] = le32_to_cpu(je->t_rate[i]); + je->m_iops[i] = le32_to_cpu(je->m_iops[i]); + je->t_iops[i] = le32_to_cpu(je->t_iops[i]); je->rate[i] = le32_to_cpu(je->rate[i]); je->iops[i] = le32_to_cpu(je->iops[i]); } @@ -699,7 +735,7 @@ static void convert_jobs_eta(struct jobs_eta *je) je->eta_sec = le64_to_cpu(je->eta_sec); } -static void sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je) +void fio_client_sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je) { int i; @@ -707,12 +743,12 @@ static void sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je) 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->m_rate[i] += je->m_rate[i]; + dst->t_rate[i] += je->t_rate[i]; + dst->m_iops[i] += je->m_iops[i]; + dst->t_iops[i] += je->t_iops[i]; dst->rate[i] += je->rate[i]; dst->iops[i] += je->iops[i]; } @@ -723,10 +759,10 @@ static void sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je) dst->eta_sec = je->eta_sec; } -static void dec_jobs_eta(struct client_eta *eta) +void fio_client_dec_jobs_eta(struct client_eta *eta, void (*fn)(struct jobs_eta *)) { if (!--eta->pending) { - display_thread_status(&eta->eta); + fn(&eta->eta); free(eta); } } @@ -767,9 +803,9 @@ static void handle_eta(struct fio_client *client, struct fio_net_cmd *cmd) 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); + fio_client_convert_jobs_eta(je); + fio_client_sum_jobs_eta(&eta->eta, je); + fio_client_dec_jobs_eta(eta, display_thread_status); } static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd) @@ -815,7 +851,7 @@ static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd) log_info("client <%s>: exited with error %d\n", client->hostname, client->error); } -static int handle_client(struct fio_client *client) +int fio_handle_client(struct fio_client *client, struct client_ops *ops) { struct fio_net_cmd *cmd; @@ -830,44 +866,37 @@ static int handle_client(struct fio_client *client) switch (cmd->opcode) { case FIO_NET_CMD_QUIT: + if (ops->quit) + ops->quit(client); 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; + ops->text_op(client, f_out, cmd->pdu_len, buf); free(cmd); break; } case FIO_NET_CMD_DU: - handle_du(client, cmd); + ops->disk_util(client, cmd); free(cmd); break; case FIO_NET_CMD_TS: - handle_ts(cmd); + ops->thread_status(cmd); free(cmd); break; case FIO_NET_CMD_GS: - handle_gs(cmd); + ops->group_stats(cmd); free(cmd); break; case FIO_NET_CMD_ETA: remove_reply_cmd(client, cmd); - handle_eta(client, cmd); + ops->eta(client, cmd); free(cmd); break; case FIO_NET_CMD_PROBE: remove_reply_cmd(client, cmd); - handle_probe(client, cmd); + ops->probe(client, cmd); free(cmd); break; case FIO_NET_CMD_RUN: @@ -882,6 +911,11 @@ static int handle_client(struct fio_client *client) handle_stop(client, cmd); free(cmd); break; + case FIO_NET_CMD_ADD_JOB: + if (ops->add_job) + ops->add_job(client, cmd); + free(cmd); + break; default: log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode)); free(cmd); @@ -922,7 +956,7 @@ static void request_client_etas(void) } while (skipped--) - dec_jobs_eta(eta); + fio_client_dec_jobs_eta(eta, display_thread_status); dprint(FD_NET, "client: requested eta tag %p\n", eta); } @@ -976,10 +1010,8 @@ static int fio_client_timed_out(void) return ret; } -int fio_handle_clients(void) +int fio_handle_clients(struct client_ops *ops) { - struct fio_client *client; - struct flist_head *entry; struct pollfd *pfds; int i, ret = 0, retval = 0; @@ -992,15 +1024,27 @@ int fio_handle_clients(void) init_group_run_stat(&client_gs); while (!exit_backend && nr_clients) { + struct flist_head *entry, *tmp; + struct fio_client *client; + i = 0; - flist_for_each(entry, &client_list) { + flist_for_each_safe(entry, tmp, &client_list) { client = flist_entry(entry, struct fio_client, list); + if (!client->sent_job && !ops->stay_connected && + flist_empty(&client->cmd_list)) { + remove_client(client); + continue; + } + pfds[i].fd = client->fd; pfds[i].events = POLLIN; i++; } + if (!nr_clients) + break; + assert(i == nr_clients); do { @@ -1034,7 +1078,7 @@ int fio_handle_clients(void) log_err("fio: unknown client fd %d\n", pfds[i].fd); continue; } - if (!handle_client(client)) { + if (!fio_handle_client(client, ops)) { log_info("client: host=%s disconnected\n", client->hostname); remove_client(client);