dedupe: fix warning and segfault on -B0
[fio.git] / t / dedupe.c
index 03c5032bcf8d79e83f7ab99ceca89d0a621291d7..37437c2c4e970825ae61c7f0378503a6f7040879 100644 (file)
 #include "../crc/md5.h"
 #include "../memalign.h"
 #include "../os/os.h"
+#include "../gettime.h"
+#include "../fio_time.h"
 
-FILE *f_err;
-struct timeval *fio_tv = NULL;
-unsigned int fio_debug = 0;
-
-void __dprint(int type, const char *str, ...)
-{
-}
+#include "../lib/bloom.h"
+#include "debug.h"
 
 struct worker_thread {
        pthread_t thread;
 
+       volatile int done;
+
        int fd;
        uint64_t cur_offset;
        uint64_t size;
 
        unsigned long items;
+       unsigned long dupes;
        int err;
 };
 
@@ -51,9 +51,9 @@ struct extent {
 
 struct chunk {
        struct rb_node rb_node;
-       struct flist_head extent_list;
        uint64_t count;
        uint32_t hash[MD5_HASH_WORDS];
+       struct flist_head extent_list[0];
 };
 
 struct item {
@@ -62,6 +62,7 @@ struct item {
 };
 
 static struct rb_root rb_root;
+static struct bloom *bloom;
 static struct fio_mutex *rb_lock;
 
 static unsigned int blocksize = 4096;
@@ -70,6 +71,8 @@ static unsigned int chunk_size = 1048576;
 static unsigned int dump_output;
 static unsigned int odirect;
 static unsigned int collision_check;
+static unsigned int print_progress = 1;
+static unsigned int use_bloom = 1;
 
 static uint64_t total_size;
 static uint64_t cur_offset;
@@ -131,11 +134,18 @@ static int read_block(int fd, void *buf, off_t offset)
 
 static void add_item(struct chunk *c, struct item *i)
 {
-       struct extent *e;
+       /*      
+        * Save some memory and don't add extent items, if we don't
+        * use them.
+        */
+       if (dump_output || collision_check) {
+               struct extent *e;
+
+               e = malloc(sizeof(*e));
+               e->offset = i->offset;
+               flist_add_tail(&e->list, &c->extent_list[0]);
+       }
 
-       e = malloc(sizeof(*e));
-       e->offset = i->offset;
-       flist_add_tail(&e->list, &c->extent_list);
        c->count++;
 }
 
@@ -148,7 +158,7 @@ static int col_check(struct chunk *c, struct item *i)
        cbuf = fio_memalign(blocksize, blocksize);
        ibuf = fio_memalign(blocksize, blocksize);
 
-       e = flist_entry(c->extent_list.next, struct extent, list);
+       e = flist_entry(c->extent_list[0].next, struct extent, list);
        if (read_block(dev_fd, cbuf, e->offset))
                goto out;
 
@@ -162,6 +172,19 @@ out:
        return ret;
 }
 
+static struct chunk *alloc_chunk(void)
+{
+       struct chunk *c;
+
+       if (collision_check || dump_output) {
+               c = malloc(sizeof(struct chunk) + sizeof(struct flist_head));
+               INIT_FLIST_HEAD(&c->extent_list[0]);
+       } else
+               c = malloc(sizeof(struct chunk));
+
+       return c;
+}
+
 static void insert_chunk(struct item *i)
 {
        struct rb_node **p, *parent;
@@ -196,9 +219,8 @@ static void insert_chunk(struct item *i)
                }
        }
 
-       c = malloc(sizeof(*c));
+       c = alloc_chunk();
        RB_CLEAR_NODE(&c->rb_node);
-       INIT_FLIST_HEAD(&c->extent_list);
        c->count = 0;
        memcpy(c->hash, i->hash, sizeof(i->hash));
        rb_link_node(&c->rb_node, parent, p);
@@ -207,14 +229,24 @@ add:
        add_item(c, i);
 }
 
-static void insert_chunks(struct item *items, unsigned int nitems)
+static void insert_chunks(struct item *items, unsigned int nitems,
+                         uint64_t *ndupes)
 {
        int i;
 
        fio_mutex_down(rb_lock);
 
-       for (i = 0; i < nitems; i++)
-               insert_chunk(&items[i]);
+       for (i = 0; i < nitems; i++) {
+               if (bloom) {
+                       unsigned int s;
+                       int r;
+
+                       s = sizeof(items[i].hash) / sizeof(uint32_t);
+                       r = bloom_set(bloom, items[i].hash, s);
+                       *ndupes += r;
+               } else
+                       insert_chunk(&items[i]);
+       }
 
        fio_mutex_up(rb_lock);
 }
@@ -233,6 +265,7 @@ static int do_work(struct worker_thread *thread, void *buf)
        unsigned int nblocks, i;
        off_t offset;
        int err = 0, nitems = 0;
+       uint64_t ndupes = 0;
        struct item *items;
 
        nblocks = thread->size / blocksize;
@@ -242,15 +275,18 @@ static int do_work(struct worker_thread *thread, void *buf)
        for (i = 0; i < nblocks; i++) {
                if (read_block(thread->fd, buf, offset))
                        break;
-               items[i].offset = offset;
+               if (items)
+                       items[i].offset = offset;
                crc_buf(buf, items[i].hash);
                offset += blocksize;
                nitems++;
        }
 
-       insert_chunks(items, nitems);
-       thread->items += nitems;
+       insert_chunks(items, nitems, &ndupes);
+
        free(items);
+       thread->items += nitems;
+       thread->dupes += ndupes;
        return err;
 }
 
@@ -272,17 +308,62 @@ static void *thread_fn(void *data)
                }
        } while (1);
 
+       thread->done = 1;
        fio_memfree(buf, blocksize);
        return NULL;
 }
 
-static int __dedupe_check(int fd, uint64_t dev_size)
+static void show_progress(struct worker_thread *threads, unsigned long total)
+{
+       unsigned long last_nitems = 0;
+       struct timeval last_tv;
+
+       fio_gettime(&last_tv, NULL);
+
+       while (print_progress) {
+               unsigned long this_items;
+               unsigned long nitems = 0;
+               uint64_t tdiff;
+               float perc;
+               int some_done;
+               int i;
+
+               for (i = 0; i < num_threads; i++) {
+                       nitems += threads[i].items;
+                       some_done = threads[i].done;
+                       if (some_done)
+                               break;
+               }
+
+               if (some_done)
+                       break;
+
+               perc = (float) nitems / (float) total;
+               perc *= 100.0;
+               this_items = nitems - last_nitems;
+               this_items *= blocksize;
+               tdiff = mtime_since_now(&last_tv);
+               if (tdiff) {
+                       this_items /= tdiff;
+                       printf("%3.2f%% done (%luKB/sec)\r", perc, this_items);
+                       last_nitems = nitems;
+                       fio_gettime(&last_tv, NULL);
+               } else
+                       printf("%3.2f%% done\r", perc);
+               fflush(stdout);
+               usleep(250000);
+       };
+}
+
+static int run_dedupe_threads(int fd, uint64_t dev_size, uint64_t *nextents,
+                               uint64_t *nchunks)
 {
        struct worker_thread *threads;
-       unsigned long nitems;
+       unsigned long nitems, total_items;
        int i, err = 0;
 
        total_size = dev_size;
+       total_items = dev_size / blocksize;
        cur_offset = 0;
        size_lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
 
@@ -291,6 +372,7 @@ static int __dedupe_check(int fd, uint64_t dev_size)
                threads[i].fd = fd;
                threads[i].items = 0;
                threads[i].err = 0;
+               threads[i].done = 0;
 
                err = pthread_create(&threads[i].thread, NULL, thread_fn, &threads[i]);
                if (err) {
@@ -299,20 +381,30 @@ static int __dedupe_check(int fd, uint64_t dev_size)
                }
        }
 
+       show_progress(threads, total_items);
+
        nitems = 0;
+       *nextents = 0;
+       *nchunks = 1;
        for (i = 0; i < num_threads; i++) {
                void *ret;
                pthread_join(threads[i].thread, &ret);
                nitems += threads[i].items;
+               *nchunks += threads[i].dupes;
        }
 
        printf("Threads(%u): %lu items processed\n", num_threads, nitems);
 
+       *nextents = nitems;
+       *nchunks = nitems - *nchunks;
+
        fio_mutex_remove(size_lock);
+       free(threads);
        return err;
 }
 
-static int dedupe_check(const char *filename)
+static int dedupe_check(const char *filename, uint64_t *nextents,
+                       uint64_t *nchunks)
 {
        uint64_t dev_size;
        struct stat sb;
@@ -340,9 +432,16 @@ static int dedupe_check(const char *filename)
                return 1;
        }
 
-       printf("Will check <%s>, size <%lu>\n", filename, dev_size);
+       if (use_bloom) {
+               uint64_t bloom_entries;
 
-       return __dedupe_check(dev_fd, dev_size);
+               bloom_entries = (3 * dev_size ) / (blocksize * 2);
+               bloom = bloom_new(bloom_entries);
+       }
+
+       printf("Will check <%s>, size <%llu>, using %u threads\n", filename, (unsigned long long) dev_size, num_threads);
+
+       return run_dedupe_threads(dev_fd, dev_size, nextents, nchunks);
 }
 
 static void show_chunk(struct chunk *c)
@@ -350,21 +449,31 @@ 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], c->count);
-       flist_for_each(n, &c->extent_list) {
+       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 %lu\n", e->offset);
+               printf("\toffset %llu\n", (unsigned long long) e->offset);
        }
 }
 
-static void iter_rb_tree(void)
+static void show_stat(uint64_t nextents, uint64_t nchunks)
 {
-       struct rb_node *n;
-       uint64_t nchunks;
-       uint64_t nextents;
        double perc;
 
-       nchunks = nextents = 0;
+       printf("Extents=%lu, Unique extents=%lu\n", (unsigned long) nextents, (unsigned long) nchunks);
+       printf("De-dupe factor: %3.2f\n", (double) nextents / (double) nchunks);
+
+       perc = 1.00 - ((double) nchunks / (double) nextents);
+       perc *= 100.0;
+       printf("Fio setting: dedupe_percentage=%u\n", (int) (perc + 0.50));
+
+}
+
+static void iter_rb_tree(uint64_t *nextents, uint64_t *nchunks)
+{
+       struct rb_node *n;
+
+       *nchunks = *nextents = 0;
 
        n = rb_first(&rb_root);
        if (!n)
@@ -374,20 +483,13 @@ static void iter_rb_tree(void)
                struct chunk *c;
 
                c = rb_entry(n, struct chunk, rb_node);
-               nchunks++;
-               nextents += c->count;
+               (*nchunks)++;
+               *nextents += c->count;
 
                if (dump_output)
                        show_chunk(c);
 
        } while ((n = rb_next(n)) != NULL);
-
-       printf("Extents=%lu, Unique extents=%lu\n", nextents, nchunks);
-       printf("De-dupe factor: %3.2f\n", (double) nextents / (double) nchunks);
-
-       perc = 1.00 - ((double) nchunks / (double) nextents);
-       perc *= 100.0;
-       printf("Fio setting: dedupe_percentage=%u\n", (int) (perc + 0.50));
 }
 
 static int usage(char *argv[])
@@ -399,14 +501,19 @@ static int usage(char *argv[])
        log_err("\t-d\tFull extent/chunk debug output\n");
        log_err("\t-o\tUse O_DIRECT\n");
        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");
        return 1;
 }
 
 int main(int argc, char *argv[])
 {
+       uint64_t nextents = 0, nchunks = 0;
        int c, ret;
 
-       while ((c = getopt(argc, argv, "b:t:d:o:c:")) != -1) {
+       debug_init();
+
+       while ((c = getopt(argc, argv, "b:t:d:o:c:p:B:")) != -1) {
                switch (c) {
                case 'b':
                        blocksize = atoi(optarg);
@@ -423,12 +530,21 @@ int main(int argc, char *argv[])
                case 'c':
                        collision_check = atoi(optarg);
                        break;
+               case 'p':
+                       print_progress = atoi(optarg);
+                       break;
+               case 'B':
+                       use_bloom = atoi(optarg);
+                       break;
                case '?':
                default:
                        return usage(argv);
                }
        }
 
+       if (collision_check || dump_output)
+               use_bloom = 0;
+
        if (!num_threads)
                num_threads = cpus_online();
 
@@ -440,10 +556,16 @@ int main(int argc, char *argv[])
        rb_root = RB_ROOT;
        rb_lock = fio_mutex_init(FIO_MUTEX_UNLOCKED);
 
-       ret = dedupe_check(argv[optind]);
+       ret = dedupe_check(argv[optind], &nextents, &nchunks);
+
+       if (!bloom)
+               iter_rb_tree(&nextents, &nchunks);
 
-       iter_rb_tree();
+       show_stat(nextents, nchunks);
 
+       fio_mutex_remove(rb_lock);
+       if (bloom)
+               bloom_free(bloom);
        scleanup();
        return ret;
 }