Manager: Short circuit AMI message processing.

Improve AMI message processing performance if there are no consumers
listening for the messages.  We now skip creating the AMI event message
text strings.

Change-Id: I7b22fc5ec4e500d00635c1a467aa8ea68a1bb2b3
changes/93/2593/2
Richard Mudgett 9 years ago
parent 6ddd856b86
commit 06632a0d11

@ -414,14 +414,14 @@ const struct timeval *stasis_message_timestamp(const struct stasis_message *msg)
* May return \c NULL, to indicate no representation. The returned object should * May return \c NULL, to indicate no representation. The returned object should
* be ast_json_unref()'ed. * be ast_json_unref()'ed.
* *
* \param message Message to convert to JSON string. * \param msg Message to convert to JSON string.
* \param sanitize Snapshot sanitization callback. * \param sanitize Snapshot sanitization callback.
* *
* \return Newly allocated string with JSON message. * \return Newly allocated string with JSON message.
* \return \c NULL on error. * \return \c NULL on error.
* \return \c NULL if JSON format is not supported. * \return \c NULL if JSON format is not supported.
*/ */
struct ast_json *stasis_message_to_json(struct stasis_message *message, struct stasis_message_sanitizer *sanitize); struct ast_json *stasis_message_to_json(struct stasis_message *msg, struct stasis_message_sanitizer *sanitize);
/*! /*!
* \brief Build the AMI representation of the message. * \brief Build the AMI representation of the message.
@ -429,12 +429,21 @@ struct ast_json *stasis_message_to_json(struct stasis_message *message, struct s
* May return \c NULL, to indicate no representation. The returned object should * May return \c NULL, to indicate no representation. The returned object should
* be ao2_cleanup()'ed. * be ao2_cleanup()'ed.
* *
* \param message Message to convert to AMI. * \param msg Message to convert to AMI.
* \return \c NULL on error. * \return \c NULL on error.
* \return \c NULL if AMI format is not supported. * \return \c NULL if AMI format is not supported.
*/ */
struct ast_manager_event_blob *stasis_message_to_ami( struct ast_manager_event_blob *stasis_message_to_ami(struct stasis_message *msg);
struct stasis_message *message);
/*!
* \brief Determine if the given message can be converted to AMI.
*
* \param msg Message to see if can be converted to AMI.
*
* \retval 0 Cannot be converted
* \retval non-zero Can be converted
*/
int stasis_message_can_be_ami(struct stasis_message *msg);
/*! /*!
* \brief Build the \ref AstGenericEvents representation of the message. * \brief Build the \ref AstGenericEvents representation of the message.
@ -442,12 +451,11 @@ struct ast_manager_event_blob *stasis_message_to_ami(
* May return \c NULL, to indicate no representation. The returned object should * May return \c NULL, to indicate no representation. The returned object should
* be disposed of via \ref ast_event_destroy. * be disposed of via \ref ast_event_destroy.
* *
* \param message Message to convert to AMI. * \param msg Message to convert to AMI.
* \return \c NULL on error. * \return \c NULL on error.
* \return \c NULL if AMI format is not supported. * \return \c NULL if AMI format is not supported.
*/ */
struct ast_event *stasis_message_to_event( struct ast_event *stasis_message_to_event(struct stasis_message *msg);
struct stasis_message *message);
/*! /*!
* \brief A topic to which messages may be posted, and subscribers, well, subscribe * \brief A topic to which messages may be posted, and subscribers, well, subscribe

@ -1549,6 +1549,17 @@ static AST_RWLIST_HEAD_STATIC(manager_hooks, manager_custom_hook);
/*! \brief A container of event documentation nodes */ /*! \brief A container of event documentation nodes */
static AO2_GLOBAL_OBJ_STATIC(event_docs); static AO2_GLOBAL_OBJ_STATIC(event_docs);
static int __attribute__((format(printf, 9, 0))) __manager_event_sessions(
struct ao2_container *sessions,
int category,
const char *event,
int chancount,
struct ast_channel **chans,
const char *file,
int line,
const char *func,
const char *fmt,
...);
static enum add_filter_result manager_add_filter(const char *filter_pattern, struct ao2_container *whitefilters, struct ao2_container *blackfilters); static enum add_filter_result manager_add_filter(const char *filter_pattern, struct ao2_container *whitefilters, struct ao2_container *blackfilters);
static int match_filter(struct mansession *s, char *eventdata); static int match_filter(struct mansession *s, char *eventdata);
@ -1687,38 +1698,75 @@ struct ast_str *ast_manager_str_from_json_object(struct ast_json *blob, key_excl
return res; return res;
} }
#define manager_event_sessions(sessions, category, event, contents , ...) \
__manager_event_sessions(sessions, category, event, 0, NULL, __FILE__, __LINE__, __PRETTY_FUNCTION__, contents , ## __VA_ARGS__)
#define any_manager_listeners(sessions) \
((sessions && ao2_container_count(sessions)) || !AST_RWLIST_EMPTY(&manager_hooks))
static void manager_default_msg_cb(void *data, struct stasis_subscription *sub, static void manager_default_msg_cb(void *data, struct stasis_subscription *sub,
struct stasis_message *message) struct stasis_message *message)
{ {
struct ao2_container *sessions;
struct ast_manager_event_blob *ev; struct ast_manager_event_blob *ev;
if (!stasis_message_can_be_ami(message)) {
/* Not an AMI message; disregard */
return;
}
sessions = ao2_global_obj_ref(mgr_sessions);
if (!any_manager_listeners(sessions)) {
/* Nobody is listening */
ao2_cleanup(sessions);
return;
}
ev = stasis_message_to_ami(message); ev = stasis_message_to_ami(message);
if (!ev) { if (!ev) {
/* Not an AMI message; disregard */ /* Conversion failure */
ao2_cleanup(sessions);
return; return;
} }
manager_event(ev->event_flags, ev->manager_event, "%s", manager_event_sessions(sessions, ev->event_flags, ev->manager_event,
ev->extra_fields); "%s", ev->extra_fields);
ao2_ref(ev, -1); ao2_ref(ev, -1);
ao2_cleanup(sessions);
} }
static void manager_generic_msg_cb(void *data, struct stasis_subscription *sub, static void manager_generic_msg_cb(void *data, struct stasis_subscription *sub,
struct stasis_message *message) struct stasis_message *message)
{ {
struct ast_json_payload *payload = stasis_message_data(message); struct ast_json_payload *payload;
int class_type = ast_json_integer_get(ast_json_object_get(payload->json, "class_type")); int class_type;
const char *type = ast_json_string_get(ast_json_object_get(payload->json, "type")); const char *type;
struct ast_json *event = ast_json_object_get(payload->json, "event"); struct ast_json *event;
struct ast_str *event_buffer; struct ast_str *event_buffer;
struct ao2_container *sessions;
sessions = ao2_global_obj_ref(mgr_sessions);
if (!any_manager_listeners(sessions)) {
/* Nobody is listening */
ao2_cleanup(sessions);
return;
}
payload = stasis_message_data(message);
class_type = ast_json_integer_get(ast_json_object_get(payload->json, "class_type"));
type = ast_json_string_get(ast_json_object_get(payload->json, "type"));
event = ast_json_object_get(payload->json, "event");
event_buffer = ast_manager_str_from_json_object(event, NULL); event_buffer = ast_manager_str_from_json_object(event, NULL);
if (!event_buffer) { if (!event_buffer) {
ast_log(AST_LOG_WARNING, "Error while creating payload for event %s\n", type); ast_log(AST_LOG_WARNING, "Error while creating payload for event %s\n", type);
ao2_cleanup(sessions);
return; return;
} }
manager_event(class_type, type, "%s", ast_str_buffer(event_buffer)); manager_event_sessions(sessions, class_type, type,
"%s", ast_str_buffer(event_buffer));
ast_free(event_buffer); ast_free(event_buffer);
ao2_cleanup(sessions);
} }
void ast_manager_publish_event(const char *type, int class_type, struct ast_json *obj) void ast_manager_publish_event(const char *type, int class_type, struct ast_json *obj)
@ -6645,22 +6693,24 @@ static void append_channel_vars(struct ast_str **pbuf, struct ast_channel *chan)
AST_THREADSTORAGE(manager_event_buf); AST_THREADSTORAGE(manager_event_buf);
#define MANAGER_EVENT_BUF_INITSIZE 256 #define MANAGER_EVENT_BUF_INITSIZE 256
int __ast_manager_event_multichan(int category, const char *event, int chancount, static int __attribute__((format(printf, 9, 0))) __manager_event_sessions_va(
struct ast_channel **chans, const char *file, int line, const char *func, struct ao2_container *sessions,
const char *fmt, ...) int category,
const char *event,
int chancount,
struct ast_channel **chans,
const char *file,
int line,
const char *func,
const char *fmt,
va_list ap)
{ {
RAII_VAR(struct ao2_container *, sessions, ao2_global_obj_ref(mgr_sessions), ao2_cleanup);
struct ast_str *auth = ast_str_alloca(MAX_AUTH_PERM_STRING); struct ast_str *auth = ast_str_alloca(MAX_AUTH_PERM_STRING);
const char *cat_str; const char *cat_str;
va_list ap;
struct timeval now; struct timeval now;
struct ast_str *buf; struct ast_str *buf;
int i; int i;
if (!(sessions && ao2_container_count(sessions)) && AST_RWLIST_EMPTY(&manager_hooks)) {
return 0;
}
buf = ast_str_thread_get(&manager_event_buf, MANAGER_EVENT_BUF_INITSIZE); buf = ast_str_thread_get(&manager_event_buf, MANAGER_EVENT_BUF_INITSIZE);
if (!buf) { if (!buf) {
return -1; return -1;
@ -6696,9 +6746,7 @@ int __ast_manager_event_multichan(int category, const char *event, int chancount
ast_config_AST_SYSTEM_NAME); ast_config_AST_SYSTEM_NAME);
} }
va_start(ap, fmt);
ast_str_append_va(&buf, 0, fmt, ap); ast_str_append_va(&buf, 0, fmt, ap);
va_end(ap);
for (i = 0; i < chancount; i++) { for (i = 0; i < chancount; i++) {
append_channel_vars(&buf, chans[i]); append_channel_vars(&buf, chans[i]);
} }
@ -6744,6 +6792,50 @@ int __ast_manager_event_multichan(int category, const char *event, int chancount
return 0; return 0;
} }
static int __attribute__((format(printf, 9, 0))) __manager_event_sessions(
struct ao2_container *sessions,
int category,
const char *event,
int chancount,
struct ast_channel **chans,
const char *file,
int line,
const char *func,
const char *fmt,
...)
{
va_list ap;
int res;
va_start(ap, fmt);
res = __manager_event_sessions_va(sessions, category, event, chancount, chans,
file, line, func, fmt, ap);
va_end(ap);
return res;
}
int __ast_manager_event_multichan(int category, const char *event, int chancount,
struct ast_channel **chans, const char *file, int line, const char *func,
const char *fmt, ...)
{
struct ao2_container *sessions = ao2_global_obj_ref(mgr_sessions);
va_list ap;
int res;
if (!any_manager_listeners(sessions)) {
/* Nobody is listening */
ao2_cleanup(sessions);
return 0;
}
va_start(ap, fmt);
res = __manager_event_sessions_va(sessions, category, event, chancount, chans,
file, line, func, fmt, ap);
va_end(ap);
ao2_cleanup(sessions);
return res;
}
/*! \brief /*! \brief
* support functions to register/unregister AMI action handlers, * support functions to register/unregister AMI action handlers,
*/ */

@ -170,17 +170,17 @@ const struct timeval *stasis_message_timestamp(const struct stasis_message *msg)
return &msg->timestamp; return &msg->timestamp;
} }
#define INVOKE_VIRTUAL(fn, ...) \ #define INVOKE_VIRTUAL(fn, ...) \
({ \ ({ \
if (msg == NULL) { \ if (!msg) { \
return NULL; \ return NULL; \
} \ } \
ast_assert(msg->type != NULL); \ ast_assert(msg->type != NULL); \
ast_assert(msg->type->vtable != NULL); \ ast_assert(msg->type->vtable != NULL); \
if (msg->type->vtable->fn == NULL) { \ if (!msg->type->vtable->fn) { \
return NULL; \ return NULL; \
} \ } \
msg->type->vtable->fn(__VA_ARGS__); \ msg->type->vtable->fn(__VA_ARGS__); \
}) })
struct ast_manager_event_blob *stasis_message_to_ami(struct stasis_message *msg) struct ast_manager_event_blob *stasis_message_to_ami(struct stasis_message *msg)
@ -199,3 +199,18 @@ struct ast_event *stasis_message_to_event(struct stasis_message *msg)
{ {
return INVOKE_VIRTUAL(to_event, msg); return INVOKE_VIRTUAL(to_event, msg);
} }
#define HAS_VIRTUAL(fn, msg) \
({ \
if (!msg) { \
return 0; \
} \
ast_assert(msg->type != NULL); \
ast_assert(msg->type->vtable != NULL); \
!!msg->type->vtable->fn; \
})
int stasis_message_can_be_ami(struct stasis_message *msg)
{
return HAS_VIRTUAL(to_ami, msg);
}

Loading…
Cancel
Save