gfio: Add support for sending logs over the network

[fio.git] / client.c

diff --git a/client.c b/client.c
index 84f13aa181d5d109dcb3f28f21c571f97684961a..c49e9a198455d72fabe318841f0f8022add89a1e 100644 (file)
--- a/client.c
+++ b/client.c
@@ -14,6 +14,7 @@
 #include <arpa/inet.h>
 #include <netdb.h>
 #include <signal.h>
 #include <arpa/inet.h>
 #include <netdb.h>
 #include <signal.h>

+#include <zlib.h>

 
 #include "fio.h"
 #include "client.h"
 
 #include "fio.h"
 #include "client.h"


@@ -39,6 +40,7 @@ struct client_ops fio_client_ops = {
        .eta            = display_thread_status,
        .probe          = handle_probe,
        .eta_msec       = FIO_CLIENT_DEF_ETA_MSEC,
        .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;
 };
 
 static struct timeval eta_tv;


@@ -206,6 +208,7 @@ struct fio_client *fio_client_add_explicit(struct client_ops *ops,
        client->fd = -1;
        client->ops = ops;
        client->refs = 1;
        client->fd = -1;
        client->ops = ops;
        client->refs = 1;

+       client->type = ops->client_type;

 
        __fio_client_add_cmd_option(client, "fio");
 
 
        __fio_client_add_cmd_option(client, "fio");
 


@@ -255,6 +258,7 @@ int fio_client_add(struct client_ops *ops, const char *hostname, void **cookie)
        client->fd = -1;
        client->ops = ops;
        client->refs = 1;
        client->fd = -1;
        client->ops = ops;
        client->refs = 1;

+       client->type = ops->client_type;

 
        __fio_client_add_cmd_option(client, "fio");
 
 
        __fio_client_add_cmd_option(client, "fio");
 


@@ -448,6 +452,7 @@ static int send_client_cmd_line(struct fio_client *client)
 
        free(lens);
        clp->lines = cpu_to_le16(client->argc);
 
        free(lens);
        clp->lines = cpu_to_le16(client->argc);

+       clp->client_type = __cpu_to_le16(client->type);

        ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
        free(pdu);
        return ret;
        ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
        free(pdu);
        return ret;


@@ -517,8 +522,11 @@ int fio_start_all_clients(void)
  */
 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;
        struct stat sb;

-       char *p, *buf;
+       char *p;
+       void *buf;

        off_t len;
        int fd, ret;
 
        off_t len;
        int fd, ret;
 


@@ -540,7 +548,9 @@ static int __fio_client_send_ini(struct fio_client *client, const char *filename
                return 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;
 
        len = sb.st_size;
        p = buf;


@@ -565,9 +575,12 @@ static int __fio_client_send_ini(struct fio_client *client, const char *filename
                return 1;
        }
 
                return 1;
        }
 

+       pdu->buf_len = __cpu_to_le32(sb.st_size);
+       pdu->client_type = cpu_to_le32(client->type);
+

        client->sent_job = 1;
        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, 0);
+       free(pdu);

        close(fd);
        return ret;
 }
        close(fd);
        return ret;
 }


@@ -938,6 +951,67 @@ static void convert_text(struct fio_net_cmd *cmd)
        pdu->log_usec   = le64_to_cpu(pdu->log_usec);
 }
 
        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;
+
+       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;
+
+       /*
+        * Everything beyond the first entry is compressed.
+        */
+       nr_samples = le32_to_cpu(pdu->nr_samples);
+
+       total = sizeof(*pdu) + nr_samples * sizeof(struct io_sample);
+       ret = malloc(total);
+       ret->nr_samples = nr_samples;
+       p = (void *) ret + sizeof(pdu->nr_samples);
+
+       stream.avail_in = cmd->pdu_len - sizeof(pdu->nr_samples);
+       stream.next_in = (void *) pdu + sizeof(pdu->nr_samples);
+       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);
+               if (err != Z_OK) {
+                       log_err("fio: inflate error %d\n", err);
+                       goto out;
+               }
+
+               this_len = this_chunk - stream.avail_out;
+               p += this_len;
+               total -= this_len;
+       }
+
+       ret->log_type = cpu_to_le32(ret->log_type);
+out:
+       inflateEnd(&stream);
+       return ret;
+}
+

 int fio_handle_client(struct fio_client *client)
 {
        struct client_ops *ops = client->ops;
 int fio_handle_client(struct fio_client *client)
 {
        struct client_ops *ops = client->ops;


@@ -1036,6 +1110,15 @@ int fio_handle_client(struct fio_client *client)
                        ops->add_job(client, cmd);
                free(cmd);
                break;
                        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;

        default:
                log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode));
                free(cmd);
        default:
                log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode));
                free(cmd);