#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;
uint64_t size;
unsigned long items;
+ unsigned long dupes;
int err;
};
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 {
};
static struct rb_root rb_root;
+static struct bloom *bloom;
static struct fio_mutex *rb_lock;
static unsigned int blocksize = 4096;
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;
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;
}
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)
{
- 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++;
}
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;
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;
}
}
- 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);
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);
}
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)
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)) {
} 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(int fd, uint64_t dev_size, uint64_t *nextents,
+ uint64_t *nchunks)
{
struct worker_thread *threads;
unsigned long nitems, total_items;
}
}
- 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;
return 1;
}
- printf("Will check <%s>, size <%llu>\n", filename, (unsigned long long) dev_size);
+ if (use_bloom) {
+ uint64_t bloom_entries;
+
+ 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);
- return __dedupe_check(dev_fd, dev_size);
+ return run_dedupe_threads(dev_fd, dev_size, nextents, nchunks);
}
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)
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[])
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);
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();
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;
}
projects / fio.git / blobdiff