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

set up new calculations compression completion progress

Browse Source
This commit is contained in:
Paul Cruz
2017-07-19 11:47:17 -07:00
parent f1ac518b59
commit 5a85c57e30
1 changed files with 10 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/adaptive-compression/adapt.c
+10 -7
View File
@@ -355,7 +355,7 @@ static void adaptCompressionLevel(adaptCCtx* ctx)
double completion; double completion;
pthread_mutex_lock(&ctx->completion_mutex.pMutex); pthread_mutex_lock(&ctx->completion_mutex.pMutex);
completion = writeSlow ? ctx->writeCompletion : ctx->createCompletion; completion = writeSlow ? ctx->writeCompletion : ctx->createCompletion;
DEBUG(2, "write completion: %f, create completion: %f\n", ctx->writeCompletion, ctx->createCompletion); DEBUG(3, "write completion: %f, create completion: %f\n", ctx->writeCompletion, ctx->createCompletion);
pthread_mutex_unlock(&ctx->completion_mutex.pMutex); pthread_mutex_unlock(&ctx->completion_mutex.pMutex);
{ {
unsigned const maxChange = (unsigned)((1-completion) * (MAX_COMPRESSION_LEVEL_CHANGE - 1)) + 1; unsigned const maxChange = (unsigned)((1-completion) * (MAX_COMPRESSION_LEVEL_CHANGE - 1)) + 1;
@@ -375,7 +375,7 @@ static void adaptCompressionLevel(adaptCCtx* ctx)
unsigned const maxChange = (unsigned)((1-completion) * (MAX_COMPRESSION_LEVEL_CHANGE-1)) + 1; unsigned const maxChange = (unsigned)((1-completion) * (MAX_COMPRESSION_LEVEL_CHANGE-1)) + 1;
unsigned const change = MIN(maxChange, ctx->compressionLevel - 1); unsigned const change = MIN(maxChange, ctx->compressionLevel - 1);
DEBUG(3, "decreasing compression level %u\n", ctx->compressionLevel); DEBUG(3, "decreasing compression level %u\n", ctx->compressionLevel);
DEBUG(3, "completion: %f\n", completion); DEBUG(2, "completion: %f\n", completion);
ctx->compressionLevel -= change; ctx->compressionLevel -= change;
reset = 1; reset = 1;
} }
@@ -493,6 +493,13 @@ static void* compressionThread(void* arg)
dstPos += ret; dstPos += ret;
dictPos += actualBlockSize; dictPos += actualBlockSize;
blockNum++; blockNum++;
/* update completion */
pthread_mutex_lock(&ctx->completion_mutex.pMutex);
if (!ctx->compressionCompletionMeasured) {
ctx->compressionCompletion = 1 - (double)remaining/job->src.size;
}
pthread_mutex_unlock(&ctx->completion_mutex.pMutex);
} }
} }
job->dst.size = job->compressedSize; job->dst.size = job->compressedSize;
@@ -549,11 +556,7 @@ static void* outputThread(void* arg)
pthread_mutex_unlock(&ctx->stats_mutex.pMutex); pthread_mutex_unlock(&ctx->stats_mutex.pMutex);
reduceCounters(ctx); reduceCounters(ctx);
pthread_mutex_lock(&ctx->completion_mutex.pMutex); pthread_mutex_lock(&ctx->completion_mutex.pMutex);
if (!ctx->compressionCompletionMeasured) { ctx->compressionCompletionMeasured = 1;
ctx->compressionCompletion = ZSTD_getCompletion(ctx->cctx);
ctx->compressionCompletionMeasured = 1;
DEBUG(3, "output detected completion: %f\n", ctx->compressionCompletion);
}
pthread_mutex_unlock(&ctx->completion_mutex.pMutex); pthread_mutex_unlock(&ctx->completion_mutex.pMutex);
DEBUG(3, "waiting on job compressed, nextJob: %u\n", currJob); DEBUG(3, "waiting on job compressed, nextJob: %u\n", currJob);
pthread_cond_wait(&ctx->jobCompressed_cond.pCond, &ctx->jobCompressed_mutex.pMutex); pthread_cond_wait(&ctx->jobCompressed_cond.pCond, &ctx->jobCompressed_mutex.pMutex);