int port;
int fd;
+ char *name;
+
int state;
+
int skip_newline;
int is_sock;
+ int waiting_eta;
uint16_t argc;
char **argv;
};
+static struct timeval eta_tv;
+
enum {
Client_created = 0,
Client_connected = 1,
Client_exited = 4,
};
+struct client_eta {
+ struct jobs_eta eta;
+ unsigned int pending;
+};
+
static FLIST_HEAD(client_list);
#define FIO_CLIENT_HASH_BITS 7
fio_client_remove_hash(client);
+ /* FIXME: check ->waiting_eta and handle it */
+
free(client->hostname);
if (client->argv)
free(client->argv);
+ if (client->name)
+ free(client->name);
free(client);
nr_clients--;
}
-static int __fio_client_add_cmd_option(struct fio_client *client,
- const char *opt)
+static void __fio_client_add_cmd_option(struct fio_client *client,
+ const char *opt)
{
int index;
- if (client->argc == FIO_NET_CMD_JOBLINE_ARGV) {
- log_err("fio: max cmd line number reached.\n");
- log_err("fio: cmd line <%s> has been ignored.\n", opt);
- return 1;
- }
-
index = client->argc++;
client->argv = realloc(client->argv, sizeof(char *) * client->argc);
client->argv[index] = strdup(opt);
dprint(FD_NET, "client: add cmd %d: %s\n", index, opt);
- return 0;
}
-int fio_client_add_cmd_option(void *cookie, const char *opt)
+void fio_client_add_cmd_option(void *cookie, const char *opt)
{
struct fio_client *client = cookie;
if (!client || !opt)
- return 0;
+ return;
- return __fio_client_add_cmd_option(client, opt);
+ __fio_client_add_cmd_option(client, opt);
}
int fio_client_add(const char *hostname, void **cookie)
dprint(FD_NET, "client: connect to host %s\n", client->hostname);
- memset(&client->addr, 0, sizeof(client->addr));
-
if (client->is_sock)
fd = fio_client_connect_sock(client);
else
static int send_client_cmd_line(struct fio_client *client)
{
- struct cmd_line_pdu *pdu;
+ struct cmd_single_line_pdu *cslp;
+ struct cmd_line_pdu *clp;
+ unsigned long offset;
+ void *pdu;
+ size_t mem;
int i, ret;
dprint(FD_NET, "client: send cmdline %d\n", client->argc);
- pdu = malloc(sizeof(*pdu));
- for (i = 0; i < client->argc; i++)
- strcpy((char *) pdu->argv[i], client->argv[i]);
+ /*
+ * Find out how much mem we need
+ */
+ for (i = 0, mem = 0; i < client->argc; i++)
+ mem += strlen(client->argv[i]) + 1;
+
+ /*
+ * We need one cmd_line_pdu, and argc number of cmd_single_line_pdu
+ */
+ mem += sizeof(*clp) + (client->argc * sizeof(*cslp));
+
+ pdu = malloc(mem);
+ clp = pdu;
+ offset = sizeof(*clp);
+
+ for (i = 0; i < client->argc; i++) {
+ uint16_t arg_len = strlen(client->argv[i]) + 1;
+
+ cslp = pdu + offset;
+ strcpy((char *) cslp->text, client->argv[i]);
+ cslp->len = cpu_to_le16(arg_len);
+ offset += sizeof(*cslp) + arg_len;
+ }
- pdu->argc = cpu_to_le16(client->argc);
- ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, sizeof(*pdu));
+ clp->lines = cpu_to_le16(client->argc);
+ 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;
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];
+ }
+
+ dst->elapsed_sec += je->elapsed_sec;
- display_thread_status(je);
+ if (je->eta_sec > dst->eta_sec)
+ dst->eta_sec = je->eta_sec;
}
-static void handle_probe(struct fio_net_cmd *cmd)
+static void handle_eta(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);
+
+ convert_jobs_eta(je);
+ sum_jobs_eta(&eta->eta, je);
+
+ if (!--eta->pending) {
+ display_thread_status(&eta->eta);
+ free(eta);
+ }
+}
+
+static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
{
struct cmd_probe_pdu *probe = (struct cmd_probe_pdu *) cmd->payload;
const char *os, *arch;
if (!arch)
os = "unknown";
- log_info("hostname=%s, be=%u, os=%s, arch=%s, fio ver %u.%u.%u\n",
+ log_info("hostname=%s, be=%u, os=%s, arch=%s, fio=%u.%u.%u\n",
probe->hostname, probe->bigendian, os, arch, probe->fio_major,
probe->fio_minor, probe->fio_patch);
+
+ if (!client->name)
+ client->name = strdup((char *) probe->hostname);
}
static int handle_client(struct fio_client *client)
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> ", client->hostname);
+ fprintf(f_out, "<%s> ", name);
ret = fwrite(buf, cmd->pdu_len, 1, f_out);
fflush(f_out);
client->skip_newline = strchr(buf, '\n') == NULL;
free(cmd);
break;
case FIO_NET_CMD_PROBE:
- handle_probe(cmd);
+ handle_probe(client, cmd);
free(cmd);
break;
case FIO_NET_CMD_START:
return 1;
}
+static void request_client_etas(void)
+{
+ struct fio_client *client;
+ struct flist_head *entry;
+ struct client_eta *eta;
+
+ 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;
+
+ flist_for_each(entry, &client_list) {
+ client = flist_entry(entry, struct fio_client, list);
+
+ client->waiting_eta = 1;
+ fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_SEND_ETA,
+ (uint64_t) eta);
+ }
+
+ dprint(FD_NET, "client: requested eta tag %p\n", eta);
+}
+
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));
while (!exit_backend && nr_clients) {
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));
+ }
+
ret = poll(pfds, nr_clients, 100);
if (ret < 0) {
if (errno == EINTR)
projects / fio.git / blobdiff