* just scans the filename for extents of the given size, checksums them,
* and orders them up.
*/
+#include <fcntl.h>
+#include <inttypes.h>
#include <stdio.h>
-#include <stdio.h>
+#include <string.h>
#include <unistd.h>
-#include <inttypes.h>
-#include <assert.h>
-#include <sys/types.h>
#include <sys/stat.h>
-#include <sys/ioctl.h>
-#include <linux/fs.h>
-#include <fcntl.h>
-#include <string.h>
-#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"
-FILE *f_err;
-struct timeval *fio_tv = NULL;
-unsigned int fio_debug = 0;
+#include "../lib/bloom.h"
+#include "debug.h"
+#include "zlib.h"
-void __dprint(int type, const char *str, ...)
-{
-}
+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 {
};
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];
};
static struct rb_root rb_root;
-static struct fio_mutex *rb_lock;
+static struct bloom *bloom;
+static struct fio_sem *rb_lock;
static unsigned int blocksize = 4096;
static unsigned int num_threads;
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));
}
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;
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;
}
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)
{
/*
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;
}
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;
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;
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)
+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++)
- insert_chunk(&items[i]);
+ for (i = 0; i < nitems; i++) {
+ if (bloom) {
+ unsigned int s;
+ int r;
- fio_mutex_up(rb_lock);
+ s = sizeof(items[i].hash) / sizeof(uint32_t);
+ r = bloom_set(bloom, items[i].hash, s);
+ *ndupes += r;
+ } else {
+ ret = insert_chunk(&items[i], unique_capacity, zc);
+ if (ret)
+ break;
+ }
+ }
+
+ fio_sem_up(rb_lock);
+ return ret;
}
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;
+ 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;
+ ret = insert_chunks(items, nitems, &ndupes, &unique_capacity, &thread->zc);
+
free(items);
- 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)
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)) {
} 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);
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++) {
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)
+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;
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) {
show_progress(threads, total_items);
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);
- fio_mutex_remove(size_lock);
+ *nextents = nitems;
+ *nchunks = nitems - *nchunks;
+
+ fio_sem_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 *unique_capacity)
{
uint64_t dev_size;
struct stat sb;
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 = 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);
+ 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)
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 iter_rb_tree(void)
+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)
{
- struct rb_node *n;
- uint64_t nchunks;
- uint64_t nextents;
- double perc;
+ uint8_t i = 0;
- nchunks = nextents = 0;
+ 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, uint64_t *ndupextents)
+{
+ struct fio_rb_node *n;
+ *nchunks = *nextents = *ndupextents = 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;
+ *ndupextents += (c->count > 1);
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");
+ log_err("\t-C\tCalculate compressible size\n");
return 1;
}
int main(int argc, char *argv[])
{
+ uint64_t nextents = 0, nchunks = 0, ndupextents = 0, unique_capacity;
int c, ret;
- while ((c = getopt(argc, argv, "b:t:d:o:c:p:")) != -1) {
+ arch_init(argv);
+ debug_init();
+
+ while ((c = getopt(argc, argv, "b:t:d:o:c:p:B:C:")) != -1) {
switch (c) {
case 'b':
blocksize = atoi(optarg);
case 'p':
print_progress = atoi(optarg);
break;
+ case 'B':
+ use_bloom = atoi(optarg);
+ break;
+ case 'C':
+ compression = atoi(optarg);
+ break;
case '?':
default:
return usage(argv);
}
}
+ 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);
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, &unique_capacity);
- ret = dedupe_check(argv[optind]);
+ if (!ret) {
+ if (!bloom)
+ iter_rb_tree(&nextents, &nchunks, &ndupextents);
- iter_rb_tree();
+ show_stat(nextents, nchunks, ndupextents, unique_capacity);
+ }
- fio_mutex_remove(rb_lock);
+ fio_sem_remove(rb_lock);
+ if (bloom)
+ bloom_free(bloom);
scleanup();
return ret;
}
projects / fio.git / blobdiff