tests/test_threadpool.c

   1 /*
   2  * Asterisk -- An open source telephony toolkit.
   3  *
   4  * Copyright (C) 2012, Digium, Inc.
   5  *
   6  * Mark Michelson <mmichelson@digium.com>
   7  *
   8  * See http://www.asterisk.org for more information about
   9  * the Asterisk project. Please do not directly contact
  10  * any of the maintainers of this project for assistance;
  11  * the project provides a web site, mailing lists and IRC
  12  * channels for your use.
  13  *
  14  * This program is free software, distributed under the terms of
  15  * the GNU General Public License Version 2. See the LICENSE file
  16  * at the top of the source tree.
  17  */
  18
  19 /*!
  20  * \file
  21  * \brief threadpool unit tests
  22  *
  23  * \author Mark Michelson <mmichelson@digium.com>
  24  *
  25  */
  26
  27 /*** MODULEINFO
  28         <depend>TEST_FRAMEWORK</depend>
  29         <support_level>core</support_level>
  30  ***/
  31
  32 #include "asterisk.h"
  33
  34 #include "asterisk/test.h"
  35 #include "asterisk/threadpool.h"
  36 #include "asterisk/module.h"
  37 #include "asterisk/lock.h"
  38 #include "asterisk/astobj2.h"
  39 #include "asterisk/logger.h"
  40
  41 struct test_listener_data {
  42         int num_active;
  43         int num_idle;
  44         int task_pushed;
  45         int num_tasks;
  46         int empty_notice;
  47         int was_empty;
  48         ast_mutex_t lock;
  49         ast_cond_t cond;
  50 };
  51
  52 static void *test_alloc(struct ast_threadpool_listener *listener)
  53 {
  54         struct test_listener_data *tld = ast_calloc(1, sizeof(*tld));
  55         if (!tld) {
  56                 return NULL;
  57         }
  58         ast_mutex_init(&tld->lock);
  59         ast_cond_init(&tld->cond, NULL);
  60         return tld;
  61 }
  62
  63 static void test_state_changed(struct ast_threadpool *pool,
  64                 struct ast_threadpool_listener *listener,
  65                 int active_threads,
  66                 int idle_threads)
  67 {
  68         struct test_listener_data *tld = listener->private_data;
  69         SCOPED_MUTEX(lock, &tld->lock);
  70         ast_log(LOG_NOTICE, "State changed: num_active: %d, num_idle: %d\n", active_threads, idle_threads);
  71         tld->num_active = active_threads;
  72         tld->num_idle = idle_threads;
  73         ast_cond_signal(&tld->cond);
  74 }
  75
  76 static void test_task_pushed(struct ast_threadpool *pool,
  77                 struct ast_threadpool_listener *listener,
  78                 int was_empty)
  79 {
  80         struct test_listener_data *tld = listener->private_data;
  81         SCOPED_MUTEX(lock, &tld->lock);
  82         tld->task_pushed = 1;
  83         ++tld->num_tasks;
  84         tld->was_empty = was_empty;
  85         ast_cond_signal(&tld->cond);
  86 }
  87
  88 static void test_emptied(struct ast_threadpool *pool,
  89                 struct ast_threadpool_listener *listener)
  90 {
  91         struct test_listener_data *tld = listener->private_data;
  92         SCOPED_MUTEX(lock, &tld->lock);
  93         tld->empty_notice = 1;
  94         ast_cond_signal(&tld->cond);
  95 }
  96
  97 static void test_destroy(void *private_data)
  98 {
  99         struct test_listener_data *tld = private_data;
 100         ast_debug(1, "Poop\n");
 101         ast_cond_destroy(&tld->cond);
 102         ast_mutex_destroy(&tld->lock);
 103         ast_free(tld);
 104 }
 105
 106 static const struct ast_threadpool_listener_callbacks test_callbacks = {
 107         .alloc = test_alloc,
 108         .state_changed = test_state_changed,
 109         .task_pushed = test_task_pushed,
 110         .emptied = test_emptied,
 111         .destroy = test_destroy,
 112 };
 113
 114 struct simple_task_data {
 115         int task_executed;
 116         ast_mutex_t lock;
 117         ast_cond_t cond;
 118 };
 119
 120 static struct simple_task_data *simple_task_data_alloc(void)
 121 {
 122         struct simple_task_data *std = ast_calloc(1, sizeof(*std));
 123
 124         if (!std) {
 125                 return NULL;
 126         }
 127         ast_mutex_init(&std->lock);
 128         ast_cond_init(&std->cond, NULL);
 129         return std;
 130 }
 131
 132 static int simple_task(void *data)
 133 {
 134         struct simple_task_data *std = data;
 135         SCOPED_MUTEX(lock, &std->lock);
 136         std->task_executed = 1;
 137         ast_cond_signal(&std->cond);
 138         return 0;
 139 }
 140
 141 #define WAIT_WHILE(tld, condition) \
 142 {\
 143         ast_mutex_lock(&tld->lock);\
 144         while ((condition)) {\
 145                 ast_cond_wait(&tld->cond, &tld->lock);\
 146         }\
 147         ast_mutex_unlock(&tld->lock);\
 148 }\
 149
 150 static void wait_for_task_pushed(struct ast_threadpool_listener *listener)
 151 {
 152         struct test_listener_data *tld = listener->private_data;
 153         struct timeval start = ast_tvnow();
 154         struct timespec end = {
 155                 .tv_sec = start.tv_sec + 5,
 156                 .tv_nsec = start.tv_usec * 1000
 157         };
 158         SCOPED_MUTEX(lock, &tld->lock);
 159
 160         while (!tld->task_pushed) {
 161                 ast_cond_timedwait(&tld->cond, lock, &end);
 162         }
 163 }
 164
 165 static enum ast_test_result_state wait_for_completion(struct simple_task_data *std)
 166 {
 167         struct timeval start = ast_tvnow();
 168         struct timespec end = {
 169                 .tv_sec = start.tv_sec + 5,
 170                 .tv_nsec = start.tv_usec * 1000
 171         };
 172         enum ast_test_result_state res = AST_TEST_PASS;
 173         SCOPED_MUTEX(lock, &std->lock);
 174
 175         while (!std->task_executed) {
 176                 ast_cond_timedwait(&std->cond, lock, &end);
 177         }
 178
 179         if (!std->task_executed) {
 180                 res = AST_TEST_FAIL;
 181         }
 182         return res;
 183 }
 184
 185 static enum ast_test_result_state wait_for_empty_notice(struct test_listener_data *tld)
 186 {
 187         struct timeval start = ast_tvnow();
 188         struct timespec end = {
 189                 .tv_sec = start.tv_sec + 5,
 190                 .tv_nsec = start.tv_usec * 1000
 191         };
 192         enum ast_test_result_state res = AST_TEST_PASS;
 193         SCOPED_MUTEX(lock, &tld->lock);
 194
 195         while (!tld->empty_notice) {
 196                 ast_cond_timedwait(&tld->cond, lock, &end);
 197         }
 198
 199         if (!tld->empty_notice) {
 200                 res = AST_TEST_FAIL;
 201         }
 202
 203         return res;
 204 }
 205
 206 static enum ast_test_result_state listener_check(
 207                 struct ast_test *test,
 208                 struct ast_threadpool_listener *listener,
 209                 int task_pushed,
 210                 int was_empty,
 211                 int num_tasks,
 212                 int num_active,
 213                 int num_idle,
 214                 int empty_notice)
 215 {
 216         struct test_listener_data *tld = listener->private_data;
 217         enum ast_test_result_state res = AST_TEST_PASS;
 218
 219         if (tld->task_pushed != task_pushed) {
 220                 ast_test_status_update(test, "Expected task %sto be pushed, but it was%s\n",
 221                                 task_pushed ? "" : "not ", tld->task_pushed ? "" : " not");
 222                 res = AST_TEST_FAIL;
 223         }
 224         if (tld->was_empty != was_empty) {
 225                 ast_test_status_update(test, "Expected %sto be empty, but it was%s\n",
 226                                 was_empty ? "" : "not ", tld->was_empty ? "" : " not");
 227                 res = AST_TEST_FAIL;
 228         }
 229         if (tld->num_tasks!= num_tasks) {
 230                 ast_test_status_update(test, "Expected %d tasks to be pushed, but got %d\n",
 231                                 num_tasks, tld->num_tasks);
 232                 res = AST_TEST_FAIL;
 233         }
 234         if (tld->num_active != num_active) {
 235                 ast_test_status_update(test, "Expected %d active threads, but got %d\n",
 236                                 num_active, tld->num_active);
 237                 res = AST_TEST_FAIL;
 238         }
 239         if (tld->num_idle != num_idle) {
 240                 ast_test_status_update(test, "Expected %d idle threads, but got %d\n",
 241                                 num_idle, tld->num_idle);
 242                 res = AST_TEST_FAIL;
 243         }
 244         if (tld->empty_notice != empty_notice) {
 245                 ast_test_status_update(test, "Expected %s empty notice, but got %s\n",
 246                                 was_empty ? "an" : "no", tld->task_pushed ? "one" : "none");
 247                 res = AST_TEST_FAIL;
 248         }
 249
 250         return res;
 251 }
 252
 253 AST_TEST_DEFINE(threadpool_push)
 254 {
 255         struct ast_threadpool *pool = NULL;
 256         struct ast_threadpool_listener *listener = NULL;
 257         struct simple_task_data *std = NULL;
 258         enum ast_test_result_state res = AST_TEST_FAIL;
 259
 260         switch (cmd) {
 261         case TEST_INIT:
 262                 info->name = "threadpool_push";
 263                 info->category = "/main/threadpool/";
 264                 info->summary = "Test task";
 265                 info->description =
 266                         "Basic threadpool test";
 267                 return AST_TEST_NOT_RUN;
 268         case TEST_EXECUTE:
 269                 break;
 270         }
 271
 272         listener = ast_threadpool_listener_alloc(&test_callbacks);
 273         if (!listener) {
 274                 return AST_TEST_FAIL;
 275         }
 276
 277         pool = ast_threadpool_create(listener, 0);
 278         if (!pool) {
 279                 goto end;
 280         }
 281
 282         std = simple_task_data_alloc();
 283         if (!std) {
 284                 goto end;
 285         }
 286
 287         ast_threadpool_push(pool, simple_task, std);
 288
 289         wait_for_task_pushed(listener);
 290
 291         res = listener_check(test, listener, 1, 1, 1, 0, 0, 0);
 292
 293 end:
 294         if (pool) {
 295                 ast_threadpool_shutdown(pool);
 296         }
 297         ao2_cleanup(listener);
 298         ast_free(std);
 299         return res;
 300 }
 301
 302 AST_TEST_DEFINE(threadpool_thread_creation)
 303 {
 304         struct ast_threadpool *pool = NULL;
 305         struct ast_threadpool_listener *listener = NULL;
 306         enum ast_test_result_state res = AST_TEST_FAIL;
 307         struct test_listener_data *tld;
 308
 309         switch (cmd) {
 310         case TEST_INIT:
 311                 info->name = "threadpool_thread_creation";
 312                 info->category = "/main/threadpool/";
 313                 info->summary = "Test threadpool thread creation";
 314                 info->description =
 315                         "Ensure that threads can be added to a threadpool";
 316                 return AST_TEST_NOT_RUN;
 317         case TEST_EXECUTE:
 318                 break;
 319         }
 320
 321         listener = ast_threadpool_listener_alloc(&test_callbacks);
 322         if (!listener) {
 323                 return AST_TEST_FAIL;
 324         }
 325         tld = listener->private_data;
 326
 327         pool = ast_threadpool_create(listener, 0);
 328         if (!pool) {
 329                 goto end;
 330         }
 331
 332         /* Now let's create a thread. It should start active, then go
 333          * idle immediately
 334          */
 335         ast_threadpool_set_size(pool, 1);
 336
 337         WAIT_WHILE(tld, tld->num_idle == 0);
 338
 339         res = listener_check(test, listener, 0, 0, 0, 0, 1, 0);
 340
 341 end:
 342         if (pool) {
 343                 ast_threadpool_shutdown(pool);
 344         }
 345         ao2_cleanup(listener);
 346         return res;
 347 }
 348
 349 AST_TEST_DEFINE(threadpool_thread_destruction)
 350 {
 351         struct ast_threadpool *pool = NULL;
 352         struct ast_threadpool_listener *listener = NULL;
 353         enum ast_test_result_state res = AST_TEST_FAIL;
 354         struct test_listener_data *tld;
 355
 356         switch (cmd) {
 357         case TEST_INIT:
 358                 info->name = "threadpool_thread_destruction";
 359                 info->category = "/main/threadpool/";
 360                 info->summary = "Test threadpool thread destruction";
 361                 info->description =
 362                         "Ensure that threads are properly destroyed in a threadpool";
 363                 return AST_TEST_NOT_RUN;
 364         case TEST_EXECUTE:
 365                 break;
 366         }
 367
 368         listener = ast_threadpool_listener_alloc(&test_callbacks);
 369         if (!listener) {
 370                 return AST_TEST_FAIL;
 371         }
 372         tld = listener->private_data;
 373
 374         pool = ast_threadpool_create(listener, 0);
 375         if (!pool) {
 376                 goto end;
 377         }
 378
 379         ast_threadpool_set_size(pool, 3);
 380
 381         WAIT_WHILE(tld, tld->num_idle < 3);
 382
 383         res = listener_check(test, listener, 0, 0, 0, 0, 3, 0);
 384         if (res == AST_TEST_FAIL) {
 385                 goto end;
 386         }
 387
 388         ast_threadpool_set_size(pool, 2);
 389
 390         WAIT_WHILE(tld, tld->num_idle > 2);
 391
 392         res = listener_check(test, listener, 0, 0, 0, 0, 2, 0);
 393
 394 end:
 395         if (pool) {
 396                 ast_threadpool_shutdown(pool);
 397         }
 398         ao2_cleanup(listener);
 399         return res;
 400 }
 401
 402 AST_TEST_DEFINE(threadpool_one_task_one_thread)
 403 {
 404         struct ast_threadpool *pool = NULL;
 405         struct ast_threadpool_listener *listener = NULL;
 406         struct simple_task_data *std = NULL;
 407         enum ast_test_result_state res = AST_TEST_FAIL;
 408         struct test_listener_data *tld;
 409
 410         switch (cmd) {
 411         case TEST_INIT:
 412                 info->name = "threadpool_one_task_one_thread";
 413                 info->category = "/main/threadpool/";
 414                 info->summary = "Test a single task with a single thread";
 415                 info->description =
 416                         "Push a task into an empty threadpool, then add a thread to the pool.";
 417                 return AST_TEST_NOT_RUN;
 418         case TEST_EXECUTE:
 419                 break;
 420         }
 421
 422         listener = ast_threadpool_listener_alloc(&test_callbacks);
 423         if (!listener) {
 424                 return AST_TEST_FAIL;
 425         }
 426         tld = listener->private_data;
 427
 428         pool = ast_threadpool_create(listener, 0);
 429         if (!pool) {
 430                 goto end;
 431         }
 432
 433         std = simple_task_data_alloc();
 434         if (!std) {
 435                 goto end;
 436         }
 437
 438         ast_threadpool_push(pool, simple_task, std);
 439
 440         ast_threadpool_set_size(pool, 1);
 441
 442         /* Threads added to the pool are active when they start,
 443          * so the newly-created thread should immediately execute
 444          * the waiting task.
 445          */
 446         res = wait_for_completion(std);
 447         if (res == AST_TEST_FAIL) {
 448                 goto end;
 449         }
 450
 451         res = wait_for_empty_notice(tld);
 452         if (res == AST_TEST_FAIL) {
 453                 goto end;
 454         }
 455
 456         /* After completing the task, the thread should go idle */
 457         WAIT_WHILE(tld, tld->num_idle == 0);
 458
 459         res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
 460
 461 end:
 462         if (pool) {
 463                 ast_threadpool_shutdown(pool);
 464         }
 465         ao2_cleanup(listener);
 466         ast_free(std);
 467         return res;
 468
 469 }
 470
 471 AST_TEST_DEFINE(threadpool_one_thread_one_task)
 472 {
 473         struct ast_threadpool *pool = NULL;
 474         struct ast_threadpool_listener *listener = NULL;
 475         struct simple_task_data *std = NULL;
 476         enum ast_test_result_state res = AST_TEST_FAIL;
 477         struct test_listener_data *tld;
 478
 479         switch (cmd) {
 480         case TEST_INIT:
 481                 info->name = "threadpool_one_thread_one_task";
 482                 info->category = "/main/threadpool/";
 483                 info->summary = "Test a single thread with a single task";
 484                 info->description =
 485                         "Add a thread to the pool and then push a task to it.";
 486                 return AST_TEST_NOT_RUN;
 487         case TEST_EXECUTE:
 488                 break;
 489         }
 490
 491         listener = ast_threadpool_listener_alloc(&test_callbacks);
 492         if (!listener) {
 493                 return AST_TEST_FAIL;
 494         }
 495         tld = listener->private_data;
 496
 497         pool = ast_threadpool_create(listener, 0);
 498         if (!pool) {
 499                 goto end;
 500         }
 501
 502         std = simple_task_data_alloc();
 503         if (!std) {
 504                 goto end;
 505         }
 506
 507         ast_threadpool_set_size(pool, 1);
 508
 509         WAIT_WHILE(tld, tld->num_idle == 0);
 510
 511         ast_threadpool_push(pool, simple_task, std);
 512
 513         res = wait_for_completion(std);
 514         if (res == AST_TEST_FAIL) {
 515                 goto end;
 516         }
 517
 518         res = wait_for_empty_notice(tld);
 519         if (res == AST_TEST_FAIL) {
 520                 goto end;
 521         }
 522
 523         /* After completing the task, the thread should go idle */
 524         WAIT_WHILE(tld, tld->num_idle == 0);
 525
 526         res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
 527
 528 end:
 529         if (pool) {
 530                 ast_threadpool_shutdown(pool);
 531         }
 532         ao2_cleanup(listener);
 533         ast_free(std);
 534         return res;
 535
 536 }
 537
 538 AST_TEST_DEFINE(threadpool_one_thread_multiple_tasks)
 539 {
 540         struct ast_threadpool *pool = NULL;
 541         struct ast_threadpool_listener *listener = NULL;
 542         struct simple_task_data *std1 = NULL;
 543         struct simple_task_data *std2 = NULL;
 544         struct simple_task_data *std3 = NULL;
 545         enum ast_test_result_state res = AST_TEST_FAIL;
 546         struct test_listener_data *tld;
 547
 548         switch (cmd) {
 549         case TEST_INIT:
 550                 info->name = "threadpool_one_thread_multiple_tasks";
 551                 info->category = "/main/threadpool/";
 552                 info->summary = "Test a single thread with multiple tasks";
 553                 info->description =
 554                         "Add a thread to the pool and then push three tasks to it.";
 555                 return AST_TEST_NOT_RUN;
 556         case TEST_EXECUTE:
 557                 break;
 558         }
 559
 560         listener = ast_threadpool_listener_alloc(&test_callbacks);
 561         if (!listener) {
 562                 return AST_TEST_FAIL;
 563         }
 564         tld = listener->private_data;
 565
 566         pool = ast_threadpool_create(listener, 0);
 567         if (!pool) {
 568                 goto end;
 569         }
 570
 571         std1 = simple_task_data_alloc();
 572         std2 = simple_task_data_alloc();
 573         std3 = simple_task_data_alloc();
 574         if (!std1 || !std2 || !std3) {
 575                 goto end;
 576         }
 577
 578         ast_threadpool_set_size(pool, 1);
 579
 580         WAIT_WHILE(tld, tld->num_idle == 0);
 581
 582         ast_threadpool_push(pool, simple_task, std1);
 583         ast_threadpool_push(pool, simple_task, std2);
 584         ast_threadpool_push(pool, simple_task, std3);
 585
 586         res = wait_for_completion(std1);
 587         if (res == AST_TEST_FAIL) {
 588                 goto end;
 589         }
 590         res = wait_for_completion(std2);
 591         if (res == AST_TEST_FAIL) {
 592                 goto end;
 593         }
 594         res = wait_for_completion(std3);
 595         if (res == AST_TEST_FAIL) {
 596                 goto end;
 597         }
 598
 599         res = wait_for_empty_notice(tld);
 600         if (res == AST_TEST_FAIL) {
 601                 goto end;
 602         }
 603
 604         WAIT_WHILE(tld, tld->num_idle == 0);
 605
 606         res = listener_check(test, listener, 1, 0, 3, 0, 1, 1);
 607
 608 end:
 609         if (pool) {
 610                 ast_threadpool_shutdown(pool);
 611         }
 612         ao2_cleanup(listener);
 613         ast_free(std1);
 614         ast_free(std2);
 615         ast_free(std3);
 616         return res;
 617
 618 }
 619
 620 static int unload_module(void)
 621 {
 622         ast_test_unregister(threadpool_push);
 623         ast_test_unregister(threadpool_thread_creation);
 624         ast_test_unregister(threadpool_thread_destruction);
 625         ast_test_unregister(threadpool_one_task_one_thread);
 626         ast_test_unregister(threadpool_one_thread_one_task);
 627         ast_test_unregister(threadpool_one_thread_multiple_tasks);
 628         return 0;
 629 }
 630
 631 static int load_module(void)
 632 {
 633         ast_test_register(threadpool_push);
 634         ast_test_register(threadpool_thread_creation);
 635         ast_test_register(threadpool_thread_destruction);
 636         ast_test_register(threadpool_one_task_one_thread);
 637         ast_test_register(threadpool_one_thread_one_task);
 638         ast_test_register(threadpool_one_thread_multiple_tasks);
 639         return AST_MODULE_LOAD_SUCCESS;
 640 }
 641
 642 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "threadpool test module");