Merge "res_pjsip/res_pjsip_callerid: NULL check on caller id name string"
[asterisk/asterisk.git] / res / res_corosync.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2007, Digium, Inc.
5  * Copyright (C) 2012, Russell Bryant
6  *
7  * Russell Bryant <russell@russellbryant.net>
8  *
9  * See http://www.asterisk.org for more information about
10  * the Asterisk project. Please do not directly contact
11  * any of the maintainers of this project for assistance;
12  * the project provides a web site, mailing lists and IRC
13  * channels for your use.
14  *
15  * This program is free software, distributed under the terms of
16  * the GNU General Public License Version 2. See the LICENSE file
17  * at the top of the source tree.
18  */
19
20 /*!
21  * \file
22  * \author Russell Bryant <russell@russellbryant.net>
23  *
24  * This module is based on and replaces the previous res_ais module.
25  */
26
27 /*** MODULEINFO
28         <depend>corosync</depend>
29         <defaultenabled>no</defaultenabled>
30         <support_level>extended</support_level>
31  ***/
32
33 #include "asterisk.h"
34
35 #include <corosync/cpg.h>
36 #include <corosync/cfg.h>
37
38 #include "asterisk/module.h"
39 #include "asterisk/logger.h"
40 #include "asterisk/poll-compat.h"
41 #include "asterisk/config.h"
42 #include "asterisk/event.h"
43 #include "asterisk/cli.h"
44 #include "asterisk/devicestate.h"
45 #include "asterisk/app.h"
46 #include "asterisk/stasis.h"
47 #include "asterisk/stasis_message_router.h"
48 #include "asterisk/stasis_system.h"
49
50 AST_RWLOCK_DEFINE_STATIC(event_types_lock);
51
52 static void publish_mwi_to_stasis(struct ast_event *event);
53 static void publish_device_state_to_stasis(struct ast_event *event);
54 static void publish_cluster_discovery_to_stasis(struct ast_event *event);
55
56 /*! \brief All the nodes that we're aware of */
57 static struct ao2_container *nodes;
58
59 /*! \brief The internal topic used for message forwarding and pings */
60 static struct stasis_topic *corosync_aggregate_topic;
61
62 /*! \brief Our \ref stasis message router */
63 static struct stasis_message_router *stasis_router;
64
65 /*! \brief Internal accessor for our topic */
66 static struct stasis_topic *corosync_topic(void)
67 {
68         return corosync_aggregate_topic;
69 }
70
71 struct corosync_node {
72         /*! The corosync ID */
73         int id;
74         /*! The Asterisk EID */
75         struct ast_eid eid;
76         /*! The IP address of the node */
77         struct ast_sockaddr addr;
78 };
79
80 static struct corosync_node *corosync_node_alloc(struct ast_event *event)
81 {
82         struct corosync_node *node;
83
84         node = ao2_alloc_options(sizeof(*node), NULL, AO2_ALLOC_OPT_LOCK_NOLOCK);
85         if (!node) {
86                 return NULL;
87         }
88
89         memcpy(&node->eid, (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID), sizeof(node->eid));
90         node->id = ast_event_get_ie_uint(event, AST_EVENT_IE_NODE_ID);
91         ast_sockaddr_parse(&node->addr, ast_event_get_ie_str(event, AST_EVENT_IE_LOCAL_ADDR), PARSE_PORT_IGNORE);
92
93         return node;
94 }
95
96 static int corosync_node_hash_fn(const void *obj, const int flags)
97 {
98         const struct corosync_node *node;
99         const int *id;
100
101         switch (flags & OBJ_SEARCH_MASK) {
102         case OBJ_SEARCH_KEY:
103                 id = obj;
104                 break;
105         case OBJ_SEARCH_OBJECT:
106                 node = obj;
107                 id = &node->id;
108                 break;
109         default:
110                 ast_assert(0);
111                 return 0;
112         }
113         return *id;
114 }
115
116 static int corosync_node_cmp_fn(void *obj, void *arg, int flags)
117 {
118         struct corosync_node *left = obj;
119         struct corosync_node *right = arg;
120         const int *id = arg;
121         int cmp;
122
123         switch (flags & OBJ_SEARCH_MASK) {
124         case OBJ_SEARCH_OBJECT:
125                 id = &right->id;
126                 /* Fall through */
127         case OBJ_SEARCH_KEY:
128                 cmp = (left->id == *id);
129                 break;
130         case OBJ_SEARCH_PARTIAL_KEY:
131                 cmp = (left->id == right->id);
132                 break;
133         default:
134                 /* Sort can only work on something with a full or partial key. */
135                 ast_assert(0);
136                 cmp = 1;
137                 break;
138         }
139         return cmp ? CMP_MATCH : 0;
140 }
141
142
143 /*! \brief A payload wrapper around a corosync ping event */
144 struct corosync_ping_payload {
145         /*! The corosync ping event being passed over \ref stasis */
146         struct ast_event *event;
147 };
148
149 /*! \brief Destructor for the \ref corosync_ping_payload wrapper object */
150 static void corosync_ping_payload_dtor(void *obj)
151 {
152         struct corosync_ping_payload *payload = obj;
153
154         ast_free(payload->event);
155 }
156
157 /*! \brief Convert a Corosync PING to a \ref ast_event */
158 static struct ast_event *corosync_ping_to_event(struct stasis_message *message)
159 {
160         struct corosync_ping_payload *payload;
161         struct ast_event *event;
162         struct ast_eid *event_eid;
163
164         if (!message) {
165                 return NULL;
166         }
167
168         payload = stasis_message_data(message);
169
170         if (!payload->event) {
171                 return NULL;
172         }
173
174         event_eid = (struct ast_eid *)ast_event_get_ie_raw(payload->event, AST_EVENT_IE_EID);
175
176         event = ast_event_new(AST_EVENT_PING,
177                                 AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, event_eid, sizeof(*event_eid),
178                                 AST_EVENT_IE_END);
179
180         return event;
181 }
182
183 STASIS_MESSAGE_TYPE_DEFN_LOCAL(corosync_ping_message_type,
184         .to_event = corosync_ping_to_event, );
185
186 /*! \brief Publish a Corosync ping to \ref stasis */
187 static void publish_corosync_ping_to_stasis(struct ast_event *event)
188 {
189         struct corosync_ping_payload *payload;
190         struct stasis_message *message;
191
192         ast_assert(ast_event_get_type(event) == AST_EVENT_PING);
193         ast_assert(event != NULL);
194
195         if (!corosync_ping_message_type()) {
196                 return;
197         }
198
199         payload = ao2_t_alloc(sizeof(*payload), corosync_ping_payload_dtor, "Create ping payload");
200         if (!payload) {
201                 return;
202         }
203         payload->event = event;
204
205         message = stasis_message_create(corosync_ping_message_type(), payload);
206         if (!message) {
207                 ao2_t_ref(payload, -1, "Destroy payload on off nominal");
208                 return;
209         }
210
211         stasis_publish(corosync_topic(), message);
212
213         ao2_t_ref(payload, -1, "Hand ref to stasis");
214         ao2_t_ref(message, -1, "Hand ref to stasis");
215 }
216
217 static struct {
218         const char *name;
219         struct stasis_forward *sub;
220         unsigned char publish;
221         unsigned char publish_default;
222         unsigned char subscribe;
223         unsigned char subscribe_default;
224         struct stasis_topic *(* topic_fn)(void);
225         struct stasis_cache *(* cache_fn)(void);
226         struct stasis_message_type *(* message_type_fn)(void);
227         void (* publish_to_stasis)(struct ast_event *);
228 } event_types[] = {
229         [AST_EVENT_MWI] = { .name = "mwi",
230                             .topic_fn = ast_mwi_topic_all,
231                             .cache_fn = ast_mwi_state_cache,
232                             .message_type_fn = ast_mwi_state_type,
233                             .publish_to_stasis = publish_mwi_to_stasis, },
234         [AST_EVENT_DEVICE_STATE_CHANGE] = { .name = "device_state",
235                                             .topic_fn = ast_device_state_topic_all,
236                                             .cache_fn = ast_device_state_cache,
237                                             .message_type_fn = ast_device_state_message_type,
238                                             .publish_to_stasis = publish_device_state_to_stasis, },
239         [AST_EVENT_PING] = { .name = "ping",
240                              .publish_default = 1,
241                              .subscribe_default = 1,
242                              .topic_fn = corosync_topic,
243                              .message_type_fn = corosync_ping_message_type,
244                              .publish_to_stasis = publish_corosync_ping_to_stasis, },
245         [AST_EVENT_CLUSTER_DISCOVERY] = { .name = "cluster_discovery",
246                                           .publish_default = 1,
247                                           .subscribe_default = 1,
248                                           .topic_fn = ast_system_topic,
249                                           .message_type_fn = ast_cluster_discovery_type,
250                                           .publish_to_stasis = publish_cluster_discovery_to_stasis, },
251 };
252
253 static struct {
254         pthread_t id;
255         int alert_pipe[2];
256         unsigned int stop:1;
257 } dispatch_thread = {
258         .id = AST_PTHREADT_NULL,
259         .alert_pipe = { -1, -1 },
260 };
261
262 static cpg_handle_t cpg_handle;
263 static corosync_cfg_handle_t cfg_handle;
264
265 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
266 static void cfg_state_track_cb(
267                 corosync_cfg_state_notification_buffer_t *notification_buffer,
268                 cs_error_t error);
269 #endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
270
271 static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
272                 corosync_cfg_shutdown_flags_t flags);
273
274 static corosync_cfg_callbacks_t cfg_callbacks = {
275 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
276         .corosync_cfg_state_track_callback = cfg_state_track_cb,
277 #endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
278         .corosync_cfg_shutdown_callback = cfg_shutdown_cb,
279 };
280
281 /*! \brief Publish cluster discovery to \ref stasis */
282 static void publish_cluster_discovery_to_stasis_full(struct corosync_node *node, int joined)
283 {
284         struct ast_json *json;
285         struct ast_json_payload *payload;
286         struct stasis_message *message;
287         char eid[18];
288         const char *addr;
289
290         ast_eid_to_str(eid, sizeof(eid), &node->eid);
291         addr = ast_sockaddr_stringify_addr(&node->addr);
292
293         ast_log(AST_LOG_NOTICE, "Node %u (%s) at %s %s the cluster\n",
294                 node->id,
295                 eid,
296                 addr,
297                 joined ? "joined" : "left");
298
299         json = ast_json_pack("{s: s, s: i, s: s, s: i}",
300                 "address", addr,
301                 "node_id", node->id,
302                 "eid", eid,
303                 "joined", joined);
304         if (!json) {
305                 return;
306         }
307
308         payload = ast_json_payload_create(json);
309         if (!payload) {
310                 ast_json_unref(json);
311                 return;
312         }
313
314         message = stasis_message_create(ast_cluster_discovery_type(), payload);
315         if (!message) {
316                 ast_json_unref(json);
317                 ao2_ref(payload, -1);
318                 return;
319         }
320
321         stasis_publish(ast_system_topic(), message);
322         ast_json_unref(json);
323         ao2_ref(payload, -1);
324         ao2_ref(message, -1);
325 }
326
327 static void send_cluster_notify(void);
328
329 /*! \brief Publish a received cluster discovery \ref ast_event to \ref stasis */
330 static void publish_cluster_discovery_to_stasis(struct ast_event *event)
331 {
332         struct corosync_node *node;
333         int id = ast_event_get_ie_uint(event, AST_EVENT_IE_NODE_ID);
334         struct ast_eid *event_eid;
335
336         ast_assert(ast_event_get_type(event) == AST_EVENT_CLUSTER_DISCOVERY);
337
338         event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
339         if (!ast_eid_cmp(&ast_eid_default, event_eid)) {
340                 /* Don't feed events back in that originated locally. */
341                 return;
342         }
343
344         ao2_lock(nodes);
345         node = ao2_find(nodes, &id, OBJ_SEARCH_KEY | OBJ_NOLOCK);
346         if (node) {
347                 /* We already know about this node */
348                 ao2_unlock(nodes);
349                 ao2_ref(node, -1);
350                 return;
351         }
352
353         node = corosync_node_alloc(event);
354         if (!node) {
355                 ao2_unlock(nodes);
356                 return;
357         }
358         ao2_link_flags(nodes, node, OBJ_NOLOCK);
359         ao2_unlock(nodes);
360
361         publish_cluster_discovery_to_stasis_full(node, 1);
362
363         ao2_ref(node, -1);
364
365         /*
366          * When we get news that someone else has joined, we need to let them
367          * know we exist as well.
368          */
369         send_cluster_notify();
370 }
371
372 /*! \brief Publish a received MWI \ref ast_event to \ref stasis */
373 static void publish_mwi_to_stasis(struct ast_event *event)
374 {
375         const char *mailbox;
376         const char *context;
377         unsigned int new_msgs;
378         unsigned int old_msgs;
379         struct ast_eid *event_eid;
380
381         ast_assert(ast_event_get_type(event) == AST_EVENT_MWI);
382
383         mailbox = ast_event_get_ie_str(event, AST_EVENT_IE_MAILBOX);
384         context = ast_event_get_ie_str(event, AST_EVENT_IE_CONTEXT);
385         new_msgs = ast_event_get_ie_uint(event, AST_EVENT_IE_NEWMSGS);
386         old_msgs = ast_event_get_ie_uint(event, AST_EVENT_IE_OLDMSGS);
387         event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
388
389         if (ast_strlen_zero(mailbox) || ast_strlen_zero(context)) {
390                 return;
391         }
392
393         if (new_msgs > INT_MAX) {
394                 new_msgs = INT_MAX;
395         }
396
397         if (old_msgs > INT_MAX) {
398                 old_msgs = INT_MAX;
399         }
400
401         if (ast_publish_mwi_state_full(mailbox, context, (int)new_msgs,
402                                        (int)old_msgs, NULL, event_eid)) {
403                 char eid[18];
404                 ast_eid_to_str(eid, sizeof(eid), event_eid);
405                 ast_log(LOG_WARNING, "Failed to publish MWI message for %s@%s from %s\n",
406                         mailbox, context, eid);
407         }
408 }
409
410 /*! \brief Publish a received device state \ref ast_event to \ref stasis */
411 static void publish_device_state_to_stasis(struct ast_event *event)
412 {
413         const char *device;
414         enum ast_device_state state;
415         unsigned int cachable;
416         struct ast_eid *event_eid;
417
418         ast_assert(ast_event_get_type(event) == AST_EVENT_DEVICE_STATE_CHANGE);
419
420         device = ast_event_get_ie_str(event, AST_EVENT_IE_DEVICE);
421         state = ast_event_get_ie_uint(event, AST_EVENT_IE_STATE);
422         cachable = ast_event_get_ie_uint(event, AST_EVENT_IE_CACHABLE);
423         event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
424
425         if (ast_strlen_zero(device)) {
426                 return;
427         }
428
429         if (ast_publish_device_state_full(device, state, cachable, event_eid)) {
430                 char eid[18];
431                 ast_eid_to_str(eid, sizeof(eid), event_eid);
432                 ast_log(LOG_WARNING, "Failed to publish device state message for %s from %s\n",
433                         device, eid);
434         }
435 }
436
437 static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
438                 uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len);
439
440 static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
441                 const struct cpg_address *member_list, size_t member_list_entries,
442                 const struct cpg_address *left_list, size_t left_list_entries,
443                 const struct cpg_address *joined_list, size_t joined_list_entries);
444
445 static cpg_callbacks_t cpg_callbacks = {
446         .cpg_deliver_fn = cpg_deliver_cb,
447         .cpg_confchg_fn = cpg_confchg_cb,
448 };
449
450 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
451 static void cfg_state_track_cb(
452                 corosync_cfg_state_notification_buffer_t *notification_buffer,
453                 cs_error_t error)
454 {
455 }
456 #endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
457
458 static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
459                 corosync_cfg_shutdown_flags_t flags)
460 {
461 }
462
463 static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
464                 uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
465 {
466         struct ast_event *event;
467         void (*publish_handler)(struct ast_event *) = NULL;
468         enum ast_event_type event_type;
469
470         if (msg_len < ast_event_minimum_length()) {
471                 ast_debug(1, "Ignoring event that's too small. %u < %u\n",
472                         (unsigned int) msg_len,
473                         (unsigned int) ast_event_minimum_length());
474                 return;
475         }
476
477         if (!ast_eid_cmp(&ast_eid_default, ast_event_get_ie_raw(msg, AST_EVENT_IE_EID))) {
478                 /* Don't feed events back in that originated locally. */
479                 return;
480         }
481
482         event_type = ast_event_get_type(msg);
483         if (event_type > AST_EVENT_TOTAL) {
484                 /* Egads, we don't support this */
485                 return;
486         }
487
488         ast_rwlock_rdlock(&event_types_lock);
489         publish_handler = event_types[event_type].publish_to_stasis;
490         if (!event_types[event_type].subscribe || !publish_handler) {
491                 /* We are not configured to subscribe to these events or
492                    we have no way to publish it internally. */
493                 ast_rwlock_unlock(&event_types_lock);
494                 return;
495         }
496         ast_rwlock_unlock(&event_types_lock);
497
498         if (!(event = ast_malloc(msg_len))) {
499                 return;
500         }
501
502         memcpy(event, msg, msg_len);
503
504         if (event_type == AST_EVENT_PING) {
505                 const struct ast_eid *eid;
506                 char buf[128] = "";
507
508                 eid = ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
509                 ast_eid_to_str(buf, sizeof(buf), (struct ast_eid *) eid);
510                 ast_log(LOG_NOTICE, "Got event PING from server with EID: '%s'\n", buf);
511         }
512         ast_debug(5, "Publishing event %s (%u) to stasis\n",
513                 ast_event_get_type_name(event), event_type);
514         publish_handler(event);
515 }
516
517 static void publish_event_to_corosync(struct ast_event *event)
518 {
519         cs_error_t cs_err;
520         struct iovec iov;
521
522         iov.iov_base = (void *)event;
523         iov.iov_len = ast_event_get_size(event);
524
525         ast_debug(5, "Publishing event %s (%u) to corosync\n",
526                 ast_event_get_type_name(event), ast_event_get_type(event));
527
528         /* The stasis subscription will only exist if we are configured to publish
529          * these events, so just send away. */
530         if ((cs_err = cpg_mcast_joined(cpg_handle, CPG_TYPE_FIFO, &iov, 1)) != CS_OK) {
531                 ast_log(LOG_WARNING, "CPG mcast failed (%u) for event %s (%u)\n",
532                         cs_err, ast_event_get_type_name(event), ast_event_get_type(event));
533         }
534 }
535
536 static void publish_to_corosync(struct stasis_message *message)
537 {
538         struct ast_event *event;
539
540         event = stasis_message_to_event(message);
541         if (!event) {
542                 return;
543         }
544
545         if (ast_eid_cmp(&ast_eid_default, ast_event_get_ie_raw(event, AST_EVENT_IE_EID))) {
546                 /* If the event didn't originate from this server, don't send it back out. */
547                 ast_event_destroy(event);
548                 return;
549         }
550
551         if (ast_event_get_type(event) == AST_EVENT_PING) {
552                 const struct ast_eid *eid;
553                 char buf[128] = "";
554
555                 eid = ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
556                 ast_eid_to_str(buf, sizeof(buf), (struct ast_eid *) eid);
557                 ast_log(LOG_NOTICE, "Sending event PING from this server with EID: '%s'\n", buf);
558         }
559
560         publish_event_to_corosync(event);
561 }
562
563 static void stasis_message_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message)
564 {
565         if (!message) {
566                 return;
567         }
568
569         publish_to_corosync(message);
570 }
571
572 static int dump_cache_cb(void *obj, void *arg, int flags)
573 {
574         struct stasis_message *message = obj;
575
576         if (!message) {
577                 return 0;
578         }
579
580         publish_to_corosync(message);
581
582         return 0;
583 }
584
585 static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
586                 const struct cpg_address *member_list, size_t member_list_entries,
587                 const struct cpg_address *left_list, size_t left_list_entries,
588                 const struct cpg_address *joined_list, size_t joined_list_entries)
589 {
590         unsigned int i;
591
592
593         for (i = 0; i < left_list_entries; i++) {
594                 const struct cpg_address *cpg_node = &left_list[i];
595                 struct corosync_node* node;
596
597                 node = ao2_find(nodes, &cpg_node->nodeid, OBJ_UNLINK | OBJ_SEARCH_KEY);
598                 if (!node) {
599                         continue;
600                 }
601
602                 publish_cluster_discovery_to_stasis_full(node, 0);
603                 ao2_ref(node, -1);
604         }
605
606         /* If any new nodes have joined, dump our cache of events we are publishing
607          * that originated from this server. */
608         if (!joined_list_entries) {
609                 return;
610         }
611
612         for (i = 0; i < ARRAY_LEN(event_types); i++) {
613                 struct ao2_container *messages;
614
615                 ast_rwlock_rdlock(&event_types_lock);
616                 if (!event_types[i].publish) {
617                         ast_rwlock_unlock(&event_types_lock);
618                         continue;
619                 }
620
621                 if (!event_types[i].cache_fn || !event_types[i].message_type_fn) {
622                         ast_rwlock_unlock(&event_types_lock);
623                         continue;
624                 }
625
626                 messages = stasis_cache_dump_by_eid(event_types[i].cache_fn(),
627                         event_types[i].message_type_fn(),
628                         &ast_eid_default);
629                 ast_rwlock_unlock(&event_types_lock);
630
631                 ao2_callback(messages, OBJ_NODATA, dump_cache_cb, NULL);
632
633                 ao2_t_ref(messages, -1, "Dispose of dumped cache");
634         }
635 }
636
637 /*! \brief Informs the cluster of our EID and our IP addresses */
638 static void send_cluster_notify(void)
639 {
640         struct ast_event *event;
641         unsigned int node_id;
642         cs_error_t cs_err;
643         corosync_cfg_node_address_t corosync_addr;
644         int num_addrs = 0;
645         struct sockaddr *sa;
646         size_t sa_len;
647         char buf[128];
648         int res;
649
650         if ((cs_err = corosync_cfg_local_get(cfg_handle, &node_id)) != CS_OK) {
651                 ast_log(LOG_WARNING, "Failed to extract Corosync node ID for this node. Not informing cluster of existance.\n");
652                 return;
653         }
654
655         if (((cs_err = corosync_cfg_get_node_addrs(cfg_handle, node_id, 1, &num_addrs, &corosync_addr)) != CS_OK) || (num_addrs < 1)) {
656                 ast_log(LOG_WARNING, "Failed to get local Corosync address. Not informing cluster of existance.\n");
657                 return;
658         }
659
660         sa = (struct sockaddr *)corosync_addr.address;
661         sa_len = (size_t)corosync_addr.address_length;
662         if ((res = getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST))) {
663                 ast_log(LOG_WARNING, "Failed to determine name of local Corosync address: %s (%d). Not informing cluster of existance.\n",
664                         gai_strerror(res), res);
665                 return;
666         }
667
668         event = ast_event_new(AST_EVENT_CLUSTER_DISCOVERY,
669                                     AST_EVENT_IE_NODE_ID, AST_EVENT_IE_PLTYPE_UINT, node_id,
670                                     AST_EVENT_IE_LOCAL_ADDR, AST_EVENT_IE_PLTYPE_STR, buf,
671                                     AST_EVENT_IE_END);
672         publish_event_to_corosync(event);
673         ast_free(event);
674 }
675
676 static void *dispatch_thread_handler(void *data)
677 {
678         cs_error_t cs_err;
679         struct pollfd pfd[3] = {
680                 { .events = POLLIN, },
681                 { .events = POLLIN, },
682                 { .events = POLLIN, },
683         };
684
685         if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
686                 ast_log(LOG_ERROR, "Failed to get CPG fd.  This module is now broken.\n");
687                 return NULL;
688         }
689
690         if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
691                 ast_log(LOG_ERROR, "Failed to get CFG fd.  This module is now broken.\n");
692                 return NULL;
693         }
694
695         pfd[2].fd = dispatch_thread.alert_pipe[0];
696
697         send_cluster_notify();
698         while (!dispatch_thread.stop) {
699                 int res;
700
701                 cs_err = CS_OK;
702
703                 pfd[0].revents = 0;
704                 pfd[1].revents = 0;
705                 pfd[2].revents = 0;
706
707                 res = ast_poll(pfd, ARRAY_LEN(pfd), -1);
708                 if (res == -1 && errno != EINTR && errno != EAGAIN) {
709                         ast_log(LOG_ERROR, "poll() error: %s (%d)\n", strerror(errno), errno);
710                         continue;
711                 }
712
713                 if (pfd[0].revents & POLLIN) {
714                         if ((cs_err = cpg_dispatch(cpg_handle, CS_DISPATCH_ALL)) != CS_OK) {
715                                 ast_log(LOG_WARNING, "Failed CPG dispatch: %u\n", cs_err);
716                         }
717                 }
718
719                 if (pfd[1].revents & POLLIN) {
720                         if ((cs_err = corosync_cfg_dispatch(cfg_handle, CS_DISPATCH_ALL)) != CS_OK) {
721                                 ast_log(LOG_WARNING, "Failed CFG dispatch: %u\n", cs_err);
722                         }
723                 }
724
725                 if (cs_err == CS_ERR_LIBRARY || cs_err == CS_ERR_BAD_HANDLE) {
726                         struct cpg_name name;
727
728                         /* If corosync gets restarted out from under Asterisk, try to recover. */
729
730                         ast_log(LOG_NOTICE, "Attempting to recover from corosync failure.\n");
731
732                         if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
733                                 ast_log(LOG_ERROR, "Failed to initialize cfg (%d)\n", (int) cs_err);
734                                 sleep(5);
735                                 continue;
736                         }
737
738                         if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks) != CS_OK)) {
739                                 ast_log(LOG_ERROR, "Failed to initialize cpg (%d)\n", (int) cs_err);
740                                 sleep(5);
741                                 continue;
742                         }
743
744                         if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
745                                 ast_log(LOG_ERROR, "Failed to get CPG fd.\n");
746                                 sleep(5);
747                                 continue;
748                         }
749
750                         if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
751                                 ast_log(LOG_ERROR, "Failed to get CFG fd.\n");
752                                 sleep(5);
753                                 continue;
754                         }
755
756                         ast_copy_string(name.value, "asterisk", sizeof(name.value));
757                         name.length = strlen(name.value);
758                         if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
759                                 ast_log(LOG_ERROR, "Failed to join cpg (%d)\n", (int) cs_err);
760                                 sleep(5);
761                                 continue;
762                         }
763
764                         ast_log(LOG_NOTICE, "Corosync recovery complete.\n");
765                         send_cluster_notify();
766                 }
767         }
768
769         return NULL;
770 }
771
772 static char *corosync_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
773 {
774         cs_error_t cs_err;
775         cpg_iteration_handle_t cpg_iter;
776         struct cpg_iteration_description_t cpg_desc;
777         unsigned int i;
778
779         switch (cmd) {
780         case CLI_INIT:
781                 e->command = "corosync show members";
782                 e->usage =
783                         "Usage: corosync show members\n"
784                         "       Show corosync cluster members\n";
785                 return NULL;
786
787         case CLI_GENERATE:
788                 return NULL;    /* no completion */
789         }
790
791         if (a->argc != e->args) {
792                 return CLI_SHOWUSAGE;
793         }
794
795         cs_err = cpg_iteration_initialize(cpg_handle, CPG_ITERATION_ALL, NULL, &cpg_iter);
796
797         if (cs_err != CS_OK) {
798                 ast_cli(a->fd, "Failed to initialize CPG iterator.\n");
799                 return CLI_FAILURE;
800         }
801
802         ast_cli(a->fd, "\n"
803                     "=============================================================\n"
804                     "=== Cluster members =========================================\n"
805                     "=============================================================\n"
806                     "===\n");
807
808         for (i = 1, cs_err = cpg_iteration_next(cpg_iter, &cpg_desc);
809                         cs_err == CS_OK;
810                         cs_err = cpg_iteration_next(cpg_iter, &cpg_desc), i++) {
811                 corosync_cfg_node_address_t addrs[8];
812                 int num_addrs = 0;
813                 unsigned int j;
814
815                 cs_err = corosync_cfg_get_node_addrs(cfg_handle, cpg_desc.nodeid,
816                                 ARRAY_LEN(addrs), &num_addrs, addrs);
817                 if (cs_err != CS_OK) {
818                         ast_log(LOG_WARNING, "Failed to get node addresses\n");
819                         continue;
820                 }
821
822                 ast_cli(a->fd, "=== Node %u\n", i);
823                 ast_cli(a->fd, "=== --> Group: %s\n", cpg_desc.group.value);
824
825                 for (j = 0; j < num_addrs; j++) {
826                         struct sockaddr *sa = (struct sockaddr *) addrs[j].address;
827                         size_t sa_len = (size_t) addrs[j].address_length;
828                         char buf[128];
829
830                         getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST);
831
832                         ast_cli(a->fd, "=== --> Address %u: %s\n", j + 1, buf);
833                 }
834
835         }
836
837         ast_cli(a->fd, "===\n"
838                        "=============================================================\n"
839                        "\n");
840
841         cpg_iteration_finalize(cpg_iter);
842
843         return CLI_SUCCESS;
844 }
845
846 static char *corosync_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
847 {
848         struct ast_event *event;
849
850         switch (cmd) {
851         case CLI_INIT:
852                 e->command = "corosync ping";
853                 e->usage =
854                         "Usage: corosync ping\n"
855                         "       Send a test ping to the cluster.\n"
856                         "A NOTICE will be in the log for every ping received\n"
857                         "on a server.\n  If you send a ping, you should see a NOTICE\n"
858                         "in the log for every server in the cluster.\n";
859                 return NULL;
860
861         case CLI_GENERATE:
862                 return NULL;    /* no completion */
863         }
864
865         if (a->argc != e->args) {
866                 return CLI_SHOWUSAGE;
867         }
868
869         event = ast_event_new(AST_EVENT_PING, AST_EVENT_IE_END);
870
871         if (!event) {
872                 return CLI_FAILURE;
873         }
874
875         ast_rwlock_rdlock(&event_types_lock);
876         event_types[AST_EVENT_PING].publish_to_stasis(event);
877         ast_rwlock_unlock(&event_types_lock);
878
879         return CLI_SUCCESS;
880 }
881
882 static char *corosync_show_config(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
883 {
884         unsigned int i;
885
886         switch (cmd) {
887         case CLI_INIT:
888                 e->command = "corosync show config";
889                 e->usage =
890                         "Usage: corosync show config\n"
891                         "       Show configuration loaded from res_corosync.conf\n";
892                 return NULL;
893
894         case CLI_GENERATE:
895                 return NULL;    /* no completion */
896         }
897
898         if (a->argc != e->args) {
899                 return CLI_SHOWUSAGE;
900         }
901
902         ast_cli(a->fd, "\n"
903                     "=============================================================\n"
904                     "=== res_corosync config =====================================\n"
905                     "=============================================================\n"
906                     "===\n");
907
908         ast_rwlock_rdlock(&event_types_lock);
909         for (i = 0; i < ARRAY_LEN(event_types); i++) {
910                 if (event_types[i].publish) {
911                         ast_cli(a->fd, "=== ==> Publishing Event Type: %s\n",
912                                         event_types[i].name);
913                 }
914                 if (event_types[i].subscribe) {
915                         ast_cli(a->fd, "=== ==> Subscribing to Event Type: %s\n",
916                                         event_types[i].name);
917                 }
918         }
919         ast_rwlock_unlock(&event_types_lock);
920
921         ast_cli(a->fd, "===\n"
922                        "=============================================================\n"
923                        "\n");
924
925         return CLI_SUCCESS;
926 }
927
928 static struct ast_cli_entry corosync_cli[] = {
929         AST_CLI_DEFINE(corosync_show_config, "Show configuration"),
930         AST_CLI_DEFINE(corosync_show_members, "Show cluster members"),
931         AST_CLI_DEFINE(corosync_ping, "Send a test ping to the cluster"),
932 };
933
934 enum {
935         PUBLISH,
936         SUBSCRIBE,
937 };
938
939 static int set_event(const char *event_type, int pubsub)
940 {
941         unsigned int i;
942
943         for (i = 0; i < ARRAY_LEN(event_types); i++) {
944                 if (!event_types[i].name || strcasecmp(event_type, event_types[i].name)) {
945                         continue;
946                 }
947
948                 switch (pubsub) {
949                 case PUBLISH:
950                         event_types[i].publish = 1;
951                         break;
952                 case SUBSCRIBE:
953                         event_types[i].subscribe = 1;
954                         break;
955                 }
956
957                 break;
958         }
959
960         return (i == ARRAY_LEN(event_types)) ? -1 : 0;
961 }
962
963 static int load_general_config(struct ast_config *cfg)
964 {
965         struct ast_variable *v;
966         int res = 0;
967         unsigned int i;
968
969         ast_rwlock_wrlock(&event_types_lock);
970
971         for (i = 0; i < ARRAY_LEN(event_types); i++) {
972                 event_types[i].publish = event_types[i].publish_default;
973                 event_types[i].subscribe = event_types[i].subscribe_default;
974         }
975
976         for (v = ast_variable_browse(cfg, "general"); v && !res; v = v->next) {
977                 if (!strcasecmp(v->name, "publish_event")) {
978                         res = set_event(v->value, PUBLISH);
979                 } else if (!strcasecmp(v->name, "subscribe_event")) {
980                         res = set_event(v->value, SUBSCRIBE);
981                 } else {
982                         ast_log(LOG_WARNING, "Unknown option '%s'\n", v->name);
983                 }
984         }
985
986         for (i = 0; i < ARRAY_LEN(event_types); i++) {
987                 if (event_types[i].publish && !event_types[i].sub) {
988                         event_types[i].sub = stasis_forward_all(event_types[i].topic_fn(),
989                                                                                                         corosync_topic());
990                         stasis_message_router_add(stasis_router,
991                                                   event_types[i].message_type_fn(),
992                                                   stasis_message_cb,
993                                                   NULL);
994                 } else if (!event_types[i].publish && event_types[i].sub) {
995                         event_types[i].sub = stasis_forward_cancel(event_types[i].sub);
996                         stasis_message_router_remove(stasis_router,
997                                                      event_types[i].message_type_fn());
998                 }
999         }
1000
1001         ast_rwlock_unlock(&event_types_lock);
1002
1003         return res;
1004 }
1005
1006 static int load_config(unsigned int reload)
1007 {
1008         static const char filename[] = "res_corosync.conf";
1009         struct ast_config *cfg;
1010         const char *cat = NULL;
1011         struct ast_flags config_flags = { 0 };
1012         int res = 0;
1013
1014         cfg = ast_config_load(filename, config_flags);
1015
1016         if (cfg == CONFIG_STATUS_FILEMISSING || cfg == CONFIG_STATUS_FILEINVALID) {
1017                 return -1;
1018         }
1019
1020         while ((cat = ast_category_browse(cfg, cat))) {
1021                 if (!strcasecmp(cat, "general")) {
1022                         res = load_general_config(cfg);
1023                 } else {
1024                         ast_log(LOG_WARNING, "Unknown configuration section '%s'\n", cat);
1025                 }
1026         }
1027
1028         ast_config_destroy(cfg);
1029
1030         return res;
1031 }
1032
1033 static void cleanup_module(void)
1034 {
1035         cs_error_t cs_err;
1036         unsigned int i;
1037
1038         if (stasis_router) {
1039
1040                 /* Unsubscribe all topic forwards and cancel all message routes */
1041                 ast_rwlock_wrlock(&event_types_lock);
1042                 for (i = 0; i < ARRAY_LEN(event_types); i++) {
1043                         if (event_types[i].sub) {
1044                                 event_types[i].sub = stasis_forward_cancel(event_types[i].sub);
1045                                 stasis_message_router_remove(stasis_router,
1046                                                              event_types[i].message_type_fn());
1047                         }
1048                         event_types[i].publish = 0;
1049                         event_types[i].subscribe = 0;
1050                 }
1051                 ast_rwlock_unlock(&event_types_lock);
1052
1053                 stasis_message_router_unsubscribe_and_join(stasis_router);
1054                 stasis_router = NULL;
1055         }
1056
1057         if (corosync_aggregate_topic) {
1058                 ao2_t_ref(corosync_aggregate_topic, -1, "Dispose of topic on cleanup");
1059                 corosync_aggregate_topic = NULL;
1060         }
1061
1062         STASIS_MESSAGE_TYPE_CLEANUP(corosync_ping_message_type);
1063
1064         if (dispatch_thread.id != AST_PTHREADT_NULL) {
1065                 char meepmeep = 'x';
1066                 dispatch_thread.stop = 1;
1067                 if (ast_carefulwrite(dispatch_thread.alert_pipe[1], &meepmeep, 1,
1068                                         5000) == -1) {
1069                         ast_log(LOG_ERROR, "Failed to write to pipe: %s (%d)\n",
1070                                         strerror(errno), errno);
1071                 }
1072                 pthread_join(dispatch_thread.id, NULL);
1073         }
1074
1075         if (dispatch_thread.alert_pipe[0] != -1) {
1076                 close(dispatch_thread.alert_pipe[0]);
1077                 dispatch_thread.alert_pipe[0] = -1;
1078         }
1079
1080         if (dispatch_thread.alert_pipe[1] != -1) {
1081                 close(dispatch_thread.alert_pipe[1]);
1082                 dispatch_thread.alert_pipe[1] = -1;
1083         }
1084
1085         if (cpg_handle && (cs_err = cpg_finalize(cpg_handle)) != CS_OK) {
1086                 ast_log(LOG_ERROR, "Failed to finalize cpg (%d)\n", (int) cs_err);
1087         }
1088         cpg_handle = 0;
1089
1090         if (cfg_handle && (cs_err = corosync_cfg_finalize(cfg_handle)) != CS_OK) {
1091                 ast_log(LOG_ERROR, "Failed to finalize cfg (%d)\n", (int) cs_err);
1092         }
1093         cfg_handle = 0;
1094
1095         ao2_cleanup(nodes);
1096         nodes = NULL;
1097 }
1098
1099 static int load_module(void)
1100 {
1101         cs_error_t cs_err;
1102         struct cpg_name name;
1103
1104         if (ast_eid_is_empty(&ast_eid_default)) {
1105                 ast_log(LOG_ERROR, "Entity ID is not set.\n");
1106                 return AST_MODULE_LOAD_DECLINE;
1107         }
1108
1109         nodes = ao2_container_alloc(23, corosync_node_hash_fn, corosync_node_cmp_fn);
1110         if (!nodes) {
1111                 goto failed;
1112         }
1113
1114         corosync_aggregate_topic = stasis_topic_create("corosync_aggregate_topic");
1115         if (!corosync_aggregate_topic) {
1116                 ast_log(AST_LOG_ERROR, "Failed to create stasis topic for corosync\n");
1117                 goto failed;
1118         }
1119
1120         stasis_router = stasis_message_router_create(corosync_aggregate_topic);
1121         if (!stasis_router) {
1122                 ast_log(AST_LOG_ERROR, "Failed to create message router for corosync topic\n");
1123                 goto failed;
1124         }
1125
1126         if (STASIS_MESSAGE_TYPE_INIT(corosync_ping_message_type) != 0) {
1127                 ast_log(AST_LOG_ERROR, "Failed to initialize corosync ping message type\n");
1128                 goto failed;
1129         }
1130
1131         if (load_config(0)) {
1132                 /* simply not configured is not a fatal error */
1133                 goto failed;
1134         }
1135
1136         if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
1137                 ast_log(LOG_ERROR, "Failed to initialize cfg: (%d)\n", (int) cs_err);
1138                 goto failed;
1139         }
1140
1141         if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks)) != CS_OK) {
1142                 ast_log(LOG_ERROR, "Failed to initialize cpg: (%d)\n", (int) cs_err);
1143                 goto failed;
1144         }
1145
1146         ast_copy_string(name.value, "asterisk", sizeof(name.value));
1147         name.length = strlen(name.value);
1148
1149         if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
1150                 ast_log(LOG_ERROR, "Failed to join: (%d)\n", (int) cs_err);
1151                 goto failed;
1152         }
1153
1154         if (pipe(dispatch_thread.alert_pipe) == -1) {
1155                 ast_log(LOG_ERROR, "Failed to create alert pipe: %s (%d)\n",
1156                                 strerror(errno), errno);
1157                 goto failed;
1158         }
1159
1160         if (ast_pthread_create_background(&dispatch_thread.id, NULL,
1161                         dispatch_thread_handler, NULL)) {
1162                 ast_log(LOG_ERROR, "Error starting CPG dispatch thread.\n");
1163                 goto failed;
1164         }
1165
1166         ast_cli_register_multiple(corosync_cli, ARRAY_LEN(corosync_cli));
1167
1168
1169         return AST_MODULE_LOAD_SUCCESS;
1170
1171 failed:
1172         cleanup_module();
1173
1174         return AST_MODULE_LOAD_DECLINE;
1175 }
1176
1177 static int unload_module(void)
1178 {
1179         ast_cli_unregister_multiple(corosync_cli, ARRAY_LEN(corosync_cli));
1180
1181         cleanup_module();
1182
1183         return 0;
1184 }
1185
1186 AST_MODULE_INFO_STANDARD_EXTENDED(ASTERISK_GPL_KEY, "Corosync");
1187