Two channel bridging module which groups bridges into batches of threads. More...
#include "asterisk.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "asterisk/module.h"
#include "asterisk/channel.h"
#include "asterisk/bridging.h"
#include "asterisk/bridging_technology.h"
#include "asterisk/frame.h"
#include "asterisk/astobj2.h"
Go to the source code of this file.
Data Structures | |
struct | multiplexed_thread |
Structure which represents a single thread handling multiple 2 channel bridges. More... | |
Defines | |
#define | MULTIPLEXED_BUCKETS 53 |
Number of buckets our multiplexed thread container can have. | |
#define | MULTIPLEXED_MAX_CHANNELS 8 |
Number of channels we handle in a single thread. | |
Functions | |
static void | __reg_module (void) |
static void | __unreg_module (void) |
static void | destroy_multiplexed_thread (void *obj) |
Destroy callback for a multiplexed thread structure. | |
static int | find_multiplexed_thread (void *obj, void *arg, int flags) |
Callback function for finding a free multiplexed thread. | |
static int | load_module (void) |
static void | multiplexed_add_or_remove (struct multiplexed_thread *multiplexed_thread, struct ast_channel *chan, int add) |
Helper function which adds or removes a channel and nudges the thread. | |
static int | multiplexed_bridge_create (struct ast_bridge *bridge) |
Create function which finds/reserves/references a multiplexed thread structure. | |
static int | multiplexed_bridge_destroy (struct ast_bridge *bridge) |
Destroy function which unreserves/unreferences/removes a multiplexed thread structure. | |
static int | multiplexed_bridge_join (struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel) |
Join function which actually adds the channel into the array to be monitored. | |
static int | multiplexed_bridge_leave (struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel) |
Leave function which actually removes the channel from the array. | |
static void | multiplexed_bridge_suspend (struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel) |
Suspend function which means control of the channel is going elsewhere. | |
static void | multiplexed_bridge_unsuspend (struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel) |
Unsuspend function which means control of the channel is coming back to us. | |
static enum ast_bridge_write_result | multiplexed_bridge_write (struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel, struct ast_frame *frame) |
Write function for writing frames into the bridge. | |
static void | multiplexed_nudge (struct multiplexed_thread *multiplexed_thread) |
Internal function which nudges the thread. | |
static void * | multiplexed_thread_function (void *data) |
Thread function that executes for multiplexed threads. | |
static int | unload_module (void) |
Variables | |
static struct ast_module_info __MODULE_INFO_SECTION | __mod_info = { __MODULE_INFO_GLOBALS .name = AST_MODULE, .flags = AST_MODFLAG_DEFAULT , .description = "Multiplexed two channel bridging module" , .key = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .load = load_module, .unload = unload_module, } |
static struct ast_module_info * | ast_module_info = &__mod_info |
static struct ast_bridge_technology | multiplexed_bridge |
static struct ao2_container * | multiplexed_threads |
Container of all operating multiplexed threads. |
Two channel bridging module which groups bridges into batches of threads.
Definition in file bridge_multiplexed.c.
#define MULTIPLEXED_BUCKETS 53 |
Number of buckets our multiplexed thread container can have.
Definition at line 47 of file bridge_multiplexed.c.
Referenced by load_module().
#define MULTIPLEXED_MAX_CHANNELS 8 |
Number of channels we handle in a single thread.
Definition at line 50 of file bridge_multiplexed.c.
Referenced by find_multiplexed_thread(), and multiplexed_add_or_remove().
static void __reg_module | ( | void | ) | [static] |
Definition at line 404 of file bridge_multiplexed.c.
static void __unreg_module | ( | void | ) | [static] |
Definition at line 404 of file bridge_multiplexed.c.
static void destroy_multiplexed_thread | ( | void * | obj | ) | [static] |
Destroy callback for a multiplexed thread structure.
Definition at line 79 of file bridge_multiplexed.c.
References multiplexed_thread::pipe.
Referenced by multiplexed_bridge_create().
{ struct multiplexed_thread *multiplexed_thread = obj; if (multiplexed_thread->pipe[0] > -1) { close(multiplexed_thread->pipe[0]); } if (multiplexed_thread->pipe[1] > -1) { close(multiplexed_thread->pipe[1]); } return; }
static int find_multiplexed_thread | ( | void * | obj, |
void * | arg, | ||
int | flags | ||
) | [static] |
Callback function for finding a free multiplexed thread.
Definition at line 72 of file bridge_multiplexed.c.
References CMP_MATCH, CMP_STOP, multiplexed_thread::count, and MULTIPLEXED_MAX_CHANNELS.
Referenced by multiplexed_bridge_create().
{ struct multiplexed_thread *multiplexed_thread = obj; return (multiplexed_thread->count <= (MULTIPLEXED_MAX_CHANNELS - 2)) ? CMP_MATCH | CMP_STOP : 0; }
static int load_module | ( | void | ) | [static] |
Definition at line 395 of file bridge_multiplexed.c.
References ao2_container_alloc, ast_bridge_technology_register, AST_MODULE_LOAD_DECLINE, and MULTIPLEXED_BUCKETS.
{ if (!(multiplexed_threads = ao2_container_alloc(MULTIPLEXED_BUCKETS, NULL, NULL))) { return AST_MODULE_LOAD_DECLINE; } return ast_bridge_technology_register(&multiplexed_bridge); }
static void multiplexed_add_or_remove | ( | struct multiplexed_thread * | multiplexed_thread, |
struct ast_channel * | chan, | ||
int | add | ||
) | [static] |
Helper function which adds or removes a channel and nudges the thread.
Definition at line 248 of file bridge_multiplexed.c.
References ao2_lock(), ao2_ref, ao2_unlock(), ast_debug, ast_pthread_create, AST_PTHREADT_NULL, AST_PTHREADT_STOP, chan, multiplexed_thread::chans, MULTIPLEXED_MAX_CHANNELS, multiplexed_nudge(), multiplexed_thread_function(), multiplexed_thread::service_count, multiplexed_thread::thread, and thread.
Referenced by multiplexed_bridge_join(), multiplexed_bridge_leave(), multiplexed_bridge_suspend(), and multiplexed_bridge_unsuspend().
{ int i, removed = 0; pthread_t thread = AST_PTHREADT_NULL; ao2_lock(multiplexed_thread); multiplexed_nudge(multiplexed_thread); for (i = 0; i < MULTIPLEXED_MAX_CHANNELS; i++) { if (multiplexed_thread->chans[i] == chan) { if (!add) { multiplexed_thread->chans[i] = NULL; multiplexed_thread->service_count--; removed = 1; } break; } else if (!multiplexed_thread->chans[i] && add) { multiplexed_thread->chans[i] = chan; multiplexed_thread->service_count++; break; } } if (multiplexed_thread->service_count && multiplexed_thread->thread == AST_PTHREADT_NULL) { ao2_ref(multiplexed_thread, +1); if (ast_pthread_create(&multiplexed_thread->thread, NULL, multiplexed_thread_function, multiplexed_thread)) { ao2_ref(multiplexed_thread, -1); ast_debug(1, "Failed to create an actual thread for multiplexed thread '%p', trying next time\n", multiplexed_thread); } } else if (!multiplexed_thread->service_count && multiplexed_thread->thread != AST_PTHREADT_NULL) { thread = multiplexed_thread->thread; multiplexed_thread->thread = AST_PTHREADT_STOP; } else if (!add && removed) { memmove(multiplexed_thread->chans + i, multiplexed_thread->chans + i + 1, sizeof(struct ast_channel *) * (MULTIPLEXED_MAX_CHANNELS - (i + 1))); } ao2_unlock(multiplexed_thread); if (thread != AST_PTHREADT_NULL) { pthread_join(thread, NULL); } return; }
static int multiplexed_bridge_create | ( | struct ast_bridge * | bridge | ) | [static] |
Create function which finds/reserves/references a multiplexed thread structure.
Definition at line 94 of file bridge_multiplexed.c.
References ao2_alloc, ao2_callback, ao2_link, ao2_lock(), ao2_ref, ao2_unlock(), ast_debug, ast_log(), AST_PTHREADT_NULL, ast_bridge::bridge_pvt, multiplexed_thread::count, destroy_multiplexed_thread(), errno, find_multiplexed_thread(), LOG_WARNING, multiplexed_thread::pipe, and multiplexed_thread::thread.
{ struct multiplexed_thread *multiplexed_thread; ao2_lock(multiplexed_threads); /* Try to find an existing thread to handle our additional channels */ if (!(multiplexed_thread = ao2_callback(multiplexed_threads, 0, find_multiplexed_thread, NULL))) { int flags; /* If we failed we will have to create a new one from scratch */ if (!(multiplexed_thread = ao2_alloc(sizeof(*multiplexed_thread), destroy_multiplexed_thread))) { ast_debug(1, "Failed to find or create a new multiplexed thread for bridge '%p'\n", bridge); ao2_unlock(multiplexed_threads); return -1; } multiplexed_thread->pipe[0] = multiplexed_thread->pipe[1] = -1; /* Setup a pipe so we can poke the thread itself when needed */ if (pipe(multiplexed_thread->pipe)) { ast_debug(1, "Failed to create a pipe for poking a multiplexed thread for bridge '%p'\n", bridge); ao2_ref(multiplexed_thread, -1); ao2_unlock(multiplexed_threads); return -1; } /* Setup each pipe for non-blocking operation */ flags = fcntl(multiplexed_thread->pipe[0], F_GETFL); if (fcntl(multiplexed_thread->pipe[0], F_SETFL, flags | O_NONBLOCK) < 0) { ast_log(LOG_WARNING, "Failed to setup first nudge pipe for non-blocking operation on %p (%d: %s)\n", bridge, errno, strerror(errno)); ao2_ref(multiplexed_thread, -1); ao2_unlock(multiplexed_threads); return -1; } flags = fcntl(multiplexed_thread->pipe[1], F_GETFL); if (fcntl(multiplexed_thread->pipe[1], F_SETFL, flags | O_NONBLOCK) < 0) { ast_log(LOG_WARNING, "Failed to setup second nudge pipe for non-blocking operation on %p (%d: %s)\n", bridge, errno, strerror(errno)); ao2_ref(multiplexed_thread, -1); ao2_unlock(multiplexed_threads); return -1; } /* Set up default parameters */ multiplexed_thread->thread = AST_PTHREADT_NULL; /* Finally link us into the container so others may find us */ ao2_link(multiplexed_threads, multiplexed_thread); ast_debug(1, "Created multiplexed thread '%p' for bridge '%p'\n", multiplexed_thread, bridge); } else { ast_debug(1, "Found multiplexed thread '%p' for bridge '%p'\n", multiplexed_thread, bridge); } /* Bump the count of the thread structure up by two since the channels for this bridge will be joining shortly */ multiplexed_thread->count += 2; ao2_unlock(multiplexed_threads); bridge->bridge_pvt = multiplexed_thread; return 0; }
static int multiplexed_bridge_destroy | ( | struct ast_bridge * | bridge | ) | [static] |
Destroy function which unreserves/unreferences/removes a multiplexed thread structure.
Definition at line 177 of file bridge_multiplexed.c.
References ao2_lock(), ao2_ref, ao2_unlink, ao2_unlock(), ast_debug, ast_bridge::bridge_pvt, multiplexed_thread::count, and multiplexed_nudge().
{ struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt; ao2_lock(multiplexed_threads); multiplexed_thread->count -= 2; if (!multiplexed_thread->count) { ast_debug(1, "Unlinking multiplexed thread '%p' since nobody is using it anymore\n", multiplexed_thread); ao2_unlink(multiplexed_threads, multiplexed_thread); } multiplexed_nudge(multiplexed_thread); ao2_unlock(multiplexed_threads); ao2_ref(multiplexed_thread, -1); return 0; }
static int multiplexed_bridge_join | ( | struct ast_bridge * | bridge, |
struct ast_bridge_channel * | bridge_channel | ||
) | [static] |
Join function which actually adds the channel into the array to be monitored.
Definition at line 295 of file bridge_multiplexed.c.
References ast_channel_make_compatible(), ast_debug, AST_LIST_FIRST, AST_LIST_LAST, ast_bridge::bridge_pvt, ast_bridge_channel::chan, ast_bridge::channels, multiplexed_add_or_remove(), ast_channel::name, ast_channel::nativeformats, ast_channel::readformat, and ast_channel::writeformat.
{ struct ast_channel *c0 = AST_LIST_FIRST(&bridge->channels)->chan, *c1 = AST_LIST_LAST(&bridge->channels)->chan; struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt; ast_debug(1, "Adding channel '%s' to multiplexed thread '%p' for monitoring\n", bridge_channel->chan->name, multiplexed_thread); multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 1); /* If the second channel has not yet joined do not make things compatible */ if (c0 == c1) { return 0; } if (((c0->writeformat == c1->readformat) && (c0->readformat == c1->writeformat) && (c0->nativeformats == c1->nativeformats))) { return 0; } return ast_channel_make_compatible(c0, c1); }
static int multiplexed_bridge_leave | ( | struct ast_bridge * | bridge, |
struct ast_bridge_channel * | bridge_channel | ||
) | [static] |
Leave function which actually removes the channel from the array.
Definition at line 317 of file bridge_multiplexed.c.
References ast_debug, ast_bridge::bridge_pvt, ast_bridge_channel::chan, multiplexed_add_or_remove(), and ast_channel::name.
{ struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt; ast_debug(1, "Removing channel '%s' from multiplexed thread '%p'\n", bridge_channel->chan->name, multiplexed_thread); multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 0); return 0; }
static void multiplexed_bridge_suspend | ( | struct ast_bridge * | bridge, |
struct ast_bridge_channel * | bridge_channel | ||
) | [static] |
Suspend function which means control of the channel is going elsewhere.
Definition at line 329 of file bridge_multiplexed.c.
References ast_debug, ast_bridge::bridge_pvt, ast_bridge_channel::chan, multiplexed_add_or_remove(), and ast_channel::name.
{ struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt; ast_debug(1, "Suspending channel '%s' from multiplexed thread '%p'\n", bridge_channel->chan->name, multiplexed_thread); multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 0); return; }
static void multiplexed_bridge_unsuspend | ( | struct ast_bridge * | bridge, |
struct ast_bridge_channel * | bridge_channel | ||
) | [static] |
Unsuspend function which means control of the channel is coming back to us.
Definition at line 341 of file bridge_multiplexed.c.
References ast_debug, ast_bridge::bridge_pvt, ast_bridge_channel::chan, multiplexed_add_or_remove(), and ast_channel::name.
{ struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt; ast_debug(1, "Unsuspending channel '%s' from multiplexed thread '%p'\n", bridge_channel->chan->name, multiplexed_thread); multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 1); return; }
static enum ast_bridge_write_result multiplexed_bridge_write | ( | struct ast_bridge * | bridge, |
struct ast_bridge_channel * | bridge_channel, | ||
struct ast_frame * | frame | ||
) | [static] |
Write function for writing frames into the bridge.
Definition at line 353 of file bridge_multiplexed.c.
References AST_BRIDGE_CHANNEL_STATE_WAIT, AST_BRIDGE_WRITE_FAILED, AST_BRIDGE_WRITE_SUCCESS, AST_LIST_FIRST, AST_LIST_LAST, ast_write(), ast_bridge_channel::chan, ast_bridge::channels, and ast_bridge_channel::state.
{ struct ast_bridge_channel *other; if (AST_LIST_FIRST(&bridge->channels) == AST_LIST_LAST(&bridge->channels)) { return AST_BRIDGE_WRITE_FAILED; } if (!(other = (AST_LIST_FIRST(&bridge->channels) == bridge_channel ? AST_LIST_LAST(&bridge->channels) : AST_LIST_FIRST(&bridge->channels)))) { return AST_BRIDGE_WRITE_FAILED; } if (other->state == AST_BRIDGE_CHANNEL_STATE_WAIT) { ast_write(other->chan, frame); } return AST_BRIDGE_WRITE_SUCCESS; }
static void multiplexed_nudge | ( | struct multiplexed_thread * | multiplexed_thread | ) | [static] |
Internal function which nudges the thread.
Definition at line 157 of file bridge_multiplexed.c.
References ast_log(), AST_PTHREADT_NULL, LOG_ERROR, multiplexed_thread::pipe, multiplexed_thread::thread, and multiplexed_thread::waiting.
Referenced by multiplexed_add_or_remove(), and multiplexed_bridge_destroy().
{ int nudge = 0; if (multiplexed_thread->thread == AST_PTHREADT_NULL) { return; } if (write(multiplexed_thread->pipe[1], &nudge, sizeof(nudge)) != sizeof(nudge)) { ast_log(LOG_ERROR, "We couldn't poke multiplexed thread '%p'... something is VERY wrong\n", multiplexed_thread); } while (multiplexed_thread->waiting) { sched_yield(); } return; }
static void* multiplexed_thread_function | ( | void * | data | ) | [static] |
Thread function that executes for multiplexed threads.
Definition at line 200 of file bridge_multiplexed.c.
References ao2_lock(), ao2_ref, ao2_unlock(), ast_bridge_handle_trip(), ast_debug, ast_log(), AST_PTHREADT_NULL, AST_PTHREADT_STOP, ast_waitfor_nandfds(), ast_channel::bridge, multiplexed_thread::chans, errno, first, LOG_WARNING, multiplexed_thread::pipe, multiplexed_thread::service_count, multiplexed_thread::thread, and multiplexed_thread::waiting.
Referenced by multiplexed_add_or_remove().
{ struct multiplexed_thread *multiplexed_thread = data; int fds = multiplexed_thread->pipe[0]; ao2_lock(multiplexed_thread); ast_debug(1, "Starting actual thread for multiplexed thread '%p'\n", multiplexed_thread); while (multiplexed_thread->thread != AST_PTHREADT_STOP) { struct ast_channel *winner = NULL, *first = multiplexed_thread->chans[0]; int to = -1, outfd = -1; /* Move channels around so not just the first one gets priority */ memmove(multiplexed_thread->chans, multiplexed_thread->chans + 1, sizeof(struct ast_channel *) * (multiplexed_thread->service_count - 1)); multiplexed_thread->chans[multiplexed_thread->service_count - 1] = first; multiplexed_thread->waiting = 1; ao2_unlock(multiplexed_thread); winner = ast_waitfor_nandfds(multiplexed_thread->chans, multiplexed_thread->service_count, &fds, 1, NULL, &outfd, &to); multiplexed_thread->waiting = 0; ao2_lock(multiplexed_thread); if (outfd > -1) { int nudge; if (read(multiplexed_thread->pipe[0], &nudge, sizeof(nudge)) < 0) { if (errno != EINTR && errno != EAGAIN) { ast_log(LOG_WARNING, "read() failed for pipe on multiplexed thread '%p': %s\n", multiplexed_thread, strerror(errno)); } } } if (winner && winner->bridge) { ast_bridge_handle_trip(winner->bridge, NULL, winner, -1); } } multiplexed_thread->thread = AST_PTHREADT_NULL; ast_debug(1, "Stopping actual thread for multiplexed thread '%p'\n", multiplexed_thread); ao2_unlock(multiplexed_thread); ao2_ref(multiplexed_thread, -1); return NULL; }
static int unload_module | ( | void | ) | [static] |
Definition at line 386 of file bridge_multiplexed.c.
References ao2_ref, and ast_bridge_technology_unregister().
{ int res = ast_bridge_technology_unregister(&multiplexed_bridge); ao2_ref(multiplexed_threads, -1); return res; }
struct ast_module_info __MODULE_INFO_SECTION __mod_info = { __MODULE_INFO_GLOBALS .name = AST_MODULE, .flags = AST_MODFLAG_DEFAULT , .description = "Multiplexed two channel bridging module" , .key = ASTERISK_GPL_KEY , .buildopt_sum = AST_BUILDOPT_SUM, .load = load_module, .unload = unload_module, } [static] |
Definition at line 404 of file bridge_multiplexed.c.
struct ast_module_info* ast_module_info = &__mod_info [static] |
Definition at line 404 of file bridge_multiplexed.c.
struct ast_bridge_technology multiplexed_bridge [static] |
Definition at line 372 of file bridge_multiplexed.c.
struct ao2_container* multiplexed_threads [static] |
Container of all operating multiplexed threads.
Definition at line 69 of file bridge_multiplexed.c.