tests/poolTests.c

   1 /*
   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 #include "pool.h"
  13 #include "threading.h"
  14 #include "util.h"
  15 #include "timefn.h"
  16 #include <stddef.h>
  17 #include <stdio.h>
  18
  19 #define ASSERT_TRUE(p)                                                       \
  20   do {                                                                       \
  21     if (!(p)) {                                                              \
  22       return 1;                                                              \
  23     }                                                                        \
  24   } while (0)
  25 #define ASSERT_FALSE(p) ASSERT_TRUE(!(p))
  26 #define ASSERT_EQ(lhs, rhs) ASSERT_TRUE((lhs) == (rhs))
  27
  28 struct data {
  29   ZSTD_pthread_mutex_t mutex;
  30   unsigned data[16];
  31   size_t i;
  32 };
  33
  34 static void fn(void *opaque)
  35 {
  36   struct data *data = (struct data *)opaque;
  37   ZSTD_pthread_mutex_lock(&data->mutex);
  38   data->data[data->i] = (unsigned)(data->i);
  39   ++data->i;
  40   ZSTD_pthread_mutex_unlock(&data->mutex);
  41 }
  42
  43 static int testOrder(size_t numThreads, size_t queueSize)
  44 {
  45   struct data data;
  46   POOL_ctx* const ctx = POOL_create(numThreads, queueSize);
  47   ASSERT_TRUE(ctx);
  48   data.i = 0;
  49   (void)ZSTD_pthread_mutex_init(&data.mutex, NULL);
  50   { size_t i;
  51     for (i = 0; i < 16; ++i) {
  52       POOL_add(ctx, &fn, &data);
  53     }
  54   }
  55   POOL_free(ctx);
  56   ASSERT_EQ(16, data.i);
  57   { size_t i;
  58     for (i = 0; i < data.i; ++i) {
  59       ASSERT_EQ(i, data.data[i]);
  60     }
  61   }
  62   ZSTD_pthread_mutex_destroy(&data.mutex);
  63   return 0;
  64 }
  65
  66
  67 /* --- test deadlocks --- */
  68
  69 static void waitFn(void *opaque) {
  70   (void)opaque;
  71   UTIL_sleepMilli(1);
  72 }
  73
  74 /* Tests for deadlock */
  75 static int testWait(size_t numThreads, size_t queueSize) {
  76   struct data data;
  77   POOL_ctx* const ctx = POOL_create(numThreads, queueSize);
  78   ASSERT_TRUE(ctx);
  79   { size_t i;
  80     for (i = 0; i < 16; ++i) {
  81         POOL_add(ctx, &waitFn, &data);
  82     }
  83   }
  84   POOL_free(ctx);
  85   return 0;
  86 }
  87
  88
  89 /* --- test POOL_resize() --- */
  90
  91 typedef struct {
  92     ZSTD_pthread_mutex_t mut;
  93     int val;
  94     int max;
  95     ZSTD_pthread_cond_t cond;
  96 } poolTest_t;
  97
  98 static void waitLongFn(void *opaque) {
  99   poolTest_t* const test = (poolTest_t*) opaque;
 100   UTIL_sleepMilli(10);
 101   ZSTD_pthread_mutex_lock(&test->mut);
 102   test->val = test->val + 1;
 103   if (test->val == test->max)
 104     ZSTD_pthread_cond_signal(&test->cond);
 105   ZSTD_pthread_mutex_unlock(&test->mut);
 106 }
 107
 108 static int testThreadReduction_internal(POOL_ctx* ctx, poolTest_t test)
 109 {
 110     int const nbWaits = 16;
 111     UTIL_time_t startTime;
 112     U64 time4threads, time2threads;
 113
 114     test.val = 0;
 115     test.max = nbWaits;
 116
 117     startTime = UTIL_getTime();
 118     {   int i;
 119         for (i=0; i<nbWaits; i++)
 120             POOL_add(ctx, &waitLongFn, &test);
 121     }
 122     ZSTD_pthread_mutex_lock(&test.mut);
 123     ZSTD_pthread_cond_wait(&test.cond, &test.mut);
 124     ASSERT_EQ(test.val, nbWaits);
 125     ZSTD_pthread_mutex_unlock(&test.mut);
 126     time4threads = UTIL_clockSpanNano(startTime);
 127
 128     ASSERT_EQ( POOL_resize(ctx, 2/*nbThreads*/) , 0 );
 129     test.val = 0;
 130     startTime = UTIL_getTime();
 131     {   int i;
 132         for (i=0; i<nbWaits; i++)
 133             POOL_add(ctx, &waitLongFn, &test);
 134     }
 135     ZSTD_pthread_mutex_lock(&test.mut);
 136     ZSTD_pthread_cond_wait(&test.cond, &test.mut);
 137     ASSERT_EQ(test.val, nbWaits);
 138     ZSTD_pthread_mutex_unlock(&test.mut);
 139     time2threads = UTIL_clockSpanNano(startTime);
 140
 141     if (time4threads >= time2threads) return 1;   /* check 4 threads were effectively faster than 2 */
 142     return 0;
 143 }
 144
 145 static int testThreadReduction(void) {
 146     int result;
 147     poolTest_t test;
 148     POOL_ctx* const ctx = POOL_create(4 /*nbThreads*/, 2 /*queueSize*/);
 149
 150     ASSERT_TRUE(ctx);
 151
 152     memset(&test, 0, sizeof(test));
 153     ASSERT_FALSE( ZSTD_pthread_mutex_init(&test.mut, NULL) );
 154     ASSERT_FALSE( ZSTD_pthread_cond_init(&test.cond, NULL) );
 155
 156     result = testThreadReduction_internal(ctx, test);
 157
 158     ZSTD_pthread_mutex_destroy(&test.mut);
 159     ZSTD_pthread_cond_destroy(&test.cond);
 160     POOL_free(ctx);
 161
 162     return result;
 163 }
 164
 165
 166 /* --- test abrupt ending --- */
 167
 168 typedef struct {
 169     ZSTD_pthread_mutex_t mut;
 170     int val;
 171 } abruptEndCanary_t;
 172
 173 static void waitIncFn(void *opaque) {
 174   abruptEndCanary_t* test = (abruptEndCanary_t*) opaque;
 175   UTIL_sleepMilli(10);
 176   ZSTD_pthread_mutex_lock(&test->mut);
 177   test->val = test->val + 1;
 178   ZSTD_pthread_mutex_unlock(&test->mut);
 179 }
 180
 181 static int testAbruptEnding_internal(abruptEndCanary_t test)
 182 {
 183     int const nbWaits = 16;
 184
 185     POOL_ctx* const ctx = POOL_create(3 /*numThreads*/, nbWaits /*queueSize*/);
 186     ASSERT_TRUE(ctx);
 187     test.val = 0;
 188
 189     {   int i;
 190         for (i=0; i<nbWaits; i++)
 191             POOL_add(ctx, &waitIncFn, &test);  /* all jobs pushed into queue */
 192     }
 193     ASSERT_EQ( POOL_resize(ctx, 1 /*numThreads*/) , 0 );   /* downsize numThreads, to try to break end condition */
 194
 195     POOL_free(ctx);  /* must finish all jobs in queue before giving back control */
 196     ASSERT_EQ(test.val, nbWaits);
 197     return 0;
 198 }
 199
 200 static int testAbruptEnding(void) {
 201     int result;
 202     abruptEndCanary_t test;
 203
 204     memset(&test, 0, sizeof(test));
 205     ASSERT_FALSE( ZSTD_pthread_mutex_init(&test.mut, NULL) );
 206
 207     result = testAbruptEnding_internal(test);
 208
 209     ZSTD_pthread_mutex_destroy(&test.mut);
 210     return result;
 211 }
 212
 213
 214
 215 /* --- test launcher --- */
 216
 217 int main(int argc, const char **argv) {
 218   size_t numThreads;
 219   (void)argc;
 220   (void)argv;
 221
 222   if (POOL_create(0, 1)) {   /* should not be possible */
 223     printf("FAIL: should not create POOL with 0 threads\n");
 224     return 1;
 225   }
 226
 227   for (numThreads = 1; numThreads <= 4; ++numThreads) {
 228     size_t queueSize;
 229     for (queueSize = 0; queueSize <= 2; ++queueSize) {
 230       printf("queueSize==%u, numThreads=%u \n",
 231             (unsigned)queueSize, (unsigned)numThreads);
 232       if (testOrder(numThreads, queueSize)) {
 233         printf("FAIL: testOrder\n");
 234         return 1;
 235       }
 236       printf("SUCCESS: testOrder\n");
 237       if (testWait(numThreads, queueSize)) {
 238         printf("FAIL: testWait\n");
 239         return 1;
 240       }
 241       printf("SUCCESS: testWait\n");
 242     }
 243   }
 244
 245   if (testThreadReduction()) {
 246       printf("FAIL: thread reduction not effective \n");
 247       return 1;
 248   } else {
 249       printf("SUCCESS: thread reduction effective (slower execution) \n");
 250   }
 251
 252   if (testAbruptEnding()) {
 253       printf("FAIL: jobs in queue not completed on early end \n");
 254       return 1;
 255   } else {
 256       printf("SUCCESS: all jobs in queue completed on early end \n");
 257   }
 258
 259   printf("PASS: all POOL tests\n");
 260
 261   return 0;
 262 }