diff options
Diffstat (limited to 'thirdparty/zstd/compress/zstdmt_compress.c')
| -rw-r--r-- | thirdparty/zstd/compress/zstdmt_compress.c | 343 |
1 files changed, 210 insertions, 133 deletions
diff --git a/thirdparty/zstd/compress/zstdmt_compress.c b/thirdparty/zstd/compress/zstdmt_compress.c index 8564bc4392..7831cd3bd8 100644 --- a/thirdparty/zstd/compress/zstdmt_compress.c +++ b/thirdparty/zstd/compress/zstdmt_compress.c @@ -5,11 +5,12 @@ * This source code is licensed under both the BSD-style license (found in the * LICENSE file in the root directory of this source tree) and the GPLv2 (found * in the COPYING file in the root directory of this source tree). + * You may select, at your option, one of the above-listed licenses. */ /* ====== Tuning parameters ====== */ -#define ZSTDMT_NBTHREADS_MAX 256 +#define ZSTDMT_NBTHREADS_MAX 200 #define ZSTDMT_OVERLAPLOG_DEFAULT 6 @@ -52,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 @@ -84,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; @@ -98,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; } @@ -112,10 +115,13 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_custo static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool) { unsigned u; + DEBUGLOG(3, "ZSTDMT_freeBufferPool (address:%08X)", (U32)(size_t)bufPool); if (!bufPool) return; /* compatibility with free on NULL */ - for (u=0; u<bufPool->totalBuffers; u++) + for (u=0; u<bufPool->totalBuffers; u++) { + DEBUGLOG(4, "free buffer %2u (address:%08X)", u, (U32)(size_t)bufPool->bTable[u].start); ZSTD_free(bufPool->bTable[u].start, bufPool->cMem); - pthread_mutex_destroy(&bufPool->poolMutex); + } + ZSTD_pthread_mutex_destroy(&bufPool->poolMutex); ZSTD_free(bufPool, bufPool->cMem); } @@ -126,10 +132,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; } @@ -145,20 +151,21 @@ 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; + bufPool->bTable[bufPool->nbBuffers] = g_nullBuffer; 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; @@ -174,24 +181,38 @@ 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); } +/* Sets parameters relevant to the compression job, initializing others to + * default values. Notably, nbThreads should probably be zero. */ +static ZSTD_CCtx_params ZSTDMT_makeJobCCtxParams(ZSTD_CCtx_params const params) +{ + ZSTD_CCtx_params jobParams; + memset(&jobParams, 0, sizeof(jobParams)); + + jobParams.cParams = params.cParams; + jobParams.fParams = params.fParams; + jobParams.compressionLevel = params.compressionLevel; + + jobParams.ldmParams = params.ldmParams; + return jobParams; +} /* ===== CCtx Pool ===== */ /* 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; @@ -204,7 +225,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); } @@ -216,7 +237,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; } @@ -232,7 +253,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*); @@ -241,7 +262,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; } } @@ -249,14 +270,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 ! */ } @@ -264,7 +285,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 { @@ -272,7 +293,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); } @@ -290,9 +311,9 @@ typedef struct { unsigned lastChunk; unsigned jobCompleted; unsigned jobScanned; - pthread_mutex_t* jobCompleted_mutex; - pthread_cond_t* jobCompleted_cond; - ZSTD_parameters params; + ZSTD_pthread_mutex_t* jobCompleted_mutex; + ZSTD_pthread_cond_t* jobCompleted_cond; + ZSTD_CCtx_params params; const ZSTD_CDict* cdict; ZSTDMT_CCtxPool* cctxPool; ZSTDMT_bufferPool* bufPool; @@ -329,10 +350,15 @@ void ZSTDMT_compressChunk(void* jobDescription) if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; } } else { /* srcStart points at reloaded section */ if (!job->firstChunk) job->params.fParams.contentSizeFlag = 0; /* ensure no srcSize control */ - { size_t const dictModeError = ZSTD_setCCtxParameter(cctx, ZSTD_p_forceRawDict, 1); /* Force loading dictionary in "content-only" mode (no header analysis) */ - size_t const initError = ZSTD_compressBegin_advanced(cctx, job->srcStart, job->dictSize, job->params, job->fullFrameSize); - if (ZSTD_isError(initError) || ZSTD_isError(dictModeError)) { job->cSize = initError; goto _endJob; } - ZSTD_setCCtxParameter(cctx, ZSTD_p_forceWindow, 1); + { ZSTD_CCtx_params jobParams = job->params; + size_t const forceWindowError = + ZSTD_CCtxParam_setParameter(&jobParams, ZSTD_p_forceMaxWindow, !job->firstChunk); + /* Force loading dictionary in "content-only" mode (no header analysis) */ + size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, job->srcStart, job->dictSize, ZSTD_dm_rawContent, jobParams, job->fullFrameSize); + if (ZSTD_isError(initError) || ZSTD_isError(forceWindowError)) { + job->cSize = initError; + goto _endJob; + } } } if (!job->firstChunk) { /* flush and overwrite frame header when it's not first segment */ size_t const hSize = ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, 0); @@ -353,11 +379,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); } @@ -375,24 +401,21 @@ 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; size_t targetDictSize; inBuff_t inBuff; - ZSTD_parameters params; + ZSTD_CCtx_params params; XXH64_state_t xxhState; - unsigned nbThreads; unsigned jobIDMask; unsigned doneJobID; unsigned nextJobID; unsigned frameEnded; unsigned allJobsCompleted; - unsigned overlapLog; unsigned long long frameContentSize; - size_t sectionSize; ZSTD_customMem cMem; ZSTD_CDict* cdictLocal; const ZSTD_CDict* cdict; @@ -407,6 +430,15 @@ static ZSTDMT_jobDescription* ZSTDMT_allocJobsTable(U32* nbJobsPtr, ZSTD_customM nbJobs * sizeof(ZSTDMT_jobDescription), cMem); } +/* Internal only */ +size_t ZSTDMT_initializeCCtxParameters(ZSTD_CCtx_params* params, unsigned nbThreads) +{ + params->nbThreads = nbThreads; + params->overlapSizeLog = ZSTDMT_OVERLAPLOG_DEFAULT; + params->jobSize = 0; + return 0; +} + ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem) { ZSTDMT_CCtx* mtctx; @@ -421,12 +453,10 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem) mtctx = (ZSTDMT_CCtx*) ZSTD_calloc(sizeof(ZSTDMT_CCtx), cMem); if (!mtctx) return NULL; + ZSTDMT_initializeCCtxParameters(&mtctx->params, nbThreads); mtctx->cMem = cMem; - mtctx->nbThreads = nbThreads; mtctx->allJobsCompleted = 1; - mtctx->sectionSize = 0; - mtctx->overlapLog = ZSTDMT_OVERLAPLOG_DEFAULT; - mtctx->factory = POOL_create(nbThreads, 0); + mtctx->factory = POOL_create_advanced(nbThreads, 0, cMem); mtctx->jobs = ZSTDMT_allocJobsTable(&nbJobs, cMem); mtctx->jobIDMask = nbJobs - 1; mtctx->bufPool = ZSTDMT_createBufferPool(nbThreads, cMem); @@ -435,11 +465,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; } @@ -459,28 +489,46 @@ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx) unsigned jobID; DEBUGLOG(3, "ZSTDMT_releaseAllJobResources"); for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) { + DEBUGLOG(4, "job%02u: release dst address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].dstBuff.start); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff); mtctx->jobs[jobID].dstBuff = g_nullBuffer; + DEBUGLOG(4, "job%02u: release src address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].src.start); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].src); mtctx->jobs[jobID].src = g_nullBuffer; } memset(mtctx->jobs, 0, (mtctx->jobIDMask+1)*sizeof(ZSTDMT_jobDescription)); + DEBUGLOG(4, "input: release address %08X", (U32)(size_t)mtctx->inBuff.buffer.start); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer); mtctx->inBuff.buffer = g_nullBuffer; mtctx->allJobsCompleted = 1; } +static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* zcs) +{ + DEBUGLOG(4, "ZSTDMT_waitForAllJobsCompleted"); + while (zcs->doneJobID < zcs->nextJobID) { + unsigned const jobID = zcs->doneJobID & zcs->jobIDMask; + 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 */ + ZSTD_pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex); + } + ZSTD_pthread_mutex_unlock(&zcs->jobCompleted_mutex); + zcs->doneJobID++; + } +} + size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx) { if (mtctx==NULL) return 0; /* compatible with free on NULL */ - POOL_free(mtctx->factory); - if (!mtctx->allJobsCompleted) ZSTDMT_releaseAllJobResources(mtctx); /* stop workers first */ - ZSTDMT_freeBufferPool(mtctx->bufPool); /* release job resources into pools first */ + POOL_free(mtctx->factory); /* stop and free worker threads */ + ZSTDMT_releaseAllJobResources(mtctx); /* release job resources into pools first */ ZSTD_free(mtctx->jobs, mtctx->cMem); + ZSTDMT_freeBufferPool(mtctx->bufPool); 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; } @@ -496,22 +544,35 @@ size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx) + ZSTD_sizeof_CDict(mtctx->cdictLocal); } -size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSDTMT_parameter parameter, unsigned value) -{ +/* Internal only */ +size_t ZSTDMT_CCtxParam_setMTCtxParameter( + ZSTD_CCtx_params* params, ZSTDMT_parameter parameter, unsigned value) { switch(parameter) { case ZSTDMT_p_sectionSize : - mtctx->sectionSize = value; + params->jobSize = value; return 0; case ZSTDMT_p_overlapSectionLog : - DEBUGLOG(5, "ZSTDMT_p_overlapSectionLog : %u", value); - mtctx->overlapLog = (value >= 9) ? 9 : value; + DEBUGLOG(4, "ZSTDMT_p_overlapSectionLog : %u", value); + params->overlapSizeLog = (value >= 9) ? 9 : value; return 0; default : return ERROR(parameter_unsupported); } } +size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSTDMT_parameter parameter, unsigned value) +{ + switch(parameter) + { + case ZSTDMT_p_sectionSize : + return ZSTDMT_CCtxParam_setMTCtxParameter(&mtctx->params, parameter, value); + case ZSTDMT_p_overlapSectionLog : + return ZSTDMT_CCtxParam_setMTCtxParameter(&mtctx->params, parameter, value); + default : + return ERROR(parameter_unsupported); + } +} /* ------------------------------------------ */ /* ===== Multi-threaded compression ===== */ @@ -528,17 +589,17 @@ static unsigned computeNbChunks(size_t srcSize, unsigned windowLog, unsigned nbT return (multiplier>1) ? nbChunksLarge : nbChunksSmall; } - -size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, - void* dst, size_t dstCapacity, - const void* src, size_t srcSize, - const ZSTD_CDict* cdict, - ZSTD_parameters const params, - unsigned overlapLog) +static size_t ZSTDMT_compress_advanced_internal( + ZSTDMT_CCtx* mtctx, + void* dst, size_t dstCapacity, + const void* src, size_t srcSize, + const ZSTD_CDict* cdict, + ZSTD_CCtx_params const params) { - unsigned const overlapRLog = (overlapLog>9) ? 0 : 9-overlapLog; + ZSTD_CCtx_params const jobParams = ZSTDMT_makeJobCCtxParams(params); + unsigned const overlapRLog = (params.overlapSizeLog>9) ? 0 : 9-params.overlapSizeLog; size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog); - unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, mtctx->nbThreads); + unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, params.nbThreads); size_t const proposedChunkSize = (srcSize + (nbChunks-1)) / nbChunks; size_t const avgChunkSize = ((proposedChunkSize & 0x1FFFF) < 0x7FFF) ? proposedChunkSize + 0xFFFF : proposedChunkSize; /* avoid too small last block */ const char* const srcStart = (const char*)src; @@ -546,12 +607,14 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, unsigned const compressWithinDst = (dstCapacity >= ZSTD_compressBound(srcSize)) ? nbChunks : (unsigned)(dstCapacity / ZSTD_compressBound(avgChunkSize)); /* presumes avgChunkSize >= 256 KB, which should be the case */ size_t frameStartPos = 0, dstBufferPos = 0; XXH64_state_t xxh64; + assert(jobParams.nbThreads == 0); + assert(mtctx->cctxPool->totalCCtx == params.nbThreads); DEBUGLOG(4, "nbChunks : %2u (chunkSize : %u bytes) ", nbChunks, (U32)avgChunkSize); if (nbChunks==1) { /* fallback to single-thread mode */ ZSTD_CCtx* const cctx = mtctx->cctxPool->cctx[0]; - if (cdict) return ZSTD_compress_usingCDict_advanced(cctx, dst, dstCapacity, src, srcSize, cdict, params.fParams); - return ZSTD_compress_advanced(cctx, dst, dstCapacity, src, srcSize, NULL, 0, params); + if (cdict) return ZSTD_compress_usingCDict_advanced(cctx, dst, dstCapacity, src, srcSize, cdict, jobParams.fParams); + return ZSTD_compress_advanced_internal(cctx, dst, dstCapacity, src, srcSize, NULL, 0, jobParams); } assert(avgChunkSize >= 256 KB); /* condition for ZSTD_compressBound(A) + ZSTD_compressBound(B) <= ZSTD_compressBound(A+B), which is required for compressWithinDst */ ZSTDMT_setBufferSize(mtctx->bufPool, ZSTD_compressBound(avgChunkSize) ); @@ -580,7 +643,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, mtctx->jobs[u].srcSize = chunkSize; mtctx->jobs[u].cdict = mtctx->nextJobID==0 ? cdict : NULL; mtctx->jobs[u].fullFrameSize = srcSize; - mtctx->jobs[u].params = params; + mtctx->jobs[u].params = jobParams; /* do not calculate checksum within sections, but write it in header for first section */ if (u!=0) mtctx->jobs[u].params.fParams.checksumFlag = 0; mtctx->jobs[u].dstBuff = dstBuffer; @@ -610,12 +673,12 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, 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; @@ -628,9 +691,8 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, if (chunkID >= compressWithinDst) { /* chunk compressed into its own buffer, which must be released */ DEBUGLOG(5, "releasing buffer %u>=%u", chunkID, compressWithinDst); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[chunkID].dstBuff); - } - mtctx->jobs[chunkID].dstBuff = g_nullBuffer; - } + } } + mtctx->jobs[chunkID].dstBuff = g_nullBuffer; dstPos += cSize ; } } /* for (chunkID=0; chunkID<nbChunks; chunkID++) */ @@ -651,6 +713,23 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, } } +size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, + void* dst, size_t dstCapacity, + const void* src, size_t srcSize, + const ZSTD_CDict* cdict, + ZSTD_parameters const params, + unsigned overlapLog) +{ + ZSTD_CCtx_params cctxParams = mtctx->params; + cctxParams.cParams = params.cParams; + cctxParams.fParams = params.fParams; + cctxParams.overlapSizeLog = overlapLog; + return ZSTDMT_compress_advanced_internal(mtctx, + dst, dstCapacity, + src, srcSize, + cdict, cctxParams); +} + size_t ZSTDMT_compressCCtx(ZSTDMT_CCtx* mtctx, void* dst, size_t dstCapacity, @@ -668,38 +747,25 @@ size_t ZSTDMT_compressCCtx(ZSTDMT_CCtx* mtctx, /* ======= Streaming API ======= */ /* ====================================== */ -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); - 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); - } - pthread_mutex_unlock(&zcs->jobCompleted_mutex); - zcs->doneJobID++; - } -} - - -/** ZSTDMT_initCStream_internal() : - * internal usage only */ -size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, - const void* dict, size_t dictSize, const ZSTD_CDict* cdict, - ZSTD_parameters params, unsigned long long pledgedSrcSize) +size_t ZSTDMT_initCStream_internal( + ZSTDMT_CCtx* zcs, + const void* dict, size_t dictSize, ZSTD_dictMode_e dictMode, + const ZSTD_CDict* cdict, ZSTD_CCtx_params params, + unsigned long long pledgedSrcSize) { DEBUGLOG(4, "ZSTDMT_initCStream_internal"); /* params are supposed to be fully validated at this point */ assert(!ZSTD_isError(ZSTD_checkCParams(params.cParams))); assert(!((dict) && (cdict))); /* either dict or cdict, not both */ + assert(zcs->cctxPool->totalCCtx == params.nbThreads); - if (zcs->nbThreads==1) { + if (params.nbThreads==1) { + ZSTD_CCtx_params const singleThreadParams = ZSTDMT_makeJobCCtxParams(params); DEBUGLOG(4, "single thread mode"); + assert(singleThreadParams.nbThreads == 0); return ZSTD_initCStream_internal(zcs->cctxPool->cctx[0], - dict, dictSize, cdict, - params, pledgedSrcSize); + dict, dictSize, cdict, + singleThreadParams, pledgedSrcSize); } if (zcs->allJobsCompleted == 0) { /* previous compression not correctly finished */ @@ -714,7 +780,7 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, DEBUGLOG(4,"cdictLocal: %08X", (U32)(size_t)zcs->cdictLocal); ZSTD_freeCDict(zcs->cdictLocal); zcs->cdictLocal = ZSTD_createCDict_advanced(dict, dictSize, - 0 /* byRef */, ZSTD_dm_auto, /* note : a loadPrefix becomes an internal CDict */ + ZSTD_dlm_byCopy, dictMode, /* note : a loadPrefix becomes an internal CDict */ params.cParams, zcs->cMem); zcs->cdict = zcs->cdictLocal; if (zcs->cdictLocal == NULL) return ERROR(memory_allocation); @@ -725,10 +791,10 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs, zcs->cdict = cdict; } - zcs->targetDictSize = (zcs->overlapLog==0) ? 0 : (size_t)1 << (zcs->params.cParams.windowLog - (9 - zcs->overlapLog)); - DEBUGLOG(4, "overlapLog : %u ", zcs->overlapLog); + zcs->targetDictSize = (params.overlapSizeLog==0) ? 0 : (size_t)1 << (params.cParams.windowLog - (9 - params.overlapSizeLog)); + DEBUGLOG(4, "overlapLog : %u ", params.overlapSizeLog); DEBUGLOG(4, "overlap Size : %u KB", (U32)(zcs->targetDictSize>>10)); - zcs->targetSectionSize = zcs->sectionSize ? zcs->sectionSize : (size_t)1 << (zcs->params.cParams.windowLog + 2); + zcs->targetSectionSize = params.jobSize ? params.jobSize : (size_t)1 << (params.cParams.windowLog + 2); zcs->targetSectionSize = MAX(ZSTDMT_SECTION_SIZE_MIN, zcs->targetSectionSize); zcs->targetSectionSize = MAX(zcs->targetDictSize, zcs->targetSectionSize); DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10)); @@ -749,8 +815,12 @@ size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx, ZSTD_parameters params, unsigned long long pledgedSrcSize) { + ZSTD_CCtx_params cctxParams = mtctx->params; DEBUGLOG(5, "ZSTDMT_initCStream_advanced"); - return ZSTDMT_initCStream_internal(mtctx, dict, dictSize, NULL, params, pledgedSrcSize); + cctxParams.cParams = params.cParams; + cctxParams.fParams = params.fParams; + return ZSTDMT_initCStream_internal(mtctx, dict, dictSize, ZSTD_dm_auto, NULL, + cctxParams, pledgedSrcSize); } size_t ZSTDMT_initCStream_usingCDict(ZSTDMT_CCtx* mtctx, @@ -758,11 +828,12 @@ size_t ZSTDMT_initCStream_usingCDict(ZSTDMT_CCtx* mtctx, ZSTD_frameParameters fParams, unsigned long long pledgedSrcSize) { - ZSTD_parameters params = ZSTD_getParamsFromCDict(cdict); + ZSTD_CCtx_params cctxParams = mtctx->params; + cctxParams.cParams = ZSTD_getCParamsFromCDict(cdict); + cctxParams.fParams = fParams; if (cdict==NULL) return ERROR(dictionary_wrong); /* method incompatible with NULL cdict */ - params.fParams = fParams; - return ZSTDMT_initCStream_internal(mtctx, NULL, 0 /*dictSize*/, cdict, - params, pledgedSrcSize); + return ZSTDMT_initCStream_internal(mtctx, NULL, 0 /*dictSize*/, ZSTD_dm_auto, cdict, + cctxParams, pledgedSrcSize); } @@ -770,14 +841,18 @@ size_t ZSTDMT_initCStream_usingCDict(ZSTDMT_CCtx* mtctx, * pledgedSrcSize is optional and can be zero == unknown */ size_t ZSTDMT_resetCStream(ZSTDMT_CCtx* zcs, unsigned long long pledgedSrcSize) { - if (zcs->nbThreads==1) + if (zcs->params.nbThreads==1) return ZSTD_resetCStream(zcs->cctxPool->cctx[0], pledgedSrcSize); - return ZSTDMT_initCStream_internal(zcs, NULL, 0, 0, zcs->params, pledgedSrcSize); + return ZSTDMT_initCStream_internal(zcs, NULL, 0, ZSTD_dm_auto, 0, zcs->params, + pledgedSrcSize); } size_t ZSTDMT_initCStream(ZSTDMT_CCtx* zcs, int compressionLevel) { ZSTD_parameters const params = ZSTD_getParams(compressionLevel, 0, 0); - return ZSTDMT_initCStream_internal(zcs, NULL, 0, NULL, params, 0); + ZSTD_CCtx_params cctxParams = zcs->params; + cctxParams.cParams = params.cParams; + cctxParams.fParams = params.fParams; + return ZSTDMT_initCStream_internal(zcs, NULL, 0, ZSTD_dm_auto, NULL, cctxParams, 0); } @@ -856,13 +931,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) { @@ -906,7 +981,7 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi /** ZSTDMT_compressStream_generic() : - * internal use only + * internal use only - exposed to be invoked from zstd_compress.c * assumption : output and input are valid (pos <= size) * @return : minimum amount of data remaining to flush, 0 if none */ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, @@ -915,25 +990,26 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, ZSTD_EndDirective endOp) { size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize; + unsigned forwardInputProgress = 0; assert(output->pos <= output->size); assert(input->pos <= input->size); if ((mtctx->frameEnded) && (endOp==ZSTD_e_continue)) { - /* current frame being ended. Only flush/end are allowed. Or start new frame with init */ + /* current frame being ended. Only flush/end are allowed */ return ERROR(stage_wrong); } - if (mtctx->nbThreads==1) { /* delegate to single-thread (synchronous) */ + if (mtctx->params.nbThreads==1) { /* delegate to single-thread (synchronous) */ return ZSTD_compressStream_generic(mtctx->cctxPool->cctx[0], output, input, endOp); } - /* single-pass shortcut (note : this is synchronous-mode) */ - if ( (mtctx->nextJobID==0) /* just started */ - && (mtctx->inBuff.filled==0) /* nothing buffered */ - && (endOp==ZSTD_e_end) /* end order */ + /* single-pass shortcut (note : synchronous-mode) */ + if ( (mtctx->nextJobID == 0) /* just started */ + && (mtctx->inBuff.filled == 0) /* nothing buffered */ + && (endOp == ZSTD_e_end) /* end order */ && (output->size - output->pos >= ZSTD_compressBound(input->size - input->pos)) ) { /* enough room */ - size_t const cSize = ZSTDMT_compress_advanced(mtctx, + size_t const cSize = ZSTDMT_compress_advanced_internal(mtctx, (char*)output->dst + output->pos, output->size - output->pos, (const char*)input->src + input->pos, input->size - input->pos, - mtctx->cdict, mtctx->params, mtctx->overlapLog); + mtctx->cdict, mtctx->params); if (ZSTD_isError(cSize)) return cSize; input->pos = input->size; output->pos += cSize; @@ -946,15 +1022,16 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, /* fill input buffer */ if (input->size > input->pos) { /* support NULL input */ if (mtctx->inBuff.buffer.start == NULL) { - mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool); - if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation); + mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool); /* note : may fail, in which case, no forward input progress */ mtctx->inBuff.filled = 0; } - { size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled); + if (mtctx->inBuff.buffer.start) { + size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled); DEBUGLOG(5, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad); memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad); input->pos += toLoad; mtctx->inBuff.filled += toLoad; + forwardInputProgress = toLoad>0; } } if ( (mtctx->inBuff.filled >= newJobThreshold) /* filled enough : let's compress */ @@ -963,7 +1040,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, } /* check for potential compressed data ready to be flushed */ - CHECK_F( ZSTDMT_flushNextJob(mtctx, output, (mtctx->inBuff.filled == mtctx->inBuffSize) /* blockToFlush */) ); /* block if it wasn't possible to create new job due to saturation */ + CHECK_F( ZSTDMT_flushNextJob(mtctx, output, !forwardInputProgress /* blockToFlush */) ); /* block if there was no forward input progress */ if (input->pos < input->size) /* input not consumed : do not flush yet */ endOp = ZSTD_e_continue; @@ -1008,7 +1085,7 @@ static size_t ZSTDMT_flushStream_internal(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* outp size_t ZSTDMT_flushStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output) { DEBUGLOG(5, "ZSTDMT_flushStream"); - if (zcs->nbThreads==1) + if (zcs->params.nbThreads==1) return ZSTD_flushStream(zcs->cctxPool->cctx[0], output); return ZSTDMT_flushStream_internal(zcs, output, 0 /* endFrame */); } @@ -1016,7 +1093,7 @@ size_t ZSTDMT_flushStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output) size_t ZSTDMT_endStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output) { DEBUGLOG(4, "ZSTDMT_endStream"); - if (zcs->nbThreads==1) + if (zcs->params.nbThreads==1) return ZSTD_endStream(zcs->cctxPool->cctx[0], output); return ZSTDMT_flushStream_internal(zcs, output, 1 /* endFrame */); } |