From ed26f948cd553ed4f35e681c06bb03be3426f2b8 Mon Sep 17 00:00:00 2001 From: Eshan05 Date: Wed, 28 Oct 2026 21:41:14 +0530 Subject: [PATCH 1/4] compress: wait for an abandoned frame's jobs before freeing what they use With nbWorkers >= 1, a job of an abandoned frame can still be running when the context frees memory it uses: - ZSTD_freeCCtx() frees the dictionary before ZSTDMT_freeCCtx() joins the workers; - ZSTD_CCtx_reset(session_and_parameters), or a session reset followed by loading another dictionary, frees it while the job runs; - with ZSTD_CCtx_refThreadPool(), ZSTDMT_freeCCtx() does not stop the shared pool and frees the jobs table and CCtx pool under the worker; - after a session reset, raising nbWorkers resizes the CCtx pool before the abandoned jobs are waited for. Move the "previous compression not correctly finished" wait of ZSTDMT_initCStream_internal() into ZSTDMT_waitForPendingJobs(), and call it on session reset, at the start of ZSTDMT_freeCCtx(), and before ZSTDMT_resize(). Free the MT context before the dictionaries. Fixes #4862 Signed-off-by: Eshan05 --- lib/compress/zstd_compress.c | 8 ++++-- lib/compress/zstdmt_compress.c | 15 +++++++---- lib/compress/zstdmt_compress.h | 6 +++++ lib/zstd.h | 1 + tests/zstreamtest.c | 47 ++++++++++++++++++++++++++++++++++ 5 files changed, 70 insertions(+), 7 deletions(-) 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..1594221c4e9 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -1043,9 +1043,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 +1271,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); 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..22b038828cd 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -2512,6 +2512,53 @@ 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; + size_t const srcSize = jobSize + 64 KB; /* starts one job, the rest stays buffered */ + size_t const largeDictSize = 1 MB; + const void* const largeDict = (const char*)CNBuffer + srcSize; + int scenario; + for (scenario = 0; scenario < 4; scenario++) { + ZSTD_CCtx* const cctx = ZSTD_createCCtx(); + ZSTD_threadPool* const pool = (scenario == 1) ? ZSTD_createThreadPool(1) : NULL; + ZSTD_inBuffer in = { CNBuffer, srcSize, 0 }; + ZSTD_outBuffer out = { compressedBuffer, compressedBufferSize, 0 }; + size_t nextDictSize = largeDictSize; + 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, largeDict, largeDictSize) ); + CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue) ); + CHECK(in.pos < jobSize, "the first job should be in progress"); + if (scenario >= 2) { + /* abandon the frame, then change the dictionary or the number of workers */ + CHECK_Z( ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only) ); + if (scenario == 2) { + nextDictSize = largeDictSize / 2; + CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, largeDict, nextDictSize) ); + } else { + CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_nbWorkers, 4) ); + } + cSize = ZSTD_compress2(cctx, compressedBuffer, compressedBufferSize, CNBuffer, srcSize); + CHECK_Z(cSize); + { size_t const dSize = ZSTD_decompress_usingDict(zd, decodedBuffer, decodedBufferSize, + compressedBuffer, cSize, largeDict, nextDictSize); + CHECK_Z(dSize); + CHECK(dSize != srcSize, "wrong decompressed size"); + CHECK(memcmp(decodedBuffer, CNBuffer, srcSize), "decompressed data must match"); + } } + ZSTD_freeCCtx(cctx); /* scenarios 0 and 1 free it with the job in progress */ + ZSTD_freeThreadPool(pool); + } + } + DISPLAYLEVEL(3, "OK \n"); +#endif + _end: FUZ_freeDictionary(dictionary); ZSTD_freeCStream(zc); From 54ade3a78b1fb2ebabfeff69012b8aafce75bb54 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 10 Nov 2026 10:02:04 -0800 Subject: [PATCH 2/4] compress: wait for an empty job before releasing it ZSTDMT_waitForAllJobsCompleted() waited while consumed < src.size. An empty job, the only job of an empty frame, meets that condition before its worker starts, so the wait returned at once. The previous commit calls ZSTDMT_waitForPendingJobs() at the start of ZSTDMT_freeCCtx(), before POOL_free() joins the workers. When an empty frame was abandoned (ZSTD_e_end returned because the output buffer was full), freeing the context cleared the job while its worker started, and the worker dereferenced a NULL pool. Before that commit, POOL_free() joined the worker first. The worker now sets a completed flag when it no longer uses the job, and the wait uses this flag. The test gets a scenario for this case. --- lib/compress/zstdmt_compress.c | 8 ++++++-- tests/zstreamtest.c | 22 ++++++++++++++++------ 2 files changed, 22 insertions(+), 8 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 1594221c4e9..272fd5d27b7 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); } @@ -1034,7 +1036,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); } @@ -1394,6 +1396,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); @@ -1427,6 +1430,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/tests/zstreamtest.c b/tests/zstreamtest.c index 22b038828cd..82d5ebf3307 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -2521,10 +2521,10 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) size_t const largeDictSize = 1 MB; const void* const largeDict = (const char*)CNBuffer + srcSize; int scenario; - for (scenario = 0; scenario < 4; 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, srcSize, 0 }; + ZSTD_inBuffer in = { CNBuffer, (scenario == 2) ? 0 : srcSize, 0 }; ZSTD_outBuffer out = { compressedBuffer, compressedBufferSize, 0 }; size_t nextDictSize = largeDictSize; CHECK(cctx == NULL, "ZSTD_createCCtx failed"); @@ -2534,11 +2534,21 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) if (pool != NULL) CHECK_Z( ZSTD_CCtx_refThreadPool(cctx, pool) ); CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, largeDict, largeDictSize) ); CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue) ); - CHECK(in.pos < jobSize, "the first job should be in progress"); - if (scenario >= 2) { + if (scenario == 2) { + /* empty frame : its only job is empty, and is still in progress + * when ZSTD_e_end returns because the output buffer is full */ + size_t remaining; + out.size = out.pos; + remaining = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_end); + CHECK_Z(remaining); + CHECK(remaining == 0, "the empty frame should not be completed"); + } else { + CHECK(in.pos < jobSize, "the first job should be in progress"); + } + if (scenario >= 3) { /* abandon the frame, then change the dictionary or the number of workers */ CHECK_Z( ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only) ); - if (scenario == 2) { + if (scenario == 3) { nextDictSize = largeDictSize / 2; CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, largeDict, nextDictSize) ); } else { @@ -2552,7 +2562,7 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) CHECK(dSize != srcSize, "wrong decompressed size"); CHECK(memcmp(decodedBuffer, CNBuffer, srcSize), "decompressed data must match"); } } - ZSTD_freeCCtx(cctx); /* scenarios 0 and 1 free it with the job in progress */ + ZSTD_freeCCtx(cctx); /* scenarios 0 to 2 free it with the job in progress */ ZSTD_freeThreadPool(pool); } } From a8af53bb32ba69a4d17f24e58f4459c64235e1a6 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 10 Nov 2026 10:04:58 -0800 Subject: [PATCH 3/4] compress: clear jobReady when the jobs are released ZSTDMT_releaseAllJobResources() clears every job descriptor, including the job prepared while no worker was free (jobReady), but it left jobReady set. After a frame was abandoned in that state, for example with ZSTD_CCtx_reset(), the first ZSTDMT_createCompressionJob() of the next frame posted the cleared descriptor, and the worker dereferenced a NULL pool. The test gets a scenario for this case. --- lib/compress/zstdmt_compress.c | 1 + tests/zstreamtest.c | 11 ++++++++--- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 272fd5d27b7..7118bbe7f4a 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -1027,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; } diff --git a/tests/zstreamtest.c b/tests/zstreamtest.c index 82d5ebf3307..b7360327ef5 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -2521,7 +2521,7 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) size_t const largeDictSize = 1 MB; const void* const largeDict = (const char*)CNBuffer + srcSize; int scenario; - for (scenario = 0; scenario < 5; scenario++) { + for (scenario = 0; scenario < 6; scenario++) { ZSTD_CCtx* const cctx = ZSTD_createCCtx(); ZSTD_threadPool* const pool = (scenario == 1) ? ZSTD_createThreadPool(1) : NULL; ZSTD_inBuffer in = { CNBuffer, (scenario == 2) ? 0 : srcSize, 0 }; @@ -2545,13 +2545,18 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) } else { CHECK(in.pos < jobSize, "the first job should be in progress"); } + if (scenario == 5) { + /* the next 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, then change the dictionary or the number of workers */ + /* abandon the frame, then change the dictionary or the number of workers, or nothing */ CHECK_Z( ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only) ); if (scenario == 3) { nextDictSize = largeDictSize / 2; CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, largeDict, nextDictSize) ); - } else { + } else if (scenario == 4) { CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_nbWorkers, 4) ); } cSize = ZSTD_compress2(cctx, compressedBuffer, compressedBufferSize, CNBuffer, srcSize); From 8136851284d424518e98cfdf1f4b370e73094c32 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 10 Nov 2026 10:57:16 -0800 Subject: [PATCH 4/4] tests: shorten the abandoned multi-threaded frame test Remove the scenario that raises nbWorkers after a reset. Since ZSTD_CCtx_reset() waits for the jobs, this scenario can only fail when the one that loads a new dictionary after a reset fails too. The new frames are compressed, but no longer decompressed: the bugs show as memory errors in the worker. Each of the five scenarios still fails when its library change is removed. --- tests/zstreamtest.c | 49 ++++++++++++++------------------------------- 1 file changed, 15 insertions(+), 34 deletions(-) diff --git a/tests/zstreamtest.c b/tests/zstreamtest.c index b7360327ef5..f671dea4451 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -2517,57 +2517,38 @@ static int basicUnitTests(U32 seed, double compressibility, int bigTests) * 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; - size_t const srcSize = jobSize + 64 KB; /* starts one job, the rest stays buffered */ - size_t const largeDictSize = 1 MB; - const void* const largeDict = (const char*)CNBuffer + srcSize; + const void* const dict = (const char*)CNBuffer + 2 * jobSize; int scenario; - for (scenario = 0; scenario < 6; 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 : srcSize, 0 }; + ZSTD_inBuffer in = { CNBuffer, (scenario == 2) ? 0 : jobSize, 0 }; ZSTD_outBuffer out = { compressedBuffer, compressedBufferSize, 0 }; - size_t nextDictSize = largeDictSize; 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, largeDict, largeDictSize) ); - CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_continue) ); + 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 is still in progress - * when ZSTD_e_end returns because the output buffer is full */ - size_t remaining; + /* 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; - remaining = ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_end); - CHECK_Z(remaining); - CHECK(remaining == 0, "the empty frame should not be completed"); - } else { - CHECK(in.pos < jobSize, "the first job should be in progress"); + CHECK_Z( ZSTD_compressStream2(cctx, &out, &in, ZSTD_e_end) ); } - if (scenario == 5) { - /* the next job is prepared, and waits until the worker is free */ + 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, then change the dictionary or the number of workers, or nothing */ + /* abandon the frame, change the dictionary, and compress a new frame */ CHECK_Z( ZSTD_CCtx_reset(cctx, ZSTD_reset_session_only) ); - if (scenario == 3) { - nextDictSize = largeDictSize / 2; - CHECK_Z( ZSTD_CCtx_loadDictionary(cctx, largeDict, nextDictSize) ); - } else if (scenario == 4) { - CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_c_nbWorkers, 4) ); - } - cSize = ZSTD_compress2(cctx, compressedBuffer, compressedBufferSize, CNBuffer, srcSize); - CHECK_Z(cSize); - { size_t const dSize = ZSTD_decompress_usingDict(zd, decodedBuffer, decodedBufferSize, - compressedBuffer, cSize, largeDict, nextDictSize); - CHECK_Z(dSize); - CHECK(dSize != srcSize, "wrong decompressed size"); - CHECK(memcmp(decodedBuffer, CNBuffer, srcSize), "decompressed data must match"); - } } - ZSTD_freeCCtx(cctx); /* scenarios 0 to 2 free it with the job in progress */ + 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); } }