X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=client.c;h=a84b92c4739ff7f83492b49d8567a70776893226;hp=8c85d2b5c6a673f433f2930fbb41c6a70e8b354b;hb=807f9971e7bfedfc905d2cb2c38a6e558db2f343;hpb=c2cb68696255a8fda93043a1f2219b60af614549;ds=sidebyside diff --git a/client.c b/client.c index 8c85d2b5..a84b92c4 100644 --- a/client.c +++ b/client.c @@ -16,47 +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; - int sent_job; +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; @@ -85,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); @@ -135,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); @@ -180,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; @@ -681,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; @@ -689,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]); } @@ -703,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; @@ -711,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]; } @@ -727,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); } } @@ -771,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) @@ -819,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; @@ -834,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: @@ -886,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); @@ -926,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); } @@ -980,7 +1010,7 @@ static int fio_client_timed_out(void) return ret; } -int fio_handle_clients(void) +int fio_handle_clients(struct client_ops *ops) { struct pollfd *pfds; int i, ret = 0, retval = 0; @@ -1001,7 +1031,7 @@ int fio_handle_clients(void) flist_for_each_safe(entry, tmp, &client_list) { client = flist_entry(entry, struct fio_client, list); - if (!client->sent_job && + if (!client->sent_job && !ops->stay_connected && flist_empty(&client->cmd_list)) { remove_client(client); continue; @@ -1048,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);