Move tp.[ch] to lib/
[fio.git] / iolog.c
diff --git a/iolog.c b/iolog.c
index a8bead11292c824eb3371f4f2c3e7a8545108cf7..950670562f61045a4256adc28ca23ccdaed4821c 100644 (file)
--- a/iolog.c
+++ b/iolog.c
@@ -6,14 +6,57 @@
 #include <stdlib.h>
 #include <libgen.h>
 #include <assert.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#ifdef CONFIG_ZLIB
+#include <zlib.h>
+#endif
+
 #include "flist.h"
 #include "fio.h"
 #include "verify.h"
 #include "trim.h"
 #include "filelock.h"
+#include "lib/tp.h"
 
 static const char iolog_ver2[] = "fio version 2 iolog";
 
+#ifdef CONFIG_ZLIB
+
+struct iolog_compress {
+       struct flist_head list;
+       void *buf;
+       size_t len;
+       unsigned int seq;
+       int nofree;
+};
+
+#define GZ_CHUNK       131072
+
+static struct iolog_compress *get_new_chunk(unsigned int seq)
+{
+       struct iolog_compress *c;
+
+       c = malloc(sizeof(*c));
+       INIT_FLIST_HEAD(&c->list);
+       c->buf = malloc(GZ_CHUNK);
+       c->len = 0;
+       c->seq = seq;
+       c->nofree = 0;
+       return c;
+}
+
+static void free_chunk(struct iolog_compress *ic)
+{
+       if (!ic->nofree) {
+               free(ic->buf);
+               free(ic);
+       }
+}
+
+#endif
+
 void queue_io_piece(struct thread_data *td, struct io_piece *ipo)
 {
        flist_add_tail(&ipo->list, &td->io_log_list);
@@ -539,19 +582,35 @@ int init_iolog(struct thread_data *td)
        return ret;
 }
 
-void setup_log(struct io_log **log, unsigned long avg_msec, int log_type,
-              int log_offset, const char *filename)
+void setup_log(struct io_log **log, struct log_params *p,
+              const char *filename)
 {
        struct io_log *l = malloc(sizeof(*l));
 
        memset(l, 0, sizeof(*l));
        l->nr_samples = 0;
        l->max_samples = 1024;
-       l->log_type = log_type;
-       l->log_offset = log_offset;
+       l->log_type = p->log_type;
+       l->log_offset = p->log_offset;
+       l->log_gz = p->log_gz;
+       l->log_gz_store = p->log_gz_store;
        l->log = malloc(l->max_samples * log_entry_sz(l));
-       l->avg_msec = avg_msec;
+       l->avg_msec = p->avg_msec;
        l->filename = strdup(filename);
+       l->td = p->td;
+
+       if (l->log_offset)
+               l->log_ddir_mask = 0x80000000;
+
+       INIT_FLIST_HEAD(&l->chunk_list);
+
+       if (l->log_gz && !p->td)
+               l->log_gz = 0;
+       else if (l->log_gz) {
+               pthread_mutex_init(&l->chunk_lock, NULL);
+               p->td->flags |= TD_F_COMPRESS_LOG;
+       }
+
        *log = l;
 }
 
@@ -581,69 +640,421 @@ static void clear_file_buffer(void *buf)
 }
 #endif
 
-static void free_log(struct io_log *log)
+void free_log(struct io_log *log)
 {
        free(log->log);
        free(log->filename);
        free(log);
 }
 
-void __finish_log(struct io_log *log)
+static void flush_samples(FILE *f, void *samples, uint64_t sample_size)
 {
-       uint64_t i;
-       void *buf;
-       FILE *f;
+       struct io_sample *s;
+       int log_offset;
+       uint64_t i, nr_samples;
 
-       f = fopen(log->filename, "a");
-       if (!f) {
-               perror("fopen log");
+       if (!sample_size)
                return;
-       }
 
-       buf = set_file_buffer(f);
+       s = __get_sample(samples, 0, 0);
+       if (s->__ddir & 0x80000000)
+               log_offset = 1;
+       else
+               log_offset = 0;
+
+       nr_samples = sample_size / __log_entry_sz(log_offset);
 
-       for (i = 0; i < log->nr_samples; i++) {
-               struct io_sample *s = get_sample(log, i);
+       for (i = 0; i < nr_samples; i++) {
+               s = __get_sample(samples, log_offset, i);
 
-               if (!log->log_offset) {
+               if (!log_offset) {
                        fprintf(f, "%lu, %lu, %u, %u\n",
                                        (unsigned long) s->time,
                                        (unsigned long) s->val,
-                                       s->ddir, s->bs);
+                                       io_sample_ddir(s), s->bs);
                } else {
                        struct io_sample_offset *so = (void *) s;
 
                        fprintf(f, "%lu, %lu, %u, %u, %llu\n",
                                        (unsigned long) s->time,
                                        (unsigned long) s->val,
-                                       s->ddir, s->bs,
+                                       io_sample_ddir(s), s->bs,
                                        (unsigned long long) so->offset);
                }
        }
+}
+
+#ifdef CONFIG_ZLIB
+static int z_stream_init(z_stream *stream, int gz_hdr)
+{
+       int wbits = 15;
+
+       stream->zalloc = Z_NULL;
+       stream->zfree = Z_NULL;
+       stream->opaque = Z_NULL;
+       stream->next_in = Z_NULL;
+
+       if (gz_hdr)
+               wbits += 32;
+
+       if (inflateInit2(stream, wbits) != Z_OK)
+               return 1;
+
+       return 0;
+}
+
+struct flush_chunk_iter {
+       unsigned int seq;
+       void *buf;
+       size_t buf_size;
+       size_t buf_used;
+       size_t chunk_sz;
+};
+
+static void finish_chunk(z_stream *stream, FILE *f,
+                        struct flush_chunk_iter *iter)
+{
+       int ret;
+
+       ret = inflateEnd(stream);
+       if (ret != Z_OK)
+               log_err("fio: failed to end log inflation (%d)\n", ret);
+
+       flush_samples(f, iter->buf, iter->buf_used);
+       free(iter->buf);
+       iter->buf = NULL;
+       iter->buf_size = iter->buf_used = 0;
+}
+
+static int flush_chunk(struct iolog_compress *ic, int gz_hdr, FILE *f,
+                      z_stream *stream, struct flush_chunk_iter *iter)
+{
+       if (ic->seq != iter->seq) {
+               if (iter->seq)
+                       finish_chunk(stream, f, iter);
+
+               z_stream_init(stream, gz_hdr);
+               iter->seq = ic->seq;
+       }
+
+       stream->avail_in = ic->len;
+       stream->next_in = ic->buf;
+
+       if (!iter->buf_size) {
+               iter->buf_size = iter->chunk_sz;
+               iter->buf = malloc(iter->buf_size);
+       }
+
+       while (stream->avail_in) {
+               size_t this_out = iter->buf_size - iter->buf_used;
+               int err;
+
+               stream->avail_out = this_out;
+               stream->next_out = iter->buf + iter->buf_used;
+
+               err = inflate(stream, Z_NO_FLUSH);
+               if (err < 0) {
+                       log_err("fio: failed inflating log: %d\n", err);
+                       break;
+               }
+
+               iter->buf_used += this_out - stream->avail_out;
+
+               if (!stream->avail_out) {
+                       iter->buf_size += iter->chunk_sz;
+                       iter->buf = realloc(iter->buf, iter->buf_size);
+                       continue;
+               }
+
+               if (err == Z_STREAM_END)
+                       break;
+       }
+
+       free_chunk(ic);
+       return 0;
+}
+
+static void flush_gz_chunks(struct io_log *log, FILE *f)
+{
+       struct flush_chunk_iter iter = { .chunk_sz = log->log_gz, };
+       struct flist_head *node;
+       z_stream stream;
+
+       while (!flist_empty(&log->chunk_list)) {
+               struct iolog_compress *ic;
+
+               node = log->chunk_list.next;
+               ic = flist_entry(node, struct iolog_compress, list);
+               flist_del(&ic->list);
+
+               if (log->log_gz_store) {
+                       fwrite(ic->buf, ic->len, 1, f);
+                       free_chunk(ic);
+               } else
+                       flush_chunk(ic, log->log_gz_store, f, &stream, &iter);
+       }
+
+       if (iter.seq) {
+               finish_chunk(&stream, f, &iter);
+               free(iter.buf);
+       }
+}
+
+int iolog_file_inflate(const char *file)
+{
+       struct flush_chunk_iter iter = { .chunk_sz = 64 * 1024 * 1024, };
+       struct iolog_compress ic;
+       z_stream stream;
+       struct stat sb;
+       size_t ret;
+       FILE *f;
+
+       if (stat(file, &sb) < 0) {
+               perror("stat");
+               return 1;
+       }
+
+       f = fopen(file, "r");
+       if (!f) {
+               perror("fopen");
+               return 1;
+       }
+
+       ic.buf = malloc(sb.st_size);
+       ic.len = sb.st_size;
+       ic.nofree = 1;
+       ic.seq = 1;
+
+       ret = fread(ic.buf, ic.len, 1, f);
+       if (ret < 0) {
+               perror("fread");
+               fclose(f);
+               return 1;
+       } else if (ret != 1) {
+               log_err("fio: short read on reading log\n");
+               fclose(f);
+               return 1;
+       }
+
+       fclose(f);
+
+       flush_chunk(&ic,  1, stdout, &stream, &iter);
+
+       if (iter.seq) {
+               finish_chunk(&stream, stdout, &iter);
+               free(iter.buf);
+       }
+
+       free(ic.buf);
+       return 0;
+}
+
+#else
+
+static void flush_gz_chunks(struct io_log *log, FILE *f)
+{
+}
+
+#endif
+
+void flush_log(struct io_log *log)
+{
+       void *buf;
+       FILE *f;
+
+       f = fopen(log->filename, "w");
+       if (!f) {
+               perror("fopen log");
+               return;
+       }
+
+       buf = set_file_buffer(f);
+
+       flush_gz_chunks(log, f);
+
+       flush_samples(f, log->log, log->nr_samples * log_entry_sz(log));
 
        fclose(f);
        clear_file_buffer(buf);
-       free_log(log);
 }
 
 static int finish_log(struct thread_data *td, struct io_log *log, int trylock)
 {
+       if (td->tp_data)
+               iolog_flush(log, 1);
+
        if (trylock) {
                if (fio_trylock_file(log->filename))
                        return 1;
        } else
                fio_lock_file(log->filename);
 
-       if (td->client_type == FIO_CLIENT_TYPE_GUI) {
+       if (td->client_type == FIO_CLIENT_TYPE_GUI)
                fio_send_iolog(td, log, log->filename);
-               free_log(log);
-       } else
-               __finish_log(log);
+       else
+               flush_log(log);
 
        fio_unlock_file(log->filename);
+       free_log(log);
        return 0;
 }
 
+#ifdef CONFIG_ZLIB
+
+struct iolog_flush_data {
+       struct tp_work work;
+       struct io_log *log;
+       void *samples;
+       uint64_t nr_samples;
+};
+
+static int gz_work(struct tp_work *work)
+{
+       struct iolog_flush_data *data;
+       struct iolog_compress *c;
+       struct flist_head list;
+       unsigned int seq;
+       z_stream stream;
+       size_t total = 0;
+       int ret, wbits;
+
+       INIT_FLIST_HEAD(&list);
+
+       data = container_of(work, struct iolog_flush_data, work);
+
+       stream.zalloc = Z_NULL;
+       stream.zfree = Z_NULL;
+       stream.opaque = Z_NULL;
+
+       /*
+        * Store gz header if storing to a file
+        */
+#if 0
+       wbits = 15;
+       if (data->log->log_gz_store)
+               wbits += 16;
+
+       ret = deflateInit2(&stream, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
+                               31, 8, Z_DEFAULT_STRATEGY);
+#else
+       ret = deflateInit(&stream, Z_DEFAULT_COMPRESSION);
+#endif
+       if (ret != Z_OK) {
+               log_err("fio: failed to init gz stream\n");
+               return 0;
+       }
+
+       seq = ++data->log->chunk_seq;
+
+       stream.next_in = (void *) data->samples;
+       stream.avail_in = data->nr_samples * log_entry_sz(data->log);
+
+       do {
+               c = get_new_chunk(seq);
+               stream.avail_out = GZ_CHUNK;
+               stream.next_out = c->buf;
+               ret = deflate(&stream, Z_NO_FLUSH);
+               if (ret < 0) {
+                       log_err("fio: deflate log (%d)\n", ret);
+                       break;
+               }
+
+               c->len = GZ_CHUNK - stream.avail_out;
+               flist_add_tail(&c->list, &list);
+               total += c->len;
+       } while (stream.avail_in);
+
+       stream.next_out = c->buf + c->len;
+       stream.avail_out = GZ_CHUNK - c->len;
+
+       ret = deflate(&stream, Z_FINISH);
+       if (ret == Z_STREAM_END)
+               c->len = GZ_CHUNK - stream.avail_out;
+       else {
+               do {
+                       c = get_new_chunk(seq);
+                       stream.avail_out = GZ_CHUNK;
+                       stream.next_out = c->buf;
+                       ret = deflate(&stream, Z_FINISH);
+                       c->len = GZ_CHUNK - stream.avail_out;
+                       flist_add_tail(&c->list, &list);
+               } while (ret != Z_STREAM_END);
+       }
+
+       ret = deflateEnd(&stream);
+       if (ret != Z_OK)
+               log_err("fio: deflateEnd %d\n", ret);
+
+       free(data->samples);
+
+       if (!flist_empty(&list)) {
+               pthread_mutex_lock(&data->log->chunk_lock);
+               flist_splice_tail(&list, &data->log->chunk_list);
+               pthread_mutex_unlock(&data->log->chunk_lock);
+       }
+
+       if (work->wait) {
+               work->done = 1;
+               pthread_cond_signal(&work->cv);
+       } else
+               free(data);
+
+       return 0;
+}
+
+int iolog_flush(struct io_log *log, int wait)
+{
+       struct tp_data *tdat = log->td->tp_data;
+       struct iolog_flush_data *data;
+       size_t sample_size;
+
+       data = malloc(sizeof(*data));
+       if (!data)
+               return 1;
+
+       data->log = log;
+
+       sample_size = log->nr_samples * log_entry_sz(log);
+       data->samples = malloc(sample_size);
+       if (!data->samples) {
+               free(data);
+               return 1;
+       }
+
+       memcpy(data->samples, log->log, sample_size);
+       data->nr_samples = log->nr_samples;
+       data->work.fn = gz_work;
+       log->nr_samples = 0;
+
+       if (wait) {
+               pthread_mutex_init(&data->work.lock, NULL);
+               pthread_cond_init(&data->work.cv, NULL);
+               data->work.wait = 1;
+       } else
+               data->work.wait = 0;
+
+       tp_queue_work(tdat, &data->work);
+
+       if (wait) {
+               pthread_mutex_lock(&data->work.lock);
+               while (!data->work.done)
+                       pthread_cond_wait(&data->work.cv, &data->work.lock);
+               pthread_mutex_unlock(&data->work.lock);
+               free(data);
+       }
+
+       return 0;
+}
+
+#else
+
+int iolog_flush(struct io_log *log, int wait)
+{
+       return 1;
+}
+
+#endif
+
 static int write_iops_log(struct thread_data *td, int try)
 {
        struct io_log *log = td->iops_log;