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	serializer_efficiency_task_data
struct	test_data
	Sorcery object created based on backend data. More...

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_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	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.

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 1107 of file test_taskpool.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 1107 of file test_taskpool.c.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 1107 of file test_taskpool.c.

AST_TEST_DEFINE() [1/10]

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/10]

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/10]

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/10]

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/10]

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/10]

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/10]

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/10]

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() [9/10]

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() [10/10]

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 1091 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);
    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().

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 1075 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);
    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 1107 of file test_taskpool.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 1107 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 1070 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().