Newer
Older
Dwayne M. Hubbard
committed
/*
* Asterisk -- An open source telephony toolkit.
*
* Copyright (C) 2007-2013, Digium, Inc.
Dwayne M. Hubbard
committed
*
* Dwayne M. Hubbard <dhubbard@digium.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.
*/
Dwayne M. Hubbard
committed
* \brief Maintain a container of uniquely-named taskprocessor threads that can be shared across modules.
*
* \author Dwayne Hubbard <dhubbard@digium.com>
*/
/*** MODULEINFO
<support_level>core</support_level>
***/
Dwayne M. Hubbard
committed
#include "asterisk.h"
Dwayne M. Hubbard
committed
#include "asterisk/_private.h"
#include "asterisk/module.h"
#include "asterisk/time.h"
#include "asterisk/astobj2.h"
#include "asterisk/cli.h"
#include "asterisk/taskprocessor.h"
#include "asterisk/sem.h"
Dwayne M. Hubbard
committed
/*!
* \brief tps_task structure is queued to a taskprocessor
Dwayne M. Hubbard
committed
*
* tps_tasks are processed in FIFO order and freed by the taskprocessing
* thread after the task handler returns. The callback function that is assigned
* to the execute() function pointer is responsible for releasing datap resources if necessary.
*/
Dwayne M. Hubbard
committed
struct tps_task {
/*! \brief The execute() task callback function pointer */
union {
int (*execute)(void *datap);
int (*execute_local)(struct ast_taskprocessor_local *local);
} callback;
Dwayne M. Hubbard
committed
/*! \brief The data pointer for the task execute() function */
void *datap;
/*! \brief AST_LIST_ENTRY overhead */
AST_LIST_ENTRY(tps_task) list;
unsigned int wants_local:1;
Dwayne M. Hubbard
committed
};
/*! \brief tps_taskprocessor_stats maintain statistics for a taskprocessor. */
struct tps_taskprocessor_stats {
/*! \brief This is the maximum number of tasks queued at any one time */
unsigned long max_qsize;
/*! \brief This is the current number of tasks processed */
unsigned long _tasks_processed_count;
};
/*! \brief A ast_taskprocessor structure is a singleton by name */
struct ast_taskprocessor {
/*! \brief Taskprocessor statistics */
struct tps_taskprocessor_stats stats;
Dwayne M. Hubbard
committed
/*! \brief Taskprocessor current queue size */
long tps_queue_size;
/*! \brief Taskprocessor low water clear alert level */
long tps_queue_low;
/*! \brief Taskprocessor high water alert trigger level */
long tps_queue_high;
Dwayne M. Hubbard
committed
/*! \brief Taskprocessor queue */
AST_LIST_HEAD_NOLOCK(tps_queue, tps_task) tps_queue;
struct ast_taskprocessor_listener *listener;
/*! Current thread executing the tasks */
pthread_t thread;
/*! Indicates if the taskprocessor is currently executing a task */
unsigned int executing:1;
/*! Indicates that a high water warning has been issued on this task processor */
unsigned int high_water_warned:1;
/*! Indicates that a high water alert is active on this taskprocessor */
unsigned int high_water_alert:1;
/*! Indicates if the taskprocessor is currently suspended */
unsigned int suspended:1;
/*! \brief Anything before the first '/' in the name (if there is one) */
char *subsystem;
/*! \brief Friendly name of the taskprocessor.
* Subsystem is appended after the name's NULL terminator.
*/
char name[0];
Dwayne M. Hubbard
committed
};
/*!
* \brief A listener for taskprocessors
*
* \since 12.0.0
*
* When a taskprocessor's state changes, the listener
* is notified of the change. This allows for tasks
* to be addressed in whatever way is appropriate for
* the module using the taskprocessor.
*/
struct ast_taskprocessor_listener {
/*! The callbacks the taskprocessor calls into to notify of state changes */
const struct ast_taskprocessor_listener_callbacks *callbacks;
/*! The taskprocessor that the listener is listening to */
struct ast_taskprocessor *tps;
/*! Data private to the listener */
void *user_data;
};
/*!
* Keep track of which subsystems are in alert
* and how many of their taskprocessors are overloaded.
*/
struct subsystem_alert {
unsigned int alert_count;
char subsystem[0];
};
static AST_VECTOR_RW(subsystem_alert_vector, struct subsystem_alert *) overloaded_subsystems;
#ifdef LOW_MEMORY
#define TPS_MAX_BUCKETS 61
#else
/*! \brief Number of buckets in the tps_singletons container. */
#define TPS_MAX_BUCKETS 1567
#endif
Dwayne M. Hubbard
committed
/*! \brief tps_singletons is the astobj2 container for taskprocessor singletons */
static struct ao2_container *tps_singletons;
/*! \brief CLI <example>taskprocessor ping <blah></example> operation requires a ping condition */
Dwayne M. Hubbard
committed
static ast_cond_t cli_ping_cond;
/*! \brief CLI <example>taskprocessor ping <blah></example> operation requires a ping condition lock */
AST_MUTEX_DEFINE_STATIC(cli_ping_cond_lock);
Dwayne M. Hubbard
committed
/*! \brief The astobj2 hash callback for taskprocessors */
static int tps_hash_cb(const void *obj, const int flags);
/*! \brief The astobj2 compare callback for taskprocessors */
static int tps_cmp_cb(void *obj, void *arg, int flags);
Dwayne M. Hubbard
committed
/*! \brief CLI <example>taskprocessor ping <blah></example> handler function */
Dwayne M. Hubbard
committed
static int tps_ping_handler(void *datap);
static char *cli_tps_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
static char *cli_subsystem_alert_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
static char *cli_tps_reset_stats(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
static char *cli_tps_reset_stats_all(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a);
Dwayne M. Hubbard
committed
static int tps_sort_cb(const void *obj_left, const void *obj_right, int flags);
Dwayne M. Hubbard
committed
static struct ast_cli_entry taskprocessor_clis[] = {
AST_CLI_DEFINE(cli_tps_ping, "Ping a named task processor"),
Dwayne M. Hubbard
committed
AST_CLI_DEFINE(cli_tps_report, "List instantiated task processors and statistics"),
AST_CLI_DEFINE(cli_subsystem_alert_report, "List task processor subsystems in alert"),
AST_CLI_DEFINE(cli_tps_reset_stats, "Reset a named task processor's stats"),
AST_CLI_DEFINE(cli_tps_reset_stats_all, "Reset all task processors' stats"),
Dwayne M. Hubbard
committed
};
struct default_taskprocessor_listener_pvt {
pthread_t poll_thread;
int dead;
struct ast_sem sem;
static void default_listener_pvt_destroy(struct default_taskprocessor_listener_pvt *pvt)
ast_assert(pvt->dead);
ast_sem_destroy(&pvt->sem);
ast_free(pvt);
static void default_listener_pvt_dtor(struct ast_taskprocessor_listener *listener)
struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
default_listener_pvt_destroy(pvt);
listener->user_data = NULL;
/*!
* \brief Function that processes tasks in the taskprocessor
* \internal
*/
static void *default_tps_processing_function(void *data)
{
struct ast_taskprocessor_listener *listener = data;
struct ast_taskprocessor *tps = listener->tps;
struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
int sem_value;
int res;
while (!pvt->dead) {
res = ast_sem_wait(&pvt->sem);
if (res != 0 && errno != EINTR) {
ast_log(LOG_ERROR, "ast_sem_wait(): %s\n",
strerror(errno));
/* Just give up */
break;
ast_taskprocessor_execute(tps);
/* No posting to a dead taskprocessor! */
res = ast_sem_getvalue(&pvt->sem, &sem_value);
ast_assert(res == 0 && sem_value == 0);
/* Free the shutdown reference (see default_listener_shutdown) */
ao2_t_ref(listener->tps, -1, "tps-shutdown");
static int default_listener_start(struct ast_taskprocessor_listener *listener)
{
struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
if (ast_pthread_create(&pvt->poll_thread, NULL, default_tps_processing_function, listener)) {
return -1;
}
return 0;
}
static void default_task_pushed(struct ast_taskprocessor_listener *listener, int was_empty)
{
struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
if (ast_sem_post(&pvt->sem) != 0) {
ast_log(LOG_ERROR, "Failed to notify of enqueued task: %s\n",
strerror(errno));
}
}
static int default_listener_die(void *data)
{
struct default_taskprocessor_listener_pvt *pvt = data;
pvt->dead = 1;
return 0;
}
static void default_listener_shutdown(struct ast_taskprocessor_listener *listener)
{
struct default_taskprocessor_listener_pvt *pvt = listener->user_data;
int res;
/* Hold a reference during shutdown */
ao2_t_ref(listener->tps, +1, "tps-shutdown");
if (ast_taskprocessor_push(listener->tps, default_listener_die, pvt)) {
/* This will cause the thread to exit early without completing tasks already
* in the queue. This is probably the least bad option in this situation. */
default_listener_die(pvt);
}
ast_assert(pvt->poll_thread != AST_PTHREADT_NULL);
if (pthread_equal(pthread_self(), pvt->poll_thread)) {
res = pthread_detach(pvt->poll_thread);
if (res != 0) {
ast_log(LOG_ERROR, "pthread_detach(): %s\n", strerror(errno));
}
} else {
res = pthread_join(pvt->poll_thread, NULL);
if (res != 0) {
ast_log(LOG_ERROR, "pthread_join(): %s\n", strerror(errno));
pvt->poll_thread = AST_PTHREADT_NULL;
}
static const struct ast_taskprocessor_listener_callbacks default_listener_callbacks = {
.start = default_listener_start,
.task_pushed = default_task_pushed,
.shutdown = default_listener_shutdown,
.dtor = default_listener_pvt_dtor,
/*! \brief How many seconds to wait for running taskprocessors to finish on shutdown. */
#define AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT 10
/*!
* \internal
* \brief Clean up resources on Asterisk shutdown
*/
static void tps_shutdown(void)
{
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
int objcount;
int tries;
struct ao2_container *sorted_tps;
struct ast_taskprocessor *tps;
struct ao2_iterator iter;
struct timespec delay = {1, 0};
/* During shutdown there may still be taskprocessor threads running and those
* tasprocessors reference tps_singletons. When those taskprocessors finish
* they will call ast_taskprocessor_unreference, creating a race condition which
* can result in tps_singletons being referenced after being deleted. To try and
* avoid this we check the container count and if greater than zero, give the
* running taskprocessors a chance to finish */
objcount = ao2_container_count(tps_singletons);
if (objcount > 0) {
ast_log(LOG_DEBUG,
"waiting for taskprocessor shutdown, %d tps object(s) still allocated.\n",
objcount);
/* give the running taskprocessors a chance to finish, up to
* AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT seconds */
for (tries = 0; tries < AST_TASKPROCESSOR_SHUTDOWN_MAX_WAIT; tries++) {
while (nanosleep(&delay, &delay));
objcount = ao2_container_count(tps_singletons);
/* if count is 0, we are done waiting */
if (objcount == 0) {
break;
}
delay.tv_sec = 1;
delay.tv_nsec = 0;
ast_log(LOG_DEBUG,
"waiting for taskprocessor shutdown, %d tps object(s) still allocated.\n",
objcount);
}
}
/* rather than try forever, risk an assertion on shutdown. This probably indicates
* a taskprocessor was not cleaned up somewhere */
if (objcount > 0) {
ast_log(LOG_ERROR,
"Asertion may occur, the following taskprocessors are still runing:\n");
sorted_tps = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, tps_sort_cb,
NULL);
if (!sorted_tps || ao2_container_dup(sorted_tps, tps_singletons, 0)) {
ast_log(LOG_ERROR, "unable to get sorted list of taskprocessors");
}
else {
iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK);
while ((tps = ao2_iterator_next(&iter))) {
ast_log(LOG_ERROR, "taskprocessor '%s'\n", tps->name);
}
}
ao2_cleanup(sorted_tps);
}
else {
ast_log(LOG_DEBUG,
"All waiting taskprocessors cleared!\n");
}
ast_cli_unregister_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
AST_VECTOR_CALLBACK_VOID(&overloaded_subsystems, ast_free);
AST_VECTOR_RW_FREE(&overloaded_subsystems);
ao2_t_ref(tps_singletons, -1, "Unref tps_singletons in shutdown");
Dwayne M. Hubbard
committed
/* initialize the taskprocessor container and register CLI operations */
int ast_tps_init(void)
{
tps_singletons = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
TPS_MAX_BUCKETS, tps_hash_cb, NULL, tps_cmp_cb);
if (!tps_singletons) {
Dwayne M. Hubbard
committed
ast_log(LOG_ERROR, "taskprocessor container failed to initialize!\n");
return -1;
}
if (AST_VECTOR_RW_INIT(&overloaded_subsystems, 10)) {
ao2_ref(tps_singletons, -1);
ast_log(LOG_ERROR, "taskprocessor subsystems vector failed to initialize!\n");
return -1;
}
ast_cond_init(&cli_ping_cond, NULL);
Dwayne M. Hubbard
committed
ast_cli_register_multiple(taskprocessor_clis, ARRAY_LEN(taskprocessor_clis));
ast_register_cleanup(tps_shutdown);
Dwayne M. Hubbard
committed
return 0;
}
/* allocate resources for the task */
static struct tps_task *tps_task_alloc(int (*task_exe)(void *datap), void *datap)
{
struct tps_task *t;
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
if (!task_exe) {
ast_log(LOG_ERROR, "task_exe is NULL!\n");
return NULL;
}
t = ast_calloc(1, sizeof(*t));
if (!t) {
ast_log(LOG_ERROR, "failed to allocate task!\n");
return NULL;
}
t->callback.execute = task_exe;
t->datap = datap;
return t;
}
static struct tps_task *tps_task_alloc_local(int (*task_exe)(struct ast_taskprocessor_local *local), void *datap)
{
struct tps_task *t;
if (!task_exe) {
ast_log(LOG_ERROR, "task_exe is NULL!\n");
return NULL;
}
t = ast_calloc(1, sizeof(*t));
if (!t) {
ast_log(LOG_ERROR, "failed to allocate task!\n");
return NULL;
Dwayne M. Hubbard
committed
}
t->callback.execute_local = task_exe;
t->datap = datap;
t->wants_local = 1;
Dwayne M. Hubbard
committed
return t;
}
Dwayne M. Hubbard
committed
static void *tps_task_free(struct tps_task *task)
{
Dwayne M. Hubbard
committed
return NULL;
}
/* Taskprocessor tab completion.
*
* The caller of this function is responsible for argument
* position checks prior to calling.
*/
static char *tps_taskprocessor_tab_complete(struct ast_cli_args *a)
Dwayne M. Hubbard
committed
{
int tklen;
struct ast_taskprocessor *p;
Dwayne M. Hubbard
committed
struct ao2_iterator i;
tklen = strlen(a->word);
i = ao2_iterator_init(tps_singletons, 0);
while ((p = ao2_iterator_next(&i))) {
if (!strncasecmp(a->word, p->name, tklen)) {
if (ast_cli_completion_add(ast_strdup(p->name))) {
ast_taskprocessor_unreference(p);
break;
}
Dwayne M. Hubbard
committed
}
ast_taskprocessor_unreference(p);
Dwayne M. Hubbard
committed
}
ao2_iterator_destroy(&i);
Dwayne M. Hubbard
committed
}
/* ping task handling function */
static int tps_ping_handler(void *datap)
{
ast_mutex_lock(&cli_ping_cond_lock);
ast_cond_signal(&cli_ping_cond);
ast_mutex_unlock(&cli_ping_cond_lock);
return 0;
}
/* ping the specified taskprocessor and display the ping time on the CLI */
static char *cli_tps_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
{
struct timeval begin, end, delta;
const char *name;
struct timeval when;
Dwayne M. Hubbard
committed
struct timespec ts;
struct ast_taskprocessor *tps;
Dwayne M. Hubbard
committed
switch (cmd) {
case CLI_INIT:
e->command = "core ping taskprocessor";
"Usage: core ping taskprocessor <taskprocessor>\n"
" Displays the time required for a task to be processed\n";
Dwayne M. Hubbard
committed
return NULL;
case CLI_GENERATE:
if (a->pos == 3) {
return tps_taskprocessor_tab_complete(a);
} else {
return NULL;
}
Dwayne M. Hubbard
committed
}
if (a->argc != 4)
Dwayne M. Hubbard
committed
return CLI_SHOWUSAGE;
name = a->argv[3];
Dwayne M. Hubbard
committed
if (!(tps = ast_taskprocessor_get(name, TPS_REF_IF_EXISTS))) {
ast_cli(a->fd, "\nping failed: %s not found\n\n", name);
return CLI_SUCCESS;
}
ast_cli(a->fd, "\npinging %s ...", name);
/*
* Wait up to 5 seconds for a ping reply.
*
* On a very busy system it could take awhile to get a
* ping response from some taskprocessors.
*/
begin = ast_tvnow();
when = ast_tvadd(begin, ast_samp2tv(5000, 1000));
ts.tv_sec = when.tv_sec;
ts.tv_nsec = when.tv_usec * 1000;
Dwayne M. Hubbard
committed
ast_mutex_lock(&cli_ping_cond_lock);
if (ast_taskprocessor_push(tps, tps_ping_handler, 0) < 0) {
ast_mutex_unlock(&cli_ping_cond_lock);
Dwayne M. Hubbard
committed
ast_cli(a->fd, "\nping failed: could not push task to %s\n\n", name);
ast_taskprocessor_unreference(tps);
Dwayne M. Hubbard
committed
return CLI_FAILURE;
}
ast_cond_timedwait(&cli_ping_cond, &cli_ping_cond_lock, &ts);
ast_mutex_unlock(&cli_ping_cond_lock);
Dwayne M. Hubbard
committed
end = ast_tvnow();
delta = ast_tvsub(end, begin);
ast_cli(a->fd, "\n\t%24s ping time: %.1ld.%.6ld sec\n\n", name, (long)delta.tv_sec, (long int)delta.tv_usec);
ast_taskprocessor_unreference(tps);
Dwayne M. Hubbard
committed
}
/*!
* \internal
* \brief Taskprocessor ao2 container sort function.
* \since 13.8.0
*
* \param obj_left pointer to the (user-defined part) of an object.
* \param obj_right pointer to the (user-defined part) of an object.
* \param flags flags from ao2_callback()
* OBJ_SEARCH_OBJECT - if set, 'obj_right', is an object.
* OBJ_SEARCH_KEY - if set, 'obj_right', is a search key item that is not an object.
* OBJ_SEARCH_PARTIAL_KEY - if set, 'obj_right', is a partial search key item that is not an object.
*
* \retval negative if obj_left < obj_right
* \retval 0 if obj_left == obj_right
* \retval positive if obj_left > obj_right
*/
static int tps_sort_cb(const void *obj_left, const void *obj_right, int flags)
{
const struct ast_taskprocessor *tps_left = obj_left;
const struct ast_taskprocessor *tps_right = obj_right;
const char *right_key = obj_right;
int cmp;
switch (flags & OBJ_SEARCH_MASK) {
default:
case OBJ_SEARCH_OBJECT:
right_key = tps_right->name;
/* Fall through */
case OBJ_SEARCH_KEY:
cmp = strcasecmp(tps_left->name, right_key);
break;
case OBJ_SEARCH_PARTIAL_KEY:
cmp = strncasecmp(tps_left->name, right_key, strlen(right_key));
break;
}
return cmp;
}
#define FMT_HEADERS "%-70s %10s %10s %10s %10s %10s\n"
#define FMT_FIELDS "%-70s %10lu %10lu %10lu %10lu %10lu\n"
Dwayne M. Hubbard
committed
/*!
* \internal
* \brief Print taskprocessor information to CLI.
* \since 13.30.0
*
* \param fd the file descriptor
* \param tps the taskprocessor
*/
static void tps_report_taskprocessor_list_helper(int fd, struct ast_taskprocessor *tps)
{
ast_cli(fd, FMT_FIELDS, tps->name, tps->stats._tasks_processed_count,
tps->tps_queue_size, tps->stats.max_qsize, tps->tps_queue_low,
tps->tps_queue_high);
}
Dwayne M. Hubbard
committed
/*!
* \internal
* \brief Prints an optionally narrowed down list of taskprocessors to the CLI.
* \since 13.30.0
*
* \param fd the file descriptor
* \param like the string we are matching on
*
* \retval 0 otherwise
*/
static int tps_report_taskprocessor_list(int fd, const char *like)
{
int tps_count = 0;
int word_len;
struct ao2_container *sorted_tps;
struct ast_taskprocessor *tps;
struct ao2_iterator iter;
sorted_tps = ao2_container_alloc_rbtree(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, tps_sort_cb,
NULL);
if (!sorted_tps
|| ao2_container_dup(sorted_tps, tps_singletons, 0)) {
ast_debug(1, "Failed to retrieve sorted taskprocessors\n");
ao2_cleanup(sorted_tps);
return 0;
Dwayne M. Hubbard
committed
word_len = strlen(like);
iter = ao2_iterator_init(sorted_tps, AO2_ITERATOR_UNLINK);
while ((tps = ao2_iterator_next(&iter))) {
if (like) {
if (!strncasecmp(like, tps->name, word_len)) {
tps_report_taskprocessor_list_helper(fd, tps);
tps_count++;
}
} else {
tps_report_taskprocessor_list_helper(fd, tps);
tps_count++;
}
ast_taskprocessor_unreference(tps);
Dwayne M. Hubbard
committed
}
ao2_iterator_destroy(&iter);
ao2_ref(sorted_tps, -1);
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
return tps_count;
}
static char *cli_tps_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
{
const char *like;
switch (cmd) {
case CLI_INIT:
e->command = "core show taskprocessors [like]";
e->usage =
"Usage: core show taskprocessors [like keyword]\n"
" Shows a list of instantiated task processors and their statistics\n";
return NULL;
case CLI_GENERATE:
if (a->pos == e->args) {
return tps_taskprocessor_tab_complete(a);
} else {
return NULL;
}
}
if (a->argc == e->args - 1) {
like = "";
} else if (a->argc == e->args + 1 && !strcasecmp(a->argv[e->args-1], "like")) {
like = a->argv[e->args];
} else {
return CLI_SHOWUSAGE;
}
ast_cli(a->fd, "\n" FMT_HEADERS, "Processor", "Processed", "In Queue", "Max Depth", "Low water", "High water");
ast_cli(a->fd, "\n%d taskprocessors\n\n", tps_report_taskprocessor_list(a->fd, like));
Dwayne M. Hubbard
committed
}
/* hash callback for astobj2 */
static int tps_hash_cb(const void *obj, const int flags)
{
const struct ast_taskprocessor *tps = obj;
const char *name = flags & OBJ_KEY ? obj : tps->name;
Dwayne M. Hubbard
committed
return ast_str_case_hash(name);
Dwayne M. Hubbard
committed
}
/* compare callback for astobj2 */
static int tps_cmp_cb(void *obj, void *arg, int flags)
Dwayne M. Hubbard
committed
{
struct ast_taskprocessor *lhs = obj, *rhs = arg;
const char *rhsname = flags & OBJ_KEY ? arg : rhs->name;
Dwayne M. Hubbard
committed
return !strcasecmp(lhs->name, rhsname) ? CMP_MATCH | CMP_STOP : 0;
Dwayne M. Hubbard
committed
}
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
static int subsystem_match(struct subsystem_alert *alert, const char *subsystem)
{
return !strcmp(alert->subsystem, subsystem);
}
static int subsystem_cmp(struct subsystem_alert *a, struct subsystem_alert *b)
{
return strcmp(a->subsystem, b->subsystem);
}
unsigned int ast_taskprocessor_get_subsystem_alert(const char *subsystem)
{
struct subsystem_alert *alert;
unsigned int count = 0;
int idx;
AST_VECTOR_RW_RDLOCK(&overloaded_subsystems);
idx = AST_VECTOR_GET_INDEX(&overloaded_subsystems, subsystem, subsystem_match);
if (idx >= 0) {
alert = AST_VECTOR_GET(&overloaded_subsystems, idx);
count = alert->alert_count;
}
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
return count;
}
static void subsystem_alert_increment(const char *subsystem)
{
struct subsystem_alert *alert;
int idx;
if (ast_strlen_zero(subsystem)) {
return;
}
AST_VECTOR_RW_WRLOCK(&overloaded_subsystems);
idx = AST_VECTOR_GET_INDEX(&overloaded_subsystems, subsystem, subsystem_match);
if (idx >= 0) {
alert = AST_VECTOR_GET(&overloaded_subsystems, idx);
alert->alert_count++;
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
return;
}
alert = ast_malloc(sizeof(*alert) + strlen(subsystem) + 1);
if (!alert) {
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
return;
}
alert->alert_count = 1;
strcpy(alert->subsystem, subsystem); /* Safe */
if (AST_VECTOR_APPEND(&overloaded_subsystems, alert)) {
ast_free(alert);
}
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
}
static void subsystem_alert_decrement(const char *subsystem)
{
struct subsystem_alert *alert;
int idx;
if (ast_strlen_zero(subsystem)) {
return;
}
AST_VECTOR_RW_WRLOCK(&overloaded_subsystems);
idx = AST_VECTOR_GET_INDEX(&overloaded_subsystems, subsystem, subsystem_match);
if (idx < 0) {
ast_log(LOG_ERROR,
"Can't decrement alert count for subsystem '%s' as it wasn't in alert\n", subsystem);
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
return;
}
alert = AST_VECTOR_GET(&overloaded_subsystems, idx);
alert->alert_count--;
if (alert->alert_count <= 0) {
AST_VECTOR_REMOVE(&overloaded_subsystems, idx, 0);
ast_free(alert);
}
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
}
static void subsystem_copy(struct subsystem_alert *alert,
struct subsystem_alert_vector *vector)
{
struct subsystem_alert *alert_copy;
alert_copy = ast_malloc(sizeof(*alert_copy) + strlen(alert->subsystem) + 1);
if (!alert_copy) {
return;
}
alert_copy->alert_count = alert->alert_count;
strcpy(alert_copy->subsystem, alert->subsystem); /* Safe */
if (AST_VECTOR_ADD_SORTED(vector, alert_copy, subsystem_cmp)) {
ast_free(alert_copy);
}
}
static char *cli_subsystem_alert_report(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
{
struct subsystem_alert_vector sorted_subsystems;
int i;
#define FMT_HEADERS_SUBSYSTEM "%-32s %12s\n"
#define FMT_FIELDS_SUBSYSTEM "%-32s %12u\n"
switch (cmd) {
case CLI_INIT:
e->command = "core show taskprocessor alerted subsystems";
e->usage =
"Usage: core show taskprocessor alerted subsystems\n"
" Shows a list of task processor subsystems that are currently alerted\n";
return NULL;
case CLI_GENERATE:
return NULL;
}
if (a->argc != e->args) {
return CLI_SHOWUSAGE;
}
if (AST_VECTOR_INIT(&sorted_subsystems, AST_VECTOR_SIZE(&overloaded_subsystems))) {
return CLI_FAILURE;
}
AST_VECTOR_RW_RDLOCK(&overloaded_subsystems);
for (i = 0; i < AST_VECTOR_SIZE(&overloaded_subsystems); i++) {
subsystem_copy(AST_VECTOR_GET(&overloaded_subsystems, i), &sorted_subsystems);
}
AST_VECTOR_RW_UNLOCK(&overloaded_subsystems);
ast_cli(a->fd, "\n" FMT_HEADERS_SUBSYSTEM, "Subsystem", "Alert Count");
for (i = 0; i < AST_VECTOR_SIZE(&sorted_subsystems); i++) {
struct subsystem_alert *alert = AST_VECTOR_GET(&sorted_subsystems, i);
ast_cli(a->fd, FMT_FIELDS_SUBSYSTEM, alert->subsystem, alert->alert_count);
}
ast_cli(a->fd, "\n%zu subsystems\n\n", AST_VECTOR_SIZE(&sorted_subsystems));
AST_VECTOR_CALLBACK_VOID(&sorted_subsystems, ast_free);
AST_VECTOR_FREE(&sorted_subsystems);
return CLI_SUCCESS;
}
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
/*! Count of the number of taskprocessors in high water alert. */
static unsigned int tps_alert_count;
/*! Access protection for tps_alert_count */
AST_RWLOCK_DEFINE_STATIC(tps_alert_lock);
/*!
* \internal
* \brief Add a delta to tps_alert_count with protection.
* \since 13.10.0
*
* \param tps Taskprocessor updating queue water mark alert trigger.
* \param delta The amount to add to tps_alert_count.
*/
static void tps_alert_add(struct ast_taskprocessor *tps, int delta)
{
unsigned int old;
ast_rwlock_wrlock(&tps_alert_lock);
old = tps_alert_count;
tps_alert_count += delta;
if (DEBUG_ATLEAST(3)
/* and tps_alert_count becomes zero or non-zero */
&& !old != !tps_alert_count) {
ast_log(LOG_DEBUG, "Taskprocessor '%s' %s the high water alert.\n",
tps->name, tps_alert_count ? "triggered" : "cleared");
}
if (tps->subsystem[0] != '\0') {
if (delta > 0) {
subsystem_alert_increment(tps->subsystem);
} else {
subsystem_alert_decrement(tps->subsystem);
}
}
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
ast_rwlock_unlock(&tps_alert_lock);
}
unsigned int ast_taskprocessor_alert_get(void)
{
unsigned int count;
ast_rwlock_rdlock(&tps_alert_lock);
count = tps_alert_count;
ast_rwlock_unlock(&tps_alert_lock);
return count;
}
int ast_taskprocessor_alert_set_levels(struct ast_taskprocessor *tps, long low_water, long high_water)
{
if (!tps || high_water < 0 || high_water < low_water) {
return -1;
}
if (low_water < 0) {
/* Set low water level to 90% of high water level */
low_water = (high_water * 9) / 10;
}
ao2_lock(tps);
tps->tps_queue_low = low_water;
tps->tps_queue_high = high_water;
if (tps->high_water_alert) {
if (!tps->tps_queue_size || tps->tps_queue_size < low_water) {
/* Update water mark alert immediately */
tps->high_water_alert = 0;
tps_alert_add(tps, -1);
}
} else {
if (high_water < tps->tps_queue_size) {
/* Update water mark alert immediately */
tps->high_water_alert = 1;
tps_alert_add(tps, +1);
}
}
ao2_unlock(tps);
return 0;
}
Dwayne M. Hubbard
committed
/* destroy the taskprocessor */
static void tps_taskprocessor_dtor(void *tps)
Dwayne M. Hubbard
committed
{
struct ast_taskprocessor *t = tps;
Mark Michelson
committed
struct tps_task *task;
while ((task = AST_LIST_REMOVE_HEAD(&t->tps_queue, list))) {
tps_task_free(task);
Dwayne M. Hubbard
committed
}
t->tps_queue_size = 0;
if (t->high_water_alert) {
t->high_water_alert = 0;
tps_alert_add(t, -1);
}
ao2_cleanup(t->listener);
t->listener = NULL;
Dwayne M. Hubbard
committed
}
/* pop the front task and return it */
static struct tps_task *tps_taskprocessor_pop(struct ast_taskprocessor *tps)
{
struct tps_task *task;
if ((task = AST_LIST_REMOVE_HEAD(&tps->tps_queue, list))) {
--tps->tps_queue_size;
if (tps->high_water_alert && tps->tps_queue_size <= tps->tps_queue_low) {
tps->high_water_alert = 0;
tps_alert_add(tps, -1);
}
Dwayne M. Hubbard
committed
}
return task;
}
long ast_taskprocessor_size(struct ast_taskprocessor *tps)
Dwayne M. Hubbard
committed
{
return (tps) ? tps->tps_queue_size : -1;
Dwayne M. Hubbard
committed
}
/* taskprocessor name accessor */
const char *ast_taskprocessor_name(struct ast_taskprocessor *tps)
{
if (!tps) {
ast_log(LOG_ERROR, "no taskprocessor specified!\n");
return NULL;
}
return tps->name;
}
static void listener_shutdown(struct ast_taskprocessor_listener *listener)
{
listener->callbacks->shutdown(listener);
ao2_ref(listener->tps, -1);
}
static void taskprocessor_listener_dtor(void *obj)
{
struct ast_taskprocessor_listener *listener = obj;
if (listener->callbacks->dtor) {
listener->callbacks->dtor(listener);
}
}
struct ast_taskprocessor_listener *ast_taskprocessor_listener_alloc(const struct ast_taskprocessor_listener_callbacks *callbacks, void *user_data)
struct ast_taskprocessor_listener *listener;
listener = ao2_alloc(sizeof(*listener), taskprocessor_listener_dtor);