X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=client.c;h=6bc114541062231f24d9bc0215e47102e05ec868;hp=f4b95d325d9f83839ce951da4967390d0f0932b8;hb=9d0ad2a56d63e7f59473f31708358a4b65d2a5e3;hpb=be6fa25809df11afff6cd767813393339ba117df diff --git a/client.c b/client.c index f4b95d32..6bc11454 100644 --- a/client.c +++ b/client.c @@ -33,6 +33,8 @@ 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); +static void convert_text(struct fio_net_cmd *cmd); + struct client_ops fio_client_ops = { .text = handle_text, .disk_util = handle_du, @@ -70,6 +72,8 @@ static int error_clients; #define FIO_CLIENT_HASH_MASK (FIO_CLIENT_HASH_SZ - 1) static struct flist_head client_hash[FIO_CLIENT_HASH_SZ]; +static struct cmd_iolog_pdu *convert_iolog(struct fio_net_cmd *, bool *); + static void fio_client_add_hash(struct fio_client *client) { int bucket = hash_long(client->fd, FIO_CLIENT_HASH_BITS); @@ -213,12 +217,32 @@ static int fio_client_dec_jobs_eta(struct client_eta *eta, client_eta_op eta_fn) return 1; } +static void fio_drain_client_text(struct fio_client *client) +{ + do { + struct fio_net_cmd *cmd; + + cmd = fio_net_recv_cmd(client->fd, false); + if (!cmd) + break; + + if (cmd->opcode == FIO_NET_CMD_TEXT) { + convert_text(cmd); + client->ops->text(client, cmd); + } + + free(cmd); + } while (1); +} + static void remove_client(struct fio_client *client) { assert(client->refs); dprint(FD_NET, "client: removed <%s>\n", client->hostname); + fio_drain_client_text(client); + if (!flist_empty(&client->list)) flist_del_init(&client->list); @@ -792,6 +816,8 @@ int fio_clients_send_ini(const char *filename) struct flist_head *entry, *tmp; flist_for_each_safe(entry, tmp, &client_list) { + bool failed = false; + client = flist_entry(entry, struct fio_client, list); if (client->nr_files) { @@ -803,12 +829,13 @@ int fio_clients_send_ini(const char *filename) cf = &client->files[i]; if (fio_client_send_cf(client, cf)) { + failed = true; remove_client(client); break; } } } - if (client->sent_job) + if (client->sent_job || failed) continue; if (!filename || fio_client_send_ini(client, filename, 0)) remove_client(client); @@ -1224,6 +1251,56 @@ static void handle_eta(struct fio_client *client, struct fio_net_cmd *cmd) fio_client_dec_jobs_eta(eta, client->ops->eta); } +static int fio_client_handle_iolog(struct fio_client *client, + struct fio_net_cmd *cmd) +{ + struct cmd_iolog_pdu *pdu; + bool store_direct; + + pdu = convert_iolog(cmd, &store_direct); + if (!pdu) { + log_err("fio: failed converting IO log\n"); + return 1; + } + + if (store_direct) { + ssize_t ret; + size_t sz; + int fd; + + fd = open((const char *) pdu->name, + O_WRONLY | O_CREAT | O_TRUNC, 0644); + if (fd < 0) { + log_err("fio: open log: %s\n", strerror(errno)); + return 1; + } + + sz = cmd->pdu_len - sizeof(*pdu); + ret = write(fd, pdu->samples, sz); + close(fd); + + if (ret != sz) { + log_err("fio: short write on compressed log\n"); + return 1; + } + + return 0; + } else { + FILE *f; + + f = fopen((const char *) pdu->name, "w"); + if (!f) { + log_err("fio: fopen log: %s\n", strerror(errno)); + return 1; + } + + flush_samples(f, pdu->samples, + pdu->nr_samples * sizeof(struct io_sample)); + fclose(f); + return 0; + } +} + static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd) { struct cmd_probe_reply_pdu *probe = (struct cmd_probe_reply_pdu *) cmd->payload; @@ -1364,18 +1441,23 @@ err: * 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) +static struct cmd_iolog_pdu *convert_iolog(struct fio_net_cmd *cmd, + bool *store_direct) { struct cmd_iolog_pdu *pdu = (struct cmd_iolog_pdu *) cmd->payload; struct cmd_iolog_pdu *ret; uint64_t i; + int compressed; void *samples; + *store_direct = false; + /* * Convert if compressed and we support it. If it's not * compressed, we need not do anything. */ - if (le32_to_cpu(pdu->compressed)) { + compressed = le32_to_cpu(pdu->compressed); + if (compressed == XMIT_COMPRESSED) { #ifndef CONFIG_ZLIB log_err("fio: server sent compressed data by mistake\n"); return NULL; @@ -1385,6 +1467,9 @@ static struct cmd_iolog_pdu *convert_iolog(struct fio_net_cmd *cmd) log_err("fio: failed decompressing log\n"); return NULL; } + } else if (compressed == STORE_COMPRESSED) { + *store_direct = true; + ret = pdu; } else ret = pdu; @@ -1394,6 +1479,9 @@ static struct cmd_iolog_pdu *convert_iolog(struct fio_net_cmd *cmd) ret->compressed = le32_to_cpu(ret->compressed); ret->log_offset = le32_to_cpu(ret->log_offset); + if (*store_direct) + return ret; + samples = &ret->samples[0]; for (i = 0; i < ret->nr_samples; i++) { struct io_sample *s; @@ -1421,8 +1509,8 @@ static void sendfile_reply(int fd, struct cmd_sendfile_reply *rep, fio_net_send_cmd(fd, FIO_NET_CMD_SENDFILE, rep, size, &tag, NULL); } -static int send_file(struct fio_client *client, struct cmd_sendfile *pdu, - uint64_t tag) +static int fio_send_file(struct fio_client *client, struct cmd_sendfile *pdu, + uint64_t tag) { struct cmd_sendfile_reply *rep; struct stat sb; @@ -1462,7 +1550,7 @@ int fio_handle_client(struct fio_client *client) dprint(FD_NET, "client: handle %s\n", client->hostname); - cmd = fio_net_recv_cmd(client->fd); + cmd = fio_net_recv_cmd(client->fd, true); if (!cmd) return 0; @@ -1550,12 +1638,7 @@ int fio_handle_client(struct fio_client *client) break; } case FIO_NET_CMD_IOLOG: - if (ops->iolog) { - struct cmd_iolog_pdu *pdu; - - pdu = convert_iolog(cmd); - ops->iolog(client, pdu); - } + fio_client_handle_iolog(client, cmd); break; case FIO_NET_CMD_UPDATE_JOB: ops->update_job(client, cmd); @@ -1577,7 +1660,7 @@ int fio_handle_client(struct fio_client *client) } case FIO_NET_CMD_SENDFILE: { struct cmd_sendfile *pdu = (struct cmd_sendfile *) cmd->payload; - send_file(client, pdu, cmd->tag); + fio_send_file(client, pdu, cmd->tag); break; } case FIO_NET_CMD_JOB_OPT: {