#include <arpa/inet.h>
#include <netdb.h>
#include <signal.h>
+#ifdef CONFIG_ZLIB
+#include <zlib.h>
+#endif
#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;
- unsigned int refs;
-
- char *name;
-
- int state;
-
- int skip_newline;
- int is_sock;
- int disk_stats_shown;
- unsigned int jobs;
- int error;
- int ipv6;
- int sent_job;
-
- struct flist_head eta_list;
- struct client_eta *eta_in_flight;
-
- struct flist_head cmd_list;
-
- uint16_t argc;
- char **argv;
-
- char **ini_file;
- unsigned int nr_ini_file;
+static void handle_du(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);
+static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd);
+static void handle_start(struct fio_client *client, struct fio_net_cmd *cmd);
+
+struct client_ops fio_client_ops = {
+ .text = handle_text,
+ .disk_util = handle_du,
+ .thread_status = handle_ts,
+ .group_stats = handle_gs,
+ .stop = handle_stop,
+ .start = handle_start,
+ .eta = display_thread_status,
+ .probe = handle_probe,
+ .eta_msec = FIO_CLIENT_DEF_ETA_MSEC,
+ .client_type = FIO_CLIENT_TYPE_CLI,
};
static struct timeval eta_tv;
-enum {
- Client_created = 0,
- Client_connected = 1,
- Client_started = 2,
- Client_running = 3,
- Client_stopped = 4,
- Client_exited = 5,
-};
-
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;
+struct thread_stat client_ts;
+struct group_run_stats client_gs;
+int sum_stat_clients;
+
static int sum_stat_nr;
+static int do_output_all_clients;
#define FIO_CLIENT_HASH_BITS 7
#define FIO_CLIENT_HASH_SZ (1 << FIO_CLIENT_HASH_BITS)
#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);
return NULL;
}
-static void remove_client(struct fio_client *client)
+void fio_put_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);
-
- 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);
if (client->ini_file)
free(client->ini_file);
+ if (!client->did_stat)
+ sum_stat_clients -= client->nr_stat;
+
free(client);
+}
+
+static void remove_client(struct fio_client *client)
+{
+ assert(client->refs);
+
+ dprint(FD_NET, "client: removed <%s>\n", client->hostname);
+
+ if (!flist_empty(&client->list))
+ flist_del_init(&client->list);
+
+ fio_client_remove_hash(client);
+
+ if (!flist_empty(&client->eta_list)) {
+ flist_del_init(&client->eta_list);
+ fio_client_dec_jobs_eta(client->eta_in_flight, client->ops->eta);
+ }
+
+ close(client->fd);
+ client->fd = -1;
+
+ if (client->ops->removed)
+ client->ops->removed(client);
+
nr_clients--;
- sum_stat_clients--;
+ fio_put_client(client);
}
-static void put_client(struct fio_client *client)
+struct fio_client *fio_get_client(struct fio_client *client)
{
- remove_client(client);
+ client->refs++;
+ return client;
}
static void __fio_client_add_cmd_option(struct fio_client *client,
}
}
+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;
+ client->type = ops->client_type;
+
+ __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;
+}
+
void fio_client_add_ini_file(void *cookie, const char *ini_file)
{
struct fio_client *client = cookie;
client->nr_ini_file++;
}
-int fio_client_add(const char *hostname, void **cookie)
+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;
+ client->type = ops->client_type;
__fio_client_add_cmd_option(client, "fio");
return 0;
}
+static void probe_client(struct fio_client *client)
+{
+ struct cmd_client_probe_pdu pdu;
+ uint64_t tag;
+
+ dprint(FD_NET, "client: send probe\n");
+
+#ifdef CONFIG_ZLIB
+ pdu.flags = __le64_to_cpu(FIO_PROBE_FLAG_ZLIB);
+#else
+ pdu.flags = 0;
+#endif
+
+ fio_net_send_cmd(client->fd, FIO_NET_CMD_PROBE, &pdu, sizeof(pdu), &tag, &client->cmd_list);
+}
+
static int fio_client_connect_ip(struct fio_client *client)
{
struct sockaddr *addr;
- fio_socklen_t socklen;
+ socklen_t socklen;
int fd, domain;
if (client->ipv6) {
fd = socket(domain, SOCK_STREAM, 0);
if (fd < 0) {
+ int ret = -errno;
+
log_err("fio: socket: %s\n", strerror(errno));
- return -1;
+ return ret;
}
if (connect(fd, addr, socklen) < 0) {
+ int ret = -errno;
+
log_err("fio: connect: %s\n", strerror(errno));
log_err("fio: failed to connect to %s:%u\n", client->hostname,
client->port);
close(fd);
- return -1;
+ return ret;
}
return fd;
static int fio_client_connect_sock(struct fio_client *client)
{
struct sockaddr_un *addr = &client->addr_un;
- fio_socklen_t len;
+ socklen_t len;
int fd;
memset(addr, 0, sizeof(*addr));
fd = socket(AF_UNIX, SOCK_STREAM, 0);
if (fd < 0) {
+ int ret = -errno;
+
log_err("fio: socket: %s\n", strerror(errno));
- return -1;
+ return ret;
}
len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
if (connect(fd, (struct sockaddr *) addr, len) < 0) {
+ int ret = -errno;
+
log_err("fio: connect; %s\n", strerror(errno));
close(fd);
- return -1;
+ return ret;
}
return fd;
}
-static int fio_client_connect(struct fio_client *client)
+int fio_client_connect(struct fio_client *client)
{
int fd;
dprint(FD_NET, "client: %s connected %d\n", client->hostname, fd);
if (fd < 0)
- return 1;
+ return fd;
client->fd = fd;
fio_client_add_hash(client);
client->state = Client_connected;
+
+ probe_client(client);
return 0;
}
+int fio_client_terminate(struct fio_client *client)
+{
+ return fio_net_send_quit(client->fd);
+}
+
void fio_clients_terminate(void)
{
struct flist_head *entry;
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, NULL);
+ fio_client_terminate(client);
}
}
act.sa_flags = SA_RESTART;
sigaction(SIGTERM, &act, NULL);
+/* Windows uses SIGBREAK as a quit signal from other applications */
+#ifdef WIN32
+ memset(&act, 0, sizeof(act));
+ act.sa_handler = sig_int;
+ act.sa_flags = SA_RESTART;
+ sigaction(SIGBREAK, &act, NULL);
+#endif
+
memset(&act, 0, sizeof(act));
act.sa_handler = sig_show_status;
act.sa_flags = SA_RESTART;
sigaction(SIGUSR1, &act, NULL);
}
-static void probe_client(struct fio_client *client)
-{
- dprint(FD_NET, "client: send probe\n");
-
- 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;
free(lens);
clp->lines = cpu_to_le16(client->argc);
- ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
+ clp->client_type = __cpu_to_le16(client->type);
+ ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, NULL, NULL);
free(pdu);
return ret;
}
#ifdef WIN32
WSADATA wsd;
- WSAStartup(MAKEWORD(2,2), &wsd);
+ WSAStartup(MAKEWORD(2, 2), &wsd);
#endif
dprint(FD_NET, "client: connect all\n");
continue;
}
- probe_client(client);
-
if (client->argc > 1)
send_client_cmd_line(client);
}
return !nr_clients;
}
+int fio_start_client(struct fio_client *client)
+{
+ dprint(FD_NET, "client: start %s\n", client->hostname);
+ return fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_RUN, 0, NULL);
+}
+
+int fio_start_all_clients(void)
+{
+ struct fio_client *client;
+ struct flist_head *entry, *tmp;
+ int ret;
+
+ dprint(FD_NET, "client: start all\n");
+
+ flist_for_each_safe(entry, tmp, &client_list) {
+ client = flist_entry(entry, struct fio_client, list);
+
+ ret = fio_start_client(client);
+ if (ret) {
+ remove_client(client);
+ continue;
+ }
+ }
+
+ return flist_empty(&client_list);
+}
+
/*
* Send file contents to server backend. We could use sendfile(), but to remain
* more portable lets just read/write the darn thing.
*/
-static int fio_client_send_ini(struct fio_client *client, const char *filename)
+static int __fio_client_send_ini(struct fio_client *client, const char *filename)
{
+ struct cmd_job_pdu *pdu;
+ size_t p_size;
struct stat sb;
- char *p, *buf;
+ char *p;
+ void *buf;
off_t len;
int fd, ret;
fd = open(filename, O_RDONLY);
if (fd < 0) {
+ int ret = -errno;
+
log_err("fio: job file <%s> open: %s\n", filename, strerror(errno));
- return 1;
+ return ret;
}
if (fstat(fd, &sb) < 0) {
+ int ret = -errno;
+
log_err("fio: job file stat: %s\n", strerror(errno));
close(fd);
- return 1;
+ return ret;
}
- buf = malloc(sb.st_size);
+ p_size = sb.st_size + sizeof(*pdu);
+ pdu = malloc(p_size);
+ buf = pdu->buf;
len = sb.st_size;
p = buf;
return 1;
}
+ pdu->buf_len = __cpu_to_le32(sb.st_size);
+ pdu->client_type = cpu_to_le32(client->type);
+
client->sent_job = 1;
- ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size, 0);
- free(buf);
+ ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, pdu, p_size, NULL, NULL);
+ free(pdu);
close(fd);
return ret;
}
+int fio_client_send_ini(struct fio_client *client, const char *filename)
+{
+ int ret;
+
+ ret = __fio_client_send_ini(client, filename);
+ if (!ret)
+ client->sent_job = 1;
+
+ return ret;
+}
+
int fio_clients_send_ini(const char *filename)
{
struct fio_client *client;
}
} else if (!filename || fio_client_send_ini(client, filename))
remove_client(client);
-
- client->sent_job = 1;
}
return !nr_clients;
}
+int fio_client_update_options(struct fio_client *client,
+ struct thread_options *o, uint64_t *tag)
+{
+ struct cmd_add_job_pdu pdu;
+
+ pdu.thread_number = cpu_to_le32(client->thread_number);
+ pdu.groupid = cpu_to_le32(client->groupid);
+ convert_thread_options_to_net(&pdu.top, o);
+
+ return fio_net_send_cmd(client->fd, FIO_NET_CMD_UPDATE_JOB, &pdu, sizeof(pdu), tag, &client->cmd_list);
+}
+
static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
{
dst->max_val = le64_to_cpu(src->max_val);
{
int i, j;
- dst->error = le32_to_cpu(src->error);
- dst->groupid = le32_to_cpu(src->groupid);
- dst->pid = le32_to_cpu(src->pid);
- dst->members = le32_to_cpu(src->members);
+ dst->error = le32_to_cpu(src->error);
+ dst->thread_number = le32_to_cpu(src->thread_number);
+ dst->groupid = le32_to_cpu(src->groupid);
+ dst->pid = le32_to_cpu(src->pid);
+ dst->members = le32_to_cpu(src->members);
+ dst->unified_rw_rep = le32_to_cpu(src->unified_rw_rep);
- for (i = 0; i < 2; i++) {
+ for (i = 0; i < DDIR_RWDIR_CNT; i++) {
convert_io_stat(&dst->clat_stat[i], &src->clat_stat[i]);
convert_io_stat(&dst->slat_stat[i], &src->slat_stat[i]);
convert_io_stat(&dst->lat_stat[i], &src->lat_stat[i]);
dst->io_u_lat_m[i] = le32_to_cpu(src->io_u_lat_m[i]);
}
- for (i = 0; i < 2; i++)
+ for (i = 0; i < DDIR_RWDIR_CNT; i++)
for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
dst->io_u_plat[i][j] = le32_to_cpu(src->io_u_plat[i][j]);
- for (i = 0; i < 3; i++) {
+ for (i = 0; i < DDIR_RWDIR_CNT; i++) {
dst->total_io_u[i] = le64_to_cpu(src->total_io_u[i]);
dst->short_io_u[i] = le64_to_cpu(src->short_io_u[i]);
}
dst->total_submit = le64_to_cpu(src->total_submit);
dst->total_complete = le64_to_cpu(src->total_complete);
- for (i = 0; i < 2; i++) {
+ for (i = 0; i < DDIR_RWDIR_CNT; i++) {
dst->io_bytes[i] = le64_to_cpu(src->io_bytes[i]);
dst->runtime[i] = le64_to_cpu(src->runtime[i]);
}
dst->total_err_count = le64_to_cpu(src->total_err_count);
dst->first_error = le32_to_cpu(src->first_error);
dst->kb_base = le32_to_cpu(src->kb_base);
+ dst->unit_base = le32_to_cpu(src->unit_base);
}
static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
{
int i;
- for (i = 0; i < 2; i++) {
+ for (i = 0; i < DDIR_RWDIR_CNT; i++) {
dst->max_run[i] = le64_to_cpu(src->max_run[i]);
dst->min_run[i] = le64_to_cpu(src->min_run[i]);
dst->max_bw[i] = le64_to_cpu(src->max_bw[i]);
}
dst->kb_base = le32_to_cpu(src->kb_base);
+ dst->unit_base = le32_to_cpu(src->unit_base);
dst->groupid = le32_to_cpu(src->groupid);
+ dst->unified_rw_rep = le32_to_cpu(src->unified_rw_rep);
}
-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);
+ client->did_stat = 1;
- if (sum_stat_clients == 1)
+ if (!do_output_all_clients)
return;
sum_thread_stats(&client_ts, &p->ts, sum_stat_nr);
sum_group_stats(&client_gs, &p->rs);
client_ts.members++;
+ client_ts.thread_number = p->ts.thread_number;
client_ts.groupid = p->ts.groupid;
+ client_ts.unified_rw_rep = p->ts.unified_rw_rep;
if (++sum_stat_nr == sum_stat_clients) {
strcpy(client_ts.name, "All clients");
}
}
-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");
}
- print_disk_util(&du->dus, &du->agg, terse_output);
+ print_disk_util(&du->dus, &du->agg, output_format == FIO_OUTPUT_TERSE);
}
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++) {
+ for (i = 0; i < DDIR_RWDIR_CNT; 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->elapsed_sec = le64_to_cpu(je->elapsed_sec);
je->eta_sec = le64_to_cpu(je->eta_sec);
+ je->nr_threads = le32_to_cpu(je->nr_threads);
je->is_pow2 = le32_to_cpu(je->is_pow2);
+ je->unit_base = le32_to_cpu(je->unit_base);
}
-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++) {
+ for (i = 0; i < DDIR_RWDIR_CNT; 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];
}
if (je->eta_sec > dst->eta_sec)
dst->eta_sec = je->eta_sec;
+
+ dst->nr_threads += je->nr_threads;
+ /* we need to handle je->run_str too ... */
}
-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);
}
}
static void remove_reply_cmd(struct fio_client *client, struct fio_net_cmd *cmd)
{
- struct fio_net_int_cmd *icmd = NULL;
+ struct fio_net_cmd_reply *reply = NULL;
struct flist_head *entry;
flist_for_each(entry, &client->cmd_list) {
- icmd = flist_entry(entry, struct fio_net_int_cmd, list);
+ reply = flist_entry(entry, struct fio_net_cmd_reply, list);
- if (cmd->tag == (uintptr_t) icmd)
+ if (cmd->tag == (uintptr_t) reply)
break;
- icmd = NULL;
+ reply = NULL;
}
- if (!icmd) {
- log_err("fio: client: unable to find matching tag\n");
+ if (!reply) {
+ log_err("fio: client: unable to find matching tag (%lx)\n", cmd->tag);
return;
}
- flist_del(&icmd->list);
- cmd->tag = icmd->saved_tag;
- free(icmd);
+ flist_del(&reply->list);
+ cmd->tag = reply->saved_tag;
+ free(reply);
+}
+
+int fio_client_wait_for_reply(struct fio_client *client, uint64_t tag)
+{
+ do {
+ struct fio_net_cmd_reply *reply = NULL;
+ struct flist_head *entry;
+
+ flist_for_each(entry, &client->cmd_list) {
+ reply = flist_entry(entry, struct fio_net_cmd_reply, list);
+
+ if (tag == (uintptr_t) reply)
+ break;
+
+ reply = NULL;
+ }
+
+ if (!reply)
+ break;
+
+ usleep(1000);
+ } while (1);
+
+ return 0;
}
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);
+ if (client->ops->jobs_eta)
+ client->ops->jobs_eta(client, je);
+
+ 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)
{
- struct cmd_probe_pdu *probe = (struct cmd_probe_pdu *) cmd->payload;
+ struct cmd_probe_reply_pdu *probe = (struct cmd_probe_reply_pdu *) cmd->payload;
const char *os, *arch;
char bit[16];
os = "unknown";
sprintf(bit, "%d-bit", probe->bpp * 8);
+ probe->flags = le64_to_cpu(probe->flags);
- log_info("hostname=%s, be=%u, %s, os=%s, arch=%s, fio=%s\n",
+ log_info("hostname=%s, be=%u, %s, os=%s, arch=%s, fio=%s, flags=%lx\n",
probe->hostname, probe->bigendian, bit, os, arch,
- probe->fio_version);
+ probe->fio_version, (unsigned long) probe->flags);
if (!client->name)
client->name = strdup((char *) probe->hostname);
client->state = Client_started;
client->jobs = le32_to_cpu(pdu->jobs);
+ client->nr_stat = le32_to_cpu(pdu->stat_outputs);
+
+ if (sum_stat_clients > 1)
+ do_output_all_clients = 1;
+
+ sum_stat_clients += client->nr_stat;
}
static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd)
+{
+ if (client->error)
+ log_info("client <%s>: exited with error %d\n", client->hostname, client->error);
+}
+
+static void convert_stop(struct fio_net_cmd *cmd)
{
struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload;
- client->state = Client_stopped;
- client->error = le32_to_cpu(pdu->error);
+ pdu->error = le32_to_cpu(pdu->error);
+}
- if (client->error)
- log_info("client <%s>: exited with error %d\n", client->hostname, client->error);
+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);
+}
+
+static struct cmd_iolog_pdu *convert_iolog_gz(struct fio_net_cmd *cmd,
+ struct cmd_iolog_pdu *pdu)
+{
+#ifdef CONFIG_ZLIB
+ struct cmd_iolog_pdu *ret;
+ z_stream stream;
+ uint32_t nr_samples;
+ size_t total;
+ void *p;
+
+ stream.zalloc = Z_NULL;
+ stream.zfree = Z_NULL;
+ stream.opaque = Z_NULL;
+ stream.avail_in = 0;
+ stream.next_in = Z_NULL;
+
+ if (inflateInit(&stream) != Z_OK)
+ return NULL;
+
+ /*
+ * Get header first, it's not compressed
+ */
+ nr_samples = le32_to_cpu(pdu->nr_samples);
+
+ total = nr_samples * sizeof(struct io_sample);
+ ret = malloc(total + sizeof(*pdu));
+ ret->nr_samples = nr_samples;
+
+ memcpy(ret, pdu, sizeof(*pdu));
+
+ p = (void *) ret + sizeof(*pdu);
+
+ stream.avail_in = cmd->pdu_len - sizeof(*pdu);
+ stream.next_in = (void *) pdu + sizeof(*pdu);
+ while (stream.avail_in) {
+ unsigned int this_chunk = 65536;
+ unsigned int this_len;
+ int err;
+
+ if (this_chunk > total)
+ this_chunk = total;
+
+ stream.avail_out = this_chunk;
+ stream.next_out = p;
+ err = inflate(&stream, Z_NO_FLUSH);
+ /* may be Z_OK, or Z_STREAM_END */
+ if (err < 0) {
+ log_err("fio: inflate error %d\n", err);
+ free(ret);
+ ret = NULL;
+ goto err;
+ }
+
+ this_len = this_chunk - stream.avail_out;
+ p += this_len;
+ total -= this_len;
+ }
+
+err:
+ inflateEnd(&stream);
+ return ret;
+#else
+ return NULL;
+#endif
+}
+
+/*
+ * This has been compressed on the server side, since it can be big.
+ * Uncompress here.
+ */
+static struct cmd_iolog_pdu *convert_iolog(struct fio_net_cmd *cmd)
+{
+ struct cmd_iolog_pdu *pdu = (struct cmd_iolog_pdu *) cmd->payload;
+ struct cmd_iolog_pdu *ret;
+ int i;
+
+ /*
+ * Convert if compressed and we support it. If it's not
+ * compressed, we need not do anything.
+ */
+ if (le32_to_cpu(pdu->compressed)) {
+#ifndef CONFIG_ZLIB
+ log_err("fio: server sent compressed data by mistake\n");
+ return NULL;
+#endif
+ ret = convert_iolog_gz(cmd, pdu);
+ if (!ret) {
+ log_err("fio: failed decompressing log\n");
+ return NULL;
+ }
+ } else
+ ret = pdu;
+
+ ret->thread_number = le32_to_cpu(ret->thread_number);
+ ret->nr_samples = le32_to_cpu(ret->nr_samples);
+ ret->log_type = le32_to_cpu(ret->log_type);
+ ret->compressed = le32_to_cpu(ret->compressed);
+
+ for (i = 0; i < ret->nr_samples; i++) {
+ struct io_sample *s = &ret->samples[i];
+
+ s->time = le64_to_cpu(s->time);
+ s->val = le64_to_cpu(s->val);
+ s->ddir = le32_to_cpu(s->ddir);
+ s->bs = le32_to_cpu(s->bs);
+ }
+
+ return ret;
}
-static int handle_client(struct fio_client *client)
+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);
if (!cmd)
return 0;
- dprint(FD_NET, "client: got cmd op %s from %s\n",
- fio_server_op(cmd->opcode), client->hostname);
+ dprint(FD_NET, "client: got cmd op %s from %s (pdu=%u)\n",
+ fio_server_op(cmd->opcode), client->hostname, cmd->pdu_len);
switch (cmd->opcode) {
case FIO_NET_CMD_QUIT:
+ if (ops->quit)
+ ops->quit(client, cmd);
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;
+ case FIO_NET_CMD_TEXT:
+ convert_text(cmd);
+ ops->text(client, cmd);
free(cmd);
break;
- }
- case FIO_NET_CMD_DU:
- handle_du(client, cmd);
+ 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:
- handle_ts(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:
- handle_gs(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);
+ convert_jobs_eta(je);
handle_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:
+ case FIO_NET_CMD_SERVER_START:
client->state = Client_running;
+ if (ops->job_start)
+ ops->job_start(client, cmd);
free(cmd);
break;
- case FIO_NET_CMD_START:
- handle_start(client, cmd);
+ case FIO_NET_CMD_START: {
+ struct cmd_start_pdu *pdu = (struct cmd_start_pdu *) cmd->payload;
+
+ pdu->jobs = le32_to_cpu(pdu->jobs);
+ ops->start(client, cmd);
free(cmd);
break;
- case FIO_NET_CMD_STOP:
- handle_stop(client, cmd);
+ }
+ case FIO_NET_CMD_STOP: {
+ struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload;
+
+ convert_stop(cmd);
+ client->state = Client_stopped;
+ client->error = le32_to_cpu(pdu->error);
+ client->signal = le32_to_cpu(pdu->signal);
+ ops->stop(client, cmd);
+ free(cmd);
+ break;
+ }
+ case FIO_NET_CMD_ADD_JOB: {
+ struct cmd_add_job_pdu *pdu = (struct cmd_add_job_pdu *) cmd->payload;
+
+ client->thread_number = le32_to_cpu(pdu->thread_number);
+ client->groupid = le32_to_cpu(pdu->groupid);
+
+ if (ops->add_job)
+ ops->add_job(client, cmd);
+ free(cmd);
+ break;
+ }
+ case FIO_NET_CMD_IOLOG:
+ if (ops->iolog) {
+ struct cmd_iolog_pdu *pdu;
+
+ pdu = convert_iolog(cmd);
+ ops->iolog(client, pdu);
+ }
+ free(cmd);
+ break;
+ case FIO_NET_CMD_UPDATE_JOB:
+ ops->update_job(client, cmd);
+ remove_reply_cmd(client, cmd);
free(cmd);
break;
default:
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);
}
static int client_check_cmd_timeout(struct fio_client *client,
struct timeval *now)
{
- struct fio_net_int_cmd *cmd;
+ struct fio_net_cmd_reply *reply;
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);
+ reply = flist_entry(entry, struct fio_net_cmd_reply, list);
- if (mtime_since(&cmd->tv, now) < FIO_NET_CLIENT_TIMEOUT)
+ if (mtime_since(&reply->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);
+ fio_server_op(reply->opcode));
+ flist_del(&reply->list);
+ free(reply);
ret = 1;
}
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;
struct timeval tv;
int ret = 0;
- gettimeofday(&tv, NULL);
+ fio_gettime(&tv, NULL);
flist_for_each_safe(entry, tmp, &client_list) {
client = flist_entry(entry, struct fio_client, list);
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;
}
return ret;
}
-int fio_handle_clients(void)
+int fio_handle_clients(struct client_ops *ops)
{
struct pollfd *pfds;
int i, ret = 0, retval = 0;
- gettimeofday(&eta_tv, NULL);
+ fio_gettime(&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);
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;
do {
struct timeval tv;
- gettimeofday(&tv, NULL);
+ fio_gettime(&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;
}
- ret = poll(pfds, nr_clients, 100);
+ ret = poll(pfds, nr_clients, ops->eta_msec);
if (ret < 0) {
if (errno == EINTR)
continue;
log_err("fio: unknown client fd %d\n", pfds[i].fd);
continue;
}
- if (!handle_client(client)) {
+ 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);
+ fio_put_client(client);
}
}
projects / fio.git / blobdiff