Asterisk - The Open Source Telephony Project  GIT-master-c559667
res_corosync.c
Go to the documentation of this file.
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/mwi.h"
46 #include "asterisk/stasis.h"
48 #include "asterisk/stasis_system.h"
49 #include "asterisk/taskprocessor.h"
50 
53 
54 /*! \brief Timeout for Corosync's poll process */
55 #define COROSYNC_POLL_TIMEOUT (10 * 1000)
56 
57 static void publish_mwi_to_stasis(struct ast_event *event);
60 
61 /*! \brief Join to corosync */
62 static int corosync_node_joined = 0;
63 
64 /*! \brief All the nodes that we're aware of */
65 static struct ao2_container *nodes;
66 
67 /*! \brief The internal topic used for message forwarding and pings */
69 
70 /*! \brief Our \ref stasis message router */
72 
73 /*! \brief Internal accessor for our topic */
74 static struct stasis_topic *corosync_topic(void)
75 {
77 }
78 
79 struct corosync_node {
80  /*! The corosync ID */
81  int id;
82  /*! The Asterisk EID */
83  struct ast_eid eid;
84  /*! The IP address of the node */
86 };
87 
88 /*! \brief Corosync ipc dispatch/request and reply size */
89 #define COROSYNC_IPC_BUFFER_SIZE (8192 * 128)
90 
91 /*! \brief Version of pthread_create to ensure stack is large enough */
92 #define corosync_pthread_create_background(a, b, c, d) \
93  ast_pthread_create_stack(a, b, c, d, \
94  (AST_BACKGROUND_STACKSIZE + (3 * COROSYNC_IPC_BUFFER_SIZE)), \
95  __FILE__, __FUNCTION__, __LINE__, #c)
96 
98 {
99  struct corosync_node *node;
100 
101  node = ao2_alloc_options(sizeof(*node), NULL, AO2_ALLOC_OPT_LOCK_NOLOCK);
102  if (!node) {
103  return NULL;
104  }
105 
106  memcpy(&node->eid, (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID), sizeof(node->eid));
109 
110  return node;
111 }
112 
113 static int corosync_node_hash_fn(const void *obj, const int flags)
114 {
115  const struct corosync_node *node;
116  const int *id;
117 
118  switch (flags & OBJ_SEARCH_MASK) {
119  case OBJ_SEARCH_KEY:
120  id = obj;
121  break;
122  case OBJ_SEARCH_OBJECT:
123  node = obj;
124  id = &node->id;
125  break;
126  default:
127  ast_assert(0);
128  return 0;
129  }
130  return *id;
131 }
132 
133 static int corosync_node_cmp_fn(void *obj, void *arg, int flags)
134 {
135  struct corosync_node *left = obj;
136  struct corosync_node *right = arg;
137  const int *id = arg;
138  int cmp;
139 
140  switch (flags & OBJ_SEARCH_MASK) {
141  case OBJ_SEARCH_OBJECT:
142  id = &right->id;
143  /* Fall through */
144  case OBJ_SEARCH_KEY:
145  cmp = (left->id == *id);
146  break;
148  cmp = (left->id == right->id);
149  break;
150  default:
151  /* Sort can only work on something with a full or partial key. */
152  ast_assert(0);
153  cmp = 1;
154  break;
155  }
156  return cmp ? CMP_MATCH : 0;
157 }
158 
159 
160 /*! \brief A payload wrapper around a corosync ping event */
162  /*! The corosync ping event being passed over \ref stasis */
163  struct ast_event *event;
164 };
165 
166 /*! \brief Destructor for the \ref corosync_ping_payload wrapper object */
167 static void corosync_ping_payload_dtor(void *obj)
168 {
169  struct corosync_ping_payload *payload = obj;
170 
171  ast_free(payload->event);
172 }
173 
174 /*! \brief Convert a Corosync PING to a \ref ast_event */
176 {
177  struct corosync_ping_payload *payload;
178  struct ast_event *event;
179  struct ast_eid *event_eid;
180 
181  if (!message) {
182  return NULL;
183  }
184 
185  payload = stasis_message_data(message);
186 
187  if (!payload->event) {
188  return NULL;
189  }
190 
191  event_eid = (struct ast_eid *)ast_event_get_ie_raw(payload->event, AST_EVENT_IE_EID);
192 
194  AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, event_eid, sizeof(*event_eid),
196 
197  return event;
198 }
199 
200 STASIS_MESSAGE_TYPE_DEFN_LOCAL(corosync_ping_message_type,
201  .to_event = corosync_ping_to_event, );
202 
203 /*! \brief Publish a Corosync ping to \ref stasis */
205 {
206  struct corosync_ping_payload *payload;
207  struct stasis_message *message;
208  struct ast_eid *event_eid;
209 
211  ast_assert(event != NULL);
212 
213  if (!corosync_ping_message_type()) {
214  return;
215  }
216 
217  payload = ao2_t_alloc(sizeof(*payload), corosync_ping_payload_dtor, "Create ping payload");
218  if (!payload) {
219  return;
220  }
221  event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
223  AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, event_eid, sizeof(*event_eid),
225 
226  message = stasis_message_create(corosync_ping_message_type(), payload);
227  if (!message) {
228  ao2_t_ref(payload, -1, "Destroy payload on off nominal");
229  return;
230  }
231 
232  stasis_publish(corosync_topic(), message);
233 
234  ao2_t_ref(payload, -1, "Hand ref to stasis");
235  ao2_t_ref(message, -1, "Hand ref to stasis");
236 }
237 
238 static struct {
239  const char *name;
241  unsigned char publish;
242  unsigned char publish_default;
243  unsigned char subscribe;
244  unsigned char subscribe_default;
245  struct stasis_topic *(* topic_fn)(void);
246  struct stasis_cache *(* cache_fn)(void);
247  struct stasis_message_type *(* message_type_fn)(void);
248  void (* publish_to_stasis)(struct ast_event *);
249 } event_types[] = {
250  [AST_EVENT_MWI] = { .name = "mwi",
251  .topic_fn = ast_mwi_topic_all,
252  .cache_fn = ast_mwi_state_cache,
253  .message_type_fn = ast_mwi_state_type,
254  .publish_to_stasis = publish_mwi_to_stasis, },
255  [AST_EVENT_DEVICE_STATE_CHANGE] = { .name = "device_state",
256  .topic_fn = ast_device_state_topic_all,
257  .cache_fn = ast_device_state_cache,
258  .message_type_fn = ast_device_state_message_type,
259  .publish_to_stasis = publish_device_state_to_stasis, },
260  [AST_EVENT_PING] = { .name = "ping",
261  .publish_default = 1,
262  .subscribe_default = 1,
263  .topic_fn = corosync_topic,
264  .message_type_fn = corosync_ping_message_type,
265  .publish_to_stasis = publish_corosync_ping_to_stasis, },
266  [AST_EVENT_CLUSTER_DISCOVERY] = { .name = "cluster_discovery",
267  .publish_default = 1,
268  .subscribe_default = 1,
269  .topic_fn = ast_system_topic,
270  .message_type_fn = ast_cluster_discovery_type,
271  .publish_to_stasis = publish_cluster_discovery_to_stasis, },
272 };
273 
274 static struct {
275  pthread_t id;
276  int alert_pipe[2];
277  unsigned int stop:1;
278 } dispatch_thread = {
279  .id = AST_PTHREADT_NULL,
280  .alert_pipe = { -1, -1 },
281 };
282 
283 static cpg_handle_t cpg_handle;
284 static corosync_cfg_handle_t cfg_handle;
285 
286 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
287 static void cfg_state_track_cb(
288  corosync_cfg_state_notification_buffer_t *notification_buffer,
289  cs_error_t error);
290 #endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
291 
292 static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
293  corosync_cfg_shutdown_flags_t flags);
294 
295 static corosync_cfg_callbacks_t cfg_callbacks = {
296 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
297  .corosync_cfg_state_track_callback = cfg_state_track_cb,
298 #endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
299  .corosync_cfg_shutdown_callback = cfg_shutdown_cb,
300 };
301 
302 /*! \brief Publish cluster discovery to \ref stasis */
304 {
305  struct ast_json *json;
306  struct ast_json_payload *payload;
307  struct stasis_message *message;
308  char eid[18];
309  const char *addr;
310 
311  ast_eid_to_str(eid, sizeof(eid), &node->eid);
312  addr = ast_sockaddr_stringify_addr(&node->addr);
313 
314  ast_log(AST_LOG_NOTICE, "Node %u (%s) at %s %s the cluster\n",
315  node->id,
316  eid,
317  addr,
318  joined ? "joined" : "left");
319 
320  json = ast_json_pack("{s: s, s: i, s: s, s: i}",
321  "address", addr,
322  "node_id", node->id,
323  "eid", eid,
324  "joined", joined);
325  if (!json) {
326  return;
327  }
328 
329  payload = ast_json_payload_create(json);
330  if (!payload) {
331  ast_json_unref(json);
332  return;
333  }
334 
336  if (!message) {
337  ast_json_unref(json);
338  ao2_ref(payload, -1);
339  return;
340  }
341 
342  stasis_publish(ast_system_topic(), message);
343  ast_json_unref(json);
344  ao2_ref(payload, -1);
345  ao2_ref(message, -1);
346 }
347 
348 static void send_cluster_notify(void);
349 
350 /*! \brief Publish a received cluster discovery \ref ast_event to \ref stasis */
352 {
353  struct corosync_node *node;
355  struct ast_eid *event_eid;
356 
358 
359  event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
360  if (!event_eid || !ast_eid_cmp(&ast_eid_default, event_eid)) {
361  /* Don't feed events back in that originated locally. */
362  return;
363  }
364 
365  ao2_lock(nodes);
366  node = ao2_find(nodes, &id, OBJ_SEARCH_KEY | OBJ_NOLOCK);
367  if (node) {
368  /* We already know about this node */
369  ao2_unlock(nodes);
370  ao2_ref(node, -1);
371  return;
372  }
373 
374  node = corosync_node_alloc(event);
375  if (!node) {
376  ao2_unlock(nodes);
377  return;
378  }
379  ao2_link_flags(nodes, node, OBJ_NOLOCK);
380  ao2_unlock(nodes);
381 
383 
384  ao2_ref(node, -1);
385 
386  /*
387  * When we get news that someone else has joined, we need to let them
388  * know we exist as well.
389  */
391 }
392 
393 /*! \brief Publish a received MWI \ref ast_event to \ref stasis */
395 {
396  const char *mailbox;
397  const char *context;
398  unsigned int new_msgs;
399  unsigned int old_msgs;
400  struct ast_eid *event_eid;
401 
403 
404  mailbox = ast_event_get_ie_str(event, AST_EVENT_IE_MAILBOX);
405  context = ast_event_get_ie_str(event, AST_EVENT_IE_CONTEXT);
406  new_msgs = ast_event_get_ie_uint(event, AST_EVENT_IE_NEWMSGS);
407  old_msgs = ast_event_get_ie_uint(event, AST_EVENT_IE_OLDMSGS);
408  event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
409 
410  if (ast_strlen_zero(mailbox) || ast_strlen_zero(context)) {
411  return;
412  }
413 
414  if (new_msgs > INT_MAX) {
415  new_msgs = INT_MAX;
416  }
417 
418  if (old_msgs > INT_MAX) {
419  old_msgs = INT_MAX;
420  }
421 
422  if (ast_publish_mwi_state_full(mailbox, context, (int)new_msgs,
423  (int)old_msgs, NULL, event_eid)) {
424  char eid[18];
425  ast_eid_to_str(eid, sizeof(eid), event_eid);
426  ast_log(LOG_WARNING, "Failed to publish MWI message for %s@%s from %s\n",
427  mailbox, context, eid);
428  }
429 }
430 
431 /*! \brief Publish a received device state \ref ast_event to \ref stasis */
433 {
434  const char *device;
435  enum ast_device_state state;
436  unsigned int cachable;
437  struct ast_eid *event_eid;
438 
440 
441  device = ast_event_get_ie_str(event, AST_EVENT_IE_DEVICE);
443  cachable = ast_event_get_ie_uint(event, AST_EVENT_IE_CACHABLE);
444  event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
445 
446  if (ast_strlen_zero(device)) {
447  return;
448  }
449 
450  if (ast_publish_device_state_full(device, state, cachable, event_eid)) {
451  char eid[18];
452  ast_eid_to_str(eid, sizeof(eid), event_eid);
453  ast_log(LOG_WARNING, "Failed to publish device state message for %s from %s\n",
454  device, eid);
455  }
456 }
457 
458 static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
459  uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len);
460 
461 static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
462  const struct cpg_address *member_list, size_t member_list_entries,
463  const struct cpg_address *left_list, size_t left_list_entries,
464  const struct cpg_address *joined_list, size_t joined_list_entries);
465 
466 static cpg_callbacks_t cpg_callbacks = {
467  .cpg_deliver_fn = cpg_deliver_cb,
468  .cpg_confchg_fn = cpg_confchg_cb,
469 };
470 
471 #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
472 static void cfg_state_track_cb(
473  corosync_cfg_state_notification_buffer_t *notification_buffer,
474  cs_error_t error)
475 {
476 }
477 #endif /* HAVE_COROSYNC_CFG_STATE_TRACK */
478 
479 static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle,
480  corosync_cfg_shutdown_flags_t flags)
481 {
482 }
483 
484 static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name,
485  uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
486 {
487  struct ast_event *event;
488  void (*publish_handler)(struct ast_event *) = NULL;
489  enum ast_event_type event_type;
490  struct ast_eid *event_eid;
491 
492  if (msg_len < ast_event_minimum_length()) {
493  ast_debug(1, "Ignoring event that's too small. %u < %u\n",
494  (unsigned int) msg_len,
495  (unsigned int) ast_event_minimum_length());
496  return;
497  }
498 
499  event_eid = (struct ast_eid *)ast_event_get_ie_raw(msg, AST_EVENT_IE_EID);
500  if (!event_eid || !ast_eid_cmp(&ast_eid_default, event_eid)) {
501  /* Don't feed events back in that originated locally. */
502  return;
503  }
504 
505  event_type = ast_event_get_type(msg);
506  if (event_type > AST_EVENT_TOTAL) {
507  /* Egads, we don't support this */
508  return;
509  }
510 
512  ast_debug(5, "cpg_deliver_cb rdlock\n");
513  publish_handler = event_types[event_type].publish_to_stasis;
514  if (!event_types[event_type].subscribe || !publish_handler) {
515  /* We are not configured to subscribe to these events or
516  we have no way to publish it internally. */
518  ast_debug(5, "cpg_deliver_cb unlock\n");
519  return;
520  }
522  ast_debug(5, "cpg_deliver_cb unlock\n");
523 
524  if (!(event = ast_malloc(msg_len))) {
525  return;
526  }
527 
528  memcpy(event, msg, msg_len);
529 
530  if (event_type == AST_EVENT_PING) {
531  const struct ast_eid *eid;
532  char buf[128] = "";
533 
534  eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
535  ast_eid_to_str(buf, sizeof(buf), (struct ast_eid *) eid);
536  ast_log(LOG_NOTICE, "Got event PING from server with EID: '%s'\n", buf);
537  }
538  ast_debug(5, "Publishing event %s (%u) to stasis\n",
539  ast_event_get_type_name(event), event_type);
540  publish_handler(event);
541  ast_free(event);
542 }
543 
545 {
546  cs_error_t cs_err;
547  struct iovec iov;
548 
549  iov.iov_base = (void *)event;
550  iov.iov_len = ast_event_get_size(event);
551 
552  ast_debug(5, "Publishing event %s (%u) to corosync\n",
554 
555  /* The stasis subscription will only exist if we are configured to publish
556  * these events, so just send away. */
558  ast_debug(5, "publish_event_to_corosync rdlock\n");
559  if ((cs_err = cpg_mcast_joined(cpg_handle, CPG_TYPE_FIFO, &iov, 1)) != CS_OK) {
560  ast_log(LOG_WARNING, "CPG mcast failed (%u) for event %s (%u)\n",
561  cs_err, ast_event_get_type_name(event), ast_event_get_type(event));
562  }
564  ast_debug(5, "publish_event_to_corosync unlock\n");
565  } else {
566  ast_log(LOG_WARNING, "CPG mcast not executed for event %s (%u): initializing CPG.\n",
568  }
569 }
570 
572 {
573  struct ast_event *event;
574  struct ast_eid *event_eid;
575 
576  event = stasis_message_to_event(message);
577  if (!event) {
578  return;
579  }
580 
581  event_eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
582  if (!event_eid || ast_eid_cmp(&ast_eid_default, event_eid)) {
583  /* If the event didn't originate from this server, don't send it back out. */
584  ast_event_destroy(event);
585  return;
586  }
587 
588  if (ast_event_get_type(event) == AST_EVENT_PING) {
589  const struct ast_eid *eid;
590  char buf[128] = "";
591 
592  eid = (struct ast_eid *)ast_event_get_ie_raw(event, AST_EVENT_IE_EID);
593  ast_eid_to_str(buf, sizeof(buf), (struct ast_eid *) eid);
594  ast_log(LOG_NOTICE, "Sending event PING from this server with EID: '%s'\n", buf);
595  }
596 
598  ast_event_destroy(event);
599 }
600 
601 static void stasis_message_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message)
602 {
603  if (!message) {
604  return;
605  }
606 
607  publish_to_corosync(message);
608 }
609 
610 static int dump_cache_cb(void *obj, void *arg, int flags)
611 {
612  struct stasis_message *message = obj;
613 
614  if (!message) {
615  return 0;
616  }
617 
618  publish_to_corosync(message);
619 
620  return 0;
621 }
622 
623 static int clear_node_cache(void *obj, void *arg, int flags)
624 {
625  struct stasis_message *cached_msg = obj;
626  struct stasis_topic *topic = arg;
627  struct stasis_message *msg;
628  struct ast_eid *msg_eid;
629 
630  if (!cached_msg) {
631  return 0;
632  }
633 
634  msg_eid = (struct ast_eid *)stasis_message_eid(cached_msg);
635  if(msg_eid && ast_eid_cmp(&ast_eid_default, msg_eid))
636  {
637  msg = stasis_cache_clear_create(cached_msg);
638  if (msg) {
639  stasis_publish(topic, msg);
640  ao2_cleanup(msg);
641  }
642  }
643 
644  return 0;
645 }
646 
647 static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name,
648  const struct cpg_address *member_list, size_t member_list_entries,
649  const struct cpg_address *left_list, size_t left_list_entries,
650  const struct cpg_address *joined_list, size_t joined_list_entries)
651 {
652  unsigned int i;
653 
654 
655  for (i = 0; i < left_list_entries; i++) {
656  const struct cpg_address *cpg_node = &left_list[i];
657  struct corosync_node* node;
658  unsigned int j;
659 
660  node = ao2_find(nodes, &cpg_node->nodeid, OBJ_UNLINK | OBJ_SEARCH_KEY);
661  if (!node) {
662  continue;
663  }
664 
665  for (j = 0; j < ARRAY_LEN(event_types); j++) {
666  struct ao2_container *messages;
667  int messages_count;
668 
670  ast_debug(5, "cpg_confchg_cb rdlock\n");
671  if (!event_types[j].subscribe) {
673  ast_debug(5, "cpg_confchg_cb unlock\n");
674  continue;
675  }
676 
679  ast_debug(5, "cpg_confchg_cb unlock\n");
680  continue;
681  }
683  ast_debug(5, "cpg_confchg_cb unlock\n");
684 
686 
687  messages_count = ao2_container_count(messages);
688  ast_log(LOG_NOTICE, "Clearing %i events of type %s of node %i from stasis cache.\n", messages_count, event_types[j].name, node->id);
690  ast_log(LOG_NOTICE, "Cleared events of type %s from stasis cache.\n", event_types[j].name);
691 
692  ao2_t_ref(messages, -1, "Dispose of flushed cache");
693  }
694 
696  ao2_ref(node, -1);
697  }
698 
699  /* If any new nodes have joined, dump our cache of events we are publishing
700  * that originated from this server. */
701  if (!joined_list_entries) {
702  return;
703  }
704 
705  for (i = 0; i < ARRAY_LEN(event_types); i++) {
706  struct ao2_container *messages;
707  int messages_count;
708 
710  ast_debug(5, "cpg_confchg_cb rdlock\n");
711  if (!event_types[i].publish) {
713  ast_debug(5, "cpg_confchg_cb unlock\n");
714  continue;
715  }
716 
719  ast_debug(5, "cpg_confchg_cb unlock\n");
720  continue;
721  }
723  ast_debug(5, "cpg_confchg_cb unlock\n");
724 
726 
727  messages_count = ao2_container_count(messages);
728  ast_log(LOG_NOTICE, "Sending %i events of type %s to corosync.\n", messages_count, event_types[i].name);
730  ast_log(LOG_NOTICE, "Sent events of type %s to corosync.\n", event_types[i].name);
731 
732  ao2_t_ref(messages, -1, "Dispose of dumped cache");
733  }
734 }
735 
736 /*! \brief Informs the cluster of our EID and our IP addresses */
737 static void send_cluster_notify(void)
738 {
739  struct ast_event *event;
740  unsigned int node_id;
741  cs_error_t cs_err;
742  corosync_cfg_node_address_t corosync_addr;
743  int num_addrs = 0;
744  struct sockaddr *sa;
745  size_t sa_len;
746  char buf[128];
747  int res;
748 
750  ast_debug(5, "send_cluster_notify rdlock\n");
751 
752  if ((cs_err = corosync_cfg_local_get(cfg_handle, &node_id)) != CS_OK) {
753  ast_log(LOG_WARNING, "Failed to extract Corosync node ID for this node. Not informing cluster of existance.\n");
754  return;
755  }
756 
757  if (((cs_err = corosync_cfg_get_node_addrs(cfg_handle, node_id, 1, &num_addrs, &corosync_addr)) != CS_OK) || (num_addrs < 1)) {
758  ast_log(LOG_WARNING, "Failed to get local Corosync address. Not informing cluster of existance.\n");
759  return;
760  }
761 
763  ast_debug(5, "send_cluster_notify unlock\n");
764  }
765 
766  sa = (struct sockaddr *)corosync_addr.address;
767  sa_len = (size_t)corosync_addr.address_length;
768  if ((res = getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST))) {
769  ast_log(LOG_WARNING, "Failed to determine name of local Corosync address: %s (%d). Not informing cluster of existance.\n",
770  gai_strerror(res), res);
771  return;
772  }
773 
779  ast_event_destroy(event);
780 }
781 
782 static void *dispatch_thread_handler(void *data)
783 {
784  cs_error_t cs_err;
785  struct pollfd pfd[3] = {
786  { .events = POLLIN, },
787  { .events = POLLIN, },
788  { .events = POLLIN, },
789  };
790 
792  ast_debug(5, "dispatch_thread_handler rdlock\n");
793  if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
794  ast_log(LOG_ERROR, "Failed to get CPG fd. This module is now broken.\n");
796  ast_debug(5, "dispatch_thread_handler unlock\n");
797  return NULL;
798  }
799 
800  if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
801  ast_log(LOG_ERROR, "Failed to get CFG fd. This module is now broken.\n");
803  ast_debug(5, "dispatch_thread_handler unlock\n");
804  return NULL;
805  }
806 
807  pfd[2].fd = dispatch_thread.alert_pipe[0];
809  ast_debug(5, "dispatch_thread_handler unlock\n");
810  } else {
811  ast_log(LOG_ERROR, "Failed to get fd: initiliazing CPG. This module is now broken.\n");
812  return NULL;
813  }
815  while (!dispatch_thread.stop) {
816  int res;
817 
818  cs_err = CS_OK;
819 
820  pfd[0].revents = 0;
821  pfd[1].revents = 0;
822  pfd[2].revents = 0;
823 
824  res = ast_poll(pfd, ARRAY_LEN(pfd), COROSYNC_POLL_TIMEOUT);
825  if (res == -1 && errno != EINTR && errno != EAGAIN) {
826  ast_log(LOG_ERROR, "poll() error: %s (%d)\n", strerror(errno), errno);
827  cs_err = CS_ERR_BAD_HANDLE;
828  } else if (res == 0) {
829  unsigned int local_nodeid;
830 
832  ast_debug(5, "dispatch_thread_handler rdlock\n");
833  if ((cs_err = cpg_local_get(cpg_handle, &local_nodeid)) == CS_OK) {
834  struct cpg_name name;
835  struct cpg_address address[CPG_MEMBERS_MAX];
836  int entries = CPG_MEMBERS_MAX;
837 
838  ast_copy_string(name.value, "asterisk", sizeof(name.value));
839  name.length = strlen(name.value);
840  if ((cs_err = cpg_membership_get(cpg_handle, &name, address, &entries)) == CS_OK) {
841  int i;
842  int found = 0;
843 
844  ast_debug(1, "CPG group has %i node membership\n", entries);
845  for (i = 0; (i < entries) && !found; i++) {
846  if (address[i].nodeid == local_nodeid)
847  found = 1;
848  }
849  if (!found) {
850  ast_log(LOG_WARNING, "Failed to check CPG node membership\n");
852  cs_err = CS_ERR_BAD_HANDLE;
853  }
854  } else {
855  ast_log(LOG_WARNING, "Failed to get CPG node membership: %u\n", cs_err);
857  cs_err = CS_ERR_BAD_HANDLE;
858  }
859  } else {
860  ast_log(LOG_WARNING, "Failed to get CPG local node id: %u\n", cs_err);
862  cs_err = CS_ERR_BAD_HANDLE;
863  }
865  ast_debug(5, "dispatch_thread_handler unlock\n");
866  } else {
867  ast_log(LOG_WARNING, "Failed to check CPG node membership: initializing CPG.\n");
869  cs_err = CS_ERR_BAD_HANDLE;
870  }
871  } else {
873  ast_debug(5, "dispatch_thread_handler rdlock\n");
874  if (pfd[0].revents & POLLIN) {
875  if ((cs_err = cpg_dispatch(cpg_handle, CS_DISPATCH_ALL)) != CS_OK) {
876  ast_log(LOG_WARNING, "Failed CPG dispatch: %u\n", cs_err);
877  }
878  }
879 
880  if (pfd[1].revents & POLLIN) {
881  if ((cs_err = corosync_cfg_dispatch(cfg_handle, CS_DISPATCH_ALL)) != CS_OK) {
882  ast_log(LOG_WARNING, "Failed CFG dispatch: %u\n", cs_err);
883  }
884  }
886  ast_debug(5, "dispatch_thread_handler unlock\n");
887  } else {
888  ast_log(LOG_WARNING, "Failed to dispatch: initializing CPG.\n");
889  }
890  }
891  if (cs_err == CS_ERR_LIBRARY || cs_err == CS_ERR_BAD_HANDLE) {
892 
893  /* If corosync gets restarted out from under Asterisk, try to recover. */
894 
895  ast_log(LOG_NOTICE, "Attempting to recover from corosync failure.\n");
896 
898  struct cpg_name name;
899  ast_debug(5, "dispatch_thread_handler wrlock\n");
900 
902  if (cpg_handle && (cs_err = cpg_finalize(cpg_handle)) != CS_OK) {
903  ast_log(LOG_ERROR, "Failed to finalize cpg (%d)\n", (int) cs_err);
904  }
905 
906  if (cfg_handle && (cs_err = corosync_cfg_finalize(cfg_handle)) != CS_OK) {
907  ast_log(LOG_ERROR, "Failed to finalize cfg (%d)\n", (int) cs_err);
908  }
909 
910  if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
911  ast_log(LOG_ERROR, "Failed to initialize cfg (%d)\n", (int) cs_err);
913  ast_debug(5, "dispatch_thread_handler unlock\n");
914  sleep(5);
915  continue;
916  }
917 
918  if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks) != CS_OK)) {
919  ast_log(LOG_ERROR, "Failed to initialize cpg (%d)\n", (int) cs_err);
921  ast_debug(5, "dispatch_thread_handler unlock\n");
922  sleep(5);
923  continue;
924  }
925 
926  if ((cs_err = cpg_fd_get(cpg_handle, &pfd[0].fd)) != CS_OK) {
927  ast_log(LOG_ERROR, "Failed to get CPG fd.\n");
929  ast_debug(5, "dispatch_thread_handler unlock\n");
930  sleep(5);
931  continue;
932  }
933 
934  if ((cs_err = corosync_cfg_fd_get(cfg_handle, &pfd[1].fd)) != CS_OK) {
935  ast_log(LOG_ERROR, "Failed to get CFG fd.\n");
937  ast_debug(5, "dispatch_thread_handler unlock\n");
938  sleep(5);
939  continue;
940  }
941 
942  ast_copy_string(name.value, "asterisk", sizeof(name.value));
943  name.length = strlen(name.value);
944  if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
945  ast_log(LOG_ERROR, "Failed to join cpg (%d)\n", (int) cs_err);
947  ast_debug(5, "dispatch_thread_handler unlock\n");
948  sleep(5);
949  continue;
950  }
953  ast_debug(5, "dispatch_thread_handler unlock\n");
954  ast_log(LOG_NOTICE, "Corosync recovery complete.\n");
956  } else {
957  ast_log(LOG_NOTICE, "Failed to recover from corosync failure: initializing CPG.\n");
958  }
959  }
960  }
961 
962  return NULL;
963 }
964 
965 static char *corosync_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
966 {
967  cs_error_t cs_err;
968  cpg_iteration_handle_t cpg_iter;
969  struct cpg_iteration_description_t cpg_desc;
970  unsigned int i;
971 
972  switch (cmd) {
973  case CLI_INIT:
974  e->command = "corosync show members";
975  e->usage =
976  "Usage: corosync show members\n"
977  " Show corosync cluster members\n";
978  return NULL;
979 
980  case CLI_GENERATE:
981  return NULL; /* no completion */
982  }
983 
984  if (a->argc != e->args) {
985  return CLI_SHOWUSAGE;
986  }
987 
989  ast_debug(5, "corosync_show_members rdlock\n");
990  cs_err = cpg_iteration_initialize(cpg_handle, CPG_ITERATION_ALL, NULL, &cpg_iter);
991 
992  if (cs_err != CS_OK) {
993  ast_cli(a->fd, "Failed to initialize CPG iterator: %u.\n", cs_err);
994  cpg_iteration_finalize(cpg_iter);
996  ast_debug(5, "corosync_show_members unlock\n");
997  return CLI_FAILURE;
998  }
999 
1000  ast_cli(a->fd, "\n"
1001  "=============================================================\n"
1002  "=== Cluster members =========================================\n"
1003  "=============================================================\n"
1004  "===\n");
1005 
1006  for (i = 1, cs_err = cpg_iteration_next(cpg_iter, &cpg_desc);
1007  cs_err == CS_OK;
1008  cs_err = cpg_iteration_next(cpg_iter, &cpg_desc), i++) {
1009  #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
1010  corosync_cfg_node_address_t addrs[8];
1011  int num_addrs = 0;
1012  unsigned int j;
1013  #endif
1014 
1015  ast_cli(a->fd, "=== Node %u\n", i);
1016  ast_cli(a->fd, "=== --> Group: %s\n", cpg_desc.group.value);
1017 
1018  #ifdef HAVE_COROSYNC_CFG_STATE_TRACK
1019  /*
1020  * Corosync 2.x cfg lib needs to allocate 1M on stack after calling
1021  * corosync_cfg_get_node_addrs. netconsole thread has allocated only 0.5M
1022  * resulting in crash.
1023  */
1024  cs_err = corosync_cfg_get_node_addrs(cfg_handle, cpg_desc.nodeid,
1025  ARRAY_LEN(addrs), &num_addrs, addrs);
1026  if (cs_err != CS_OK) {
1027  ast_log(LOG_WARNING, "Failed to get node addresses\n");
1028  continue;
1029  }
1030 
1031  for (j = 0; j < num_addrs; j++) {
1032  struct sockaddr *sa = (struct sockaddr *) addrs[j].address;
1033  size_t sa_len = (size_t) addrs[j].address_length;
1034  char buf[128];
1035 
1036  getnameinfo(sa, sa_len, buf, sizeof(buf), NULL, 0, NI_NUMERICHOST);
1037 
1038  ast_cli(a->fd, "=== --> Address %u: %s\n", j + 1, buf);
1039  }
1040  #else
1041  ast_cli(a->fd, "=== --> Nodeid: %"PRIu32"\n", cpg_desc.nodeid);
1042  #endif
1043  }
1044 
1045  ast_cli(a->fd, "===\n"
1046  "=============================================================\n"
1047  "\n");
1048 
1049  cpg_iteration_finalize(cpg_iter);
1051  ast_debug(5, "corosync_show_members unlock\n");
1052  } else {
1053  ast_cli(a->fd, "Failed to initialize CPG iterator: initializing CPG.\n");
1054  }
1055 
1056  return CLI_SUCCESS;
1057 }
1058 
1059 static char *corosync_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
1060 {
1061  struct ast_event *event;
1062 
1063  switch (cmd) {
1064  case CLI_INIT:
1065  e->command = "corosync ping";
1066  e->usage =
1067  "Usage: corosync ping\n"
1068  " Send a test ping to the cluster.\n"
1069  "A NOTICE will be in the log for every ping received\n"
1070  "on a server.\n If you send a ping, you should see a NOTICE\n"
1071  "in the log for every server in the cluster.\n";
1072  return NULL;
1073 
1074  case CLI_GENERATE:
1075  return NULL; /* no completion */
1076  }
1077 
1078  if (a->argc != e->args) {
1079  return CLI_SHOWUSAGE;
1080  }
1081 
1083 
1084  if (!event) {
1085  return CLI_FAILURE;
1086  }
1087 
1088  event_types[AST_EVENT_PING].publish_to_stasis(event);
1089 
1090  ast_event_destroy(event);
1091  return CLI_SUCCESS;
1092 }
1093 
1094 static char *corosync_show_config(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
1095 {
1096  unsigned int i;
1097 
1098  switch (cmd) {
1099  case CLI_INIT:
1100  e->command = "corosync show config";
1101  e->usage =
1102  "Usage: corosync show config\n"
1103  " Show configuration loaded from res_corosync.conf\n";
1104  return NULL;
1105 
1106  case CLI_GENERATE:
1107  return NULL; /* no completion */
1108  }
1109 
1110  if (a->argc != e->args) {
1111  return CLI_SHOWUSAGE;
1112  }
1113 
1114  ast_cli(a->fd, "\n"
1115  "=============================================================\n"
1116  "=== res_corosync config =====================================\n"
1117  "=============================================================\n"
1118  "===\n");
1119 
1121  ast_debug(5, "corosync_show_config rdlock\n");
1122  for (i = 0; i < ARRAY_LEN(event_types); i++) {
1123  if (event_types[i].publish) {
1124  ast_cli(a->fd, "=== ==> Publishing Event Type: %s\n",
1125  event_types[i].name);
1126  }
1127  if (event_types[i].subscribe) {
1128  ast_cli(a->fd, "=== ==> Subscribing to Event Type: %s\n",
1129  event_types[i].name);
1130  }
1131  }
1133  ast_debug(5, "corosync_show_config unlock\n");
1134 
1135  ast_cli(a->fd, "===\n"
1136  "=============================================================\n"
1137  "\n");
1138 
1139  return CLI_SUCCESS;
1140 }
1141 
1142 static struct ast_cli_entry corosync_cli[] = {
1143  AST_CLI_DEFINE(corosync_show_config, "Show configuration"),
1144  AST_CLI_DEFINE(corosync_show_members, "Show cluster members"),
1145  AST_CLI_DEFINE(corosync_ping, "Send a test ping to the cluster"),
1146 };
1147 
1148 enum {
1151 };
1152 
1153 static int set_event(const char *event_type, int pubsub)
1154 {
1155  unsigned int i;
1156 
1157  for (i = 0; i < ARRAY_LEN(event_types); i++) {
1158  if (!event_types[i].name || strcasecmp(event_type, event_types[i].name)) {
1159  continue;
1160  }
1161 
1162  switch (pubsub) {
1163  case PUBLISH:
1164  event_types[i].publish = 1;
1165  break;
1166  case SUBSCRIBE:
1167  event_types[i].subscribe = 1;
1168  break;
1169  }
1170 
1171  break;
1172  }
1173 
1174  return (i == ARRAY_LEN(event_types)) ? -1 : 0;
1175 }
1176 
1177 static int load_general_config(struct ast_config *cfg)
1178 {
1179  struct ast_variable *v;
1180  int res = 0;
1181  unsigned int i;
1182 
1184  ast_debug(5, "load_general_config wrlock\n");
1185 
1186  for (i = 0; i < ARRAY_LEN(event_types); i++) {
1187  event_types[i].publish = event_types[i].publish_default;
1188  event_types[i].subscribe = event_types[i].subscribe_default;
1189  }
1190 
1191  for (v = ast_variable_browse(cfg, "general"); v && !res; v = v->next) {
1192  if (!strcasecmp(v->name, "publish_event")) {
1193  res = set_event(v->value, PUBLISH);
1194  } else if (!strcasecmp(v->name, "subscribe_event")) {
1195  res = set_event(v->value, SUBSCRIBE);
1196  } else {
1197  ast_log(LOG_WARNING, "Unknown option '%s'\n", v->name);
1198  }
1199  }
1200 
1201  for (i = 0; i < ARRAY_LEN(event_types); i++) {
1202  if (event_types[i].publish && !event_types[i].sub) {
1204  corosync_topic());
1205  stasis_message_router_add(stasis_router,
1208  NULL);
1209  } else if (!event_types[i].publish && event_types[i].sub) {
1211  stasis_message_router_remove(stasis_router,
1213  }
1214  }
1215 
1217  ast_debug(5, "load_general_config unlock\n");
1218 
1219  return res;
1220 }
1221 
1222 static int load_config(unsigned int reload)
1223 {
1224  static const char filename[] = "res_corosync.conf";
1225  struct ast_config *cfg;
1226  const char *cat = NULL;
1227  struct ast_flags config_flags = { 0 };
1228  int res = 0;
1229 
1230  cfg = ast_config_load(filename, config_flags);
1231 
1233  return -1;
1234  }
1235 
1236  while ((cat = ast_category_browse(cfg, cat))) {
1237  if (!strcasecmp(cat, "general")) {
1238  res = load_general_config(cfg);
1239  } else {
1240  ast_log(LOG_WARNING, "Unknown configuration section '%s'\n", cat);
1241  }
1242  }
1243 
1244  ast_config_destroy(cfg);
1245 
1246  return res;
1247 }
1248 
1249 static void cleanup_module(void)
1250 {
1251  cs_error_t cs_err;
1252  unsigned int i;
1253 
1254  if (stasis_router) {
1255 
1256  /* Unsubscribe all topic forwards and cancel all message routes */
1257  for (i = 0; i < ARRAY_LEN(event_types); i++) {
1258  struct ao2_container *messages = NULL;
1259  int messages_count;
1260  unsigned char subscribe = 0;
1261 
1263  ast_debug(5, "cleanup_module wrlock\n");
1264  subscribe = event_types[i].subscribe;
1265 
1266  if (event_types[i].sub) {
1269  }
1270  event_types[i].publish = 0;
1271  event_types[i].subscribe = 0;
1273  ast_debug(5, "cleanup_module unlock\n");
1274 
1275  if (subscribe && event_types[i].cache_fn && event_types[i].message_type_fn) {
1277  messages_count = ao2_container_count(messages);
1278  ast_log(LOG_NOTICE, "Clearing %i events of type %s of other nodes from stasis cache.\n", messages_count, event_types[i].name);
1280  ast_log(LOG_NOTICE, "Cleared events of type %s from stasis cache.\n", event_types[i].name);
1281  ao2_t_ref(messages, -1, "Dispose of flushed cache");
1282  }
1283  }
1284 
1286  stasis_router = NULL;
1287  }
1288 
1289  if (corosync_aggregate_topic) {
1290  ao2_t_ref(corosync_aggregate_topic, -1, "Dispose of topic on cleanup");
1291  corosync_aggregate_topic = NULL;
1292  }
1293 
1294  STASIS_MESSAGE_TYPE_CLEANUP(corosync_ping_message_type);
1295 
1296  if (dispatch_thread.id != AST_PTHREADT_NULL) {
1297  char meepmeep = 'x';
1298  dispatch_thread.stop = 1;
1299  if (ast_carefulwrite(dispatch_thread.alert_pipe[1], &meepmeep, 1,
1300  5000) == -1) {
1301  ast_log(LOG_ERROR, "Failed to write to pipe: %s (%d)\n",
1302  strerror(errno), errno);
1303  }
1304  pthread_join(dispatch_thread.id, NULL);
1305  }
1306 
1307  if (dispatch_thread.alert_pipe[0] != -1) {
1308  close(dispatch_thread.alert_pipe[0]);
1309  dispatch_thread.alert_pipe[0] = -1;
1310  }
1311 
1312  if (dispatch_thread.alert_pipe[1] != -1) {
1313  close(dispatch_thread.alert_pipe[1]);
1314  dispatch_thread.alert_pipe[1] = -1;
1315  }
1316 
1318  ast_debug(5, "cleanup_module wrlock\n");
1319  if (cpg_handle && (cs_err = cpg_finalize(cpg_handle)) != CS_OK) {
1320  ast_log(LOG_ERROR, "Failed to finalize cpg (%d)\n", (int) cs_err);
1321  }
1322  cpg_handle = 0;
1323 
1324  if (cfg_handle && (cs_err = corosync_cfg_finalize(cfg_handle)) != CS_OK) {
1325  ast_log(LOG_ERROR, "Failed to finalize cfg (%d)\n", (int) cs_err);
1326  }
1327  cfg_handle = 0;
1330  ast_debug(5, "cleanup_module unlock\n");
1331  }
1332  ao2_cleanup(nodes);
1333  nodes = NULL;
1334 }
1335 
1336 static int load_module(void)
1337 {
1338  cs_error_t cs_err;
1339  struct cpg_name name;
1340 
1342  ast_log(LOG_ERROR, "Entity ID is not set.\n");
1343  return AST_MODULE_LOAD_DECLINE;
1344  }
1345 
1348  if (!nodes) {
1349  goto failed;
1350  }
1351 
1352  corosync_aggregate_topic = stasis_topic_create("corosync:aggregator");
1353  if (!corosync_aggregate_topic) {
1354  ast_log(AST_LOG_ERROR, "Failed to create stasis topic for corosync\n");
1355  goto failed;
1356  }
1357 
1358  stasis_router = stasis_message_router_create(corosync_aggregate_topic);
1359  if (!stasis_router) {
1360  ast_log(AST_LOG_ERROR, "Failed to create message router for corosync topic\n");
1361  goto failed;
1362  }
1365 
1366  if (STASIS_MESSAGE_TYPE_INIT(corosync_ping_message_type) != 0) {
1367  ast_log(AST_LOG_ERROR, "Failed to initialize corosync ping message type\n");
1368  goto failed;
1369  }
1370 
1371  if (load_config(0)) {
1372  /* simply not configured is not a fatal error */
1373  goto failed;
1374  }
1375 
1378  ast_debug(5, "load_module wrlock\n");
1379  if ((cs_err = corosync_cfg_initialize(&cfg_handle, &cfg_callbacks)) != CS_OK) {
1380  ast_log(LOG_ERROR, "Failed to initialize cfg: (%d)\n", (int) cs_err);
1382  ast_debug(5, "load_module unlock\n");
1383  goto failed;
1384  }
1385 
1386  if ((cs_err = cpg_initialize(&cpg_handle, &cpg_callbacks)) != CS_OK) {
1387  ast_log(LOG_ERROR, "Failed to initialize cpg: (%d)\n", (int) cs_err);
1389  ast_debug(5, "load_module unlock\n");
1390  goto failed;
1391  }
1392 
1393  ast_copy_string(name.value, "asterisk", sizeof(name.value));
1394  name.length = strlen(name.value);
1395 
1396  if ((cs_err = cpg_join(cpg_handle, &name)) != CS_OK) {
1397  ast_log(LOG_ERROR, "Failed to join: (%d)\n", (int) cs_err);
1399  ast_debug(5, "load_module unlock\n");
1400  goto failed;
1401  }
1402 
1403  if (pipe(dispatch_thread.alert_pipe) == -1) {
1404  ast_log(LOG_ERROR, "Failed to create alert pipe: %s (%d)\n",
1405  strerror(errno), errno);
1407  ast_debug(5, "load_module unlock\n");
1408  goto failed;
1409  }
1411 
1413  ast_debug(5, "load_module unlock\n");
1416  ast_log(LOG_ERROR, "Error starting CPG dispatch thread.\n");
1417  goto failed;
1418  }
1419 
1420  ast_cli_register_multiple(corosync_cli, ARRAY_LEN(corosync_cli));
1421  } else {
1422  goto failed;
1423  }
1424 
1425  return AST_MODULE_LOAD_SUCCESS;
1426 
1427 failed:
1428  cleanup_module();
1429 
1430  return AST_MODULE_LOAD_DECLINE;
1431 }
1432 
1433 static int unload_module(void)
1434 {
1435  ast_cli_unregister_multiple(corosync_cli, ARRAY_LEN(corosync_cli));
1436 
1437  cleanup_module();
1438 
1439  return 0;
1440 }
1441 
static char * ast_sockaddr_stringify_addr(const struct ast_sockaddr *addr)
Wrapper around ast_sockaddr_stringify_fmt() to return an address only.
Definition: netsock2.h:290
struct ao2_container * stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
Dump all entity items from the cache to a subscription.
Definition: stasis_cache.c:757
#define ao2_t_ref(o, delta, tag)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:463
struct ast_variable * next
static void stasis_message_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message)
Definition: res_corosync.c:601
enum sip_cc_notify_state state
Definition: chan_sip.c:957
#define ast_rwlock_rdlock(a)
Definition: lock.h:233
Definition: test_heap.c:38
#define AST_CLI_DEFINE(fn, txt,...)
Definition: cli.h:197
struct ast_eid eid
Definition: res_corosync.c:83
ast_device_state
Device States.
Definition: devicestate.h:52
An event.
Definition: event.c:81
static struct corosync_node * corosync_node_alloc(struct ast_event *event)
Definition: res_corosync.c:97
static void * dispatch_thread_handler(void *data)
Definition: res_corosync.c:782
Asterisk main include file. File version handling, generic pbx functions.
int ao2_container_count(struct ao2_container *c)
Returns the number of elements in a container.
#define AST_RWLOCK_DEFINE_STATIC(rwlock)
Definition: lock.h:541
#define ARRAY_LEN(a)
Definition: isdn_lib.c:42
static cpg_handle_t cpg_handle
Definition: res_corosync.c:283
static void publish_to_corosync(struct stasis_message *message)
Definition: res_corosync.c:571
struct ast_json * ast_json_pack(char const *format,...)
Helper for creating complex JSON values.
Definition: json.c:591
enum ast_event_type ast_event_get_type(const struct ast_event *event)
Get the type for an event.
Definition: event.c:288
static char * corosync_show_config(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
#define ao2_t_alloc(data_size, destructor_fn, debug_msg)
Definition: astobj2.h:409
int ast_sockaddr_parse(struct ast_sockaddr *addr, const char *str, int flags)
Parse an IPv4 or IPv6 address string.
Definition: netsock2.c:230
int ast_cli_unregister_multiple(struct ast_cli_entry *e, int len)
Unregister multiple commands.
Definition: clicompat.c:30
#define COROSYNC_POLL_TIMEOUT
Timeout for Corosync&#39;s poll process.
Definition: res_corosync.c:55
char * ast_eid_to_str(char *s, int maxlen, struct ast_eid *eid)
Convert an EID to a string.
Definition: main/utils.c:2458
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1105
struct ast_json_payload * ast_json_payload_create(struct ast_json *json)
Create an ao2 object to pass json blobs as data payloads for stasis.
Definition: json.c:735
struct ast_variable * ast_variable_browse(const struct ast_config *config, const char *category_name)
Definition: extconf.c:1216
Device state management.
void ast_json_unref(struct ast_json *value)
Decrease refcount on value. If refcount reaches zero, value is freed.
Definition: json.c:73
char * address
Definition: f2c.h:59
char buf[BUFSIZE]
Definition: eagi_proxy.c:66
int ast_carefulwrite(int fd, char *s, int len, int timeoutms)
Try to write string, but wait no more than ms milliseconds before timing out.
Definition: main/utils.c:1463
int stasis_message_router_add(struct stasis_message_router *router, struct stasis_message_type *message_type, stasis_subscription_cb callback, void *data)
Add a route to a message router.
struct stasis_cache * ast_mwi_state_cache(void)
Backend cache for ast_mwi_topic_cached().
Definition: mwi.c:90
#define STASIS_MESSAGE_TYPE_INIT(name)
Boiler-plate messaging macro for initializing message types.
Definition: stasis.h:1501
static int dump_cache_cb(void *obj, void *arg, int flags)
Definition: res_corosync.c:610
static void corosync_ping_payload_dtor(void *obj)
Destructor for the corosync_ping_payload wrapper object.
Definition: res_corosync.c:167
static struct @449 dispatch_thread
Stasis Message Bus API. See Stasis Message Bus API for detailed documentation.
descriptor for a cli entry.
Definition: cli.h:171
const int argc
Definition: cli.h:160
#define LOG_WARNING
Definition: logger.h:274
static ast_rwlock_t event_types_lock
Definition: res_corosync.c:51
static struct ast_cli_entry corosync_cli[]
#define AST_TASKPROCESSOR_HIGH_WATER_LEVEL
Definition: taskprocessor.h:63
static int clear_node_cache(void *obj, void *arg, int flags)
Definition: res_corosync.c:623
#define ao2_callback(c, flags, cb_fn, arg)
Definition: astobj2.h:1716
#define CONFIG_STATUS_FILEINVALID
void stasis_message_router_unsubscribe_and_join(struct stasis_message_router *router)
Unsubscribe the router from the upstream topic, blocking until the final message has been processed...
static void send_cluster_notify(void)
Informs the cluster of our EID and our IP addresses.
Definition: res_corosync.c:737
Structure for variables, used for configurations and for channel variables.
#define STASIS_MESSAGE_TYPE_CLEANUP(name)
Boiler-plate messaging macro for cleaning up message types.
Definition: stasis.h:1523
Definition: cli.h:152
Assume that the ao2_container is already locked.
Definition: astobj2.h:1067
Definition: astman.c:222
struct stasis_message * stasis_cache_clear_create(struct stasis_message *message)
A message which instructs the caching topic to remove an entry from its cache.
Definition: stasis_cache.c:778
#define ast_cli_register_multiple(e, len)
Register multiple commands.
Definition: cli.h:265
Number of new messages Used by: AST_EVENT_MWI Payload type: UINT.
Definition: event_defs.h:77
Number of Used by: AST_EVENT_MWI Payload type: UINT.
Definition: event_defs.h:83
#define ao2_alloc_options(data_size, destructor_fn, options)
Definition: astobj2.h:406
#define ast_assert(a)
Definition: utils.h:650
#define ao2_link_flags(container, obj, flags)
Definition: astobj2.h:1572
#define ao2_unlock(a)
Definition: astobj2.h:730
static void publish_cluster_discovery_to_stasis(struct ast_event *event)
Publish a received cluster discovery ast_event to Stasis Message Bus API.
Definition: res_corosync.c:351
struct stasis_topic * ast_mwi_topic_all(void)
Get the Stasis Message Bus API topic for MWI messages.
Definition: mwi.c:85
char * ast_category_browse(struct ast_config *config, const char *prev_name)
Browse categories.
Definition: extconf.c:3328
#define NULL
Definition: resample.c:96
void ast_cli(int fd, const char *fmt,...)
Definition: clicompat.c:6
#define ast_rwlock_unlock(a)
Definition: lock.h:232
static void publish_cluster_discovery_to_stasis_full(struct corosync_node *node, int joined)
Publish cluster discovery to Stasis Message Bus API.
Definition: res_corosync.c:303
Socket address structure.
Definition: netsock2.h:97
An Entity ID is essentially a MAC address, brief and unique.
Definition: utils.h:726
unsigned char subscribe_default
Definition: res_corosync.c:244
static int unload_module(void)
Entity ID Used by All events Payload type: RAW This IE indicates which server the event originated fr...
Definition: event_defs.h:272
int ast_eid_cmp(const struct ast_eid *eid1, const struct ast_eid *eid2)
Compare two EIDs.
Definition: main/utils.c:2713
unsigned char publish
Definition: res_corosync.c:241
const char * name
Definition: res_corosync.c:239
int args
This gets set in ast_cli_register()
Definition: cli.h:185
static int load_general_config(struct ast_config *cfg)
static char * corosync_ping(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
static void cpg_confchg_cb(cpg_handle_t handle, const struct cpg_name *group_name, const struct cpg_address *member_list, size_t member_list_entries, const struct cpg_address *left_list, size_t left_list_entries, const struct cpg_address *joined_list, size_t joined_list_entries)
Definition: res_corosync.c:647
#define ast_rwlock_tryrdlock(a)
Definition: lock.h:235
#define AST_LOG_NOTICE
Definition: logger.h:268
Configuration File Parser.
static char mailbox[AST_MAX_MAILBOX_UNIQUEID]
Definition: chan_mgcp.c:204
struct stasis_message_type * ast_device_state_message_type(void)
Get the Stasis message type for device state messages.
static void publish_corosync_ping_to_stasis(struct ast_event *event)
Publish a Corosync ping to Stasis Message Bus API.
Definition: res_corosync.c:204
#define ast_debug(level,...)
Log a DEBUG message.
Definition: logger.h:444
#define ast_log
Definition: astobj2.c:42
ast_event_type
Definition: event_defs.h:28
Context IE Used by AST_EVENT_MWI Payload type: str.
Definition: event_defs.h:127
The arg parameter is a partial search key similar to OBJ_SEARCH_KEY.
Definition: astobj2.h:1120
#define ast_config_load(filename, flags)
Load a config file.
#define AST_LOG_ERROR
Definition: logger.h:290
unsigned int stop
Definition: res_corosync.c:277
const int fd
Definition: cli.h:159
const struct ast_eid * stasis_message_eid(const struct stasis_message *msg)
Get the entity id for a stasis_message.
#define AST_PTHREADT_NULL
Definition: lock.h:66
#define ast_poll(a, b, c)
Definition: poll-compat.h:88
static struct ast_event * corosync_ping_to_event(struct stasis_message *message)
Convert a Corosync PING to a ast_event.
Definition: res_corosync.c:175
#define ao2_ref(o, delta)
Definition: astobj2.h:464
void ast_config_destroy(struct ast_config *config)
Destroys a config.
Definition: extconf.c:1290
#define ao2_lock(a)
Definition: astobj2.h:718
#define ast_malloc(len)
A wrapper for malloc()
Definition: astmm.h:193
static void cpg_deliver_cb(cpg_handle_t handle, const struct cpg_name *group_name, uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
Definition: res_corosync.c:484
struct stasis_topic * stasis_topic_create(const char *name)
Create a new topic.
Definition: stasis.c:614
static void publish_device_state_to_stasis(struct ast_event *event)
Publish a received device state ast_event to Stasis Message Bus API.
Definition: res_corosync.c:432
struct ast_sockaddr addr
Definition: res_corosync.c:85
struct ao2_container * stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
Dump cached items to a subscription for a specific entity.
Definition: stasis_cache.c:718
Event non-cachability flag Used by: All events Payload type: UINT.
Definition: event_defs.h:306
#define stasis_message_router_create(topic)
const void * ast_event_get_ie_raw(const struct ast_event *event, enum ast_event_ie_type ie_type)
Get the value of an information element that has a raw payload.
Definition: event.c:311
static int corosync_node_joined
Join to corosync.
Definition: res_corosync.c:62
static int corosync_node_cmp_fn(void *obj, void *arg, int flags)
Definition: res_corosync.c:133
int ast_publish_mwi_state_full(const char *mailbox, const char *context, int new_msgs, int old_msgs, const char *channel_id, struct ast_eid *eid)
Publish a MWI state update via stasis with all parameters.
Definition: mwi.c:388
static cpg_callbacks_t cpg_callbacks
Definition: res_corosync.c:466
struct ast_event * event
Definition: res_corosync.c:163
int ast_publish_device_state_full(const char *device, enum ast_device_state state, enum ast_devstate_cache cachable, struct ast_eid *eid)
Publish a device state update with EID.
Definition: devicestate.c:709
A payload wrapper around a corosync ping event.
Definition: res_corosync.c:161
#define LOG_ERROR
Definition: logger.h:285
#define ao2_container_alloc_hash(ao2_options, container_options, n_buckets, hash_fn, sort_fn, cmp_fn)
Definition: astobj2.h:1310
struct stasis_cache * ast_device_state_cache(void)
Backend cache for ast_device_state_topic_cached()
Definition: devicestate.c:673
#define CLI_SHOWUSAGE
Definition: cli.h:45
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.
struct stasis_message * stasis_message_create(struct stasis_message_type *type, void *data)
Create a new message.
void stasis_message_router_remove(struct stasis_message_router *router, struct stasis_message_type *message_type)
Remove a route from a message router.
int alert_pipe[2]
Definition: res_corosync.c:276
int errno
#define corosync_pthread_create_background(a, b, c, d)
Version of pthread_create to ensure stack is large enough.
Definition: res_corosync.c:92
static void cleanup_module(void)
#define LOG_NOTICE
Definition: logger.h:263
struct stasis_topic * ast_device_state_topic_all(void)
Get the Stasis topic for device state messages.
Definition: devicestate.c:668
struct stasis_topic * ast_system_topic(void)
A Stasis Message Bus API topic which publishes messages regarding system changes. ...
struct stasis_topic *(* topic_fn)(void)
Definition: res_corosync.c:245
static ast_rwlock_t init_cpg_lock
Definition: res_corosync.c:52
#define ast_strlen_zero(a)
Definition: muted.c:73
struct ast_event * stasis_message_to_event(struct stasis_message *msg)
Build the Generic event system representation of the message.
#define CLI_FAILURE
Definition: cli.h:46
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
Publish a message to a topic&#39;s subscribers.
Definition: stasis.c:1507
#define ast_free(a)
Definition: astmm.h:182
char * command
Definition: cli.h:186
size_t ast_event_minimum_length(void)
Get the minimum length of an ast_event.
Definition: event.c:529
static int reload(void)
Definition: cdr_mysql.c:741
static corosync_cfg_callbacks_t cfg_callbacks
Definition: res_corosync.c:295
struct stasis_message_type * ast_cluster_discovery_type(void)
A stasis_message_type for Cluster discovery.
struct stasis_forward * stasis_forward_cancel(struct stasis_forward *forward)
Definition: stasis.c:1544
STASIS_MESSAGE_TYPE_DEFN_LOCAL(corosync_ping_message_type,.to_event=corosync_ping_to_event,)
struct stasis_message_type * ast_mwi_state_type(void)
Get the Stasis Message Bus API message type for MWI messages.
Module has failed to load, may be in an inconsistent state.
Definition: module.h:78
#define ao2_find(container, arg, flags)
Definition: astobj2.h:1756
An API for managing task processing threads that can be shared across modules.
static int load_module(void)
Structure used to handle boolean flags.
Definition: utils.h:199
Support for logging to various files, console and syslog Configuration in file logger.conf.
static void publish_mwi_to_stasis(struct ast_event *event)
Publish a received MWI ast_event to Stasis Message Bus API.
Definition: res_corosync.c:394
const char * usage
Definition: cli.h:177
uint32_t ast_event_get_ie_uint(const struct ast_event *event, enum ast_event_ie_type ie_type)
Get the value of an information element that has an integer payload.
Definition: event.c:293
#define CONFIG_STATUS_FILEMISSING
static corosync_cfg_handle_t cfg_handle
Definition: res_corosync.c:284
#define ast_rwlock_wrlock(a)
Definition: lock.h:234
struct ast_eid ast_eid_default
Global EID.
Definition: options.c:93
#define CLI_SUCCESS
Definition: cli.h:44
The arg parameter is an object of the same type.
Definition: astobj2.h:1091
void ast_event_destroy(struct ast_event *event)
Destroy an event.
Definition: event.c:524
unsigned char subscribe
Definition: res_corosync.c:243
#define ao2_cleanup(obj)
Definition: astobj2.h:1958
Standard Command Line Interface.
struct ast_event * ast_event_new(enum ast_event_type event_type,...)
Create a new event.
Definition: event.c:402
void ast_copy_string(char *dst, const char *src, size_t size)
Size-limited null-terminating string copy.
Definition: strings.h:401
Asterisk MWI API.
AST_MODULE_INFO_STANDARD_EXTENDED(ASTERISK_GPL_KEY, "Corosync")
#define ast_rwlock_trywrlock(a)
Definition: lock.h:236
static int set_event(const char *event_type, int pubsub)
static struct stasis_message_router * stasis_router
Our Stasis Message Bus API message router.
Definition: res_corosync.c:71
struct stasis_forward * sub
Definition: res_corosync.c:240
Abstract JSON element (object, array, string, int, ...).
int error(const char *format,...)
Definition: utils/frame.c:999
Forwarding information.
Definition: stasis.c:1527
static int load_config(unsigned int reload)
size_t ast_event_get_size(const struct ast_event *event)
Get the size of an event.
Definition: event.c:228
static struct stasis_topic * corosync_aggregate_topic
The internal topic used for message forwarding and pings.
Definition: res_corosync.c:68
const char * ast_event_get_type_name(const struct ast_event *event)
Get the string representation of the type of the given event.
Definition: event.c:194
int ast_eid_is_empty(const struct ast_eid *eid)
Check if EID is empty.
Definition: main/utils.c:2718
Generic container type.
static void publish_event_to_corosync(struct ast_event *event)
Definition: res_corosync.c:544
Search option field mask.
Definition: astobj2.h:1076
static char context[AST_MAX_CONTEXT]
Definition: chan_alsa.c:116
struct stasis_forward * stasis_forward_all(struct stasis_topic *from_topic, struct stasis_topic *to_topic)
Create a subscription which forwards all messages from one topic to another.
Definition: stasis.c:1574
Generic State IE Used by AST_EVENT_DEVICE_STATE_CHANGE Payload type: UINT The actual state values dep...
Definition: event_defs.h:121
const char * ast_event_get_ie_str(const struct ast_event *event, enum ast_event_ie_type ie_type)
Get the value of an information element that has a string payload.
Definition: event.c:302
int stasis_message_router_set_congestion_limits(struct stasis_message_router *router, long low_water, long high_water)
Set the high and low alert water marks of the stasis message router.
unsigned char publish_default
Definition: res_corosync.c:242
static struct @448 event_types[]
#define ASTERISK_GPL_KEY
The text the key() function should return.
Definition: module.h:46
static int corosync_node_hash_fn(const void *obj, const int flags)
Definition: res_corosync.c:113
Asterisk module definitions.
struct stasis_message_type *(* message_type_fn)(void)
Definition: res_corosync.c:247
void(* publish_to_stasis)(struct ast_event *)
Definition: res_corosync.c:248
Device Name Used by AST_EVENT_DEVICE_STATE_CHANGE Payload type: STR.
Definition: event_defs.h:113
struct stasis_cache *(* cache_fn)(void)
Definition: res_corosync.c:246
Cluster node ID Used by: Corosync Payload type: UINT.
Definition: event_defs.h:313
Mailbox name.
Definition: event_defs.h:89
static struct test_val a
static char * corosync_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
Definition: res_corosync.c:965
static struct stasis_topic * corosync_topic(void)
Internal accessor for our topic.
Definition: res_corosync.c:74
static void cfg_shutdown_cb(corosync_cfg_handle_t cfg_handle, corosync_cfg_shutdown_flags_t flags)
Definition: res_corosync.c:479
static struct ao2_container * nodes
All the nodes that we&#39;re aware of.
Definition: res_corosync.c:65