Merge "res_pjsip: Add serialized scheduler (res_pjsip/pjsip_scheduler.c)"

This commit is contained in:
zuul
2016-04-27 11:14:11 -05:00
committed by Gerrit Code Review
6 changed files with 1291 additions and 9 deletions

View File

@@ -20,6 +20,7 @@
#include "asterisk/compat.h"
#include "asterisk/lock.h"
#include "asterisk/linkedlists.h"
#include "asterisk/inline_api.h"
/*! \file
* \ref AstObj2
@@ -725,6 +726,46 @@ int __ao2_trylock(void *a, enum ao2_lock_req lock_how, const char *file, const c
void *ao2_object_get_lockaddr(void *obj);
/*!
* \brief Increment reference count on an object and lock it
* \since 13.9.0
*
* \param[in] obj A pointer to the ao2 object
* \retval 0 The object is not an ao2 object or wasn't locked successfully
* \retval 1 The object's reference count was incremented and was locked
*/
AST_INLINE_API(
int ao2_ref_and_lock(void *obj),
{
ao2_ref(obj, +1);
if (ao2_lock(obj)) {
ao2_ref(obj, -1);
return 0;
}
return 1;
}
)
/*!
* \brief Unlock an object and decrement its reference count
* \since 13.9.0
*
* \param[in] obj A pointer to the ao2 object
* \retval 0 The object is not an ao2 object or wasn't unlocked successfully
* \retval 1 The object was unlocked and it's reference count was decremented
*/
AST_INLINE_API(
int ao2_unlock_and_unref(void *obj),
{
if (ao2_unlock(obj)) {
return 0;
}
ao2_ref(obj, -1);
return 1;
}
)
/*! Global ao2 object holder structure. */
struct ao2_global_obj {
/*! Access lock to the held ao2 object. */
@@ -1903,4 +1944,97 @@ void ao2_iterator_cleanup(struct ao2_iterator *iter);
*/
int ao2_iterator_count(struct ao2_iterator *iter);
/*!
* \brief Creates a hash function for a structure string field.
* \param stype The structure type
* \param field The string field in the structure to hash
*
* AO2_STRING_FIELD_HASH_CB(mystruct, myfield) will produce a function
* named mystruct_hash_fn which hashes mystruct->myfield.
*/
#define AO2_STRING_FIELD_HASH_FN(stype, field) \
static int stype ## _hash_fn(const void *obj, const int flags) \
{ \
const struct stype *object = obj; \
const char *key; \
switch (flags & OBJ_SEARCH_MASK) { \
case OBJ_SEARCH_KEY: \
key = obj; \
break; \
case OBJ_SEARCH_OBJECT: \
key = object->field; \
break; \
default: \
ast_assert(0); \
return 0; \
} \
return ast_str_hash(key); \
}
/*!
* \brief Creates a compare function for a structure string field.
* \param stype The structure type
* \param field The string field in the structure to compare
*
* AO2_STRING_FIELD_CMP_FN(mystruct, myfield) will produce a function
* named mystruct_cmp_fn which compares mystruct->myfield.
*/
#define AO2_STRING_FIELD_CMP_FN(stype, field) \
static int stype ## _cmp_fn(void *obj, void *arg, int flags) \
{ \
const struct stype *object_left = obj, *object_right = arg; \
const char *right_key = arg; \
int cmp; \
switch (flags & OBJ_SEARCH_MASK) { \
case OBJ_SEARCH_OBJECT: \
right_key = object_right->field; \
case OBJ_SEARCH_KEY: \
cmp = strcmp(object_left->field, right_key); \
break; \
case OBJ_SEARCH_PARTIAL_KEY: \
cmp = strncmp(object_left->field, right_key, strlen(right_key)); \
break; \
default: \
cmp = 0; \
break; \
} \
if (cmp) { \
return 0; \
} \
return CMP_MATCH; \
}
/*!
* \brief Creates a sort function for a structure string field.
* \param stype The structure type
* \param field The string field in the structure to compare
*
* AO2_STRING_FIELD_SORT_FN(mystruct, myfield) will produce a function
* named mystruct_sort_fn which compares mystruct->myfield.
*/
#define AO2_STRING_FIELD_SORT_FN(stype, field) \
static int stype ## _sort_fn(const void *obj, const void *arg, int flags) \
{ \
const struct stype *object_left = obj; \
const struct stype *object_right = arg; \
const char *right_key = arg; \
int cmp; \
\
switch (flags & OBJ_SEARCH_MASK) { \
case OBJ_SEARCH_OBJECT: \
right_key = object_right->field; \
/* Fall through */ \
case OBJ_SEARCH_KEY: \
cmp = strcmp(object_left->field, right_key); \
break; \
case OBJ_SEARCH_PARTIAL_KEY: \
cmp = strncmp(object_left->field, right_key, strlen(right_key)); \
break; \
default: \
cmp = 0; \
break; \
} \
return cmp; \
}
#endif /* _ASTERISK_ASTOBJ2_H */

View File

@@ -19,6 +19,13 @@
#ifndef _RES_PJSIP_H
#define _RES_PJSIP_H
#include <pjsip.h>
/* Needed for SUBSCRIBE, NOTIFY, and PUBLISH method definitions */
#include <pjsip_simple.h>
#include <pjsip/sip_transaction.h>
#include <pj/timer.h>
#include <pjlib.h>
#include "asterisk/stringfields.h"
/* Needed for struct ast_sockaddr */
#include "asterisk/netsock2.h"
@@ -1166,8 +1173,9 @@ struct ast_sip_auth *ast_sip_get_artificial_auth(void);
*/
struct ast_sip_endpoint *ast_sip_get_artificial_endpoint(void);
/*!
* \page Threading model for SIP
/*! \defgroup pjsip_threading PJSIP Threading Model
* @{
* \page PJSIP PJSIP Threading Model
*
* There are three major types of threads that SIP will have to deal with:
* \li Asterisk threads
@@ -1216,6 +1224,19 @@ struct ast_sip_endpoint *ast_sip_get_artificial_endpoint(void);
* previous tasks pushed with the same serializer have completed. For more information
* on serializers and the benefits they provide, see \ref ast_threadpool_serializer
*
* \par Scheduler
*
* Some situations require that a task run periodically or at a future time. Normally
* the ast_sched functionality would be used but ast_sched only uses 1 thread for all
* tasks and that thread isn't registered with PJLIB and therefore can't do any PJSIP
* related work.
*
* ast_sip_sched uses ast_sched only as a scheduled queue. When a task is ready to run,
* it's pushed to a Serializer to be invoked asynchronously by a Servant. This ensures
* that the task is executed in a PJLIB registered thread and allows the ast_sched thread
* to immediately continue processing the queue. The Serializer used by ast_sip_sched
* is one of your choosing or a random one from the res_pjsip pool if you don't choose one.
*
* \note
*
* Do not make assumptions about individual threads based on a corresponding serializer.
@@ -1224,6 +1245,8 @@ struct ast_sip_endpoint *ast_sip_get_artificial_endpoint(void);
* tasks, even though they are all guaranteed to be executed in sequence.
*/
typedef int (*ast_sip_task)(void *user_data);
/*!
* \brief Create a new serializer for SIP tasks
* \since 13.8.0
@@ -1333,6 +1356,214 @@ int ast_sip_push_task_synchronous(struct ast_taskprocessor *serializer, int (*si
*/
int ast_sip_thread_is_servant(void);
/*!
* \brief Task flags for the res_pjsip scheduler
*
* The default is AST_SIP_SCHED_TASK_FIXED
* | AST_SIP_SCHED_TASK_DATA_NOT_AO2
* | AST_SIP_SCHED_TASK_DATA_NO_CLEANUP
* | AST_SIP_SCHED_TASK_PERIODIC
*/
enum ast_sip_scheduler_task_flags {
/*!
* The defaults
*/
AST_SIP_SCHED_TASK_DEFAULTS = (0 << 0),
/*!
* Run at a fixed interval.
* Stop scheduling if the callback returns 0.
* Any other value is ignored.
*/
AST_SIP_SCHED_TASK_FIXED = (0 << 0),
/*!
* Run at a variable interval.
* Stop scheduling if the callback returns 0.
* Any other return value is used as the new interval.
*/
AST_SIP_SCHED_TASK_VARIABLE = (1 << 0),
/*!
* The task data is not an AO2 object.
*/
AST_SIP_SCHED_TASK_DATA_NOT_AO2 = (0 << 1),
/*!
* The task data is an AO2 object.
* A reference count will be held by the scheduler until
* after the task has run for the final time (if ever).
*/
AST_SIP_SCHED_TASK_DATA_AO2 = (1 << 1),
/*!
* Don't take any cleanup action on the data
*/
AST_SIP_SCHED_TASK_DATA_NO_CLEANUP = (0 << 3),
/*!
* If AST_SIP_SCHED_TASK_DATA_AO2 is set, decrement the reference count
* otherwise call ast_free on it.
*/
AST_SIP_SCHED_TASK_DATA_FREE = ( 1 << 3 ),
/*! \brief AST_SIP_SCHED_TASK_PERIODIC
* The task is scheduled at multiples of interval
* \see Interval
*/
AST_SIP_SCHED_TASK_PERIODIC = (0 << 4),
/*! \brief AST_SIP_SCHED_TASK_DELAY
* The next invocation of the task is at last finish + interval
* \see Interval
*/
AST_SIP_SCHED_TASK_DELAY = (1 << 4),
};
/*!
* \brief Scheduler task data structure
*/
struct ast_sip_sched_task;
/*!
* \brief Schedule a task to run in the res_pjsip thread pool
* \since 13.9.0
*
* \param serializer The serializer to use. If NULL, don't use a serializer (see note below)
* \param interval The invocation interval in milliseconds (see note below)
* \param sip_task The task to invoke
* \param name An optional name to associate with the task
* \param task_data Optional data to pass to the task
* \param flags One of enum ast_sip_scheduler_task_type
*
* \returns Pointer to \ref ast_sip_sched_task ao2 object which must be dereferenced when done.
*
* \paragraph Serialization
*
* Specifying a serializer guarantees serialized execution but NOT specifying a serializer
* may still result in tasks being effectively serialized if the thread pool is busy.
* The point of the serializer BTW is not to prevent parallel executions of the SAME task.
* That happens automatically (see below). It's to prevent the task from running at the same
* time as other work using the same serializer, whether or not it's being run by the scheduler.
*
* \paragraph Interval
*
* The interval is used to calculate the next time the task should run. There are two models.
*
* \ref AST_SIP_SCHED_TASK_PERIODIC specifies that the invocations of the task occur at the
* specific interval. That is, every \ref "interval" milliseconds, regardless of how long the task
* takes. If the task takes longer than \ref interval, it will be scheduled at the next available
* multiple of \ref interval. For exmaple: If the task has an interval of 60 seconds and the task
* takes 70 seconds, the next invocation will happen at 120 seconds.
*
* \ref AST_SIP_SCHED_TASK_DELAY specifies that the next invocation of the task should start
* at \ref interval milliseconds after the current invocation has finished.
*
*/
struct ast_sip_sched_task *ast_sip_schedule_task(struct ast_taskprocessor *serializer,
int interval, ast_sip_task sip_task, char *name, void *task_data,
enum ast_sip_scheduler_task_flags flags);
/*!
* \brief Cancels the next invocation of a task
* \since 13.9.0
*
* \param schtd The task structure pointer
* \retval 0 Success
* \retval -1 Failure
* \note Only cancels future invocations not the currently running invocation.
*/
int ast_sip_sched_task_cancel(struct ast_sip_sched_task *schtd);
/*!
* \brief Cancels the next invocation of a task by name
* \since 13.9.0
*
* \param name The task name
* \retval 0 Success
* \retval -1 Failure
* \note Only cancels future invocations not the currently running invocation.
*/
int ast_sip_sched_task_cancel_by_name(const char *name);
/*!
* \brief Gets the last start and end times of the task
* \since 13.9.0
*
* \param schtd The task structure pointer
* \param[out] when_queued Pointer to a timeval structure to contain the time when queued
* \param[out] last_start Pointer to a timeval structure to contain the time when last started
* \param[out] last_end Pointer to a timeval structure to contain the time when last ended
* \retval 0 Success
* \retval -1 Failure
* \note Any of the pointers can be NULL if you don't need them.
*/
int ast_sip_sched_task_get_times(struct ast_sip_sched_task *schtd,
struct timeval *when_queued, struct timeval *last_start, struct timeval *last_end);
/*!
* \brief Gets the last start and end times of the task by name
* \since 13.9.0
*
* \param name The task name
* \param[out] when_queued Pointer to a timeval structure to contain the time when queued
* \param[out] last_start Pointer to a timeval structure to contain the time when last started
* \param[out] last_end Pointer to a timeval structure to contain the time when last ended
* \retval 0 Success
* \retval -1 Failure
* \note Any of the pointers can be NULL if you don't need them.
*/
int ast_sip_sched_task_get_times_by_name(const char *name,
struct timeval *when_queued, struct timeval *last_start, struct timeval *last_end);
/*!
* \brief Gets the number of milliseconds until the next invocation
* \since 13.9.0
*
* \param schtd The task structure pointer
* \return The number of milliseconds until the next invocation or -1 if the task isn't scheduled
*/
int ast_sip_sched_task_get_next_run(struct ast_sip_sched_task *schtd);
/*!
* \brief Gets the number of milliseconds until the next invocation
* \since 13.9.0
*
* \param name The task name
* \return The number of milliseconds until the next invocation or -1 if the task isn't scheduled
*/
int ast_sip_sched_task_get_next_run_by_name(const char *name);
/*!
* \brief Checks if the task is currently running
* \since 13.9.0
*
* \param schtd The task structure pointer
* \retval 0 not running
* \retval 1 running
*/
int ast_sip_sched_is_task_running(struct ast_sip_sched_task *schtd);
/*!
* \brief Checks if the task is currently running
* \since 13.9.0
*
* \param name The task name
* \retval 0 not running or not found
* \retval 1 running
*/
int ast_sip_sched_is_task_running_by_name(const char *name);
/*!
* \brief Gets the task name
* \since 13.9.0
*
* \param schtd The task structure pointer
* \retval 0 success
* \retval 1 failure
*/
int ast_sip_sched_task_get_name(struct ast_sip_sched_task *schtd, char *name, size_t maxlen);
/*!
* @}
*/
/*!
* \brief SIP body description
*

View File

@@ -3636,11 +3636,7 @@ int ast_sip_push_task(struct ast_taskprocessor *serializer, int (*sip_task)(void
serializer = serializer_pool[pos];
}
if (serializer) {
return ast_taskprocessor_push(serializer, sip_task, task_data);
} else {
return ast_threadpool_push(sip_threadpool, sip_task, task_data);
}
}
struct sync_task_data {
@@ -4158,6 +4154,11 @@ static int load_module(void)
goto error;
}
if (ast_sip_initialize_scheduler()) {
ast_log(LOG_ERROR, "Failed to start scheduler. Aborting load\n");
goto error;
}
/* Now load all the pjproject infrastructure. */
if (load_pjsip()) {
goto error;
@@ -4196,8 +4197,10 @@ static int load_module(void)
return AST_MODULE_LOAD_SUCCESS;
error:
/* These functions all check for NULLs and are safe to call at any time */
unload_pjsip(NULL);
/* These functions all check for NULLs and are safe to call at any time */
ast_sip_destroy_scheduler();
serializer_pool_shutdown();
ast_threadpool_shutdown(sip_threadpool);
@@ -4228,7 +4231,7 @@ static int unload_module(void)
* so we have to push the work to the threadpool to handle
*/
ast_sip_push_task_synchronous(NULL, unload_pjsip, NULL);
ast_sip_destroy_scheduler();
serializer_pool_shutdown();
ast_threadpool_shutdown(sip_threadpool);

View File

@@ -313,4 +313,23 @@ struct ast_sip_contact_status *ast_res_pjsip_find_or_create_contact_status(const
*/
int ast_sip_validate_uri_length(const char *uri);
/*!
* \brief Initialize scheduler
* \since 13.9.0
*
* \retval -1 failure
* \retval 0 success
*/
int ast_sip_initialize_scheduler(void);
/*!
* \internal
* \brief Destroy scheduler
* \since 13.9.0
*
* \retval -1 failure
* \retval 0 success
*/
int ast_sip_destroy_scheduler(void);
#endif /* RES_PJSIP_PRIVATE_H_ */

View File

@@ -0,0 +1,495 @@
/*
* Asterisk -- An open source telephony toolkit.
*
* Copyright (C) 2016, Fairview 5 Engineering, LLC
*
* George Joseph <george.joseph@fairview5.com>
*
* See http://www.asterisk.org for more information about
* the Asterisk project. Please do not directly contact
* any of the maintainers of this project for assistance;
* the project provides a web site, mailing lists and IRC
* channels for your use.
*
* This program is free software, distributed under the terms of
* the GNU General Public License Version 2. See the LICENSE file
* at the top of the source tree.
*/
/*! \file
*
* \brief res_pjsip Scheduler
*
* \author George Joseph <george.joseph@fairview5.com>
*/
#include "asterisk.h"
ASTERISK_REGISTER_FILE()
#include "asterisk/res_pjsip.h"
#include "include/res_pjsip_private.h"
#include "asterisk/res_pjsip_cli.h"
#define TASK_BUCKETS 53
static struct ast_sched_context *scheduler_context;
static struct ao2_container *tasks;
static int task_count;
struct ast_sip_sched_task {
/*! ast_sip_sched task id */
uint32_t task_id;
/*! ast_sched scheudler id */
int current_scheduler_id;
/*! task is currently running */
int is_running;
/*! task */
ast_sip_task task;
/*! task data */
void *task_data;
/*! reschedule interval in milliseconds */
int interval;
/*! the time the task was queued */
struct timeval when_queued;
/*! the last time the task was started */
struct timeval last_start;
/*! the last time the task was ended */
struct timeval last_end;
/*! times run */
int run_count;
/*! the task reschedule, cleanup and policy flags */
enum ast_sip_scheduler_task_flags flags;
/*! the serializer to be used (if any) */
struct ast_taskprocessor *serializer;
/* A name to be associated with the task */
char name[0];
};
AO2_STRING_FIELD_HASH_FN(ast_sip_sched_task, name);
AO2_STRING_FIELD_CMP_FN(ast_sip_sched_task, name);
AO2_STRING_FIELD_SORT_FN(ast_sip_sched_task, name);
static int push_to_serializer(const void *data);
/*
* This function is run in the context of the serializer.
* It runs the task with a simple call and reschedules based on the result.
*/
static int run_task(void *data)
{
RAII_VAR(struct ast_sip_sched_task *, schtd, ao2_bump(data), ao2_cleanup);
int res;
int delay;
ao2_lock(schtd);
schtd->last_start = ast_tvnow();
schtd->is_running = 1;
schtd->run_count++;
ao2_unlock(schtd);
res = schtd->task(schtd->task_data);
ao2_lock(schtd);
schtd->is_running = 0;
schtd->last_end = ast_tvnow();
/*
* Don't restart if the task returned 0 or if the interval
* was set to 0 while the task was running
*/
if (!res || !schtd->interval) {
schtd->interval = 0;
ao2_unlock(schtd);
ao2_unlink(tasks, schtd);
return -1;
}
if (schtd->flags & AST_SIP_SCHED_TASK_VARIABLE) {
schtd->interval = res;
}
if (schtd->flags & AST_SIP_SCHED_TASK_DELAY) {
delay = schtd->interval;
} else {
delay = schtd->interval - (ast_tvdiff_ms(schtd->last_end, schtd->last_start) % schtd->interval);
}
schtd->current_scheduler_id = ast_sched_add(scheduler_context, delay, push_to_serializer, (const void *)schtd);
if (schtd->current_scheduler_id < 0) {
schtd->interval = 0;
ao2_unlock(schtd);
ao2_unlink(tasks, schtd);
return -1;
}
ao2_unlock(schtd);
return 0;
}
/*
* This function is run by the scheduler thread. Its only job is to push the task
* to the serialize and return. It returns 0 so it's not rescheduled.
*/
static int push_to_serializer(const void *data)
{
struct ast_sip_sched_task *schtd = (struct ast_sip_sched_task *)data;
if (ast_sip_push_task(schtd->serializer, run_task, schtd)) {
ao2_ref(schtd, -1);
}
return 0;
}
int ast_sip_sched_task_cancel(struct ast_sip_sched_task *schtd)
{
int res;
if (!ao2_ref_and_lock(schtd)) {
return -1;
}
if (schtd->current_scheduler_id < 0 || schtd->interval <= 0) {
ao2_unlock_and_unref(schtd);
return 0;
}
schtd->interval = 0;
ao2_unlock_and_unref(schtd);
ao2_unlink(tasks, schtd);
res = ast_sched_del(scheduler_context, schtd->current_scheduler_id);
return res;
}
int ast_sip_sched_task_cancel_by_name(const char *name)
{
RAII_VAR(struct ast_sip_sched_task *, schtd, NULL, ao2_cleanup);
if (ast_strlen_zero(name)) {
return -1;
}
schtd = ao2_find(tasks, name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
if (!schtd) {
return -1;
}
return ast_sip_sched_task_cancel(schtd);
}
int ast_sip_sched_task_get_times(struct ast_sip_sched_task *schtd,
struct timeval *queued, struct timeval *last_start, struct timeval *last_end)
{
if (!ao2_ref_and_lock(schtd)) {
return -1;
}
if (queued) {
memcpy(queued, &schtd->when_queued, sizeof(struct timeval));
}
if (last_start) {
memcpy(last_start, &schtd->last_start, sizeof(struct timeval));
}
if (last_end) {
memcpy(last_end, &schtd->last_end, sizeof(struct timeval));
}
ao2_unlock_and_unref(schtd);
return 0;
}
int ast_sip_sched_task_get_times_by_name(const char *name,
struct timeval *queued, struct timeval *last_start, struct timeval *last_end)
{
RAII_VAR(struct ast_sip_sched_task *, schtd, NULL, ao2_cleanup);
if (ast_strlen_zero(name)) {
return -1;
}
schtd = ao2_find(tasks, name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
if (!schtd) {
return -1;
}
return ast_sip_sched_task_get_times(schtd, queued, last_start, last_end);
}
int ast_sip_sched_task_get_name(struct ast_sip_sched_task *schtd, char *name, size_t maxlen)
{
if (maxlen <= 0) {
return -1;
}
if (!ao2_ref_and_lock(schtd)) {
return -1;
}
ast_copy_string(name, schtd->name, maxlen);
ao2_unlock_and_unref(schtd);
return 0;
}
int ast_sip_sched_task_get_next_run(struct ast_sip_sched_task *schtd)
{
int delay;
struct timeval since_when;
struct timeval now;
if (!ao2_ref_and_lock(schtd)) {
return -1;
}
if (schtd->interval) {
delay = schtd->interval;
now = ast_tvnow();
if (schtd->flags & AST_SIP_SCHED_TASK_DELAY) {
since_when = schtd->is_running ? now : schtd->last_end;
} else {
since_when = schtd->last_start.tv_sec ? schtd->last_start : schtd->when_queued;
}
delay -= ast_tvdiff_ms(now, since_when);
delay = delay < 0 ? 0 : delay;
} else {
delay = -1;
}
ao2_unlock_and_unref(schtd);
return delay;
}
int ast_sip_sched_task_get_next_run_by_name(const char *name)
{
RAII_VAR(struct ast_sip_sched_task *, schtd, NULL, ao2_cleanup);
if (ast_strlen_zero(name)) {
return -1;
}
schtd = ao2_find(tasks, name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
if (!schtd) {
return -1;
}
return ast_sip_sched_task_get_next_run(schtd);
}
int ast_sip_sched_is_task_running(struct ast_sip_sched_task *schtd)
{
if (!schtd) {
return 0;
}
return schtd->is_running;
}
int ast_sip_sched_is_task_running_by_name(const char *name)
{
RAII_VAR(struct ast_sip_sched_task *, schtd, NULL, ao2_cleanup);
if (ast_strlen_zero(name)) {
return 0;
}
schtd = ao2_find(tasks, name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
if (!schtd) {
return 0;
}
return schtd->is_running;
}
static void schtd_destructor(void *data)
{
struct ast_sip_sched_task *schtd = data;
if (schtd->flags & AST_SIP_SCHED_TASK_DATA_AO2) {
/* release our own ref, then release the callers if asked to do so */
ao2_ref(schtd->task_data, (schtd->flags & AST_SIP_SCHED_TASK_DATA_FREE) ? -2 : -1);
} else if (schtd->task_data && (schtd->flags & AST_SIP_SCHED_TASK_DATA_FREE)) {
ast_free(schtd->task_data);
}
}
struct ast_sip_sched_task *ast_sip_schedule_task(struct ast_taskprocessor *serializer,
int interval, ast_sip_task sip_task, char *name, void *task_data, enum ast_sip_scheduler_task_flags flags)
{
#define ID_LEN 13 /* task_deadbeef */
struct ast_sip_sched_task *schtd;
int res;
if (interval < 0) {
return NULL;
}
schtd = ao2_alloc((sizeof(*schtd) + (!ast_strlen_zero(name) ? strlen(name) : ID_LEN) + 1), schtd_destructor);
if (!schtd) {
return NULL;
}
schtd->task_id = ast_atomic_fetchadd_int(&task_count, 1);
schtd->serializer = serializer;
schtd->task = sip_task;
if (!ast_strlen_zero(name)) {
strcpy(schtd->name, name); /* Safe */
} else {
sprintf(schtd->name, "task_%08x", schtd->task_id);
}
schtd->task_data = task_data;
schtd->flags = flags;
schtd->interval = interval;
schtd->when_queued = ast_tvnow();
if (flags & AST_SIP_SCHED_TASK_DATA_AO2) {
ao2_ref(task_data, +1);
}
res = ast_sched_add(scheduler_context, interval, push_to_serializer, (const void *)schtd);
if (res < 0) {
ao2_ref(schtd, -1);
return NULL;
} else {
schtd->current_scheduler_id = res;
ao2_link(tasks, schtd);
}
return schtd;
#undef ID_LEN
}
static char *cli_show_tasks(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
{
struct ao2_iterator i;
struct ast_sip_sched_task *schtd;
const char *log_format = ast_logger_get_dateformat();
struct ast_tm tm;
char queued[32];
char last_start[32];
char last_end[32];
int datelen;
struct timeval now = ast_tvnow();
const char *separator = "======================================";
switch (cmd) {
case CLI_INIT:
e->command = "pjsip show scheduled_tasks";
e->usage = "Usage: pjsip show scheduled_tasks\n"
" Show all scheduled tasks\n";
return NULL;
case CLI_GENERATE:
return NULL;
}
if (a->argc != 3) {
return CLI_SHOWUSAGE;
}
ast_localtime(&now, &tm, NULL);
datelen = ast_strftime(queued, sizeof(queued), log_format, &tm);
ast_cli(a->fd, "PJSIP Scheduled Tasks:\n\n");
ast_cli(a->fd, " %1$-24s %2$-8s %3$-9s %4$-7s %6$-*5$s %7$-*5$s %8$-*5$s\n",
"Task Name", "Interval", "Times Run", "State",
datelen, "Queued", "Last Started", "Last Ended");
ast_cli(a->fd, " %1$-24.24s %2$-8.8s %3$-9.9s %4$-7.7s %6$-*5$.*5$s %7$-*5$.*5$s %8$-*5$.*5$s\n",
separator, separator, separator, separator,
datelen, separator, separator, separator);
ao2_ref(tasks, +1);
ao2_rdlock(tasks);
i = ao2_iterator_init(tasks, 0);
while ((schtd = ao2_iterator_next(&i))) {
ast_localtime(&schtd->when_queued, &tm, NULL);
ast_strftime(queued, sizeof(queued), log_format, &tm);
if (ast_tvzero(schtd->last_start)) {
strcpy(last_start, "not yet started");
} else {
ast_localtime(&schtd->last_start, &tm, NULL);
ast_strftime(last_start, sizeof(last_start), log_format, &tm);
}
if (ast_tvzero(schtd->last_end)) {
if (ast_tvzero(schtd->last_start)) {
strcpy(last_end, "not yet started");
} else {
strcpy(last_end, "running");
}
} else {
ast_localtime(&schtd->last_end, &tm, NULL);
ast_strftime(last_end, sizeof(last_end), log_format, &tm);
}
ast_cli(a->fd, " %1$-24.24s %2$-8.3f %3$-9d %4$-7s %6$-*5$s %7$-*5$s %8$-*5$s\n",
schtd->name,
schtd->interval / 1000.0,
schtd->run_count,
schtd->is_running ? "running" : "waiting",
datelen, queued, last_start, last_end);
ao2_cleanup(schtd);
}
ao2_iterator_destroy(&i);
ao2_unlock(tasks);
ao2_ref(tasks, -1);
ast_cli(a->fd, "\n");
return CLI_SUCCESS;
}
static struct ast_cli_entry cli_commands[] = {
AST_CLI_DEFINE(cli_show_tasks, "Show all scheduled tasks"),
};
int ast_sip_initialize_scheduler(void)
{
if (!(scheduler_context = ast_sched_context_create())) {
ast_log(LOG_ERROR, "Failed to create scheduler. Aborting load\n");
return -1;
}
if (ast_sched_start_thread(scheduler_context)) {
ast_log(LOG_ERROR, "Failed to start scheduler. Aborting load\n");
ast_sched_context_destroy(scheduler_context);
return -1;
}
tasks = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, AO2_CONTAINER_ALLOC_OPT_DUPS_REJECT,
TASK_BUCKETS, ast_sip_sched_task_hash_fn, ast_sip_sched_task_sort_fn, ast_sip_sched_task_cmp_fn);
if (!tasks) {
ast_log(LOG_ERROR, "Failed to allocate task container. Aborting load\n");
ast_sched_context_destroy(scheduler_context);
return -1;
}
ast_cli_register_multiple(cli_commands, ARRAY_LEN(cli_commands));
return 0;
}
int ast_sip_destroy_scheduler(void)
{
ast_cli_unregister_multiple(cli_commands, ARRAY_LEN(cli_commands));
if (scheduler_context) {
ast_sched_context_destroy(scheduler_context);
}
ao2_cleanup(tasks);
tasks = NULL;
return 0;
}

View File

@@ -0,0 +1,400 @@
/*
* Asterisk -- An open source telephony toolkit.
*
* Copyright (C) 2016, Fairview 5 Engineering, LLC
*
* George Joseph <george.joseph@fairview5.com>
*
* See http://www.asterisk.org for more information about
* the Asterisk project. Please do not directly contact
* any of the maintainers of this project for assistance;
* the project provides a web site, mailing lists and IRC
* channels for your use.
*
* This program is free software, distributed under the terms of
* the GNU General Public License Version 2. See the LICENSE file
* at the top of the source tree.
*/
/*!
* \file
* \brief res_pjsip scheduler tests
*
* \author George Joseph <george.joseph@fairview5.com>
*
*/
/*** MODULEINFO
<depend>TEST_FRAMEWORK</depend>
<depend>res_pjsip</depend>
<support_level>core</support_level>
***/
#include "asterisk.h"
ASTERISK_REGISTER_FILE()
#include <pjsip.h>
#include "asterisk/test.h"
#include "asterisk/module.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/res_pjsip.h"
#include "asterisk/utils.h"
#define CATEGORY "/res/res_pjsip/scheduler/"
struct test_data {
ast_mutex_t lock;
ast_cond_t cond;
pthread_t tid;
struct timeval test_start;
struct timeval task_start;
struct timeval task_end;
int is_servant;
int interval;
int sleep;
int done;
struct ast_test *test;
};
#define S2U(x) (long int)(x * 1000 * 1000)
#define M2U(x) (long int)(x * 1000)
static int task_1(void *data)
{
struct test_data *test = data;
test->done = 0;
test->task_start = ast_tvnow();
test->tid = pthread_self();
test->is_servant = ast_sip_thread_is_servant();
usleep(M2U(test->sleep));
test->task_end = ast_tvnow();
ast_mutex_lock(&test->lock);
test->done = 1;
ast_mutex_unlock(&test->lock);
ast_cond_signal(&test->cond);
return test->interval;
}
static void data_cleanup(void *data)
{
struct test_data *test_data = data;
ast_mutex_destroy(&test_data->lock);
ast_cond_destroy(&test_data->cond);
}
#define waitfor(x) \
{ \
ast_mutex_lock(&(x)->lock); \
while (!(x)->done) { \
ast_cond_wait(&(x)->cond, &(x)->lock); \
} \
(x)->done = 0; \
ast_mutex_unlock(&(x)->lock); \
}
static int scheduler(struct ast_test *test, int serialized)
{
RAII_VAR(struct ast_taskprocessor *, tp1, NULL, ast_taskprocessor_unreference);
RAII_VAR(struct test_data *, test_data1, ao2_alloc(sizeof(*test_data1), data_cleanup), ao2_cleanup);
RAII_VAR(struct test_data *, test_data2, ao2_alloc(sizeof(*test_data2), data_cleanup), ao2_cleanup);
RAII_VAR(struct ast_sip_sched_task *, task1, NULL, ao2_cleanup);
RAII_VAR(struct ast_sip_sched_task *, task2, NULL, ao2_cleanup);
int duration;
int delay;
struct timeval task1_start;
ast_test_validate(test, test_data1 != NULL);
ast_test_validate(test, test_data2 != NULL);
test_data1->test = test;
test_data1->test_start = ast_tvnow();
test_data1->interval = 2000;
test_data1->sleep = 1000;
ast_mutex_init(&test_data1->lock);
ast_cond_init(&test_data1->cond, NULL);
test_data2->test = test;
test_data2->test_start = ast_tvnow();
test_data2->interval = 2000;
test_data2->sleep = 1000;
ast_mutex_init(&test_data2->lock);
ast_cond_init(&test_data2->cond, NULL);
if (serialized) {
ast_test_status_update(test, "This test will take about %3.1f seconds\n",
(test_data1->interval + test_data1->sleep + (MAX(test_data1->interval - test_data2->interval, 0)) + test_data2->sleep) / 1000.0);
tp1 = ast_sip_create_serializer("test-scheduler-serializer");
ast_test_validate(test, (tp1 != NULL));
} else {
ast_test_status_update(test, "This test will take about %3.1f seconds\n",
((MAX(test_data1->interval, test_data2->interval) + MAX(test_data1->sleep, test_data2->sleep)) / 1000.0));
}
task1 = ast_sip_schedule_task(tp1, test_data1->interval, task_1, NULL, test_data1, AST_SIP_SCHED_TASK_FIXED);
ast_test_validate(test, task1 != NULL);
task2 = ast_sip_schedule_task(tp1, test_data2->interval, task_1, NULL, test_data2, AST_SIP_SCHED_TASK_FIXED);
ast_test_validate(test, task2 != NULL);
waitfor(test_data1);
ast_sip_sched_task_cancel(task1);
ast_test_validate(test, test_data1->is_servant);
duration = ast_tvdiff_ms(test_data1->task_end, test_data1->test_start);
ast_test_validate(test, (duration > ((test_data1->interval + test_data1->sleep) * 0.9))
&& (duration < ((test_data1->interval + test_data1->sleep) * 1.1)));
ast_sip_sched_task_get_times(task1, NULL, &task1_start, NULL);
delay = ast_tvdiff_ms(task1_start, test_data1->test_start);
ast_test_validate(test, (delay > (test_data1->interval * 0.9)
&& (delay < (test_data1->interval * 1.1))));
waitfor(test_data2);
ast_sip_sched_task_cancel(task2);
ast_test_validate(test, test_data2->is_servant);
if (serialized) {
ast_test_validate(test, test_data1->tid == test_data2->tid);
ast_test_validate(test, ast_tvdiff_ms(test_data2->task_start, test_data1->task_end) >= 0);
} else {
ast_test_validate(test, test_data1->tid != test_data2->tid);
}
return AST_TEST_PASS;
}
AST_TEST_DEFINE(serialized_scheduler)
{
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = CATEGORY;
info->summary = "Test res_pjsip serialized scheduler";
info->description = "Test res_pjsip serialized scheduler";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
return scheduler(test, 1);
}
AST_TEST_DEFINE(unserialized_scheduler)
{
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = CATEGORY;
info->summary = "Test res_pjsip unserialized scheduler";
info->description = "Test res_pjsip unserialized scheduler";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
return scheduler(test, 0);
}
static int run_count;
static int destruct_count;
static int dummy_task(void *data)
{
int *sleep = data;
usleep(M2U(*sleep));
run_count++;
return 0;
}
static void test_destructor(void *data)
{
destruct_count++;
}
AST_TEST_DEFINE(scheduler_cleanup)
{
RAII_VAR(int *, sleep, NULL, ao2_cleanup);
RAII_VAR(struct ast_sip_sched_task *, task, NULL, ao2_cleanup);
int interval;
int when;
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = CATEGORY;
info->summary = "Test res_pjsip scheduler cleanup";
info->description = "Test res_pjsip scheduler cleanup";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
destruct_count = 0;
interval = 1000;
sleep = ao2_alloc(sizeof(*sleep), test_destructor);
ast_test_validate(test, sleep != NULL);
*sleep = 500;
ast_test_status_update(test, "This test will take about %3.1f seconds\n",
((interval * 1.1) + *sleep) / 1000.0);
task = ast_sip_schedule_task(NULL, interval, dummy_task, "dummy", sleep,
AST_SIP_SCHED_TASK_DATA_AO2 | AST_SIP_SCHED_TASK_DATA_FREE);
ast_test_validate(test, task != NULL);
usleep(M2U(interval * 0.5));
when = ast_sip_sched_task_get_next_run(task);
ast_test_validate(test, (when > (interval * 0.4) && when < (interval * 0.6)));
usleep(M2U(interval * 0.6));
ast_test_validate(test, ast_sip_sched_is_task_running(task));
usleep(M2U(*sleep));
ast_test_validate(test, (ast_sip_sched_is_task_running(task) == 0));
when = ast_sip_sched_task_get_next_run(task);
ast_test_validate(test, (when < 0), res, error);
ast_test_validate(test, (ao2_ref(task, 0) == 1));
ao2_ref(task, -1);
task = NULL;
ast_test_validate(test, (destruct_count == 1));
sleep = NULL;
return AST_TEST_PASS;
}
AST_TEST_DEFINE(scheduler_cancel)
{
RAII_VAR(int *, sleep, NULL, ao2_cleanup);
RAII_VAR(struct ast_sip_sched_task *, task, NULL, ao2_cleanup);
int interval;
int when;
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = CATEGORY;
info->summary = "Test res_pjsip scheduler cancel task";
info->description = "Test res_pjsip scheduler cancel task";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
destruct_count = 0;
interval = 1000;
sleep = ao2_alloc(sizeof(*sleep), test_destructor);
ast_test_validate(test, sleep != NULL);
*sleep = 500;
ast_test_status_update(test, "This test will take about %3.1f seconds\n",
(interval + *sleep) / 1000.0);
task = ast_sip_schedule_task(NULL, interval, dummy_task, "dummy", sleep, AST_SIP_SCHED_TASK_DATA_NO_CLEANUP);
ast_test_validate(test, task != NULL);
usleep(M2U(interval * 0.5));
when = ast_sip_sched_task_get_next_run_by_name("dummy");
ast_test_validate(test, (when > (interval * 0.4) && when < (interval * 0.6)));
ast_test_validate(test, !ast_sip_sched_is_task_running_by_name("dummy"));
ast_test_validate(test, ao2_ref(task, 0) == 2);
ast_sip_sched_task_cancel_by_name("dummy");
when = ast_sip_sched_task_get_next_run(task);
ast_test_validate(test, when < 0);
usleep(M2U(interval));
ast_test_validate(test, run_count == 0);
ast_test_validate(test, destruct_count == 0);
ast_test_validate(test, ao2_ref(task, 0) == 1);
return AST_TEST_PASS;
}
AST_TEST_DEFINE(scheduler_policy)
{
RAII_VAR(struct test_data *, test_data1, ao2_alloc(sizeof(*test_data1), data_cleanup), ao2_cleanup);
RAII_VAR(struct ast_sip_sched_task *, task, NULL, ao2_cleanup);
int when;
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = CATEGORY;
info->summary = "Test res_pjsip scheduler cancel task";
info->description = "Test res_pjsip scheduler cancel task";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
ast_test_validate(test, test_data1 != NULL);
destruct_count = 0;
run_count = 0;
test_data1->test = test;
test_data1->test_start = ast_tvnow();
test_data1->interval = 1000;
test_data1->sleep = 500;
ast_mutex_init(&test_data1->lock);
ast_cond_init(&test_data1->cond, NULL);
ast_test_status_update(test, "This test will take about %3.1f seconds\n",
((test_data1->interval * 3) + test_data1->sleep) / 1000.0);
task = ast_sip_schedule_task(NULL, test_data1->interval, task_1, "test_1", test_data1,
AST_SIP_SCHED_TASK_DATA_NO_CLEANUP | AST_SIP_SCHED_TASK_PERIODIC);
ast_test_validate(test, task != NULL);
waitfor(test_data1);
when = ast_tvdiff_ms(test_data1->task_start, test_data1->test_start);
ast_test_validate(test, when > test_data1->interval * 0.9 && when < test_data1->interval * 1.1);
waitfor(test_data1);
when = ast_tvdiff_ms(test_data1->task_start, test_data1->test_start);
ast_test_validate(test, when > test_data1->interval * 2 * 0.9 && when < test_data1->interval * 2 * 1.1);
waitfor(test_data1);
when = ast_tvdiff_ms(test_data1->task_start, test_data1->test_start);
ast_test_validate(test, when > test_data1->interval * 3 * 0.9 && when < test_data1->interval * 3 * 1.1);
ast_sip_sched_task_cancel(task);
ao2_ref(task, -1);
task = NULL;
return AST_TEST_PASS;
}
static int load_module(void)
{
CHECK_PJSIP_MODULE_LOADED();
AST_TEST_REGISTER(serialized_scheduler);
AST_TEST_REGISTER(unserialized_scheduler);
AST_TEST_REGISTER(scheduler_cleanup);
AST_TEST_REGISTER(scheduler_cancel);
AST_TEST_REGISTER(scheduler_policy);
return AST_MODULE_LOAD_SUCCESS;
}
static int unload_module(void)
{
AST_TEST_UNREGISTER(scheduler_cancel);
AST_TEST_UNREGISTER(scheduler_cleanup);
AST_TEST_UNREGISTER(unserialized_scheduler);
AST_TEST_UNREGISTER(serialized_scheduler);
AST_TEST_UNREGISTER(scheduler_policy);
return 0;
}
AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "res_pjsip scheduler test module");