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

Merge branch 'dev' into shorterTests

Browse Source 
fixed conflicts
This commit is contained in:
Yann Collet
2017-09-28 12:19:28 -07:00
parent bbef058ae6 824f75ea7c
commit e4ec427720
25 changed files with 933 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstd_compress.c
+30 -16
View File
@@ -256,6 +256,9 @@ size_t ZSTD_CCtx_setParameter(ZSTD_CCtx* cctx, ZSTD_cParameter param, unsigned v
switch(param)
{
case ZSTD_p_format :
return ZSTD_CCtxParam_setParameter(&cctx->requestedParams, param, value);
case ZSTD_p_compressionLevel:
if (value == 0) return 0; /* special value : 0 means "don't change anything" */
if (cctx->cdict) return ERROR(stage_wrong);
@@ -326,6 +329,12 @@ size_t ZSTD_CCtxParam_setParameter(
{
switch(param)
{
case ZSTD_p_format :
if (value > (unsigned)ZSTD_f_zstd1_magicless)
return ERROR(parameter_unsupported);
params->format = (ZSTD_format_e)value;
return 0;
case ZSTD_p_compressionLevel :
if ((int)value > ZSTD_maxCLevel()) value = ZSTD_maxCLevel();
if (value == 0) return 0;
@@ -669,7 +678,7 @@ ZSTD_compressionParameters ZSTD_adjustCParams(ZSTD_compressionParameters cPar, u
return ZSTD_adjustCParams_internal(cPar, srcSize, dictSize);
}
size_t ZSTD_estimateCCtxSize_advanced_usingCCtxParams(const ZSTD_CCtx_params* params)
size_t ZSTD_estimateCCtxSize_usingCCtxParams(const ZSTD_CCtx_params* params)
{
/* Estimate CCtx size is supported for single-threaded compression only. */
if (params->nbThreads > 1) { return ERROR(GENERIC); }
@@ -706,22 +715,22 @@ size_t ZSTD_estimateCCtxSize_advanced_usingCCtxParams(const ZSTD_CCtx_params* pa
}
}
size_t ZSTD_estimateCCtxSize_advanced_usingCParams(ZSTD_compressionParameters cParams)
size_t ZSTD_estimateCCtxSize_usingCParams(ZSTD_compressionParameters cParams)
{
ZSTD_CCtx_params const params = ZSTD_makeCCtxParamsFromCParams(cParams);
return ZSTD_estimateCCtxSize_advanced_usingCCtxParams(&params);
return ZSTD_estimateCCtxSize_usingCCtxParams(&params);
}
size_t ZSTD_estimateCCtxSize(int compressionLevel)
{
ZSTD_compressionParameters const cParams = ZSTD_getCParams(compressionLevel, 0, 0);
return ZSTD_estimateCCtxSize_advanced_usingCParams(cParams);
return ZSTD_estimateCCtxSize_usingCParams(cParams);
}
size_t ZSTD_estimateCStreamSize_advanced_usingCCtxParams(const ZSTD_CCtx_params* params)
size_t ZSTD_estimateCStreamSize_usingCCtxParams(const ZSTD_CCtx_params* params)
{
if (params->nbThreads > 1) { return ERROR(GENERIC); }
{ size_t const CCtxSize = ZSTD_estimateCCtxSize_advanced_usingCCtxParams(params);
{ size_t const CCtxSize = ZSTD_estimateCCtxSize_usingCCtxParams(params);
size_t const blockSize = MIN(ZSTD_BLOCKSIZE_MAX, (size_t)1 << params->cParams.windowLog);
size_t const inBuffSize = ((size_t)1 << params->cParams.windowLog) + blockSize;
size_t const outBuffSize = ZSTD_compressBound(blockSize) + 1;
@@ -731,15 +740,15 @@ size_t ZSTD_estimateCStreamSize_advanced_usingCCtxParams(const ZSTD_CCtx_params*
}
}
size_t ZSTD_estimateCStreamSize_advanced_usingCParams(ZSTD_compressionParameters cParams)
size_t ZSTD_estimateCStreamSize_usingCParams(ZSTD_compressionParameters cParams)
{
ZSTD_CCtx_params const params = ZSTD_makeCCtxParamsFromCParams(cParams);
return ZSTD_estimateCStreamSize_advanced_usingCCtxParams(&params);
return ZSTD_estimateCStreamSize_usingCCtxParams(&params);
}
size_t ZSTD_estimateCStreamSize(int compressionLevel) {
ZSTD_compressionParameters const cParams = ZSTD_getCParams(compressionLevel, 0, 0);
return ZSTD_estimateCStreamSize_advanced_usingCParams(cParams);
return ZSTD_estimateCStreamSize_usingCParams(cParams);
}
static U32 ZSTD_equivalentCParams(ZSTD_compressionParameters cParams1,
@@ -1699,14 +1708,18 @@ static size_t ZSTD_writeFrameHeader(void* dst, size_t dstCapacity,
U32 const fcsCode = params.fParams.contentSizeFlag ?
(pledgedSrcSize>=256) + (pledgedSrcSize>=65536+256) + (pledgedSrcSize>=0xFFFFFFFFU) : 0; /* 0-3 */
BYTE const frameHeaderDecriptionByte = (BYTE)(dictIDSizeCode + (checksumFlag<<2) + (singleSegment<<5) + (fcsCode<<6) );
size_t pos;
size_t pos=0;
if (dstCapacity < ZSTD_frameHeaderSize_max) return ERROR(dstSize_tooSmall);
DEBUGLOG(5, "ZSTD_writeFrameHeader : dictIDFlag : %u ; dictID : %u ; dictIDSizeCode : %u",
DEBUGLOG(4, "ZSTD_writeFrameHeader : dictIDFlag : %u ; dictID : %u ; dictIDSizeCode : %u",
!params.fParams.noDictIDFlag, dictID, dictIDSizeCode);
MEM_writeLE32(dst, ZSTD_MAGICNUMBER);
op[4] = frameHeaderDecriptionByte; pos=5;
if (params.format == ZSTD_f_zstd1) {
DEBUGLOG(4, "writing zstd magic number");
MEM_writeLE32(dst, ZSTD_MAGICNUMBER);
pos = 4;
}
op[pos++] = frameHeaderDecriptionByte;
if (!singleSegment) op[pos++] = windowLogByte;
switch(dictIDSizeCode)
{
@@ -2191,8 +2204,8 @@ size_t ZSTD_estimateCDictSize_advanced(
{
DEBUGLOG(5, "sizeof(ZSTD_CDict) : %u", (U32)sizeof(ZSTD_CDict));
DEBUGLOG(5, "CCtx estimate : %u",
(U32)ZSTD_estimateCCtxSize_advanced_usingCParams(cParams));
return sizeof(ZSTD_CDict) + ZSTD_estimateCCtxSize_advanced_usingCParams(cParams)
(U32)ZSTD_estimateCCtxSize_usingCParams(cParams));
return sizeof(ZSTD_CDict) + ZSTD_estimateCCtxSize_usingCParams(cParams)
+ (dictLoadMethod == ZSTD_dlm_byRef ? 0 : dictSize);
}
@@ -2317,7 +2330,7 @@ ZSTD_CDict* ZSTD_initStaticCDict(void* workspace, size_t workspaceSize,
ZSTD_dictMode_e dictMode,
ZSTD_compressionParameters cParams)
{
size_t const cctxSize = ZSTD_estimateCCtxSize_advanced_usingCParams(cParams);
size_t const cctxSize = ZSTD_estimateCCtxSize_usingCParams(cParams);
size_t const neededSize = sizeof(ZSTD_CDict) + (dictLoadMethod == ZSTD_dlm_byRef ? 0 : dictSize)
+ cctxSize;
ZSTD_CDict* const cdict = (ZSTD_CDict*) workspace;
@@ -2579,6 +2592,7 @@ MEM_STATIC size_t ZSTD_limitCopy(void* dst, size_t dstCapacity,
/** ZSTD_compressStream_generic():
* internal function for all *compressStream*() variants and *compress_generic()
* non-static, because can be called from zstdmt.c
* @return : hint size for next input */
size_t ZSTD_compressStream_generic(ZSTD_CStream* zcs,
ZSTD_outBuffer* output,
lib/compress/zstd_opt.c
+10 -2
View File
@@ -529,7 +529,14 @@ size_t ZSTD_compressBlock_opt_generic(ZSTD_CCtx* ctx,
} else {
opt[cur].rep[2] = (opt[cur].off > 1) ? opt[cur-mlen].rep[1] : opt[cur-mlen].rep[2];
opt[cur].rep[1] = (opt[cur].off > 0) ? opt[cur-mlen].rep[0] : opt[cur-mlen].rep[1];
opt[cur].rep[0] = ((opt[cur].off==ZSTD_REP_MOVE_OPT) && (mlen != 1)) ? (opt[cur-mlen].rep[0] - 1) : (opt[cur-mlen].rep[opt[cur].off]);
/* If opt[cur].off == ZSTD_REP_MOVE_OPT, then mlen != 1.
* offset ZSTD_REP_MOVE_OPT is used for the special case
* litLength == 0, where offset 0 means something special.
* mlen == 1 means the previous byte was stored as a literal,
* so they are mutually exclusive.
*/
assert(!(opt[cur].off == ZSTD_REP_MOVE_OPT && mlen == 1));
opt[cur].rep[0] = (opt[cur].off == ZSTD_REP_MOVE_OPT) ? (opt[cur-mlen].rep[0] - 1) : (opt[cur-mlen].rep[opt[cur].off]);
}
best_mlen = minMatch;
@@ -804,7 +811,8 @@ size_t ZSTD_compressBlock_opt_extDict_generic(ZSTD_CCtx* ctx,
} else {
opt[cur].rep[2] = (opt[cur].off > 1) ? opt[cur-mlen].rep[1] : opt[cur-mlen].rep[2];
opt[cur].rep[1] = (opt[cur].off > 0) ? opt[cur-mlen].rep[0] : opt[cur-mlen].rep[1];
opt[cur].rep[0] = ((opt[cur].off==ZSTD_REP_MOVE_OPT) && (mlen != 1)) ? (opt[cur-mlen].rep[0] - 1) : (opt[cur-mlen].rep[opt[cur].off]);
assert(!(opt[cur].off == ZSTD_REP_MOVE_OPT && mlen == 1));
opt[cur].rep[0] = (opt[cur].off == ZSTD_REP_MOVE_OPT) ? (opt[cur-mlen].rep[0] - 1) : (opt[cur-mlen].rep[opt[cur].off]);
}
best_mlen = minMatch;
lib/compress/zstdmt_compress.c
+49 -47
View File
@@ -53,22 +53,24 @@ static unsigned long long GetCurrentClockTimeMicroseconds(void)
}
#define MUTEX_WAIT_TIME_DLEVEL 6
#define PTHREAD_MUTEX_LOCK(mutex) { \
if (ZSTD_DEBUG>=MUTEX_WAIT_TIME_DLEVEL) { \
#define ZSTD_PTHREAD_MUTEX_LOCK(mutex) { \
if (ZSTD_DEBUG >= MUTEX_WAIT_TIME_DLEVEL) { \
unsigned long long const beforeTime = GetCurrentClockTimeMicroseconds(); \
pthread_mutex_lock(mutex); \
ZSTD_pthread_mutex_lock(mutex); \
{ unsigned long long const afterTime = GetCurrentClockTimeMicroseconds(); \
unsigned long long const elapsedTime = (afterTime-beforeTime); \
if (elapsedTime > 1000) { /* or whatever threshold you like; I'm using 1 millisecond here */ \
DEBUGLOG(MUTEX_WAIT_TIME_DLEVEL, "Thread took %llu microseconds to acquire mutex %s \n", \
elapsedTime, #mutex); \
} } \
} else pthread_mutex_lock(mutex); \
} else { \
ZSTD_pthread_mutex_lock(mutex); \
} \
}
#else
# define PTHREAD_MUTEX_LOCK(m) pthread_mutex_lock(m)
# define ZSTD_PTHREAD_MUTEX_LOCK(m) ZSTD_pthread_mutex_lock(m)
# define DEBUG_PRINTHEX(l,p,n) {}
#endif
@@ -85,7 +87,7 @@ typedef struct buffer_s {
static const buffer_t g_nullBuffer = { NULL, 0 };
typedef struct ZSTDMT_bufferPool_s {
pthread_mutex_t poolMutex;
ZSTD_pthread_mutex_t poolMutex;
size_t bufferSize;
unsigned totalBuffers;
unsigned nbBuffers;
@@ -99,7 +101,7 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_custo
ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc(
sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem);
if (bufPool==NULL) return NULL;
if (pthread_mutex_init(&bufPool->poolMutex, NULL)) {
if (ZSTD_pthread_mutex_init(&bufPool->poolMutex, NULL)) {
ZSTD_free(bufPool, cMem);
return NULL;
}
@@ -116,7 +118,7 @@ static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool)
if (!bufPool) return; /* compatibility with free on NULL */
for (u=0; u<bufPool->totalBuffers; u++)
ZSTD_free(bufPool->bTable[u].start, bufPool->cMem);
pthread_mutex_destroy(&bufPool->poolMutex);
ZSTD_pthread_mutex_destroy(&bufPool->poolMutex);
ZSTD_free(bufPool, bufPool->cMem);
}
@@ -127,10 +129,10 @@ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)
+ (bufPool->totalBuffers - 1) * sizeof(buffer_t);
unsigned u;
size_t totalBufferSize = 0;
pthread_mutex_lock(&bufPool->poolMutex);
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
for (u=0; u<bufPool->totalBuffers; u++)
totalBufferSize += bufPool->bTable[u].size;
pthread_mutex_unlock(&bufPool->poolMutex);
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
return poolSize + totalBufferSize;
}
@@ -146,20 +148,20 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
{
size_t const bSize = bufPool->bufferSize;
DEBUGLOG(5, "ZSTDMT_getBuffer");
pthread_mutex_lock(&bufPool->poolMutex);
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers) { /* try to use an existing buffer */
buffer_t const buf = bufPool->bTable[--(bufPool->nbBuffers)];
size_t const availBufferSize = buf.size;
if ((availBufferSize >= bSize) & (availBufferSize <= 10*bSize)) {
/* large enough, but not too much */
pthread_mutex_unlock(&bufPool->poolMutex);
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
return buf;
}
/* size conditions not respected : scratch this buffer, create new one */
DEBUGLOG(5, "existing buffer does not meet size conditions => freeing");
ZSTD_free(buf.start, bufPool->cMem);
}
pthread_mutex_unlock(&bufPool->poolMutex);
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
/* create new buffer */
DEBUGLOG(5, "create a new buffer");
{ buffer_t buffer;
@@ -175,13 +177,13 @@ static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf)
{
if (buf.start == NULL) return; /* compatible with release on NULL */
DEBUGLOG(5, "ZSTDMT_releaseBuffer");
pthread_mutex_lock(&bufPool->poolMutex);
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers < bufPool->totalBuffers) {
bufPool->bTable[bufPool->nbBuffers++] = buf; /* stored for later use */
pthread_mutex_unlock(&bufPool->poolMutex);
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
return;
}
pthread_mutex_unlock(&bufPool->poolMutex);
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
/* Reached bufferPool capacity (should not happen) */
DEBUGLOG(5, "buffer pool capacity reached => freeing ");
ZSTD_free(buf.start, bufPool->cMem);
@@ -206,7 +208,7 @@ static ZSTD_CCtx_params ZSTDMT_makeJobCCtxParams(ZSTD_CCtx_params const params)
/* a single CCtx Pool can be invoked from multiple threads in parallel */
typedef struct {
pthread_mutex_t poolMutex;
ZSTD_pthread_mutex_t poolMutex;
unsigned totalCCtx;
unsigned availCCtx;
ZSTD_customMem cMem;
@@ -219,7 +221,7 @@ static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool)
unsigned u;
for (u=0; u<pool->totalCCtx; u++)
ZSTD_freeCCtx(pool->cctx[u]); /* note : compatible with free on NULL */
pthread_mutex_destroy(&pool->poolMutex);
ZSTD_pthread_mutex_destroy(&pool->poolMutex);
ZSTD_free(pool, pool->cMem);
}
@@ -231,7 +233,7 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads,
ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_calloc(
sizeof(ZSTDMT_CCtxPool) + (nbThreads-1)*sizeof(ZSTD_CCtx*), cMem);
if (!cctxPool) return NULL;
if (pthread_mutex_init(&cctxPool->poolMutex, NULL)) {
if (ZSTD_pthread_mutex_init(&cctxPool->poolMutex, NULL)) {
ZSTD_free(cctxPool, cMem);
return NULL;
}
@@ -247,7 +249,7 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads,
/* only works during initialization phase, not during compression */
static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool)
{
pthread_mutex_lock(&cctxPool->poolMutex);
ZSTD_pthread_mutex_lock(&cctxPool->poolMutex);
{ unsigned const nbThreads = cctxPool->totalCCtx;
size_t const poolSize = sizeof(*cctxPool)
+ (nbThreads-1)*sizeof(ZSTD_CCtx*);
@@ -256,7 +258,7 @@ static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool)
for (u=0; u<nbThreads; u++) {
totalCCtxSize += ZSTD_sizeof_CCtx(cctxPool->cctx[u]);
}
pthread_mutex_unlock(&cctxPool->poolMutex);
ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);
return poolSize + totalCCtxSize;
}
}
@@ -264,14 +266,14 @@ static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool)
static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool)
{
DEBUGLOG(5, "ZSTDMT_getCCtx");
pthread_mutex_lock(&cctxPool->poolMutex);
ZSTD_pthread_mutex_lock(&cctxPool->poolMutex);
if (cctxPool->availCCtx) {
cctxPool->availCCtx--;
{ ZSTD_CCtx* const cctx = cctxPool->cctx[cctxPool->availCCtx];
pthread_mutex_unlock(&cctxPool->poolMutex);
ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);
return cctx;
} }
pthread_mutex_unlock(&cctxPool->poolMutex);
ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);
DEBUGLOG(5, "create one more CCtx");
return ZSTD_createCCtx_advanced(cctxPool->cMem); /* note : can be NULL, when creation fails ! */
}
@@ -279,7 +281,7 @@ static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool)
static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
{
if (cctx==NULL) return; /* compatibility with release on NULL */
pthread_mutex_lock(&pool->poolMutex);
ZSTD_pthread_mutex_lock(&pool->poolMutex);
if (pool->availCCtx < pool->totalCCtx)
pool->cctx[pool->availCCtx++] = cctx;
else {
@@ -287,7 +289,7 @@ static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
DEBUGLOG(5, "CCtx pool overflow : free cctx");
ZSTD_freeCCtx(cctx);
}
pthread_mutex_unlock(&pool->poolMutex);
ZSTD_pthread_mutex_unlock(&pool->poolMutex);
}
@@ -305,8 +307,8 @@ typedef struct {
unsigned lastChunk;
unsigned jobCompleted;
unsigned jobScanned;
pthread_mutex_t* jobCompleted_mutex;
pthread_cond_t* jobCompleted_cond;
ZSTD_pthread_mutex_t* jobCompleted_mutex;
ZSTD_pthread_cond_t* jobCompleted_cond;
ZSTD_CCtx_params params;
const ZSTD_CDict* cdict;
ZSTDMT_CCtxPool* cctxPool;
@@ -373,11 +375,11 @@ _endJob:
ZSTDMT_releaseCCtx(job->cctxPool, cctx);
ZSTDMT_releaseBuffer(job->bufPool, job->src);
job->src = g_nullBuffer; job->srcStart = NULL;
PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex);
ZSTD_PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex);
job->jobCompleted = 1;
job->jobScanned = 0;
pthread_cond_signal(job->jobCompleted_cond);
pthread_mutex_unlock(job->jobCompleted_mutex);
ZSTD_pthread_cond_signal(job->jobCompleted_cond);
ZSTD_pthread_mutex_unlock(job->jobCompleted_mutex);
}
@@ -395,8 +397,8 @@ struct ZSTDMT_CCtx_s {
ZSTDMT_jobDescription* jobs;
ZSTDMT_bufferPool* bufPool;
ZSTDMT_CCtxPool* cctxPool;
pthread_mutex_t jobCompleted_mutex;
pthread_cond_t jobCompleted_cond;
ZSTD_pthread_mutex_t jobCompleted_mutex;
ZSTD_pthread_cond_t jobCompleted_cond;
size_t targetSectionSize;
size_t inBuffSize;
size_t dictSize;
@@ -459,11 +461,11 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem)
ZSTDMT_freeCCtx(mtctx);
return NULL;
}
if (pthread_mutex_init(&mtctx->jobCompleted_mutex, NULL)) {
if (ZSTD_pthread_mutex_init(&mtctx->jobCompleted_mutex, NULL)) {
ZSTDMT_freeCCtx(mtctx);
return NULL;
}
if (pthread_cond_init(&mtctx->jobCompleted_cond, NULL)) {
if (ZSTD_pthread_cond_init(&mtctx->jobCompleted_cond, NULL)) {
ZSTDMT_freeCCtx(mtctx);
return NULL;
}
@@ -499,12 +501,12 @@ static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* zcs)
DEBUGLOG(4, "ZSTDMT_waitForAllJobsCompleted");
while (zcs->doneJobID < zcs->nextJobID) {
unsigned const jobID = zcs->doneJobID & zcs->jobIDMask;
PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
ZSTD_PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
while (zcs->jobs[jobID].jobCompleted==0) {
DEBUGLOG(5, "waiting for jobCompleted signal from chunk %u", zcs->doneJobID); /* we want to block when waiting for data to flush */
pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex);
ZSTD_pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex);
}
pthread_mutex_unlock(&zcs->jobCompleted_mutex);
ZSTD_pthread_mutex_unlock(&zcs->jobCompleted_mutex);
zcs->doneJobID++;
}
}
@@ -521,8 +523,8 @@ size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx)
ZSTD_free(mtctx->jobs, mtctx->cMem);
ZSTDMT_freeCCtxPool(mtctx->cctxPool);
ZSTD_freeCDict(mtctx->cdictLocal);
pthread_mutex_destroy(&mtctx->jobCompleted_mutex);
pthread_cond_destroy(&mtctx->jobCompleted_cond);
ZSTD_pthread_mutex_destroy(&mtctx->jobCompleted_mutex);
ZSTD_pthread_cond_destroy(&mtctx->jobCompleted_cond);
ZSTD_free(mtctx, mtctx->cMem);
return 0;
}
@@ -667,12 +669,12 @@ static size_t ZSTDMT_compress_advanced_internal(
unsigned chunkID;
for (chunkID=0; chunkID<nbChunks; chunkID++) {
DEBUGLOG(5, "waiting for chunk %u ", chunkID);
PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex);
ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex);
while (mtctx->jobs[chunkID].jobCompleted==0) {
DEBUGLOG(5, "waiting for jobCompleted signal from chunk %u", chunkID);
pthread_cond_wait(&mtctx->jobCompleted_cond, &mtctx->jobCompleted_mutex);
ZSTD_pthread_cond_wait(&mtctx->jobCompleted_cond, &mtctx->jobCompleted_mutex);
}
pthread_mutex_unlock(&mtctx->jobCompleted_mutex);
ZSTD_pthread_mutex_unlock(&mtctx->jobCompleted_mutex);
DEBUGLOG(5, "ready to write chunk %u ", chunkID);
mtctx->jobs[chunkID].srcStart = NULL;
@@ -926,13 +928,13 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
{
unsigned const wJobID = zcs->doneJobID & zcs->jobIDMask;
if (zcs->doneJobID == zcs->nextJobID) return 0; /* all flushed ! */
PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
ZSTD_PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
while (zcs->jobs[wJobID].jobCompleted==0) {
DEBUGLOG(5, "waiting for jobCompleted signal from job %u", zcs->doneJobID);
if (!blockToFlush) { pthread_mutex_unlock(&zcs->jobCompleted_mutex); return 0; } /* nothing ready to be flushed => skip */
pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex); /* block when nothing available to flush */
if (!blockToFlush) { ZSTD_pthread_mutex_unlock(&zcs->jobCompleted_mutex); return 0; } /* nothing ready to be flushed => skip */
ZSTD_pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex); /* block when nothing available to flush */
}
pthread_mutex_unlock(&zcs->jobCompleted_mutex);
ZSTD_pthread_mutex_unlock(&zcs->jobCompleted_mutex);
/* compression job completed : output can be flushed */
{ ZSTDMT_jobDescription job = zcs->jobs[wJobID];
if (!job.jobScanned) {