Merge "res_pjsip: New endpoint option "refer_blind_progress""
[asterisk/asterisk.git] / res / res_pjsip_transport_management.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2014, 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 /*** MODULEINFO
20         <depend>pjproject</depend>
21         <depend>res_pjsip</depend>
22         <support_level>core</support_level>
23  ***/
24
25 #include "asterisk.h"
26
27 #include <signal.h>
28
29 #include <pjsip.h>
30 #include <pjsip_ua.h>
31
32 #include "asterisk/res_pjsip.h"
33 #include "asterisk/module.h"
34 #include "asterisk/astobj2.h"
35
36 /*! \brief Number of buckets for keepalive transports */
37 #define TRANSPORTS_BUCKETS 53
38
39 #define IDLE_TIMEOUT (pjsip_cfg()->tsx.td)
40
41 /*! \brief The keep alive packet to send */
42 static const pj_str_t keepalive_packet = { "\r\n\r\n", 4 };
43
44 /*! \brief Global container of active transports */
45 static AO2_GLOBAL_OBJ_STATIC(monitored_transports);
46
47 /*! \brief Scheduler context for timing out connections with no data received */
48 static struct ast_sched_context *sched;
49
50 /*! \brief Thread keeping things alive */
51 static pthread_t keepalive_thread = AST_PTHREADT_NULL;
52
53 /*! \brief The global interval at which to send keepalives */
54 static unsigned int keepalive_interval;
55
56 /*! \brief Existing transport manager callback that we need to invoke */
57 static pjsip_tp_state_callback tpmgr_state_callback;
58
59 /*! \brief Structure for transport to be monitored */
60 struct monitored_transport {
61         /*! \brief The underlying PJSIP transport */
62         pjsip_transport *transport;
63         /*! \brief Non-zero if a PJSIP request was received */
64         int sip_received;
65 };
66
67 /*! \brief Callback function to send keepalive */
68 static int keepalive_transport_cb(void *obj, void *arg, int flags)
69 {
70         struct monitored_transport *monitored = obj;
71         pjsip_tpselector selector = {
72                 .type = PJSIP_TPSELECTOR_TRANSPORT,
73                 .u.transport = monitored->transport,
74         };
75
76         pjsip_tpmgr_send_raw(pjsip_endpt_get_tpmgr(ast_sip_get_pjsip_endpoint()),
77                 monitored->transport->key.type, &selector, NULL, keepalive_packet.ptr, keepalive_packet.slen,
78                 &monitored->transport->key.rem_addr, pj_sockaddr_get_len(&monitored->transport->key.rem_addr),
79                 NULL, NULL);
80
81         return 0;
82 }
83
84 /*! \brief Thread which sends keepalives to all active connection-oriented transports */
85 static void *keepalive_transport_thread(void *data)
86 {
87         struct ao2_container *transports;
88         pj_thread_desc desc;
89         pj_thread_t *thread;
90
91         if (pj_thread_register("Asterisk Keepalive Thread", desc, &thread) != PJ_SUCCESS) {
92                 ast_log(LOG_ERROR, "Could not register keepalive thread with PJLIB, keepalives will not occur.\n");
93                 return NULL;
94         }
95
96         transports = ao2_global_obj_ref(monitored_transports);
97         if (!transports) {
98                 return NULL;
99         }
100
101         /* Once loaded this module just keeps on going as it is unsafe to stop and change the underlying
102          * callback for the transport manager.
103          */
104         while (keepalive_interval) {
105                 sleep(keepalive_interval);
106                 ao2_callback(transports, OBJ_NODATA, keepalive_transport_cb, NULL);
107         }
108
109         ao2_ref(transports, -1);
110         return NULL;
111 }
112
113 AST_THREADSTORAGE(desc_storage);
114
115 static int idle_sched_cb(const void *data)
116 {
117         struct monitored_transport *keepalive = (struct monitored_transport *) data;
118
119         if (!pj_thread_is_registered()) {
120                 pj_thread_t *thread;
121                 pj_thread_desc *desc;
122
123                 desc = ast_threadstorage_get(&desc_storage, sizeof(pj_thread_desc));
124                 if (!desc) {
125                         ast_log(LOG_ERROR, "Could not get thread desc from thread-local storage.\n");
126                         ao2_ref(keepalive, -1);
127                         return 0;
128                 }
129
130                 pj_bzero(*desc, sizeof(*desc));
131
132                 pj_thread_register("Transport Monitor", *desc, &thread);
133         }
134
135         if (!keepalive->sip_received) {
136                 ast_log(LOG_NOTICE, "Shutting down transport '%s' since no request was received in %d seconds\n",
137                                 keepalive->transport->info, IDLE_TIMEOUT / 1000);
138                 pjsip_transport_shutdown(keepalive->transport);
139         }
140
141         ao2_ref(keepalive, -1);
142         return 0;
143 }
144
145 /*! \brief Destructor for keepalive transport */
146 static void monitored_transport_destroy(void *obj)
147 {
148         struct monitored_transport *keepalive = obj;
149
150         pjsip_transport_dec_ref(keepalive->transport);
151 }
152
153 /*! \brief Callback invoked when transport changes occur */
154 static void monitored_transport_state_callback(pjsip_transport *transport, pjsip_transport_state state,
155         const pjsip_transport_state_info *info)
156 {
157         struct ao2_container *transports;
158
159         /* We only care about reliable transports */
160         if (PJSIP_TRANSPORT_IS_RELIABLE(transport)
161                 && (transport->dir == PJSIP_TP_DIR_INCOMING || keepalive_interval)
162                 && (transports = ao2_global_obj_ref(monitored_transports))) {
163                 struct monitored_transport *monitored;
164
165                 switch (state) {
166                 case PJSIP_TP_STATE_CONNECTED:
167                         monitored = ao2_alloc_options(sizeof(*monitored),
168                                 monitored_transport_destroy, AO2_ALLOC_OPT_LOCK_NOLOCK);
169                         if (!monitored) {
170                                 break;
171                         }
172                         monitored->transport = transport;
173                         pjsip_transport_add_ref(monitored->transport);
174
175                         ao2_link(transports, monitored);
176
177                         if (transport->dir == PJSIP_TP_DIR_INCOMING) {
178                                 /* Let the scheduler inherit the reference from allocation */
179                                 if (ast_sched_add_variable(sched, IDLE_TIMEOUT, idle_sched_cb, monitored, 1) < 0) {
180                                         /* Uh Oh.  Could not schedule the idle check.  Kill the transport. */
181                                         ao2_unlink(transports, monitored);
182                                         ao2_ref(monitored, -1);
183                                         pjsip_transport_shutdown(transport);
184                                 }
185                         } else {
186                                 /* No scheduled task, so get rid of the allocation reference */
187                                 ao2_ref(monitored, -1);
188                         }
189                         break;
190                 case PJSIP_TP_STATE_SHUTDOWN:
191                 case PJSIP_TP_STATE_DISCONNECTED:
192                         ao2_find(transports, transport->obj_name, OBJ_SEARCH_KEY | OBJ_NODATA | OBJ_UNLINK);
193                         break;
194                 default:
195                         break;
196                 }
197
198                 ao2_ref(transports, -1);
199         }
200
201         /* Forward to the old state callback if present */
202         if (tpmgr_state_callback) {
203                 tpmgr_state_callback(transport, state, info);
204         }
205 }
206
207 /*! \brief Hashing function for monitored transport */
208 static int monitored_transport_hash_fn(const void *obj, int flags)
209 {
210         const struct monitored_transport *object;
211         const char *key;
212
213         switch (flags & OBJ_SEARCH_MASK) {
214         case OBJ_SEARCH_KEY:
215                 key = obj;
216                 break;
217         case OBJ_SEARCH_OBJECT:
218                 object = obj;
219                 key = object->transport->obj_name;
220                 break;
221         default:
222                 /* Hash can only work on something with a full key. */
223                 ast_assert(0);
224                 return 0;
225         }
226         return ast_str_hash(key);
227 }
228
229 /*! \brief Comparison function for monitored transport */
230 static int monitored_transport_cmp_fn(void *obj, void *arg, int flags)
231 {
232         const struct monitored_transport *object_left = obj;
233         const struct monitored_transport *object_right = arg;
234         const char *right_key = arg;
235         int cmp;
236
237         switch (flags & OBJ_SEARCH_MASK) {
238         case OBJ_SEARCH_OBJECT:
239                 right_key = object_right->transport->obj_name;
240                 /* Fall through */
241         case OBJ_SEARCH_KEY:
242                 cmp = strcmp(object_left->transport->obj_name, right_key);
243                 break;
244         case OBJ_SEARCH_PARTIAL_KEY:
245                 /*
246                  * We could also use a partial key struct containing a length
247                  * so strlen() does not get called for every comparison instead.
248                  */
249                 cmp = strncmp(object_left->transport->obj_name, right_key, strlen(right_key));
250                 break;
251         default:
252                 /*
253                  * What arg points to is specific to this traversal callback
254                  * and has no special meaning to astobj2.
255                  */
256                 cmp = 0;
257                 break;
258         }
259
260         return !cmp ? CMP_MATCH : 0;
261 }
262
263 static void keepalive_global_loaded(const char *object_type)
264 {
265         unsigned int new_interval = ast_sip_get_keep_alive_interval();
266
267         if (new_interval) {
268                 keepalive_interval = new_interval;
269         } else if (keepalive_interval) {
270                 ast_log(LOG_NOTICE, "Keepalive support can not be disabled once activated.\n");
271                 return;
272         } else {
273                 /* This will occur if no keepalive interval has been specified at initial start */
274                 return;
275         }
276
277         if (keepalive_thread != AST_PTHREADT_NULL) {
278                 return;
279         }
280
281         if (ast_pthread_create(&keepalive_thread, NULL, keepalive_transport_thread, NULL)) {
282                 ast_log(LOG_ERROR, "Could not create thread for sending keepalive messages.\n");
283                 keepalive_thread = AST_PTHREADT_NULL;
284                 keepalive_interval = 0;
285         }
286 }
287
288 /*! \brief Observer which is used to update our interval when the global setting changes */
289 static struct ast_sorcery_observer keepalive_global_observer = {
290         .loaded = keepalive_global_loaded,
291 };
292
293 /*!
294  * \brief
295  * On incoming TCP connections, when we receive a SIP request, we mark that we have
296  * received a valid SIP request. This way, we will not shut the transport down for
297  * idleness
298  */
299 static pj_bool_t idle_monitor_on_rx_request(pjsip_rx_data *rdata)
300 {
301         struct ao2_container *transports;
302         struct monitored_transport *idle_trans;
303
304         transports = ao2_global_obj_ref(monitored_transports);
305         if (!transports) {
306                 return PJ_FALSE;
307         }
308
309         idle_trans = ao2_find(transports, rdata->tp_info.transport->obj_name, OBJ_SEARCH_KEY);
310         ao2_ref(transports, -1);
311         if (!idle_trans) {
312                 return PJ_FALSE;
313         }
314
315         idle_trans->sip_received = 1;
316         ao2_ref(idle_trans, -1);
317
318         return PJ_FALSE;
319 }
320
321 static pjsip_module idle_monitor_module = {
322         .name = {"idle monitor module", 19},
323         .priority = PJSIP_MOD_PRIORITY_TRANSPORT_LAYER + 3,
324         .on_rx_request = idle_monitor_on_rx_request,
325 };
326
327 static int load_module(void)
328 {
329         struct ao2_container *transports;
330         pjsip_tpmgr *tpmgr;
331
332         CHECK_PJSIP_MODULE_LOADED();
333
334         tpmgr = pjsip_endpt_get_tpmgr(ast_sip_get_pjsip_endpoint());
335         if (!tpmgr) {
336                 ast_log(LOG_ERROR, "No transport manager to attach keepalive functionality to.\n");
337                 return AST_MODULE_LOAD_DECLINE;
338         }
339
340         transports = ao2_container_alloc(TRANSPORTS_BUCKETS, monitored_transport_hash_fn,
341                 monitored_transport_cmp_fn);
342         if (!transports) {
343                 ast_log(LOG_ERROR, "Could not create container for transports to perform keepalive on.\n");
344                 return AST_MODULE_LOAD_DECLINE;
345         }
346         ao2_global_obj_replace_unref(monitored_transports, transports);
347         ao2_ref(transports, -1);
348
349         sched = ast_sched_context_create();
350         if (!sched) {
351                 ast_log(LOG_ERROR, "Failed to create keepalive scheduler context.\n");
352                 ao2_global_obj_release(monitored_transports);
353                 return AST_MODULE_LOAD_DECLINE;
354         }
355
356         if (ast_sched_start_thread(sched)) {
357                 ast_log(LOG_ERROR, "Failed to start keepalive scheduler thread\n");
358                 ast_sched_context_destroy(sched);
359                 sched = NULL;
360                 ao2_global_obj_release(monitored_transports);
361                 return AST_MODULE_LOAD_DECLINE;
362         }
363
364         ast_sip_register_service(&idle_monitor_module);
365
366         tpmgr_state_callback = pjsip_tpmgr_get_state_cb(tpmgr);
367         pjsip_tpmgr_set_state_cb(tpmgr, &monitored_transport_state_callback);
368
369         ast_sorcery_observer_add(ast_sip_get_sorcery(), "global", &keepalive_global_observer);
370         ast_sorcery_reload_object(ast_sip_get_sorcery(), "global");
371
372         ast_module_shutdown_ref(ast_module_info->self);
373         return AST_MODULE_LOAD_SUCCESS;
374 }
375
376 static int unload_module(void)
377 {
378         pjsip_tpmgr *tpmgr;
379
380         if (keepalive_interval) {
381                 keepalive_interval = 0;
382                 if (keepalive_thread != AST_PTHREADT_NULL) {
383                         pthread_kill(keepalive_thread, SIGURG);
384                         pthread_join(keepalive_thread, NULL);
385                         keepalive_thread = AST_PTHREADT_NULL;
386                 }
387         }
388
389         ast_sorcery_observer_remove(ast_sip_get_sorcery(), "global", &keepalive_global_observer);
390
391         tpmgr = pjsip_endpt_get_tpmgr(ast_sip_get_pjsip_endpoint());
392         if (tpmgr) {
393                 pjsip_tpmgr_set_state_cb(tpmgr, tpmgr_state_callback);
394         }
395
396         ast_sip_unregister_service(&idle_monitor_module);
397
398         ast_sched_context_destroy(sched);
399         sched = NULL;
400
401         ao2_global_obj_release(monitored_transports);
402
403         return 0;
404 }
405
406 static int reload_module(void)
407 {
408         ast_sorcery_reload_object(ast_sip_get_sorcery(), "global");
409         return 0;
410 }
411
412 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_LOAD_ORDER, "PJSIP Reliable Transport Management",
413         .support_level = AST_MODULE_SUPPORT_CORE,
414         .load = load_module,
415         .reload = reload_module,
416         .unload = unload_module,
417         .load_pri = AST_MODPRI_CHANNEL_DEPEND - 4,
418 );