Sapian
/
asterisk
mirror of http://gerrit.asterisk.org/asterisk
3
0
Fork 0
Code Issues Releases Wiki Activity

app_voicemail: Remove dependency on the stasis cache

Browse Source 
app_voicemail utilized the stasis cache when polling mailboxes for MWI. This
caused a memory leak (items were not being appropriately removed from the
cache), and subsequent slowdown in system processing. This patch removes the
stasis cache dependency, thus alleviating the memory leak. It does this by
utilizing the new MWI API that better manages state lifetime.

ASTERISK-28443
ASTERISK-27121

Change-Id: Ie89fedaca81ea1fd03d150d9d3a1ef3d53740e46
changes/66/11466/7
Kevin Harwell 7 years ago
parent 9637e1dfdc
commit c93c579190
1 changed files with 89 additions and 220 deletions
Show Stats Download Patch File Download Diff File

309
apps/app_voicemail.c
Unescape View File

@ -1016,43 +1016,8 @@ static ast_cond_t poll_cond = PTHREAD_COND_INITIALIZER;
static pthread_t poll_thread = AST_PTHREADT_NULL;
static unsigned char poll_thread_run;
/*! Subscription to MWI event subscription changes */
static struct stasis_subscription *mwi_sub_sub;
/*!
* \brief An MWI subscription
*
* This is so we can keep track of which mailboxes are subscribed to.
* This way, we know which mailboxes to poll when the pollmailboxes
* option is being used.
*/
struct mwi_sub {
AST_RWLIST_ENTRY(mwi_sub) entry;
int old_urgent;
int old_new;
int old_old;
char *uniqueid;
char mailbox[0];
};
struct mwi_sub_task {
const char *mailbox;
const char *context;
const char *uniqueid;
};
static void mwi_sub_task_dtor(struct mwi_sub_task *mwist)
{
ast_free((void *) mwist->mailbox);
ast_free((void *) mwist->context);
ast_free((void *) mwist->uniqueid);
ast_free(mwist);
}
static struct ast_taskprocessor *mwi_subscription_tps;
static AST_RWLIST_HEAD_STATIC(mwi_subs, mwi_sub);
struct alias_mailbox_mapping {
char *alias;
char *mailbox;
@ -6316,7 +6281,7 @@ static int inboxcount(const char *mailbox, int *newmsgs, int *oldmsgs)
return res;
}
static void run_externnotify(char *context, char *extension, const char *flag)
static void run_externnotify(const char *context, const char *extension, const char *flag)
{
char arguments[255];
char ext_context[256] = "";
@ -13197,38 +13162,29 @@ static struct ast_cli_entry cli_voicemail[] = {
AST_CLI_DEFINE(handle_voicemail_reload, "Reload voicemail configuration"),
};
static void poll_subscribed_mailbox(struct mwi_sub *mwi_sub)
static int poll_subscribed_mailbox(struct ast_mwi_state *mwi_state, void *data)
{
int new = 0, old = 0, urgent = 0;
inboxcount2(mwi_sub->mailbox, &urgent, &new, &old);
if (!mwi_state) {
/* This should only occur due to allocation failure of a default mwi state object */
return 0;
}
inboxcount2(mwi_state->uniqueid, &urgent, &new, &old);
#ifdef IMAP_STORAGE
if (imap_poll_logout) {
imap_logout(mwi_sub->mailbox);
imap_logout(mwi_state->uniqueid);
}
#endif
if (urgent != mwi_sub->old_urgent || new != mwi_sub->old_new || old != mwi_sub->old_old) {
mwi_sub->old_urgent = urgent;
mwi_sub->old_new = new;
mwi_sub->old_old = old;
queue_mwi_event(NULL, mwi_sub->mailbox, urgent, new, old);
run_externnotify(NULL, mwi_sub->mailbox, NULL);
if (urgent != mwi_state->urgent_msgs || new != mwi_state->new_msgs || old != mwi_state->old_msgs) {
queue_mwi_event(NULL, mwi_state->uniqueid, urgent, new, old);
run_externnotify(NULL, mwi_state->uniqueid, NULL);
}
}
static void poll_subscribed_mailboxes(void)
{
struct mwi_sub *mwi_sub;
AST_RWLIST_RDLOCK(&mwi_subs);
AST_RWLIST_TRAVERSE(&mwi_subs, mwi_sub, entry) {
if (!ast_strlen_zero(mwi_sub->mailbox)) {
poll_subscribed_mailbox(mwi_sub);
}
}
AST_RWLIST_UNLOCK(&mwi_subs);
return 0;
}
static void *mb_poll_thread(void *data)
@ -13237,6 +13193,12 @@ static void *mb_poll_thread(void *data)
struct timespec ts = { 0, };
struct timeval wait;
ast_mwi_state_callback_subscribed(poll_subscribed_mailbox, NULL);
if (!poll_thread_run) {
break;
}
wait = ast_tvadd(ast_tvnow(), ast_samp2tv(poll_freq, 1));
ts.tv_sec = wait.tv_sec;
ts.tv_nsec = wait.tv_usec * 1000;
@ -13244,22 +13206,11 @@ static void *mb_poll_thread(void *data)
ast_mutex_lock(&poll_lock);
ast_cond_timedwait(&poll_cond, &poll_lock, &ts);
ast_mutex_unlock(&poll_lock);
if (!poll_thread_run)
break;
poll_subscribed_mailboxes();
}
return NULL;
}
static void mwi_sub_destroy(struct mwi_sub *mwi_sub)
{
ast_free(mwi_sub->uniqueid);
ast_free(mwi_sub);
}
#ifdef IMAP_STORAGE
static void imap_logout(const char *mailbox_id)
{
@ -13295,157 +13246,74 @@ static void imap_logout(const char *mailbox_id)
vmstate_delete(vms);
}
static void imap_close_subscribed_mailboxes(void)
{
struct mwi_sub *mwi_sub;
AST_RWLIST_RDLOCK(&mwi_subs);
AST_RWLIST_TRAVERSE(&mwi_subs, mwi_sub, entry) {
if (!ast_strlen_zero(mwi_sub->mailbox)) {
imap_logout(mwi_sub->mailbox);
}
}
AST_RWLIST_UNLOCK(&mwi_subs);
}
#endif
static int handle_unsubscribe(void *datap)
static int imap_close_subscribed_mailbox(struct ast_mwi_state *mwi_state, void *data)
{
struct mwi_sub *mwi_sub;
char *uniqueid = datap;
AST_RWLIST_WRLOCK(&mwi_subs);
AST_RWLIST_TRAVERSE_SAFE_BEGIN(&mwi_subs, mwi_sub, entry) {
if (!strcmp(mwi_sub->uniqueid, uniqueid)) {
AST_LIST_REMOVE_CURRENT(entry);
/* Don't break here since a duplicate uniqueid
* may have been added as a result of a cache dump. */
#ifdef IMAP_STORAGE
imap_logout(mwi_sub->mailbox);
#endif
mwi_sub_destroy(mwi_sub);
}
if (mwi_state && !ast_strlen_zero(mwi_state->uniqueid)) {
imap_logout(mwi_state->uniqueid);
}
AST_RWLIST_TRAVERSE_SAFE_END
AST_RWLIST_UNLOCK(&mwi_subs);
ast_free(uniqueid);
return 0;
}
static int handle_subscribe(void *datap)
{
unsigned int len;
struct mwi_sub *mwi_sub;
struct mwi_sub_task *p = datap;
len = sizeof(*mwi_sub) + 1;
if (!ast_strlen_zero(p->mailbox))
len += strlen(p->mailbox);
if (!ast_strlen_zero(p->context))
len += strlen(p->context) + 1; /* Allow for seperator */
#endif
if (!(mwi_sub = ast_calloc(1, len)))
return -1;
static int mwi_handle_unsubscribe2(void *data)
{
struct ast_mwi_state *mwi_state = data;
mwi_sub->uniqueid = ast_strdup(p->uniqueid);
if (!ast_strlen_zero(p->mailbox))
strcpy(mwi_sub->mailbox, p->mailbox);
/*
* Go ahead and clear the implicit MWI publisher here to avoid a leak. If a backing
* configuration is available it'll re-initialize (reset the cached state) on its
* next publish.
*/
ast_delete_mwi_state_full(mwi_state->uniqueid, NULL, NULL);
if (!ast_strlen_zero(p->context)) {
strcat(mwi_sub->mailbox, "@");
strcat(mwi_sub->mailbox, p->context);
}
#ifdef IMAP_STORAGE
imap_close_subscribed_mailbox(mwi_state, NULL);
#endif
AST_RWLIST_WRLOCK(&mwi_subs);
AST_RWLIST_INSERT_TAIL(&mwi_subs, mwi_sub, entry);
AST_RWLIST_UNLOCK(&mwi_subs);
mwi_sub_task_dtor(p);
poll_subscribed_mailbox(mwi_sub);
ao2_ref(mwi_state, -1);
return 0;
}
static void mwi_unsub_event_cb(struct stasis_subscription_change *change)
static void mwi_handle_unsubscribe(const char *id, struct ast_mwi_subscriber *sub)
{
char *uniqueid = ast_strdup(change->uniqueid);
if (!uniqueid) {
ast_log(LOG_ERROR, "Unable to allocate memory for uniqueid\n");
return;
}
void *data = ast_mwi_subscriber_data(sub);
if (ast_taskprocessor_push(mwi_subscription_tps, handle_unsubscribe, uniqueid) < 0) {
ast_free(uniqueid);
/* Don't bump data's reference. We'll just use the one returned above */
if (ast_taskprocessor_push(mwi_subscription_tps, mwi_handle_unsubscribe2, data) < 0) {
/* A reference was returned for data when retrieving, so remove it on error */
ao2_ref(data, -1);
}
}
static void mwi_sub_event_cb(struct stasis_subscription_change *change)
static int mwi_handle_subscribe2(void *data)
{
struct mwi_sub_task *mwist;
const char *topic;
char *context;
char *mailbox;
mwist = ast_calloc(1, (sizeof(*mwist)));
if (!mwist) {
return;
}
/* The topic name is prefixed with "mwi:all/" as this is a pool topic */
topic = stasis_topic_name(change->topic) + 8;
if (separate_mailbox(ast_strdupa(topic), &mailbox, &context)) {
ast_free(mwist);
return;
}
mwist->mailbox = ast_strdup(mailbox);
mwist->context = ast_strdup(context);
mwist->uniqueid = ast_strdup(change->uniqueid);
if (ast_taskprocessor_push(mwi_subscription_tps, handle_subscribe, mwist) < 0) {
mwi_sub_task_dtor(mwist);
}
poll_subscribed_mailbox(data, NULL);
ao2_ref(data, -1);
return 0;
}
static void mwi_event_cb(void *userdata, struct stasis_subscription *sub, struct stasis_message *msg)
static void mwi_handle_subscribe(const char *id, struct ast_mwi_subscriber *sub)
{
struct stasis_subscription_change *change;
/* Only looking for subscription change notices here */
if (stasis_message_type(msg) != stasis_subscription_change_type()) {
return;
}
change = stasis_message_data(msg);
if (change->topic == ast_mwi_topic_all()) {
return;
}
void *data = ast_mwi_subscriber_data(sub);
if (!strcmp(change->description, "Subscribe")) {
mwi_sub_event_cb(change);
} else if (!strcmp(change->description, "Unsubscribe")) {
mwi_unsub_event_cb(change);
/* Don't bump data's reference. We'll just use the one returned above */
if (ast_taskprocessor_push(mwi_subscription_tps, mwi_handle_subscribe2, data) < 0) {
/* A reference was returned for data when retrieving, so remove it on error */
ao2_ref(data, -1);
}
}
static int dump_cache(void *obj, void *arg, int flags)
{
struct stasis_message *msg = obj;
mwi_event_cb(NULL, NULL, msg);
return 0;
}
struct ast_mwi_observer mwi_observer = {
.on_subscribe = mwi_handle_subscribe,
.on_unsubscribe = mwi_handle_unsubscribe,
};
static void start_poll_thread(void)
{
int errcode;
mwi_sub_sub = stasis_subscribe(ast_mwi_topic_all(), mwi_event_cb, NULL);
if (mwi_sub_sub) {
struct ao2_container *cached = stasis_cache_dump(ast_mwi_state_cache(), stasis_subscription_change_type());
if (cached) {
ao2_callback(cached, OBJ_MULTIPLE | OBJ_NODATA, dump_cache, NULL);
}
ao2_cleanup(cached);
}
ast_mwi_add_observer(&mwi_observer);
poll_thread_run = 1;
@ -13458,15 +13326,14 @@ static void stop_poll_thread(void)
{
poll_thread_run = 0;
mwi_sub_sub = stasis_unsubscribe_and_join(mwi_sub_sub);
ast_mutex_lock(&poll_lock);
ast_cond_signal(&poll_cond);
ast_mutex_unlock(&poll_lock);
pthread_join(poll_thread, NULL);
poll_thread = AST_PTHREADT_NULL;
ast_mwi_remove_observer(&mwi_observer);
}
/*!
@ -13590,38 +13457,40 @@ static int append_vmu_info_astman(
}
static int manager_voicemail_refresh(struct mansession *s, const struct message *m)
static int manager_match_mailbox(struct ast_mwi_state *mwi_state, void *data)
{
const char *context = astman_get_header(m, "Context");
const char *mailbox = astman_get_header(m, "Mailbox");
struct mwi_sub *mwi_sub;
const char *context = astman_get_header(data, "Context");
const char *mailbox = astman_get_header(data, "Mailbox");
const char *at;
AST_RWLIST_RDLOCK(&mwi_subs);
AST_RWLIST_TRAVERSE(&mwi_subs, mwi_sub, entry) {
if (!ast_strlen_zero(mwi_sub->mailbox)) {
if (
/* First case: everything matches */
(ast_strlen_zero(context) && ast_strlen_zero(mailbox)) ||
/* Second case: match the mailbox only */
(ast_strlen_zero(context) && !ast_strlen_zero(mailbox) &&
(at = strchr(mwi_sub->mailbox, '@')) &&
strncmp(mailbox, mwi_sub->mailbox, at - mwi_sub->mailbox) == 0) ||
/* Third case: match the context only */
(!ast_strlen_zero(context) && ast_strlen_zero(mailbox) &&
(at = strchr(mwi_sub->mailbox, '@')) &&
strcmp(context, at + 1) == 0) ||
/* Final case: match an exact specified mailbox */
(!ast_strlen_zero(context) && !ast_strlen_zero(mailbox) &&
(at = strchr(mwi_sub->mailbox, '@')) &&
strncmp(mailbox, mwi_sub->mailbox, at - mwi_sub->mailbox) == 0 &&
strcmp(context, at + 1) == 0)
if (!ast_strlen_zero(mwi_state->uniqueid)) {
if (
/* First case: everything matches */
(ast_strlen_zero(context) && ast_strlen_zero(mailbox)) ||
/* Second case: match the mailbox only */
(ast_strlen_zero(context) && !ast_strlen_zero(mailbox) &&
(at = strchr(mwi_state->uniqueid, '@')) &&
strncmp(mailbox, mwi_state->uniqueid, at - mwi_state->uniqueid) == 0) ||
/* Third case: match the context only */
(!ast_strlen_zero(context) && ast_strlen_zero(mailbox) &&
(at = strchr(mwi_state->uniqueid, '@')) &&
strcmp(context, at + 1) == 0) ||
/* Final case: match an exact specified mailbox */
(!ast_strlen_zero(context) && !ast_strlen_zero(mailbox) &&
(at = strchr(mwi_state->uniqueid, '@')) &&
strncmp(mailbox, mwi_state->uniqueid, at - mwi_state->uniqueid) == 0 &&
strcmp(context, at + 1) == 0)
) {
poll_subscribed_mailbox(mwi_sub);
}
poll_subscribed_mailbox(mwi_state, NULL);
}
}
AST_RWLIST_UNLOCK(&mwi_subs);
return 0;
}
static int manager_voicemail_refresh(struct mansession *s, const struct message *m)
{
ast_mwi_state_callback_all(manager_match_mailbox, (void *)m);
astman_send_ack(s, m, "Refresh sent");
return RESULT_SUCCESS;
}
@ -13943,7 +13812,7 @@ static int actual_load_config(int reload, struct ast_config *cfg, struct ast_con
strcpy(listen_control_stop_key, DEFAULT_LISTEN_CONTROL_STOP_KEY);
#ifdef IMAP_STORAGE
imap_close_subscribed_mailboxes();
ast_mwi_state_callback_all(imap_close_subscribed_mailbox, NULL);
#endif
/* Free all the users structure */
@ -15326,7 +15195,7 @@ static int unload_module(void)
ast_unload_realtime("voicemail_data");
#ifdef IMAP_STORAGE
imap_close_subscribed_mailboxes();
ast_mwi_state_callback_all(imap_close_subscribed_mailbox, NULL);
#endif
free_vm_users();
free_vm_zones();

Write Preview
Loading…
Cancel
Save