corosync  2.4.6
totemsrp.c
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2003-2006 MontaVista Software, Inc.
3  * Copyright (c) 2006-2009 Red Hat, Inc.
4  *
5  * All rights reserved.
6  *
7  * Author: Steven Dake (sdake@redhat.com)
8  *
9  * This software licensed under BSD license, the text of which follows:
10  *
11  * Redistribution and use in source and binary forms, with or without
12  * modification, are permitted provided that the following conditions are met:
13  *
14  * - Redistributions of source code must retain the above copyright notice,
15  * this list of conditions and the following disclaimer.
16  * - Redistributions in binary form must reproduce the above copyright notice,
17  * this list of conditions and the following disclaimer in the documentation
18  * and/or other materials provided with the distribution.
19  * - Neither the name of the MontaVista Software, Inc. nor the names of its
20  * contributors may be used to endorse or promote products derived from this
21  * software without specific prior written permission.
22  *
23  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
24  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
25  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
26  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
27  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
28  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
29  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
30  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
31  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
32  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
33  * THE POSSIBILITY OF SUCH DAMAGE.
34  */
35 
36 /*
37  * The first version of this code was based upon Yair Amir's PhD thesis:
38  * http://www.cs.jhu.edu/~yairamir/phd.ps) (ch4,5).
39  *
40  * The current version of totemsrp implements the Totem protocol specified in:
41  * http://citeseer.ist.psu.edu/amir95totem.html
42  *
43  * The deviations from the above published protocols are:
44  * - encryption of message contents with nss
45  * - authentication of meessage contents with SHA1/HMAC
46  * - token hold mode where token doesn't rotate on unused ring - reduces cpu
47  * usage on 1.6ghz xeon from 35% to less then .1 % as measured by top
48  */
49 
50 #include <config.h>
51 
52 #include <assert.h>
53 #ifdef HAVE_ALLOCA_H
54 #include <alloca.h>
55 #endif
56 #include <sys/mman.h>
57 #include <sys/types.h>
58 #include <sys/stat.h>
59 #include <sys/socket.h>
60 #include <netdb.h>
61 #include <sys/un.h>
62 #include <sys/ioctl.h>
63 #include <sys/param.h>
64 #include <netinet/in.h>
65 #include <arpa/inet.h>
66 #include <unistd.h>
67 #include <fcntl.h>
68 #include <stdlib.h>
69 #include <stdio.h>
70 #include <errno.h>
71 #include <sched.h>
72 #include <time.h>
73 #include <sys/time.h>
74 #include <sys/poll.h>
75 #include <sys/uio.h>
76 #include <limits.h>
77 
78 #include <qb/qbdefs.h>
79 #include <qb/qbutil.h>
80 #include <qb/qbloop.h>
81 
82 #include <corosync/swab.h>
83 #include <corosync/sq.h>
84 #include <corosync/list.h>
85 
86 #define LOGSYS_UTILS_ONLY 1
87 #include <corosync/logsys.h>
88 
89 #include "totemsrp.h"
90 #include "totemrrp.h"
91 #include "totemnet.h"
92 
93 #include "cs_queue.h"
94 
95 #define LOCALHOST_IP inet_addr("127.0.0.1")
96 #define QUEUE_RTR_ITEMS_SIZE_MAX 16384 /* allow 16384 retransmit items */
97 #define RETRANS_MESSAGE_QUEUE_SIZE_MAX 16384 /* allow 500 messages to be queued */
98 #define RECEIVED_MESSAGE_QUEUE_SIZE_MAX 500 /* allow 500 messages to be queued */
99 #define MAXIOVS 5
100 #define RETRANSMIT_ENTRIES_MAX 30
101 #define TOKEN_SIZE_MAX 64000 /* bytes */
102 #define LEAVE_DUMMY_NODEID 0
103 
104 /*
105  * Rollover handling:
106  * SEQNO_START_MSG is the starting sequence number after a new configuration
107  * This should remain zero, unless testing overflow in which case
108  * 0x7ffff000 and 0xfffff000 are good starting values.
109  *
110  * SEQNO_START_TOKEN is the starting sequence number after a new configuration
111  * for a token. This should remain zero, unless testing overflow in which
112  * case 07fffff00 or 0xffffff00 are good starting values.
113  */
114 #define SEQNO_START_MSG 0x0
115 #define SEQNO_START_TOKEN 0x0
116 
117 /*
118  * These can be used ot test different rollover points
119  * #define SEQNO_START_MSG 0xfffffe00
120  * #define SEQNO_START_TOKEN 0xfffffe00
121  */
122 
123 /*
124  * These can be used to test the error recovery algorithms
125  * #define TEST_DROP_ORF_TOKEN_PERCENTAGE 30
126  * #define TEST_DROP_COMMIT_TOKEN_PERCENTAGE 30
127  * #define TEST_DROP_MCAST_PERCENTAGE 50
128  * #define TEST_RECOVERY_MSG_COUNT 300
129  */
130 
131 /*
132  * we compare incoming messages to determine if their endian is
133  * different - if so convert them
134  *
135  * do not change
136  */
137 #define ENDIAN_LOCAL 0xff22
138 
140  MESSAGE_TYPE_ORF_TOKEN = 0, /* Ordering, Reliability, Flow (ORF) control Token */
141  MESSAGE_TYPE_MCAST = 1, /* ring ordered multicast message */
142  MESSAGE_TYPE_MEMB_MERGE_DETECT = 2, /* merge rings if there are available rings */
143  MESSAGE_TYPE_MEMB_JOIN = 3, /* membership join message */
144  MESSAGE_TYPE_MEMB_COMMIT_TOKEN = 4, /* membership commit token */
145  MESSAGE_TYPE_TOKEN_HOLD_CANCEL = 5, /* cancel the holding of the token */
146 };
147 
151 };
152 
153 /*
154  * New membership algorithm local variables
155  */
157  struct srp_addr addr;
158  int set;
159 };
160 
161 
163  struct list_head list;
164  int (*callback_fn) (enum totem_callback_token_type type, const void *);
166  int delete;
167  void *data;
168 };
169 
170 
172  int mcast;
173  int token;
174 };
175 
176 struct message_header {
177  char type;
178  char encapsulated;
179  unsigned short endian_detector;
180  unsigned int nodeid;
181 } __attribute__((packed));
182 
183 
184 struct mcast {
187  unsigned int seq;
190  unsigned int node_id;
192 } __attribute__((packed));
193 
194 
195 struct rtr_item {
197  unsigned int seq;
198 }__attribute__((packed));
199 
200 
201 struct orf_token {
203  unsigned int seq;
204  unsigned int token_seq;
205  unsigned int aru;
206  unsigned int aru_addr;
208  unsigned int backlog;
209  unsigned int fcc;
212  struct rtr_item rtr_list[0];
213 }__attribute__((packed));
214 
215 
216 struct memb_join {
219  unsigned int proc_list_entries;
220  unsigned int failed_list_entries;
221  unsigned long long ring_seq;
222  unsigned char end_of_memb_join[0];
223 /*
224  * These parts of the data structure are dynamic:
225  * struct srp_addr proc_list[];
226  * struct srp_addr failed_list[];
227  */
228 } __attribute__((packed));
229 
230 
235 } __attribute__((packed));
236 
237 
241 } __attribute__((packed));
242 
243 
246  unsigned int aru;
247  unsigned int high_delivered;
248  unsigned int received_flg;
249 }__attribute__((packed));
250 
251 
254  unsigned int token_seq;
256  unsigned int retrans_flg;
259  unsigned char end_of_commit_token[0];
260 /*
261  * These parts of the data structure are dynamic:
262  *
263  * struct srp_addr addr[PROCESSOR_COUNT_MAX];
264  * struct memb_commit_token_memb_entry memb_list[PROCESSOR_COUNT_MAX];
265  */
266 }__attribute__((packed));
267 
268 struct message_item {
269  struct mcast *mcast;
270  unsigned int msg_len;
271 };
272 
274  struct mcast *mcast;
275  unsigned int msg_len;
276 };
277 
283 };
284 
287 
289 
290  /*
291  * Flow control mcasts and remcasts on last and current orf_token
292  */
294 
296 
298 
300 
302 
303  struct srp_addr my_id;
304 
306 
308 
310 
312 
314 
316 
318 
320 
322 
324 
326 
328 
330 
332 
334 
336 
338 
340 
342 
344 
345  unsigned int my_last_aru;
346 
348 
350 
351  unsigned int my_high_seq_received;
352 
353  unsigned int my_install_seq;
354 
356 
358 
360 
362 
364 
365  /*
366  * Queues used to order, deliver, and recover messages
367  */
369 
371 
373 
375 
377 
378  /*
379  * Received up to and including
380  */
381  unsigned int my_aru;
382 
383  unsigned int my_high_delivered;
384 
386 
388 
390 
392 
393  unsigned int my_token_seq;
394 
395  /*
396  * Timers
397  */
398  qb_loop_timer_handle timer_pause_timeout;
399 
400  qb_loop_timer_handle timer_orf_token_timeout;
401 
403 
405 
406  qb_loop_timer_handle timer_merge_detect_timeout;
407 
409 
411 
412  qb_loop_timer_handle memb_timer_state_commit_timeout;
413 
414  qb_loop_timer_handle timer_heartbeat_timeout;
415 
416  /*
417  * Function and data used to log messages
418  */
420 
422 
424 
426 
428 
430 
432 
434  int level,
435  int subsys,
436  const char *function,
437  const char *file,
438  int line,
439  const char *format, ...)__attribute__((format(printf, 6, 7)));;
440 
442 
443 //TODO struct srp_addr next_memb;
444 
446 
448 
450  unsigned int nodeid,
451  const void *msg,
452  unsigned int msg_len,
453  int endian_conversion_required);
454 
456  enum totem_configuration_type configuration_type,
457  const unsigned int *member_list, size_t member_list_entries,
458  const unsigned int *left_list, size_t left_list_entries,
459  const unsigned int *joined_list, size_t joined_list_entries,
460  const struct memb_ring_id *ring_id);
461 
463 
465  int waiting_trans_ack);
466 
468  struct memb_ring_id *memb_ring_id,
469  const struct totem_ip_address *addr);
470 
472  const struct memb_ring_id *memb_ring_id,
473  const struct totem_ip_address *addr);
474 
476 
478 
479  unsigned long long token_ring_id_seq;
480 
481  unsigned int last_released;
482 
483  unsigned int set_aru;
484 
486 
488 
490 
491  unsigned int my_last_seq;
492 
493  struct timeval tv_old;
494 
496 
498 
499  unsigned int use_heartbeat;
500 
501  unsigned int my_trc;
502 
503  unsigned int my_pbl;
504 
505  unsigned int my_cbl;
506 
507  uint64_t pause_timestamp;
508 
510 
512 
514 
516 
518 
520 
521  int flushing;
522 
525  char commit_token_storage[40000];
526 };
527 
529  int count;
530  int (*handler_functions[6]) (
531  struct totemsrp_instance *instance,
532  const void *msg,
533  size_t msg_len,
534  int endian_conversion_needed);
535 };
536 
555 };
556 
557 const char* gather_state_from_desc [] = {
558  [TOTEMSRP_GSFROM_CONSENSUS_TIMEOUT] = "consensus timeout",
559  [TOTEMSRP_GSFROM_GATHER_MISSING1] = "MISSING",
560  [TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_OPERATIONAL_STATE] = "The token was lost in the OPERATIONAL state.",
561  [TOTEMSRP_GSFROM_THE_CONSENSUS_TIMEOUT_EXPIRED] = "The consensus timeout expired.",
562  [TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_COMMIT_STATE] = "The token was lost in the COMMIT state.",
563  [TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_RECOVERY_STATE] = "The token was lost in the RECOVERY state.",
564  [TOTEMSRP_GSFROM_FAILED_TO_RECEIVE] = "failed to receive",
565  [TOTEMSRP_GSFROM_FOREIGN_MESSAGE_IN_OPERATIONAL_STATE] = "foreign message in operational state",
566  [TOTEMSRP_GSFROM_FOREIGN_MESSAGE_IN_GATHER_STATE] = "foreign message in gather state",
567  [TOTEMSRP_GSFROM_MERGE_DURING_OPERATIONAL_STATE] = "merge during operational state",
568  [TOTEMSRP_GSFROM_MERGE_DURING_GATHER_STATE] = "merge during gather state",
569  [TOTEMSRP_GSFROM_MERGE_DURING_JOIN] = "merge during join",
570  [TOTEMSRP_GSFROM_JOIN_DURING_OPERATIONAL_STATE] = "join during operational state",
571  [TOTEMSRP_GSFROM_JOIN_DURING_COMMIT_STATE] = "join during commit state",
572  [TOTEMSRP_GSFROM_JOIN_DURING_RECOVERY] = "join during recovery",
573  [TOTEMSRP_GSFROM_INTERFACE_CHANGE] = "interface change",
574 };
575 
576 /*
577  * forward decls
578  */
579 static int message_handler_orf_token (
580  struct totemsrp_instance *instance,
581  const void *msg,
582  size_t msg_len,
583  int endian_conversion_needed);
584 
585 static int message_handler_mcast (
586  struct totemsrp_instance *instance,
587  const void *msg,
588  size_t msg_len,
589  int endian_conversion_needed);
590 
591 static int message_handler_memb_merge_detect (
592  struct totemsrp_instance *instance,
593  const void *msg,
594  size_t msg_len,
595  int endian_conversion_needed);
596 
597 static int message_handler_memb_join (
598  struct totemsrp_instance *instance,
599  const void *msg,
600  size_t msg_len,
601  int endian_conversion_needed);
602 
603 static int message_handler_memb_commit_token (
604  struct totemsrp_instance *instance,
605  const void *msg,
606  size_t msg_len,
607  int endian_conversion_needed);
608 
609 static int message_handler_token_hold_cancel (
610  struct totemsrp_instance *instance,
611  const void *msg,
612  size_t msg_len,
613  int endian_conversion_needed);
614 
615 static void totemsrp_instance_initialize (struct totemsrp_instance *instance);
616 
617 static unsigned int main_msgs_missing (void);
618 
619 static void main_token_seqid_get (
620  const void *msg,
621  unsigned int *seqid,
622  unsigned int *token_is);
623 
624 static void srp_addr_copy (struct srp_addr *dest, const struct srp_addr *src);
625 
626 static void srp_addr_to_nodeid (
627  unsigned int *nodeid_out,
628  struct srp_addr *srp_addr_in,
629  unsigned int entries);
630 
631 static int srp_addr_equal (const struct srp_addr *a, const struct srp_addr *b);
632 
633 static void memb_leave_message_send (struct totemsrp_instance *instance);
634 
635 static void token_callbacks_execute (struct totemsrp_instance *instance, enum totem_callback_token_type type);
636 static void memb_state_gather_enter (struct totemsrp_instance *instance, enum gather_state_from gather_from);
637 static void messages_deliver_to_app (struct totemsrp_instance *instance, int skip, unsigned int end_point);
638 static int orf_token_mcast (struct totemsrp_instance *instance, struct orf_token *oken,
639  int fcc_mcasts_allowed);
640 static void messages_free (struct totemsrp_instance *instance, unsigned int token_aru);
641 
642 static void memb_ring_id_set (struct totemsrp_instance *instance,
643  const struct memb_ring_id *ring_id);
644 static void target_set_completed (void *context);
645 static void memb_state_commit_token_update (struct totemsrp_instance *instance);
646 static void memb_state_commit_token_target_set (struct totemsrp_instance *instance);
647 static int memb_state_commit_token_send (struct totemsrp_instance *instance);
648 static int memb_state_commit_token_send_recovery (struct totemsrp_instance *instance, struct memb_commit_token *memb_commit_token);
649 static void memb_state_commit_token_create (struct totemsrp_instance *instance);
650 static int token_hold_cancel_send (struct totemsrp_instance *instance);
651 static void orf_token_endian_convert (const struct orf_token *in, struct orf_token *out);
652 static void memb_commit_token_endian_convert (const struct memb_commit_token *in, struct memb_commit_token *out);
653 static void memb_join_endian_convert (const struct memb_join *in, struct memb_join *out);
654 static void mcast_endian_convert (const struct mcast *in, struct mcast *out);
655 static void memb_merge_detect_endian_convert (
656  const struct memb_merge_detect *in,
657  struct memb_merge_detect *out);
658 static void srp_addr_copy_endian_convert (struct srp_addr *out, const struct srp_addr *in);
659 static void timer_function_orf_token_timeout (void *data);
660 static void timer_function_pause_timeout (void *data);
661 static void timer_function_heartbeat_timeout (void *data);
662 static void timer_function_token_retransmit_timeout (void *data);
663 static void timer_function_token_hold_retransmit_timeout (void *data);
664 static void timer_function_merge_detect_timeout (void *data);
665 static void *totemsrp_buffer_alloc (struct totemsrp_instance *instance);
666 static void totemsrp_buffer_release (struct totemsrp_instance *instance, void *ptr);
667 static const char* gsfrom_to_msg(enum gather_state_from gsfrom);
668 
669 void main_deliver_fn (
670  void *context,
671  const void *msg,
672  unsigned int msg_len);
673 
675  void *context,
676  const struct totem_ip_address *iface_address,
677  unsigned int iface_no);
678 
680  6,
681  {
682  message_handler_orf_token, /* MESSAGE_TYPE_ORF_TOKEN */
683  message_handler_mcast, /* MESSAGE_TYPE_MCAST */
684  message_handler_memb_merge_detect, /* MESSAGE_TYPE_MEMB_MERGE_DETECT */
685  message_handler_memb_join, /* MESSAGE_TYPE_MEMB_JOIN */
686  message_handler_memb_commit_token, /* MESSAGE_TYPE_MEMB_COMMIT_TOKEN */
687  message_handler_token_hold_cancel /* MESSAGE_TYPE_TOKEN_HOLD_CANCEL */
688  }
689 };
690 
691 #define log_printf(level, format, args...) \
692 do { \
693  instance->totemsrp_log_printf ( \
694  level, instance->totemsrp_subsys_id, \
695  __FUNCTION__, __FILE__, __LINE__, \
696  format, ##args); \
697 } while (0);
698 #define LOGSYS_PERROR(err_num, level, fmt, args...) \
699 do { \
700  char _error_str[LOGSYS_MAX_PERROR_MSG_LEN]; \
701  const char *_error_ptr = qb_strerror_r(err_num, _error_str, sizeof(_error_str)); \
702  instance->totemsrp_log_printf ( \
703  level, instance->totemsrp_subsys_id, \
704  __FUNCTION__, __FILE__, __LINE__, \
705  fmt ": %s (%d)\n", ##args, _error_ptr, err_num); \
706  } while(0)
707 
708 static const char* gsfrom_to_msg(enum gather_state_from gsfrom)
709 {
710  if (gsfrom <= TOTEMSRP_GSFROM_MAX) {
711  return gather_state_from_desc[gsfrom];
712  }
713  else {
714  return "UNKNOWN";
715  }
716 }
717 
718 static void totemsrp_instance_initialize (struct totemsrp_instance *instance)
719 {
720  memset (instance, 0, sizeof (struct totemsrp_instance));
721 
722  list_init (&instance->token_callback_received_listhead);
723 
724  list_init (&instance->token_callback_sent_listhead);
725 
726  instance->my_received_flg = 1;
727 
728  instance->my_token_seq = SEQNO_START_TOKEN - 1;
729 
731 
732  instance->set_aru = -1;
733 
734  instance->my_aru = SEQNO_START_MSG;
735 
737 
739 
740  instance->orf_token_discard = 0;
741 
742  instance->originated_orf_token = 0;
743 
744  instance->commit_token = (struct memb_commit_token *)instance->commit_token_storage;
745 
746  instance->my_id.no_addrs = INTERFACE_MAX;
747 
748  instance->waiting_trans_ack = 1;
749 }
750 
751 static void main_token_seqid_get (
752  const void *msg,
753  unsigned int *seqid,
754  unsigned int *token_is)
755 {
756  const struct orf_token *token = msg;
757 
758  *seqid = 0;
759  *token_is = 0;
760  if (token->header.type == MESSAGE_TYPE_ORF_TOKEN) {
761  *seqid = token->token_seq;
762  *token_is = 1;
763  }
764 }
765 
766 static unsigned int main_msgs_missing (void)
767 {
768 // TODO
769  return (0);
770 }
771 
772 static int pause_flush (struct totemsrp_instance *instance)
773 {
774  uint64_t now_msec;
775  uint64_t timestamp_msec;
776  int res = 0;
777 
778  now_msec = (qb_util_nano_current_get () / QB_TIME_NS_IN_MSEC);
779  timestamp_msec = instance->pause_timestamp / QB_TIME_NS_IN_MSEC;
780 
781  if ((now_msec - timestamp_msec) > (instance->totem_config->token_timeout / 2)) {
783  "Process pause detected for %d ms, flushing membership messages.", (unsigned int)(now_msec - timestamp_msec));
784  /*
785  * -1 indicates an error from recvmsg
786  */
787  do {
789  } while (res == -1);
790  }
791  return (res);
792 }
793 
794 static int token_event_stats_collector (enum totem_callback_token_type type, const void *void_instance)
795 {
796  struct totemsrp_instance *instance = (struct totemsrp_instance *)void_instance;
797  uint32_t time_now;
798  unsigned long long nano_secs = qb_util_nano_current_get ();
799 
800  time_now = (nano_secs / QB_TIME_NS_IN_MSEC);
801 
802  if (type == TOTEM_CALLBACK_TOKEN_RECEIVED) {
803  /* incr latest token the index */
804  if (instance->stats.latest_token == (TOTEM_TOKEN_STATS_MAX - 1))
805  instance->stats.latest_token = 0;
806  else
807  instance->stats.latest_token++;
808 
809  if (instance->stats.earliest_token == instance->stats.latest_token) {
810  /* we have filled up the array, start overwriting */
811  if (instance->stats.earliest_token == (TOTEM_TOKEN_STATS_MAX - 1))
812  instance->stats.earliest_token = 0;
813  else
814  instance->stats.earliest_token++;
815 
816  instance->stats.token[instance->stats.earliest_token].rx = 0;
817  instance->stats.token[instance->stats.earliest_token].tx = 0;
818  instance->stats.token[instance->stats.earliest_token].backlog_calc = 0;
819  }
820 
821  instance->stats.token[instance->stats.latest_token].rx = time_now;
822  instance->stats.token[instance->stats.latest_token].tx = 0; /* in case we drop the token */
823  } else {
824  instance->stats.token[instance->stats.latest_token].tx = time_now;
825  }
826  return 0;
827 }
828 
829 /*
830  * Exported interfaces
831  */
833  qb_loop_t *poll_handle,
834  void **srp_context,
835  struct totem_config *totem_config,
837 
838  void (*deliver_fn) (
839  unsigned int nodeid,
840  const void *msg,
841  unsigned int msg_len,
842  int endian_conversion_required),
843 
844  void (*confchg_fn) (
845  enum totem_configuration_type configuration_type,
846  const unsigned int *member_list, size_t member_list_entries,
847  const unsigned int *left_list, size_t left_list_entries,
848  const unsigned int *joined_list, size_t joined_list_entries,
849  const struct memb_ring_id *ring_id),
850  void (*waiting_trans_ack_cb_fn) (
851  int waiting_trans_ack))
852 {
853  struct totemsrp_instance *instance;
854  int res;
855 
856  instance = malloc (sizeof (struct totemsrp_instance));
857  if (instance == NULL) {
858  goto error_exit;
859  }
860 
861  totemsrp_instance_initialize (instance);
862 
863  instance->totemsrp_waiting_trans_ack_cb_fn = waiting_trans_ack_cb_fn;
864  instance->totemsrp_waiting_trans_ack_cb_fn (1);
865 
866  stats->srp = &instance->stats;
867  instance->stats.latest_token = 0;
868  instance->stats.earliest_token = 0;
869 
870  instance->totem_config = totem_config;
871 
872  /*
873  * Configure logging
874  */
883 
884  /*
885  * Configure totem store and load functions
886  */
888  instance->memb_ring_id_store = totem_config->totem_memb_ring_id_store;
889 
890  /*
891  * Initialize local variables for totemsrp
892  */
893  totemip_copy (&instance->mcast_address, &totem_config->interfaces[0].mcast_addr);
894 
895  /*
896  * Display totem configuration
897  */
899  "Token Timeout (%d ms) retransmit timeout (%d ms)",
900  totem_config->token_timeout, totem_config->token_retransmit_timeout);
902  "token hold (%d ms) retransmits before loss (%d retrans)",
903  totem_config->token_hold_timeout, totem_config->token_retransmits_before_loss_const);
905  "join (%d ms) send_join (%d ms) consensus (%d ms) merge (%d ms)",
906  totem_config->join_timeout,
907  totem_config->send_join_timeout,
908  totem_config->consensus_timeout,
909 
910  totem_config->merge_timeout);
912  "downcheck (%d ms) fail to recv const (%d msgs)",
913  totem_config->downcheck_timeout, totem_config->fail_to_recv_const);
915  "seqno unchanged const (%d rotations) Maximum network MTU %d", totem_config->seqno_unchanged_const, totem_config->net_mtu);
916 
918  "window size per rotation (%d messages) maximum messages per rotation (%d messages)",
919  totem_config->window_size, totem_config->max_messages);
920 
922  "missed count const (%d messages)",
923  totem_config->miss_count_const);
924 
926  "send threads (%d threads)", totem_config->threads);
928  "RRP token expired timeout (%d ms)",
929  totem_config->rrp_token_expired_timeout);
931  "RRP token problem counter (%d ms)",
932  totem_config->rrp_problem_count_timeout);
934  "RRP threshold (%d problem count)",
935  totem_config->rrp_problem_count_threshold);
937  "RRP multicast threshold (%d problem count)",
938  totem_config->rrp_problem_count_mcast_threshold);
940  "RRP automatic recovery check timeout (%d ms)",
941  totem_config->rrp_autorecovery_check_timeout);
943  "RRP mode set to %s.", instance->totem_config->rrp_mode);
944 
946  "heartbeat_failures_allowed (%d)", totem_config->heartbeat_failures_allowed);
948  "max_network_delay (%d ms)", totem_config->max_network_delay);
949 
950 
951  cs_queue_init (&instance->retrans_message_queue, RETRANS_MESSAGE_QUEUE_SIZE_MAX,
952  sizeof (struct message_item), instance->threaded_mode_enabled);
953 
954  sq_init (&instance->regular_sort_queue,
955  QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0);
956 
957  sq_init (&instance->recovery_sort_queue,
958  QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0);
959 
960  instance->totemsrp_poll_handle = poll_handle;
961 
962  instance->totemsrp_deliver_fn = deliver_fn;
963 
964  instance->totemsrp_confchg_fn = confchg_fn;
965  instance->use_heartbeat = 1;
966 
967  timer_function_pause_timeout (instance);
968 
969  if ( totem_config->heartbeat_failures_allowed == 0 ) {
971  "HeartBeat is Disabled. To enable set heartbeat_failures_allowed > 0");
972  instance->use_heartbeat = 0;
973  }
974 
975  if (instance->use_heartbeat) {
976  instance->heartbeat_timeout
977  = (totem_config->heartbeat_failures_allowed) * totem_config->token_retransmit_timeout
978  + totem_config->max_network_delay;
979 
980  if (instance->heartbeat_timeout >= totem_config->token_timeout) {
982  "total heartbeat_timeout (%d ms) is not less than token timeout (%d ms)",
983  instance->heartbeat_timeout,
984  totem_config->token_timeout);
986  "heartbeat_timeout = heartbeat_failures_allowed * token_retransmit_timeout + max_network_delay");
988  "heartbeat timeout should be less than the token timeout. Heartbeat is disabled!!");
989  instance->use_heartbeat = 0;
990  }
991  else {
993  "total heartbeat_timeout (%d ms)", instance->heartbeat_timeout);
994  }
995  }
996 
997  res = totemrrp_initialize (
998  poll_handle,
999  &instance->totemrrp_context,
1000  totem_config,
1001  stats->srp,
1002  instance,
1005  main_token_seqid_get,
1006  main_msgs_missing,
1007  target_set_completed);
1008  if (res == -1) {
1009  goto error_exit;
1010  }
1011 
1012  /*
1013  * Must have net_mtu adjusted by totemrrp_initialize first
1014  */
1015  cs_queue_init (&instance->new_message_queue,
1017  sizeof (struct message_item), instance->threaded_mode_enabled);
1018 
1019  cs_queue_init (&instance->new_message_queue_trans,
1021  sizeof (struct message_item), instance->threaded_mode_enabled);
1022 
1024  &instance->token_recv_event_handle,
1026  0,
1027  token_event_stats_collector,
1028  instance);
1030  &instance->token_sent_event_handle,
1032  0,
1033  token_event_stats_collector,
1034  instance);
1035  *srp_context = instance;
1036  return (0);
1037 
1038 error_exit:
1039  return (-1);
1040 }
1041 
1043  void *srp_context)
1044 {
1045  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1046 
1047 
1048  memb_leave_message_send (instance);
1049  totemrrp_finalize (instance->totemrrp_context);
1050  cs_queue_free (&instance->new_message_queue);
1051  cs_queue_free (&instance->new_message_queue_trans);
1052  cs_queue_free (&instance->retrans_message_queue);
1053  sq_free (&instance->regular_sort_queue);
1054  sq_free (&instance->recovery_sort_queue);
1055  free (instance);
1056 }
1057 
1058 /*
1059  * Return configured interfaces. interfaces is array of totem_ip addresses allocated by caller,
1060  * with interaces_size number of items. iface_count is final number of interfaces filled by this
1061  * function.
1062  *
1063  * Function returns 0 on success, otherwise if interfaces array is not big enough, -2 is returned,
1064  * and if interface was not found, -1 is returned.
1065  */
1067  void *srp_context,
1068  unsigned int nodeid,
1069  struct totem_ip_address *interfaces,
1070  unsigned int interfaces_size,
1071  char ***status,
1072  unsigned int *iface_count)
1073 {
1074  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1075  int res = 0;
1076  unsigned int found = 0;
1077  unsigned int i;
1078 
1079  for (i = 0; i < instance->my_memb_entries; i++) {
1080  if (instance->my_memb_list[i].addr[0].nodeid == nodeid) {
1081  found = 1;
1082  break;
1083  }
1084  }
1085 
1086  if (found) {
1087  *iface_count = instance->totem_config->interface_count;
1088 
1089  if (interfaces_size >= *iface_count) {
1090  memcpy (interfaces, instance->my_memb_list[i].addr,
1091  sizeof (struct totem_ip_address) * *iface_count);
1092  } else {
1093  res = -2;
1094  }
1095 
1096  goto finish;
1097  }
1098 
1099  for (i = 0; i < instance->my_left_memb_entries; i++) {
1100  if (instance->my_left_memb_list[i].addr[0].nodeid == nodeid) {
1101  found = 1;
1102  break;
1103  }
1104  }
1105 
1106  if (found) {
1107  *iface_count = instance->totem_config->interface_count;
1108 
1109  if (interfaces_size >= *iface_count) {
1110  memcpy (interfaces, instance->my_left_memb_list[i].addr,
1111  sizeof (struct totem_ip_address) * *iface_count);
1112  } else {
1113  res = -2;
1114  }
1115  } else {
1116  res = -1;
1117  }
1118 
1119 finish:
1120  totemrrp_ifaces_get (instance->totemrrp_context, status, NULL);
1121  return (res);
1122 }
1123 
1125  void *srp_context,
1126  const char *cipher_type,
1127  const char *hash_type)
1128 {
1129  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1130  int res;
1131 
1132  res = totemrrp_crypto_set(instance->totemrrp_context, cipher_type, hash_type);
1133 
1134  return (res);
1135 }
1136 
1137 
1139  void *srp_context)
1140 {
1141  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1142  unsigned int res;
1143 
1144  res = instance->totem_config->interfaces[0].boundto.nodeid;
1145 
1146  return (res);
1147 }
1148 
1150  void *srp_context)
1151 {
1152  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1153  int res;
1154 
1155  res = instance->totem_config->interfaces[0].boundto.family;
1156 
1157  return (res);
1158 }
1159 
1160 
1162  void *srp_context)
1163 {
1164  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1165 
1167  instance->totem_config->interface_count);
1168 
1169  return (0);
1170 }
1171 
1172 
1173 /*
1174  * Set operations for use by the membership algorithm
1175  */
1176 static int srp_addr_equal (const struct srp_addr *a, const struct srp_addr *b)
1177 {
1178  unsigned int i;
1179  unsigned int res;
1180 
1181  for (i = 0; i < 1; i++) {
1182  res = totemip_equal (&a->addr[i], &b->addr[i]);
1183  if (res == 0) {
1184  return (0);
1185  }
1186  }
1187  return (1);
1188 }
1189 
1190 static void srp_addr_copy (struct srp_addr *dest, const struct srp_addr *src)
1191 {
1192  unsigned int i;
1193 
1194  dest->no_addrs = src->no_addrs;
1195 
1196  for (i = 0; i < INTERFACE_MAX; i++) {
1197  totemip_copy (&dest->addr[i], &src->addr[i]);
1198  }
1199 }
1200 
1201 static void srp_addr_to_nodeid (
1202  unsigned int *nodeid_out,
1203  struct srp_addr *srp_addr_in,
1204  unsigned int entries)
1205 {
1206  unsigned int i;
1207 
1208  for (i = 0; i < entries; i++) {
1209  nodeid_out[i] = srp_addr_in[i].addr[0].nodeid;
1210  }
1211 }
1212 
1213 static void srp_addr_copy_endian_convert (struct srp_addr *out, const struct srp_addr *in)
1214 {
1215  int i;
1216 
1217  for (i = 0; i < INTERFACE_MAX; i++) {
1218  totemip_copy_endian_convert (&out->addr[i], &in->addr[i]);
1219  }
1220 }
1221 
1222 static void memb_consensus_reset (struct totemsrp_instance *instance)
1223 {
1224  instance->consensus_list_entries = 0;
1225 }
1226 
1227 static void memb_set_subtract (
1228  struct srp_addr *out_list, int *out_list_entries,
1229  struct srp_addr *one_list, int one_list_entries,
1230  struct srp_addr *two_list, int two_list_entries)
1231 {
1232  int found = 0;
1233  int i;
1234  int j;
1235 
1236  *out_list_entries = 0;
1237 
1238  for (i = 0; i < one_list_entries; i++) {
1239  for (j = 0; j < two_list_entries; j++) {
1240  if (srp_addr_equal (&one_list[i], &two_list[j])) {
1241  found = 1;
1242  break;
1243  }
1244  }
1245  if (found == 0) {
1246  srp_addr_copy (&out_list[*out_list_entries], &one_list[i]);
1247  *out_list_entries = *out_list_entries + 1;
1248  }
1249  found = 0;
1250  }
1251 }
1252 
1253 /*
1254  * Set consensus for a specific processor
1255  */
1256 static void memb_consensus_set (
1257  struct totemsrp_instance *instance,
1258  const struct srp_addr *addr)
1259 {
1260  int found = 0;
1261  int i;
1262 
1263  if (addr->addr[0].nodeid == LEAVE_DUMMY_NODEID)
1264  return;
1265 
1266  for (i = 0; i < instance->consensus_list_entries; i++) {
1267  if (srp_addr_equal(addr, &instance->consensus_list[i].addr)) {
1268  found = 1;
1269  break; /* found entry */
1270  }
1271  }
1272  srp_addr_copy (&instance->consensus_list[i].addr, addr);
1273  instance->consensus_list[i].set = 1;
1274  if (found == 0) {
1275  instance->consensus_list_entries++;
1276  }
1277  return;
1278 }
1279 
1280 /*
1281  * Is consensus set for a specific processor
1282  */
1283 static int memb_consensus_isset (
1284  struct totemsrp_instance *instance,
1285  const struct srp_addr *addr)
1286 {
1287  int i;
1288 
1289  for (i = 0; i < instance->consensus_list_entries; i++) {
1290  if (srp_addr_equal (addr, &instance->consensus_list[i].addr)) {
1291  return (instance->consensus_list[i].set);
1292  }
1293  }
1294  return (0);
1295 }
1296 
1297 /*
1298  * Is consensus agreed upon based upon consensus database
1299  */
1300 static int memb_consensus_agreed (
1301  struct totemsrp_instance *instance)
1302 {
1303  struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
1304  int token_memb_entries = 0;
1305  int agreed = 1;
1306  int i;
1307 
1308  memb_set_subtract (token_memb, &token_memb_entries,
1309  instance->my_proc_list, instance->my_proc_list_entries,
1310  instance->my_failed_list, instance->my_failed_list_entries);
1311 
1312  for (i = 0; i < token_memb_entries; i++) {
1313  if (memb_consensus_isset (instance, &token_memb[i]) == 0) {
1314  agreed = 0;
1315  break;
1316  }
1317  }
1318 
1319  if (agreed && instance->failed_to_recv == 1) {
1320  /*
1321  * Both nodes agreed on our failure. We don't care how many proc list items left because we
1322  * will create single ring anyway.
1323  */
1324 
1325  return (agreed);
1326  }
1327 
1328  assert (token_memb_entries >= 1);
1329 
1330  return (agreed);
1331 }
1332 
1333 static void memb_consensus_notset (
1334  struct totemsrp_instance *instance,
1335  struct srp_addr *no_consensus_list,
1336  int *no_consensus_list_entries,
1337  struct srp_addr *comparison_list,
1338  int comparison_list_entries)
1339 {
1340  int i;
1341 
1342  *no_consensus_list_entries = 0;
1343 
1344  for (i = 0; i < instance->my_proc_list_entries; i++) {
1345  if (memb_consensus_isset (instance, &instance->my_proc_list[i]) == 0) {
1346  srp_addr_copy (&no_consensus_list[*no_consensus_list_entries], &instance->my_proc_list[i]);
1347  *no_consensus_list_entries = *no_consensus_list_entries + 1;
1348  }
1349  }
1350 }
1351 
1352 /*
1353  * Is set1 equal to set2 Entries can be in different orders
1354  */
1355 static int memb_set_equal (
1356  struct srp_addr *set1, int set1_entries,
1357  struct srp_addr *set2, int set2_entries)
1358 {
1359  int i;
1360  int j;
1361 
1362  int found = 0;
1363 
1364  if (set1_entries != set2_entries) {
1365  return (0);
1366  }
1367  for (i = 0; i < set2_entries; i++) {
1368  for (j = 0; j < set1_entries; j++) {
1369  if (srp_addr_equal (&set1[j], &set2[i])) {
1370  found = 1;
1371  break;
1372  }
1373  }
1374  if (found == 0) {
1375  return (0);
1376  }
1377  found = 0;
1378  }
1379  return (1);
1380 }
1381 
1382 /*
1383  * Is subset fully contained in fullset
1384  */
1385 static int memb_set_subset (
1386  const struct srp_addr *subset, int subset_entries,
1387  const struct srp_addr *fullset, int fullset_entries)
1388 {
1389  int i;
1390  int j;
1391  int found = 0;
1392 
1393  if (subset_entries > fullset_entries) {
1394  return (0);
1395  }
1396  for (i = 0; i < subset_entries; i++) {
1397  for (j = 0; j < fullset_entries; j++) {
1398  if (srp_addr_equal (&subset[i], &fullset[j])) {
1399  found = 1;
1400  }
1401  }
1402  if (found == 0) {
1403  return (0);
1404  }
1405  found = 0;
1406  }
1407  return (1);
1408 }
1409 /*
1410  * merge subset into fullset taking care not to add duplicates
1411  */
1412 static void memb_set_merge (
1413  const struct srp_addr *subset, int subset_entries,
1414  struct srp_addr *fullset, int *fullset_entries)
1415 {
1416  int found = 0;
1417  int i;
1418  int j;
1419 
1420  for (i = 0; i < subset_entries; i++) {
1421  for (j = 0; j < *fullset_entries; j++) {
1422  if (srp_addr_equal (&fullset[j], &subset[i])) {
1423  found = 1;
1424  break;
1425  }
1426  }
1427  if (found == 0) {
1428  srp_addr_copy (&fullset[*fullset_entries], &subset[i]);
1429  *fullset_entries = *fullset_entries + 1;
1430  }
1431  found = 0;
1432  }
1433  return;
1434 }
1435 
1436 static void memb_set_and_with_ring_id (
1437  struct srp_addr *set1,
1438  struct memb_ring_id *set1_ring_ids,
1439  int set1_entries,
1440  struct srp_addr *set2,
1441  int set2_entries,
1442  struct memb_ring_id *old_ring_id,
1443  struct srp_addr *and,
1444  int *and_entries)
1445 {
1446  int i;
1447  int j;
1448  int found = 0;
1449 
1450  *and_entries = 0;
1451 
1452  for (i = 0; i < set2_entries; i++) {
1453  for (j = 0; j < set1_entries; j++) {
1454  if (srp_addr_equal (&set1[j], &set2[i])) {
1455  if (memcmp (&set1_ring_ids[j], old_ring_id, sizeof (struct memb_ring_id)) == 0) {
1456  found = 1;
1457  }
1458  break;
1459  }
1460  }
1461  if (found) {
1462  srp_addr_copy (&and[*and_entries], &set1[j]);
1463  *and_entries = *and_entries + 1;
1464  }
1465  found = 0;
1466  }
1467  return;
1468 }
1469 
1470 #ifdef CODE_COVERAGE
1471 static void memb_set_print (
1472  char *string,
1473  struct srp_addr *list,
1474  int list_entries)
1475 {
1476  int i;
1477  int j;
1478  printf ("List '%s' contains %d entries:\n", string, list_entries);
1479 
1480  for (i = 0; i < list_entries; i++) {
1481  printf ("Address %d with %d rings\n", i, list[i].no_addrs);
1482  for (j = 0; j < list[i].no_addrs; j++) {
1483  printf ("\tiface %d %s\n", j, totemip_print (&list[i].addr[j]));
1484  printf ("\tfamily %d\n", list[i].addr[j].family);
1485  }
1486  }
1487 }
1488 #endif
1489 static void my_leave_memb_clear(
1490  struct totemsrp_instance *instance)
1491 {
1492  memset(instance->my_leave_memb_list, 0, sizeof(instance->my_leave_memb_list));
1493  instance->my_leave_memb_entries = 0;
1494 }
1495 
1496 static unsigned int my_leave_memb_match(
1497  struct totemsrp_instance *instance,
1498  unsigned int nodeid)
1499 {
1500  int i;
1501  unsigned int ret = 0;
1502 
1503  for (i = 0; i < instance->my_leave_memb_entries; i++){
1504  if (instance->my_leave_memb_list[i] == nodeid){
1505  ret = nodeid;
1506  break;
1507  }
1508  }
1509  return ret;
1510 }
1511 
1512 static void my_leave_memb_set(
1513  struct totemsrp_instance *instance,
1514  unsigned int nodeid)
1515 {
1516  int i, found = 0;
1517  for (i = 0; i < instance->my_leave_memb_entries; i++){
1518  if (instance->my_leave_memb_list[i] == nodeid){
1519  found = 1;
1520  break;
1521  }
1522  }
1523  if (found == 1) {
1524  return;
1525  }
1526  if (instance->my_leave_memb_entries < (PROCESSOR_COUNT_MAX - 1)) {
1527  instance->my_leave_memb_list[instance->my_leave_memb_entries] = nodeid;
1528  instance->my_leave_memb_entries++;
1529  } else {
1531  "Cannot set LEAVE nodeid=%d", nodeid);
1532  }
1533 }
1534 
1535 
1536 static void *totemsrp_buffer_alloc (struct totemsrp_instance *instance)
1537 {
1538  assert (instance != NULL);
1539  return totemrrp_buffer_alloc (instance->totemrrp_context);
1540 }
1541 
1542 static void totemsrp_buffer_release (struct totemsrp_instance *instance, void *ptr)
1543 {
1544  assert (instance != NULL);
1545  totemrrp_buffer_release (instance->totemrrp_context, ptr);
1546 }
1547 
1548 static void reset_token_retransmit_timeout (struct totemsrp_instance *instance)
1549 {
1550  int32_t res;
1551 
1552  qb_loop_timer_del (instance->totemsrp_poll_handle,
1554  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1555  QB_LOOP_MED,
1556  instance->totem_config->token_retransmit_timeout*QB_TIME_NS_IN_MSEC,
1557  (void *)instance,
1558  timer_function_token_retransmit_timeout,
1560  if (res != 0) {
1561  log_printf(instance->totemsrp_log_level_error, "reset_token_retransmit_timeout - qb_loop_timer_add error : %d", res);
1562  }
1563 
1564 }
1565 
1566 static void start_merge_detect_timeout (struct totemsrp_instance *instance)
1567 {
1568  int32_t res;
1569 
1570  if (instance->my_merge_detect_timeout_outstanding == 0) {
1571  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1572  QB_LOOP_MED,
1573  instance->totem_config->merge_timeout*QB_TIME_NS_IN_MSEC,
1574  (void *)instance,
1575  timer_function_merge_detect_timeout,
1576  &instance->timer_merge_detect_timeout);
1577  if (res != 0) {
1578  log_printf(instance->totemsrp_log_level_error, "start_merge_detect_timeout - qb_loop_timer_add error : %d", res);
1579  }
1580 
1582  }
1583 }
1584 
1585 static void cancel_merge_detect_timeout (struct totemsrp_instance *instance)
1586 {
1587  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_merge_detect_timeout);
1589 }
1590 
1591 /*
1592  * ring_state_* is used to save and restore the sort queue
1593  * state when a recovery operation fails (and enters gather)
1594  */
1595 static void old_ring_state_save (struct totemsrp_instance *instance)
1596 {
1597  if (instance->old_ring_state_saved == 0) {
1598  instance->old_ring_state_saved = 1;
1599  memcpy (&instance->my_old_ring_id, &instance->my_ring_id,
1600  sizeof (struct memb_ring_id));
1601  instance->old_ring_state_aru = instance->my_aru;
1604  "Saving state aru %x high seq received %x",
1605  instance->my_aru, instance->my_high_seq_received);
1606  }
1607 }
1608 
1609 static void old_ring_state_restore (struct totemsrp_instance *instance)
1610 {
1611  instance->my_aru = instance->old_ring_state_aru;
1614  "Restoring instance->my_aru %x my high seq received %x",
1615  instance->my_aru, instance->my_high_seq_received);
1616 }
1617 
1618 static void old_ring_state_reset (struct totemsrp_instance *instance)
1619 {
1621  "Resetting old ring state");
1622  instance->old_ring_state_saved = 0;
1623 }
1624 
1625 static void reset_pause_timeout (struct totemsrp_instance *instance)
1626 {
1627  int32_t res;
1628 
1629  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_pause_timeout);
1630  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1631  QB_LOOP_MED,
1632  instance->totem_config->token_timeout * QB_TIME_NS_IN_MSEC / 5,
1633  (void *)instance,
1634  timer_function_pause_timeout,
1635  &instance->timer_pause_timeout);
1636  if (res != 0) {
1637  log_printf(instance->totemsrp_log_level_error, "reset_pause_timeout - qb_loop_timer_add error : %d", res);
1638  }
1639 }
1640 
1641 static void reset_token_timeout (struct totemsrp_instance *instance) {
1642  int32_t res;
1643 
1644  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout);
1645  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1646  QB_LOOP_MED,
1647  instance->totem_config->token_timeout*QB_TIME_NS_IN_MSEC,
1648  (void *)instance,
1649  timer_function_orf_token_timeout,
1650  &instance->timer_orf_token_timeout);
1651  if (res != 0) {
1652  log_printf(instance->totemsrp_log_level_error, "reset_token_timeout - qb_loop_timer_add error : %d", res);
1653  }
1654 }
1655 
1656 static void reset_heartbeat_timeout (struct totemsrp_instance *instance) {
1657  int32_t res;
1658 
1659  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout);
1660  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1661  QB_LOOP_MED,
1662  instance->heartbeat_timeout*QB_TIME_NS_IN_MSEC,
1663  (void *)instance,
1664  timer_function_heartbeat_timeout,
1665  &instance->timer_heartbeat_timeout);
1666  if (res != 0) {
1667  log_printf(instance->totemsrp_log_level_error, "reset_heartbeat_timeout - qb_loop_timer_add error : %d", res);
1668  }
1669 }
1670 
1671 
1672 static void cancel_token_timeout (struct totemsrp_instance *instance) {
1673  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout);
1674 }
1675 
1676 static void cancel_heartbeat_timeout (struct totemsrp_instance *instance) {
1677  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout);
1678 }
1679 
1680 static void cancel_token_retransmit_timeout (struct totemsrp_instance *instance)
1681 {
1682  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_retransmit_timeout);
1683 }
1684 
1685 static void start_token_hold_retransmit_timeout (struct totemsrp_instance *instance)
1686 {
1687  int32_t res;
1688 
1689  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1690  QB_LOOP_MED,
1691  instance->totem_config->token_hold_timeout*QB_TIME_NS_IN_MSEC,
1692  (void *)instance,
1693  timer_function_token_hold_retransmit_timeout,
1695  if (res != 0) {
1696  log_printf(instance->totemsrp_log_level_error, "start_token_hold_retransmit_timeout - qb_loop_timer_add error : %d", res);
1697  }
1698 }
1699 
1700 static void cancel_token_hold_retransmit_timeout (struct totemsrp_instance *instance)
1701 {
1702  qb_loop_timer_del (instance->totemsrp_poll_handle,
1704 }
1705 
1706 static void memb_state_consensus_timeout_expired (
1707  struct totemsrp_instance *instance)
1708 {
1709  struct srp_addr no_consensus_list[PROCESSOR_COUNT_MAX];
1710  int no_consensus_list_entries;
1711 
1712  instance->stats.consensus_timeouts++;
1713  if (memb_consensus_agreed (instance)) {
1714  memb_consensus_reset (instance);
1715 
1716  memb_consensus_set (instance, &instance->my_id);
1717 
1718  reset_token_timeout (instance); // REVIEWED
1719  } else {
1720  memb_consensus_notset (
1721  instance,
1722  no_consensus_list,
1723  &no_consensus_list_entries,
1724  instance->my_proc_list,
1725  instance->my_proc_list_entries);
1726 
1727  memb_set_merge (no_consensus_list, no_consensus_list_entries,
1728  instance->my_failed_list, &instance->my_failed_list_entries);
1729  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_CONSENSUS_TIMEOUT);
1730  }
1731 }
1732 
1733 static void memb_join_message_send (struct totemsrp_instance *instance);
1734 
1735 static void memb_merge_detect_transmit (struct totemsrp_instance *instance);
1736 
1737 /*
1738  * Timers used for various states of the membership algorithm
1739  */
1740 static void timer_function_pause_timeout (void *data)
1741 {
1742  struct totemsrp_instance *instance = data;
1743 
1744  instance->pause_timestamp = qb_util_nano_current_get ();
1745  reset_pause_timeout (instance);
1746 }
1747 
1748 static void memb_recovery_state_token_loss (struct totemsrp_instance *instance)
1749 {
1750  old_ring_state_restore (instance);
1751  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_RECOVERY_STATE);
1752  instance->stats.recovery_token_lost++;
1753 }
1754 
1755 static void timer_function_orf_token_timeout (void *data)
1756 {
1757  struct totemsrp_instance *instance = data;
1758 
1759  switch (instance->memb_state) {
1762  "The token was lost in the OPERATIONAL state.");
1764  "A processor failed, forming new configuration:"
1765  " token timed out (%ums), waiting %ums for consensus.",
1766  instance->totem_config->token_timeout,
1767  instance->totem_config->consensus_timeout);
1769  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_OPERATIONAL_STATE);
1770  instance->stats.operational_token_lost++;
1771  break;
1772 
1773  case MEMB_STATE_GATHER:
1775  "The consensus timeout expired (%ums).",
1776  instance->totem_config->consensus_timeout);
1777  memb_state_consensus_timeout_expired (instance);
1778  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_CONSENSUS_TIMEOUT_EXPIRED);
1779  instance->stats.gather_token_lost++;
1780  break;
1781 
1782  case MEMB_STATE_COMMIT:
1784  "The token was lost in the COMMIT state.");
1785  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_COMMIT_STATE);
1786  instance->stats.commit_token_lost++;
1787  break;
1788 
1789  case MEMB_STATE_RECOVERY:
1791  "The token was lost in the RECOVERY state.");
1792  memb_recovery_state_token_loss (instance);
1793  instance->orf_token_discard = 1;
1794  break;
1795  }
1796 }
1797 
1798 static void timer_function_heartbeat_timeout (void *data)
1799 {
1800  struct totemsrp_instance *instance = data;
1802  "HeartBeat Timer expired Invoking token loss mechanism in state %d ", instance->memb_state);
1803  timer_function_orf_token_timeout(data);
1804 }
1805 
1806 static void memb_timer_function_state_gather (void *data)
1807 {
1808  struct totemsrp_instance *instance = data;
1809  int32_t res;
1810 
1811  switch (instance->memb_state) {
1813  case MEMB_STATE_RECOVERY:
1814  assert (0); /* this should never happen */
1815  break;
1816  case MEMB_STATE_GATHER:
1817  case MEMB_STATE_COMMIT:
1818  memb_join_message_send (instance);
1819 
1820  /*
1821  * Restart the join timeout
1822  `*/
1823  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);
1824 
1825  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
1826  QB_LOOP_MED,
1827  instance->totem_config->join_timeout*QB_TIME_NS_IN_MSEC,
1828  (void *)instance,
1829  memb_timer_function_state_gather,
1831 
1832  if (res != 0) {
1833  log_printf(instance->totemsrp_log_level_error, "memb_timer_function_state_gather - qb_loop_timer_add error : %d", res);
1834  }
1835  break;
1836  }
1837 }
1838 
1839 static void memb_timer_function_gather_consensus_timeout (void *data)
1840 {
1841  struct totemsrp_instance *instance = data;
1842  memb_state_consensus_timeout_expired (instance);
1843 }
1844 
1845 static void deliver_messages_from_recovery_to_regular (struct totemsrp_instance *instance)
1846 {
1847  unsigned int i;
1848  struct sort_queue_item *recovery_message_item;
1849  struct sort_queue_item regular_message_item;
1850  unsigned int range = 0;
1851  int res;
1852  void *ptr;
1853  struct mcast *mcast;
1854 
1856  "recovery to regular %x-%x", SEQNO_START_MSG + 1, instance->my_aru);
1857 
1858  range = instance->my_aru - SEQNO_START_MSG;
1859  /*
1860  * Move messages from recovery to regular sort queue
1861  */
1862 // todo should i be initialized to 0 or 1 ?
1863  for (i = 1; i <= range; i++) {
1864  res = sq_item_get (&instance->recovery_sort_queue,
1865  i + SEQNO_START_MSG, &ptr);
1866  if (res != 0) {
1867  continue;
1868  }
1869  recovery_message_item = ptr;
1870 
1871  /*
1872  * Convert recovery message into regular message
1873  */
1874  mcast = recovery_message_item->mcast;
1875  if (mcast->header.encapsulated == MESSAGE_ENCAPSULATED) {
1876  /*
1877  * Message is a recovery message encapsulated
1878  * in a new ring message
1879  */
1880  regular_message_item.mcast =
1881  (struct mcast *)(((char *)recovery_message_item->mcast) + sizeof (struct mcast));
1882  regular_message_item.msg_len =
1883  recovery_message_item->msg_len - sizeof (struct mcast);
1884  mcast = regular_message_item.mcast;
1885  } else {
1886  /*
1887  * TODO this case shouldn't happen
1888  */
1889  continue;
1890  }
1891 
1893  "comparing if ring id is for this processors old ring seqno %d",
1894  mcast->seq);
1895 
1896  /*
1897  * Only add this message to the regular sort
1898  * queue if it was originated with the same ring
1899  * id as the previous ring
1900  */
1901  if (memcmp (&instance->my_old_ring_id, &mcast->ring_id,
1902  sizeof (struct memb_ring_id)) == 0) {
1903 
1904  res = sq_item_inuse (&instance->regular_sort_queue, mcast->seq);
1905  if (res == 0) {
1906  sq_item_add (&instance->regular_sort_queue,
1907  &regular_message_item, mcast->seq);
1908  if (sq_lt_compare (instance->old_ring_state_high_seq_received, mcast->seq)) {
1909  instance->old_ring_state_high_seq_received = mcast->seq;
1910  }
1911  }
1912  } else {
1914  "-not adding msg with seq no %x", mcast->seq);
1915  }
1916  }
1917 }
1918 
1919 /*
1920  * Change states in the state machine of the membership algorithm
1921  */
1922 static void memb_state_operational_enter (struct totemsrp_instance *instance)
1923 {
1924  struct srp_addr joined_list[PROCESSOR_COUNT_MAX];
1925  int joined_list_entries = 0;
1926  unsigned int aru_save;
1927  unsigned int joined_list_totemip[PROCESSOR_COUNT_MAX];
1928  unsigned int trans_memb_list_totemip[PROCESSOR_COUNT_MAX];
1929  unsigned int new_memb_list_totemip[PROCESSOR_COUNT_MAX];
1930  unsigned int left_list[PROCESSOR_COUNT_MAX];
1931  unsigned int i;
1932  unsigned int res;
1933  char left_node_msg[1024];
1934  char joined_node_msg[1024];
1935  char failed_node_msg[1024];
1936 
1937  instance->originated_orf_token = 0;
1938 
1939  memb_consensus_reset (instance);
1940 
1941  old_ring_state_reset (instance);
1942 
1943  deliver_messages_from_recovery_to_regular (instance);
1944 
1946  "Delivering to app %x to %x",
1947  instance->my_high_delivered + 1, instance->old_ring_state_high_seq_received);
1948 
1949  aru_save = instance->my_aru;
1950  instance->my_aru = instance->old_ring_state_aru;
1951 
1952  messages_deliver_to_app (instance, 0, instance->old_ring_state_high_seq_received);
1953 
1954  /*
1955  * Calculate joined and left list
1956  */
1957  memb_set_subtract (instance->my_left_memb_list,
1958  &instance->my_left_memb_entries,
1959  instance->my_memb_list, instance->my_memb_entries,
1960  instance->my_trans_memb_list, instance->my_trans_memb_entries);
1961 
1962  memb_set_subtract (joined_list, &joined_list_entries,
1963  instance->my_new_memb_list, instance->my_new_memb_entries,
1964  instance->my_trans_memb_list, instance->my_trans_memb_entries);
1965 
1966  /*
1967  * Install new membership
1968  */
1969  instance->my_memb_entries = instance->my_new_memb_entries;
1970  memcpy (&instance->my_memb_list, instance->my_new_memb_list,
1971  sizeof (struct srp_addr) * instance->my_memb_entries);
1972  instance->last_released = 0;
1973  instance->my_set_retrans_flg = 0;
1974 
1975  /*
1976  * Inform RRP about transitional change
1977  */
1979  instance->totemrrp_context,
1981  instance->my_trans_memb_list, instance->my_trans_memb_entries,
1982  instance->my_left_memb_list, instance->my_left_memb_entries,
1983  NULL, 0,
1984  &instance->my_ring_id);
1985  /*
1986  * Deliver transitional configuration to application
1987  */
1988  srp_addr_to_nodeid (left_list, instance->my_left_memb_list,
1989  instance->my_left_memb_entries);
1990  srp_addr_to_nodeid (trans_memb_list_totemip,
1991  instance->my_trans_memb_list, instance->my_trans_memb_entries);
1993  trans_memb_list_totemip, instance->my_trans_memb_entries,
1994  left_list, instance->my_left_memb_entries,
1995  0, 0, &instance->my_ring_id);
1996  /*
1997  * Switch new totemsrp messages queue. Messages sent from now on are stored
1998  * in different queue so synchronization messages are delivered first. Totempg
1999  * buffers will be switched later.
2000  */
2001  instance->waiting_trans_ack = 1;
2002 
2003 // TODO we need to filter to ensure we only deliver those
2004 // messages which are part of instance->my_deliver_memb
2005  messages_deliver_to_app (instance, 1, instance->old_ring_state_high_seq_received);
2006 
2007  /*
2008  * Switch totempg buffers. This used to be right after
2009  * instance->waiting_trans_ack = 1;
2010  * line. This was causing problem, because there may be not yet
2011  * processed parts of messages in totempg buffers.
2012  * So when buffers were switched and recovered messages
2013  * got delivered it was not possible to assemble them.
2014  */
2015  instance->totemsrp_waiting_trans_ack_cb_fn (1);
2016 
2017  instance->my_aru = aru_save;
2018 
2019  /*
2020  * Inform RRP about regular membership change
2021  */
2023  instance->totemrrp_context,
2025  instance->my_new_memb_list, instance->my_new_memb_entries,
2026  NULL, 0,
2027  joined_list, joined_list_entries,
2028  &instance->my_ring_id);
2029  /*
2030  * Deliver regular configuration to application
2031  */
2032  srp_addr_to_nodeid (new_memb_list_totemip,
2033  instance->my_new_memb_list, instance->my_new_memb_entries);
2034  srp_addr_to_nodeid (joined_list_totemip, joined_list,
2035  joined_list_entries);
2037  new_memb_list_totemip, instance->my_new_memb_entries,
2038  0, 0,
2039  joined_list_totemip, joined_list_entries, &instance->my_ring_id);
2040 
2041  /*
2042  * The recovery sort queue now becomes the regular
2043  * sort queue. It is necessary to copy the state
2044  * into the regular sort queue.
2045  */
2046  sq_copy (&instance->regular_sort_queue, &instance->recovery_sort_queue);
2047  instance->my_last_aru = SEQNO_START_MSG;
2048 
2049  /* When making my_proc_list smaller, ensure that the
2050  * now non-used entries are zero-ed out. There are some suspect
2051  * assert's that assume that there is always 2 entries in the list.
2052  * These fail when my_proc_list is reduced to 1 entry (and the
2053  * valid [0] entry is the same as the 'unused' [1] entry).
2054  */
2055  memset(instance->my_proc_list, 0,
2056  sizeof (struct srp_addr) * instance->my_proc_list_entries);
2057 
2058  instance->my_proc_list_entries = instance->my_new_memb_entries;
2059  memcpy (instance->my_proc_list, instance->my_new_memb_list,
2060  sizeof (struct srp_addr) * instance->my_memb_entries);
2061 
2062  instance->my_failed_list_entries = 0;
2063  /*
2064  * TODO Not exactly to spec
2065  *
2066  * At the entry to this function all messages without a gap are
2067  * deliered.
2068  *
2069  * This code throw away messages from the last gap in the sort queue
2070  * to my_high_seq_received
2071  *
2072  * What should really happen is we should deliver all messages up to
2073  * a gap, then delier the transitional configuration, then deliver
2074  * the messages between the first gap and my_high_seq_received, then
2075  * deliver a regular configuration, then deliver the regular
2076  * configuration
2077  *
2078  * Unfortunately totempg doesn't appear to like this operating mode
2079  * which needs more inspection
2080  */
2081  i = instance->my_high_seq_received + 1;
2082  do {
2083  void *ptr;
2084 
2085  i -= 1;
2086  res = sq_item_get (&instance->regular_sort_queue, i, &ptr);
2087  if (i == 0) {
2088  break;
2089  }
2090  } while (res);
2091 
2092  instance->my_high_delivered = i;
2093 
2094  for (i = 0; i <= instance->my_high_delivered; i++) {
2095  void *ptr;
2096 
2097  res = sq_item_get (&instance->regular_sort_queue, i, &ptr);
2098  if (res == 0) {
2099  struct sort_queue_item *regular_message;
2100 
2101  regular_message = ptr;
2102  free (regular_message->mcast);
2103  }
2104  }
2105  sq_items_release (&instance->regular_sort_queue, instance->my_high_delivered);
2106  instance->last_released = instance->my_high_delivered;
2107 
2108  if (joined_list_entries) {
2109  int sptr = 0;
2110  sptr += snprintf(joined_node_msg, sizeof(joined_node_msg)-sptr, " joined:");
2111  for (i=0; i< joined_list_entries; i++) {
2112  sptr += snprintf(joined_node_msg+sptr, sizeof(joined_node_msg)-sptr, " %u", joined_list_totemip[i]);
2113  }
2114  }
2115  else {
2116  joined_node_msg[0] = '\0';
2117  }
2118 
2119  if (instance->my_left_memb_entries) {
2120  int sptr = 0;
2121  int sptr2 = 0;
2122  sptr += snprintf(left_node_msg, sizeof(left_node_msg)-sptr, " left:");
2123  for (i=0; i< instance->my_left_memb_entries; i++) {
2124  sptr += snprintf(left_node_msg+sptr, sizeof(left_node_msg)-sptr, " %u", left_list[i]);
2125  }
2126  for (i=0; i< instance->my_left_memb_entries; i++) {
2127  if (my_leave_memb_match(instance, left_list[i]) == 0) {
2128  if (sptr2 == 0) {
2129  sptr2 += snprintf(failed_node_msg, sizeof(failed_node_msg)-sptr2, " failed:");
2130  }
2131  sptr2 += snprintf(failed_node_msg+sptr2, sizeof(left_node_msg)-sptr2, " %u", left_list[i]);
2132  }
2133  }
2134  if (sptr2 == 0) {
2135  failed_node_msg[0] = '\0';
2136  }
2137  }
2138  else {
2139  left_node_msg[0] = '\0';
2140  failed_node_msg[0] = '\0';
2141  }
2142 
2143  my_leave_memb_clear(instance);
2144 
2146  "entering OPERATIONAL state.");
2148  "A new membership (%s:%lld) was formed. Members%s%s",
2149  totemip_print (&instance->my_ring_id.rep),
2150  instance->my_ring_id.seq,
2151  joined_node_msg,
2152  left_node_msg);
2153 
2154  if (strlen(failed_node_msg)) {
2156  "Failed to receive the leave message.%s",
2157  failed_node_msg);
2158  }
2159 
2160  instance->memb_state = MEMB_STATE_OPERATIONAL;
2161 
2162  instance->stats.operational_entered++;
2163  instance->stats.continuous_gather = 0;
2164 
2165  instance->my_received_flg = 1;
2166 
2167  reset_pause_timeout (instance);
2168 
2169  /*
2170  * Save ring id information from this configuration to determine
2171  * which processors are transitioning from old regular configuration
2172  * in to new regular configuration on the next configuration change
2173  */
2174  memcpy (&instance->my_old_ring_id, &instance->my_ring_id,
2175  sizeof (struct memb_ring_id));
2176 
2177  return;
2178 }
2179 
2180 static void memb_state_gather_enter (
2181  struct totemsrp_instance *instance,
2182  enum gather_state_from gather_from)
2183 {
2184  int32_t res;
2185 
2186  instance->orf_token_discard = 1;
2187 
2188  instance->originated_orf_token = 0;
2189 
2190  memb_set_merge (
2191  &instance->my_id, 1,
2192  instance->my_proc_list, &instance->my_proc_list_entries);
2193 
2194  memb_join_message_send (instance);
2195 
2196  /*
2197  * Restart the join timeout
2198  */
2199  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);
2200 
2201  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
2202  QB_LOOP_MED,
2203  instance->totem_config->join_timeout*QB_TIME_NS_IN_MSEC,
2204  (void *)instance,
2205  memb_timer_function_state_gather,
2207  if (res != 0) {
2208  log_printf(instance->totemsrp_log_level_error, "memb_state_gather_enter - qb_loop_timer_add error(1) : %d", res);
2209  }
2210 
2211  /*
2212  * Restart the consensus timeout
2213  */
2214  qb_loop_timer_del (instance->totemsrp_poll_handle,
2216 
2217  res = qb_loop_timer_add (instance->totemsrp_poll_handle,
2218  QB_LOOP_MED,
2219  instance->totem_config->consensus_timeout*QB_TIME_NS_IN_MSEC,
2220  (void *)instance,
2221  memb_timer_function_gather_consensus_timeout,
2223  if (res != 0) {
2224  log_printf(instance->totemsrp_log_level_error, "memb_state_gather_enter - qb_loop_timer_add error(2) : %d", res);
2225  }
2226 
2227  /*
2228  * Cancel the token loss and token retransmission timeouts
2229  */
2230  cancel_token_retransmit_timeout (instance); // REVIEWED
2231  cancel_token_timeout (instance); // REVIEWED
2232  cancel_merge_detect_timeout (instance);
2233 
2234  memb_consensus_reset (instance);
2235 
2236  memb_consensus_set (instance, &instance->my_id);
2237 
2239  "entering GATHER state from %d(%s).",
2240  gather_from, gsfrom_to_msg(gather_from));
2241 
2242  instance->memb_state = MEMB_STATE_GATHER;
2243  instance->stats.gather_entered++;
2244 
2245  if (gather_from == TOTEMSRP_GSFROM_THE_CONSENSUS_TIMEOUT_EXPIRED) {
2246  /*
2247  * State 3 means gather, so we are continuously gathering.
2248  */
2249  instance->stats.continuous_gather++;
2250  }
2251 
2252  return;
2253 }
2254 
2255 static void timer_function_token_retransmit_timeout (void *data);
2256 
2257 static void target_set_completed (
2258  void *context)
2259 {
2260  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
2261 
2262  memb_state_commit_token_send (instance);
2263 
2264 }
2265 
2266 static void memb_state_commit_enter (
2267  struct totemsrp_instance *instance)
2268 {
2269  old_ring_state_save (instance);
2270 
2271  memb_state_commit_token_update (instance);
2272 
2273  memb_state_commit_token_target_set (instance);
2274 
2275  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);
2276 
2278 
2279  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_consensus_timeout);
2280 
2282 
2283  memb_ring_id_set (instance, &instance->commit_token->ring_id);
2284  instance->memb_ring_id_store (&instance->my_ring_id, &instance->my_id.addr[0]);
2285 
2286  instance->token_ring_id_seq = instance->my_ring_id.seq;
2287 
2289  "entering COMMIT state.");
2290 
2291  instance->memb_state = MEMB_STATE_COMMIT;
2292  reset_token_retransmit_timeout (instance); // REVIEWED
2293  reset_token_timeout (instance); // REVIEWED
2294 
2295  instance->stats.commit_entered++;
2296  instance->stats.continuous_gather = 0;
2297 
2298  /*
2299  * reset all flow control variables since we are starting a new ring
2300  */
2301  instance->my_trc = 0;
2302  instance->my_pbl = 0;
2303  instance->my_cbl = 0;
2304  /*
2305  * commit token sent after callback that token target has been set
2306  */
2307 }
2308 
2309 static void memb_state_recovery_enter (
2310  struct totemsrp_instance *instance,
2312 {
2313  int i;
2314  int local_received_flg = 1;
2315  unsigned int low_ring_aru;
2316  unsigned int range = 0;
2317  unsigned int messages_originated = 0;
2318  const struct srp_addr *addr;
2319  struct memb_commit_token_memb_entry *memb_list;
2320  struct memb_ring_id my_new_memb_ring_id_list[PROCESSOR_COUNT_MAX];
2321 
2322  addr = (const struct srp_addr *)commit_token->end_of_commit_token;
2323  memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries);
2324 
2326  "entering RECOVERY state.");
2327 
2328  instance->orf_token_discard = 0;
2329 
2330  instance->my_high_ring_delivered = 0;
2331 
2332  sq_reinit (&instance->recovery_sort_queue, SEQNO_START_MSG);
2333  cs_queue_reinit (&instance->retrans_message_queue);
2334 
2335  low_ring_aru = instance->old_ring_state_high_seq_received;
2336 
2337  memb_state_commit_token_send_recovery (instance, commit_token);
2338 
2339  instance->my_token_seq = SEQNO_START_TOKEN - 1;
2340 
2341  /*
2342  * Build regular configuration
2343  */
2345  instance->totemrrp_context,
2346  commit_token->addr_entries);
2347 
2348  /*
2349  * Build transitional configuration
2350  */
2351  for (i = 0; i < instance->my_new_memb_entries; i++) {
2352  memcpy (&my_new_memb_ring_id_list[i],
2353  &memb_list[i].ring_id,
2354  sizeof (struct memb_ring_id));
2355  }
2356  memb_set_and_with_ring_id (
2357  instance->my_new_memb_list,
2358  my_new_memb_ring_id_list,
2359  instance->my_new_memb_entries,
2360  instance->my_memb_list,
2361  instance->my_memb_entries,
2362  &instance->my_old_ring_id,
2363  instance->my_trans_memb_list,
2364  &instance->my_trans_memb_entries);
2365 
2366  for (i = 0; i < instance->my_trans_memb_entries; i++) {
2368  "TRANS [%d] member %s:", i, totemip_print (&instance->my_trans_memb_list[i].addr[0]));
2369  }
2370  for (i = 0; i < instance->my_new_memb_entries; i++) {
2372  "position [%d] member %s:", i, totemip_print (&addr[i].addr[0]));
2374  "previous ring seq %llx rep %s",
2375  memb_list[i].ring_id.seq,
2376  totemip_print (&memb_list[i].ring_id.rep));
2377 
2379  "aru %x high delivered %x received flag %d",
2380  memb_list[i].aru,
2381  memb_list[i].high_delivered,
2382  memb_list[i].received_flg);
2383 
2384  // assert (totemip_print (&memb_list[i].ring_id.rep) != 0);
2385  }
2386  /*
2387  * Determine if any received flag is false
2388  */
2389  for (i = 0; i < commit_token->addr_entries; i++) {
2390  if (memb_set_subset (&instance->my_new_memb_list[i], 1,
2391  instance->my_trans_memb_list, instance->my_trans_memb_entries) &&
2392 
2393  memb_list[i].received_flg == 0) {
2394  instance->my_deliver_memb_entries = instance->my_trans_memb_entries;
2395  memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list,
2396  sizeof (struct srp_addr) * instance->my_trans_memb_entries);
2397  local_received_flg = 0;
2398  break;
2399  }
2400  }
2401  if (local_received_flg == 1) {
2402  goto no_originate;
2403  } /* Else originate messages if we should */
2404 
2405  /*
2406  * Calculate my_low_ring_aru, instance->my_high_ring_delivered for the transitional membership
2407  */
2408  for (i = 0; i < commit_token->addr_entries; i++) {
2409  if (memb_set_subset (&instance->my_new_memb_list[i], 1,
2410  instance->my_deliver_memb_list,
2411  instance->my_deliver_memb_entries) &&
2412 
2413  memcmp (&instance->my_old_ring_id,
2414  &memb_list[i].ring_id,
2415  sizeof (struct memb_ring_id)) == 0) {
2416 
2417  if (sq_lt_compare (memb_list[i].aru, low_ring_aru)) {
2418 
2419  low_ring_aru = memb_list[i].aru;
2420  }
2421  if (sq_lt_compare (instance->my_high_ring_delivered, memb_list[i].high_delivered)) {
2422  instance->my_high_ring_delivered = memb_list[i].high_delivered;
2423  }
2424  }
2425  }
2426 
2427  /*
2428  * Copy all old ring messages to instance->retrans_message_queue
2429  */
2430  range = instance->old_ring_state_high_seq_received - low_ring_aru;
2431  if (range == 0) {
2432  /*
2433  * No messages to copy
2434  */
2435  goto no_originate;
2436  }
2437  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
2438 
2440  "copying all old ring messages from %x-%x.",
2441  low_ring_aru + 1, instance->old_ring_state_high_seq_received);
2442 
2443  for (i = 1; i <= range; i++) {
2445  struct message_item message_item;
2446  void *ptr;
2447  int res;
2448 
2449  res = sq_item_get (&instance->regular_sort_queue,
2450  low_ring_aru + i, &ptr);
2451  if (res != 0) {
2452  continue;
2453  }
2454  sort_queue_item = ptr;
2455  messages_originated++;
2456  memset (&message_item, 0, sizeof (struct message_item));
2457  // TODO LEAK
2458  message_item.mcast = totemsrp_buffer_alloc (instance);
2459  assert (message_item.mcast);
2461  srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);
2463  message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid;
2464  assert (message_item.mcast->header.nodeid);
2466  memcpy (&message_item.mcast->ring_id, &instance->my_ring_id,
2467  sizeof (struct memb_ring_id));
2468  message_item.msg_len = sort_queue_item->msg_len + sizeof (struct mcast);
2469  memcpy (((char *)message_item.mcast) + sizeof (struct mcast),
2470  sort_queue_item->mcast,
2471  sort_queue_item->msg_len);
2472  cs_queue_item_add (&instance->retrans_message_queue, &message_item);
2473  }
2475  "Originated %d messages in RECOVERY.", messages_originated);
2476  goto originated;
2477 
2478 no_originate:
2480  "Did not need to originate any messages in recovery.");
2481 
2482 originated:
2483  instance->my_aru = SEQNO_START_MSG;
2484  instance->my_aru_count = 0;
2485  instance->my_seq_unchanged = 0;
2487  instance->my_install_seq = SEQNO_START_MSG;
2488  instance->last_released = SEQNO_START_MSG;
2489 
2490  reset_token_timeout (instance); // REVIEWED
2491  reset_token_retransmit_timeout (instance); // REVIEWED
2492 
2493  instance->memb_state = MEMB_STATE_RECOVERY;
2494  instance->stats.recovery_entered++;
2495  instance->stats.continuous_gather = 0;
2496 
2497  return;
2498 }
2499 
2500 void totemsrp_event_signal (void *srp_context, enum totem_event_type type, int value)
2501 {
2502  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
2503 
2504  token_hold_cancel_send (instance);
2505 
2506  return;
2507 }
2508 
2510  void *srp_context,
2511  struct iovec *iovec,
2512  unsigned int iov_len,
2513  int guarantee)
2514 {
2515  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
2516  int i;
2517  struct message_item message_item;
2518  char *addr;
2519  unsigned int addr_idx;
2520  struct cs_queue *queue_use;
2521 
2522  if (instance->waiting_trans_ack) {
2523  queue_use = &instance->new_message_queue_trans;
2524  } else {
2525  queue_use = &instance->new_message_queue;
2526  }
2527 
2528  if (cs_queue_is_full (queue_use)) {
2529  log_printf (instance->totemsrp_log_level_debug, "queue full");
2530  return (-1);
2531  }
2532 
2533  memset (&message_item, 0, sizeof (struct message_item));
2534 
2535  /*
2536  * Allocate pending item
2537  */
2538  message_item.mcast = totemsrp_buffer_alloc (instance);
2539  if (message_item.mcast == 0) {
2540  goto error_mcast;
2541  }
2542 
2543  /*
2544  * Set mcast header
2545  */
2546  memset(message_item.mcast, 0, sizeof (struct mcast));
2547  message_item.mcast->header.type = MESSAGE_TYPE_MCAST;
2548  message_item.mcast->header.endian_detector = ENDIAN_LOCAL;
2550  message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid;
2551  assert (message_item.mcast->header.nodeid);
2552 
2553  message_item.mcast->guarantee = guarantee;
2554  srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);
2555 
2556  addr = (char *)message_item.mcast;
2557  addr_idx = sizeof (struct mcast);
2558  for (i = 0; i < iov_len; i++) {
2559  memcpy (&addr[addr_idx], iovec[i].iov_base, iovec[i].iov_len);
2560  addr_idx += iovec[i].iov_len;
2561  }
2562 
2563  message_item.msg_len = addr_idx;
2564 
2565  log_printf (instance->totemsrp_log_level_trace, "mcasted message added to pending queue");
2566  instance->stats.mcast_tx++;
2567  cs_queue_item_add (queue_use, &message_item);
2568 
2569  return (0);
2570 
2571 error_mcast:
2572  return (-1);
2573 }
2574 
2575 /*
2576  * Determine if there is room to queue a new message
2577  */
2578 int totemsrp_avail (void *srp_context)
2579 {
2580  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
2581  int avail;
2582  struct cs_queue *queue_use;
2583 
2584  if (instance->waiting_trans_ack) {
2585  queue_use = &instance->new_message_queue_trans;
2586  } else {
2587  queue_use = &instance->new_message_queue;
2588  }
2589  cs_queue_avail (queue_use, &avail);
2590 
2591  return (avail);
2592 }
2593 
2594 /*
2595  * ORF Token Management
2596  */
2597 /*
2598  * Recast message to mcast group if it is available
2599  */
2600 static int orf_token_remcast (
2601  struct totemsrp_instance *instance,
2602  int seq)
2603 {
2604  struct sort_queue_item *sort_queue_item;
2605  int res;
2606  void *ptr;
2607 
2608  struct sq *sort_queue;
2609 
2610  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2611  sort_queue = &instance->recovery_sort_queue;
2612  } else {
2613  sort_queue = &instance->regular_sort_queue;
2614  }
2615 
2616  res = sq_in_range (sort_queue, seq);
2617  if (res == 0) {
2618  log_printf (instance->totemsrp_log_level_debug, "sq not in range");
2619  return (-1);
2620  }
2621 
2622  /*
2623  * Get RTR item at seq, if not available, return
2624  */
2625  res = sq_item_get (sort_queue, seq, &ptr);
2626  if (res != 0) {
2627  return -1;
2628  }
2629 
2630  sort_queue_item = ptr;
2631 
2633  instance->totemrrp_context,
2634  sort_queue_item->mcast,
2635  sort_queue_item->msg_len);
2636 
2637  return (0);
2638 }
2639 
2640 
2641 /*
2642  * Free all freeable messages from ring
2643  */
2644 static void messages_free (
2645  struct totemsrp_instance *instance,
2646  unsigned int token_aru)
2647 {
2648  struct sort_queue_item *regular_message;
2649  unsigned int i;
2650  int res;
2651  int log_release = 0;
2652  unsigned int release_to;
2653  unsigned int range = 0;
2654 
2655  release_to = token_aru;
2656  if (sq_lt_compare (instance->my_last_aru, release_to)) {
2657  release_to = instance->my_last_aru;
2658  }
2659  if (sq_lt_compare (instance->my_high_delivered, release_to)) {
2660  release_to = instance->my_high_delivered;
2661  }
2662 
2663  /*
2664  * Ensure we dont try release before an already released point
2665  */
2666  if (sq_lt_compare (release_to, instance->last_released)) {
2667  return;
2668  }
2669 
2670  range = release_to - instance->last_released;
2671  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
2672 
2673  /*
2674  * Release retransmit list items if group aru indicates they are transmitted
2675  */
2676  for (i = 1; i <= range; i++) {
2677  void *ptr;
2678 
2679  res = sq_item_get (&instance->regular_sort_queue,
2680  instance->last_released + i, &ptr);
2681  if (res == 0) {
2682  regular_message = ptr;
2683  totemsrp_buffer_release (instance, regular_message->mcast);
2684  }
2685  sq_items_release (&instance->regular_sort_queue,
2686  instance->last_released + i);
2687 
2688  log_release = 1;
2689  }
2690  instance->last_released += range;
2691 
2692  if (log_release) {
2694  "releasing messages up to and including %x", release_to);
2695  }
2696 }
2697 
2698 static void update_aru (
2699  struct totemsrp_instance *instance)
2700 {
2701  unsigned int i;
2702  int res;
2703  struct sq *sort_queue;
2704  unsigned int range;
2705  unsigned int my_aru_saved = 0;
2706 
2707  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2708  sort_queue = &instance->recovery_sort_queue;
2709  } else {
2710  sort_queue = &instance->regular_sort_queue;
2711  }
2712 
2713  range = instance->my_high_seq_received - instance->my_aru;
2714 
2715  my_aru_saved = instance->my_aru;
2716  for (i = 1; i <= range; i++) {
2717 
2718  void *ptr;
2719 
2720  res = sq_item_get (sort_queue, my_aru_saved + i, &ptr);
2721  /*
2722  * If hole, stop updating aru
2723  */
2724  if (res != 0) {
2725  break;
2726  }
2727  }
2728  instance->my_aru += i - 1;
2729 }
2730 
2731 /*
2732  * Multicasts pending messages onto the ring (requires orf_token possession)
2733  */
2734 static int orf_token_mcast (
2735  struct totemsrp_instance *instance,
2736  struct orf_token *token,
2737  int fcc_mcasts_allowed)
2738 {
2739  struct message_item *message_item = 0;
2740  struct cs_queue *mcast_queue;
2741  struct sq *sort_queue;
2742  struct sort_queue_item sort_queue_item;
2743  struct mcast *mcast;
2744  unsigned int fcc_mcast_current;
2745 
2746  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2747  mcast_queue = &instance->retrans_message_queue;
2748  sort_queue = &instance->recovery_sort_queue;
2749  reset_token_retransmit_timeout (instance); // REVIEWED
2750  } else {
2751  if (instance->waiting_trans_ack) {
2752  mcast_queue = &instance->new_message_queue_trans;
2753  } else {
2754  mcast_queue = &instance->new_message_queue;
2755  }
2756 
2757  sort_queue = &instance->regular_sort_queue;
2758  }
2759 
2760  for (fcc_mcast_current = 0; fcc_mcast_current < fcc_mcasts_allowed; fcc_mcast_current++) {
2761  if (cs_queue_is_empty (mcast_queue)) {
2762  break;
2763  }
2764  message_item = (struct message_item *)cs_queue_item_get (mcast_queue);
2765 
2766  message_item->mcast->seq = ++token->seq;
2767  message_item->mcast->this_seqno = instance->global_seqno++;
2768 
2769  /*
2770  * Build IO vector
2771  */
2772  memset (&sort_queue_item, 0, sizeof (struct sort_queue_item));
2773  sort_queue_item.mcast = message_item->mcast;
2774  sort_queue_item.msg_len = message_item->msg_len;
2775 
2776  mcast = sort_queue_item.mcast;
2777 
2778  memcpy (&mcast->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id));
2779 
2780  /*
2781  * Add message to retransmit queue
2782  */
2783  sq_item_add (sort_queue, &sort_queue_item, message_item->mcast->seq);
2784 
2786  instance->totemrrp_context,
2787  message_item->mcast,
2788  message_item->msg_len);
2789 
2790  /*
2791  * Delete item from pending queue
2792  */
2793  cs_queue_item_remove (mcast_queue);
2794 
2795  /*
2796  * If messages mcasted, deliver any new messages to totempg
2797  */
2798  instance->my_high_seq_received = token->seq;
2799  }
2800 
2801  update_aru (instance);
2802 
2803  /*
2804  * Return 1 if more messages are available for single node clusters
2805  */
2806  return (fcc_mcast_current);
2807 }
2808 
2809 /*
2810  * Remulticasts messages in orf_token's retransmit list (requires orf_token)
2811  * Modify's orf_token's rtr to include retransmits required by this process
2812  */
2813 static int orf_token_rtr (
2814  struct totemsrp_instance *instance,
2815  struct orf_token *orf_token,
2816  unsigned int *fcc_allowed)
2817 {
2818  unsigned int res;
2819  unsigned int i, j;
2820  unsigned int found;
2821  struct sq *sort_queue;
2822  struct rtr_item *rtr_list;
2823  unsigned int range = 0;
2824  char retransmit_msg[1024];
2825  char value[64];
2826 
2827  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2828  sort_queue = &instance->recovery_sort_queue;
2829  } else {
2830  sort_queue = &instance->regular_sort_queue;
2831  }
2832 
2833  rtr_list = &orf_token->rtr_list[0];
2834 
2835  strcpy (retransmit_msg, "Retransmit List: ");
2836  if (orf_token->rtr_list_entries) {
2838  "Retransmit List %d", orf_token->rtr_list_entries);
2839  for (i = 0; i < orf_token->rtr_list_entries; i++) {
2840  sprintf (value, "%x ", rtr_list[i].seq);
2841  strcat (retransmit_msg, value);
2842  }
2843  strcat (retransmit_msg, "");
2845  "%s", retransmit_msg);
2846  }
2847 
2848  /*
2849  * Retransmit messages on orf_token's RTR list from RTR queue
2850  */
2851  for (instance->fcc_remcast_current = 0, i = 0;
2852  instance->fcc_remcast_current < *fcc_allowed && i < orf_token->rtr_list_entries;) {
2853 
2854  /*
2855  * If this retransmit request isn't from this configuration,
2856  * try next rtr entry
2857  */
2858  if (memcmp (&rtr_list[i].ring_id, &instance->my_ring_id,
2859  sizeof (struct memb_ring_id)) != 0) {
2860 
2861  i += 1;
2862  continue;
2863  }
2864 
2865  res = orf_token_remcast (instance, rtr_list[i].seq);
2866  if (res == 0) {
2867  /*
2868  * Multicasted message, so no need to copy to new retransmit list
2869  */
2870  orf_token->rtr_list_entries -= 1;
2871  assert (orf_token->rtr_list_entries >= 0);
2872  memmove (&rtr_list[i], &rtr_list[i + 1],
2873  sizeof (struct rtr_item) * (orf_token->rtr_list_entries - i));
2874 
2875  instance->stats.mcast_retx++;
2876  instance->fcc_remcast_current++;
2877  } else {
2878  i += 1;
2879  }
2880  }
2881  *fcc_allowed = *fcc_allowed - instance->fcc_remcast_current;
2882 
2883  /*
2884  * Add messages to retransmit to RTR list
2885  * but only retry if there is room in the retransmit list
2886  */
2887 
2888  range = orf_token->seq - instance->my_aru;
2889  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
2890 
2891  for (i = 1; (orf_token->rtr_list_entries < RETRANSMIT_ENTRIES_MAX) &&
2892  (i <= range); i++) {
2893 
2894  /*
2895  * Ensure message is within the sort queue range
2896  */
2897  res = sq_in_range (sort_queue, instance->my_aru + i);
2898  if (res == 0) {
2899  break;
2900  }
2901 
2902  /*
2903  * Find if a message is missing from this processor
2904  */
2905  res = sq_item_inuse (sort_queue, instance->my_aru + i);
2906  if (res == 0) {
2907  /*
2908  * Determine how many times we have missed receiving
2909  * this sequence number. sq_item_miss_count increments
2910  * a counter for the sequence number. The miss count
2911  * will be returned and compared. This allows time for
2912  * delayed multicast messages to be received before
2913  * declaring the message is missing and requesting a
2914  * retransmit.
2915  */
2916  res = sq_item_miss_count (sort_queue, instance->my_aru + i);
2917  if (res < instance->totem_config->miss_count_const) {
2918  continue;
2919  }
2920 
2921  /*
2922  * Determine if missing message is already in retransmit list
2923  */
2924  found = 0;
2925  for (j = 0; j < orf_token->rtr_list_entries; j++) {
2926  if (instance->my_aru + i == rtr_list[j].seq) {
2927  found = 1;
2928  }
2929  }
2930  if (found == 0) {
2931  /*
2932  * Missing message not found in current retransmit list so add it
2933  */
2934  memcpy (&rtr_list[orf_token->rtr_list_entries].ring_id,
2935  &instance->my_ring_id, sizeof (struct memb_ring_id));
2936  rtr_list[orf_token->rtr_list_entries].seq = instance->my_aru + i;
2937  orf_token->rtr_list_entries++;
2938  }
2939  }
2940  }
2941  return (instance->fcc_remcast_current);
2942 }
2943 
2944 static void token_retransmit (struct totemsrp_instance *instance)
2945 {
2947  instance->orf_token_retransmit,
2948  instance->orf_token_retransmit_size);
2949 }
2950 
2951 /*
2952  * Retransmit the regular token if no mcast or token has
2953  * been received in retransmit token period retransmit
2954  * the token to the next processor
2955  */
2956 static void timer_function_token_retransmit_timeout (void *data)
2957 {
2958  struct totemsrp_instance *instance = data;
2959 
2960  switch (instance->memb_state) {
2961  case MEMB_STATE_GATHER:
2962  break;
2963  case MEMB_STATE_COMMIT:
2965  case MEMB_STATE_RECOVERY:
2966  token_retransmit (instance);
2967  reset_token_retransmit_timeout (instance); // REVIEWED
2968  break;
2969  }
2970 }
2971 
2972 static void timer_function_token_hold_retransmit_timeout (void *data)
2973 {
2974  struct totemsrp_instance *instance = data;
2975 
2976  switch (instance->memb_state) {
2977  case MEMB_STATE_GATHER:
2978  break;
2979  case MEMB_STATE_COMMIT:
2980  break;
2982  case MEMB_STATE_RECOVERY:
2983  token_retransmit (instance);
2984  break;
2985  }
2986 }
2987 
2988 static void timer_function_merge_detect_timeout(void *data)
2989 {
2990  struct totemsrp_instance *instance = data;
2991 
2993 
2994  switch (instance->memb_state) {
2996  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
2997  memb_merge_detect_transmit (instance);
2998  }
2999  break;
3000  case MEMB_STATE_GATHER:
3001  case MEMB_STATE_COMMIT:
3002  case MEMB_STATE_RECOVERY:
3003  break;
3004  }
3005 }
3006 
3007 /*
3008  * Send orf_token to next member (requires orf_token)
3009  */
3010 static int token_send (
3011  struct totemsrp_instance *instance,
3012  struct orf_token *orf_token,
3013  int forward_token)
3014 {
3015  int res = 0;
3016  unsigned int orf_token_size;
3017 
3018  orf_token_size = sizeof (struct orf_token) +
3019  (orf_token->rtr_list_entries * sizeof (struct rtr_item));
3020 
3021  orf_token->header.nodeid = instance->my_id.addr[0].nodeid;
3022  memcpy (instance->orf_token_retransmit, orf_token, orf_token_size);
3023  instance->orf_token_retransmit_size = orf_token_size;
3024  assert (orf_token->header.nodeid);
3025 
3026  if (forward_token == 0) {
3027  return (0);
3028  }
3029 
3031  orf_token,
3032  orf_token_size);
3033 
3034  return (res);
3035 }
3036 
3037 static int token_hold_cancel_send (struct totemsrp_instance *instance)
3038 {
3040 
3041  /*
3042  * Only cancel if the token is currently held
3043  */
3044  if (instance->my_token_held == 0) {
3045  return (0);
3046  }
3047  instance->my_token_held = 0;
3048 
3049  /*
3050  * Build message
3051  */
3056  memcpy (&token_hold_cancel.ring_id, &instance->my_ring_id,
3057  sizeof (struct memb_ring_id));
3058  assert (token_hold_cancel.header.nodeid);
3059 
3060  instance->stats.token_hold_cancel_tx++;
3061 
3063  sizeof (struct token_hold_cancel));
3064 
3065  return (0);
3066 }
3067 
3068 static int orf_token_send_initial (struct totemsrp_instance *instance)
3069 {
3070  struct orf_token orf_token;
3071  int res;
3072 
3073  orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN;
3074  orf_token.header.endian_detector = ENDIAN_LOCAL;
3075  orf_token.header.encapsulated = 0;
3076  orf_token.header.nodeid = instance->my_id.addr[0].nodeid;
3077  assert (orf_token.header.nodeid);
3078  orf_token.seq = SEQNO_START_MSG;
3079  orf_token.token_seq = SEQNO_START_TOKEN;
3080  orf_token.retrans_flg = 1;
3081  instance->my_set_retrans_flg = 1;
3082  instance->stats.orf_token_tx++;
3083 
3084  if (cs_queue_is_empty (&instance->retrans_message_queue) == 1) {
3085  orf_token.retrans_flg = 0;
3086  instance->my_set_retrans_flg = 0;
3087  } else {
3088  orf_token.retrans_flg = 1;
3089  instance->my_set_retrans_flg = 1;
3090  }
3091 
3092  orf_token.aru = 0;
3093  orf_token.aru = SEQNO_START_MSG - 1;
3094  orf_token.aru_addr = instance->my_id.addr[0].nodeid;
3095 
3096  memcpy (&orf_token.ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id));
3097  orf_token.fcc = 0;
3098  orf_token.backlog = 0;
3099 
3100  orf_token.rtr_list_entries = 0;
3101 
3102  res = token_send (instance, &orf_token, 1);
3103 
3104  return (res);
3105 }
3106 
3107 static void memb_state_commit_token_update (
3108  struct totemsrp_instance *instance)
3109 {
3110  struct srp_addr *addr;
3111  struct memb_commit_token_memb_entry *memb_list;
3112  unsigned int high_aru;
3113  unsigned int i;
3114 
3115  addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
3116  memb_list = (struct memb_commit_token_memb_entry *)(addr + instance->commit_token->addr_entries);
3117 
3118  memcpy (instance->my_new_memb_list, addr,
3119  sizeof (struct srp_addr) * instance->commit_token->addr_entries);
3120 
3121  instance->my_new_memb_entries = instance->commit_token->addr_entries;
3122 
3123  memcpy (&memb_list[instance->commit_token->memb_index].ring_id,
3124  &instance->my_old_ring_id, sizeof (struct memb_ring_id));
3125 
3126  memb_list[instance->commit_token->memb_index].aru = instance->old_ring_state_aru;
3127  /*
3128  * TODO high delivered is really instance->my_aru, but with safe this
3129  * could change?
3130  */
3131  instance->my_received_flg =
3132  (instance->my_aru == instance->my_high_seq_received);
3133 
3134  memb_list[instance->commit_token->memb_index].received_flg = instance->my_received_flg;
3135 
3136  memb_list[instance->commit_token->memb_index].high_delivered = instance->my_high_delivered;
3137  /*
3138  * find high aru up to current memb_index for all matching ring ids
3139  * if any ring id matching memb_index has aru less then high aru set
3140  * received flag for that entry to false
3141  */
3142  high_aru = memb_list[instance->commit_token->memb_index].aru;
3143  for (i = 0; i <= instance->commit_token->memb_index; i++) {
3144  if (memcmp (&memb_list[instance->commit_token->memb_index].ring_id,
3145  &memb_list[i].ring_id,
3146  sizeof (struct memb_ring_id)) == 0) {
3147 
3148  if (sq_lt_compare (high_aru, memb_list[i].aru)) {
3149  high_aru = memb_list[i].aru;
3150  }
3151  }
3152  }
3153 
3154  for (i = 0; i <= instance->commit_token->memb_index; i++) {
3155  if (memcmp (&memb_list[instance->commit_token->memb_index].ring_id,
3156  &memb_list[i].ring_id,
3157  sizeof (struct memb_ring_id)) == 0) {
3158 
3159  if (sq_lt_compare (memb_list[i].aru, high_aru)) {
3160  memb_list[i].received_flg = 0;
3161  if (i == instance->commit_token->memb_index) {
3162  instance->my_received_flg = 0;
3163  }
3164  }
3165  }
3166  }
3167 
3168  instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3169  instance->commit_token->memb_index += 1;
3170  assert (instance->commit_token->memb_index <= instance->commit_token->addr_entries);
3171  assert (instance->commit_token->header.nodeid);
3172 }
3173 
3174 static void memb_state_commit_token_target_set (
3175  struct totemsrp_instance *instance)
3176 {
3177  struct srp_addr *addr;
3178  unsigned int i;
3179 
3180  addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
3181 
3182  for (i = 0; i < instance->totem_config->interface_count; i++) {
3184  instance->totemrrp_context,
3185  &addr[instance->commit_token->memb_index %
3186  instance->commit_token->addr_entries].addr[i],
3187  i);
3188  }
3189 }
3190 
3191 static int memb_state_commit_token_send_recovery (
3192  struct totemsrp_instance *instance,
3193  struct memb_commit_token *commit_token)
3194 {
3195  unsigned int commit_token_size;
3196 
3197  commit_token->token_seq++;
3198  commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3199  commit_token_size = sizeof (struct memb_commit_token) +
3200  ((sizeof (struct srp_addr) +
3201  sizeof (struct memb_commit_token_memb_entry)) * commit_token->addr_entries);
3202  /*
3203  * Make a copy for retransmission if necessary
3204  */
3205  memcpy (instance->orf_token_retransmit, commit_token, commit_token_size);
3206  instance->orf_token_retransmit_size = commit_token_size;
3207 
3208  instance->stats.memb_commit_token_tx++;
3209 
3211  commit_token,
3212  commit_token_size);
3213 
3214  /*
3215  * Request retransmission of the commit token in case it is lost
3216  */
3217  reset_token_retransmit_timeout (instance);
3218  return (0);
3219 }
3220 
3221 static int memb_state_commit_token_send (
3222  struct totemsrp_instance *instance)
3223 {
3224  unsigned int commit_token_size;
3225 
3226  instance->commit_token->token_seq++;
3227  instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3228  commit_token_size = sizeof (struct memb_commit_token) +
3229  ((sizeof (struct srp_addr) +
3230  sizeof (struct memb_commit_token_memb_entry)) * instance->commit_token->addr_entries);
3231  /*
3232  * Make a copy for retransmission if necessary
3233  */
3234  memcpy (instance->orf_token_retransmit, instance->commit_token, commit_token_size);
3235  instance->orf_token_retransmit_size = commit_token_size;
3236 
3237  instance->stats.memb_commit_token_tx++;
3238 
3240  instance->commit_token,
3241  commit_token_size);
3242 
3243  /*
3244  * Request retransmission of the commit token in case it is lost
3245  */
3246  reset_token_retransmit_timeout (instance);
3247  return (0);
3248 }
3249 
3250 
3251 static int memb_lowest_in_config (struct totemsrp_instance *instance)
3252 {
3253  struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
3254  int token_memb_entries = 0;
3255  int i;
3256  struct totem_ip_address *lowest_addr;
3257 
3258  memb_set_subtract (token_memb, &token_memb_entries,
3259  instance->my_proc_list, instance->my_proc_list_entries,
3260  instance->my_failed_list, instance->my_failed_list_entries);
3261 
3262  /*
3263  * find representative by searching for smallest identifier
3264  */
3265  assert(token_memb_entries > 0);
3266 
3267  lowest_addr = &token_memb[0].addr[0];
3268  for (i = 1; i < token_memb_entries; i++) {
3269  if (totemip_compare(lowest_addr, &token_memb[i].addr[0]) > 0) {
3270  totemip_copy (lowest_addr, &token_memb[i].addr[0]);
3271  }
3272  }
3273  return (totemip_compare (lowest_addr, &instance->my_id.addr[0]) == 0);
3274 }
3275 
3276 static int srp_addr_compare (const void *a, const void *b)
3277 {
3278  const struct srp_addr *srp_a = (const struct srp_addr *)a;
3279  const struct srp_addr *srp_b = (const struct srp_addr *)b;
3280 
3281  return (totemip_compare (&srp_a->addr[0], &srp_b->addr[0]));
3282 }
3283 
3284 static void memb_state_commit_token_create (
3285  struct totemsrp_instance *instance)
3286 {
3287  struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
3288  struct srp_addr *addr;
3289  struct memb_commit_token_memb_entry *memb_list;
3290  int token_memb_entries = 0;
3291 
3293  "Creating commit token because I am the rep.");
3294 
3295  memb_set_subtract (token_memb, &token_memb_entries,
3296  instance->my_proc_list, instance->my_proc_list_entries,
3297  instance->my_failed_list, instance->my_failed_list_entries);
3298 
3299  memset (instance->commit_token, 0, sizeof (struct memb_commit_token));
3302  instance->commit_token->header.encapsulated = 0;
3303  instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3304  assert (instance->commit_token->header.nodeid);
3305 
3306  totemip_copy(&instance->commit_token->ring_id.rep, &instance->my_id.addr[0]);
3307 
3308  instance->commit_token->ring_id.seq = instance->token_ring_id_seq + 4;
3309 
3310  /*
3311  * This qsort is necessary to ensure the commit token traverses
3312  * the ring in the proper order
3313  */
3314  qsort (token_memb, token_memb_entries, sizeof (struct srp_addr),
3315  srp_addr_compare);
3316 
3317  instance->commit_token->memb_index = 0;
3318  instance->commit_token->addr_entries = token_memb_entries;
3319 
3320  addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
3321  memb_list = (struct memb_commit_token_memb_entry *)(addr + instance->commit_token->addr_entries);
3322 
3323  memcpy (addr, token_memb,
3324  token_memb_entries * sizeof (struct srp_addr));
3325  memset (memb_list, 0,
3326  sizeof (struct memb_commit_token_memb_entry) * token_memb_entries);
3327 }
3328 
3329 static void memb_join_message_send (struct totemsrp_instance *instance)
3330 {
3331  char memb_join_data[40000];
3332  struct memb_join *memb_join = (struct memb_join *)memb_join_data;
3333  char *addr;
3334  unsigned int addr_idx;
3335  size_t msg_len;
3336 
3337  memb_join->header.type = MESSAGE_TYPE_MEMB_JOIN;
3338  memb_join->header.endian_detector = ENDIAN_LOCAL;
3339  memb_join->header.encapsulated = 0;
3340  memb_join->header.nodeid = instance->my_id.addr[0].nodeid;
3341  assert (memb_join->header.nodeid);
3342 
3343  msg_len = sizeof(struct memb_join) +
3344  ((instance->my_proc_list_entries + instance->my_failed_list_entries) * sizeof(struct srp_addr));
3345 
3346  if (msg_len > sizeof(memb_join_data)) {
3348  "memb_join_message too long. Ignoring message.");
3349 
3350  return ;
3351  }
3352 
3353  memb_join->ring_seq = instance->my_ring_id.seq;
3354  memb_join->proc_list_entries = instance->my_proc_list_entries;
3355  memb_join->failed_list_entries = instance->my_failed_list_entries;
3356  srp_addr_copy (&memb_join->system_from, &instance->my_id);
3357 
3358  /*
3359  * This mess adds the joined and failed processor lists into the join
3360  * message
3361  */
3362  addr = (char *)memb_join;
3363  addr_idx = sizeof (struct memb_join);
3364  memcpy (&addr[addr_idx],
3365  instance->my_proc_list,
3366  instance->my_proc_list_entries *
3367  sizeof (struct srp_addr));
3368  addr_idx +=
3369  instance->my_proc_list_entries *
3370  sizeof (struct srp_addr);
3371  memcpy (&addr[addr_idx],
3372  instance->my_failed_list,
3373  instance->my_failed_list_entries *
3374  sizeof (struct srp_addr));
3375  addr_idx +=
3376  instance->my_failed_list_entries *
3377  sizeof (struct srp_addr);
3378 
3379  if (instance->totem_config->send_join_timeout) {
3380  usleep (random() % (instance->totem_config->send_join_timeout * 1000));
3381  }
3382 
3383  instance->stats.memb_join_tx++;
3384 
3386  instance->totemrrp_context,
3387  memb_join,
3388  addr_idx);
3389 }
3390 
3391 static void memb_leave_message_send (struct totemsrp_instance *instance)
3392 {
3393  char memb_join_data[40000];
3394  struct memb_join *memb_join = (struct memb_join *)memb_join_data;
3395  char *addr;
3396  unsigned int addr_idx;
3397  int active_memb_entries;
3398  struct srp_addr active_memb[PROCESSOR_COUNT_MAX];
3399  size_t msg_len;
3400 
3402  "sending join/leave message");
3403 
3404  /*
3405  * add us to the failed list, and remove us from
3406  * the members list
3407  */
3408  memb_set_merge(
3409  &instance->my_id, 1,
3410  instance->my_failed_list, &instance->my_failed_list_entries);
3411 
3412  memb_set_subtract (active_memb, &active_memb_entries,
3413  instance->my_proc_list, instance->my_proc_list_entries,
3414  &instance->my_id, 1);
3415 
3416  msg_len = sizeof(struct memb_join) +
3417  ((active_memb_entries + instance->my_failed_list_entries) * sizeof(struct srp_addr));
3418 
3419  if (msg_len > sizeof(memb_join_data)) {
3421  "memb_leave message too long. Ignoring message.");
3422 
3423  return ;
3424  }
3425 
3426  memb_join->header.type = MESSAGE_TYPE_MEMB_JOIN;
3427  memb_join->header.endian_detector = ENDIAN_LOCAL;
3428  memb_join->header.encapsulated = 0;
3429  memb_join->header.nodeid = LEAVE_DUMMY_NODEID;
3430 
3431  memb_join->ring_seq = instance->my_ring_id.seq;
3432  memb_join->proc_list_entries = active_memb_entries;
3433  memb_join->failed_list_entries = instance->my_failed_list_entries;
3434  srp_addr_copy (&memb_join->system_from, &instance->my_id);
3435  memb_join->system_from.addr[0].nodeid = LEAVE_DUMMY_NODEID;
3436 
3437  // TODO: CC Maybe use the actual join send routine.
3438  /*
3439  * This mess adds the joined and failed processor lists into the join
3440  * message
3441  */
3442  addr = (char *)memb_join;
3443  addr_idx = sizeof (struct memb_join);
3444  memcpy (&addr[addr_idx],
3445  active_memb,
3446  active_memb_entries *
3447  sizeof (struct srp_addr));
3448  addr_idx +=
3449  active_memb_entries *
3450  sizeof (struct srp_addr);
3451  memcpy (&addr[addr_idx],
3452  instance->my_failed_list,
3453  instance->my_failed_list_entries *
3454  sizeof (struct srp_addr));
3455  addr_idx +=
3456  instance->my_failed_list_entries *
3457  sizeof (struct srp_addr);
3458 
3459 
3460  if (instance->totem_config->send_join_timeout) {
3461  usleep (random() % (instance->totem_config->send_join_timeout * 1000));
3462  }
3463  instance->stats.memb_join_tx++;
3464 
3466  instance->totemrrp_context,
3467  memb_join,
3468  addr_idx);
3469 }
3470 
3471 static void memb_merge_detect_transmit (struct totemsrp_instance *instance)
3472 {
3474 
3479  srp_addr_copy (&memb_merge_detect.system_from, &instance->my_id);
3480  memcpy (&memb_merge_detect.ring_id, &instance->my_ring_id,
3481  sizeof (struct memb_ring_id));
3482  assert (memb_merge_detect.header.nodeid);
3483 
3484  instance->stats.memb_merge_detect_tx++;
3487  sizeof (struct memb_merge_detect));
3488 }
3489 
3490 static void memb_ring_id_set (
3491  struct totemsrp_instance *instance,
3492  const struct memb_ring_id *ring_id)
3493 {
3494 
3495  memcpy (&instance->my_ring_id, ring_id, sizeof (struct memb_ring_id));
3496 }
3497 
3499  void *srp_context,
3500  void **handle_out,
3501  enum totem_callback_token_type type,
3502  int delete,
3503  int (*callback_fn) (enum totem_callback_token_type type, const void *),
3504  const void *data)
3505 {
3506  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
3507  struct token_callback_instance *callback_handle;
3508 
3509  token_hold_cancel_send (instance);
3510 
3511  callback_handle = malloc (sizeof (struct token_callback_instance));
3512  if (callback_handle == 0) {
3513  return (-1);
3514  }
3515  *handle_out = (void *)callback_handle;
3516  list_init (&callback_handle->list);
3517  callback_handle->callback_fn = callback_fn;
3518  callback_handle->data = (void *) data;
3519  callback_handle->callback_type = type;
3520  callback_handle->delete = delete;
3521  switch (type) {
3523  list_add (&callback_handle->list, &instance->token_callback_received_listhead);
3524  break;
3526  list_add (&callback_handle->list, &instance->token_callback_sent_listhead);
3527  break;
3528  }
3529 
3530  return (0);
3531 }
3532 
3533 void totemsrp_callback_token_destroy (void *srp_context, void **handle_out)
3534 {
3535  struct token_callback_instance *h;
3536 
3537  if (*handle_out) {
3538  h = (struct token_callback_instance *)*handle_out;
3539  list_del (&h->list);
3540  free (h);
3541  h = NULL;
3542  *handle_out = 0;
3543  }
3544 }
3545 
3546 static void token_callbacks_execute (
3547  struct totemsrp_instance *instance,
3548  enum totem_callback_token_type type)
3549 {
3550  struct list_head *list;
3551  struct list_head *list_next;
3552  struct list_head *callback_listhead = 0;
3554  int res;
3555  int del;
3556 
3557  switch (type) {
3559  callback_listhead = &instance->token_callback_received_listhead;
3560  break;
3562  callback_listhead = &instance->token_callback_sent_listhead;
3563  break;
3564  default:
3565  assert (0);
3566  }
3567 
3568  for (list = callback_listhead->next; list != callback_listhead;
3569  list = list_next) {
3570 
3571  token_callback_instance = list_entry (list, struct token_callback_instance, list);
3572 
3573  list_next = list->next;
3574  del = token_callback_instance->delete;
3575  if (del == 1) {
3576  list_del (list);
3577  }
3578 
3579  res = token_callback_instance->callback_fn (
3580  token_callback_instance->callback_type,
3581  token_callback_instance->data);
3582  /*
3583  * This callback failed to execute, try it again on the next token
3584  */
3585  if (res == -1 && del == 1) {
3586  list_add (list, callback_listhead);
3587  } else if (del) {
3588  free (token_callback_instance);
3589  }
3590  }
3591 }
3592 
3593 /*
3594  * Flow control functions
3595  */
3596 static unsigned int backlog_get (struct totemsrp_instance *instance)
3597 {
3598  unsigned int backlog = 0;
3599  struct cs_queue *queue_use = NULL;
3600 
3601  if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
3602  if (instance->waiting_trans_ack) {
3603  queue_use = &instance->new_message_queue_trans;
3604  } else {
3605  queue_use = &instance->new_message_queue;
3606  }
3607  } else
3608  if (instance->memb_state == MEMB_STATE_RECOVERY) {
3609  queue_use = &instance->retrans_message_queue;
3610  }
3611 
3612  if (queue_use != NULL) {
3613  backlog = cs_queue_used (queue_use);
3614  }
3615 
3616  instance->stats.token[instance->stats.latest_token].backlog_calc = backlog;
3617  return (backlog);
3618 }
3619 
3620 static int fcc_calculate (
3621  struct totemsrp_instance *instance,
3622  struct orf_token *token)
3623 {
3624  unsigned int transmits_allowed;
3625  unsigned int backlog_calc;
3626 
3627  transmits_allowed = instance->totem_config->max_messages;
3628 
3629  if (transmits_allowed > instance->totem_config->window_size - token->fcc) {
3630  transmits_allowed = instance->totem_config->window_size - token->fcc;
3631  }
3632 
3633  instance->my_cbl = backlog_get (instance);
3634 
3635  /*
3636  * Only do backlog calculation if there is a backlog otherwise
3637  * we would result in div by zero
3638  */
3639  if (token->backlog + instance->my_cbl - instance->my_pbl) {
3640  backlog_calc = (instance->totem_config->window_size * instance->my_pbl) /
3641  (token->backlog + instance->my_cbl - instance->my_pbl);
3642  if (backlog_calc > 0 && transmits_allowed > backlog_calc) {
3643  transmits_allowed = backlog_calc;
3644  }
3645  }
3646 
3647  return (transmits_allowed);
3648 }
3649 
3650 /*
3651  * don't overflow the RTR sort queue
3652  */
3653 static void fcc_rtr_limit (
3654  struct totemsrp_instance *instance,
3655  struct orf_token *token,
3656  unsigned int *transmits_allowed)
3657 {
3658  int check = QUEUE_RTR_ITEMS_SIZE_MAX;
3659  check -= (*transmits_allowed + instance->totem_config->window_size);
3660  assert (check >= 0);
3661  if (sq_lt_compare (instance->last_released +
3662  QUEUE_RTR_ITEMS_SIZE_MAX - *transmits_allowed -
3663  instance->totem_config->window_size,
3664 
3665  token->seq)) {
3666 
3667  *transmits_allowed = 0;
3668  }
3669 }
3670 
3671 static void fcc_token_update (
3672  struct totemsrp_instance *instance,
3673  struct orf_token *token,
3674  unsigned int msgs_transmitted)
3675 {
3676  token->fcc += msgs_transmitted - instance->my_trc;
3677  token->backlog += instance->my_cbl - instance->my_pbl;
3678  instance->my_trc = msgs_transmitted;
3679  instance->my_pbl = instance->my_cbl;
3680 }
3681 
3682 /*
3683  * Sanity checkers
3684  */
3685 static int check_totemip_sanity(
3686  const struct totemsrp_instance *instance,
3687  const struct totem_ip_address *addr,
3688  int endian_conversion_needed)
3689 {
3690  unsigned short family;
3691 
3692  family = addr->family;
3693  if (endian_conversion_needed) {
3694  family = swab16(family);
3695  }
3696 
3697  if (family != AF_INET && family != AF_INET6) {
3699  "Received message corrupted... ignoring.");
3700 
3701  return (-1);
3702  }
3703 
3704  return (0);
3705 }
3706 
3707 static int check_srpaddr_sanity(
3708  const struct totemsrp_instance *instance,
3709  const struct srp_addr *addr,
3710  int endian_conversion_needed)
3711 {
3712  int i;
3713 
3714  if (addr->no_addrs < 1 || addr->no_addrs > INTERFACE_MAX) {
3715  return (-1);
3716  }
3717 
3718  for (i = 0; i < addr->no_addrs; i++) {
3719  if (i == 0 || addr->addr[i].family != 0) {
3720  if (check_totemip_sanity(instance, &addr->addr[i], endian_conversion_needed) == -1) {
3721  return (-1);
3722  }
3723  }
3724  }
3725 
3726  return (0);
3727 }
3728 
3729 static int check_orf_token_sanity(
3730  const struct totemsrp_instance *instance,
3731  const void *msg,
3732  size_t msg_len,
3733  int endian_conversion_needed)
3734 {
3735  int rtr_entries;
3736  const struct orf_token *token = (const struct orf_token *)msg;
3737  size_t required_len;
3738  int i;
3739 
3740  if (msg_len < sizeof(struct orf_token)) {
3742  "Received orf_token message is too short... ignoring.");
3743 
3744  return (-1);
3745  }
3746 
3747  if (check_totemip_sanity(instance, &token->ring_id.rep, endian_conversion_needed) == -1) {
3748  return (-1);
3749  }
3750 
3751  if (endian_conversion_needed) {
3752  rtr_entries = swab32(token->rtr_list_entries);
3753  } else {
3754  rtr_entries = token->rtr_list_entries;
3755  }
3756 
3757  required_len = sizeof(struct orf_token) + rtr_entries * sizeof(struct rtr_item);
3758  if (msg_len < required_len) {
3760  "Received orf_token message is too short... ignoring.");
3761 
3762  return (-1);
3763  }
3764 
3765  for (i = 0; i < rtr_entries; i++) {
3766  if (check_totemip_sanity(instance, &token->rtr_list[i].ring_id.rep,
3767  endian_conversion_needed) == -1) {
3768  return (-1);
3769  }
3770  }
3771 
3772  return (0);
3773 }
3774 
3775 static int check_mcast_sanity(
3776  struct totemsrp_instance *instance,
3777  const void *msg,
3778  size_t msg_len,
3779  int endian_conversion_needed)
3780 {
3781  const struct mcast *mcast_msg = (const struct mcast *)msg;
3782 
3783  if (msg_len < sizeof(struct mcast)) {
3785  "Received mcast message is too short... ignoring.");
3786 
3787  return (-1);
3788  }
3789 
3790  if ((check_totemip_sanity(instance, &mcast_msg->ring_id.rep, endian_conversion_needed) == -1) ||
3791  (check_srpaddr_sanity(instance, &mcast_msg->system_from, endian_conversion_needed) == -1)) {
3792  return (-1);
3793  }
3794 
3795  return (0);
3796 }
3797 
3798 static int check_memb_merge_detect_sanity(
3799  struct totemsrp_instance *instance,
3800  const void *msg,
3801  size_t msg_len,
3802  int endian_conversion_needed)
3803 {
3804  const struct memb_merge_detect *mmd_msg = (const struct memb_merge_detect *)msg;
3805 
3806  if (msg_len < sizeof(struct memb_merge_detect)) {
3808  "Received memb_merge_detect message is too short... ignoring.");
3809 
3810  return (-1);
3811  }
3812 
3813  if ((check_totemip_sanity(instance, &mmd_msg->ring_id.rep, endian_conversion_needed) == -1) ||
3814  (check_srpaddr_sanity(instance, &mmd_msg->system_from, endian_conversion_needed) == -1)) {
3815  return (-1);
3816  }
3817 
3818  return (0);
3819 }
3820 
3821 static int check_memb_join_sanity(
3822  struct totemsrp_instance *instance,
3823  const void *msg,
3824  size_t msg_len,
3825  int endian_conversion_needed)
3826 {
3827  const struct memb_join *mj_msg = (const struct memb_join *)msg;
3828  unsigned int proc_list_entries;
3829  unsigned int failed_list_entries;
3830  size_t required_len;
3831  const struct srp_addr *proc_list;
3832  const struct srp_addr *failed_list;
3833  int i;
3834 
3835  if (msg_len < sizeof(struct memb_join)) {
3837  "Received memb_join message is too short... ignoring.");
3838 
3839  return (-1);
3840  }
3841 
3842  if (check_srpaddr_sanity(instance, &mj_msg->system_from, endian_conversion_needed) == -1) {
3843  return (-1);
3844  }
3845 
3846  proc_list_entries = mj_msg->proc_list_entries;
3847  failed_list_entries = mj_msg->failed_list_entries;
3848 
3849  if (endian_conversion_needed) {
3850  proc_list_entries = swab32(proc_list_entries);
3851  failed_list_entries = swab32(failed_list_entries);
3852  }
3853 
3854  required_len = sizeof(struct memb_join) + ((proc_list_entries + failed_list_entries) * sizeof(struct srp_addr));
3855  if (msg_len < required_len) {
3857  "Received memb_join message is too short... ignoring.");
3858 
3859  return (-1);
3860  }
3861 
3862  proc_list = (struct srp_addr *)mj_msg->end_of_memb_join;
3863  failed_list = proc_list + proc_list_entries;
3864 
3865  for (i = 0; i < proc_list_entries; i++) {
3866  if (check_srpaddr_sanity(instance, &proc_list[i], endian_conversion_needed) == -1) {
3867  return (-1);
3868  }
3869  }
3870 
3871  for (i = 0; i < failed_list_entries; i++) {
3872  if (check_srpaddr_sanity(instance, &failed_list[i], endian_conversion_needed) == -1) {
3873  return (-1);
3874  }
3875  }
3876 
3877  return (0);
3878 }
3879 
3880 static int check_memb_commit_token_sanity(
3881  struct totemsrp_instance *instance,
3882  const void *msg,
3883  size_t msg_len,
3884  int endian_conversion_needed)
3885 {
3886  const struct memb_commit_token *mct_msg = (const struct memb_commit_token *)msg;
3887  unsigned int addr_entries;
3888  const struct srp_addr *addr;
3889  const struct memb_commit_token_memb_entry *memb_list;
3890  size_t required_len;
3891  int i;
3892 
3893  if (msg_len < sizeof(struct memb_commit_token)) {
3895  "Received memb_commit_token message is too short... ignoring.");
3896 
3897  return (0);
3898  }
3899 
3900  if (check_totemip_sanity(instance, &mct_msg->ring_id.rep, endian_conversion_needed) == -1) {
3901  return (-1);
3902  }
3903 
3904  addr_entries= mct_msg->addr_entries;
3905  if (endian_conversion_needed) {
3906  addr_entries = swab32(addr_entries);
3907  }
3908 
3909  required_len = sizeof(struct memb_commit_token) +
3910  (addr_entries * (sizeof(struct srp_addr) + sizeof(struct memb_commit_token_memb_entry)));
3911  if (msg_len < required_len) {
3913  "Received memb_commit_token message is too short... ignoring.");
3914 
3915  return (-1);
3916  }
3917 
3918  addr = (const struct srp_addr *)mct_msg->end_of_commit_token;
3919  memb_list = (const struct memb_commit_token_memb_entry *)(addr + addr_entries);
3920 
3921  for (i = 0; i < addr_entries; i++) {
3922  if (check_srpaddr_sanity(instance, &addr[i], endian_conversion_needed) == -1) {
3923  return (-1);
3924  }
3925 
3926  if (memb_list[i].ring_id.rep.family != 0) {
3927  if (check_totemip_sanity(instance, &memb_list[i].ring_id.rep,
3928  endian_conversion_needed) == -1) {
3929  return (-1);
3930  }
3931  }
3932  }
3933 
3934  return (0);
3935 }
3936 
3937 static int check_token_hold_cancel_sanity(
3938  struct totemsrp_instance *instance,
3939  const void *msg,
3940  size_t msg_len,
3941  int endian_conversion_needed)
3942 {
3943  const struct token_hold_cancel *thc_msg = (const struct token_hold_cancel *)msg;
3944 
3945  if (msg_len < sizeof(struct token_hold_cancel)) {
3947  "Received token_hold_cancel message is too short... ignoring.");
3948 
3949  return (-1);
3950  }
3951 
3952  if (check_totemip_sanity(instance, &thc_msg->ring_id.rep, endian_conversion_needed) == -1) {
3953  return (-1);
3954  }
3955 
3956  return (0);
3957 }
3958 
3959 /*
3960  * Message Handlers
3961  */
3962 
3963 unsigned long long int tv_old;
3964 /*
3965  * message handler called when TOKEN message type received
3966  */
3967 static int message_handler_orf_token (
3968  struct totemsrp_instance *instance,
3969  const void *msg,
3970  size_t msg_len,
3971  int endian_conversion_needed)
3972 {
3973  char token_storage[1500];
3974  char token_convert[1500];
3975  struct orf_token *token = NULL;
3976  int forward_token;
3977  unsigned int transmits_allowed;
3978  unsigned int mcasted_retransmit;
3979  unsigned int mcasted_regular;
3980  unsigned int last_aru;
3981 
3982 #ifdef GIVEINFO
3983  unsigned long long tv_current;
3984  unsigned long long tv_diff;
3985 
3986  tv_current = qb_util_nano_current_get ();
3987  tv_diff = tv_current - tv_old;
3988  tv_old = tv_current;
3989 
3991  "Time since last token %0.4f ms", ((float)tv_diff) / 1000000.0);
3992 #endif
3993 
3994  if (check_orf_token_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
3995  return (0);
3996  }
3997 
3998  if (instance->orf_token_discard) {
3999  return (0);
4000  }
4001 #ifdef TEST_DROP_ORF_TOKEN_PERCENTAGE
4002  if (random()%100 < TEST_DROP_ORF_TOKEN_PERCENTAGE) {
4003  return (0);
4004  }
4005 #endif
4006 
4007  if (endian_conversion_needed) {
4008  orf_token_endian_convert ((struct orf_token *)msg,
4009  (struct orf_token *)token_convert);
4010  msg = (struct orf_token *)token_convert;
4011  }
4012 
4013  /*
4014  * Make copy of token and retransmit list in case we have
4015  * to flush incoming messages from the kernel queue
4016  */
4017  token = (struct orf_token *)token_storage;
4018  memcpy (token, msg, sizeof (struct orf_token));
4019  memcpy (&token->rtr_list[0], (char *)msg + sizeof (struct orf_token),
4020  sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX);
4021 
4022 
4023  /*
4024  * Handle merge detection timeout
4025  */
4026  if (token->seq == instance->my_last_seq) {
4027  start_merge_detect_timeout (instance);
4028  instance->my_seq_unchanged += 1;
4029  } else {
4030  cancel_merge_detect_timeout (instance);
4031  cancel_token_hold_retransmit_timeout (instance);
4032  instance->my_seq_unchanged = 0;
4033  }
4034 
4035  instance->my_last_seq = token->seq;
4036 
4037 #ifdef TEST_RECOVERY_MSG_COUNT
4038  if (instance->memb_state == MEMB_STATE_OPERATIONAL && token->seq > TEST_RECOVERY_MSG_COUNT) {
4039  return (0);
4040  }
4041 #endif
4042  instance->flushing = 1;
4044  instance->flushing = 0;
4045 
4046  /*
4047  * Determine if we should hold (in reality drop) the token
4048  */
4049  instance->my_token_held = 0;
4050  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) &&
4051  instance->my_seq_unchanged > instance->totem_config->seqno_unchanged_const) {
4052  instance->my_token_held = 1;
4053  } else
4054  if (!totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) &&
4055  instance->my_seq_unchanged >= instance->totem_config->seqno_unchanged_const) {
4056  instance->my_token_held = 1;
4057  }
4058 
4059  /*
4060  * Hold onto token when there is no activity on ring and
4061  * this processor is the ring rep
4062  */
4063  forward_token = 1;
4064  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
4065  if (instance->my_token_held) {
4066  forward_token = 0;
4067  }
4068  }
4069 
4070  switch (instance->memb_state) {
4071  case MEMB_STATE_COMMIT:
4072  /* Discard token */
4073  break;
4074 
4076  messages_free (instance, token->aru);
4077  /*
4078  * Do NOT add break, this case should also execute code in gather case.
4079  */
4080 
4081  case MEMB_STATE_GATHER:
4082  /*
4083  * DO NOT add break, we use different free mechanism in recovery state
4084  */
4085 
4086  case MEMB_STATE_RECOVERY:
4087  /*
4088  * Discard tokens from another configuration
4089  */
4090  if (memcmp (&token->ring_id, &instance->my_ring_id,
4091  sizeof (struct memb_ring_id)) != 0) {
4092 
4093  if ((forward_token)
4094  && instance->use_heartbeat) {
4095  reset_heartbeat_timeout(instance);
4096  }
4097  else {
4098  cancel_heartbeat_timeout(instance);
4099  }
4100 
4101  return (0); /* discard token */
4102  }
4103 
4104  /*
4105  * Discard retransmitted tokens
4106  */
4107  if (sq_lte_compare (token->token_seq, instance->my_token_seq)) {
4108  return (0); /* discard token */
4109  }
4110 
4111  /*
4112  * Token is valid so trigger callbacks
4113  */
4114  token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_RECEIVED);
4115 
4116  last_aru = instance->my_last_aru;
4117  instance->my_last_aru = token->aru;
4118 
4119  transmits_allowed = fcc_calculate (instance, token);
4120  mcasted_retransmit = orf_token_rtr (instance, token, &transmits_allowed);
4121 
4123  instance->my_token_held == 1 &&
4124  (token->rtr_list_entries > 0 || mcasted_retransmit > 0)) {
4125  instance->my_token_held = 0;
4126  forward_token = 1;
4127  }
4128 
4129  fcc_rtr_limit (instance, token, &transmits_allowed);
4130  mcasted_regular = orf_token_mcast (instance, token, transmits_allowed);
4131 /*
4132 if (mcasted_regular) {
4133 printf ("mcasted regular %d\n", mcasted_regular);
4134 printf ("token seq %d\n", token->seq);
4135 }
4136 */
4137  fcc_token_update (instance, token, mcasted_retransmit +
4138  mcasted_regular);
4139 
4140  if (sq_lt_compare (instance->my_aru, token->aru) ||
4141  instance->my_id.addr[0].nodeid == token->aru_addr ||
4142  token->aru_addr == 0) {
4143 
4144  token->aru = instance->my_aru;
4145  if (token->aru == token->seq) {
4146  token->aru_addr = 0;
4147  } else {
4148  token->aru_addr = instance->my_id.addr[0].nodeid;
4149  }
4150  }
4151  if (token->aru == last_aru && token->aru_addr != 0) {
4152  instance->my_aru_count += 1;
4153  } else {
4154  instance->my_aru_count = 0;
4155  }
4156 
4157  /*
4158  * We really don't follow specification there. In specification, OTHER nodes
4159  * detect failure of one node (based on aru_count) and my_id IS NEVER added
4160  * to failed list (so node never mark itself as failed)
4161  */
4162  if (instance->my_aru_count > instance->totem_config->fail_to_recv_const &&
4163  token->aru_addr == instance->my_id.addr[0].nodeid) {
4164 
4166  "FAILED TO RECEIVE");
4167 
4168  instance->failed_to_recv = 1;
4169 
4170  memb_set_merge (&instance->my_id, 1,
4171  instance->my_failed_list,
4172  &instance->my_failed_list_entries);
4173 
4174  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_FAILED_TO_RECEIVE);
4175  } else {
4176  instance->my_token_seq = token->token_seq;
4177  token->token_seq += 1;
4178 
4179  if (instance->memb_state == MEMB_STATE_RECOVERY) {
4180  /*
4181  * instance->my_aru == instance->my_high_seq_received means this processor
4182  * has recovered all messages it can recover
4183  * (ie: its retrans queue is empty)
4184  */
4185  if (cs_queue_is_empty (&instance->retrans_message_queue) == 0) {
4186 
4187  if (token->retrans_flg == 0) {
4188  token->retrans_flg = 1;
4189  instance->my_set_retrans_flg = 1;
4190  }
4191  } else
4192  if (token->retrans_flg == 1 && instance->my_set_retrans_flg) {
4193  token->retrans_flg = 0;
4194  instance->my_set_retrans_flg = 0;
4195  }
4197  "token retrans flag is %d my set retrans flag%d retrans queue empty %d count %d, aru %x",
4198  token->retrans_flg, instance->my_set_retrans_flg,
4199  cs_queue_is_empty (&instance->retrans_message_queue),
4200  instance->my_retrans_flg_count, token->aru);
4201  if (token->retrans_flg == 0) {
4202  instance->my_retrans_flg_count += 1;
4203  } else {
4204  instance->my_retrans_flg_count = 0;
4205  }
4206  if (instance->my_retrans_flg_count == 2) {
4207  instance->my_install_seq = token->seq;
4208  }
4210  "install seq %x aru %x high seq received %x",
4211  instance->my_install_seq, instance->my_aru, instance->my_high_seq_received);
4212  if (instance->my_retrans_flg_count >= 2 &&
4213  instance->my_received_flg == 0 &&
4214  sq_lte_compare (instance->my_install_seq, instance->my_aru)) {
4215  instance->my_received_flg = 1;
4216  instance->my_deliver_memb_entries = instance->my_trans_memb_entries;
4217  memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list,
4218  sizeof (struct totem_ip_address) * instance->my_trans_memb_entries);
4219  }
4220  if (instance->my_retrans_flg_count >= 3 &&
4221  sq_lte_compare (instance->my_install_seq, token->aru)) {
4222  instance->my_rotation_counter += 1;
4223  } else {
4224  instance->my_rotation_counter = 0;
4225  }
4226  if (instance->my_rotation_counter == 2) {
4228  "retrans flag count %x token aru %x install seq %x aru %x %x",
4229  instance->my_retrans_flg_count, token->aru, instance->my_install_seq,
4230  instance->my_aru, token->seq);
4231 
4232  memb_state_operational_enter (instance);
4233  instance->my_rotation_counter = 0;
4234  instance->my_retrans_flg_count = 0;
4235  }
4236  }
4237 
4239  token_send (instance, token, forward_token);
4240 
4241 #ifdef GIVEINFO
4242  tv_current = qb_util_nano_current_get ();
4243  tv_diff = tv_current - tv_old;
4244  tv_old = tv_current;
4246  "I held %0.4f ms",
4247  ((float)tv_diff) / 1000000.0);
4248 #endif
4249  if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
4250  messages_deliver_to_app (instance, 0,
4251  instance->my_high_seq_received);
4252  }
4253 
4254  /*
4255  * Deliver messages after token has been transmitted
4256  * to improve performance
4257  */
4258  reset_token_timeout (instance); // REVIEWED
4259  reset_token_retransmit_timeout (instance); // REVIEWED
4260  if (totemip_equal(&instance->my_id.addr[0], &instance->my_ring_id.rep) &&
4261  instance->my_token_held == 1) {
4262 
4263  start_token_hold_retransmit_timeout (instance);
4264  }
4265 
4266  token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_SENT);
4267  }
4268  break;
4269  }
4270 
4271  if ((forward_token)
4272  && instance->use_heartbeat) {
4273  reset_heartbeat_timeout(instance);
4274  }
4275  else {
4276  cancel_heartbeat_timeout(instance);
4277  }
4278 
4279  return (0);
4280 }
4281 
4282 static void messages_deliver_to_app (
4283  struct totemsrp_instance *instance,
4284  int skip,
4285  unsigned int end_point)
4286 {
4287  struct sort_queue_item *sort_queue_item_p;
4288  unsigned int i;
4289  int res;
4290  struct mcast *mcast_in;
4291  struct mcast mcast_header;
4292  unsigned int range = 0;
4293  int endian_conversion_required;
4294  unsigned int my_high_delivered_stored = 0;
4295 
4296 
4297  range = end_point - instance->my_high_delivered;
4298 
4299  if (range) {
4301  "Delivering %x to %x", instance->my_high_delivered,
4302  end_point);
4303  }
4304  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
4305  my_high_delivered_stored = instance->my_high_delivered;
4306 
4307  /*
4308  * Deliver messages in order from rtr queue to pending delivery queue
4309  */
4310  for (i = 1; i <= range; i++) {
4311 
4312  void *ptr = 0;
4313 
4314  /*
4315  * If out of range of sort queue, stop assembly
4316  */
4317  res = sq_in_range (&instance->regular_sort_queue,
4318  my_high_delivered_stored + i);
4319  if (res == 0) {
4320  break;
4321  }
4322 
4323  res = sq_item_get (&instance->regular_sort_queue,
4324  my_high_delivered_stored + i, &ptr);
4325  /*
4326  * If hole, stop assembly
4327  */
4328  if (res != 0 && skip == 0) {
4329  break;
4330  }
4331 
4332  instance->my_high_delivered = my_high_delivered_stored + i;
4333 
4334  if (res != 0) {
4335  continue;
4336 
4337  }
4338 
4339  sort_queue_item_p = ptr;
4340 
4341  mcast_in = sort_queue_item_p->mcast;
4342  assert (mcast_in != (struct mcast *)0xdeadbeef);
4343 
4344  endian_conversion_required = 0;
4345  if (mcast_in->header.endian_detector != ENDIAN_LOCAL) {
4346  endian_conversion_required = 1;
4347  mcast_endian_convert (mcast_in, &mcast_header);
4348  } else {
4349  memcpy (&mcast_header, mcast_in, sizeof (struct mcast));
4350  }
4351 
4352  /*
4353  * Skip messages not originated in instance->my_deliver_memb
4354  */
4355  if (skip &&
4356  memb_set_subset (&mcast_header.system_from,
4357  1,
4358  instance->my_deliver_memb_list,
4359  instance->my_deliver_memb_entries) == 0) {
4360 
4361  instance->my_high_delivered = my_high_delivered_stored + i;
4362 
4363  continue;
4364  }
4365 
4366  /*
4367  * Message found
4368  */
4370  "Delivering MCAST message with seq %x to pending delivery queue",
4371  mcast_header.seq);
4372 
4373  /*
4374  * Message is locally originated multicast
4375  */
4376  instance->totemsrp_deliver_fn (
4377  mcast_header.header.nodeid,
4378  ((char *)sort_queue_item_p->mcast) + sizeof (struct mcast),
4379  sort_queue_item_p->msg_len - sizeof (struct mcast),
4380  endian_conversion_required);
4381  }
4382 }
4383 
4384 /*
4385  * recv message handler called when MCAST message type received
4386  */
4387 static int message_handler_mcast (
4388  struct totemsrp_instance *instance,
4389  const void *msg,
4390  size_t msg_len,
4391  int endian_conversion_needed)
4392 {
4393  struct sort_queue_item sort_queue_item;
4394  struct sq *sort_queue;
4395  struct mcast mcast_header;
4396 
4397  if (check_mcast_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4398  return (0);
4399  }
4400 
4401  if (endian_conversion_needed) {
4402  mcast_endian_convert (msg, &mcast_header);
4403  } else {
4404  memcpy (&mcast_header, msg, sizeof (struct mcast));
4405  }
4406 
4407  if (mcast_header.header.encapsulated == MESSAGE_ENCAPSULATED) {
4408  sort_queue = &instance->recovery_sort_queue;
4409  } else {
4410  sort_queue = &instance->regular_sort_queue;
4411  }
4412 
4413  assert (msg_len <= FRAME_SIZE_MAX);
4414 
4415 #ifdef TEST_DROP_MCAST_PERCENTAGE
4416  if (random()%100 < TEST_DROP_MCAST_PERCENTAGE) {
4417  return (0);
4418  }
4419 #endif
4420 
4421  /*
4422  * If the message is foreign execute the switch below
4423  */
4424  if (memcmp (&instance->my_ring_id, &mcast_header.ring_id,
4425  sizeof (struct memb_ring_id)) != 0) {
4426 
4427  switch (instance->memb_state) {
4429  memb_set_merge (
4430  &mcast_header.system_from, 1,
4431  instance->my_proc_list, &instance->my_proc_list_entries);
4432  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_FOREIGN_MESSAGE_IN_OPERATIONAL_STATE);
4433  break;
4434 
4435  case MEMB_STATE_GATHER:
4436  if (!memb_set_subset (
4437  &mcast_header.system_from,
4438  1,
4439  instance->my_proc_list,
4440  instance->my_proc_list_entries)) {
4441 
4442  memb_set_merge (&mcast_header.system_from, 1,
4443  instance->my_proc_list, &instance->my_proc_list_entries);
4444  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_FOREIGN_MESSAGE_IN_GATHER_STATE);
4445  return (0);
4446  }
4447  break;
4448 
4449  case MEMB_STATE_COMMIT:
4450  /* discard message */
4451  instance->stats.rx_msg_dropped++;
4452  break;
4453 
4454  case MEMB_STATE_RECOVERY:
4455  /* discard message */
4456  instance->stats.rx_msg_dropped++;
4457  break;
4458  }
4459  return (0);
4460  }
4461 
4463  "Received ringid(%s:%lld) seq %x",
4464  totemip_print (&mcast_header.ring_id.rep),
4465  mcast_header.ring_id.seq,
4466  mcast_header.seq);
4467 
4468  /*
4469  * Add mcast message to rtr queue if not already in rtr queue
4470  * otherwise free io vectors
4471  */
4472  if (msg_len > 0 && msg_len <= FRAME_SIZE_MAX &&
4473  sq_in_range (sort_queue, mcast_header.seq) &&
4474  sq_item_inuse (sort_queue, mcast_header.seq) == 0) {
4475 
4476  /*
4477  * Allocate new multicast memory block
4478  */
4479 // TODO LEAK
4480  sort_queue_item.mcast = totemsrp_buffer_alloc (instance);
4481  if (sort_queue_item.mcast == NULL) {
4482  return (-1); /* error here is corrected by the algorithm */
4483  }
4484  memcpy (sort_queue_item.mcast, msg, msg_len);
4485  sort_queue_item.msg_len = msg_len;
4486 
4487  if (sq_lt_compare (instance->my_high_seq_received,
4488  mcast_header.seq)) {
4489  instance->my_high_seq_received = mcast_header.seq;
4490  }
4491 
4492  sq_item_add (sort_queue, &sort_queue_item, mcast_header.seq);
4493  }
4494 
4495  update_aru (instance);
4496  if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
4497  messages_deliver_to_app (instance, 0, instance->my_high_seq_received);
4498  }
4499 
4500 /* TODO remove from retrans message queue for old ring in recovery state */
4501  return (0);
4502 }
4503 
4504 static int message_handler_memb_merge_detect (
4505  struct totemsrp_instance *instance,
4506  const void *msg,
4507  size_t msg_len,
4508  int endian_conversion_needed)
4509 {
4511 
4512  if (check_memb_merge_detect_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4513  return (0);
4514  }
4515 
4516  if (endian_conversion_needed) {
4517  memb_merge_detect_endian_convert (msg, &memb_merge_detect);
4518  } else {
4519  memcpy (&memb_merge_detect, msg,
4520  sizeof (struct memb_merge_detect));
4521  }
4522 
4523  /*
4524  * do nothing if this is a merge detect from this configuration
4525  */
4526  if (memcmp (&instance->my_ring_id, &memb_merge_detect.ring_id,
4527  sizeof (struct memb_ring_id)) == 0) {
4528 
4529  return (0);
4530  }
4531 
4532  /*
4533  * Execute merge operation
4534  */
4535  switch (instance->memb_state) {
4537  memb_set_merge (&memb_merge_detect.system_from, 1,
4538  instance->my_proc_list, &instance->my_proc_list_entries);
4539  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_MERGE_DURING_OPERATIONAL_STATE);
4540  break;
4541 
4542  case MEMB_STATE_GATHER:
4543  if (!memb_set_subset (
4545  1,
4546  instance->my_proc_list,
4547  instance->my_proc_list_entries)) {
4548 
4549  memb_set_merge (&memb_merge_detect.system_from, 1,
4550  instance->my_proc_list, &instance->my_proc_list_entries);
4551  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_MERGE_DURING_GATHER_STATE);
4552  return (0);
4553  }
4554  break;
4555 
4556  case MEMB_STATE_COMMIT:
4557  /* do nothing in commit */
4558  break;
4559 
4560  case MEMB_STATE_RECOVERY:
4561  /* do nothing in recovery */
4562  break;
4563  }
4564  return (0);
4565 }
4566 
4567 static void memb_join_process (
4568  struct totemsrp_instance *instance,
4569  const struct memb_join *memb_join)
4570 {
4571  struct srp_addr *proc_list;
4572  struct srp_addr *failed_list;
4573  int gather_entered = 0;
4574  int fail_minus_memb_entries = 0;
4575  struct srp_addr fail_minus_memb[PROCESSOR_COUNT_MAX];
4576 
4577  proc_list = (struct srp_addr *)memb_join->end_of_memb_join;
4578  failed_list = proc_list + memb_join->proc_list_entries;
4579 
4580 /*
4581  memb_set_print ("proclist", proc_list, memb_join->proc_list_entries);
4582  memb_set_print ("faillist", failed_list, memb_join->failed_list_entries);
4583  memb_set_print ("my_proclist", instance->my_proc_list, instance->my_proc_list_entries);
4584  memb_set_print ("my_faillist", instance->my_failed_list, instance->my_failed_list_entries);
4585 -*/
4586 
4587  if (memb_join->header.type == MESSAGE_TYPE_MEMB_JOIN) {
4588  if (instance->flushing) {
4589  if (memb_join->header.nodeid == LEAVE_DUMMY_NODEID) {
4591  "Discarding LEAVE message during flush, nodeid=%u",
4592  memb_join->failed_list_entries > 0 ? failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid : LEAVE_DUMMY_NODEID);
4593  if (memb_join->failed_list_entries > 0) {
4594  my_leave_memb_set(instance, failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid);
4595  }
4596  } else {
4598  "Discarding JOIN message during flush, nodeid=%d", memb_join->header.nodeid);
4599  }
4600  return;
4601  } else {
4602  if (memb_join->header.nodeid == LEAVE_DUMMY_NODEID) {
4604  "Received LEAVE message from %u", memb_join->failed_list_entries > 0 ? failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid : LEAVE_DUMMY_NODEID);
4605  if (memb_join->failed_list_entries > 0) {
4606  my_leave_memb_set(instance, failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid);
4607  }
4608  }
4609  }
4610 
4611  }
4612 
4613  if (memb_set_equal (proc_list,
4614  memb_join->proc_list_entries,
4615  instance->my_proc_list,
4616  instance->my_proc_list_entries) &&
4617 
4618  memb_set_equal (failed_list,
4619  memb_join->failed_list_entries,
4620  instance->my_failed_list,
4621  instance->my_failed_list_entries)) {
4622 
4623  memb_consensus_set (instance, &memb_join->system_from);
4624 
4625  if (memb_consensus_agreed (instance) && instance->failed_to_recv == 1) {
4626  instance->failed_to_recv = 0;
4627  srp_addr_copy (&instance->my_proc_list[0],
4628  &instance->my_id);
4629  instance->my_proc_list_entries = 1;
4630  instance->my_failed_list_entries = 0;
4631 
4632  memb_state_commit_token_create (instance);
4633 
4634  memb_state_commit_enter (instance);
4635  return;
4636  }
4637  if (memb_consensus_agreed (instance) &&
4638  memb_lowest_in_config (instance)) {
4639 
4640  memb_state_commit_token_create (instance);
4641 
4642  memb_state_commit_enter (instance);
4643  } else {
4644  goto out;
4645  }
4646  } else
4647  if (memb_set_subset (proc_list,
4648  memb_join->proc_list_entries,
4649  instance->my_proc_list,
4650  instance->my_proc_list_entries) &&
4651 
4652  memb_set_subset (failed_list,
4653  memb_join->failed_list_entries,
4654  instance->my_failed_list,
4655  instance->my_failed_list_entries)) {
4656 
4657  goto out;
4658  } else
4659  if (memb_set_subset (&memb_join->system_from, 1,
4660  instance->my_failed_list, instance->my_failed_list_entries)) {
4661 
4662  goto out;
4663  } else {
4664  memb_set_merge (proc_list,
4665  memb_join->proc_list_entries,
4666  instance->my_proc_list, &instance->my_proc_list_entries);
4667 
4668  if (memb_set_subset (
4669  &instance->my_id, 1,
4670  failed_list, memb_join->failed_list_entries)) {
4671 
4672  memb_set_merge (
4673  &memb_join->system_from, 1,
4674  instance->my_failed_list, &instance->my_failed_list_entries);
4675  } else {
4676  if (memb_set_subset (
4677  &memb_join->system_from, 1,
4678  instance->my_memb_list,
4679  instance->my_memb_entries)) {
4680 
4681  if (memb_set_subset (
4682  &memb_join->system_from, 1,
4683  instance->my_failed_list,
4684  instance->my_failed_list_entries) == 0) {
4685 
4686  memb_set_merge (failed_list,
4687  memb_join->failed_list_entries,
4688  instance->my_failed_list, &instance->my_failed_list_entries);
4689  } else {
4690  memb_set_subtract (fail_minus_memb,
4691  &fail_minus_memb_entries,
4692  failed_list,
4693  memb_join->failed_list_entries,
4694  instance->my_memb_list,
4695  instance->my_memb_entries);
4696 
4697  memb_set_merge (fail_minus_memb,
4698  fail_minus_memb_entries,
4699  instance->my_failed_list,
4700  &instance->my_failed_list_entries);
4701  }
4702  }
4703  }
4704  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_MERGE_DURING_JOIN);
4705  gather_entered = 1;
4706  }
4707 
4708 out:
4709  if (gather_entered == 0 &&
4710  instance->memb_state == MEMB_STATE_OPERATIONAL) {
4711 
4712  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_JOIN_DURING_OPERATIONAL_STATE);
4713  }
4714 }
4715 
4716 static void memb_join_endian_convert (const struct memb_join *in, struct memb_join *out)
4717 {
4718  int i;
4719  struct srp_addr *in_proc_list;
4720  struct srp_addr *in_failed_list;
4721  struct srp_addr *out_proc_list;
4722  struct srp_addr *out_failed_list;
4723 
4724  out->header.type = in->header.type;
4726  out->header.nodeid = swab32 (in->header.nodeid);
4727  srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
4730  out->ring_seq = swab64 (in->ring_seq);
4731 
4732  in_proc_list = (struct srp_addr *)in->end_of_memb_join;
4733  in_failed_list = in_proc_list + out->proc_list_entries;
4734  out_proc_list = (struct srp_addr *)out->end_of_memb_join;
4735  out_failed_list = out_proc_list + out->proc_list_entries;
4736 
4737  for (i = 0; i < out->proc_list_entries; i++) {
4738  srp_addr_copy_endian_convert (&out_proc_list[i], &in_proc_list[i]);
4739  }
4740  for (i = 0; i < out->failed_list_entries; i++) {
4741  srp_addr_copy_endian_convert (&out_failed_list[i], &in_failed_list[i]);
4742  }
4743 }
4744 
4745 static void memb_commit_token_endian_convert (const struct memb_commit_token *in, struct memb_commit_token *out)
4746 {
4747  int i;
4748  struct srp_addr *in_addr = (struct srp_addr *)in->end_of_commit_token;
4749  struct srp_addr *out_addr = (struct srp_addr *)out->end_of_commit_token;
4750  struct memb_commit_token_memb_entry *in_memb_list;
4751  struct memb_commit_token_memb_entry *out_memb_list;
4752 
4753  out->header.type = in->header.type;
4755  out->header.nodeid = swab32 (in->header.nodeid);
4756  out->token_seq = swab32 (in->token_seq);
4758  out->ring_id.seq = swab64 (in->ring_id.seq);
4759  out->retrans_flg = swab32 (in->retrans_flg);
4760  out->memb_index = swab32 (in->memb_index);
4761  out->addr_entries = swab32 (in->addr_entries);
4762 
4763  in_memb_list = (struct memb_commit_token_memb_entry *)(in_addr + out->addr_entries);
4764  out_memb_list = (struct memb_commit_token_memb_entry *)(out_addr + out->addr_entries);
4765  for (i = 0; i < out->addr_entries; i++) {
4766  srp_addr_copy_endian_convert (&out_addr[i], &in_addr[i]);
4767 
4768  /*
4769  * Only convert the memb entry if it has been set
4770  */
4771  if (in_memb_list[i].ring_id.rep.family != 0) {
4772  totemip_copy_endian_convert (&out_memb_list[i].ring_id.rep,
4773  &in_memb_list[i].ring_id.rep);
4774 
4775  out_memb_list[i].ring_id.seq =
4776  swab64 (in_memb_list[i].ring_id.seq);
4777  out_memb_list[i].aru = swab32 (in_memb_list[i].aru);
4778  out_memb_list[i].high_delivered = swab32 (in_memb_list[i].high_delivered);
4779  out_memb_list[i].received_flg = swab32 (in_memb_list[i].received_flg);
4780  }
4781  }
4782 }
4783 
4784 static void orf_token_endian_convert (const struct orf_token *in, struct orf_token *out)
4785 {
4786  int i;
4787 
4788  out->header.type = in->header.type;
4790  out->header.nodeid = swab32 (in->header.nodeid);
4791  out->seq = swab32 (in->seq);
4792  out->token_seq = swab32 (in->token_seq);
4793  out->aru = swab32 (in->aru);
4795  out->aru_addr = swab32(in->aru_addr);
4796  out->ring_id.seq = swab64 (in->ring_id.seq);
4797  out->fcc = swab32 (in->fcc);
4798  out->backlog = swab32 (in->backlog);
4799  out->retrans_flg = swab32 (in->retrans_flg);
4801  for (i = 0; i < out->rtr_list_entries; i++) {
4803  out->rtr_list[i].ring_id.seq = swab64 (in->rtr_list[i].ring_id.seq);
4804  out->rtr_list[i].seq = swab32 (in->rtr_list[i].seq);
4805  }
4806 }
4807 
4808 static void mcast_endian_convert (const struct mcast *in, struct mcast *out)
4809 {
4810  out->header.type = in->header.type;
4812  out->header.nodeid = swab32 (in->header.nodeid);
4814 
4815  out->seq = swab32 (in->seq);
4816  out->this_seqno = swab32 (in->this_seqno);
4818  out->ring_id.seq = swab64 (in->ring_id.seq);
4819  out->node_id = swab32 (in->node_id);
4820  out->guarantee = swab32 (in->guarantee);
4821  srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
4822 }
4823 
4824 static void memb_merge_detect_endian_convert (
4825  const struct memb_merge_detect *in,
4826  struct memb_merge_detect *out)
4827 {
4828  out->header.type = in->header.type;
4830  out->header.nodeid = swab32 (in->header.nodeid);
4832  out->ring_id.seq = swab64 (in->ring_id.seq);
4833  srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
4834 }
4835 
4836 static int ignore_join_under_operational (
4837  struct totemsrp_instance *instance,
4838  const struct memb_join *memb_join)
4839 {
4840  struct srp_addr *proc_list;
4841  struct srp_addr *failed_list;
4842  unsigned long long ring_seq;
4843 
4844  proc_list = (struct srp_addr *)memb_join->end_of_memb_join;
4845  failed_list = proc_list + memb_join->proc_list_entries;
4846  ring_seq = memb_join->ring_seq;
4847 
4848  if (memb_set_subset (&instance->my_id, 1,
4849  failed_list, memb_join->failed_list_entries)) {
4850  return (1);
4851  }
4852 
4853  /*
4854  * In operational state, my_proc_list is exactly the same as
4855  * my_memb_list.
4856  */
4857  if ((memb_set_subset (&memb_join->system_from, 1,
4858  instance->my_memb_list, instance->my_memb_entries)) &&
4859  (ring_seq < instance->my_ring_id.seq)) {
4860  return (1);
4861  }
4862 
4863  return (0);
4864 }
4865 
4866 static int message_handler_memb_join (
4867  struct totemsrp_instance *instance,
4868  const void *msg,
4869  size_t msg_len,
4870  int endian_conversion_needed)
4871 {
4872  const struct memb_join *memb_join;
4873  struct memb_join *memb_join_convert = alloca (msg_len);
4874 
4875  if (check_memb_join_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4876  return (0);
4877  }
4878 
4879  if (endian_conversion_needed) {
4880  memb_join = memb_join_convert;
4881  memb_join_endian_convert (msg, memb_join_convert);
4882 
4883  } else {
4884  memb_join = msg;
4885  }
4886  /*
4887  * If the process paused because it wasn't scheduled in a timely
4888  * fashion, flush the join messages because they may be queued
4889  * entries
4890  */
4891  if (pause_flush (instance)) {
4892  return (0);
4893  }
4894 
4895  if (instance->token_ring_id_seq < memb_join->ring_seq) {
4896  instance->token_ring_id_seq = memb_join->ring_seq;
4897  }
4898  switch (instance->memb_state) {
4900  if (!ignore_join_under_operational (instance, memb_join)) {
4901  memb_join_process (instance, memb_join);
4902  }
4903  break;
4904 
4905  case MEMB_STATE_GATHER:
4906  memb_join_process (instance, memb_join);
4907  break;
4908 
4909  case MEMB_STATE_COMMIT:
4910  if (memb_set_subset (&memb_join->system_from,
4911  1,
4912  instance->my_new_memb_list,
4913  instance->my_new_memb_entries) &&
4914 
4915  memb_join->ring_seq >= instance->my_ring_id.seq) {
4916 
4917  memb_join_process (instance, memb_join);
4918  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_JOIN_DURING_COMMIT_STATE);
4919  }
4920  break;
4921 
4922  case MEMB_STATE_RECOVERY:
4923  if (memb_set_subset (&memb_join->system_from,
4924  1,
4925  instance->my_new_memb_list,
4926  instance->my_new_memb_entries) &&
4927 
4928  memb_join->ring_seq >= instance->my_ring_id.seq) {
4929 
4930  memb_join_process (instance, memb_join);
4931  memb_recovery_state_token_loss (instance);
4932  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_JOIN_DURING_RECOVERY);
4933  }
4934  break;
4935  }
4936  return (0);
4937 }
4938 
4939 static int message_handler_memb_commit_token (
4940  struct totemsrp_instance *instance,
4941  const void *msg,
4942  size_t msg_len,
4943  int endian_conversion_needed)
4944 {
4945  struct memb_commit_token *memb_commit_token_convert = alloca (msg_len);
4947  struct srp_addr sub[PROCESSOR_COUNT_MAX];
4948  int sub_entries;
4949 
4950  struct srp_addr *addr;
4951 
4953  "got commit token");
4954 
4955  if (check_memb_commit_token_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4956  return (0);
4957  }
4958 
4959  if (endian_conversion_needed) {
4960  memb_commit_token_endian_convert (msg, memb_commit_token_convert);
4961  } else {
4962  memcpy (memb_commit_token_convert, msg, msg_len);
4963  }
4964  memb_commit_token = memb_commit_token_convert;
4965  addr = (struct srp_addr *)memb_commit_token->end_of_commit_token;
4966 
4967 #ifdef TEST_DROP_COMMIT_TOKEN_PERCENTAGE
4968  if (random()%100 < TEST_DROP_COMMIT_TOKEN_PERCENTAGE) {
4969  return (0);
4970  }
4971 #endif
4972  switch (instance->memb_state) {
4974  /* discard token */
4975  break;
4976 
4977  case MEMB_STATE_GATHER:
4978  memb_set_subtract (sub, &sub_entries,
4979  instance->my_proc_list, instance->my_proc_list_entries,
4980  instance->my_failed_list, instance->my_failed_list_entries);
4981 
4982  if (memb_set_equal (addr,
4983  memb_commit_token->addr_entries,
4984  sub,
4985  sub_entries) &&
4986 
4987  memb_commit_token->ring_id.seq > instance->my_ring_id.seq) {
4988  memcpy (instance->commit_token, memb_commit_token, msg_len);
4989  memb_state_commit_enter (instance);
4990  }
4991  break;
4992 
4993  case MEMB_STATE_COMMIT:
4994  /*
4995  * If retransmitted commit tokens are sent on this ring
4996  * filter them out and only enter recovery once the
4997  * commit token has traversed the array. This is
4998  * determined by :
4999  * memb_commit_token->memb_index == memb_commit_token->addr_entries) {
5000  */
5001  if (memb_commit_token->ring_id.seq == instance->my_ring_id.seq &&
5002  memb_commit_token->memb_index == memb_commit_token->addr_entries) {
5003  memb_state_recovery_enter (instance, memb_commit_token);
5004  }
5005  break;
5006 
5007  case MEMB_STATE_RECOVERY:
5008  if (totemip_equal (&instance->my_id.addr[0], &instance->my_ring_id.rep)) {
5009 
5010  /* Filter out duplicated tokens */
5011  if (instance->originated_orf_token) {
5012  break;
5013  }
5014 
5015  instance->originated_orf_token = 1;
5016 
5018  "Sending initial ORF token");
5019 
5020  // TODO convert instead of initiate
5021  orf_token_send_initial (instance);
5022  reset_token_timeout (instance); // REVIEWED
5023  reset_token_retransmit_timeout (instance); // REVIEWED
5024  }
5025  break;
5026  }
5027  return (0);
5028 }
5029 
5030 static int message_handler_token_hold_cancel (
5031  struct totemsrp_instance *instance,
5032  const void *msg,
5033  size_t msg_len,
5034  int endian_conversion_needed)
5035 {
5036  const struct token_hold_cancel *token_hold_cancel = msg;
5037 
5038  if (check_token_hold_cancel_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
5039  return (0);
5040  }
5041 
5042  if (memcmp (&token_hold_cancel->ring_id, &instance->my_ring_id,
5043  sizeof (struct memb_ring_id)) == 0) {
5044 
5045  instance->my_seq_unchanged = 0;
5046  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
5047  timer_function_token_retransmit_timeout (instance);
5048  }
5049  }
5050  return (0);
5051 }
5052 
5054  void *context,
5055  const void *msg,
5056  unsigned int msg_len)
5057 {
5058  struct totemsrp_instance *instance = context;
5059  const struct message_header *message_header = msg;
5060 
5061  if (msg_len < sizeof (struct message_header)) {
5063  "Received message is too short... ignoring %u.",
5064  (unsigned int)msg_len);
5065  return;
5066  }
5067 
5068 
5069  switch (message_header->type) {
5071  instance->stats.orf_token_rx++;
5072  break;
5073  case MESSAGE_TYPE_MCAST:
5074  instance->stats.mcast_rx++;
5075  break;
5077  instance->stats.memb_merge_detect_rx++;
5078  break;
5080  instance->stats.memb_join_rx++;
5081  break;
5083  instance->stats.memb_commit_token_rx++;
5084  break;
5086  instance->stats.token_hold_cancel_rx++;
5087  break;
5088  default:
5089  log_printf (instance->totemsrp_log_level_security, "Type of received message is wrong... ignoring %d.\n", (int)message_header->type);
5090 printf ("wrong message type\n");
5091  instance->stats.rx_msg_dropped++;
5092  return;
5093  }
5094  /*
5095  * Handle incoming message
5096  */
5097  totemsrp_message_handlers.handler_functions[(int)message_header->type] (
5098  instance,
5099  msg,
5100  msg_len,
5101  message_header->endian_detector != ENDIAN_LOCAL);
5102 }
5103 
5105  void *context,
5106  const struct totem_ip_address *iface_addr,
5107  unsigned int iface_no)
5108 {
5109  struct totemsrp_instance *instance = context;
5110  int i;
5111 
5112  totemip_copy (&instance->my_id.addr[iface_no], iface_addr);
5113  assert (instance->my_id.addr[iface_no].nodeid);
5114 
5115  totemip_copy (&instance->my_memb_list[0].addr[iface_no], iface_addr);
5116 
5117  if (instance->iface_changes++ == 0) {
5118  instance->memb_ring_id_create_or_load (&instance->my_ring_id,
5119  &instance->my_id.addr[0]);
5120  /*
5121  * Increase the ring_id sequence number. This doesn't follow specification.
5122  * Solves problem with restarted leader node (node with lowest nodeid) before
5123  * rest of the cluster forms new membership and guarantees unique ring_id for
5124  * new singleton configuration.
5125  */
5126  instance->my_ring_id.seq++;
5127 
5128  instance->token_ring_id_seq = instance->my_ring_id.seq;
5129  log_printf (
5130  instance->totemsrp_log_level_debug,
5131  "Created or loaded sequence id %llx.%s for this ring.",
5132  instance->my_ring_id.seq,
5133  totemip_print (&instance->my_ring_id.rep));
5134 
5135  if (instance->totemsrp_service_ready_fn) {
5136  instance->totemsrp_service_ready_fn ();
5137  }
5138 
5139  }
5140 
5141  for (i = 0; i < instance->totem_config->interfaces[iface_no].member_count; i++) {
5142  totemsrp_member_add (instance,
5143  &instance->totem_config->interfaces[iface_no].member_list[i],
5144  iface_no);
5145  }
5146 
5147  if (instance->iface_changes >= instance->totem_config->interface_count) {
5148  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_INTERFACE_CHANGE);
5149  }
5150 }
5151 
5152 void totemsrp_net_mtu_adjust (struct totem_config *totem_config) {
5153  totem_config->net_mtu -= 2 * sizeof (struct mcast);
5154 }
5155 
5157  void *context,
5158  void (*totem_service_ready) (void))
5159 {
5160  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5161 
5162  instance->totemsrp_service_ready_fn = totem_service_ready;
5163 }
5164 
5166  void *context,
5167  const struct totem_ip_address *member,
5168  int ring_no)
5169 {
5170  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5171  int res;
5172 
5173  res = totemrrp_member_add (instance->totemrrp_context, member, ring_no);
5174 
5175  return (res);
5176 }
5177 
5179  void *context,
5180  const struct totem_ip_address *member,
5181  int ring_no)
5182 {
5183  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5184  int res;
5185 
5186  res = totemrrp_member_remove (instance->totemrrp_context, member, ring_no);
5187 
5188  return (res);
5189 }
5190 
5191 void totemsrp_threaded_mode_enable (void *context)
5192 {
5193  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5194 
5195  instance->threaded_mode_enabled = 1;
5196 }
5197 
5198 void totemsrp_trans_ack (void *context)
5199 {
5200  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5201 
5202  instance->waiting_trans_ack = 0;
5203  instance->totemsrp_waiting_trans_ack_cb_fn (0);
5204 }
void(* totemsrp_service_ready_fn)(void)
Definition: totemsrp.c:462
unsigned int backlog
Definition: totemsrp.c:208
void(* totemsrp_deliver_fn)(unsigned int nodeid, const void *msg, unsigned int msg_len, int endian_conversion_required)
Definition: totemsrp.c:449
void(*) enum memb_stat memb_state)
Definition: totemsrp.c:441
uint8_t no_addrs
Definition: totemrrp.h:59
unsigned short family
Definition: coroapi.h:113
gather_state_from
Definition: totemsrp.c:537
int totemrrp_iface_check(void *rrp_context)
Definition: totemrrp.c:2277
void main_iface_change_fn(void *context, const struct totem_ip_address *iface_address, unsigned int iface_no)
Definition: totemsrp.c:5104
void totemip_copy_endian_convert(struct totem_ip_address *addr1, const struct totem_ip_address *addr2)
Definition: totemip.c:129
struct srp_addr system_from
Definition: totemsrp.c:218
#define ENDIAN_LOCAL
Definition: totemsrp.c:137
uint64_t gather_entered
Definition: totem.h:267
struct memb_ring_id ring_id
Definition: totemsrp.c:196
struct list_head list
Definition: totemsrp.c:163
uint32_t waiting_trans_ack
Definition: totemsrp.c:519
struct srp_addr system_from
Definition: totemsrp.c:186
struct memb_ring_id ring_id
Definition: totemsrp.c:255
int totemsrp_log_level_debug
Definition: totemsrp.c:427
struct memb_ring_id my_ring_id
Definition: totemsrp.c:337
Totem Single Ring Protocol.
uint64_t memb_commit_token_rx
Definition: totem.h:262
int my_leave_memb_entries
Definition: totemsrp.c:335
struct message_header header
Definition: totemsrp.c:185
unsigned int old_ring_state_high_seq_received
Definition: totemsrp.c:489
unsigned int proc_list_entries
Definition: totemsrp.c:219
uint32_t value
struct totem_interface * interfaces
Definition: totem.h:117
unsigned int interface_count
Definition: totem.h:118
int totemsrp_my_family_get(void *srp_context)
Definition: totemsrp.c:1149
struct list_head * next
Definition: list.h:47
uint64_t memb_join_tx
Definition: totem.h:256
void(* totemsrp_confchg_fn)(enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id)
Definition: totemsrp.c:455
The totem_ip_address struct.
Definition: coroapi.h:111
unsigned int seq
Definition: totemsrp.c:62
totemsrp_token_stats_t token[TOTEM_TOKEN_STATS_MAX]
Definition: totem.h:281
const char * totemip_print(const struct totem_ip_address *addr)
Definition: totemip.c:264
unsigned char addr[TOTEMIP_ADDRLEN]
Definition: coroapi.h:114
int totemsrp_log_level_error
Definition: totemsrp.c:421
int old_ring_state_aru
Definition: totemsrp.c:487
#define LEAVE_DUMMY_NODEID
Definition: totemsrp.c:102
unsigned int seq
Definition: totemsrp.c:203
struct memb_ring_id ring_id
Definition: totemsrp.c:245
int fcc_remcast_current
Definition: totemsrp.c:297
qb_loop_timer_handle timer_heartbeat_timeout
Definition: totemsrp.c:414
unsigned int failed_list_entries
Definition: totemsrp.c:220
uint64_t mcast_rx
Definition: totem.h:260
unsigned long long int tv_old
Definition: totemsrp.c:3963
#define SEQNO_START_TOKEN
Definition: totemsrp.c:115
void(* memb_ring_id_store)(const struct memb_ring_id *memb_ring_id, const struct totem_ip_address *addr)
Definition: totemsrp.c:471
unsigned int token_hold_timeout
Definition: totem.h:136
int member_count
Definition: totem.h:73
unsigned int msg_len
Definition: totemsrp.c:270
struct memb_ring_id ring_id
Definition: totemsrp.c:207
struct totem_ip_address member_list[PROCESSOR_COUNT_MAX]
Definition: totem.h:74
int totemip_compare(const void *a, const void *b)
Definition: totemip.c:158
int totemsrp_member_add(void *context, const struct totem_ip_address *member, int ring_no)
Definition: totemsrp.c:5165
void * token_sent_event_handle
Definition: totemsrp.c:524
struct timeval tv_old
Definition: totemsrp.c:493
int retrans_flg
Definition: totemsrp.c:210
struct srp_addr system_from
Definition: totemsrp.c:233
int my_new_memb_entries
Definition: totemsrp.c:325
totem_configuration_type
The totem_configuration_type enum.
Definition: coroapi.h:132
int addr_entries
Definition: totemsrp.c:65
int totemsrp_log_level_notice
Definition: totemsrp.c:425
unsigned int proc_list_entries
Definition: totemsrp.c:62
unsigned int totemsrp_my_nodeid_get(void *srp_context)
Definition: totemsrp.c:1138
unsigned int my_pbl
Definition: totemsrp.c:503
char rrp_mode[TOTEM_RRP_MODE_BYTES]
Definition: totem.h:164
void totemsrp_net_mtu_adjust(struct totem_config *totem_config)
Definition: totemsrp.c:5152
int totemsrp_log_level_warning
Definition: totemsrp.c:423
int totemsrp_crypto_set(void *srp_context, const char *cipher_type, const char *hash_type)
Definition: totemsrp.c:1124
void totemrrp_membership_changed(void *rrp_context, enum totem_configuration_type configuration_type, const struct srp_addr *member_list, size_t member_list_entries, const struct srp_addr *left_list, size_t left_list_entries, const struct srp_addr *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id)
Definition: totemrrp.c:2380
unsigned int my_aru
Definition: totemsrp.c:381
uint64_t memb_merge_detect_rx
Definition: totem.h:255
int totemsrp_ifaces_get(void *srp_context, unsigned int nodeid, struct totem_ip_address *interfaces, unsigned int interfaces_size, char ***status, unsigned int *iface_count)
Definition: totemsrp.c:1066
int guarantee
Definition: totemsrp.c:66
struct cs_queue new_message_queue_trans
Definition: totemsrp.c:370
struct message_header header
Definition: totemsrp.c:232
unsigned char end_of_commit_token[0]
Definition: totemsrp.c:259
unsigned int seq
Definition: totemsrp.c:187
unsigned char addr[TOTEMIP_ADDRLEN]
Definition: coroapi.h:77
char commit_token_storage[40000]
Definition: totemsrp.c:525
unsigned int rrp_problem_count_timeout
Definition: totem.h:156
struct list_head token_callback_sent_listhead
Definition: totemsrp.c:387
The sq struct.
Definition: sq.h:43
unsigned int set_aru
Definition: totemsrp.c:483
struct cs_queue new_message_queue
Definition: totemsrp.c:368
int my_rotation_counter
Definition: totemsrp.c:355
int earliest_token
Definition: totem.h:278
struct srp_addr my_deliver_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:315
uint64_t orf_token_tx
Definition: totem.h:252
void totemsrp_callback_token_destroy(void *srp_context, void **handle_out)
Definition: totemsrp.c:3533
uint64_t gather_token_lost
Definition: totem.h:268
int totemsrp_log_level_trace
Definition: totemsrp.c:429
void totemip_copy(struct totem_ip_address *addr1, const struct totem_ip_address *addr2)
Definition: totemip.c:123
int totemrrp_ifaces_get(void *rrp_context, char ***status, unsigned int *iface_count)
Definition: totemrrp.c:2286
struct memb_ring_id my_old_ring_id
Definition: totemsrp.c:339
memb_state
Definition: totemsrp.c:278
void * totemrrp_buffer_alloc(void *rrp_context)
Definition: totemrrp.c:2179
unsigned int downcheck_timeout
Definition: totem.h:148
struct srp_addr my_new_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:309
#define TOKEN_SIZE_MAX
Definition: totemsrp.c:101
uint64_t memb_commit_token_tx
Definition: totem.h:261
Definition: list.h:46
int my_deliver_memb_entries
Definition: totemsrp.c:331
unsigned int max_network_delay
Definition: totem.h:174
unsigned int heartbeat_failures_allowed
Definition: totem.h:172
#define TOTEM_TOKEN_STATS_MAX
Definition: totem.h:280
unsigned int my_last_seq
Definition: totemsrp.c:491
int my_left_memb_entries
Definition: totemsrp.c:333
#define swab64(x)
The swab64 macro.
Definition: swab.h:65
struct message_item __attribute__
unsigned long long token_ring_id_seq
Definition: totemsrp.c:479
struct totem_ip_address mcast_address
Definition: totemsrp.c:447
void(* totemsrp_log_printf)(int level, int subsys, const char *function, const char *file, int line, const char *format,...) __attribute__((format(printf
Definition: totemsrp.c:433
int totemsrp_callback_token_create(void *srp_context, void **handle_out, enum totem_callback_token_type type, int delete, int(*callback_fn)(enum totem_callback_token_type type, const void *), const void *data)
Definition: totemsrp.c:3498
unsigned int send_join_timeout
Definition: totem.h:142
unsigned int window_size
Definition: totem.h:176
int guarantee
Definition: totemsrp.c:191
unsigned int seq
Definition: totemsrp.c:197
void totemsrp_service_ready_register(void *context, void(*totem_service_ready)(void))
Definition: totemsrp.c:5156
unsigned int rrp_problem_count_threshold
Definition: totem.h:158
struct mcast * mcast
Definition: totemsrp.c:274
struct srp_addr my_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:313
uint64_t operational_entered
Definition: totem.h:265
void(*) in log_level_security)
Definition: totem.h:85
unsigned long long ring_seq
Definition: totemsrp.c:221
#define INTERFACE_MAX
Definition: coroapi.h:88
int totemsrp_mcast(void *srp_context, struct iovec *iovec, unsigned int iov_len, int guarantee)
Multicast a message.
Definition: totemsrp.c:2509
message_type
Definition: totemsrp.c:139
int latest_token
Definition: totem.h:279
uint64_t operational_token_lost
Definition: totem.h:266
unsigned int received_flg
Definition: totemsrp.c:63
uint64_t consensus_timeouts
Definition: totem.h:273
unsigned int aru_addr
Definition: totemsrp.c:206
Totem Network interface - also does encryption/decryption.
unsigned int my_high_delivered
Definition: totemsrp.c:383
struct message_handlers totemsrp_message_handlers
Definition: totemsrp.c:679
qb_loop_timer_handle memb_timer_state_gather_consensus_timeout
Definition: totemsrp.c:410
uint64_t recovery_token_lost
Definition: totem.h:272
unsigned int backlog
Definition: totemsrp.c:66
int this_seqno
Definition: totemsrp.c:188
unsigned int token_retransmits_before_loss_const
Definition: totem.h:138
unsigned char end_of_memb_join[0]
Definition: totemsrp.c:222
struct message_header header
Definition: totemsrp.c:239
int totemrrp_finalize(void *rrp_context)
Definition: totemrrp.c:2031
struct list_head token_callback_received_listhead
Definition: totemsrp.c:385
int totemrrp_member_remove(void *rrp_context, const struct totem_ip_address *member, int iface_no)
Definition: totemrrp.c:2367
struct rtr_item rtr_list[0]
Definition: totemsrp.c:70
unsigned int retrans_flg
Definition: totemsrp.c:256
int totemsrp_ring_reenable(void *srp_context)
Definition: totemsrp.c:1161
struct memb_ring_id ring_id
Definition: totemsrp.c:189
unsigned int seqno_unchanged_const
Definition: totem.h:152
uint64_t commit_token_lost
Definition: totem.h:270
unsigned int miss_count_const
Definition: totem.h:190
int totemrrp_crypto_set(void *rrp_context, const char *cipher_type, const char *hash_type)
Definition: totemrrp.c:2301
uint64_t token_hold_cancel_rx
Definition: totem.h:264
unsigned int join_timeout
Definition: totem.h:140
unsigned int aru
Definition: totemsrp.c:246
uint32_t originated_orf_token
Definition: totemsrp.c:515
unsigned int nodeid
Definition: coroapi.h:112
int totemrrp_send_flush(void *rrp_context)
Definition: totemrrp.c:2224
uint64_t pause_timestamp
Definition: totemsrp.c:507
int my_set_retrans_flg
Definition: totemsrp.c:357
struct message_header header
Definition: totemsrp.c:202
struct totem_ip_address mcast_addr
Definition: totem.h:70
char encapsulated
Definition: totemrrp.c:554
#define MESSAGE_QUEUE_MAX
Definition: coroapi.h:98
int totemrrp_member_add(void *rrp_context, const struct totem_ip_address *member, int iface_no)
Definition: totemrrp.c:2354
Linked list API.
unsigned int received_flg
Definition: totemsrp.c:248
unsigned int my_cbl
Definition: totemsrp.c:505
struct totem_ip_address rep
Definition: coroapi.h:123
unsigned int last_released
Definition: totemsrp.c:481
int orf_token_retransmit_size
Definition: totemsrp.c:391
int totemsrp_avail(void *srp_context)
Return number of available messages that can be queued.
Definition: totemsrp.c:2578
unsigned int rrp_autorecovery_check_timeout
Definition: totem.h:162
uint64_t mcast_retx
Definition: totem.h:259
unsigned int msg_len
Definition: totemsrp.c:275
#define RETRANS_MESSAGE_QUEUE_SIZE_MAX
Definition: totemsrp.c:97
void(* memb_ring_id_create_or_load)(struct memb_ring_id *memb_ring_id, const struct totem_ip_address *addr)
Definition: totemsrp.c:467
unsigned int fail_to_recv_const
Definition: totem.h:150
unsigned int token_seq
Definition: totemsrp.c:204
struct mcast * mcast
Definition: totemsrp.c:269
void * token_recv_event_handle
Definition: totemsrp.c:523
struct totem_ip_address boundto
Definition: totem.h:69
unsigned int my_high_seq_received
Definition: totemsrp.c:351
int totemrrp_initialize(qb_loop_t *poll_handle, void **rrp_context, struct totem_config *totem_config, totemsrp_stats_t *stats, void *context, void(*deliver_fn)(void *context, const void *msg, unsigned int msg_len), void(*iface_change_fn)(void *context, const struct totem_ip_address *iface_addr, unsigned int iface_no), void(*token_seqid_get)(const void *msg, unsigned int *seqid, unsigned int *token_is), unsigned int(*msgs_missing)(void), void(*target_set_completed)(void *context))
Create an instance.
Definition: totemrrp.c:2060
qb_loop_t * totemsrp_poll_handle
Definition: totemsrp.c:445
totem_event_type
Definition: totem.h:219
qb_loop_timer_handle timer_pause_timeout
Definition: totemsrp.c:398
qb_loop_timer_handle timer_merge_detect_timeout
Definition: totemsrp.c:406
int old_ring_state_saved
Definition: totemsrp.c:485
int my_merge_detect_timeout_outstanding
Definition: totemsrp.c:343
uint64_t rx_msg_dropped
Definition: totem.h:274
void(* log_printf)(int level, int subsys, const char *function_name, const char *file_name, int file_line, const char *format,...) __attribute__((format(printf
Definition: totem.h:78
int totemsrp_log_level_security
Definition: totemsrp.c:419
qb_loop_timer_handle timer_orf_token_retransmit_timeout
Definition: totemsrp.c:402
struct totem_config * totem_config
Definition: totemsrp.c:497
int(* callback_fn)(enum totem_callback_token_type type, const void *)
Definition: totemsrp.c:164
#define swab32(x)
The swab32 macro.
Definition: swab.h:51
qb_loop_timer_handle timer_orf_token_timeout
Definition: totemsrp.c:400
uint32_t continuous_gather
Definition: totem.h:275
void totemsrp_threaded_mode_enable(void *context)
Definition: totemsrp.c:5191
unsigned int aru
Definition: totemsrp.c:63
encapsulation_type
Definition: totemsrp.c:148
unsigned int net_mtu
Definition: totem.h:168
int totemsrp_initialize(qb_loop_t *poll_handle, void **srp_context, struct totem_config *totem_config, totemmrp_stats_t *stats, void(*deliver_fn)(unsigned int nodeid, const void *msg, unsigned int msg_len, int endian_conversion_required), void(*confchg_fn)(enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id), void(*waiting_trans_ack_cb_fn)(int waiting_trans_ack))
Create a protocol instance.
Definition: totemsrp.c:832
void totemsrp_event_signal(void *srp_context, enum totem_event_type type, int value)
Definition: totemsrp.c:2500
unsigned int node_id
Definition: totemsrp.c:190
int totemrrp_recv_flush(void *rrp_context)
Definition: totemrrp.c:2215
uint32_t orf_token_discard
Definition: totemsrp.c:513
int my_failed_list_entries
Definition: totemsrp.c:323
struct srp_addr my_id
Definition: totemsrp.c:303
struct srp_addr my_left_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:317
uint64_t token_hold_cancel_tx
Definition: totem.h:263
void(* totem_memb_ring_id_create_or_load)(struct memb_ring_id *memb_ring_id, const struct totem_ip_address *addr)
Definition: totem.h:198
unsigned int token_timeout
Definition: totem.h:132
Definition: totemsrp.c:244
unsigned int high_delivered
Definition: totemsrp.c:247
unsigned int consensus_timeout
Definition: totem.h:144
totemsrp_stats_t stats
Definition: totemsrp.c:511
void main_deliver_fn(void *context, const void *msg, unsigned int msg_len)
Definition: totemsrp.c:5053
#define PROCESSOR_COUNT_MAX
Definition: coroapi.h:96
uint64_t mcast_tx
Definition: totem.h:258
void totemrrp_buffer_release(void *rrp_context, void *ptr)
Definition: totemrrp.c:2186
void * totemrrp_context
Definition: totemsrp.c:495
Totem Network interface - also does encryption/decryption.
char orf_token_retransmit[TOKEN_SIZE_MAX]
Definition: totemsrp.c:389
struct message_header header
Definition: totemsrp.c:217
struct sq regular_sort_queue
Definition: totemsrp.c:374
int my_retrans_flg_count
Definition: totemsrp.c:359
unsigned int nodeid
Definition: totemsrp.c:63
The memb_ring_id struct.
Definition: coroapi.h:122
#define SEQNO_START_MSG
Definition: totemsrp.c:114
#define swab16(x)
The swab16 macro.
Definition: swab.h:39
void totemsrp_finalize(void *srp_context)
Definition: totemsrp.c:1042
#define QUEUE_RTR_ITEMS_SIZE_MAX
Definition: totemsrp.c:96
struct srp_addr my_failed_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:307
unsigned short family
Definition: coroapi.h:76
struct cs_queue retrans_message_queue
Definition: totemsrp.c:372
unsigned int aru
Definition: totemsrp.c:205
const char * gather_state_from_desc[]
Definition: totemsrp.c:557
qb_loop_timer_handle memb_timer_state_gather_join_timeout
Definition: totemsrp.c:408
int my_trans_memb_entries
Definition: totemsrp.c:327
unsigned int my_trc
Definition: totemsrp.c:501
void(* totemsrp_waiting_trans_ack_cb_fn)(int waiting_trans_ack)
Definition: totemsrp.c:464
uint64_t memb_merge_detect_tx
Definition: totem.h:254
unsigned int high_delivered
Definition: totemsrp.c:62
struct rtr_item rtr_list[0]
Definition: totemsrp.c:212
totemsrp_stats_t * srp
Definition: totem.h:290
int consensus_list_entries
Definition: totemsrp.c:301
unsigned int rrp_problem_count_mcast_threshold
Definition: totem.h:160
int totemrrp_processor_count_set(void *rrp_context, unsigned int processor_count)
Definition: totemrrp.c:2193
char type
Definition: totemsrp.c:60
uint64_t memb_join_rx
Definition: totem.h:257
int totemrrp_mcast_noflush_send(void *rrp_context, const void *msg, unsigned int msg_len)
Definition: totemrrp.c:2257
unsigned int cancel_token_hold_on_retransmit
Definition: totem.h:196
#define FRAME_SIZE_MAX
Definition: totem.h:50
int rtr_list_entries
Definition: totemsrp.c:211
uint32_t threaded_mode_enabled
Definition: totemsrp.c:517
enum totem_callback_token_type callback_type
Definition: totemsrp.c:165
int totemrrp_mcast_recv_empty(void *rrp_context)
Definition: totemrrp.c:2343
int my_proc_list_entries
Definition: totemsrp.c:321
#define list_entry(ptr, type, member)
Definition: list.h:84
unsigned long long ring_seq
Definition: totemsrp.c:64
struct totem_logging_configuration totem_logging_configuration
Definition: totem.h:166
unsigned short endian_detector
Definition: totemrrp.c:555
int totemrrp_mcast_flush_send(void *rrp_context, const void *msg, unsigned int msg_len)
Definition: totemrrp.c:2243
struct memb_ring_id ring_id
Definition: totemsrp.c:240
#define log_printf(level, format, args...)
Definition: totemsrp.c:691
unsigned long long seq
Definition: coroapi.h:124
void totemsrp_trans_ack(void *context)
Definition: totemsrp.c:5198
unsigned int max_messages
Definition: totem.h:178
uint64_t recovery_entered
Definition: totem.h:271
qb_loop_timer_handle memb_timer_state_commit_timeout
Definition: totemsrp.c:412
struct memb_commit_token * commit_token
Definition: totemsrp.c:509
struct consensus_list_item consensus_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:299
struct srp_addr addr
Definition: totemsrp.c:157
struct srp_addr my_proc_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:305
int(* handler_functions[6])(struct totemsrp_instance *instance, const void *msg, size_t msg_len, int endian_conversion_needed)
Definition: totemsrp.c:530
int totemsrp_subsys_id
Definition: totemsrp.c:431
unsigned int merge_timeout
Definition: totem.h:146
unsigned int use_heartbeat
Definition: totemsrp.c:499
struct message_header header
Definition: totemsrp.c:253
int totemsrp_member_remove(void *context, const struct totem_ip_address *member, int ring_no)
Definition: totemsrp.c:5178
unsigned int token_retransmit_timeout
Definition: totem.h:134
int rtr_list_entries
Definition: totemsrp.c:69
struct srp_addr my_trans_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:311
#define RETRANSMIT_ENTRIES_MAX
Definition: totemsrp.c:100
unsigned int token_seq
Definition: totemsrp.c:254
int totemip_equal(const struct totem_ip_address *addr1, const struct totem_ip_address *addr2)
Definition: totemip.c:72
unsigned int my_token_seq
Definition: totemsrp.c:393
struct memb_ring_id ring_id
Definition: totemsrp.c:64
unsigned int my_last_aru
Definition: totemsrp.c:345
int totemrrp_ring_reenable(void *rrp_context, unsigned int iface_no)
Definition: totemrrp.c:2320
unsigned int my_leave_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:319
uint64_t commit_entered
Definition: totem.h:269
qb_loop_timer_handle timer_orf_token_hold_retransmit_timeout
Definition: totemsrp.c:404
struct totem_ip_address addr[INTERFACE_MAX]
Definition: totemrrp.h:60
unsigned int rrp_token_expired_timeout
Definition: totem.h:154
struct memb_ring_id ring_id
Definition: totemsrp.c:234
unsigned int my_install_seq
Definition: totemsrp.c:353
uint64_t orf_token_rx
Definition: totem.h:253
unsigned int nodeid
Definition: totemsrp.c:180
int totemrrp_token_send(void *rrp_context, const void *msg, unsigned int msg_len)
Definition: totemrrp.c:2232
unsigned int threads
Definition: totem.h:170
unsigned int failed_list_entries
Definition: totemsrp.c:63
void(* totem_memb_ring_id_store)(const struct memb_ring_id *memb_ring_id, const struct totem_ip_address *addr)
Definition: totem.h:202
struct sq recovery_sort_queue
Definition: totemsrp.c:376
int totemrrp_token_target_set(void *rrp_context, struct totem_ip_address *addr, unsigned int iface_no)
Definition: totemrrp.c:2205
totem_callback_token_type
The totem_callback_token_type enum.
Definition: coroapi.h:142
unsigned int my_high_ring_delivered
Definition: totemsrp.c:361
unsigned int fcc
Definition: totemsrp.c:209