Media Project Phase2: SILK 8khz-24khz, SLINEAR 8khz-192khz, SPEEX 32khz, hd audio...
[asterisk/asterisk.git] / 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                 if (multiplexed_thread->thread == AST_PTHREADT_STOP) {
223                         break;
224                 }
225
226                 if (outfd > -1) {
227                         int nudge;
228
229                         if (read(multiplexed_thread->pipe[0], &nudge, sizeof(nudge)) < 0) {
230                                 if (errno != EINTR && errno != EAGAIN) {
231                                         ast_log(LOG_WARNING, "read() failed for pipe on multiplexed thread '%p': %s\n", multiplexed_thread, strerror(errno));
232                                 }
233                         }
234                 }
235                 if (winner && winner->bridge) {
236                         struct ast_bridge *bridge = winner->bridge;
237                         int stop = 0;
238                         ao2_unlock(multiplexed_thread);
239                         while ((bridge = winner->bridge) && ao2_trylock(bridge)) {
240                                 sched_yield();
241                                 if (multiplexed_thread->thread == AST_PTHREADT_STOP) {
242                                         stop = 1;
243                                         break;
244                                 }
245                         }
246                         if (!stop && bridge) {
247                                 ast_bridge_handle_trip(bridge, NULL, winner, -1);
248                                 ao2_unlock(bridge);
249                         }
250                         ao2_lock(multiplexed_thread);
251                 }
252         }
253
254         multiplexed_thread->thread = AST_PTHREADT_NULL;
255
256         ast_debug(1, "Stopping actual thread for multiplexed thread '%p'\n", multiplexed_thread);
257
258         ao2_unlock(multiplexed_thread);
259         ao2_ref(multiplexed_thread, -1);
260
261         return NULL;
262 }
263
264 /*! \brief Helper function which adds or removes a channel and nudges the thread */
265 static void multiplexed_add_or_remove(struct multiplexed_thread *multiplexed_thread, struct ast_channel *chan, int add)
266 {
267         int i, removed = 0;
268         pthread_t thread = AST_PTHREADT_NULL;
269
270         ao2_lock(multiplexed_thread);
271
272         multiplexed_nudge(multiplexed_thread);
273
274         for (i = 0; i < MULTIPLEXED_MAX_CHANNELS; i++) {
275                 if (multiplexed_thread->chans[i] == chan) {
276                         if (!add) {
277                                 multiplexed_thread->chans[i] = NULL;
278                                 multiplexed_thread->service_count--;
279                                 removed = 1;
280                         }
281                         break;
282                 } else if (!multiplexed_thread->chans[i] && add) {
283                         multiplexed_thread->chans[i] = chan;
284                         multiplexed_thread->service_count++;
285                         break;
286                 }
287         }
288
289         if (multiplexed_thread->service_count && multiplexed_thread->thread == AST_PTHREADT_NULL) {
290                 ao2_ref(multiplexed_thread, +1);
291                 if (ast_pthread_create(&multiplexed_thread->thread, NULL, multiplexed_thread_function, multiplexed_thread)) {
292                         ao2_ref(multiplexed_thread, -1);
293                         ast_debug(1, "Failed to create an actual thread for multiplexed thread '%p', trying next time\n", multiplexed_thread);
294                 }
295         } else if (!multiplexed_thread->service_count && multiplexed_thread->thread != AST_PTHREADT_NULL) {
296                 thread = multiplexed_thread->thread;
297                 multiplexed_thread->thread = AST_PTHREADT_STOP;
298         } else if (!add && removed) {
299                 memmove(multiplexed_thread->chans + i, multiplexed_thread->chans + i + 1, sizeof(struct ast_channel *) * (MULTIPLEXED_MAX_CHANNELS - (i + 1)));
300         }
301
302         ao2_unlock(multiplexed_thread);
303
304         if (thread != AST_PTHREADT_NULL) {
305                 pthread_join(thread, NULL);
306         }
307
308         return;
309 }
310
311 /*! \brief Join function which actually adds the channel into the array to be monitored */
312 static int multiplexed_bridge_join(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
313 {
314         struct ast_channel *c0 = AST_LIST_FIRST(&bridge->channels)->chan, *c1 = AST_LIST_LAST(&bridge->channels)->chan;
315         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
316
317         ast_debug(1, "Adding channel '%s' to multiplexed thread '%p' for monitoring\n", bridge_channel->chan->name, multiplexed_thread);
318
319         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 1);
320
321         /* If the second channel has not yet joined do not make things compatible */
322         if (c0 == c1) {
323                 return 0;
324         }
325
326         if ((ast_format_cmp(&c0->writeformat, &c1->readformat) == AST_FORMAT_CMP_EQUAL) &&
327                 (ast_format_cmp(&c0->readformat, &c1->writeformat) == AST_FORMAT_CMP_EQUAL) &&
328                 (ast_format_cap_identical(c0->nativeformats, c1->nativeformats))) {
329                 return 0;
330         }
331
332         return ast_channel_make_compatible(c0, c1);
333 }
334
335 /*! \brief Leave function which actually removes the channel from the array */
336 static int multiplexed_bridge_leave(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
337 {
338         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
339
340         ast_debug(1, "Removing channel '%s' from multiplexed thread '%p'\n", bridge_channel->chan->name, multiplexed_thread);
341
342         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 0);
343
344         return 0;
345 }
346
347 /*! \brief Suspend function which means control of the channel is going elsewhere */
348 static void multiplexed_bridge_suspend(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
349 {
350         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
351
352         ast_debug(1, "Suspending channel '%s' from multiplexed thread '%p'\n", bridge_channel->chan->name, multiplexed_thread);
353
354         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 0);
355
356         return;
357 }
358
359 /*! \brief Unsuspend function which means control of the channel is coming back to us */
360 static void multiplexed_bridge_unsuspend(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel)
361 {
362         struct multiplexed_thread *multiplexed_thread = bridge->bridge_pvt;
363
364         ast_debug(1, "Unsuspending channel '%s' from multiplexed thread '%p'\n", bridge_channel->chan->name, multiplexed_thread);
365
366         multiplexed_add_or_remove(multiplexed_thread, bridge_channel->chan, 1);
367
368         return;
369 }
370
371 /*! \brief Write function for writing frames into the bridge */
372 static enum ast_bridge_write_result multiplexed_bridge_write(struct ast_bridge *bridge, struct ast_bridge_channel *bridge_channel, struct ast_frame *frame)
373 {
374         struct ast_bridge_channel *other;
375
376         if (AST_LIST_FIRST(&bridge->channels) == AST_LIST_LAST(&bridge->channels)) {
377                 return AST_BRIDGE_WRITE_FAILED;
378         }
379
380         if (!(other = (AST_LIST_FIRST(&bridge->channels) == bridge_channel ? AST_LIST_LAST(&bridge->channels) : AST_LIST_FIRST(&bridge->channels)))) {
381                 return AST_BRIDGE_WRITE_FAILED;
382         }
383
384         if (other->state == AST_BRIDGE_CHANNEL_STATE_WAIT) {
385                 ast_write(other->chan, frame);
386         }
387
388         return AST_BRIDGE_WRITE_SUCCESS;
389 }
390
391 static struct ast_bridge_technology multiplexed_bridge = {
392         .name = "multiplexed_bridge",
393         .capabilities = AST_BRIDGE_CAPABILITY_1TO1MIX,
394         .preference = AST_BRIDGE_PREFERENCE_HIGH,
395         .create = multiplexed_bridge_create,
396         .destroy = multiplexed_bridge_destroy,
397         .join = multiplexed_bridge_join,
398         .leave = multiplexed_bridge_leave,
399         .suspend = multiplexed_bridge_suspend,
400         .unsuspend = multiplexed_bridge_unsuspend,
401         .write = multiplexed_bridge_write,
402 };
403
404 static int unload_module(void)
405 {
406         int res = ast_bridge_technology_unregister(&multiplexed_bridge);
407
408         ao2_ref(multiplexed_threads, -1);
409         multiplexed_bridge.format_capabilities = ast_format_cap_destroy(multiplexed_bridge.format_capabilities);
410
411         return res;
412 }
413
414 static int load_module(void)
415 {
416         if (!(multiplexed_threads = ao2_container_alloc(MULTIPLEXED_BUCKETS, NULL, NULL))) {
417                 return AST_MODULE_LOAD_DECLINE;
418         }
419         if (!(multiplexed_bridge.format_capabilities = ast_format_cap_alloc())) {
420                 return AST_MODULE_LOAD_DECLINE;
421         }
422         ast_format_cap_add_all_by_type(multiplexed_bridge.format_capabilities, AST_FORMAT_TYPE_AUDIO);
423         ast_format_cap_add_all_by_type(multiplexed_bridge.format_capabilities, AST_FORMAT_TYPE_VIDEO);
424         ast_format_cap_add_all_by_type(multiplexed_bridge.format_capabilities, AST_FORMAT_TYPE_TEXT);
425         return ast_bridge_technology_register(&multiplexed_bridge);
426 }
427
428 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Multiplexed two channel bridging module");