#include <arpa/inet.h>
#include <netdb.h>
#include <signal.h>
+#include <zlib.h>
#include "fio.h"
#include "client.h"
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_op = handle_text,
+ .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;
flist_for_each(entry, &client_hash[bucket]) {
client = flist_entry(entry, struct fio_client, hash_list);
- if (client->fd == fd)
- return fio_get_client(client);
+ if (client->fd == fd) {
+ client->refs++;
+ return client;
+ }
}
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;
+ free(client->hostname);
+ if (client->argv)
+ free(client->argv);
+ if (client->name)
+ free(client->name);
+
+ free(client);
+}
+
+static void remove_client(struct fio_client *client)
+{
+ assert(client->refs);
+
dprint(FD_NET, "client: removed <%s>\n", client->hostname);
- flist_del(&client->list);
+
+ if (!flist_empty(&client->list))
+ flist_del_init(&client->list);
fio_client_remove_hash(client);
fio_client_dec_jobs_eta(client->eta_in_flight, client->ops->eta);
}
- free(client->hostname);
- if (client->argv)
- free(client->argv);
- if (client->name)
- free(client->name);
+ close(client->fd);
+ client->fd = -1;
+
+ if (client->ops->removed)
+ client->ops->removed(client);
- free(client);
nr_clients--;
sum_stat_clients--;
-}
-void fio_put_client(struct fio_client *client)
-{
- remove_client(client);
+ fio_put_client(client);
}
struct fio_client *fio_get_client(struct fio_client *client)
client->fd = -1;
client->ops = ops;
client->refs = 1;
+ client->type = ops->client_type;
__fio_client_add_cmd_option(client, "fio");
client->fd = -1;
client->ops = ops;
client->refs = 1;
+ client->type = ops->client_type;
__fio_client_add_cmd_option(client, "fio");
return 0;
}
-void fio_client_terminate(struct fio_client *client)
+int fio_client_terminate(struct fio_client *client)
{
- fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0, NULL);
+ return fio_net_send_quit(client->fd);
}
void fio_clients_terminate(void)
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");
*/
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;
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 ret;
ret = __fio_client_send_ini(client, filename);
- if (ret) {
- remove_client(client);
- return ret;
- }
+ if (!ret)
+ client->sent_job = 1;
- client->sent_job = 1;
return ret;
}
flist_for_each_safe(entry, tmp, &client_list) {
client = flist_entry(entry, struct fio_client, list);
- fio_client_send_ini(client, filename);
+ if (fio_client_send_ini(client, filename))
+ remove_client(client);
}
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);
for (i = 0; i < 2; i++) {
convert_io_stat(&dst->clat_stat[i], &src->clat_stat[i]);
sum_group_stats(&client_gs, &p->rs);
client_ts.members++;
+ client_ts.thread_number = p->ts.thread_number;
client_ts.groupid = p->ts.groupid;
if (++sum_stat_nr == sum_stat_clients) {
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)
struct cmd_start_pdu *pdu = (struct cmd_start_pdu *) cmd->payload;
client->state = Client_started;
- client->jobs = le32_to_cpu(pdu->jobs);
+ client->jobs = pdu->jobs;
}
static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd)
pdu->log_usec = le64_to_cpu(pdu->log_usec);
}
+/*
+ * 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;
+ uint32_t nr_samples;
+ unsigned long total;
+ z_stream stream;
+ void *p;
+ int i;
+
+ 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->thread_number = le32_to_cpu(pdu->thread_number);
+ ret->nr_samples = nr_samples;
+ ret->log_type = le32_to_cpu(pdu->log_type);
+ strcpy((char *) ret->name, (char *) pdu->name);
+
+ 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 out;
+ }
+
+ this_len = this_chunk - stream.avail_out;
+ p += this_len;
+ total -= this_len;
+ }
+
+ 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);
+ }
+
+out:
+ inflateEnd(&stream);
+ return ret;
+}
+
int fio_handle_client(struct fio_client *client)
{
struct client_ops *ops = client->ops;
switch (cmd->opcode) {
case FIO_NET_CMD_QUIT:
if (ops->quit)
- ops->quit(client);
+ ops->quit(client, cmd);
remove_client(client);
free(cmd);
break;
case FIO_NET_CMD_TEXT:
convert_text(cmd);
- ops->text_op(client, cmd);
+ ops->text(client, cmd);
free(cmd);
break;
case FIO_NET_CMD_DU: {
break;
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: {
struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload;
convert_stop(cmd);
client->state = Client_stopped;
- client->error = pdu->error;
+ 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:
+ 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:
+ remove_reply_cmd(client, cmd);
+ ops->update_job(client, cmd);
+ free(cmd);
+ break;
default:
log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode));
free(cmd);
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;
}
break;
}
- ret = poll(pfds, nr_clients, 100);
+ ret = poll(pfds, nr_clients, ops->eta_msec);
if (ret < 0) {
if (errno == EINTR)
continue;
projects / fio.git / blobdiff