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.
This commit is contained in:
Yann Collet
2018-06-22 12:14:59 -07:00
parent 243cd9d8bb
commit fbd5dfc1b1
4 changed files with 29 additions and 37 deletions
+17 -23
View File
@@ -190,20 +190,19 @@ size_t POOL_sizeof(POOL_ctx *ctx) {
} }
/* @return : a working pool on success, NULL on failure /* @return : 0 on success, 1 on error */
* note : starting context is considered consumed. */ static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads)
static POOL_ctx* POOL_resize_internal(POOL_ctx* ctx, size_t numThreads)
{ {
if (numThreads <= ctx->threadCapacity) { if (numThreads <= ctx->threadCapacity) {
if (!numThreads) return NULL; if (!numThreads) return 1;
ctx->threadLimit = numThreads; ctx->threadLimit = numThreads;
return ctx; return 0;
} }
/* numThreads > threadCapacity */ /* numThreads > threadCapacity */
{ ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem); { 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 */ /* 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); ZSTD_free(ctx->threads, ctx->customMem);
ctx->threads = threadPool; ctx->threads = threadPool;
/* Initialize additional threads */ /* 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) { for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) {
if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) { if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) {
ctx->threadCapacity = threadId; ctx->threadCapacity = threadId;
ctx->threadLimit = threadId; return 1;
return NULL; /* will release the pool */
} } } }
} } } }
/* successfully expanded */ /* successfully expanded */
ctx->threadCapacity = numThreads; ctx->threadCapacity = numThreads;
ctx->threadLimit = numThreads; ctx->threadLimit = numThreads;
return ctx; return 0;
} }
/* @return : a working pool on success, NULL on failure /* @return : 0 on success, 1 on error */
* note : starting context is considered consumed. */ int POOL_resize(POOL_ctx* ctx, size_t numThreads)
POOL_ctx* 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); ZSTD_pthread_mutex_lock(&ctx->queueMutex);
{ POOL_ctx* const newCtx = POOL_resize_internal(ctx, numThreads); result = POOL_resize_internal(ctx, numThreads);
if (newCtx!=ctx) {
POOL_free(ctx);
return newCtx;
} }
ZSTD_pthread_cond_broadcast(&ctx->queuePopCond); ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
ZSTD_pthread_mutex_unlock(&ctx->queueMutex); ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
return ctx; return result;
} }
/** /**
@@ -321,9 +315,9 @@ void POOL_free(POOL_ctx* ctx) {
(void)ctx; (void)ctx;
} }
POOL_ctx* POOL_resize(POOL_ctx* ctx, size_t numThreads) { int POOL_resize(POOL_ctx* ctx, size_t numThreads) {
(void)numThreads; (void)ctx; (void)numThreads;
return ctx; return 0;
} }
void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque) { void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque) {
+7 -7
View File
@@ -40,14 +40,14 @@ void POOL_free(POOL_ctx* ctx);
/*! POOL_resize() : /*! POOL_resize() :
* Expands or shrinks pool's number of threads. * Expands or shrinks pool's number of threads.
* This is more efficient than releasing and creating a new context. * This is more efficient than releasing + creating a new context,
* @return : a new pool context on success, NULL on failure * since it tries to preserve and re-use existing threads.
* note : new pool context might have same address as original one, but it's not guaranteed. * `numThreads` must be at least 1.
* consider starting context as consumed, only rely on returned one. * @return : 0 when resize was successful,
* note 2 : only numThreads can be resized, queueSize is unchanged. * !0 (typically 1) if there is an error.
* note 3 : `numThreads` must be at least 1 * 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() : /*! POOL_sizeof() :
* @return threadpool memory usage * @return threadpool memory usage
+1 -2
View File
@@ -1018,8 +1018,7 @@ static ZSTD_CCtx_params ZSTDMT_initJobCCtxParams(ZSTD_CCtx_params const params)
* @return : error code if fails, 0 on success */ * @return : error code if fails, 0 on success */
static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers) static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers)
{ {
mtctx->factory = POOL_resize(mtctx->factory, nbWorkers); if (POOL_resize(mtctx->factory, nbWorkers)) return ERROR(memory_allocation);
if (mtctx->factory == NULL) return ERROR(memory_allocation);
CHECK_F( ZSTDMT_expandJobsTable(mtctx, nbWorkers) ); CHECK_F( ZSTDMT_expandJobsTable(mtctx, nbWorkers) );
mtctx->bufPool = ZSTDMT_expandBufferPool(mtctx->bufPool, nbWorkers); mtctx->bufPool = ZSTDMT_expandBufferPool(mtctx->bufPool, nbWorkers);
if (mtctx->bufPool == NULL) return ERROR(memory_allocation); if (mtctx->bufPool == NULL) return ERROR(memory_allocation);
+4 -5
View File
@@ -121,8 +121,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test)
ZSTD_pthread_mutex_unlock(&test.mut); ZSTD_pthread_mutex_unlock(&test.mut);
time4threads = UTIL_clockSpanNano(startTime); time4threads = UTIL_clockSpanNano(startTime);
ctx = POOL_resize(ctx, 2/*nbThreads*/); ASSERT_EQ( POOL_resize(ctx, 2/*nbThreads*/) , 0 );
ASSERT_TRUE(ctx);
test.val = 0; test.val = 0;
startTime = UTIL_getTime(); startTime = UTIL_getTime();
{ int i; { int i;
@@ -142,7 +141,7 @@ static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test)
static int testThreadReduction(void) { static int testThreadReduction(void) {
int result; int result;
poolTest_t test; 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); ASSERT_TRUE(ctx);
@@ -179,7 +178,7 @@ static int testAbruptEnding_internal(abruptEndCanary_t test)
{ {
int const nbWaits = 16; 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); ASSERT_TRUE(ctx);
test.val = 0; test.val = 0;
@@ -187,7 +186,7 @@ static int testAbruptEnding_internal(abruptEndCanary_t test)
for (i=0; i<nbWaits; i++) for (i=0; i<nbWaits; i++)
POOL_add(ctx, &waitIncFn, &test); /* all jobs pushed into queue */ POOL_add(ctx, &waitIncFn, &test); /* all jobs pushed into queue */
} }
ctx = POOL_resize(ctx, 1 /*numThreads*/); /* downsize numThreads, to try to break end condition */ ASSERT_EQ( POOL_resize(ctx, 1 /*numThreads*/) , 0 ); /* downsize numThreads, to try to break end condition */
POOL_free(ctx); /* must finish all jobs in queue before giving back control */ POOL_free(ctx); /* must finish all jobs in queue before giving back control */
ASSERT_EQ(test.val, nbWaits); ASSERT_EQ(test.val, nbWaits);