X-Git-Url: https://git.kernel.dk/?a=blobdiff_plain;f=t%2Fdedupe.c;h=02e52b742ee33730e94c42c4ca6741f9b17f5eaf;hb=7d7a704638a1e957c845c04eeac82bdeda0c674c;hp=37437c2c4e970825ae61c7f0378503a6f7040879;hpb=9b36c122ab218c278651d0565d69339eb9948808;p=fio.git diff --git a/t/dedupe.c b/t/dedupe.c index 37437c2c..02e52b74 100644 --- a/t/dedupe.c +++ b/t/dedupe.c @@ -3,45 +3,46 @@ * just scans the filename for extents of the given size, checksums them, * and orders them up. */ +#include +#include #include -#include +#include #include -#include -#include -#include #include -#include -#include -#include -#include -#include "../lib/rbtree.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 "../memalign.h" #include "../os/os.h" #include "../gettime.h" #include "../fio_time.h" +#include "../lib/rbtree.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 { @@ -50,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]; @@ -63,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; @@ -73,24 +74,29 @@ 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 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 = 0; + + if (blockdev_size(f, &bytes)) { + log_err("dedupe: failed getting bdev size\n"); return 0; } - } else + ret = bytes; + } else { ret = sb->st_size; + } return (ret & ~((uint64_t)blocksize - 1)); } @@ -100,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; @@ -110,21 +116,21 @@ static int get_work(uint64_t *offset, uint64_t *size) ret = 0; } - fio_mutex_up(size_lock); + fio_sem_up(size_lock); 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) + } 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 +138,41 @@ 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 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) { /* @@ -155,20 +196,20 @@ 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(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); out: - fio_memfree(cbuf, blocksize); - fio_memfree(ibuf, blocksize); + fio_memfree(cbuf, blocksize, false); + fio_memfree(ibuf, blocksize, false); return ret; } @@ -179,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; @@ -198,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; @@ -225,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) { @@ -244,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) @@ -260,34 +312,70 @@ 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; + uint64_t unique_capacity = 0; struct item *items; + int ret; - 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; - if (items) - items[i].offset = offset; - crc_buf(buf, items[i].hash); + void *thisptr = buf + (i * blocksize); + + items[i].offset = offset; + crc_buf(thisptr, items[i].hash); offset += blocksize; 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 err; + 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) @@ -295,7 +383,8 @@ static void *thread_fn(void *data) struct worker_thread *thread = data; void *buf; - buf = fio_memalign(blocksize, blocksize); + buf = fio_memalign(blocksize, chunk_size, false); + thread_init_zlib_control(thread); do { if (get_work(&thread->cur_offset, &thread->size)) { @@ -309,14 +398,14 @@ static void *thread_fn(void *data) } while (1); thread->done = 1; - fio_memfree(buf, blocksize); + fio_memfree(buf, chunk_size, false); return NULL; } static void show_progress(struct worker_thread *threads, unsigned long total) { unsigned long last_nitems = 0; - struct timeval last_tv; + struct timespec last_tv; fio_gettime(&last_tv, NULL); @@ -325,7 +414,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++) { @@ -344,19 +433,21 @@ 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; - printf("%3.2f%% done (%luKB/sec)\r", perc, this_items); + this_items = (this_items * 1000) / (tdiff * 1024); + 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(int fd, uint64_t dev_size, uint64_t *nextents, - uint64_t *nchunks) +static int run_dedupe_threads(struct fio_file *f, uint64_t dev_size, + uint64_t *nextents, uint64_t *nchunks, + uint64_t *unique_capacity) { struct worker_thread *threads; unsigned long nitems, total_items; @@ -365,14 +456,12 @@ static int run_dedupe_threads(int fd, uint64_t dev_size, uint64_t *nextents, 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++) { - 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) { @@ -386,11 +475,13 @@ static int run_dedupe_threads(int fd, uint64_t dev_size, uint64_t *nextents, 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); @@ -398,13 +489,13 @@ static int run_dedupe_threads(int fd, uint64_t dev_size, uint64_t *nextents, *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; @@ -412,36 +503,43 @@ static int dedupe_check(const char *filename, uint64_t *nextents, flags = O_RDONLY; if (odirect) - flags |= O_DIRECT; + flags |= OS_O_DIRECT; - dev_fd = open(filename, flags); - if (dev_fd == -1) { + memset(&file, 0, sizeof(file)); + file.file_name = strdup(filename); + + 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 = (3 * dev_size ) / (blocksize * 2); + bloom_entries = 8 * (dev_size / blocksize); 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(dev_fd, dev_size, nextents, nchunks); + return run_dedupe_threads(&file, dev_size, nextents, nchunks, + unique_capacity); +err: + if (file.fd != -1) + close(file.fd); + free(file.file_name); + return 1; } static void show_chunk(struct chunk *c) @@ -449,31 +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) { - double perc; + uint8_t i = 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); + 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", (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 { + 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) @@ -485,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); @@ -503,17 +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); @@ -536,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); @@ -554,16 +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 (!bloom) - iter_rb_tree(&nextents, &nchunks); + if (!ret) { + if (!bloom) + 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();