bridges/bridge_multiplexed.c

   1 /*
   2  * Asterisk -- An open source telephony toolkit.
   3  *
   4  * Copyright (C) 2008, Digium, Inc.
   5  *
   6  * Joshua Colp <jcolp@digium.com>
   7  *
   8  * See http://www.asterisk.org for more information about
   9  * the Asterisk project. Please do not directly contact
  10  * any of the maintainers of this project for assistance;
  11  * the project provides a web site, mailing lists and IRC
  12  * channels for your use.
  13  *
  14  * This program is free software, distributed under the terms of
  15  * the GNU General Public License Version 2. See the LICENSE file
  16  * at the top of the source tree.
  17  */
  18
  19 /*! \file
  20  *
  21  * \brief Two channel bridging module which groups bridges into batches of threads
  22  *
  23  * \author Joshua Colp <jcolp@digium.com>
  24  *
  25  * \ingroup bridges
  26  */
  27
  28 /*** MODULEINFO
  29         <support_level>core</support_level>
  30  ***/
  31
  32 #include "asterisk.h"
  33
  34 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
  35
  36 #include <stdio.h>
  37 #include <stdlib.h>
  38 #include <string.h>
  39 #include <sys/types.h>
  40 #include <sys/stat.h>
  41 #include <fcntl.h>
  42
  43 #include "asterisk/module.h"
  44 #include "asterisk/channel.h"
  45 #include "asterisk/bridging.h"
  46 #include "asterisk/bridging_technology.h"
  47 #include "asterisk/frame.h"
  48 #include "asterisk/astobj2.h"
  49
  50 /*! \brief Number of buckets our multiplexed thread container can have */
  51 #define MULTIPLEXED_BUCKETS 53
  52
  53 /*! \brief Number of channels we handle in a single thread */
  54 #define MULTIPLEXED_MAX_CHANNELS 8
  55
  56 /*! \brief Structure which represents a single thread handling multiple 2 channel bridges */
  57 struct multiplexed_thread {
  58         /*! Thread itself */
  59         pthread_t thread;
  60         /*! Pipe used to wake up the multiplexed thread */
  61         int pipe[2];
  62         /*! Channels in this thread */
  63         struct ast_channel *chans[MULTIPLEXED_MAX_CHANNELS];
  64         /*! Number of channels in this thread */
  65         unsigned int count;
  66         /*! Bit used to indicate that the thread is waiting on channels */
  67         unsigned int waiting:1;
  68         /*! Number of channels actually being serviced by this thread */
  69         unsigned int service_count;
  70 };
  71
  72 /*! \brief Container of all operating multiplexed threads */
  73 static struct ao2_container *multiplexed_threads;
  74
  75 /*! \brief Callback function for finding a free multiplexed thread */
  76 static int find_multiplexed_thread(void *obj, void *arg, int flags)
  77 {
  78         struct multiplexed_thread *multiplexed_thread = obj;
  79         return (multiplexed_thread->count <= (MULTIPLEXED_MAX_CHANNELS - 2)) ? CMP_MATCH | CMP_STOP : 0;
  80 }
  81
  82 /*! \brief Destroy callback for a multiplexed thread structure */
  83 static void destroy_multiplexed_thread(void *obj)
  84 {
  85         struct multiplexed_thread *multiplexed_thread = obj;
  86
  87         if (multiplexed_thread->pipe[0] > -1) {
  88                 close(multiplexed_thread->pipe[0]);
  89         }
  90         if (multiplexed_thread->pipe[1] > -1) {
  91                 close(multiplexed_thread->pipe[1]);
  92         }
  93
  94         return;
  95 }
  96
  97 /*! \brief Create function which finds/reserves/references a multiplexed thread structure */
  98 static int multiplexed_bridge_create(struct ast_bridge *bridge)
  99 {
 100         struct multiplexed_thread *multiplexed_thread;
 101
 102         ao2_lock(multiplexed_threads);
 103
 104         /* Try to find an existing thread to handle our additional channels */
 105         if (!(multiplexed_thread = ao2_callback(multiplexed_threads, 0, find_multiplexed_thread, NULL))) {
 106                 int flags;
 107
 108                 /* If we failed we will have to create a new one from scratch */
 109                 if (!(multiplexed_thread = ao2_alloc(sizeof(*multiplexed_thread), destroy_multiplexed_thread))) {
 110                         ast_debug(1, "Failed to find or create a new multiplexed thread for bridge '%p'\n", bridge);
 111                         ao2_unlock(multiplexed_threads);
 112                         return -1;
 113                 }
 114
 115                 multiplexed_thread->pipe[0] = multiplexed_thread->pipe[1] = -1;
 116                 /* Setup a pipe so we can poke the thread itself when needed */
 117                 if (pipe(multiplexed_thread->pipe)) {
 118                         ast_debug(1, "Failed to create a pipe for poking a multiplexed thread for bridge '%p'\n", bridge);
 119                         ao2_ref(multiplexed_thread, -1);
 120                         ao2_unlock(multiplexed_threads);
 121                         return -1;
 122                 }
 123
 124                 /* Setup each pipe for non-blocking operation */
 125                 flags = fcntl(multiplexed_thread->pipe[0], F_GETFL);
 126                 if (fcntl(multiplexed_thread->pipe[0], F_SETFL, flags | O_NONBLOCK) < 0) {
 127                         ast_log(LOG_WARNING, "Failed to setup first nudge pipe for non-blocking operation on %p (%d: %s)\n", bridge, errno, strerror(errno));
 128                         ao2_ref(multiplexed_thread, -1);
 129                         ao2_unlock(multiplexed_threads);
 130                         return -1;
 131                 }
 132                 flags = fcntl(multiplexed_thread->pipe[1], F_GETFL);
 133                 if (fcntl(multiplexed_thread->pipe[1], F_SETFL, flags | O_NONBLOCK) < 0) {
 134                         ast_log(LOG_WARNING, "Failed to setup second nudge pipe for non-blocking operation on %p (%d: %s)\n", bridge, errno, strerror(errno));
 135                         ao2_ref(multiplexed_thread, -1);
 136                         ao2_unlock(multiplexed_threads);
 137                         return -1;
 138                 }
 139
 140                 /* Set up default parameters */
 141                 multiplexed_thread->thread = AST_PTHREADT_NULL;
 142
 143                 /* Finally link us into the container so others may find us */
 144                 ao2_link(multiplexed_threads, multiplexed_thread);
 145                 ast_debug(1, "Created multiplexed thread '%p' for bridge '%p'\n", multiplexed_thread, bridge);
 146         } else {
 147                 ast_debug(1, "Found multiplexed thread '%p' for bridge '%p'\n", multiplexed_thread, bridge);
 148         }
 149
 150         /* Bump the count of the thread structure up by two since the channels for this bridge will be joining shortly */
 151         multiplexed_thread->count += 2;
 152
 153         ao2_unlock(multiplexed_threads);
 154
 155         bridge->bridge_pvt = multiplexed_thread;
 156
 157         return 0;
 158 }
 159
 160 /*! \brief Internal function which nudges the thread */
 161 static void multiplexed_nudge(struct multiplexed_thread *multiplexed_thread)
 162 {
 163         int nudge = 0;
 164
 165         if (multiplexed_thread->thread == AST_PTHREADT_NULL) {
 166                 return;
 167         }
 168
 169         if (write(multiplexed_thread->pipe[1], &nudge, sizeof(nudge)) != sizeof(nudge)) {
 170                 ast_log(LOG_ERROR, "We couldn't poke multiplexed thread '%p'... something is VERY wrong\n", multiplexed_thread);
 171         }
 172
 173         while (multiplexed_thread->waiting) {
 174                 sched_yield();
 175         }
 176
 177         return;
 178 }
 179
 180 /*! \brief Destroy function which unreserves/unreferences/removes a multiplexed thread structure */
 181 static int multiplexed_bridge_destroy(struct ast_bridge *bridge)
 182 {
 183         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
 184
 185         ao2_lock(multiplexed_threads);
 186
 187         multiplexed_thread->count -= 2;
 188
 189         if (!multiplexed_thread->count) {
 190                 ast_debug(1, "Unlinking multiplexed thread '%p' since nobody is using it anymore\n", multiplexed_thread);
 191                 ao2_unlink(multiplexed_threads, multiplexed_thread);
 192         }
 193
 194         multiplexed_nudge(multiplexed_thread);
 195
 196         ao2_unlock(multiplexed_threads);
 197
 198         ao2_ref(multiplexed_thread, -1);
 199
 200         return 0;
 201 }
 202
 203 /*! \brief Thread function that executes for multiplexed threads */
 204 static void *multiplexed_thread_function(void *data)
 205 {
 206         struct multiplexed_thread *multiplexed_thread = data;
 207         int fds = multiplexed_thread->pipe[0];
 208
 209         ao2_lock(multiplexed_thread);
 210
 211         ast_debug(1, "Starting actual thread for multiplexed thread '%p'\n", multiplexed_thread);
 212
 213         while (multiplexed_thread->thread != AST_PTHREADT_STOP) {
 214                 struct ast_channel *winner = NULL, *first = multiplexed_thread->chans[0];
 215                 int to = -1, outfd = -1;
 216
 217                 /* Move channels around so not just the first one gets priority */
 218                 memmove(multiplexed_thread->chans, multiplexed_thread->chans + 1, sizeof(struct ast_channel *) * (multiplexed_thread->service_count - 1));
 219                 multiplexed_thread->chans[multiplexed_thread->service_count - 1] = first;
 220
 221                 multiplexed_thread->waiting = 1;
 222                 ao2_unlock(multiplexed_thread);
 223                 winner = ast_waitfor_nandfds(multiplexed_thread->chans, multiplexed_thread->service_count, &fds, 1, NULL, &outfd, &to);
 224                 multiplexed_thread->waiting = 0;
 225                 ao2_lock(multiplexed_thread);
 226                 if (multiplexed_thread->thread == AST_PTHREADT_STOP) {
 227                         break;
 228                 }
 229
 230                 if (outfd > -1) {
 231                         int nudge;
 232
 233                         if (read(multiplexed_thread->pipe[0], &nudge, sizeof(nudge)) < 0) {
 234                                 if (errno != EINTR && errno != EAGAIN) {
 235                                         ast_log(LOG_WARNING, "read() failed for pipe on multiplexed thread '%p': %s\n", multiplexed_thread, strerror(errno));
 236                                 }
 237                         }
 238                 }
 239                 if (winner && winner->bridge) {
 240                         struct ast_bridge *bridge = winner->bridge;
 241                         int stop = 0;
 242                         ao2_unlock(multiplexed_thread);
 243                         while ((bridge = winner->bridge) && ao2_trylock(bridge)) {
 244                                 sched_yield();
 245                                 if (multiplexed_thread->thread == AST_PTHREADT_STOP) {
 246                                         stop = 1;
 247                                         break;
 248                                 }
 249                         }
 250                         if (!stop && bridge) {
 251                                 ast_bridge_handle_trip(bridge, NULL, winner, -1);
 252                                 ao2_unlock(bridge);
 253                         }
 254                         ao2_lock(multiplexed_thread);
 255                 }
 256         }
 257
 258         multiplexed_thread->thread = AST_PTHREADT_NULL;
 259
 260         ast_debug(1, "Stopping actual thread for multiplexed thread '%p'\n", multiplexed_thread);
 261
 262         ao2_unlock(multiplexed_thread);
 263         ao2_ref(multiplexed_thread, -1);
 264
 265         return NULL;
 266 }
 267
 268 /*! \brief Helper function which adds or removes a channel and nudges the thread */
 269 static void multiplexed_add_or_remove(struct multiplexed_thread *multiplexed_thread, struct ast_channel *chan, int add)
 270 {
 271         int i, removed = 0;
 272         pthread_t thread = AST_PTHREADT_NULL;
 273
 274         ao2_lock(multiplexed_thread);
 275
 276         multiplexed_nudge(multiplexed_thread);
 277
 278         for (i = 0; i < MULTIPLEXED_MAX_CHANNELS; i++) {
 279                 if (multiplexed_thread->chans[i] == chan) {
 280                         if (!add) {
 281                                 multiplexed_thread->chans[i] = NULL;
 282                                 multiplexed_thread->service_count--;
 283                                 removed = 1;
 284                         }
 285                         break;
 286                 } else if (!multiplexed_thread->chans[i] && add) {
 287                         multiplexed_thread->chans[i] = chan;
 288                         multiplexed_thread->service_count++;
 289                         break;
 290                 }
 291         }
 292
 293         if (multiplexed_thread->service_count && multiplexed_thread->thread == AST_PTHREADT_NULL) {
 294                 ao2_ref(multiplexed_thread, +1);
 295                 if (ast_pthread_create(&multiplexed_thread->thread, NULL, multiplexed_thread_function, multiplexed_thread)) {
 296                         ao2_ref(multiplexed_thread, -1);
 297                         ast_debug(1, "Failed to create an actual thread for multiplexed thread '%p', trying next time\n", multiplexed_thread);
 298                 }
 299         } else if (!multiplexed_thread->service_count && multiplexed_thread->thread != AST_PTHREADT_NULL) {
 300                 thread = multiplexed_thread->thread;
 301                 multiplexed_thread->thread = AST_PTHREADT_STOP;
 302         } else if (!add && removed) {
 303                 memmove(multiplexed_thread->chans + i, multiplexed_thread->chans + i + 1, sizeof(struct ast_channel *) * (MULTIPLEXED_MAX_CHANNELS - (i + 1)));
 304         }
 305
 306         ao2_unlock(multiplexed_thread);
 307
 308         if (thread != AST_PTHREADT_NULL) {
 309                 pthread_join(thread, NULL);
 310         }
 311
 312         return;
 313 }
 314
 315 /*! \brief Join function which actually adds the channel into the array to be monitored */
 316 static int multiplexed_bridge_join(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
 317 {
 318         struct ast_channel *c0 = AST_LIST_FIRST(&bridge->channels)->chan, *c1 = AST_LIST_LAST(&bridge->channels)->chan;
 319         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
 320
 321         ast_debug(1, "Adding channel '%s' to multiplexed thread '%p' for monitoring\n", ast_channel_name(bridge_channel->chan), multiplexed_thread);
 322
 323         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 1);
 324
 325         /* If the second channel has not yet joined do not make things compatible */
 326         if (c0 == c1) {
 327                 return 0;
 328         }
 329
 330         if ((ast_format_cmp(&c0->writeformat, &c1->readformat) == AST_FORMAT_CMP_EQUAL) &&
 331                 (ast_format_cmp(&c0->readformat, &c1->writeformat) == AST_FORMAT_CMP_EQUAL) &&
 332                 (ast_format_cap_identical(ast_channel_nativeformats(c0), ast_channel_nativeformats(c1)))) {
 333                 return 0;
 334         }
 335
 336         return ast_channel_make_compatible(c0, c1);
 337 }
 338
 339 /*! \brief Leave function which actually removes the channel from the array */
 340 static int multiplexed_bridge_leave(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
 341 {
 342         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
 343
 344         ast_debug(1, "Removing channel '%s' from multiplexed thread '%p'\n", ast_channel_name(bridge_channel->chan), multiplexed_thread);
 345
 346         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 0);
 347
 348         return 0;
 349 }
 350
 351 /*! \brief Suspend function which means control of the channel is going elsewhere */
 352 static void multiplexed_bridge_suspend(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
 353 {
 354         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
 355
 356         ast_debug(1, "Suspending channel '%s' from multiplexed thread '%p'\n", ast_channel_name(bridge_channel->chan), multiplexed_thread);
 357
 358         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 0);
 359
 360         return;
 361 }
 362
 363 /*! \brief Unsuspend function which means control of the channel is coming back to us */
 364 static void multiplexed_bridge_unsuspend(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
 365 {
 366         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
 367
 368         ast_debug(1, "Unsuspending channel '%s' from multiplexed thread '%p'\n", ast_channel_name(bridge_channel->chan), multiplexed_thread);
 369
 370         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 1);
 371
 372         return;
 373 }
 374
 375 /*! \brief Write function for writing frames into the bridge */
 376 static enum ast_bridge_write_result multiplexed_bridge_write(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel, struct ast_frame *frame)
 377 {
 378         struct ast_bridge_channel *other;
 379
 380         if (AST_LIST_FIRST(&bridge->channels) == AST_LIST_LAST(&bridge->channels)) {
 381                 return AST_BRIDGE_WRITE_FAILED;
 382         }
 383
 384         if (!(other = (AST_LIST_FIRST(&bridge->channels) == bridge_channel ? AST_LIST_LAST(&bridge->channels) : AST_LIST_FIRST(&bridge->channels)))) {
 385                 return AST_BRIDGE_WRITE_FAILED;
 386         }
 387
 388         if (other->state == AST_BRIDGE_CHANNEL_STATE_WAIT) {
 389                 ast_write(other->chan, frame);
 390         }
 391
 392         return AST_BRIDGE_WRITE_SUCCESS;
 393 }
 394
 395 static struct ast_bridge_technology multiplexed_bridge = {
 396         .name = "multiplexed_bridge",
 397         .capabilities = AST_BRIDGE_CAPABILITY_1TO1MIX,
 398         .preference = AST_BRIDGE_PREFERENCE_HIGH,
 399         .create = multiplexed_bridge_create,
 400         .destroy = multiplexed_bridge_destroy,
 401         .join = multiplexed_bridge_join,
 402         .leave = multiplexed_bridge_leave,
 403         .suspend = multiplexed_bridge_suspend,
 404         .unsuspend = multiplexed_bridge_unsuspend,
 405         .write = multiplexed_bridge_write,
 406 };
 407
 408 static int unload_module(void)
 409 {
 410         int res = ast_bridge_technology_unregister(&multiplexed_bridge);
 411
 412         ao2_ref(multiplexed_threads, -1);
 413         multiplexed_bridge.format_capabilities = ast_format_cap_destroy(multiplexed_bridge.format_capabilities);
 414
 415         return res;
 416 }
 417
 418 static int load_module(void)
 419 {
 420         if (!(multiplexed_threads = ao2_container_alloc(MULTIPLEXED_BUCKETS, NULL, NULL))) {
 421                 return AST_MODULE_LOAD_DECLINE;
 422         }
 423         if (!(multiplexed_bridge.format_capabilities = ast_format_cap_alloc())) {
 424                 return AST_MODULE_LOAD_DECLINE;
 425         }
 426         ast_format_cap_add_all_by_type(multiplexed_bridge.format_capabilities, AST_FORMAT_TYPE_AUDIO);
 427         ast_format_cap_add_all_by_type(multiplexed_bridge.format_capabilities, AST_FORMAT_TYPE_VIDEO);
 428         ast_format_cap_add_all_by_type(multiplexed_bridge.format_capabilities, AST_FORMAT_TYPE_TEXT);
 429         return ast_bridge_technology_register(&multiplexed_bridge);
 430 }
 431
 432 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Multiplexed two channel bridging module");