#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;
char *name;
int state;
+
int skip_newline;
int is_sock;
+ 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,
};
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)
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)
{
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);
free(client);
nr_clients--;
+ sum_stat_clients--;
}
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,
else
fd = fio_client_connect_ip(client);
+ dprint(FD_NET, "client: %s connected %d\n", client->hostname, fd);
+
if (fd < 0)
return 1;
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);
}
}
{
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)
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
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]);
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);
+ ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
free(pdu);
return ret;
}
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;
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)
show_group_stats(gs);
}
-static void handle_eta(struct fio_net_cmd *cmd)
+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);
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];
+ }
- display_thread_status(je);
+ 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 == (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);
+ dec_jobs_eta(eta);
}
static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
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:
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;
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;
}
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;
+ }
+
+ 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, &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;
struct pollfd *pfds;
int i, ret = 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) {
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)
projects / fio.git / blobdiff