2 * Copyright (c) Meta Platforms, Inc. and affiliates.
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.
13 #include "threading.h"
19 #define ASSERT_TRUE(p) \
25 #define ASSERT_FALSE(p) ASSERT_TRUE(!(p))
26 #define ASSERT_EQ(lhs, rhs) ASSERT_TRUE((lhs) == (rhs))
29 ZSTD_pthread_mutex_t mutex;
34 static void fn(void *opaque)
36 struct data *data = (struct data *)opaque;
37 ZSTD_pthread_mutex_lock(&data->mutex);
38 data->data[data->i] = (unsigned)(data->i);
40 ZSTD_pthread_mutex_unlock(&data->mutex);
43 static int testOrder(size_t numThreads, size_t queueSize)
46 POOL_ctx* const ctx = POOL_create(numThreads, queueSize);
49 ASSERT_FALSE(ZSTD_pthread_mutex_init(&data.mutex, NULL));
51 for (i = 0; i < 16; ++i) {
52 POOL_add(ctx, &fn, &data);
56 ASSERT_EQ(16, data.i);
58 for (i = 0; i < data.i; ++i) {
59 ASSERT_EQ(i, data.data[i]);
62 ZSTD_pthread_mutex_destroy(&data.mutex);
67 /* --- test deadlocks --- */
69 static void waitFn(void *opaque) {
74 /* Tests for deadlock */
75 static int testWait(size_t numThreads, size_t queueSize) {
77 POOL_ctx* const ctx = POOL_create(numThreads, queueSize);
80 for (i = 0; i < 16; ++i) {
81 POOL_add(ctx, &waitFn, &data);
89 /* --- test POOL_resize() --- */
92 ZSTD_pthread_mutex_t mut;
96 ZSTD_pthread_cond_t cond;
99 static void waitLongFn(void *opaque) {
100 poolTest_t* const test = (poolTest_t*) opaque;
101 ZSTD_pthread_mutex_lock(&test->mut);
103 if (test->val > test->max)
104 test->max = test->val;
105 ZSTD_pthread_mutex_unlock(&test->mut);
109 ZSTD_pthread_mutex_lock(&test->mut);
112 if (test->countdown == 0)
113 ZSTD_pthread_cond_signal(&test->cond);
114 ZSTD_pthread_mutex_unlock(&test->mut);
117 static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test)
119 int const nbWaits = 16;
121 test.countdown = nbWaits;
126 for (i=0; i<nbWaits; i++)
127 POOL_add(ctx, &waitLongFn, &test);
129 ZSTD_pthread_mutex_lock(&test.mut);
130 while (test.countdown > 0)
131 ZSTD_pthread_cond_wait(&test.cond, &test.mut);
132 ASSERT_EQ(test.val, 0);
133 ASSERT_EQ(test.max, 4);
134 ZSTD_pthread_mutex_unlock(&test.mut);
136 ASSERT_EQ( POOL_resize(ctx, 2/*nbThreads*/) , 0 );
137 test.countdown = nbWaits;
141 for (i=0; i<nbWaits; i++)
142 POOL_add(ctx, &waitLongFn, &test);
144 ZSTD_pthread_mutex_lock(&test.mut);
145 while (test.countdown > 0)
146 ZSTD_pthread_cond_wait(&test.cond, &test.mut);
147 ASSERT_EQ(test.val, 0);
148 ASSERT_EQ(test.max, 2);
149 ZSTD_pthread_mutex_unlock(&test.mut);
154 static int testThreadReduction(void) {
157 POOL_ctx* const ctx = POOL_create(4 /*nbThreads*/, 2 /*queueSize*/);
161 memset(&test, 0, sizeof(test));
162 ASSERT_FALSE( ZSTD_pthread_mutex_init(&test.mut, NULL) );
163 ASSERT_FALSE( ZSTD_pthread_cond_init(&test.cond, NULL) );
165 result = testThreadReduction_internal(ctx, test);
167 ZSTD_pthread_mutex_destroy(&test.mut);
168 ZSTD_pthread_cond_destroy(&test.cond);
175 /* --- test abrupt ending --- */
178 ZSTD_pthread_mutex_t mut;
182 static void waitIncFn(void *opaque) {
183 abruptEndCanary_t* test = (abruptEndCanary_t*) opaque;
185 ZSTD_pthread_mutex_lock(&test->mut);
186 test->val = test->val + 1;
187 ZSTD_pthread_mutex_unlock(&test->mut);
190 static int testAbruptEnding_internal(abruptEndCanary_t test)
192 int const nbWaits = 16;
194 POOL_ctx* const ctx = POOL_create(3 /*numThreads*/, nbWaits /*queueSize*/);
199 for (i=0; i<nbWaits; i++)
200 POOL_add(ctx, &waitIncFn, &test); /* all jobs pushed into queue */
202 ASSERT_EQ( POOL_resize(ctx, 1 /*numThreads*/) , 0 ); /* downsize numThreads, to try to break end condition */
204 POOL_free(ctx); /* must finish all jobs in queue before giving back control */
205 ASSERT_EQ(test.val, nbWaits);
209 static int testAbruptEnding(void) {
211 abruptEndCanary_t test;
213 memset(&test, 0, sizeof(test));
214 ASSERT_FALSE( ZSTD_pthread_mutex_init(&test.mut, NULL) );
216 result = testAbruptEnding_internal(test);
218 ZSTD_pthread_mutex_destroy(&test.mut);
224 /* --- test launcher --- */
226 int main(int argc, const char **argv) {
231 if (POOL_create(0, 1)) { /* should not be possible */
232 printf("FAIL: should not create POOL with 0 threads\n");
236 for (numThreads = 1; numThreads <= 4; ++numThreads) {
238 for (queueSize = 0; queueSize <= 2; ++queueSize) {
239 printf("queueSize==%u, numThreads=%u \n",
240 (unsigned)queueSize, (unsigned)numThreads);
241 if (testOrder(numThreads, queueSize)) {
242 printf("FAIL: testOrder\n");
245 printf("SUCCESS: testOrder\n");
246 if (testWait(numThreads, queueSize)) {
247 printf("FAIL: testWait\n");
250 printf("SUCCESS: testWait\n");
254 if (testThreadReduction()) {
255 printf("FAIL: thread reduction not effective \n");
258 printf("SUCCESS: thread reduction effective \n");
261 if (testAbruptEnding()) {
262 printf("FAIL: jobs in queue not completed on early end \n");
265 printf("SUCCESS: all jobs in queue completed on early end \n");
268 printf("PASS: all POOL tests\n");