lib/pattern: fix formatting
[fio.git] / t / dedupe.c
index c3b837f7b698cdedd027f20be12e085ffe21faa1..02e52b742ee33730e94c42c4ca6741f9b17f5eaf 100644 (file)
@@ -3,24 +3,20 @@
  * just scans the filename for extents of the given size, checksums them,
  * and orders them up.
  */
+#include <fcntl.h>
+#include <inttypes.h>
 #include <stdio.h>
-#include <stdio.h>
+#include <string.h>
 #include <unistd.h>
-#include <inttypes.h>
-#include <assert.h>
-#include <sys/types.h>
 #include <sys/stat.h>
-#include <sys/ioctl.h>
-#include <fcntl.h>
-#include <string.h>
 
+#include "../fio.h"
 #include "../flist.h"
 #include "../log.h"
-#include "../mutex.h"
+#include "../fio_sem.h"
 #include "../smalloc.h"
 #include "../minmax.h"
 #include "../crc/md5.h"
-#include "../lib/memalign.h"
 #include "../os/os.h"
 #include "../gettime.h"
 #include "../fio_time.h"
 
 #include "../lib/bloom.h"
 #include "debug.h"
+#include "zlib.h"
+
+struct zlib_ctrl {
+       z_stream stream;
+       unsigned char *buf_in;
+       unsigned char *buf_out;
+};
 
 struct worker_thread {
+       struct zlib_ctrl zc;
        pthread_t thread;
-
-       volatile int done;
-
-       int fd;
        uint64_t cur_offset;
        uint64_t size;
-
+       unsigned long long unique_capacity;
        unsigned long items;
        unsigned long dupes;
        int err;
+       int fd;
+       volatile int done;
 };
 
 struct extent {
@@ -49,7 +51,7 @@ struct extent {
 };
 
 struct chunk {
-       struct rb_node rb_node;
+       struct fio_rb_node rb_node;
        uint64_t count;
        uint32_t hash[MD5_HASH_WORDS];
        struct flist_head extent_list[0];
@@ -62,7 +64,7 @@ struct item {
 
 static struct rb_root rb_root;
 static struct bloom *bloom;
-static struct fio_mutex *rb_lock;
+static struct fio_sem *rb_lock;
 
 static unsigned int blocksize = 4096;
 static unsigned int num_threads;
@@ -72,10 +74,11 @@ static unsigned int odirect;
 static unsigned int collision_check;
 static unsigned int print_progress = 1;
 static unsigned int use_bloom = 1;
+static unsigned int compression = 0;
 
 static uint64_t total_size;
 static uint64_t cur_offset;
-static struct fio_mutex *size_lock;
+static struct fio_sem *size_lock;
 
 static struct fio_file file;
 
@@ -91,8 +94,9 @@ static uint64_t get_size(struct fio_file *f, struct stat *sb)
                        return 0;
                }
                ret = bytes;
-       } else
+       } else {
                ret = sb->st_size;
+       }
 
        return (ret & ~((uint64_t)blocksize - 1));
 }
@@ -102,7 +106,7 @@ static int get_work(uint64_t *offset, uint64_t *size)
        uint64_t this_chunk;
        int ret = 1;
 
-       fio_mutex_down(size_lock);
+       fio_sem_down(size_lock);
 
        if (cur_offset < total_size) {
                *offset = cur_offset;
@@ -112,7 +116,7 @@ static int get_work(uint64_t *offset, uint64_t *size)
                ret = 0;
        }
 
-       fio_mutex_up(size_lock);
+       fio_sem_up(size_lock);
        return ret;
 }
 
@@ -124,9 +128,9 @@ static int __read_block(int fd, void *buf, off_t offset, size_t count)
        if (ret < 0) {
                perror("pread");
                return 1;
-       } else if (!ret)
+       } else if (!ret) {
                return 1;
-       else if (ret != count) {
+       else if (ret != count) {
                log_err("dedupe: short read on block\n");
                return 1;
        }
@@ -139,6 +143,36 @@ static int read_block(int fd, void *buf, off_t offset)
        return __read_block(fd, buf, offset, blocksize);
 }
 
+static int account_unique_capacity(uint64_t offset, uint64_t *unique_capacity,
+                                  struct zlib_ctrl *zc)
+{
+       z_stream *stream = &zc->stream;
+       unsigned int compressed_len;
+       int ret;
+
+       if (read_block(file.fd, zc->buf_in, offset))
+               return 1;
+
+       stream->next_in = zc->buf_in;
+       stream->avail_in = blocksize;
+       stream->avail_out = deflateBound(stream, blocksize);
+       stream->next_out = zc->buf_out;
+
+       ret = deflate(stream, Z_FINISH);
+       if (ret == Z_STREAM_ERROR)
+               return 1;
+       compressed_len = blocksize - stream->avail_out;
+
+       if (dump_output)
+               printf("offset 0x%lx compressed to %d blocksize %d ratio %.2f \n",
+                               (unsigned long) offset, compressed_len, blocksize,
+                               (float)compressed_len / (float)blocksize);
+
+       *unique_capacity += compressed_len;
+       deflateReset(stream);
+       return 0;
+}
+
 static void add_item(struct chunk *c, struct item *i)
 {
        /*      
@@ -162,8 +196,8 @@ static int col_check(struct chunk *c, struct item *i)
        char *cbuf, *ibuf;
        int ret = 1;
 
-       cbuf = fio_memalign(blocksize, blocksize);
-       ibuf = fio_memalign(blocksize, blocksize);
+       cbuf = fio_memalign(blocksize, blocksize, false);
+       ibuf = fio_memalign(blocksize, blocksize, false);
 
        e = flist_entry(c->extent_list[0].next, struct extent, list);
        if (read_block(file.fd, cbuf, e->offset))
@@ -174,8 +208,8 @@ static int col_check(struct chunk *c, struct item *i)
 
        ret = memcmp(ibuf, cbuf, blocksize);
 out:
-       fio_memfree(cbuf, blocksize);
-       fio_memfree(ibuf, blocksize);
+       fio_memfree(cbuf, blocksize, false);
+       fio_memfree(ibuf, blocksize, false);
        return ret;
 }
 
@@ -186,17 +220,19 @@ static struct chunk *alloc_chunk(void)
        if (collision_check || dump_output) {
                c = malloc(sizeof(struct chunk) + sizeof(struct flist_head));
                INIT_FLIST_HEAD(&c->extent_list[0]);
-       } else
+       } else {
                c = malloc(sizeof(struct chunk));
+       }
 
        return c;
 }
 
-static void insert_chunk(struct item *i)
+static int insert_chunk(struct item *i, uint64_t *unique_capacity,
+                       struct zlib_ctrl *zc)
 {
-       struct rb_node **p, *parent;
+       struct fio_rb_node **p, *parent;
        struct chunk *c;
-       int diff;
+       int ret, diff;
 
        p = &rb_root.rb_node;
        parent = NULL;
@@ -205,19 +241,17 @@ static void insert_chunk(struct item *i)
 
                c = rb_entry(parent, struct chunk, rb_node);
                diff = memcmp(i->hash, c->hash, sizeof(i->hash));
-               if (diff < 0)
+               if (diff < 0) {
                        p = &(*p)->rb_left;
-               else if (diff > 0)
+               } else if (diff > 0) {
                        p = &(*p)->rb_right;
-               else {
-                       int ret;
-
+               } else {
                        if (!collision_check)
                                goto add;
 
-                       fio_mutex_up(rb_lock);
+                       fio_sem_up(rb_lock);
                        ret = col_check(c, i);
-                       fio_mutex_down(rb_lock);
+                       fio_sem_down(rb_lock);
 
                        if (!ret)
                                goto add;
@@ -232,16 +266,23 @@ static void insert_chunk(struct item *i)
        memcpy(c->hash, i->hash, sizeof(i->hash));
        rb_link_node(&c->rb_node, parent, p);
        rb_insert_color(&c->rb_node, &rb_root);
+       if (compression) {
+               ret = account_unique_capacity(i->offset, unique_capacity, zc);
+               if (ret)
+                       return ret;
+       }
 add:
        add_item(c, i);
+       return 0;
 }
 
-static void insert_chunks(struct item *items, unsigned int nitems,
-                         uint64_t *ndupes)
+static int insert_chunks(struct item *items, unsigned int nitems,
+                        uint64_t *ndupes, uint64_t *unique_capacity,
+                        struct zlib_ctrl *zc)
 {
-       int i;
+       int i, ret = 0;
 
-       fio_mutex_down(rb_lock);
+       fio_sem_down(rb_lock);
 
        for (i = 0; i < nitems; i++) {
                if (bloom) {
@@ -251,11 +292,15 @@ static void insert_chunks(struct item *items, unsigned int nitems,
                        s = sizeof(items[i].hash) / sizeof(uint32_t);
                        r = bloom_set(bloom, items[i].hash, s);
                        *ndupes += r;
-               } else
-                       insert_chunk(&items[i]);
+               } else {
+                       ret = insert_chunk(&items[i], unique_capacity, zc);
+                       if (ret)
+                               break;
+               }
        }
 
-       fio_mutex_up(rb_lock);
+       fio_sem_up(rb_lock);
+       return ret;
 }
 
 static void crc_buf(void *buf, uint32_t *hash)
@@ -281,11 +326,14 @@ static int do_work(struct worker_thread *thread, void *buf)
        off_t offset;
        int nitems = 0;
        uint64_t ndupes = 0;
+       uint64_t unique_capacity = 0;
        struct item *items;
+       int ret;
 
        offset = thread->cur_offset;
 
-       nblocks = read_blocks(thread->fd, buf, offset, min(thread->size, (uint64_t)chunk_size));
+       nblocks = read_blocks(thread->fd, buf, offset,
+                               min(thread->size, (uint64_t) chunk_size));
        if (!nblocks)
                return 1;
 
@@ -300,12 +348,34 @@ static int do_work(struct worker_thread *thread, void *buf)
                nitems++;
        }
 
-       insert_chunks(items, nitems, &ndupes);
+       ret = insert_chunks(items, nitems, &ndupes, &unique_capacity, &thread->zc);
 
        free(items);
-       thread->items += nitems;
-       thread->dupes += ndupes;
-       return 0;
+       if (!ret) {
+               thread->items += nitems;
+               thread->dupes += ndupes;
+               thread->unique_capacity += unique_capacity;
+               return 0;
+       }
+
+       return ret;
+}
+
+static void thread_init_zlib_control(struct worker_thread *thread)
+{
+       size_t sz;
+
+       z_stream *stream = &thread->zc.stream;
+       stream->zalloc = Z_NULL;
+       stream->zfree = Z_NULL;
+       stream->opaque = Z_NULL;
+
+       if (deflateInit(stream, Z_DEFAULT_COMPRESSION) != Z_OK)
+               return;
+
+       thread->zc.buf_in = fio_memalign(blocksize, blocksize, false);
+       sz = deflateBound(stream, blocksize);
+       thread->zc.buf_out = fio_memalign(blocksize, sz, false);
 }
 
 static void *thread_fn(void *data)
@@ -313,7 +383,8 @@ static void *thread_fn(void *data)
        struct worker_thread *thread = data;
        void *buf;
 
-       buf = fio_memalign(blocksize, chunk_size);
+       buf = fio_memalign(blocksize, chunk_size, false);
+       thread_init_zlib_control(thread);
 
        do {
                if (get_work(&thread->cur_offset, &thread->size)) {
@@ -327,7 +398,7 @@ static void *thread_fn(void *data)
        } while (1);
 
        thread->done = 1;
-       fio_memfree(buf, chunk_size);
+       fio_memfree(buf, chunk_size, false);
        return NULL;
 }
 
@@ -366,15 +437,17 @@ static void show_progress(struct worker_thread *threads, unsigned long total)
                        printf("%3.2f%% done (%luKiB/sec)\r", perc, this_items);
                        last_nitems = nitems;
                        fio_gettime(&last_tv, NULL);
-               } else
+               } else {
                        printf("%3.2f%% done\r", perc);
+               }
                fflush(stdout);
                usleep(250000);
        };
 }
 
 static int run_dedupe_threads(struct fio_file *f, uint64_t dev_size,
-                             uint64_t *nextents, uint64_t *nchunks)
+                             uint64_t *nextents, uint64_t *nchunks,
+                             uint64_t *unique_capacity)
 {
        struct worker_thread *threads;
        unsigned long nitems, total_items;
@@ -383,7 +456,7 @@ static int run_dedupe_threads(struct fio_file *f, uint64_t dev_size,
        total_size = dev_size;
        total_items = dev_size / blocksize;
        cur_offset = 0;
-       size_lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
+       size_lock = fio_sem_init(FIO_SEM_UNLOCKED);
 
        threads = malloc(num_threads * sizeof(struct worker_thread));
        for (i = 0; i < num_threads; i++) {
@@ -402,11 +475,13 @@ static int run_dedupe_threads(struct fio_file *f, uint64_t dev_size,
        nitems = 0;
        *nextents = 0;
        *nchunks = 1;
+       *unique_capacity = 0;
        for (i = 0; i < num_threads; i++) {
                void *ret;
                pthread_join(threads[i].thread, &ret);
                nitems += threads[i].items;
                *nchunks += threads[i].dupes;
+               *unique_capacity += threads[i].unique_capacity;
        }
 
        printf("Threads(%u): %lu items processed\n", num_threads, nitems);
@@ -414,13 +489,13 @@ static int run_dedupe_threads(struct fio_file *f, uint64_t dev_size,
        *nextents = nitems;
        *nchunks = nitems - *nchunks;
 
-       fio_mutex_remove(size_lock);
+       fio_sem_remove(size_lock);
        free(threads);
        return err;
 }
 
 static int dedupe_check(const char *filename, uint64_t *nextents,
-                       uint64_t *nchunks)
+                       uint64_t *nchunks, uint64_t *unique_capacity)
 {
        uint64_t dev_size;
        struct stat sb;
@@ -455,9 +530,11 @@ static int dedupe_check(const char *filename, uint64_t *nextents,
                bloom = bloom_new(bloom_entries);
        }
 
-       printf("Will check <%s>, size <%llu>, using %u threads\n", filename, (unsigned long long) dev_size, num_threads);
+       printf("Will check <%s>, size <%llu>, using %u threads\n", filename,
+                               (unsigned long long) dev_size, num_threads);
 
-       return run_dedupe_threads(&file, dev_size, nextents, nchunks);
+       return run_dedupe_threads(&file, dev_size, nextents, nchunks,
+                                       unique_capacity);
 err:
        if (file.fd != -1)
                close(file.fd);
@@ -470,36 +547,69 @@ static void show_chunk(struct chunk *c)
        struct flist_head *n;
        struct extent *e;
 
-       printf("c hash %8x %8x %8x %8x, count %lu\n", c->hash[0], c->hash[1], c->hash[2], c->hash[3], (unsigned long) c->count);
+       printf("c hash %8x %8x %8x %8x, count %lu\n", c->hash[0], c->hash[1],
+                       c->hash[2], c->hash[3], (unsigned long) c->count);
        flist_for_each(n, &c->extent_list[0]) {
                e = flist_entry(n, struct extent, list);
                printf("\toffset %llu\n", (unsigned long long) e->offset);
        }
 }
 
-static void show_stat(uint64_t nextents, uint64_t nchunks)
+static const char *capacity_unit[] = {"b","KB", "MB", "GB", "TB", "PB", "EB"};
+
+static uint64_t bytes_to_human_readable_unit(uint64_t n, const char **unit_out)
+{
+       uint8_t i = 0;
+
+       while (n >= 1024) {
+               i++;
+               n /= 1024;
+       }
+
+       *unit_out = capacity_unit[i];
+       return n;
+}
+
+static void show_stat(uint64_t nextents, uint64_t nchunks, uint64_t ndupextents,
+                     uint64_t unique_capacity)
 {
        double perc, ratio;
+       const char *unit;
+       uint64_t uc_human;
 
-       printf("Extents=%lu, Unique extents=%lu\n", (unsigned long) nextents, (unsigned long) nchunks);
+       printf("Extents=%lu, Unique extents=%lu", (unsigned long) nextents,
+                                               (unsigned long) nchunks);
+       if (!bloom)
+               printf(" Duplicated extents=%lu", (unsigned long) ndupextents);
+       printf("\n");
 
        if (nchunks) {
                ratio = (double) nextents / (double) nchunks;
                printf("De-dupe ratio: 1:%3.2f\n", ratio - 1.0);
-       } else
+       } else {
                printf("De-dupe ratio: 1:infinite\n");
+       }
+
+       if (ndupextents) {
+               printf("De-dupe working set at least: %3.2f%%\n",
+                       100.0 * (double) ndupextents / (double) nextents);
+       }
 
        perc = 1.00 - ((double) nchunks / (double) nextents);
        perc *= 100.0;
        printf("Fio setting: dedupe_percentage=%u\n", (int) (perc + 0.50));
 
+
+       if (compression) {
+               uc_human = bytes_to_human_readable_unit(unique_capacity, &unit);
+               printf("Unique capacity %lu%s\n", (unsigned long) uc_human, unit);
+       }
 }
 
-static void iter_rb_tree(uint64_t *nextents, uint64_t *nchunks)
+static void iter_rb_tree(uint64_t *nextents, uint64_t *nchunks, uint64_t *ndupextents)
 {
-       struct rb_node *n;
-
-       *nchunks = *nextents = 0;
+       struct fio_rb_node *n;
+       *nchunks = *nextents = *ndupextents = 0;
 
        n = rb_first(&rb_root);
        if (!n)
@@ -511,6 +621,7 @@ static void iter_rb_tree(uint64_t *nextents, uint64_t *nchunks)
                c = rb_entry(n, struct chunk, rb_node);
                (*nchunks)++;
                *nextents += c->count;
+               *ndupextents += (c->count > 1);
 
                if (dump_output)
                        show_chunk(c);
@@ -529,18 +640,19 @@ static int usage(char *argv[])
        log_err("\t-c\tFull collision check\n");
        log_err("\t-B\tUse probabilistic bloom filter\n");
        log_err("\t-p\tPrint progress indicator\n");
+       log_err("\t-C\tCalculate compressible size\n");
        return 1;
 }
 
 int main(int argc, char *argv[])
 {
-       uint64_t nextents = 0, nchunks = 0;
+       uint64_t nextents = 0, nchunks = 0, ndupextents = 0, unique_capacity;
        int c, ret;
 
        arch_init(argv);
        debug_init();
 
-       while ((c = getopt(argc, argv, "b:t:d:o:c:p:B:")) != -1) {
+       while ((c = getopt(argc, argv, "b:t:d:o:c:p:B:C:")) != -1) {
                switch (c) {
                case 'b':
                        blocksize = atoi(optarg);
@@ -563,17 +675,20 @@ int main(int argc, char *argv[])
                case 'B':
                        use_bloom = atoi(optarg);
                        break;
+               case 'C':
+                       compression = atoi(optarg);
+                       break;
                case '?':
                default:
                        return usage(argv);
                }
        }
 
-       if (collision_check || dump_output)
+       if (collision_check || dump_output || compression)
                use_bloom = 0;
 
        if (!num_threads)
-               num_threads = cpus_online();
+               num_threads = cpus_configured();
 
        if (argc == optind)
                return usage(argv);
@@ -581,18 +696,18 @@ int main(int argc, char *argv[])
        sinit();
 
        rb_root = RB_ROOT;
-       rb_lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
+       rb_lock = fio_sem_init(FIO_SEM_UNLOCKED);
 
-       ret = dedupe_check(argv[optind], &nextents, &nchunks);
+       ret = dedupe_check(argv[optind], &nextents, &nchunks, &unique_capacity);
 
        if (!ret) {
                if (!bloom)
-                       iter_rb_tree(&nextents, &nchunks);
+                       iter_rb_tree(&nextents, &nchunks, &ndupextents);
 
-               show_stat(nextents, nchunks);
+               show_stat(nextents, nchunks, ndupextents, unique_capacity);
        }
 
-       fio_mutex_remove(rb_lock);
+       fio_sem_remove(rb_lock);
        if (bloom)
                bloom_free(bloom);
        scleanup();