From 1c714fda3fe0932faff5c1a535a2f4a597b8d0bb Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Mon, 18 Jun 2018 20:46:39 -0700 Subject: [PATCH 01/11] introduced POOL_resize() not complete yet : finalize behavior in case of unfinished expansion --- lib/common/pool.c | 89 +++++++++++++++++++++++++++++++++++++---------- lib/common/pool.h | 37 ++++++++++---------- 2 files changed, 88 insertions(+), 38 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index 773488b07..6795f25eb 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -33,8 +33,9 @@ typedef struct POOL_job_s { struct POOL_ctx_s { ZSTD_customMem customMem; /* Keep track of the threads */ - ZSTD_pthread_t *threads; - size_t numThreads; + ZSTD_pthread_t* threads; + size_t threadCapacity; + size_t threadLimit; /* The queue is a circular buffer */ POOL_job *queue; @@ -58,10 +59,10 @@ struct POOL_ctx_s { }; /* POOL_thread() : - Work thread for the thread pool. - Waits for jobs and executes them. - @returns : NULL on failure else non-null. -*/ + * Work thread for the thread pool. + * Waits for jobs and executes them. + * @returns : NULL on failure else non-null. + */ static void* POOL_thread(void* opaque) { POOL_ctx* const ctx = (POOL_ctx*)opaque; if (!ctx) { return NULL; } @@ -103,16 +104,17 @@ POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) { return POOL_create_advanced(numThreads, queueSize, ZSTD_defaultCMem); } -POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, ZSTD_customMem customMem) { +POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, + ZSTD_customMem customMem) { POOL_ctx* ctx; - /* Check the parameters */ + /* Check parameters */ if (!numThreads) { return NULL; } /* Allocate the context and zero initialize */ ctx = (POOL_ctx*)ZSTD_calloc(sizeof(POOL_ctx), customMem); if (!ctx) { return NULL; } /* Initialize the job queue. - * It needs one extra space since one space is wasted to differentiate empty - * and full queues. + * It needs one extra space since one space is wasted to differentiate + * empty and full queues. */ ctx->queueSize = queueSize + 1; ctx->queue = (POOL_job*)ZSTD_malloc(ctx->queueSize * sizeof(POOL_job), customMem); @@ -126,7 +128,7 @@ POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, ZSTD_customM ctx->shutdown = 0; /* Allocate space for the thread handles */ ctx->threads = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), customMem); - ctx->numThreads = 0; + ctx->threadCapacity = 0; ctx->customMem = customMem; /* Check for errors */ if (!ctx->threads || !ctx->queue) { POOL_free(ctx); return NULL; } @@ -134,11 +136,12 @@ POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, ZSTD_customM { size_t i; for (i = 0; i < numThreads; ++i) { if (ZSTD_pthread_create(&ctx->threads[i], NULL, &POOL_thread, ctx)) { - ctx->numThreads = i; + ctx->threadCapacity = i; POOL_free(ctx); return NULL; } } - ctx->numThreads = numThreads; + ctx->threadCapacity = numThreads; + ctx->threadLimit = numThreads; } return ctx; } @@ -156,8 +159,8 @@ static void POOL_join(POOL_ctx* ctx) { ZSTD_pthread_cond_broadcast(&ctx->queuePopCond); /* Join all of the threads */ { size_t i; - for (i = 0; i < ctx->numThreads; ++i) { - ZSTD_pthread_join(ctx->threads[i], NULL); + for (i = 0; i < ctx->threadCapacity; ++i) { + ZSTD_pthread_join(ctx->threads[i], NULL); /* note : could fail */ } } } @@ -172,24 +175,72 @@ void POOL_free(POOL_ctx *ctx) { ZSTD_free(ctx, ctx->customMem); } + + size_t POOL_sizeof(POOL_ctx *ctx) { if (ctx==NULL) return 0; /* supports sizeof NULL */ return sizeof(*ctx) + ctx->queueSize * sizeof(POOL_job) - + ctx->numThreads * sizeof(ZSTD_pthread_t); + + ctx->threadCapacity * sizeof(ZSTD_pthread_t); +} + + +/* note : only works if no job is running ! + * return : 1 on success, 0 on failure */ +static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) +{ + if (ctx->numThreadsBusy > 0) return 0; + if (numThreads <= ctx->threadCapacity) { + ctx->threadLimit = numThreads; + return 1; + } + /* numThreads > threadCapacity */ + { ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem); + if (!threadPool) return 0; + /* Initialize additional threads */ + { size_t threadId; + for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) { + if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) { + break; + } } + if (threadId != numThreads) { /* not all threads successfully init */ + /* how to destroy existing threads ? */ + /* POOL_join destroy all existing threads, not just newly created ones */ + return 0; + } + } + /* replace existing thread pool */ + memcpy(threadPool, ctx->threads, ctx->threadCapacity); + ZSTD_free(ctx->threads, ctx->customMem); + ctx->threads = threadPool; + } + ctx->threadCapacity = numThreads; + ctx->threadLimit = numThreads; + return 1; +} + +/* return : 1 on success, 0 on failure */ +int POOL_resize(POOL_ctx* ctx, size_t numThreads) +{ + int result; + if (!ctx) return 0; + ZSTD_pthread_mutex_lock(&ctx->queueMutex); + result = POOL_resize_internal(ctx, numThreads); + ZSTD_pthread_mutex_unlock(&ctx->queueMutex); + return result; } /** * Returns 1 if the queue is full and 0 otherwise. * - * If the queueSize is 1 (the pool was created with an intended queueSize of 0), - * then a queue is empty if there is a thread free and no job is waiting. + * When queueSize is 1 (pool was created with an intended queueSize of 0), + * then a queue is empty if there is a thread free _and_ no job is waiting. */ static int isQueueFull(POOL_ctx const* ctx) { if (ctx->queueSize > 1) { return ctx->queueHead == ((ctx->queueTail + 1) % ctx->queueSize); } else { - return ctx->numThreadsBusy == ctx->numThreads || + return ctx->numThreadsBusy == ctx->threadLimit || !ctx->queueEmpty; } } diff --git a/lib/common/pool.h b/lib/common/pool.h index a57e9b4fa..dba28859c 100644 --- a/lib/common/pool.h +++ b/lib/common/pool.h @@ -30,40 +30,39 @@ typedef struct POOL_ctx_s POOL_ctx; */ POOL_ctx* POOL_create(size_t numThreads, size_t queueSize); -POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, ZSTD_customMem customMem); +POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, + ZSTD_customMem customMem); /*! POOL_free() : - Free a thread pool returned by POOL_create(). -*/ + * Free a thread pool returned by POOL_create(). + */ void POOL_free(POOL_ctx* ctx); /*! POOL_sizeof() : - return memory usage of pool returned by POOL_create(). -*/ + * @return threadpool memory usage + * note : compatible with NULL (returns 0 in this case) + */ size_t POOL_sizeof(POOL_ctx* ctx); /*! POOL_function : - The function type that can be added to a thread pool. -*/ + * The function type that can be added to a thread pool. + */ typedef void (*POOL_function)(void*); -/*! POOL_add_function : - The function type for a generic thread pool add function. -*/ -typedef void (*POOL_add_function)(void*, POOL_function, void*); /*! POOL_add() : - Add the job `function(opaque)` to the thread pool. `ctx` must be valid. - Possibly blocks until there is room in the queue. - Note : The function may be executed asynchronously, so `opaque` must live until the function has been completed. -*/ + * Add the job `function(opaque)` to the thread pool. `ctx` must be valid. + * Possibly blocks until there is room in the queue. + * Note : The function may be executed asynchronously, + * therefore, `opaque` must live until function has been completed. + */ void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque); /*! POOL_tryAdd() : - Add the job `function(opaque)` to the thread pool if a worker is available. - return immediately otherwise. - @return : 1 if successful, 0 if not. -*/ + * Add the job `function(opaque)` to thread pool _if_ a worker is available. + * Returns immediately even if not (does not block). + * @return : 1 if successful, 0 if not. + */ int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque); From 4567c571994428ce19710b9b03ff3b28cfaa7c38 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 19 Jun 2018 16:03:12 -0700 Subject: [PATCH 02/11] finalized POOL_resize() POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) The function may fail, and returns a NULL pointer in this case. --- lib/common/pool.c | 50 +++++++++++++++++++++++++---------------------- lib/common/pool.h | 10 ++++++++++ 2 files changed, 37 insertions(+), 23 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index 6795f25eb..e64833f87 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -185,18 +185,21 @@ size_t POOL_sizeof(POOL_ctx *ctx) { } -/* note : only works if no job is running ! - * return : 1 on success, 0 on failure */ -static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) +/* note : only works if no job is running ! */ +static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) { - if (ctx->numThreadsBusy > 0) return 0; + if (ctx->numThreadsBusy > 0) return NULL; if (numThreads <= ctx->threadCapacity) { ctx->threadLimit = numThreads; - return 1; + return ctx; } /* numThreads > threadCapacity */ { ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem); - if (!threadPool) return 0; + if (!threadPool) return NULL; + /* replace existing thread pool */ + memcpy(threadPool, ctx->threads, ctx->threadCapacity); + ZSTD_free(ctx->threads, ctx->customMem); + ctx->threads = threadPool; /* Initialize additional threads */ { size_t threadId; for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) { @@ -204,30 +207,26 @@ static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) break; } } if (threadId != numThreads) { /* not all threads successfully init */ - /* how to destroy existing threads ? */ - /* POOL_join destroy all existing threads, not just newly created ones */ - return 0; - } - } - /* replace existing thread pool */ - memcpy(threadPool, ctx->threads, ctx->threadCapacity); - ZSTD_free(ctx->threads, ctx->customMem); - ctx->threads = threadPool; - } + ctx->threadCapacity = threadId; + return NULL; /* will release the pool */ + } } } + /* successfully expanded */ ctx->threadCapacity = numThreads; ctx->threadLimit = numThreads; - return 1; + return ctx; } -/* return : 1 on success, 0 on failure */ -int POOL_resize(POOL_ctx* ctx, size_t numThreads) +/* @return : a working pool on success, NULL on failure + * note : starting context is considered consumed. */ +POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) { - int result; - if (!ctx) return 0; + POOL_ctx* newCtx; + if (ctx==NULL) return NULL; ZSTD_pthread_mutex_lock(&ctx->queueMutex); - result = POOL_resize_internal(ctx, numThreads); + newCtx = POOL_resize_internal(ctx, numThreads); ZSTD_pthread_mutex_unlock(&ctx->queueMutex); - return result; + if (newCtx!=ctx) POOL_free(ctx); + return newCtx; } /** @@ -314,6 +313,11 @@ void POOL_free(POOL_ctx* ctx) { (void)ctx; } +POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) { + (void)numThreads; + return ctx; +} + void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque) { (void)ctx; function(opaque); diff --git a/lib/common/pool.h b/lib/common/pool.h index dba28859c..00ea76631 100644 --- a/lib/common/pool.h +++ b/lib/common/pool.h @@ -38,6 +38,16 @@ POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, */ void POOL_free(POOL_ctx* ctx); +/*! POOL_resize() : + * Expands or shrinks pool's number of threads. + * This is more efficient than releasing and creating a new context. + * @return : a new pool context on success, NULL on failure + * note : new pool context might have same address as original one, but it's not guaranteed. + * consider starting context as consumed, only rely on returned one. + * note 2 : only numThreads can be resized, queueSize is unchanged. + */ +POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads); + /*! POOL_sizeof() : * @return threadpool memory usage * note : compatible with NULL (returns 0 in this case) From 066fbbfe1c8f849420ff9f10711bd1d11b8ce15b Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 19 Jun 2018 17:28:56 -0700 Subject: [PATCH 03/11] make zstdmt resize its context when nbThreads change. Technically, it only expands. But when instructed to use less threads, the thread pool will limit nb of concurrent threads. --- lib/compress/zstd_compress.c | 6 +-- lib/compress/zstdmt_compress.c | 98 ++++++++++++++++++++++++++-------- lib/compress/zstdmt_compress.h | 5 -- 3 files changed, 77 insertions(+), 32 deletions(-) diff --git a/lib/compress/zstd_compress.c b/lib/compress/zstd_compress.c index c071c5738..d268cef08 100644 --- a/lib/compress/zstd_compress.c +++ b/lib/compress/zstd_compress.c @@ -3645,13 +3645,9 @@ size_t ZSTD_compress_generic (ZSTD_CCtx* cctx, } if (params.nbWorkers > 0) { /* mt context creation */ - if (cctx->mtctx == NULL || (params.nbWorkers != ZSTDMT_getNbWorkers(cctx->mtctx))) { + if (cctx->mtctx == NULL) { DEBUGLOG(4, "ZSTD_compress_generic: creating new mtctx for nbWorkers=%u", params.nbWorkers); - if (cctx->mtctx != NULL) - DEBUGLOG(4, "ZSTD_compress_generic: previous nbWorkers was %u", - ZSTDMT_getNbWorkers(cctx->mtctx)); - ZSTDMT_freeCCtx(cctx->mtctx); cctx->mtctx = ZSTDMT_createCCtx_advanced(params.nbWorkers, cctx->customMem); if (cctx->mtctx == NULL) return ERROR(memory_allocation); } diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 0c59c6a72..4a2a4cdd5 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -159,6 +159,25 @@ static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* const bufPool, size_t const ZSTD_pthread_mutex_unlock(&bufPool->poolMutex); } + +static ZSTDMT_bufferPool* ZSTDMT_expandBufferPool(ZSTDMT_bufferPool* srcBufPool, U32 nbWorkers) +{ + unsigned const maxNbBuffers = 2*nbWorkers + 3; + if (srcBufPool==NULL) return NULL; + if (srcBufPool->totalBuffers >= maxNbBuffers) /* good enough */ + return srcBufPool; + /* need a larger buffer pool */ + { ZSTD_customMem const cMem = srcBufPool->cMem; + size_t const bSize = srcBufPool->bufferSize; /* forward parameters */ + ZSTDMT_bufferPool* newBufPool; + ZSTDMT_freeBufferPool(srcBufPool); + newBufPool = ZSTDMT_createBufferPool(nbWorkers, cMem); + if (newBufPool==NULL) return newBufPool; + ZSTDMT_setBufferSize(newBufPool, bSize); + return newBufPool; + } +} + /** ZSTDMT_getBuffer() : * assumption : bufPool must be valid * @return : a buffer, with start pointer and size @@ -309,6 +328,10 @@ static void ZSTDMT_freeSeqPool(ZSTDMT_seqPool* seqPool) ZSTDMT_freeBufferPool(seqPool); } +static ZSTDMT_seqPool* ZSTDMT_expandSeqPool(ZSTDMT_seqPool* pool, U32 nbWorkers) +{ + return ZSTDMT_expandBufferPool(pool, nbWorkers); +} /* ===== CCtx Pool ===== */ @@ -354,6 +377,18 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbWorkers, return cctxPool; } +static ZSTDMT_CCtxPool* ZSTDMT_expandCCtxPool(ZSTDMT_CCtxPool* srcPool, + unsigned nbWorkers) +{ + if (srcPool==NULL) return NULL; + if (nbWorkers <= srcPool->totalCCtx) return srcPool; /* good enough */ + /* need a larger cctx pool */ + { ZSTD_customMem const cMem = srcPool->cMem; + ZSTDMT_freeCCtxPool(srcPool); + return ZSTDMT_createCCtxPool(nbWorkers, cMem); + } +} + /* only works during initialization phase, not during compression */ static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool) { @@ -745,9 +780,9 @@ struct ZSTDMT_CCtx_s { ZSTD_CCtx_params params; size_t targetSectionSize; size_t targetPrefixSize; - roundBuff_t roundBuff; + int jobReady; /* 1 => one job is already prepared, but pool has shortage of workers. Don't create a new job. */ inBuff_t inBuff; - int jobReady; /* 1 => one job is already prepared, but pool has shortage of workers. Don't create another one. */ + roundBuff_t roundBuff; serialState_t serial; unsigned singleBlockingThread; unsigned jobIDMask; @@ -798,6 +833,20 @@ static ZSTDMT_jobDescription* ZSTDMT_createJobsTable(U32* nbJobsPtr, ZSTD_custom return jobTable; } +static size_t ZSTDMT_expandJobsTable (ZSTDMT_CCtx* mtctx, U32 nbWorkers) { + U32 nbJobs = nbWorkers + 2; + if (nbJobs > mtctx->jobIDMask+1) { /* need more job capacity */ + ZSTDMT_freeJobsTable(mtctx->jobs, mtctx->jobIDMask+1, mtctx->cMem); + mtctx->jobIDMask = 0; + mtctx->jobs = ZSTDMT_createJobsTable(&nbJobs, mtctx->cMem); + if (mtctx->jobs==NULL) return ERROR(memory_allocation); + assert((nbJobs != 0) && ((nbJobs & (nbJobs - 1)) == 0)); /* ensure nbJobs is a power of 2 */ + mtctx->jobIDMask = nbJobs - 1; + } + return 0; +} + + /* ZSTDMT_CCtxParam_setNbWorkers(): * Internal use only */ size_t ZSTDMT_CCtxParam_setNbWorkers(ZSTD_CCtx_params* params, unsigned nbWorkers) @@ -964,6 +1013,25 @@ static ZSTD_CCtx_params ZSTDMT_initJobCCtxParams(ZSTD_CCtx_params const params) return jobParams; } + +/* ZSTDMT_resize() : + * @return : error code if fails, 0 on success */ +static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers) +{ + mtctx->factory = POOL_resize(mtctx->factory, nbWorkers); + if (mtctx->factory == NULL) return ERROR(memory_allocation); + CHECK_F( ZSTDMT_expandJobsTable(mtctx, nbWorkers) ); + mtctx->bufPool = ZSTDMT_expandBufferPool(mtctx->bufPool, nbWorkers); + if (mtctx->bufPool == NULL) return ERROR(memory_allocation); + mtctx->cctxPool = ZSTDMT_expandCCtxPool(mtctx->cctxPool, nbWorkers); + if (mtctx->cctxPool == NULL) return ERROR(memory_allocation); + mtctx->seqPool = ZSTDMT_expandSeqPool(mtctx->seqPool, nbWorkers); + if (mtctx->seqPool == NULL) return ERROR(memory_allocation); + ZSTDMT_CCtxParam_setNbWorkers(&mtctx->params, nbWorkers); + return 0; +} + + /*! ZSTDMT_updateCParams_whileCompressing() : * Updates only a selected set of compression parameters, to remain compatible with current frame. * New parameters will be applied to next compression job. */ @@ -980,15 +1048,6 @@ void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_p } } -/* ZSTDMT_getNbWorkers(): - * @return nb threads currently active in mtctx. - * mtctx must be valid */ -unsigned ZSTDMT_getNbWorkers(const ZSTDMT_CCtx* mtctx) -{ - assert(mtctx != NULL); - return mtctx->params.nbWorkers; -} - /* ZSTDMT_getFrameProgression(): * tells how much data has been consumed (input) and produced (output) for current frame. * able to count progression inside worker threads. @@ -1089,15 +1148,7 @@ static size_t ZSTDMT_compress_advanced_internal( if (ZSTDMT_serialState_reset(&mtctx->serial, mtctx->seqPool, params)) return ERROR(memory_allocation); - if (nbJobs > mtctx->jobIDMask+1) { /* enlarge job table */ - U32 jobsTableSize = nbJobs; - ZSTDMT_freeJobsTable(mtctx->jobs, mtctx->jobIDMask+1, mtctx->cMem); - mtctx->jobIDMask = 0; - mtctx->jobs = ZSTDMT_createJobsTable(&jobsTableSize, mtctx->cMem); - if (mtctx->jobs==NULL) return ERROR(memory_allocation); - assert((jobsTableSize != 0) && ((jobsTableSize & (jobsTableSize - 1)) == 0)); /* ensure jobsTableSize is a power of 2 */ - mtctx->jobIDMask = jobsTableSize - 1; - } + CHECK_F( ZSTDMT_expandJobsTable(mtctx, nbJobs) ); /* only expands if necessary */ { unsigned u; for (u=0; ucctxPool->totalCCtx); - /* params are supposed to be fully validated at this point */ + + /* params supposed partially fully validated at this point */ assert(!ZSTD_isError(ZSTD_checkCParams(params.cParams))); assert(!((dict) && (cdict))); /* either dict or cdict, not both */ - assert(mtctx->cctxPool->totalCCtx == params.nbWorkers); /* init */ + if (params.nbWorkers != mtctx->params.nbWorkers) + ZSTDMT_resize(mtctx, params.nbWorkers); + if (params.jobSize == 0) { params.jobSize = 1U << ZSTDMT_computeTargetJobLog(params); } diff --git a/lib/compress/zstdmt_compress.h b/lib/compress/zstdmt_compress.h index f79e3b441..4249a82de 100644 --- a/lib/compress/zstdmt_compress.h +++ b/lib/compress/zstdmt_compress.h @@ -126,11 +126,6 @@ size_t ZSTDMT_CCtxParam_setNbWorkers(ZSTD_CCtx_params* params, unsigned nbWorker * New parameters will be applied to next compression job. */ void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_params* cctxParams); -/* ZSTDMT_getNbWorkers(): - * @return nb threads currently active in mtctx. - * mtctx must be valid */ -unsigned ZSTDMT_getNbWorkers(const ZSTDMT_CCtx* mtctx); - /* ZSTDMT_getFrameProgression(): * tells how much data has been consumed (input) and produced (output) for current frame. * able to count progression inside worker threads. From 166901dc726691e31de143b9f570693256a1c953 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 19 Jun 2018 18:07:18 -0700 Subject: [PATCH 04/11] reduced POOL_resize() restriction It's not necessary to ensure that no job is ongoing. The pool is only expanded, existing threads are preserved. In case of error, the only option is to return NULL and terminate the thread pool anyway. --- lib/common/pool.c | 1 - lib/compress/zstdmt_compress.c | 2 +- lib/decompress/zstd_decompress.c | 4 ---- 3 files changed, 1 insertion(+), 6 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index e64833f87..a0e2bedf0 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -188,7 +188,6 @@ size_t POOL_sizeof(POOL_ctx *ctx) { /* note : only works if no job is running ! */ static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) { - if (ctx->numThreadsBusy > 0) return NULL; if (numThreads <= ctx->threadCapacity) { ctx->threadLimit = numThreads; return ctx; diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index 4a2a4cdd5..bbbdc5cd4 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -1280,7 +1280,7 @@ size_t ZSTDMT_initCStream_internal( /* init */ if (params.nbWorkers != mtctx->params.nbWorkers) - ZSTDMT_resize(mtctx, params.nbWorkers); + CHECK_F( ZSTDMT_resize(mtctx, params.nbWorkers) ); if (params.jobSize == 0) { params.jobSize = 1U << ZSTDMT_computeTargetJobLog(params); diff --git a/lib/decompress/zstd_decompress.c b/lib/decompress/zstd_decompress.c index 148c87f9d..4de98a8e6 100644 --- a/lib/decompress/zstd_decompress.c +++ b/lib/decompress/zstd_decompress.c @@ -1108,15 +1108,11 @@ size_t ZSTD_execSequence(BYTE* op, return sequenceLength; } /* span extDict & currentPrefixSegment */ - DEBUGLOG(2, "ZSTD_execSequence: found a 2-segments match") { size_t const length1 = dictEnd - match; - DEBUGLOG(2, "first part (extDict) is %zu bytes long", length1); memmove(oLitEnd, match, length1); op = oLitEnd + length1; sequence.matchLength -= length1; - DEBUGLOG(2, "second part (prefix) is %zu bytes long", sequence.matchLength); match = prefixStart; - DEBUGLOG(2, "first byte of 2nd part : %02X", *prefixStart); if (op > oend_w || sequence.matchLength < MINMATCH) { U32 i; for (i = 0; i < sequence.matchLength; ++i) op[i] = match[i]; From 62469c9f41747ba78a8c616d84d7b36ffc2b7b41 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Tue, 19 Jun 2018 20:14:03 -0700 Subject: [PATCH 05/11] fixed wrong size in pthread struct transfer --- lib/common/pool.c | 11 +++++------ tests/zstreamtest.c | 6 +++--- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index a0e2bedf0..7f9219e36 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -196,19 +196,18 @@ static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) { ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem); if (!threadPool) return NULL; /* replace existing thread pool */ - memcpy(threadPool, ctx->threads, ctx->threadCapacity); + memcpy(threadPool, ctx->threads, ctx->threadCapacity * sizeof(ctx->threads[0])); ZSTD_free(ctx->threads, ctx->customMem); ctx->threads = threadPool; /* Initialize additional threads */ { size_t threadId; for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) { if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) { - break; + ctx->threadCapacity = threadId; + ctx->threadLimit = threadId; + return NULL; /* will release the pool */ } } - if (threadId != numThreads) { /* not all threads successfully init */ - ctx->threadCapacity = threadId; - return NULL; /* will release the pool */ - } } } + } } /* successfully expanded */ ctx->threadCapacity = numThreads; ctx->threadLimit = numThreads; diff --git a/tests/zstreamtest.c b/tests/zstreamtest.c index 425fbab39..fba99aa34 100644 --- a/tests/zstreamtest.c +++ b/tests/zstreamtest.c @@ -1643,13 +1643,13 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double DISPLAYLEVEL(5, "Creating new context \n"); ZSTD_freeCCtx(zc); zc = ZSTD_createCCtx(); - CHECK(zc==NULL, "ZSTD_createCCtx allocation error"); - resetAllowed=0; + CHECK(zc == NULL, "ZSTD_createCCtx allocation error"); + resetAllowed = 0; } if ((FUZ_rand(&lseed) & 0xFF) == 132) { ZSTD_freeDStream(zd); zd = ZSTD_createDStream(); - CHECK(zd==NULL, "ZSTD_createDStream allocation error"); + CHECK(zd == NULL, "ZSTD_createDStream allocation error"); ZSTD_initDStream_usingDict(zd, NULL, 0); /* ensure at least one init */ } From 6b48eb12c04faa34e6c950c0206b149edcc1f828 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Wed, 20 Jun 2018 14:35:39 -0700 Subject: [PATCH 06/11] change control of threadLimit now limits maximum nb of active threads even when queueSize > 1. --- lib/common/pool.c | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index 7f9219e36..ca5a38ee5 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -10,9 +10,10 @@ /* ====== Dependencies ======= */ -#include /* size_t */ -#include "pool.h" +#include /* size_t */ +#include "debug.h" /* assert */ #include "zstd_internal.h" /* ZSTD_malloc, ZSTD_free */ +#include "pool.h" /* ====== Compiler specifics ====== */ #if defined(_MSC_VER) @@ -70,14 +71,14 @@ static void* POOL_thread(void* opaque) { /* Lock the mutex and wait for a non-empty queue or until shutdown */ ZSTD_pthread_mutex_lock(&ctx->queueMutex); - while (ctx->queueEmpty && !ctx->shutdown) { + while ( ctx->queueEmpty + || (ctx->numThreadsBusy >= ctx->threadLimit) ) { + if (ctx->shutdown) { + ZSTD_pthread_mutex_unlock(&ctx->queueMutex); + return opaque; + } ZSTD_pthread_cond_wait(&ctx->queuePopCond, &ctx->queueMutex); } - /* empty => shutting down: so stop */ - if (ctx->queueEmpty) { - ZSTD_pthread_mutex_unlock(&ctx->queueMutex); - return opaque; - } /* Pop a job off the queue */ { POOL_job const job = ctx->queue[ctx->queueHead]; ctx->queueHead = (ctx->queueHead + 1) % ctx->queueSize; @@ -97,7 +98,7 @@ static void* POOL_thread(void* opaque) { ZSTD_pthread_cond_signal(&ctx->queuePushCond); } } } /* for (;;) */ - /* Unreachable */ + assert(0); /* Unreachable */ } POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) { @@ -237,7 +238,7 @@ static int isQueueFull(POOL_ctx const* ctx) { if (ctx->queueSize > 1) { return ctx->queueHead == ((ctx->queueTail + 1) % ctx->queueSize); } else { - return ctx->numThreadsBusy == ctx->threadLimit || + return (ctx->numThreadsBusy == ctx->threadLimit) || !ctx->queueEmpty; } } From 6de249c1c61e19071af1b28d09f39b41484772d4 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Wed, 20 Jun 2018 17:18:57 -0700 Subject: [PATCH 07/11] fixed: bug when counting nb of active threads when queueSize > 1 also : added a test in testpool.c verifying resizing is effective. --- lib/common/pool.c | 9 ++-- tests/poolTests.c | 130 +++++++++++++++++++++++++++++++++++++++------- 2 files changed, 115 insertions(+), 24 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index ca5a38ee5..d08b1de79 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -91,12 +91,13 @@ static void* POOL_thread(void* opaque) { job.function(job.opaque); /* If the intended queue size was 0, signal after finishing job */ + ZSTD_pthread_mutex_lock(&ctx->queueMutex); + ctx->numThreadsBusy--; if (ctx->queueSize == 1) { - ZSTD_pthread_mutex_lock(&ctx->queueMutex); - ctx->numThreadsBusy--; - ZSTD_pthread_mutex_unlock(&ctx->queueMutex); ZSTD_pthread_cond_signal(&ctx->queuePushCond); - } } + } + ZSTD_pthread_mutex_unlock(&ctx->queueMutex); + } } /* for (;;) */ assert(0); /* Unreachable */ } diff --git a/tests/poolTests.c b/tests/poolTests.c index 00ee83015..23061b568 100644 --- a/tests/poolTests.c +++ b/tests/poolTests.c @@ -15,11 +15,11 @@ #include #include -#define ASSERT_TRUE(p) \ - do { \ - if (!(p)) { \ - return 1; \ - } \ +#define ASSERT_TRUE(p) \ + do { \ + if (!(p)) { \ + return 1; \ + } \ } while (0) #define ASSERT_FALSE(p) ASSERT_TRUE(!(p)) #define ASSERT_EQ(lhs, rhs) ASSERT_TRUE((lhs) == (rhs)) @@ -32,10 +32,10 @@ struct data { void fn(void *opaque) { struct data *data = (struct data *)opaque; - pthread_mutex_lock(&data->mutex); + ZSTD_pthread_mutex_lock(&data->mutex); data->data[data->i] = data->i; ++data->i; - pthread_mutex_unlock(&data->mutex); + ZSTD_pthread_mutex_unlock(&data->mutex); } int testOrder(size_t numThreads, size_t queueSize) { @@ -43,25 +43,26 @@ int testOrder(size_t numThreads, size_t queueSize) { POOL_ctx *ctx = POOL_create(numThreads, queueSize); ASSERT_TRUE(ctx); data.i = 0; - pthread_mutex_init(&data.mutex, NULL); - { - size_t i; + ZSTD_pthread_mutex_init(&data.mutex, NULL); + { size_t i; for (i = 0; i < 16; ++i) { POOL_add(ctx, &fn, &data); } } POOL_free(ctx); ASSERT_EQ(16, data.i); - { - size_t i; + { size_t i; for (i = 0; i < data.i; ++i) { ASSERT_EQ(i, data.data[i]); } } - pthread_mutex_destroy(&data.mutex); + ZSTD_pthread_mutex_destroy(&data.mutex); return 0; } + +/* --- test deadlocks --- */ + void waitFn(void *opaque) { (void)opaque; UTIL_sleepMilli(1); @@ -72,8 +73,7 @@ int testWait(size_t numThreads, size_t queueSize) { struct data data; POOL_ctx *ctx = POOL_create(numThreads, queueSize); ASSERT_TRUE(ctx); - { - size_t i; + { size_t i; for (i = 0; i < 16; ++i) { POOL_add(ctx, &waitFn, &data); } @@ -82,25 +82,115 @@ int testWait(size_t numThreads, size_t queueSize) { return 0; } + +/* --- test POOL_resize() --- */ + +typedef struct { + ZSTD_pthread_mutex_t mut; + int val; + int max; + ZSTD_pthread_cond_t cond; +} test_t; + +void waitLongFn(void *opaque) { + test_t* test = (test_t*) opaque; + UTIL_sleepMilli(10); + ZSTD_pthread_mutex_lock(&test->mut); + test->val = test->val + 1; + if (test->val == test->max) + ZSTD_pthread_cond_signal(&test->cond); + ZSTD_pthread_mutex_unlock(&test->mut); +} + +static int testThreadReduction_internal(POOL_ctx* ctx, test_t test) +{ + int const nbWaits = 16; + UTIL_time_t startTime, time4threads, time2threads; + + test.val = 0; + test.max = nbWaits; + + startTime = UTIL_getTime(); + { int i; + for (i=0; i= time2threads) return 1; /* check 4 threads were effectively faster than 2 */ + return 0; +} + +static int testThreadReduction(void) { + int result; + test_t test; + POOL_ctx* ctx = POOL_create(4 /*nbThreads*/, 2 /*queueSize*/); + + ASSERT_TRUE(ctx); + + memset(&test, 0, sizeof(test)); + ASSERT_FALSE( ZSTD_pthread_mutex_init(&test.mut, NULL) ); + ASSERT_FALSE( ZSTD_pthread_cond_init(&test.cond, NULL) ); + + result = testThreadReduction_internal(ctx, test); + + ZSTD_pthread_mutex_destroy(&test.mut); + ZSTD_pthread_cond_destroy(&test.cond); + POOL_free(ctx); + return result; +} + + +/* --- test launcher --- */ + int main(int argc, const char **argv) { size_t numThreads; + (void)argc; + (void)argv; + + if (POOL_create(0, 1)) { /* should not be possible */ + printf("FAIL: should not create POOL with 0 threads\n"); + return 1; + } + for (numThreads = 1; numThreads <= 4; ++numThreads) { size_t queueSize; for (queueSize = 0; queueSize <= 2; ++queueSize) { + printf("queueSize==%u, numThreads=%u \n", + (unsigned)queueSize, (unsigned)numThreads); if (testOrder(numThreads, queueSize)) { printf("FAIL: testOrder\n"); return 1; } + printf("SUCCESS: testOrder\n"); if (testWait(numThreads, queueSize)) { printf("FAIL: testWait\n"); return 1; } + printf("SUCCESS: testWait\n"); } } - printf("PASS: testOrder\n"); - (void)argc; - (void)argv; - return (POOL_create(0, 1)) ? printf("FAIL: testInvalid\n"), 1 - : printf("PASS: testInvalid\n"), 0; + + if (testThreadReduction()) return 1; + printf("PASS: all POOL tests\n"); + return 0; } From 7d80ada5caa31c1d964e960f86670238473041a5 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Thu, 21 Jun 2018 12:24:36 -0700 Subject: [PATCH 08/11] added a test for POOL (multithreading) ensuring all jobs in queue are nonetheless completed when POOL is instructed to end abruptly (POOL_free()) --- tests/poolTests.c | 76 ++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 69 insertions(+), 7 deletions(-) diff --git a/tests/poolTests.c b/tests/poolTests.c index 23061b568..d7ff70ac9 100644 --- a/tests/poolTests.c +++ b/tests/poolTests.c @@ -90,10 +90,10 @@ typedef struct { int val; int max; ZSTD_pthread_cond_t cond; -} test_t; +} poolTest_t; void waitLongFn(void *opaque) { - test_t* test = (test_t*) opaque; + poolTest_t* test = (poolTest_t*) opaque; UTIL_sleepMilli(10); ZSTD_pthread_mutex_lock(&test->mut); test->val = test->val + 1; @@ -102,7 +102,7 @@ void waitLongFn(void *opaque) { ZSTD_pthread_mutex_unlock(&test->mut); } -static int testThreadReduction_internal(POOL_ctx* ctx, test_t test) +static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test) { int const nbWaits = 16; UTIL_time_t startTime, time4threads, time2threads; @@ -117,7 +117,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, test_t test) } ZSTD_pthread_mutex_lock(&test.mut); ZSTD_pthread_cond_wait(&test.cond, &test.mut); - ASSERT_TRUE(test.val == nbWaits); + ASSERT_EQ(test.val, nbWaits); ZSTD_pthread_mutex_unlock(&test.mut); time4threads = UTIL_clockSpanNano(startTime); @@ -131,7 +131,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, test_t test) } ZSTD_pthread_mutex_lock(&test.mut); ZSTD_pthread_cond_wait(&test.cond, &test.mut); - ASSERT_TRUE(test.val == nbWaits); + ASSERT_EQ(test.val, nbWaits); ZSTD_pthread_mutex_unlock(&test.mut); time2threads = UTIL_clockSpanNano(startTime); @@ -141,7 +141,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, test_t test) static int testThreadReduction(void) { int result; - test_t test; + poolTest_t test; POOL_ctx* ctx = POOL_create(4 /*nbThreads*/, 2 /*queueSize*/); ASSERT_TRUE(ctx); @@ -155,10 +155,59 @@ static int testThreadReduction(void) { ZSTD_pthread_mutex_destroy(&test.mut); ZSTD_pthread_cond_destroy(&test.cond); POOL_free(ctx); + return result; } +/* --- test abrupt ending --- */ + +typedef struct { + ZSTD_pthread_mutex_t mut; + int val; +} abruptEndCanary_t; + +void waitIncFn(void *opaque) { + abruptEndCanary_t* test = (abruptEndCanary_t*) opaque; + UTIL_sleepMilli(1); + ZSTD_pthread_mutex_lock(&test->mut); + test->val = test->val + 1; + ZSTD_pthread_mutex_unlock(&test->mut); +} + +static int testAbruptEnding_internal(abruptEndCanary_t test) +{ + int const nbWaits = 16; + + POOL_ctx* const ctx = POOL_create(2 /*numThreads*/, nbWaits /*queueSize*/); + ASSERT_TRUE(ctx); + test.val = 0; + + { int i; + for (i=0; i Date: Thu, 21 Jun 2018 14:58:59 -0700 Subject: [PATCH 09/11] added extended POOL test abrupt end + downsizing with running jobs remaining in queue. also : POOL_resize() requires numThreads >= 1 --- lib/common/pool.c | 17 +++++++++++------ lib/common/pool.h | 1 + tests/poolTests.c | 7 ++++--- 3 files changed, 16 insertions(+), 9 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index d08b1de79..bd31f032f 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -73,10 +73,13 @@ static void* POOL_thread(void* opaque) { while ( ctx->queueEmpty || (ctx->numThreadsBusy >= ctx->threadLimit) ) { - if (ctx->shutdown) { - ZSTD_pthread_mutex_unlock(&ctx->queueMutex); - return opaque; - } + if (ctx->shutdown) { + /* even if !queueEmpty, (possible if numThreadsBusy >= threadLimit), + * a few threads will be shutdown while !queueEmpty, + * but enough threads will remain active to finish the queue */ + ZSTD_pthread_mutex_unlock(&ctx->queueMutex); + return opaque; + } ZSTD_pthread_cond_wait(&ctx->queuePopCond, &ctx->queueMutex); } /* Pop a job off the queue */ @@ -99,7 +102,7 @@ static void* POOL_thread(void* opaque) { ZSTD_pthread_mutex_unlock(&ctx->queueMutex); } } /* for (;;) */ - assert(0); /* Unreachable */ + assert(0); /* Unreachable */ } POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) { @@ -187,10 +190,12 @@ size_t POOL_sizeof(POOL_ctx *ctx) { } -/* note : only works if no job is running ! */ +/* @return : a working pool on success, NULL on failure + * note : starting context is considered consumed. */ static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) { if (numThreads <= ctx->threadCapacity) { + if (!numThreads) return NULL; ctx->threadLimit = numThreads; return ctx; } diff --git a/lib/common/pool.h b/lib/common/pool.h index 00ea76631..caf514907 100644 --- a/lib/common/pool.h +++ b/lib/common/pool.h @@ -45,6 +45,7 @@ void POOL_free(POOL_ctx* ctx); * note : new pool context might have same address as original one, but it's not guaranteed. * consider starting context as consumed, only rely on returned one. * note 2 : only numThreads can be resized, queueSize is unchanged. + * note 3 : `numThreads` must be at least 1 */ POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads); diff --git a/tests/poolTests.c b/tests/poolTests.c index d7ff70ac9..d5768967a 100644 --- a/tests/poolTests.c +++ b/tests/poolTests.c @@ -169,7 +169,7 @@ typedef struct { void waitIncFn(void *opaque) { abruptEndCanary_t* test = (abruptEndCanary_t*) opaque; - UTIL_sleepMilli(1); + UTIL_sleepMilli(10); ZSTD_pthread_mutex_lock(&test->mut); test->val = test->val + 1; ZSTD_pthread_mutex_unlock(&test->mut); @@ -179,14 +179,15 @@ static int testAbruptEnding_internal(abruptEndCanary_t test) { int const nbWaits = 16; - POOL_ctx* const ctx = POOL_create(2 /*numThreads*/, nbWaits /*queueSize*/); + POOL_ctx* ctx = POOL_create(3 /*numThreads*/, nbWaits /*queueSize*/); ASSERT_TRUE(ctx); test.val = 0; { int i; for (i=0; i Date: Thu, 21 Jun 2018 18:04:58 -0700 Subject: [PATCH 10/11] add a cond_broadcast after resize to make sure all threads (notably newly available threads) get awaken to immediately process potential items in the queue. --- lib/common/pool.c | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index bd31f032f..41a216f16 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -225,13 +225,16 @@ static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) * note : starting context is considered consumed. */ POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) { - POOL_ctx* newCtx; if (ctx==NULL) return NULL; ZSTD_pthread_mutex_lock(&ctx->queueMutex); - newCtx = POOL_resize_internal(ctx, numThreads); + { POOL_ctx* const newCtx = POOL_resize_internal(ctx, numThreads); + if (newCtx!=ctx) { + POOL_free(ctx); + return newCtx; + } } + ZSTD_pthread_cond_broadcast(&ctx->queuePopCond); ZSTD_pthread_mutex_unlock(&ctx->queueMutex); - if (newCtx!=ctx) POOL_free(ctx); - return newCtx; + return ctx; } /** From fbd5dfc1b163b462e2aa75a8b224d24938da81f3 Mon Sep 17 00:00:00 2001 From: Yann Collet Date: Fri, 22 Jun 2018 12:14:59 -0700 Subject: [PATCH 11/11] changed POOL_resize() return type to int return is now just en error code. This guarantee that `ctx` remains valid after POOL_resize(). Gets rid of internal POOL_free() operation. --- lib/common/pool.c | 40 +++++++++++++++------------------- lib/common/pool.h | 14 ++++++------ lib/compress/zstdmt_compress.c | 3 +-- tests/poolTests.c | 9 ++++---- 4 files changed, 29 insertions(+), 37 deletions(-) diff --git a/lib/common/pool.c b/lib/common/pool.c index 41a216f16..281b3824a 100644 --- a/lib/common/pool.c +++ b/lib/common/pool.c @@ -190,20 +190,19 @@ size_t POOL_sizeof(POOL_ctx *ctx) { } -/* @return : a working pool on success, NULL on failure - * note : starting context is considered consumed. */ -static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) +/* @return : 0 on success, 1 on error */ +static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) { if (numThreads <= ctx->threadCapacity) { - if (!numThreads) return NULL; + if (!numThreads) return 1; ctx->threadLimit = numThreads; - return ctx; + return 0; } /* numThreads > threadCapacity */ { ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem); - if (!threadPool) return NULL; + if (!threadPool) return 1; /* replace existing thread pool */ - memcpy(threadPool, ctx->threads, ctx->threadCapacity * sizeof(ctx->threads[0])); + memcpy(threadPool, ctx->threads, ctx->threadCapacity * sizeof(*threadPool)); ZSTD_free(ctx->threads, ctx->customMem); ctx->threads = threadPool; /* Initialize additional threads */ @@ -211,30 +210,25 @@ static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads) for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) { if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) { ctx->threadCapacity = threadId; - ctx->threadLimit = threadId; - return NULL; /* will release the pool */ + return 1; } } } } /* successfully expanded */ ctx->threadCapacity = numThreads; ctx->threadLimit = numThreads; - return ctx; + return 0; } -/* @return : a working pool on success, NULL on failure - * note : starting context is considered consumed. */ -POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) +/* @return : 0 on success, 1 on error */ +int POOL_resize(POOL_ctx* ctx, size_t numThreads) { - if (ctx==NULL) return NULL; + int result; + if (ctx==NULL) return 1; ZSTD_pthread_mutex_lock(&ctx->queueMutex); - { POOL_ctx* const newCtx = POOL_resize_internal(ctx, numThreads); - if (newCtx!=ctx) { - POOL_free(ctx); - return newCtx; - } } + result = POOL_resize_internal(ctx, numThreads); ZSTD_pthread_cond_broadcast(&ctx->queuePopCond); ZSTD_pthread_mutex_unlock(&ctx->queueMutex); - return ctx; + return result; } /** @@ -321,9 +315,9 @@ void POOL_free(POOL_ctx* ctx) { (void)ctx; } -POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) { - (void)numThreads; - return ctx; +int POOL_resize(POOL_ctx* ctx, size_t numThreads) { + (void)ctx; (void)numThreads; + return 0; } void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque) { diff --git a/lib/common/pool.h b/lib/common/pool.h index caf514907..458d37f13 100644 --- a/lib/common/pool.h +++ b/lib/common/pool.h @@ -40,14 +40,14 @@ void POOL_free(POOL_ctx* ctx); /*! POOL_resize() : * Expands or shrinks pool's number of threads. - * This is more efficient than releasing and creating a new context. - * @return : a new pool context on success, NULL on failure - * note : new pool context might have same address as original one, but it's not guaranteed. - * consider starting context as consumed, only rely on returned one. - * note 2 : only numThreads can be resized, queueSize is unchanged. - * note 3 : `numThreads` must be at least 1 + * This is more efficient than releasing + creating a new context, + * since it tries to preserve and re-use existing threads. + * `numThreads` must be at least 1. + * @return : 0 when resize was successful, + * !0 (typically 1) if there is an error. + * note : only numThreads can be resized, queueSize remains unchanged. */ -POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads); +int POOL_resize(POOL_ctx* ctx, size_t numThreads); /*! POOL_sizeof() : * @return threadpool memory usage diff --git a/lib/compress/zstdmt_compress.c b/lib/compress/zstdmt_compress.c index bbbdc5cd4..dc025e2a5 100644 --- a/lib/compress/zstdmt_compress.c +++ b/lib/compress/zstdmt_compress.c @@ -1018,8 +1018,7 @@ static ZSTD_CCtx_params ZSTDMT_initJobCCtxParams(ZSTD_CCtx_params const params) * @return : error code if fails, 0 on success */ static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers) { - mtctx->factory = POOL_resize(mtctx->factory, nbWorkers); - if (mtctx->factory == NULL) return ERROR(memory_allocation); + if (POOL_resize(mtctx->factory, nbWorkers)) return ERROR(memory_allocation); CHECK_F( ZSTDMT_expandJobsTable(mtctx, nbWorkers) ); mtctx->bufPool = ZSTDMT_expandBufferPool(mtctx->bufPool, nbWorkers); if (mtctx->bufPool == NULL) return ERROR(memory_allocation); diff --git a/tests/poolTests.c b/tests/poolTests.c index d5768967a..6a058a5a3 100644 --- a/tests/poolTests.c +++ b/tests/poolTests.c @@ -121,8 +121,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test) ZSTD_pthread_mutex_unlock(&test.mut); time4threads = UTIL_clockSpanNano(startTime); - ctx = POOL_resize(ctx, 2/*nbThreads*/); - ASSERT_TRUE(ctx); + ASSERT_EQ( POOL_resize(ctx, 2/*nbThreads*/) , 0 ); test.val = 0; startTime = UTIL_getTime(); { int i; @@ -142,7 +141,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test) static int testThreadReduction(void) { int result; poolTest_t test; - POOL_ctx* ctx = POOL_create(4 /*nbThreads*/, 2 /*queueSize*/); + POOL_ctx* const ctx = POOL_create(4 /*nbThreads*/, 2 /*queueSize*/); ASSERT_TRUE(ctx); @@ -179,7 +178,7 @@ static int testAbruptEnding_internal(abruptEndCanary_t test) { int const nbWaits = 16; - POOL_ctx* ctx = POOL_create(3 /*numThreads*/, nbWaits /*queueSize*/); + POOL_ctx* const ctx = POOL_create(3 /*numThreads*/, nbWaits /*queueSize*/); ASSERT_TRUE(ctx); test.val = 0; @@ -187,7 +186,7 @@ static int testAbruptEnding_internal(abruptEndCanary_t test) for (i=0; i