pthread_workqueue-0.8.2/testing/api/test.c

   1 #include <limits.h>
   2 #include <stdlib.h>
   3 #include <stdio.h>
   4 #include <string.h>
   5
   6 #if !defined(_WIN32)
   7 # include <sys/wait.h>
   8 # if !defined(NO_CONFIG_H)
   9 #  include "config.h"
  10 # endif
  11 # include <semaphore.h>
  12 #else
  13 # define inline _inline
  14 # include "../../src/windows/platform.h"
  15 # include "posix_semaphore.h"
  16 #endif
  17 #include "../../src/private.h"
  18
  19 #if HAVE_ERR_H
  20 # include <err.h>
  21 #else
  22 # define err(rc,msg,...) do { perror(msg); exit(rc); } while (0)
  23 # define errx(rc,msg,...) do { puts(msg); exit(rc); } while (0)
  24 #endif
  25
  26 #include <pthread_workqueue.h>
  27
  28 static int work_cnt;
  29
  30 /* If non-zero, extra debugging statements will be printed */
  31 static int dbg = 0;
  32
  33 static sem_t test_complete;
  34 static int test_rounds;
  35 #undef dbg_puts
  36 #define dbg_puts(s) if (dbg) puts(s)
  37 #undef dbg_printf
  38 #define dbg_printf(fmt,...) if (dbg) fprintf(stderr, fmt, __VA_ARGS__)
  39
  40 void additem(pthread_workqueue_t wq, void (*func)(void *),
  41              void * arg)
  42 {
  43
  44     int rv;
  45
  46     rv = pthread_workqueue_additem_np(wq, *func, arg, NULL, NULL);
  47     if (rv != 0)
  48         errx(1, "unable to add item: %s", strerror(rv));
  49     dbg_puts("added item\n");
  50 }
  51
  52 void
  53 mark_progress(void)
  54 {
  55     static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
  56
  57     pthread_mutex_lock(&mtx);
  58     test_rounds--;
  59     pthread_mutex_unlock(&mtx);
  60
  61     dbg_printf("rounds = %d\n", test_rounds);
  62     if (test_rounds == 0) {
  63         sem_post(&test_complete);
  64     }
  65 }
  66
  67 void
  68 sem_up(void *arg)
  69 {
  70     dbg_puts("semaphore UP\n");
  71     sem_post((sem_t *) arg);
  72     mark_progress();
  73 }
  74
  75 void
  76 sem_down(void *arg)
  77 {
  78     dbg_puts("semaphore DOWN\n");
  79     sem_wait((sem_t *) arg);
  80     dbg_puts("semaphore UP\n");
  81     sem_post((sem_t *) arg);
  82     mark_progress();
  83 }
  84
  85 void
  86 compute(void *arg)
  87 {
  88     int *count = (int *) arg;
  89 #define nval 5000
  90     int val[nval];
  91     int i,j;
  92
  93     /* Do some useless computation */
  94     for (i = 0; i < nval; i++) {
  95         val[i] = INT_MAX;
  96     }
  97     for (j = 0; j < nval; j++) {
  98         for (i = 0; i < nval; i++) {
  99             val[i] /= 3;
 100             val[i] *= 2;
 101             val[i] /= 4;
 102             val[i] *= 5;
 103         }
 104     }
 105
 106     if (count != NULL)
 107         (*count)--;
 108 }
 109
 110
 111 void
 112 sleepy(void *msg)
 113 {
 114     printf("%s\n", (char *) msg);
 115     if (strcmp(msg, "done") == 0)
 116         exit(0);
 117     sleep(random() % 6);
 118 }
 119
 120 void
 121 lazy(void *arg)
 122 {
 123     sleep(3);
 124     dbg_printf("item %lu complete\n", (unsigned long) arg);
 125         work_cnt--;
 126 }
 127
 128 void
 129 run_blocking_test(pthread_workqueue_t wq, int rounds)
 130 {
 131         long i = 0;
 132         work_cnt = rounds;
 133     for (i = 0; i < rounds; i++) {
 134         additem(wq, lazy, (void *) i);
 135     }
 136         while (work_cnt > 0)
 137                 sleep(1);
 138 }
 139
 140 void
 141 run_cond_wait_test(pthread_workqueue_t wq)
 142 {
 143         const int rounds = 10;
 144         long i = 0;
 145
 146         sleep(3);       /* Allow time for the workers to enter pthread_cond_wait() */
 147         work_cnt = rounds;
 148     for (i = 0; i < rounds; i++) {
 149         additem(wq, lazy, (void *) i);
 150                 sleep(1);
 151     }
 152         while (work_cnt > 0)
 153                 sleep(1);
 154 }
 155
 156 void
 157 run_load_test(pthread_workqueue_t wq)
 158 {
 159     char buf[16];
 160         int i = 0;
 161     for (i = 0; i < 1024; i++) {
 162         sprintf(buf, "%d", i);
 163         additem(wq, sleepy, strdup(buf));
 164         additem(wq, compute, NULL);
 165     }
 166     additem(wq, sleepy, "done");
 167 }
 168
 169 /* Try to overwhelm the CPU with computation requests */
 170 void
 171 run_stress_test(pthread_workqueue_t wq, int rounds)
 172 {
 173         int i = 0;
 174         work_cnt = rounds;
 175     for (i = 0; i < rounds; i++) {
 176         additem(wq, compute, &work_cnt);
 177     }
 178         while (work_cnt > 0)
 179                 sleep(1);
 180 }
 181
 182 /*
 183  * Ensure that the library is reinitialized after fork(2) is called.
 184  */
 185 void
 186 run_fork_test(pthread_workqueue_t wq)
 187 {
 188 #if !defined(_WIN32)
 189     pid_t pid;
 190     int   rv, status, timeout;
 191
 192     puts("fork test... ");
 193     pid = fork();
 194     if (pid < 0)
 195         err(1, "fork");
 196
 197     if (pid == 0) {
 198         /* Child */
 199         wq = NULL;
 200         rv = pthread_workqueue_create_np(&wq, NULL);
 201         if (rv < 0)
 202             errx(1, "pthread_workqueue_create_np");
 203         work_cnt = 1;
 204         timeout = 5;
 205         additem(wq, compute, &work_cnt);
 206         while (work_cnt > 0) {
 207             sleep(1);
 208             if (--timeout == 0)
 209                 errx(1, "work was not completed");
 210         }
 211         exit(0);
 212     } else {
 213         /* Parent */
 214         if (wait(&status) != pid)
 215             err(1, "waitpid");
 216         if (WEXITSTATUS(status) != 0)
 217             errx(1, "fork test failed");
 218         puts("ok\n");
 219     }
 220 #else
 221     puts("fork test... N/A\n");
 222 #endif
 223 }
 224
 225 void
 226 run_overcommit_test(pthread_workqueue_t wq)
 227 {
 228     sem_t sem;
 229     pthread_workqueue_t ocwq;
 230     pthread_workqueue_attr_t attr;
 231     int i, rv;
 232
 233     (void)wq;
 234     sem_init(&sem, 0, 0);
 235
 236     printf("pthread_workqueue_create_np() - overcommit enabled ");
 237     pthread_workqueue_attr_init_np(&attr);
 238     pthread_workqueue_attr_setovercommit_np(&attr, 1);
 239     rv = pthread_workqueue_create_np(&ocwq, &attr);
 240     if (rv != 0)
 241         err(1, "failed");
 242
 243     puts("ok\n");
 244
 245     printf("stress test - overcommit enabled ");
 246     run_stress_test(ocwq, 25);
 247     puts("ok\n");
 248
 249     /* FIXME: should use a multiple of the number of CPUs instead of magic number */
 250     printf("deadlock test - overcommit enabled ");
 251     test_rounds = 41;
 252     for (i = 0; i < 40; i++) {
 253         additem(ocwq, sem_down, &sem);
 254     }
 255     additem(ocwq, sem_up, &sem);
 256     sem_wait(&test_complete);
 257     puts("ok\n");
 258 }
 259
 260 int main() {
 261     pthread_workqueue_t wq;
 262     int rv;
 263
 264 #ifdef MAKE_STATIC
 265         pthread_workqueue_init_np();
 266 #endif
 267
 268     sem_init(&test_complete, 0, 0);
 269
 270     run_overcommit_test(NULL);
 271
 272     printf("pthread_workqueue_create_np().. ");
 273     rv = pthread_workqueue_create_np(&wq, NULL);
 274     if (rv != 0)
 275         err(1, "failed");
 276     printf("ok\n");
 277
 278     printf("stress test.. ");
 279     run_stress_test(wq, 25);
 280     printf("ok\n");
 281
 282     run_fork_test(wq);
 283
 284     //run_deadlock_test();
 285 //    run_cond_wait_test();
 286 //    run_blocking_test();
 287     //run_load_test();
 288
 289
 290         puts("All tests completed.\n");
 291     exit(0);
 292 }