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