[Asterisk-code-review] bridge: Add a deferred queue. (asterisk[14])
Jenkins2
asteriskteam at digium.com
Thu Jun 15 14:58:24 CDT 2017
Jenkins2 has submitted this change and it was merged. ( https://gerrit.asterisk.org/5800 )
Change subject: bridge: Add a deferred queue.
......................................................................
bridge: Add a deferred queue.
This change adds a deferred queue to bridging. If a bridge
technology determines that a frame can not be written and
should be deferred it can indicate back to bridging to do so.
Bridging will then requeue any deferred frames upon a new
channel joining the bridge.
This change has been leveraged for T.38 request negotiate
control frames. Without the deferred queue there is a race
condition between the bridge receiving the T.38 request
negotiate and the second channel joining and being in the
bridge. If the channel is not yet in the bridge then the T.38
negotiation fails.
A unit test has also been added that confirms that a T.38
request negotiate control frame is deferred when no other
channel is in the bridge and that it is requeued when a new
channel joins the bridge.
ASTERISK-26923
Change-Id: Ie05b08523f399eae579130f4a5f562a344d2e415
---
M bridges/bridge_native_rtp.c
M bridges/bridge_simple.c
M include/asterisk/bridge_channel.h
M include/asterisk/bridge_channel_internal.h
M include/asterisk/bridge_technology.h
M main/bridge.c
M main/bridge_channel.c
A tests/test_bridging.c
8 files changed, 397 insertions(+), 10 deletions(-)
Approvals:
George Joseph: Looks good to me, but someone else must approve
Kevin Harwell: Looks good to me, approved
Jenkins2: Approved for Submit
diff --git a/bridges/bridge_native_rtp.c b/bridges/bridge_native_rtp.c
index 919df83..832e397 100644
--- a/bridges/bridge_native_rtp.c
+++ b/bridges/bridge_native_rtp.c
@@ -510,7 +510,37 @@
static int native_rtp_bridge_write(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel, struct ast_frame *frame)
{
- return ast_bridge_queue_everyone_else(bridge, bridge_channel, frame);
+ const struct ast_control_t38_parameters *t38_parameters;
+ int defer = 0;
+
+ if (!ast_bridge_queue_everyone_else(bridge, bridge_channel, frame)) {
+ /* This frame was successfully queued so no need to defer */
+ return 0;
+ }
+
+ /* Depending on the frame defer it so when the next channel joins it receives it */
+ switch (frame->frametype) {
+ case AST_FRAME_CONTROL:
+ switch (frame->subclass.integer) {
+ case AST_CONTROL_T38_PARAMETERS:
+ t38_parameters = frame->data.ptr;
+ switch (t38_parameters->request_response) {
+ case AST_T38_REQUEST_NEGOTIATE:
+ defer = -1;
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+
+ return defer;
}
static struct ast_bridge_technology native_rtp_bridge = {
diff --git a/bridges/bridge_simple.c b/bridges/bridge_simple.c
index fc12bd1..1358056 100644
--- a/bridges/bridge_simple.c
+++ b/bridges/bridge_simple.c
@@ -63,7 +63,37 @@
static int simple_bridge_write(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel, struct ast_frame *frame)
{
- return ast_bridge_queue_everyone_else(bridge, bridge_channel, frame);
+ const struct ast_control_t38_parameters *t38_parameters;
+ int defer = 0;
+
+ if (!ast_bridge_queue_everyone_else(bridge, bridge_channel, frame)) {
+ /* This frame was successfully queued so no need to defer */
+ return 0;
+ }
+
+ /* Depending on the frame defer it so when the next channel joins it receives it */
+ switch (frame->frametype) {
+ case AST_FRAME_CONTROL:
+ switch (frame->subclass.integer) {
+ case AST_CONTROL_T38_PARAMETERS:
+ t38_parameters = frame->data.ptr;
+ switch (t38_parameters->request_response) {
+ case AST_T38_REQUEST_NEGOTIATE:
+ defer = -1;
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+
+ return defer;
}
static struct ast_bridge_technology simple_bridge = {
diff --git a/include/asterisk/bridge_channel.h b/include/asterisk/bridge_channel.h
index 797be4e..bead207 100644
--- a/include/asterisk/bridge_channel.h
+++ b/include/asterisk/bridge_channel.h
@@ -145,6 +145,8 @@
AST_LIST_ENTRY(ast_bridge_channel) entry;
/*! Queue of outgoing frames to the channel. */
AST_LIST_HEAD_NOLOCK(, ast_frame) wr_queue;
+ /*! Queue of deferred frames, queued onto channel when other party joins. */
+ AST_LIST_HEAD_NOLOCK(, ast_frame) deferred_queue;
/*! Pipe to alert thread when frames are put into the wr_queue. */
int alert_pipe[2];
/*!
diff --git a/include/asterisk/bridge_channel_internal.h b/include/asterisk/bridge_channel_internal.h
index fb8e781..ba71e9f 100644
--- a/include/asterisk/bridge_channel_internal.h
+++ b/include/asterisk/bridge_channel_internal.h
@@ -98,6 +98,17 @@
/*!
* \internal
+ * \brief Queue any deferred frames on the channel.
+ * \since 13.17.0
+ *
+ * \param bridge_channel Channel that the deferred frames should be pulled from and queued to.
+ *
+ * \return Nothing
+ */
+void bridge_channel_queue_deferred_frames(struct ast_bridge_channel *bridge_channel);
+
+/*!
+ * \internal
* \brief Push the bridge channel into its specified bridge.
* \since 12.0.0
*
diff --git a/include/asterisk/bridge_technology.h b/include/asterisk/bridge_technology.h
index 843d93c..fb275c1 100644
--- a/include/asterisk/bridge_technology.h
+++ b/include/asterisk/bridge_technology.h
@@ -156,6 +156,9 @@
* \retval -1 Frame needs to be deferred.
*
* \note On entry, bridge is already locked.
+ *
+ * \note Deferred frames will be automatically queued onto the channel when another
+ * channel joins the bridge.
*/
int (*write)(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel, struct ast_frame *frame);
/*! TRUE if the bridge technology is currently suspended. */
diff --git a/main/bridge.c b/main/bridge.c
index ef0da01..740a88f 100644
--- a/main/bridge.c
+++ b/main/bridge.c
@@ -471,6 +471,7 @@
}
AST_LIST_TRAVERSE(&bridge->channels, bridge_channel, entry) {
+ bridge_channel_queue_deferred_frames(bridge_channel);
if (!bridge_channel->just_joined) {
continue;
}
diff --git a/main/bridge_channel.c b/main/bridge_channel.c
index f954688..875d37d 100644
--- a/main/bridge_channel.c
+++ b/main/bridge_channel.c
@@ -639,18 +639,21 @@
static int bridge_channel_write_frame(struct ast_bridge_channel *bridge_channel, struct ast_frame *frame)
{
const struct ast_control_t38_parameters *t38_parameters;
+ int deferred;
ast_assert(frame->frametype != AST_FRAME_BRIDGE_ACTION_SYNC);
ast_bridge_channel_lock_bridge(bridge_channel);
-/*
- * XXX need to implement a deferred write queue for when there
- * is no peer channel in the bridge (yet or it was kicked).
- *
- * The tech decides if a frame needs to be pushed back for deferral.
- * simple_bridge/native_bridge are likely the only techs that will do this.
- */
- bridge_channel->bridge->technology->write(bridge_channel->bridge, bridge_channel, frame);
+
+ deferred = bridge_channel->bridge->technology->write(bridge_channel->bridge, bridge_channel, frame);
+ if (deferred) {
+ struct ast_frame *dup;
+
+ dup = ast_frdup(frame);
+ if (dup) {
+ AST_LIST_INSERT_HEAD(&bridge_channel->deferred_queue, dup, frame_list);
+ }
+ }
/* Remember any owed events to the bridge. */
switch (frame->frametype) {
@@ -752,6 +755,18 @@
bridge_channel->owed.t38_terminate = 0;
orig_bridge->technology->write(orig_bridge, NULL, &frame);
}
+}
+
+void bridge_channel_queue_deferred_frames(struct ast_bridge_channel *bridge_channel)
+{
+ struct ast_frame *frame;
+
+ ast_channel_lock(bridge_channel->chan);
+ while ((frame = AST_LIST_REMOVE_HEAD(&bridge_channel->deferred_queue, frame_list))) {
+ ast_queue_frame_head(bridge_channel->chan, frame);
+ ast_frfree(frame);
+ }
+ ast_channel_unlock(bridge_channel->chan);
}
/*!
@@ -2883,6 +2898,11 @@
}
ast_alertpipe_close(bridge_channel->alert_pipe);
+ /* Flush any unhandled deferred_queue frames. */
+ while ((fr = AST_LIST_REMOVE_HEAD(&bridge_channel->deferred_queue, frame_list))) {
+ ast_frfree(fr);
+ }
+
ast_cond_destroy(&bridge_channel->cond);
ao2_cleanup(bridge_channel->write_format);
diff --git a/tests/test_bridging.c b/tests/test_bridging.c
new file mode 100644
index 0000000..74595c4
--- /dev/null
+++ b/tests/test_bridging.c
@@ -0,0 +1,290 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2017, Digium, Inc.
+ *
+ * Joshua Colp <jcolp at 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 Bridging unit tests
+ *
+ * \author Joshua Colp <jcolp at digium.com>
+ *
+ */
+
+/*** MODULEINFO
+ <depend>TEST_FRAMEWORK</depend>
+ <support_level>core</support_level>
+ ***/
+
+#include "asterisk.h"
+
+#include "asterisk/module.h"
+#include "asterisk/test.h"
+#include "asterisk/channel.h"
+#include "asterisk/time.h"
+#include "asterisk/bridge.h"
+#include "asterisk/bridge_basic.h"
+#include "asterisk/features.h"
+#include "asterisk/format_cache.h"
+
+#define TEST_CATEGORY "/main/bridging/"
+
+#define CHANNEL_TECH_NAME "BridgingTestChannel"
+
+#define TEST_CHANNEL_FORMAT ast_format_slin
+
+/*! \brief A private structure for the test channel */
+struct test_bridging_chan_pvt {
+ /* \brief The expected indication */
+ int condition;
+ /*! \brief The number of indicated things */
+ unsigned int indicated;
+};
+
+/*! \brief Callback function for when a frame is written to a channel */
+static int test_bridging_chan_indicate(struct ast_channel *chan, int condition, const void *data, size_t datalen)
+{
+ struct test_bridging_chan_pvt *test_pvt = ast_channel_tech_pvt(chan);
+
+ if (condition == test_pvt->condition) {
+ test_pvt->indicated++;
+ }
+
+ return 0;
+}
+
+/*! \brief Callback function for when a channel is hung up */
+static int test_bridging_chan_hangup(struct ast_channel *chan)
+{
+ struct test_bridging_chan_pvt *test_pvt = ast_channel_tech_pvt(chan);
+
+ ast_free(test_pvt);
+ ast_channel_tech_pvt_set(chan, NULL);
+
+ return 0;
+}
+
+/*! \brief A channel technology used for the unit tests */
+static struct ast_channel_tech test_bridging_chan_tech = {
+ .type = CHANNEL_TECH_NAME,
+ .description = "Mock channel technology for bridge tests",
+ .indicate = test_bridging_chan_indicate,
+ .hangup = test_bridging_chan_hangup,
+ .properties = AST_CHAN_TP_INTERNAL,
+};
+
+static void test_nanosleep(int secs, long nanosecs)
+{
+ struct timespec sleep_time = {secs, nanosecs};
+
+ while ((nanosleep(&sleep_time, &sleep_time) == -1) && (errno == EINTR)) {
+ }
+}
+
+/*! \brief Wait until a channel is bridged */
+static void wait_for_bridged(struct ast_channel *channel)
+{
+ ast_channel_lock(channel);
+ while (!ast_channel_is_bridged(channel)) {
+ ast_channel_unlock(channel);
+ test_nanosleep(0, 1000000);
+ ast_channel_lock(channel);
+ }
+ ast_channel_unlock(channel);
+}
+
+/*! \brief Wait until a channel is not bridged */
+static void wait_for_unbridged(struct ast_channel *channel)
+{
+ ast_channel_lock(channel);
+ while (ast_channel_is_bridged(channel)) {
+ ast_channel_unlock(channel);
+ test_nanosleep(0, 1000000);
+ ast_channel_lock(channel);
+ }
+ ast_channel_unlock(channel);
+}
+
+/*! \brief Wait until a channel has no frames on its read queue */
+static void wait_for_empty_queue(struct ast_channel *channel)
+{
+ ast_channel_lock(channel);
+ while (!AST_LIST_EMPTY(ast_channel_readq(channel))) {
+ ast_channel_unlock(channel);
+ test_nanosleep(0, 1000000);
+ ast_channel_lock(channel);
+ }
+ ast_channel_unlock(channel);
+}
+
+/*! \brief Create a \ref test_bridging_chan_tech for Alice. */
+#define START_ALICE(channel, pvt) START_CHANNEL(channel, pvt, "Alice", "100")
+
+/*! \brief Create a \ref test_bridging_chan_tech for Bob. */
+#define START_BOB(channel, pvt) START_CHANNEL(channel, pvt, "Bob", "200")
+
+#define START_CHANNEL(channel, pvt, name, number) do { \
+ channel = ast_channel_alloc(0, AST_STATE_UP, number, name, number, number, \
+ "default", NULL, NULL, 0, CHANNEL_TECH_NAME "/" name); \
+ pvt = ast_calloc(1, sizeof(*pvt)); \
+ ast_channel_tech_pvt_set(channel, pvt); \
+ ast_channel_nativeformats_set(channel, test_bridging_chan_tech.capabilities); \
+ ast_channel_set_rawwriteformat(channel, TEST_CHANNEL_FORMAT); \
+ ast_channel_set_rawreadformat(channel, TEST_CHANNEL_FORMAT); \
+ ast_channel_set_writeformat(channel, TEST_CHANNEL_FORMAT); \
+ ast_channel_set_readformat(channel, TEST_CHANNEL_FORMAT); \
+ ast_channel_unlock(channel); \
+ } while (0)
+
+/*! \brief Hang up a test channel safely */
+#define HANGUP_CHANNEL(channel) do { \
+ ao2_ref(channel, +1); \
+ ast_hangup((channel)); \
+ ao2_cleanup(channel); \
+ channel = NULL; \
+ } while (0)
+
+static void safe_channel_release(struct ast_channel *chan)
+{
+ if (!chan) {
+ return;
+ }
+ ast_channel_release(chan);
+}
+
+static void safe_bridge_destroy(struct ast_bridge *bridge)
+{
+ if (!bridge) {
+ return;
+ }
+ ast_bridge_destroy(bridge, 0);
+}
+
+static void stream_periodic_frames(struct ast_channel *chan, int ms, int interval_ms)
+{
+ long nanosecs;
+
+ ast_assert(chan != NULL);
+ ast_assert(0 < ms);
+ ast_assert(0 < interval_ms);
+
+ nanosecs = interval_ms * 1000000L;
+ while (0 < ms) {
+ ast_queue_frame(chan, &ast_null_frame);
+
+ if (interval_ms < ms) {
+ ms -= interval_ms;
+ } else {
+ nanosecs = ms * 1000000L;
+ ms = 0;
+ }
+ test_nanosleep(0, nanosecs);
+ }
+}
+
+AST_TEST_DEFINE(test_bridging_deferred_queue)
+{
+ RAII_VAR(struct ast_channel *, chan_alice, NULL, safe_channel_release);
+ struct test_bridging_chan_pvt *alice_pvt;
+ struct ast_control_t38_parameters t38_parameters = {
+ .request_response = AST_T38_REQUEST_NEGOTIATE,
+ };
+ struct ast_frame frame = {
+ .frametype = AST_FRAME_CONTROL,
+ .subclass.integer = AST_CONTROL_T38_PARAMETERS,
+ .data.ptr = &t38_parameters,
+ .datalen = sizeof(t38_parameters),
+ };
+ RAII_VAR(struct ast_channel *, chan_bob, NULL, safe_channel_release);
+ struct test_bridging_chan_pvt *bob_pvt;
+ RAII_VAR(struct ast_bridge *, bridge1, NULL, safe_bridge_destroy);
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = TEST_CATEGORY;
+ info->summary = "Test that deferred frames from a channel in a bridge get written";
+ info->description =
+ "This test creates two channels, queues a deferrable frame on one, places it into\n"
+ "a bridge, confirms the frame was read by the bridge, adds the second channel to the\n"
+ "bridge, and makes sure the deferred frame is written to it.";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ /* Create the bridges */
+ bridge1 = ast_bridge_basic_new();
+ ast_test_validate(test, bridge1 != NULL);
+
+ /* Create channels that will go into the bridge */
+ START_ALICE(chan_alice, alice_pvt);
+ START_BOB(chan_bob, bob_pvt);
+ bob_pvt->condition = AST_CONTROL_T38_PARAMETERS;
+
+ /* Bridge alice and wait for the frame to be deferred */
+ ast_test_validate(test, !ast_bridge_impart(bridge1, chan_alice, NULL, NULL, AST_BRIDGE_IMPART_CHAN_DEPARTABLE));
+ wait_for_bridged(chan_alice);
+ ast_queue_frame(chan_alice, &frame);
+ wait_for_empty_queue(chan_alice);
+
+ /* Bridge bob for a second so it can receive the deferred T.38 request negotiate frame */
+ ast_test_validate(test, !ast_bridge_impart(bridge1, chan_bob, NULL, NULL, AST_BRIDGE_IMPART_CHAN_DEPARTABLE));
+ wait_for_bridged(chan_bob);
+ stream_periodic_frames(chan_alice, 1000, 20);
+ ast_test_validate(test, !ast_bridge_depart(chan_bob));
+ wait_for_unbridged(chan_bob);
+
+ /* Ensure that we received the expected indications while it was in there (request to negotiate, and to terminate) */
+ ast_test_validate(test, bob_pvt->indicated == 2);
+
+ /* Now remove alice since we are done */
+ ast_test_validate(test, !ast_bridge_depart(chan_alice));
+ wait_for_unbridged(chan_alice);
+
+ /* Hangup the channels */
+ HANGUP_CHANNEL(chan_alice);
+ HANGUP_CHANNEL(chan_bob);
+
+ return AST_TEST_PASS;
+}
+
+static int unload_module(void)
+{
+ AST_TEST_UNREGISTER(test_bridging_deferred_queue);
+
+ ast_channel_unregister(&test_bridging_chan_tech);
+ ao2_cleanup(test_bridging_chan_tech.capabilities);
+ test_bridging_chan_tech.capabilities = NULL;
+
+ return 0;
+}
+
+static int load_module(void)
+{
+ test_bridging_chan_tech.capabilities = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
+ if (!test_bridging_chan_tech.capabilities) {
+ return AST_MODULE_LOAD_DECLINE;
+ }
+ ast_format_cap_append(test_bridging_chan_tech.capabilities, TEST_CHANNEL_FORMAT, 0);
+ ast_channel_register(&test_bridging_chan_tech);
+
+ AST_TEST_REGISTER(test_bridging_deferred_queue);
+
+ return AST_MODULE_LOAD_SUCCESS;
+}
+
+AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Bridging Unit Tests");
--
To view, visit https://gerrit.asterisk.org/5800
To unsubscribe, visit https://gerrit.asterisk.org/settings
Gerrit-Project: asterisk
Gerrit-Branch: 14
Gerrit-MessageType: merged
Gerrit-Change-Id: Ie05b08523f399eae579130f4a5f562a344d2e415
Gerrit-Change-Number: 5800
Gerrit-PatchSet: 3
Gerrit-Owner: Joshua Colp <jcolp at digium.com>
Gerrit-Reviewer: George Joseph <gjoseph at digium.com>
Gerrit-Reviewer: Jenkins2
Gerrit-Reviewer: Kevin Harwell <kharwell at digium.com>
Gerrit-Reviewer: Mark Michelson <mmichelson at digium.com>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.digium.com/pipermail/asterisk-code-review/attachments/20170615/fdc66a91/attachment-0001.html>
More information about the asterisk-code-review
mailing list