#include "flist.h"
#include "hash.h"
-extern void (*update_thread_status)(char *status_message, double perc);
-
-struct client_eta {
- struct jobs_eta eta;
- unsigned int pending;
-};
-
-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;
-
- name = client->name ? client->name : client->hostname;
-
- if (!client->skip_newline)
- fprintf(f, "<%s> ", name);
- ret = fwrite(buf, pdu_len, 1, f);
- fflush(f);
- client->skip_newline = strchr(buf, '\n') == NULL;
-}
-
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_ts(struct fio_client *client, struct fio_net_cmd *cmd);
+static void handle_gs(struct fio_client *client, struct fio_net_cmd *cmd);
static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd);
+static void handle_text(struct fio_client *client, struct fio_net_cmd *cmd);
struct client_ops fio_client_ops = {
- .text_op = fio_client_text_op,
+ .text_op = handle_text,
.disk_util = handle_du,
.thread_status = handle_ts,
.group_stats = handle_gs,
- .eta = handle_eta,
+ .eta = display_thread_status,
.probe = handle_probe,
- /* status display, if NULL, printf is used */
};
static struct timeval eta_tv;
static FLIST_HEAD(arg_list);
-static struct thread_stat client_ts;
-static struct group_run_stats client_gs;
-static int sum_stat_clients;
+struct thread_stat client_ts;
+struct group_run_stats client_gs;
+int sum_stat_clients;
+
static int sum_stat_nr;
#define FIO_CLIENT_HASH_BITS 7
#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, struct client_ops *ops);
-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);
flist_for_each(entry, &client_hash[bucket]) {
client = flist_entry(entry, struct fio_client, hash_list);
- if (client->fd == fd)
+ if (client->fd == fd) {
+ client->refs++;
return client;
+ }
}
return NULL;
static void remove_client(struct fio_client *client)
{
+ assert(client->refs);
+
+ if (--client->refs)
+ return;
+
dprint(FD_NET, "client: removed <%s>\n", client->hostname);
flist_del(&client->list);
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, client->ops->eta);
}
free(client->hostname);
sum_stat_clients--;
}
+static void put_client(struct fio_client *client)
+{
+ remove_client(client);
+}
+
static void __fio_client_add_cmd_option(struct fio_client *client,
const char *opt)
{
}
}
-int fio_client_add(const char *hostname, void **cookie)
+struct fio_client *fio_client_add_explicit(struct client_ops *ops,
+ 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;
+ client->ops = ops;
+ client->refs = 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(struct client_ops *ops, const char *hostname, void **cookie)
{
struct fio_client *existing = *cookie;
struct fio_client *client;
return -1;
client->fd = -1;
+ client->ops = ops;
+ client->refs = 1;
__fio_client_add_cmd_option(client, "fio");
dst->groupid = le32_to_cpu(src->groupid);
}
-static void handle_ts(struct fio_net_cmd *cmd)
+static void handle_ts(struct fio_client *client, struct fio_net_cmd *cmd)
{
struct cmd_ts_pdu *p = (struct cmd_ts_pdu *) cmd->payload;
- convert_ts(&p->ts, &p->ts);
- convert_gs(&p->rs, &p->rs);
-
show_thread_status(&p->ts, &p->rs);
if (sum_stat_clients == 1)
}
}
-static void handle_gs(struct fio_net_cmd *cmd)
+static void handle_gs(struct fio_client *client, struct fio_net_cmd *cmd)
{
struct group_run_stats *gs = (struct group_run_stats *) cmd->payload;
- convert_gs(gs, gs);
show_group_stats(gs);
}
+static void handle_text(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmd->payload;
+ const char *buf = (const char *) pdu->buf;
+ 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, pdu->buf_len, 1, f_out);
+ fflush(f_out);
+ client->skip_newline = strchr(buf, '\n') == NULL;
+}
+
static void convert_agg(struct disk_util_agg *agg)
{
int i;
{
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");
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]);
}
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;
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];
}
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, client_eta_op eta_fn)
{
if (!--eta->pending) {
- display_thread_status(&eta->eta);
+ eta_fn(&eta->eta);
free(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);
+ fio_client_sum_jobs_eta(&eta->eta, je);
+ fio_client_dec_jobs_eta(eta, client->ops->eta);
}
static void handle_probe(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, struct client_ops *ops)
+static void convert_text(struct fio_net_cmd *cmd)
{
+ struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmd->payload;
+
+ pdu->level = le32_to_cpu(pdu->level);
+ pdu->buf_len = le32_to_cpu(pdu->buf_len);
+ pdu->log_sec = le64_to_cpu(pdu->log_sec);
+ pdu->log_usec = le64_to_cpu(pdu->log_usec);
+}
+
+int fio_handle_client(struct fio_client *client)
+{
+ struct client_ops *ops = client->ops;
struct fio_net_cmd *cmd;
dprint(FD_NET, "client: handle %s\n", client->hostname);
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;
- ops->text_op(client, f_out, cmd->pdu_len, buf);
+ case FIO_NET_CMD_TEXT:
+ convert_text(cmd);
+ ops->text_op(client, cmd);
free(cmd);
break;
- }
- case FIO_NET_CMD_DU:
+ case FIO_NET_CMD_DU: {
+ struct cmd_du_pdu *du = (struct cmd_du_pdu *) cmd->payload;
+
+ convert_dus(&du->dus);
+ convert_agg(&du->agg);
+
ops->disk_util(client, cmd);
free(cmd);
break;
- case FIO_NET_CMD_TS:
- ops->thread_status(cmd);
+ }
+ case FIO_NET_CMD_TS: {
+ struct cmd_ts_pdu *p = (struct cmd_ts_pdu *) cmd->payload;
+
+ convert_ts(&p->ts, &p->ts);
+ convert_gs(&p->rs, &p->rs);
+
+ ops->thread_status(client, cmd);
free(cmd);
break;
- case FIO_NET_CMD_GS:
- ops->group_stats(cmd);
+ }
+ case FIO_NET_CMD_GS: {
+ struct group_run_stats *gs = (struct group_run_stats *) cmd->payload;
+
+ convert_gs(gs, gs);
+
+ ops->group_stats(client, cmd);
free(cmd);
break;
- case FIO_NET_CMD_ETA:
+ }
+ case FIO_NET_CMD_ETA: {
+ struct jobs_eta *je = (struct jobs_eta *) cmd->payload;
+
remove_reply_cmd(client, cmd);
- ops->eta(client, cmd);
+ convert_jobs_eta(je);
+ handle_eta(client, cmd);
free(cmd);
break;
+ }
case FIO_NET_CMD_PROBE:
remove_reply_cmd(client, cmd);
ops->probe(client, cmd);
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);
return 1;
}
-static void request_client_etas(void)
+static void request_client_etas(struct client_ops *ops)
{
struct fio_client *client;
struct flist_head *entry;
}
while (skipped--)
- dec_jobs_eta(eta);
+ fio_client_dec_jobs_eta(eta, ops->eta);
dprint(FD_NET, "client: requested eta tag %p\n", eta);
}
return flist_empty(&client->cmd_list) && ret;
}
-static int fio_client_timed_out(void)
+static int fio_check_clients_timed_out(void)
{
struct fio_client *client;
struct flist_head *entry, *tmp;
if (!client_check_cmd_timeout(client, &tv))
continue;
- log_err("fio: client %s timed out\n", client->hostname);
+ if (client->ops->timed_out)
+ client->ops->timed_out(client);
+ else
+ log_err("fio: client %s timed out\n", client->hostname);
+
remove_client(client);
ret = 1;
}
init_thread_stat(&client_ts);
init_group_run_stat(&client_gs);
- /* Used by eta.c:display_thread_status() */
- update_thread_status = ops->thread_status_display;
-
while (!exit_backend && nr_clients) {
struct flist_head *entry, *tmp;
struct fio_client *client;
flist_for_each_safe(entry, tmp, &client_list) {
client = flist_entry(entry, struct fio_client, list);
- if (!client->sent_job &&
+ if (!client->sent_job && !client->ops->stay_connected &&
flist_empty(&client->cmd_list)) {
remove_client(client);
continue;
gettimeofday(&tv, NULL);
if (mtime_since(&eta_tv, &tv) >= 900) {
- request_client_etas();
+ request_client_etas(ops);
memcpy(&eta_tv, &tv, sizeof(tv));
- if (fio_client_timed_out())
+ if (fio_check_clients_timed_out())
break;
}
log_err("fio: unknown client fd %d\n", pfds[i].fd);
continue;
}
- if (!handle_client(client, ops)) {
+ if (!fio_handle_client(client)) {
log_info("client: host=%s disconnected\n",
client->hostname);
remove_client(client);
retval = 1;
} else if (client->error)
retval = 1;
+ put_client(client);
}
}
projects / fio.git / blobdiff