ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

split compression into smaller blocks

Browse Source
This commit is contained in:
Paul Cruz
2017-07-19 11:23:40 -07:00
parent 338951cd48
commit f1ac518b59
1 changed files with 53 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/adaptive-compression/adapt.c
+30 -8
View File
@@ -437,15 +437,27 @@ static void* compressionThread(void* arg)
/* compress the data */ /* compress the data */
{ {
size_t const compressionBlockSize = 4 << 17; /* 128 KB */
unsigned const cLevel = ctx->compressionLevel; unsigned const cLevel = ctx->compressionLevel;
unsigned blockNum = 0;
size_t remaining = job->src.size;
size_t srcPos = 0;
size_t dstPos = 0;
size_t dictPos = 0;
DEBUG(3, "cLevel used: %u\n", cLevel); DEBUG(3, "cLevel used: %u\n", cLevel);
DEBUG(3, "compression level used: %u\n", cLevel); DEBUG(3, "compression level used: %u\n", cLevel);
/* reset compressed size */
job->compressedSize = 0;
while (remaining != 0) {
size_t const actualBlockSize = MIN(remaining, compressionBlockSize);
/* begin compression */ /* begin compression */
{ {
size_t const useDictSize = MIN(getUseableDictSize(cLevel), job->dictSize); size_t const useDictSize = MIN(getUseableDictSize(cLevel), job->dictSize);
DEBUG(3, "useDictSize: %zu, job->dictSize: %zu\n", useDictSize, job->dictSize); DEBUG(3, "useDictSize: %zu, job->dictSize: %zu\n", useDictSize, job->dictSize);
size_t const dictModeError = ZSTD_setCCtxParameter(ctx->cctx, ZSTD_p_forceRawDict, 1); size_t const dictModeError = ZSTD_setCCtxParameter(ctx->cctx, ZSTD_p_forceRawDict, 1);
size_t const initError = ZSTD_compressBegin_usingDict(ctx->cctx, job->src.start + job->dictSize - useDictSize, useDictSize, cLevel); size_t const initError = ZSTD_compressBegin_usingDict(ctx->cctx, job->src.start + job->dictSize - useDictSize + dictPos, useDictSize, cLevel);
size_t const windowSizeError = ZSTD_setCCtxParameter(ctx->cctx, ZSTD_p_forceWindow, 1); size_t const windowSizeError = ZSTD_setCCtxParameter(ctx->cctx, ZSTD_p_forceWindow, 1);
if (ZSTD_isError(dictModeError) || ZSTD_isError(initError) || ZSTD_isError(windowSizeError)) { if (ZSTD_isError(dictModeError) || ZSTD_isError(initError) || ZSTD_isError(windowSizeError)) {
DISPLAY("Error: something went wrong while starting compression\n"); DISPLAY("Error: something went wrong while starting compression\n");
@@ -455,8 +467,8 @@ static void* compressionThread(void* arg)
} }
/* continue compression */ /* continue compression */
if (currJob != 0) { /* not first job flush/overwrite the frame header */ if (currJob != 0 || blockNum != 0) { /* not first block of first job flush/overwrite the frame header */
size_t const hSize = ZSTD_compressContinue(ctx->cctx, job->dst.start, job->dst.capacity, job->src.start + job->dictSize, 0); size_t const hSize = ZSTD_compressContinue(ctx->cctx, job->dst.start + dstPos, job->dst.capacity - dstPos, job->src.start + job->dictSize + srcPos, 0);
if (ZSTD_isError(hSize)) { if (ZSTD_isError(hSize)) {
DISPLAY("Error: something went wrong while continuing compression\n"); DISPLAY("Error: something went wrong while continuing compression\n");
job->compressedSize = hSize; job->compressedSize = hSize;
@@ -465,14 +477,24 @@ static void* compressionThread(void* arg)
} }
ZSTD_invalidateRepCodes(ctx->cctx); ZSTD_invalidateRepCodes(ctx->cctx);
} }
job->compressedSize = (job->lastJob) ? {
ZSTD_compressEnd (ctx->cctx, job->dst.start, job->dst.capacity, job->src.start + job->dictSize, job->src.size) :
ZSTD_compressContinue(ctx->cctx, job->dst.start, job->dst.capacity, job->src.start + job->dictSize, job->src.size); size_t const ret = (job->lastJob && remaining <= compressionBlockSize) ?
if (ZSTD_isError(job->compressedSize)) { ZSTD_compressEnd (ctx->cctx, job->dst.start + dstPos, job->dst.capacity - dstPos, job->src.start + job->dictSize + srcPos, actualBlockSize) :
DISPLAY("Error: something went wrong during compression: %s\n", ZSTD_getErrorName(job->compressedSize)); ZSTD_compressContinue(ctx->cctx, job->dst.start + dstPos, job->dst.capacity - dstPos, job->src.start + job->dictSize + srcPos, actualBlockSize);
if (ZSTD_isError(ret)) {
DISPLAY("Error: something went wrong during compression: %s\n", ZSTD_getErrorName(ret));
signalErrorToThreads(ctx); signalErrorToThreads(ctx);
return arg; return arg;
} }
job->compressedSize += ret;
remaining -= actualBlockSize;
srcPos += actualBlockSize;
dstPos += ret;
dictPos += actualBlockSize;
blockNum++;
}
}
job->dst.size = job->compressedSize; job->dst.size = job->compressedSize;
} }
pthread_mutex_lock(&ctx->jobCompressed_mutex.pMutex); pthread_mutex_lock(&ctx->jobCompressed_mutex.pMutex);