Test for Asterisk Version info
[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 /*** 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 && ast_channel_internal_bridge(winner)) {
240                         struct ast_bridge *bridge = ast_channel_internal_bridge(winner);
241                         int stop = 0;
242                         ao2_unlock(multiplexed_thread);
243                         while ((bridge = ast_channel_internal_bridge(winner)) && 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(ast_channel_writeformat(c0), ast_channel_readformat(c1)) == AST_FORMAT_CMP_EQUAL) &&
331                 (ast_format_cmp(ast_channel_readformat(c0), ast_channel_writeformat(c1)) == 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");