Add test where we add a task then create a thread for it.
[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 listener_check(
186                 struct ast_test *test,
187                 struct ast_threadpool_listener *listener,
188                 int task_pushed,
189                 int was_empty,
190                 int num_tasks,
191                 int num_active,
192                 int num_idle,
193                 int empty_notice)
194 {
195         struct test_listener_data *tld = listener->private_data;
196         enum ast_test_result_state res = AST_TEST_PASS;
197
198         if (tld->task_pushed != task_pushed) {
199                 ast_test_status_update(test, "Expected task %sto be pushed, but it was%s\n",
200                                 task_pushed ? "" : "not ", tld->task_pushed ? "" : " not");
201                 res = AST_TEST_FAIL;
202         }
203         if (tld->was_empty != was_empty) {
204                 ast_test_status_update(test, "Expected %sto be empty, but it was%s\n",
205                                 was_empty ? "" : "not ", tld->task_pushed ? "" : " not");
206                 res = AST_TEST_FAIL;
207         }
208         if (tld->num_tasks!= num_tasks) {
209                 ast_test_status_update(test, "Expected %d tasks to be pushed, but got %d\n",
210                                 num_tasks, tld->num_tasks);
211                 res = AST_TEST_FAIL;
212         }
213         if (tld->num_active != num_active) {
214                 ast_test_status_update(test, "Expected %d active threads, but got %d\n",
215                                 num_active, tld->num_active);
216                 res = AST_TEST_FAIL;
217         }
218         if (tld->num_idle != num_idle) {
219                 ast_test_status_update(test, "Expected %d idle threads, but got %d\n",
220                                 num_idle, tld->num_idle);
221                 res = AST_TEST_FAIL;
222         }
223         if (tld->empty_notice != empty_notice) {
224                 ast_test_status_update(test, "Expected %s empty notice, but got %s\n",
225                                 was_empty ? "an" : "no", tld->task_pushed ? "one" : "none");
226                 res = AST_TEST_FAIL;
227         }
228
229         return res;
230 }
231
232 AST_TEST_DEFINE(threadpool_push)
233 {
234         struct ast_threadpool *pool = NULL;
235         struct ast_threadpool_listener *listener = NULL;
236         struct simple_task_data *std = NULL;
237         enum ast_test_result_state res = AST_TEST_FAIL;
238
239         switch (cmd) {
240         case TEST_INIT:
241                 info->name = "threadpool_push";
242                 info->category = "/main/threadpool/";
243                 info->summary = "Test task";
244                 info->description =
245                         "Basic threadpool test";
246                 return AST_TEST_NOT_RUN;
247         case TEST_EXECUTE:
248                 break;
249         }
250
251         listener = ast_threadpool_listener_alloc(&test_callbacks);
252         if (!listener) {
253                 return AST_TEST_FAIL;
254         }
255
256         pool = ast_threadpool_create(listener, 0);
257         if (!pool) {
258                 goto end;
259         }
260
261         std = simple_task_data_alloc();
262         if (!std) {
263                 goto end;
264         }
265
266         ast_threadpool_push(pool, simple_task, std);
267
268         wait_for_task_pushed(listener);
269
270         res = listener_check(test, listener, 1, 1, 1, 0, 0, 0);
271
272 end:
273         if (pool) {
274                 ast_threadpool_shutdown(pool);
275         }
276         ao2_cleanup(listener);
277         ast_free(std);
278         return res;
279 }
280
281 AST_TEST_DEFINE(threadpool_thread_creation)
282 {
283         struct ast_threadpool *pool = NULL;
284         struct ast_threadpool_listener *listener = NULL;
285         enum ast_test_result_state res = AST_TEST_FAIL;
286         struct test_listener_data *tld;
287
288         switch (cmd) {
289         case TEST_INIT:
290                 info->name = "threadpool_thread_creation";
291                 info->category = "/main/threadpool/";
292                 info->summary = "Test threadpool thread creation";
293                 info->description =
294                         "Ensure that threads can be added to a threadpool";
295                 return AST_TEST_NOT_RUN;
296         case TEST_EXECUTE:
297                 break;
298         }
299
300         listener = ast_threadpool_listener_alloc(&test_callbacks);
301         if (!listener) {
302                 return AST_TEST_FAIL;
303         }
304         tld = listener->private_data;
305
306         pool = ast_threadpool_create(listener, 0);
307         if (!pool) {
308                 goto end;
309         }
310
311         /* Now let's create a thread. It should start active, then go
312          * idle immediately
313          */
314         ast_threadpool_set_size(pool, 1);
315
316         WAIT_WHILE(tld, tld->num_idle == 0);
317
318         res = listener_check(test, listener, 0, 0, 0, 0, 1, 0);
319
320 end:
321         if (pool) {
322                 ast_threadpool_shutdown(pool);
323         }
324         ao2_cleanup(listener);
325         return res;
326 }
327
328 AST_TEST_DEFINE(threadpool_thread_destruction)
329 {
330         struct ast_threadpool *pool = NULL;
331         struct ast_threadpool_listener *listener = NULL;
332         enum ast_test_result_state res = AST_TEST_FAIL;
333         struct test_listener_data *tld;
334
335         switch (cmd) {
336         case TEST_INIT:
337                 info->name = "threadpool_thread_destruction";
338                 info->category = "/main/threadpool/";
339                 info->summary = "Test threadpool thread destruction";
340                 info->description =
341                         "Ensure that threads are properly destroyed in a threadpool";
342                 return AST_TEST_NOT_RUN;
343         case TEST_EXECUTE:
344                 break;
345         }
346
347         listener = ast_threadpool_listener_alloc(&test_callbacks);
348         if (!listener) {
349                 return AST_TEST_FAIL;
350         }
351         tld = listener->private_data;
352
353         pool = ast_threadpool_create(listener, 0);
354         if (!pool) {
355                 goto end;
356         }
357
358         ast_threadpool_set_size(pool, 3);
359
360         WAIT_WHILE(tld, tld->num_idle < 3);
361
362         res = listener_check(test, listener, 0, 0, 0, 0, 3, 0);
363         if (res == AST_TEST_FAIL) {
364                 goto end;
365         }
366
367         ast_threadpool_set_size(pool, 2);
368
369         WAIT_WHILE(tld, tld->num_idle > 2);
370
371         res = listener_check(test, listener, 0, 0, 0, 0, 2, 0);
372
373 end:
374         if (pool) {
375                 ast_threadpool_shutdown(pool);
376         }
377         ao2_cleanup(listener);
378         return res;
379 }
380
381 AST_TEST_DEFINE(threadpool_one_task_one_thread)
382 {
383         struct ast_threadpool *pool = NULL;
384         struct ast_threadpool_listener *listener = NULL;
385         struct simple_task_data *std = NULL;
386         enum ast_test_result_state res = AST_TEST_FAIL;
387         struct test_listener_data *tld;
388
389         switch (cmd) {
390         case TEST_INIT:
391                 info->name = "threadpool_one_task_one_thread";
392                 info->category = "/main/threadpool/";
393                 info->summary = "Test a single task with a single thread";
394                 info->description =
395                         "Ensure that a thread executes the added task";
396                 return AST_TEST_NOT_RUN;
397         case TEST_EXECUTE:
398                 break;
399         }
400
401         listener = ast_threadpool_listener_alloc(&test_callbacks);
402         if (!listener) {
403                 return AST_TEST_FAIL;
404         }
405         tld = listener->private_data;
406
407         pool = ast_threadpool_create(listener, 0);
408         if (!pool) {
409                 goto end;
410         }
411
412         std = simple_task_data_alloc();
413         if (!std) {
414                 goto end;
415         }
416
417         ast_threadpool_push(pool, simple_task, std);
418
419         ast_threadpool_set_size(pool, 1);
420
421         /* Threads added to the pool are active when they start,
422          * so the newly-created thread should immediately execute
423          * the waiting task.
424          */
425         res = wait_for_completion(std);
426
427         if (res == AST_TEST_FAIL) {
428                 goto end;
429         }
430
431         /* After completing the task, the thread should go idle */
432         WAIT_WHILE(tld, tld->num_idle == 0);
433
434         res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
435
436 end:
437         if (pool) {
438                 ast_threadpool_shutdown(pool);
439         }
440         ao2_cleanup(listener);
441         ast_free(std);
442         return res;
443
444 }
445
446 static int unload_module(void)
447 {
448         ast_test_unregister(threadpool_push);
449         ast_test_unregister(threadpool_thread_creation);
450         ast_test_unregister(threadpool_thread_destruction);
451         ast_test_unregister(threadpool_one_task_one_thread);
452         return 0;
453 }
454
455 static int load_module(void)
456 {
457         ast_test_register(threadpool_push);
458         ast_test_register(threadpool_thread_creation);
459         ast_test_register(threadpool_thread_destruction);
460         ast_test_register(threadpool_one_task_one_thread);
461         return AST_MODULE_LOAD_SUCCESS;
462 }
463
464 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "threadpool test module");