VICIfast Research, Report 01: The Two Bottlenecks Inside a Busy VICIdial Server Test harness. FOR A TEST SERVER, NOT A LIVE FLOOR. Read it before you apply it. The experimental Asterisk 22.5.1 patch from section 7.6 (eventthreads in manager.conf). Test servers only. diff -ru a/include/asterisk/stasis_message_router.h b/include/asterisk/stasis_message_router.h --- a/include/asterisk/stasis_message_router.h 2026-10-22 01:50:15.749310435 +0000 +++ b/include/asterisk/stasis_message_router.h 2026-10-22 01:50:28.677337612 +0000 @@ -129,6 +129,34 @@ struct stasis_message *message); /*! + * \brief Receives a routed message in place of the route's callback. + * + * \param data The pointer given to \ref stasis_message_router_set_dispatcher + * \param callback The route's callback, to be called later with the arguments below + * \param cb_data The route's data + * \param sub The router's subscription + * \param message The message (borrowed: take a reference to keep it) + */ +typedef void (*stasis_message_router_dispatcher)(void *data, + stasis_subscription_cb callback, void *cb_data, + struct stasis_subscription *sub, struct stasis_message *message); + +/*! + * \brief Hand routed messages to a dispatcher instead of calling routes inline. + * + * The router still picks the route on its subscription's thread, in order; + * the dispatcher decides where and when the route's callback runs. The + * subscription's final message is always handled inline. NULL restores + * inline dispatch. + * + * \param router Router + * \param dispatcher Dispatcher, or NULL + * \param data Passed to the dispatcher + */ +void stasis_message_router_set_dispatcher(struct stasis_message_router *router, + stasis_message_router_dispatcher dispatcher, void *data); + +/*! * \brief Set the high and low alert water marks of the stasis message router. * \since 13.10.0 * diff -ru a/main/manager.c b/main/manager.c --- a/main/manager.c 2026-10-22 01:50:15.749310435 +0000 +++ b/main/manager.c 2026-10-22 02:29:46.555072517 +0000 @@ -102,6 +102,8 @@ #include "asterisk/format_cache.h" #include "asterisk/translate.h" #include "asterisk/taskprocessor.h" +#include "asterisk/threadpool.h" +#include "asterisk/vector.h" #include "asterisk/message.h" /*! \addtogroup Group_AMI AMI functions @@ -175,6 +177,8 @@ static int authlimit; static char *manager_channelvars; static char *manager_disabledevents; +/*! manager.conf [general] eventthreads: AMI event formatting threads, read at start (0 = inline) */ +static int manager_event_threads; #define DEFAULT_REALM "asterisk" static char global_realm[MAXHOSTNAMELEN]; /*!< Default realm */ @@ -7535,6 +7539,201 @@ ao2_ref(vars, -1); } +/* + * Parallel AMI event formatting (manager.conf [general] eventthreads = N). + * + * Stock, the manager's message router turns every stasis message into AMI + * text on its one subscription thread; on a busy system that thread is the + * ceiling (its queue grows and every AMI client sees events late). With N > 0 + * the router thread only queues each message, in arrival order, as a batch and + * hands it to a pool of N threads. They run the same route callbacks; the + * events those callbacks build are held in the batch instead of being + * appended. One committer then appends batches to the event queue strictly in + * arrival order, so AMI clients see the same events in the same order as + * stock - only the formatting runs in parallel. + */ +struct deferred_event { + int category; + int event_name_hash; + char *event; + char *text; +}; + +struct event_batch { + stasis_subscription_cb callback; + void *cb_data; + struct stasis_subscription *sub; + struct stasis_message *message; + /*! Set once the callback has run; read and written under the list lock */ + int done; + AST_VECTOR(, struct deferred_event) events; + AST_LIST_ENTRY(event_batch) list; +}; + +/*! Batches in the order the router received their messages */ +static AST_LIST_HEAD_STATIC(event_batches, event_batch); +static unsigned long event_batches_pending; +static unsigned long event_batches_pending_max; +static unsigned long event_batches_total; +static int event_threads_running; +static struct ast_threadpool *event_format_pool; +static struct ast_taskprocessor *event_committer; + +/*! The batch the current thread is formatting, if any */ +AST_THREADSTORAGE_CUSTOM_SCOPE(event_batch_current, NULL, THREADSTORAGE_RAW_CLEANUP, static); + +static void manager_event_commit(struct ao2_container *sessions, int category, + int event_name_hash, const char *event, const char *text); + +/*! Committer: append every finished batch at the head of the list, in order. */ +static int event_batches_commit(void *unused) +{ + for (;;) { + struct event_batch *batch; + struct ao2_container *sessions; + size_t i; + + AST_LIST_LOCK(&event_batches); + batch = AST_LIST_FIRST(&event_batches); + if (!batch || !batch->done) { + AST_LIST_UNLOCK(&event_batches); + break; + } + AST_LIST_REMOVE_HEAD(&event_batches, list); + event_batches_pending--; + AST_LIST_UNLOCK(&event_batches); + + sessions = ao2_global_obj_ref(mgr_sessions); + for (i = 0; i < AST_VECTOR_SIZE(&batch->events); i++) { + struct deferred_event *de = AST_VECTOR_GET_ADDR(&batch->events, i); + + if (sessions) { + manager_event_commit(sessions, de->category, de->event_name_hash, de->event, de->text); + } + ast_free(de->event); + ast_free(de->text); + } + ao2_cleanup(sessions); + AST_VECTOR_FREE(&batch->events); + ast_free(batch); + } + return 0; +} + +/*! Run one batch's route callback with its events held back, then queue a commit. */ +static void event_batch_run(struct event_batch *batch) +{ + ast_threadstorage_set_ptr(&event_batch_current, batch); + batch->callback(batch->cb_data, batch->sub, batch->message); + ast_threadstorage_set_ptr(&event_batch_current, NULL); + ao2_cleanup(batch->message); + batch->message = NULL; + + AST_LIST_LOCK(&event_batches); + batch->done = 1; + AST_LIST_UNLOCK(&event_batches); + + /* Only the committer appends, so order holds. If this push fails the batch + * waits, done, for the next successful one. */ + if (ast_taskprocessor_push(event_committer, event_batches_commit, NULL)) { + ast_log(LOG_WARNING, "Could not queue an AMI event commit\n"); + } +} + +static int event_batch_work(void *data) +{ + event_batch_run(data); + return 0; +} + +/*! Router dispatcher: runs on the router's thread, in arrival order. */ +static void manager_event_dispatch(void *data, stasis_subscription_cb callback, + void *cb_data, struct stasis_subscription *sub, struct stasis_message *message) +{ + struct event_batch *batch = ast_calloc(1, sizeof(*batch)); + + if (!batch || AST_VECTOR_INIT(&batch->events, 2)) { + ast_free(batch); + callback(cb_data, sub, message); + return; + } + batch->callback = callback; + batch->cb_data = cb_data; + batch->sub = sub; + batch->message = ao2_bump(message); + + AST_LIST_LOCK(&event_batches); + AST_LIST_INSERT_TAIL(&event_batches, batch, list); + if (++event_batches_pending > event_batches_pending_max) { + event_batches_pending_max = event_batches_pending; + } + event_batches_total++; + AST_LIST_UNLOCK(&event_batches); + + if (ast_threadpool_push(event_format_pool, event_batch_work, batch)) { + /* Format here instead; the committer still appends it in order. */ + event_batch_run(batch); + } +} + +static void manager_event_threads_start(void) +{ + struct ast_threadpool_options options = { + .version = AST_THREADPOOL_OPTIONS_VERSION, + .idle_timeout = 0, + .auto_increment = 0, + .initial_size = manager_event_threads, + .max_size = manager_event_threads, + }; + + if (manager_event_threads <= 0 || event_threads_running || !stasis_router) { + return; + } + event_committer = ast_taskprocessor_get("manager:eventcommit", TPS_REF_DEFAULT); + if (!event_committer) { + ast_log(LOG_ERROR, "AMI event threads: no committer; formatting stays inline\n"); + return; + } + event_format_pool = ast_threadpool_create("manager:eventfmt", NULL, &options); + if (!event_format_pool) { + ast_taskprocessor_unreference(event_committer); + event_committer = NULL; + ast_log(LOG_ERROR, "AMI event threads: no thread pool; formatting stays inline\n"); + return; + } + event_threads_running = 1; + stasis_message_router_set_dispatcher(stasis_router, manager_event_dispatch, NULL); + ast_log(LOG_NOTICE, "AMI events formatted on %d threads, appended in arrival order\n", + manager_event_threads); +} + +/*! Back to inline formatting, after letting held batches drain (up to 2 s). */ +static void manager_event_threads_stop(void) +{ + int i; + + if (!event_threads_running) { + return; + } + stasis_message_router_set_dispatcher(stasis_router, NULL, NULL); + for (i = 0; i < 200; i++) { + int empty; + + AST_LIST_LOCK(&event_batches); + empty = AST_LIST_EMPTY(&event_batches); + AST_LIST_UNLOCK(&event_batches); + if (empty) { + break; + } + usleep(10000); + } + ast_threadpool_shutdown(event_format_pool); + event_format_pool = NULL; + ast_taskprocessor_unreference(event_committer); + event_committer = NULL; + event_threads_running = 0; +} + /* XXX see if can be moved inside the function */ AST_THREADSTORAGE(manager_event_buf); #define MANAGER_EVENT_BUF_INITSIZE 256 @@ -7610,7 +7809,36 @@ event_name_hash = ast_str_hash(event); - append_event(ast_str_buffer(buf), event_name_hash, category); + if (event_threads_running) { + struct event_batch *batch = ast_threadstorage_get_ptr(&event_batch_current); + + if (batch) { + /* Formatted on an event thread: the committer appends it, in order. */ + struct deferred_event de = { + .category = category, + .event_name_hash = event_name_hash, + .event = ast_strdup(event), + .text = ast_strdup(ast_str_buffer(buf)), + }; + + if (de.event && de.text && !AST_VECTOR_APPEND(&batch->events, de)) { + return 0; + } + ast_free(de.event); + ast_free(de.text); + /* Out of memory: append it now, out of order rather than lost. */ + } + } + + manager_event_commit(sessions, category, event_name_hash, event, ast_str_buffer(buf)); + return 0; +} + +/*! Append a built event to the queue, wake the sessions and run the hooks. */ +static void manager_event_commit(struct ao2_container *sessions, int category, + int event_name_hash, const char *event, const char *text) +{ + append_event(text, event_name_hash, category); /* Wake up any sleeping sessions */ if (sessions) { @@ -7619,6 +7847,16 @@ iter = ao2_iterator_init(sessions, 0); while ((session = ao2_iterator_next(&iter))) { + /* Wake only sessions that can receive this event: process_events() + * delivers nothing else, so the rest would wake for nothing. On a + * VICIdial box ~85 sendcron sessions (read = command) used to wake + * for every call event. */ + if (category != EVENT_FLAG_SHUTDOWN + && ((session->readperm & category) != category + || (session->send_events & category) != category)) { + unref_mansession(session); + continue; + } ast_mutex_lock(&session->notify_lock); if (session->waiting_thread != AST_PTHREADT_NULL) { pthread_kill(session->waiting_thread, SIGURG); @@ -7638,15 +7876,15 @@ if (category != EVENT_FLAG_SHUTDOWN && !AST_RWLIST_EMPTY(&manager_hooks)) { struct manager_custom_hook *hook; + /* Hooks take a writable body: a copy, as they were given a scratch buffer before. */ + char *body = ast_strdupa(text); AST_RWLIST_RDLOCK(&manager_hooks); AST_RWLIST_TRAVERSE(&manager_hooks, hook, list) { - hook->helper(category, event, ast_str_buffer(buf)); + hook->helper(category, event, body); } AST_RWLIST_UNLOCK(&manager_hooks); } - - return 0; } static int __attribute__((format(printf, 9, 0))) __manager_event_sessions( @@ -9164,6 +9402,11 @@ ast_cli(a->fd, FORMAT, "Timestamp events:", AST_CLI_YESNO(timestampevents)); ast_cli(a->fd, FORMAT3, "Channel vars:", S_OR(manager_channelvars, "")); ast_cli(a->fd, FORMAT3, "Disabled events:", S_OR(manager_disabledevents, "")); + ast_cli(a->fd, FORMAT2, "Event threads:", event_threads_running ? manager_event_threads : 0); + if (event_threads_running) { + ast_cli(a->fd, " %-25.25s %lu now, %lu max, %lu total\n", "Event batches:", + event_batches_pending, event_batches_pending_max, event_batches_total); + } ast_cli(a->fd, FORMAT, "Debug:", AST_CLI_YESNO(manager_debug)); #undef FORMAT #undef FORMAT2 @@ -9511,6 +9754,7 @@ test_suite_forwarder = NULL; #endif + manager_event_threads_stop(); if (stasis_router) { stasis_message_router_unsubscribe_and_join(stasis_router); stasis_router = NULL; @@ -9844,6 +10088,18 @@ load_channelvars(var); } else if (!strcasecmp(var->name, "disabledevents")) { load_disabledevents(var); + } else if (!strcasecmp(var->name, "eventthreads")) { + int threads; + + if (sscanf(val, "%30d", &threads) != 1 || threads < 0 || threads > 64) { + ast_log(LOG_WARNING, "Invalid eventthreads '%s' in manager.conf [general] (0-64); using 0\n", val); + threads = 0; + } + if (event_threads_running && threads != manager_event_threads) { + ast_log(LOG_NOTICE, "manager.conf eventthreads takes effect at the next restart\n"); + } else { + manager_event_threads = threads; + } } else { ast_log(LOG_NOTICE, "Invalid keyword <%s> = <%s> in manager.conf [general]\n", var->name, val); @@ -9855,6 +10111,7 @@ ast_log(LOG_ERROR, "Manager subscription error\n"); return -1; } + manager_event_threads_start(); } ast_sockaddr_copy(&amis_desc_local_address_tmp, &amis_desc.local_address); diff -ru a/main/stasis_message_router.c b/main/stasis_message_router.c --- a/main/stasis_message_router.c 2026-10-22 01:50:15.749310435 +0000 +++ b/main/stasis_message_router.c 2026-10-22 01:50:36.081353205 +0000 @@ -135,6 +135,10 @@ struct route_table cache_routes; /*! Route of last resort */ struct stasis_message_route default_route; + /*! When set, routed messages go here instead of to the route inline */ + stasis_message_router_dispatcher dispatcher; + /*! Data for the dispatcher */ + void *dispatcher_data; }; static void router_dtor(void *obj) @@ -192,16 +196,41 @@ { struct stasis_message_router *router = data; struct stasis_message_route route; + stasis_message_router_dispatcher dispatcher; + void *dispatcher_data; + int final = stasis_subscription_final_message(sub, message); if (find_route(router, message, &route) == 0) { - route.callback(route.data, sub, message); + ao2_lock(router); + dispatcher = router->dispatcher; + dispatcher_data = router->dispatcher_data; + ao2_unlock(router); + + if (dispatcher && !final) { + dispatcher(dispatcher_data, route.callback, route.data, sub, message); + } else { + route.callback(route.data, sub, message); + } } - if (stasis_subscription_final_message(sub, message)) { + if (final) { ao2_cleanup(router); } } +void stasis_message_router_set_dispatcher(struct stasis_message_router *router, + stasis_message_router_dispatcher dispatcher, void *data) +{ + if (!router) { + return; + } + + ao2_lock(router); + router->dispatcher = dispatcher; + router->dispatcher_data = data; + ao2_unlock(router); +} + static struct stasis_message_router *stasis_message_router_create_internal( struct stasis_topic *topic, int use_thread_pool, const char *file, int lineno, const char *func)