3d64fe9140c09deef728313cac3e77c86ff49b7a
[asterisk/asterisk.git] / 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 AST_TEST_DEFINE(threadpool_reactivation)
621 {
622         struct ast_threadpool *pool = NULL;
623         struct ast_threadpool_listener *listener = NULL;
624         struct simple_task_data *std1 = NULL;
625         struct simple_task_data *std2 = NULL;
626         enum ast_test_result_state res = AST_TEST_FAIL;
627         struct test_listener_data *tld;
628
629         switch (cmd) {
630         case TEST_INIT:
631                 info->name = "threadpool_reactivation";
632                 info->category = "/main/threadpool/";
633                 info->summary = "Test that a threadpool reactivates when work is added";
634                 info->description =
635                         "Push a task into a threadpool. Make sure the task executes and the\n"
636                         "thread goes idle. Then push a second task and ensure that the thread\n"
637                         "awakens and executes the second task.\n";
638                 return AST_TEST_NOT_RUN;
639         case TEST_EXECUTE:
640                 break;
641         }
642
643         listener = ast_threadpool_listener_alloc(&test_callbacks);
644         if (!listener) {
645                 return AST_TEST_FAIL;
646         }
647         tld = listener->private_data;
648
649         pool = ast_threadpool_create(listener, 0);
650         if (!pool) {
651                 goto end;
652         }
653
654         std1 = simple_task_data_alloc();
655         std2 = simple_task_data_alloc();
656         if (!std1 || !std2) {
657                 goto end;
658         }
659
660         ast_threadpool_push(pool, simple_task, std1);
661
662         ast_threadpool_set_size(pool, 1);
663
664         res = wait_for_completion(std1);
665         if (res == AST_TEST_FAIL) {
666                 goto end;
667         }
668
669         res = wait_for_empty_notice(tld);
670         if (res == AST_TEST_FAIL) {
671                 goto end;
672         }
673         
674         WAIT_WHILE(tld, tld->num_idle == 0);
675
676         res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
677
678         /* Now make sure the threadpool reactivates when we add a second task */
679         ast_threadpool_push(pool, simple_task, std2);
680
681         res = wait_for_completion(std2);
682         if (res == AST_TEST_FAIL) {
683                 goto end;
684         }
685
686         res = wait_for_empty_notice(tld);
687         if (res == AST_TEST_FAIL) {
688                 goto end;
689         }
690         
691         WAIT_WHILE(tld, tld->num_idle == 0);
692
693         res = listener_check(test, listener, 1, 1, 2, 0, 1, 1);
694
695 end:
696         if (pool) {
697                 ast_threadpool_shutdown(pool);
698         }
699         ao2_cleanup(listener);
700         ast_free(std1);
701         ast_free(std2);
702         return res;
703
704 }
705
706 struct complex_task_data {
707         int task_executed;
708         int continue_task;
709         ast_mutex_t lock;
710         ast_cond_t stall_cond;
711         ast_cond_t done_cond;
712 };
713
714 static struct complex_task_data *complex_task_data_alloc(void)
715 {
716         struct complex_task_data *ctd = ast_calloc(1, sizeof(*ctd));
717
718         if (!ctd) {
719                 return NULL;
720         }
721         ast_mutex_init(&ctd->lock);
722         ast_cond_init(&ctd->stall_cond, NULL);
723         ast_cond_init(&ctd->done_cond, NULL);
724         return ctd;
725 }
726
727 static int complex_task(void *data)
728 {
729         struct complex_task_data *ctd = data;
730         SCOPED_MUTEX(lock, &ctd->lock);
731         while (!ctd->continue_task) {
732                 ast_cond_wait(&ctd->stall_cond, lock);
733         }
734         /* We got poked. Finish up */
735         ctd->task_executed = 1;
736         ast_cond_signal(&ctd->done_cond);
737         return 0;
738 }
739
740 static void poke_worker(struct complex_task_data *ctd)
741 {
742         SCOPED_MUTEX(lock, &ctd->lock);
743         ctd->continue_task = 1;
744         ast_cond_signal(&ctd->stall_cond);
745 }
746
747 static enum ast_test_result_state wait_for_complex_completion(struct complex_task_data *ctd)
748 {
749         struct timeval start = ast_tvnow();
750         struct timespec end = {
751                 .tv_sec = start.tv_sec + 5,
752                 .tv_nsec = start.tv_usec * 1000
753         };
754         enum ast_test_result_state res = AST_TEST_PASS;
755         SCOPED_MUTEX(lock, &ctd->lock);
756
757         while (!ctd->task_executed) {
758                 ast_cond_timedwait(&ctd->done_cond, lock, &end);
759         }
760
761         if (!ctd->task_executed) {
762                 res = AST_TEST_FAIL;
763         }
764         return res;
765 }
766
767 AST_TEST_DEFINE(threadpool_task_distribution)
768 {
769         struct ast_threadpool *pool = NULL;
770         struct ast_threadpool_listener *listener = NULL;
771         struct complex_task_data *ctd1 = NULL;
772         struct complex_task_data *ctd2 = NULL;
773         enum ast_test_result_state res = AST_TEST_FAIL;
774         struct test_listener_data *tld;
775
776         switch (cmd) {
777         case TEST_INIT:
778                 info->name = "threadpool_task_distribution";
779                 info->category = "/main/threadpool/";
780                 info->summary = "Test that tasks are evenly distributed to threads";
781                 info->description =
782                         "Push two tasks into a threadpool. Ensure that each is handled by\n"
783                         "a separate thread\n";
784                 return AST_TEST_NOT_RUN;
785         case TEST_EXECUTE:
786                 break;
787         }
788
789         listener = ast_threadpool_listener_alloc(&test_callbacks);
790         if (!listener) {
791                 return AST_TEST_FAIL;
792         }
793         tld = listener->private_data;
794
795         pool = ast_threadpool_create(listener, 0);
796         if (!pool) {
797                 goto end;
798         }
799
800         ctd1 = complex_task_data_alloc();
801         ctd2 = complex_task_data_alloc();
802         if (!ctd1 || !ctd2) {
803                 goto end;
804         }
805
806         ast_threadpool_push(pool, complex_task, ctd1);
807         ast_threadpool_push(pool, complex_task, ctd2);
808
809         ast_threadpool_set_size(pool, 2);
810
811         WAIT_WHILE(tld, tld->num_active < 2);
812
813         res = listener_check(test, listener, 1, 0, 2, 2, 0, 0);
814         if (res == AST_TEST_FAIL) {
815                 goto end;
816         }
817
818         /* The tasks are stalled until we poke them */
819         poke_worker(ctd1);
820         poke_worker(ctd2);
821
822         res = wait_for_complex_completion(ctd1);
823         if (res == AST_TEST_FAIL) {
824                 goto end;
825         }
826         res = wait_for_complex_completion(ctd2);
827         if (res == AST_TEST_FAIL) {
828                 goto end;
829         }
830
831         WAIT_WHILE(tld, tld->num_idle < 2);
832
833         res = listener_check(test, listener, 1, 0, 2, 0, 2, 1);
834
835 end:
836         if (pool) {
837                 ast_threadpool_shutdown(pool);
838         }
839         ao2_cleanup(listener);
840         ast_free(ctd1);
841         ast_free(ctd2);
842         return res;
843 }
844
845 AST_TEST_DEFINE(threadpool_more_destruction)
846 {
847         struct ast_threadpool *pool = NULL;
848         struct ast_threadpool_listener *listener = NULL;
849         struct complex_task_data *ctd1 = NULL;
850         struct complex_task_data *ctd2 = NULL;
851         enum ast_test_result_state res = AST_TEST_FAIL;
852         struct test_listener_data *tld;
853
854         switch (cmd) {
855         case TEST_INIT:
856                 info->name = "threadpool_more_destruction";
857                 info->category = "/main/threadpool/";
858                 info->summary = "Test that threads are destroyed as expected";
859                 info->description =
860                         "Push two tasks into a threadpool. Set the threadpool size to 4\n"
861                         "Ensure that there are 2 active and 2 idle threads. Then shrink the\n"
862                         "threadpool down to 1 thread. Ensure that the thread leftove is active\n"
863                         "and ensure that both tasks complete.\n";
864                 return AST_TEST_NOT_RUN;
865         case TEST_EXECUTE:
866                 break;
867         }
868
869         listener = ast_threadpool_listener_alloc(&test_callbacks);
870         if (!listener) {
871                 return AST_TEST_FAIL;
872         }
873         tld = listener->private_data;
874
875         pool = ast_threadpool_create(listener, 0);
876         if (!pool) {
877                 goto end;
878         }
879
880         ctd1 = complex_task_data_alloc();
881         ctd2 = complex_task_data_alloc();
882         if (!ctd1 || !ctd2) {
883                 goto end;
884         }
885
886         ast_threadpool_push(pool, complex_task, ctd1);
887         ast_threadpool_push(pool, complex_task, ctd2);
888
889         ast_threadpool_set_size(pool, 4);
890
891         WAIT_WHILE(tld, tld->num_idle < 2);
892
893         res = listener_check(test, listener, 1, 0, 2, 2, 2, 0);
894         if (res == AST_TEST_FAIL) {
895                 goto end;
896         }
897
898         ast_threadpool_set_size(pool, 1);
899
900         /* Shrinking the threadpool should kill off the two idle threads
901          * and one of the active threads.
902          */
903         WAIT_WHILE(tld, tld->num_idle > 0 || tld->num_active > 1);
904
905         res = listener_check(test, listener, 1, 0, 2, 1, 0, 0);
906         if (res == AST_TEST_FAIL) {
907                 goto end;
908         }
909
910         /* The tasks are stalled until we poke them */
911         poke_worker(ctd1);
912         poke_worker(ctd2);
913
914         res = wait_for_complex_completion(ctd1);
915         if (res == AST_TEST_FAIL) {
916                 goto end;
917         }
918         res = wait_for_complex_completion(ctd2);
919         if (res == AST_TEST_FAIL) {
920                 goto end;
921         }
922
923         WAIT_WHILE(tld, tld->num_idle < 1);
924
925         res = listener_check(test, listener, 1, 0, 2, 0, 1, 1);
926
927 end:
928         if (pool) {
929                 ast_threadpool_shutdown(pool);
930         }
931         ao2_cleanup(listener);
932         ast_free(ctd1);
933         ast_free(ctd2);
934         return res;
935 }
936
937 static int unload_module(void)
938 {
939         ast_test_unregister(threadpool_push);
940         ast_test_unregister(threadpool_thread_creation);
941         ast_test_unregister(threadpool_thread_destruction);
942         ast_test_unregister(threadpool_one_task_one_thread);
943         ast_test_unregister(threadpool_one_thread_one_task);
944         ast_test_unregister(threadpool_one_thread_multiple_tasks);
945         ast_test_unregister(threadpool_reactivation);
946         ast_test_unregister(threadpool_task_distribution);
947         ast_test_unregister(threadpool_more_destruction);
948         return 0;
949 }
950
951 static int load_module(void)
952 {
953         ast_test_register(threadpool_push);
954         ast_test_register(threadpool_thread_creation);
955         ast_test_register(threadpool_thread_destruction);
956         ast_test_register(threadpool_one_task_one_thread);
957         ast_test_register(threadpool_one_thread_one_task);
958         ast_test_register(threadpool_one_thread_multiple_tasks);
959         ast_test_register(threadpool_reactivation);
960         ast_test_register(threadpool_task_distribution);
961         ast_test_register(threadpool_more_destruction);
962         return AST_MODULE_LOAD_SUCCESS;
963 }
964
965 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "threadpool test module");