test_taskpool.c File Reference

taskpool 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/taskpool.h"
#include "asterisk/cli.h"

Include dependency graph for test_taskpool.c:

Go to the source code of this file.

Data Structures
struct	efficiency_task_data
struct	sel_data
struct	serializer_efficiency_task_data
struct	test_data
	Sorcery object created based on backend data. More...

Macros
#define	SEL_POOL   4

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
struct ast_module *	AST_MODULE_SELF_SYM (void)
	AST_TEST_DEFINE (taskpool_push)
	AST_TEST_DEFINE (taskpool_push_grow)
	AST_TEST_DEFINE (taskpool_push_serializer)
	AST_TEST_DEFINE (taskpool_push_serializer_synchronous)
	AST_TEST_DEFINE (taskpool_push_serializer_synchronous_requeue)
	AST_TEST_DEFINE (taskpool_push_shrink)
	AST_TEST_DEFINE (taskpool_push_synchronous)
	AST_TEST_DEFINE (taskpool_selector_skips_busy_executor)
	AST_TEST_DEFINE (taskpool_serializer_multiple_suspension)
	AST_TEST_DEFINE (taskpool_serializer_push_wait_while_suspended_from_other_serializer)
	AST_TEST_DEFINE (taskpool_serializer_suspension)
static int	cascade_task (void *data)
static int	efficiency_task (void *data)
static char *	handle_cli_taskpool_push_efficiency (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static char *	handle_cli_taskpool_push_serializer_efficiency (struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static int	load_module (void)
static int	requeue_task (void *data)
static int	sel_blocking_task (void *data)
static int	sel_probe_task (void *data)
static int	sel_run_case (struct ast_test *test, int probe_first, const char *label)
static int	serializer_efficiency_task (void *data)
static int	simple_task (void *data)
static struct test_data *	test_alloc (void)
static void	test_destroy (struct test_data *td)
static int	unload_module (void)

Variables
static struct ast_module_info	__mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "taskpool test module" , .key = ASTERISK_GPL_KEY , .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 []

Detailed Description

taskpool unit tests

Author

Joshua Colp jcolp.nosp@m.@san.nosp@m.goma..nosp@m.com

Definition in file test_taskpool.c.

Macro Definition Documentation

SEL_POOL

#define SEL_POOL   4

Definition at line 1070 of file test_taskpool.c.

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 1317 of file test_taskpool.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 1317 of file test_taskpool.c.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 1317 of file test_taskpool.c.

AST_TEST_DEFINE() [1/11]

AST_TEST_DEFINE

(

taskpool_push

)

Definition at line 79 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct timeval start;
    struct timespec end;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool pushing test";
        info->description =
            "Pushes a single task into a taskpool asynchronously and ensures it is executed.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    if (ast_taskpool_push(pool, simple_task, td)) {
        goto end;
    }
 
    /* It should not take more than 5 seconds for a single simple task to execute */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_push, ast_taskpool_shutdown(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [2/11]

AST_TEST_DEFINE

(

taskpool_push_grow

)

Definition at line 394 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 1,
        .minimum_size = 0,
        .initial_size = 0,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct timeval start;
    struct timespec end;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push_grow";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool pushing test with auto-grow enabled";
        info->description =
            "Pushes a single task into a taskpool asynchronously, ensures it is executed and the pool grows.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    if (ast_taskpool_taskprocessors_count(pool) != 0) {
        ast_test_status_update(test, "Expected taskpool to have 0 taskprocessors but it has %zu\n", ast_taskpool_taskprocessors_count(pool));
        res = AST_TEST_FAIL;
        goto end;
    }
 
    if (ast_taskpool_push(pool, simple_task, td)) {
        goto end;
    }
 
    if (ast_taskpool_taskprocessors_count(pool) != 1) {
        ast_test_status_update(test, "Expected taskpool to have 1 taskprocessor but it has %zu\n", ast_taskpool_taskprocessors_count(pool));
        res = AST_TEST_FAIL;
        goto end;
    }
 
    /* It should not take more than 5 seconds for a single simple task to execute */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_push, ast_taskpool_shutdown(), ast_taskpool_taskprocessors_count(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [3/11]

AST_TEST_DEFINE

(

taskpool_push_serializer

)

Definition at line 191 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct ast_taskprocessor *serializer = NULL;
    struct timeval start;
    struct timespec end;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push_serializer";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool serializer pushing test";
        info->description =
            "Pushes a single task into a taskpool serializer and ensures it is executed.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    serializer = ast_taskpool_serializer("serializer", pool);
    if (!serializer) {
        goto end;
    }
 
    if (ast_taskprocessor_push(serializer, simple_task, td)) {
        goto end;
    }
 
    /* It should not take more than 5 seconds for a single simple task to execute */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
    if (td->taskprocessor != serializer) {
        ast_test_status_update(test, "Expected taskprocessor to be same as serializer but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskprocessor_unreference(serializer);
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_shutdown(), ast_taskprocessor_push, ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, simple_task(), test_data::taskprocessor, test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [4/11]

AST_TEST_DEFINE

(

taskpool_push_serializer_synchronous

)

Definition at line 265 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct ast_taskprocessor *serializer = NULL;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push_serializer_synchronous";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool serializer synchronous pushing test";
        info->description =
            "Pushes a single task into a taskpool serializer synchronously and ensures it is executed.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    serializer = ast_taskpool_serializer("serializer", pool);
    if (!serializer) {
        goto end;
    }
 
    if (ast_taskpool_serializer_push_wait(serializer, simple_task, td)) {
        goto end;
    }
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
    if (td->taskprocessor != serializer) {
        ast_test_status_update(test, "Expected taskprocessor to be same as serializer but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskprocessor_unreference(serializer);
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_serializer_push_wait, ast_taskpool_shutdown(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, end, test_data::executed, NULL, options, simple_task(), test_data::taskprocessor, test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [5/11]

AST_TEST_DEFINE

(

taskpool_push_serializer_synchronous_requeue

)

Definition at line 332 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct ast_taskprocessor *serializer = NULL;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push_serializer_synchronous_requeue";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool serializer synchronous requeueing test";
        info->description =
            "Pushes a single task into a taskpool serializer synchronously and ensures it is requeued and executed.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    serializer = ast_taskpool_serializer("serializer", pool);
    if (!serializer) {
        goto end;
    }
 
    if (ast_taskpool_serializer_push_wait(serializer, requeue_task, td)) {
        goto end;
    }
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
    if (td->taskprocessor != serializer) {
        ast_test_status_update(test, "Expected taskprocessor to be same as serializer but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskprocessor_unreference(serializer);
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_serializer_push_wait, ast_taskpool_shutdown(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, end, test_data::executed, NULL, options, requeue_task(), test_data::taskprocessor, test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [6/11]

AST_TEST_DEFINE

(

taskpool_push_shrink

)

Definition at line 468 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 1,
        .auto_increment = 1,
        .minimum_size = 0,
        .initial_size = 0,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct timeval start;
    struct timespec end;
    int iterations = 0;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push_shrink";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool pushing test with auto-shrink enabled";
        info->description =
            "Pushes a single task into a taskpool asynchronously, ensures it is executed and the pool shrinks.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    if (ast_taskpool_taskprocessors_count(pool) != 0) {
        ast_test_status_update(test, "Expected taskpool to have 0 taskprocessors but it has %zu\n", ast_taskpool_taskprocessors_count(pool));
        res = AST_TEST_FAIL;
        goto end;
    }
 
    if (ast_taskpool_push(pool, simple_task, td)) {
        res = AST_TEST_FAIL;
        goto end;
    }
 
    if (ast_taskpool_taskprocessors_count(pool) != 1) {
        ast_test_status_update(test, "Expected taskpool to have 1 taskprocessor but it has %zu\n", ast_taskpool_taskprocessors_count(pool));
        res = AST_TEST_FAIL;
        goto end;
    }
 
    /* We give 10 seconds for the pool to shrink back to normal, but if it happens earlier we
     * stop our check early.
     */
    ast_mutex_lock(&td->lock);
    do {
        start = ast_tvnow();
        end.tv_sec = start.tv_sec + 1;
        end.tv_nsec = start.tv_usec * 1000;
 
        if (ast_cond_timedwait(&td->cond, &td->lock, &end) == ETIMEDOUT) {
            iterations++;
        }
    } while (ast_taskpool_taskprocessors_count(pool) != 0 && iterations != 10);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
    if (ast_taskpool_taskprocessors_count(pool) != 0) {
        ast_test_status_update(test, "Expected taskpool to have 0 taskprocessors but it has %zu\n", ast_taskpool_taskprocessors_count(pool));
        res = AST_TEST_FAIL;
        goto end;
    }
 
end:
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_push, ast_taskpool_shutdown(), ast_taskpool_taskprocessors_count(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [7/11]

AST_TEST_DEFINE

(

taskpool_push_synchronous

)

Definition at line 141 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "push_synchronous";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool synchronous pushing test";
        info->description =
            "Pushes a single task into a taskpool synchronously and ensures it is executed.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    if (ast_taskpool_push_wait(pool, simple_task, td)) {
        goto end;
    }
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_push_wait, ast_taskpool_shutdown(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, end, test_data::executed, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [8/11]

AST_TEST_DEFINE

(

taskpool_selector_skips_busy_executor

)

Definition at line 1242 of file test_taskpool.c.

{
    int ran_after;
    int ran_before;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "selector_skips_busy_executor";
        info->category = "/main/taskpool/";
        info->summary = "A task should not be assigned to an executor that is busy";
        info->description =
            "Occupies one executor of a four-executor pool with a task that "
            "blocks until released, then pushes a trivial task on a different "
            "serializer while three executors are free. Runs the scenario "
            "twice: once with the second serializer created after the blocking "
            "task is already running, and once with it created beforehand, so "
            "the result cannot be attributed to both serializers having been "
            "created while the pool was idle.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
 
    ran_after = sel_run_case(test, 0, "serializer created while an executor is busy");
    ran_before = sel_run_case(test, 1, "serializer created while the pool is idle");
 
    if (!ran_after || !ran_before) {
        ast_test_status_update(test,
            "a trivial task was queued behind the blocked executor while %d "
            "executors were free\n", SEL_POOL - 1);
        return AST_TEST_FAIL;
    }
 
    return AST_TEST_PASS;
}

References AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, SEL_POOL, sel_run_case(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [9/11]

AST_TEST_DEFINE

(

taskpool_serializer_multiple_suspension

)

Definition at line 642 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskprocessor *serializer = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct timeval start;
    struct timespec end;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "serializer_multiple_suspension";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool serializer multiple suspension test";
        info->description =
            "Pushes a single task into a taskpool serializer asynchronously while suspended multiple times, and ensures it only executes after unsuspension.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    serializer = ast_taskpool_serializer("serializer", pool);
    if (!serializer) {
        goto end;
    }
 
    ast_taskpool_serializer_suspend(serializer);
    ast_taskpool_serializer_suspend(serializer);
 
    if (ast_taskprocessor_push(serializer, simple_task, td)) {
        goto end;
    }
 
    /* Give 5 seconds to ensure the task isn't executed */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (td->executed) {
        ast_test_status_update(test, "Expected simple task to not be executed but it was\n");
        res = AST_TEST_FAIL;
    }
 
    ast_taskpool_serializer_unsuspend(serializer);
    ast_taskpool_serializer_unsuspend(serializer);
 
    /* Give 5 seconds to ensure the task is executed */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskprocessor_unreference(serializer);
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_serializer_suspend(), ast_taskpool_serializer_unsuspend(), ast_taskpool_shutdown(), ast_taskprocessor_push, ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [10/11]

AST_TEST_DEFINE

(

taskpool_serializer_push_wait_while_suspended_from_other_serializer

)

Definition at line 743 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskprocessor *suspended_serializer = NULL;
    struct ast_taskprocessor *push_serializer = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct timeval start;
    struct timespec end;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "serializer_push_wait_while_suspended_from_other_serializer";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool serializer push wait while suspended from other serializer test";
        info->description =
            "Pushes a single task into a taskpool serializer synchronously from another serializer while suspended, and ensures it only executes after unsuspension.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    suspended_serializer = ast_taskpool_serializer("suspended_serializer", pool);
    if (!suspended_serializer) {
        goto end;
    }
 
    td->test_specific_data = suspended_serializer;
 
    push_serializer = ast_taskpool_serializer("push_serializer", pool);
    if (!push_serializer) {
        goto end;
    }
 
    ast_taskpool_serializer_suspend(suspended_serializer);
 
    /* We push a task to the unsuspended serializer which will push a task to the suspended serializer
     * in a synchronous fashion. That task can not execute until the suspended serializer becomes
     * unsuspended.
     */
    if (ast_taskprocessor_push(push_serializer, cascade_task, td)) {
        goto end;
    }
 
    /* Give 5 seconds to ensure the task isn't executed */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (td->executed) {
        ast_test_status_update(test, "Expected simple task to not be executed but it was\n");
        res = AST_TEST_FAIL;
    }
 
    ast_taskpool_serializer_unsuspend(suspended_serializer);
 
    /* Give 5 seconds to ensure the task is executed */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskprocessor_unreference(suspended_serializer);
    ast_taskprocessor_unreference(push_serializer);
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_serializer_suspend(), ast_taskpool_serializer_unsuspend(), ast_taskpool_shutdown(), ast_taskprocessor_push, ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), cascade_task(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, test_alloc(), test_destroy(), TEST_EXECUTE, TEST_INIT, and test_data::test_specific_data.

AST_TEST_DEFINE() [11/11]

AST_TEST_DEFINE

(

taskpool_serializer_suspension

)

Definition at line 554 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskprocessor *serializer = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 1,
        .initial_size = 1,
        .max_size = 1,
    };
    enum ast_test_result_state res = AST_TEST_PASS;
    struct timeval start;
    struct timespec end;
 
    switch (cmd) {
    case TEST_INIT:
        info->name = "serializer_suspension";
        info->category = "/main/taskpool/";
        info->summary = "Taskpool serializer suspension test";
        info->description =
            "Pushes a single task into a taskpool serializer asynchronously while suspended, and ensures it only executes after unsuspension.";
        return AST_TEST_NOT_RUN;
    case TEST_EXECUTE:
        break;
    }
    td = test_alloc();
    if (!td) {
        return AST_TEST_FAIL;
    }
 
    pool = ast_taskpool_create(info->name, &options);
    if (!pool) {
        goto end;
    }
 
    serializer = ast_taskpool_serializer("serializer", pool);
    if (!serializer) {
        goto end;
    }
 
    ast_taskpool_serializer_suspend(serializer);
 
    if (ast_taskprocessor_push(serializer, simple_task, td)) {
        goto end;
    }
 
    /* Give 5 seconds to ensure the task isn't executed */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (td->executed) {
        ast_test_status_update(test, "Expected simple task to not be executed but it was\n");
        res = AST_TEST_FAIL;
    }
 
    ast_taskpool_serializer_unsuspend(serializer);
 
    /* Give 5 seconds to ensure the task is executed */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&td->lock);
    while (!td->executed && ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    if (!td->executed) {
        ast_test_status_update(test, "Expected simple task to be executed but it was not\n");
        res = AST_TEST_FAIL;
    }
 
end:
    ast_taskprocessor_unreference(serializer);
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return res;
}

References ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_serializer_suspend(), ast_taskpool_serializer_unsuspend(), ast_taskpool_shutdown(), ast_taskprocessor_push, ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), test_data::cond, end, test_data::executed, test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

cascade_task()

static int cascade_task ( void *  data )

static

Definition at line 732 of file test_taskpool.c.

{
    struct test_data *td = data;
 
    if (ast_taskpool_serializer_push_wait(td->test_specific_data, simple_task, td)) {
        return -1;
    }
 
    return 0;
}

References ast_taskpool_serializer_push_wait, simple_task(), and test_data::test_specific_data.

Referenced by AST_TEST_DEFINE().

efficiency_task()

static int efficiency_task ( void *  data )

static

Definition at line 850 of file test_taskpool.c.

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

References ao2_ref, ast_atomic_fetchadd_int(), ast_taskpool_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_taskpool_push_efficiency().

handle_cli_taskpool_push_efficiency()

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

static

Definition at line 869 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 5,
        .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 = "taskpool push efficiency";
        e->usage =
            "Usage: taskpool push efficiency\n"
            "       Pushes 200 tasks to a taskpool and measures\n"
            "       the number of tasks executed within 30 seconds.\n";
        return NULL;
    case CLI_GENERATE:
        return NULL;
    }
 
    td = test_alloc();
    if (!td) {
        return CLI_SUCCESS;
    }
 
    pool = ast_taskpool_create("taskpool_push_efficiency", &options);
    if (!pool) {
        goto end;
    }
 
    etd.pool = pool;
 
    /* Push in 200 tasks, cause why not */
    for (i = 0; i < 200; i++) {
        /* Ensure that the task has a reference to the pool */
        ao2_bump(pool);
        if (ast_taskpool_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(&td->lock);
    while (ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    /* Give the total tasks executed, and tell each task to not requeue */
    ast_cli(a->fd, "Total tasks executed in 30 seconds: %d (%d per second)\n", etd.num_tasks_executed, etd.num_tasks_executed / 30);
 
end:
    etd.shutdown = 1;
    ast_taskpool_shutdown(pool);
    test_destroy(td);
    return CLI_SUCCESS;
}

References a, ao2_bump, ast_cli(), ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_push, ast_taskpool_shutdown(), ast_tvnow(), CLI_GENERATE, CLI_INIT, CLI_SUCCESS, ast_cli_entry::command, test_data::cond, efficiency_task(), end, test_data::lock, NULL, efficiency_task_data::num_tasks_executed, options, efficiency_task_data::pool, efficiency_task_data::shutdown, test_alloc(), test_destroy(), and ast_cli_entry::usage.

handle_cli_taskpool_push_serializer_efficiency()

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

static

Definition at line 976 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct test_data *td = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = 5,
        .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 = "taskpool push serializer efficiency";
        e->usage =
            "Usage: taskpool push serializer efficiency\n"
            "       Pushes 200 tasks to a taskpool in serializers and measures\n"
            "       the number of tasks executed within 30 seconds.\n";
        return NULL;
    case CLI_GENERATE:
        return NULL;
    }
    td = test_alloc();
    if (!td) {
        return CLI_SUCCESS;
    }
 
    memset(&etd, 0, sizeof(etd));
 
    pool = ast_taskpool_create("taskpool_push_serializer_efficiency", &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_taskpool_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_taskpool_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(&td->lock);
    while (ast_cond_timedwait(&td->cond, &td->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&td->lock);
 
    /* Give the total tasks executed, and tell each task to not requeue */
    ast_cli(a->fd, "Total tasks executed in 30 seconds: %d (%d per second)\n", num_tasks_executed, num_tasks_executed / 30);
    shutdown = 1;
 
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_taskpool_shutdown(pool);
    test_destroy(td);
    return CLI_SUCCESS;
}

References a, ast_cli(), ast_cond_timedwait, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_shutdown(), ast_taskprocessor_build_name(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_push, ast_taskprocessor_unreference(), ast_tvnow(), CLI_GENERATE, CLI_INIT, CLI_SUCCESS, ast_cli_entry::command, test_data::cond, end, test_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_destroy(), and ast_cli_entry::usage.

load_module()

static int load_module ( void  )

static

Definition at line 1300 of file test_taskpool.c.

{
    ast_cli_register_multiple(cli, ARRAY_LEN(cli));
    AST_TEST_REGISTER(taskpool_push);
    AST_TEST_REGISTER(taskpool_push_synchronous);
    AST_TEST_REGISTER(taskpool_push_serializer);
    AST_TEST_REGISTER(taskpool_push_serializer_synchronous);
    AST_TEST_REGISTER(taskpool_push_serializer_synchronous_requeue);
    AST_TEST_REGISTER(taskpool_push_grow);
    AST_TEST_REGISTER(taskpool_push_shrink);
    AST_TEST_REGISTER(taskpool_serializer_suspension);
    AST_TEST_REGISTER(taskpool_serializer_multiple_suspension);
    AST_TEST_REGISTER(taskpool_serializer_push_wait_while_suspended_from_other_serializer);
    AST_TEST_REGISTER(taskpool_selector_skips_busy_executor);
    return AST_MODULE_LOAD_SUCCESS;
}

References ARRAY_LEN, ast_cli_register_multiple, AST_MODULE_LOAD_SUCCESS, AST_TEST_REGISTER, and cli.

requeue_task()

static int requeue_task ( void *  data )

static

Definition at line 327 of file test_taskpool.c.

{
    return ast_taskpool_serializer_push_wait(ast_taskpool_serializer_get_current(), simple_task, data);
}

References ast_taskpool_serializer_get_current(), ast_taskpool_serializer_push_wait, and simple_task().

Referenced by AST_TEST_DEFINE().

sel_blocking_task()

static int sel_blocking_task ( void *  data )

static

Definition at line 1081 of file test_taskpool.c.

{
    struct sel_data *sd = data;
 
    ast_mutex_lock(&sd->lock);
    sd->blocker_running = 1;
    ast_cond_broadcast(&sd->cond);
    while (!sd->blocker_release) {
        ast_cond_wait(&sd->cond, &sd->lock);
    }
    sd->blocker_done = 1;
    ast_cond_broadcast(&sd->cond);
    ast_mutex_unlock(&sd->lock);
    return 0;
}

References ast_cond_broadcast, ast_cond_wait, ast_mutex_lock, ast_mutex_unlock, sel_data::blocker_done, sel_data::blocker_release, sel_data::blocker_running, sel_data::cond, and sel_data::lock.

Referenced by sel_run_case().

sel_probe_task()

static int sel_probe_task ( void *  data )

static

Definition at line 1097 of file test_taskpool.c.

{
    struct sel_data *sd = data;
 
    ast_mutex_lock(&sd->lock);
    sd->probe_ran = 1;
    ast_cond_broadcast(&sd->cond);
    ast_mutex_unlock(&sd->lock);
    return 0;
}

References ast_cond_broadcast, ast_mutex_lock, ast_mutex_unlock, sel_data::cond, sel_data::lock, and sel_data::probe_ran.

Referenced by sel_run_case().

sel_run_case()

static int sel_run_case ( struct ast_test *  test, int  probe_first, const char *  label  )

static

Definition at line 1118 of file test_taskpool.c.

{
    struct ast_taskpool *pool = NULL;
    struct sel_data *sd = NULL;
    struct ast_taskpool_options options = {
        .version = AST_TASKPOOL_OPTIONS_VERSION,
        .idle_timeout = 0,
        .auto_increment = 0,
        .minimum_size = SEL_POOL,
        .initial_size = SEL_POOL,
        .max_size = SEL_POOL,
    };
    struct ast_taskprocessor *blocker_ser = NULL;
    struct ast_taskprocessor *probe_ser = NULL;
    int ran = 0;
    int running;
    struct timeval start;
    struct timespec end;
 
    sd = ast_calloc(1, sizeof(*sd));
    if (!sd) {
        return 0;
    }
    ast_mutex_init(&sd->lock);
    ast_cond_init(&sd->cond, NULL);
 
    pool = ast_taskpool_create(label, &options);
    if (!pool) {
        ast_test_status_update(test, "%s: could not create pool\n", label);
        goto cleanup;
    }
 
    blocker_ser = ast_taskpool_serializer("blocker", pool);
    if (!blocker_ser) {
        goto cleanup;
    }
 
    if (probe_first) {
        probe_ser = ast_taskpool_serializer("probe", pool);
        if (!probe_ser) {
            goto cleanup;
        }
    }
 
    if (ast_taskprocessor_push(blocker_ser, sel_blocking_task, sd)) {
        goto cleanup;
    }
 
    /* Wait until the blocker is definitely inside its task. */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 5;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&sd->lock);
    while (!sd->blocker_running
        && ast_cond_timedwait(&sd->cond, &sd->lock, &end) != ETIMEDOUT) {
    }
    running = sd->blocker_running;
    ast_mutex_unlock(&sd->lock);
 
    if (!running) {
        ast_test_status_update(test, "%s: blocking task never started\n", label);
        goto release;
    }
 
    /*
     * With probe_first == 0 the probe serializer is created here, while one
     * executor is already busy and three are free.
     */
    if (!probe_first) {
        probe_ser = ast_taskpool_serializer("probe", pool);
        if (!probe_ser) {
            goto release;
        }
    }
 
    if (ast_taskprocessor_push(probe_ser, sel_probe_task, sd)) {
        goto release;
    }
 
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 3;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&sd->lock);
    while (!sd->probe_ran
        && ast_cond_timedwait(&sd->cond, &sd->lock, &end) != ETIMEDOUT) {
    }
    ran = sd->probe_ran;
    ast_mutex_unlock(&sd->lock);
 
    ast_test_status_update(test,
        "%s: one executor busy, %d of %d free -> probe %s\n",
        label, SEL_POOL - 1, SEL_POOL,
        ran ? "ran on a free executor" : "did NOT run within 3s");
 
release:
    /* Wait for the blocking task to leave so the pool is idle before shutdown */
    start = ast_tvnow();
    end.tv_sec = start.tv_sec + 3;
    end.tv_nsec = start.tv_usec * 1000;
 
    ast_mutex_lock(&sd->lock);
    sd->blocker_release = 1;
    ast_cond_broadcast(&sd->cond);
    while (!sd->blocker_done
        && ast_cond_timedwait(&sd->cond, &sd->lock, &end) != ETIMEDOUT) {
    }
    ast_mutex_unlock(&sd->lock);
 
cleanup:
    ast_taskprocessor_unreference(blocker_ser);
    ast_taskprocessor_unreference(probe_ser);
    if (pool) {
        ast_taskpool_shutdown(pool);
    }
    if (sd) {
        ast_mutex_destroy(&sd->lock);
        ast_cond_destroy(&sd->cond);
        ast_free(sd);
    }
    return ran;
}

References ast_calloc, ast_cond_broadcast, ast_cond_destroy, ast_cond_init, ast_cond_timedwait, ast_free, ast_mutex_destroy, ast_mutex_init, ast_mutex_lock, ast_mutex_unlock, ast_taskpool_create(), AST_TASKPOOL_OPTIONS_VERSION, ast_taskpool_serializer(), ast_taskpool_shutdown(), ast_taskprocessor_push, ast_taskprocessor_unreference(), ast_test_status_update, ast_tvnow(), sel_data::blocker_done, sel_data::blocker_release, sel_data::blocker_running, cleanup(), sel_data::cond, end, sel_data::lock, NULL, options, sel_data::probe_ran, sel_blocking_task(), SEL_POOL, and sel_probe_task().

Referenced by AST_TEST_DEFINE().

serializer_efficiency_task()

static int serializer_efficiency_task ( void *  data )

static

Definition at line 949 of file test_taskpool.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_taskpool_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_taskpool_serializer_get_current(), ast_taskprocessor_push, 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_taskpool_push_serializer_efficiency(), and serializer_efficiency_task().

simple_task()

static int simple_task ( void *  data )

static

Definition at line 69 of file test_taskpool.c.

{
    struct test_data *td = data;
    SCOPED_MUTEX(lock, &td->lock);
    td->taskprocessor = ast_taskpool_serializer_get_current();
    td->executed = 1;
    ast_cond_signal(&td->cond);
    return 0;
}

References ast_cond_signal, ast_taskpool_serializer_get_current(), test_data::cond, test_data::executed, lock, test_data::lock, SCOPED_MUTEX, and test_data::taskprocessor.

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), cascade_task(), and requeue_task().

test_alloc()

static struct test_data * test_alloc ( void  )

static

Definition at line 51 of file test_taskpool.c.

{
    struct test_data *td = ast_calloc(1, sizeof(*td));
    if (!td) {
        return NULL;
    }
    ast_mutex_init(&td->lock);
    ast_cond_init(&td->cond, NULL);
    return td;
}

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

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), handle_cli_taskpool_push_efficiency(), and handle_cli_taskpool_push_serializer_efficiency().

test_destroy()

static void test_destroy ( struct test_data *  td )

static

Definition at line 62 of file test_taskpool.c.

{
    ast_mutex_destroy(&td->lock);
    ast_cond_destroy(&td->cond);
    ast_free(td);
}

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

Referenced by AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), AST_TEST_DEFINE(), handle_cli_taskpool_push_efficiency(), and handle_cli_taskpool_push_serializer_efficiency().

unload_module()

static int unload_module ( void  )

static

Definition at line 1283 of file test_taskpool.c.

{
    ast_cli_unregister_multiple(cli, ARRAY_LEN(cli));
    AST_TEST_UNREGISTER(taskpool_push);
    AST_TEST_UNREGISTER(taskpool_push_synchronous);
    AST_TEST_UNREGISTER(taskpool_push_serializer);
    AST_TEST_UNREGISTER(taskpool_push_serializer_synchronous);
    AST_TEST_UNREGISTER(taskpool_push_serializer_synchronous_requeue);
    AST_TEST_UNREGISTER(taskpool_push_grow);
    AST_TEST_UNREGISTER(taskpool_push_shrink);
    AST_TEST_UNREGISTER(taskpool_serializer_suspension);
    AST_TEST_UNREGISTER(taskpool_serializer_multiple_suspension);
    AST_TEST_UNREGISTER(taskpool_serializer_push_wait_while_suspended_from_other_serializer);
    AST_TEST_UNREGISTER(taskpool_selector_skips_busy_executor);
    return 0;
}

References ARRAY_LEN, ast_cli_unregister_multiple(), AST_TEST_UNREGISTER, and cli.

Variable Documentation

__mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "taskpool test module" , .key = ASTERISK_GPL_KEY , .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 1317 of file test_taskpool.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 1317 of file test_taskpool.c.

cli

struct ast_cli_entry cli[]

static

Initial value:

= {
    { .handler =  handle_cli_taskpool_push_efficiency , .summary =  "Push tasks to a taskpool and measure efficiency" ,},
    { .handler =  handle_cli_taskpool_push_serializer_efficiency , .summary =  "Push tasks to a taskpool in serializers and measure efficiency" ,},
}

Definition at line 1278 of file test_taskpool.c.

                                    {
    AST_CLI_DEFINE(handle_cli_taskpool_push_efficiency, "Push tasks to a taskpool and measure efficiency"),
    AST_CLI_DEFINE(handle_cli_taskpool_push_serializer_efficiency, "Push tasks to a taskpool in serializers and measure efficiency"),
};

Referenced by load_module(), and unload_module().