greg/pcompress
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity
pcompress/rabin/rabin_dedup.c
Moinak Ghosh d49a088eea Fixes and performance improvements for Dedupe Delta Compression 
Avoid using fingerprints in minhash computation and fix write amplification
Modify min-heap to use 64bit values
Improve bsdiff performance
Fix pointer comparison in bsdiff
Use 32bit offsets in bsdiff to reduce memory usage
Improve Zero RLE Encoder performance
Add more buffer overflow checks in Zero RLE Decoder
2013-01-13 22:04:59 +05:30

856 lines
25 KiB
C
Executable file
Raw Blame History

/*
* The rabin polynomial computation is derived from:
* http://code.google.com/p/rabin-fingerprint-c/
*
* originally created by Joel Lawrence Tucci on 09-March-2011.
*
* Rabin polynomial portions Copyright (c) 2011 Joel Lawrence Tucci
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
*
* Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
* Neither the name of the project's author nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
* FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
* TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
* NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
/*
* This file is a part of Pcompress, a chunked parallel multi-
* algorithm lossless compression and decompression program.
*
* Copyright (C) 2012 Moinak Ghosh. All rights reserved.
* Use is subject to license terms.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* moinakg@belenix.org, http://moinakg.wordpress.com/
*
*/
#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS 1
#endif
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <allocator.h>
#include <utils.h>
#include <pthread.h>
#include <heapq.h>
#include <xxhash.h>
#include "rabin_dedup.h"
#define DELTA_EXTRA2_PCT(x) ((x) >> 1)
#define DELTA_EXTRA_PCT(x) (((x) >> 1) + ((x) >> 3))
#define DELTA_NORMAL_PCT(x) (((x) >> 1) + ((x) >> 2) + ((x) >> 3))
extern int lzma_init(void **data, int *level, int nthreads, int64_t chunksize,
int file_version, compress_op_t op);
extern int lzma_compress(void *src, uint64_t srclen, void *dst,
uint64_t *destlen, int level, uchar_t chdr, void *data);
extern int lzma_decompress(void *src, uint64_t srclen, void *dst,
uint64_t *dstlen, int level, uchar_t chdr, void *data);
extern int lzma_deinit(void **data);
extern int bsdiff(u_char *oldbuf, bsize_t oldsize, u_char *newbuf, bsize_t newsize,
u_char *diff, u_char *scratch, bsize_t scratchsize);
extern bsize_t get_bsdiff_sz(u_char *pbuf);
extern int bspatch(u_char *pbuf, u_char *oldbuf, bsize_t oldsize, u_char *newbuf,
bsize_t *_newsize);
static pthread_mutex_t init_lock = PTHREAD_MUTEX_INITIALIZER;
uint64_t ir[256], out[256];
static int inited = 0;
static uint32_t
dedupe_min_blksz(int rab_blk_sz)
{
uint32_t min_blk;
min_blk = (1 << (rab_blk_sz + RAB_BLK_MIN_BITS)) - 1024;
return (min_blk);
}
uint32_t
dedupe_buf_extra(uint64_t chunksize, int rab_blk_sz, const char *algo, int delta_flag)
{
if (rab_blk_sz < 1 || rab_blk_sz > 5)
rab_blk_sz = RAB_BLK_DEFAULT;
return ((chunksize / dedupe_min_blksz(rab_blk_sz)) * sizeof (uint32_t));
}
/*
* Initialize the algorithm with the default params.
*/
dedupe_context_t *
create_dedupe_context(uint64_t chunksize, uint64_t real_chunksize, int rab_blk_sz,
const char *algo, int delta_flag, int fixed_flag, int file_version, compress_op_t op) {
dedupe_context_t *ctx;
uint32_t i;
if (rab_blk_sz < 1 || rab_blk_sz > 5)
rab_blk_sz = RAB_BLK_DEFAULT;
if (fixed_flag) {
delta_flag = 0;
inited = 1;
}
/*
* Pre-compute a table of irreducible polynomial evaluations for each
* possible byte value.
*/
pthread_mutex_lock(&init_lock);
if (!inited) {
int term, pow, j;
uint64_t val, poly_pow;
poly_pow = 1;
for (j = 0; j < RAB_POLYNOMIAL_WIN_SIZE; j++) {
poly_pow = (poly_pow * RAB_POLYNOMIAL_CONST) & POLY_MASK;
}
for (j = 0; j < 256; j++) {
term = 1;
pow = 1;
val = 1;
out[j] = (j * poly_pow) & POLY_MASK;
for (i=0; i<RAB_POLYNOMIAL_WIN_SIZE; i++) {
if (term & FP_POLY) {
val += ((pow * j) & POLY_MASK);
}
pow = (pow * RAB_POLYNOMIAL_CONST) & POLY_MASK;
term <<= 1;
}
ir[j] = val;
}
inited = 1;
}
pthread_mutex_unlock(&init_lock);
/*
* Rabin window size must be power of 2 for optimization.
*/
if (!ISP2(RAB_POLYNOMIAL_WIN_SIZE)) {
fprintf(stderr, "Rabin window size must be a power of 2 in range 4 <= x <= 64\n");
return (NULL);
}
if (chunksize < RAB_MIN_CHUNK_SIZE) {
fprintf(stderr, "Minimum chunk size for Dedup must be %" PRIu64 " bytes\n",
RAB_MIN_CHUNK_SIZE);
return (NULL);
}
/*
* For LZMA with chunksize <= LZMA Window size and/or Delta enabled we
* use 4K minimum Rabin block size. For everything else it is 2K based
* on experimentation.
*/
ctx = (dedupe_context_t *)slab_alloc(NULL, sizeof (dedupe_context_t));
ctx->rabin_poly_max_block_size = RAB_POLYNOMIAL_MAX_BLOCK_SIZE;
ctx->current_window_data = NULL;
ctx->fixed_flag = fixed_flag;
ctx->rabin_break_patt = 0;
ctx->rabin_poly_avg_block_size = RAB_BLK_AVG_SZ(rab_blk_sz);
ctx->rabin_avg_block_mask = RAB_BLK_MASK;
ctx->rabin_poly_min_block_size = dedupe_min_blksz(rab_blk_sz);
ctx->delta_flag = 0;
/*
* Scale down similarity percentage based on avg block size unless user specified
* argument '-EE' in which case fixed 40% match is used for Delta compression.
*/
if (delta_flag == DELTA_NORMAL) {
if (ctx->rabin_poly_avg_block_size < (1 << 14)) {
ctx->delta_flag = 1;
} else if (ctx->rabin_poly_avg_block_size < (1 << 16)) {
ctx->delta_flag = 2;
} else {
ctx->delta_flag = 3;
}
} else if (delta_flag == DELTA_EXTRA) {
ctx->delta_flag = 2;
}
if (!fixed_flag)
ctx->blknum = chunksize / ctx->rabin_poly_min_block_size;
else
ctx->blknum = chunksize / ctx->rabin_poly_avg_block_size;
if (chunksize % ctx->rabin_poly_min_block_size)
ctx->blknum++;
if (ctx->blknum > RABIN_MAX_BLOCKS) {
fprintf(stderr, "Chunk size too large for dedup.\n");
destroy_dedupe_context(ctx);
return (NULL);
}
ctx->current_window_data = (uchar_t *)slab_alloc(NULL, RAB_POLYNOMIAL_WIN_SIZE);
ctx->blocks = NULL;
if (real_chunksize > 0) {
ctx->blocks = (rabin_blockentry_t **)slab_calloc(NULL,
ctx->blknum, sizeof (rabin_blockentry_t *));
}
if(ctx == NULL || ctx->current_window_data == NULL ||
(ctx->blocks == NULL && real_chunksize > 0)) {
fprintf(stderr,
"Could not allocate rabin polynomial context, out of memory\n");
destroy_dedupe_context(ctx);
return (NULL);
}
ctx->lzma_data = NULL;
ctx->level = 14;
if (real_chunksize > 0) {
lzma_init(&(ctx->lzma_data), &(ctx->level), 1, chunksize, file_version, op);
// The lzma_data member is not needed during decompression
if (!(ctx->lzma_data) && op == COMPRESS) {
fprintf(stderr,
"Could not initialize LZMA data for dedupe index, out of memory\n");
destroy_dedupe_context(ctx);
return (NULL);
}
}
slab_cache_add(sizeof (rabin_blockentry_t));
ctx->real_chunksize = real_chunksize;
reset_dedupe_context(ctx);
return (ctx);
}
void
reset_dedupe_context(dedupe_context_t *ctx)
{
memset(ctx->current_window_data, 0, RAB_POLYNOMIAL_WIN_SIZE);
ctx->window_pos = 0;
ctx->valid = 0;
}
void
destroy_dedupe_context(dedupe_context_t *ctx)
{
if (ctx) {
uint32_t i;
if (ctx->current_window_data) slab_free(NULL, ctx->current_window_data);
if (ctx->blocks) {
for (i=0; i<ctx->blknum && ctx->blocks[i] != NULL; i++) {
slab_free(NULL, ctx->blocks[i]);
}
slab_free(NULL, ctx->blocks);
}
if (ctx->lzma_data) lzma_deinit(&(ctx->lzma_data));
slab_free(NULL, ctx);
}
}
/**
* Perform Deduplication.
* Both Semi-Rabin fingerprinting based and Fixed Block Deduplication are supported.
* A 16-byte window is used for the rolling checksum and dedup blocks can vary in size
* from 4K-128K.
*/
uint32_t
dedupe_compress(dedupe_context_t *ctx, uchar_t *buf, uint64_t *size, uint64_t offset, uint64_t *rabin_pos)
{
uint64_t i, last_offset, j, ary_sz;
uint32_t blknum;
uchar_t *buf1 = (uchar_t *)buf;
uint32_t length;
uint64_t cur_roll_checksum, cur_pos_checksum;
uint32_t *ctx_heap;
rabin_blockentry_t **htab;
heap_t heap;
DEBUG_STAT_EN(uint32_t max_count);
DEBUG_STAT_EN(max_count = 0);
DEBUG_STAT_EN(double strt, en_1, en);
length = offset;
last_offset = 0;
blknum = 0;
ctx->valid = 0;
cur_roll_checksum = 0;
if (*size < ctx->rabin_poly_avg_block_size) return (0);
DEBUG_STAT_EN(strt = get_wtime_millis());
if (ctx->fixed_flag) {
blknum = *size / ctx->rabin_poly_avg_block_size;
j = *size % ctx->rabin_poly_avg_block_size;
if (j)
blknum++;
else
j = ctx->rabin_poly_avg_block_size;
last_offset = 0;
length = ctx->rabin_poly_avg_block_size;
for (i=0; i<blknum; i++) {
if (i == blknum-1) {
length = j;
}
if (ctx->blocks[i] == 0) {
ctx->blocks[i] = (rabin_blockentry_t *)slab_alloc(NULL,
sizeof (rabin_blockentry_t));
}
ctx->blocks[i]->offset = last_offset;
ctx->blocks[i]->index = i; // Need to store for sorting
ctx->blocks[i]->length = length;
ctx->blocks[i]->similar = 0;
ctx->blocks[i]->hash = XXH32(buf1+last_offset, length, 0);
ctx->blocks[i]->similarity_hash = ctx->blocks[i]->hash;
last_offset += length;
}
goto process_blocks;
}
if (rabin_pos == NULL) {
/*
* Initialize arrays for sketch computation. We re-use memory allocated
* for the compressed chunk temporarily.
*/
ary_sz = ctx->rabin_poly_max_block_size;
ctx_heap = (uint32_t *)(ctx->cbuf + ctx->real_chunksize - ary_sz);
}
memset(ctx->current_window_data, 0, RAB_POLYNOMIAL_WIN_SIZE);
/*
* If rabin_pos is non-zero then we are being asked to scan for the last rabin boundary
* in the chunk. We start scanning at chunk end - max rabin block size. We avoid doing
* a full chunk scan.
*
* !!!NOTE!!!: Code duplication below for performance.
*/
if (rabin_pos) {
offset = *size - ctx->rabin_poly_max_block_size;
length = 0;
for (i=offset; i<*size; i++) {
uchar_t cur_byte = buf1[i];
uint64_t pushed_out = ctx->current_window_data[ctx->window_pos];
ctx->current_window_data[ctx->window_pos] = cur_byte;
cur_roll_checksum = (cur_roll_checksum * RAB_POLYNOMIAL_CONST) & POLY_MASK;
cur_roll_checksum += cur_byte;
cur_roll_checksum -= out[pushed_out];
cur_pos_checksum = cur_roll_checksum ^ ir[pushed_out];
ctx->window_pos = (ctx->window_pos + 1) & (RAB_POLYNOMIAL_WIN_SIZE-1);
length++;
if (length < ctx->rabin_poly_min_block_size) continue;
// If we hit our special value update block offset
if ((cur_pos_checksum & ctx->rabin_avg_block_mask) == ctx->rabin_break_patt) {
last_offset = i;
length = 0;
}
}
if (last_offset < *size) {
*rabin_pos = last_offset;
}
return (0);
}
j = 0;
for (i=offset; i<*size; i++) {
uint64_t pc[4];
uchar_t cur_byte = buf1[i];
uint64_t pushed_out = ctx->current_window_data[ctx->window_pos];
ctx->current_window_data[ctx->window_pos] = cur_byte;
cur_roll_checksum = (cur_roll_checksum * RAB_POLYNOMIAL_CONST) & POLY_MASK;
cur_roll_checksum += cur_byte;
cur_roll_checksum -= out[pushed_out];
cur_pos_checksum = cur_roll_checksum ^ ir[pushed_out];
/*
* Window pos has to rotate from 0 .. RAB_POLYNOMIAL_WIN_SIZE-1
* We avoid a branch here by masking. This requires RAB_POLYNOMIAL_WIN_SIZE
* to be power of 2
*/
ctx->window_pos = (ctx->window_pos + 1) & (RAB_POLYNOMIAL_WIN_SIZE-1);
length++;
if (length < ctx->rabin_poly_min_block_size) continue;
// If we hit our special value or reached the max block size update block offset
if ((cur_pos_checksum & ctx->rabin_avg_block_mask) == ctx->rabin_break_patt ||
length >= ctx->rabin_poly_max_block_size) {
if (ctx->blocks[blknum] == 0)
ctx->blocks[blknum] = (rabin_blockentry_t *)slab_alloc(NULL,
sizeof (rabin_blockentry_t));
ctx->blocks[blknum]->offset = last_offset;
ctx->blocks[blknum]->index = blknum; // Need to store for sorting
ctx->blocks[blknum]->length = length;
DEBUG_STAT_EN(if (length >= ctx->rabin_poly_max_block_size) max_count++);
/*
* Reset the heap structure and find the K min values if Delta Compression
* is enabled. We use a min heap mechanism taken from the heap based priority
* queue implementation in Python.
* Here K = similarity extent = 87% or 62% or 50%.
*
* Once block contents are arranged in a min heap we compute the K min values
* sketch by hashing over the heap till K%. We interpret the raw bytes as a
* sequence of 64-bit integers.
* This is called minhashing and is used widely, for example in various
* search engines to detect similar documents.
*/
if (ctx->delta_flag) {
memcpy(ctx_heap, buf1+last_offset, length);
length /= 8;
pc[1] = DELTA_NORMAL_PCT(length);
pc[2] = DELTA_EXTRA_PCT(length);
pc[3] = DELTA_EXTRA2_PCT(length);
reset_heap(&heap, pc[ctx->delta_flag]);
ksmallest((int64_t *)ctx_heap, length, &heap);
ctx->blocks[blknum]->similarity_hash =
XXH32((const uchar_t *)ctx_heap, pc[ctx->delta_flag]*8, 0);
}
blknum++;
last_offset = i+1;
length = 0;
j = 0;
}
}
// Insert the last left-over trailing bytes, if any, into a block.
if (last_offset < *size) {
if (ctx->blocks[blknum] == 0)
ctx->blocks[blknum] = (rabin_blockentry_t *)slab_alloc(NULL,
sizeof (rabin_blockentry_t));
ctx->blocks[blknum]->offset = last_offset;
ctx->blocks[blknum]->index = blknum;
length = *size - last_offset;
ctx->blocks[blknum]->length = length;
if (ctx->delta_flag) {
uint64_t cur_sketch;
uint64_t pc[3];
if (length > ctx->rabin_poly_min_block_size) {
memcpy(ctx_heap, buf1+last_offset, length);
length /= 8;
pc[1] = DELTA_NORMAL_PCT(length);
pc[2] = DELTA_EXTRA_PCT(length);
pc[3] = DELTA_EXTRA2_PCT(length);
reset_heap(&heap, pc[ctx->delta_flag]);
ksmallest((int64_t *)ctx_heap, length, &heap);
cur_sketch =
XXH32((const uchar_t *)ctx_heap, pc[ctx->delta_flag]*8, 0);
ctx->blocks[blknum]->similarity_hash = cur_sketch;
} else {
cur_sketch =
XXH32((const uchar_t *)(buf1+last_offset), length, 0);
ctx->blocks[blknum]->similarity_hash = cur_sketch;
}
}
blknum++;
last_offset = *size;
}
process_blocks:
// If we found at least a few chunks, perform dedup.
DEBUG_STAT_EN(en_1 = get_wtime_millis());
DEBUG_STAT_EN(fprintf(stderr, "Original size: %" PRId64 ", blknum: %u\n", *size, blknum));
DEBUG_STAT_EN(fprintf(stderr, "Number of maxlen blocks: %u\n", max_count));
if (blknum > 2) {
uint64_t pos, matchlen, pos1;
int valid = 1;
uint32_t *dedupe_index;
uint64_t dedupe_index_sz;
rabin_blockentry_t *be;
DEBUG_STAT_EN(uint32_t delta_calls, delta_fails, merge_count, hash_collisions);
DEBUG_STAT_EN(delta_calls = 0);
DEBUG_STAT_EN(delta_fails = 0);
DEBUG_STAT_EN(hash_collisions = 0);
ary_sz = blknum * sizeof (rabin_blockentry_t *);
htab = (rabin_blockentry_t **)(ctx->cbuf + ctx->real_chunksize - ary_sz);
memset(htab, 0, ary_sz);
/*
* Compute hash signature for each block. We do this in a separate loop to
* have a fast linear scan through the buffer.
*/
if (ctx->delta_flag) {
for (i=0; i<blknum; i++) {
ctx->blocks[i]->hash = XXH32(buf1+ctx->blocks[i]->offset,
ctx->blocks[i]->length, 0);
}
} else {
for (i=0; i<blknum; i++) {
ctx->blocks[i]->hash = XXH32(buf1+ctx->blocks[i]->offset,
ctx->blocks[i]->length, 0);
ctx->blocks[i]->similarity_hash = ctx->blocks[i]->hash;
}
}
/*
* Perform hash-matching of blocks and use a bucket-chained hashtable to match
* for duplicates and similar blocks. Unique blocks are inserted and duplicates
* and similar ones are marked in the block array.
*
* Hashtable memory is not allocated. We just use available space in the
* target buffer.
*/
matchlen = 0;
for (i=0; i<blknum; i++) {
uint64_t ck;
/*
* Bias hash with length for fewer collisions. If Delta Compression is
* not enabled then value of similarity_hash == hash.
*/
ck = ctx->blocks[i]->similarity_hash;
ck += (ck / ctx->blocks[i]->length);
j = ck % blknum;
if (htab[j] == 0) {
/*
* Hash bucket empty. So add block into table.
*/
htab[j] = ctx->blocks[i];
ctx->blocks[i]->other = 0;
ctx->blocks[i]->next = 0;
ctx->blocks[i]->similar = 0;
} else {
be = htab[j];
length = 0;
/*
* Look for exact duplicates. Same cksum, length and memcmp()\
*/
while (1) {
if (be->hash == ctx->blocks[i]->hash &&
be->length == ctx->blocks[i]->length &&
memcmp(buf1 + be->offset, buf1 + ctx->blocks[i]->offset,
be->length) == 0) {
ctx->blocks[i]->similar = SIMILAR_EXACT;
ctx->blocks[i]->other = be;
be->similar = SIMILAR_REF;
matchlen += be->length;
length = 1;
break;
}
if (be->next)
be = be->next;
else
break;
}
if (ctx->delta_flag && !length) {
/*
* Look for similar blocks.
*/
be = htab[j];
while (1) {
if (be->similarity_hash == ctx->blocks[i]->similarity_hash &&
be->length == ctx->blocks[i]->length) {
ctx->blocks[i]->similar = SIMILAR_PARTIAL;
ctx->blocks[i]->other = be;
be->similar = SIMILAR_REF;
matchlen += (be->length>>1);
length = 1;
break;
}
if (be->next)
be = be->next;
else
break;
}
}
/*
* No duplicate in table for this block. So add it to
* the bucket chain.
*/
if (!length) {
ctx->blocks[i]->other = 0;
ctx->blocks[i]->next = 0;
ctx->blocks[i]->similar = 0;
be->next = ctx->blocks[i];
DEBUG_STAT_EN(hash_collisions++);
}
}
}
DEBUG_STAT_EN(fprintf(stderr, "Total Hashtable bucket collisions: %u\n", hash_collisions));
dedupe_index_sz = (uint64_t)blknum * RABIN_ENTRY_SIZE;
if (matchlen < dedupe_index_sz) {
DEBUG_STAT_EN(en = get_wtime_millis());
DEBUG_STAT_EN(fprintf(stderr, "Chunking speed %.3f MB/s, Overall Dedupe speed %.3f MB/s\n",
get_mb_s(*size, strt, en_1), get_mb_s(*size, strt, en)));
DEBUG_STAT_EN(fprintf(stderr, "No Dedupe possible.\n"));
ctx->valid = 0;
return (0);
}
dedupe_index = (uint32_t *)(ctx->cbuf + RABIN_HDR_SIZE);
pos = 0;
DEBUG_STAT_EN(merge_count = 0);
/*
* Merge runs of unique blocks into a single block entry to reduce
* dedupe index size.
*/
for (i=0; i<blknum;) {
dedupe_index[pos] = i;
ctx->blocks[i]->index = pos;
pos++;
length = 0;
j = i;
if (ctx->blocks[i]->similar == 0) {
while (i< blknum && ctx->blocks[i]->similar == 0 &&
length < RABIN_MAX_BLOCK_SIZE) {
length += ctx->blocks[i]->length;
i++;
DEBUG_STAT_EN(merge_count++);
}
ctx->blocks[j]->length = length;
} else {
i++;
}
}
DEBUG_STAT_EN(fprintf(stderr, "Merge count: %u\n", merge_count));
/*
* Final pass update dedupe index and copy data.
*/
blknum = pos;
dedupe_index_sz = (uint64_t)blknum * RABIN_ENTRY_SIZE;
pos1 = dedupe_index_sz + RABIN_HDR_SIZE;
matchlen = ctx->real_chunksize - *size;
for (i=0; i<blknum; i++) {
be = ctx->blocks[dedupe_index[i]];
if (be->similar == 0 || be->similar == SIMILAR_REF) {
/* Just copy. */
dedupe_index[i] = htonl(be->length);
memcpy(ctx->cbuf + pos1, buf1 + be->offset, be->length);
pos1 += be->length;
} else {
if (be->similar == SIMILAR_EXACT) {
dedupe_index[i] = htonl((be->other->index | RABIN_INDEX_FLAG) &
CLEAR_SIMILARITY_FLAG);
} else {
uchar_t *oldbuf, *newbuf;
int32_t bsz;
/*
* Perform bsdiff.
*/
oldbuf = buf1 + be->other->offset;
newbuf = buf1 + be->offset;
DEBUG_STAT_EN(delta_calls++);
bsz = bsdiff(oldbuf, be->other->length, newbuf, be->length,
ctx->cbuf + pos1, buf1 + *size, matchlen);
if (bsz == 0) {
DEBUG_STAT_EN(delta_fails++);
memcpy(ctx->cbuf + pos1, newbuf, be->length);
dedupe_index[i] = htonl(be->length);
pos1 += be->length;
} else {
dedupe_index[i] = htonl(be->other->index |
RABIN_INDEX_FLAG | SET_SIMILARITY_FLAG);
pos1 += bsz;
}
}
}
}
if (valid) {
uchar_t *cbuf = ctx->cbuf;
uint64_t *entries;
DEBUG_STAT_EN(uint64_t sz);
DEBUG_STAT_EN(sz = *size);
*((uint32_t *)cbuf) = htonl(blknum);
cbuf += sizeof (uint32_t);
entries = (uint64_t *)cbuf;
entries[0] = htonll(*size);
entries[1] = 0;
entries[2] = htonll(pos1 - dedupe_index_sz - RABIN_HDR_SIZE);
*size = pos1;
ctx->valid = 1;
DEBUG_STAT_EN(en = get_wtime_millis());
DEBUG_STAT_EN(fprintf(stderr, "Deduped size: %" PRId64 ", blknum: %u, delta_calls: %u, delta_fails: %u\n",
*size, blknum, delta_calls, delta_fails));
DEBUG_STAT_EN(fprintf(stderr, "Chunking speed %.3f MB/s, Overall Dedupe speed %.3f MB/s\n",
get_mb_s(sz, strt, en_1), get_mb_s(sz, strt, en)));
/*
* Remaining header entries: size of compressed index and size of
* compressed data are inserted later via rabin_update_hdr, after actual compression!
*/
return (dedupe_index_sz);
}
}
return (0);
}
void
update_dedupe_hdr(uchar_t *buf, uint64_t dedupe_index_sz_cmp, uint64_t dedupe_data_sz_cmp)
{
uint64_t *entries;
buf += sizeof (uint32_t);
entries = (uint64_t *)buf;
entries[1] = htonll(dedupe_index_sz_cmp);
entries[3] = htonll(dedupe_data_sz_cmp);
}
void
parse_dedupe_hdr(uchar_t *buf, uint32_t *blknum, uint64_t *dedupe_index_sz,
uint64_t *dedupe_data_sz, uint64_t *dedupe_index_sz_cmp,
uint64_t *dedupe_data_sz_cmp, uint64_t *deduped_size)
{
uint64_t *entries;
*blknum = ntohl(*((uint32_t *)(buf)));
buf += sizeof (uint32_t);
entries = (uint64_t *)buf;
*dedupe_data_sz = ntohll(entries[0]);
*dedupe_index_sz = (uint64_t)(*blknum) * RABIN_ENTRY_SIZE;
*dedupe_index_sz_cmp = ntohll(entries[1]);
*deduped_size = ntohll(entries[2]);
*dedupe_data_sz_cmp = ntohll(entries[3]);
}
void
dedupe_decompress(dedupe_context_t *ctx, uchar_t *buf, uint64_t *size)
{
uint32_t blknum, blk, oblk, len;
uint32_t *dedupe_index;
uint64_t data_sz, sz, indx_cmp, data_sz_cmp, deduped_sz;
uint64_t dedupe_index_sz, pos1;
uchar_t *pos2;
parse_dedupe_hdr(buf, &blknum, &dedupe_index_sz, &data_sz, &indx_cmp, &data_sz_cmp, &deduped_sz);
dedupe_index = (uint32_t *)(buf + RABIN_HDR_SIZE);
pos1 = dedupe_index_sz + RABIN_HDR_SIZE;
pos2 = ctx->cbuf;
sz = 0;
ctx->valid = 1;
slab_cache_add(sizeof (rabin_blockentry_t));
for (blk = 0; blk < blknum; blk++) {
if (ctx->blocks[blk] == 0)
ctx->blocks[blk] = (rabin_blockentry_t *)slab_alloc(NULL, sizeof (rabin_blockentry_t));
len = ntohl(dedupe_index[blk]);
ctx->blocks[blk]->hash = 0;
if (len == 0) {
ctx->blocks[blk]->hash = 1;
} else if (!(len & RABIN_INDEX_FLAG)) {
ctx->blocks[blk]->length = len;
ctx->blocks[blk]->offset = pos1;
pos1 += len;
} else {
bsize_t blen;
ctx->blocks[blk]->length = 0;
if (len & GET_SIMILARITY_FLAG) {
ctx->blocks[blk]->offset = pos1;
ctx->blocks[blk]->index = (len & RABIN_INDEX_VALUE) | SET_SIMILARITY_FLAG;
blen = get_bsdiff_sz(buf + pos1);
pos1 += blen;
} else {
ctx->blocks[blk]->index = len & RABIN_INDEX_VALUE;
}
}
}
for (blk = 0; blk < blknum; blk++) {
int rv;
bsize_t newsz;
if (ctx->blocks[blk]->hash == 1) continue;
if (ctx->blocks[blk]->length > 0) {
len = ctx->blocks[blk]->length;
pos1 = ctx->blocks[blk]->offset;
} else {
oblk = ctx->blocks[blk]->index;
if (oblk & GET_SIMILARITY_FLAG) {
oblk = oblk & CLEAR_SIMILARITY_FLAG;
len = ctx->blocks[oblk]->length;
pos1 = ctx->blocks[oblk]->offset;
newsz = data_sz - sz;
rv = bspatch(buf + ctx->blocks[blk]->offset, buf + pos1, len, pos2, &newsz);
if (rv == 0) {
fprintf(stderr, "Failed to bspatch block.\n");
ctx->valid = 0;
break;
}
pos2 += newsz;
sz += newsz;
if (sz > data_sz) {
fprintf(stderr, "Dedup data overflows chunk.\n");
ctx->valid = 0;
break;
}
continue;
} else {
len = ctx->blocks[oblk]->length;
pos1 = ctx->blocks[oblk]->offset;
}
}
memcpy(pos2, buf + pos1, len);
pos2 += len;
sz += len;
if (sz > data_sz) {
fprintf(stderr, "Dedup data overflows chunk.\n");
ctx->valid = 0;
break;
}
}
if (ctx->valid && sz < data_sz) {
fprintf(stderr, "Too little dedup data processed.\n");
ctx->valid = 0;
}
*size = data_sz;
}
/*
* TODO: Consolidate rabin dedup and compression/decompression in functions here rather than
* messy code in main program.
int
rabin_compress(dedupe_context_t *ctx, uchar_t *from, uint64_t fromlen, uchar_t *to, uint64_t *tolen,
int level, char chdr, void *data, compress_func_ptr cmp)
{
}
*/
Reference in a new issue View git blame Copy permalink