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