test_taskprocessor.c File Reference

taskprocessor unit tests More...

#include "asterisk.h"
#include "asterisk/test.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/module.h"
#include "asterisk/astobj2.h"
#include "asterisk/serializer.h"
#include "asterisk/threadpool.h"

Include dependency graph for test_taskprocessor.c:

Go to the source code of this file.

Data Structures
struct	load_task_data
	Relevant data associated with taskprocessor load test. More...
struct	shutdown_data
struct	task_data
	userdata associated with baseline taskprocessor test More...
struct	test_listener_pvt
	Private data for the test taskprocessor listener. More...

Macros
#define	HIGH_WATER_MARK   6
#define	LOW_WATER_MARK   3
#define	NUM_TASKS   20000
#define	TEST_DATA_ARRAY_SIZE   10

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
struct ast_module *	AST_MODULE_SELF_SYM (void)
	AST_TEST_DEFINE (default_taskprocessor)
	Baseline test for default taskprocessor. More...
	AST_TEST_DEFINE (subsystem_alert)
	Baseline test for subsystem alert. More...
	AST_TEST_DEFINE (default_taskprocessor_load)
	Load test for taskprocessor with default listener. More...
	AST_TEST_DEFINE (taskprocessor_listener)
	Test for a taskprocessor with custom listener. More...
	AST_TEST_DEFINE (taskprocessor_shutdown)
	AST_TEST_DEFINE (taskprocessor_push_local)
	AST_TEST_DEFINE (serializer_pool)
	Baseline test for a serializer pool. More...
static int	check_stats (struct ast_test *test, const struct test_listener_pvt *pvt, int num_pushed, int num_emptied, int num_was_empty)
	helper to ensure that statistics the listener is keeping are what we expect More...
static int	listener_test_task (void *ignore)
	Queued task for taskprocessor listener test. More...
static int	load_module (void)
static int	load_task (void *data)
	a queued task to be used in the taskprocessor load test More...
static int	local_task_exe (struct ast_taskprocessor_local *local)
static struct shutdown_data *	shutdown_data_create (int dont_wait)
static void	shutdown_data_dtor (void *data)
static int	shutdown_has_completed (struct shutdown_data *shutdown_data)
static void	shutdown_poke (struct shutdown_data *shutdown_data)
static int	shutdown_task_exec (void *data)
static int	shutdown_waitfor_completion (struct shutdown_data *shutdown_data)
static int	shutdown_waitfor_start (struct shutdown_data *shutdown_data)
static int	task (void *data)
	Queued task for baseline test. More...
static struct task_data *	task_data_create (void)
	Create a task_data object. More...
static void	task_data_dtor (void *obj)
static int	task_wait (struct task_data *task_data)
	Wait for a task to execute. More...
static void	test_emptied (struct ast_taskprocessor_listener *listener)
	test taskprocessor listener's emptied callback. More...
static void *	test_listener_pvt_alloc (void)
	test taskprocessor listener's alloc callback More...
static void	test_shutdown (struct ast_taskprocessor_listener *listener)
	test taskprocessor listener's shutdown callback. More...
static int	test_start (struct ast_taskprocessor_listener *listener)
	test taskprocessor listener's start callback More...
static void	test_task_pushed (struct ast_taskprocessor_listener *listener, int was_empty)
	test taskprocessor listener's task_pushed callback More...
static void *	tps_shutdown_thread (void *data)
static int	unload_module (void)

Variables
static struct ast_module_info	__mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "taskprocessor 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 load_task_data	load_task_results
static const struct ast_taskprocessor_listener_callbacks	test_callbacks

Detailed Description

taskprocessor unit tests

Author

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

Definition in file test_taskprocessor.c.

Macro Definition Documentation

HIGH_WATER_MARK

#define HIGH_WATER_MARK   6

Referenced by AST_TEST_DEFINE().

LOW_WATER_MARK

#define LOW_WATER_MARK   3

Referenced by AST_TEST_DEFINE().

NUM_TASKS

#define NUM_TASKS   20000

Definition at line 314 of file test_taskprocessor.c.

Referenced by AST_TEST_DEFINE().

TEST_DATA_ARRAY_SIZE

#define TEST_DATA_ARRAY_SIZE   10

Referenced by AST_TEST_DEFINE().

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 990 of file test_taskprocessor.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 990 of file test_taskprocessor.c.

AST_MODULE_SELF_SYM()

struct ast_module* AST_MODULE_SELF_SYM

(

void

)

Definition at line 990 of file test_taskprocessor.c.

AST_TEST_DEFINE() [1/7]

AST_TEST_DEFINE

(

default_taskprocessor

)

Baseline test for default taskprocessor.

This test ensures that when a task is added to a taskprocessor that has been allocated with a default listener that the task gets executed as expected

Definition at line 132 of file test_taskprocessor.c.

References ao2_cleanup, ast_taskprocessor_get(), ast_taskprocessor_push(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, sip_to_pjsip::info(), NULL, RAII_VAR, task(), task_data_create(), task_wait(), TEST_EXECUTE, TEST_INIT, and TPS_REF_DEFAULT.

{
   RAII_VAR(struct ast_taskprocessor *, tps, NULL, ast_taskprocessor_unreference);
   RAII_VAR(struct task_data *, task_data, NULL, ao2_cleanup);
   int res;

   switch (cmd) {
   case TEST_INIT:
      info->name = "default_taskprocessor";
      info->category = "/main/taskprocessor/";
      info->summary = "Test of default taskproccesor";
      info->description =
         "Ensures that a queued task gets executed.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }

   tps = ast_taskprocessor_get("test", TPS_REF_DEFAULT);

   if (!tps) {
      ast_test_status_update(test, "Unable to create test taskprocessor\n");
      return AST_TEST_FAIL;
   }

   task_data = task_data_create();
   if (!task_data) {
      ast_test_status_update(test, "Unable to create task_data\n");
      return AST_TEST_FAIL;
   }

   if (ast_taskprocessor_push(tps, task, task_data)) {
      ast_test_status_update(test, "Failed to queue task\n");
      return AST_TEST_FAIL;
   }

   res = task_wait(task_data);
   if (res != 0) {
      ast_test_status_update(test, "Queued task did not execute!\n");
      return AST_TEST_FAIL;
   }

   return AST_TEST_PASS;
}

AST_TEST_DEFINE() [2/7]

AST_TEST_DEFINE

(

subsystem_alert

)

Baseline test for subsystem alert.

Definition at line 180 of file test_taskprocessor.c.

References ao2_cleanup, ast_taskprocessor_alert_get(), ast_taskprocessor_alert_set_levels(), ast_taskprocessor_get(), ast_taskprocessor_get_subsystem_alert(), ast_taskprocessor_push(), ast_taskprocessor_size(), ast_taskprocessor_suspend(), ast_taskprocessor_unreference(), ast_taskprocessor_unsuspend(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, data_cleanup(), HIGH_WATER_MARK, sip_to_pjsip::info(), LOW_WATER_MARK, NULL, RAII_VAR, task(), task_data_create(), task_wait(), TEST_DATA_ARRAY_SIZE, TEST_EXECUTE, TEST_INIT, TPS_REF_DEFAULT, and task_data::wait_time.

{
   RAII_VAR(struct ast_taskprocessor *, tps, NULL, ast_taskprocessor_unreference);
#define TEST_DATA_ARRAY_SIZE 10
#define LOW_WATER_MARK 3
#define HIGH_WATER_MARK 6
   struct task_data *task_data[(TEST_DATA_ARRAY_SIZE + 1)] = { 0 };
   int res = 0;
   int i;
   long queue_count;
   unsigned int alert_level;
   unsigned int subsystem_alert_level;

   switch (cmd) {
   case TEST_INIT:
      info->name = "subsystem_alert";
      info->category = "/main/taskprocessor/";
      info->summary = "Test of subsystem alerts";
      info->description =
         "Ensures alerts are generated properly.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }

   tps = ast_taskprocessor_get("test_subsystem/test", TPS_REF_DEFAULT);

   if (!tps) {
      ast_test_status_update(test, "Unable to create test taskprocessor\n");
      return AST_TEST_FAIL;
   }

   ast_taskprocessor_alert_set_levels(tps, LOW_WATER_MARK, HIGH_WATER_MARK);
   ast_taskprocessor_suspend(tps);

   for (i = 1; i <= TEST_DATA_ARRAY_SIZE; i++) {
      task_data[i] = task_data_create();
      if (!task_data[i]) {
         ast_test_status_update(test, "Unable to create task_data\n");
         res = -1;
         goto data_cleanup;
      }
      task_data[i]->wait_time = 500;

      ast_test_status_update(test, "Pushing task %d\n", i);
      if (ast_taskprocessor_push(tps, task, task_data[i])) {
         ast_test_status_update(test, "Failed to queue task\n");
         res = -1;
         goto data_cleanup;
      }

      queue_count = ast_taskprocessor_size(tps);
      alert_level = ast_taskprocessor_alert_get();
      subsystem_alert_level = ast_taskprocessor_get_subsystem_alert("test_subsystem");

      if (queue_count == HIGH_WATER_MARK) {
         if (subsystem_alert_level) {
            ast_test_status_update(test, "Subsystem alert triggered correctly at %ld\n", queue_count);
         }
         if (alert_level) {
            ast_test_status_update(test, "Global alert triggered correctly at %ld\n", queue_count);
         }
      } else if (queue_count < HIGH_WATER_MARK) {
         if (subsystem_alert_level > 0) {
            ast_test_status_update(test, "Subsystem alert triggered unexpectedly at %ld\n", queue_count);
            res = -1;
         }
         if (alert_level > 0) {
            ast_test_status_update(test, "Global alert triggered unexpectedly at %ld\n", queue_count);
            res = -1;
         }
      } else {
         if (subsystem_alert_level == 0) {
            ast_test_status_update(test, "Subsystem alert failed to trigger at %ld\n", queue_count);
            res = -1;
         }
         if (alert_level == 0) {
            ast_test_status_update(test, "Global alert failed to trigger at %ld\n", queue_count);
            res = -1;
         }
      }
   }

   ast_taskprocessor_unsuspend(tps);

   for (i = 1; i <= TEST_DATA_ARRAY_SIZE; i++) {
      ast_test_status_update(test, "Waiting on task %d\n", i);
      if (task_wait(task_data[i])) {
         ast_test_status_update(test, "Queued task '%d' did not execute!\n", i);
         res = -1;
         goto data_cleanup;
      }

      queue_count = ast_taskprocessor_size(tps);
      alert_level = ast_taskprocessor_alert_get();
      subsystem_alert_level = ast_taskprocessor_get_subsystem_alert("test_subsystem");

      if (queue_count == LOW_WATER_MARK) {
         if (!subsystem_alert_level) {
            ast_test_status_update(test, "Subsystem alert cleared correctly at %ld\n", queue_count);
         }
         if (!alert_level) {
            ast_test_status_update(test, "Global alert cleared correctly at %ld\n", queue_count);
         }
      } else if (queue_count > LOW_WATER_MARK) {
         if (subsystem_alert_level == 0) {
            ast_test_status_update(test, "Subsystem alert cleared unexpectedly at %ld\n", queue_count);
            res = -1;
         }
         if (alert_level == 0) {
            ast_test_status_update(test, "Global alert cleared unexpectedly at %ld\n", queue_count);
            res = -1;
         }
      } else {
         if (subsystem_alert_level > 0) {
            ast_test_status_update(test, "Subsystem alert failed to clear at %ld\n", queue_count);
            res = -1;
         }
         if (alert_level > 0) {
            ast_test_status_update(test, "Global alert failed to clear at %ld\n", queue_count);
            res = -1;
         }
      }

   }

data_cleanup:
   for (i = 1; i <= TEST_DATA_ARRAY_SIZE; i++) {
      ao2_cleanup(task_data[i]);
   }

   return res ? AST_TEST_FAIL : AST_TEST_PASS;
}

AST_TEST_DEFINE() [3/7]

AST_TEST_DEFINE

(

default_taskprocessor_load

)

Load test for taskprocessor with default listener.

This test queues a large number of tasks, each with random data associated. The test ensures that all of the tasks are run and that the tasks are executed in the same order that they were queued

Definition at line 352 of file test_taskprocessor.c.

References ast_cond_destroy, ast_cond_init, ast_cond_timedwait, ast_mutex_destroy, ast_mutex_init, ast_mutex_lock, ast_mutex_unlock, ast_random(), ast_taskprocessor_get(), ast_taskprocessor_push(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_tvnow(), load_task_data::cond, sip_to_pjsip::info(), load_task(), load_task_results, load_task_data::lock, NULL, NUM_TASKS, load_task_data::task_rand, load_task_data::tasks_completed, TEST_EXECUTE, TEST_INIT, and TPS_REF_DEFAULT.

{
   struct ast_taskprocessor *tps;
   struct timeval start;
   struct timespec ts;
   enum ast_test_result_state res = AST_TEST_PASS;
   int timedwait_res;
   int i;
   int rand_data[NUM_TASKS];

   switch (cmd) {
   case TEST_INIT:
      info->name = "default_taskprocessor_load";
      info->category = "/main/taskprocessor/";
      info->summary = "Load test of default taskproccesor";
      info->description =
         "Ensure that a large number of queued tasks are executed in the proper order.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }

   tps = ast_taskprocessor_get("test", TPS_REF_DEFAULT);

   if (!tps) {
      ast_test_status_update(test, "Unable to create test taskprocessor\n");
      return AST_TEST_FAIL;
   }

   start = ast_tvnow();

   ts.tv_sec = start.tv_sec + 60;
   ts.tv_nsec = start.tv_usec * 1000;

   ast_cond_init(&load_task_results.cond, NULL);
   ast_mutex_init(&load_task_results.lock);
   load_task_results.tasks_completed = 0;

   for (i = 0; i < NUM_TASKS; ++i) {
      rand_data[i] = ast_random();
      if (ast_taskprocessor_push(tps, load_task, &rand_data[i])) {
         ast_test_status_update(test, "Failed to queue task\n");
         res = AST_TEST_FAIL;
         goto test_end;
      }
   }

   ast_mutex_lock(&load_task_results.lock);
   while (load_task_results.tasks_completed < NUM_TASKS) {
      timedwait_res = ast_cond_timedwait(&load_task_results.cond, &load_task_results.lock, &ts);
      if (timedwait_res == ETIMEDOUT) {
         break;
      }
   }
   ast_mutex_unlock(&load_task_results.lock);

   if (load_task_results.tasks_completed != NUM_TASKS) {
      ast_test_status_update(test, "Unexpected number of tasks executed. Expected %d but got %d\n",
            NUM_TASKS, load_task_results.tasks_completed);
      res = AST_TEST_FAIL;
      goto test_end;
   }

   for (i = 0; i < NUM_TASKS; ++i) {
      if (rand_data[i] != load_task_results.task_rand[i]) {
         ast_test_status_update(test, "Queued tasks did not execute in order\n");
         res = AST_TEST_FAIL;
         goto test_end;
      }
   }

test_end:
   tps = ast_taskprocessor_unreference(tps);
   ast_mutex_destroy(&load_task_results.lock);
   ast_cond_destroy(&load_task_results.cond);
   return res;
}

AST_TEST_DEFINE() [4/7]

AST_TEST_DEFINE

(

taskprocessor_listener

)

Test for a taskprocessor with custom listener.

This test pushes tasks to a taskprocessor with a custom listener, executes the taskss, and destroys the taskprocessor.

The test ensures that the listener's callbacks are called when expected and that the data being passed in is accurate.

Definition at line 555 of file test_taskprocessor.c.

References ao2_cleanup, ast_free, ast_taskprocessor_create_with_listener(), ast_taskprocessor_execute(), ast_taskprocessor_listener_alloc(), ast_taskprocessor_push(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, check_stats(), sip_to_pjsip::info(), listener(), listener_test_task(), NULL, test_listener_pvt::shutdown, TEST_EXECUTE, TEST_INIT, and test_listener_pvt_alloc().

{
   struct ast_taskprocessor *tps = NULL;
   struct ast_taskprocessor_listener *listener = NULL;
   struct test_listener_pvt *pvt = NULL;
   enum ast_test_result_state res = AST_TEST_PASS;

   switch (cmd) {
   case TEST_INIT:
      info->name = "taskprocessor_listener";
      info->category = "/main/taskprocessor/";
      info->summary = "Test of taskproccesor listeners";
      info->description =
         "Ensures that listener callbacks are called when expected.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }

   pvt = test_listener_pvt_alloc();
   if (!pvt) {
      ast_test_status_update(test, "Unable to allocate test taskprocessor listener user data\n");
      return AST_TEST_FAIL;
   }

   listener = ast_taskprocessor_listener_alloc(&test_callbacks, pvt);
   if (!listener) {
      ast_test_status_update(test, "Unable to allocate test taskprocessor listener\n");
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   tps = ast_taskprocessor_create_with_listener("test_listener", listener);
   if (!tps) {
      ast_test_status_update(test, "Unable to allocate test taskprocessor\n");
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   if (ast_taskprocessor_push(tps, listener_test_task, NULL)) {
      ast_test_status_update(test, "Failed to queue task\n");
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   if (check_stats(test, pvt, 1, 0, 1) < 0) {
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   if (ast_taskprocessor_push(tps, listener_test_task, NULL)) {
      ast_test_status_update(test, "Failed to queue task\n");
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   if (check_stats(test, pvt, 2, 0, 1) < 0) {
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   ast_taskprocessor_execute(tps);

   if (check_stats(test, pvt, 2, 0, 1) < 0) {
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   ast_taskprocessor_execute(tps);

   if (check_stats(test, pvt, 2, 1, 1) < 0) {
      res = AST_TEST_FAIL;
      goto test_exit;
   }

   tps = ast_taskprocessor_unreference(tps);

   if (!pvt->shutdown) {
      res = AST_TEST_FAIL;
      goto test_exit;
   }

test_exit:
   ao2_cleanup(listener);
   /* This is safe even if tps is NULL */
   ast_taskprocessor_unreference(tps);
   ast_free(pvt);
   return res;
}

AST_TEST_DEFINE() [5/7]

AST_TEST_DEFINE

(

taskprocessor_shutdown

)

Definition at line 749 of file test_taskprocessor.c.

References ao2_cleanup, ast_pthread_create, ast_taskprocessor_get(), ast_taskprocessor_push(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, sip_to_pjsip::info(), NULL, RAII_VAR, shutdown_data_create(), shutdown_has_completed(), shutdown_poke(), shutdown_task_exec(), shutdown_waitfor_completion(), shutdown_waitfor_start(), TEST_EXECUTE, TEST_INIT, TPS_REF_DEFAULT, and tps_shutdown_thread().

{
   RAII_VAR(struct ast_taskprocessor *, tps, NULL, ast_taskprocessor_unreference);
   RAII_VAR(struct shutdown_data *, task1, NULL, ao2_cleanup);
   RAII_VAR(struct shutdown_data *, task2, NULL, ao2_cleanup);
   int push_res;
   int wait_res;
   int pthread_res;
   pthread_t shutdown_thread;

   switch (cmd) {
   case TEST_INIT:
      info->name = "taskprocessor_shutdown";
      info->category = "/main/taskprocessor/";
      info->summary = "Test of taskproccesor shutdown sequence";
      info->description =
         "Ensures that all tasks run to completion after the taskprocessor has been unref'ed.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }

   tps = ast_taskprocessor_get("test_shutdown", TPS_REF_DEFAULT);
   task1 = shutdown_data_create(0); /* task1 waits to be poked */
   task2 = shutdown_data_create(1); /* task2 waits for nothing */

   if (!tps || !task1 || !task2) {
      ast_test_status_update(test, "Allocation error\n");
      return AST_TEST_FAIL;
   }

   push_res = ast_taskprocessor_push(tps, shutdown_task_exec, task1);
   if (push_res != 0) {
      ast_test_status_update(test, "Could not push task1\n");
      return AST_TEST_FAIL;
   }

   push_res = ast_taskprocessor_push(tps, shutdown_task_exec, task2);
   if (push_res != 0) {
      ast_test_status_update(test, "Could not push task2\n");
      return AST_TEST_FAIL;
   }

   wait_res = shutdown_waitfor_start(task1);
   if (!wait_res) {
      ast_test_status_update(test, "Task1 didn't start\n");
      return AST_TEST_FAIL;
   }

   pthread_res = ast_pthread_create(&shutdown_thread, NULL, tps_shutdown_thread, tps);
   if (pthread_res != 0) {
      ast_test_status_update(test, "Failed to create shutdown thread\n");
      return AST_TEST_FAIL;
   }
   tps = NULL;

   /* Wakeup task1; it should complete */
   shutdown_poke(task1);
   wait_res = shutdown_waitfor_completion(task1);
   if (!wait_res) {
      ast_test_status_update(test, "Task1 didn't complete\n");
      return AST_TEST_FAIL;
   }

   /* Wait for shutdown to complete */
   pthread_join(shutdown_thread, NULL);

   /* Should have also completed task2 */
   wait_res = shutdown_has_completed(task2);
   if (!wait_res) {
      ast_test_status_update(test, "Task2 didn't finish\n");
      return AST_TEST_FAIL;
   }

   return AST_TEST_PASS;
}

AST_TEST_DEFINE() [6/7]

AST_TEST_DEFINE

(

taskprocessor_push_local

)

Definition at line 837 of file test_taskprocessor.c.

References ao2_cleanup, ast_taskprocessor_get(), ast_taskprocessor_push_local(), ast_taskprocessor_set_local(), ast_taskprocessor_unreference(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, sip_to_pjsip::info(), local_task_exe(), NULL, RAII_VAR, task_data_create(), task_wait(), TEST_EXECUTE, TEST_INIT, and TPS_REF_DEFAULT.

{
   RAII_VAR(struct ast_taskprocessor *, tps, NULL,
      ast_taskprocessor_unreference);
   RAII_VAR(struct task_data *, task_data, NULL, ao2_cleanup);
   int local_data;
   int res;

   switch (cmd) {
   case TEST_INIT:
      info->name = __func__;
      info->category = "/main/taskprocessor/";
      info->summary = "Test of pushing local data";
      info->description =
         "Ensures that local data is passed along.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }


   tps = ast_taskprocessor_get("test", TPS_REF_DEFAULT);
   if (!tps) {
      ast_test_status_update(test, "Unable to create test taskprocessor\n");
      return AST_TEST_FAIL;
   }


   task_data = task_data_create();
   if (!task_data) {
      ast_test_status_update(test, "Unable to create task_data\n");
      return AST_TEST_FAIL;
   }

   local_data = 0;
   ast_taskprocessor_set_local(tps, &local_data);

   if (ast_taskprocessor_push_local(tps, local_task_exe, task_data)) {
      ast_test_status_update(test, "Failed to queue task\n");
      return AST_TEST_FAIL;
   }

   res = task_wait(task_data);
   if (res != 0) {
      ast_test_status_update(test, "Queued task did not execute!\n");
      return AST_TEST_FAIL;
   }

   if (local_data != 1) {
      ast_test_status_update(test,
         "Queued task did not set local_data!\n");
      return AST_TEST_FAIL;
   }

   return AST_TEST_PASS;
}

AST_TEST_DEFINE() [7/7]

AST_TEST_DEFINE

(

serializer_pool

)

Baseline test for a serializer pool.

This test ensures that when a task is added to a taskprocessor that has been allocated with a default listener that the task gets executed as expected

Definition at line 901 of file test_taskprocessor.c.

References ao2_cleanup, ast_serializer_pool_create(), ast_serializer_pool_destroy(), ast_serializer_pool_get(), ast_serializer_pool_name(), ast_serializer_pool_set_alerts(), ast_taskprocessor_push(), AST_TEST_FAIL, AST_TEST_NOT_RUN, AST_TEST_PASS, ast_test_status_update, ast_threadpool_create(), AST_THREADPOOL_OPTIONS_VERSION, ast_threadpool_shutdown(), sip_to_pjsip::info(), NULL, RAII_VAR, task(), task_data_create(), task_wait(), TEST_EXECUTE, TEST_INIT, threadpool, ast_threadpool_options::version, and task_data::wait_time.

{
   RAII_VAR(struct ast_threadpool *, threadpool, NULL, ast_threadpool_shutdown);
   RAII_VAR(struct ast_serializer_pool *, serializer_pool, NULL, ast_serializer_pool_destroy);
   RAII_VAR(struct task_data *, task_data, NULL, ao2_cleanup);
   struct ast_threadpool_options options = {
      .version = AST_THREADPOOL_OPTIONS_VERSION,
      .idle_timeout = 0,
      .auto_increment = 0,
      .initial_size = 1,
      .max_size = 0,
   };
   /* struct ast_taskprocessor *tps; */

   switch (cmd) {
   case TEST_INIT:
      info->name = "serializer_pool";
      info->category = "/main/taskprocessor/";
      info->summary = "Test using a serializer pool";
      info->description =
         "Ensures that a queued task gets executed.";
      return AST_TEST_NOT_RUN;
   case TEST_EXECUTE:
      break;
   }

   ast_test_validate(test, threadpool = ast_threadpool_create("test", NULL, &options));
   ast_test_validate(test, serializer_pool = ast_serializer_pool_create(
                    "test/test", 5, threadpool, 2)); /* 2 second shutdown group time out */
   ast_test_validate(test, !strcmp(ast_serializer_pool_name(serializer_pool), "test/test"));
   ast_test_validate(test, !ast_serializer_pool_set_alerts(serializer_pool, 5, 0));
   ast_test_validate(test, task_data = task_data_create());

   task_data->wait_time = 4000; /* task takes 4 seconds */
   ast_test_validate(test, !ast_taskprocessor_push(
                    ast_serializer_pool_get(serializer_pool), task, task_data));

   if (!ast_serializer_pool_destroy(serializer_pool)) {
      ast_test_status_update(test, "Unexpected pool destruction!\n");
      /*
       * The pool should have timed out, so if it destruction reports success
       * we need to fail.
       */
      serializer_pool = NULL;
      return AST_TEST_FAIL;
   }

   ast_test_validate(test, !task_wait(task_data));

   /* The first attempt should have failed. Second try should destroy successfully */
   if (ast_serializer_pool_destroy(serializer_pool)) {
      ast_test_status_update(test, "Unable to destroy serializer pool in allotted time!\n");
      /*
       * If this fails we'll try again on return to hopefully avoid a memory leak.
       * If it again times out a third time, well not much we can do.
       */
      return AST_TEST_FAIL;
   }

   /* Test passed, so set pool to NULL to avoid "re-running" destroy */
   serializer_pool = NULL;

   return AST_TEST_PASS;
}

check_stats()

static int check_stats ( struct ast_test *  test, const struct test_listener_pvt *  pvt, int  num_pushed, int  num_emptied, int  num_was_empty  )

static

helper to ensure that statistics the listener is keeping are what we expect

Parameters

test	The currently-running test
pvt	The private data for the taskprocessor listener
num_pushed	The expected current number of tasks pushed to the processor
num_emptied	The expected current number of times the taskprocessor has become empty
num_was_empty	The expected current number of times that tasks were pushed to an empty taskprocessor

Return values

-1	Stats were not as expected
0	Stats were as expected

Definition at line 523 of file test_taskprocessor.c.

References ast_test_status_update, test_listener_pvt::num_emptied, test_listener_pvt::num_pushed, and test_listener_pvt::num_was_empty.

Referenced by AST_TEST_DEFINE().

{
   if (pvt->num_pushed != num_pushed) {
      ast_test_status_update(test, "Unexpected number of tasks pushed. Expected %d but got %d\n",
            num_pushed, pvt->num_pushed);
      return -1;
   }

   if (pvt->num_emptied != num_emptied) {
      ast_test_status_update(test, "Unexpected number of empties. Expected %d but got %d\n",
            num_emptied, pvt->num_emptied);
      return -1;
   }

   if (pvt->num_was_empty != num_was_empty) {
      ast_test_status_update(test, "Unexpected number of empties. Expected %d but got %d\n",
            num_was_empty, pvt->num_emptied);
      return -1;
   }

   return 0;
}

listener_test_task()

static int listener_test_task ( void *  ignore )

static

Queued task for taskprocessor listener test.

Does nothing.

Definition at line 507 of file test_taskprocessor.c.

Referenced by AST_TEST_DEFINE().

{
   return 0;
}

load_module()

static int load_module ( void  )

static

Definition at line 978 of file test_taskprocessor.c.

References AST_MODULE_LOAD_SUCCESS, and ast_test_register().

{
   ast_test_register(default_taskprocessor);
   ast_test_register(default_taskprocessor_load);
   ast_test_register(subsystem_alert);
   ast_test_register(taskprocessor_listener);
   ast_test_register(taskprocessor_shutdown);
   ast_test_register(taskprocessor_push_local);
   ast_test_register(serializer_pool);
   return AST_MODULE_LOAD_SUCCESS;
}

load_task()

static int load_task ( void *  data )

static

a queued task to be used in the taskprocessor load test

The task increments the number of tasks executed and puts the passed-in data into the next slot in the array of random data.

Definition at line 336 of file test_taskprocessor.c.

References ast_cond_signal, load_task_data::cond, load_task_results, task_data::lock, load_task_data::lock, SCOPED_MUTEX, load_task_data::task_rand, and load_task_data::tasks_completed.

Referenced by AST_TEST_DEFINE().

{
   int *randdata = data;
   SCOPED_MUTEX(lock, &load_task_results.lock);
   load_task_results.task_rand[load_task_results.tasks_completed++] = *randdata;
   ast_cond_signal(&load_task_results.cond);
   return 0;
}

local_task_exe()

static int local_task_exe ( struct ast_taskprocessor_local *  local )

static

Definition at line 826 of file test_taskprocessor.c.

References ast_taskprocessor_local::data, ast_taskprocessor::local_data, ast_taskprocessor_local::local_data, and task().

Referenced by AST_TEST_DEFINE().

{
   int *local_data = local->local_data;
   struct task_data *task_data = local->data;

   *local_data = 1;
   task(task_data);

   return 0;
}

shutdown_data_create()

static struct shutdown_data* shutdown_data_create ( int  dont_wait )

static

Definition at line 662 of file test_taskprocessor.c.

References ao2_alloc, ao2_cleanup, ao2_ref, ast_cond_init, ast_mutex_init, shutdown_data::in, shutdown_data::lock, NULL, shutdown_data::out, RAII_VAR, shutdown_data_dtor(), and shutdown_data::task_stop_waiting.

Referenced by AST_TEST_DEFINE().

{
   RAII_VAR(struct shutdown_data *, shutdown_data, NULL, ao2_cleanup);

   shutdown_data = ao2_alloc(sizeof(*shutdown_data), shutdown_data_dtor);
   if (!shutdown_data) {
      return NULL;
   }

   ast_mutex_init(&shutdown_data->lock);
   ast_cond_init(&shutdown_data->in, NULL);
   ast_cond_init(&shutdown_data->out, NULL);
   shutdown_data->task_stop_waiting = dont_wait;
   ao2_ref(shutdown_data, +1);
   return shutdown_data;
}

shutdown_data_dtor()

static void shutdown_data_dtor ( void *  data )

static

Definition at line 654 of file test_taskprocessor.c.

References ast_cond_destroy, ast_mutex_destroy, shutdown_data::in, shutdown_data::lock, and shutdown_data::out.

Referenced by shutdown_data_create().

{
   struct shutdown_data *shutdown_data = data;
   ast_mutex_destroy(&shutdown_data->lock);
   ast_cond_destroy(&shutdown_data->in);
   ast_cond_destroy(&shutdown_data->out);
}

shutdown_has_completed()

static int shutdown_has_completed ( struct shutdown_data *  shutdown_data )

static

Definition at line 711 of file test_taskprocessor.c.

References task_data::lock, shutdown_data::lock, SCOPED_MUTEX, and shutdown_data::task_complete.

Referenced by AST_TEST_DEFINE().

{
   SCOPED_MUTEX(lock, &shutdown_data->lock);
   return shutdown_data->task_complete;
}

shutdown_poke()

static void shutdown_poke ( struct shutdown_data *  shutdown_data )

static

Definition at line 735 of file test_taskprocessor.c.

References ast_cond_signal, shutdown_data::in, task_data::lock, shutdown_data::lock, SCOPED_MUTEX, and shutdown_data::task_stop_waiting.

Referenced by AST_TEST_DEFINE().

{
   SCOPED_MUTEX(lock, &shutdown_data->lock);
   shutdown_data->task_stop_waiting = 1;
   ast_cond_signal(&shutdown_data->in);
}

shutdown_task_exec()

static int shutdown_task_exec ( void *  data )

static

Definition at line 679 of file test_taskprocessor.c.

References ast_cond_signal, ast_cond_wait, shutdown_data::in, task_data::lock, shutdown_data::lock, shutdown_data::out, SCOPED_MUTEX, shutdown_data::task_complete, shutdown_data::task_started, and shutdown_data::task_stop_waiting.

Referenced by AST_TEST_DEFINE().

{
   struct shutdown_data *shutdown_data = data;
   SCOPED_MUTEX(lock, &shutdown_data->lock);
   shutdown_data->task_started = 1;
   ast_cond_signal(&shutdown_data->out);
   while (!shutdown_data->task_stop_waiting) {
      ast_cond_wait(&shutdown_data->in, &shutdown_data->lock);
   }
   shutdown_data->task_complete = 1;
   ast_cond_signal(&shutdown_data->out);
   return 0;
}

shutdown_waitfor_completion()

static int shutdown_waitfor_completion ( struct shutdown_data *  shutdown_data )

static

Definition at line 693 of file test_taskprocessor.c.

References ast_cond_timedwait, ast_tvnow(), task_data::lock, shutdown_data::lock, shutdown_data::out, SCOPED_MUTEX, and shutdown_data::task_complete.

Referenced by AST_TEST_DEFINE().

{
   struct timeval start = ast_tvnow();
   struct timespec end = {
      .tv_sec = start.tv_sec + 5,
      .tv_nsec = start.tv_usec * 1000
   };
   SCOPED_MUTEX(lock, &shutdown_data->lock);

   while (!shutdown_data->task_complete) {
      if (ast_cond_timedwait(&shutdown_data->out, &shutdown_data->lock, &end) == ETIMEDOUT) {
         break;
      }
   }

   return shutdown_data->task_complete;
}

shutdown_waitfor_start()

static int shutdown_waitfor_start ( struct shutdown_data *  shutdown_data )

static

Definition at line 717 of file test_taskprocessor.c.

References ast_cond_timedwait, ast_tvnow(), task_data::lock, shutdown_data::lock, shutdown_data::out, SCOPED_MUTEX, and shutdown_data::task_started.

Referenced by AST_TEST_DEFINE().

{
   struct timeval start = ast_tvnow();
   struct timespec end = {
      .tv_sec = start.tv_sec + 5,
      .tv_nsec = start.tv_usec * 1000
   };
   SCOPED_MUTEX(lock, &shutdown_data->lock);

   while (!shutdown_data->task_started) {
      if (ast_cond_timedwait(&shutdown_data->out, &shutdown_data->lock, &end) == ETIMEDOUT) {
         break;
      }
   }

   return shutdown_data->task_started;
}

task()

static int task ( void *  data )

static

Queued task for baseline test.

The task simply sets a boolean to indicate the task has been run and then signals a condition saying it's complete

Definition at line 88 of file test_taskprocessor.c.

References ast_cond_signal, task_data::cond, task_data::lock, SCOPED_MUTEX, task_data::task_complete, and task_data::wait_time.

Referenced by AST_TEST_DEFINE(), ast_threadpool_push(), local_task_exe(), tps_taskprocessor_dtor(), and tps_taskprocessor_pop().

{
   struct task_data *task_data = data;

   SCOPED_MUTEX(lock, &task_data->lock);
   if (task_data->wait_time > 0) {
      usleep(task_data->wait_time * 1000);
   }
   task_data->task_complete = 1;
   ast_cond_signal(&task_data->cond);
   return 0;
}

task_data_create()

static struct task_data* task_data_create ( void  )

static

Create a task_data object.

Definition at line 64 of file test_taskprocessor.c.

References ao2_alloc, ast_cond_init, ast_mutex_init, task_data::cond, task_data::lock, NULL, task_data::task_complete, task_data_dtor(), and task_data::wait_time.

Referenced by AST_TEST_DEFINE().

{
   struct task_data *task_data =
      ao2_alloc(sizeof(*task_data), task_data_dtor);

   if (!task_data) {
      return NULL;
   }

   ast_cond_init(&task_data->cond, NULL);
   ast_mutex_init(&task_data->lock);
   task_data->task_complete = 0;
   task_data->wait_time = 0;

   return task_data;
}

task_data_dtor()

static void task_data_dtor ( void *  obj )

static

Definition at line 55 of file test_taskprocessor.c.

References ast_cond_destroy, ast_mutex_destroy, task_data::cond, and task_data::lock.

Referenced by task_data_create().

{
   struct task_data *task_data = obj;

   ast_mutex_destroy(&task_data->lock);
   ast_cond_destroy(&task_data->cond);
}

task_wait()

static int task_wait ( struct task_data *  task_data )

static

Wait for a task to execute.

Definition at line 104 of file test_taskprocessor.c.

References ast_cond_timedwait, ast_tvnow(), task_data::cond, task_data::lock, SCOPED_MUTEX, and task_data::task_complete.

Referenced by AST_TEST_DEFINE().

{
   struct timeval start = ast_tvnow();
   struct timespec end;
   SCOPED_MUTEX(lock, &task_data->lock);

   end.tv_sec = start.tv_sec + 30;
   end.tv_nsec = start.tv_usec * 1000;

   while (!task_data->task_complete) {
      int res;
      res = ast_cond_timedwait(&task_data->cond, &task_data->lock,
         &end);
      if (res == ETIMEDOUT) {
         return -1;
      }
   }

   return 0;
}

test_emptied()

static void test_emptied ( struct ast_taskprocessor_listener *  listener )

static

test taskprocessor listener's emptied callback.

Definition at line 480 of file test_taskprocessor.c.

References ast_taskprocessor_listener_get_user_data(), and test_listener_pvt::num_emptied.

{
   struct test_listener_pvt *pvt = ast_taskprocessor_listener_get_user_data(listener);
   ++pvt->num_emptied;
}

test_listener_pvt_alloc()

static void* test_listener_pvt_alloc ( void  )

static

test taskprocessor listener's alloc callback

Definition at line 447 of file test_taskprocessor.c.

References ast_calloc.

Referenced by AST_TEST_DEFINE().

{
   struct test_listener_pvt *pvt;

   pvt = ast_calloc(1, sizeof(*pvt));
   return pvt;
}

test_shutdown()

static void test_shutdown ( struct ast_taskprocessor_listener *  listener )

static

test taskprocessor listener's shutdown callback.

Definition at line 489 of file test_taskprocessor.c.

References ast_taskprocessor_listener_get_user_data(), and test_listener_pvt::shutdown.

{
   struct test_listener_pvt *pvt = ast_taskprocessor_listener_get_user_data(listener);
   pvt->shutdown = 1;
}

test_start()

static int test_start ( struct ast_taskprocessor_listener *  listener )

static

test taskprocessor listener's start callback

Definition at line 458 of file test_taskprocessor.c.

{
   return 0;
}

test_task_pushed()

static void test_task_pushed ( struct ast_taskprocessor_listener *  listener, int  was_empty  )

static

test taskprocessor listener's task_pushed callback

Adjusts private data's stats as indicated by the parameters.

Definition at line 468 of file test_taskprocessor.c.

References ast_taskprocessor_listener_get_user_data(), test_listener_pvt::num_pushed, and test_listener_pvt::num_was_empty.

{
   struct test_listener_pvt *pvt = ast_taskprocessor_listener_get_user_data(listener);
   ++pvt->num_pushed;
   if (was_empty) {
      ++pvt->num_was_empty;
   }
}

tps_shutdown_thread()

static void* tps_shutdown_thread ( void *  data )

static

Definition at line 742 of file test_taskprocessor.c.

References ast_taskprocessor_unreference(), and NULL.

Referenced by AST_TEST_DEFINE().

{
   struct ast_taskprocessor *tps = data;
   ast_taskprocessor_unreference(tps);
   return NULL;
}

unload_module()

static int unload_module ( void  )

static

Definition at line 966 of file test_taskprocessor.c.

References ast_test_unregister().

{
   ast_test_unregister(default_taskprocessor);
   ast_test_unregister(default_taskprocessor_load);
   ast_test_unregister(subsystem_alert);
   ast_test_unregister(taskprocessor_listener);
   ast_test_unregister(taskprocessor_shutdown);
   ast_test_unregister(taskprocessor_push_local);
   ast_test_unregister(serializer_pool);
   return 0;
}

Variable Documentation

__mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "taskprocessor 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 990 of file test_taskprocessor.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 990 of file test_taskprocessor.c.

load_task_results

struct load_task_data load_task_results

static

Referenced by AST_TEST_DEFINE(), and load_task().

test_callbacks

const struct ast_taskprocessor_listener_callbacks test_callbacks

static

Definition at line 495 of file test_taskprocessor.c.