diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index 38c3fc0d4cd..88e1df51232 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -184,10 +184,10 @@ static void ZSTD_freeCCtxContent(ZSTD_CCtx* cctx) { assert(cctx != NULL); assert(cctx->staticSize == 0); - ZSTD_clearAllDicts(cctx); #ifdef ZSTD_MULTITHREAD - ZSTDMT_freeCCtx(cctx->mtctx); cctx->mtctx = NULL; + ZSTDMT_freeCCtx(cctx->mtctx); cctx->mtctx = NULL; /* pending jobs may use the dictionary */ #endif + ZSTD_clearAllDicts(cctx); ZSTD_cwksp_free(&cctx->workspace, cctx->customMem); } @@ -1389,6 +1389,10 @@ size_t ZSTD_CCtx_reset(ZSTD_CCtx* cctx, ZSTD_ResetDirective reset) { if ( (reset == ZSTD_reset_session_only) || (reset == ZSTD_reset_session_and_parameters) ) { +#ifdef ZSTD_MULTITHREAD + /* jobs of an abandoned frame may still reference the dictionary */ + if (cctx->mtctx != NULL) ZSTDMT_waitForPendingJobs(cctx->mtctx); +#endif cctx->streamStage = zcss_init; cctx->pledgedSrcSizePlusOne = 0; } diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index cb710475c2a..7118bbe7f4a 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -662,6 +662,7 @@ static const Range kNullRange = { NULL, 0 }; typedef struct { size_t consumed; /* SHARED - set0 by mtctx, then modified by worker AND read by mtctx */ size_t cSize; /* SHARED - set0 by mtctx, then modified by worker AND read by mtctx, then set0 by mtctx */ + unsigned completed; /* SHARED - set0 by mtctx, then set1 by worker when it no longer uses the job */ ZSTD_pthread_mutex_t job_mutex; /* Thread-safe - used by mtctx and worker */ ZSTD_pthread_cond_t job_cond; /* Thread-safe - used by mtctx and worker */ ZSTDMT_CCtxPool* cctxPool; /* Thread-safe - used by mtctx and (all) workers */ @@ -814,7 +815,8 @@ static void ZSTDMT_compressionJob(void* jobDescription) ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); if (ZSTD_isError(job->cSize)) assert(lastCBlockSize == 0); job->cSize += lastCBlockSize; - job->consumed = job->src.size; /* when job->consumed == job->src.size , compression job is presumed completed */ + job->consumed = job->src.size; + job->completed = 1; /* also true for an empty job, unlike consumed == src.size */ ZSTD_pthread_cond_signal(&job->job_cond); ZSTD_pthread_mutex_unlock(&job->job_mutex); } @@ -1025,6 +1027,7 @@ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx) } mtctx->inBuff.buffer = g_nullBuffer; mtctx->inBuff.filled = 0; + mtctx->jobReady = 0; /* the prepared job was cleared above */ mtctx->allJobsCompleted = 1; } @@ -1034,7 +1037,7 @@ static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* mtctx) while (mtctx->doneJobID < mtctx->nextJobID) { unsigned const jobID = mtctx->doneJobID & mtctx->jobIDMask; ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->jobs[jobID].job_mutex); - while (mtctx->jobs[jobID].consumed < mtctx->jobs[jobID].src.size) { + while (!mtctx->jobs[jobID].completed) { DEBUGLOG(4, "waiting for jobCompleted signal from job %u", mtctx->doneJobID); /* we want to block when waiting for data to flush */ ZSTD_pthread_cond_wait(&mtctx->jobs[jobID].job_cond, &mtctx->jobs[jobID].job_mutex); } @@ -1043,9 +1046,18 @@ static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* mtctx) } } +void ZSTDMT_waitForPendingJobs(ZSTDMT_CCtx* mtctx) +{ + if (mtctx->allJobsCompleted == 0) { /* previous compression not correctly finished */ + ZSTDMT_waitForAllJobsCompleted(mtctx); + ZSTDMT_releaseAllJobResources(mtctx); /* Will set allJobsCompleted to 1 */ + } +} + size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx) { if (mtctx==NULL) return 0; /* compatible with free on NULL */ + ZSTDMT_waitForPendingJobs(mtctx); /* a shared pool is not stopped below */ if (!mtctx->providedFactory) POOL_free(mtctx->factory); /* stop and free worker threads */ ZSTDMT_releaseAllJobResources(mtctx); /* release job resources into pools first */ @@ -1262,17 +1274,13 @@ size_t ZSTDMT_initCStream_internal( assert(!((dict) && (cdict))); /* either dict or cdict, not both */ /* init */ + ZSTDMT_waitForPendingJobs(mtctx); /* before resizing the pools they use */ if (params.nbWorkers != mtctx->params.nbWorkers) FORWARD_IF_ERROR( ZSTDMT_resize(mtctx, (unsigned)params.nbWorkers) , ""); if (params.jobSize != 0 && params.jobSize < ZSTDMT_JOBSIZE_MIN) params.jobSize = ZSTDMT_JOBSIZE_MIN; if (params.jobSize > (size_t)ZSTDMT_JOBSIZE_MAX) params.jobSize = (size_t)ZSTDMT_JOBSIZE_MAX; - if (mtctx->allJobsCompleted == 0) { /* previous compression not correctly finished */ - ZSTDMT_waitForAllJobsCompleted(mtctx); - ZSTDMT_releaseAllJobResources(mtctx); /* Will set allJobsCompleted to 1 */ - } - mtctx->params = params; mtctx->frameContentSize = pledgedSrcSize; ZSTD_freeCDict(mtctx->cdictLocal); @@ -1389,6 +1397,7 @@ static void ZSTDMT_writeLastEmptyBlock(ZSTDMT_jobDescription* job) assert(job->src.size == 0); /* last job is empty -> will be simplified into a last empty block */ assert(job->firstJob == 0); /* cannot be first job, as it also needs to create frame header */ assert(job->dstBuff.start == NULL); /* invoked from streaming variant only (otherwise, dstBuff might be user's output) */ + job->completed = 1; /* written here, never posted to a worker */ job->dstBuff = ZSTDMT_getBuffer(job->bufPool); if (job->dstBuff.start == NULL) { job->cSize = ERROR(memory_allocation); @@ -1422,6 +1431,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* mtctx, size_t srcSize, ZS mtctx->jobs[jobID].prefix = mtctx->inBuff.prefix; mtctx->jobs[jobID].consumed = 0; mtctx->jobs[jobID].cSize = 0; + mtctx->jobs[jobID].completed = 0; mtctx->jobs[jobID].params = mtctx->params; mtctx->jobs[jobID].cdict = mtctx->nextJobID==0 ? mtctx->cdict : NULL; mtctx->jobs[jobID].fullFrameSize = mtctx->frameContentSize; diff --git a/lib/compress/zstdmt_compress.h b/lib/compress/zstdmt_compress.h index 91b489b9cb4..3e2ee664ed0 100644 --- a/lib/compress/zstdmt_compress.h +++ b/lib/compress/zstdmt_compress.h @@ -49,6 +49,12 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbWorkers, ZSTD_threadPool *pool); size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx); +/*! ZSTDMT_waitForPendingJobs() : + * Waits for the jobs of an unfinished frame to complete, then releases their resources. + * Must be called before freeing anything these jobs reference, such as the dictionary. + * Does nothing if the last frame was completed. */ +void ZSTDMT_waitForPendingJobs(ZSTDMT_CCtx* mtctx); + size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx); /* === Streaming functions === */ diff --git a/lib/zstd.h b/lib/zstd.h index 97fef316fd6..3c24acc230d 100644 --- a/lib/zstd.h +++ b/lib/zstd.h @@ -597,6 +597,7 @@ typedef enum { * - The session : will stop compressing current frame, and make CCtx ready to start a new one. * Useful after an error, or to interrupt any ongoing compression. * Any internal data not yet flushed is cancelled. + * With nbWorkers >= 1, it waits for the jobs already started to complete. * Compression parameters and dictionary remain unchanged. * They will be used to compress next frame. * Resetting session never fails. diff --git a/tests/zstreamtest.c b/tests/zstreamtest.c index 88a20a3e270..f671dea4451 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -2512,6 +2512,49 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) } DISPLAYLEVEL(3, "OK \n"); +#ifdef ZSTD_MULTITHREAD + /* Freeing or resetting the context must wait for the jobs still running, + * since they use its dictionary, pools and jobs table */ + DISPLAYLEVEL(3, "test%3i : abandon a multi-threaded frame with a job in progress : ", testNb++); + { size_t const jobSize = 1 MB; + const void* const dict = (const char*)CNBuffer + 2 * jobSize; + int scenario; + for (scenario = 0; scenario < 5; scenario++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx(); + ZSTD_threadPool* const pool = (scenario == 1) ? ZSTD_createThreadPool(1) : NULL; + ZSTD_inBuffer in = { CNBuffer, (scenario == 2) ? 0 : jobSize, 0 }; + ZSTD_outBuffer out = { compressedBuffer, compressedBufferSize, 0 }; + CHECK(cctx == NULL, "ZSTD_createCCtx failed"); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_compressionLevel, 10) ); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_nbWorkers, 1) ); + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_jobSize, (int)jobSize) ); + if (pool != NULL) CHECK_Z( ZSTD_CCtx_refThreadPool(cctx, pool) ); + CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, dict, 1 MB) ); /* makes the first job slow */ + CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue) ); /* starts the first job, if any */ + if (scenario == 2) { + /* empty frame : its only job is empty, and still runs when + * ZSTD_e_end returns because the output buffer is full */ + out.size = out.pos; + CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_end) ); + } + if (scenario == 4) { + /* the second job is prepared, and waits until the worker is free */ + in.size = 2 * jobSize; + CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue) ); + } + if (scenario >= 3) { + /* abandon the frame, change the dictionary, and compress a new frame */ + CHECK_Z( ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only) ); + CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, dict, 512 KB) ); + CHECK_Z( ZSTD_compress2(cctx, compressedBuffer, compressedBufferSize, CNBuffer, jobSize) ); + } + ZSTD_freeCCtx(cctx); /* scenarios 0 to 2 free it with a job in progress */ + ZSTD_freeThreadPool(pool); + } + } + DISPLAYLEVEL(3, "OK \n"); +#endif + _end: FUZ_freeDictionary(dictionary); ZSTD_freeCStream(zc);