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

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 810 of file test_taskpool.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 810 of file test_taskpool.c.

AST_MODULE_SELF_SYM()

struct ast_module * AST_MODULE_SELF_SYM

(

void

)

Definition at line 810 of file test_taskpool.c.

AST_TEST_DEFINE() [1/7]

AST_TEST_DEFINE

(

taskpool_push

)

Definition at line 78 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, sip_to_pjsip::info(), test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [2/7]

AST_TEST_DEFINE

(

taskpool_push_grow

)

Definition at line 393 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, sip_to_pjsip::info(), test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [3/7]

AST_TEST_DEFINE

(

taskpool_push_serializer

)

Definition at line 190 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, sip_to_pjsip::info(), test_data::lock, NULL, options, simple_task(), test_data::taskprocessor, test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [4/7]

AST_TEST_DEFINE

(

taskpool_push_serializer_synchronous

)

Definition at line 264 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, sip_to_pjsip::info(), NULL, options, simple_task(), test_data::taskprocessor, test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [5/7]

AST_TEST_DEFINE

(

taskpool_push_serializer_synchronous_requeue

)

Definition at line 331 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, sip_to_pjsip::info(), NULL, options, requeue_task(), test_data::taskprocessor, test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [6/7]

AST_TEST_DEFINE

(

taskpool_push_shrink

)

Definition at line 467 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, sip_to_pjsip::info(), test_data::lock, NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

AST_TEST_DEFINE() [7/7]

AST_TEST_DEFINE

(

taskpool_push_synchronous

)

Definition at line 140 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, sip_to_pjsip::info(), NULL, options, simple_task(), test_alloc(), test_destroy(), TEST_EXECUTE, and TEST_INIT.

efficiency_task()

static int efficiency_task ( void *  data )

static

Definition at line 559 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 578 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\n", etd.num_tasks_executed);
 
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 685 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\n", num_tasks_executed);
    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 797 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);
    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 326 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 658 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 68 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(), and requeue_task().

test_alloc()

static struct test_data * test_alloc ( void  )

static

Definition at line 50 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(), 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 61 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(), handle_cli_taskpool_push_efficiency(), and handle_cli_taskpool_push_serializer_efficiency().

unload_module()

static int unload_module ( void  )

static

Definition at line 784 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);
    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 = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = AST_BUILDOPT_SUM, .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_DEFAULT, .support_level = AST_MODULE_SUPPORT_CORE, }

static

Definition at line 810 of file test_taskpool.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 810 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 779 of file test_taskpool.c.

Referenced by load_module(), and unload_module().