t/dedupe: use fio abstracted OS_O_DIRECT
[fio.git] / t / dedupe.c
index 16c1748d86f19131f7985daa9b6eecda168fbbf0..030b4fc9d00f167b2d0eea363f03e73fdb469060 100644 (file)
@@ -11,7 +11,6 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <sys/ioctl.h>
-#include <linux/fs.h>
 #include <fcntl.h>
 #include <string.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;
@@ -45,6 +39,7 @@ struct worker_thread {
        uint64_t size;
 
        unsigned long items;
+       unsigned long dupes;
        int err;
 };
 
@@ -66,6 +61,7 @@ struct item {
 };
 
 static struct rb_root rb_root;
+static struct bloom *bloom;
 static struct fio_mutex *rb_lock;
 
 static unsigned int blocksize = 4096;
@@ -75,22 +71,26 @@ 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;
 static struct fio_mutex *size_lock;
 
-static int dev_fd;
+static struct fio_file file;
 
-static uint64_t get_size(int fd, struct stat *sb)
+static uint64_t get_size(struct fio_file *f, struct stat *sb)
 {
        uint64_t ret;
 
        if (S_ISBLK(sb->st_mode)) {
-               if (ioctl(fd, BLKGETSIZE64, &ret) < 0) {
-                       perror("ioctl");
+               unsigned long long bytes;
+
+               if (blockdev_size(f, &bytes)) {
+                       log_err("dedupe: failed getting bdev size\n");
                        return 0;
                }
+               ret = bytes;
        } else
                ret = sb->st_size;
 
@@ -116,17 +116,17 @@ static int get_work(uint64_t *offset, uint64_t *size)
        return ret;
 }
 
-static int read_block(int fd, void *buf, off_t offset)
+static int __read_block(int fd, void *buf, off_t offset, size_t count)
 {
        ssize_t ret;
 
-       ret = pread(fd, buf, blocksize, offset);
+       ret = pread(fd, buf, count, offset);
        if (ret < 0) {
                perror("pread");
                return 1;
        } else if (!ret)
                return 1;
-       else if (ret != blocksize) {
+       else if (ret != count) {
                log_err("dedupe: short read on block\n");
                return 1;
        }
@@ -134,6 +134,11 @@ static int read_block(int fd, void *buf, off_t offset)
        return 0;
 }
 
+static int read_block(int fd, void *buf, off_t offset)
+{
+       return __read_block(fd, buf, offset, blocksize);
+}
+
 static void add_item(struct chunk *c, struct item *i)
 {
        /*      
@@ -161,10 +166,10 @@ static int col_check(struct chunk *c, struct item *i)
        ibuf = fio_memalign(blocksize, blocksize);
 
        e = flist_entry(c->extent_list[0].next, struct extent, list);
-       if (read_block(dev_fd, cbuf, e->offset))
+       if (read_block(file.fd, cbuf, e->offset))
                goto out;
 
-       if (read_block(dev_fd, ibuf, i->offset))
+       if (read_block(file.fd, ibuf, i->offset))
                goto out;
 
        ret = memcmp(ibuf, cbuf, blocksize);
@@ -231,14 +236,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);
 }
@@ -252,30 +267,46 @@ static void crc_buf(void *buf, uint32_t *hash)
        fio_md5_final(&ctx);
 }
 
+static unsigned int read_blocks(int fd, void *buf, off_t offset, size_t size)
+{
+       if (__read_block(fd, buf, offset, size))
+               return 0;
+
+       return size / blocksize;
+}
+
 static int do_work(struct worker_thread *thread, void *buf)
 {
        unsigned int nblocks, i;
        off_t offset;
-       int err = 0, nitems = 0;
+       int nitems = 0;
+       uint64_t ndupes = 0;
        struct item *items;
 
-       nblocks = thread->size / blocksize;
        offset = thread->cur_offset;
+
+       nblocks = read_blocks(thread->fd, buf, offset, min(thread->size, (uint64_t)chunk_size));
+       if (!nblocks)
+               return 1;
+
        items = malloc(sizeof(*items) * nblocks);
 
        for (i = 0; i < nblocks; i++) {
-               if (read_block(thread->fd, buf, offset))
-                       break;
-               items[i].offset = offset;
-               crc_buf(buf, items[i].hash);
+               void *thisptr = buf + (i * blocksize);
+
+               if (items)
+                       items[i].offset = offset;
+               crc_buf(thisptr, items[i].hash);
                offset += blocksize;
                nitems++;
        }
 
-       insert_chunks(items, nitems);
-       thread->items += nitems;
+       insert_chunks(items, nitems, &ndupes);
+
        free(items);
-       return err;
+       thread->items += nitems;
+       thread->dupes += ndupes;
+       return 0;
 }
 
 static void *thread_fn(void *data)
@@ -283,7 +314,7 @@ static void *thread_fn(void *data)
        struct worker_thread *thread = data;
        void *buf;
 
-       buf = fio_memalign(blocksize, blocksize);
+       buf = fio_memalign(blocksize, chunk_size);
 
        do {
                if (get_work(&thread->cur_offset, &thread->size)) {
@@ -297,7 +328,7 @@ static void *thread_fn(void *data)
        } while (1);
 
        thread->done = 1;
-       fio_memfree(buf, blocksize);
+       fio_memfree(buf, chunk_size);
        return NULL;
 }
 
@@ -313,7 +344,7 @@ static void show_progress(struct worker_thread *threads, unsigned long total)
                unsigned long nitems = 0;
                uint64_t tdiff;
                float perc;
-               int some_done;
+               int some_done = 0;
                int i;
 
                for (i = 0; i < num_threads; i++) {
@@ -332,7 +363,7 @@ static void show_progress(struct worker_thread *threads, unsigned long total)
                this_items *= blocksize;
                tdiff = mtime_since_now(&last_tv);
                if (tdiff) {
-                       this_items /= tdiff;
+                       this_items = (this_items * 1000) / (tdiff * 1024);
                        printf("%3.2f%% done (%luKB/sec)\r", perc, this_items);
                        last_nitems = nitems;
                        fio_gettime(&last_tv, NULL);
@@ -343,7 +374,8 @@ static void show_progress(struct worker_thread *threads, unsigned long total)
        };
 }
 
-static int run_dedupe_threads(int fd, uint64_t dev_size)
+static int run_dedupe_threads(struct fio_file *f, uint64_t dev_size,
+                             uint64_t *nextents, uint64_t *nchunks)
 {
        struct worker_thread *threads;
        unsigned long nitems, total_items;
@@ -356,7 +388,7 @@ static int run_dedupe_threads(int fd, uint64_t dev_size)
 
        threads = malloc(num_threads * sizeof(struct worker_thread));
        for (i = 0; i < num_threads; i++) {
-               threads[i].fd = fd;
+               threads[i].fd = f->fd;
                threads[i].items = 0;
                threads[i].err = 0;
                threads[i].done = 0;
@@ -371,20 +403,27 @@ static int run_dedupe_threads(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;
@@ -392,29 +431,41 @@ static int dedupe_check(const char *filename)
 
        flags = O_RDONLY;
        if (odirect)
-               flags |= O_DIRECT;
+               flags |= OS_O_DIRECT;
+
+       memset(&file, 0, sizeof(file));
+       file.file_name = strdup(filename);
 
-       dev_fd = open(filename, flags);
-       if (dev_fd == -1) {
+       file.fd = open(filename, flags);
+       if (file.fd == -1) {
                perror("open");
-               return 1;
+               goto err;
        }
 
-       if (fstat(dev_fd, &sb) < 0) {
+       if (fstat(file.fd, &sb) < 0) {
                perror("fstat");
-               close(dev_fd);
-               return 1;
+               goto err;
        }
 
-       dev_size = get_size(dev_fd, &sb);
-       if (!dev_size) {
-               close(dev_fd);
-               return 1;
+       dev_size = get_size(&file, &sb);
+       if (!dev_size)
+               goto err;
+
+       if (use_bloom) {
+               uint64_t bloom_entries;
+
+               bloom_entries = 8 * (dev_size / blocksize);
+               bloom = bloom_new(bloom_entries);
        }
 
-       printf("Will check <%s>, size <%llu>\n", filename, (unsigned long long) dev_size);
+       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);
+       return run_dedupe_threads(&file, dev_size, nextents, nchunks);
+err:
+       if (file.fd != -1)
+               close(file.fd);
+       free(file.file_name);
+       return 1;
 }
 
 static void show_chunk(struct chunk *c)
@@ -429,14 +480,29 @@ static void show_chunk(struct chunk *c)
        }
 }
 
-static void iter_rb_tree(void)
+static void show_stat(uint64_t nextents, uint64_t nchunks)
+{
+       double perc, ratio;
+
+       printf("Extents=%lu, Unique extents=%lu\n", (unsigned long) nextents, (unsigned long) nchunks);
+
+       if (nchunks) {
+               ratio = (double) nextents / (double) nchunks;
+               printf("De-dupe ratio: 1:%3.2f\n", ratio - 1.0);
+       } else
+               printf("De-dupe ratio: 1:infinite\n");
+
+       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;
-       uint64_t nchunks;
-       uint64_t nextents;
-       double perc;
 
-       nchunks = nextents = 0;
+       *nchunks = *nextents = 0;
 
        n = rb_first(&rb_root);
        if (!n)
@@ -446,20 +512,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", (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 int usage(char *argv[])
@@ -471,15 +530,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:p:")) != -1) {
+       debug_init();
+
+       while ((c = getopt(argc, argv, "b:t:d:o:c:p:B:")) != -1) {
                switch (c) {
                case 'b':
                        blocksize = atoi(optarg);
@@ -499,12 +562,18 @@ int main(int argc, char *argv[])
                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();
 
@@ -516,11 +585,18 @@ 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 (!ret) {
+               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;
 }