/src/zstd/lib/common/pool.c

Source (jump to first uncovered line)
/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 * All rights reserved.
 *
 * 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.
 */


/* ======   Dependencies   ======= */
#include "../common/allocations.h"  /* ZSTD_customCalloc, ZSTD_customFree */
#include "zstd_deps.h" /* size_t */
#include "debug.h"     /* assert */
#include "pool.h"

/* ======   Compiler specifics   ====== */
#if defined(_MSC_VER)
#  pragma warning(disable : 4204)        /* disable: C4204: non-constant aggregate initializer */
#endif


#ifdef ZSTD_MULTITHREAD

#include "threading.h"   /* pthread adaptation */

/* A job is a function and an opaque argument */
typedef struct POOL_job_s {
    POOL_function function;
    void *opaque;
} POOL_job;

struct POOL_ctx_s {
    ZSTD_customMem customMem;
    /* Keep track of the threads */
    ZSTD_pthread_t* threads;
    size_t threadCapacity;
    size_t threadLimit;

    /* The queue is a circular buffer */
    POOL_job *queue;
    size_t queueHead;
    size_t queueTail;
    size_t queueSize;

    /* The number of threads working on jobs */
    size_t numThreadsBusy;
    /* Indicates if the queue is empty */
    int queueEmpty;

    /* The mutex protects the queue */
    ZSTD_pthread_mutex_t queueMutex;
    /* Condition variable for pushers to wait on when the queue is full */
    ZSTD_pthread_cond_t queuePushCond;
    /* Condition variables for poppers to wait on when the queue is empty */
    ZSTD_pthread_cond_t queuePopCond;
    /* Indicates if the queue is shutting down */
    int shutdown;
};

/* POOL_thread() :
 * 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; }
    for (;;) {
        /* Lock the mutex and wait for a non-empty queue or until shutdown */
        ZSTD_pthread_mutex_lock(&ctx->queueMutex);

        while ( ctx->queueEmpty
            || (ctx->numThreadsBusy >= ctx->threadLimit) ) {
            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 */
        {   POOL_job const job = ctx->queue[ctx->queueHead];
            ctx->queueHead = (ctx->queueHead + 1) % ctx->queueSize;
            ctx->numThreadsBusy++;
            ctx->queueEmpty = (ctx->queueHead == ctx->queueTail);
            /* Unlock the mutex, signal a pusher, and run the job */
            ZSTD_pthread_cond_signal(&ctx->queuePushCond);
            ZSTD_pthread_mutex_unlock(&ctx->queueMutex);

            job.function(job.opaque);

            /* If the intended queue size was 0, signal after finishing job */
            ZSTD_pthread_mutex_lock(&ctx->queueMutex);
            ctx->numThreadsBusy--;
            ZSTD_pthread_cond_signal(&ctx->queuePushCond);
            ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
        }
    }  /* for (;;) */
    assert(0);  /* Unreachable */
}

/* ZSTD_createThreadPool() : public access point */
POOL_ctx* ZSTD_createThreadPool(size_t numThreads) {
    return POOL_create (numThreads, 0);
}

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* ctx;
    /* Check parameters */
    if (!numThreads) { return NULL; }
    /* Allocate the context and zero initialize */
    ctx = (POOL_ctx*)ZSTD_customCalloc(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.
     */
    ctx->queueSize = queueSize + 1;
    ctx->queue = (POOL_job*)ZSTD_customCalloc(ctx->queueSize * sizeof(POOL_job), customMem);
    ctx->queueHead = 0;
    ctx->queueTail = 0;
    ctx->numThreadsBusy = 0;
    ctx->queueEmpty = 1;
    {
        int error = 0;
        error |= ZSTD_pthread_mutex_init(&ctx->queueMutex, NULL);
        error |= ZSTD_pthread_cond_init(&ctx->queuePushCond, NULL);
        error |= ZSTD_pthread_cond_init(&ctx->queuePopCond, NULL);
        if (error) { POOL_free(ctx); return NULL; }
    }
    ctx->shutdown = 0;
    /* Allocate space for the thread handles */
    ctx->threads = (ZSTD_pthread_t*)ZSTD_customCalloc(numThreads * sizeof(ZSTD_pthread_t), customMem);
    ctx->threadCapacity = 0;
    ctx->customMem = customMem;
    /* Check for errors */
    if (!ctx->threads || !ctx->queue) { POOL_free(ctx); return NULL; }
    /* Initialize the threads */
    {   size_t i;
        for (i = 0; i < numThreads; ++i) {
            if (ZSTD_pthread_create(&ctx->threads[i], NULL, &POOL_thread, ctx)) {
                ctx->threadCapacity = i;
                POOL_free(ctx);
                return NULL;
        }   }
        ctx->threadCapacity = numThreads;
        ctx->threadLimit = numThreads;
    }
    return ctx;
}

/*! POOL_join() :
    Shutdown the queue, wake any sleeping threads, and join all of the threads.
*/
static void POOL_join(POOL_ctx* ctx) {
    /* Shut down the queue */
    ZSTD_pthread_mutex_lock(&ctx->queueMutex);
    ctx->shutdown = 1;
    ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
    /* Wake up sleeping threads */
    ZSTD_pthread_cond_broadcast(&ctx->queuePushCond);
    ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
    /* Join all of the threads */
    {   size_t i;
        for (i = 0; i < ctx->threadCapacity; ++i) {
            ZSTD_pthread_join(ctx->threads[i]);  /* note : could fail */
    }   }
}

void POOL_free(POOL_ctx *ctx) {
    if (!ctx) { return; }
    POOL_join(ctx);
    ZSTD_pthread_mutex_destroy(&ctx->queueMutex);
    ZSTD_pthread_cond_destroy(&ctx->queuePushCond);
    ZSTD_pthread_cond_destroy(&ctx->queuePopCond);
    ZSTD_customFree(ctx->queue, ctx->customMem);
    ZSTD_customFree(ctx->threads, ctx->customMem);
    ZSTD_customFree(ctx, ctx->customMem);
}

/*! POOL_joinJobs() :
 *  Waits for all queued jobs to finish executing.
 */
void POOL_joinJobs(POOL_ctx* ctx) {
    ZSTD_pthread_mutex_lock(&ctx->queueMutex);
    while(!ctx->queueEmpty || ctx->numThreadsBusy > 0) {
        ZSTD_pthread_cond_wait(&ctx->queuePushCond, &ctx->queueMutex);
    }
    ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
}

void ZSTD_freeThreadPool (ZSTD_threadPool* pool) {
  POOL_free (pool);
}

size_t POOL_sizeof(const POOL_ctx* ctx) {
    if (ctx==NULL) return 0;  /* supports sizeof NULL */
    return sizeof(*ctx)
        + ctx->queueSize * sizeof(POOL_job)
        + ctx->threadCapacity * sizeof(ZSTD_pthread_t);
}


/* @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 1;
        ctx->threadLimit = numThreads;
        return 0;
    }
    /* numThreads > threadCapacity */
    {   ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_customCalloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem);
        if (!threadPool) return 1;
        /* replace existing thread pool */
        ZSTD_memcpy(threadPool, ctx->threads, ctx->threadCapacity * sizeof(ZSTD_pthread_t));
        ZSTD_customFree(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)) {
                    ctx->threadCapacity = threadId;
                    return 1;
            }   }
    }   }
    /* successfully expanded */
    ctx->threadCapacity = numThreads;
    ctx->threadLimit = numThreads;
    return 0;
}

/* @return : 0 on success, 1 on error */
int POOL_resize(POOL_ctx* ctx, size_t numThreads)
{
    int result;
    if (ctx==NULL) return 1;
    ZSTD_pthread_mutex_lock(&ctx->queueMutex);
    result = POOL_resize_internal(ctx, numThreads);
    ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
    ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
    return result;
}

/**
 * Returns 1 if the queue is full and 0 otherwise.
 *
 * 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->threadLimit) ||
               !ctx->queueEmpty;
    }
}


static void
POOL_add_internal(POOL_ctx* ctx, POOL_function function, void *opaque)
{
    POOL_job job;
    job.function = function;
    job.opaque = opaque;
    assert(ctx != NULL);
    if (ctx->shutdown) return;

    ctx->queueEmpty = 0;
    ctx->queue[ctx->queueTail] = job;
    ctx->queueTail = (ctx->queueTail + 1) % ctx->queueSize;
    ZSTD_pthread_cond_signal(&ctx->queuePopCond);
}

void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque)
{
    assert(ctx != NULL);
    ZSTD_pthread_mutex_lock(&ctx->queueMutex);
    /* Wait until there is space in the queue for the new job */
    while (isQueueFull(ctx) && (!ctx->shutdown)) {
        ZSTD_pthread_cond_wait(&ctx->queuePushCond, &ctx->queueMutex);
    }
    POOL_add_internal(ctx, function, opaque);
    ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
}


int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque)
{
    assert(ctx != NULL);
    ZSTD_pthread_mutex_lock(&ctx->queueMutex);
    if (isQueueFull(ctx)) {
        ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
        return 0;
    }
    POOL_add_internal(ctx, function, opaque);
    ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
    return 1;
}


#else  /* ZSTD_MULTITHREAD  not defined */

/* ========================== */
/* No multi-threading support */
/* ========================== */


/* We don't need any data, but if it is empty, malloc() might return NULL. */
struct POOL_ctx_s {
    int dummy;
};
static POOL_ctx g_poolCtx;

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)
{
    (void)numThreads;
    (void)queueSize;
    (void)customMem;
    return &g_poolCtx;
}

void POOL_free(POOL_ctx* ctx) {
    assert(!ctx || ctx == &g_poolCtx);
    (void)ctx;
}

void POOL_joinJobs(POOL_ctx* ctx){
    assert(!ctx || ctx == &g_poolCtx);
    (void)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) {
    (void)ctx;
    function(opaque);
}

int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque) {
    (void)ctx;
    function(opaque);
    return 1;
}

size_t POOL_sizeof(const POOL_ctx* ctx) {
    if (ctx==NULL) return 0;  /* supports sizeof NULL */
    assert(ctx == &g_poolCtx);
    return sizeof(*ctx);
}

#endif  /* ZSTD_MULTITHREAD */

Coverage Report

Created: 2025-07-11 06:35

Line	Count	Source (jump to first uncovered line)
1		/*
2		* Copyright (c) Meta Platforms, Inc. and affiliates.
3		* All rights reserved.
4		*
5		* This source code is licensed under both the BSD-style license (found in the
6		* LICENSE file in the root directory of this source tree) and the GPLv2 (found
7		* in the COPYING file in the root directory of this source tree).
8		* You may select, at your option, one of the above-listed licenses.
9		*/
10
11
12		/* ====== Dependencies ======= */
13		#include "../common/allocations.h" /* ZSTD_customCalloc, ZSTD_customFree */
14		#include "zstd_deps.h" /* size_t */
15		#include "debug.h" /* assert */
16		#include "pool.h"
17
18		/* ====== Compiler specifics ====== */
19		#if defined(_MSC_VER)
20		# pragma warning(disable : 4204) /* disable: C4204: non-constant aggregate initializer */
21		#endif
22
23
24		#ifdef ZSTD_MULTITHREAD
25
26		#include "threading.h" /* pthread adaptation */
27
28		/* A job is a function and an opaque argument */
29		typedef struct POOL_job_s {
30		POOL_function function;
31		void *opaque;
32		} POOL_job;
33
34		struct POOL_ctx_s {
35		ZSTD_customMem customMem;
36		/* Keep track of the threads */
37		ZSTD_pthread_t* threads;
38		size_t threadCapacity;
39		size_t threadLimit;
40
41		/* The queue is a circular buffer */
42		POOL_job *queue;
43		size_t queueHead;
44		size_t queueTail;
45		size_t queueSize;
46
47		/* The number of threads working on jobs */
48		size_t numThreadsBusy;
49		/* Indicates if the queue is empty */
50		int queueEmpty;
51
52		/* The mutex protects the queue */
53		ZSTD_pthread_mutex_t queueMutex;
54		/* Condition variable for pushers to wait on when the queue is full */
55		ZSTD_pthread_cond_t queuePushCond;
56		/* Condition variables for poppers to wait on when the queue is empty */
57		ZSTD_pthread_cond_t queuePopCond;
58		/* Indicates if the queue is shutting down */
59		int shutdown;
60		};
61
62		/* POOL_thread() :
63		* Work thread for the thread pool.
64		* Waits for jobs and executes them.
65		* @returns : NULL on failure else non-null.
66		*/
67	388	static void* POOL_thread(void* opaque) {
68	388	POOL_ctx* const ctx = (POOL_ctx*)opaque;
69	388	if (!ctx) { return NULL; }
70	1.28k	for (;;) {
71		/* Lock the mutex and wait for a non-empty queue or until shutdown */
72	1.28k	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
73
74	2.55k	while ( ctx->queueEmpty
75	2.55k	\|\| (ctx->numThreadsBusy >= ctx->threadLimit) ) {
76	1.66k	if (ctx->shutdown) {
77		/* even if !queueEmpty, (possible if numThreadsBusy >= threadLimit),
78		* a few threads will be shutdown while !queueEmpty,
79		* but enough threads will remain active to finish the queue */
80	388	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
81	388	return opaque;
82	388	}
83	1.27k	ZSTD_pthread_cond_wait(&ctx->queuePopCond, &ctx->queueMutex);
84	1.27k	}
85		/* Pop a job off the queue */
86	896	{ POOL_job const job = ctx->queue[ctx->queueHead];
87	896	ctx->queueHead = (ctx->queueHead + 1) % ctx->queueSize;
88	896	ctx->numThreadsBusy++;
89	896	ctx->queueEmpty = (ctx->queueHead == ctx->queueTail);
90		/* Unlock the mutex, signal a pusher, and run the job */
91	896	ZSTD_pthread_cond_signal(&ctx->queuePushCond);
92	896	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
93
94	896	job.function(job.opaque);
95
96		/* If the intended queue size was 0, signal after finishing job */
97	896	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
98	896	ctx->numThreadsBusy--;
99	896	ZSTD_pthread_cond_signal(&ctx->queuePushCond);
100	896	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
101	896	}
102	896	} /* for (;;) */
103	0	assert(0); /* Unreachable */
104	0	}
105
106		/* ZSTD_createThreadPool() : public access point */
107	0	POOL_ctx* ZSTD_createThreadPool(size_t numThreads) {
108	0	return POOL_create (numThreads, 0);
109	0	}
110
111	0	POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) {
112	0	return POOL_create_advanced(numThreads, queueSize, ZSTD_defaultCMem);
113	0	}
114
115		POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize,
116		ZSTD_customMem customMem)
117	217	{
118	217	POOL_ctx* ctx;
119		/* Check parameters */
120	217	if (!numThreads) { return NULL; }
121		/* Allocate the context and zero initialize */
122	217	ctx = (POOL_ctx*)ZSTD_customCalloc(sizeof(POOL_ctx), customMem);
123	217	if (!ctx) { return NULL; }
124		/* Initialize the job queue.
125		* It needs one extra space since one space is wasted to differentiate
126		* empty and full queues.
127		*/
128	217	ctx->queueSize = queueSize + 1;
129	217	ctx->queue = (POOL_job)ZSTD_customCalloc(ctx->queueSize sizeof(POOL_job), customMem);
130	217	ctx->queueHead = 0;
131	217	ctx->queueTail = 0;
132	217	ctx->numThreadsBusy = 0;
133	217	ctx->queueEmpty = 1;
134	217	{
135	217	int error = 0;
136	217	error \|= ZSTD_pthread_mutex_init(&ctx->queueMutex, NULL);
137	217	error \|= ZSTD_pthread_cond_init(&ctx->queuePushCond, NULL);
138	217	error \|= ZSTD_pthread_cond_init(&ctx->queuePopCond, NULL);
139	217	if (error) { POOL_free(ctx); return NULL; }
140	217	}
141	217	ctx->shutdown = 0;
142		/* Allocate space for the thread handles */
143	217	ctx->threads = (ZSTD_pthread_t)ZSTD_customCalloc(numThreads sizeof(ZSTD_pthread_t), customMem);
144	217	ctx->threadCapacity = 0;
145	217	ctx->customMem = customMem;
146		/* Check for errors */
147	217	if (!ctx->threads \|\| !ctx->queue) { POOL_free(ctx); return NULL; }
148		/* Initialize the threads */
149	217	{ size_t i;
150	605	for (i = 0; i < numThreads; ++i) {
151	388	if (ZSTD_pthread_create(&ctx->threads[i], NULL, &POOL_thread, ctx)) {
152	0	ctx->threadCapacity = i;
153	0	POOL_free(ctx);
154	0	return NULL;
155	0	} }
156	217	ctx->threadCapacity = numThreads;
157	217	ctx->threadLimit = numThreads;
158	217	}
159	0	return ctx;
160	217	}
161
162		/*! POOL_join() :
163		Shutdown the queue, wake any sleeping threads, and join all of the threads.
164		*/
165	217	static void POOL_join(POOL_ctx* ctx) {
166		/* Shut down the queue */
167	217	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
168	217	ctx->shutdown = 1;
169	217	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
170		/* Wake up sleeping threads */
171	217	ZSTD_pthread_cond_broadcast(&ctx->queuePushCond);
172	217	ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
173		/* Join all of the threads */
174	217	{ size_t i;
175	605	for (i = 0; i < ctx->threadCapacity; ++i) {
176	388	ZSTD_pthread_join(ctx->threads[i]); /* note : could fail */
177	388	} }
178	217	}
179
180	217	void POOL_free(POOL_ctx *ctx) {
181	217	if (!ctx) { return; }
182	217	POOL_join(ctx);
183	217	ZSTD_pthread_mutex_destroy(&ctx->queueMutex);
184	217	ZSTD_pthread_cond_destroy(&ctx->queuePushCond);
185	217	ZSTD_pthread_cond_destroy(&ctx->queuePopCond);
186	217	ZSTD_customFree(ctx->queue, ctx->customMem);
187	217	ZSTD_customFree(ctx->threads, ctx->customMem);
188	217	ZSTD_customFree(ctx, ctx->customMem);
189	217	}
190
191		/*! POOL_joinJobs() :
192		* Waits for all queued jobs to finish executing.
193		*/
194	0	void POOL_joinJobs(POOL_ctx* ctx) {
195	0	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
196	0	while(!ctx->queueEmpty \|\| ctx->numThreadsBusy > 0) {
197	0	ZSTD_pthread_cond_wait(&ctx->queuePushCond, &ctx->queueMutex);
198	0	}
199	0	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
200	0	}
201
202	0	void ZSTD_freeThreadPool (ZSTD_threadPool* pool) {
203	0	POOL_free (pool);
204	0	}
205
206	0	size_t POOL_sizeof(const POOL_ctx* ctx) {
207	0	if (ctx==NULL) return 0; /* supports sizeof NULL */
208	0	return sizeof(*ctx)
209	0	+ ctx->queueSize * sizeof(POOL_job)
210	0	+ ctx->threadCapacity * sizeof(ZSTD_pthread_t);
211	0	}
212
213
214		/* @return : 0 on success, 1 on error */
215		static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads)
216	0	{
217	0	if (numThreads <= ctx->threadCapacity) {
218	0	if (!numThreads) return 1;
219	0	ctx->threadLimit = numThreads;
220	0	return 0;
221	0	}
222		/* numThreads > threadCapacity */
223	0	{ ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t)ZSTD_customCalloc(numThreads sizeof(ZSTD_pthread_t), ctx->customMem);
224	0	if (!threadPool) return 1;
225		/* replace existing thread pool */
226	0	ZSTD_memcpy(threadPool, ctx->threads, ctx->threadCapacity * sizeof(ZSTD_pthread_t));
227	0	ZSTD_customFree(ctx->threads, ctx->customMem);
228	0	ctx->threads = threadPool;
229		/* Initialize additional threads */
230	0	{ size_t threadId;
231	0	for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) {
232	0	if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) {
233	0	ctx->threadCapacity = threadId;
234	0	return 1;
235	0	} }
236	0	} }
237		/* successfully expanded */
238	0	ctx->threadCapacity = numThreads;
239	0	ctx->threadLimit = numThreads;
240	0	return 0;
241	0	}
242
243		/* @return : 0 on success, 1 on error */
244		int POOL_resize(POOL_ctx* ctx, size_t numThreads)
245	0	{
246	0	int result;
247	0	if (ctx==NULL) return 1;
248	0	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
249	0	result = POOL_resize_internal(ctx, numThreads);
250	0	ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
251	0	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
252	0	return result;
253	0	}
254
255		/**
256		* Returns 1 if the queue is full and 0 otherwise.
257		*
258		* When queueSize is 1 (pool was created with an intended queueSize of 0),
259		* then a queue is empty if there is a thread free _and_ no job is waiting.
260		*/
261	1.29k	static int isQueueFull(POOL_ctx const* ctx) {
262	1.29k	if (ctx->queueSize > 1) {
263	0	return ctx->queueHead == ((ctx->queueTail + 1) % ctx->queueSize);
264	1.29k	} else {
265	1.29k	return (ctx->numThreadsBusy == ctx->threadLimit) \|\|
266	1.29k	!ctx->queueEmpty;
267	1.29k	}
268	1.29k	}
269
270
271		static void
272		POOL_add_internal(POOL_ctx* ctx, POOL_function function, void *opaque)
273	896	{
274	896	POOL_job job;
275	896	job.function = function;
276	896	job.opaque = opaque;
277	896	assert(ctx != NULL);
278	896	if (ctx->shutdown) return;
279
280	896	ctx->queueEmpty = 0;
281	896	ctx->queue[ctx->queueTail] = job;
282	896	ctx->queueTail = (ctx->queueTail + 1) % ctx->queueSize;
283	896	ZSTD_pthread_cond_signal(&ctx->queuePopCond);
284	896	}
285
286		void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque)
287	0	{
288	0	assert(ctx != NULL);
289	0	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
290		/* Wait until there is space in the queue for the new job */
291	0	while (isQueueFull(ctx) && (!ctx->shutdown)) {
292	0	ZSTD_pthread_cond_wait(&ctx->queuePushCond, &ctx->queueMutex);
293	0	}
294	0	POOL_add_internal(ctx, function, opaque);
295	0	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
296	0	}
297
298
299		int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque)
300	1.29k	{
301	1.29k	assert(ctx != NULL);
302	1.29k	ZSTD_pthread_mutex_lock(&ctx->queueMutex);
303	1.29k	if (isQueueFull(ctx)) {
304	396	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
305	396	return 0;
306	396	}
307	896	POOL_add_internal(ctx, function, opaque);
308	896	ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
309	896	return 1;
310	1.29k	}
311
312
313		#else /* ZSTD_MULTITHREAD not defined */
314
315		/* ========================== */
316		/* No multi-threading support */
317		/* ========================== */
318
319
320		/* We don't need any data, but if it is empty, malloc() might return NULL. */
321		struct POOL_ctx_s {
322		int dummy;
323		};
324		static POOL_ctx g_poolCtx;
325
326		POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) {
327		return POOL_create_advanced(numThreads, queueSize, ZSTD_defaultCMem);
328		}
329
330		POOL_ctx*
331		POOL_create_advanced(size_t numThreads, size_t queueSize, ZSTD_customMem customMem)
332		{
333		(void)numThreads;
334		(void)queueSize;
335		(void)customMem;
336		return &g_poolCtx;
337		}
338
339		void POOL_free(POOL_ctx* ctx) {
340		assert(!ctx \|\| ctx == &g_poolCtx);
341		(void)ctx;
342		}
343
344		void POOL_joinJobs(POOL_ctx* ctx){
345		assert(!ctx \|\| ctx == &g_poolCtx);
346		(void)ctx;
347		}
348
349		int POOL_resize(POOL_ctx* ctx, size_t numThreads) {
350		(void)ctx; (void)numThreads;
351		return 0;
352		}
353
354		void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque) {
355		(void)ctx;
356		function(opaque);
357		}
358
359		int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque) {
360		(void)ctx;
361		function(opaque);
362		return 1;
363		}
364
365		size_t POOL_sizeof(const POOL_ctx* ctx) {
366		if (ctx==NULL) return 0; /* supports sizeof NULL */
367		assert(ctx == &g_poolCtx);
368		return sizeof(*ctx);
369		}
370
371		#endif /* ZSTD_MULTITHREAD */