X-Git-Url: https://git.kernel.dk/?p=fio.git;a=blobdiff_plain;f=t%2Fdedupe.c;h=5b88fcb8961ad6904b096c29082534ec59228cc3;hp=7acb090a63e4a0a080f083b715a7c7d9df9ba3c2;hb=60a5a18b811cd60430d2912b49fa3970df0d5077;hpb=03ecad7b9d6f8a793c669b5e15fc776b388b6edf diff --git a/t/dedupe.c b/t/dedupe.c index 7acb090a..5b88fcb8 100644 --- a/t/dedupe.c +++ b/t/dedupe.c @@ -11,7 +11,6 @@ #include #include #include -#include #include #include @@ -24,14 +23,11 @@ #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; @@ -43,6 +39,7 @@ struct worker_thread { uint64_t size; unsigned long items; + unsigned long dupes; int err; }; @@ -53,9 +50,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 { @@ -64,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; @@ -73,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; @@ -114,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; } @@ -132,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) { /* @@ -143,7 +150,7 @@ static void add_item(struct chunk *c, struct item *i) e = malloc(sizeof(*e)); e->offset = i->offset; - flist_add_tail(&e->list, &c->extent_list); + flist_add_tail(&e->list, &c->extent_list[0]); } c->count++; @@ -158,11 +165,11 @@ 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); - if (read_block(dev_fd, cbuf, e->offset)) + e = flist_entry(c->extent_list[0].next, struct extent, list); + 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); @@ -172,6 +179,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; @@ -206,9 +226,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); @@ -217,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); } @@ -238,30 +267,45 @@ 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; + void *thisptr = buf + (i * blocksize); + items[i].offset = offset; - crc_buf(buf, items[i].hash); + 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) @@ -269,7 +313,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)) { @@ -283,11 +327,54 @@ static void *thread_fn(void *data) } while (1); thread->done = 1; - fio_memfree(buf, blocksize); + fio_memfree(buf, chunk_size); 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 = 0; + 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 = (this_items * 1000) / (tdiff * 1024); + 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(struct fio_file *f, uint64_t dev_size, + uint64_t *nextents, uint64_t *nchunks) { struct worker_thread *threads; unsigned long nitems, total_items; @@ -300,10 +387,8 @@ static int __dedupe_check(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].items = 0; - threads[i].err = 0; - threads[i].done = 0; + memset(&threads[i], 0, sizeof(struct worker_thread)); + threads[i].fd = f->fd; err = pthread_create(&threads[i].thread, NULL, thread_fn, &threads[i]); if (err) { @@ -312,42 +397,30 @@ static int __dedupe_check(int fd, uint64_t dev_size) } } - while (print_progress) { - float perc; - int some_done; - - nitems = 0; - 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_items; - perc *= 100.0; - printf("%3.2f%% done\r", perc); - fflush(stdout); - usleep(200000); - }; + 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; @@ -355,29 +428,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 __dedupe_check(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) @@ -386,20 +471,35 @@ static void show_chunk(struct chunk *c) 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); - flist_for_each(n, &c->extent_list) { + 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 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) @@ -409,20 +509,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[]) @@ -434,15 +527,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); @@ -462,12 +559,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(); @@ -479,10 +582,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; }