test_threadpool.c File Reference

threadpool unit tests More...

#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"

Include dependency graph for test_threadpool.c:

Go to the source code of this file.

Data Structures
struct	complex_task_data
struct	efficiency_task_data
struct	serializer_efficiency_task_data
struct	simple_task_data
struct	test_listener_data

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
struct ast_module *	AST_MODULE_SELF_SYM (void)
	AST_TEST_DEFINE (threadpool_auto_increment)
	AST_TEST_DEFINE (threadpool_initial_threads)
	AST_TEST_DEFINE (threadpool_max_size)
	AST_TEST_DEFINE (threadpool_more_destruction)
	AST_TEST_DEFINE (threadpool_one_task_one_thread)
	AST_TEST_DEFINE (threadpool_one_thread_multiple_tasks)
	AST_TEST_DEFINE (threadpool_one_thread_one_task)
	AST_TEST_DEFINE (threadpool_push)
	AST_TEST_DEFINE (threadpool_reactivation)
	AST_TEST_DEFINE (threadpool_serializer)
	AST_TEST_DEFINE (threadpool_serializer_dupe)
	AST_TEST_DEFINE (threadpool_task_distribution)
	AST_TEST_DEFINE (threadpool_thread_creation)
	AST_TEST_DEFINE (threadpool_thread_destruction)
	AST_TEST_DEFINE (threadpool_thread_timeout)
	AST_TEST_DEFINE (threadpool_thread_timeout_thrash)
static int	complex_task (void *data)
static struct complex_task_data *	complex_task_data_alloc (void)
static void	complex_task_data_free (struct complex_task_data *ctd)
static int	efficiency_task (void *data)
static char *	handle_cli_threadpool_push_efficiency (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	handle_cli_threadpool_push_serializer_efficiency (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static int	has_complex_started (struct complex_task_data *ctd)
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)
static int	load_module (void)
static void	poke_worker (struct complex_task_data *ctd)
static int	serializer_efficiency_task (void *data)
static int	simple_task (void *data)
static struct simple_task_data *	simple_task_data_alloc (void)
static void	simple_task_data_free (struct simple_task_data *std)
static struct test_listener_data *	test_alloc (void)
static void	test_emptied (struct ast_threadpool *pool, struct ast_threadpool_listener *listener)
static void	test_shutdown (struct ast_threadpool_listener *listener)
static void	test_state_changed (struct ast_threadpool *pool, struct ast_threadpool_listener *listener, int active_threads, int idle_threads)
static void	test_task_pushed (struct ast_threadpool *pool, struct ast_threadpool_listener *listener, int was_empty)
static int	unload_module (void)
static enum ast_test_result_state	wait_for_completion (struct ast_test *test, struct simple_task_data *std)
static enum ast_test_result_state	wait_for_complex_completion (struct complex_task_data *ctd)
static int	wait_for_complex_start (struct complex_task_data *ctd)
static enum ast_test_result_state	wait_for_empty_notice (struct ast_test *test, struct test_listener_data *tld)
static void	wait_for_task_pushed (struct ast_threadpool_listener *listener)
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)
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)

Variables
static struct ast_module_info	__mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "threadpool test module" , .key = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = AST_BUILDOPT_SUM, .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_DEFAULT, .support_level = AST_MODULE_SUPPORT_CORE, }
static const struct ast_module_info *	ast_module_info = &__mod_info
static struct ast_cli_entry	cli []
static const struct ast_threadpool_listener_callbacks	test_callbacks

Detailed Description

threadpool unit tests

Author

Mark Michelson mmich.nosp@m.elso.nosp@m.n@dig.nosp@m.ium..nosp@m.com

Definition in file test_threadpool.c.

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 2100 of file test_threadpool.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 2100 of file test_threadpool.c.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 2100 of file test_threadpool.c.

AST_TEST_DEFINE() [1/16]

AST_TEST_DEFINE

(

threadpool_auto_increment

)

Definition at line 1098 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), wait_for_empty_notice(), wait_until_thread_state(), and wait_until_thread_state_task_pushed().

AST_TEST_DEFINE() [2/16]

AST_TEST_DEFINE

(

threadpool_initial_threads

)

Definition at line 459 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), NULL, options, test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, and wait_until_thread_state().

AST_TEST_DEFINE() [3/16]

AST_TEST_DEFINE

(

threadpool_max_size

)

Definition at line 1226 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), and wait_until_thread_state().

AST_TEST_DEFINE() [4/16]

AST_TEST_DEFINE

(

threadpool_more_destruction

)

Definition at line 1623 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), complex_task(), complex_task_data_alloc(), complex_task_data_free(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, poke_worker(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_complex_completion(), and wait_until_thread_state().

AST_TEST_DEFINE() [5/16]

AST_TEST_DEFINE

(

threadpool_one_task_one_thread

)

Definition at line 788 of file test_threadpool.c.

{
    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;
 
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), wait_for_empty_notice(), and wait_until_thread_state().

AST_TEST_DEFINE() [6/16]

AST_TEST_DEFINE

(

threadpool_one_thread_multiple_tasks

)

Definition at line 957 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), wait_for_empty_notice(), and wait_until_thread_state().

AST_TEST_DEFINE() [7/16]

AST_TEST_DEFINE

(

threadpool_one_thread_one_task

)

Definition at line 872 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), wait_for_empty_notice(), and wait_until_thread_state().

AST_TEST_DEFINE() [8/16]

AST_TEST_DEFINE

(

threadpool_push

)

Definition at line 290 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, and wait_for_task_pushed().

AST_TEST_DEFINE() [9/16]

AST_TEST_DEFINE

(

threadpool_reactivation

)

Definition at line 1298 of file test_threadpool.c.

{
    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;
 
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), wait_for_empty_notice(), and wait_until_thread_state().

AST_TEST_DEFINE() [10/16]

AST_TEST_DEFINE

(

threadpool_serializer

)

Definition at line 1739 of file test_threadpool.c.

{
    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;
}

References ast_taskprocessor_push(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_threadpool_create(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_serializer(), ast_threadpool_shutdown(), complex_task(), complex_task_data_alloc(), complex_task_data_free(), end, has_complex_started(), sip_to_pjsip::info(), NULL, options, poke_worker(), TEST_EXECUTE, TEST_INIT, wait_for_complex_completion(), and wait_for_complex_start().

AST_TEST_DEFINE() [11/16]

AST_TEST_DEFINE

(

threadpool_serializer_dupe

)

Definition at line 1997 of file test_threadpool.c.

{
    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;
}

References ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_threadpool_create(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_serializer(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), NULL, options, TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [12/16]

AST_TEST_DEFINE

(

threadpool_task_distribution

)

Definition at line 1524 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), complex_task(), complex_task_data_alloc(), complex_task_data_free(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, poke_worker(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_complex_completion(), and wait_until_thread_state().

AST_TEST_DEFINE() [13/16]

AST_TEST_DEFINE

(

threadpool_thread_creation

)

Definition at line 511 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), NULL, options, test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, and wait_until_thread_state().

AST_TEST_DEFINE() [14/16]

AST_TEST_DEFINE

(

threadpool_thread_destruction

)

Definition at line 566 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, and wait_until_thread_state().

AST_TEST_DEFINE() [15/16]

AST_TEST_DEFINE

(

threadpool_thread_timeout

)

Definition at line 630 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_free, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_set_size(), ast_threadpool_shutdown(), end, sip_to_pjsip::info(), listener(), listener_check(), NULL, options, test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, and wait_until_thread_state().

AST_TEST_DEFINE() [16/16]

AST_TEST_DEFINE

(

threadpool_thread_timeout_thrash

)

Definition at line 697 of file test_threadpool.c.

{
    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;
}

References ao2_cleanup, ast_cond_timedwait, ast_free, ast_mutex_lock, ast_mutex_unlock, AST_TEST_FAIL, AST_TEST_NOT_RUN, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_set_size(), ast_threadpool_shutdown(), ast_tvnow(), test_listener_data::cond, end, sip_to_pjsip::info(), listener(), listener_check(), test_listener_data::lock, NULL, options, simple_task(), simple_task_data_alloc(), simple_task_data_free(), test_alloc(), test_callbacks, TEST_EXECUTE, TEST_INIT, wait_for_completion(), and wait_until_thread_state().

complex_task()

static int complex_task ( void *  data )

static

Definition at line 1443 of file test_threadpool.c.

{
    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;
}

References ast_cond_signal, ast_cond_wait, complex_task_data::continue_task, lock, complex_task_data::lock, complex_task_data::notify_cond, SCOPED_MUTEX, complex_task_data::stall_cond, complex_task_data::task_executed, and complex_task_data::task_started.

Referenced by AST_TEST_DEFINE().

complex_task_data_alloc()

static struct complex_task_data * complex_task_data_alloc ( void  )

static

Definition at line 1417 of file test_threadpool.c.

{
    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;
}

References ast_calloc, ast_cond_init, ast_mutex_init, complex_task_data::lock, complex_task_data::notify_cond, NULL, and complex_task_data::stall_cond.

Referenced by AST_TEST_DEFINE().

complex_task_data_free()

static void complex_task_data_free ( struct complex_task_data *  ctd )

static

Definition at line 1430 of file test_threadpool.c.

{
    if (!ctd) {
        return;
    }
 
    ast_mutex_destroy(&ctd->lock);
    ast_cond_destroy(&ctd->stall_cond);
    ast_cond_destroy(&ctd->notify_cond);
 
    ast_free(ctd);
}

References ast_cond_destroy, ast_free, ast_mutex_destroy, complex_task_data::lock, complex_task_data::notify_cond, and complex_task_data::stall_cond.

Referenced by AST_TEST_DEFINE().

efficiency_task()

static int efficiency_task ( void *  data )

static

Definition at line 358 of file test_threadpool.c.

{
    struct efficiency_task_data *etd = data;
 
    if (etd->shutdown) {
        return 0;
    }
 
    ast_atomic_fetchadd_int(&etd->num_tasks_executed, +1);
 
    if (ast_threadpool_push(etd->pool, efficiency_task, etd)) {
        return -1;
    }
 
    return 0;
}

References ast_atomic_fetchadd_int(), ast_threadpool_push(), efficiency_task(), efficiency_task_data::num_tasks_executed, efficiency_task_data::pool, and efficiency_task_data::shutdown.

Referenced by efficiency_task(), and handle_cli_threadpool_push_efficiency().

handle_cli_threadpool_push_efficiency()

static char * handle_cli_threadpool_push_efficiency ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 375 of file test_threadpool.c.

{
    struct ast_threadpool *pool = NULL;
    struct ast_threadpool_listener *listener = 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 = 5,
        .max_size = 5,
    };
    struct efficiency_task_data etd = {
        .pool = NULL,
        .num_tasks_executed = 0,
        .shutdown = 0,
    };
    struct timeval start;
    struct timespec end;
    int i;
 
    switch (cmd) {
    case CLI_INIT:
        e->command = "threadpool push efficiency";
        e->usage =
            "Usage: threadpool push efficiency\n"
            "       Pushes 200 tasks to a threadpool and measures\n"
            "       the number of tasks executed within 30 seconds.\n";
        return NULL;
    case CLI_GENERATE:
        return NULL;
    }
 
    tld = test_alloc();
    if (!tld) {
        return CLI_SUCCESS;
    }
 
    listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
    if (!listener) {
        goto end;
    }
 
    pool = ast_threadpool_create("threadpool_push_efficiency", listener, &options);
    if (!pool) {
        goto end;
    }
 
    etd.pool = pool;
 
    /* Push in 200 tasks, cause why not */
    for (i = 0; i < 200; i++) {
        if (ast_threadpool_push(pool, efficiency_task, &etd)) {
            goto end;
        }
    }
 
    /* Wait for 30 seconds */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 30;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&tld->lock);
    while (ast_cond_timedwait(&tld->cond, &tld->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&tld->lock);
 
    /* Give the total tasks executed, and tell each task to not requeue */
    ast_cli(a->fd, "Total tasks executed in 30 seconds: %d\n", etd.num_tasks_executed);
    etd.shutdown = 1;
 
    res = wait_for_empty_notice(NULL, tld);
    if (res == AST_TEST_FAIL) {
        goto end;
    }
 
end:
    ast_threadpool_shutdown(pool);
    ao2_cleanup(listener);
    ast_free(tld);
    return CLI_SUCCESS;
}

References a, ao2_cleanup, ast_cli(), ast_cond_timedwait, ast_free, ast_mutex_lock, ast_mutex_unlock, AST_TEST_FAIL, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_push(), ast_threadpool_shutdown(), ast_tvnow(), CLI_GENERATE, CLI_INIT, CLI_SUCCESS, ast_cli_entry::command, test_listener_data::cond, efficiency_task(), end, listener(), test_listener_data::lock, NULL, efficiency_task_data::num_tasks_executed, options, efficiency_task_data::pool, efficiency_task_data::shutdown, test_alloc(), test_callbacks, ast_cli_entry::usage, and wait_for_empty_notice().

handle_cli_threadpool_push_serializer_efficiency()

static char * handle_cli_threadpool_push_serializer_efficiency ( struct ast_cli_entry *  e, int  cmd, struct ast_cli_args *  a  )

static

Definition at line 1889 of file test_threadpool.c.

{
    struct ast_threadpool *pool = NULL;
    struct ast_threadpool_listener *listener = 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 = 5,
        .max_size = 5,
    };
    struct serializer_efficiency_task_data etd[200];
    struct timeval start;
    struct timespec end;
    int i;
    int num_tasks_executed = 0;
    int shutdown = 0;
 
    switch (cmd) {
        case CLI_INIT:
            e->command = "threadpool push serializer efficiency";
            e->usage =
                "Usage: threadpool push serializer efficiency\n"
                "       Pushes 200 tasks to a threadpool in serializers and measures\n"
                "       the number of tasks executed within 30 seconds.\n";
            return NULL;
        case CLI_GENERATE:
            return NULL;
        }
 
    tld = test_alloc();
    if (!tld) {
        return CLI_SUCCESS;
    }
 
    memset(&etd, 0, sizeof(etd));
 
    listener = ast_threadpool_listener_alloc(&test_callbacks, tld);
    if (!listener) {
        goto end;
    }
 
    pool = ast_threadpool_create("threadpool_push_serializer_efficiency", listener, &options);
    if (!pool) {
        goto end;
    }
 
    /* We create 400 (200 pairs) of serializers */
    for (i = 0; i < 200; i++) {
        char serializer_name[AST_TASKPROCESSOR_MAX_NAME + 1];
 
        ast_taskprocessor_build_name(serializer_name, sizeof(serializer_name), "serializer%d", i);
        etd[i].serializer[0] = ast_threadpool_serializer(serializer_name, pool);
        if (!etd[i].serializer[0]) {
            goto end;
        }
 
        ast_taskprocessor_build_name(serializer_name, sizeof(serializer_name), "serializer%d", i);
        etd[i].serializer[1] = ast_threadpool_serializer(serializer_name, pool);
        if (!etd[i].serializer[1]) {
            goto end;
        }
 
        etd[i].num_tasks_executed = &num_tasks_executed;
        etd[i].shutdown = &shutdown;
    }
 
    /* And once created we push in 200 tasks */
    for (i = 0; i < 200; i++) {
        if (ast_taskprocessor_push(etd[i].serializer[0], serializer_efficiency_task, &etd[i])) {
            goto end;
        }
    }
 
    /* Wait for 30 seconds */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 30;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&tld->lock);
    while (ast_cond_timedwait(&tld->cond, &tld->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&tld->lock);
 
    /* Give the total tasks executed, and tell each task to not requeue */
    ast_cli(a->fd, "Total tasks executed in 30 seconds: %d\n", num_tasks_executed);
    shutdown = 1;
 
    res = wait_for_empty_notice(NULL, tld);
    if (res == AST_TEST_FAIL) {
        goto end;
    }
 
end:
    /* We need to unreference each serializer */
    for (i = 0; i < 200; i++) {
        ast_taskprocessor_unreference(etd[i].serializer[0]);
        ast_taskprocessor_unreference(etd[i].serializer[1]);
    }
    ast_threadpool_shutdown(pool);
    ao2_cleanup(listener);
    ast_free(tld);
    return CLI_SUCCESS;
}

References a, ao2_cleanup, ast_cli(), ast_cond_timedwait, ast_free, ast_mutex_lock, ast_mutex_unlock, ast_taskprocessor_build_name(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_push(), ast_taskprocessor_unreference(), AST_TEST_FAIL, ast_threadpool_create(), ast_threadpool_listener_alloc(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_serializer(), ast_threadpool_shutdown(), ast_tvnow(), CLI_GENERATE, CLI_INIT, CLI_SUCCESS, ast_cli_entry::command, test_listener_data::cond, end, listener(), test_listener_data::lock, NULL, serializer_efficiency_task_data::num_tasks_executed, options, serializer_efficiency_task_data::serializer, serializer_efficiency_task(), serializer_efficiency_task_data::shutdown, test_alloc(), test_callbacks, ast_cli_entry::usage, and wait_for_empty_notice().

has_complex_started()

static int has_complex_started ( struct complex_task_data *  ctd )

static

Definition at line 1484 of file test_threadpool.c.

{
    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;
}

References ast_cond_timedwait, ast_tvnow(), end, lock, complex_task_data::lock, complex_task_data::notify_cond, SCOPED_MUTEX, and complex_task_data::task_started.

Referenced by AST_TEST_DEFINE().

listener_check()

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  )

static

Definition at line 243 of file test_threadpool.c.

{
    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;
}

References AST_TEST_FAIL, AST_TEST_PASS, ast_test_status_update, ast_threadpool_listener_get_user_data(), test_listener_data::empty_notice, listener(), test_listener_data::num_active, test_listener_data::num_idle, test_listener_data::num_tasks, test_listener_data::task_pushed, and test_listener_data::was_empty.

Referenced by AST_TEST_DEFINE().

load_module()

static int load_module ( void  )

static

Definition at line 2078 of file test_threadpool.c.

{
    ast_cli_register_multiple(cli, ARRAY_LEN(cli));
    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;
}

References ARRAY_LEN, ast_cli_register_multiple, AST_MODULE_LOAD_SUCCESS, and cli.

poke_worker()

static void poke_worker ( struct complex_task_data *  ctd )

static

Definition at line 1459 of file test_threadpool.c.

{
    SCOPED_MUTEX(lock, &ctd->lock);
    ctd->continue_task = 1;
    ast_cond_signal(&ctd->stall_cond);
}

References ast_cond_signal, complex_task_data::continue_task, lock, complex_task_data::lock, SCOPED_MUTEX, and complex_task_data::stall_cond.

Referenced by AST_TEST_DEFINE().

serializer_efficiency_task()

static int serializer_efficiency_task ( void *  data )

static

Definition at line 1862 of file test_threadpool.c.

{
    struct serializer_efficiency_task_data *etd = data;
    struct ast_taskprocessor *taskprocessor = etd->serializer[0];
 
    if (*etd->shutdown) {
        return 0;
    }
 
    ast_atomic_fetchadd_int(etd->num_tasks_executed, +1);
 
    /* We ping pong a task between a pair of taskprocessors to ensure that
     * a single taskprocessor does not receive a thread from the threadpool
     * exclusively.
     */
    if (taskprocessor == ast_threadpool_serializer_get_current()) {
        taskprocessor = etd->serializer[1];
    }
 
    if (ast_taskprocessor_push(taskprocessor,
        serializer_efficiency_task, etd)) {
        return -1;
    }
 
    return 0;
}

References ast_atomic_fetchadd_int(), ast_taskprocessor_push(), ast_threadpool_serializer_get_current(), serializer_efficiency_task_data::num_tasks_executed, serializer_efficiency_task_data::serializer, serializer_efficiency_task(), and serializer_efficiency_task_data::shutdown.

Referenced by handle_cli_threadpool_push_serializer_efficiency(), and serializer_efficiency_task().

simple_task()

static int simple_task ( void *  data )

static

Definition at line 142 of file test_threadpool.c.

{
    struct simple_task_data *std = data;
    SCOPED_MUTEX(lock, &std->lock);
    std->task_executed = 1;
    ast_cond_signal(&std->cond);
    return 0;
}

References ast_cond_signal, simple_task_data::cond, lock, simple_task_data::lock, SCOPED_MUTEX, and simple_task_data::task_executed.

Referenced by AST_TEST_DEFINE().

simple_task_data_alloc()

static struct simple_task_data * simple_task_data_alloc ( void  )

static

Definition at line 118 of file test_threadpool.c.

{
    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;
}

References ast_calloc, ast_cond_init, ast_mutex_init, simple_task_data::cond, simple_task_data::lock, and NULL.

Referenced by AST_TEST_DEFINE().

simple_task_data_free()

static void simple_task_data_free ( struct simple_task_data *  std )

static

Definition at line 130 of file test_threadpool.c.

{
    if (!std) {
        return;
    }
 
    ast_mutex_destroy(&std->lock);
    ast_cond_destroy(&std->cond);
 
    ast_free(std);
}

References ast_cond_destroy, ast_free, ast_mutex_destroy, simple_task_data::cond, and simple_task_data::lock.

Referenced by AST_TEST_DEFINE().

test_alloc()

static struct test_listener_data * test_alloc ( void  )

static

Definition at line 53 of file test_threadpool.c.

{
    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;
}

References ast_calloc, ast_cond_init, ast_mutex_init, test_listener_data::cond, test_listener_data::lock, and NULL.

Referenced by AST_TEST_DEFINE(), handle_cli_threadpool_push_efficiency(), and handle_cli_threadpool_push_serializer_efficiency().

test_emptied()

static void test_emptied ( struct ast_threadpool *  pool, struct ast_threadpool_listener *  listener  )

static

Definition at line 89 of file test_threadpool.c.

{
    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);
}

References ast_cond_signal, ast_threadpool_listener_get_user_data(), test_listener_data::cond, test_listener_data::empty_notice, listener(), lock, test_listener_data::lock, and SCOPED_MUTEX.

test_shutdown()

static void test_shutdown ( struct ast_threadpool_listener *  listener )

static

Definition at line 98 of file test_threadpool.c.

{
    struct test_listener_data *tld = ast_threadpool_listener_get_user_data(listener);
    ast_cond_destroy(&tld->cond);
    ast_mutex_destroy(&tld->lock);
}

References ast_cond_destroy, ast_mutex_destroy, ast_threadpool_listener_get_user_data(), test_listener_data::cond, listener(), and test_listener_data::lock.

test_state_changed()

static void test_state_changed ( struct ast_threadpool *  pool, struct ast_threadpool_listener *  listener, int  active_threads, int  idle_threads  )

static

Definition at line 64 of file test_threadpool.c.

{
    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);
}

References ast_cond_signal, ast_log, ast_threadpool_listener_get_user_data(), test_listener_data::cond, listener(), lock, test_listener_data::lock, LOG_NOTICE, test_listener_data::num_active, test_listener_data::num_idle, and SCOPED_MUTEX.

test_task_pushed()

static void test_task_pushed ( struct ast_threadpool *  pool, struct ast_threadpool_listener *  listener, int  was_empty  )

static

Definition at line 77 of file test_threadpool.c.

{
    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);
}

References ast_cond_signal, ast_threadpool_listener_get_user_data(), test_listener_data::cond, listener(), lock, test_listener_data::lock, test_listener_data::num_tasks, SCOPED_MUTEX, test_listener_data::task_pushed, and test_listener_data::was_empty.

unload_module()

static int unload_module ( void  )

static

Definition at line 2056 of file test_threadpool.c.

{
    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);
    ast_cli_unregister_multiple(cli, ARRAY_LEN(cli));
    return 0;
}

References ARRAY_LEN, ast_cli_unregister_multiple(), and cli.

wait_for_completion()

static enum ast_test_result_state wait_for_completion ( struct ast_test *  test, struct simple_task_data *  std  )

static

Definition at line 194 of file test_threadpool.c.

{
    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;
}

References ast_cond_timedwait, AST_TEST_FAIL, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), simple_task_data::cond, end, lock, simple_task_data::lock, SCOPED_MUTEX, and simple_task_data::task_executed.

Referenced by AST_TEST_DEFINE().

wait_for_complex_completion()

static enum ast_test_result_state wait_for_complex_completion ( struct complex_task_data *  ctd )

static

Definition at line 1502 of file test_threadpool.c.

{
    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;
}

References ast_cond_timedwait, AST_TEST_FAIL, AST_TEST_PASS, ast_tvnow(), end, lock, complex_task_data::lock, complex_task_data::notify_cond, SCOPED_MUTEX, and complex_task_data::task_executed.

Referenced by AST_TEST_DEFINE().

wait_for_complex_start()

static int wait_for_complex_start ( struct complex_task_data *  ctd )

static

Definition at line 1466 of file test_threadpool.c.

{
    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;
}

References ast_cond_timedwait, ast_tvnow(), end, lock, complex_task_data::lock, complex_task_data::notify_cond, SCOPED_MUTEX, and complex_task_data::task_started.

Referenced by AST_TEST_DEFINE().

wait_for_empty_notice()

static enum ast_test_result_state wait_for_empty_notice ( struct ast_test *  test, struct test_listener_data *  tld  )

static

Definition at line 217 of file test_threadpool.c.

{
    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) {
        if (test) {
            ast_test_status_update(test, "Test listener not notified that threadpool is empty\n");
        }
        res = AST_TEST_FAIL;
    }
 
    return res;
}

References ast_cond_timedwait, AST_TEST_FAIL, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_listener_data::cond, test_listener_data::empty_notice, end, lock, test_listener_data::lock, and SCOPED_MUTEX.

Referenced by AST_TEST_DEFINE(), handle_cli_threadpool_push_efficiency(), and handle_cli_threadpool_push_serializer_efficiency().

wait_for_task_pushed()

static void wait_for_task_pushed ( struct ast_threadpool_listener *  listener )

static

Definition at line 177 of file test_threadpool.c.

{
    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;
        }
    }
}

References ast_cond_timedwait, ast_threadpool_listener_get_user_data(), ast_tvnow(), test_listener_data::cond, end, listener(), lock, test_listener_data::lock, SCOPED_MUTEX, and test_listener_data::task_pushed.

Referenced by AST_TEST_DEFINE().

wait_until_thread_state()

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  )

static

Definition at line 151 of file test_threadpool.c.

{
    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;
}

References ast_cond_timedwait, AST_TEST_FAIL, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_listener_data::cond, end, lock, test_listener_data::lock, test_listener_data::num_active, test_listener_data::num_idle, and SCOPED_MUTEX.

Referenced by AST_TEST_DEFINE(), and wait_until_thread_state_task_pushed().

wait_until_thread_state_task_pushed()

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  )

static

Definition at line 1063 of file test_threadpool.c.

{
    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;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, AST_TEST_FAIL, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_listener_data::cond, end, test_listener_data::lock, test_listener_data::num_tasks, and wait_until_thread_state().

Referenced by AST_TEST_DEFINE().

Variable Documentation

__mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "threadpool test module" , .key = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = AST_BUILDOPT_SUM, .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_DEFAULT, .support_level = AST_MODULE_SUPPORT_CORE, }

static

Definition at line 2100 of file test_threadpool.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 2100 of file test_threadpool.c.

cli

struct ast_cli_entry cli[]

static

Initial value:

= {
    AST_CLI_DEFINE(handle_cli_threadpool_push_efficiency, "Push tasks to a threadpool and measure efficiency"),
    AST_CLI_DEFINE(handle_cli_threadpool_push_serializer_efficiency, "Push tasks to a threadpool in serializers and measure efficiency"),
}

Definition at line 2051 of file test_threadpool.c.

Referenced by load_module(), and unload_module().

test_callbacks

const struct ast_threadpool_listener_callbacks test_callbacks

static

Definition at line 105 of file test_threadpool.c.

Referenced by AST_TEST_DEFINE(), handle_cli_threadpool_push_efficiency(), and handle_cli_threadpool_push_serializer_efficiency().