blob: 7a829454328022ccab351661bd7d49102cdf4c09 [file] [log] [blame]
khenaidooac637102019-01-14 15:44:34 -05001/*
2 * Copyright (c) 2016-present, Yann Collet, Facebook, Inc.
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 ======= */
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070013#include <stddef.h> /* size_t */
14#include "debug.h" /* assert */
khenaidooac637102019-01-14 15:44:34 -050015#include "zstd_internal.h" /* ZSTD_malloc, ZSTD_free */
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070016#include "pool.h"
khenaidooac637102019-01-14 15:44:34 -050017
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 */
29typedef struct POOL_job_s {
30 POOL_function function;
31 void *opaque;
32} POOL_job;
33
34struct POOL_ctx_s {
35 ZSTD_customMem customMem;
36 /* Keep track of the threads */
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070037 ZSTD_pthread_t* threads;
38 size_t threadCapacity;
39 size_t threadLimit;
khenaidooac637102019-01-14 15:44:34 -050040
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() :
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070063 * Work thread for the thread pool.
64 * Waits for jobs and executes them.
65 * @returns : NULL on failure else non-null.
66 */
khenaidooac637102019-01-14 15:44:34 -050067static void* POOL_thread(void* opaque) {
68 POOL_ctx* const ctx = (POOL_ctx*)opaque;
69 if (!ctx) { return NULL; }
70 for (;;) {
71 /* Lock the mutex and wait for a non-empty queue or until shutdown */
72 ZSTD_pthread_mutex_lock(&ctx->queueMutex);
73
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070074 while ( ctx->queueEmpty
75 || (ctx->numThreadsBusy >= ctx->threadLimit) ) {
76 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 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
81 return opaque;
82 }
khenaidooac637102019-01-14 15:44:34 -050083 ZSTD_pthread_cond_wait(&ctx->queuePopCond, &ctx->queueMutex);
84 }
khenaidooac637102019-01-14 15:44:34 -050085 /* Pop a job off the queue */
86 { POOL_job const job = ctx->queue[ctx->queueHead];
87 ctx->queueHead = (ctx->queueHead + 1) % ctx->queueSize;
88 ctx->numThreadsBusy++;
89 ctx->queueEmpty = ctx->queueHead == ctx->queueTail;
90 /* Unlock the mutex, signal a pusher, and run the job */
khenaidooac637102019-01-14 15:44:34 -050091 ZSTD_pthread_cond_signal(&ctx->queuePushCond);
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070092 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
khenaidooac637102019-01-14 15:44:34 -050093
94 job.function(job.opaque);
95
96 /* If the intended queue size was 0, signal after finishing job */
Scott Bakerbeb3cfa2019-10-01 14:44:30 -070097 ZSTD_pthread_mutex_lock(&ctx->queueMutex);
98 ctx->numThreadsBusy--;
khenaidooac637102019-01-14 15:44:34 -050099 if (ctx->queueSize == 1) {
khenaidooac637102019-01-14 15:44:34 -0500100 ZSTD_pthread_cond_signal(&ctx->queuePushCond);
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700101 }
102 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
103 }
khenaidooac637102019-01-14 15:44:34 -0500104 } /* for (;;) */
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700105 assert(0); /* Unreachable */
khenaidooac637102019-01-14 15:44:34 -0500106}
107
108POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) {
109 return POOL_create_advanced(numThreads, queueSize, ZSTD_defaultCMem);
110}
111
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700112POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize,
113 ZSTD_customMem customMem) {
khenaidooac637102019-01-14 15:44:34 -0500114 POOL_ctx* ctx;
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700115 /* Check parameters */
khenaidooac637102019-01-14 15:44:34 -0500116 if (!numThreads) { return NULL; }
117 /* Allocate the context and zero initialize */
118 ctx = (POOL_ctx*)ZSTD_calloc(sizeof(POOL_ctx), customMem);
119 if (!ctx) { return NULL; }
120 /* Initialize the job queue.
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700121 * It needs one extra space since one space is wasted to differentiate
122 * empty and full queues.
khenaidooac637102019-01-14 15:44:34 -0500123 */
124 ctx->queueSize = queueSize + 1;
125 ctx->queue = (POOL_job*)ZSTD_malloc(ctx->queueSize * sizeof(POOL_job), customMem);
126 ctx->queueHead = 0;
127 ctx->queueTail = 0;
128 ctx->numThreadsBusy = 0;
129 ctx->queueEmpty = 1;
130 (void)ZSTD_pthread_mutex_init(&ctx->queueMutex, NULL);
131 (void)ZSTD_pthread_cond_init(&ctx->queuePushCond, NULL);
132 (void)ZSTD_pthread_cond_init(&ctx->queuePopCond, NULL);
133 ctx->shutdown = 0;
134 /* Allocate space for the thread handles */
135 ctx->threads = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), customMem);
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700136 ctx->threadCapacity = 0;
khenaidooac637102019-01-14 15:44:34 -0500137 ctx->customMem = customMem;
138 /* Check for errors */
139 if (!ctx->threads || !ctx->queue) { POOL_free(ctx); return NULL; }
140 /* Initialize the threads */
141 { size_t i;
142 for (i = 0; i < numThreads; ++i) {
143 if (ZSTD_pthread_create(&ctx->threads[i], NULL, &POOL_thread, ctx)) {
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700144 ctx->threadCapacity = i;
khenaidooac637102019-01-14 15:44:34 -0500145 POOL_free(ctx);
146 return NULL;
147 } }
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700148 ctx->threadCapacity = numThreads;
149 ctx->threadLimit = numThreads;
khenaidooac637102019-01-14 15:44:34 -0500150 }
151 return ctx;
152}
153
154/*! POOL_join() :
155 Shutdown the queue, wake any sleeping threads, and join all of the threads.
156*/
157static void POOL_join(POOL_ctx* ctx) {
158 /* Shut down the queue */
159 ZSTD_pthread_mutex_lock(&ctx->queueMutex);
160 ctx->shutdown = 1;
161 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
162 /* Wake up sleeping threads */
163 ZSTD_pthread_cond_broadcast(&ctx->queuePushCond);
164 ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
165 /* Join all of the threads */
166 { size_t i;
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700167 for (i = 0; i < ctx->threadCapacity; ++i) {
168 ZSTD_pthread_join(ctx->threads[i], NULL); /* note : could fail */
khenaidooac637102019-01-14 15:44:34 -0500169 } }
170}
171
172void POOL_free(POOL_ctx *ctx) {
173 if (!ctx) { return; }
174 POOL_join(ctx);
175 ZSTD_pthread_mutex_destroy(&ctx->queueMutex);
176 ZSTD_pthread_cond_destroy(&ctx->queuePushCond);
177 ZSTD_pthread_cond_destroy(&ctx->queuePopCond);
178 ZSTD_free(ctx->queue, ctx->customMem);
179 ZSTD_free(ctx->threads, ctx->customMem);
180 ZSTD_free(ctx, ctx->customMem);
181}
182
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700183
184
khenaidooac637102019-01-14 15:44:34 -0500185size_t POOL_sizeof(POOL_ctx *ctx) {
186 if (ctx==NULL) return 0; /* supports sizeof NULL */
187 return sizeof(*ctx)
188 + ctx->queueSize * sizeof(POOL_job)
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700189 + ctx->threadCapacity * sizeof(ZSTD_pthread_t);
190}
191
192
193/* @return : 0 on success, 1 on error */
194static int POOL_resize_internal(POOL_ctx* ctx, size_t numThreads)
195{
196 if (numThreads <= ctx->threadCapacity) {
197 if (!numThreads) return 1;
198 ctx->threadLimit = numThreads;
199 return 0;
200 }
201 /* numThreads > threadCapacity */
202 { ZSTD_pthread_t* const threadPool = (ZSTD_pthread_t*)ZSTD_malloc(numThreads * sizeof(ZSTD_pthread_t), ctx->customMem);
203 if (!threadPool) return 1;
204 /* replace existing thread pool */
205 memcpy(threadPool, ctx->threads, ctx->threadCapacity * sizeof(*threadPool));
206 ZSTD_free(ctx->threads, ctx->customMem);
207 ctx->threads = threadPool;
208 /* Initialize additional threads */
209 { size_t threadId;
210 for (threadId = ctx->threadCapacity; threadId < numThreads; ++threadId) {
211 if (ZSTD_pthread_create(&threadPool[threadId], NULL, &POOL_thread, ctx)) {
212 ctx->threadCapacity = threadId;
213 return 1;
214 } }
215 } }
216 /* successfully expanded */
217 ctx->threadCapacity = numThreads;
218 ctx->threadLimit = numThreads;
219 return 0;
220}
221
222/* @return : 0 on success, 1 on error */
223int POOL_resize(POOL_ctx* ctx, size_t numThreads)
224{
225 int result;
226 if (ctx==NULL) return 1;
227 ZSTD_pthread_mutex_lock(&ctx->queueMutex);
228 result = POOL_resize_internal(ctx, numThreads);
229 ZSTD_pthread_cond_broadcast(&ctx->queuePopCond);
230 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
231 return result;
khenaidooac637102019-01-14 15:44:34 -0500232}
233
234/**
235 * Returns 1 if the queue is full and 0 otherwise.
236 *
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700237 * When queueSize is 1 (pool was created with an intended queueSize of 0),
238 * then a queue is empty if there is a thread free _and_ no job is waiting.
khenaidooac637102019-01-14 15:44:34 -0500239 */
240static int isQueueFull(POOL_ctx const* ctx) {
241 if (ctx->queueSize > 1) {
242 return ctx->queueHead == ((ctx->queueTail + 1) % ctx->queueSize);
243 } else {
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700244 return (ctx->numThreadsBusy == ctx->threadLimit) ||
khenaidooac637102019-01-14 15:44:34 -0500245 !ctx->queueEmpty;
246 }
247}
248
249
250static void POOL_add_internal(POOL_ctx* ctx, POOL_function function, void *opaque)
251{
252 POOL_job const job = {function, opaque};
253 assert(ctx != NULL);
254 if (ctx->shutdown) return;
255
256 ctx->queueEmpty = 0;
257 ctx->queue[ctx->queueTail] = job;
258 ctx->queueTail = (ctx->queueTail + 1) % ctx->queueSize;
259 ZSTD_pthread_cond_signal(&ctx->queuePopCond);
260}
261
262void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque)
263{
264 assert(ctx != NULL);
265 ZSTD_pthread_mutex_lock(&ctx->queueMutex);
266 /* Wait until there is space in the queue for the new job */
267 while (isQueueFull(ctx) && (!ctx->shutdown)) {
268 ZSTD_pthread_cond_wait(&ctx->queuePushCond, &ctx->queueMutex);
269 }
270 POOL_add_internal(ctx, function, opaque);
271 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
272}
273
274
275int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque)
276{
277 assert(ctx != NULL);
278 ZSTD_pthread_mutex_lock(&ctx->queueMutex);
279 if (isQueueFull(ctx)) {
280 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
281 return 0;
282 }
283 POOL_add_internal(ctx, function, opaque);
284 ZSTD_pthread_mutex_unlock(&ctx->queueMutex);
285 return 1;
286}
287
288
289#else /* ZSTD_MULTITHREAD not defined */
290
291/* ========================== */
292/* No multi-threading support */
293/* ========================== */
294
295
296/* We don't need any data, but if it is empty, malloc() might return NULL. */
297struct POOL_ctx_s {
298 int dummy;
299};
300static POOL_ctx g_ctx;
301
302POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) {
303 return POOL_create_advanced(numThreads, queueSize, ZSTD_defaultCMem);
304}
305
306POOL_ctx* POOL_create_advanced(size_t numThreads, size_t queueSize, ZSTD_customMem customMem) {
307 (void)numThreads;
308 (void)queueSize;
309 (void)customMem;
310 return &g_ctx;
311}
312
313void POOL_free(POOL_ctx* ctx) {
314 assert(!ctx || ctx == &g_ctx);
315 (void)ctx;
316}
317
Scott Bakerbeb3cfa2019-10-01 14:44:30 -0700318int POOL_resize(POOL_ctx* ctx, size_t numThreads) {
319 (void)ctx; (void)numThreads;
320 return 0;
321}
322
khenaidooac637102019-01-14 15:44:34 -0500323void POOL_add(POOL_ctx* ctx, POOL_function function, void* opaque) {
324 (void)ctx;
325 function(opaque);
326}
327
328int POOL_tryAdd(POOL_ctx* ctx, POOL_function function, void* opaque) {
329 (void)ctx;
330 function(opaque);
331 return 1;
332}
333
334size_t POOL_sizeof(POOL_ctx* ctx) {
335 if (ctx==NULL) return 0; /* supports sizeof NULL */
336 assert(ctx == &g_ctx);
337 return sizeof(*ctx);
338}
339
340#endif /* ZSTD_MULTITHREAD */