2013-03-08 15:15:13 +00:00
|
|
|
/*
|
|
|
|
* Asterisk -- An open source telephony toolkit.
|
|
|
|
*
|
|
|
|
* Copyright (C) 2013, Digium, Inc.
|
|
|
|
*
|
|
|
|
* David M. Lee, II <dlee@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.
|
|
|
|
*/
|
|
|
|
|
|
|
|
/*!
|
|
|
|
* \file \brief Test Stasis message bus.
|
|
|
|
*
|
|
|
|
* \author\verbatim David M. Lee, II <dlee@digium.com> \endverbatim
|
|
|
|
*
|
|
|
|
* \ingroup tests
|
|
|
|
*/
|
|
|
|
|
|
|
|
/*** MODULEINFO
|
|
|
|
<depend>TEST_FRAMEWORK</depend>
|
|
|
|
<support_level>core</support_level>
|
|
|
|
***/
|
|
|
|
|
|
|
|
#include "asterisk.h"
|
|
|
|
|
|
|
|
ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
|
|
|
|
|
|
|
|
#include "asterisk/astobj2.h"
|
|
|
|
#include "asterisk/module.h"
|
|
|
|
#include "asterisk/stasis.h"
|
2013-03-15 17:35:16 +00:00
|
|
|
#include "asterisk/stasis_message_router.h"
|
2013-03-08 15:15:13 +00:00
|
|
|
#include "asterisk/test.h"
|
|
|
|
|
|
|
|
static const char *test_category = "/stasis/core/";
|
|
|
|
|
2013-11-22 23:57:45 +00:00
|
|
|
static struct ast_json *fake_json(struct stasis_message *message, const struct stasis_message_sanitizer *sanitize)
|
2013-06-11 15:46:35 +00:00
|
|
|
{
|
|
|
|
const char *text = stasis_message_data(message);
|
|
|
|
|
|
|
|
return ast_json_string_create(text);
|
|
|
|
}
|
|
|
|
|
|
|
|
static struct ast_manager_event_blob *fake_ami(struct stasis_message *message)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct ast_manager_event_blob *, res, NULL, ao2_cleanup);
|
|
|
|
const char *text = stasis_message_data(message);
|
|
|
|
|
|
|
|
res = ast_manager_event_blob_create(EVENT_FLAG_TEST, "FakeMI",
|
2013-08-08 20:52:49 +00:00
|
|
|
"Message: %s\r\n", text);
|
2013-06-11 15:46:35 +00:00
|
|
|
|
|
|
|
if (res == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ao2_ref(res, +1);
|
|
|
|
return res;
|
|
|
|
}
|
|
|
|
|
|
|
|
static struct stasis_message_vtable fake_vtable = {
|
|
|
|
.to_json = fake_json,
|
|
|
|
.to_ami = fake_ami
|
|
|
|
};
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
AST_TEST_DEFINE(message_type)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, uut, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test basic message_type functions";
|
|
|
|
info->description = "Test basic message_type functions";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create(NULL, NULL, NULL) == STASIS_MESSAGE_TYPE_ERROR);
|
|
|
|
ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &uut) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_test_validate(test, 0 == strcmp(stasis_message_type_name(uut), "SomeMessage"));
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(message)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
|
2014-03-07 20:41:13 +00:00
|
|
|
RAII_VAR(struct stasis_message *, uut1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, uut2, NULL, ao2_cleanup);
|
2013-03-08 15:15:13 +00:00
|
|
|
RAII_VAR(char *, data, NULL, ao2_cleanup);
|
|
|
|
char *expected = "SomeData";
|
|
|
|
struct timeval expected_timestamp;
|
|
|
|
struct timeval time_diff;
|
2014-03-07 20:41:13 +00:00
|
|
|
struct ast_eid foreign_eid;
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test basic message functions";
|
|
|
|
info->description = "Test basic message functions";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
memset(&foreign_eid, 0xFF, sizeof(foreign_eid));
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
ast_test_validate(test, NULL == stasis_message_create_full(NULL, NULL, NULL));
|
|
|
|
ast_test_validate(test, NULL == stasis_message_create_full(type, NULL, NULL));
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
data = ao2_alloc(strlen(expected) + 1, NULL);
|
2014-03-07 20:41:13 +00:00
|
|
|
strcpy(data, expected);/* Safe */
|
2013-03-08 15:15:13 +00:00
|
|
|
expected_timestamp = ast_tvnow();
|
2014-03-07 20:41:13 +00:00
|
|
|
uut1 = stasis_message_create_full(type, data, &foreign_eid);
|
|
|
|
uut2 = stasis_message_create_full(type, data, NULL);
|
|
|
|
|
|
|
|
ast_test_validate(test, NULL != uut1);
|
|
|
|
ast_test_validate(test, NULL != uut2);
|
|
|
|
ast_test_validate(test, type == stasis_message_type(uut1));
|
|
|
|
ast_test_validate(test, type == stasis_message_type(uut2));
|
|
|
|
ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut1)));
|
|
|
|
ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut2)));
|
|
|
|
ast_test_validate(test, NULL != stasis_message_eid(uut1));
|
|
|
|
ast_test_validate(test, NULL == stasis_message_eid(uut2));
|
|
|
|
ast_test_validate(test, !ast_eid_cmp(&foreign_eid, stasis_message_eid(uut1)));
|
|
|
|
|
|
|
|
ast_test_validate(test, 3 == ao2_ref(data, 0)); /* uut1 and uut2 have ref to data */
|
|
|
|
|
|
|
|
time_diff = ast_tvsub(*stasis_message_timestamp(uut1), expected_timestamp);
|
2013-03-08 15:15:13 +00:00
|
|
|
/* 10ms is certainly long enough for the two calls to complete */
|
|
|
|
ast_test_validate(test, time_diff.tv_sec == 0);
|
|
|
|
ast_test_validate(test, time_diff.tv_usec < 10000);
|
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
ao2_ref(uut1, -1);
|
|
|
|
uut1 = NULL;
|
|
|
|
ast_test_validate(test, 2 == ao2_ref(data, 0)); /* uut1 unreffed data */
|
|
|
|
ao2_ref(uut2, -1);
|
|
|
|
uut2 = NULL;
|
|
|
|
ast_test_validate(test, 1 == ao2_ref(data, 0)); /* uut2 unreffed data */
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
struct consumer {
|
|
|
|
ast_cond_t out;
|
|
|
|
struct stasis_message **messages_rxed;
|
|
|
|
size_t messages_rxed_len;
|
|
|
|
int ignore_subscriptions;
|
|
|
|
int complete;
|
|
|
|
};
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
static void consumer_dtor(void *obj)
|
|
|
|
{
|
2013-03-08 15:15:13 +00:00
|
|
|
struct consumer *consumer = obj;
|
|
|
|
|
|
|
|
ast_cond_destroy(&consumer->out);
|
|
|
|
|
|
|
|
while (consumer->messages_rxed_len > 0) {
|
|
|
|
ao2_cleanup(consumer->messages_rxed[--consumer->messages_rxed_len]);
|
|
|
|
}
|
|
|
|
ast_free(consumer->messages_rxed);
|
|
|
|
consumer->messages_rxed = NULL;
|
|
|
|
}
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
static struct consumer *consumer_create(int ignore_subscriptions)
|
|
|
|
{
|
|
|
|
struct consumer *consumer;
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
|
|
|
|
if (!consumer) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
consumer->ignore_subscriptions = ignore_subscriptions;
|
2014-02-26 23:35:20 +00:00
|
|
|
consumer->messages_rxed = ast_malloc(sizeof(*consumer->messages_rxed));
|
2013-03-08 15:15:13 +00:00
|
|
|
if (!consumer->messages_rxed) {
|
2014-02-26 23:35:20 +00:00
|
|
|
ao2_cleanup(consumer);
|
2013-03-08 15:15:13 +00:00
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ast_cond_init(&consumer->out, NULL);
|
|
|
|
|
|
|
|
return consumer;
|
|
|
|
}
|
|
|
|
|
Multiple revisions 399887,400138,400178,400180-400181
........
r399887 | dlee | 2013-09-26 10:41:47 -0500 (Thu, 26 Sep 2013) | 1 line
Minor performance bump by not allocate manager variable struct if we don't need it
........
r400138 | dlee | 2013-09-30 10:24:00 -0500 (Mon, 30 Sep 2013) | 23 lines
Stasis performance improvements
This patch addresses several performance problems that were found in
the initial performance testing of Asterisk 12.
The Stasis dispatch object was allocated as an AO2 object, even though
it has a very confined lifecycle. This was replaced with a straight
ast_malloc().
The Stasis message router was spending an inordinate amount of time
searching hash tables. In this case, most of our routers had 6 or
fewer routes in them to begin with. This was replaced with an array
that's searched linearly for the route.
We more heavily rely on AO2 objects in Asterisk 12, and the memset()
in ao2_ref() actually became noticeable on the profile. This was
#ifdef'ed to only run when AO2_DEBUG was enabled.
After being misled by an erroneous comment in taskprocessor.c during
profiling, the wrong comment was removed.
Review: https://reviewboard.asterisk.org/r/2873/
........
r400178 | dlee | 2013-09-30 13:26:27 -0500 (Mon, 30 Sep 2013) | 24 lines
Taskprocessor optimization; switch Stasis to use taskprocessors
This patch optimizes taskprocessor to use a semaphore for signaling,
which the OS can do a better job at managing contention and waiting
that we can with a mutex and condition.
The taskprocessor execution was also slightly optimized to reduce the
number of locks taken.
The only observable difference in the taskprocessor implementation is
that when the final reference to the taskprocessor goes away, it will
execute all tasks to completion instead of discarding the unexecuted
tasks.
For systems where unnamed semaphores are not supported, a really
simple semaphore implementation is provided. (Which gives identical
performance as the original taskprocessor implementation).
The way we ended up implementing Stasis caused the threadpool to be a
burden instead of a boost to performance. This was switched to just
use taskprocessors directly for subscriptions.
Review: https://reviewboard.asterisk.org/r/2881/
........
r400180 | dlee | 2013-09-30 13:39:34 -0500 (Mon, 30 Sep 2013) | 28 lines
Optimize how Stasis forwards are dispatched
This patch optimizes how forwards are dispatched in Stasis.
Originally, forwards were dispatched as subscriptions that are invoked
on the publishing thread. This did not account for the vast number of
forwards we would end up having in the system, and the amount of work it
would take to walk though the forward subscriptions.
This patch modifies Stasis so that rather than walking the tree of
forwards on every dispatch, when forwards and subscriptions are changed,
the subscriber list for every topic in the tree is changed.
This has a couple of benefits. First, this reduces the workload of
dispatching messages. It also reduces contention when dispatching to
different topics that happen to forward to the same aggregation topic
(as happens with all of the channel, bridge and endpoint topics).
Since forwards are no longer subscriptions, the bulk of this patch is
simply changing stasis_subscription objects to stasis_forward objects
(which, admittedly, I should have done in the first place.)
Since this required me to yet again put in a growing array, I finally
abstracted that out into a set of ast_vector macros in
asterisk/vector.h.
Review: https://reviewboard.asterisk.org/r/2883/
........
r400181 | dlee | 2013-09-30 13:48:57 -0500 (Mon, 30 Sep 2013) | 28 lines
Remove dispatch object allocation from Stasis publishing
While looking for areas for performance improvement, I realized that an
unused feature in Stasis was negatively impacting performance.
When a message is sent to a subscriber, a dispatch object is allocated
for the dispatch, containing the topic the message was published to, the
subscriber the message is being sent to, and the message itself.
The topic is actually unused by any subscriber in Asterisk today. And
the subscriber is associated with the taskprocessor the message is being
dispatched to.
First, this patch removes the unused topic parameter from Stasis
subscription callbacks.
Second, this patch introduces the concept of taskprocessor local data,
data that may be set on a taskprocessor and provided along with the data
pointer when a task is pushed using the ast_taskprocessor_push_local()
call. This allows the task to have both data specific to that
taskprocessor, in addition to data specific to that invocation.
With those two changes, the dispatch object can be removed completely,
and the message is simply refcounted and sent directly to the
taskprocessor.
Review: https://reviewboard.asterisk.org/r/2884/
........
Merged revisions 399887,400138,400178,400180-400181 from http://svn.asterisk.org/svn/asterisk/branches/12
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@400186 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-09-30 18:55:27 +00:00
|
|
|
static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_message *message)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
struct consumer *consumer = data;
|
|
|
|
RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
|
2014-02-26 23:35:20 +00:00
|
|
|
SCOPED_AO2LOCK(lock, consumer);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2013-03-28 15:45:18 +00:00
|
|
|
if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change_type()) {
|
2013-03-08 15:15:13 +00:00
|
|
|
++consumer->messages_rxed_len;
|
|
|
|
consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
|
|
|
|
ast_assert(consumer->messages_rxed != NULL);
|
|
|
|
consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
|
|
|
|
ao2_ref(message, +1);
|
|
|
|
}
|
|
|
|
|
|
|
|
if (stasis_subscription_final_message(sub, message)) {
|
|
|
|
consumer->complete = 1;
|
|
|
|
consumer_needs_cleanup = consumer;
|
|
|
|
}
|
|
|
|
|
|
|
|
ast_cond_signal(&consumer->out);
|
|
|
|
}
|
|
|
|
|
2014-01-12 22:07:01 +00:00
|
|
|
static void consumer_exec_sync(void *data, struct stasis_subscription *sub, struct stasis_message *message)
|
|
|
|
{
|
|
|
|
struct consumer *consumer = data;
|
|
|
|
RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
|
2014-02-26 23:35:20 +00:00
|
|
|
SCOPED_AO2LOCK(lock, consumer);
|
2014-01-12 22:07:01 +00:00
|
|
|
|
|
|
|
if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change_type()) {
|
|
|
|
++consumer->messages_rxed_len;
|
|
|
|
consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
|
|
|
|
ast_assert(consumer->messages_rxed != NULL);
|
|
|
|
consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
|
|
|
|
ao2_ref(message, +1);
|
|
|
|
}
|
|
|
|
|
|
|
|
if (stasis_subscription_final_message(sub, message)) {
|
|
|
|
consumer->complete = 1;
|
|
|
|
consumer_needs_cleanup = consumer;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
static int consumer_wait_for(struct consumer *consumer, size_t expected_len)
|
|
|
|
{
|
|
|
|
struct timeval start = ast_tvnow();
|
|
|
|
struct timespec end = {
|
|
|
|
.tv_sec = start.tv_sec + 30,
|
|
|
|
.tv_nsec = start.tv_usec * 1000
|
|
|
|
};
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
SCOPED_AO2LOCK(lock, consumer);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
while (consumer->messages_rxed_len < expected_len) {
|
2014-02-26 23:35:20 +00:00
|
|
|
int r = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
if (r == ETIMEDOUT) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
ast_assert(r == 0); /* Not expecting any othet types of errors */
|
|
|
|
}
|
|
|
|
return consumer->messages_rxed_len;
|
|
|
|
}
|
|
|
|
|
|
|
|
static int consumer_wait_for_completion(struct consumer *consumer)
|
|
|
|
{
|
|
|
|
struct timeval start = ast_tvnow();
|
|
|
|
struct timespec end = {
|
2013-07-23 13:39:50 +00:00
|
|
|
.tv_sec = start.tv_sec + 3,
|
2013-03-08 15:15:13 +00:00
|
|
|
.tv_nsec = start.tv_usec * 1000
|
|
|
|
};
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
SCOPED_AO2LOCK(lock, consumer);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
while (!consumer->complete) {
|
2014-02-26 23:35:20 +00:00
|
|
|
int r = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
if (r == ETIMEDOUT) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
ast_assert(r == 0); /* Not expecting any othet types of errors */
|
|
|
|
}
|
|
|
|
return consumer->complete;
|
|
|
|
}
|
|
|
|
|
|
|
|
static int consumer_should_stay(struct consumer *consumer, size_t expected_len)
|
|
|
|
{
|
|
|
|
struct timeval start = ast_tvnow();
|
|
|
|
struct timeval diff = {
|
|
|
|
.tv_sec = 0,
|
|
|
|
.tv_usec = 100000 /* wait for 100ms */
|
|
|
|
};
|
|
|
|
struct timeval end_tv = ast_tvadd(start, diff);
|
|
|
|
struct timespec end = {
|
|
|
|
.tv_sec = end_tv.tv_sec,
|
|
|
|
.tv_nsec = end_tv.tv_usec * 1000
|
|
|
|
};
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
SCOPED_AO2LOCK(lock, consumer);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
while (consumer->messages_rxed_len == expected_len) {
|
2014-02-26 23:35:20 +00:00
|
|
|
int r = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
if (r == ETIMEDOUT) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
ast_assert(r == 0); /* Not expecting any othet types of errors */
|
|
|
|
}
|
|
|
|
return consumer->messages_rxed_len;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(subscription_messages)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(char *, expected_uniqueid, NULL, ast_free);
|
|
|
|
int complete;
|
|
|
|
struct stasis_subscription_change *change;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test subscribe/unsubscribe messages";
|
|
|
|
info->description = "Test subscribe/unsubscribe messages";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
consumer = consumer_create(0);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
|
|
|
|
uut = stasis_subscribe(topic, consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
|
|
|
|
|
2013-03-15 13:04:52 +00:00
|
|
|
uut = stasis_unsubscribe(uut);
|
2013-03-08 15:15:13 +00:00
|
|
|
complete = consumer_wait_for_completion(consumer);
|
|
|
|
ast_test_validate(test, 1 == complete);
|
|
|
|
|
|
|
|
ast_test_validate(test, 2 == consumer->messages_rxed_len);
|
2013-03-28 15:45:18 +00:00
|
|
|
ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[0]));
|
|
|
|
ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[1]));
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
change = stasis_message_data(consumer->messages_rxed[0]);
|
|
|
|
ast_test_validate(test, topic == change->topic);
|
|
|
|
ast_test_validate(test, 0 == strcmp("Subscribe", change->description));
|
|
|
|
ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
|
|
|
|
|
|
|
|
change = stasis_message_data(consumer->messages_rxed[1]);
|
|
|
|
ast_test_validate(test, topic == change->topic);
|
|
|
|
ast_test_validate(test, 0 == strcmp("Unsubscribe", change->description));
|
|
|
|
ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(publish)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
const char *actual;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
2014-01-12 22:07:01 +00:00
|
|
|
info->summary = "Test publishing";
|
|
|
|
info->description = "Test publishing";
|
2013-03-08 15:15:13 +00:00
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
|
|
|
|
uut = stasis_subscribe(topic, consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
test_message = stasis_message_create(test_message_type, test_data);
|
|
|
|
|
|
|
|
stasis_publish(topic, test_message);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
actual = stasis_message_data(consumer->messages_rxed[0]);
|
|
|
|
ast_test_validate(test, test_data == actual);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2014-01-12 22:07:01 +00:00
|
|
|
AST_TEST_DEFINE(publish_sync)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
const char *actual;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test synchronous publishing";
|
|
|
|
info->description = "Test synchronous publishing";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
|
|
|
|
uut = stasis_subscribe(topic, consumer_exec_sync, consumer);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2014-01-12 22:07:01 +00:00
|
|
|
test_message = stasis_message_create(test_message_type, test_data);
|
|
|
|
|
|
|
|
stasis_publish_sync(uut, test_message);
|
|
|
|
|
|
|
|
actual_len = consumer->messages_rxed_len;
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
actual = stasis_message_data(consumer->messages_rxed[0]);
|
|
|
|
ast_test_validate(test, test_data == actual);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
AST_TEST_DEFINE(unsubscribe_stops_messages)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test simple subscriptions";
|
|
|
|
info->description = "Test simple subscriptions";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
|
|
|
|
uut = stasis_subscribe(topic, consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
2013-03-15 13:04:52 +00:00
|
|
|
uut = stasis_unsubscribe(uut);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
test_message = stasis_message_create(test_message_type, test_data);
|
|
|
|
|
|
|
|
stasis_publish(topic, test_message);
|
|
|
|
|
|
|
|
actual_len = consumer_should_stay(consumer, 0);
|
|
|
|
ast_test_validate(test, 0 == actual_len);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(forward)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
RAII_VAR(struct consumer *, parent_consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
|
Multiple revisions 399887,400138,400178,400180-400181
........
r399887 | dlee | 2013-09-26 10:41:47 -0500 (Thu, 26 Sep 2013) | 1 line
Minor performance bump by not allocate manager variable struct if we don't need it
........
r400138 | dlee | 2013-09-30 10:24:00 -0500 (Mon, 30 Sep 2013) | 23 lines
Stasis performance improvements
This patch addresses several performance problems that were found in
the initial performance testing of Asterisk 12.
The Stasis dispatch object was allocated as an AO2 object, even though
it has a very confined lifecycle. This was replaced with a straight
ast_malloc().
The Stasis message router was spending an inordinate amount of time
searching hash tables. In this case, most of our routers had 6 or
fewer routes in them to begin with. This was replaced with an array
that's searched linearly for the route.
We more heavily rely on AO2 objects in Asterisk 12, and the memset()
in ao2_ref() actually became noticeable on the profile. This was
#ifdef'ed to only run when AO2_DEBUG was enabled.
After being misled by an erroneous comment in taskprocessor.c during
profiling, the wrong comment was removed.
Review: https://reviewboard.asterisk.org/r/2873/
........
r400178 | dlee | 2013-09-30 13:26:27 -0500 (Mon, 30 Sep 2013) | 24 lines
Taskprocessor optimization; switch Stasis to use taskprocessors
This patch optimizes taskprocessor to use a semaphore for signaling,
which the OS can do a better job at managing contention and waiting
that we can with a mutex and condition.
The taskprocessor execution was also slightly optimized to reduce the
number of locks taken.
The only observable difference in the taskprocessor implementation is
that when the final reference to the taskprocessor goes away, it will
execute all tasks to completion instead of discarding the unexecuted
tasks.
For systems where unnamed semaphores are not supported, a really
simple semaphore implementation is provided. (Which gives identical
performance as the original taskprocessor implementation).
The way we ended up implementing Stasis caused the threadpool to be a
burden instead of a boost to performance. This was switched to just
use taskprocessors directly for subscriptions.
Review: https://reviewboard.asterisk.org/r/2881/
........
r400180 | dlee | 2013-09-30 13:39:34 -0500 (Mon, 30 Sep 2013) | 28 lines
Optimize how Stasis forwards are dispatched
This patch optimizes how forwards are dispatched in Stasis.
Originally, forwards were dispatched as subscriptions that are invoked
on the publishing thread. This did not account for the vast number of
forwards we would end up having in the system, and the amount of work it
would take to walk though the forward subscriptions.
This patch modifies Stasis so that rather than walking the tree of
forwards on every dispatch, when forwards and subscriptions are changed,
the subscriber list for every topic in the tree is changed.
This has a couple of benefits. First, this reduces the workload of
dispatching messages. It also reduces contention when dispatching to
different topics that happen to forward to the same aggregation topic
(as happens with all of the channel, bridge and endpoint topics).
Since forwards are no longer subscriptions, the bulk of this patch is
simply changing stasis_subscription objects to stasis_forward objects
(which, admittedly, I should have done in the first place.)
Since this required me to yet again put in a growing array, I finally
abstracted that out into a set of ast_vector macros in
asterisk/vector.h.
Review: https://reviewboard.asterisk.org/r/2883/
........
r400181 | dlee | 2013-09-30 13:48:57 -0500 (Mon, 30 Sep 2013) | 28 lines
Remove dispatch object allocation from Stasis publishing
While looking for areas for performance improvement, I realized that an
unused feature in Stasis was negatively impacting performance.
When a message is sent to a subscriber, a dispatch object is allocated
for the dispatch, containing the topic the message was published to, the
subscriber the message is being sent to, and the message itself.
The topic is actually unused by any subscriber in Asterisk today. And
the subscriber is associated with the taskprocessor the message is being
dispatched to.
First, this patch removes the unused topic parameter from Stasis
subscription callbacks.
Second, this patch introduces the concept of taskprocessor local data,
data that may be set on a taskprocessor and provided along with the data
pointer when a task is pushed using the ast_taskprocessor_push_local()
call. This allows the task to have both data specific to that
taskprocessor, in addition to data specific to that invocation.
With those two changes, the dispatch object can be removed completely,
and the message is simply refcounted and sent directly to the
taskprocessor.
Review: https://reviewboard.asterisk.org/r/2884/
........
Merged revisions 399887,400138,400178,400180-400181 from http://svn.asterisk.org/svn/asterisk/branches/12
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@400186 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-09-30 18:55:27 +00:00
|
|
|
RAII_VAR(struct stasis_forward *, forward_sub, NULL, stasis_forward_cancel);
|
2013-03-08 15:15:13 +00:00
|
|
|
RAII_VAR(struct stasis_subscription *, parent_sub, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test sending events to a parent topic";
|
|
|
|
info->description = "Test sending events to a parent topic.\n"
|
|
|
|
"This test creates three topics (one parent, two children)\n"
|
|
|
|
"and publishes a message to one child, and verifies it's\n"
|
|
|
|
"only seen by that child and the parent";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
parent_topic = stasis_topic_create("ParentTestTopic");
|
|
|
|
ast_test_validate(test, NULL != parent_topic);
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
forward_sub = stasis_forward_all(topic, parent_topic);
|
|
|
|
ast_test_validate(test, NULL != forward_sub);
|
|
|
|
|
|
|
|
parent_consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != parent_consumer);
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
|
|
|
|
parent_sub = stasis_subscribe(parent_topic, consumer_exec, parent_consumer);
|
|
|
|
ast_test_validate(test, NULL != parent_sub);
|
|
|
|
ao2_ref(parent_consumer, +1);
|
|
|
|
sub = stasis_subscribe(topic, consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
test_message = stasis_message_create(test_message_type, test_data);
|
|
|
|
|
|
|
|
stasis_publish(topic, test_message);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(parent_consumer, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2013-04-01 13:37:51 +00:00
|
|
|
AST_TEST_DEFINE(interleaving)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic2, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
|
|
|
|
|
Multiple revisions 399887,400138,400178,400180-400181
........
r399887 | dlee | 2013-09-26 10:41:47 -0500 (Thu, 26 Sep 2013) | 1 line
Minor performance bump by not allocate manager variable struct if we don't need it
........
r400138 | dlee | 2013-09-30 10:24:00 -0500 (Mon, 30 Sep 2013) | 23 lines
Stasis performance improvements
This patch addresses several performance problems that were found in
the initial performance testing of Asterisk 12.
The Stasis dispatch object was allocated as an AO2 object, even though
it has a very confined lifecycle. This was replaced with a straight
ast_malloc().
The Stasis message router was spending an inordinate amount of time
searching hash tables. In this case, most of our routers had 6 or
fewer routes in them to begin with. This was replaced with an array
that's searched linearly for the route.
We more heavily rely on AO2 objects in Asterisk 12, and the memset()
in ao2_ref() actually became noticeable on the profile. This was
#ifdef'ed to only run when AO2_DEBUG was enabled.
After being misled by an erroneous comment in taskprocessor.c during
profiling, the wrong comment was removed.
Review: https://reviewboard.asterisk.org/r/2873/
........
r400178 | dlee | 2013-09-30 13:26:27 -0500 (Mon, 30 Sep 2013) | 24 lines
Taskprocessor optimization; switch Stasis to use taskprocessors
This patch optimizes taskprocessor to use a semaphore for signaling,
which the OS can do a better job at managing contention and waiting
that we can with a mutex and condition.
The taskprocessor execution was also slightly optimized to reduce the
number of locks taken.
The only observable difference in the taskprocessor implementation is
that when the final reference to the taskprocessor goes away, it will
execute all tasks to completion instead of discarding the unexecuted
tasks.
For systems where unnamed semaphores are not supported, a really
simple semaphore implementation is provided. (Which gives identical
performance as the original taskprocessor implementation).
The way we ended up implementing Stasis caused the threadpool to be a
burden instead of a boost to performance. This was switched to just
use taskprocessors directly for subscriptions.
Review: https://reviewboard.asterisk.org/r/2881/
........
r400180 | dlee | 2013-09-30 13:39:34 -0500 (Mon, 30 Sep 2013) | 28 lines
Optimize how Stasis forwards are dispatched
This patch optimizes how forwards are dispatched in Stasis.
Originally, forwards were dispatched as subscriptions that are invoked
on the publishing thread. This did not account for the vast number of
forwards we would end up having in the system, and the amount of work it
would take to walk though the forward subscriptions.
This patch modifies Stasis so that rather than walking the tree of
forwards on every dispatch, when forwards and subscriptions are changed,
the subscriber list for every topic in the tree is changed.
This has a couple of benefits. First, this reduces the workload of
dispatching messages. It also reduces contention when dispatching to
different topics that happen to forward to the same aggregation topic
(as happens with all of the channel, bridge and endpoint topics).
Since forwards are no longer subscriptions, the bulk of this patch is
simply changing stasis_subscription objects to stasis_forward objects
(which, admittedly, I should have done in the first place.)
Since this required me to yet again put in a growing array, I finally
abstracted that out into a set of ast_vector macros in
asterisk/vector.h.
Review: https://reviewboard.asterisk.org/r/2883/
........
r400181 | dlee | 2013-09-30 13:48:57 -0500 (Mon, 30 Sep 2013) | 28 lines
Remove dispatch object allocation from Stasis publishing
While looking for areas for performance improvement, I realized that an
unused feature in Stasis was negatively impacting performance.
When a message is sent to a subscriber, a dispatch object is allocated
for the dispatch, containing the topic the message was published to, the
subscriber the message is being sent to, and the message itself.
The topic is actually unused by any subscriber in Asterisk today. And
the subscriber is associated with the taskprocessor the message is being
dispatched to.
First, this patch removes the unused topic parameter from Stasis
subscription callbacks.
Second, this patch introduces the concept of taskprocessor local data,
data that may be set on a taskprocessor and provided along with the data
pointer when a task is pushed using the ast_taskprocessor_push_local()
call. This allows the task to have both data specific to that
taskprocessor, in addition to data specific to that invocation.
With those two changes, the dispatch object can be removed completely,
and the message is simply refcounted and sent directly to the
taskprocessor.
Review: https://reviewboard.asterisk.org/r/2884/
........
Merged revisions 399887,400138,400178,400180-400181 from http://svn.asterisk.org/svn/asterisk/branches/12
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@400186 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-09-30 18:55:27 +00:00
|
|
|
RAII_VAR(struct stasis_forward *, forward_sub1, NULL, stasis_forward_cancel);
|
|
|
|
RAII_VAR(struct stasis_forward *, forward_sub2, NULL, stasis_forward_cancel);
|
2013-04-01 13:37:51 +00:00
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
int actual_len;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test sending interleaved events to a parent topic";
|
|
|
|
info->description = "Test sending events to a parent topic.\n"
|
|
|
|
"This test creates three topics (one parent, two children)\n"
|
|
|
|
"and publishes messages alternately between the children.\n"
|
|
|
|
"It verifies that the messages are received in the expected\n"
|
|
|
|
"order.";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("test", NULL, &test_message_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-04-01 13:37:51 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type);
|
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
|
|
|
|
|
|
|
test_message1 = stasis_message_create(test_message_type, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message1);
|
|
|
|
test_message2 = stasis_message_create(test_message_type, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message2);
|
|
|
|
test_message3 = stasis_message_create(test_message_type, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message3);
|
|
|
|
|
|
|
|
parent_topic = stasis_topic_create("ParentTestTopic");
|
|
|
|
ast_test_validate(test, NULL != parent_topic);
|
|
|
|
topic1 = stasis_topic_create("Topic1");
|
|
|
|
ast_test_validate(test, NULL != topic1);
|
|
|
|
topic2 = stasis_topic_create("Topic2");
|
|
|
|
ast_test_validate(test, NULL != topic2);
|
|
|
|
|
|
|
|
forward_sub1 = stasis_forward_all(topic1, parent_topic);
|
|
|
|
ast_test_validate(test, NULL != forward_sub1);
|
|
|
|
forward_sub2 = stasis_forward_all(topic2, parent_topic);
|
|
|
|
ast_test_validate(test, NULL != forward_sub2);
|
|
|
|
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
|
|
|
|
sub = stasis_subscribe(parent_topic, consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
stasis_publish(topic1, test_message1);
|
|
|
|
stasis_publish(topic2, test_message2);
|
|
|
|
stasis_publish(topic1, test_message3);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 3);
|
|
|
|
ast_test_validate(test, 3 == actual_len);
|
|
|
|
|
|
|
|
ast_test_validate(test, test_message1 == consumer->messages_rxed[0]);
|
|
|
|
ast_test_validate(test, test_message2 == consumer->messages_rxed[1]);
|
|
|
|
ast_test_validate(test, test_message3 == consumer->messages_rxed[2]);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
struct cache_test_data {
|
|
|
|
char *id;
|
|
|
|
char *value;
|
|
|
|
};
|
|
|
|
|
|
|
|
static void cache_test_data_dtor(void *obj)
|
|
|
|
{
|
|
|
|
struct cache_test_data *data = obj;
|
2014-03-07 20:41:13 +00:00
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_free(data->id);
|
|
|
|
ast_free(data->value);
|
|
|
|
}
|
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
static struct stasis_message *cache_test_message_create_full(struct stasis_message_type *type, const char *name, const char *value, struct ast_eid *eid)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct cache_test_data *, data, NULL, ao2_cleanup);
|
|
|
|
|
|
|
|
data = ao2_alloc(sizeof(*data), cache_test_data_dtor);
|
|
|
|
if (data == NULL) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
ast_assert(name != NULL);
|
|
|
|
ast_assert(value != NULL);
|
|
|
|
|
|
|
|
data->id = ast_strdup(name);
|
|
|
|
data->value = ast_strdup(value);
|
|
|
|
if (!data->id || !data->value) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
return stasis_message_create_full(type, data, eid);
|
|
|
|
}
|
|
|
|
|
|
|
|
static struct stasis_message *cache_test_message_create(struct stasis_message_type *type, const char *name, const char *value)
|
|
|
|
{
|
|
|
|
return cache_test_message_create_full(type, name, value, &ast_eid_default);
|
2013-03-08 15:15:13 +00:00
|
|
|
}
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
static const char *cache_test_data_id(struct stasis_message *message)
|
|
|
|
{
|
2013-03-08 15:15:13 +00:00
|
|
|
struct cache_test_data *cachable = stasis_message_data(message);
|
|
|
|
|
|
|
|
if (0 != strcmp("Cacheable", stasis_message_type_name(stasis_message_type(message)))) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
return cachable->id;
|
|
|
|
}
|
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
static struct stasis_message *cache_test_aggregate_calc_fn(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
|
|
|
|
{
|
|
|
|
struct stasis_message *aggregate_snapshot;
|
|
|
|
struct stasis_message *snapshot;
|
|
|
|
struct stasis_message_type *type = NULL;
|
|
|
|
struct cache_test_data *test_data = NULL;
|
|
|
|
int idx;
|
|
|
|
int accumulated = 0;
|
|
|
|
char aggregate_str[30];
|
|
|
|
|
|
|
|
/* Accumulate the aggregate value. */
|
|
|
|
snapshot = stasis_cache_entry_get_local(entry);
|
|
|
|
if (snapshot) {
|
|
|
|
type = stasis_message_type(snapshot);
|
|
|
|
test_data = stasis_message_data(snapshot);
|
|
|
|
accumulated += atoi(test_data->value);
|
|
|
|
}
|
|
|
|
for (idx = 0; ; ++idx) {
|
|
|
|
snapshot = stasis_cache_entry_get_remote(entry, idx);
|
|
|
|
if (!snapshot) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
type = stasis_message_type(snapshot);
|
|
|
|
test_data = stasis_message_data(snapshot);
|
|
|
|
accumulated += atoi(test_data->value);
|
|
|
|
}
|
|
|
|
|
|
|
|
if (!test_data) {
|
|
|
|
/* There are no test entries cached. Delete the aggregate. */
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
snapshot = stasis_cache_entry_get_aggregate(entry);
|
|
|
|
if (snapshot) {
|
|
|
|
type = stasis_message_type(snapshot);
|
|
|
|
test_data = stasis_message_data(snapshot);
|
|
|
|
if (accumulated == atoi(test_data->value)) {
|
|
|
|
/* Aggregate test entry did not change. */
|
|
|
|
return ao2_bump(snapshot);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
snprintf(aggregate_str, sizeof(aggregate_str), "%d", accumulated);
|
|
|
|
aggregate_snapshot = cache_test_message_create_full(type, test_data->id, aggregate_str, NULL);
|
|
|
|
if (!aggregate_snapshot) {
|
|
|
|
/* Bummer. We have to keep the old aggregate snapshot. */
|
|
|
|
ast_log(LOG_ERROR, "Could not create aggregate snapshot.\n");
|
|
|
|
return ao2_bump(snapshot);
|
|
|
|
}
|
|
|
|
|
|
|
|
return aggregate_snapshot;
|
|
|
|
}
|
|
|
|
|
|
|
|
static void cache_test_aggregate_publish_fn(struct stasis_topic *topic, struct stasis_message *aggregate)
|
|
|
|
{
|
|
|
|
stasis_publish(topic, aggregate);
|
|
|
|
}
|
|
|
|
|
|
|
|
static int check_cache_aggregate(struct stasis_cache *cache, struct stasis_message_type *cache_type, const char *id, const char *value)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message *, aggregate, NULL, ao2_cleanup);
|
|
|
|
struct cache_test_data *test_data;
|
|
|
|
|
|
|
|
aggregate = stasis_cache_get_by_eid(cache, cache_type, id, NULL);
|
|
|
|
if (!aggregate) {
|
|
|
|
/* No aggregate, return true if given no value. */
|
|
|
|
return !value;
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Return true if the given value matches the aggregate value. */
|
|
|
|
test_data = stasis_message_data(aggregate);
|
|
|
|
return value && !strcmp(value, test_data->value);
|
|
|
|
}
|
|
|
|
|
2013-08-06 14:28:23 +00:00
|
|
|
AST_TEST_DEFINE(cache_filter)
|
2013-03-08 15:15:13 +00:00
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
2013-08-01 13:49:34 +00:00
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
2013-03-08 15:15:13 +00:00
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
2013-08-06 14:28:23 +00:00
|
|
|
info->summary = "Test caching topics only forward cache_update messages.";
|
|
|
|
info->description = "Test caching topics only forward cache_update messages.";
|
2013-03-08 15:15:13 +00:00
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("NonCacheable", NULL, &non_cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_test_validate(test, NULL != non_cache_type);
|
|
|
|
topic = stasis_topic_create("SomeTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache = stasis_cache_create(cache_test_data_id);
|
|
|
|
ast_test_validate(test, NULL != cache);
|
|
|
|
caching_topic = stasis_caching_topic_create(topic, cache);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_test_validate(test, NULL != caching_topic);
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
test_message = cache_test_message_create(non_cache_type, "1", "1");
|
|
|
|
ast_test_validate(test, NULL != test_message);
|
|
|
|
|
|
|
|
stasis_publish(topic, test_message);
|
|
|
|
|
2013-08-06 14:28:23 +00:00
|
|
|
actual_len = consumer_should_stay(consumer, 0);
|
|
|
|
ast_test_validate(test, 0 == actual_len);
|
2013-03-08 15:15:13 +00:00
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(cache)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
2013-08-01 13:49:34 +00:00
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
2013-03-08 15:15:13 +00:00
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
struct stasis_cache_update *actual_update;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test passing messages through cache topic unscathed.";
|
|
|
|
info->description = "Test passing messages through cache topic unscathed.";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("Cacheable", NULL, &cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_test_validate(test, NULL != cache_type);
|
|
|
|
topic = stasis_topic_create("SomeTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache = stasis_cache_create(cache_test_data_id);
|
|
|
|
ast_test_validate(test, NULL != cache);
|
|
|
|
caching_topic = stasis_caching_topic_create(topic, cache);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_test_validate(test, NULL != caching_topic);
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
test_message1_1 = cache_test_message_create(cache_type, "1", "1");
|
|
|
|
ast_test_validate(test, NULL != test_message1_1);
|
|
|
|
test_message2_1 = cache_test_message_create(cache_type, "2", "1");
|
|
|
|
ast_test_validate(test, NULL != test_message2_1);
|
|
|
|
|
|
|
|
/* Post a couple of snapshots */
|
|
|
|
stasis_publish(topic, test_message1_1);
|
|
|
|
stasis_publish(topic, test_message2_1);
|
|
|
|
actual_len = consumer_wait_for(consumer, 2);
|
|
|
|
ast_test_validate(test, 2 == actual_len);
|
|
|
|
|
|
|
|
/* Check for new snapshot messages */
|
2013-03-28 15:45:18 +00:00
|
|
|
ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(consumer->messages_rxed[0]));
|
2013-03-08 15:15:13 +00:00
|
|
|
actual_update = stasis_message_data(consumer->messages_rxed[0]);
|
|
|
|
ast_test_validate(test, NULL == actual_update->old_snapshot);
|
|
|
|
ast_test_validate(test, test_message1_1 == actual_update->new_snapshot);
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_test_validate(test, test_message1_1 == stasis_cache_get(cache, cache_type, "1"));
|
2013-03-08 15:15:13 +00:00
|
|
|
/* stasis_cache_get returned a ref, so unref test_message1_1 */
|
|
|
|
ao2_ref(test_message1_1, -1);
|
|
|
|
|
2013-03-28 15:45:18 +00:00
|
|
|
ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(consumer->messages_rxed[1]));
|
2013-03-08 15:15:13 +00:00
|
|
|
actual_update = stasis_message_data(consumer->messages_rxed[1]);
|
|
|
|
ast_test_validate(test, NULL == actual_update->old_snapshot);
|
|
|
|
ast_test_validate(test, test_message2_1 == actual_update->new_snapshot);
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_test_validate(test, test_message2_1 == stasis_cache_get(cache, cache_type, "2"));
|
2013-03-08 15:15:13 +00:00
|
|
|
/* stasis_cache_get returned a ref, so unref test_message2_1 */
|
|
|
|
ao2_ref(test_message2_1, -1);
|
|
|
|
|
|
|
|
/* Update snapshot 2 */
|
|
|
|
test_message2_2 = cache_test_message_create(cache_type, "2", "2");
|
|
|
|
ast_test_validate(test, NULL != test_message2_2);
|
|
|
|
stasis_publish(topic, test_message2_2);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 3);
|
|
|
|
ast_test_validate(test, 3 == actual_len);
|
|
|
|
|
|
|
|
actual_update = stasis_message_data(consumer->messages_rxed[2]);
|
|
|
|
ast_test_validate(test, test_message2_1 == actual_update->old_snapshot);
|
|
|
|
ast_test_validate(test, test_message2_2 == actual_update->new_snapshot);
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_test_validate(test, test_message2_2 == stasis_cache_get(cache, cache_type, "2"));
|
2013-03-08 15:15:13 +00:00
|
|
|
/* stasis_cache_get returned a ref, so unref test_message2_2 */
|
|
|
|
ao2_ref(test_message2_2, -1);
|
|
|
|
|
|
|
|
/* Clear snapshot 1 */
|
2013-06-07 12:56:56 +00:00
|
|
|
test_message1_clear = stasis_cache_clear_create(test_message1_1);
|
2013-03-08 15:15:13 +00:00
|
|
|
ast_test_validate(test, NULL != test_message1_clear);
|
|
|
|
stasis_publish(topic, test_message1_clear);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 4);
|
|
|
|
ast_test_validate(test, 4 == actual_len);
|
|
|
|
|
|
|
|
actual_update = stasis_message_data(consumer->messages_rxed[3]);
|
|
|
|
ast_test_validate(test, test_message1_1 == actual_update->old_snapshot);
|
|
|
|
ast_test_validate(test, NULL == actual_update->new_snapshot);
|
2013-08-01 13:49:34 +00:00
|
|
|
ast_test_validate(test, NULL == stasis_cache_get(cache, cache_type, "1"));
|
2013-03-08 15:15:13 +00:00
|
|
|
|
2013-04-22 14:58:53 +00:00
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(cache_dump)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
2013-08-01 13:49:34 +00:00
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
2013-04-22 14:58:53 +00:00
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
|
|
|
|
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
struct ao2_iterator i;
|
|
|
|
void *obj;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
2014-03-07 20:41:13 +00:00
|
|
|
info->summary = "Test cache dump routines.";
|
|
|
|
info->description = "Test cache dump routines.";
|
2013-04-22 14:58:53 +00:00
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("Cacheable", NULL, &cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-04-22 14:58:53 +00:00
|
|
|
ast_test_validate(test, NULL != cache_type);
|
|
|
|
topic = stasis_topic_create("SomeTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache = stasis_cache_create(cache_test_data_id);
|
|
|
|
ast_test_validate(test, NULL != cache);
|
|
|
|
caching_topic = stasis_caching_topic_create(topic, cache);
|
2013-04-22 14:58:53 +00:00
|
|
|
ast_test_validate(test, NULL != caching_topic);
|
|
|
|
consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer);
|
|
|
|
sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
ao2_ref(consumer, +1);
|
|
|
|
|
|
|
|
test_message1_1 = cache_test_message_create(cache_type, "1", "1");
|
|
|
|
ast_test_validate(test, NULL != test_message1_1);
|
|
|
|
test_message2_1 = cache_test_message_create(cache_type, "2", "1");
|
|
|
|
ast_test_validate(test, NULL != test_message2_1);
|
|
|
|
|
|
|
|
/* Post a couple of snapshots */
|
|
|
|
stasis_publish(topic, test_message1_1);
|
|
|
|
stasis_publish(topic, test_message2_1);
|
|
|
|
actual_len = consumer_wait_for(consumer, 2);
|
|
|
|
ast_test_validate(test, 2 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
2014-01-03 18:33:19 +00:00
|
|
|
ao2_cleanup(cache_dump);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache_dump = stasis_cache_dump(cache, NULL);
|
2013-04-22 14:58:53 +00:00
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 2 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
ast_test_validate(test, actual_cache_entry == test_message1_1 || actual_cache_entry == test_message2_1);
|
|
|
|
}
|
2013-12-20 20:00:50 +00:00
|
|
|
ao2_iterator_destroy(&i);
|
2013-04-22 14:58:53 +00:00
|
|
|
|
|
|
|
/* Update snapshot 2 */
|
|
|
|
test_message2_2 = cache_test_message_create(cache_type, "2", "2");
|
|
|
|
ast_test_validate(test, NULL != test_message2_2);
|
|
|
|
stasis_publish(topic, test_message2_2);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 3);
|
|
|
|
ast_test_validate(test, 3 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
2014-01-03 18:33:19 +00:00
|
|
|
ao2_cleanup(cache_dump);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache_dump = stasis_cache_dump(cache, NULL);
|
2013-04-22 14:58:53 +00:00
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 2 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
ast_test_validate(test, actual_cache_entry == test_message1_1 || actual_cache_entry == test_message2_2);
|
|
|
|
}
|
2013-12-20 20:00:50 +00:00
|
|
|
ao2_iterator_destroy(&i);
|
2013-04-22 14:58:53 +00:00
|
|
|
|
|
|
|
/* Clear snapshot 1 */
|
2013-06-07 12:56:56 +00:00
|
|
|
test_message1_clear = stasis_cache_clear_create(test_message1_1);
|
2013-04-22 14:58:53 +00:00
|
|
|
ast_test_validate(test, NULL != test_message1_clear);
|
|
|
|
stasis_publish(topic, test_message1_clear);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer, 4);
|
|
|
|
ast_test_validate(test, 4 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
2014-01-03 18:33:19 +00:00
|
|
|
ao2_cleanup(cache_dump);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache_dump = stasis_cache_dump(cache, NULL);
|
2013-04-22 14:58:53 +00:00
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
2013-03-08 16:00:14 +00:00
|
|
|
ast_test_validate(test, 1 == ao2_container_count(cache_dump));
|
2013-04-22 14:58:53 +00:00
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
ast_test_validate(test, actual_cache_entry == test_message2_2);
|
|
|
|
}
|
2013-12-20 20:00:50 +00:00
|
|
|
ao2_iterator_destroy(&i);
|
2013-03-08 16:00:14 +00:00
|
|
|
|
|
|
|
/* Dump the cache to ensure that it has no subscription change items in it since those aren't cached */
|
2013-04-22 14:58:53 +00:00
|
|
|
ao2_cleanup(cache_dump);
|
2013-08-01 13:49:34 +00:00
|
|
|
cache_dump = stasis_cache_dump(cache, stasis_subscription_change_type());
|
2013-03-08 16:00:14 +00:00
|
|
|
ast_test_validate(test, 0 == ao2_container_count(cache_dump));
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2014-03-07 20:41:13 +00:00
|
|
|
AST_TEST_DEFINE(cache_eid_aggregate)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
|
|
|
|
RAII_VAR(struct consumer *, cache_consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, topic_consumer, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, topic_sub, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_subscription *, cache_sub, NULL, stasis_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_3, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_4, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2_clear, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
|
|
|
|
int actual_len;
|
|
|
|
struct ao2_iterator i;
|
|
|
|
void *obj;
|
|
|
|
struct ast_eid foreign_eid1;
|
|
|
|
struct ast_eid foreign_eid2;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test cache eid and aggregate support.";
|
|
|
|
info->description = "Test cache eid and aggregate support.";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
memset(&foreign_eid1, 0xAA, sizeof(foreign_eid1));
|
|
|
|
memset(&foreign_eid2, 0xBB, sizeof(foreign_eid2));
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("Cacheable", NULL, &cache_type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2014-03-07 20:41:13 +00:00
|
|
|
ast_test_validate(test, NULL != cache_type);
|
|
|
|
|
|
|
|
topic = stasis_topic_create("SomeTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
/* To consume events published to the topic. */
|
|
|
|
topic_consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != topic_consumer);
|
|
|
|
|
|
|
|
topic_sub = stasis_subscribe(topic, consumer_exec, topic_consumer);
|
|
|
|
ast_test_validate(test, NULL != topic_sub);
|
|
|
|
ao2_ref(topic_consumer, +1);
|
|
|
|
|
|
|
|
cache = stasis_cache_create_full(cache_test_data_id,
|
|
|
|
cache_test_aggregate_calc_fn, cache_test_aggregate_publish_fn);
|
|
|
|
ast_test_validate(test, NULL != cache);
|
|
|
|
|
|
|
|
caching_topic = stasis_caching_topic_create(topic, cache);
|
|
|
|
ast_test_validate(test, NULL != caching_topic);
|
|
|
|
|
|
|
|
/* To consume update events published to the caching_topic. */
|
|
|
|
cache_consumer = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != cache_consumer);
|
|
|
|
|
|
|
|
cache_sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, cache_consumer);
|
|
|
|
ast_test_validate(test, NULL != cache_sub);
|
|
|
|
ao2_ref(cache_consumer, +1);
|
|
|
|
|
|
|
|
/* Create test messages. */
|
|
|
|
test_message1_1 = cache_test_message_create_full(cache_type, "1", "1", &ast_eid_default);
|
|
|
|
ast_test_validate(test, NULL != test_message1_1);
|
|
|
|
test_message2_1 = cache_test_message_create_full(cache_type, "2", "1", &ast_eid_default);
|
|
|
|
ast_test_validate(test, NULL != test_message2_1);
|
|
|
|
test_message2_2 = cache_test_message_create_full(cache_type, "2", "2", &foreign_eid1);
|
|
|
|
ast_test_validate(test, NULL != test_message2_2);
|
|
|
|
test_message2_3 = cache_test_message_create_full(cache_type, "2", "3", &foreign_eid2);
|
|
|
|
ast_test_validate(test, NULL != test_message2_3);
|
|
|
|
test_message2_4 = cache_test_message_create_full(cache_type, "2", "4", &foreign_eid2);
|
|
|
|
ast_test_validate(test, NULL != test_message2_4);
|
|
|
|
|
|
|
|
/* Post some snapshots */
|
|
|
|
stasis_publish(topic, test_message1_1);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "1", "1"));
|
|
|
|
stasis_publish(topic, test_message2_1);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "1"));
|
|
|
|
stasis_publish(topic, test_message2_2);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "3"));
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(cache_consumer, 6);
|
|
|
|
ast_test_validate(test, 6 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(topic_consumer, 6);
|
|
|
|
ast_test_validate(test, 6 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_all(cache, NULL);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 3 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message1_1
|
|
|
|
|| actual_cache_entry == test_message2_1
|
|
|
|
|| actual_cache_entry == test_message2_2);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Check the local cached items */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_by_eid(cache, NULL, &ast_eid_default);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 2 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message1_1
|
|
|
|
|| actual_cache_entry == test_message2_1);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Post snapshot 2 from another eid. */
|
|
|
|
stasis_publish(topic, test_message2_3);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "6"));
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(cache_consumer, 8);
|
|
|
|
ast_test_validate(test, 8 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(topic_consumer, 8);
|
|
|
|
ast_test_validate(test, 8 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_all(cache, NULL);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 4 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message1_1
|
|
|
|
|| actual_cache_entry == test_message2_1
|
|
|
|
|| actual_cache_entry == test_message2_2
|
|
|
|
|| actual_cache_entry == test_message2_3);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Check the remote cached items */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_by_eid(cache, NULL, &foreign_eid1);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 1 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test, actual_cache_entry == test_message2_2);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Post snapshot 2 from a repeated eid. */
|
|
|
|
stasis_publish(topic, test_message2_4);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "7"));
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(cache_consumer, 10);
|
|
|
|
ast_test_validate(test, 10 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(topic_consumer, 10);
|
|
|
|
ast_test_validate(test, 10 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_all(cache, NULL);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 4 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message1_1
|
|
|
|
|| actual_cache_entry == test_message2_1
|
|
|
|
|| actual_cache_entry == test_message2_2
|
|
|
|
|| actual_cache_entry == test_message2_4);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Check all snapshot 2 cache entries. */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_get_all(cache, cache_type, "2");
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 3 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message2_1
|
|
|
|
|| actual_cache_entry == test_message2_2
|
|
|
|
|| actual_cache_entry == test_message2_4);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Clear snapshot 1 */
|
|
|
|
test_message1_clear = stasis_cache_clear_create(test_message1_1);
|
|
|
|
ast_test_validate(test, NULL != test_message1_clear);
|
|
|
|
stasis_publish(topic, test_message1_clear);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "1", NULL));
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(cache_consumer, 12);
|
|
|
|
ast_test_validate(test, 12 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(topic_consumer, 11);
|
|
|
|
ast_test_validate(test, 11 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_all(cache, NULL);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 3 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message2_1
|
|
|
|
|| actual_cache_entry == test_message2_2
|
|
|
|
|| actual_cache_entry == test_message2_4);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
/* Clear snapshot 2 from a remote eid */
|
|
|
|
test_message2_clear = stasis_cache_clear_create(test_message2_2);
|
|
|
|
ast_test_validate(test, NULL != test_message2_clear);
|
|
|
|
stasis_publish(topic, test_message2_clear);
|
|
|
|
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "5"));
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(cache_consumer, 14);
|
|
|
|
ast_test_validate(test, 14 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(topic_consumer, 13);
|
|
|
|
ast_test_validate(test, 13 == actual_len);
|
|
|
|
|
|
|
|
/* Check the cache */
|
|
|
|
ao2_cleanup(cache_dump);
|
|
|
|
cache_dump = stasis_cache_dump_all(cache, NULL);
|
|
|
|
ast_test_validate(test, NULL != cache_dump);
|
|
|
|
ast_test_validate(test, 2 == ao2_container_count(cache_dump));
|
|
|
|
i = ao2_iterator_init(cache_dump, 0);
|
|
|
|
while ((obj = ao2_iterator_next(&i))) {
|
|
|
|
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
|
|
|
|
|
|
|
|
ast_test_validate(test,
|
|
|
|
actual_cache_entry == test_message2_1
|
|
|
|
|| actual_cache_entry == test_message2_4);
|
|
|
|
}
|
|
|
|
ao2_iterator_destroy(&i);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2013-03-15 17:35:16 +00:00
|
|
|
AST_TEST_DEFINE(router)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
2013-07-23 13:39:50 +00:00
|
|
|
RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
|
2013-03-15 17:35:16 +00:00
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
|
|
|
|
int actual_len, ret;
|
|
|
|
struct stasis_message *actual;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test simple message routing";
|
|
|
|
info->description = "Test simple message routing";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
consumer1 = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer1);
|
|
|
|
consumer2 = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer2);
|
|
|
|
consumer3 = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer3);
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage1", NULL, &test_message_type1) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-15 17:35:16 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type1);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage2", NULL, &test_message_type2) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-15 17:35:16 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type2);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("TestMessage3", NULL, &test_message_type3) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-03-15 17:35:16 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type3);
|
|
|
|
|
|
|
|
uut = stasis_message_router_create(topic);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
|
|
|
|
ret = stasis_message_router_add(
|
|
|
|
uut, test_message_type1, consumer_exec, consumer1);
|
|
|
|
ast_test_validate(test, 0 == ret);
|
|
|
|
ao2_ref(consumer1, +1);
|
|
|
|
ret = stasis_message_router_add(
|
|
|
|
uut, test_message_type2, consumer_exec, consumer2);
|
|
|
|
ast_test_validate(test, 0 == ret);
|
|
|
|
ao2_ref(consumer2, +1);
|
|
|
|
ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
|
|
|
|
ast_test_validate(test, 0 == ret);
|
|
|
|
ao2_ref(consumer3, +1);
|
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
|
|
|
test_message1 = stasis_message_create(test_message_type1, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message1);
|
|
|
|
test_message2 = stasis_message_create(test_message_type2, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message2);
|
|
|
|
test_message3 = stasis_message_create(test_message_type3, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message3);
|
|
|
|
|
|
|
|
stasis_publish(topic, test_message1);
|
|
|
|
stasis_publish(topic, test_message2);
|
|
|
|
stasis_publish(topic, test_message3);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer1, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(consumer2, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(consumer3, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
|
|
|
|
actual = consumer1->messages_rxed[0];
|
|
|
|
ast_test_validate(test, test_message1 == actual);
|
|
|
|
|
|
|
|
actual = consumer2->messages_rxed[0];
|
|
|
|
ast_test_validate(test, test_message2 == actual);
|
|
|
|
|
|
|
|
actual = consumer3->messages_rxed[0];
|
|
|
|
ast_test_validate(test, test_message3 == actual);
|
|
|
|
|
|
|
|
/* consumer1 and consumer2 do not get the final message. */
|
|
|
|
ao2_cleanup(consumer1);
|
|
|
|
ao2_cleanup(consumer2);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2014-02-26 23:35:20 +00:00
|
|
|
static const char *cache_simple(struct stasis_message *message)
|
|
|
|
{
|
2013-07-23 13:39:50 +00:00
|
|
|
const char *type_name =
|
|
|
|
stasis_message_type_name(stasis_message_type(message));
|
|
|
|
if (!ast_begins_with(type_name, "Cache")) {
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
return "cached";
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(router_cache_updates)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
2013-08-01 13:49:34 +00:00
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe_and_join);
|
2013-07-23 13:39:50 +00:00
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
|
|
|
|
RAII_VAR(char *, test_data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, message1, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, message2, NULL, ao2_cleanup);
|
|
|
|
struct stasis_cache_update *update;
|
|
|
|
int actual_len, ret;
|
|
|
|
struct stasis_message *actual;
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test special handling cache_update messages";
|
|
|
|
info->description = "Test special handling cache_update messages";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("TestTopic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
2013-08-01 13:49:34 +00:00
|
|
|
cache = stasis_cache_create(cache_simple);
|
|
|
|
ast_test_validate(test, NULL != cache);
|
|
|
|
caching_topic = stasis_caching_topic_create(topic, cache);
|
2013-07-23 13:39:50 +00:00
|
|
|
ast_test_validate(test, NULL != caching_topic);
|
|
|
|
|
|
|
|
consumer1 = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer1);
|
|
|
|
consumer2 = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer2);
|
|
|
|
consumer3 = consumer_create(1);
|
|
|
|
ast_test_validate(test, NULL != consumer3);
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("Cache1", NULL, &test_message_type1) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-07-23 13:39:50 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type1);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("Cache2", NULL, &test_message_type2) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-07-23 13:39:50 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type2);
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("NonCache", NULL, &test_message_type3) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-07-23 13:39:50 +00:00
|
|
|
ast_test_validate(test, NULL != test_message_type3);
|
|
|
|
|
|
|
|
uut = stasis_message_router_create(
|
|
|
|
stasis_caching_get_topic(caching_topic));
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
|
|
|
|
ret = stasis_message_router_add_cache_update(
|
|
|
|
uut, test_message_type1, consumer_exec, consumer1);
|
|
|
|
ast_test_validate(test, 0 == ret);
|
|
|
|
ao2_ref(consumer1, +1);
|
|
|
|
ret = stasis_message_router_add(
|
|
|
|
uut, stasis_cache_update_type(), consumer_exec, consumer2);
|
|
|
|
ast_test_validate(test, 0 == ret);
|
|
|
|
ao2_ref(consumer2, +1);
|
|
|
|
ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
|
|
|
|
ast_test_validate(test, 0 == ret);
|
|
|
|
ao2_ref(consumer3, +1);
|
|
|
|
|
|
|
|
test_data = ao2_alloc(1, NULL);
|
|
|
|
ast_test_validate(test, NULL != test_data);
|
|
|
|
test_message1 = stasis_message_create(test_message_type1, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message1);
|
|
|
|
test_message2 = stasis_message_create(test_message_type2, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message2);
|
|
|
|
test_message3 = stasis_message_create(test_message_type3, test_data);
|
|
|
|
ast_test_validate(test, NULL != test_message3);
|
|
|
|
|
|
|
|
stasis_publish(topic, test_message1);
|
|
|
|
stasis_publish(topic, test_message2);
|
|
|
|
stasis_publish(topic, test_message3);
|
|
|
|
|
|
|
|
actual_len = consumer_wait_for(consumer1, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
|
|
|
actual_len = consumer_wait_for(consumer2, 1);
|
|
|
|
ast_test_validate(test, 1 == actual_len);
|
2013-08-06 14:28:23 +00:00
|
|
|
/* Uncacheable message should not be passed through */
|
|
|
|
actual_len = consumer_should_stay(consumer3, 0);
|
|
|
|
ast_test_validate(test, 0 == actual_len);
|
2013-07-23 13:39:50 +00:00
|
|
|
|
|
|
|
actual = consumer1->messages_rxed[0];
|
|
|
|
ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
|
|
|
|
update = stasis_message_data(actual);
|
|
|
|
ast_test_validate(test, test_message_type1 == update->type);
|
|
|
|
ast_test_validate(test, test_message1 == update->new_snapshot);
|
|
|
|
|
|
|
|
actual = consumer2->messages_rxed[0];
|
|
|
|
ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
|
|
|
|
update = stasis_message_data(actual);
|
|
|
|
ast_test_validate(test, test_message_type2 == update->type);
|
|
|
|
ast_test_validate(test, test_message2 == update->new_snapshot);
|
|
|
|
|
|
|
|
/* consumer1 and consumer2 do not get the final message. */
|
|
|
|
ao2_cleanup(consumer1);
|
|
|
|
ao2_cleanup(consumer2);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2013-06-11 15:46:35 +00:00
|
|
|
AST_TEST_DEFINE(no_to_json)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(char *, data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct ast_json *, actual, NULL, ast_json_unref);
|
|
|
|
char *expected = "SomeData";
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test message to_json function";
|
|
|
|
info->description = "Test message to_json function";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Test NULL */
|
2013-11-22 23:57:45 +00:00
|
|
|
actual = stasis_message_to_json(NULL, NULL);
|
2013-06-11 15:46:35 +00:00
|
|
|
ast_test_validate(test, NULL == actual);
|
|
|
|
|
|
|
|
/* Test message with NULL to_json function */
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-06-11 15:46:35 +00:00
|
|
|
|
|
|
|
data = ao2_alloc(strlen(expected) + 1, NULL);
|
|
|
|
strcpy(data, expected);
|
|
|
|
uut = stasis_message_create(type, data);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
|
2013-11-22 23:57:45 +00:00
|
|
|
actual = stasis_message_to_json(uut, NULL);
|
2013-06-11 15:46:35 +00:00
|
|
|
ast_test_validate(test, NULL == actual);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(to_json)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(char *, data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct ast_json *, actual, NULL, ast_json_unref);
|
|
|
|
const char *expected_text = "SomeData";
|
|
|
|
RAII_VAR(struct ast_json *, expected, NULL, ast_json_unref);
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test message to_json function when NULL";
|
|
|
|
info->description = "Test message to_json function when NULL";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("SomeMessage", &fake_vtable, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-06-11 15:46:35 +00:00
|
|
|
|
|
|
|
data = ao2_alloc(strlen(expected_text) + 1, NULL);
|
|
|
|
strcpy(data, expected_text);
|
|
|
|
uut = stasis_message_create(type, data);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
|
|
|
|
expected = ast_json_string_create(expected_text);
|
2013-11-22 23:57:45 +00:00
|
|
|
actual = stasis_message_to_json(uut, NULL);
|
2013-06-11 15:46:35 +00:00
|
|
|
ast_test_validate(test, ast_json_equal(expected, actual));
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(no_to_ami)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(char *, data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct ast_manager_event_blob *, actual, NULL, ao2_cleanup);
|
|
|
|
char *expected = "SomeData";
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test message to_ami function when NULL";
|
|
|
|
info->description = "Test message to_ami function when NULL";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
/* Test NULL */
|
|
|
|
actual = stasis_message_to_ami(NULL);
|
|
|
|
ast_test_validate(test, NULL == actual);
|
|
|
|
|
|
|
|
/* Test message with NULL to_ami function */
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("SomeMessage", NULL, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-06-11 15:46:35 +00:00
|
|
|
|
|
|
|
data = ao2_alloc(strlen(expected) + 1, NULL);
|
|
|
|
strcpy(data, expected);
|
|
|
|
uut = stasis_message_create(type, data);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
|
|
|
|
actual = stasis_message_to_ami(uut);
|
|
|
|
ast_test_validate(test, NULL == actual);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(to_ami)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(char *, data, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct ast_manager_event_blob *, actual, NULL, ao2_cleanup);
|
|
|
|
const char *expected_text = "SomeData";
|
2013-08-08 22:57:06 +00:00
|
|
|
const char *expected = "Message: SomeData\r\n";
|
2013-06-11 15:46:35 +00:00
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test message to_ami function";
|
|
|
|
info->description = "Test message to_ami function";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2014-08-06 12:55:28 +00:00
|
|
|
ast_test_validate(test, stasis_message_type_create("SomeMessage", &fake_vtable, &type) == STASIS_MESSAGE_TYPE_SUCCESS);
|
2013-06-11 15:46:35 +00:00
|
|
|
|
|
|
|
data = ao2_alloc(strlen(expected_text) + 1, NULL);
|
|
|
|
strcpy(data, expected_text);
|
|
|
|
uut = stasis_message_create(type, data);
|
|
|
|
ast_test_validate(test, NULL != uut);
|
|
|
|
|
|
|
|
actual = stasis_message_to_ami(uut);
|
|
|
|
ast_test_validate(test, strcmp(expected, actual->extra_fields) == 0);
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
static void noop(void *data, struct stasis_subscription *sub,
|
Multiple revisions 399887,400138,400178,400180-400181
........
r399887 | dlee | 2013-09-26 10:41:47 -0500 (Thu, 26 Sep 2013) | 1 line
Minor performance bump by not allocate manager variable struct if we don't need it
........
r400138 | dlee | 2013-09-30 10:24:00 -0500 (Mon, 30 Sep 2013) | 23 lines
Stasis performance improvements
This patch addresses several performance problems that were found in
the initial performance testing of Asterisk 12.
The Stasis dispatch object was allocated as an AO2 object, even though
it has a very confined lifecycle. This was replaced with a straight
ast_malloc().
The Stasis message router was spending an inordinate amount of time
searching hash tables. In this case, most of our routers had 6 or
fewer routes in them to begin with. This was replaced with an array
that's searched linearly for the route.
We more heavily rely on AO2 objects in Asterisk 12, and the memset()
in ao2_ref() actually became noticeable on the profile. This was
#ifdef'ed to only run when AO2_DEBUG was enabled.
After being misled by an erroneous comment in taskprocessor.c during
profiling, the wrong comment was removed.
Review: https://reviewboard.asterisk.org/r/2873/
........
r400178 | dlee | 2013-09-30 13:26:27 -0500 (Mon, 30 Sep 2013) | 24 lines
Taskprocessor optimization; switch Stasis to use taskprocessors
This patch optimizes taskprocessor to use a semaphore for signaling,
which the OS can do a better job at managing contention and waiting
that we can with a mutex and condition.
The taskprocessor execution was also slightly optimized to reduce the
number of locks taken.
The only observable difference in the taskprocessor implementation is
that when the final reference to the taskprocessor goes away, it will
execute all tasks to completion instead of discarding the unexecuted
tasks.
For systems where unnamed semaphores are not supported, a really
simple semaphore implementation is provided. (Which gives identical
performance as the original taskprocessor implementation).
The way we ended up implementing Stasis caused the threadpool to be a
burden instead of a boost to performance. This was switched to just
use taskprocessors directly for subscriptions.
Review: https://reviewboard.asterisk.org/r/2881/
........
r400180 | dlee | 2013-09-30 13:39:34 -0500 (Mon, 30 Sep 2013) | 28 lines
Optimize how Stasis forwards are dispatched
This patch optimizes how forwards are dispatched in Stasis.
Originally, forwards were dispatched as subscriptions that are invoked
on the publishing thread. This did not account for the vast number of
forwards we would end up having in the system, and the amount of work it
would take to walk though the forward subscriptions.
This patch modifies Stasis so that rather than walking the tree of
forwards on every dispatch, when forwards and subscriptions are changed,
the subscriber list for every topic in the tree is changed.
This has a couple of benefits. First, this reduces the workload of
dispatching messages. It also reduces contention when dispatching to
different topics that happen to forward to the same aggregation topic
(as happens with all of the channel, bridge and endpoint topics).
Since forwards are no longer subscriptions, the bulk of this patch is
simply changing stasis_subscription objects to stasis_forward objects
(which, admittedly, I should have done in the first place.)
Since this required me to yet again put in a growing array, I finally
abstracted that out into a set of ast_vector macros in
asterisk/vector.h.
Review: https://reviewboard.asterisk.org/r/2883/
........
r400181 | dlee | 2013-09-30 13:48:57 -0500 (Mon, 30 Sep 2013) | 28 lines
Remove dispatch object allocation from Stasis publishing
While looking for areas for performance improvement, I realized that an
unused feature in Stasis was negatively impacting performance.
When a message is sent to a subscriber, a dispatch object is allocated
for the dispatch, containing the topic the message was published to, the
subscriber the message is being sent to, and the message itself.
The topic is actually unused by any subscriber in Asterisk today. And
the subscriber is associated with the taskprocessor the message is being
dispatched to.
First, this patch removes the unused topic parameter from Stasis
subscription callbacks.
Second, this patch introduces the concept of taskprocessor local data,
data that may be set on a taskprocessor and provided along with the data
pointer when a task is pushed using the ast_taskprocessor_push_local()
call. This allows the task to have both data specific to that
taskprocessor, in addition to data specific to that invocation.
With those two changes, the dispatch object can be removed completely,
and the message is simply refcounted and sent directly to the
taskprocessor.
Review: https://reviewboard.asterisk.org/r/2884/
........
Merged revisions 399887,400138,400178,400180-400181 from http://svn.asterisk.org/svn/asterisk/branches/12
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@400186 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-09-30 18:55:27 +00:00
|
|
|
struct stasis_message *message)
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
{
|
|
|
|
/* no-op */
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(dtor_order)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test that destruction order doesn't bomb stuff";
|
|
|
|
info->description = "Test that destruction order doesn't bomb stuff";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
topic = stasis_topic_create("test-topic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
sub = stasis_subscribe(topic, noop, NULL);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
|
|
|
|
/* With any luck, this won't completely blow everything up */
|
|
|
|
ao2_cleanup(topic);
|
|
|
|
stasis_unsubscribe(sub);
|
|
|
|
|
|
|
|
/* These refs were cleaned up manually */
|
|
|
|
topic = NULL;
|
|
|
|
sub = NULL;
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
|
|
|
static const char *noop_get_id(struct stasis_message *message)
|
|
|
|
{
|
|
|
|
return NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_TEST_DEFINE(caching_dtor_order)
|
|
|
|
{
|
|
|
|
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
|
|
|
|
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL,
|
|
|
|
stasis_caching_unsubscribe);
|
|
|
|
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
|
|
|
|
|
|
|
|
switch (cmd) {
|
|
|
|
case TEST_INIT:
|
|
|
|
info->name = __func__;
|
|
|
|
info->category = test_category;
|
|
|
|
info->summary = "Test that destruction order doesn't bomb stuff";
|
|
|
|
info->description = "Test that destruction order doesn't bomb stuff";
|
|
|
|
return AST_TEST_NOT_RUN;
|
|
|
|
case TEST_EXECUTE:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
cache = stasis_cache_create(noop_get_id);
|
|
|
|
ast_test_validate(test, NULL != cache);
|
|
|
|
|
|
|
|
topic = stasis_topic_create("test-topic");
|
|
|
|
ast_test_validate(test, NULL != topic);
|
|
|
|
|
|
|
|
caching_topic = stasis_caching_topic_create(topic, cache);
|
|
|
|
ast_test_validate(test, NULL != caching_topic);
|
|
|
|
|
|
|
|
sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), noop,
|
|
|
|
NULL);
|
|
|
|
ast_test_validate(test, NULL != sub);
|
|
|
|
|
|
|
|
/* With any luck, this won't completely blow everything up */
|
|
|
|
ao2_cleanup(cache);
|
|
|
|
ao2_cleanup(topic);
|
|
|
|
stasis_caching_unsubscribe(caching_topic);
|
|
|
|
stasis_unsubscribe(sub);
|
|
|
|
|
|
|
|
/* These refs were cleaned up manually */
|
|
|
|
cache = NULL;
|
|
|
|
topic = NULL;
|
|
|
|
caching_topic = NULL;
|
|
|
|
sub = NULL;
|
|
|
|
|
|
|
|
return AST_TEST_PASS;
|
|
|
|
}
|
|
|
|
|
2013-03-08 15:15:13 +00:00
|
|
|
static int unload_module(void)
|
|
|
|
{
|
|
|
|
AST_TEST_UNREGISTER(message_type);
|
|
|
|
AST_TEST_UNREGISTER(message);
|
|
|
|
AST_TEST_UNREGISTER(subscription_messages);
|
|
|
|
AST_TEST_UNREGISTER(publish);
|
2014-01-12 22:07:01 +00:00
|
|
|
AST_TEST_UNREGISTER(publish_sync);
|
2013-03-08 15:15:13 +00:00
|
|
|
AST_TEST_UNREGISTER(unsubscribe_stops_messages);
|
|
|
|
AST_TEST_UNREGISTER(forward);
|
2013-08-06 14:28:23 +00:00
|
|
|
AST_TEST_UNREGISTER(cache_filter);
|
2013-03-08 15:15:13 +00:00
|
|
|
AST_TEST_UNREGISTER(cache);
|
2013-04-22 14:58:53 +00:00
|
|
|
AST_TEST_UNREGISTER(cache_dump);
|
2014-03-07 20:41:13 +00:00
|
|
|
AST_TEST_UNREGISTER(cache_eid_aggregate);
|
2013-03-15 17:35:16 +00:00
|
|
|
AST_TEST_UNREGISTER(router);
|
2013-07-23 13:39:50 +00:00
|
|
|
AST_TEST_UNREGISTER(router_cache_updates);
|
2013-04-01 13:37:51 +00:00
|
|
|
AST_TEST_UNREGISTER(interleaving);
|
2013-06-11 15:46:35 +00:00
|
|
|
AST_TEST_UNREGISTER(no_to_json);
|
|
|
|
AST_TEST_UNREGISTER(to_json);
|
|
|
|
AST_TEST_UNREGISTER(no_to_ami);
|
|
|
|
AST_TEST_UNREGISTER(to_ami);
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
AST_TEST_UNREGISTER(dtor_order);
|
|
|
|
AST_TEST_UNREGISTER(caching_dtor_order);
|
2013-03-08 15:15:13 +00:00
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
static int load_module(void)
|
|
|
|
{
|
|
|
|
AST_TEST_REGISTER(message_type);
|
|
|
|
AST_TEST_REGISTER(message);
|
|
|
|
AST_TEST_REGISTER(subscription_messages);
|
|
|
|
AST_TEST_REGISTER(publish);
|
2014-01-12 22:07:01 +00:00
|
|
|
AST_TEST_REGISTER(publish_sync);
|
2013-03-08 15:15:13 +00:00
|
|
|
AST_TEST_REGISTER(unsubscribe_stops_messages);
|
|
|
|
AST_TEST_REGISTER(forward);
|
2013-08-06 14:28:23 +00:00
|
|
|
AST_TEST_REGISTER(cache_filter);
|
2013-03-08 15:15:13 +00:00
|
|
|
AST_TEST_REGISTER(cache);
|
2013-04-22 14:58:53 +00:00
|
|
|
AST_TEST_REGISTER(cache_dump);
|
2014-03-07 20:41:13 +00:00
|
|
|
AST_TEST_REGISTER(cache_eid_aggregate);
|
2013-03-15 17:35:16 +00:00
|
|
|
AST_TEST_REGISTER(router);
|
2013-07-23 13:39:50 +00:00
|
|
|
AST_TEST_REGISTER(router_cache_updates);
|
2013-04-01 13:37:51 +00:00
|
|
|
AST_TEST_REGISTER(interleaving);
|
2013-06-11 15:46:35 +00:00
|
|
|
AST_TEST_REGISTER(no_to_json);
|
|
|
|
AST_TEST_REGISTER(to_json);
|
|
|
|
AST_TEST_REGISTER(no_to_ami);
|
|
|
|
AST_TEST_REGISTER(to_ami);
|
Stasis: address refcount races; implementation comments
Change r395954 reordered some stasis object destruction, which should
have been fine. Unfortunately, it caused some hard to reproduce issues
related to objects being accessed after they had been destroyed. The
patch in r396329 fixed the destruction order problem; this patch
addresses the underlying issue. A few other stasis-related fixes were
also added.
* Add ref-bumps around areas where objects may get transitively
destroyed. (For example, where we lock a topic, unref a subscription,
which unrefs the topic, which explodes the topic when we try to
unlock it.)
* Wrote an extensive doxygen page about Stasis implementation,
relationships between objects, lifecycles of objects, how the
refcounting works, etc. Many other comments were added, corrected, or
cleaned up.
* Added an assert to the topic dtor to catch extra ref decrements.
* Fixed type used after destruction errors for graceful shutdown in
stasis_channels.c.
* I added two unit tests in an attempt to catch destruction order
issues. Since the underlying cause is a race condition, though, the
tests rarely failed even when the code was wrong.
* Fixed a leak in stasis_cache_pattern.c.
(closes issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2746/
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396842 65c4cc65-6c06-0410-ace0-fbb531ad65f3
2013-08-16 16:03:34 +00:00
|
|
|
AST_TEST_REGISTER(dtor_order);
|
|
|
|
AST_TEST_REGISTER(caching_dtor_order);
|
2013-03-08 15:15:13 +00:00
|
|
|
return AST_MODULE_LOAD_SUCCESS;
|
|
|
|
}
|
|
|
|
|
|
|
|
AST_MODULE_INFO(ASTERISK_GPL_KEY, 0, "Stasis testing",
|
|
|
|
.load = load_module,
|
|
|
|
.unload = unload_module
|
|
|
|
);
|