corosync  2.3.5
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 *);
165  enum totem_callback_token_type callback_type;
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 
299  struct consensus_list_item consensus_list[PROCESSOR_COUNT_MAX];
300 
302 
303  struct srp_addr my_id;
304 
305  struct srp_addr my_proc_list[PROCESSOR_COUNT_MAX];
306 
307  struct srp_addr my_failed_list[PROCESSOR_COUNT_MAX];
308 
309  struct srp_addr my_new_memb_list[PROCESSOR_COUNT_MAX];
310 
311  struct srp_addr my_trans_memb_list[PROCESSOR_COUNT_MAX];
312 
313  struct srp_addr my_memb_list[PROCESSOR_COUNT_MAX];
314 
315  struct srp_addr my_deliver_memb_list[PROCESSOR_COUNT_MAX];
316 
317  struct srp_addr my_left_memb_list[PROCESSOR_COUNT_MAX];
318 
319  unsigned int my_leave_memb_list[PROCESSOR_COUNT_MAX];
320 
322 
324 
326 
328 
330 
332 
334 
336 
337  struct memb_ring_id my_ring_id;
338 
339  struct memb_ring_id my_old_ring_id;
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  */
368  struct cs_queue new_message_queue;
369 
370  struct cs_queue new_message_queue_trans;
371 
372  struct cs_queue retrans_message_queue;
373 
374  struct sq regular_sort_queue;
375 
376  struct sq recovery_sort_queue;
377 
378  /*
379  * Received up to and including
380  */
381  unsigned int my_aru;
382 
383  unsigned int my_high_delivered;
384 
385  struct list_head token_callback_received_listhead;
386 
387  struct list_head token_callback_sent_listhead;
388 
389  char orf_token_retransmit[TOKEN_SIZE_MAX];
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 
433  void (*totemsrp_log_printf) (
434  int level,
435  int sybsys,
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 
447  struct totem_ip_address mcast_address;
448 
449  void (*totemsrp_deliver_fn) (
450  unsigned int nodeid,
451  const void *msg,
452  unsigned int msg_len,
453  int endian_conversion_required);
454 
455  void (*totemsrp_confchg_fn) (
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 
462  void (*totemsrp_service_ready_fn) (void);
463 
464  void (*totemsrp_waiting_trans_ack_cb_fn) (
465  int waiting_trans_ack);
466 
467  void (*memb_ring_id_create_or_load) (
468  struct memb_ring_id *memb_ring_id,
469  const struct totem_ip_address *addr);
470 
471  void (*memb_ring_id_store) (
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 (0 <= gsfrom && 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 
855  instance = malloc (sizeof (struct totemsrp_instance));
856  if (instance == NULL) {
857  goto error_exit;
858  }
859 
860  totemsrp_instance_initialize (instance);
861 
862  instance->totemsrp_waiting_trans_ack_cb_fn = waiting_trans_ack_cb_fn;
863  instance->totemsrp_waiting_trans_ack_cb_fn (1);
864 
865  stats->srp = &instance->stats;
866  instance->stats.latest_token = 0;
867  instance->stats.earliest_token = 0;
868 
869  instance->totem_config = totem_config;
870 
871  /*
872  * Configure logging
873  */
882 
883  /*
884  * Configure totem store and load functions
885  */
887  instance->memb_ring_id_store = totem_config->totem_memb_ring_id_store;
888 
889  /*
890  * Initialize local variables for totemsrp
891  */
892  totemip_copy (&instance->mcast_address, &totem_config->interfaces[0].mcast_addr);
893 
894  /*
895  * Display totem configuration
896  */
898  "Token Timeout (%d ms) retransmit timeout (%d ms)",
899  totem_config->token_timeout, totem_config->token_retransmit_timeout);
901  "token hold (%d ms) retransmits before loss (%d retrans)",
902  totem_config->token_hold_timeout, totem_config->token_retransmits_before_loss_const);
904  "join (%d ms) send_join (%d ms) consensus (%d ms) merge (%d ms)",
905  totem_config->join_timeout,
906  totem_config->send_join_timeout,
907  totem_config->consensus_timeout,
908 
909  totem_config->merge_timeout);
911  "downcheck (%d ms) fail to recv const (%d msgs)",
912  totem_config->downcheck_timeout, totem_config->fail_to_recv_const);
914  "seqno unchanged const (%d rotations) Maximum network MTU %d", totem_config->seqno_unchanged_const, totem_config->net_mtu);
915 
917  "window size per rotation (%d messages) maximum messages per rotation (%d messages)",
918  totem_config->window_size, totem_config->max_messages);
919 
921  "missed count const (%d messages)",
922  totem_config->miss_count_const);
923 
925  "send threads (%d threads)", totem_config->threads);
927  "RRP token expired timeout (%d ms)",
928  totem_config->rrp_token_expired_timeout);
930  "RRP token problem counter (%d ms)",
931  totem_config->rrp_problem_count_timeout);
933  "RRP threshold (%d problem count)",
934  totem_config->rrp_problem_count_threshold);
936  "RRP multicast threshold (%d problem count)",
937  totem_config->rrp_problem_count_mcast_threshold);
939  "RRP automatic recovery check timeout (%d ms)",
940  totem_config->rrp_autorecovery_check_timeout);
942  "RRP mode set to %s.", instance->totem_config->rrp_mode);
943 
945  "heartbeat_failures_allowed (%d)", totem_config->heartbeat_failures_allowed);
947  "max_network_delay (%d ms)", totem_config->max_network_delay);
948 
949 
950  cs_queue_init (&instance->retrans_message_queue, RETRANS_MESSAGE_QUEUE_SIZE_MAX,
951  sizeof (struct message_item), instance->threaded_mode_enabled);
952 
953  sq_init (&instance->regular_sort_queue,
954  QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0);
955 
956  sq_init (&instance->recovery_sort_queue,
957  QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0);
958 
959  instance->totemsrp_poll_handle = poll_handle;
960 
961  instance->totemsrp_deliver_fn = deliver_fn;
962 
963  instance->totemsrp_confchg_fn = confchg_fn;
964  instance->use_heartbeat = 1;
965 
966  timer_function_pause_timeout (instance);
967 
968  if ( totem_config->heartbeat_failures_allowed == 0 ) {
970  "HeartBeat is Disabled. To enable set heartbeat_failures_allowed > 0");
971  instance->use_heartbeat = 0;
972  }
973 
974  if (instance->use_heartbeat) {
975  instance->heartbeat_timeout
976  = (totem_config->heartbeat_failures_allowed) * totem_config->token_retransmit_timeout
977  + totem_config->max_network_delay;
978 
979  if (instance->heartbeat_timeout >= totem_config->token_timeout) {
981  "total heartbeat_timeout (%d ms) is not less than token timeout (%d ms)",
982  instance->heartbeat_timeout,
983  totem_config->token_timeout);
985  "heartbeat_timeout = heartbeat_failures_allowed * token_retransmit_timeout + max_network_delay");
987  "heartbeat timeout should be less than the token timeout. HeartBeat is Diabled !!");
988  instance->use_heartbeat = 0;
989  }
990  else {
992  "total heartbeat_timeout (%d ms)", instance->heartbeat_timeout);
993  }
994  }
995 
997  poll_handle,
998  &instance->totemrrp_context,
999  totem_config,
1000  stats->srp,
1001  instance,
1002  main_deliver_fn,
1003  main_iface_change_fn,
1004  main_token_seqid_get,
1005  main_msgs_missing,
1006  target_set_completed);
1007 
1008  /*
1009  * Must have net_mtu adjusted by totemrrp_initialize first
1010  */
1011  cs_queue_init (&instance->new_message_queue,
1013  sizeof (struct message_item), instance->threaded_mode_enabled);
1014 
1015  cs_queue_init (&instance->new_message_queue_trans,
1017  sizeof (struct message_item), instance->threaded_mode_enabled);
1018 
1020  &instance->token_recv_event_handle,
1022  0,
1023  token_event_stats_collector,
1024  instance);
1026  &instance->token_sent_event_handle,
1028  0,
1029  token_event_stats_collector,
1030  instance);
1031  *srp_context = instance;
1032  return (0);
1033 
1034 error_exit:
1035  return (-1);
1036 }
1037 
1039  void *srp_context)
1040 {
1041  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1042 
1043 
1044  memb_leave_message_send (instance);
1045  totemrrp_finalize (instance->totemrrp_context);
1046  cs_queue_free (&instance->new_message_queue);
1047  cs_queue_free (&instance->new_message_queue_trans);
1048  cs_queue_free (&instance->retrans_message_queue);
1049  sq_free (&instance->regular_sort_queue);
1050  sq_free (&instance->recovery_sort_queue);
1051  free (instance);
1052 }
1053 
1054 /*
1055  * Return configured interfaces. interfaces is array of totem_ip addresses allocated by caller,
1056  * with interaces_size number of items. iface_count is final number of interfaces filled by this
1057  * function.
1058  *
1059  * Function returns 0 on success, otherwise if interfaces array is not big enough, -2 is returned,
1060  * and if interface was not found, -1 is returned.
1061  */
1063  void *srp_context,
1064  unsigned int nodeid,
1065  struct totem_ip_address *interfaces,
1066  unsigned int interfaces_size,
1067  char ***status,
1068  unsigned int *iface_count)
1069 {
1070  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1071  int res = 0;
1072  unsigned int found = 0;
1073  unsigned int i;
1074 
1075  for (i = 0; i < instance->my_memb_entries; i++) {
1076  if (instance->my_memb_list[i].addr[0].nodeid == nodeid) {
1077  found = 1;
1078  break;
1079  }
1080  }
1081 
1082  if (found) {
1083  *iface_count = instance->totem_config->interface_count;
1084 
1085  if (interfaces_size >= *iface_count) {
1086  memcpy (interfaces, instance->my_memb_list[i].addr,
1087  sizeof (struct totem_ip_address) * *iface_count);
1088  } else {
1089  res = -2;
1090  }
1091 
1092  goto finish;
1093  }
1094 
1095  for (i = 0; i < instance->my_left_memb_entries; i++) {
1096  if (instance->my_left_memb_list[i].addr[0].nodeid == nodeid) {
1097  found = 1;
1098  break;
1099  }
1100  }
1101 
1102  if (found) {
1103  *iface_count = instance->totem_config->interface_count;
1104 
1105  if (interfaces_size >= *iface_count) {
1106  memcpy (interfaces, instance->my_left_memb_list[i].addr,
1107  sizeof (struct totem_ip_address) * *iface_count);
1108  } else {
1109  res = -2;
1110  }
1111  } else {
1112  res = -1;
1113  }
1114 
1115 finish:
1116  totemrrp_ifaces_get (instance->totemrrp_context, status, NULL);
1117  return (res);
1118 }
1119 
1121  void *srp_context,
1122  const char *cipher_type,
1123  const char *hash_type)
1124 {
1125  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1126  int res;
1127 
1128  res = totemrrp_crypto_set(instance->totemrrp_context, cipher_type, hash_type);
1129 
1130  return (res);
1131 }
1132 
1133 
1135  void *srp_context)
1136 {
1137  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1138  unsigned int res;
1139 
1140  res = instance->totem_config->interfaces[0].boundto.nodeid;
1141 
1142  return (res);
1143 }
1144 
1146  void *srp_context)
1147 {
1148  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1149  int res;
1150 
1151  res = instance->totem_config->interfaces[0].boundto.family;
1152 
1153  return (res);
1154 }
1155 
1156 
1158  void *srp_context)
1159 {
1160  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
1161 
1163  instance->totem_config->interface_count);
1164 
1165  return (0);
1166 }
1167 
1168 
1169 /*
1170  * Set operations for use by the membership algorithm
1171  */
1172 static int srp_addr_equal (const struct srp_addr *a, const struct srp_addr *b)
1173 {
1174  unsigned int i;
1175  unsigned int res;
1176 
1177  for (i = 0; i < 1; i++) {
1178  res = totemip_equal (&a->addr[i], &b->addr[i]);
1179  if (res == 0) {
1180  return (0);
1181  }
1182  }
1183  return (1);
1184 }
1185 
1186 static void srp_addr_copy (struct srp_addr *dest, const struct srp_addr *src)
1187 {
1188  unsigned int i;
1189 
1190  dest->no_addrs = src->no_addrs;
1191 
1192  for (i = 0; i < INTERFACE_MAX; i++) {
1193  totemip_copy (&dest->addr[i], &src->addr[i]);
1194  }
1195 }
1196 
1197 static void srp_addr_to_nodeid (
1198  unsigned int *nodeid_out,
1199  struct srp_addr *srp_addr_in,
1200  unsigned int entries)
1201 {
1202  unsigned int i;
1203 
1204  for (i = 0; i < entries; i++) {
1205  nodeid_out[i] = srp_addr_in[i].addr[0].nodeid;
1206  }
1207 }
1208 
1209 static void srp_addr_copy_endian_convert (struct srp_addr *out, const struct srp_addr *in)
1210 {
1211  int i;
1212 
1213  for (i = 0; i < INTERFACE_MAX; i++) {
1214  totemip_copy_endian_convert (&out->addr[i], &in->addr[i]);
1215  }
1216 }
1217 
1218 static void memb_consensus_reset (struct totemsrp_instance *instance)
1219 {
1220  instance->consensus_list_entries = 0;
1221 }
1222 
1223 static void memb_set_subtract (
1224  struct srp_addr *out_list, int *out_list_entries,
1225  struct srp_addr *one_list, int one_list_entries,
1226  struct srp_addr *two_list, int two_list_entries)
1227 {
1228  int found = 0;
1229  int i;
1230  int j;
1231 
1232  *out_list_entries = 0;
1233 
1234  for (i = 0; i < one_list_entries; i++) {
1235  for (j = 0; j < two_list_entries; j++) {
1236  if (srp_addr_equal (&one_list[i], &two_list[j])) {
1237  found = 1;
1238  break;
1239  }
1240  }
1241  if (found == 0) {
1242  srp_addr_copy (&out_list[*out_list_entries], &one_list[i]);
1243  *out_list_entries = *out_list_entries + 1;
1244  }
1245  found = 0;
1246  }
1247 }
1248 
1249 /*
1250  * Set consensus for a specific processor
1251  */
1252 static void memb_consensus_set (
1253  struct totemsrp_instance *instance,
1254  const struct srp_addr *addr)
1255 {
1256  int found = 0;
1257  int i;
1258 
1259  if (addr->addr[0].nodeid == LEAVE_DUMMY_NODEID)
1260  return;
1261 
1262  for (i = 0; i < instance->consensus_list_entries; i++) {
1263  if (srp_addr_equal(addr, &instance->consensus_list[i].addr)) {
1264  found = 1;
1265  break; /* found entry */
1266  }
1267  }
1268  srp_addr_copy (&instance->consensus_list[i].addr, addr);
1269  instance->consensus_list[i].set = 1;
1270  if (found == 0) {
1271  instance->consensus_list_entries++;
1272  }
1273  return;
1274 }
1275 
1276 /*
1277  * Is consensus set for a specific processor
1278  */
1279 static int memb_consensus_isset (
1280  struct totemsrp_instance *instance,
1281  const struct srp_addr *addr)
1282 {
1283  int i;
1284 
1285  for (i = 0; i < instance->consensus_list_entries; i++) {
1286  if (srp_addr_equal (addr, &instance->consensus_list[i].addr)) {
1287  return (instance->consensus_list[i].set);
1288  }
1289  }
1290  return (0);
1291 }
1292 
1293 /*
1294  * Is consensus agreed upon based upon consensus database
1295  */
1296 static int memb_consensus_agreed (
1297  struct totemsrp_instance *instance)
1298 {
1299  struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
1300  int token_memb_entries = 0;
1301  int agreed = 1;
1302  int i;
1303 
1304  memb_set_subtract (token_memb, &token_memb_entries,
1305  instance->my_proc_list, instance->my_proc_list_entries,
1306  instance->my_failed_list, instance->my_failed_list_entries);
1307 
1308  for (i = 0; i < token_memb_entries; i++) {
1309  if (memb_consensus_isset (instance, &token_memb[i]) == 0) {
1310  agreed = 0;
1311  break;
1312  }
1313  }
1314 
1315  if (agreed && instance->failed_to_recv == 1) {
1316  /*
1317  * Both nodes agreed on our failure. We don't care how many proc list items left because we
1318  * will create single ring anyway.
1319  */
1320 
1321  return (agreed);
1322  }
1323 
1324  assert (token_memb_entries >= 1);
1325 
1326  return (agreed);
1327 }
1328 
1329 static void memb_consensus_notset (
1330  struct totemsrp_instance *instance,
1331  struct srp_addr *no_consensus_list,
1332  int *no_consensus_list_entries,
1333  struct srp_addr *comparison_list,
1334  int comparison_list_entries)
1335 {
1336  int i;
1337 
1338  *no_consensus_list_entries = 0;
1339 
1340  for (i = 0; i < instance->my_proc_list_entries; i++) {
1341  if (memb_consensus_isset (instance, &instance->my_proc_list[i]) == 0) {
1342  srp_addr_copy (&no_consensus_list[*no_consensus_list_entries], &instance->my_proc_list[i]);
1343  *no_consensus_list_entries = *no_consensus_list_entries + 1;
1344  }
1345  }
1346 }
1347 
1348 /*
1349  * Is set1 equal to set2 Entries can be in different orders
1350  */
1351 static int memb_set_equal (
1352  struct srp_addr *set1, int set1_entries,
1353  struct srp_addr *set2, int set2_entries)
1354 {
1355  int i;
1356  int j;
1357 
1358  int found = 0;
1359 
1360  if (set1_entries != set2_entries) {
1361  return (0);
1362  }
1363  for (i = 0; i < set2_entries; i++) {
1364  for (j = 0; j < set1_entries; j++) {
1365  if (srp_addr_equal (&set1[j], &set2[i])) {
1366  found = 1;
1367  break;
1368  }
1369  }
1370  if (found == 0) {
1371  return (0);
1372  }
1373  found = 0;
1374  }
1375  return (1);
1376 }
1377 
1378 /*
1379  * Is subset fully contained in fullset
1380  */
1381 static int memb_set_subset (
1382  const struct srp_addr *subset, int subset_entries,
1383  const struct srp_addr *fullset, int fullset_entries)
1384 {
1385  int i;
1386  int j;
1387  int found = 0;
1388 
1389  if (subset_entries > fullset_entries) {
1390  return (0);
1391  }
1392  for (i = 0; i < subset_entries; i++) {
1393  for (j = 0; j < fullset_entries; j++) {
1394  if (srp_addr_equal (&subset[i], &fullset[j])) {
1395  found = 1;
1396  }
1397  }
1398  if (found == 0) {
1399  return (0);
1400  }
1401  found = 0;
1402  }
1403  return (1);
1404 }
1405 /*
1406  * merge subset into fullset taking care not to add duplicates
1407  */
1408 static void memb_set_merge (
1409  const struct srp_addr *subset, int subset_entries,
1410  struct srp_addr *fullset, int *fullset_entries)
1411 {
1412  int found = 0;
1413  int i;
1414  int j;
1415 
1416  for (i = 0; i < subset_entries; i++) {
1417  for (j = 0; j < *fullset_entries; j++) {
1418  if (srp_addr_equal (&fullset[j], &subset[i])) {
1419  found = 1;
1420  break;
1421  }
1422  }
1423  if (found == 0) {
1424  srp_addr_copy (&fullset[*fullset_entries], &subset[i]);
1425  *fullset_entries = *fullset_entries + 1;
1426  }
1427  found = 0;
1428  }
1429  return;
1430 }
1431 
1432 static void memb_set_and_with_ring_id (
1433  struct srp_addr *set1,
1434  struct memb_ring_id *set1_ring_ids,
1435  int set1_entries,
1436  struct srp_addr *set2,
1437  int set2_entries,
1438  struct memb_ring_id *old_ring_id,
1439  struct srp_addr *and,
1440  int *and_entries)
1441 {
1442  int i;
1443  int j;
1444  int found = 0;
1445 
1446  *and_entries = 0;
1447 
1448  for (i = 0; i < set2_entries; i++) {
1449  for (j = 0; j < set1_entries; j++) {
1450  if (srp_addr_equal (&set1[j], &set2[i])) {
1451  if (memcmp (&set1_ring_ids[j], old_ring_id, sizeof (struct memb_ring_id)) == 0) {
1452  found = 1;
1453  }
1454  break;
1455  }
1456  }
1457  if (found) {
1458  srp_addr_copy (&and[*and_entries], &set1[j]);
1459  *and_entries = *and_entries + 1;
1460  }
1461  found = 0;
1462  }
1463  return;
1464 }
1465 
1466 #ifdef CODE_COVERAGE
1467 static void memb_set_print (
1468  char *string,
1469  struct srp_addr *list,
1470  int list_entries)
1471 {
1472  int i;
1473  int j;
1474  printf ("List '%s' contains %d entries:\n", string, list_entries);
1475 
1476  for (i = 0; i < list_entries; i++) {
1477  printf ("Address %d with %d rings\n", i, list[i].no_addrs);
1478  for (j = 0; j < list[i].no_addrs; j++) {
1479  printf ("\tiface %d %s\n", j, totemip_print (&list[i].addr[j]));
1480  printf ("\tfamily %d\n", list[i].addr[j].family);
1481  }
1482  }
1483 }
1484 #endif
1485 static void my_leave_memb_clear(
1486  struct totemsrp_instance *instance)
1487 {
1488  memset(instance->my_leave_memb_list, 0, sizeof(instance->my_leave_memb_list));
1489  instance->my_leave_memb_entries = 0;
1490 }
1491 
1492 static unsigned int my_leave_memb_match(
1493  struct totemsrp_instance *instance,
1494  unsigned int nodeid)
1495 {
1496  int i;
1497  unsigned int ret = 0;
1498 
1499  for (i = 0; i < instance->my_leave_memb_entries; i++){
1500  if (instance->my_leave_memb_list[i] == nodeid){
1501  ret = nodeid;
1502  break;
1503  }
1504  }
1505  return ret;
1506 }
1507 
1508 static void my_leave_memb_set(
1509  struct totemsrp_instance *instance,
1510  unsigned int nodeid)
1511 {
1512  int i, found = 0;
1513  for (i = 0; i < instance->my_leave_memb_entries; i++){
1514  if (instance->my_leave_memb_list[i] == nodeid){
1515  found = 1;
1516  break;
1517  }
1518  }
1519  if (found == 1) {
1520  return;
1521  }
1522  if (instance->my_leave_memb_entries < (PROCESSOR_COUNT_MAX - 1)) {
1523  instance->my_leave_memb_list[instance->my_leave_memb_entries] = nodeid;
1524  instance->my_leave_memb_entries++;
1525  } else {
1527  "Cannot set LEAVE nodeid=%d", nodeid);
1528  }
1529 }
1530 
1531 
1532 static void *totemsrp_buffer_alloc (struct totemsrp_instance *instance)
1533 {
1534  assert (instance != NULL);
1535  return totemrrp_buffer_alloc (instance->totemrrp_context);
1536 }
1537 
1538 static void totemsrp_buffer_release (struct totemsrp_instance *instance, void *ptr)
1539 {
1540  assert (instance != NULL);
1541  totemrrp_buffer_release (instance->totemrrp_context, ptr);
1542 }
1543 
1544 static void reset_token_retransmit_timeout (struct totemsrp_instance *instance)
1545 {
1546  qb_loop_timer_del (instance->totemsrp_poll_handle,
1548  qb_loop_timer_add (instance->totemsrp_poll_handle,
1549  QB_LOOP_MED,
1550  instance->totem_config->token_retransmit_timeout*QB_TIME_NS_IN_MSEC,
1551  (void *)instance,
1552  timer_function_token_retransmit_timeout,
1554 
1555 }
1556 
1557 static void start_merge_detect_timeout (struct totemsrp_instance *instance)
1558 {
1559  if (instance->my_merge_detect_timeout_outstanding == 0) {
1560  qb_loop_timer_add (instance->totemsrp_poll_handle,
1561  QB_LOOP_MED,
1562  instance->totem_config->merge_timeout*QB_TIME_NS_IN_MSEC,
1563  (void *)instance,
1564  timer_function_merge_detect_timeout,
1565  &instance->timer_merge_detect_timeout);
1566 
1568  }
1569 }
1570 
1571 static void cancel_merge_detect_timeout (struct totemsrp_instance *instance)
1572 {
1573  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_merge_detect_timeout);
1575 }
1576 
1577 /*
1578  * ring_state_* is used to save and restore the sort queue
1579  * state when a recovery operation fails (and enters gather)
1580  */
1581 static void old_ring_state_save (struct totemsrp_instance *instance)
1582 {
1583  if (instance->old_ring_state_saved == 0) {
1584  instance->old_ring_state_saved = 1;
1585  memcpy (&instance->my_old_ring_id, &instance->my_ring_id,
1586  sizeof (struct memb_ring_id));
1587  instance->old_ring_state_aru = instance->my_aru;
1590  "Saving state aru %x high seq received %x",
1591  instance->my_aru, instance->my_high_seq_received);
1592  }
1593 }
1594 
1595 static void old_ring_state_restore (struct totemsrp_instance *instance)
1596 {
1597  instance->my_aru = instance->old_ring_state_aru;
1600  "Restoring instance->my_aru %x my high seq received %x",
1601  instance->my_aru, instance->my_high_seq_received);
1602 }
1603 
1604 static void old_ring_state_reset (struct totemsrp_instance *instance)
1605 {
1607  "Resetting old ring state");
1608  instance->old_ring_state_saved = 0;
1609 }
1610 
1611 static void reset_pause_timeout (struct totemsrp_instance *instance)
1612 {
1613  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_pause_timeout);
1614  qb_loop_timer_add (instance->totemsrp_poll_handle,
1615  QB_LOOP_MED,
1616  instance->totem_config->token_timeout * QB_TIME_NS_IN_MSEC / 5,
1617  (void *)instance,
1618  timer_function_pause_timeout,
1619  &instance->timer_pause_timeout);
1620 }
1621 
1622 static void reset_token_timeout (struct totemsrp_instance *instance) {
1623  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout);
1624  qb_loop_timer_add (instance->totemsrp_poll_handle,
1625  QB_LOOP_MED,
1626  instance->totem_config->token_timeout*QB_TIME_NS_IN_MSEC,
1627  (void *)instance,
1628  timer_function_orf_token_timeout,
1629  &instance->timer_orf_token_timeout);
1630 }
1631 
1632 static void reset_heartbeat_timeout (struct totemsrp_instance *instance) {
1633  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout);
1634  qb_loop_timer_add (instance->totemsrp_poll_handle,
1635  QB_LOOP_MED,
1636  instance->heartbeat_timeout*QB_TIME_NS_IN_MSEC,
1637  (void *)instance,
1638  timer_function_heartbeat_timeout,
1639  &instance->timer_heartbeat_timeout);
1640 }
1641 
1642 
1643 static void cancel_token_timeout (struct totemsrp_instance *instance) {
1644  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout);
1645 }
1646 
1647 static void cancel_heartbeat_timeout (struct totemsrp_instance *instance) {
1648  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout);
1649 }
1650 
1651 static void cancel_token_retransmit_timeout (struct totemsrp_instance *instance)
1652 {
1653  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->timer_orf_token_retransmit_timeout);
1654 }
1655 
1656 static void start_token_hold_retransmit_timeout (struct totemsrp_instance *instance)
1657 {
1658  qb_loop_timer_add (instance->totemsrp_poll_handle,
1659  QB_LOOP_MED,
1660  instance->totem_config->token_hold_timeout*QB_TIME_NS_IN_MSEC,
1661  (void *)instance,
1662  timer_function_token_hold_retransmit_timeout,
1664 }
1665 
1666 static void cancel_token_hold_retransmit_timeout (struct totemsrp_instance *instance)
1667 {
1668  qb_loop_timer_del (instance->totemsrp_poll_handle,
1670 }
1671 
1672 static void memb_state_consensus_timeout_expired (
1673  struct totemsrp_instance *instance)
1674 {
1675  struct srp_addr no_consensus_list[PROCESSOR_COUNT_MAX];
1676  int no_consensus_list_entries;
1677 
1678  instance->stats.consensus_timeouts++;
1679  if (memb_consensus_agreed (instance)) {
1680  memb_consensus_reset (instance);
1681 
1682  memb_consensus_set (instance, &instance->my_id);
1683 
1684  reset_token_timeout (instance); // REVIEWED
1685  } else {
1686  memb_consensus_notset (
1687  instance,
1688  no_consensus_list,
1689  &no_consensus_list_entries,
1690  instance->my_proc_list,
1691  instance->my_proc_list_entries);
1692 
1693  memb_set_merge (no_consensus_list, no_consensus_list_entries,
1694  instance->my_failed_list, &instance->my_failed_list_entries);
1695  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_CONSENSUS_TIMEOUT);
1696  }
1697 }
1698 
1699 static void memb_join_message_send (struct totemsrp_instance *instance);
1700 
1701 static void memb_merge_detect_transmit (struct totemsrp_instance *instance);
1702 
1703 /*
1704  * Timers used for various states of the membership algorithm
1705  */
1706 static void timer_function_pause_timeout (void *data)
1707 {
1708  struct totemsrp_instance *instance = data;
1709 
1710  instance->pause_timestamp = qb_util_nano_current_get ();
1711  reset_pause_timeout (instance);
1712 }
1713 
1714 static void memb_recovery_state_token_loss (struct totemsrp_instance *instance)
1715 {
1716  old_ring_state_restore (instance);
1717  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_RECOVERY_STATE);
1718  instance->stats.recovery_token_lost++;
1719 }
1720 
1721 static void timer_function_orf_token_timeout (void *data)
1722 {
1723  struct totemsrp_instance *instance = data;
1724 
1725  switch (instance->memb_state) {
1728  "The token was lost in the OPERATIONAL state.");
1730  "A processor failed, forming new configuration.");
1732  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_OPERATIONAL_STATE);
1733  instance->stats.operational_token_lost++;
1734  break;
1735 
1736  case MEMB_STATE_GATHER:
1738  "The consensus timeout expired.");
1739  memb_state_consensus_timeout_expired (instance);
1740  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_CONSENSUS_TIMEOUT_EXPIRED);
1741  instance->stats.gather_token_lost++;
1742  break;
1743 
1744  case MEMB_STATE_COMMIT:
1746  "The token was lost in the COMMIT state.");
1747  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_THE_TOKEN_WAS_LOST_IN_THE_COMMIT_STATE);
1748  instance->stats.commit_token_lost++;
1749  break;
1750 
1751  case MEMB_STATE_RECOVERY:
1753  "The token was lost in the RECOVERY state.");
1754  memb_recovery_state_token_loss (instance);
1755  instance->orf_token_discard = 1;
1756  break;
1757  }
1758 }
1759 
1760 static void timer_function_heartbeat_timeout (void *data)
1761 {
1762  struct totemsrp_instance *instance = data;
1764  "HeartBeat Timer expired Invoking token loss mechanism in state %d ", instance->memb_state);
1765  timer_function_orf_token_timeout(data);
1766 }
1767 
1768 static void memb_timer_function_state_gather (void *data)
1769 {
1770  struct totemsrp_instance *instance = data;
1771 
1772  switch (instance->memb_state) {
1774  case MEMB_STATE_RECOVERY:
1775  assert (0); /* this should never happen */
1776  break;
1777  case MEMB_STATE_GATHER:
1778  case MEMB_STATE_COMMIT:
1779  memb_join_message_send (instance);
1780 
1781  /*
1782  * Restart the join timeout
1783  `*/
1784  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);
1785 
1786  qb_loop_timer_add (instance->totemsrp_poll_handle,
1787  QB_LOOP_MED,
1788  instance->totem_config->join_timeout*QB_TIME_NS_IN_MSEC,
1789  (void *)instance,
1790  memb_timer_function_state_gather,
1792  break;
1793  }
1794 }
1795 
1796 static void memb_timer_function_gather_consensus_timeout (void *data)
1797 {
1798  struct totemsrp_instance *instance = data;
1799  memb_state_consensus_timeout_expired (instance);
1800 }
1801 
1802 static void deliver_messages_from_recovery_to_regular (struct totemsrp_instance *instance)
1803 {
1804  unsigned int i;
1805  struct sort_queue_item *recovery_message_item;
1806  struct sort_queue_item regular_message_item;
1807  unsigned int range = 0;
1808  int res;
1809  void *ptr;
1810  struct mcast *mcast;
1811 
1813  "recovery to regular %x-%x", SEQNO_START_MSG + 1, instance->my_aru);
1814 
1815  range = instance->my_aru - SEQNO_START_MSG;
1816  /*
1817  * Move messages from recovery to regular sort queue
1818  */
1819 // todo should i be initialized to 0 or 1 ?
1820  for (i = 1; i <= range; i++) {
1821  res = sq_item_get (&instance->recovery_sort_queue,
1822  i + SEQNO_START_MSG, &ptr);
1823  if (res != 0) {
1824  continue;
1825  }
1826  recovery_message_item = ptr;
1827 
1828  /*
1829  * Convert recovery message into regular message
1830  */
1831  mcast = recovery_message_item->mcast;
1832  if (mcast->header.encapsulated == MESSAGE_ENCAPSULATED) {
1833  /*
1834  * Message is a recovery message encapsulated
1835  * in a new ring message
1836  */
1837  regular_message_item.mcast =
1838  (struct mcast *)(((char *)recovery_message_item->mcast) + sizeof (struct mcast));
1839  regular_message_item.msg_len =
1840  recovery_message_item->msg_len - sizeof (struct mcast);
1841  mcast = regular_message_item.mcast;
1842  } else {
1843  /*
1844  * TODO this case shouldn't happen
1845  */
1846  continue;
1847  }
1848 
1850  "comparing if ring id is for this processors old ring seqno %d",
1851  mcast->seq);
1852 
1853  /*
1854  * Only add this message to the regular sort
1855  * queue if it was originated with the same ring
1856  * id as the previous ring
1857  */
1858  if (memcmp (&instance->my_old_ring_id, &mcast->ring_id,
1859  sizeof (struct memb_ring_id)) == 0) {
1860 
1861  res = sq_item_inuse (&instance->regular_sort_queue, mcast->seq);
1862  if (res == 0) {
1863  sq_item_add (&instance->regular_sort_queue,
1864  &regular_message_item, mcast->seq);
1865  if (sq_lt_compare (instance->old_ring_state_high_seq_received, mcast->seq)) {
1866  instance->old_ring_state_high_seq_received = mcast->seq;
1867  }
1868  }
1869  } else {
1871  "-not adding msg with seq no %x", mcast->seq);
1872  }
1873  }
1874 }
1875 
1876 /*
1877  * Change states in the state machine of the membership algorithm
1878  */
1879 static void memb_state_operational_enter (struct totemsrp_instance *instance)
1880 {
1881  struct srp_addr joined_list[PROCESSOR_COUNT_MAX];
1882  int joined_list_entries = 0;
1883  unsigned int aru_save;
1884  unsigned int joined_list_totemip[PROCESSOR_COUNT_MAX];
1885  unsigned int trans_memb_list_totemip[PROCESSOR_COUNT_MAX];
1886  unsigned int new_memb_list_totemip[PROCESSOR_COUNT_MAX];
1887  unsigned int left_list[PROCESSOR_COUNT_MAX];
1888  unsigned int i;
1889  unsigned int res;
1890  char left_node_msg[1024];
1891  char joined_node_msg[1024];
1892  char failed_node_msg[1024];
1893 
1894  instance->originated_orf_token = 0;
1895 
1896  memb_consensus_reset (instance);
1897 
1898  old_ring_state_reset (instance);
1899 
1900  deliver_messages_from_recovery_to_regular (instance);
1901 
1903  "Delivering to app %x to %x",
1904  instance->my_high_delivered + 1, instance->old_ring_state_high_seq_received);
1905 
1906  aru_save = instance->my_aru;
1907  instance->my_aru = instance->old_ring_state_aru;
1908 
1909  messages_deliver_to_app (instance, 0, instance->old_ring_state_high_seq_received);
1910 
1911  /*
1912  * Calculate joined and left list
1913  */
1914  memb_set_subtract (instance->my_left_memb_list,
1915  &instance->my_left_memb_entries,
1916  instance->my_memb_list, instance->my_memb_entries,
1917  instance->my_trans_memb_list, instance->my_trans_memb_entries);
1918 
1919  memb_set_subtract (joined_list, &joined_list_entries,
1920  instance->my_new_memb_list, instance->my_new_memb_entries,
1921  instance->my_trans_memb_list, instance->my_trans_memb_entries);
1922 
1923  /*
1924  * Install new membership
1925  */
1926  instance->my_memb_entries = instance->my_new_memb_entries;
1927  memcpy (&instance->my_memb_list, instance->my_new_memb_list,
1928  sizeof (struct srp_addr) * instance->my_memb_entries);
1929  instance->last_released = 0;
1930  instance->my_set_retrans_flg = 0;
1931 
1932  /*
1933  * Inform RRP about transitional change
1934  */
1936  instance->totemrrp_context,
1938  instance->my_trans_memb_list, instance->my_trans_memb_entries,
1939  instance->my_left_memb_list, instance->my_left_memb_entries,
1940  NULL, 0,
1941  &instance->my_ring_id);
1942  /*
1943  * Deliver transitional configuration to application
1944  */
1945  srp_addr_to_nodeid (left_list, instance->my_left_memb_list,
1946  instance->my_left_memb_entries);
1947  srp_addr_to_nodeid (trans_memb_list_totemip,
1948  instance->my_trans_memb_list, instance->my_trans_memb_entries);
1950  trans_memb_list_totemip, instance->my_trans_memb_entries,
1951  left_list, instance->my_left_memb_entries,
1952  0, 0, &instance->my_ring_id);
1953  instance->waiting_trans_ack = 1;
1954  instance->totemsrp_waiting_trans_ack_cb_fn (1);
1955 
1956 // TODO we need to filter to ensure we only deliver those
1957 // messages which are part of instance->my_deliver_memb
1958  messages_deliver_to_app (instance, 1, instance->old_ring_state_high_seq_received);
1959 
1960  instance->my_aru = aru_save;
1961 
1962  /*
1963  * Inform RRP about regular membership change
1964  */
1966  instance->totemrrp_context,
1968  instance->my_new_memb_list, instance->my_new_memb_entries,
1969  NULL, 0,
1970  joined_list, joined_list_entries,
1971  &instance->my_ring_id);
1972  /*
1973  * Deliver regular configuration to application
1974  */
1975  srp_addr_to_nodeid (new_memb_list_totemip,
1976  instance->my_new_memb_list, instance->my_new_memb_entries);
1977  srp_addr_to_nodeid (joined_list_totemip, joined_list,
1978  joined_list_entries);
1980  new_memb_list_totemip, instance->my_new_memb_entries,
1981  0, 0,
1982  joined_list_totemip, joined_list_entries, &instance->my_ring_id);
1983 
1984  /*
1985  * The recovery sort queue now becomes the regular
1986  * sort queue. It is necessary to copy the state
1987  * into the regular sort queue.
1988  */
1989  sq_copy (&instance->regular_sort_queue, &instance->recovery_sort_queue);
1990  instance->my_last_aru = SEQNO_START_MSG;
1991 
1992  /* When making my_proc_list smaller, ensure that the
1993  * now non-used entries are zero-ed out. There are some suspect
1994  * assert's that assume that there is always 2 entries in the list.
1995  * These fail when my_proc_list is reduced to 1 entry (and the
1996  * valid [0] entry is the same as the 'unused' [1] entry).
1997  */
1998  memset(instance->my_proc_list, 0,
1999  sizeof (struct srp_addr) * instance->my_proc_list_entries);
2000 
2001  instance->my_proc_list_entries = instance->my_new_memb_entries;
2002  memcpy (instance->my_proc_list, instance->my_new_memb_list,
2003  sizeof (struct srp_addr) * instance->my_memb_entries);
2004 
2005  instance->my_failed_list_entries = 0;
2006  /*
2007  * TODO Not exactly to spec
2008  *
2009  * At the entry to this function all messages without a gap are
2010  * deliered.
2011  *
2012  * This code throw away messages from the last gap in the sort queue
2013  * to my_high_seq_received
2014  *
2015  * What should really happen is we should deliver all messages up to
2016  * a gap, then delier the transitional configuration, then deliver
2017  * the messages between the first gap and my_high_seq_received, then
2018  * deliver a regular configuration, then deliver the regular
2019  * configuration
2020  *
2021  * Unfortunately totempg doesn't appear to like this operating mode
2022  * which needs more inspection
2023  */
2024  i = instance->my_high_seq_received + 1;
2025  do {
2026  void *ptr;
2027 
2028  i -= 1;
2029  res = sq_item_get (&instance->regular_sort_queue, i, &ptr);
2030  if (i == 0) {
2031  break;
2032  }
2033  } while (res);
2034 
2035  instance->my_high_delivered = i;
2036 
2037  for (i = 0; i <= instance->my_high_delivered; i++) {
2038  void *ptr;
2039 
2040  res = sq_item_get (&instance->regular_sort_queue, i, &ptr);
2041  if (res == 0) {
2042  struct sort_queue_item *regular_message;
2043 
2044  regular_message = ptr;
2045  free (regular_message->mcast);
2046  }
2047  }
2048  sq_items_release (&instance->regular_sort_queue, instance->my_high_delivered);
2049  instance->last_released = instance->my_high_delivered;
2050 
2051  if (joined_list_entries) {
2052  int sptr = 0;
2053  sptr += snprintf(joined_node_msg, sizeof(joined_node_msg)-sptr, " joined:");
2054  for (i=0; i< joined_list_entries; i++) {
2055  sptr += snprintf(joined_node_msg+sptr, sizeof(joined_node_msg)-sptr, " %u", joined_list_totemip[i]);
2056  }
2057  }
2058  else {
2059  joined_node_msg[0] = '\0';
2060  }
2061 
2062  if (instance->my_left_memb_entries) {
2063  int sptr = 0;
2064  int sptr2 = 0;
2065  sptr += snprintf(left_node_msg, sizeof(left_node_msg)-sptr, " left:");
2066  for (i=0; i< instance->my_left_memb_entries; i++) {
2067  sptr += snprintf(left_node_msg+sptr, sizeof(left_node_msg)-sptr, " %u", left_list[i]);
2068  }
2069  for (i=0; i< instance->my_left_memb_entries; i++) {
2070  if (my_leave_memb_match(instance, left_list[i]) == 0) {
2071  if (sptr2 == 0) {
2072  sptr2 += snprintf(failed_node_msg, sizeof(failed_node_msg)-sptr2, " failed:");
2073  }
2074  sptr2 += snprintf(failed_node_msg+sptr2, sizeof(left_node_msg)-sptr2, " %u", left_list[i]);
2075  }
2076  }
2077  if (sptr2 == 0) {
2078  failed_node_msg[0] = '\0';
2079  }
2080  }
2081  else {
2082  left_node_msg[0] = '\0';
2083  failed_node_msg[0] = '\0';
2084  }
2085 
2086  my_leave_memb_clear(instance);
2087 
2089  "entering OPERATIONAL state.");
2091  "A new membership (%s:%lld) was formed. Members%s%s",
2092  totemip_print (&instance->my_ring_id.rep),
2093  instance->my_ring_id.seq,
2094  joined_node_msg,
2095  left_node_msg);
2096 
2097  if (strlen(failed_node_msg)) {
2099  "Failed to receive the leave message.%s",
2100  failed_node_msg);
2101  }
2102 
2103  instance->memb_state = MEMB_STATE_OPERATIONAL;
2104 
2105  instance->stats.operational_entered++;
2106  instance->stats.continuous_gather = 0;
2107 
2108  instance->my_received_flg = 1;
2109 
2110  reset_pause_timeout (instance);
2111 
2112  /*
2113  * Save ring id information from this configuration to determine
2114  * which processors are transitioning from old regular configuration
2115  * in to new regular configuration on the next configuration change
2116  */
2117  memcpy (&instance->my_old_ring_id, &instance->my_ring_id,
2118  sizeof (struct memb_ring_id));
2119 
2120  return;
2121 }
2122 
2123 static void memb_state_gather_enter (
2124  struct totemsrp_instance *instance,
2125  enum gather_state_from gather_from)
2126 {
2127  instance->orf_token_discard = 1;
2128 
2129  instance->originated_orf_token = 0;
2130 
2131  memb_set_merge (
2132  &instance->my_id, 1,
2133  instance->my_proc_list, &instance->my_proc_list_entries);
2134 
2135  memb_join_message_send (instance);
2136 
2137  /*
2138  * Restart the join timeout
2139  */
2140  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);
2141 
2142  qb_loop_timer_add (instance->totemsrp_poll_handle,
2143  QB_LOOP_MED,
2144  instance->totem_config->join_timeout*QB_TIME_NS_IN_MSEC,
2145  (void *)instance,
2146  memb_timer_function_state_gather,
2148 
2149  /*
2150  * Restart the consensus timeout
2151  */
2152  qb_loop_timer_del (instance->totemsrp_poll_handle,
2154 
2155  qb_loop_timer_add (instance->totemsrp_poll_handle,
2156  QB_LOOP_MED,
2157  instance->totem_config->consensus_timeout*QB_TIME_NS_IN_MSEC,
2158  (void *)instance,
2159  memb_timer_function_gather_consensus_timeout,
2161 
2162  /*
2163  * Cancel the token loss and token retransmission timeouts
2164  */
2165  cancel_token_retransmit_timeout (instance); // REVIEWED
2166  cancel_token_timeout (instance); // REVIEWED
2167  cancel_merge_detect_timeout (instance);
2168 
2169  memb_consensus_reset (instance);
2170 
2171  memb_consensus_set (instance, &instance->my_id);
2172 
2174  "entering GATHER state from %d(%s).",
2175  gather_from, gsfrom_to_msg(gather_from));
2176 
2177  instance->memb_state = MEMB_STATE_GATHER;
2178  instance->stats.gather_entered++;
2179 
2180  if (gather_from == TOTEMSRP_GSFROM_THE_CONSENSUS_TIMEOUT_EXPIRED) {
2181  /*
2182  * State 3 means gather, so we are continuously gathering.
2183  */
2184  instance->stats.continuous_gather++;
2185  }
2186 
2187  return;
2188 }
2189 
2190 static void timer_function_token_retransmit_timeout (void *data);
2191 
2192 static void target_set_completed (
2193  void *context)
2194 {
2195  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
2196 
2197  memb_state_commit_token_send (instance);
2198 
2199 }
2200 
2201 static void memb_state_commit_enter (
2202  struct totemsrp_instance *instance)
2203 {
2204  old_ring_state_save (instance);
2205 
2206  memb_state_commit_token_update (instance);
2207 
2208  memb_state_commit_token_target_set (instance);
2209 
2210  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout);
2211 
2213 
2214  qb_loop_timer_del (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_consensus_timeout);
2215 
2217 
2218  memb_ring_id_set (instance, &instance->commit_token->ring_id);
2219  instance->memb_ring_id_store (&instance->my_ring_id, &instance->my_id.addr[0]);
2220 
2221  instance->token_ring_id_seq = instance->my_ring_id.seq;
2222 
2224  "entering COMMIT state.");
2225 
2226  instance->memb_state = MEMB_STATE_COMMIT;
2227  reset_token_retransmit_timeout (instance); // REVIEWED
2228  reset_token_timeout (instance); // REVIEWED
2229 
2230  instance->stats.commit_entered++;
2231  instance->stats.continuous_gather = 0;
2232 
2233  /*
2234  * reset all flow control variables since we are starting a new ring
2235  */
2236  instance->my_trc = 0;
2237  instance->my_pbl = 0;
2238  instance->my_cbl = 0;
2239  /*
2240  * commit token sent after callback that token target has been set
2241  */
2242 }
2243 
2244 static void memb_state_recovery_enter (
2245  struct totemsrp_instance *instance,
2246  struct memb_commit_token *commit_token)
2247 {
2248  int i;
2249  int local_received_flg = 1;
2250  unsigned int low_ring_aru;
2251  unsigned int range = 0;
2252  unsigned int messages_originated = 0;
2253  const struct srp_addr *addr;
2254  struct memb_commit_token_memb_entry *memb_list;
2255  struct memb_ring_id my_new_memb_ring_id_list[PROCESSOR_COUNT_MAX];
2256 
2257  addr = (const struct srp_addr *)commit_token->end_of_commit_token;
2258  memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries);
2259 
2261  "entering RECOVERY state.");
2262 
2263  instance->orf_token_discard = 0;
2264 
2265  instance->my_high_ring_delivered = 0;
2266 
2267  sq_reinit (&instance->recovery_sort_queue, SEQNO_START_MSG);
2268  cs_queue_reinit (&instance->retrans_message_queue);
2269 
2270  low_ring_aru = instance->old_ring_state_high_seq_received;
2271 
2272  memb_state_commit_token_send_recovery (instance, commit_token);
2273 
2274  instance->my_token_seq = SEQNO_START_TOKEN - 1;
2275 
2276  /*
2277  * Build regular configuration
2278  */
2280  instance->totemrrp_context,
2281  commit_token->addr_entries);
2282 
2283  /*
2284  * Build transitional configuration
2285  */
2286  for (i = 0; i < instance->my_new_memb_entries; i++) {
2287  memcpy (&my_new_memb_ring_id_list[i],
2288  &memb_list[i].ring_id,
2289  sizeof (struct memb_ring_id));
2290  }
2291  memb_set_and_with_ring_id (
2292  instance->my_new_memb_list,
2293  my_new_memb_ring_id_list,
2294  instance->my_new_memb_entries,
2295  instance->my_memb_list,
2296  instance->my_memb_entries,
2297  &instance->my_old_ring_id,
2298  instance->my_trans_memb_list,
2299  &instance->my_trans_memb_entries);
2300 
2301  for (i = 0; i < instance->my_trans_memb_entries; i++) {
2303  "TRANS [%d] member %s:", i, totemip_print (&instance->my_trans_memb_list[i].addr[0]));
2304  }
2305  for (i = 0; i < instance->my_new_memb_entries; i++) {
2307  "position [%d] member %s:", i, totemip_print (&addr[i].addr[0]));
2309  "previous ring seq %llx rep %s",
2310  memb_list[i].ring_id.seq,
2311  totemip_print (&memb_list[i].ring_id.rep));
2312 
2314  "aru %x high delivered %x received flag %d",
2315  memb_list[i].aru,
2316  memb_list[i].high_delivered,
2317  memb_list[i].received_flg);
2318 
2319  // assert (totemip_print (&memb_list[i].ring_id.rep) != 0);
2320  }
2321  /*
2322  * Determine if any received flag is false
2323  */
2324  for (i = 0; i < commit_token->addr_entries; i++) {
2325  if (memb_set_subset (&instance->my_new_memb_list[i], 1,
2326  instance->my_trans_memb_list, instance->my_trans_memb_entries) &&
2327 
2328  memb_list[i].received_flg == 0) {
2329  instance->my_deliver_memb_entries = instance->my_trans_memb_entries;
2330  memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list,
2331  sizeof (struct srp_addr) * instance->my_trans_memb_entries);
2332  local_received_flg = 0;
2333  break;
2334  }
2335  }
2336  if (local_received_flg == 1) {
2337  goto no_originate;
2338  } /* Else originate messages if we should */
2339 
2340  /*
2341  * Calculate my_low_ring_aru, instance->my_high_ring_delivered for the transitional membership
2342  */
2343  for (i = 0; i < commit_token->addr_entries; i++) {
2344  if (memb_set_subset (&instance->my_new_memb_list[i], 1,
2345  instance->my_deliver_memb_list,
2346  instance->my_deliver_memb_entries) &&
2347 
2348  memcmp (&instance->my_old_ring_id,
2349  &memb_list[i].ring_id,
2350  sizeof (struct memb_ring_id)) == 0) {
2351 
2352  if (sq_lt_compare (memb_list[i].aru, low_ring_aru)) {
2353 
2354  low_ring_aru = memb_list[i].aru;
2355  }
2356  if (sq_lt_compare (instance->my_high_ring_delivered, memb_list[i].high_delivered)) {
2357  instance->my_high_ring_delivered = memb_list[i].high_delivered;
2358  }
2359  }
2360  }
2361 
2362  /*
2363  * Copy all old ring messages to instance->retrans_message_queue
2364  */
2365  range = instance->old_ring_state_high_seq_received - low_ring_aru;
2366  if (range == 0) {
2367  /*
2368  * No messages to copy
2369  */
2370  goto no_originate;
2371  }
2372  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
2373 
2375  "copying all old ring messages from %x-%x.",
2376  low_ring_aru + 1, instance->old_ring_state_high_seq_received);
2377 
2378  for (i = 1; i <= range; i++) {
2380  struct message_item message_item;
2381  void *ptr;
2382  int res;
2383 
2384  res = sq_item_get (&instance->regular_sort_queue,
2385  low_ring_aru + i, &ptr);
2386  if (res != 0) {
2387  continue;
2388  }
2389  sort_queue_item = ptr;
2390  messages_originated++;
2391  memset (&message_item, 0, sizeof (struct message_item));
2392  // TODO LEAK
2393  message_item.mcast = totemsrp_buffer_alloc (instance);
2394  assert (message_item.mcast);
2395  message_item.mcast->header.type = MESSAGE_TYPE_MCAST;
2396  srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);
2398  message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid;
2399  assert (message_item.mcast->header.nodeid);
2400  message_item.mcast->header.endian_detector = ENDIAN_LOCAL;
2401  memcpy (&message_item.mcast->ring_id, &instance->my_ring_id,
2402  sizeof (struct memb_ring_id));
2403  message_item.msg_len = sort_queue_item->msg_len + sizeof (struct mcast);
2404  memcpy (((char *)message_item.mcast) + sizeof (struct mcast),
2405  sort_queue_item->mcast,
2406  sort_queue_item->msg_len);
2407  cs_queue_item_add (&instance->retrans_message_queue, &message_item);
2408  }
2410  "Originated %d messages in RECOVERY.", messages_originated);
2411  goto originated;
2412 
2413 no_originate:
2415  "Did not need to originate any messages in recovery.");
2416 
2417 originated:
2418  instance->my_aru = SEQNO_START_MSG;
2419  instance->my_aru_count = 0;
2420  instance->my_seq_unchanged = 0;
2422  instance->my_install_seq = SEQNO_START_MSG;
2423  instance->last_released = SEQNO_START_MSG;
2424 
2425  reset_token_timeout (instance); // REVIEWED
2426  reset_token_retransmit_timeout (instance); // REVIEWED
2427 
2428  instance->memb_state = MEMB_STATE_RECOVERY;
2429  instance->stats.recovery_entered++;
2430  instance->stats.continuous_gather = 0;
2431 
2432  return;
2433 }
2434 
2435 void totemsrp_event_signal (void *srp_context, enum totem_event_type type, int value)
2436 {
2437  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
2438 
2439  token_hold_cancel_send (instance);
2440 
2441  return;
2442 }
2443 
2445  void *srp_context,
2446  struct iovec *iovec,
2447  unsigned int iov_len,
2448  int guarantee)
2449 {
2450  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
2451  int i;
2452  struct message_item message_item;
2453  char *addr;
2454  unsigned int addr_idx;
2455  struct cs_queue *queue_use;
2456 
2457  if (instance->waiting_trans_ack) {
2458  queue_use = &instance->new_message_queue_trans;
2459  } else {
2460  queue_use = &instance->new_message_queue;
2461  }
2462 
2463  if (cs_queue_is_full (queue_use)) {
2464  log_printf (instance->totemsrp_log_level_debug, "queue full");
2465  return (-1);
2466  }
2467 
2468  memset (&message_item, 0, sizeof (struct message_item));
2469 
2470  /*
2471  * Allocate pending item
2472  */
2473  message_item.mcast = totemsrp_buffer_alloc (instance);
2474  if (message_item.mcast == 0) {
2475  goto error_mcast;
2476  }
2477 
2478  /*
2479  * Set mcast header
2480  */
2481  memset(message_item.mcast, 0, sizeof (struct mcast));
2482  message_item.mcast->header.type = MESSAGE_TYPE_MCAST;
2483  message_item.mcast->header.endian_detector = ENDIAN_LOCAL;
2485  message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid;
2486  assert (message_item.mcast->header.nodeid);
2487 
2488  message_item.mcast->guarantee = guarantee;
2489  srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);
2490 
2491  addr = (char *)message_item.mcast;
2492  addr_idx = sizeof (struct mcast);
2493  for (i = 0; i < iov_len; i++) {
2494  memcpy (&addr[addr_idx], iovec[i].iov_base, iovec[i].iov_len);
2495  addr_idx += iovec[i].iov_len;
2496  }
2497 
2498  message_item.msg_len = addr_idx;
2499 
2500  log_printf (instance->totemsrp_log_level_trace, "mcasted message added to pending queue");
2501  instance->stats.mcast_tx++;
2502  cs_queue_item_add (queue_use, &message_item);
2503 
2504  return (0);
2505 
2506 error_mcast:
2507  return (-1);
2508 }
2509 
2510 /*
2511  * Determine if there is room to queue a new message
2512  */
2513 int totemsrp_avail (void *srp_context)
2514 {
2515  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
2516  int avail;
2517  struct cs_queue *queue_use;
2518 
2519  if (instance->waiting_trans_ack) {
2520  queue_use = &instance->new_message_queue_trans;
2521  } else {
2522  queue_use = &instance->new_message_queue;
2523  }
2524  cs_queue_avail (queue_use, &avail);
2525 
2526  return (avail);
2527 }
2528 
2529 /*
2530  * ORF Token Management
2531  */
2532 /*
2533  * Recast message to mcast group if it is available
2534  */
2535 static int orf_token_remcast (
2536  struct totemsrp_instance *instance,
2537  int seq)
2538 {
2540  int res;
2541  void *ptr;
2542 
2543  struct sq *sort_queue;
2544 
2545  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2546  sort_queue = &instance->recovery_sort_queue;
2547  } else {
2548  sort_queue = &instance->regular_sort_queue;
2549  }
2550 
2551  res = sq_in_range (sort_queue, seq);
2552  if (res == 0) {
2553  log_printf (instance->totemsrp_log_level_debug, "sq not in range");
2554  return (-1);
2555  }
2556 
2557  /*
2558  * Get RTR item at seq, if not available, return
2559  */
2560  res = sq_item_get (sort_queue, seq, &ptr);
2561  if (res != 0) {
2562  return -1;
2563  }
2564 
2565  sort_queue_item = ptr;
2566 
2568  instance->totemrrp_context,
2569  sort_queue_item->mcast,
2570  sort_queue_item->msg_len);
2571 
2572  return (0);
2573 }
2574 
2575 
2576 /*
2577  * Free all freeable messages from ring
2578  */
2579 static void messages_free (
2580  struct totemsrp_instance *instance,
2581  unsigned int token_aru)
2582 {
2583  struct sort_queue_item *regular_message;
2584  unsigned int i;
2585  int res;
2586  int log_release = 0;
2587  unsigned int release_to;
2588  unsigned int range = 0;
2589 
2590  release_to = token_aru;
2591  if (sq_lt_compare (instance->my_last_aru, release_to)) {
2592  release_to = instance->my_last_aru;
2593  }
2594  if (sq_lt_compare (instance->my_high_delivered, release_to)) {
2595  release_to = instance->my_high_delivered;
2596  }
2597 
2598  /*
2599  * Ensure we dont try release before an already released point
2600  */
2601  if (sq_lt_compare (release_to, instance->last_released)) {
2602  return;
2603  }
2604 
2605  range = release_to - instance->last_released;
2606  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
2607 
2608  /*
2609  * Release retransmit list items if group aru indicates they are transmitted
2610  */
2611  for (i = 1; i <= range; i++) {
2612  void *ptr;
2613 
2614  res = sq_item_get (&instance->regular_sort_queue,
2615  instance->last_released + i, &ptr);
2616  if (res == 0) {
2617  regular_message = ptr;
2618  totemsrp_buffer_release (instance, regular_message->mcast);
2619  }
2620  sq_items_release (&instance->regular_sort_queue,
2621  instance->last_released + i);
2622 
2623  log_release = 1;
2624  }
2625  instance->last_released += range;
2626 
2627  if (log_release) {
2629  "releasing messages up to and including %x", release_to);
2630  }
2631 }
2632 
2633 static void update_aru (
2634  struct totemsrp_instance *instance)
2635 {
2636  unsigned int i;
2637  int res;
2638  struct sq *sort_queue;
2639  unsigned int range;
2640  unsigned int my_aru_saved = 0;
2641 
2642  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2643  sort_queue = &instance->recovery_sort_queue;
2644  } else {
2645  sort_queue = &instance->regular_sort_queue;
2646  }
2647 
2648  range = instance->my_high_seq_received - instance->my_aru;
2649 
2650  my_aru_saved = instance->my_aru;
2651  for (i = 1; i <= range; i++) {
2652 
2653  void *ptr;
2654 
2655  res = sq_item_get (sort_queue, my_aru_saved + i, &ptr);
2656  /*
2657  * If hole, stop updating aru
2658  */
2659  if (res != 0) {
2660  break;
2661  }
2662  }
2663  instance->my_aru += i - 1;
2664 }
2665 
2666 /*
2667  * Multicasts pending messages onto the ring (requires orf_token possession)
2668  */
2669 static int orf_token_mcast (
2670  struct totemsrp_instance *instance,
2671  struct orf_token *token,
2672  int fcc_mcasts_allowed)
2673 {
2674  struct message_item *message_item = 0;
2675  struct cs_queue *mcast_queue;
2676  struct sq *sort_queue;
2677  struct sort_queue_item sort_queue_item;
2678  struct mcast *mcast;
2679  unsigned int fcc_mcast_current;
2680 
2681  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2682  mcast_queue = &instance->retrans_message_queue;
2683  sort_queue = &instance->recovery_sort_queue;
2684  reset_token_retransmit_timeout (instance); // REVIEWED
2685  } else {
2686  if (instance->waiting_trans_ack) {
2687  mcast_queue = &instance->new_message_queue_trans;
2688  } else {
2689  mcast_queue = &instance->new_message_queue;
2690  }
2691 
2692  sort_queue = &instance->regular_sort_queue;
2693  }
2694 
2695  for (fcc_mcast_current = 0; fcc_mcast_current < fcc_mcasts_allowed; fcc_mcast_current++) {
2696  if (cs_queue_is_empty (mcast_queue)) {
2697  break;
2698  }
2699  message_item = (struct message_item *)cs_queue_item_get (mcast_queue);
2700 
2701  message_item->mcast->seq = ++token->seq;
2702  message_item->mcast->this_seqno = instance->global_seqno++;
2703 
2704  /*
2705  * Build IO vector
2706  */
2707  memset (&sort_queue_item, 0, sizeof (struct sort_queue_item));
2708  sort_queue_item.mcast = message_item->mcast;
2709  sort_queue_item.msg_len = message_item->msg_len;
2710 
2711  mcast = sort_queue_item.mcast;
2712 
2713  memcpy (&mcast->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id));
2714 
2715  /*
2716  * Add message to retransmit queue
2717  */
2718  sq_item_add (sort_queue, &sort_queue_item, message_item->mcast->seq);
2719 
2721  instance->totemrrp_context,
2722  message_item->mcast,
2723  message_item->msg_len);
2724 
2725  /*
2726  * Delete item from pending queue
2727  */
2728  cs_queue_item_remove (mcast_queue);
2729 
2730  /*
2731  * If messages mcasted, deliver any new messages to totempg
2732  */
2733  instance->my_high_seq_received = token->seq;
2734  }
2735 
2736  update_aru (instance);
2737 
2738  /*
2739  * Return 1 if more messages are available for single node clusters
2740  */
2741  return (fcc_mcast_current);
2742 }
2743 
2744 /*
2745  * Remulticasts messages in orf_token's retransmit list (requires orf_token)
2746  * Modify's orf_token's rtr to include retransmits required by this process
2747  */
2748 static int orf_token_rtr (
2749  struct totemsrp_instance *instance,
2750  struct orf_token *orf_token,
2751  unsigned int *fcc_allowed)
2752 {
2753  unsigned int res;
2754  unsigned int i, j;
2755  unsigned int found;
2756  struct sq *sort_queue;
2757  struct rtr_item *rtr_list;
2758  unsigned int range = 0;
2759  char retransmit_msg[1024];
2760  char value[64];
2761 
2762  if (instance->memb_state == MEMB_STATE_RECOVERY) {
2763  sort_queue = &instance->recovery_sort_queue;
2764  } else {
2765  sort_queue = &instance->regular_sort_queue;
2766  }
2767 
2768  rtr_list = &orf_token->rtr_list[0];
2769 
2770  strcpy (retransmit_msg, "Retransmit List: ");
2771  if (orf_token->rtr_list_entries) {
2773  "Retransmit List %d", orf_token->rtr_list_entries);
2774  for (i = 0; i < orf_token->rtr_list_entries; i++) {
2775  sprintf (value, "%x ", rtr_list[i].seq);
2776  strcat (retransmit_msg, value);
2777  }
2778  strcat (retransmit_msg, "");
2780  "%s", retransmit_msg);
2781  }
2782 
2783  /*
2784  * Retransmit messages on orf_token's RTR list from RTR queue
2785  */
2786  for (instance->fcc_remcast_current = 0, i = 0;
2787  instance->fcc_remcast_current < *fcc_allowed && i < orf_token->rtr_list_entries;) {
2788 
2789  /*
2790  * If this retransmit request isn't from this configuration,
2791  * try next rtr entry
2792  */
2793  if (memcmp (&rtr_list[i].ring_id, &instance->my_ring_id,
2794  sizeof (struct memb_ring_id)) != 0) {
2795 
2796  i += 1;
2797  continue;
2798  }
2799 
2800  res = orf_token_remcast (instance, rtr_list[i].seq);
2801  if (res == 0) {
2802  /*
2803  * Multicasted message, so no need to copy to new retransmit list
2804  */
2805  orf_token->rtr_list_entries -= 1;
2806  assert (orf_token->rtr_list_entries >= 0);
2807  memmove (&rtr_list[i], &rtr_list[i + 1],
2808  sizeof (struct rtr_item) * (orf_token->rtr_list_entries - i));
2809 
2810  instance->stats.mcast_retx++;
2811  instance->fcc_remcast_current++;
2812  } else {
2813  i += 1;
2814  }
2815  }
2816  *fcc_allowed = *fcc_allowed - instance->fcc_remcast_current;
2817 
2818  /*
2819  * Add messages to retransmit to RTR list
2820  * but only retry if there is room in the retransmit list
2821  */
2822 
2823  range = orf_token->seq - instance->my_aru;
2824  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
2825 
2826  for (i = 1; (orf_token->rtr_list_entries < RETRANSMIT_ENTRIES_MAX) &&
2827  (i <= range); i++) {
2828 
2829  /*
2830  * Ensure message is within the sort queue range
2831  */
2832  res = sq_in_range (sort_queue, instance->my_aru + i);
2833  if (res == 0) {
2834  break;
2835  }
2836 
2837  /*
2838  * Find if a message is missing from this processor
2839  */
2840  res = sq_item_inuse (sort_queue, instance->my_aru + i);
2841  if (res == 0) {
2842  /*
2843  * Determine how many times we have missed receiving
2844  * this sequence number. sq_item_miss_count increments
2845  * a counter for the sequence number. The miss count
2846  * will be returned and compared. This allows time for
2847  * delayed multicast messages to be received before
2848  * declaring the message is missing and requesting a
2849  * retransmit.
2850  */
2851  res = sq_item_miss_count (sort_queue, instance->my_aru + i);
2852  if (res < instance->totem_config->miss_count_const) {
2853  continue;
2854  }
2855 
2856  /*
2857  * Determine if missing message is already in retransmit list
2858  */
2859  found = 0;
2860  for (j = 0; j < orf_token->rtr_list_entries; j++) {
2861  if (instance->my_aru + i == rtr_list[j].seq) {
2862  found = 1;
2863  }
2864  }
2865  if (found == 0) {
2866  /*
2867  * Missing message not found in current retransmit list so add it
2868  */
2869  memcpy (&rtr_list[orf_token->rtr_list_entries].ring_id,
2870  &instance->my_ring_id, sizeof (struct memb_ring_id));
2871  rtr_list[orf_token->rtr_list_entries].seq = instance->my_aru + i;
2872  orf_token->rtr_list_entries++;
2873  }
2874  }
2875  }
2876  return (instance->fcc_remcast_current);
2877 }
2878 
2879 static void token_retransmit (struct totemsrp_instance *instance)
2880 {
2882  instance->orf_token_retransmit,
2883  instance->orf_token_retransmit_size);
2884 }
2885 
2886 /*
2887  * Retransmit the regular token if no mcast or token has
2888  * been received in retransmit token period retransmit
2889  * the token to the next processor
2890  */
2891 static void timer_function_token_retransmit_timeout (void *data)
2892 {
2893  struct totemsrp_instance *instance = data;
2894 
2895  switch (instance->memb_state) {
2896  case MEMB_STATE_GATHER:
2897  break;
2898  case MEMB_STATE_COMMIT:
2900  case MEMB_STATE_RECOVERY:
2901  token_retransmit (instance);
2902  reset_token_retransmit_timeout (instance); // REVIEWED
2903  break;
2904  }
2905 }
2906 
2907 static void timer_function_token_hold_retransmit_timeout (void *data)
2908 {
2909  struct totemsrp_instance *instance = data;
2910 
2911  switch (instance->memb_state) {
2912  case MEMB_STATE_GATHER:
2913  break;
2914  case MEMB_STATE_COMMIT:
2915  break;
2917  case MEMB_STATE_RECOVERY:
2918  token_retransmit (instance);
2919  break;
2920  }
2921 }
2922 
2923 static void timer_function_merge_detect_timeout(void *data)
2924 {
2925  struct totemsrp_instance *instance = data;
2926 
2928 
2929  switch (instance->memb_state) {
2931  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
2932  memb_merge_detect_transmit (instance);
2933  }
2934  break;
2935  case MEMB_STATE_GATHER:
2936  case MEMB_STATE_COMMIT:
2937  case MEMB_STATE_RECOVERY:
2938  break;
2939  }
2940 }
2941 
2942 /*
2943  * Send orf_token to next member (requires orf_token)
2944  */
2945 static int token_send (
2946  struct totemsrp_instance *instance,
2947  struct orf_token *orf_token,
2948  int forward_token)
2949 {
2950  int res = 0;
2951  unsigned int orf_token_size;
2952 
2953  orf_token_size = sizeof (struct orf_token) +
2954  (orf_token->rtr_list_entries * sizeof (struct rtr_item));
2955 
2956  orf_token->header.nodeid = instance->my_id.addr[0].nodeid;
2957  memcpy (instance->orf_token_retransmit, orf_token, orf_token_size);
2958  instance->orf_token_retransmit_size = orf_token_size;
2959  assert (orf_token->header.nodeid);
2960 
2961  if (forward_token == 0) {
2962  return (0);
2963  }
2964 
2966  orf_token,
2967  orf_token_size);
2968 
2969  return (res);
2970 }
2971 
2972 static int token_hold_cancel_send (struct totemsrp_instance *instance)
2973 {
2974  struct token_hold_cancel token_hold_cancel;
2975 
2976  /*
2977  * Only cancel if the token is currently held
2978  */
2979  if (instance->my_token_held == 0) {
2980  return (0);
2981  }
2982  instance->my_token_held = 0;
2983 
2984  /*
2985  * Build message
2986  */
2987  token_hold_cancel.header.type = MESSAGE_TYPE_TOKEN_HOLD_CANCEL;
2988  token_hold_cancel.header.endian_detector = ENDIAN_LOCAL;
2989  token_hold_cancel.header.encapsulated = 0;
2990  token_hold_cancel.header.nodeid = instance->my_id.addr[0].nodeid;
2991  memcpy (&token_hold_cancel.ring_id, &instance->my_ring_id,
2992  sizeof (struct memb_ring_id));
2993  assert (token_hold_cancel.header.nodeid);
2994 
2995  instance->stats.token_hold_cancel_tx++;
2996 
2997  totemrrp_mcast_flush_send (instance->totemrrp_context, &token_hold_cancel,
2998  sizeof (struct token_hold_cancel));
2999 
3000  return (0);
3001 }
3002 
3003 static int orf_token_send_initial (struct totemsrp_instance *instance)
3004 {
3005  struct orf_token orf_token;
3006  int res;
3007 
3008  orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN;
3009  orf_token.header.endian_detector = ENDIAN_LOCAL;
3010  orf_token.header.encapsulated = 0;
3011  orf_token.header.nodeid = instance->my_id.addr[0].nodeid;
3012  assert (orf_token.header.nodeid);
3013  orf_token.seq = SEQNO_START_MSG;
3014  orf_token.token_seq = SEQNO_START_TOKEN;
3015  orf_token.retrans_flg = 1;
3016  instance->my_set_retrans_flg = 1;
3017  instance->stats.orf_token_tx++;
3018 
3019  if (cs_queue_is_empty (&instance->retrans_message_queue) == 1) {
3020  orf_token.retrans_flg = 0;
3021  instance->my_set_retrans_flg = 0;
3022  } else {
3023  orf_token.retrans_flg = 1;
3024  instance->my_set_retrans_flg = 1;
3025  }
3026 
3027  orf_token.aru = 0;
3028  orf_token.aru = SEQNO_START_MSG - 1;
3029  orf_token.aru_addr = instance->my_id.addr[0].nodeid;
3030 
3031  memcpy (&orf_token.ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id));
3032  orf_token.fcc = 0;
3033  orf_token.backlog = 0;
3034 
3035  orf_token.rtr_list_entries = 0;
3036 
3037  res = token_send (instance, &orf_token, 1);
3038 
3039  return (res);
3040 }
3041 
3042 static void memb_state_commit_token_update (
3043  struct totemsrp_instance *instance)
3044 {
3045  struct srp_addr *addr;
3046  struct memb_commit_token_memb_entry *memb_list;
3047  unsigned int high_aru;
3048  unsigned int i;
3049 
3050  addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
3051  memb_list = (struct memb_commit_token_memb_entry *)(addr + instance->commit_token->addr_entries);
3052 
3053  memcpy (instance->my_new_memb_list, addr,
3054  sizeof (struct srp_addr) * instance->commit_token->addr_entries);
3055 
3056  instance->my_new_memb_entries = instance->commit_token->addr_entries;
3057 
3058  memcpy (&memb_list[instance->commit_token->memb_index].ring_id,
3059  &instance->my_old_ring_id, sizeof (struct memb_ring_id));
3060 
3061  memb_list[instance->commit_token->memb_index].aru = instance->old_ring_state_aru;
3062  /*
3063  * TODO high delivered is really instance->my_aru, but with safe this
3064  * could change?
3065  */
3066  instance->my_received_flg =
3067  (instance->my_aru == instance->my_high_seq_received);
3068 
3069  memb_list[instance->commit_token->memb_index].received_flg = instance->my_received_flg;
3070 
3071  memb_list[instance->commit_token->memb_index].high_delivered = instance->my_high_delivered;
3072  /*
3073  * find high aru up to current memb_index for all matching ring ids
3074  * if any ring id matching memb_index has aru less then high aru set
3075  * received flag for that entry to false
3076  */
3077  high_aru = memb_list[instance->commit_token->memb_index].aru;
3078  for (i = 0; i <= instance->commit_token->memb_index; i++) {
3079  if (memcmp (&memb_list[instance->commit_token->memb_index].ring_id,
3080  &memb_list[i].ring_id,
3081  sizeof (struct memb_ring_id)) == 0) {
3082 
3083  if (sq_lt_compare (high_aru, memb_list[i].aru)) {
3084  high_aru = memb_list[i].aru;
3085  }
3086  }
3087  }
3088 
3089  for (i = 0; i <= instance->commit_token->memb_index; i++) {
3090  if (memcmp (&memb_list[instance->commit_token->memb_index].ring_id,
3091  &memb_list[i].ring_id,
3092  sizeof (struct memb_ring_id)) == 0) {
3093 
3094  if (sq_lt_compare (memb_list[i].aru, high_aru)) {
3095  memb_list[i].received_flg = 0;
3096  if (i == instance->commit_token->memb_index) {
3097  instance->my_received_flg = 0;
3098  }
3099  }
3100  }
3101  }
3102 
3103  instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3104  instance->commit_token->memb_index += 1;
3105  assert (instance->commit_token->memb_index <= instance->commit_token->addr_entries);
3106  assert (instance->commit_token->header.nodeid);
3107 }
3108 
3109 static void memb_state_commit_token_target_set (
3110  struct totemsrp_instance *instance)
3111 {
3112  struct srp_addr *addr;
3113  unsigned int i;
3114 
3115  addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
3116 
3117  for (i = 0; i < instance->totem_config->interface_count; i++) {
3119  instance->totemrrp_context,
3120  &addr[instance->commit_token->memb_index %
3121  instance->commit_token->addr_entries].addr[i],
3122  i);
3123  }
3124 }
3125 
3126 static int memb_state_commit_token_send_recovery (
3127  struct totemsrp_instance *instance,
3128  struct memb_commit_token *commit_token)
3129 {
3130  unsigned int commit_token_size;
3131 
3132  commit_token->token_seq++;
3133  commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3134  commit_token_size = sizeof (struct memb_commit_token) +
3135  ((sizeof (struct srp_addr) +
3136  sizeof (struct memb_commit_token_memb_entry)) * commit_token->addr_entries);
3137  /*
3138  * Make a copy for retransmission if necessary
3139  */
3140  memcpy (instance->orf_token_retransmit, commit_token, commit_token_size);
3141  instance->orf_token_retransmit_size = commit_token_size;
3142 
3143  instance->stats.memb_commit_token_tx++;
3144 
3146  commit_token,
3147  commit_token_size);
3148 
3149  /*
3150  * Request retransmission of the commit token in case it is lost
3151  */
3152  reset_token_retransmit_timeout (instance);
3153  return (0);
3154 }
3155 
3156 static int memb_state_commit_token_send (
3157  struct totemsrp_instance *instance)
3158 {
3159  unsigned int commit_token_size;
3160 
3161  instance->commit_token->token_seq++;
3162  instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3163  commit_token_size = sizeof (struct memb_commit_token) +
3164  ((sizeof (struct srp_addr) +
3165  sizeof (struct memb_commit_token_memb_entry)) * instance->commit_token->addr_entries);
3166  /*
3167  * Make a copy for retransmission if necessary
3168  */
3169  memcpy (instance->orf_token_retransmit, instance->commit_token, commit_token_size);
3170  instance->orf_token_retransmit_size = commit_token_size;
3171 
3172  instance->stats.memb_commit_token_tx++;
3173 
3175  instance->commit_token,
3176  commit_token_size);
3177 
3178  /*
3179  * Request retransmission of the commit token in case it is lost
3180  */
3181  reset_token_retransmit_timeout (instance);
3182  return (0);
3183 }
3184 
3185 
3186 static int memb_lowest_in_config (struct totemsrp_instance *instance)
3187 {
3188  struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
3189  int token_memb_entries = 0;
3190  int i;
3191  struct totem_ip_address *lowest_addr;
3192 
3193  memb_set_subtract (token_memb, &token_memb_entries,
3194  instance->my_proc_list, instance->my_proc_list_entries,
3195  instance->my_failed_list, instance->my_failed_list_entries);
3196 
3197  /*
3198  * find representative by searching for smallest identifier
3199  */
3200 
3201  lowest_addr = &token_memb[0].addr[0];
3202  for (i = 1; i < token_memb_entries; i++) {
3203  if (totemip_compare(lowest_addr, &token_memb[i].addr[0]) > 0) {
3204  totemip_copy (lowest_addr, &token_memb[i].addr[0]);
3205  }
3206  }
3207  return (totemip_compare (lowest_addr, &instance->my_id.addr[0]) == 0);
3208 }
3209 
3210 static int srp_addr_compare (const void *a, const void *b)
3211 {
3212  const struct srp_addr *srp_a = (const struct srp_addr *)a;
3213  const struct srp_addr *srp_b = (const struct srp_addr *)b;
3214 
3215  return (totemip_compare (&srp_a->addr[0], &srp_b->addr[0]));
3216 }
3217 
3218 static void memb_state_commit_token_create (
3219  struct totemsrp_instance *instance)
3220 {
3221  struct srp_addr token_memb[PROCESSOR_COUNT_MAX];
3222  struct srp_addr *addr;
3223  struct memb_commit_token_memb_entry *memb_list;
3224  int token_memb_entries = 0;
3225 
3227  "Creating commit token because I am the rep.");
3228 
3229  memb_set_subtract (token_memb, &token_memb_entries,
3230  instance->my_proc_list, instance->my_proc_list_entries,
3231  instance->my_failed_list, instance->my_failed_list_entries);
3232 
3233  memset (instance->commit_token, 0, sizeof (struct memb_commit_token));
3236  instance->commit_token->header.encapsulated = 0;
3237  instance->commit_token->header.nodeid = instance->my_id.addr[0].nodeid;
3238  assert (instance->commit_token->header.nodeid);
3239 
3240  totemip_copy(&instance->commit_token->ring_id.rep, &instance->my_id.addr[0]);
3241 
3242  instance->commit_token->ring_id.seq = instance->token_ring_id_seq + 4;
3243 
3244  /*
3245  * This qsort is necessary to ensure the commit token traverses
3246  * the ring in the proper order
3247  */
3248  qsort (token_memb, token_memb_entries, sizeof (struct srp_addr),
3249  srp_addr_compare);
3250 
3251  instance->commit_token->memb_index = 0;
3252  instance->commit_token->addr_entries = token_memb_entries;
3253 
3254  addr = (struct srp_addr *)instance->commit_token->end_of_commit_token;
3255  memb_list = (struct memb_commit_token_memb_entry *)(addr + instance->commit_token->addr_entries);
3256 
3257  memcpy (addr, token_memb,
3258  token_memb_entries * sizeof (struct srp_addr));
3259  memset (memb_list, 0,
3260  sizeof (struct memb_commit_token_memb_entry) * token_memb_entries);
3261 }
3262 
3263 static void memb_join_message_send (struct totemsrp_instance *instance)
3264 {
3265  char memb_join_data[40000];
3266  struct memb_join *memb_join = (struct memb_join *)memb_join_data;
3267  char *addr;
3268  unsigned int addr_idx;
3269  size_t msg_len;
3270 
3271  memb_join->header.type = MESSAGE_TYPE_MEMB_JOIN;
3272  memb_join->header.endian_detector = ENDIAN_LOCAL;
3273  memb_join->header.encapsulated = 0;
3274  memb_join->header.nodeid = instance->my_id.addr[0].nodeid;
3275  assert (memb_join->header.nodeid);
3276 
3277  msg_len = sizeof(struct memb_join) +
3278  ((instance->my_proc_list_entries + instance->my_failed_list_entries) * sizeof(struct srp_addr));
3279 
3280  if (msg_len > sizeof(memb_join_data)) {
3281  log_printf (instance->totemsrp_log_level_error,
3282  "memb_join_message too long. Ignoring message.");
3283 
3284  return ;
3285  }
3286 
3287  memb_join->ring_seq = instance->my_ring_id.seq;
3288  memb_join->proc_list_entries = instance->my_proc_list_entries;
3289  memb_join->failed_list_entries = instance->my_failed_list_entries;
3290  srp_addr_copy (&memb_join->system_from, &instance->my_id);
3291 
3292  /*
3293  * This mess adds the joined and failed processor lists into the join
3294  * message
3295  */
3296  addr = (char *)memb_join;
3297  addr_idx = sizeof (struct memb_join);
3298  memcpy (&addr[addr_idx],
3299  instance->my_proc_list,
3300  instance->my_proc_list_entries *
3301  sizeof (struct srp_addr));
3302  addr_idx +=
3303  instance->my_proc_list_entries *
3304  sizeof (struct srp_addr);
3305  memcpy (&addr[addr_idx],
3306  instance->my_failed_list,
3307  instance->my_failed_list_entries *
3308  sizeof (struct srp_addr));
3309  addr_idx +=
3310  instance->my_failed_list_entries *
3311  sizeof (struct srp_addr);
3312 
3313  if (instance->totem_config->send_join_timeout) {
3314  usleep (random() % (instance->totem_config->send_join_timeout * 1000));
3315  }
3316 
3317  instance->stats.memb_join_tx++;
3318 
3320  instance->totemrrp_context,
3321  memb_join,
3322  addr_idx);
3323 }
3324 
3325 static void memb_leave_message_send (struct totemsrp_instance *instance)
3326 {
3327  char memb_join_data[40000];
3328  struct memb_join *memb_join = (struct memb_join *)memb_join_data;
3329  char *addr;
3330  unsigned int addr_idx;
3331  int active_memb_entries;
3332  struct srp_addr active_memb[PROCESSOR_COUNT_MAX];
3333  size_t msg_len;
3334 
3336  "sending join/leave message");
3337 
3338  /*
3339  * add us to the failed list, and remove us from
3340  * the members list
3341  */
3342  memb_set_merge(
3343  &instance->my_id, 1,
3344  instance->my_failed_list, &instance->my_failed_list_entries);
3345 
3346  memb_set_subtract (active_memb, &active_memb_entries,
3347  instance->my_proc_list, instance->my_proc_list_entries,
3348  &instance->my_id, 1);
3349 
3350  msg_len = sizeof(struct memb_join) +
3351  ((active_memb_entries + instance->my_failed_list_entries) * sizeof(struct srp_addr));
3352 
3353  if (msg_len > sizeof(memb_join_data)) {
3354  log_printf (instance->totemsrp_log_level_error,
3355  "memb_leave message too long. Ignoring message.");
3356 
3357  return ;
3358  }
3359 
3360  memb_join->header.type = MESSAGE_TYPE_MEMB_JOIN;
3361  memb_join->header.endian_detector = ENDIAN_LOCAL;
3362  memb_join->header.encapsulated = 0;
3363  memb_join->header.nodeid = LEAVE_DUMMY_NODEID;
3364 
3365  memb_join->ring_seq = instance->my_ring_id.seq;
3366  memb_join->proc_list_entries = active_memb_entries;
3367  memb_join->failed_list_entries = instance->my_failed_list_entries;
3368  srp_addr_copy (&memb_join->system_from, &instance->my_id);
3369  memb_join->system_from.addr[0].nodeid = LEAVE_DUMMY_NODEID;
3370 
3371  // TODO: CC Maybe use the actual join send routine.
3372  /*
3373  * This mess adds the joined and failed processor lists into the join
3374  * message
3375  */
3376  addr = (char *)memb_join;
3377  addr_idx = sizeof (struct memb_join);
3378  memcpy (&addr[addr_idx],
3379  active_memb,
3380  active_memb_entries *
3381  sizeof (struct srp_addr));
3382  addr_idx +=
3383  active_memb_entries *
3384  sizeof (struct srp_addr);
3385  memcpy (&addr[addr_idx],
3386  instance->my_failed_list,
3387  instance->my_failed_list_entries *
3388  sizeof (struct srp_addr));
3389  addr_idx +=
3390  instance->my_failed_list_entries *
3391  sizeof (struct srp_addr);
3392 
3393 
3394  if (instance->totem_config->send_join_timeout) {
3395  usleep (random() % (instance->totem_config->send_join_timeout * 1000));
3396  }
3397  instance->stats.memb_join_tx++;
3398 
3400  instance->totemrrp_context,
3401  memb_join,
3402  addr_idx);
3403 }
3404 
3405 static void memb_merge_detect_transmit (struct totemsrp_instance *instance)
3406 {
3407  struct memb_merge_detect memb_merge_detect;
3408 
3409  memb_merge_detect.header.type = MESSAGE_TYPE_MEMB_MERGE_DETECT;
3410  memb_merge_detect.header.endian_detector = ENDIAN_LOCAL;
3411  memb_merge_detect.header.encapsulated = 0;
3412  memb_merge_detect.header.nodeid = instance->my_id.addr[0].nodeid;
3413  srp_addr_copy (&memb_merge_detect.system_from, &instance->my_id);
3414  memcpy (&memb_merge_detect.ring_id, &instance->my_ring_id,
3415  sizeof (struct memb_ring_id));
3416  assert (memb_merge_detect.header.nodeid);
3417 
3418  instance->stats.memb_merge_detect_tx++;
3420  &memb_merge_detect,
3421  sizeof (struct memb_merge_detect));
3422 }
3423 
3424 static void memb_ring_id_set (
3425  struct totemsrp_instance *instance,
3426  const struct memb_ring_id *ring_id)
3427 {
3428 
3429  memcpy (&instance->my_ring_id, ring_id, sizeof (struct memb_ring_id));
3430 }
3431 
3433  void *srp_context,
3434  void **handle_out,
3435  enum totem_callback_token_type type,
3436  int delete,
3437  int (*callback_fn) (enum totem_callback_token_type type, const void *),
3438  const void *data)
3439 {
3440  struct totemsrp_instance *instance = (struct totemsrp_instance *)srp_context;
3441  struct token_callback_instance *callback_handle;
3442 
3443  token_hold_cancel_send (instance);
3444 
3445  callback_handle = malloc (sizeof (struct token_callback_instance));
3446  if (callback_handle == 0) {
3447  return (-1);
3448  }
3449  *handle_out = (void *)callback_handle;
3450  list_init (&callback_handle->list);
3451  callback_handle->callback_fn = callback_fn;
3452  callback_handle->data = (void *) data;
3453  callback_handle->callback_type = type;
3454  callback_handle->delete = delete;
3455  switch (type) {
3457  list_add (&callback_handle->list, &instance->token_callback_received_listhead);
3458  break;
3460  list_add (&callback_handle->list, &instance->token_callback_sent_listhead);
3461  break;
3462  }
3463 
3464  return (0);
3465 }
3466 
3467 void totemsrp_callback_token_destroy (void *srp_context, void **handle_out)
3468 {
3469  struct token_callback_instance *h;
3470 
3471  if (*handle_out) {
3472  h = (struct token_callback_instance *)*handle_out;
3473  list_del (&h->list);
3474  free (h);
3475  h = NULL;
3476  *handle_out = 0;
3477  }
3478 }
3479 
3480 static void token_callbacks_execute (
3481  struct totemsrp_instance *instance,
3482  enum totem_callback_token_type type)
3483 {
3484  struct list_head *list;
3485  struct list_head *list_next;
3486  struct list_head *callback_listhead = 0;
3488  int res;
3489  int del;
3490 
3491  switch (type) {
3493  callback_listhead = &instance->token_callback_received_listhead;
3494  break;
3496  callback_listhead = &instance->token_callback_sent_listhead;
3497  break;
3498  default:
3499  assert (0);
3500  }
3501 
3502  for (list = callback_listhead->next; list != callback_listhead;
3503  list = list_next) {
3504 
3505  token_callback_instance = list_entry (list, struct token_callback_instance, list);
3506 
3507  list_next = list->next;
3508  del = token_callback_instance->delete;
3509  if (del == 1) {
3510  list_del (list);
3511  }
3512 
3513  res = token_callback_instance->callback_fn (
3514  token_callback_instance->callback_type,
3515  token_callback_instance->data);
3516  /*
3517  * This callback failed to execute, try it again on the next token
3518  */
3519  if (res == -1 && del == 1) {
3520  list_add (list, callback_listhead);
3521  } else if (del) {
3522  free (token_callback_instance);
3523  }
3524  }
3525 }
3526 
3527 /*
3528  * Flow control functions
3529  */
3530 static unsigned int backlog_get (struct totemsrp_instance *instance)
3531 {
3532  unsigned int backlog = 0;
3533  struct cs_queue *queue_use = NULL;
3534 
3535  if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
3536  if (instance->waiting_trans_ack) {
3537  queue_use = &instance->new_message_queue_trans;
3538  } else {
3539  queue_use = &instance->new_message_queue;
3540  }
3541  } else
3542  if (instance->memb_state == MEMB_STATE_RECOVERY) {
3543  queue_use = &instance->retrans_message_queue;
3544  }
3545 
3546  if (queue_use != NULL) {
3547  backlog = cs_queue_used (queue_use);
3548  }
3549 
3550  instance->stats.token[instance->stats.latest_token].backlog_calc = backlog;
3551  return (backlog);
3552 }
3553 
3554 static int fcc_calculate (
3555  struct totemsrp_instance *instance,
3556  struct orf_token *token)
3557 {
3558  unsigned int transmits_allowed;
3559  unsigned int backlog_calc;
3560 
3561  transmits_allowed = instance->totem_config->max_messages;
3562 
3563  if (transmits_allowed > instance->totem_config->window_size - token->fcc) {
3564  transmits_allowed = instance->totem_config->window_size - token->fcc;
3565  }
3566 
3567  instance->my_cbl = backlog_get (instance);
3568 
3569  /*
3570  * Only do backlog calculation if there is a backlog otherwise
3571  * we would result in div by zero
3572  */
3573  if (token->backlog + instance->my_cbl - instance->my_pbl) {
3574  backlog_calc = (instance->totem_config->window_size * instance->my_pbl) /
3575  (token->backlog + instance->my_cbl - instance->my_pbl);
3576  if (backlog_calc > 0 && transmits_allowed > backlog_calc) {
3577  transmits_allowed = backlog_calc;
3578  }
3579  }
3580 
3581  return (transmits_allowed);
3582 }
3583 
3584 /*
3585  * don't overflow the RTR sort queue
3586  */
3587 static void fcc_rtr_limit (
3588  struct totemsrp_instance *instance,
3589  struct orf_token *token,
3590  unsigned int *transmits_allowed)
3591 {
3592  int check = QUEUE_RTR_ITEMS_SIZE_MAX;
3593  check -= (*transmits_allowed + instance->totem_config->window_size);
3594  assert (check >= 0);
3595  if (sq_lt_compare (instance->last_released +
3596  QUEUE_RTR_ITEMS_SIZE_MAX - *transmits_allowed -
3597  instance->totem_config->window_size,
3598 
3599  token->seq)) {
3600 
3601  *transmits_allowed = 0;
3602  }
3603 }
3604 
3605 static void fcc_token_update (
3606  struct totemsrp_instance *instance,
3607  struct orf_token *token,
3608  unsigned int msgs_transmitted)
3609 {
3610  token->fcc += msgs_transmitted - instance->my_trc;
3611  token->backlog += instance->my_cbl - instance->my_pbl;
3612  instance->my_trc = msgs_transmitted;
3613  instance->my_pbl = instance->my_cbl;
3614 }
3615 
3616 /*
3617  * Sanity checkers
3618  */
3619 static int check_totemip_sanity(
3620  const struct totemsrp_instance *instance,
3621  const struct totem_ip_address *addr,
3622  int endian_conversion_needed)
3623 {
3624  unsigned short family;
3625 
3626  family = addr->family;
3627  if (endian_conversion_needed) {
3628  family = swab16(family);
3629  }
3630 
3631  if (family != AF_INET && family != AF_INET6) {
3633  "Received message corrupted... ignoring.");
3634 
3635  return (-1);
3636  }
3637 
3638  return (0);
3639 }
3640 
3641 static int check_srpaddr_sanity(
3642  const struct totemsrp_instance *instance,
3643  const struct srp_addr *addr,
3644  int endian_conversion_needed)
3645 {
3646  int i;
3647 
3648  if (addr->no_addrs < 1 || addr->no_addrs > INTERFACE_MAX) {
3649  return (-1);
3650  }
3651 
3652  for (i = 0; i < addr->no_addrs; i++) {
3653  if (i == 0 || addr->addr[i].family != 0) {
3654  if (check_totemip_sanity(instance, &addr->addr[i], endian_conversion_needed) == -1) {
3655  return (-1);
3656  }
3657  }
3658  }
3659 
3660  return (0);
3661 }
3662 
3663 static int check_orf_token_sanity(
3664  const struct totemsrp_instance *instance,
3665  const void *msg,
3666  size_t msg_len,
3667  int endian_conversion_needed)
3668 {
3669  int rtr_entries;
3670  const struct orf_token *token = (const struct orf_token *)msg;
3671  size_t required_len;
3672  int i;
3673 
3674  if (msg_len < sizeof(struct orf_token)) {
3676  "Received orf_token message is too short... ignoring.");
3677 
3678  return (-1);
3679  }
3680 
3681  if (check_totemip_sanity(instance, &token->ring_id.rep, endian_conversion_needed) == -1) {
3682  return (-1);
3683  }
3684 
3685  if (endian_conversion_needed) {
3686  rtr_entries = swab32(token->rtr_list_entries);
3687  } else {
3688  rtr_entries = token->rtr_list_entries;
3689  }
3690 
3691  required_len = sizeof(struct orf_token) + rtr_entries * sizeof(struct rtr_item);
3692  if (msg_len < required_len) {
3694  "Received orf_token message is too short... ignoring.");
3695 
3696  return (-1);
3697  }
3698 
3699  for (i = 0; i < rtr_entries; i++) {
3700  if (check_totemip_sanity(instance, &token->rtr_list[i].ring_id.rep,
3701  endian_conversion_needed) == -1) {
3702  return (-1);
3703  }
3704  }
3705 
3706  return (0);
3707 }
3708 
3709 static int check_mcast_sanity(
3710  struct totemsrp_instance *instance,
3711  const void *msg,
3712  size_t msg_len,
3713  int endian_conversion_needed)
3714 {
3715  const struct mcast *mcast_msg = (const struct mcast *)msg;
3716 
3717  if (msg_len < sizeof(struct mcast)) {
3719  "Received mcast message is too short... ignoring.");
3720 
3721  return (-1);
3722  }
3723 
3724  if ((check_totemip_sanity(instance, &mcast_msg->ring_id.rep, endian_conversion_needed) == -1) ||
3725  (check_srpaddr_sanity(instance, &mcast_msg->system_from, endian_conversion_needed) == -1)) {
3726  return (-1);
3727  }
3728 
3729  return (0);
3730 }
3731 
3732 static int check_memb_merge_detect_sanity(
3733  struct totemsrp_instance *instance,
3734  const void *msg,
3735  size_t msg_len,
3736  int endian_conversion_needed)
3737 {
3738  const struct memb_merge_detect *mmd_msg = (const struct memb_merge_detect *)msg;
3739 
3740  if (msg_len < sizeof(struct memb_merge_detect)) {
3742  "Received memb_merge_detect message is too short... ignoring.");
3743 
3744  return (-1);
3745  }
3746 
3747  if ((check_totemip_sanity(instance, &mmd_msg->ring_id.rep, endian_conversion_needed) == -1) ||
3748  (check_srpaddr_sanity(instance, &mmd_msg->system_from, endian_conversion_needed) == -1)) {
3749  return (-1);
3750  }
3751 
3752  return (0);
3753 }
3754 
3755 static int check_memb_join_sanity(
3756  struct totemsrp_instance *instance,
3757  const void *msg,
3758  size_t msg_len,
3759  int endian_conversion_needed)
3760 {
3761  const struct memb_join *mj_msg = (const struct memb_join *)msg;
3762  unsigned int proc_list_entries;
3763  unsigned int failed_list_entries;
3764  size_t required_len;
3765  const struct srp_addr *proc_list;
3766  const struct srp_addr *failed_list;
3767  int i;
3768 
3769  if (msg_len < sizeof(struct memb_join)) {
3771  "Received memb_join message is too short... ignoring.");
3772 
3773  return (-1);
3774  }
3775 
3776  if (check_srpaddr_sanity(instance, &mj_msg->system_from, endian_conversion_needed) == -1) {
3777  return (-1);
3778  }
3779 
3780  proc_list_entries = mj_msg->proc_list_entries;
3781  failed_list_entries = mj_msg->failed_list_entries;
3782 
3783  if (endian_conversion_needed) {
3784  proc_list_entries = swab32(proc_list_entries);
3785  failed_list_entries = swab32(failed_list_entries);
3786  }
3787 
3788  required_len = sizeof(struct memb_join) + ((proc_list_entries + failed_list_entries) * sizeof(struct srp_addr));
3789  if (msg_len < required_len) {
3791  "Received memb_join message is too short... ignoring.");
3792 
3793  return (-1);
3794  }
3795 
3796  proc_list = (struct srp_addr *)mj_msg->end_of_memb_join;
3797  failed_list = proc_list + proc_list_entries;
3798 
3799  for (i = 0; i < proc_list_entries; i++) {
3800  if (check_srpaddr_sanity(instance, &proc_list[i], endian_conversion_needed) == -1) {
3801  return (-1);
3802  }
3803  }
3804 
3805  for (i = 0; i < failed_list_entries; i++) {
3806  if (check_srpaddr_sanity(instance, &failed_list[i], endian_conversion_needed) == -1) {
3807  return (-1);
3808  }
3809  }
3810 
3811  return (0);
3812 }
3813 
3814 static int check_memb_commit_token_sanity(
3815  struct totemsrp_instance *instance,
3816  const void *msg,
3817  size_t msg_len,
3818  int endian_conversion_needed)
3819 {
3820  const struct memb_commit_token *mct_msg = (const struct memb_commit_token *)msg;
3821  unsigned int addr_entries;
3822  const struct srp_addr *addr;
3823  const struct memb_commit_token_memb_entry *memb_list;
3824  size_t required_len;
3825  int i;
3826 
3827  if (msg_len < sizeof(struct memb_commit_token)) {
3829  "Received memb_commit_token message is too short... ignoring.");
3830 
3831  return (0);
3832  }
3833 
3834  if (check_totemip_sanity(instance, &mct_msg->ring_id.rep, endian_conversion_needed) == -1) {
3835  return (-1);
3836  }
3837 
3838  addr_entries= mct_msg->addr_entries;
3839  if (endian_conversion_needed) {
3840  addr_entries = swab32(addr_entries);
3841  }
3842 
3843  required_len = sizeof(struct memb_commit_token) +
3844  (addr_entries * (sizeof(struct srp_addr) + sizeof(struct memb_commit_token_memb_entry)));
3845  if (msg_len < required_len) {
3847  "Received memb_commit_token message is too short... ignoring.");
3848 
3849  return (-1);
3850  }
3851 
3852  addr = (const struct srp_addr *)mct_msg->end_of_commit_token;
3853  memb_list = (const struct memb_commit_token_memb_entry *)(addr + addr_entries);
3854 
3855  for (i = 0; i < addr_entries; i++) {
3856  if (check_srpaddr_sanity(instance, &addr[i], endian_conversion_needed) == -1) {
3857  return (-1);
3858  }
3859 
3860  if (memb_list[i].ring_id.rep.family != 0) {
3861  if (check_totemip_sanity(instance, &memb_list[i].ring_id.rep,
3862  endian_conversion_needed) == -1) {
3863  return (-1);
3864  }
3865  }
3866  }
3867 
3868  return (0);
3869 }
3870 
3871 static int check_token_hold_cancel_sanity(
3872  struct totemsrp_instance *instance,
3873  const void *msg,
3874  size_t msg_len,
3875  int endian_conversion_needed)
3876 {
3877  const struct token_hold_cancel *thc_msg = (const struct token_hold_cancel *)msg;
3878 
3879  if (msg_len < sizeof(struct token_hold_cancel)) {
3881  "Received token_hold_cancel message is too short... ignoring.");
3882 
3883  return (-1);
3884  }
3885 
3886  if (check_totemip_sanity(instance, &thc_msg->ring_id.rep, endian_conversion_needed) == -1) {
3887  return (-1);
3888  }
3889 
3890  return (0);
3891 }
3892 
3893 /*
3894  * Message Handlers
3895  */
3896 
3897 unsigned long long int tv_old;
3898 /*
3899  * message handler called when TOKEN message type received
3900  */
3901 static int message_handler_orf_token (
3902  struct totemsrp_instance *instance,
3903  const void *msg,
3904  size_t msg_len,
3905  int endian_conversion_needed)
3906 {
3907  char token_storage[1500];
3908  char token_convert[1500];
3909  struct orf_token *token = NULL;
3910  int forward_token;
3911  unsigned int transmits_allowed;
3912  unsigned int mcasted_retransmit;
3913  unsigned int mcasted_regular;
3914  unsigned int last_aru;
3915 
3916 #ifdef GIVEINFO
3917  unsigned long long tv_current;
3918  unsigned long long tv_diff;
3919 
3920  tv_current = qb_util_nano_current_get ();
3921  tv_diff = tv_current - tv_old;
3922  tv_old = tv_current;
3923 
3925  "Time since last token %0.4f ms", ((float)tv_diff) / 1000000.0);
3926 #endif
3927 
3928  if (check_orf_token_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
3929  return (0);
3930  }
3931 
3932  if (instance->orf_token_discard) {
3933  return (0);
3934  }
3935 #ifdef TEST_DROP_ORF_TOKEN_PERCENTAGE
3936  if (random()%100 < TEST_DROP_ORF_TOKEN_PERCENTAGE) {
3937  return (0);
3938  }
3939 #endif
3940 
3941  if (endian_conversion_needed) {
3942  orf_token_endian_convert ((struct orf_token *)msg,
3943  (struct orf_token *)token_convert);
3944  msg = (struct orf_token *)token_convert;
3945  }
3946 
3947  /*
3948  * Make copy of token and retransmit list in case we have
3949  * to flush incoming messages from the kernel queue
3950  */
3951  token = (struct orf_token *)token_storage;
3952  memcpy (token, msg, sizeof (struct orf_token));
3953  memcpy (&token->rtr_list[0], (char *)msg + sizeof (struct orf_token),
3954  sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX);
3955 
3956 
3957  /*
3958  * Handle merge detection timeout
3959  */
3960  if (token->seq == instance->my_last_seq) {
3961  start_merge_detect_timeout (instance);
3962  instance->my_seq_unchanged += 1;
3963  } else {
3964  cancel_merge_detect_timeout (instance);
3965  cancel_token_hold_retransmit_timeout (instance);
3966  instance->my_seq_unchanged = 0;
3967  }
3968 
3969  instance->my_last_seq = token->seq;
3970 
3971 #ifdef TEST_RECOVERY_MSG_COUNT
3972  if (instance->memb_state == MEMB_STATE_OPERATIONAL && token->seq > TEST_RECOVERY_MSG_COUNT) {
3973  return (0);
3974  }
3975 #endif
3976  instance->flushing = 1;
3978  instance->flushing = 0;
3979 
3980  /*
3981  * Determine if we should hold (in reality drop) the token
3982  */
3983  instance->my_token_held = 0;
3984  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) &&
3985  instance->my_seq_unchanged > instance->totem_config->seqno_unchanged_const) {
3986  instance->my_token_held = 1;
3987  } else
3988  if (!totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) &&
3989  instance->my_seq_unchanged >= instance->totem_config->seqno_unchanged_const) {
3990  instance->my_token_held = 1;
3991  }
3992 
3993  /*
3994  * Hold onto token when there is no activity on ring and
3995  * this processor is the ring rep
3996  */
3997  forward_token = 1;
3998  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
3999  if (instance->my_token_held) {
4000  forward_token = 0;
4001  }
4002  }
4003 
4004  token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_RECEIVED);
4005 
4006  switch (instance->memb_state) {
4007  case MEMB_STATE_COMMIT:
4008  /* Discard token */
4009  break;
4010 
4012  messages_free (instance, token->aru);
4013  /*
4014  * Do NOT add break, this case should also execute code in gather case.
4015  */
4016 
4017  case MEMB_STATE_GATHER:
4018  /*
4019  * DO NOT add break, we use different free mechanism in recovery state
4020  */
4021 
4022  case MEMB_STATE_RECOVERY:
4023  /*
4024  * Discard tokens from another configuration
4025  */
4026  if (memcmp (&token->ring_id, &instance->my_ring_id,
4027  sizeof (struct memb_ring_id)) != 0) {
4028 
4029  if ((forward_token)
4030  && instance->use_heartbeat) {
4031  reset_heartbeat_timeout(instance);
4032  }
4033  else {
4034  cancel_heartbeat_timeout(instance);
4035  }
4036 
4037  return (0); /* discard token */
4038  }
4039 
4040  /*
4041  * Discard retransmitted tokens
4042  */
4043  if (sq_lte_compare (token->token_seq, instance->my_token_seq)) {
4044  return (0); /* discard token */
4045  }
4046  last_aru = instance->my_last_aru;
4047  instance->my_last_aru = token->aru;
4048 
4049  transmits_allowed = fcc_calculate (instance, token);
4050  mcasted_retransmit = orf_token_rtr (instance, token, &transmits_allowed);
4051 
4052  if (instance->my_token_held == 1 &&
4053  (token->rtr_list_entries > 0 || mcasted_retransmit > 0)) {
4054  instance->my_token_held = 0;
4055  forward_token = 1;
4056  }
4057 
4058  fcc_rtr_limit (instance, token, &transmits_allowed);
4059  mcasted_regular = orf_token_mcast (instance, token, transmits_allowed);
4060 /*
4061 if (mcasted_regular) {
4062 printf ("mcasted regular %d\n", mcasted_regular);
4063 printf ("token seq %d\n", token->seq);
4064 }
4065 */
4066  fcc_token_update (instance, token, mcasted_retransmit +
4067  mcasted_regular);
4068 
4069  if (sq_lt_compare (instance->my_aru, token->aru) ||
4070  instance->my_id.addr[0].nodeid == token->aru_addr ||
4071  token->aru_addr == 0) {
4072 
4073  token->aru = instance->my_aru;
4074  if (token->aru == token->seq) {
4075  token->aru_addr = 0;
4076  } else {
4077  token->aru_addr = instance->my_id.addr[0].nodeid;
4078  }
4079  }
4080  if (token->aru == last_aru && token->aru_addr != 0) {
4081  instance->my_aru_count += 1;
4082  } else {
4083  instance->my_aru_count = 0;
4084  }
4085 
4086  /*
4087  * We really don't follow specification there. In specification, OTHER nodes
4088  * detect failure of one node (based on aru_count) and my_id IS NEVER added
4089  * to failed list (so node never mark itself as failed)
4090  */
4091  if (instance->my_aru_count > instance->totem_config->fail_to_recv_const &&
4092  token->aru_addr == instance->my_id.addr[0].nodeid) {
4093 
4095  "FAILED TO RECEIVE");
4096 
4097  instance->failed_to_recv = 1;
4098 
4099  memb_set_merge (&instance->my_id, 1,
4100  instance->my_failed_list,
4101  &instance->my_failed_list_entries);
4102 
4103  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_FAILED_TO_RECEIVE);
4104  } else {
4105  instance->my_token_seq = token->token_seq;
4106  token->token_seq += 1;
4107 
4108  if (instance->memb_state == MEMB_STATE_RECOVERY) {
4109  /*
4110  * instance->my_aru == instance->my_high_seq_received means this processor
4111  * has recovered all messages it can recover
4112  * (ie: its retrans queue is empty)
4113  */
4114  if (cs_queue_is_empty (&instance->retrans_message_queue) == 0) {
4115 
4116  if (token->retrans_flg == 0) {
4117  token->retrans_flg = 1;
4118  instance->my_set_retrans_flg = 1;
4119  }
4120  } else
4121  if (token->retrans_flg == 1 && instance->my_set_retrans_flg) {
4122  token->retrans_flg = 0;
4123  instance->my_set_retrans_flg = 0;
4124  }
4126  "token retrans flag is %d my set retrans flag%d retrans queue empty %d count %d, aru %x",
4127  token->retrans_flg, instance->my_set_retrans_flg,
4128  cs_queue_is_empty (&instance->retrans_message_queue),
4129  instance->my_retrans_flg_count, token->aru);
4130  if (token->retrans_flg == 0) {
4131  instance->my_retrans_flg_count += 1;
4132  } else {
4133  instance->my_retrans_flg_count = 0;
4134  }
4135  if (instance->my_retrans_flg_count == 2) {
4136  instance->my_install_seq = token->seq;
4137  }
4139  "install seq %x aru %x high seq received %x",
4140  instance->my_install_seq, instance->my_aru, instance->my_high_seq_received);
4141  if (instance->my_retrans_flg_count >= 2 &&
4142  instance->my_received_flg == 0 &&
4143  sq_lte_compare (instance->my_install_seq, instance->my_aru)) {
4144  instance->my_received_flg = 1;
4145  instance->my_deliver_memb_entries = instance->my_trans_memb_entries;
4146  memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list,
4147  sizeof (struct totem_ip_address) * instance->my_trans_memb_entries);
4148  }
4149  if (instance->my_retrans_flg_count >= 3 &&
4150  sq_lte_compare (instance->my_install_seq, token->aru)) {
4151  instance->my_rotation_counter += 1;
4152  } else {
4153  instance->my_rotation_counter = 0;
4154  }
4155  if (instance->my_rotation_counter == 2) {
4157  "retrans flag count %x token aru %x install seq %x aru %x %x",
4158  instance->my_retrans_flg_count, token->aru, instance->my_install_seq,
4159  instance->my_aru, token->seq);
4160 
4161  memb_state_operational_enter (instance);
4162  instance->my_rotation_counter = 0;
4163  instance->my_retrans_flg_count = 0;
4164  }
4165  }
4166 
4168  token_send (instance, token, forward_token);
4169 
4170 #ifdef GIVEINFO
4171  tv_current = qb_util_nano_current_get ();
4172  tv_diff = tv_current - tv_old;
4173  tv_old = tv_current;
4175  "I held %0.4f ms",
4176  ((float)tv_diff) / 1000000.0);
4177 #endif
4178  if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
4179  messages_deliver_to_app (instance, 0,
4180  instance->my_high_seq_received);
4181  }
4182 
4183  /*
4184  * Deliver messages after token has been transmitted
4185  * to improve performance
4186  */
4187  reset_token_timeout (instance); // REVIEWED
4188  reset_token_retransmit_timeout (instance); // REVIEWED
4189  if (totemip_equal(&instance->my_id.addr[0], &instance->my_ring_id.rep) &&
4190  instance->my_token_held == 1) {
4191 
4192  start_token_hold_retransmit_timeout (instance);
4193  }
4194 
4195  token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_SENT);
4196  }
4197  break;
4198  }
4199 
4200  if ((forward_token)
4201  && instance->use_heartbeat) {
4202  reset_heartbeat_timeout(instance);
4203  }
4204  else {
4205  cancel_heartbeat_timeout(instance);
4206  }
4207 
4208  return (0);
4209 }
4210 
4211 static void messages_deliver_to_app (
4212  struct totemsrp_instance *instance,
4213  int skip,
4214  unsigned int end_point)
4215 {
4216  struct sort_queue_item *sort_queue_item_p;
4217  unsigned int i;
4218  int res;
4219  struct mcast *mcast_in;
4220  struct mcast mcast_header;
4221  unsigned int range = 0;
4222  int endian_conversion_required;
4223  unsigned int my_high_delivered_stored = 0;
4224 
4225 
4226  range = end_point - instance->my_high_delivered;
4227 
4228  if (range) {
4230  "Delivering %x to %x", instance->my_high_delivered,
4231  end_point);
4232  }
4233  assert (range < QUEUE_RTR_ITEMS_SIZE_MAX);
4234  my_high_delivered_stored = instance->my_high_delivered;
4235 
4236  /*
4237  * Deliver messages in order from rtr queue to pending delivery queue
4238  */
4239  for (i = 1; i <= range; i++) {
4240 
4241  void *ptr = 0;
4242 
4243  /*
4244  * If out of range of sort queue, stop assembly
4245  */
4246  res = sq_in_range (&instance->regular_sort_queue,
4247  my_high_delivered_stored + i);
4248  if (res == 0) {
4249  break;
4250  }
4251 
4252  res = sq_item_get (&instance->regular_sort_queue,
4253  my_high_delivered_stored + i, &ptr);
4254  /*
4255  * If hole, stop assembly
4256  */
4257  if (res != 0 && skip == 0) {
4258  break;
4259  }
4260 
4261  instance->my_high_delivered = my_high_delivered_stored + i;
4262 
4263  if (res != 0) {
4264  continue;
4265 
4266  }
4267 
4268  sort_queue_item_p = ptr;
4269 
4270  mcast_in = sort_queue_item_p->mcast;
4271  assert (mcast_in != (struct mcast *)0xdeadbeef);
4272 
4273  endian_conversion_required = 0;
4274  if (mcast_in->header.endian_detector != ENDIAN_LOCAL) {
4275  endian_conversion_required = 1;
4276  mcast_endian_convert (mcast_in, &mcast_header);
4277  } else {
4278  memcpy (&mcast_header, mcast_in, sizeof (struct mcast));
4279  }
4280 
4281  /*
4282  * Skip messages not originated in instance->my_deliver_memb
4283  */
4284  if (skip &&
4285  memb_set_subset (&mcast_header.system_from,
4286  1,
4287  instance->my_deliver_memb_list,
4288  instance->my_deliver_memb_entries) == 0) {
4289 
4290  instance->my_high_delivered = my_high_delivered_stored + i;
4291 
4292  continue;
4293  }
4294 
4295  /*
4296  * Message found
4297  */
4299  "Delivering MCAST message with seq %x to pending delivery queue",
4300  mcast_header.seq);
4301 
4302  /*
4303  * Message is locally originated multicast
4304  */
4305  instance->totemsrp_deliver_fn (
4306  mcast_header.header.nodeid,
4307  ((char *)sort_queue_item_p->mcast) + sizeof (struct mcast),
4308  sort_queue_item_p->msg_len - sizeof (struct mcast),
4309  endian_conversion_required);
4310  }
4311 }
4312 
4313 /*
4314  * recv message handler called when MCAST message type received
4315  */
4316 static int message_handler_mcast (
4317  struct totemsrp_instance *instance,
4318  const void *msg,
4319  size_t msg_len,
4320  int endian_conversion_needed)
4321 {
4322  struct sort_queue_item sort_queue_item;
4323  struct sq *sort_queue;
4324  struct mcast mcast_header;
4325 
4326  if (check_mcast_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4327  return (0);
4328  }
4329 
4330  if (endian_conversion_needed) {
4331  mcast_endian_convert (msg, &mcast_header);
4332  } else {
4333  memcpy (&mcast_header, msg, sizeof (struct mcast));
4334  }
4335 
4336  if (mcast_header.header.encapsulated == MESSAGE_ENCAPSULATED) {
4337  sort_queue = &instance->recovery_sort_queue;
4338  } else {
4339  sort_queue = &instance->regular_sort_queue;
4340  }
4341 
4342  assert (msg_len <= FRAME_SIZE_MAX);
4343 
4344 #ifdef TEST_DROP_MCAST_PERCENTAGE
4345  if (random()%100 < TEST_DROP_MCAST_PERCENTAGE) {
4346  return (0);
4347  }
4348 #endif
4349 
4350  /*
4351  * If the message is foreign execute the switch below
4352  */
4353  if (memcmp (&instance->my_ring_id, &mcast_header.ring_id,
4354  sizeof (struct memb_ring_id)) != 0) {
4355 
4356  switch (instance->memb_state) {
4358  memb_set_merge (
4359  &mcast_header.system_from, 1,
4360  instance->my_proc_list, &instance->my_proc_list_entries);
4361  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_FOREIGN_MESSAGE_IN_OPERATIONAL_STATE);
4362  break;
4363 
4364  case MEMB_STATE_GATHER:
4365  if (!memb_set_subset (
4366  &mcast_header.system_from,
4367  1,
4368  instance->my_proc_list,
4369  instance->my_proc_list_entries)) {
4370 
4371  memb_set_merge (&mcast_header.system_from, 1,
4372  instance->my_proc_list, &instance->my_proc_list_entries);
4373  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_FOREIGN_MESSAGE_IN_GATHER_STATE);
4374  return (0);
4375  }
4376  break;
4377 
4378  case MEMB_STATE_COMMIT:
4379  /* discard message */
4380  instance->stats.rx_msg_dropped++;
4381  break;
4382 
4383  case MEMB_STATE_RECOVERY:
4384  /* discard message */
4385  instance->stats.rx_msg_dropped++;
4386  break;
4387  }
4388  return (0);
4389  }
4390 
4392  "Received ringid(%s:%lld) seq %x",
4393  totemip_print (&mcast_header.ring_id.rep),
4394  mcast_header.ring_id.seq,
4395  mcast_header.seq);
4396 
4397  /*
4398  * Add mcast message to rtr queue if not already in rtr queue
4399  * otherwise free io vectors
4400  */
4401  if (msg_len > 0 && msg_len <= FRAME_SIZE_MAX &&
4402  sq_in_range (sort_queue, mcast_header.seq) &&
4403  sq_item_inuse (sort_queue, mcast_header.seq) == 0) {
4404 
4405  /*
4406  * Allocate new multicast memory block
4407  */
4408 // TODO LEAK
4409  sort_queue_item.mcast = totemsrp_buffer_alloc (instance);
4410  if (sort_queue_item.mcast == NULL) {
4411  return (-1); /* error here is corrected by the algorithm */
4412  }
4413  memcpy (sort_queue_item.mcast, msg, msg_len);
4414  sort_queue_item.msg_len = msg_len;
4415 
4416  if (sq_lt_compare (instance->my_high_seq_received,
4417  mcast_header.seq)) {
4418  instance->my_high_seq_received = mcast_header.seq;
4419  }
4420 
4421  sq_item_add (sort_queue, &sort_queue_item, mcast_header.seq);
4422  }
4423 
4424  update_aru (instance);
4425  if (instance->memb_state == MEMB_STATE_OPERATIONAL) {
4426  messages_deliver_to_app (instance, 0, instance->my_high_seq_received);
4427  }
4428 
4429 /* TODO remove from retrans message queue for old ring in recovery state */
4430  return (0);
4431 }
4432 
4433 static int message_handler_memb_merge_detect (
4434  struct totemsrp_instance *instance,
4435  const void *msg,
4436  size_t msg_len,
4437  int endian_conversion_needed)
4438 {
4439  struct memb_merge_detect memb_merge_detect;
4440 
4441  if (check_memb_merge_detect_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4442  return (0);
4443  }
4444 
4445  if (endian_conversion_needed) {
4446  memb_merge_detect_endian_convert (msg, &memb_merge_detect);
4447  } else {
4448  memcpy (&memb_merge_detect, msg,
4449  sizeof (struct memb_merge_detect));
4450  }
4451 
4452  /*
4453  * do nothing if this is a merge detect from this configuration
4454  */
4455  if (memcmp (&instance->my_ring_id, &memb_merge_detect.ring_id,
4456  sizeof (struct memb_ring_id)) == 0) {
4457 
4458  return (0);
4459  }
4460 
4461  /*
4462  * Execute merge operation
4463  */
4464  switch (instance->memb_state) {
4466  memb_set_merge (&memb_merge_detect.system_from, 1,
4467  instance->my_proc_list, &instance->my_proc_list_entries);
4468  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_MERGE_DURING_OPERATIONAL_STATE);
4469  break;
4470 
4471  case MEMB_STATE_GATHER:
4472  if (!memb_set_subset (
4473  &memb_merge_detect.system_from,
4474  1,
4475  instance->my_proc_list,
4476  instance->my_proc_list_entries)) {
4477 
4478  memb_set_merge (&memb_merge_detect.system_from, 1,
4479  instance->my_proc_list, &instance->my_proc_list_entries);
4480  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_MERGE_DURING_GATHER_STATE);
4481  return (0);
4482  }
4483  break;
4484 
4485  case MEMB_STATE_COMMIT:
4486  /* do nothing in commit */
4487  break;
4488 
4489  case MEMB_STATE_RECOVERY:
4490  /* do nothing in recovery */
4491  break;
4492  }
4493  return (0);
4494 }
4495 
4496 static void memb_join_process (
4497  struct totemsrp_instance *instance,
4498  const struct memb_join *memb_join)
4499 {
4500  struct srp_addr *proc_list;
4501  struct srp_addr *failed_list;
4502  int gather_entered = 0;
4503  int fail_minus_memb_entries = 0;
4504  struct srp_addr fail_minus_memb[PROCESSOR_COUNT_MAX];
4505 
4506  proc_list = (struct srp_addr *)memb_join->end_of_memb_join;
4507  failed_list = proc_list + memb_join->proc_list_entries;
4508 
4509 /*
4510  memb_set_print ("proclist", proc_list, memb_join->proc_list_entries);
4511  memb_set_print ("faillist", failed_list, memb_join->failed_list_entries);
4512  memb_set_print ("my_proclist", instance->my_proc_list, instance->my_proc_list_entries);
4513  memb_set_print ("my_faillist", instance->my_failed_list, instance->my_failed_list_entries);
4514 -*/
4515 
4516  if (memb_join->header.type == MESSAGE_TYPE_MEMB_JOIN) {
4517  if (instance->flushing) {
4518  if (memb_join->header.nodeid == LEAVE_DUMMY_NODEID) {
4520  "Discarding LEAVE message during flush, nodeid=%u",
4521  memb_join->failed_list_entries > 0 ? failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid : LEAVE_DUMMY_NODEID);
4522  if (memb_join->failed_list_entries > 0) {
4523  my_leave_memb_set(instance, failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid);
4524  }
4525  } else {
4527  "Discarding JOIN message during flush, nodeid=%d", memb_join->header.nodeid);
4528  }
4529  return;
4530  } else {
4531  if (memb_join->header.nodeid == LEAVE_DUMMY_NODEID) {
4533  "Recieve LEAVE message from %u", memb_join->failed_list_entries > 0 ? failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid : LEAVE_DUMMY_NODEID);
4534  if (memb_join->failed_list_entries > 0) {
4535  my_leave_memb_set(instance, failed_list[memb_join->failed_list_entries - 1 ].addr[0].nodeid);
4536  }
4537  }
4538  }
4539 
4540  }
4541 
4542  if (memb_set_equal (proc_list,
4543  memb_join->proc_list_entries,
4544  instance->my_proc_list,
4545  instance->my_proc_list_entries) &&
4546 
4547  memb_set_equal (failed_list,
4548  memb_join->failed_list_entries,
4549  instance->my_failed_list,
4550  instance->my_failed_list_entries)) {
4551 
4552  memb_consensus_set (instance, &memb_join->system_from);
4553 
4554  if (memb_consensus_agreed (instance) && instance->failed_to_recv == 1) {
4555  instance->failed_to_recv = 0;
4556  srp_addr_copy (&instance->my_proc_list[0],
4557  &instance->my_id);
4558  instance->my_proc_list_entries = 1;
4559  instance->my_failed_list_entries = 0;
4560 
4561  memb_state_commit_token_create (instance);
4562 
4563  memb_state_commit_enter (instance);
4564  return;
4565  }
4566  if (memb_consensus_agreed (instance) &&
4567  memb_lowest_in_config (instance)) {
4568 
4569  memb_state_commit_token_create (instance);
4570 
4571  memb_state_commit_enter (instance);
4572  } else {
4573  goto out;
4574  }
4575  } else
4576  if (memb_set_subset (proc_list,
4577  memb_join->proc_list_entries,
4578  instance->my_proc_list,
4579  instance->my_proc_list_entries) &&
4580 
4581  memb_set_subset (failed_list,
4582  memb_join->failed_list_entries,
4583  instance->my_failed_list,
4584  instance->my_failed_list_entries)) {
4585 
4586  goto out;
4587  } else
4588  if (memb_set_subset (&memb_join->system_from, 1,
4589  instance->my_failed_list, instance->my_failed_list_entries)) {
4590 
4591  goto out;
4592  } else {
4593  memb_set_merge (proc_list,
4594  memb_join->proc_list_entries,
4595  instance->my_proc_list, &instance->my_proc_list_entries);
4596 
4597  if (memb_set_subset (
4598  &instance->my_id, 1,
4599  failed_list, memb_join->failed_list_entries)) {
4600 
4601  memb_set_merge (
4602  &memb_join->system_from, 1,
4603  instance->my_failed_list, &instance->my_failed_list_entries);
4604  } else {
4605  if (memb_set_subset (
4606  &memb_join->system_from, 1,
4607  instance->my_memb_list,
4608  instance->my_memb_entries)) {
4609 
4610  if (memb_set_subset (
4611  &memb_join->system_from, 1,
4612  instance->my_failed_list,
4613  instance->my_failed_list_entries) == 0) {
4614 
4615  memb_set_merge (failed_list,
4616  memb_join->failed_list_entries,
4617  instance->my_failed_list, &instance->my_failed_list_entries);
4618  } else {
4619  memb_set_subtract (fail_minus_memb,
4620  &fail_minus_memb_entries,
4621  failed_list,
4622  memb_join->failed_list_entries,
4623  instance->my_memb_list,
4624  instance->my_memb_entries);
4625 
4626  memb_set_merge (fail_minus_memb,
4627  fail_minus_memb_entries,
4628  instance->my_failed_list,
4629  &instance->my_failed_list_entries);
4630  }
4631  }
4632  }
4633  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_MERGE_DURING_JOIN);
4634  gather_entered = 1;
4635  }
4636 
4637 out:
4638  if (gather_entered == 0 &&
4639  instance->memb_state == MEMB_STATE_OPERATIONAL) {
4640 
4641  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_JOIN_DURING_OPERATIONAL_STATE);
4642  }
4643 }
4644 
4645 static void memb_join_endian_convert (const struct memb_join *in, struct memb_join *out)
4646 {
4647  int i;
4648  struct srp_addr *in_proc_list;
4649  struct srp_addr *in_failed_list;
4650  struct srp_addr *out_proc_list;
4651  struct srp_addr *out_failed_list;
4652 
4653  out->header.type = in->header.type;
4655  out->header.nodeid = swab32 (in->header.nodeid);
4656  srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
4659  out->ring_seq = swab64 (in->ring_seq);
4660 
4661  in_proc_list = (struct srp_addr *)in->end_of_memb_join;
4662  in_failed_list = in_proc_list + out->proc_list_entries;
4663  out_proc_list = (struct srp_addr *)out->end_of_memb_join;
4664  out_failed_list = out_proc_list + out->proc_list_entries;
4665 
4666  for (i = 0; i < out->proc_list_entries; i++) {
4667  srp_addr_copy_endian_convert (&out_proc_list[i], &in_proc_list[i]);
4668  }
4669  for (i = 0; i < out->failed_list_entries; i++) {
4670  srp_addr_copy_endian_convert (&out_failed_list[i], &in_failed_list[i]);
4671  }
4672 }
4673 
4674 static void memb_commit_token_endian_convert (const struct memb_commit_token *in, struct memb_commit_token *out)
4675 {
4676  int i;
4677  struct srp_addr *in_addr = (struct srp_addr *)in->end_of_commit_token;
4678  struct srp_addr *out_addr = (struct srp_addr *)out->end_of_commit_token;
4679  struct memb_commit_token_memb_entry *in_memb_list;
4680  struct memb_commit_token_memb_entry *out_memb_list;
4681 
4682  out->header.type = in->header.type;
4684  out->header.nodeid = swab32 (in->header.nodeid);
4685  out->token_seq = swab32 (in->token_seq);
4687  out->ring_id.seq = swab64 (in->ring_id.seq);
4688  out->retrans_flg = swab32 (in->retrans_flg);
4689  out->memb_index = swab32 (in->memb_index);
4690  out->addr_entries = swab32 (in->addr_entries);
4691 
4692  in_memb_list = (struct memb_commit_token_memb_entry *)(in_addr + out->addr_entries);
4693  out_memb_list = (struct memb_commit_token_memb_entry *)(out_addr + out->addr_entries);
4694  for (i = 0; i < out->addr_entries; i++) {
4695  srp_addr_copy_endian_convert (&out_addr[i], &in_addr[i]);
4696 
4697  /*
4698  * Only convert the memb entry if it has been set
4699  */
4700  if (in_memb_list[i].ring_id.rep.family != 0) {
4701  totemip_copy_endian_convert (&out_memb_list[i].ring_id.rep,
4702  &in_memb_list[i].ring_id.rep);
4703 
4704  out_memb_list[i].ring_id.seq =
4705  swab64 (in_memb_list[i].ring_id.seq);
4706  out_memb_list[i].aru = swab32 (in_memb_list[i].aru);
4707  out_memb_list[i].high_delivered = swab32 (in_memb_list[i].high_delivered);
4708  out_memb_list[i].received_flg = swab32 (in_memb_list[i].received_flg);
4709  }
4710  }
4711 }
4712 
4713 static void orf_token_endian_convert (const struct orf_token *in, struct orf_token *out)
4714 {
4715  int i;
4716 
4717  out->header.type = in->header.type;
4719  out->header.nodeid = swab32 (in->header.nodeid);
4720  out->seq = swab32 (in->seq);
4721  out->token_seq = swab32 (in->token_seq);
4722  out->aru = swab32 (in->aru);
4724  out->aru_addr = swab32(in->aru_addr);
4725  out->ring_id.seq = swab64 (in->ring_id.seq);
4726  out->fcc = swab32 (in->fcc);
4727  out->backlog = swab32 (in->backlog);
4728  out->retrans_flg = swab32 (in->retrans_flg);
4730  for (i = 0; i < out->rtr_list_entries; i++) {
4732  out->rtr_list[i].ring_id.seq = swab64 (in->rtr_list[i].ring_id.seq);
4733  out->rtr_list[i].seq = swab32 (in->rtr_list[i].seq);
4734  }
4735 }
4736 
4737 static void mcast_endian_convert (const struct mcast *in, struct mcast *out)
4738 {
4739  out->header.type = in->header.type;
4741  out->header.nodeid = swab32 (in->header.nodeid);
4743 
4744  out->seq = swab32 (in->seq);
4745  out->this_seqno = swab32 (in->this_seqno);
4747  out->ring_id.seq = swab64 (in->ring_id.seq);
4748  out->node_id = swab32 (in->node_id);
4749  out->guarantee = swab32 (in->guarantee);
4750  srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
4751 }
4752 
4753 static void memb_merge_detect_endian_convert (
4754  const struct memb_merge_detect *in,
4755  struct memb_merge_detect *out)
4756 {
4757  out->header.type = in->header.type;
4759  out->header.nodeid = swab32 (in->header.nodeid);
4761  out->ring_id.seq = swab64 (in->ring_id.seq);
4762  srp_addr_copy_endian_convert (&out->system_from, &in->system_from);
4763 }
4764 
4765 static int ignore_join_under_operational (
4766  struct totemsrp_instance *instance,
4767  const struct memb_join *memb_join)
4768 {
4769  struct srp_addr *proc_list;
4770  struct srp_addr *failed_list;
4771  unsigned long long ring_seq;
4772 
4773  proc_list = (struct srp_addr *)memb_join->end_of_memb_join;
4774  failed_list = proc_list + memb_join->proc_list_entries;
4775  ring_seq = memb_join->ring_seq;
4776 
4777  if (memb_set_subset (&instance->my_id, 1,
4778  failed_list, memb_join->failed_list_entries)) {
4779  return (1);
4780  }
4781 
4782  /*
4783  * In operational state, my_proc_list is exactly the same as
4784  * my_memb_list.
4785  */
4786  if ((memb_set_subset (&memb_join->system_from, 1,
4787  instance->my_memb_list, instance->my_memb_entries)) &&
4788  (ring_seq < instance->my_ring_id.seq)) {
4789  return (1);
4790  }
4791 
4792  return (0);
4793 }
4794 
4795 static int message_handler_memb_join (
4796  struct totemsrp_instance *instance,
4797  const void *msg,
4798  size_t msg_len,
4799  int endian_conversion_needed)
4800 {
4801  const struct memb_join *memb_join;
4802  struct memb_join *memb_join_convert = alloca (msg_len);
4803 
4804  if (check_memb_join_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4805  return (0);
4806  }
4807 
4808  if (endian_conversion_needed) {
4809  memb_join = memb_join_convert;
4810  memb_join_endian_convert (msg, memb_join_convert);
4811 
4812  } else {
4813  memb_join = msg;
4814  }
4815  /*
4816  * If the process paused because it wasn't scheduled in a timely
4817  * fashion, flush the join messages because they may be queued
4818  * entries
4819  */
4820  if (pause_flush (instance)) {
4821  return (0);
4822  }
4823 
4824  if (instance->token_ring_id_seq < memb_join->ring_seq) {
4825  instance->token_ring_id_seq = memb_join->ring_seq;
4826  }
4827  switch (instance->memb_state) {
4829  if (!ignore_join_under_operational (instance, memb_join)) {
4830  memb_join_process (instance, memb_join);
4831  }
4832  break;
4833 
4834  case MEMB_STATE_GATHER:
4835  memb_join_process (instance, memb_join);
4836  break;
4837 
4838  case MEMB_STATE_COMMIT:
4839  if (memb_set_subset (&memb_join->system_from,
4840  1,
4841  instance->my_new_memb_list,
4842  instance->my_new_memb_entries) &&
4843 
4844  memb_join->ring_seq >= instance->my_ring_id.seq) {
4845 
4846  memb_join_process (instance, memb_join);
4847  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_JOIN_DURING_COMMIT_STATE);
4848  }
4849  break;
4850 
4851  case MEMB_STATE_RECOVERY:
4852  if (memb_set_subset (&memb_join->system_from,
4853  1,
4854  instance->my_new_memb_list,
4855  instance->my_new_memb_entries) &&
4856 
4857  memb_join->ring_seq >= instance->my_ring_id.seq) {
4858 
4859  memb_join_process (instance, memb_join);
4860  memb_recovery_state_token_loss (instance);
4861  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_JOIN_DURING_RECOVERY);
4862  }
4863  break;
4864  }
4865  return (0);
4866 }
4867 
4868 static int message_handler_memb_commit_token (
4869  struct totemsrp_instance *instance,
4870  const void *msg,
4871  size_t msg_len,
4872  int endian_conversion_needed)
4873 {
4874  struct memb_commit_token *memb_commit_token_convert = alloca (msg_len);
4875  struct memb_commit_token *memb_commit_token;
4876  struct srp_addr sub[PROCESSOR_COUNT_MAX];
4877  int sub_entries;
4878 
4879  struct srp_addr *addr;
4880 
4882  "got commit token");
4883 
4884  if (check_memb_commit_token_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4885  return (0);
4886  }
4887 
4888  if (endian_conversion_needed) {
4889  memb_commit_token_endian_convert (msg, memb_commit_token_convert);
4890  } else {
4891  memcpy (memb_commit_token_convert, msg, msg_len);
4892  }
4893  memb_commit_token = memb_commit_token_convert;
4894  addr = (struct srp_addr *)memb_commit_token->end_of_commit_token;
4895 
4896 #ifdef TEST_DROP_COMMIT_TOKEN_PERCENTAGE
4897  if (random()%100 < TEST_DROP_COMMIT_TOKEN_PERCENTAGE) {
4898  return (0);
4899  }
4900 #endif
4901  switch (instance->memb_state) {
4903  /* discard token */
4904  break;
4905 
4906  case MEMB_STATE_GATHER:
4907  memb_set_subtract (sub, &sub_entries,
4908  instance->my_proc_list, instance->my_proc_list_entries,
4909  instance->my_failed_list, instance->my_failed_list_entries);
4910 
4911  if (memb_set_equal (addr,
4912  memb_commit_token->addr_entries,
4913  sub,
4914  sub_entries) &&
4915 
4916  memb_commit_token->ring_id.seq > instance->my_ring_id.seq) {
4917  memcpy (instance->commit_token, memb_commit_token, msg_len);
4918  memb_state_commit_enter (instance);
4919  }
4920  break;
4921 
4922  case MEMB_STATE_COMMIT:
4923  /*
4924  * If retransmitted commit tokens are sent on this ring
4925  * filter them out and only enter recovery once the
4926  * commit token has traversed the array. This is
4927  * determined by :
4928  * memb_commit_token->memb_index == memb_commit_token->addr_entries) {
4929  */
4930  if (memb_commit_token->ring_id.seq == instance->my_ring_id.seq &&
4931  memb_commit_token->memb_index == memb_commit_token->addr_entries) {
4932  memb_state_recovery_enter (instance, memb_commit_token);
4933  }
4934  break;
4935 
4936  case MEMB_STATE_RECOVERY:
4937  if (totemip_equal (&instance->my_id.addr[0], &instance->my_ring_id.rep)) {
4938 
4939  /* Filter out duplicated tokens */
4940  if (instance->originated_orf_token) {
4941  break;
4942  }
4943 
4944  instance->originated_orf_token = 1;
4945 
4947  "Sending initial ORF token");
4948 
4949  // TODO convert instead of initiate
4950  orf_token_send_initial (instance);
4951  reset_token_timeout (instance); // REVIEWED
4952  reset_token_retransmit_timeout (instance); // REVIEWED
4953  }
4954  break;
4955  }
4956  return (0);
4957 }
4958 
4959 static int message_handler_token_hold_cancel (
4960  struct totemsrp_instance *instance,
4961  const void *msg,
4962  size_t msg_len,
4963  int endian_conversion_needed)
4964 {
4965  const struct token_hold_cancel *token_hold_cancel = msg;
4966 
4967  if (check_token_hold_cancel_sanity(instance, msg, msg_len, endian_conversion_needed) == -1) {
4968  return (0);
4969  }
4970 
4971  if (memcmp (&token_hold_cancel->ring_id, &instance->my_ring_id,
4972  sizeof (struct memb_ring_id)) == 0) {
4973 
4974  instance->my_seq_unchanged = 0;
4975  if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) {
4976  timer_function_token_retransmit_timeout (instance);
4977  }
4978  }
4979  return (0);
4980 }
4981 
4983  void *context,
4984  const void *msg,
4985  unsigned int msg_len)
4986 {
4987  struct totemsrp_instance *instance = context;
4988  const struct message_header *message_header = msg;
4989 
4990  if (msg_len < sizeof (struct message_header)) {
4992  "Received message is too short... ignoring %u.",
4993  (unsigned int)msg_len);
4994  return;
4995  }
4996 
4997 
4998  switch (message_header->type) {
5000  instance->stats.orf_token_rx++;
5001  break;
5002  case MESSAGE_TYPE_MCAST:
5003  instance->stats.mcast_rx++;
5004  break;
5006  instance->stats.memb_merge_detect_rx++;
5007  break;
5009  instance->stats.memb_join_rx++;
5010  break;
5012  instance->stats.memb_commit_token_rx++;
5013  break;
5015  instance->stats.token_hold_cancel_rx++;
5016  break;
5017  default:
5018  log_printf (instance->totemsrp_log_level_security, "Type of received message is wrong... ignoring %d.\n", (int)message_header->type);
5019 printf ("wrong message type\n");
5020  instance->stats.rx_msg_dropped++;
5021  return;
5022  }
5023  /*
5024  * Handle incoming message
5025  */
5026  totemsrp_message_handlers.handler_functions[(int)message_header->type] (
5027  instance,
5028  msg,
5029  msg_len,
5030  message_header->endian_detector != ENDIAN_LOCAL);
5031 }
5032 
5034  void *context,
5035  const struct totem_ip_address *iface_addr,
5036  unsigned int iface_no)
5037 {
5038  struct totemsrp_instance *instance = context;
5039  int i;
5040 
5041  totemip_copy (&instance->my_id.addr[iface_no], iface_addr);
5042  assert (instance->my_id.addr[iface_no].nodeid);
5043 
5044  totemip_copy (&instance->my_memb_list[0].addr[iface_no], iface_addr);
5045 
5046  if (instance->iface_changes++ == 0) {
5047  instance->memb_ring_id_create_or_load (&instance->my_ring_id,
5048  &instance->my_id.addr[0]);
5049  instance->token_ring_id_seq = instance->my_ring_id.seq;
5050  log_printf (
5051  instance->totemsrp_log_level_debug,
5052  "Created or loaded sequence id %llx.%s for this ring.",
5053  instance->my_ring_id.seq,
5054  totemip_print (&instance->my_ring_id.rep));
5055 
5056  if (instance->totemsrp_service_ready_fn) {
5057  instance->totemsrp_service_ready_fn ();
5058  }
5059 
5060  }
5061 
5062  for (i = 0; i < instance->totem_config->interfaces[iface_no].member_count; i++) {
5063  totemsrp_member_add (instance,
5064  &instance->totem_config->interfaces[iface_no].member_list[i],
5065  iface_no);
5066  }
5067 
5068  if (instance->iface_changes >= instance->totem_config->interface_count) {
5069  memb_state_gather_enter (instance, TOTEMSRP_GSFROM_INTERFACE_CHANGE);
5070  }
5071 }
5072 
5073 void totemsrp_net_mtu_adjust (struct totem_config *totem_config) {
5074  totem_config->net_mtu -= sizeof (struct mcast);
5075 }
5076 
5078  void *context,
5079  void (*totem_service_ready) (void))
5080 {
5081  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5082 
5083  instance->totemsrp_service_ready_fn = totem_service_ready;
5084 }
5085 
5087  void *context,
5088  const struct totem_ip_address *member,
5089  int ring_no)
5090 {
5091  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5092  int res;
5093 
5094  res = totemrrp_member_add (instance->totemrrp_context, member, ring_no);
5095 
5096  return (res);
5097 }
5098 
5100  void *context,
5101  const struct totem_ip_address *member,
5102  int ring_no)
5103 {
5104  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5105  int res;
5106 
5107  res = totemrrp_member_remove (instance->totemrrp_context, member, ring_no);
5108 
5109  return (res);
5110 }
5111 
5112 void totemsrp_threaded_mode_enable (void *context)
5113 {
5114  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5115 
5116  instance->threaded_mode_enabled = 1;
5117 }
5118 
5119 void totemsrp_trans_ack (void *context)
5120 {
5121  struct totemsrp_instance *instance = (struct totemsrp_instance *)context;
5122 
5123  instance->waiting_trans_ack = 0;
5124  instance->totemsrp_waiting_trans_ack_cb_fn (0);
5125 }
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:97
char encapsulated
Definition: totemsrp.c:61
gather_state_from
Definition: totemsrp.c:537
int totemrrp_iface_check(void *rrp_context)
Definition: totemrrp.c:2216
void main_iface_change_fn(void *context, const struct totem_ip_address *iface_address, unsigned int iface_no)
Definition: totemsrp.c:5033
void totemip_copy_endian_convert(struct totem_ip_address *addr1, const struct totem_ip_address *addr2)
Definition: totemip.c:101
struct srp_addr system_from
Definition: totemsrp.c:218
#define ENDIAN_LOCAL
Definition: totemsrp.c:137
uint64_t gather_entered
Definition: totem.h:260
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:255
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:114
unsigned int interface_count
Definition: totem.h:115
int totemsrp_my_family_get(void *srp_context)
Definition: totemsrp.c:1145
struct list_head * next
Definition: list.h:47
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
unsigned int seq
Definition: totemsrp.c:62
totemsrp_token_stats_t token[TOTEM_TOKEN_STATS_MAX]
Definition: totem.h:274
const char * totemip_print(const struct totem_ip_address *addr)
Definition: totemip.c:214
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
unsigned char end_of_memb_join[0]
Definition: totemsrp.c:65
uint64_t mcast_rx
Definition: totem.h:253
unsigned long long int tv_old
Definition: totemsrp.c:3897
#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:133
int member_count
Definition: totem.h:70
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:71
int totemip_compare(const void *a, const void *b)
Definition: totemip.c:130
int totemsrp_member_add(void *context, const struct totem_ip_address *member, int ring_no)
Definition: totemsrp.c:5086
void * token_sent_event_handle
Definition: totemsrp.c:524
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
Definition: coroapi.h:110
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:1134
unsigned int my_pbl
Definition: totemsrp.c:503
char rrp_mode[TOTEM_RRP_MODE_BYTES]
Definition: totem.h:161
void totemsrp_net_mtu_adjust(struct totem_config *totem_config)
Definition: totemsrp.c:5073
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:1120
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:2319
unsigned int my_aru
Definition: totemsrp.c:381
struct message_header header
Definition: totemsrp.c:60
uint64_t memb_merge_detect_rx
Definition: totem.h:248
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:1062
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
char commit_token_storage[40000]
Definition: totemsrp.c:525
unsigned int rrp_problem_count_timeout
Definition: totem.h:153
struct list_head token_callback_sent_listhead
Definition: totemsrp.c:387
Definition: sq.h:40
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:271
struct srp_addr my_deliver_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:315
uint64_t orf_token_tx
Definition: totem.h:245
void totemsrp_callback_token_destroy(void *srp_context, void **handle_out)
Definition: totemsrp.c:3467
uint64_t gather_token_lost
Definition: totem.h:261
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:95
int totemrrp_ifaces_get(void *rrp_context, char ***status, unsigned int *iface_count)
Definition: totemrrp.c:2225
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:2118
unsigned int downcheck_timeout
Definition: totem.h:145
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:254
Definition: list.h:46
int my_deliver_memb_entries
Definition: totemsrp.c:331
unsigned int max_network_delay
Definition: totem.h:171
unsigned int heartbeat_failures_allowed
Definition: totem.h:169
#define TOTEM_TOKEN_STATS_MAX
Definition: totem.h:273
unsigned int my_last_seq
Definition: totemsrp.c:491
int my_left_memb_entries
Definition: totemsrp.c:333
#define swab64(x)
Definition: swab.h:52
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
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:3432
unsigned int send_join_timeout
Definition: totem.h:139
unsigned int window_size
Definition: totem.h:173
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:5077
unsigned int rrp_problem_count_threshold
Definition: totem.h:155
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:258
void(*) in log_level_security)
Definition: totem.h:82
unsigned long long ring_seq
Definition: totemsrp.c:221
#define INTERFACE_MAX
Definition: coroapi.h:75
int totemsrp_mcast(void *srp_context, struct iovec *iovec, unsigned int iov_len, int guarantee)
Multicast a message.
Definition: totemsrp.c:2444
message_type
Definition: totemsrp.c:139
int latest_token
Definition: totem.h:272
uint64_t operational_token_lost
Definition: totem.h:259
unsigned int received_flg
Definition: totemsrp.c:63
uint64_t consensus_timeouts
Definition: totem.h:266
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:265
unsigned int backlog
Definition: totemsrp.c:66
int this_seqno
Definition: totemsrp.c:188
unsigned char end_of_memb_join[0]
Definition: totemsrp.c:222
unsigned int token_retransmits_before_loss_const
Definition: totem.h:135
struct message_header header
Definition: totemsrp.c:239
int totemrrp_finalize(void *rrp_context)
Definition: totemrrp.c:1974
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:2306
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:1157
struct memb_ring_id ring_id
Definition: totemsrp.c:189
unsigned int seqno_unchanged_const
Definition: totem.h:149
uint64_t commit_token_lost
Definition: totem.h:263
unsigned int miss_count_const
Definition: totem.h:187
int totemrrp_crypto_set(void *rrp_context, const char *cipher_type, const char *hash_type)
Definition: totemrrp.c:2240
uint64_t token_hold_cancel_rx
Definition: totem.h:257
unsigned int join_timeout
Definition: totem.h:137
unsigned int aru
Definition: totemsrp.c:246
uint32_t originated_orf_token
Definition: totemsrp.c:515
unsigned int nodeid
Definition: coroapi.h:96
int totemrrp_send_flush(void *rrp_context)
Definition: totemrrp.c:2163
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:67
char encapsulated
Definition: totemrrp.c:554
#define MESSAGE_QUEUE_MAX
Definition: coroapi.h:85
int totemrrp_member_add(void *rrp_context, const struct totem_ip_address *member, int iface_no)
Definition: totemrrp.c:2293
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:104
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:2513
unsigned int rrp_autorecovery_check_timeout
Definition: totem.h:159
uint64_t mcast_retx
Definition: totem.h:252
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:147
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:66
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:2003
qb_loop_t * totemsrp_poll_handle
Definition: totemsrp.c:445
totem_event_type
Definition: totem.h:212
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:267
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:75
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)
Definition: swab.h:43
qb_loop_timer_handle timer_orf_token_timeout
Definition: totemsrp.c:400
uint32_t continuous_gather
Definition: totem.h:268
void totemsrp_threaded_mode_enable(void *context)
Definition: totemsrp.c:5112
unsigned int aru
Definition: totemsrp.c:63
encapsulation_type
Definition: totemsrp.c:148
unsigned int net_mtu
Definition: totem.h:165
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:2435
unsigned int node_id
Definition: totemsrp.c:190
int totemrrp_recv_flush(void *rrp_context)
Definition: totemrrp.c:2154
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:256
void(* totem_memb_ring_id_create_or_load)(struct memb_ring_id *memb_ring_id, const struct totem_ip_address *addr)
Definition: totem.h:191
unsigned int token_timeout
Definition: totem.h:129
Definition: totemsrp.c:244
unsigned int high_delivered
Definition: totemsrp.c:247
unsigned int consensus_timeout
Definition: totem.h:141
totemsrp_stats_t stats
Definition: totemsrp.c:511
void main_deliver_fn(void *context, const void *msg, unsigned int msg_len)
Definition: totemsrp.c:4982
#define PROCESSOR_COUNT_MAX
Definition: coroapi.h:83
unsigned short endian_detector
Definition: totemsrp.c:62
uint64_t mcast_tx
Definition: totem.h:251
void totemrrp_buffer_release(void *rrp_context, void *ptr)
Definition: totemrrp.c:2125
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
#define SEQNO_START_MSG
Definition: totemsrp.c:114
#define swab16(x)
Definition: swab.h:35
void totemsrp_finalize(void *srp_context)
Definition: totemsrp.c:1038
#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:66
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:247
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:283
int consensus_list_entries
Definition: totemsrp.c:301
unsigned int rrp_problem_count_mcast_threshold
Definition: totem.h:157
int totemrrp_processor_count_set(void *rrp_context, unsigned int processor_count)
Definition: totemrrp.c:2132
char type
Definition: totemsrp.c:60
uint64_t memb_join_rx
Definition: totem.h:250
int totemrrp_mcast_noflush_send(void *rrp_context, const void *msg, unsigned int msg_len)
Definition: totemrrp.c:2196
unsigned char end_of_commit_token[0]
Definition: totemsrp.c:66
#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:2282
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:163
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:2182
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:105
void totemsrp_trans_ack(void *context)
Definition: totemsrp.c:5119
unsigned int max_messages
Definition: totem.h:175
uint64_t recovery_entered
Definition: totem.h:264
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 system_from
Definition: totemsrp.c:61
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:143
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:5099
unsigned int token_retransmit_timeout
Definition: totem.h:131
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
void(* totemsrp_log_printf)(int level, int sybsys, const char *function, const char *file, int line, const char *format,...) __attribute__((format(printf
Definition: totemsrp.c:433
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:71
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:2259
unsigned int my_leave_memb_list[PROCESSOR_COUNT_MAX]
Definition: totemsrp.c:319
uint64_t commit_entered
Definition: totem.h:262
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:151
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:246
unsigned int nodeid
Definition: totemsrp.c:180
int totemrrp_token_send(void *rrp_context, const void *msg, unsigned int msg_len)
Definition: totemrrp.c:2171
unsigned int threads
Definition: totem.h:167
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:195
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:2144
totem_callback_token_type
Definition: coroapi.h:117
unsigned int my_high_ring_delivered
Definition: totemsrp.c:361
unsigned int fcc
Definition: totemsrp.c:209