diff options
author | Jenkins2 <jenkins2@gerrit.asterisk.org> | 2017-06-15 14:48:39 -0500 |
---|---|---|
committer | Gerrit Code Review <gerrit2@gerrit.digium.api> | 2017-06-15 14:48:39 -0500 |
commit | 5fa52f0b5a9c55746ad60403940903a0a7b8f276 (patch) | |
tree | b1943adf587ed4ec2fa3a50ff632111315a16b78 | |
parent | 8b12ad2f61e807f2138d0a80927e1c4859c86c4c (diff) | |
parent | e414833f6e77345f4969116972e9cf1ad9b595fd (diff) |
Merge "bridge: Add a deferred queue." into 13
-rw-r--r-- | bridges/bridge_native_rtp.c | 32 | ||||
-rw-r--r-- | bridges/bridge_simple.c | 32 | ||||
-rw-r--r-- | include/asterisk/bridge_channel.h | 2 | ||||
-rw-r--r-- | include/asterisk/bridge_channel_internal.h | 11 | ||||
-rw-r--r-- | include/asterisk/bridge_technology.h | 3 | ||||
-rw-r--r-- | main/bridge.c | 1 | ||||
-rw-r--r-- | main/bridge_channel.c | 36 | ||||
-rw-r--r-- | tests/test_bridging.c | 292 |
8 files changed, 399 insertions, 10 deletions
diff --git a/bridges/bridge_native_rtp.c b/bridges/bridge_native_rtp.c index a80ef4c5a..78e35a16b 100644 --- a/bridges/bridge_native_rtp.c +++ b/bridges/bridge_native_rtp.c @@ -510,7 +510,37 @@ static void native_rtp_bridge_leave(struct ast_bridge *bridge, struct ast_bridge 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 570453500..3e2a73e46 100644 --- a/bridges/bridge_simple.c +++ b/bridges/bridge_simple.c @@ -63,7 +63,37 @@ static int simple_bridge_join(struct ast_bridge *bridge, struct ast_bridge_chann 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 55c2b3a76..a16695e07 100644 --- a/include/asterisk/bridge_channel.h +++ b/include/asterisk/bridge_channel.h @@ -145,6 +145,8 @@ struct ast_bridge_channel { 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 fb8e781e8..ba71e9fc4 100644 --- a/include/asterisk/bridge_channel_internal.h +++ b/include/asterisk/bridge_channel_internal.h @@ -98,6 +98,17 @@ void bridge_channel_settle_owed_events(struct ast_bridge *orig_bridge, struct as /*! * \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 402b54e98..5add4551f 100644 --- a/include/asterisk/bridge_technology.h +++ b/include/asterisk/bridge_technology.h @@ -156,6 +156,9 @@ struct ast_bridge_technology { * \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); /*! diff --git a/main/bridge.c b/main/bridge.c index 7f6fbbef9..b2beb8637 100644 --- a/main/bridge.c +++ b/main/bridge.c @@ -471,6 +471,7 @@ static void bridge_complete_join(struct ast_bridge *bridge) } 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 eba5ae40a..0af688ad4 100644 --- a/main/bridge_channel.c +++ b/main/bridge_channel.c @@ -639,18 +639,21 @@ void ast_bridge_channel_kick(struct ast_bridge_channel *bridge_channel, int caus 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) { @@ -754,6 +757,18 @@ void bridge_channel_settle_owed_events(struct ast_bridge *orig_bridge, struct as } } +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); +} + /*! * \internal * \brief Suspend a channel from a bridge. @@ -2854,6 +2869,11 @@ static void bridge_channel_destroy(void *obj) } 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 000000000..08a2fcc1f --- /dev/null +++ b/tests/test_bridging.c @@ -0,0 +1,292 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2017, Digium, Inc. + * + * Joshua Colp <jcolp@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@digium.com> + * + */ + +/*** MODULEINFO + <depend>TEST_FRAMEWORK</depend> + <support_level>core</support_level> + ***/ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$") + +#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"); |