1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843 |
- /*
- * Asterisk -- An open source telephony toolkit.
- *
- * Copyright (C) 2012-2013, Digium, Inc.
- *
- * Mark Michelson <mmichelson@digium.com>
- *
- * See http://www.asterisk.org for more information about
- * the Asterisk project. Please do not directly contact
- * any of the maintainers of this project for assistance;
- * the project provides a web site, mailing lists and IRC
- * channels for your use.
- *
- * This program is free software, distributed under the terms of
- * the GNU General Public License Version 2. See the LICENSE file
- * at the top of the source tree.
- */
- /*!
- * \file
- * \brief threadpool unit tests
- *
- * \author Mark Michelson <mmichelson@digium.com>
- *
- */
- /*** MODULEINFO
- <depend>TEST_FRAMEWORK</depend>
- <support_level>core</support_level>
- ***/
- #include "asterisk.h"
- #include "asterisk/astobj2.h"
- #include "asterisk/lock.h"
- #include "asterisk/logger.h"
- #include "asterisk/module.h"
- #include "asterisk/taskprocessor.h"
- #include "asterisk/test.h"
- #include "asterisk/threadpool.h"
- struct test_listener_data {
- int num_active;
- int num_idle;
- int task_pushed;
- int num_tasks;
- int empty_notice;
- int was_empty;
- ast_mutex_t lock;
- ast_cond_t cond;
- };
- static struct test_listener_data *test_alloc(void)
- {
- struct test_listener_data *tld = ast_calloc(1, sizeof(*tld));
- if (!tld) {
- return NULL;
- }
- ast_mutex_init(&tld->lock);
- ast_cond_init(&tld->cond, NULL);
- return tld;
- }
- static void test_state_changed(struct ast_threadpool *pool,
- struct ast_threadpool_listener *listener,
- int active_threads,
- int idle_threads)
- {
- struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
- SCOPED_MUTEX(lock, &tld->lock);
- tld->num_active = active_threads;
- tld->num_idle = idle_threads;
- ast_log(LOG_NOTICE, "Thread state: %d active, %d idle\n", tld->num_active, tld->num_idle);
- ast_cond_signal(&tld->cond);
- }
- static void test_task_pushed(struct ast_threadpool *pool,
- struct ast_threadpool_listener *listener,
- int was_empty)
- {
- struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
- SCOPED_MUTEX(lock, &tld->lock);
- tld->task_pushed = 1;
- ++tld->num_tasks;
- tld->was_empty = was_empty;
- ast_cond_signal(&tld->cond);
- }
- static void test_emptied(struct ast_threadpool *pool,
- struct ast_threadpool_listener *listener)
- {
- struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
- SCOPED_MUTEX(lock, &tld->lock);
- tld->empty_notice = 1;
- ast_cond_signal(&tld->cond);
- }
- static void test_shutdown(struct ast_threadpool_listener *listener)
- {
- struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
- ast_cond_destroy(&tld->cond);
- ast_mutex_destroy(&tld->lock);
- }
- static const struct ast_threadpool_listener_callbacks test_callbacks = {
- .state_changed = test_state_changed,
- .task_pushed = test_task_pushed,
- .emptied = test_emptied,
- .shutdown = test_shutdown,
- };
- struct simple_task_data {
- int task_executed;
- ast_mutex_t lock;
- ast_cond_t cond;
- };
- static struct simple_task_data *simple_task_data_alloc(void)
- {
- struct simple_task_data *std = ast_calloc(1, sizeof(*std));
- if (!std) {
- return NULL;
- }
- ast_mutex_init(&std->lock);
- ast_cond_init(&std->cond, NULL);
- return std;
- }
- static void simple_task_data_free(struct simple_task_data *std)
- {
- if (!std) {
- return;
- }
- ast_mutex_destroy(&std->lock);
- ast_cond_destroy(&std->cond);
- ast_free(std);
- }
- static int simple_task(void *data)
- {
- struct simple_task_data *std = data;
- SCOPED_MUTEX(lock, &std->lock);
- std->task_executed = 1;
- ast_cond_signal(&std->cond);
- return 0;
- }
- static enum ast_test_result_state wait_until_thread_state(struct ast_test *test, struct test_listener_data *tld, int num_active, int num_idle)
- {
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 5,
- .tv_nsec = start.tv_usec * 1000
- };
- enum ast_test_result_state res = AST_TEST_PASS;
- SCOPED_MUTEX(lock, &tld->lock);
- while (!(tld->num_active == num_active && tld->num_idle == num_idle)) {
- if (ast_cond_timedwait(&tld->cond, &tld->lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- if (tld->num_active != num_active && tld->num_idle != num_idle) {
- ast_test_status_update(test, "Number of active threads and idle threads not what was expected.\n");
- ast_test_status_update(test, "Expected %d active threads but got %d\n", num_active, tld->num_active);
- ast_test_status_update(test, "Expected %d idle threads but got %d\n", num_idle, tld->num_idle);
- res = AST_TEST_FAIL;
- }
- return res;
- }
- static void wait_for_task_pushed(struct ast_threadpool_listener *listener)
- {
- struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 5,
- .tv_nsec = start.tv_usec * 1000
- };
- SCOPED_MUTEX(lock, &tld->lock);
- while (!tld->task_pushed) {
- if (ast_cond_timedwait(&tld->cond, lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- }
- static enum ast_test_result_state wait_for_completion(struct ast_test *test, struct simple_task_data *std)
- {
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 5,
- .tv_nsec = start.tv_usec * 1000
- };
- enum ast_test_result_state res = AST_TEST_PASS;
- SCOPED_MUTEX(lock, &std->lock);
- while (!std->task_executed) {
- if (ast_cond_timedwait(&std->cond, lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- if (!std->task_executed) {
- ast_test_status_update(test, "Task execution did not occur\n");
- res = AST_TEST_FAIL;
- }
- return res;
- }
- static enum ast_test_result_state wait_for_empty_notice(struct ast_test *test, struct test_listener_data *tld)
- {
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 5,
- .tv_nsec = start.tv_usec * 1000
- };
- enum ast_test_result_state res = AST_TEST_PASS;
- SCOPED_MUTEX(lock, &tld->lock);
- while (!tld->empty_notice) {
- if (ast_cond_timedwait(&tld->cond, lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- if (!tld->empty_notice) {
- ast_test_status_update(test, "Test listener not notified that threadpool is empty\n");
- res = AST_TEST_FAIL;
- }
- return res;
- }
- static enum ast_test_result_state listener_check(
- struct ast_test *test,
- struct ast_threadpool_listener *listener,
- int task_pushed,
- int was_empty,
- int num_tasks,
- int num_active,
- int num_idle,
- int empty_notice)
- {
- struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
- enum ast_test_result_state res = AST_TEST_PASS;
- if (tld->task_pushed != task_pushed) {
- ast_test_status_update(test, "Expected task %sto be pushed, but it was%s\n",
- task_pushed ? "" : "not ", tld->task_pushed ? "" : " not");
- res = AST_TEST_FAIL;
- }
- if (tld->was_empty != was_empty) {
- ast_test_status_update(test, "Expected %sto be empty, but it was%s\n",
- was_empty ? "" : "not ", tld->was_empty ? "" : " not");
- res = AST_TEST_FAIL;
- }
- if (tld->num_tasks!= num_tasks) {
- ast_test_status_update(test, "Expected %d tasks to be pushed, but got %d\n",
- num_tasks, tld->num_tasks);
- res = AST_TEST_FAIL;
- }
- if (tld->num_active != num_active) {
- ast_test_status_update(test, "Expected %d active threads, but got %d\n",
- num_active, tld->num_active);
- res = AST_TEST_FAIL;
- }
- if (tld->num_idle != num_idle) {
- ast_test_status_update(test, "Expected %d idle threads, but got %d\n",
- num_idle, tld->num_idle);
- res = AST_TEST_FAIL;
- }
- if (tld->empty_notice != empty_notice) {
- ast_test_status_update(test, "Expected %s empty notice, but got %s\n",
- was_empty ? "an" : "no", tld->task_pushed ? "one" : "none");
- res = AST_TEST_FAIL;
- }
- return res;
- }
- AST_TEST_DEFINE(threadpool_push)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std = NULL;
- struct test_listener_data *tld = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "push";
- info->category = "/main/threadpool/";
- info->summary = "Test task";
- info->description =
- "Basic threadpool test";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std = simple_task_data_alloc();
- if (!std) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std)) {
- goto end;
- }
- wait_for_task_pushed(listener);
- res = listener_check(test, listener, 1, 1, 1, 0, 0, 0);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_initial_threads)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 3,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "initial_threads";
- info->category = "/main/threadpool/";
- info->summary = "Test threadpool initialization state";
- info->description =
- "Ensure that a threadpool created with a specific size contains the\n"
- "proper number of idle threads.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 3);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_thread_creation)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "thread_creation";
- info->category = "/main/threadpool/";
- info->summary = "Test threadpool thread creation";
- info->description =
- "Ensure that threads can be added to a threadpool";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- /* Now let's create a thread. It should start active, then go
- * idle immediately
- */
- ast_threadpool_set_size(pool, 1);
- res = wait_until_thread_state(test, tld, 0, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_thread_destruction)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "thread_destruction";
- info->category = "/main/threadpool/";
- info->summary = "Test threadpool thread destruction";
- info->description =
- "Ensure that threads are properly destroyed in a threadpool";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- ast_threadpool_set_size(pool, 3);
- res = wait_until_thread_state(test, tld, 0, 3);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 0, 0, 0, 0, 3, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- ast_threadpool_set_size(pool, 2);
- res = wait_until_thread_state(test, tld, 0, 2);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_thread_timeout)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 2,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "thread_timeout";
- info->category = "/main/threadpool/";
- info->summary = "Test threadpool thread timeout";
- info->description =
- "Ensure that a thread with a two second timeout dies as expected.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 0, 0, 0, 0, 1, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 0, 0, 0, 0, 0, 0);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_thread_timeout_thrash)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 1,
- .auto_increment = 1,
- .initial_size = 0,
- .max_size = 1,
- };
- int iteration;
- switch (cmd) {
- case TEST_INIT:
- info->name = "thread_timeout_thrash";
- info->category = "/main/threadpool/";
- info->summary = "Thrash threadpool thread timeout";
- info->description =
- "Repeatedly queue a task when a threadpool thread should timeout.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- for (iteration = 0; iteration < 30; ++iteration) {
- struct simple_task_data *std = NULL;
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + options.idle_timeout,
- .tv_nsec = start.tv_usec * 1000
- };
- std = simple_task_data_alloc();
- if (!std) {
- goto end;
- }
- /* Wait until the threadpool thread should timeout due to being idle */
- ast_mutex_lock(&tld->lock);
- while (ast_cond_timedwait(&tld->cond, &tld->lock, &end) != ETIMEDOUT) {
- /* This purposely left empty as we want to loop waiting for a time out */
- }
- ast_mutex_unlock(&tld->lock);
- if (ast_threadpool_push(pool, simple_task, std)) {
- res = AST_TEST_FAIL;
- } else {
- res = wait_for_completion(test, std);
- }
- simple_task_data_free(std);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- }
- res = wait_until_thread_state(test, tld, 0, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 1, 30, 0, 0, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_one_task_one_thread)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "one_task_one_thread";
- info->category = "/main/threadpool/";
- info->summary = "Test a single task with a single thread";
- info->description =
- "Push a task into an empty threadpool, then add a thread to the pool.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std = simple_task_data_alloc();
- if (!std) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std)) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- /* Threads added to the pool are active when they start,
- * so the newly-created thread should immediately execute
- * the waiting task.
- */
- res = wait_for_completion(test, std);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- /* After completing the task, the thread should go idle */
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_one_thread_one_task)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "one_thread_one_task";
- info->category = "/main/threadpool/";
- info->summary = "Test a single thread with a single task";
- info->description =
- "Add a thread to the pool and then push a task to it.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std = simple_task_data_alloc();
- if (!std) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std)) {
- res = AST_TEST_FAIL;
- goto end;
- }
- res = wait_for_completion(test, std);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- /* After completing the task, the thread should go idle */
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_one_thread_multiple_tasks)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std1 = NULL;
- struct simple_task_data *std2 = NULL;
- struct simple_task_data *std3 = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "one_thread_multiple_tasks";
- info->category = "/main/threadpool/";
- info->summary = "Test a single thread with multiple tasks";
- info->description =
- "Add a thread to the pool and then push three tasks to it.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std1 = simple_task_data_alloc();
- std2 = simple_task_data_alloc();
- std3 = simple_task_data_alloc();
- if (!std1 || !std2 || !std3) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = AST_TEST_FAIL;
- if (ast_threadpool_push(pool, simple_task, std1)) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std2)) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std3)) {
- goto end;
- }
- res = wait_for_completion(test, std1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_completion(test, std2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_completion(test, std3);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 0, 3, 0, 1, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std1);
- simple_task_data_free(std2);
- simple_task_data_free(std3);
- ast_free(tld);
- return res;
- }
- static enum ast_test_result_state wait_until_thread_state_task_pushed(struct ast_test *test,
- struct test_listener_data *tld, int num_active, int num_idle, int num_tasks)
- {
- enum ast_test_result_state res = AST_TEST_PASS;
- struct timeval start;
- struct timespec end;
- res = wait_until_thread_state(test, tld, num_active, num_idle);
- if (res == AST_TEST_FAIL) {
- return res;
- }
- start = ast_tvnow();
- end.tv_sec = start.tv_sec + 5;
- end.tv_nsec = start.tv_usec * 1000;
- ast_mutex_lock(&tld->lock);
- while (tld->num_tasks != num_tasks) {
- if (ast_cond_timedwait(&tld->cond, &tld->lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- if (tld->num_tasks != num_tasks) {
- ast_test_status_update(test, "Number of tasks pushed %d does not match expected %d\n",
- tld->num_tasks, num_tasks);
- res = AST_TEST_FAIL;
- }
- ast_mutex_unlock(&tld->lock);
- return res;
- }
- AST_TEST_DEFINE(threadpool_auto_increment)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std1 = NULL;
- struct simple_task_data *std2 = NULL;
- struct simple_task_data *std3 = NULL;
- struct simple_task_data *std4 = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 3,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "auto_increment";
- info->category = "/main/threadpool/";
- info->summary = "Test that the threadpool grows as tasks are added";
- info->description =
- "Create an empty threadpool and push a task to it. Once the task is\n"
- "pushed, the threadpool should add three threads and be able to\n"
- "handle the task. The threads should then go idle";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std1 = simple_task_data_alloc();
- std2 = simple_task_data_alloc();
- std3 = simple_task_data_alloc();
- std4 = simple_task_data_alloc();
- if (!std1 || !std2 || !std3 || !std4) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std1)) {
- goto end;
- }
- /* Pushing the task should result in the threadpool growing
- * by three threads. This will allow the task to actually execute
- */
- res = wait_for_completion(test, std1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 3);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- /* Now push three tasks into the pool and ensure the pool does not
- * grow.
- */
- res = AST_TEST_FAIL;
- if (ast_threadpool_push(pool, simple_task, std2)) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std3)) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std4)) {
- goto end;
- }
- res = wait_for_completion(test, std2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_completion(test, std3);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_completion(test, std4);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state_task_pushed(test, tld, 0, 3, 4);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std1);
- simple_task_data_free(std2);
- simple_task_data_free(std3);
- simple_task_data_free(std4);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_max_size)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 3,
- .initial_size = 0,
- .max_size = 2,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "max_size";
- info->category = "/main/threadpool/";
- info->summary = "Test that the threadpool does not exceed its maximum size restriction";
- info->description =
- "Create an empty threadpool and push a task to it. Once the task is\n"
- "pushed, the threadpool should attempt to grow by three threads, but the\n"
- "pool's restrictions should only allow two threads to be added.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std = simple_task_data_alloc();
- if (!std) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std)) {
- goto end;
- }
- res = wait_for_completion(test, std);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 1, 1, 0, 2, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_reactivation)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct simple_task_data *std1 = NULL;
- struct simple_task_data *std2 = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "reactivation";
- info->category = "/main/threadpool/";
- info->summary = "Test that a threadpool reactivates when work is added";
- info->description =
- "Push a task into a threadpool. Make sure the task executes and the\n"
- "thread goes idle. Then push a second task and ensure that the thread\n"
- "awakens and executes the second task.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- std1 = simple_task_data_alloc();
- std2 = simple_task_data_alloc();
- if (!std1 || !std2) {
- goto end;
- }
- if (ast_threadpool_push(pool, simple_task, std1)) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- res = wait_for_completion(test, std1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 1, 1, 0, 1, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- /* Now make sure the threadpool reactivates when we add a second task */
- if (ast_threadpool_push(pool, simple_task, std2)) {
- res = AST_TEST_FAIL;
- goto end;
- }
- res = wait_for_completion(test, std2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_empty_notice(test, tld);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 1, 2, 0, 1, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- simple_task_data_free(std1);
- simple_task_data_free(std2);
- ast_free(tld);
- return res;
- }
- struct complex_task_data {
- int task_started;
- int task_executed;
- int continue_task;
- ast_mutex_t lock;
- ast_cond_t stall_cond;
- ast_cond_t notify_cond;
- };
- static struct complex_task_data *complex_task_data_alloc(void)
- {
- struct complex_task_data *ctd = ast_calloc(1, sizeof(*ctd));
- if (!ctd) {
- return NULL;
- }
- ast_mutex_init(&ctd->lock);
- ast_cond_init(&ctd->stall_cond, NULL);
- ast_cond_init(&ctd->notify_cond, NULL);
- return ctd;
- }
- static void complex_task_data_free(struct complex_task_data *ctd)
- {
- if (!ctd) {
- return;
- }
- ast_mutex_destroy(&ctd->lock);
- ast_cond_destroy(&ctd->stall_cond);
- ast_cond_destroy(&ctd->notify_cond);
- ast_free(ctd);
- }
- static int complex_task(void *data)
- {
- struct complex_task_data *ctd = data;
- SCOPED_MUTEX(lock, &ctd->lock);
- /* Notify that we started */
- ctd->task_started = 1;
- ast_cond_signal(&ctd->notify_cond);
- while (!ctd->continue_task) {
- ast_cond_wait(&ctd->stall_cond, lock);
- }
- /* We got poked. Finish up */
- ctd->task_executed = 1;
- ast_cond_signal(&ctd->notify_cond);
- return 0;
- }
- static void poke_worker(struct complex_task_data *ctd)
- {
- SCOPED_MUTEX(lock, &ctd->lock);
- ctd->continue_task = 1;
- ast_cond_signal(&ctd->stall_cond);
- }
- static int wait_for_complex_start(struct complex_task_data *ctd)
- {
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 5,
- .tv_nsec = start.tv_usec * 1000
- };
- SCOPED_MUTEX(lock, &ctd->lock);
- while (!ctd->task_started) {
- if (ast_cond_timedwait(&ctd->notify_cond, lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- return ctd->task_started;
- }
- static int has_complex_started(struct complex_task_data *ctd)
- {
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 1,
- .tv_nsec = start.tv_usec * 1000
- };
- SCOPED_MUTEX(lock, &ctd->lock);
- while (!ctd->task_started) {
- if (ast_cond_timedwait(&ctd->notify_cond, lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- return ctd->task_started;
- }
- static enum ast_test_result_state wait_for_complex_completion(struct complex_task_data *ctd)
- {
- struct timeval start = ast_tvnow();
- struct timespec end = {
- .tv_sec = start.tv_sec + 5,
- .tv_nsec = start.tv_usec * 1000
- };
- enum ast_test_result_state res = AST_TEST_PASS;
- SCOPED_MUTEX(lock, &ctd->lock);
- while (!ctd->task_executed) {
- if (ast_cond_timedwait(&ctd->notify_cond, lock, &end) == ETIMEDOUT) {
- break;
- }
- }
- if (!ctd->task_executed) {
- res = AST_TEST_FAIL;
- }
- return res;
- }
- AST_TEST_DEFINE(threadpool_task_distribution)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct complex_task_data *ctd1 = NULL;
- struct complex_task_data *ctd2 = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "task_distribution";
- info->category = "/main/threadpool/";
- info->summary = "Test that tasks are evenly distributed to threads";
- info->description =
- "Push two tasks into a threadpool. Ensure that each is handled by\n"
- "a separate thread";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- ctd1 = complex_task_data_alloc();
- ctd2 = complex_task_data_alloc();
- if (!ctd1 || !ctd2) {
- goto end;
- }
- if (ast_threadpool_push(pool, complex_task, ctd1)) {
- goto end;
- }
- if (ast_threadpool_push(pool, complex_task, ctd2)) {
- goto end;
- }
- ast_threadpool_set_size(pool, 2);
- res = wait_until_thread_state(test, tld, 2, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 0, 2, 2, 0, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- /* The tasks are stalled until we poke them */
- poke_worker(ctd1);
- poke_worker(ctd2);
- res = wait_for_complex_completion(ctd1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_complex_completion(ctd2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 0, 2, 0, 2, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- complex_task_data_free(ctd1);
- complex_task_data_free(ctd2);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_more_destruction)
- {
- struct ast_threadpool *pool = NULL;
- struct ast_threadpool_listener *listener = NULL;
- struct complex_task_data *ctd1 = NULL;
- struct complex_task_data *ctd2 = NULL;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct test_listener_data *tld = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 0,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "more_destruction";
- info->category = "/main/threadpool/";
- info->summary = "Test that threads are destroyed as expected";
- info->description =
- "Push two tasks into a threadpool. Set the threadpool size to 4\n"
- "Ensure that there are 2 active and 2 idle threads. Then shrink the\n"
- "threadpool down to 1 thread. Ensure that the thread leftover is active\n"
- "and ensure that both tasks complete.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- tld = test_alloc();
- if (!tld) {
- return AST_TEST_FAIL;
- }
- listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
- if (!listener) {
- goto end;
- }
- pool = ast_threadpool_create(info->name, listener, &options);
- if (!pool) {
- goto end;
- }
- ctd1 = complex_task_data_alloc();
- ctd2 = complex_task_data_alloc();
- if (!ctd1 || !ctd2) {
- goto end;
- }
- if (ast_threadpool_push(pool, complex_task, ctd1)) {
- goto end;
- }
- if (ast_threadpool_push(pool, complex_task, ctd2)) {
- goto end;
- }
- ast_threadpool_set_size(pool, 4);
- res = wait_until_thread_state(test, tld, 2, 2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 0, 2, 2, 2, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- ast_threadpool_set_size(pool, 1);
- /* Shrinking the threadpool should kill off the two idle threads
- * and one of the active threads.
- */
- res = wait_until_thread_state(test, tld, 1, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 0, 2, 1, 0, 0);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- /* The tasks are stalled until we poke them */
- poke_worker(ctd1);
- poke_worker(ctd2);
- res = wait_for_complex_completion(ctd1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_for_complex_completion(ctd2);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = wait_until_thread_state(test, tld, 0, 1);
- if (res == AST_TEST_FAIL) {
- goto end;
- }
- res = listener_check(test, listener, 1, 0, 2, 0, 1, 1);
- end:
- ast_threadpool_shutdown(pool);
- ao2_cleanup(listener);
- complex_task_data_free(ctd1);
- complex_task_data_free(ctd2);
- ast_free(tld);
- return res;
- }
- AST_TEST_DEFINE(threadpool_serializer)
- {
- int started = 0;
- int finished = 0;
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct ast_threadpool *pool = NULL;
- struct ast_taskprocessor *uut = NULL;
- struct complex_task_data *data1 = NULL;
- struct complex_task_data *data2 = NULL;
- struct complex_task_data *data3 = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 2,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "threadpool_serializer";
- info->category = "/main/threadpool/";
- info->summary = "Test that serializers";
- info->description =
- "Ensures that tasks enqueued to a serialize execute in sequence.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- pool = ast_threadpool_create("threadpool_serializer", NULL, &options);
- if (!pool) {
- ast_test_status_update(test, "Could not create threadpool\n");
- goto end;
- }
- uut = ast_threadpool_serializer("ser1", pool);
- data1 = complex_task_data_alloc();
- data2 = complex_task_data_alloc();
- data3 = complex_task_data_alloc();
- if (!uut || !data1 || !data2 || !data3) {
- ast_test_status_update(test, "Allocation failed\n");
- goto end;
- }
- /* This should start right away */
- if (ast_taskprocessor_push(uut, complex_task, data1)) {
- ast_test_status_update(test, "Failed to enqueue data1\n");
- goto end;
- }
- started = wait_for_complex_start(data1);
- if (!started) {
- ast_test_status_update(test, "Failed to start data1\n");
- goto end;
- }
- /* This should not start until data 1 is complete */
- if (ast_taskprocessor_push(uut, complex_task, data2)) {
- ast_test_status_update(test, "Failed to enqueue data2\n");
- goto end;
- }
- started = has_complex_started(data2);
- if (started) {
- ast_test_status_update(test, "data2 started out of order\n");
- goto end;
- }
- /* But the free thread in the pool can still run */
- if (ast_threadpool_push(pool, complex_task, data3)) {
- ast_test_status_update(test, "Failed to enqueue data3\n");
- }
- started = wait_for_complex_start(data3);
- if (!started) {
- ast_test_status_update(test, "Failed to start data3\n");
- goto end;
- }
- /* Finishing data1 should allow data2 to start */
- poke_worker(data1);
- finished = wait_for_complex_completion(data1) == AST_TEST_PASS;
- if (!finished) {
- ast_test_status_update(test, "data1 couldn't finish\n");
- goto end;
- }
- started = wait_for_complex_start(data2);
- if (!started) {
- ast_test_status_update(test, "Failed to start data2\n");
- goto end;
- }
- /* Finish up */
- poke_worker(data2);
- finished = wait_for_complex_completion(data2) == AST_TEST_PASS;
- if (!finished) {
- ast_test_status_update(test, "data2 couldn't finish\n");
- goto end;
- }
- poke_worker(data3);
- finished = wait_for_complex_completion(data3) == AST_TEST_PASS;
- if (!finished) {
- ast_test_status_update(test, "data3 couldn't finish\n");
- goto end;
- }
- res = AST_TEST_PASS;
- end:
- poke_worker(data1);
- poke_worker(data2);
- poke_worker(data3);
- ast_taskprocessor_unreference(uut);
- ast_threadpool_shutdown(pool);
- complex_task_data_free(data1);
- complex_task_data_free(data2);
- complex_task_data_free(data3);
- return res;
- }
- AST_TEST_DEFINE(threadpool_serializer_dupe)
- {
- enum ast_test_result_state res = AST_TEST_FAIL;
- struct ast_threadpool *pool = NULL;
- struct ast_taskprocessor *uut = NULL;
- struct ast_taskprocessor *there_can_be_only_one = NULL;
- struct ast_threadpool_options options = {
- .version = AST_THREADPOOL_OPTIONS_VERSION,
- .idle_timeout = 0,
- .auto_increment = 0,
- .initial_size = 2,
- .max_size = 0,
- };
- switch (cmd) {
- case TEST_INIT:
- info->name = "threadpool_serializer_dupe";
- info->category = "/main/threadpool/";
- info->summary = "Test that serializers are uniquely named";
- info->description =
- "Creating two serializers with the same name should\n"
- "result in error.";
- return AST_TEST_NOT_RUN;
- case TEST_EXECUTE:
- break;
- }
- pool = ast_threadpool_create("threadpool_serializer", NULL, &options);
- if (!pool) {
- ast_test_status_update(test, "Could not create threadpool\n");
- goto end;
- }
- uut = ast_threadpool_serializer("highlander", pool);
- if (!uut) {
- ast_test_status_update(test, "Allocation failed\n");
- goto end;
- }
- there_can_be_only_one = ast_threadpool_serializer("highlander", pool);
- if (there_can_be_only_one) {
- ast_taskprocessor_unreference(there_can_be_only_one);
- ast_test_status_update(test, "Duplicate name error\n");
- goto end;
- }
- res = AST_TEST_PASS;
- end:
- ast_taskprocessor_unreference(uut);
- ast_threadpool_shutdown(pool);
- return res;
- }
- static int unload_module(void)
- {
- ast_test_unregister(threadpool_push);
- ast_test_unregister(threadpool_initial_threads);
- ast_test_unregister(threadpool_thread_creation);
- ast_test_unregister(threadpool_thread_destruction);
- ast_test_unregister(threadpool_thread_timeout);
- ast_test_unregister(threadpool_thread_timeout_thrash);
- ast_test_unregister(threadpool_one_task_one_thread);
- ast_test_unregister(threadpool_one_thread_one_task);
- ast_test_unregister(threadpool_one_thread_multiple_tasks);
- ast_test_unregister(threadpool_auto_increment);
- ast_test_unregister(threadpool_max_size);
- ast_test_unregister(threadpool_reactivation);
- ast_test_unregister(threadpool_task_distribution);
- ast_test_unregister(threadpool_more_destruction);
- ast_test_unregister(threadpool_serializer);
- ast_test_unregister(threadpool_serializer_dupe);
- return 0;
- }
- static int load_module(void)
- {
- ast_test_register(threadpool_push);
- ast_test_register(threadpool_initial_threads);
- ast_test_register(threadpool_thread_creation);
- ast_test_register(threadpool_thread_destruction);
- ast_test_register(threadpool_thread_timeout);
- ast_test_register(threadpool_thread_timeout_thrash);
- ast_test_register(threadpool_one_task_one_thread);
- ast_test_register(threadpool_one_thread_one_task);
- ast_test_register(threadpool_one_thread_multiple_tasks);
- ast_test_register(threadpool_auto_increment);
- ast_test_register(threadpool_max_size);
- ast_test_register(threadpool_reactivation);
- ast_test_register(threadpool_task_distribution);
- ast_test_register(threadpool_more_destruction);
- ast_test_register(threadpool_serializer);
- ast_test_register(threadpool_serializer_dupe);
- return AST_MODULE_LOAD_SUCCESS;
- }
- AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "threadpool test module");
|