source3/lib/pthreadpool/tests.c

   1 #include <stdio.h>
   2 #include <string.h>
   3 #include <poll.h>
   4 #include <errno.h>
   5 #include <stdlib.h>
   6 #include <pthread.h>
   7 #include <unistd.h>
   8 #include <sys/types.h>
   9 #include <sys/wait.h>
  10 #include "pthreadpool.h"
  11
  12 static int test_init(void)
  13 {
  14         struct pthreadpool *p;
  15         int ret;
  16
  17         ret = pthreadpool_init(1, &p);
  18         if (ret != 0) {
  19                 fprintf(stderr, "pthreadpool_init failed: %s\n",
  20                         strerror(ret));
  21                 return -1;
  22         }
  23         ret = pthreadpool_destroy(p);
  24         if (ret != 0) {
  25                 fprintf(stderr, "pthreadpool_init failed: %s\n",
  26                         strerror(ret));
  27                 return -1;
  28         }
  29         return 0;
  30 }
  31
  32 static void test_sleep(void *ptr)
  33 {
  34         int *ptimeout = (int *)ptr;
  35         int ret;
  36         ret = poll(NULL, 0, *ptimeout);
  37         if (ret != 0) {
  38                 fprintf(stderr, "poll returned %d (%s)\n",
  39                         ret, strerror(errno));
  40         }
  41 }
  42
  43 static int test_jobs(int num_threads, int num_jobs)
  44 {
  45         char *finished;
  46         struct pthreadpool *p;
  47         int timeout = 1;
  48         int i, ret;
  49
  50         finished = (char *)calloc(1, num_jobs);
  51         if (finished == NULL) {
  52                 fprintf(stderr, "calloc failed\n");
  53                 return -1;
  54         }
  55
  56         ret = pthreadpool_init(num_threads, &p);
  57         if (ret != 0) {
  58                 fprintf(stderr, "pthreadpool_init failed: %s\n",
  59                         strerror(ret));
  60                 return -1;
  61         }
  62
  63         for (i=0; i<num_jobs; i++) {
  64                 ret = pthreadpool_add_job(p, i, test_sleep, &timeout);
  65                 if (ret != 0) {
  66                         fprintf(stderr, "pthreadpool_add_job failed: %s\n",
  67                                 strerror(ret));
  68                         return -1;
  69                 }
  70         }
  71
  72         for (i=0; i<num_jobs; i++) {
  73                 int jobid = -1;
  74                 ret = pthreadpool_finished_job(p, &jobid);
  75                 if ((ret != 0) || (jobid >= num_jobs)) {
  76                         fprintf(stderr, "invalid job number %d\n", jobid);
  77                         return -1;
  78                 }
  79                 finished[jobid] += 1;
  80         }
  81
  82         for (i=0; i<num_jobs; i++) {
  83                 if (finished[i] != 1) {
  84                         fprintf(stderr, "finished[%d] = %d\n",
  85                                 i, finished[i]);
  86                         return -1;
  87                 }
  88         }
  89
  90         ret = pthreadpool_destroy(p);
  91         if (ret != 0) {
  92                 fprintf(stderr, "pthreadpool_destroy failed: %s\n",
  93                         strerror(ret));
  94                 return -1;
  95         }
  96
  97         free(finished);
  98         return 0;
  99 }
 100
 101 static int test_busydestroy(void)
 102 {
 103         struct pthreadpool *p;
 104         int timeout = 50;
 105         struct pollfd pfd;
 106         int ret;
 107
 108         ret = pthreadpool_init(1, &p);
 109         if (ret != 0) {
 110                 fprintf(stderr, "pthreadpool_init failed: %s\n",
 111                         strerror(ret));
 112                 return -1;
 113         }
 114         ret = pthreadpool_add_job(p, 1, test_sleep, &timeout);
 115         if (ret != 0) {
 116                 fprintf(stderr, "pthreadpool_add_job failed: %s\n",
 117                         strerror(ret));
 118                 return -1;
 119         }
 120         ret = pthreadpool_destroy(p);
 121         if (ret != EBUSY) {
 122                 fprintf(stderr, "Could destroy a busy pool\n");
 123                 return -1;
 124         }
 125
 126         pfd.fd = pthreadpool_signal_fd(p);
 127         pfd.events = POLLIN|POLLERR;
 128
 129         poll(&pfd, 1, -1);
 130
 131         ret = pthreadpool_destroy(p);
 132         if (ret != 0) {
 133                 fprintf(stderr, "pthreadpool_destroy failed: %s\n",
 134                         strerror(ret));
 135                 return -1;
 136         }
 137         return 0;
 138 }
 139
 140 struct threaded_state {
 141         pthread_t tid;
 142         struct pthreadpool *p;
 143         int start_job;
 144         int num_jobs;
 145         int timeout;
 146 };
 147
 148 static void *test_threaded_worker(void *p)
 149 {
 150         struct threaded_state *state = (struct threaded_state *)p;
 151         int i;
 152
 153         for (i=0; i<state->num_jobs; i++) {
 154                 int ret = pthreadpool_add_job(state->p, state->start_job + i,
 155                                               test_sleep, &state->timeout);
 156                 if (ret != 0) {
 157                         fprintf(stderr, "pthreadpool_add_job failed: %s\n",
 158                                 strerror(ret));
 159                         return NULL;
 160                 }
 161         }
 162         return NULL;
 163 }
 164
 165 static int test_threaded_addjob(int num_pools, int num_threads, int poolsize,
 166                                 int num_jobs)
 167 {
 168         struct pthreadpool **pools;
 169         struct threaded_state *states;
 170         struct threaded_state *state;
 171         struct pollfd *pfds;
 172         char *finished;
 173         pid_t child;
 174         int i, ret, poolnum;
 175         int received;
 176
 177         states = calloc(num_threads, sizeof(struct threaded_state));
 178         if (states == NULL) {
 179                 fprintf(stderr, "calloc failed\n");
 180                 return -1;
 181         }
 182
 183         finished = calloc(num_threads * num_jobs, 1);
 184         if (finished == NULL) {
 185                 fprintf(stderr, "calloc failed\n");
 186                 return -1;
 187         }
 188
 189         pools = calloc(num_pools, sizeof(struct pthreadpool *));
 190         if (pools == NULL) {
 191                 fprintf(stderr, "calloc failed\n");
 192                 return -1;
 193         }
 194
 195         pfds = calloc(num_pools, sizeof(struct pollfd));
 196         if (pfds == NULL) {
 197                 fprintf(stderr, "calloc failed\n");
 198                 return -1;
 199         }
 200
 201         for (i=0; i<num_pools; i++) {
 202                 ret = pthreadpool_init(poolsize, &pools[i]);
 203                 if (ret != 0) {
 204                         fprintf(stderr, "pthreadpool_init failed: %s\n",
 205                                 strerror(ret));
 206                         return -1;
 207                 }
 208                 pfds[i].fd = pthreadpool_signal_fd(pools[i]);
 209                 pfds[i].events = POLLIN|POLLHUP;
 210         }
 211
 212         poolnum = 0;
 213
 214         for (i=0; i<num_threads; i++) {
 215                 state = &states[i];
 216
 217                 state->p = pools[poolnum];
 218                 poolnum = (poolnum + 1) % num_pools;
 219
 220                 state->num_jobs = num_jobs;
 221                 state->timeout = 1;
 222                 state->start_job = i * num_jobs;
 223
 224                 ret = pthread_create(&state->tid, NULL, test_threaded_worker,
 225                                      state);
 226                 if (ret != 0) {
 227                         fprintf(stderr, "pthread_create failed: %s\n",
 228                                 strerror(ret));
 229                         return -1;
 230                 }
 231         }
 232
 233         if (random() % 1) {
 234                 poll(NULL, 0, 1);
 235         }
 236
 237         child = fork();
 238         if (child < 0) {
 239                 fprintf(stderr, "fork failed: %s\n", strerror(errno));
 240                 return -1;
 241         }
 242         if (child == 0) {
 243                 for (i=0; i<num_pools; i++) {
 244                         ret = pthreadpool_destroy(pools[i]);
 245                         if (ret != 0) {
 246                                 fprintf(stderr, "pthreadpool_destroy failed: "
 247                                         "%s\n", strerror(ret));
 248                                 exit(1);
 249                         }
 250                 }
 251                 /* child */
 252                 exit(0);
 253         }
 254
 255         for (i=0; i<num_threads; i++) {
 256                 ret = pthread_join(states[i].tid, NULL);
 257                 if (ret != 0) {
 258                         fprintf(stderr, "pthread_join(%d) failed: %s\n",
 259                                 i, strerror(ret));
 260                         return -1;
 261                 }
 262         }
 263
 264         received = 0;
 265
 266         while (received < num_threads*num_jobs) {
 267                 int j;
 268
 269                 ret = poll(pfds, num_pools, 1000);
 270                 if (ret == -1) {
 271                         fprintf(stderr, "poll failed: %s\n",
 272                                 strerror(errno));
 273                         return -1;
 274                 }
 275                 if (ret == 0) {
 276                         fprintf(stderr, "\npoll timed out\n");
 277                         break;
 278                 }
 279
 280                 for (j=0; j<num_pools; j++) {
 281                         int jobid = -1;
 282
 283                         if ((pfds[j].revents & (POLLIN|POLLHUP)) == 0) {
 284                                 continue;
 285                         }
 286
 287                         ret = pthreadpool_finished_job(pools[j], &jobid);
 288                         if ((ret != 0) || (jobid >= num_jobs * num_threads)) {
 289                                 fprintf(stderr, "invalid job number %d\n",
 290                                         jobid);
 291                                 return -1;
 292                         }
 293                         finished[jobid] += 1;
 294                         received += 1;
 295                 }
 296         }
 297
 298         for (i=0; i<num_threads*num_jobs; i++) {
 299                 if (finished[i] != 1) {
 300                         fprintf(stderr, "finished[%d] = %d\n",
 301                                 i, finished[i]);
 302                         return -1;
 303                 }
 304         }
 305
 306         for (i=0; i<num_pools; i++) {
 307                 ret = pthreadpool_destroy(pools[i]);
 308                 if (ret != 0) {
 309                         fprintf(stderr, "pthreadpool_destroy failed: %s\n",
 310                                 strerror(ret));
 311                         return -1;
 312                 }
 313         }
 314
 315         free(pfds);
 316         free(pools);
 317         free(states);
 318         free(finished);
 319
 320         return 0;
 321 }
 322
 323 static int test_fork(void)
 324 {
 325         struct pthreadpool *p;
 326         pid_t child, waited;
 327         int status, ret;
 328
 329         ret = pthreadpool_init(1, &p);
 330         if (ret != 0) {
 331                 fprintf(stderr, "pthreadpool_init failed: %s\n",
 332                         strerror(ret));
 333                 return -1;
 334         }
 335         ret = pthreadpool_destroy(p);
 336         if (ret != 0) {
 337                 fprintf(stderr, "pthreadpool_destroy failed: %s\n",
 338                         strerror(ret));
 339                 return -1;
 340         }
 341
 342         child = fork();
 343         if (child < 0) {
 344                 perror("fork failed");
 345                 return -1;
 346         }
 347         if (child == 0) {
 348                 exit(0);
 349         }
 350         waited = wait(&status);
 351         if (waited == -1) {
 352                 perror("wait failed");
 353                 return -1;
 354         }
 355         if (waited != child) {
 356                 fprintf(stderr, "expected child %d, got %d\n",
 357                         (int)child, (int)waited);
 358                 return -1;
 359         }
 360         return 0;
 361 }
 362
 363 int main(void)
 364 {
 365         int ret;
 366
 367         ret = test_init();
 368         if (ret != 0) {
 369                 fprintf(stderr, "test_init failed\n");
 370                 return 1;
 371         }
 372
 373         ret = test_fork();
 374         if (ret != 0) {
 375                 fprintf(stderr, "test_fork failed\n");
 376                 return 1;
 377         }
 378
 379         ret = test_jobs(10, 10000);
 380         if (ret != 0) {
 381                 fprintf(stderr, "test_jobs failed\n");
 382                 return 1;
 383         }
 384
 385         ret = test_busydestroy();
 386         if (ret != 0) {
 387                 fprintf(stderr, "test_busydestroy failed\n");
 388                 return 1;
 389         }
 390
 391         /*
 392          * Test 10 threads adding jobs on a single pool
 393          */
 394         ret = test_threaded_addjob(1, 10, 5, 5000);
 395         if (ret != 0) {
 396                 fprintf(stderr, "test_jobs failed\n");
 397                 return 1;
 398         }
 399
 400         /*
 401          * Test 10 threads on 3 pools to verify our fork handling
 402          * works right.
 403          */
 404         ret = test_threaded_addjob(3, 10, 5, 5000);
 405         if (ret != 0) {
 406                 fprintf(stderr, "test_jobs failed\n");
 407                 return 1;
 408         }
 409
 410         printf("success\n");
 411         return 0;
 412 }