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

Introduce ZSTD_getFrameProgression()

Browse Source 
Produces 3 statistics for ongoing frame compression :
- ingested
- consumed (effectively compressed)
- produced

Ingested can be larger than consumed due to buffering effect.

For the time being, this patch mostly fixes the % ratio issue,
since it computes consumed / produced,
instead of ingested / produced.

That being said, update is not "smooth",
because on a slow enough setting,
fileio spends most of its time waiting for a worker to complete its job.

This could be improved thanks to more granular flushing
i.e. start flushing before ongoing job is fully completed.
This commit is contained in:
Yann Collet
2018-01-17 16:39:02 -08:00
parent 592ce5a042
commit 394eec697b
7 changed files with 208 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+49 -15
View File
@@ -308,7 +308,7 @@ typedef struct {
const void* srcStart;
size_t prefixSize;
size_t srcSize;
size_t readSize;
size_t consumed;
buffer_t dstBuff;
size_t cSize;
size_t dstFlushed;
@@ -382,8 +382,7 @@ void ZSTDMT_compressChunk(void* jobDescription)
ZSTD_compressEnd (cctx, dstBuff.start, dstBuff.size, src, job->srcSize) :
ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, job->srcSize);
#else
if (sizeof(size_t) > sizeof(int))
assert(job->srcSize < ((size_t)INT_MAX) * ZSTD_BLOCKSIZE_MAX); /* check overflow */
if (sizeof(size_t) > sizeof(int)) assert(job->srcSize < ((size_t)INT_MAX) * ZSTD_BLOCKSIZE_MAX); /* check overflow */
{ int const nbBlocks = (int)((job->srcSize + (ZSTD_BLOCKSIZE_MAX-1)) / ZSTD_BLOCKSIZE_MAX);
const BYTE* ip = (const BYTE*) src;
BYTE* const ostart = (BYTE*)dstBuff.start;
@@ -399,7 +398,7 @@ void ZSTDMT_compressChunk(void* jobDescription)
op += cSize; assert(op < oend);
/* stats */
job->cSize += cSize;
job->readSize = ZSTD_BLOCKSIZE_MAX * blockNb;
job->consumed = ZSTD_BLOCKSIZE_MAX * blockNb;
}
/* last block */
if ((nbBlocks > 0) | job->lastChunk /*need to output a "last block" flag*/ ) {
@@ -412,7 +411,7 @@ void ZSTDMT_compressChunk(void* jobDescription)
/* stats */
job->cSize += cSize;
}
job->readSize = job->srcSize;
job->consumed = job->srcSize;
}
#endif
@@ -460,6 +459,8 @@ struct ZSTDMT_CCtx_s {
unsigned frameEnded;
unsigned allJobsCompleted;
unsigned long long frameContentSize;
unsigned long long consumed;
unsigned long long produced;
ZSTD_customMem cMem;
ZSTD_CDict* cdictLocal;
const ZSTD_CDict* cdict;
@@ -501,15 +502,6 @@ size_t ZSTDMT_CCtxParam_setNbThreads(ZSTD_CCtx_params* params, unsigned nbThread
return nbThreads;
}
/* ZSTDMT_getNbThreads():
* @return nb threads currently active in mtctx.
* mtctx must be valid */
unsigned ZSTDMT_getNbThreads(const ZSTDMT_CCtx* mtctx)
{
assert(mtctx != NULL);
return mtctx->params.nbThreads;
}
ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem)
{
ZSTDMT_CCtx* mtctx;
@@ -553,6 +545,7 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx(unsigned nbThreads)
return ZSTDMT_createCCtx_advanced(nbThreads, ZSTD_defaultCMem);
}
/* ZSTDMT_releaseAllJobResources() :
* note : ensure all workers are killed first ! */
static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx)
@@ -652,6 +645,43 @@ size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSTDMT_parameter parameter,
}
}
/* ZSTDMT_getNbThreads():
* @return nb threads currently active in mtctx.
* mtctx must be valid */
unsigned ZSTDMT_getNbThreads(const ZSTDMT_CCtx* mtctx)
{
assert(mtctx != NULL);
return mtctx->params.nbThreads;
}
/* ZSTDMT_getFrameProgression():
* tells how much data has been consumed (input) and produced (output) for current frame.
* able to count progression inside worker threads.
* Note : mutex will be triggered during statistics collection. */
ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx)
{
ZSTD_frameProgression fs;
DEBUGLOG(5, "ZSTDMT_getFrameProgression");
ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex);
fs.consumed = mtctx->consumed;
fs.produced = mtctx->produced;
assert(mtctx->inBuff.filled >= mtctx->prefixSize);
fs.ingested = mtctx->consumed + (mtctx->inBuff.filled - mtctx->prefixSize);
{ unsigned jobNb;
for (jobNb = mtctx->doneJobID ; jobNb < mtctx->nextJobID ; jobNb++) {
unsigned const wJobID = jobNb & mtctx->jobIDMask;
size_t const cResult = mtctx->jobs[wJobID].cSize;
size_t const produced = ZSTD_isError(cResult) ? 0 : cResult;
fs.consumed += mtctx->jobs[wJobID].consumed;
fs.ingested += mtctx->jobs[wJobID].srcSize;
fs.produced += produced;
}
}
ZSTD_pthread_mutex_unlock(&mtctx->jobCompleted_mutex);
return fs;
}
/* ------------------------------------------ */
/* ===== Multi-threaded compression ===== */
/* ------------------------------------------ */
@@ -900,6 +930,8 @@ size_t ZSTDMT_initCStream_internal(
zcs->nextJobID = 0;
zcs->frameEnded = 0;
zcs->allJobsCompleted = 0;
zcs->consumed = 0;
zcs->produced = 0;
if (params.fParams.checksumFlag) XXH64_reset(&zcs->xxhState, 0);
return 0;
}
@@ -963,7 +995,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
zcs->jobs[jobID].src = zcs->inBuff.buffer;
zcs->jobs[jobID].srcStart = zcs->inBuff.buffer.start;
zcs->jobs[jobID].srcSize = srcSize;
zcs->jobs[jobID].readSize = 0;
zcs->jobs[jobID].consumed = 0;
zcs->jobs[jobID].prefixSize = zcs->prefixSize;
assert(zcs->inBuff.filled >= srcSize + zcs->prefixSize);
zcs->jobs[jobID].params = zcs->params;
@@ -1070,6 +1102,8 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
zcs->jobs[wJobID].dstBuff = g_nullBuffer;
zcs->jobs[wJobID].jobCompleted = 0;
zcs->doneJobID++;
zcs->consumed += job.srcSize;
zcs->produced += job.cSize;
} else {
zcs->jobs[wJobID].dstFlushed = job.dstFlushed;
}