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);
struct client_ops fio_client_ops = {
.text_op = handle_text,
.disk_util = handle_du,
.thread_status = handle_ts,
.group_stats = handle_gs,
+ .stop = handle_stop,
.eta = display_thread_status,
.probe = handle_probe,
+ .eta_msec = FIO_CLIENT_DEF_ETA_MSEC,
};
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);
flist_for_each(entry, &client_hash[bucket]) {
client = flist_entry(entry, struct fio_client, hash_list);
- if (client->fd == fd) {
- client->refs++;
- return client;
- }
+ if (client->fd == fd)
+ return fio_get_client(client);
}
return NULL;
sum_stat_clients--;
}
-static void put_client(struct fio_client *client)
+void fio_put_client(struct fio_client *client)
{
remove_client(client);
}
+struct fio_client *fio_get_client(struct fio_client *client)
+{
+ client->refs++;
+ return client;
+}
+
static void __fio_client_add_cmd_option(struct fio_client *client,
const char *opt)
{
return 0;
}
+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 fio_client_connect_ip(struct fio_client *client)
{
struct sockaddr *addr;
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;
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;
}
+void fio_client_terminate(struct fio_client *client)
+{
+ fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0, NULL);
+}
+
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);
}
}
sigaction(SIGTERM, &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;
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 stat sb;
char *p, *buf;
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);
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) {
+ remove_client(client);
+ return ret;
+ }
+
+ client->sent_job = 1;
+ return ret;
+}
+
int fio_clients_send_ini(const char *filename)
{
struct fio_client *client;
flist_for_each_safe(entry, tmp, &client_list) {
client = flist_entry(entry, struct fio_client, list);
- if (fio_client_send_ini(client, filename))
- remove_client(client);
-
- client->sent_job = 1;
+ fio_client_send_ini(client, filename);
}
return !nr_clients;
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);
}
void fio_client_sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je)
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 ... */
}
void fio_client_dec_jobs_eta(struct client_eta *eta, client_eta_op eta_fn)
client->eta_in_flight = NULL;
flist_del_init(&client->eta_list);
+ 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_stop(struct fio_client *client, 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);
-
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;
+
+ pdu->error = le32_to_cpu(pdu->error);
+}
+
static void convert_text(struct fio_net_cmd *cmd)
{
struct cmd_text_pdu *pdu = (struct cmd_text_pdu *) cmd->payload;
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:
ops->probe(client, cmd);
free(cmd);
break;
- case FIO_NET_CMD_RUN:
+ case FIO_NET_CMD_SERVER_START:
client->state = Client_running;
free(cmd);
break;
handle_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 = pdu->error;
+ ops->stop(client, cmd);
free(cmd);
break;
+ }
case FIO_NET_CMD_ADD_JOB:
if (ops->add_job)
ops->add_job(client, cmd);
struct timeval tv;
gettimeofday(&tv, NULL);
- if (mtime_since(&eta_tv, &tv) >= 900) {
+ if (mtime_since(&eta_tv, &tv) >= ops->eta_msec) {
request_client_etas(ops);
memcpy(&eta_tv, &tv, sizeof(tv));
retval = 1;
} else if (client->error)
retval = 1;
- put_client(client);
+ fio_put_client(client);
}
}
projects / fio.git / blobdiff